企业智能日报系统：飞书+多数据源自动汇总全链路

某50人技术团队的真实场景：每天早上站会前，每个同学要花15-20分钟手动收集昨天的工作内容——翻飞书消息、查GitHub提交记录、看TAPD任务状态、整理JIRA工单。50个人每天浪费约15小时在信息收集上。

团队负责人找到我，需求很明确：

：这个项目需要定时调度（RRULE）、飞书连接器、错误重试、记忆系统（记住每个成员的日报偏好）——这些全是WorkBuddy内置能力，不用从零造轮子。

┌──────────────┐   ┌──────────────┐   ┌──────────────┐   ┌──────────────┐
│  飞书消息API  │   │ GitHub API   │   │  TAPD API    │   │  JIRA API    │
└──────┬───────┘   └──────┬───────┘   └──────┬───────┘   └──────┬───────┘
       │                  │                  │                  │
       └──────────┬───────┴──────────┬───────┘                  │
                  │   数据采集层       │                          │
                  ▼                  ▼                          │
         ┌────────────────────────────────────────────────────────┘
         │              WorkBuddy 自动化引擎
         │  ┌──────────────────────────────────────────────┐
         │  │ 1. RRULE触发 (每天8:30)                       │
         │  │ 2. 并行采集4个数据源 (隔离运行时)              │
         │  │ 3. AI总结生成日报 (记忆系统读取个人偏好)        │
         │  │ 4. 飞书群发送 (失败自动重试3次)                │
         │  │ 5. 日报存入记忆系统 (供历史查询)               │
         │  └──────────────────────────────────────────────┘
         └────────────────────┬─────────────────────────────
                              ▼
                   ┌─────────────────────┐
                   │  飞书群机器人通知     │
                   └─────────────────────┘


本项目综合应用了专题1前7篇文章的核心知识：

上一篇我们讲了工作区与自动化管理——如何组织多个自动化任务、配置多项目调度。今天把这个知识用到真实企业项目中：搭建一个完整的智能日报系统，从需求分析到上线验收，完整走一遍。

# 创建日报系统专用工作区
mkdir -p ~/WorkBuddy/daily-report-system
cd ~/WorkBuddy/daily-report-system

# 初始化WorkBuddy项目
# 通过WorkBuddy桌面端创建工作区，或CLI初始化


# .workbuddy/runtime.yml
node:
  version: "22.22.2"
  workspace: "./node_modules"

python:
  version: "3.13.12"
  venv_path: "./.venv"
  pip_index: "https://pypi.tuna.tsinghua.edu.cn/simple"


# 创建Python虚拟环境并安装依赖
C:\Users\fumingcai\.workbuddy\binaries\python\versions\3.13.12\python.exe -m venv .venv

# 激活venv后安装依赖
pip install requests openai pytz

# 验证安装
python -c "import requests; import openai; print('依赖安装成功')"


API密钥绝对不能硬编码到代码里。

 利用WorkBuddy的安全沙箱机制，通过环境变量注入：

# .workbuddy/secrets.yml（此文件不提交到Git）
feishu:
  app_id: "cli_xxxxxxxxxxxx"
  app_secret: "xxxxxxxxxxxxxxxxxxxxxxxx"
  group_chat_id: "oc_xxxxxxxxxxxxxxxx"

github:
  token: "ghp_xxxxxxxxxxxxxxxxxxxx"

tapd:
  app_id: "xxxxxxxx"
  app_key: "xxxxxxxx-xxxx-xxxx-xxxx-xxxxxxxxxxxx"

jira:
  base_url: "https://yourcompany.atlassian.net"
  email: "reporter@company.com"
  api_token: "ATATT3xxxxxxxxxxxxxxxx"

openai:
  api_key: "sk-proj-xxxxxxxxxxxxxxxx"


# collectors/feishu_collector.py
# 飞书消息采集器：获取指定用户昨天在指定群的发言

import requests
import os
from datetime import datetime, timedelta, timezone

