当前位置：首页 > news >正文

墨语灵犀自动化办公实战：Python脚本批量处理文档与邮件

news 2026/7/31 21:10:07

墨语灵犀自动化办公实战：Python脚本批量处理文档与邮件

你是不是也经常被这些重复性的办公任务搞得焦头烂额？每天要打开几十个Excel表格，手动汇总数据写报告；收到一堆格式五花八门的Word文档，得一个个调整格式、提取关键信息；邮箱里塞满了各种邮件，光是分类和回复就要花掉大半天时间。

这些工作不仅枯燥，还特别容易出错。今天，我就来分享一套结合“墨语灵犀”大模型和Python脚本的自动化办公方案，让你彻底告别这些繁琐的手工操作。这套方案的核心思路很简单：让Python负责“动手”（读取文件、调用接口），让墨语灵犀负责“动脑”（理解内容、生成文本），两者结合，效率直接起飞。

接下来，我会通过三个具体的实战案例，手把手带你搭建这套自动化系统。你会发现，原来那些让人头疼的办公任务，现在只需要运行一个脚本就能轻松搞定。

1. 案例一：自动读取Excel并生成分析报告摘要

想象一下这个场景：每周一早上，你都要打开销售部门发来的十几个Excel文件，手动计算本周的销售额、增长率、Top 10产品，然后把这些数据整理成一份分析报告摘要，发给老板。这个过程至少要花掉你两个小时。

现在，我们可以用Python脚本自动完成这一切。

1.1 准备工作与环境搭建

首先，你需要安装几个必要的Python库。打开你的命令行工具，输入以下命令：

pip install pandas openpyxl requests

pandas：用来读取和处理Excel数据的神器。
openpyxl：专门处理.xlsx格式的Excel文件。
requests：用来和墨语灵犀的API进行通信。

假设你已经部署好了墨语灵犀服务，并且知道它的API地址（比如http://localhost:8000/v1/chat/completions）和API密钥。如果没有，你需要先完成部署，这部分网上有很多教程，这里就不展开了。

1.2 编写数据读取与处理脚本

我们先写一个脚本来读取Excel文件。假设我们有一个名为weekly_sales.xlsx的文件，里面有一个叫SalesData的工作表。

import pandas as pd import requests import json # 1. 读取Excel文件 def read_excel_data(file_path, sheet_name='SalesData'): """ 读取指定Excel文件和工作表的数据。 """ try: df = pd.read_excel(file_path, sheet_name=sheet_name) print(f"成功读取文件: {file_path}") print(f"数据形状: {df.shape}") # 显示行数和列数 return df except Exception as e: print(f"读取文件时出错: {e}") return None # 2. 对数据进行简单的统计分析 def analyze_sales_data(df): """ 对销售数据进行基础分析，生成统计字符串。 """ if df is None or df.empty: return "数据为空，无法进行分析。" # 确保必要的列存在 required_columns = ['产品', '销售额', '日期'] for col in required_columns: if col not in df.columns: return f"数据中缺少必要的列: {col}" # 进行一些基础计算 total_sales = df['销售额'].sum() avg_sales = df['销售额'].mean() top_product = df.loc[df['销售额'].idxmax(), '产品'] sales_by_date = df.groupby('日期')['销售额'].sum().to_dict() # 将分析结果组织成一段文本 analysis_text = f""" 本周销售数据概览： - 总销售额：{total_sales:,.2f} 元 - 平均每单销售额：{avg_sales:,.2f} 元 - 最畅销产品：{top_product} - 每日销售额趋势：{sales_by_date} """ return analysis_text # 测试读取和分析 if __name__ == "__main__": file_path = "weekly_sales.xlsx" sales_df = read_excel_data(file_path) if sales_df is not None: print(sales_df.head()) # 预览前几行数据 analysis_result = analyze_sales_data(sales_df) print(analysis_result)

运行这个脚本，你应该能看到Excel文件的内容和基本的统计结果。这只是第一步，我们得到了结构化的数据摘要。

