scripts/ci/chatops/__init__.py

@@ -0,0 +1,19 @@
+"""CI ChatOps 工具包 - 飞书机器人对接 Gitea Actions
+
+模块:
+    config         - 配置管理（环境变量）
+    gitea_client   - Gitea API 客户端封装
+    feishu_notify  - 飞书通知（失败/恢复/E2E摘要）
+    ci_query       - CI 状态查询
+    ci_trigger     - CI 重跑触发
+    webhook_server - Gitea webhook 接收服务（FastAPI）
+"""
+
+__all__ = [
+    "config",
+    "gitea_client",
+    "feishu_notify",
+    "ci_query",
+    "ci_trigger",
+    "webhook_server",
+]

scripts/ci/chatops/ci_query.py

@@ -0,0 +1,296 @@
+#!/usr/bin/env python3
+"""
+CI 状态查询模块 - 查询 run 列表、某分支/某 PR 的 CI 状态、失败详情
+
+支持查询类型:
+  - list_runs: 列出最近的 workflow runs
+  - branch_status: 某分支最新 CI 状态
+  - pr_status: 某 PR 的 CI 状态
+  - failure_detail: 某次 run 的失败详情
+
+用法:
+  python3 scripts/ci/chatops/ci_query.py --branch develop
+  python3 scripts/ci/chatops/ci_query.py --pr 123
+  python3 scripts/ci/chatops/ci_query.py --run-id 456 --detail
+
+设计:
+  - 与飞书机器人 /ci status 命令对接
+  - 返回结构化数据，上层负责格式化输出
+"""
+
+import argparse
+import sys
+
+from . import config
+from .gitea_client import GiteaClient
+
+
+class CIQuery:
+    """CI 状态查询器"""
+
+    def __init__(self, gitea_client=None):
+        self.gitea = gitea_client or GiteaClient()
+
+    # ── 查询方法 ──────────────────────────────────────
+
+    def get_branch_status(self, branch, limit=5):
+        """获取指定分支最新的 CI 状态
+
+        Returns:
+            dict: {branch, latest_run, recent_runs, overall_status}
+        """
+        runs, total = self.gitea.list_runs(branch=branch, limit=limit)
+        if not runs:
+            return {
+                "branch": branch,
+                "latest_run": None,
+                "recent_runs": [],
+                "overall_status": "no_runs",
+                "total_count": total,
+            }
+
+        latest = runs[0]
+        overall = self._derive_overall_status(runs)
+
+        return {
+            "branch": branch,
+            "latest_run": latest,
+            "recent_runs": runs,
+            "overall_status": overall,
+            "total_count": total,
+        }
+
+    def get_pr_status(self, pr_number):
+        """获取指定 PR 的 CI 状态
+
+        Returns:
+            dict: {pr_number, pr_title, runs, overall_status}
+        """
+        pr = self.gitea.get_pr(pr_number)
+        if not pr:
+            return {
+                "pr_number": pr_number,
+                "pr_title": "未知",
+                "runs": [],
+                "overall_status": "pr_not_found",
+            }
+
+        pr_title = pr.get("title", "")
+        runs = self.gitea.get_pr_ci_runs(pr_number, limit=10)
+        overall = self._derive_overall_status(runs) if runs else "no_runs"
+
+        return {
+            "pr_number": pr_number,
+            "pr_title": pr_title,
+            "runs": runs,
+            "overall_status": overall,
+            "head_sha": pr.get("head", {}).get("sha", ""),
+        }
+
+    def get_failure_detail(self, run_id):
+        """获取某次 run 的失败详情
+
+        Returns:
+            dict: {run_info, failed_jobs, summary}
+        """
+        run = self.gitea.get_run(run_id)
+        if not run:
+            return {"run_info": None, "failed_jobs": [], "summary": "Run not found"}
+
+        failed_jobs = self.gitea.get_failed_jobs_summary(run_id, max_lines_per_job=30)
+
+        summary_parts = []
+        for job in failed_jobs:
+            step = f"（步骤: {job['failed_step']}）" if job["failed_step"] else ""
+            summary_parts.append(f"• {job['name']}{step}")
+
+        summary = "\n".join(summary_parts) if summary_parts else "无失败 job（可能还在运行中）"
+
+        return {
+            "run_info": run,
+            "failed_jobs": failed_jobs,
+            "summary": summary,
+            "total_jobs": len(self.gitea.get_run_jobs(run_id)),
+        }
+
+    def list_recent_runs(self, status=None, branch=None, limit=10):
+        """列出最近的 runs"""
+        runs, total = self.gitea.list_runs(status=status, branch=branch, limit=limit)
+        return {"runs": runs, "total_count": total}
+
+    # ── 辅助方法 ────────────────────────────────────
+
+    @staticmethod
+    def _derive_overall_status(runs):
+        """根据最近 runs 推导整体状态
+
+        Returns:
+            success: 最近一次成功
+            failing: 最近一次失败（连续失败）
+            flaky: 有失败有成功（最近一次失败
+            running: 有正在运行的
+            unknown: 未知
+        """
+        if not runs:
+            return "no_runs"
+
+        # 检查是否有运行中的
+        running = [r for r in runs if r.get("status") != "completed"]
+        if running:
+            return "running"
+
+        # 看最近一次
+        latest = runs[0]
+        latest_conclusion = latest.get("conclusion", "unknown")
+
+        if latest_conclusion == "success":
+            return "success"
+
+        if latest_conclusion == "failure":
+            # 检查是否连续失败
+            consecutive_failures = 0
+            for r in runs:
+                if r.get("conclusion") == "failure":
+                    consecutive_failures += 1
+                else:
+                    break
+
+            # 看之前有没有成功
+            has_success = any(r.get("conclusion") == "success" for r in runs)
+
+            if has_success:
+                return "flaky"
+            return "failing"
+
+        return "unknown"
+
+    # ── 格式化输出 ────────────────────────────────────
+
+    @staticmethod
+    def format_branch_status(status_data):
+        """格式化分支状态为人类可读文本"""
+        branch = status_data["branch"]
+        latest = status_data["latest_run"]
+        overall = status_data["overall_status"]
+
+        status_emoji = {
+            "success": "✅",
+            "failing": "🔴",
+            "flaky": "🟡",
+            "running": "🔄",
+            "no_runs": "⚪",
+            "unknown": "❓",
+        }.get(overall, "❓")
+
+        lines = [f"**CI 状态：{branch} 分支**", f"整体状态: {status_emoji} {overall}"]
+
+        if latest:
+            name = latest.get("name", "Unknown")
+            conclusion = latest.get("conclusion", latest.get("status", "unknown"))
+            run_id = latest.get("id", "")
+            created = latest.get("created_at", "")[:16].replace("T", " ")
+            run_url = f"{config.GITEA_URL}/{config.GITEA_REPO}/actions/runs/{run_id}"
+            lines.append(f"最新: [{name} #{run_id}]({run_url}) - {conclusion} ({created})")
+
+        recent = status_data["recent_runs"]
+        if len(recent) > 1:
+            lines.append(f"\n最近 {len(recent)} 次:")
+            for r in recent[:5]:
+                c = r.get("conclusion", r.get("status", "?"))
+                emoji = {"success": "✅", "failure": "❌", "skipped": "⏭️"}.get(c, "🔄")
+                lines.append(f"  {emoji} #{r.get('id', '?')} {r.get('name', '?')[:30]} - {c}")
+
+        return "\n".join(lines)
+
+    @staticmethod
+    def format_pr_status(status_data):
+        """格式化 PR 状态为人类可读文本"""
+        pr_num = status_data["pr_number"]
+        pr_title = status_data["pr_title"]
+        overall = status_data["overall_status"]
+
+        status_emoji = {
+            "success": "✅",
+            "failing": "🔴",
+            "flaky": "🟡",
+            "running": "🔄",
+            "no_runs": "⚪",
+            "pr_not_found": "❓",
+            "unknown": "❓",
+        }.get(overall, "❓")
+
+        pr_url = f"{config.GITEA_URL}/{config.GITEA_REPO}/pulls/{pr_num}"
+        lines = [
+            f"**CI 状态：PR #{pr_num}**",
+            f"标题: [{pr_title}]({pr_url})",
+            f"状态: {status_emoji} {overall}",
+        ]
+
+        runs = status_data["runs"]
+        if runs:
+            lines.append(f"\nCI Runs ({len(runs)}):")
+            for r in runs[:5]:
+                c = r.get("conclusion", r.get("status", "?"))
+                emoji = {"success": "✅", "failure": "❌", "skipped": "⏭️"}.get(c, "🔄")
+                run_id = r.get("id", "?")
+                run_url = f"{config.GITEA_URL}/{config.GITEA_REPO}/actions/runs/{run_id}"
+                lines.append(f"  {emoji} [{r.get('name', '?')[:30]} #{run_id}]({run_url}) - {c}")
+
+        return "\n".join(lines)
+
+
+# ── CLI 入口 ──────────────────────────────────────────
+
+
+def main():
+    parser = argparse.ArgumentParser(description="CI 状态查询")
+    parser.add_argument("--branch", help="查询指定分支的 CI 状态")
+    parser.add_argument("--pr", type=int, help="查询指定 PR 的 CI 状态")
+    parser.add_argument("--run-id", help="查询指定 run 的详情")
+    parser.add_argument("--detail", action="store_true", help="显示失败详情")
+    parser.add_argument("--limit", type=int, default=5, help="返回数量限制")
+    parser.add_argument("--status", help="按状态过滤: success/failure/running")
+
+    args = parser.parse_args()
+
+    query = CIQuery()
+
+    if args.run_id:
+        if args.detail:
+            result = query.get_failure_detail(args.run_id)
+            print(f"Run #{args.run_id} 失败详情:")
+            print(result["summary"])
+            if result["failed_jobs"]:
+                print("\n详细日志尾部:")
+                for job in result["failed_jobs"]:
+                    print(f"\n--- {job['name']} ---")
+                    print(job["log_tail"][:500] if job["log_tail"] else "无日志")
+        else:
+            run = query.gitea.get_run(args.run_id)
+            if run:
+                print(f"Run #{args.run_id}: {run.get('name')} - {run.get('conclusion', run.get('status'))}")
+                print(f"分支: {run.get('head_branch', '?')}")
+                print(f"触发: {run.get('event', '?')}")
+            else:
+                print(f"Run {args.run_id} 不存在")
+    elif args.pr:
+        result = query.get_pr_status(args.pr)
+        print(CIQuery.format_pr_status(result))
+    elif args.branch:
+        result = query.get_branch_status(args.branch, limit=args.limit)
+        print(CIQuery.format_branch_status(result))
+    elif args.status:
+        result = query.list_recent_runs(status=args.status, limit=args.limit)
+        for r in result["runs"]:
+            print(
+                f"#{r.get('id')} {r.get('name')[:40]} - {r.get('conclusion', r.get('status'))} ({r.get('head_branch', '?')})"
+            )
+    else:
+        parser.print_help()
+        return 1
+
+    return 0
+
+
+if __name__ == "__main__":
+    sys.exit(main())