class FeishuCollector:
    """飞书消息采集器"""
    
    def __init__(self):
        self.app_id = os.environ.get("FEISHU_APP_ID")
        self.app_secret = os.environ.get("FEISHU_APP_SECRET")
        self._token = None
    
    def _get_tenant_token(self):
        """获取飞书 tenant_access_token"""
        url = "https://open.feishu.cn/open-apis/auth/v3/tenant_access_token/internal"
        resp = requests.post(url, json={
            "app_id": self.app_id,
            "app_secret": self.app_secret
        }, timeout=10)
        self._token = resp.json()["tenant_access_token"]
        return self._token
    
    def collect_messages(self, chat_id, user_id=None):
        """
        采集昨天指定群聊的消息
        :param chat_id: 飞书群聊ID
        :param user_id: 可选，指定用户的消息
        :return: 消息文本列表
        """
        token = self._get_tenant_token()
        
        # 计算昨天的时间范围（UTC+8）
        now_cst = datetime.now(timezone(timedelta(hours=8)))
        yesterday_start = (now_cst - timedelta(days=1)).replace(
            hour=0, minute=0, second=0, microsecond=0
        )
        yesterday_end = yesterday_start.replace(
            hour=23, minute=59, second=59
        )
        
        # 转为毫秒时间戳
        start_time = int(yesterday_start.timestamp())
        end_time = int(yesterday_end.timestamp())
        
        url = "https://open.feishu.cn/open-apis/im/v1/messages"
        params = {
            "container_id_type": "chat",
            "container_id": chat_id,
            "start_time": str(start_time),
            "end_time": str(end_time),
            "page_size": 50
        }
        if user_id:
            params["user_id"] = user_id
        
        headers = {"Authorization": f"Bearer {token}"}
        resp = requests.get(url, params=params, headers=headers, timeout=15)
        
        messages = []
        for item in resp.json().get("data", {}).get("items", []):
            # 提取文本消息内容
            if item.get("msg_type") == "text":
                import json
                content = json.loads(item["body"]["content"])
                messages.append(content.get("text", ""))
        
        return messages


# collectors/github_collector.py
# GitHub提交采集器：获取指定仓库昨天的commits

class GitHubCollector:
    """GitHub Commits 采集器"""
    
    def __init__(self):
        self.token = os.environ.get("GITHUB_TOKEN")
        self.headers = {
            "Authorization": f"token {self.token}",
            "Accept": "application/vnd.github.v3+json"
        }
    
    def collect_commits(self, repo, author=None):
        """
        采集指定仓库昨天的commits
        :param repo: 仓库名，格式 "owner/repo"
        :param author: 可选，指定提交者GitHub用户名
        :return: commit信息列表 [{"message": ..., "author": ..., "time": ...}]
        """
        from datetime import datetime, timedelta, timezone
        
        now_utc = datetime.now(timezone.utc)
        yesterday = (now_utc - timedelta(days=1)).strftime("%Y-%m-%d")
        today = now_utc.strftime("%Y-%m-%d")
        
        url = f"https://api.github.com/repos/{repo}/commits"
        params = {"since": f"{yesterday}T00:00:00Z", "until": f"{today}T00:00:00Z"}
        if author:
            params["author"] = author
        
        resp = requests.get(url, params=params, headers=self.headers, timeout=15)
        
        commits = []
        for c in resp.json():
            commits.append({
                "message": c["commit"]["message"].split("\n")[0],  # 只取第一行
                "author": c["commit"]["author"]["name"],
                "time": c["commit"]["author"]["date"]
            })
        
        return commits


# collectors/tapd_collector.py
# TAPD任务变更采集器