1.3 调用墨语灵犀生成报告摘要

接下来是魔法时刻：我们把上面生成的analysis_text交给墨语灵犀，让它写一份有洞察力的报告摘要。

# 3. 调用墨语灵犀API生成分析报告 def generate_report_with_llm(analysis_text, api_url, api_key): """ 将数据分析文本发送给墨语灵犀，让其生成一份正式的报告摘要。 """ # 构建提示词（Prompt） prompt = f""" 你是一位专业的商业分析师。请根据以下提供的本周销售数据统计信息，撰写一份简洁、专业、有洞察力的销售分析报告摘要。 报告需要面向公司管理层，突出关键发现、主要趋势和潜在问题。 数据统计信息如下： {analysis_text} 请生成报告摘要。 """ # 准备请求数据 headers = { "Content-Type": "application/json", "Authorization": f"Bearer {api_key}" } data = { "model": "moyu-lingxi", # 根据你的模型名称调整 "messages": [ {"role": "user", "content": prompt} ], "temperature": 0.7, # 控制创造性，0.7比较平衡 "max_tokens": 500 # 生成文本的最大长度 } try: response = requests.post(api_url, headers=headers, data=json.dumps(data)) response.raise_for_status() # 检查请求是否成功 result = response.json() # 提取生成的报告内容 report_content = result['choices'][0]['message']['content'] return report_content.strip() except requests.exceptions.RequestException as e: print(f"API请求失败: {e}") return None except (KeyError, IndexError) as e: print(f"解析API响应时出错: {e}") return None # 主函数：串联整个流程 def main(): # 你的配置信息 EXCEL_FILE = "weekly_sales.xlsx" API_URL = "http://localhost:8000/v1/chat/completions" # 替换为你的API地址 API_KEY = "your-api-key-here" # 替换为你的API密钥 print("开始自动化销售报告生成流程...") print("-" * 50) # 步骤1: 读取数据 df = read_excel_data(EXCEL_FILE) if df is None: print("流程终止：无法读取数据。") return # 步骤2: 分析数据 print("\n正在分析销售数据...") analysis_text = analyze_sales_data(df) print("基础数据分析完成。") # 步骤3: 生成报告 print("\n正在调用墨语灵犀生成报告摘要...") final_report = generate_report_with_llm(analysis_text, API_URL, API_KEY) if final_report: print("\n" + "="*50) print("生成的销售分析报告摘要：") print("="*50) print(final_report) print("="*50) # 可选：将报告保存到文件 with open("weekly_sales_report.txt", "w", encoding="utf-8") as f: f.write(final_report) print("\n报告已保存至 'weekly_sales_report.txt'") else: print("报告生成失败。") if __name__ == "__main__": main()

把这个脚本保存为auto_sales_report.py，替换里面的API_URL和API_KEY，然后运行它。几分钟后，一份结构清晰、语言专业的销售分析报告摘要就诞生了。原本需要两小时的工作，现在不到一分钟就完成了。

2. 案例二：批量处理Word文档格式与信息提取

第二个常见痛点是处理大量Word文档。比如，法务部门发来一堆合同草案，格式乱七八糟，你需要统一字体、段落间距，还要把里面的“甲方”、“乙方”、“合同金额”和“签署日期”这些关键信息摘出来，整理成表格。

手动操作？一份合同就得折腾十几分钟。我们用Python+墨语灵犀来批量处理。

2.1 使用python-docx处理Word文档

首先，安装处理Word的库：

pip install python-docx

我们先写一个函数来统一文档格式。假设公司标准格式是：标题用黑体三号，正文用宋体小四，段落行距1.5倍。