scripts/ci/chatops/ci_trigger.py

@@ -0,0 +1,169 @@
+#!/usr/bin/env python3
+"""
+CI 触发模块 - 重新运行失败 job、重跑整个 workflow、取消 run
+
+支持操作:
+  - rerun_failed: 重跑失败的 jobs
+  - rerun_all: 重跑整个 workflow
+  - cancel: 取消运行中的 run
+
+用法:
+  python3 scripts/ci/chatops/ci_trigger.py --run-id 456 --action rerun_failed
+  python3 scripts/ci/chatops/ci_trigger.py --run-id 456 --action rerun_all
+  python3 scripts/ci/chatops/ci_trigger.py --run-id 456 --action cancel
+
+设计:
+  - 与飞书机器人 /ci rerun 命令对接
+  - 操作前自动校验 run 状态，避免无效操作
+"""
+
+import argparse
+import sys
+
+from . import config
+from .gitea_client import GiteaClient
+
+
+class CITrigger:
+    """CI 操作触发器"""
+
+    def __init__(self, gitea_client=None):
+        self.gitea = gitea_client or GiteaClient()
+
+    # ── 触发操作 ─────────────────────────────────────
+
+    def rerun_failed(self, run_id):
+        """重跑失败的 jobs
+
+        Returns:
+            dict: {success, message, new_run_id?}
+        """
+        run = self.gitea.get_run(run_id)
+        if not run:
+            return {"success": False, "message": f"Run {run_id} 不存在"}
+
+        status = run.get("status", "")
+        if status != "completed":
+            return {
+                "success": False,
+                "message": f"Run {run_id} 当前状态为 {status}，仅 completed 状态才能重跑",
+            }
+
+        result = self.gitea.rerun_failed_jobs(run_id)
+        if result is None:
+            return {"success": False, "message": "重跑请求失败"}
+
+        # Gitea rerun 后返回的 run id 通常不变（复用原 run）
+        return {
+            "success": True,
+            "message": f"已触发重跑失败 jobs: Run #{run_id}",
+            "run_id": run_id,
+            "run_url": f"{config.GITEA_URL}/{config.GITEA_REPO}/actions/runs/{run_id}",
+        }
+
+    def rerun_all(self, run_id):
+        """重跑整个 workflow run
+
+        Returns:
+            dict: {success, message, run_id, run_url}
+        """
+        run = self.gitea.get_run(run_id)
+        if not run:
+            return {"success": False, "message": f"Run {run_id} 不存在"}
+
+        status = run.get("status", "")
+        if status == "running" or status == "pending":
+            return {
+                "success": False,
+                "message": f"Run {run_id} 正在运行中，无需重跑",
+            }
+
+        result = self.gitea.rerun_run(run_id)
+        if result is None:
+            return {"success": False, "message": "重跑请求失败"}
+
+        return {
+            "success": True,
+            "message": f"已触发完整重跑: Run #{run_id}",
+            "run_id": run_id,
+            "run_url": f"{config.GITEA_URL}/{config.GITEA_REPO}/actions/runs/{run_id}",
+        }
+
+    def cancel_run(self, run_id):
+        """取消运行中的 run
+
+        Returns:
+            dict: {success, message}
+        """
+        run = self.gitea.get_run(run_id)
+        if not run:
+            return {"success": False, "message": f"Run {run_id} 不存在"}
+
+        status = run.get("status", "")
+        if status == "completed":
+            return {
+                "success": False,
+                "message": f"Run {run_id} 已完成，无需取消",
+            }
+
+        result = self.gitea.cancel_run(run_id)
+        if result is None:
+            return {"success": False, "message": "取消请求失败"}
+
+        return {
+            "success": True,
+            "message": f"已取消 Run #{run_id}",
+            "run_id": run_id,
+        }
+
+    def rerun_latest_failed(self, branch="develop", workflow_id=None):
+        """重跑指定分支最近一次失败的 run
+
+        用于快速恢复场景，不需要先查 run_id
+        """
+        runs, _ = self.gitea.list_runs(branch=branch, workflow_id=workflow_id, status="failure", limit=5)
+        if not runs:
+            return {"success": False, "message": f"{branch} 分支没有失败的 run"}
+
+        latest = runs[0]
+        run_id = latest.get("id")
+        return self.rerun_failed(run_id)
+
+
+# ── CLI 入口 ──────────────────────────────────────────
+
+
+def main():
+    parser = argparse.ArgumentParser(description="CI 触发操作")
+    parser.add_argument("--run-id", required=True, help="Workflow Run ID")
+    parser.add_argument(
+        "--action",
+        required=True,
+        choices=["rerun_failed", "rerun_all", "cancel"],
+        help="操作类型",
+    )
+
+    args = parser.parse_args()
+
+    trigger = CITrigger()
+
+    if args.action == "rerun_failed":
+        result = trigger.rerun_failed(args.run_id)
+    elif args.action == "rerun_all":
+        result = trigger.rerun_all(args.run_id)
+    elif args.action == "cancel":
+        result = trigger.cancel_run(args.run_id)
+    else:
+        print(f"未知操作: {args.action}")
+        return 1
+
+    status = "✅" if result["success"] else "❌"
+    print(f"{status} {result['message']}")
+    if result.get("run_url"):
+        print(f"   {result['run_url']}")
+
+    return 0 if result["success"] else 1
+
+
+if __name__ == "__main__":
+    sys.exit(main())

