AI 表格做 Scrum 团队的大脑

┌──────────────────────────────────────────────────────────┐
│                    钉钉 AI 表格（大脑）                      │
│                                                          │
│  任务数据 ←→ AI 分析 ←→ 方案建议 ←→ 状态追踪                │
│     ↑           ↑          ↓           ↑                  │
└─────┼───────────┼──────────┼───────────┼──────────────────┘
      │           │          │           │
      ▼           │          ▼           │
  dws aitable     │    dws chat send     │  CI/CD Webhook
  (读写数据)      │    (推送通知到群)      │  (构建/部署回调)
      │           │          │           │
      ▼           │          ▼           │
   CI/CD 系统     │      钉钉群沟通       │  代码仓库
   (构建/测试/    │      (讨论/决策/      │  (提交/MR/
    部署)         │       确认)           │   Review)

task_id:     BUG-042
type:        bug
priority:    P0
status:      开发中
assignee:    张三
context:     用户支付时频繁超时，错误码 PAYMENT_TIMEOUT
ai_analysis: 支付超时集中在支付网关模块，关联 v2.3.1 发布。
             根因可能是 PaymentClient 超时配置过短。
             建议方案：添加重试策略（3 次，间隔 2s/5s/10s），
             调整 connectTimeout 从 3s 到 5s。预计 30 分钟。
ci_status:   构建中

import subprocess, json
from datetime import datetime

def dws_run(args: list[str]) -> dict:
    """执行 dws 命令并解析 JSON 输出"""
    result = subprocess.run(
        ["dws"] + args + ["-f", "json"],
        capture_output=True, text=True
    )
    return json.loads(result.stdout)

def send_group_message(text: str) -> None:
    """发送消息到 Scrum 团队群"""
    subprocess.run([
        "dws", "chat", "message", "send",
        "--group-id", "$SCRUM_GROUP_ID",
        "--content", text,
        "-f", "json"
    ])
    # generated by AI

def daily_standup_summary():
    """每日晨会摘要：读取 AI 表格，推送分析到群"""
    records = dws_run([
        "aitable", "record", "list",
        "--base-id", "$BASE_ID",
        "--table-id", "$TABLE_ID",
        "--filter", "status!=已完成 AND sprint=当前"
    ])

    p0 = [r for r in records if r["priority"] == "P0"]
    p1 = [r for r in records if r["priority"] == "P1"]
    bugs = [r for r in records if r["type"] == "bug"]

    summary = (
        f"📊 **{datetime.now().strftime('%m/%d')} 晨会摘要**\n\n"
        f"今日待办 {len(records)} 项 "
        f"（P0×{len(p0)} / P1×{len(p1)}）\n\n"
    )

    if len(bugs) >= 3:
        modules = {}
        for b in bugs:
            mod = b.get("module", "未知")
            modules[mod] = modules.get(mod, 0) + 1
        top_module = max(modules, key=modules.get)
        if modules[top_module] >= 2:
            summary += (
                f"⚠️ {top_module} 模块有 {modules[top_module]} 个 bug，"
                f"建议优先排查\n\n"
            )

    for r in records[:5]:
        emoji = "🔴" if r["priority"] == "P0" else "🟡"
        summary += (
            f"{emoji} [{r['task_id']}] {r['context'][:30]}...\n"
            f"   → {r.get('ai_analysis', '待分析')[:60]}\n"
        )

    send_group_message(summary)
    # generated by AI

from dataclasses import dataclass

@dataclass
class CIBuildResult:
    task_id: str
    status: str       # "success" | "failed"
    duration: int     # 秒
    deploy_env: str   # "staging" | "production"
    commit_sha: str

def on_ci_complete(build: CIBuildResult):
    """CI/CD 完成回调：更新 AI 表格 + 通知群"""
    # 更新 AI 表格
    new_status = "测试中" if build.status == "success" else "构建失败"
    dws_run([
        "aitable", "record", "update",
        "--base-id", "$BASE_ID",
        "--table-id", "$TABLE_ID",
        "--record-id", build.task_id,
        "--fields", json.dumps({
            "ci_status": "通过" if build.status == "success" else "失败",
            "status": new_status,
        })
    ])

    # 推送到群
    if build.status == "success":
        msg = (
            f"✅ {build.task_id} 构建通过 "
            f"（{build.duration // 60} 分钟）"
            f"→ 已部署 {build.deploy_env}"
        )
    else:
        msg = (
            f"❌ {build.task_id} 构建失败，"
            f"请 {get_assignee(build.task_id)} 检查"
        )

    send_group_message(msg)
    # generated by AI

def on_status_change(task_id: str, old: str, new: str, assignee: str):
    """任务状态变更 → 自动通知群"""
    emoji_map = {
        "开发中": "🔧", "测试中": "🧪",
        "已部署": "🚀", "已完成": "✅"
    }
    emoji = emoji_map.get(new, "📋")
    send_group_message(
        f"{emoji} {task_id}：{old} → **{new}**（{assignee}）"
    )
    # generated by AI