from docx import Document from docx.shared import Pt, RGBColor from docx.enum.text import WD_ALIGN_PARAGRAPH import os def standardize_word_format(doc_path, output_path): """ 将Word文档格式标准化。 """ try: doc = Document(doc_path) # 遍历所有段落，应用格式 for paragraph in doc.paragraphs: # 根据段落样式判断是否是标题 if paragraph.style.name.startswith('Heading'): # 标题格式：黑体，三号，加粗，居中 run = paragraph.runs[0] if paragraph.runs else paragraph.add_run() run.font.name = '黑体' run.font.size = Pt(16) # 三号约16磅 run.font.bold = True paragraph.alignment = WD_ALIGN_PARAGRAPH.CENTER else: # 正文格式：宋体，小四，黑色 for run in paragraph.runs: run.font.name = '宋体' run.font.size = Pt(12) # 小四约12磅 run.font.color.rgb = RGBColor(0, 0, 0) # 黑色 # 设置段落行距为1.5倍 paragraph.paragraph_format.line_spacing = 1.5 # 保存标准化后的文档 doc.save(output_path) print(f"文档已标准化并保存至: {output_path}") return True except Exception as e: print(f"处理文档 {doc_path} 时出错: {e}") return False

这个函数能搞定格式统一。接下来，我们需要从文档里提取关键信息。单纯用规则提取（比如找“甲方：”后面的文字）在复杂文档里容易出错。这时，就该墨语灵犀上场了。

2.2 提取文档内容并发送给墨语灵犀

我们先从Word里把文字读出来，然后让墨语灵犀帮我们识别和提取。