scripts/ci/chatops/config.py

@@ -0,0 +1,55 @@
+#!/usr/bin/env python3
+"""
+ChatOps 配置管理 - 统一从环境变量读取配置，不硬编码任何敏感信息
+
+环境变量:
+  GITEA_URL              Gitea 地址 (默认 https://git.xiaoxiajianji.com)
+  GITEA_REPO             仓库路径 (默认 xiaoxia/xiaoxia-saas)
+  GITEA_TOKEN            Gitea API Token (优先使用)
+  GITEA_USERNAME         Gitea 用户名 (密码认证时)
+  GITEA_PASSWORD         Gitea 密码 (密码认证时)
+  FEISHU_WEBHOOK_URL     飞书自定义机器人 webhook 地址
+  FEISHU_APP_ID          飞书应用 App ID (应用机器人模式，预留)
+  FEISHU_APP_SECRET      飞书应用 App Secret (应用机器人模式，预留)
+  CHATOPS_NOTIFY_BRANCHES  触发通知的分支，逗号分隔 (默认 main,develop)
+  CHATOPS_WEBHOOK_PORT   webhook 服务监听端口 (默认 8090)
+  CHATOPS_WEBHOOK_SECRET Gitea webhook 密钥 (校验签名，可选)
+"""
+
+import os
+
+# ── Gitea 配置 ────────────────────────────────────────
+GITEA_URL = os.environ.get("GITEA_URL", "https://git.xiaoxiajianji.com").rstrip("/")
+GITEA_REPO = os.environ.get("GITEA_REPO", "xiaoxia/xiaoxia-saas")
+GITEA_TOKEN = os.environ.get("GITEA_TOKEN", "")
+GITEA_USERNAME = os.environ.get("GITEA_USERNAME", "")
+GITEA_PASSWORD = os.environ.get("GITEA_PASSWORD", "")
+
+# ── 飞书配置 ──────────────────────────────────────────
+FEISHU_WEBHOOK_URL = os.environ.get("FEISHU_WEBHOOK_URL", "")
+FEISHU_APP_ID = os.environ.get("FEISHU_APP_ID", "")
+FEISHU_APP_SECRET = os.environ.get("FEISHU_APP_SECRET", "")
+
+# ── 通知配置 ──────────────────────────────────────────
+NOTIFY_BRANCHES = [b.strip() for b in os.environ.get("CHATOPS_NOTIFY_BRANCHES", "main,develop").split(",") if b.strip()]
+
+# ── Webhook 服务配置 ──────────────────────────────────
+WEBHOOK_PORT = int(os.environ.get("CHATOPS_WEBHOOK_PORT", "8090"))
+WEBHOOK_SECRET = os.environ.get("CHATOPS_WEBHOOK_SECRET", "")
+
+# ── 常量 ──────────────────────────────────────────────
+PAGE_LIMIT = 50  # Gitea API 每页最大数量
+
+
+def has_gitea_auth() -> bool:
+    """检查是否配置了 Gitea 认证信息"""
+    if GITEA_TOKEN:
+        return True
+    if GITEA_USERNAME and GITEA_PASSWORD:
+        return True
+    return False
+
+
+def has_feishu_webhook() -> bool:
+    """检查是否配置了飞书 webhook"""
+    return bool(FEISHU_WEBHOOK_URL)

scripts/ci/chatops/feishu_notify.py

