#!/usr/bin/env python3 """ Gitea Webhook 接收服务 - FastAPI 实现 功能: 1. 接收 Gitea Actions webhook 事件,触发飞书通知 2. 接收飞书机器人回调消息,处理 /ci 交互命令 3. 维护简单的状态缓存,检测分支恢复等状态变化 部署: 部署到构建服务器,监听 8090 端口(可配置) Gitea webhook 指向: http://:8090/webhook/gitea 飞书消息回调指向: http://: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 - 重跑失败的 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 或 /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 的失败 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.CHATOPS_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())