def extract_text_from_docx(doc_path): """ 从Word文档中提取纯文本内容。 """ try: doc = Document(doc_path) full_text = [] for para in doc.paragraphs: full_text.append(para.text) return '\n'.join(full_text) except Exception as e: print(f"从 {doc_path} 提取文本时出错: {e}") return "" def extract_contract_info_with_llm(doc_text, api_url, api_key): """ 使用墨语灵犀从合同文本中提取结构化信息。 """ prompt = f""" 你是一位专业的法务助理。请从以下合同文档内容中，准确提取以下关键信息： 1. 甲方全称 2. 乙方全称 3. 合同总金额（包括货币单位） 4. 合同签署日期 5. 合同有效期 请以JSON格式返回结果，键名使用英文：party_a, party_b, total_amount, signing_date, contract_term。 如果某项信息在文中未找到，其值设为 null。 合同内容如下： {doc_text} """ headers = { "Content-Type": "application/json", "Authorization": f"Bearer {api_key}" } data = { "model": "moyu-lingxi", "messages": [{"role": "user", "content": prompt}], "temperature": 0.1, # 提取信息要求准确，降低创造性 "max_tokens": 300 } try: response = requests.post(api_url, headers=headers, json=data) response.raise_for_status() result = response.json() # 尝试解析返回的JSON reply_content = result['choices'][0]['message']['content'] # 清理回复内容，只提取JSON部分（模型有时会在JSON外加说明文字） import re json_match = re.search(r'\{.*\}', reply_content, re.DOTALL) if json_match: json_str = json_match.group() import json as json_module info_dict = json_module.loads(json_str) return info_dict else: print("未能从回复中解析出JSON。") return None except Exception as e: print(f"调用API提取信息时出错: {e}") return None

2.3 批量处理整个文件夹的文档

现在，我们把格式化和信息提取结合起来，写一个批量处理的脚本。

import os import json def batch_process_word_docs(input_folder, output_folder, api_url, api_key): """ 批量处理一个文件夹内的所有Word文档。 """ # 确保输出文件夹存在 os.makedirs(output_folder, exist_ok=True) # 用于保存所有提取的信息 all_contracts_info = [] # 遍历输入文件夹中的所有.docx文件 for filename in os.listdir(input_folder): if filename.endswith('.docx'): input_path = os.path.join(input_folder, filename) output_path = os.path.join(output_folder, f"standardized_{filename}") print(f"\n正在处理: {filename}") print("-" * 30) # 1. 标准化格式 print("步骤1: 标准化文档格式...") format_success = standardize_word_format(input_path, output_path) # 2. 提取文本 print("步骤2: 提取文档文本内容...") doc_text = extract_text_from_docx(input_path) if not doc_text: print("警告: 文本内容为空，跳过信息提取。") continue # 3. 提取关键信息 print("步骤3: 调用墨语灵犀提取关键信息...") contract_info = extract_contract_info_with_llm(doc_text, api_url, api_key) if contract_info: contract_info['source_file'] = filename all_contracts_info.append(contract_info) print(f"提取成功: {contract_info}") else: print("信息提取失败。") # 将所有提取的信息保存到一个JSON文件中 if all_contracts_info: info_output_path = os.path.join(output_folder, "extracted_contract_info.json") with open(info_output_path, 'w', encoding='utf-8') as f: json.dump(all_contracts_info, f, ensure_ascii=False, indent=2) print(f"\n所有合同信息已汇总保存至: {info_output_path}") return all_contracts_info # 使用示例 if __name__ == "__main__": INPUT_FOLDER = "./raw_contracts" # 存放原始合同的文件夹 OUTPUT_FOLDER = "./processed_contracts" # 处理后的输出文件夹 API_URL = "http://localhost:8000/v1/chat/completions" API_KEY = "your-api-key-here" print("开始批量处理Word文档...") results = batch_process_word_docs(INPUT_FOLDER, OUTPUT_FOLDER, API_URL, API_KEY) print(f"\n处理完成！共处理了 {len(results)} 份文档。")

把这个脚本指向存放合同草案的文件夹，运行一下。它会自动完成三件事：第一，把所有文档的格式统一成公司标准；第二，从每份合同里提取出关键信息；第三，把这些信息整理成一个清晰的JSON文件。原本需要一整天的工作，现在喝杯咖啡的功夫就搞定了。

3. 案例三：智能分析邮箱历史邮件并自动处理

最后一个场景可能更普遍：邮箱管理。每天收到大量邮件，有些需要立即回复，有些可以稍后处理，有些则是订阅的新闻稿。手动分类和回复效率极低。我们来构建一个智能邮件助手。

3.1 使用imaplib和email库读取邮件

Python标准库里的imaplib和email可以帮我们连接邮箱并读取邮件。这里以IMAP协议为例。

import imaplib import email from email.header import decode_header import re def connect_to_mailbox(email_user, email_pass, imap_server='imap.example.com'): """ 连接到IMAP邮箱服务器。 """ try: mail = imaplib.IMAP4_SSL(imap_server) mail.login(email_user, email_pass) print("邮箱登录成功。") return mail except Exception as e: print(f"连接邮箱失败: {e}") return None def fetch_recent_emails(mail, mailbox='INBOX', num_emails=10): """ 获取最近的邮件。 """ try: # 选择邮箱文件夹 mail.select(mailbox) # 搜索所有邮件，按日期倒序排序（最新的在前） status, messages = mail.search(None, 'ALL') if status != 'OK': print("搜索邮件失败。") return [] # 获取邮件ID列表 email_ids = messages[0].split() # 取最近N封 recent_email_ids = email_ids[-num_emails:] if len(email_ids) >= num_emails else email_ids emails_data = [] for e_id in recent_email_ids: # 获取邮件内容 status, msg_data = mail.fetch(e_id, '(RFC822)') if status != 'OK': continue # 解析邮件 raw_email = msg_data[0][1] msg = email.message_from_bytes(raw_email) # 解码主题 subject, encoding = decode_header(msg['Subject'])[0] if isinstance(subject, bytes): subject = subject.decode(encoding if encoding else 'utf-8') # 获取发件人 from_ = msg.get('From') # 提取纯文本正文 body = "" if msg.is_multipart(): for part in msg.walk(): content_type = part.get_content_type() content_disposition = str(part.get("Content-Disposition")) # 跳过附件 if "attachment" in content_disposition: continue if content_type == "text/plain": # 解码正文 try: body_bytes = part.get_payload(decode=True) charset = part.get_content_charset() or 'utf-8' body = body_bytes.decode(charset, errors='ignore') break # 优先取text/plain部分 except: pass else: # 非多部分邮件 content_type = msg.get_content_type() if content_type == "text/plain": try: body_bytes = msg.get_payload(decode=True) charset = msg.get_content_charset() or 'utf-8' body = body_bytes.decode(charset, errors='ignore') except: pass # 简单清理正文（去除过长引用和签名） cleaned_body = clean_email_body(body) emails_data.append({ 'id': e_id.decode(), 'subject': subject, 'from': from_, 'body_preview': cleaned_body[:500] # 只取前500字符用于分析 }) return emails_data except Exception as e: print(f"获取邮件时出错: {e}") return [] def clean_email_body(body): """ 简单清理邮件正文，去除常见的引用块和签名。 """ if not body: return "" # 去除以 `>` 开头的引用行 lines = body.split('\n') cleaned_lines = [] for line in lines: stripped = line.lstrip() if not stripped.startswith('>') and not stripped.startswith('On') and 'wrote:' not in line: cleaned_lines.append(line) cleaned = '\n'.join(cleaned_lines) # 截取到常见签名分隔符 for marker in ['--', '___', 'Best regards', 'Thanks', '签名']: if marker in cleaned: cleaned = cleaned.split(marker)[0] return cleaned.strip()

这段代码能帮你把最近10封邮件的关键信息（主题、发件人、正文预览）提取出来。接下来，我们让墨语灵犀当你的邮件秘书。

3.2 使用墨语灵犀进行邮件分类与回复草稿生成

我们可以设计两个功能：一是自动判断邮件的紧急程度和类别；二是为需要回复的邮件生成回复草稿。

def categorize_email_with_llm(email_data, api_url, api_key): """ 使用墨语灵犀对邮件进行分类。 """ prompt = f""" 请分析以下邮件内容，并对其进行分类。 邮件主题：{email_data['subject']} 发件人：{email_data['from']} 邮件正文预览：{email_data['body_preview']} 请从以下类别中选择最合适的一个： 1. 紧急待办 - 需要今天内回复或处理的重要邮件（如客户投诉、老板指令、紧急会议） 2. 常规工作 - 与当前项目或常规工作相关的邮件（如项目讨论、任务分配、进度汇报） 3. 订阅通知 - 新闻稿、促销信息、系统通知等无需立即处理的邮件 4. 社交联络 - 同事或朋友的问候、邀请、非工作闲聊等 5. 未知/其他 - 无法明确分类的邮件 同时，请判断这封邮件是否需要回复（是/否）。 请以JSON格式返回，包含两个键：category 和 need_reply。 """ headers = {"Content-Type": "application/json", "Authorization": f"Bearer {api_key}"} data = { "model": "moyu-lingxi", "messages": [{"role": "user", "content": prompt}], "temperature": 0.1, "max_tokens": 150 } try: response = requests.post(api_url, headers=headers, json=data) response.raise_for_status() result = response.json() reply = result['choices'][0]['message']['content'] # 提取JSON import re json_match = re.search(r'\{.*\}', reply, re.DOTALL) if json_match: import json as json_module category_info = json_module.loads(json_match.group()) return category_info except Exception as e: print(f"邮件分类失败: {e}") return {"category": "未知/其他", "need_reply": "否"} def generate_reply_draft_with_llm(email_data, api_url, api_key): """ 使用墨语灵犀生成邮件回复草稿。 """ prompt = f""" 你是一位专业的职场人士。请根据以下收到的邮件，撰写一封礼貌、专业、得体的回复草稿。 回复语言应与来邮件语言一致（中文来邮则用中文回复）。 收到的邮件信息： 主题：{email_data['subject']} 发件人：{email_data['from']} 正文：{email_data['body_preview']} 请生成回复草稿。如果邮件内容不完整或无法判断如何回复，请说明“需要更多信息”。 """ headers = {"Content-Type": "application/json", "Authorization": f"Bearer {api_key}"} data = { "model": "moyu-lingxi", "messages": [{"role": "user", "content": prompt}], "temperature": 0.7, "max_tokens": 300 } try: response = requests.post(api_url, headers=headers, json=data) response.raise_for_status() result = response.json() draft = result['choices'][0]['message']['content'] return draft.strip() except Exception as e: print(f"生成回复草稿失败: {e}") return "抱歉，未能生成回复草稿。"

3.3 构建完整的邮件处理工作流

最后，我们把所有功能串起来，创建一个完整的邮件处理脚本。

def process_email_inbox(email_user, email_pass, api_url, api_key, imap_server='imap.example.com'): """ 完整的邮件处理工作流：连接邮箱、获取邮件、分类、生成回复草稿。 """ print("开始处理邮箱...") # 1. 连接邮箱 mail = connect_to_mailbox(email_user, email_pass, imap_server) if not mail: return # 2. 获取最近邮件 print("获取最近邮件...") recent_emails = fetch_recent_emails(mail, num_emails=15) print(f"共获取到 {len(recent_emails)} 封邮件。") # 3. 对每封邮件进行分类和处理 processed_results = [] for idx, email_data in enumerate(recent_emails, 1): print(f"\n处理第 {idx} 封邮件: {email_data['subject'][:50]}...") # 分类 print(" 正在分类...") category_info = categorize_email_with_llm(email_data, api_url, api_key) email_data.update(category_info) # 如果需要回复，则生成草稿 if category_info.get('need_reply') == '是': print(" 需要回复，正在生成草稿...") reply_draft = generate_reply_draft_with_llm(email_data, api_url, api_key) email_data['reply_draft'] = reply_draft else: email_data['reply_draft'] = None processed_results.append(email_data) # 简单打印结果 print(f" 分类: {category_info.get('category')}, 需回复: {category_info.get('need_reply')}") if email_data['reply_draft']: print(f" 回复草稿预览: {email_data['reply_draft'][:100]}...") # 4. 断开连接 mail.logout() # 5. 输出处理摘要 print("\n" + "="*60) print("邮件处理摘要") print("="*60) from collections import Counter category_counter = Counter([e.get('category', '未知') for e in processed_results]) need_reply_count = sum(1 for e in processed_results if e.get('need_reply') == '是') print(f"共处理邮件: {len(processed_results)} 封") print("分类统计:") for category, count in category_counter.items(): print(f" {category}: {count} 封") print(f"需要回复的邮件: {need_reply_count} 封") # 6. 将结果保存为JSON文件，方便后续查看 import json from datetime import datetime timestamp = datetime.now().strftime("%Y%m%d_%H%M%S") output_filename = f"email_processing_report_{timestamp}.json" with open(output_filename, 'w', encoding='utf-8') as f: json.dump(processed_results, f, ensure_ascii=False, indent=2) print(f"\n详细处理结果已保存至: {output_filename}") # 7. 特别提示紧急邮件 urgent_emails = [e for e in processed_results if e.get('category') == '紧急待办'] if urgent_emails: print("\n⚠️ 紧急待办邮件提醒:") for email in urgent_emails: print(f" - {email['subject']} (来自: {email['from']})") return processed_results # 使用示例（注意：需要替换为真实的邮箱信息和API信息） if __name__ == "__main__": # 你的邮箱配置（出于安全考虑，建议从环境变量或配置文件中读取） EMAIL_USER = "your-email@example.com" EMAIL_PASSWORD = "your-email-password-or-app-specific-password" IMAP_SERVER = "imap.example.com" # 例如：imap.qq.com, imap.gmail.com # 墨语灵犀API配置 API_URL = "http://localhost:8000/v1/chat/completions" API_KEY = "your-api-key-here" # 运行处理流程 results = process_email_inbox(EMAIL_USER, EMAIL_PASSWORD, API_URL, API_KEY, IMAP_SERVER)

运行这个脚本，它会自动登录你的邮箱，分析最近的邮件，给每封邮件打上分类标签，并为需要回复的邮件生成回复草稿。最后，它会给你一份清晰的摘要报告，并特别提醒你有哪些紧急邮件需要优先处理。你只需要检查一下草稿，稍作修改就可以发送，省去了大量阅读、思考和打字的时间。