@@ -0,0 +1,390 @@
+#!/usr/bin/env python3
+"""
+飞书通知模块 - CI 关键事件推送
+
+支持通知类型:
+  - branch_failure: main/develop 分支 CI 失败
+  - branch_recovery: main/develop 分支 CI 从失败恢复（绿色恢复）
+  - e2e_failure: E2E 测试失败摘要
+  - pr_failure: PR CI 失败（可选）
+
+用法:
+  # 命令行直接调用（供 CI workflow 使用）
+  python3 -m scripts.ci.chatops.feishu_notify --mode failure --run-id 12345
+  python3 scripts/ci/chatops/feishu_notify.py --mode recovery --run-id 12345
+
+  # Python 模块调用
+  from scripts.ci.chatops.feishu_notify import FeishuNotifier
+  notifier = FeishuNotifier()
+  notifier.notify_branch_failure(run_id=12345, branch="develop")
+
+设计原则:
+  1. 通知失败永远不阻断主流程（返回 0）
+  2. 卡片信息丰富，一键跳转 Gitea 详情页
+  3. 失败通知包含错误摘要，不用点进去就能判断严重程度
+"""
+
+import argparse
+import json
+import sys
+import urllib.request
+from datetime import datetime
+
+from . import config
+from .gitea_client import GiteaClient
+
+
+class FeishuNotifier:
+    """飞书通知发送器"""
+
+    def __init__(self, webhook_url=None, gitea_client=None):
+        self.webhook_url = webhook_url or config.FEISHU_WEBHOOK_URL
+        self.gitea = gitea_client or GiteaClient()
+
+    def _send_card(self, card_payload):
+        """发送飞书卡片消息
+
+        Returns:
+            True 表示发送成功（飞书返回 code=0）
+        """
+        if not self.webhook_url:
+            print("[INFO] 未配置 FEISHU_WEBHOOK_URL，跳过飞书通知")
+            return False
+
+        payload = {"msg_type": "interactive", "card": card_payload}
+        data = json.dumps(payload).encode("utf-8")
+        req = urllib.request.Request(
+            self.webhook_url,
+            data=data,
+            headers={"Content-Type": "application/json"},
+            method="POST",
+        )
+        try:
+            with urllib.request.urlopen(req, timeout=10) as resp:
+                resp_body = resp.read().decode("utf-8")
+            result = json.loads(resp_body)
+            if result.get("code", 0) != 0:
+                print(
+                    f"[WARN] 飞书通知返回错误: {result.get('msg', resp_body)}",
+                    file=sys.stderr,
+                )
+                return False
+            return True
+        except Exception as e:
+            print(f"[WARN] 飞书通知发送失败: {e}", file=sys.stderr)
+            return False
+
+    # ── 通知模板 ──────────────────────────────────────
+
+    def _run_url(self, run_id):
+        return f"{config.GITEA_URL}/{config.GITEA_REPO}/actions/runs/{run_id}"
+
+    def _pr_url(self, pr_number):
+        return f"{config.GITEA_URL}/{config.GITEA_REPO}/pulls/{pr_number}"
+
+    def notify_branch_failure(self, run_id, branch, run_data=None):
+        """main/develop 分支 CI 失败通知
+
+        包含: 失败 job 列表、错误摘要、一键重跑链接
+        """
+        run = run_data or self.gitea.get_run(run_id)
+        if not run:
+            print(f"[WARN] 无法获取 run {run_id} 详情", file=sys.stderr)
+            return False
+
+        workflow_name = run.get("name", "Unknown Workflow")
+        commit_msg = run.get("head_commit", {}).get("message", "未知").splitlines()[0][:60]
+        commit_sha = run.get("head_sha", "")[:8]
+        actor = (
+            run.get("trigger_event", {}).get("actor", {}).get("login", "unknown")
+            if isinstance(run.get("trigger_event"), dict)
+            else run.get("actor", "unknown")
+        )
+        run_url = self._run_url(run_id)
+
+        # 获取失败 job 摘要
+        failed_jobs = self.gitea.get_failed_jobs_summary(run_id, max_lines_per_job=15)
+
+        # 构建失败摘要
+        failure_summary = ""
+        if failed_jobs:
+            job_lines = []
+            for job in failed_jobs[:3]:  # 最多显示 3 个
+                step_info = f"（{job['failed_step']}）" if job["failed_step"] else ""
+                job_lines.append(f"• **{job['name']}**{step_info}")
+                if job["log_tail"]:
+                    # 取最后 3 行日志
+                    tail_lines = job["log_tail"].strip().splitlines()[-3:]
+                    for line in tail_lines:
+                        clean_line = line.strip()[:100]
+                        if clean_line:
+                            job_lines.append(f"  `{clean_line}`")
+            failure_summary = "\n".join(job_lines)
+        else:
+            failure_summary = "（获取失败详情中，点击查看日志）"
+
+        # 字段
+        fields = [
+            {"is_short": True, "text": {"tag": "lark_md", "content": f"**分支**\n{branch}"}},
+            {"is_short": True, "text": {"tag": "lark_md", "content": f"**Workflow**\n{workflow_name}"}},
+            {"is_short": True, "text": {"tag": "lark_md", "content": f"**提交**\n`{commit_sha}`"}},
+            {"is_short": True, "text": {"tag": "lark_md", "content": f"**触发者**\n{actor}"}},
+            {"is_short": False, "text": {"tag": "lark_md", "content": f"**提交信息**\n{commit_msg}"}},
+            {"is_short": False, "text": {"tag": "lark_md", "content": f"**失败详情**\n{failure_summary}"}},
+        ]
+
+        card = {
+            "header": {
+                "title": {
+                    "tag": "plain_text",
+                    "content": f"❌ CI告警：{branch} 分支构建失败",
+                },
+                "status": "red",
+            },
+            "elements": [
+                {"tag": "div", "fields": fields},
+                {
+                    "tag": "action",
+                    "actions": [
+                        {
+                            "tag": "button",
+                            "text": {"tag": "plain_text", "content": "查看失败日志"},
+                            "url": run_url,
+                            "type": "danger",
+                        },
+                        {
+                            "tag": "button",
+                            "text": {"tag": "plain_text", "content": "重跑失败Job"},
+                            "url": f"{run_url}/rerun-failed-jobs",
+                            "type": "default",
+                        },
+                    ],
+                },
+            ],
+        }
+
+        result = self._send_card(card)
+        print(f"[INFO] 分支失败通知已发送: {branch} run={run_id}")
+        return result
+
+    def notify_branch_recovery(self, run_id, branch, previous_failure_run_id=None):
+        """分支 CI 恢复通知（从失败变成功）"""
+        run = self.gitea.get_run(run_id)
+        if not run:
+            print(f"[WARN] 无法获取 run {run_id} 详情", file=sys.stderr)
+            return False
+
+        workflow_name = run.get("name", "Unknown Workflow")
+        commit_sha = run.get("head_sha", "")[:8]
+        run_url = self._run_url(run_id)
+
+        # 计算恢复耗时
+        duration_text = "已恢复"
+        if previous_failure_run_id:
+            prev_run = self.gitea.get_run(previous_failure_run_id)
+            if prev_run:
+                # 简单计算两个 run 的时间差
+                prev_time = prev_run.get("created_at", "")
+                cur_time = run.get("created_at", "")
+                if prev_time and cur_time:
+                    try:
+                        t1 = datetime.fromisoformat(prev_time.replace("Z", "+00:00"))
+                        t2 = datetime.fromisoformat(cur_time.replace("Z", "+00:00"))
+                        diff = (t2 - t1).total_seconds() / 60
+                        duration_text = f"故障时长约 {diff:.0f} 分钟"
+                    except Exception:
+                        pass
+
+        fields = [
+            {"is_short": True, "text": {"tag": "lark_md", "content": f"**分支**\n{branch}"}},
+            {"is_short": True, "text": {"tag": "lark_md", "content": f"**Workflow**\n{workflow_name}"}},
+            {"is_short": True, "text": {"tag": "lark_md", "content": f"**提交**\n`{commit_sha}`"}},
+            {"is_short": True, "text": {"tag": "lark_md", "content": "**状态**\n✅ 已恢复"}},
+            {"is_short": False, "text": {"tag": "lark_md", "content": f"**说明**\n{duration_text}"}},
+        ]
+
+        card = {
+            "header": {
+                "title": {
+                    "tag": "plain_text",
+                    "content": f"✅ CI通知：{branch} 分支构建已恢复",
+                },
+                "status": "green",
+            },
+            "elements": [
+                {"tag": "div", "fields": fields},
+                {
+                    "tag": "action",
+                    "actions": [
+                        {
+                            "tag": "button",
+                            "text": {"tag": "plain_text", "content": "查看详情"},
+                            "url": run_url,
+                            "type": "primary",
+                        }
+                    ],
+                },
+            ],
+        }
+
+        result = self._send_card(card)
+        print(f"[INFO] 分支恢复通知已发送: {branch} run={run_id}")
+        return result
+
+    def notify_e2e_failure(self, run_id, branch="develop", pr_number=None):
+        """E2E 测试失败摘要通知"""
+        failed_jobs = self.gitea.get_failed_jobs_summary(run_id, max_lines_per_job=50)
+        e2e_jobs = [j for j in failed_jobs if "e2e" in j["name"].lower() or "test" in j["name"].lower()]
+
+        if not e2e_jobs:
+            # 没有明确的 e2e job，取所有失败的
+            e2e_jobs = failed_jobs
+
+        run_url = self._run_url(run_id)
+        title_suffix = f"PR #{pr_number}" if pr_number else f"{branch} 分支"
+
+        # 构建失败用例摘要
+        case_summary = ""
+        for job in e2e_jobs[:3]:
+            case_summary += f"**{job['name']}**\n"
+            if job["log_tail"]:
+                # 尝试提取 FAIL 行
+                fail_lines = [
+                    line.strip()
+                    for line in job["log_tail"].splitlines()
+                    if "FAIL" in line or "fail" in line.lower() or "✗" in line or "●" in line
+                ][:5]
+                if fail_lines:
+                    for line in fail_lines:
+                        case_summary += f"  • {line[:120]}\n"
+                else:
+                    tail = job["log_tail"].strip().splitlines()[-5:]
+                    for line in tail:
+                        case_summary += f"  `{line.strip()[:100]}`\n"
+            case_summary += "\n"
+
+        if not case_summary:
+            case_summary = "（点击查看完整测试报告）"
+
+        fields = [
+            {"is_short": True, "text": {"tag": "lark_md", "content": f"**来源**\n{title_suffix}"}},
+            {"is_short": True, "text": {"tag": "lark_md", "content": f"**失败Job数**\n{len(e2e_jobs)}"}},
+            {"is_short": False, "text": {"tag": "lark_md", "content": f"**失败摘要**\n{case_summary}"}},
+        ]
+
+        card = {
+            "header": {
+                "title": {
+                    "tag": "plain_text",
+                    "content": f"🧪 CI告警：E2E 测试失败 - {title_suffix}",
+                },
+                "status": "orange",
+            },
+            "elements": [
+                {"tag": "div", "fields": fields},
+                {
+                    "tag": "action",
+                    "actions": [
+                        {
+                            "tag": "button",
+                            "text": {"tag": "plain_text", "content": "查看完整报告"},
+                            "url": run_url,
+                            "type": "danger",
+                        }
+                    ],
+                },
+            ],
+        }
+
+        result = self._send_card(card)
+        print(f"[INFO] E2E失败通知已发送: run={run_id}")
+        return result
+
+    def notify_pr_failure(self, run_id, pr_number, pr_title=""):
+        """PR CI 失败通知（轻量版，可选开启）"""
+        run_url = self._run_url(run_id)
+        pr_url = self._pr_url(pr_number)
+
+        failed_jobs = self.gitea.get_failed_jobs_summary(run_id, max_lines_per_job=10)
+        failure_names = [j["name"] for j in failed_jobs[:3]]
+        failure_text = "、".join(failure_names) if failure_names else "未知"
+
+        fields = [
+            {
+                "is_short": False,
+                "text": {"tag": "lark_md", "content": f"**PR**\n[#{pr_number} {pr_title[:50]}]({pr_url})"},
+            },
+            {"is_short": False, "text": {"tag": "lark_md", "content": f"**失败任务**\n{failure_text}"}},
+        ]
+
+        card = {
+            "header": {
+                "title": {
+                    "tag": "plain_text",
+                    "content": f"⚠️ CI通知：PR #{pr_number} 构建失败",
+                },
+                "status": "yellow",
+            },
+            "elements": [
+                {"tag": "div", "fields": fields},
+                {
+                    "tag": "action",
+                    "actions": [
+                        {
+                            "tag": "button",
+                            "text": {"tag": "plain_text", "content": "查看日志"},
+                            "url": run_url,
+                            "type": "default",
+                        },
+                        {
+                            "tag": "button",
+                            "text": {"tag": "plain_text", "content": "查看PR"},
+                            "url": pr_url,
+                            "type": "default",
+                        },
+                    ],
+                },
+            ],
+        }
+
+        result = self._send_card(card)
+        print(f"[INFO] PR失败通知已发送: PR #{pr_number} run={run_id}")
+        return result
+
+
+# ── CLI 入口 ──────────────────────────────────────────
+
+
+def main():
+    parser = argparse.ArgumentParser(description="飞书 CI 通知")
+    parser.add_argument(
+        "--mode",
+        required=True,
+        choices=["failure", "recovery", "e2e_failure", "pr_failure"],
+        help="通知模式",
+    )
+    parser.add_argument("--run-id", required=True, help="Workflow Run ID")
+    parser.add_argument("--branch", default="develop", help="分支名")
+    parser.add_argument("--pr-number", type=int, help="PR 编号")
+    parser.add_argument("--pr-title", default="", help="PR 标题")
+    parser.add_argument("--prev-run-id", help="上一个失败的 run ID（恢复通知用）")
+    parser.add_argument("--webhook", help="飞书 webhook URL（覆盖环境变量）")
+
+    args = parser.parse_args()
+
+    notifier = FeishuNotifier(webhook_url=args.webhook)
+
+    if args.mode == "failure":
+        notifier.notify_branch_failure(args.run_id, args.branch)
+    elif args.mode == "recovery":
+        notifier.notify_branch_recovery(args.run_id, args.branch, previous_failure_run_id=args.prev_run_id)
+    elif args.mode == "e2e_failure":
+        notifier.notify_e2e_failure(args.run_id, branch=args.branch, pr_number=args.pr_number)
+    elif args.mode == "pr_failure":
+        notifier.notify_pr_failure(args.run_id, args.pr_number, pr_title=args.pr_title)
+
+    return 0
+
+
+if __name__ == "__main__":
+    sys.exit(main())