class TAPDCollector:
    """TAPD任务变更采集器"""
    
    def __init__(self):
        self.app_id = os.environ.get("TAPD_APP_ID")
        self.app_key = os.environ.get("TAPD_APP_KEY")
        self.base_url = "https://api.tapd.cn"
    
    def collect_task_changes(self, workspace_id, user_id=None):
        """
        采集昨天TAPD任务状态变更
        :param workspace_id: TAPD项目ID
        :return: 任务变更列表
        """
        from datetime import datetime, timedelta, timezone
        
        now_cst = datetime.now(timezone(timedelta(hours=8)))
        yesterday = (now_cst - timedelta(days=1)).strftime("%Y-%m-%d")
        
        url = f"{self.base_url}/changes"
        params = {
            "workspace_id": workspace_id,
            "created": f"{yesterday}",
            "limit": 200
        }
        if user_id:
            params["owner"] = user_id
        
        # TAPD使用Basic Auth
        import base64
        auth = base64.b64encode(f"{self.app_id}:{self.app_key}".encode()).decode()
        headers = {"Authorization": f"Basic {auth}"}
        
        resp = requests.get(url, params=params, headers=headers, timeout=15)
        
        changes = []
        for item in resp.json().get("data", []):
            change = item.get("Change", {})
            changes.append({
                "task_name": change.get("name", ""),
                "field": change.get("field", ""),
                "old_value": change.get("old_value", ""),
                "new_value": change.get("new_value", ""),
                "author": change.get("author", "")
            })
        
        return changes


# summarizer/report_generator.py
# AI日报生成器：将多源数据汇总为结构化日报

import os
import json
from datetime import datetime, timezone, timedelta
from openai import OpenAI

class ReportGenerator:
    """AI日报生成器"""
    
    def __init__(self):
        self.client = OpenAI(api_key=os.environ.get("OPENAI_API_KEY"))
        self.model = "gpt-4o-mini"  # 日常日报用mini降本
    
    def generate_report(self, member_name, collected_data, preferences=None):
        """
        根据采集数据生成结构化日报
        :param member_name: 成员姓名
        :param collected_data: 多源采集数据字典
        :param preferences: 个人偏好（来自记忆系统）
        :return: 日报Markdown文本
        """
        today = datetime.now(timezone(timedelta(hours=8))).strftime("%Y-%m-%d")
        
        # 构建数据摘要
        data_summary = self._build_data_summary(collected_data)
        
        # 构建偏好提示
        pref_text = ""
        if preferences:
            pref_text = f"\n\n**该成员的日报偏好**：\n{preferences}"
        
        prompt = f"""你是日报生成助手。请根据以下数据，为成员「{member_name}」生成{today}的日报。

**采集到的原始数据**：
{data_summary}
{pref_text}

**输出要求**：
1. 使用Markdown格式
2. 结构：昨日完成 → 今日计划 → 风险/阻塞
3. 语言简洁专业，每条不超过两行
4. 如果某个数据源没有数据，不要编造，直接跳过
5. 合并相关联的工作项（如GitHub提交和TAPD任务如果指向同一件事，合并描述）
"""

        resp = self.client.chat.completions.create(
            model=self.model,
            messages=[
                {"role": "system", "content": "你是专业的技术日报生成助手，输出结构清晰、信息准确。"},
                {"role": "user", "content": prompt}
            ],
            temperature=0.3,  # 低温度保证稳定性
            max_tokens=800
        )
        
        return resp.choices[0].message.content
    
    def _build_data_summary(self, data):
        """将多源数据构建为文本摘要"""
        parts = []
        
        if data.get("feishu_messages"):
            parts.append("【飞书消息】")
            for msg in data["feishu_messages"][:20]:  # 限制数量控制token
                parts.append(f"  - {msg[:100]}")
        
        if data.get("github_commits"):
            parts.append("【GitHub提交】")
            for c in data["github_commits"][:15]:
                parts.append(f"  - [{c['author']}] {c['message'][:80]}")
        
        if data.get("tapd_changes"):
            parts.append("【TAPD任务变更】")
            for t in data["tapd_changes"][:15]:
                parts.append(f"  - {t['task_name'][:60]} | {t['field']}: {t['old_value']}→{t['new_value']}")
        
        if data.get("jira_issues"):
            parts.append("【JIRA工单】")
            for j in data["jira_issues"][:10]:
                parts.append(f"  - {j.get('key', '')} {j.get('summary', '')[:60]} | 状态: {j.get('status', '')}")
        
        return "\n".join(parts) if parts else "（昨日无数据）"


