#!/usr/bin/env python3 """ CI健康度每日巡检报告脚本 - 调用ci_health_check.py获取数据 - 有失败时生成飞书卡片通知并发送 - 无失败时静默退出（不打扰） - 用于每日定时巡检 用法: python3 scripts/ci/ci_health_report.py [--limit 30] [--dry-run] 环境变量: GITEA_TOKEN API token（必需） CI_NOTIFY_WEBHOOK 飞书webhook地址（必需，用于发报告） GITEA_API_URL Gitea API 地址 GITEA_REPO 仓库 """ import argparse import json import os import subprocess import sys import urllib.request from datetime import datetime, timedelta, timezone def run_health_check(limit: int) -> dict: """调用ci_health_check.py获取JSON结果""" script_dir = os.path.dirname(os.path.abspath(__file__)) cmd = [ sys.executable, os.path.join(script_dir, "ci_health_check.py"), "--json", "--limit", str(limit), ] env = os.environ.copy() # 确保GITEA_TOKEN传递 if not env.get("GITEA_TOKEN") and env.get("GITHUB_TOKEN"): env["GITEA_TOKEN"] = env["GITHUB_TOKEN"] result = subprocess.run(cmd, capture_output=True, text=True, env=env) if result.returncode != 0: print(f"health check failed: {result.stderr}") return {"workflows": {}, "failed_runs": []} try: return json.loads(result.stdout) except json.JSONDecodeError: print(f"failed to parse health check output: {result.stdout[:200]}") return {"workflows": {}, "failed_runs": []} def build_feishu_card(data: dict) -> dict: """构建飞书卡片消息""" wf_stats = data.get("workflows", {}) failed_runs = data.get("failed_runs", []) # 统计数据 total_all = sum(s["total"] for s in wf_stats.values()) succ_all = sum(s["success"] for s in wf_stats.values()) fail_all = sum(s["failure"] for s in wf_stats.values()) rate_all = (succ_all / total_all * 100) if total_all > 0 else 0 # 失败分类 infra_fail = 0 biz_fail = 0 unknown_fail = 0 for run in failed_runs: for job in run.get("jobs", []): cat = job.get("category", "unknown") if cat == "infra": infra_fail += 1 elif cat == "business": biz_fail += 1 else: unknown_fail += 1 now = datetime.now(timezone(timedelta(hours=8))).strftime("%Y-%m-%d %H:%M") # 各workflow成功率行 wf_lines = [] for wf, s in sorted(wf_stats.items()): total = s["total"] succ = s["success"] fail = s["failure"] rate = (succ / total * 100) if total > 0 else 0 icon = "🟢" if rate >= 90 else ("🟡" if rate >= 70 else "🔴") wf_name = ( wf.replace("ci-pipeline.yml", "CI Pipeline") .replace("code-review.yml", "Code Review") .replace("daily-check.yml", "Daily Check") .replace("preview-deploy.yml", "Preview Deploy") ) wf_lines.append(f"{icon} **{wf_name}**: {rate:.0f}% ({succ}/{total}，失败{fail})") # 失败详情（最多显示5条） fail_detail_lines = [] for _i, run in enumerate(failed_runs[:5]): run_id = run["id"] title = run.get("title", "")[:35] branch = run.get("branch", "") jobs_str = ", ".join(j["name"][:15] for j in run.get("jobs", [])[:3]) fail_detail_lines.append(f"• **#{run_id}** {title}\n 分支: {branch} | 失败: {jobs_str}") if len(failed_runs) > 5: fail_detail_lines.append(f"... 还有 {len(failed_runs) - 5} 条失败记录") # 整体状态 if fail_all == 0: status_text = "✅ 全部通过" status_color = "green" elif infra_fail > biz_fail: status_text = "⚠️ 基础设施问题为主" status_color = "yellow" else: status_text = "🔴 存在业务失败" status_color = "red" card = { "config": {"wide_screen_mode": True}, "header": { "title": {"tag": "plain_text", "content": f"CI告警 - 每日健康度巡检 ({now})"}, "template": status_color, }, "elements": [ { "tag": "div", "text": { "tag": "lark_md", "content": f"**统计范围**: 最近 {total_all} 条 run\n**整体状态**: {status_text}\n**总成功率**: {rate_all:.1f}% ({succ_all}/{total_all})", }, }, {"tag": "hr"}, { "tag": "div", "text": { "tag": "lark_md", "content": "**📊 各Workflow成功率**\n" + "\n".join(wf_lines) if wf_lines else "暂无数据", }, }, ], } # 失败分类统计 if fail_all > 0: card["elements"].append({"tag": "hr"}) card["elements"].append( { "tag": "div", "text": { "tag": "lark_md", "content": f"**失败原因分类**\n🏗️ 基础设施: {infra_fail} 个\n🐛 业务代码: {biz_fail} 个\n❓ 待确认: {unknown_fail} 个", }, } ) # 失败详情 if fail_detail_lines: card["elements"].append({"tag": "hr"}) card["elements"].append( { "tag": "div", "text": { "tag": "lark_md", "content": "**❌ 失败详情**\n" + "\n\n".join(fail_detail_lines), }, } ) # 查看更多 card["elements"].append({"tag": "hr"}) base_url = os.environ.get("GITEA_BASE_URL", "https://git.xiaoxiajianji.com") repo = os.environ.get("GITEA_REPO", "xiaoxia/xiaoxia-saas") card["elements"].append( { "tag": "action", "actions": [ { "tag": "button", "text": {"tag": "plain_text", "content": "查看CI面板"}, "type": "primary", "url": f"{base_url}/{repo}/actions", } ], } ) return {"msg_type": "interactive", "card": card} def send_feishu(webhook: str, payload: dict) -> bool: """发送飞书webhook""" data = json.dumps(payload).encode() req = urllib.request.Request( webhook, data=data, headers={"Content-Type": "application/json"}, ) try: with urllib.request.urlopen(req, timeout=10) as resp: result = json.loads(resp.read().decode()) return result.get("code", -1) == 0 or result.get("StatusCode", -1) == 0 except Exception as e: print(f"send feishu failed: {e}") return False def main(): parser = argparse.ArgumentParser(description="CI健康度每日巡检报告") parser.add_argument("--limit", type=int, default=30, help="统计最近N条run") parser.add_argument("--dry-run", action="store_true", help="只打印不发送") parser.add_argument("--always-notify", action="store_true", help="即使全部通过也发送通知") args = parser.parse_args() webhook = os.environ.get("CI_NOTIFY_WEBHOOK", "") if not webhook and not args.dry_run: print("未配置 CI_NOTIFY_WEBHOOK，跳过通知") # 还是执行健康检查输出到日志，方便排查 data = run_health_check(args.limit) print(f"health check done: {len(data.get('failed_runs', []))} failed") return 0 # 执行健康检查 data = run_health_check(args.limit) failed_count = len(data.get("failed_runs", [])) # 无失败且不强制通知 → 静默退出 if failed_count == 0 and not args.always_notify: print("✅ 全部通过，静默退出") return 0 # 构建并发送卡片 card = build_feishu_card(data) if args.dry_run: print(json.dumps(card, ensure_ascii=False, indent=2)) return 0 success = send_feishu(webhook, card) if success: print(f"📤 已发送健康度报告，失败 {failed_count} 条") else: print("❌ 发送飞书通知失败") # 通知失败不阻断流程 return 0 if __name__ == "__main__": sys.exit(main())