Python周报自动化闭环：数据库+Dify+python-docx+SMTP

简介这是一份面向具备Python基础的中初级研发人员与技术管理者的办公自动化实践方案聚焦解决周报编写耗时重复的痛点提供从数据库拉取、Dify智能分析、Word文档生成到邮件自动发送的端到端实现。资源为单文件PDF文档179KB完整呈现系统模块化设计逻辑、核心代码实现含DataFetcher数据获取、DifyAnalyzer调用封装、python-docx模板填充、smtplib邮件发送及关键细节如时间范围计算、异常处理建议与定时调度集成。内容预览显示其覆盖环境配置、四大模块分步实现、SQL查询构造、Dify API请求封装与Prompt工程设计特别强调模块间数据传递与生产级健壮性考量。目前已有293人学习下载适合希望将AI能力嵌入业务流程、掌握Python在数据—文档—通信全链路自动化应用的技术团队快速落地实践。1. 周报自动化不是“写个脚本发邮件”而是打通数据、分析、文档与分发的闭环链路很多团队还在用 Excel 手动汇总各系统数据、复制粘贴到 Word 模板、再逐条核对关键指标、最后手动点开 Outlook 发送——这个过程平均耗时 2.7 小时/人/周据 2024 年《国内中型企业办公效率调研》抽样统计。而真正有效的周报自动化核心不在“自动发邮件”而在让数据源头可信、分析逻辑可复用、文档生成可验证、分发动作可追溯。本方案聚焦 Python 生态下可落地的最小闭环从 MySQL/PostgreSQL 等关系型数据库直接拉取结构化业务数据非 CSV 导入接入 Dify 作为智能分析层处理自然语言查询与摘要生成非调用通用大模型 API用 python-docx 精确控制 Word 格式避开 COM 接口导致的关闭卡顿、宏安全警告等 Windows 特有顽疾并通过 SMTP 协议原生发送带附件的正式邮件不依赖 Outlook 客户端。适合已有数据库权限、能部署轻量 Dify 实例、且需交付可审计 Word 文档的技术运营、数据分析或项目管理岗位——它不追求“全自动无人值守”而是把人工干预点压缩到仅剩“确认分析结论”和“点击发送”两个动作。2. 数据库拉取用 SQLAlchemy pandas 构建健壮、可参数化的查询管道周报数据源往往分散在多个表甚至多个库中硬编码 SQL 易出错、难维护。本环节采用 SQLAlchemy ORM Core 混合模式兼顾可读性与灵活性同时规避 raw SQL 注入风险。2.1 建立带连接池与超时控制的数据库会话# db_config.py from sqlalchemy import create_engine from sqlalchemy.orm import sessionmaker from urllib.parse import quote_plus def get_db_engine(db_typemysql, hostlocalhost, port3306, databasereport_db, usernamereader, password): 构建带连接池与超时的 SQLAlchemy 引擎 支持 mysql/postgresql密码含特殊字符时自动 URL 编码 if db_type mysql: url fmysqlpymysql://{username}:{quote_plus(password)}{host}:{port}/{database} elif db_type postgresql: url fpostgresqlpsycopg2://{username}:{quote_plus(password)}{host}:{port}/{database} else: raise ValueError(仅支持 mysql 或 postgresql) return create_engine( url, pool_size5, # 连接池大小 max_overflow10, # 超出池大小后允许的最大额外连接数 pool_timeout30, # 获取连接超时秒 pool_recycle3600, # 连接回收时间秒避免 MySQL 默认 8 小时断连 echoFalse # 生产环境关闭 SQL 日志 ) # 初始化引擎全局单例 engine get_db_engine( db_typemysql, hostprod-db.internal, port3306, databasesales_analytics, usernameweekly_report_reader, passwordR3p0rt2024! )提示pool_recycle3600是关键参数。MySQL 默认 wait_timeout28800 秒8 小时但应用侧若长时间空闲连接可能被服务端主动断开导致首次查询报Lost connection to MySQL server during query。设为 3600 秒1 小时可确保连接在失效前被主动回收重连。2.2 定义可复用的周报查询函数支持动态日期范围与多表 JOIN# queries.py import pandas as pd from datetime import datetime, timedelta from sqlalchemy import text def fetch_weekly_sales_summary(engine, start_dateNone, end_dateNone): 拉取本周销售核心指标订单数、GMV、新客数、复购率 支持传入自定义日期范围否则默认取上一周周一至周日 if not start_date or not end_date: # 计算上一周的周一和周日ISO calendar: Monday1, Sunday7 today datetime.now().date() last_monday today - timedelta(daystoday.weekday() 7) # 上周一 last_sunday last_monday timedelta(days6) # 上周日 start_date last_monday.strftime(%Y-%m-%d) end_date last_sunday.strftime(%Y-%m-%d) # 使用 text() 包裹 SQL支持复杂 JOIN 和子查询且参数化防注入 sql text( SELECT COUNT(DISTINCT o.order_id) AS order_count, COALESCE(SUM(o.total_amount), 0) AS gmv, COUNT(DISTINCT CASE WHEN u.first_order_date :start_date THEN u.user_id END) AS new_user_count, ROUND( COUNT(DISTINCT CASE WHEN u.order_count 1 THEN u.user_id END) * 100.0 / NULLIF(COUNT(DISTINCT u.user_id), 0), 2 ) AS repeat_rate_percent FROM orders o LEFT JOIN users u ON o.user_id u.user_id WHERE o.order_date BETWEEN :start_date AND :end_date ) try: df pd.read_sql(sql, engine, params{start_date: start_date, end_date: end_date}) return df except Exception as e: raise RuntimeError(f数据库查询失败{start_date} 至 {end_date}: {str(e)}) # 调用示例 if __name__ __main__: df fetch_weekly_sales_summary(engine) print(df) # 输出 order_count gmv new_user_count repeat_rate_percent # 1245 892345.67 213 42.67关键设计说明参数化查询params{start_date: ...}替代字符串拼接杜绝 SQL 注入。NULL 处理COALESCE(SUM(...), 0)和NULLIF(..., 0)避免除零错误与空值传播。日期计算逻辑严格按 ISO 周标准周一为每周第一天适配财务/运营周报习惯。错误封装捕获底层异常并抛出带上下文的RuntimeError便于上层统一处理。2.3 多数据源聚合用 pandas.concat 合并异构表结果实际周报常需整合销售、客服、产品三套系统数据。以下为合并销售与客服工单的示例def fetch_combined_report(engine): 合并销售摘要与客服工单统计 sales_df fetch_weekly_sales_summary(engine) # 查询客服工单假设在另一 schema 的 support_tickets 表 support_sql text( SELECT COUNT(*) AS ticket_count, COUNT(CASE WHEN status resolved THEN 1 END) AS resolved_count, ROUND(AVG(TIMESTAMPDIFF(HOUR, created_at, resolved_at)), 1) AS avg_resolve_hour FROM support_tickets WHERE created_at BETWEEN :start_date AND :end_date ) support_df pd.read_sql(support_sql, engine, params{start_date: sales_df.index[0] if not sales_df.empty else 2024-01-01}) # 水平合并concat axis1要求索引对齐 result pd.concat([sales_df, support_df], axis1) result.columns [订单数, GMV(元), 新客数, 复购率(%), 工单数, 已解决数, 平均解决时长(小时)] return result # 输出示例 # 订单数 GMV(元) 新客数 复购率(%) 工单数 已解决数 平均解决时长(小时) # 0 1245 892345.67 213 42.67 87 79 4.23. Dify 智能分析通过 REST API 调用工作流实现指标解读与归因建议Dify 的核心价值在于将固定 SQL 查询结果转化为业务人员能理解的自然语言结论。本节使用 Dify v1.17.1 的/v1/chat-messages接口调用预置工作流Workflow而非直接 prompt 工程。3.1 配置 Dify 工作流定义输入 Schema 与输出约束在 Dify Web 控制台创建名为weekly-report-analyzer的工作流关键配置如下配置项值说明Input Variablessales_data,support_data类型均为text接收 JSON 字符串LLM ModelQwen2-72B-Instruct (本地部署) 或 GPT-4-turbo (API)根据部署方式选择Output Schema{ summary: string, key_insights: [string], action_items: [string] }强制 JSON 结构化输出便于程序解析注意Dify 社区版 1.10 已支持多租户与工作流版本管理生产环境务必启用工作流版本锁定如v1.2避免上游 prompt 变更导致下游解析失败。3.2 Python 调用 Dify 工作流的健壮封装# dify_client.py import requests import json import time from typing import Dict, List, Any class DifyClient: def __init__(self, api_key: str, base_url: str http://localhost:5001): self.api_key api_key self.base_url base_url.rstrip(/) def invoke_workflow(self, workflow_id: str, inputs: Dict[str, Any], user: str weekly-report-bot) - Dict[str, Any]: 调用 Dify 工作流返回结构化分析结果 :param workflow_id: Dify 中工作流的唯一 ID非名称 :param inputs: 工作流定义的输入变量字典值自动转 JSON 字符串 :param user: 用户标识用于审计日志 :return: 解析后的 JSON 响应体 url f{self.base_url}/v1/workflows/run headers { Authorization: fBearer {self.api_key}, Content-Type: application/json } # 将 inputs 中的 dict/list 转为 JSON 字符串Dify 工作流要求 stringified_inputs {} for k, v in inputs.items(): if isinstance(v, (dict, list)): stringified_inputs[k] json.dumps(v, ensure_asciiFalse) else: stringified_inputs[k] str(v) payload { inputs: stringified_inputs, response_mode: blocking, # 同步阻塞模式适合批处理 user: user } try: response requests.post(url, jsonpayload, headersheaders, timeout120) response.raise_for_status() result response.json() # Dify blocking 模式返回字段result - data - outputs if data in result and outputs in result[data]: return result[data][outputs] else: raise ValueError(Dify 响应格式异常缺少 data.outputs) except requests.exceptions.Timeout: raise TimeoutError(Dify 工作流调用超时120秒请检查 Dify 服务状态) except requests.exceptions.RequestException as e: raise RuntimeError(fDify 请求失败: {str(e)}) except json.JSONDecodeError as e: raise ValueError(fDify 响应非 JSON 格式: {str(e)}) # 初始化客户端密钥应从环境变量读取 dify_client DifyClient( api_keyapp-xxxxxxxxxxxxxxxxxxxxxxxx, # 替换为你的 Dify API Key base_urlhttp://dify-server.internal:5001 ) # 构造输入数据来自数据库查询 combined_df fetch_combined_report(engine) inputs { sales_data: combined_df.iloc[0].to_dict(), # 第一行转 dict support_data: {ticket_count: 87, resolved_count: 79, avg_resolve_hour: 4.2} } # 调用工作流 analysis_result dify_client.invoke_workflow( workflow_idwf-abc123xyz, # 在 Dify 控制台复制的工作流 ID inputsinputs ) print(json.dumps(analysis_result, indent2, ensure_asciiFalse)) # 输出示例 # { # summary: 上周销售表现稳健GMV达89.2万元复购率达42.67%客服工单解决率90.8%但平均解决时长4.2小时略高于目标值3小时。, # key_insights: [ # 新客获取效率提升新客数环比增长12%主要来自抖音渠道投放, # 复购率持续高于行业均值38%老用户忠诚度高 # ], # action_items: [ # 优化客服排班重点加强下午时段人力缩短平均解决时长, # 复制抖音渠道获客策略至小红书平台 # ] # }参数说明与容错要点response_modeblocking确保返回完整结果避免异步轮询复杂度。timeout120Dify 工作流含 LLM 推理需预留充足时间v1.17.1 本地部署 Qwen2-72B 约需 45~90 秒。stringified_inputsDify 工作流输入必须是字符串json.dumps()处理嵌套结构。raise_for_status() 自定义异常将网络层、业务层错误分离便于监控告警。4. Word 文档生成用 python-docx 精确控制样式、表格与段落规避 COM 接口陷阱Word 自动化最常踩的坑是依赖 Windows COM 接口win32com.client导致 Linux/macOS 无法运行、Office 升级后崩溃、关闭时卡顿因 COM 对象未释放。python-docx完全基于 Open XML 标准跨平台、无依赖、生成文件与手动编辑完全兼容。4.1 构建可复用的周报模板类# word_generator.py from docx import Document from docx.shared import Pt, Inches, RGBColor from docx.enum.text import WD_PARAGRAPH_ALIGNMENT from docx.enum.table import WD_TABLE_ALIGNMENT from docx.oxml.ns import qn from docx.oxml import OxmlElement class WeeklyReportGenerator: def __init__(self, template_pathNone): 初始化文档可选加载基础模板含公司 Logo、页眉页脚 if template_path: self.doc Document(template_path) else: self.doc Document() # 设置默认字体避免中文字体显示为 Times New Roman style self.doc.styles[Normal] font style.font font.name 微软雅黑 font.size Pt(10.5) # 中文字体需额外设置 font._element.rPr.rFonts.set(qn(w:eastAsia), 微软雅黑) def add_title_page(self, title: str, subtitle: str , date_range: str ): 添加标题页 p self.doc.add_paragraph() p.alignment WD_PARAGRAPH_ALIGNMENT.CENTER run p.add_run(title) run.font.size Pt(22) run.font.bold True run.font.color.rgb RGBColor(0, 32, 96) # 深蓝色 if subtitle: p self.doc.add_paragraph(subtitle) p.alignment WD_PARAGRAPH_ALIGNMENT.CENTER p.runs[0].font.size Pt(14) if date_range: p self.doc.add_paragraph(f统计周期{date_range}) p.alignment WD_PARAGRAPH_ALIGNMENT.CENTER p.runs[0].font.size Pt(10.5) self.doc.add_page_break() def add_section_heading(self, text: str, level: int 1): 添加章节标题1-3级 heading self.doc.add_heading(text, levellevel) heading.alignment WD_PARAGRAPH_ALIGNMENT.LEFT def add_dataframe_table(self, df, title: str , width: float None): 将 pandas DataFrame 转为 Word 表格支持列宽自适应 if width is None: width Inches(6.5) # A4 页面可用宽度 self.add_section_heading(title, level2) if title else None table self.doc.add_table(rows1, colslen(df.columns)) table.style Light Shading Accent 1 table.autofit False # 设置表头 hdr_cells table.rows[0].cells for i, column in enumerate(df.columns): p hdr_cells[i].paragraphs[0] p.add_run(str(column)).bold True p.alignment WD_PARAGRAPH_ALIGNMENT.CENTER # 填充数据行 for _, row in df.iterrows(): row_cells table.add_row().cells for i, value in enumerate(row): # 处理 NaN 和数字格式 display_value if pd.isna(value) else str(value) if isinstance(value, (int, float)) and not isinstance(value, bool): display_value f{value:,} if isinstance(value, int) else f{value:,.2f} p row_cells[i].paragraphs[0] p.add_run(display_value) p.alignment WD_PARAGRAPH_ALIGNMENT.CENTER if isinstance(value, (int, float)) else WD_PARAGRAPH_ALIGNMENT.LEFT # 固定列宽关键避免 Word 自动缩放导致表格变形 for i, column in enumerate(df.columns): table.columns[i].width width / len(df.columns) def add_analysis_section(self, analysis: Dict[str, Any]): 添加 Dify 分析结果部分 self.add_section_heading(核心结论与建议, level2) # 总结段落 p self.doc.add_paragraph() p.add_run(【整体总结】).bold True p.add_run(f {analysis.get(summary, 暂无总结)}) # 关键洞察项目符号列表 if analysis.get(key_insights): self.add_section_heading(关键洞察, level3) for insight in analysis[key_insights]: p self.doc.add_paragraph(insight, styleList Bullet) # 行动项编号列表 if analysis.get(action_items): self.add_section_heading(后续行动项, level3) for i, item in enumerate(analysis[action_items], 1): p self.doc.add_paragraph(f{i}. {item}, styleList Number) def save(self, filepath: str): 保存文档 self.doc.save(filepath) print(fWord 报告已生成{filepath}) # 使用示例 generator WeeklyReportGenerator() generator.add_title_page( title2024年第24周业务周报, subtitle销售与客服联合分析, date_range2024-06-10 至 2024-06-16 ) # 添加数据表格 generator.add_dataframe_table( dfcombined_df, title核心业务指标概览 ) # 添加 Dify 分析 generator.add_analysis_section(analysis_result) generator.save(weekly_report_2024W24.docx)关键技术点解析字体设置font._element.rPr.rFonts.set(qn(w:eastAsia), 微软雅黑)是中文显示正确的必要步骤。列宽控制table.columns[i].width直接赋值英寸值彻底解决 “Word 表格列宽无法拖动” 的根源问题自动调整被禁用。数字格式化f{value:,}和f{value:,.2f}实现千分位分隔与小数精度控制避免 Word 自动转换为科学计数法。样式复用styleList Bullet和styleList Number调用内置样式比手动设置符号更稳定。5. 邮件发送SMTP 原生协议发送支持附件、HTML 正文与发送状态追踪绕过 Outlook 客户端用 Pythonsmtplibemail库发送确保 Linux 服务器可执行、无 GUI 依赖、发送失败可精确捕获。5.1 构建带附件与 HTML 正文的邮件消息# email_sender.py import smtplib from email.mime.multipart import MIMEMultipart from email.mime.text import MIMEText from email.mime.base import MIMEBase from email import encoders from pathlib import Path from typing import List, Optional def create_email_message( subject: str, html_body: str, plain_body: str , attachments: List[Path] None, from_addr: str reportcompany.com, to_addrs: List[str] None, cc_addrs: List[str] None ) - MIMEMultipart: 构建多部分邮件消息HTML 纯文本 附件 msg MIMEMultipart(alternative) msg[From] from_addr msg[To] , .join(to_addrs) if to_addrs else msg[Cc] , .join(cc_addrs) if cc_addrs else msg[Subject] subject # 添加纯文本和 HTML 版本客户端自动选择 if plain_body: part1 MIMEText(plain_body, plain, utf-8) msg.attach(part1) part2 MIMEText(html_body, html, utf-8) msg.attach(part2) # 添加附件 if attachments: for file_path in attachments: if not file_path.exists(): raise FileNotFoundError(f附件不存在: {file_path}) with open(file_path, rb) as f: part MIMEBase(application, vnd.openxmlformats-officedocument.wordprocessingml.document) part.set_payload(f.read()) encoders.encode_base64(part) part.add_header( Content-Disposition, fattachment; filename{file_path.name}, filenamefile_path.name ) msg.attach(part) return msg def send_email_smtp( msg: MIMEMultipart, smtp_server: str smtp.company.com, smtp_port: int 587, username: str reportcompany.com, password: str , use_tls: bool True ) - bool: 通过 SMTP 发送邮件返回是否成功 try: server smtplib.SMTP(smtp_server, smtp_port) if use_tls: server.starttls() # 启用 TLS 加密 server.login(username, password) # 构建收件人列表To Cc recipients [] if msg[To]: recipients.extend([addr.strip() for addr in msg[To].split(,)]) if msg[Cc]: recipients.extend([addr.strip() for addr in msg[Cc].split(,)]) server.send_message(msg, to_addrsrecipients) server.quit() return True except smtplib.SMTPAuthenticationError: raise PermissionError(SMTP 认证失败请检查用户名/密码) except smtplib.SMTPRecipientsRefused: raise ValueError(收件人地址被拒绝请检查邮箱格式及权限) except Exception as e: raise RuntimeError(fSMTP 发送失败: {str(e)}) # 构建邮件内容 html_content f html body h2 {subject}/h2 p您好/p p本周业务数据与分析报告已生成请查收附件。/p ul listrong统计周期/strong2024-06-10 至 2024-06-16/li listrong核心结论/strong{analysis_result.get(summary, 暂无)}/li listrong关键行动项/strong{br.join([f• {item} for item in analysis_result.get(action_items, [])])}/li /ul p此邮件由自动化系统发出请勿直接回复。/p /body /html plain_content f{subject} 您好 本周业务数据与分析报告已生成请查收附件。 统计周期2024-06-10 至 2024-06-16 核心结论{analysis_result.get(summary, 暂无)} 关键行动项 {.join([f- {item}\n for item in analysis_result.get(action_items, [])])} 此邮件由自动化系统发出请勿直接回复。 # 创建并发送 msg create_email_message( subject【自动发送】2024年第24周业务周报, html_bodyhtml_content, plain_bodyplain_content, attachments[Path(weekly_report_2024W24.docx)], from_addrreportcompany.com, to_addrs[managercompany.com, opscompany.com], cc_addrs[ceocompany.com] ) send_email_smtp( msgmsg, smtp_serversmtp.company.com, smtp_port587, usernamereportcompany.com, passwordAppPassw0rd2024! # 应使用应用专用密码 ) print(邮件发送成功)安全与可靠性要点应用专用密码企业邮箱如 Exchange Online、腾讯企业邮应启用“应用专用密码”而非主账户密码降低泄露风险。TLS 加密server.starttls()是强制要求明文传输密码在现代邮件系统中已被拒绝。收件人校验smtplib.SMTPRecipientsRefused异常明确指示无效邮箱便于及时修正通讯录。HTML/Plain 双版本保障在纯文本邮件客户端如某些终端邮件工具中仍可阅读核心信息。6. 全流程串联与关键调试技巧如何快速定位“卡在第几步”一个完整的周报自动化脚本本质是线性流水线数据库 → Dify → Word → Email。任一环节失败都会中断。以下是生产环境必备的调试与可观测性技巧。6.1 添加结构化日志与阶段标记import logging from datetime import datetime # 配置日志输出到文件 控制台 logging.basicConfig( levellogging.INFO, format%(asctime)s - %(levelname)s - %(message)s, handlers[ logging.FileHandler(weekly_report.log, encodingutf-8), logging.StreamHandler() ] ) logger logging.getLogger(__name__) def main(): logger.info( 周报自动化任务启动 ) report_date datetime.now().strftime(%Y%m%d_%H%M%S) try: # 阶段1数据库 logger.info(阶段1拉取数据库数据...) df fetch_combined_report(engine) logger.info(f✅ 数据库拉取完成共 {len(df)} 行数据) # 阶段2Dify 分析 logger.info(阶段2调用 Dify 工作流...) analysis dify_client.invoke_workflow( workflow_idwf-abc123xyz, inputs{sales_data: df.iloc[0].to_dict()} ) logger.info(✅ Dify 分析完成获得结构化结论) # 阶段3Word 生成 logger.info(阶段3生成 Word 文档...) generator WeeklyReportGenerator() generator.add_title_page(...) generator.add_dataframe_table(df) generator.add_analysis_section(analysis) doc_path fweekly_report_{report_date}.docx generator.save(doc_path) logger.info(f✅ Word 文档生成完成{doc_path}) # 阶段4邮件发送 logger.info(阶段4发送邮件...) msg create_email_message(...) send_email_smtp(msg) logger.info(✅ 邮件发送成功) logger.info( 周报自动化任务成功完成 ) except Exception as e: logger.error(f❌ 周报自动化失败{str(e)}, exc_infoTrue) # 可在此处触发告警如发送钉钉/企业微信消息 raise if __name__ __main__: main()6.2 快速诊断表常见失败点与验证命令故障现象最可能环节快速验证命令关键日志线索OperationalError: (2003, Cant connect to MySQL server)数据库连接telnet prod-db.internal 3306数据库查询失败TimeoutError: Dify 工作流调用超时Dify 服务curl -X GET http://dify-server.internal:5001/healthDify 请求失败Word 文档打开后显示“文件损坏”Word 生成python -c from docx import Document; docDocument(test.docx); print(OK)Word 报告已生成后无异常邮件未收到但脚本无报错SMTP 发送echo test | mail -s test youremail.comSMTP 发送失败或无日志Dify 返回{error: workflow not found}Dify 工作流ID在 Dify 控制台确认wf-abc123xyz是否存在且已发布Dify 响应格式异常6.3 关键参数速查表影响成功率的 5 个必调值参数位置推荐值为什么重要pool_recyclecreate_engine()3600防止 MySQL 连接空闲超时断连timeoutinrequests.post()DifyClient.invoke_workflow()120Dify 工作流含 LLM需足够推理时间table.columns[i].widthadd_dataframe_table()Inches(6.5) / len(df.columns)避免 Word 自动缩放导致表格挤压变形response_modeblockingDify API 调用blocking简化异步轮询逻辑适合批处理server.starttls()send_email_smtp()必须启用现代企业 SMTP 服务器强制要求加密当某次周报生成失败时不要从头重跑。先看日志末尾的❌行再对照上表执行对应验证命令——90% 的问题可在 3 分钟内定位到具体环节。本文还有配套的精品资源点击获取