# notifier/feishu_notifier.py
# 飞书群消息发送器

import os
import requests
import json
import time

class FeishuNotifier:
    """飞书群消息发送器"""
    
    def __init__(self):
        self.app_id = os.environ.get("FEISHU_APP_ID")
        self.app_secret = os.environ.get("FEISHU_APP_SECRET")
        self.group_chat_id = os.environ.get("FEISHU_GROUP_CHAT_ID")
        self._token = None
    
    def _get_token(self):
        """获取tenant_access_token（带缓存）"""
        if self._token:
            return self._token
        
        resp = requests.post(
            "https://open.feishu.cn/open-apis/auth/v3/tenant_access_token/internal",
            json={"app_id": self.app_id, "app_secret": self.app_secret},
            timeout=10
        )
        self._token = resp.json()["tenant_access_token"]
        return self._token
    
    def send_report(self, report_md, member_name):
        """
        发送日报到飞书群
        :param report_md: 日报Markdown文本
        :param member_name: 成员姓名
        """
        token = self._get_token()
        
        # 飞书支持发送富文本卡片消息
        card_content = {
            "config": {"wide_screen_mode": True},
            "header": {
                "title": {"tag": "plain_text", "content": f"📋 {member_name}的日报"},
                "template": "blue"
            },
            "elements": [
                {
                    "tag": "markdown",
                    "content": report_md
                },
                {
                    "tag": "hr"
                },
                {
                    "tag": "note",
                    "elements": [
                        {
                            "tag": "plain_text",
                            "content": "本日报由WorkBuddy智能日报系统自动生成"
                        }
                    ]
                }
            ]
        }
        
        url = "https://open.feishu.cn/open-apis/im/v1/messages"
        headers = {
            "Authorization": f"Bearer {token}",
            "Content-Type": "application/json"
        }
        
        body = {
            "receive_id": self.group_chat_id,
            "msg_type": "interactive",
            "content": json.dumps(card_content)
        }
        
        params = {"receive_id_type": "chat_id"}
        
        resp = requests.post(
            url, params=params, headers=headers, 
            json=body, timeout=15
        )
        
        if resp.status_code == 200 and resp.json().get("code") == 0:
            print(f"✅ {member_name}日报发送成功")
            return True
        else:
            print(f"❌ {member_name}日报发送失败: {resp.text}")
            return False


4.4 主调度模块（整合所有模块）

# main.py
# 日报系统主调度入口

import os
import sys
import json
import traceback
from datetime import datetime, timezone, timedelta

# 加载Secrets到环境变量（安全沙箱模式下运行）
def load_secrets():
    """从secrets.yml加载API密钥到环境变量"""
    import yaml
    secrets_path = os.path.join(os.path.dirname(__file__), ".workbuddy", "secrets.yml")
    if os.path.exists(secrets_path):
        with open(secrets_path, "r", encoding="utf-8") as f:
            secrets = yaml.safe_load(f)
            # 扁平化注入环境变量
            for category, keys in secrets.items():
                for key, value in keys.items():
                    env_key = f"{category.upper()}_{key.upper()}"
                    os.environ[env_key] = str(value)

# 团队成员配置
TEAM_MEMBERS = [
    {
        "name": "张三",
        "github": "zhangsan",
        "tapd_user": "zhangsan@company.com",
        "feishu_user": "ou_xxxxxxxxxxxx",
        "preferences": "日报格式偏好：精简版，每条一行，不需要今日计划"
    },
    {
        "name": "李四",
        "github": "lisi",
        "tapd_user": "lisi@company.com",
        "feishu_user": "ou_yyyyyyyyyyyy",
        "preferences": "日报格式偏好：详细版，包含代码改动行数统计"
    }
]

