#!/usr/bin/env python3 """ CI 健康度快速检查脚本 - 统计最近 N 条 run 的成功率（按 workflow 分类） - 列出失败的 run 和失败的 job/step - 区分基础设施问题 vs 业务代码问题 - 输出简洁的健康度报告 用法: python3 scripts/ci/ci_health_check.py [--limit 20] [--workflow ci-pipeline.yml] [--json] 环境变量: GITEA_TOKEN API token（必需） GITEA_API_URL Gitea API 地址，默认 https://git.xiaoxiajianji.com/api/v1 GITEA_REPO 仓库，默认 xiaoxia/xiaoxia-saas """ import argparse import json import os import sys import urllib.request from datetime import datetime, timedelta, timezone # ---- 基础设施问题关键词（命中即判定为基础设施问题）---- INFRA_KEYWORDS = [ # 网络/连接 "Couldn't connect to server", "Connection refused", "Connection reset", "Connection timed out", "Failed to connect to", "network is unreachable", "TLS handshake timeout", "SSL certificate problem", # 容器/Runner "No such container", "container already exists", "docker: not found", "no space left on device", "out of memory", "OOMKilled", "pull access denied", "manifest unknown", "Error response from daemon", "runner", "runner is not online", "no matching runners", # Checkout/Git "Could not resolve host", "fatal: unable to access", "The remote end hung up unexpectedly", "early EOF", "index-pack failed", "git fetch", "checkout failed", "ETXTBSY", "text file busy", # 镜像/环境 "No module named pip", "pip: not found", "command not found: python", "python3: not found", "node: not found", "npm: not found", "exec format error", "standard_init_linux.go", # 系统/资源 "Input/output error", "device or resource busy", "No space left on device", "Disk full", # 鉴权/配置 "401 Unauthorized", "403 Forbidden", "404 Not Found", "identity_sign: private key", "Permission denied", ] def api_get(path: str) -> dict: base = os.environ.get("GITEA_API_URL", "https://git.xiaoxiajianji.com/api/v1") repo = os.environ.get("GITEA_REPO", "xiaoxia/xiaoxia-saas") token = os.environ.get("GITEA_TOKEN", "") url = f"{base}/repos/{repo}/{path}" req = urllib.request.Request(url, headers={"Authorization": f"token {token}"}) with urllib.request.urlopen(req, timeout=30) as resp: return json.loads(resp.read().decode()) def get_run_jobs(run_id: int) -> list: return api_get(f"actions/runs/{run_id}/jobs").get("jobs", []) def get_job_log(job_id: int) -> str: try: return api_get(f"actions/jobs/{job_id}/logs") except Exception: return "" def classify_failure(job: dict) -> str: """判断失败原因类型: infra / business / unknown""" name = job.get("name", "") # 仅根据 job 名称做初步分类（更精确需读日志，但代价高） infra_jobs = ["Checkout", "Build", "Deploy", "Cleanup"] business_jobs = [ "Unit Tests", "Integration Tests", "Frontend Lint", "Frontend Unit Tests", "Staging E2E", "E2E", "Validate Code Quality", ] name_lower = name.lower() if ( any(k.lower() in name_lower for k in infra_jobs) and "Test" not in name and "Lint" not in name and "Validate" not in name ): return "infra" if any(k.lower() in name_lower for k in business_jobs): return "business" return "unknown" def analyze_with_log(job_id: int) -> str: """通过日志关键词精确分类""" log = get_job_log(job_id) log_lower = log.lower() for kw in INFRA_KEYWORDS: if kw.lower() in log_lower: return "infra" return "business" def fmt_time(t: str) -> str: if not t or t.startswith("1970"): return "-" try: dt = datetime.fromisoformat(t.replace("Z", "+00:00")) bj = dt.astimezone(timezone(timedelta(hours=8))) return bj.strftime("%m-%d %H:%M") except Exception: return t[:16] def main(): parser = argparse.ArgumentParser(description="CI 健康度快速检查") parser.add_argument("--limit", type=int, default=20, help="最近多少条 run") parser.add_argument("--workflow", type=str, default="", help="只看某个 workflow") parser.add_argument("--json", action="store_true", help="JSON 输出") parser.add_argument("--deep", action="store_true", help="深度检查（读日志，较慢）") args = parser.parse_args() if not os.environ.get("GITEA_TOKEN"): print("错误: 请设置 GITEA_TOKEN 环境变量", file=sys.stderr) sys.exit(1) # 1. 获取最近 run runs = api_get(f"actions/runs?limit={args.limit}").get("workflow_runs", []) if args.workflow: runs = [r for r in runs if args.workflow in r.get("path", "")] if not runs: print("没有找到匹配的 run") return # 按 workflow 分组统计 wf_stats = {} failed_runs = [] for r in runs: path = r.get("path", "unknown") # 提取 workflow 文件名，兼容各种 path 格式 if ".yml" in path or ".yaml" in path: # ci-pipeline.yml@refs/heads/develop -> ci-pipeline.yml wf = path.split("@")[0].split("/")[-1] else: wf = path.split("/")[-1] if "/" in path else path if wf not in wf_stats: wf_stats[wf] = {"total": 0, "success": 0, "failure": 0, "cancelled": 0, "others": 0} wf_stats[wf]["total"] += 1 status = r.get("status", "") conc = r.get("conclusion", "") if status != "completed": wf_stats[wf]["others"] += 1 continue if conc == "success": wf_stats[wf]["success"] += 1 elif conc == "failure": wf_stats[wf]["failure"] += 1 failed_runs.append(r) elif conc == "cancelled": wf_stats[wf]["cancelled"] += 1 else: wf_stats[wf]["others"] += 1 # 2. 失败 run 详情 failed_details = [] for r in failed_runs[:10]: # 最多看10个失败的 jobs = get_run_jobs(r["id"]) failed_jobs = [j for j in jobs if j.get("conclusion") == "failure"] job_infos = [] for j in failed_jobs: cat = classify_failure(j) if args.deep and cat == "unknown": cat = analyze_with_log(j["id"]) # 找失败的 step failed_steps = [] for step in j.get("steps", []): if step.get("conclusion") == "failure": failed_steps.append(step.get("name", "?")) job_infos.append( { "name": j.get("name", ""), "category": cat, "failed_steps": failed_steps, "runner": j.get("runner_name", ""), } ) failed_details.append( { "id": r["id"], "title": r.get("display_title", ""), "branch": r.get("head_branch", ""), "time": fmt_time(r.get("updated_at", "")), "jobs": job_infos, } ) # 3. 输出 if args.json: result = {"workflows": wf_stats, "failed_runs": failed_details} print(json.dumps(result, ensure_ascii=False, indent=2)) return # 文本报告 print("=" * 60) print(" CI 健康度报告") print("=" * 60) print(f"统计范围: 最近 {len(runs)} 条 run") print(f"时间: {datetime.now(timezone(timedelta(hours=8))).strftime('%Y-%m-%d %H:%M:%S')}") print() print("📊 各 Workflow 成功率:") print("-" * 60) for wf, s in sorted(wf_stats.items()): total = s["total"] succ = s["success"] rate = (succ / total * 100) if total > 0 else 0 bar = "█" * int(rate / 5) + "░" * (20 - int(rate / 5)) icon = "🟢" if rate >= 90 else ("🟡" if rate >= 70 else "🔴") print(f" {icon} {wf:35s} {rate:5.1f}% {bar} ({succ}/{total})") if s["failure"]: print(f" 失败: {s['failure']} 取消: {s['cancelled']} 进行中: {s['others']}") if failed_details: print() print("❌ 失败详情:") print("-" * 60) for d in failed_details: print(f" #{d['id']} [{d['time']}] {d['title'][:45]}") print(f" 分支: {d['branch']}") for j in d["jobs"]: cat_icon = "🏗️" if j["category"] == "infra" else ("🐛" if j["category"] == "business" else "❓") steps = ", ".join(j["failed_steps"][:3]) if j["failed_steps"] else "未知" print(f" {cat_icon} {j['name'][:30]:30s} 失败步骤: {steps}") if j["runner"]: print(f" runner: {j['runner']}") else: print() print("✅ 最近没有失败的 run") # 总结 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()) infra_fail = sum(1 for d in failed_details for j in d["jobs"] if j["category"] == "infra") biz_fail = sum(1 for d in failed_details for j in d["jobs"] if j["category"] == "business") rate_all = (succ_all / total_all * 100) if total_all > 0 else 0 print() print("=" * 60) print(f" 总结: 总成功率 {rate_all:.1f}% ({succ_all}/{total_all})") if fail_all > 0: print(f" 失败job分类: 基础设施 {infra_fail} 个 | 业务代码 {biz_fail} 个") if infra_fail > biz_fail: print(" ⚠️ 主要是基础设施问题，建议优先排查 CI 环境") else: print(" 💡 主要是业务代码问题，建议关注业务侧修复") print("=" * 60) if __name__ == "__main__": main()