scripts/ci/chatops/gitea_client.py

@@ -0,0 +1,243 @@
+#!/usr/bin/env python3
+"""
+Gitea API 客户端封装 - Actions + PR + Webhook 相关接口
+
+基于 urllib 实现，无第三方依赖，与 ci_dashboard.py 风格一致。
+支持 token 和 basic auth 两种认证方式。
+"""
+
+import base64
+import json
+import sys
+import urllib.error
+import urllib.request
+
+from . import config
+
+
+class GiteaClient:
+    """Gitea API 客户端"""
+
+    def __init__(
+        self,
+        base_url=None,
+        repo=None,
+        token=None,
+        username=None,
+        password=None,
+    ):
+        self.base_url = (base_url or config.GITEA_URL).rstrip("/")
+        self.repo = repo or config.GITEA_REPO
+        self.token = token or config.GITEA_TOKEN
+        self.username = username or config.GITEA_USERNAME
+        self.password = password or config.GITEA_PASSWORD
+        self.api_base = f"{self.base_url}/api/v1/repos/{self.repo}"
+
+    def _request(self, path, method="GET", data=None):
+        """通用 HTTP 请求
+
+        Args:
+            path: API 路径（相对于 /api/v1/repos/{repo}/）
+            method: HTTP 方法
+            data: 请求体（dict 或 bytes）
+
+        Returns:
+            解析后的 JSON 数据，失败返回 None
+        """
+        url = f"{self.api_base}/{path}"
+        body = None
+        if data is not None:
+            if isinstance(data, (dict, list)):
+                body = json.dumps(data).encode("utf-8")
+            else:
+                body = data if isinstance(data, bytes) else str(data).encode()
+
+        req = urllib.request.Request(url, data=body, method=method)
+        req.add_header("Content-Type", "application/json")
+
+        if self.token:
+            req.add_header("Authorization", f"token {self.token}")
+        elif self.username and self.password:
+            auth = base64.b64encode(f"{self.username}:{self.password}".encode()).decode()
+            req.add_header("Authorization", f"Basic {auth}")
+
+        try:
+            with urllib.request.urlopen(req, timeout=30) as resp:
+                resp_body = resp.read().decode()
+                if not resp_body:
+                    return {}
+                return json.loads(resp_body)
+        except urllib.error.HTTPError as e:
+            err_body = ""
+            try:
+                err_body = e.read().decode()
+            except Exception:
+                pass
+            print(
+                f"[WARN] HTTP {e.code}: {url} - {err_body[:200]}",
+                file=sys.stderr,
+            )
+            return None
+        except Exception as e:
+            print(f"[WARN] 请求失败 {url}: {e}", file=sys.stderr)
+            return None
+
+    # ── Actions: Workflow Runs ────────────────────────
+
+    def list_runs(
+        self,
+        status=None,
+        branch=None,
+        event=None,
+        workflow_id=None,
+        page=1,
+        limit=config.PAGE_LIMIT,
+    ):
+        """获取 workflow runs 列表
+
+        Returns:
+            (runs列表, 总数)
+        """
+        params = []
+        if status:
+            params.append(f"status={status}")
+        if branch:
+            params.append(f"branch={branch}")
+        if event:
+            params.append(f"event={event}")
+        if workflow_id:
+            params.append(f"workflow_id={workflow_id}")
+        params.append(f"page={page}")
+        params.append(f"limit={limit}")
+        path = f"actions/runs?{'&'.join(params)}"
+        data = self._request(path)
+        if not data:
+            return [], 0
+        runs = data.get("workflow_runs", [])
+        total = data.get("total_count", 0)
+        return runs, total
+
+    def get_run(self, run_id):
+        """获取单个 run 详情"""
+        return self._request(f"actions/runs/{run_id}")
+
+    def get_run_jobs(self, run_id):
+        """获取 run 的 jobs 列表"""
+        data = self._request(f"actions/runs/{run_id}/jobs")
+        if not data:
+            return []
+        return data.get("jobs", [])
+
+    def get_job_log(self, run_id, job_id):
+        """获取 job 日志（纯文本）"""
+        url = f"{self.api_base}/actions/runs/{run_id}/jobs/{job_id}/logs"
+        req = urllib.request.Request(url)
+        if self.token:
+            req.add_header("Authorization", f"token {self.token}")
+        elif self.username and self.password:
+            auth = base64.b64encode(f"{self.username}:{self.password}".encode()).decode()
+            req.add_header("Authorization", f"Basic {auth}")
+        try:
+            with urllib.request.urlopen(req, timeout=30) as resp:
+                return resp.read().decode("utf-8", errors="replace")
+        except Exception as e:
+            print(f"[WARN] 获取日志失败 job={job_id}: {e}", file=sys.stderr)
+            return ""
+
+    def rerun_run(self, run_id):
+        """重新运行整个 workflow run"""
+        return self._request(f"actions/runs/{run_id}/rerun", method="POST")
+
+    def rerun_failed_jobs(self, run_id):
+        """重新运行失败的 jobs"""
+        return self._request(f"actions/runs/{run_id}/rerun-failed-jobs", method="POST")
+
+    def cancel_run(self, run_id):
+        """取消 run"""
+        return self._request(f"actions/runs/{run_id}/cancel", method="POST")
+
+    # ── Actions: Workflows ────────────────────────────
+
+    def list_workflows(self):
+        """获取 workflow 列表"""
+        data = self._request("actions/workflows")
+        if not data:
+            return []
+        return data.get("workflows", [])
+
+    def get_workflow(self, workflow_id):
+        """获取单个 workflow 详情"""
+        return self._request(f"actions/workflows/{workflow_id}")
+
+    # ── Pull Requests ─────────────────────────────────
+
+    def get_pr(self, pr_number):
+        """获取 PR 详情"""
+        return self._request(f"pulls/{pr_number}")
+
+    def get_pr_ci_runs(self, pr_number, limit=20):
+        """获取 PR 关联的 CI runs（通过 head_sha 查询）"""
+        pr = self.get_pr(pr_number)
+        if not pr:
+            return []
+        head_sha = pr.get("head", {}).get("sha", "")
+        if not head_sha:
+            return []
+        # 用 head_sha 过滤 runs
+        runs, _ = self.list_runs(limit=limit)
+        return [r for r in runs if r.get("head_sha", "") == head_sha]
+
+    # ── 便捷方法 ──────────────────────────────────────
+
+    def get_latest_run(self, branch, workflow_id=None, status=None):
+        """获取指定分支最新的 run"""
+        runs, _ = self.list_runs(branch=branch, workflow_id=workflow_id, status=status, limit=5)
+        return runs[0] if runs else None
+
+    def get_failed_jobs_summary(self, run_id, max_lines_per_job=30):
+        """获取失败 job 的摘要信息（用于通知）
+
+        Returns:
+            list[dict]: 每个失败 job 的 {name, conclusion, failed_step, log_tail}
+        """
+        jobs = self.get_run_jobs(run_id)
+        if not jobs:
+            return []
+
+        failed = [j for j in jobs if j.get("status") == "completed" and j.get("conclusion") == "failure"]
+        if not failed:
+            # 运行中的也返回，方便定位
+            failed = [j for j in jobs if j.get("status") != "completed"]
+
+        result = []
+        for job in failed[:5]:  # 最多取 5 个失败 job
+            job_id = job.get("id", "")
+            name = job.get("name", "Unknown")
+            conclusion = job.get("conclusion", job.get("status", "unknown"))
+
+            # 找失败的 step
+            failed_step = ""
+            steps = job.get("steps", [])
+            for step in steps:
+                if step.get("conclusion") == "failure":
+                    failed_step = step.get("name", "")
+                    break
+
+            # 取日志尾部
+            log_tail = ""
+            if job_id:
+                log = self.get_job_log(run_id, job_id)
+                if log:
+                    lines = log.strip().splitlines()
+                    log_tail = "\n".join(lines[-max_lines_per_job:])
+
+            result.append(
+                {
+                    "name": name,
+                    "conclusion": conclusion,
+                    "failed_step": failed_step,
+                    "log_tail": log_tail,
+                    "job_id": job_id,
+                }
+            )
+        return result

