# 从基础配置到架构设计：JiuwenSwarm 进阶版日报生成器开发实践 ## 引言｜当"日报自动化"真正走进日常工作场景 过去一年，AI Agent 的能力被讨论得很多。从简单的问答到复杂的工作流编排，看起来 AI 已经无所不能。但当我们真正把 Agent 带到日常办公场景时，会发现一个很现实的问题： > 一个能对话的 Agent，并不等于一个能帮你自动化工作的助手。 > 在日常办公场景里，我们面对的不是简单的"问答交互"，而是更多的重复性工作，例如：每天下班前要写日报、每周要汇总周报、每月要整理月度总结。这些工作虽然不复杂，但需要耗费大量时间和精力。 如果一个 Agent 只是"能聊天"，那它只是一个有趣的玩具。 但如果我们希望它成为一套真正可交付的"工作自动化助手"，那它必须具备三件能力：1.自动收集多源数据，2.智能分析工作效率，3.多渠道主动推送。 在这篇文章里，我会完整拆解： + 如何设计多数据源采集架构（Git 提交 + 邮件统计 + 记忆系统 + 待办事项） + 如何实现工作分析引擎（效率指标计算 + 趋势对比 + 关键词提取） + 如何构建报告生成器（日报 + 周报 + 月报） + 如何配置定时任务实现自动推送 + 完整的代码实现与测试验证流程 如果你也在思考： + 如何把 Agent 从 Demo 变成生产力工具？ + 如何让 Agent 主动推送信息而不是被动响应？ + 如何构建可复用的模块化技能体系？ 那接下来的内容，或许会给你一些新的视角。 --- ## 项目环境说明 > **本文档基于真实开发项目编写**，所有配置和代码均为实际使用的版本，非示例占位符。 > ### 实际运行环境 | 项目 | 配置值 | | --- | --- | | **项目路径** | `D:\Download\jiuwenswarm` | | **操作系统** | Windows 10 | | **Python** | 3.10+ | | **模型服务** | ModelScope (Qwen/Qwen3-235B-A22B-Instruct-2507) | ### 实际数据源配置 | 数据源 | 配置值 | | --- | --- | | **Git 仓库** | `D:\Download\jiuwenswarm` (当前项目) | | **邮箱** | `zxworkem@163.com` (网易163邮箱) | | **推送渠道** | 飞书 (`cli_a92035b1823a9cd2`) | | **心跳时间** | 每天 18:00-18:30 | ### 核心文件位置 ```plain D:\Download\jiuwenswarm\ ├── .env # 环境变量配置 ├── config/config.yaml # 应用配置（心跳、飞书频道） ├── workspace/ │ ├── HEARTBEAT.md # 心跳任务配置 │ └── agent/skills/daily-report/ # 技能模块 │ ├── SKILL.md # 技能定义 v2.0.0 │ ├── collectors/ # 数据采集模块 │ ├── analyzers/ # 工作分析模块 │ └── generators/ # 报告生成模块 ``` ![](https://cdn.nlark.com/yuque/0/2026/png/27326384/1772822490912-87defae8-95ee-4c68-890b-d5082761d783.png) ## 一、问题背景 ### 1.1 基础版日报生成器的局限性 说起构建一个**真正智能的日报生成器**，大家第一反应往往是： > "那不就是把记忆系统和待办事项拼一拼，再加个模板就完事？" > 在简单场景里，这种做法确实能用。 可一到真实工作现场，你就会碰到三类大坑： 1. **数据源单一** 只能从记忆和待办获取数据，无法感知代码提交、邮件处理等实际工作产出。 2. **缺乏分析能力** 只是简单罗列任务，无法计算效率指标、无法进行趋势对比、无法给出工作建议。 3. **报告类型固定** 只能生成日报，无法汇总周报、月报，无法满足不同汇报周期的需求。 举个例子，用户想看"本周代码提交趋势"或"与上周相比的工作效率变化"，基础版完全做不到。因为这些需要： + 获取 Git 提交历史数据 + 进行跨时间段的数据对比 + 计算效率指标和趋势分析 **进阶版日报生成器** 就不一样了——它从多个数据源采集信息，通过工作分析引擎处理，最终生成包含效率指标、趋势对比、工作建议的完整报告。 ### 1.2 日常办公的实际痛点 做办公自动化项目时，踩过不少坑。 最典型的就是**每天写日报太费时间**。下班前本该是整理一天工作、准备下班的时候，却要花 15-20 分钟回想今天做了什么、写到哪了、有什么成果。有时候忙了一天，反而想不起来具体干了啥。 还有一个问题是**日报格式不统一**。有时候写成流水账，有时候写成要点，有时候干脆忘了写。团队协作时，每个人的日报格式都不一样，leader 看起来也头疼。 最麻烦的是**容易遗漏重要事项**。明明今天解决了一个关键 bug，或者完成了一个重要功能，结果写日报的时候忘了提，等于白干了。如果有系统能自动采集 Git 提交记录，就不会遗漏这些重要产出。 **邮件沟通**也是一个容易被忽略的工作内容。今天处理了多少邮件、有多少未读、有哪些重要邮件需要跟进，这些信息写日报时往往想不起来。 用了进阶版日报生成器之后，这些问题确实解决了很多。系统会自动采集 Git 提交、邮件统计、记忆记录、待办事项，然后通过工作分析引擎计算效率指标，生成趋势对比，给出工作建议。 ### 1.3 JiuwenSwarm 的技能系统特性 JiuwenSwarm 是一个开源的 Agent 开发框架，其技能系统支持： | 能力 | 说明 | | --- | --- | | **模块化技能** | 每个 Skill 可以包含多个 Python 模块 | | **工具集成** | 可声明 `allowed_tools` 获取系统工具权限 | | **心跳触发** | 支持定时任务自动执行技能 | | **多频道推送** | 支持飞书、Web 等多种渠道 | 这个框架的价值在哪？简单说就是**能力可扩展**。通过模块化设计，我们可以把数据采集、工作分析、报告生成分别封装，形成清晰的责任边界。 ## 二、技术方案 ### 2.1 JiuwenSwarm 的分层设计 JiuwenSwarm 是分层架构的，我们的进阶版日报生成器在 Application Layer（应用层）： ![](https://cdn.nlark.com/yuque/0/2026/png/27326384/1772822901009-32c4955a-17b4-47f8-81f3-22283748f998.png) ### 2.2 三层数据处理流程 ![](https://cdn.nlark.com/yuque/0/2026/png/27326384/1772861778248-67064c9c-4de8-41c7-850a-699ca4205cd1.png) ### 2.3 核心组件概览 本项目的核心组件及其职责： | 组件 | 类型 | 职责 | 所在模块 | | --- | --- | --- | --- | | **GitCollector** | Collector | Git 提交记录采集 | `collectors/git_collector.py` | | **EmailCollector** | Collector | 网易邮箱统计采集 | `collectors/email_collector.py` | | **MemoryCollector** | Collector | 记忆数据采集 | `collectors/memory_collector.py` | | **TodoCollector** | Collector | 待办事项采集 | `collectors/todo_collector.py` | | **DataAggregator** | Aggregator | 数据聚合器 | `collectors/aggregator.py` | | **WorkAnalyzer** | Analyzer | 工作分析引擎 | `analyzers/work_analyzer.py` | | **ReportGenerator** | Generator | 报告生成器 | `generators/report_generator.py` | 组件依赖关系： ![](https://cdn.nlark.com/yuque/0/2026/png/27326384/1772861858079-e087425b-3bfd-40de-8567-92c874d0b396.png) ### 2.4 数据流与交互设计 完整的报告请求处理流程： ![](https://cdn.nlark.com/yuque/0/2026/png/27326384/1772861961611-4d5aae74-099e-4cc4-aec5-622b14fc515e.png) 关键技术决策： | 决策点 | 选择 | 原因 | | --- | --- | --- | | Git 采集方式 | 命令行 git log | 无需额外依赖，直接调用系统 Git | | 邮件采集方式 | IMAP 协议 | 网易邮箱支持，可获取邮件统计 | | 分词工具 | jieba（可选） | 中文效果好，无依赖也可降级 | | 报告格式 | Markdown | 兼容性好，飞书可渲染 | | 触发方式 | 心跳 + 手动 | 定时自动 + 按需执行 | ## 第三章｜Skills 技能系统工程化设计 ### 3.1 Skills 目录结构 ```plain workspace/agent/skills/daily-report/ ├── SKILL.md # 技能定义（必须） ├── collectors/ # 数据采集模块 │ ├── __init__.py │ ├── git_collector.py # Git 提交采集 │ ├── email_collector.py # 邮件统计采集 │ ├── memory_collector.py # 记忆数据采集 │ ├── todo_collector.py # 待办事项采集 │ └── aggregator.py # 数据聚合器 ├── analyzers/ # 工作分析模块 │ ├── __init__.py │ └── work_analyzer.py # 工作分析引擎 ├── generators/ # 报告生成模块 │ ├── __init__.py │ └── report_generator.py # 报告生成器 └── report_helper.py # 兼容旧版脚本 ``` ![](https://cdn.nlark.com/yuque/0/2026/png/27326384/1772862008911-7b9d807a-9175-4c62-a208-c3b6381169be.png) ### 3.2 advanced-daily-report 的 SKILL.md ```markdown --- name: advanced-daily-report version: 2.0.0 description: 进阶版日报生成器，支持多数据源采集、工作分析、趋势对比、周报月报聚合 tags: [report, automation, productivity, daily, weekly, monthly, advanced] allowed_tools: [read_memory, write_memory, bash, read_file, write_file] --- # 进阶版日报生成器 自动采集多源数据，智能分析工作效率，生成日报/周报/月报并推送到飞书。 ## 核心能力 ### 1. 多数据源采集 | 数据源 | 采集内容 | 频率 | |--------|----------|------| | **Git 仓库** | 提交记录、代码变更统计 | 实时 | | **网易邮箱** | 收发邮件统计、未读提醒 | 实时 | | **记忆系统** | 今日工作记录、长期记忆 | 实时 | | **待办事项** | 任务状态、完成率 | 实时 | ### 2. 智能工作分析 - **效率指标计算** - 任务完成率 = 已完成 / 总任务 - 生产力得分（0-100） - 专注度得分（0-100） - **趋势对比** - 与昨日对比 - 与上周同期对比 - 周趋势图 - **关键词提取** - 自动提取今日工作关键词 - 工作主题聚类 ### 3. 多报告类型 | 类型 | 触发方式 | 推送时间 | |------|----------|----------| | **日报** | 手动/定时 | 每天 18:00 | | **周报** | 定时 | 每周五 18:00 | | **月报** | 定时 | 每月最后一天 18:00 | ## 目录结构 ``` daily-report/ ├── SKILL.md # 技能定义（本文件） ├── collectors/ # 数据采集模块 │ ├── __init__.py │ ├── git_collector.py # Git 提交采集 │ ├── email_collector.py # 邮件统计采集 │ ├── memory_collector.py # 记忆数据采集 │ ├── todo_collector.py # 待办事项采集 │ └── aggregator.py # 数据聚合器 ├── analyzers/ # 分析模块 │ ├── __init__.py │ └── work_analyzer.py # 工作分析引擎 ├── generators/ # 报告生成模块 │ ├── __init__.py │ └── report_generator.py # 报告生成器 └── report_helper.py # 兼容旧版脚本 ``` ## 使用方式 ### ⚠️ 重要：执行方式 本技能通过执行 Python 脚本来采集数据（Git提交、邮箱邮件、记忆、待办）。 **必须使用 `bash` 工具执行脚本**，而不是直接回复用户。 **脚本会自动采集以下数据**： - **Git 提交记录**：通过 `git log` 命令读取 `D:/Download/jiuwenswarm` 仓库的提交历史 - **邮箱邮件统计**：通过 IMAP 协议连接 `zxworkem@163.com` 读取邮件统计（需要邮箱授权码） - **记忆系统**：读取 `workspace/agent/memory/` 目录下的每日记忆文件 - **待办事项**：读取 `workspace/session/` 目录下的 todo.md 文件 ### 手动触发 当用户请求生成日报/周报/月报时，**执行以下命令**： ```bash # 生成今日日报（包含Git提交、待办任务、记忆数据） cd D:/Download/jiuwenswarm && python workspace/agent/skills/daily-report/run_report.py daily --save # 生成指定日期日报 cd D:/Download/jiuwenswarm && python workspace/agent/skills/daily-report/run_report.py daily --date 2026-03-06 --save # 生成周报（聚合一周数据） cd D:/Download/jiuwenswarm && python workspace/agent/skills/daily-report/run_report.py weekly --save # 生成月报（聚合一月数据，包含每日Git提交统计） cd D:/Download/jiuwenswarm && python workspace/agent/skills/daily-report/run_report.py monthly --save # 生成月报（指定月份） cd D:/Download/jiuwenswarm && python workspace/agent/skills/daily-report/run_report.py monthly --year 2026 --month 3 --save ``` ### 执行步骤 1. 用户发送 "生成日报" / "生成周报" / "生成月报" 等指令 2. **使用 bash 执行上述命令** 3. 脚本自动采集数据： - Git: 执行 `git log` 获取提交记录、代码变更统计 - 邮箱: 通过 IMAP 连接获取邮件统计（如果配置了邮箱） - 记忆: 读取记忆文件获取工作记录 - 待办: 解析 todo.md 获取任务状态 4. 等待脚本执行完成，获取输出内容 5. 将报告内容发送给用户 ### 触发关键词 - 日报：生成今日日报、生成昨天日报、查看今日工作、查看代码提交 - 周报：生成本周周报、周报汇总、本周工作总结 - 月报：生成本月月报、月度总结、读取邮箱中本月的内容整理成月报、本月代码提交统计 ### 数据源说明 | 数据源 | 采集方式 | 配置位置 | |--------|----------|----------| | **Git 仓库** | `git log` 命令 | 仓库路径: `D:/Download/jiuwenswarm` | | **网易邮箱** | IMAP 协议 | `.env`: `EMAIL_ADDRESS`, `EMAIL_TOKEN` | | **记忆系统** | 读取 MD 文件 | `workspace/agent/memory/YYYY-MM-DD.md` | | **待办事项** | 解析 todo.md | `workspace/session/*/todo.md` | ### 定时触发 通过 `HEARTBEAT.md` 配置定时执行： ```markdown ## 活跃的任务项 - 生成今日工作日报 # 每天执行 - 每周五生成周报 # 周报 - 每月末生成月报 # 月报 ``` ## 日报模板 ```markdown # 📋 工作日报 - 2026-03-06 ## 📊 今日概览 | 指标 | 数值 | |------|------| | 提交次数 | 5 | | 任务完成 | 3/8 | | 代码变更 | +350/-80 | | 邮件处理 | 收 12 / 发 3 | | 生产力得分 | 78.5 | ## ✅ 已完成任务 - 完成日报生成器技能开发 - 配置飞书频道推送 - 测试心跳触发功能 ## 🔄 进行中任务 - 编写开发文档 - 添加周报聚合功能 ## 💻 代码提交 | 时间 | 提交信息 | 变更 | |------|----------|------| | 09:30 | feat: 添加日报生成功能 | +120/-30 | | 14:15 | fix: 修复邮件采集bug | +45/-12 | ## 📧 邮件概况 - 今日收件: 12 封 - 今日发件: 3 封 - 未读邮件: 2 封 ## 📈 趋势对比 - 提交: ↑ 2 次 - 效率: ↑ 5.2 分 ## 💡 工作建议 1. 专注度较低，建议减少干扰 2. 任务完成率有待提高 ## 🔜 明日计划 - 完善日报模板 - 添加周报聚合功能 ``` ## 配置说明 ### Git 仓库配置 本项目监控的 Git 仓库（脚本会自动读取）： ``` 仓库路径: D:/Download/jiuwenswarm ``` 脚本通过 `git log` 命令采集以下数据： - 提交哈希、提交信息、作者、时间 - 每次提交的文件变更数、新增行数、删除行数 ### 邮箱配置 在 `.env` 文件中配置（本项目实际配置）： ```env EMAIL_ADDRESS=zxworkem@163.com EMAIL_TOKEN=UFvBrE7ZDAsT2xdP EMAIL_PROVIDER=163 ``` **注意**：`EMAIL_TOKEN` 是邮箱授权码，不是登录密码。 获取方式：登录163邮箱 → 设置 → POP3/SMTP/IMAP → 开启IMAP服务 → 获取授权码 ### 心跳配置 ```yaml heartbeat: every: 3600 target: feishu active_hours: start: 18:00 end: 18:30 ``` ## API 参考 ### 数据采集器 ```python from collectors import DataAggregator aggregator = DataAggregator( workspace_dir="workspace", git_repo="path/to/repo", email_config={ "address": "xxx@163.com", "auth_code": "xxx", "provider": "163" } ) # 采集今日数据 data = aggregator.collect() # 采集一周数据 week_data = aggregator.collect_week() ``` ### 工作分析器 ```python from analyzers import WorkAnalyzer analyzer = WorkAnalyzer() result = analyzer.analyze(data.to_dict()) print(f"生产力得分: {result.metrics.productivity_score}") print(f"关键词: {result.keywords}") print(f"建议: {result.suggestions}") ``` ### 报告生成器 ```python from generators import ReportGenerator generator = ReportGenerator(aggregator) # 生成日报 daily = generator.generate_daily() # 生成周报 weekly = generator.generate_weekly() # 生成月报 monthly = generator.generate_monthly(2026, 3) ``` ## 注意事项 1. **Git 仓库**: 确保仓库路径正确且有访问权限 2. **邮箱授权**: 使用授权码而非登录密码 3. **心跳时间**: 修改后需重启服务 4. **数据存储**: 报告保存到 `workspace/agent/reports/` ## 更新日志 - **v2.0.0** (2026-03-06): 进阶版，支持多数据源、趋势对比、周报月报 - **v1.0.0** (2026-03-06): 初始版本，基础日报生成 ``` ![](https://cdn.nlark.com/yuque/0/2026/png/27326384/1772862026601-dd7ed3ff-26da-4182-8f6f-c45c437feffc.png) ### 3.3 日报模板 ```markdown # 📋 工作日报 - 2026-03-06 ## 📊 今日概览 | 指标 | 数值 | |------|------| | 提交次数 | 5 | | 任务完成 | 3/8 | | 代码变更 | +350/-80 | | 邮件处理 | 收 12 / 发 3 | | 生产力得分 | 78.5 | ## ✅ 已完成任务 - 完成日报生成器技能开发 - 配置飞书频道推送 ## 💻 代码提交 | 时间 | 提交信息 | 变更 | |------|----------|------| | 09:30 | feat: 添加日报生成功能 | +120/-30 | ## 📧 邮件概况 - 今日收件: 12 封 - 今日发件: 3 封 - 未读邮件: 2 封 ## 📈 趋势对比 - 提交: ↑ 2 次 - 效率: ↑ 5.2 分 ## 💡 工作建议 1. 专注度较低，建议减少干扰 ## 🔜 明日计划 - 完善日报模板 ``` ## 第四章｜数据采集层完整实现 ### 4.1 Git 提交采集器 ```python # collectors/git_collector.py # -*- coding: utf-8 -*- """ Git 提交记录采集器 功能： - 获取指定日期的 Git 提交记录 - 统计提交次数、修改文件数、代码行数变化 - 支持多个仓库 """ import subprocess from dataclasses import dataclass, field from datetime import datetime, timedelta from pathlib import Path from typing import Optional @dataclass class GitCommit: """Git 提交记录""" hash: str # 提交哈希 message: str # 提交信息 author: str # 作者 date: datetime # 提交时间 files_changed: int = 0 # 修改文件数 insertions: int = 0 # 新增行数 deletions: int = 0 # 删除行数 def to_dict(self) -> dict: return { "hash": self.hash, "message": self.message, "author": self.author, "date": self.date.isoformat(), "files_changed": self.files_changed, "insertions": self.insertions, "deletions": self.deletions, } @dataclass class GitStats: """Git 统计数据""" commits: list[GitCommit] = field(default_factory=list) total_commits: int = 0 total_files_changed: int = 0 total_insertions: int = 0 total_deletions: int = 0 @property def net_lines(self) -> int: """净增行数""" return self.total_insertions - self.total_deletions def to_dict(self) -> dict: return { "total_commits": self.total_commits, "total_files_changed": self.total_files_changed, "total_insertions": self.total_insertions, "total_deletions": self.total_deletions, "net_lines": self.net_lines, "commits": [c.to_dict() for c in self.commits], } class GitCollector: """Git 提交记录采集器""" def __init__(self, repo_path: str | Path): """ 初始化 Git 采集器 Args: repo_path: Git 仓库路径 """ self.repo_path = Path(repo_path).resolve() def _run_git_command(self, args: list[str], timeout: int = 30) -> str: """执行 Git 命令""" try: result = subprocess.run( ["git", "-C", str(self.repo_path)] + args, capture_output=True, text=True, timeout=timeout, encoding="utf-8", errors="replace", ) return result.stdout.strip() except subprocess.TimeoutExpired: return "" except Exception as e: return "" def get_commits(self, date: Optional[str] = None, author: Optional[str] = None) -> GitStats: """ 获取指定日期的提交记录 Args: date: 日期字符串 (YYYY-MM-DD)，默认今天 author: 作者名称过滤，默认不过滤 Returns: GitStats: Git 统计数据 """ if date is None: date = datetime.now().strftime("%Y-%m-%d") stats = GitStats() # 获取提交列表 log_format = "%H|%s|%an|%ai" since = f"{date} 00:00:00" until = f"{date} 23:59:59" cmd_args = [ "log", f"--since={since}", f"--until={until}", f"--format={log_format}", ] if author: cmd_args.append(f"--author={author}") log_output = self._run_git_command(cmd_args) if not log_output: return stats # 解析提交记录 for line in log_output.split("\n"): if not line.strip(): continue parts = line.split("|", 3) if len(parts) < 4: continue commit_hash, message, author_name, date_str = parts try: commit_date = datetime.fromisoformat( date_str.replace(" ", "T").split("+")[0] ) except ValueError: commit_date = datetime.now() # 获取每个提交的文件变更统计 numstat = self._run_git_command( ["show", "--numstat", "--format=", commit_hash] ) files_changed = 0 insertions = 0 deletions = 0 for stat_line in numstat.split("\n"): if not stat_line.strip(): continue stat_parts = stat_line.split("\t") if len(stat_parts) >= 2: try: ins = int(stat_parts[0]) if stat_parts[0] != "-" else 0 dels = int(stat_parts[1]) if stat_parts[1] != "-" else 0 insertions += ins deletions += dels files_changed += 1 except ValueError: continue commit = GitCommit( hash=commit_hash[:8], message=message.strip(), author=author_name, date=commit_date, files_changed=files_changed, insertions=insertions, deletions=deletions, ) stats.commits.append(commit) stats.total_commits += 1 stats.total_files_changed += files_changed stats.total_insertions += insertions stats.total_deletions += deletions return stats ``` ### 4.2 邮件统计采集器 > **重要说明**：163邮箱的 IMAP 服务有特殊限制，直接使用 `SELECT` 命令会返回 "Unsafe Login" 错误。 > 解决方案： > 1. 注册 ID 命令到 imaplib：`imaplib.Commands['ID'] = ('NONAUTH', 'AUTH', 'SELECTED')` > 2. 登录后发送 ID 命令声明客户端身份 > 3. 使用 `STATUS` 命令获取邮件统计（绕过 SELECT 限制） > 4. 如需读取邮件内容，ID 命令发送成功后可正常使用 SELECT ```python # collectors/email_collector.py # -*- coding: utf-8 -*- """ 邮件统计采集器 支持： - 网易邮箱 (163/126/yeah) - 通过 IMAP 协议读取邮件 功能： - 统计收件箱邮件数量 - 获取未读邮件数 - 读取邮件内容摘要 163邮箱特殊处理： - 必须注册 ID 命令并登录后发送 - 使用 STATUS 命令获取统计（绕过 Unsafe Login 限制） """ import email import re from dataclasses import dataclass, field from datetime import datetime, timedelta from email.header import decode_header from typing import Optional try: import imaplib IMAP_AVAILABLE = True # 163邮箱必须：注册ID命令到imaplib imaplib.Commands['ID'] = ('NONAUTH', 'AUTH', 'SELECTED') except ImportError: IMAP_AVAILABLE = False imaplib = None # 网易邮箱 IMAP 服务器配置 NETEASE_IMAP_SERVERS = { "163": "imap.163.com", "126": "imap.126.com", "yeah": "imap.yeah.net", } @dataclass class EmailInfo: """邮件信息""" subject: str = "" # 主题 sender: str = "" # 发件人 date: str = "" # 日期 body_preview: str = "" # 正文预览 def to_dict(self) -> dict: return { "subject": self.subject, "sender": self.sender, "date": self.date, "body_preview": self.body_preview[:200] if self.body_preview else "", } @dataclass class EmailStats: """邮件统计数据""" total_emails: int = 0 # 邮箱总邮件数 unread: int = 0 # 未读邮件数 recent_emails: list[EmailInfo] = field(default_factory=list) # 近期邮件 def to_dict(self) -> dict: return { "total_emails": self.total_emails, "unread": self.unread, "recent_emails": [e.to_dict() for e in self.recent_emails], } class EmailCollector: """邮件统计采集器""" def __init__( self, email_address: str, auth_code: str, provider: str = "163", ): """ 初始化邮件采集器 Args: email_address: 邮箱地址 auth_code: 授权码（不是登录密码） provider: 邮箱提供商 (163/126/yeah) """ if not IMAP_AVAILABLE: raise ImportError("imaplib 模块不可用") self.email_address = email_address self.auth_code = auth_code self.provider = provider.lower() if self.provider not in NETEASE_IMAP_SERVERS: raise ValueError(f"不支持的邮箱提供商: {provider}") self.imap_server = NETEASE_IMAP_SERVERS[self.provider] self._connection = None def _decode_str(self, s: str) -> str: """解码邮件字符串""" if s is None: return "" decoded_parts = decode_header(s) result = [] for part, encoding in decoded_parts: if isinstance(part, bytes): try: result.append(part.decode(encoding or "utf-8", errors="ignore")) except Exception: result.append(part.decode("utf-8", errors="ignore")) else: result.append(part) return "".join(result) def connect(self) -> bool: """连接 IMAP 服务器并发送ID命令""" try: self._connection = imaplib.IMAP4_SSL(self.imap_server, 993) self._connection.login(self.email_address, self.auth_code) # 163邮箱必须：登录后立即发送ID命令 args = '("name" "python" "version" "1.0" "vendor" "python-imap")' self._connection._simple_command("ID", args) return True except Exception as e: print(f"连接邮箱失败: {e}") return False def disconnect(self): """断开连接""" if self._connection: try: self._connection.logout() except Exception: pass self._connection = None def get_stats(self) -> EmailStats: """ 获取邮件统计（使用STATUS命令，绕过SELECT限制） Returns: EmailStats: 邮件统计数据 """ stats = EmailStats() if not self._connection: if not self.connect(): return stats try: # 使用 STATUS 命令获取统计（163邮箱的SELECT会报Unsafe Login） status, data = self._connection.status("INBOX", "(MESSAGES UNSEEN)") if status == "OK" and data: # 解析响应: b'"INBOX" (MESSAGES 39 UNSEEN 32)' response = data[0].decode() if isinstance(data[0], bytes) else str(data[0]) messages_match = re.search(r'MESSAGES\s+(\d+)', response) unseen_match = re.search(r'UNSEEN\s+(\d+)', response) if messages_match: stats.total_emails = int(messages_match.group(1)) if unseen_match: stats.unread = int(unseen_match.group(1)) except Exception as e: print(f"获取邮件统计失败: {e}") return stats def get_recent_emails(self, limit: int = 10, days: int = 30) -> list[EmailInfo]: """ 读取近期邮件内容（ID命令发送后可正常使用SELECT） Args: limit: 最多读取邮件数量 days: 只读取最近N天内的邮件 Returns: 邮件列表 """ if not self._connection: if not self.connect(): return [] emails = [] try: # ID命令已发送，现在可以正常使用SELECT typ, dat = self._connection.select("INBOX") if typ != "OK": return [] # 搜索最近N天的邮件 since_date = (datetime.now() - timedelta(days=days)).strftime("%d-%b-%Y") typ, msg_ids = self._connection.search(None, f'(SINCE {since_date})') if typ != "OK" or not msg_ids[0]: return [] ids = msg_ids[0].split()[-limit:] # 获取最新的N封 for msg_id in reversed(ids): try: typ, msg_data = self._connection.fetch(msg_id, "(RFC822)") if typ != "OK": continue raw_email = msg_data[0][1] msg = email.message_from_bytes(raw_email) # 解码主题 subject = self._decode_str(msg["Subject"]) or "(无主题)" from_addr = self._decode_str(msg.get("From", "")) date_str = msg.get("Date", "") # 提取正文 body = "" if msg.is_multipart(): for part in msg.walk(): content_type = part.get_content_type() if content_type == "text/plain": payload = part.get_payload(decode=True) charset = part.get_content_charset() or "utf-8" body = payload.decode(charset, errors="ignore") break elif content_type == "text/html" and not body: payload = part.get_payload(decode=True) charset = part.get_content_charset() or "utf-8" html_body = payload.decode(charset, errors="ignore") body = re.sub(r'<[^>]+>', ' ', html_body) body = re.sub(r'\s+', ' ', body).strip() else: payload = msg.get_payload(decode=True) charset = msg.get_content_charset() or "utf-8" body = payload.decode(charset, errors="ignore") if payload else "" emails.append(EmailInfo( subject=subject[:100], sender=from_addr[:80], date=date_str, body_preview=body[:500] if body else "" )) except Exception: continue except Exception as e: print(f"读取邮件内容失败: {e}") return emails def __enter__(self): self.connect() return self def __exit__(self, exc_type, exc_val, exc_tb): self.disconnect() return False ``` ### 4.3 记忆数据采集器 ```python # collectors/memory_collector.py # -*- coding: utf-8 -*- """ 记忆数据采集器 功能： - 读取今日记忆文件 - 读取长期记忆 - 提取工作摘要 """ import re from dataclasses import dataclass, field from datetime import datetime, timedelta from pathlib import Path from typing import Optional @dataclass class MemoryData: """记忆数据""" today_content: str = "" # 今日记忆内容 long_term_content: str = "" # 长期记忆内容 work_summaries: list[str] = field(default_factory=list) # 工作摘要列表 key_decisions: list[str] = field(default_factory=list) # 关键决策 def to_dict(self) -> dict: return { "today_content": self.today_content[:500] if self.today_content else "", "work_summaries": self.work_summaries, "key_decisions": self.key_decisions, } class MemoryCollector: """记忆数据采集器""" def __init__(self, workspace_dir: str | Path): """ 初始化记忆采集器 Args: workspace_dir: workspace 目录路径 """ self.workspace_dir = Path(workspace_dir) self.memory_dir = self.workspace_dir / "agent" / "memory" def _read_file_safe(self, file_path: Path) -> str: """安全读取文件""" if not file_path.exists(): return "" try: return file_path.read_text(encoding="utf-8") except Exception: return "" def _extract_list_items(self, content: str) -> list[str]: """提取列表项（以 - 或 * 开头的行）""" items = [] for line in content.split("\n"): stripped = line.strip() if stripped.startswith("-") or stripped.startswith("*"): item = stripped.lstrip("-* ").strip() # 跳过注释和空项 if item and not item.startswith(" - 生成今日工作日报 --- ## 已完成的任务项 --- ## 任务说明 ### 日报任务 - **触发时间**: 每天 18:00 - 18:30（根据 config.yaml 配置） - **推送目标**: 飞书 - **内容**: 今日 Git 提交、任务完成情况、邮件统计、工作效率分析 ### 周报任务 - **触发时间**: 每周五 18:00 - 18:30 - **推送目标**: 飞书 - **内容**: 本周数据聚合、趋势分析、下周计划 ### 月报任务 - **触发时间**: 每月最后一天 18:00 - 18:30 - **推送目标**: 飞书 - **内容**: 本月数据聚合、成果总结、下月计划 --- ## 配置方式 修改 `config/config.yaml` 中的 `heartbeat` 配置： ```yaml heartbeat: every: 3600 # 心跳间隔（秒） target: feishu # 推送目标 active_hours: start: 18:00 # 生效开始时间 end: 18:30 # 生效结束时间 ``` **修改后重启服务生效。** ``` ### 6.3 Git 仓库配置 本项目监控的 Git 仓库（脚本会自动读取）： ```plain 仓库路径: D:/Download/jiuwenswarm ``` **采集方式**：脚本通过 `git log` 命令采集数据，无需额外配置。 **采集内容**： + 提交哈希（commit hash） + 提交信息（commit message） + 提交作者、提交时间 + 文件变更数、新增行数、删除行数 **执行命令**： ```bash # 脚本内部执行的 git 命令 git -C D:/Download/jiuwenswarm log --since="2026-03-07 00:00:00" --until="2026-03-07 23:59:59" --format="%H|%s|%an|%ai" --numstat ``` **多仓库支持**：如需监控多个仓库，可扩展 `DataAggregator`： ```python # 扩展配置示例（需自行实现） git_repos: - path: "D:/Download/jiuwenswarm" name: "jiuwenswarm" - path: "D:/Projects/another-repo" name: "another-project" ``` ### 6.4 飞书频道配置 (config.yaml) 本项目实际使用的 `config.yaml` 飞书配置： ```yaml heartbeat: # 心跳间隔（秒），默认 3600 (1小时) every: 3600 # 心跳结果回传的 channel target: feishu # 心跳生效时间段（本地时间） # 18:00-18:30 期间会触发日报生成 active_hours: start: 18:00 end: 18:30 channels: feishu: # 飞书应用配置 # 获取方式：飞书开放平台 → 创建企业自建应用 → 获取 App ID 和 App Secret app_id: cli_a92035b1823a9cd2 app_secret: ***** encrypt_key: # 加密 key（可选） verification_token: # 验证 token（可选） allow_from: # IP 白名单（可选） enabled: true ``` **飞书应用配置步骤**： 1. 访问 [飞书开放平台](https://open.feishu.cn/) 2. 创建企业自建应用，获取 `app_id` 和 `app_secret` 3. 添加「机器人」能力 4. 配置事件订阅：`im.message.receive_v1` 5. 发布应用到所有成员 --- ## 第七章｜测试验证 ### 7.1 测试数据采集器 ```bash # 测试 Git 采集（采集指定日期的提交记录） cd D:\Download\jiuwenswarm python workspace/agent/skills/daily-report/run_report.py daily --date 2026-03-07 # 测试月报生成（采集整月数据） python workspace/agent/skills/daily-report/run_report.py monthly --year 2026 --month 3 # 测试保存到文件 python workspace/agent/skills/daily-report/run_report.py daily --save ``` ![](https://cdn.nlark.com/yuque/0/2026/png/27326384/1772862327714-d3e2c75f-83bc-46fc-8c25-e95e723f9dee.png) ### 7.2 完整测试流程 #### 步骤1：创建待办清单（测试待办数据采集） 在飞书或 Web 前端发送： ```plain 帮我创建一个待办清单： 1. 完成日报生成器技能开发 2. 实现Git提交数据采集模块 3. 实现邮箱统计数据采集模块 4. 配置飞书频道推送 5. 测试心跳触发功能 6. 编写开发文档 ``` ![](https://cdn.nlark.com/yuque/0/2026/png/27326384/1772862423989-5bdc4508-5cfe-4465-86c1-ba696f0246f8.png) #### 步骤2：模拟工作记录（测试记忆数据采集） ```plain 帮我记录今天的工作： - 上午完成了 SKILL.md 技能定义文件编写 - 创建了 Git 提交采集器 git_collector.py - 创建了邮箱统计采集器 email_collector.py - 创建了记忆数据采集器 memory_collector.py - 创建了待办事项采集器 todo_collector.py - 下午完成了工作分析引擎 work_analyzer.py - 实现了报告生成器 report_generator.py - 配置了心跳和飞书推送 - 进行了功能测试和调试 ``` ![](https://cdn.nlark.com/yuque/0/2026/png/27326384/1772862487637-8206f051-3e56-42b9-acc5-fa6a9d2ae819.png) ![](https://cdn.nlark.com/yuque/0/2026/png/27326384/1772862510145-a395e7f8-075a-4f4c-8757-10d4ba4074d9.png) #### 步骤3：提交代码（测试 Git 数据采集） ```bash # 在项目中提交一些代码，用于测试 Git 采集 git add . git commit -m "feat: 添加日报生成器完整功能 - 实现多数据源采集（Git/邮箱/记忆/待办） - 添加工作分析引擎 - 支持日报/周报/月报生成 - 配置飞书推送和心跳触发" ``` ![](https://cdn.nlark.com/yuque/0/2026/png/27326384/1772862545512-3cf0d5fd-ab1e-4d4a-bea2-520a843302f4.png) ![](https://cdn.nlark.com/yuque/0/2026/png/27326384/1772862651273-81909a36-6233-46a6-b35f-2e21f783f7cc.png) #### 步骤4：生成日报 ```plain 生成今日日报 ``` ![](https://cdn.nlark.com/yuque/0/2026/png/27326384/1772863455503-33da3fbd-859f-4ea0-8578-7a846c5438c4.png) #### 步骤5：生成月报（测试邮箱数据采集） ```plain 读取邮箱中本月的内容整理成月报 ``` ![](https://cdn.nlark.com/yuque/0/2026/png/27326384/1772866734099-2059a8b7-632a-4e42-978b-0bdc80f9c2b5.png) ## 第九章｜扩展方向 ### 9.1 更多数据源 | 数据源 | 采集方式 | 价值 | | --- | --- | --- | | 企业微信/钉钉 | API | 消息沟通统计 | | 日程/日历 | CalDAV/iCal | 会议时间分析 | | Jira/飞书任务 | API | 项目进度跟踪 | | 浏览器历史 | 本地数据库 | 工作内容追溯 | ### 9.2 更智能分析 + **工作模式识别**：识别高效时段、低效时段 + **疲劳度预警**：基于连续工作时长 + **时间分配建议**：优化任务优先级 ### 9.3 更丰富交互 + **飞书按钮交互**：编辑、重新生成、推送 + **日报编辑功能**：在线修改后保存 + **审批流程**：Leader 审阅确认 --- ## 写在最后 从最初的一个简单想法——"能不能让 AI 帮我写日报"，到如今这套完整的多数据源日报生成系统，这个项目经历了多次迭代和优化。 开发过程中遇到的最大挑战是 **163邮箱的 IMAP 协议适配**。网易邮箱的安全限制导致 `SELECT` 命令返回 "Unsafe Login" 错误，经过反复调试和资料查阅，最终通过以下方案解决： 1. **注册 ID 命令**：`imaplib.Commands['ID'] = ('NONAUTH', 'AUTH', 'SELECTED')` 2. **发送身份声明**：登录后立即发送 `ID` 命令 3. **使用 STATUS 命令**：绕过 SELECT 限制获取邮件统计 这套系统现在可以： - 自动采集 Git 提交、邮件统计、记忆记录、待办事项 - 生成日报、周报、月报 - 通过飞书定时推送 - 读取邮件内容并生成摘要 如果你也在尝试构建类似的 AI Agent 应用，希望这篇文章能给你一些参考。 > **让 AI Agent 真正成为智能工作助手，从进阶版日报生成器开始。** > > — JiuwenSwarm 进阶版日报生成器开发实践 --- **参考资料**： - [163邮箱 IMAP ID 命令解决方案](https://github.com/HKUDS/nanobot/issues/1123) - [网易邮箱帮助中心](https://help.mail.163.com/) - [Python imaplib 文档](https://docs.python.org/3/library/imaplib.html)