feat(ci): P2-2 ChatOps集成 - 飞书机器人对接Gitea Actions

新增 scripts/ci/chatops/ 目录，Python实现，纯工具链扩展不动业务代码。 3个核心模块： - ci_query.py - 查询CI状态（分支/PR/失败详情） - ci_trigger.py - 触发重跑（rerun失败job/整个workflow/取消） - feishu_notify.py - CI事件飞书通知（失败/恢复/E2E摘要）配套基础设施： - config.py - 配置化，所有token/webhook走环境变量 - gitea_client.py - Gitea API统一封装（urllib实现，无第三方依赖） - webhook_server.py - FastAPI webhook接收服务（可选依赖）通知规则（最高频使用场景）： - main/develop分支CI失败 → 飞书告警卡片（含失败摘要+一键重跑） - 分支从失败恢复 → 绿色恢复通知（含故障时长） - E2E测试失败 → 单独摘要通知 - 连续失败不重复告警（状态缓存去重）飞书交互命令（第一版已实现解析+执行）： - /ci status [branch] - 查询分支CI状态 - /ci rerun <run-id> - 重跑失败jobs - /ci rerun latest [branch] - 重跑最近一次失败的run - /ci help - 帮助 black+ruff全绿，与现有ci脚本风格一致。
2026-07-18 18:38:11 +08:00
parent 2e9478bb18
commit 55347f962c
7 changed files with 1618 additions and 0 deletions
@@ -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())