# 数据源配置
DATA_SOURCES = {
    "feishu_chats": ["oc_chat1_id", "oc_chat2_id"],
    "github_repos": ["company/backend", "company/frontend"],
    "tapd_workspace": "12345678"
}

def run_daily_report():
    """主执行函数"""
    print(f"🚀 日报系统启动 - {datetime.now(timezone(timedelta(hours=8))).strftime('%Y-%m-%d %H:%M')}")
    
    load_secrets()
    
    from collectors.feishu_collector import FeishuCollector
    from collectors.github_collector import GitHubCollector
    from collectors.tapd_collector import TAPDCollector
    from summarizer.report_generator import ReportGenerator
    from notifier.feishu_notifier import FeishuNotifier
    
    # 初始化各模块
    feishu_collector = FeishuCollector()
    github_collector = GitHubCollector()
    tapd_collector = TAPDCollector()
    report_gen = ReportGenerator()
    notifier = FeishuNotifier()
    
    success_count = 0
    fail_count = 0
    
    for member in TEAM_MEMBERS:
        try:
            print(f"\n📊 正在处理: {member['name']}")
            
            # 并行采集数据（这里用串行简化，生产环境可用concurrent.futures）
            collected = {}
            
            # 1. 飞书消息
            try:
                all_msgs = []
                for chat_id in DATA_SOURCES["feishu_chats"]:
                    msgs = feishu_collector.collect_messages(chat_id, member["feishu_user"])
                    all_msgs.extend(msgs)
                collected["feishu_messages"] = all_msgs
            except Exception as e:
                print(f"  ⚠️ 飞书采集失败: {e}")
                collected["feishu_messages"] = []
            
            # 2. GitHub提交
            try:
                all_commits = []
                for repo in DATA_SOURCES["github_repos"]:
                    commits = github_collector.collect_commits(repo, member["github"])
                    all_commits.extend(commits)
                collected["github_commits"] = all_commits
            except Exception as e:
                print(f"  ⚠️ GitHub采集失败: {e}")
                collected["github_commits"] = []
            
            # 3. TAPD任务变更
            try:
                changes = tapd_collector.collect_task_changes(
                    DATA_SOURCES["tapd_workspace"], member["tapd_user"]
                )
                collected["tapd_changes"] = changes
            except Exception as e:
                print(f"  ⚠️ TAPD采集失败: {e}")
                collected["tapd_changes"] = []
            
            # AI生成日报
            report = report_gen.generate_report(
                member["name"], collected, member.get("preferences")
            )
            
            # 发送到飞书群
            sent = notifier.send_report(report, member["name"])
            if sent:
                success_count += 1
                # 记录到记忆系统（供历史查询）
                save_to_memory(member["name"], report)
            else:
                fail_count += 1
                
        except Exception as e:
            print(f"  ❌ {member['name']}日报生成失败: {e}")
            traceback.print_exc()
            fail_count += 1
    
    print(f"\n{'='*50}")
    print(f"✅ 成功: {success_count}人 | ❌ 失败: {fail_count}人")
    
    if fail_count > 0:
        sys.exit(1)  # 非零退出码触发WorkBuddy重试机制


def save_to_memory(name, report):
    """将日报保存到工作区记忆系统"""
    today = datetime.now(timezone(timedelta(hours=8))).strftime("%Y-%m-%d")
    memory_dir = os.path.join(os.path.dirname(__file__), ".workbuddy", "memory")
    os.makedirs(memory_dir, exist_ok=True)
    
    log_path = os.path.join(memory_dir, f"{today}.md")
    with open(log_path, "a", encoding="utf-8") as f:
        f.write(f"\n### {name}的日报\n{report}\n")


if __name__ == "__main__":
    run_daily_report()


# 日报系统的容错策略
retry:
  max_attempts: 3           # 最多重试3次
  backoff: exponential      # 指数退避：30s → 60s → 120s
  retry_on_exit_code: [1]   # 退出码1时重试（main.py失败时exit(1)）