scripts/ci/chatops/webhook_server.py

@@ -0,0 +1,446 @@
+#!/usr/bin/env python3
+"""
+Gitea Webhook 接收服务 - FastAPI 实现
+
+功能:
+  1. 接收 Gitea Actions webhook 事件，触发飞书通知
+  2. 接收飞书机器人回调消息，处理 /ci 交互命令
+  3. 维护简单的状态缓存，检测分支恢复等状态变化
+
+部署:
+  部署到构建服务器，监听 8090 端口（可配置）
+  Gitea webhook 指向: http://<server>:8090/webhook/gitea
+  飞书消息回调指向: http://<server>:8090/webhook/feishu
+
+依赖:
+  fastapi + uvicorn（可选，未安装时仅模块可用，服务不可启动）
+
+注意:
+  本文件为第一版骨架，通知逻辑已实现，飞书交互命令待后续完善。
+"""
+
+import hashlib
+import hmac
+import json
+import sys
+import threading
+import time
+from typing import Optional
+
+from . import config
+from .gitea_client import GiteaClient
+
+# FastAPI 是可选依赖，未安装时仅导出类不启动服务
+try:
+    from fastapi import FastAPI, Header, HTTPException, Request
+    from fastapi.responses import JSONResponse
+
+    FASTAPI_AVAILABLE = True
+except ImportError:
+    FASTAPI_AVAILABLE = False
+    FastAPI = None  # type: ignore
+
+
+# ── 状态缓存 ──────────────────────────────────────────
+
+
+class StateCache:
+    """简单的内存状态缓存，用于检测状态变化
+
+    记录每个分支最后一次 run 的状态，用于判断:
+    - 是否从失败变成功（恢复通知）
+    - 是否连续失败（避免重复告警）
+    """
+
+    def __init__(self, max_entries=100):
+        self._cache = {}  # {branch: {last_status, last_run_id, last_notified_failure}}
+        self._lock = threading.Lock()
+        self._max = max_entries
+
+    def get(self, key):
+        with self._lock:
+            return self._cache.get(key)
+
+    def set(self, key, value):
+        with self._lock:
+            self._cache[key] = value
+            # 简单的淘汰策略
+            if len(self._cache) > self._max:
+                oldest_key = next(iter(self._cache))
+                del self._cache[oldest_key]
+
+    def check_and_update(self, branch, run_id, conclusion):
+        """检查状态变化并更新缓存
+
+        Returns:
+            dict: {is_new_failure, is_recovery, previous_status, previous_run_id}
+        """
+        prev = self.get(branch) or {}
+        prev_status = prev.get("last_status", "unknown")
+        prev_run_id = prev.get("last_run_id")
+
+        is_new_failure = False
+        is_recovery = False
+
+        if conclusion == "failure" and prev_status != "failure":
+            is_new_failure = True
+        if conclusion == "success" and prev_status == "failure":
+            is_recovery = True
+
+        self.set(
+            branch,
+            {
+                "last_status": conclusion,
+                "last_run_id": run_id,
+                "last_updated": time.time(),
+                "last_notified_failure": run_id if is_new_failure else prev.get("last_notified_failure"),
+            },
+        )
+
+        return {
+            "is_new_failure": is_new_failure,
+            "is_recovery": is_recovery,
+            "previous_status": prev_status,
+            "previous_run_id": prev_run_id,
+        }
+
+
+state_cache = StateCache()
+
+
+# ── Gitea Webhook 处理 ────────────────────────────────
+
+
+def verify_gitea_signature(payload: bytes, signature: str) -> bool:
+    """校验 Gitea webhook 签名（X-Gitea-Signature）
+
+    Gitea 使用 HMAC-SHA256 签名，格式: sha256=xxx
+    """
+    if not config.WEBHOOK_SECRET:
+        return True  # 未配置密钥则跳过校验
+
+    if not signature:
+        return False
+
+    try:
+        algo, sig_hex = signature.split("=", 1)
+        if algo != "sha256":
+            return False
+        expected = hmac.new(config.WEBHOOK_SECRET.encode(), payload, hashlib.sha256).hexdigest()
+        return hmac.compare_digest(expected, sig_hex)
+    except Exception:
+        return False
+
+
+def handle_gitea_webhook(payload: dict, event_type: str) -> dict:
+    """处理 Gitea webhook 事件
+
+    Args:
+        payload: webhook 请求体
+        event_type: X-Gitea-Event 头
+
+    Returns:
+        dict: {handled, notifications_sent, message}
+    """
+    if event_type != "create" and event_type != "push":
+        # 我们主要关心 push 和 actions 事件
+        # Gitea Actions 的 webhook 事件类型可能是 "push" 或专门的 actions 事件
+        pass
+
+    # 尝试提取 run 信息
+    run_info = _extract_run_info(payload)
+    if not run_info:
+        return {"handled": False, "notifications_sent": 0, "message": "非 CI 事件，跳过"}
+
+    branch = run_info["branch"]
+    run_id = run_info["run_id"]
+    status = run_info["status"]
+    conclusion = run_info.get("conclusion", "")
+
+    # 只处理已完成的 run
+    if status != "completed":
+        return {"handled": True, "notifications_sent": 0, "message": f"Run {run_id} 仍在运行中 ({status})"}
+
+    # 检查是否在通知分支列表中
+    if branch not in config.NOTIFY_BRANCHES:
+        return {
+            "handled": True,
+            "notifications_sent": 0,
+            "message": f"分支 {branch} 不在通知列表中",
+        }
+
+    # 检查状态变化
+    change_info = state_cache.check_and_update(branch, run_id, conclusion)
+    notifications = 0
+
+    # 延迟导入，避免循环依赖
+    from .feishu_notify import FeishuNotifier
+
+    notifier = FeishuNotifier()
+
+    if conclusion == "failure" and change_info["is_new_failure"]:
+        # 新失败 → 发失败通知
+        notifier.notify_branch_failure(run_id, branch)
+        notifications += 1
+
+        # 检查是否是 E2E 失败
+        gitea = GiteaClient()
+        failed_jobs = gitea.get_failed_jobs_summary(run_id)
+        has_e2e = any("e2e" in j["name"].lower() for j in failed_jobs)
+        if has_e2e:
+            notifier.notify_e2e_failure(run_id, branch=branch)
+            notifications += 1
+
+    elif conclusion == "success" and change_info["is_recovery"]:
+        # 从失败恢复 → 发恢复通知
+        prev_run_id = change_info.get("previous_run_id")
+        notifier.notify_branch_recovery(run_id, branch, previous_failure_run_id=prev_run_id)
+        notifications += 1
+
+    return {
+        "handled": True,
+        "notifications_sent": notifications,
+        "message": f"分支 {branch} run {run_id} {conclusion}",
+    }
+
+
+def _extract_run_info(payload: dict) -> Optional[dict]:
+    """从 webhook payload 中提取 run 信息
+
+    Gitea Actions webhook 的 payload 结构可能不同，这里做兼容处理。
+    如果 payload 不是 run 事件，返回 None。
+    """
+    # 尝试多种可能的结构
+    if "workflow_run" in payload:
+        wr = payload["workflow_run"]
+        return {
+            "run_id": wr.get("id"),
+            "branch": wr.get("head_branch", ""),
+            "status": wr.get("status", ""),
+            "conclusion": wr.get("conclusion", ""),
+            "name": wr.get("name", ""),
+        }
+
+    if "action" in payload and "pull_request" in payload:
+        # PR 事件，暂不处理
+        return None
+
+    if "ref" in payload and "head_commit" in payload:
+        # push 事件，不是 run 事件
+        return None
+
+    return None
+
+
+# ── 飞书消息处理 ──────────────────────────────────────
+
+
+def handle_feishu_message(payload: dict) -> dict:
+    """处理飞书机器人回调消息
+
+    支持命令:
+      /ci status [branch] - 查询分支 CI 状态
+      /ci rerun <run-id>  - 重跑失败的 jobs
+      /ci help            - 帮助
+
+    注意: 第一版骨架，仅解析命令，实际执行逻辑待完善。
+    """
+    # 飞书消息回调格式
+    header = payload.get("header", {})
+    event_type = header.get("event_type", "")
+
+    if event_type == "url_verification":
+        # 飞书 URL 验证
+        return {"challenge": payload.get("challenge", "")}
+
+    if event_type != "im.message.receive_v1":
+        return {"handled": False, "message": f"非消息事件: {event_type}"}
+
+    event = payload.get("event", {})
+    message = event.get("message", {})
+    content_str = message.get("content", "{}")
+
+    try:
+        content = json.loads(content_str)
+    except json.JSONDecodeError:
+        content = {}
+
+    text = content.get("text", "")
+    if not text:
+        return {"handled": False, "message": "空消息"}
+
+    # 解析命令
+    text = text.strip()
+    if not text.startswith("/ci"):
+        return {"handled": False, "message": "非 CI 命令"}
+
+    parts = text.split()
+    if len(parts) < 2:
+        return _help_response()
+
+    cmd = parts[1].lower()
+
+    if cmd == "status":
+        branch = parts[2] if len(parts) > 2 else "develop"
+        return _handle_status_command(branch)
+
+    elif cmd == "rerun":
+        if len(parts) < 3:
+            return {"text": "用法: /ci rerun <run-id> 或 /ci rerun latest [branch]"}
+        arg = parts[2]
+        if arg == "latest":
+            branch = parts[3] if len(parts) > 3 else "develop"
+            return _handle_rerun_latest(branch)
+        return _handle_rerun_command(arg)
+
+    elif cmd == "help":
+        return _help_response()
+
+    else:
+        return {"text": f"未知命令: {cmd}\n输入 /ci help 查看帮助"}
+
+
+def _handle_status_command(branch: str) -> dict:
+    """处理 /ci status 命令"""
+    from .ci_query import CIQuery
+
+    query = CIQuery()
+    result = query.get_branch_status(branch)
+    reply = CIQuery.format_branch_status(result)
+    return {"text": reply}
+
+
+def _handle_rerun_command(run_id: str) -> dict:
+    """处理 /ci rerun 命令"""
+    from .ci_trigger import CITrigger
+
+    trigger = CITrigger()
+    try:
+        result = trigger.rerun_failed(int(run_id))
+    except ValueError:
+        return {"text": f"无效的 run id: {run_id}"}
+
+    if result["success"]:
+        return {"text": f"✅ {result['message']}\n{result.get('run_url', '')}"}
+    return {"text": f"❌ {result['message']}"}
+
+
+def _handle_rerun_latest(branch: str) -> dict:
+    """处理 /ci rerun latest 命令"""
+    from .ci_trigger import CITrigger
+
+    trigger = CITrigger()
+    result = trigger.rerun_latest_failed(branch=branch)
+
+    if result["success"]:
+        return {"text": f"✅ {result['message']}\n{result.get('run_url', '')}"}
+    return {"text": f"❌ {result['message']}"}
+
+
+def _help_response() -> dict:
+    """返回帮助信息"""
+    help_text = """**CI ChatOps 命令帮助**
+
+`/ci status [branch]`  查询分支 CI 状态（默认 develop）
+`/ci rerun <run-id>`   重跑指定 run 的失败 jobs
+`/ci rerun latest [branch]`  重跑分支最近一次失败的 run
+`/ci help`             显示此帮助
+
+**环境变量配置:**
+  `GITEA_TOKEN` / `GITEA_USERNAME + GITEA_PASSWORD`
+  `FEISHU_WEBHOOK_URL`
+  `CHATOPS_NOTIFY_BRANCHES=main,develop`
+"""
+    return {"text": help_text}
+
+
+# ── FastAPI 应用 ──────────────────────────────────────
+
+
+def create_app():
+    """创建 FastAPI 应用
+
+    如果 FastAPI 未安装，返回 None
+    """
+    if not FASTAPI_AVAILABLE:
+        print(
+            "[WARN] FastAPI 未安装，无法启动 webhook 服务。" " 请运行: pip install fastapi uvicorn",
+            file=sys.stderr,
+        )
+        return None
+
+    app = FastAPI(title="CI ChatOps Webhook", version="0.1.0")
+
+    @app.post("/webhook/gitea")
+    async def gitea_webhook(
+        request: Request,
+        x_gitea_event: str = Header(default=""),
+        x_gitea_signature: str = Header(default=""),
+    ):
+        body = await request.body()
+
+        # 签名校验
+        if not verify_gitea_signature(body, x_gitea_signature):
+            raise HTTPException(status_code=401, detail="Invalid signature")
+
+        try:
+            payload = json.loads(body.decode())
+        except json.JSONDecodeError as e:
+            raise HTTPException(status_code=400, detail="Invalid JSON") from e
+
+        result = handle_gitea_webhook(payload, x_gitea_event)
+        return JSONResponse(content=result)
+
+    @app.post("/webhook/feishu")
+    async def feishu_webhook(request: Request):
+        body = await request.body()
+        try:
+            payload = json.loads(body.decode())
+        except json.JSONDecodeError as e:
+            raise HTTPException(status_code=400, detail="Invalid JSON") from e
+
+        result = handle_feishu_message(payload)
+        return JSONResponse(content=result)
+
+    @app.get("/health")
+    async def health():
+        return {"status": "ok", "service": "ci-chatops"}
+
+    return app
+
+
+# ── CLI 入口 ──────────────────────────────────────────
+
+
+def main():
+    """启动 webhook 服务"""
+    import argparse
+
+    parser = argparse.ArgumentParser(description="CI ChatOps Webhook 服务")
+    parser.add_argument("--port", type=int, default=config.WEBHOOK_PORT, help="监听端口")
+    parser.add_argument("--host", default="0.0.0.0", help="监听地址")
+    args = parser.parse_args()
+
+    app = create_app()
+    if not app:
+        print("[ERROR] FastAPI 不可用，请先安装: pip install fastapi uvicorn")
+        return 1
+
+    try:
+        import uvicorn
+    except ImportError:
+        print("[ERROR] uvicorn 未安装，请先安装: pip install uvicorn")
+        return 1
+
+    print(f"[INFO] CI ChatOps Webhook 服务启动: http://{args.host}:{args.port}")
+    print("[INFO] Gitea webhook:   POST /webhook/gitea")
+    print("[INFO] 飞书 webhook:    POST /webhook/feishu")
+    print("[INFO] 健康检查:        GET  /health")
+    print(f"[INFO] 通知分支:        {', '.join(config.NOTIFY_BRANCHES)}")
+
+    uvicorn.run(app, host=args.host, port=args.port)
+    return 0
+
+
+if __name__ == "__main__":
+    sys.exit(main())