From 55347f962c638ca3553d0f538b0ecfd6eb82f47d Mon Sep 17 00:00:00 2001 From: CI Bot Date: Sat, 18 Jul 2026 18:38:11 +0800 Subject: [PATCH] =?UTF-8?q?feat(ci):=20P2-2=20ChatOps=E9=9B=86=E6=88=90=20?= =?UTF-8?q?-=20=E9=A3=9E=E4=B9=A6=E6=9C=BA=E5=99=A8=E4=BA=BA=E5=AF=B9?= =?UTF-8?q?=E6=8E=A5Gitea=20Actions?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 新增 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 - 重跑失败jobs - /ci rerun latest [branch] - 重跑最近一次失败的run - /ci help - 帮助 black+ruff全绿，与现有ci脚本风格一致。 --- scripts/ci/chatops/__init__.py | 19 ++ scripts/ci/chatops/ci_query.py | 296 ++++++++++++++++++ scripts/ci/chatops/ci_trigger.py | 169 ++++++++++ scripts/ci/chatops/config.py | 55 ++++ scripts/ci/chatops/feishu_notify.py | 390 +++++++++++++++++++++++ scripts/ci/chatops/gitea_client.py | 243 +++++++++++++++ scripts/ci/chatops/webhook_server.py | 446 +++++++++++++++++++++++++++ 7 files changed, 1618 insertions(+) create mode 100755 scripts/ci/chatops/__init__.py create mode 100755 scripts/ci/chatops/ci_query.py create mode 100755 scripts/ci/chatops/ci_trigger.py create mode 100755 scripts/ci/chatops/config.py create mode 100755 scripts/ci/chatops/feishu_notify.py create mode 100755 scripts/ci/chatops/gitea_client.py create mode 100755 scripts/ci/chatops/webhook_server.py diff --git a/scripts/ci/chatops/__init__.py b/scripts/ci/chatops/__init__.py new file mode 100755 index 000000000..ab54c0fa0 --- /dev/null +++ b/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", +] diff --git a/scripts/ci/chatops/ci_query.py b/scripts/ci/chatops/ci_query.py new file mode 100755 index 000000000..d4a9c0a34 --- /dev/null +++ b/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()) diff --git a/scripts/ci/chatops/ci_trigger.py b/scripts/ci/chatops/ci_trigger.py new file mode 100755 index 000000000..4ad7f1271 --- /dev/null +++ b/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()) diff --git a/scripts/ci/chatops/config.py b/scripts/ci/chatops/config.py new file mode 100755 index 000000000..008bb7f6d --- /dev/null +++ b/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) diff --git a/scripts/ci/chatops/feishu_notify.py b/scripts/ci/chatops/feishu_notify.py new file mode 100755 index 000000000..fefc48dca --- /dev/null +++ b/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()) diff --git a/scripts/ci/chatops/gitea_client.py b/scripts/ci/chatops/gitea_client.py new file mode 100755 index 000000000..49f10ac68 --- /dev/null +++ b/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 diff --git a/scripts/ci/chatops/webhook_server.py b/scripts/ci/chatops/webhook_server.py new file mode 100755 index 000000000..0cbcc5d14 --- /dev/null +++ b/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://: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.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()) -- 2.54.0