# 单数据源失败不阻断整体流程
# （在main.py中已实现：每个数据源try-catch独立处理）


：50人×15小时/天×22天 = 16500小时/月的人力浪费消除，成本仅¥110/月。

# 日报系统生产配置模板
schedule:
  rrule: "FREQ=DAILY;BYHOUR=8;BYMINUTE=30;BYDAY=MO,TU,WE,TH,FR"
runtime:
  model: "gpt-4o-mini"       # 日报用mini降本
  timeout: 300000             # 5分钟超时
  retry:
    max_attempts: 3
    backoff: exponential
security:
  secrets_file: ".workbuddy/secrets.yml"
  gitignore: true
  rotate_keys: "quarterly"    # 每季度轮换密钥
memory:
  daily_log: true             # 日报存入daily log
  preferences: true           # 记录个人偏好


专题2开篇：AI自动化 vs RPA vs 传统脚本：企业该怎么选？

专题1完整闭环！从环境搭建到企业级项目落地，你已经掌握了WorkBuddy的核心能力。接下来进入专题2——AI Skills/Agent规范编写与优化实战，从"用自动化"进化到"造自动化"。

专题2第1篇（通识引流）：三种自动化方案的技术对比、成本分析、场景适配决策矩阵——什么时候用RPA、什么时候用传统脚本、什么时候用AI自动化。

：回顾本项目中的main.py，思考如果要把数据采集器封装成可复用的Skill，需要怎么设计？

本文首发于「AI技术专家」公众号，转载请注明出处。

从需求分析到上线验收，完整走通一个企业AI自动化项目。涵盖多数据源采集（飞书/GitHub/TAPD）、AI日报生成、飞书群发送、错误重试、记忆系统、以及4个真实踩坑复盘。50人团队，月成本仅¥110。

后端研发专家

编程语言

开发工具

后端

企业智能日报系统通过飞书+多数据源自动汇总，实现高效日报生成。系统整合飞书消息、GitHub提交、TAPD任务和JIRA工单数据，利用AI自动生成结构化日报，每天8:30发送至飞书群。50人团队每月可节省16500小时人力成本，仅需110元/月。系统支持失败重试、历史查询和个人偏好记忆，是企业自动化办公的实用解决方案。

2026上云采购 | AI焕新·智启新局

tapd

文章

问答

视频

教程

学习中心

腾讯云实验室

直播

竞赛

腾讯云代码分析专区

腾讯iOA零信任安全管理系统专区

腾讯云架构师技术同盟交流圈

腾讯云数据库专区

腾讯云智能顾问专区

腾讯云原生专区

腾讯混元专区

腾讯云TCE专区

腾讯云Lighthouse专区

腾讯云HAI专区

腾讯云Edgeone专区

腾讯云存储专区

腾讯云智能专区

腾讯轻联专区 

腾讯云开发专区

TAPD专区

腾讯轻量云游戏服专区

EdgeOne AI 安全实战专区

腾讯云最具价值专家

腾讯云架构师技术同盟

腾讯云创作之星

腾讯云开发者先锋

腾讯云代码助手

云原生构建

TAPD 敏捷项目管理

Cloud Studio

SDK中心

API中心

命令行工具

涵盖代码开发、场景应用、自动测试全流程，助你从零构建专属AI助手

一站式MCP教程库，解锁AI应用新玩法

聚焦“写作效率、视觉美观与运行性能”三方面进行全面升级，为您提供更高效、稳定的创作环境

社区富文本&Markdown编辑器全新改版上线，欢迎大家体验!

诚挚邀请您参与本次调研，分享您的真实使用感受与建议。您的反馈至关重要，感谢您的支持与参与！

企业智能日报系统：飞书+多数据源自动汇总全链路

企业智能日报系统：飞书+多数据源自动汇总全链路

社区

活动

圈层

关于

腾讯云开发者

热门产品

热门推荐

更多推荐