From 2cb273b4720cecf98f1591dcc52548312a9aaaee Mon Sep 17 00:00:00 2001 From: xiaoxia Date: Wed, 22 Jul 2026 11:04:31 +0800 Subject: [PATCH] style: black format ci_repeated_failure_detector.py --- scripts/ci/ci_repeated_failure_detector.py | 194 ++++++++++++--------- 1 file changed, 113 insertions(+), 81 deletions(-) diff --git a/scripts/ci/ci_repeated_failure_detector.py b/scripts/ci/ci_repeated_failure_detector.py index e12d78536..3d6e02631 100644 --- a/scripts/ci/ci_repeated_failure_detector.py +++ b/scripts/ci/ci_repeated_failure_detector.py @@ -31,7 +31,9 @@ REPO = get_env("GITEA_REPO", "xiaoxia/xiaoxia-saas") DAYS = int(get_env("FAIL_CHECK_DAYS", "7")) FAIL_THRESHOLD = int(get_env("FAIL_THRESHOLD", 3)) # 失败次数阈值 FAIL_RATE_THRESHOLD = float(get_env("FAIL_RATE_THRESHOLD", "30")) # 失败率阈值% -CONSECUTIVE_FAIL_THRESHOLD = int(get_env("CONSECUTIVE_FAIL_THRESHOLD", "3")) # 连续失败阈值 +CONSECUTIVE_FAIL_THRESHOLD = int( + get_env("CONSECUTIVE_FAIL_THRESHOLD", "3") +) # 连续失败阈值 WEBHOOK = get_env("CI_NOTIFY_WEBHOOK", "") @@ -56,17 +58,17 @@ def fetch_recent_runs(days=7, per_page=50, max_pages=10): """获取最近N天的runs""" since = (datetime.now(timezone.utc) - timedelta(days=days)).isoformat() all_runs = [] - + for page in range(1, max_pages + 1): path = f"/repos/{REPO}/actions/runs?page={page}&limit={per_page}" data = api_get(path) if not data: break - + runs = data.get("workflow_runs", data.get("runs", [])) if not runs: break - + # 检查时间范围（Gitea用started_at，格式2026-07-22T10:58:10+08:00） oldest = None for r in runs: @@ -75,13 +77,13 @@ def fetch_recent_runs(days=7, per_page=50, max_pages=10): all_runs.append(r) else: oldest = started - + if oldest and oldest < since: break - + if len(runs) < per_page: break - + return all_runs @@ -97,58 +99,69 @@ def fetch_run_jobs(run_id): def analyze_failures(runs): """ 分析失败情况 - + 返回: - job_stats: {job_name: {total, success, failure, skipped, failure_rate, failures: [...]}} - consecutive_failures: {job_name: current_streak, max_streak, last_status} """ - job_stats = defaultdict(lambda: { - "total": 0, "success": 0, "failure": 0, "error": 0, - "skipped": 0, "cancelled": 0, "failures": [] - }) - + job_stats = defaultdict( + lambda: { + "total": 0, + "success": 0, + "failure": 0, + "error": 0, + "skipped": 0, + "cancelled": 0, + "failures": [], + } + ) + # 按时间正序排列（旧→新）用于连续失败计算 - sorted_runs = sorted(runs, key=lambda r: r.get("started_at", r.get("created_at", ""))) - + sorted_runs = sorted( + runs, key=lambda r: r.get("started_at", r.get("created_at", "")) + ) + # 连续失败跟踪 {job_name: streak} consecutive = defaultdict(lambda: {"current": 0, "max": 0, "last_run": None}) - + for run in sorted_runs: run_id = run.get("id") run_status = run.get("status", "") run_conclusion = run.get("conclusion", "") run_started = run.get("started_at", run.get("created_at", "")) event = run.get("event", "") - + # 只统计pull_request和push事件的CI if event not in ("pull_request", "push"): continue - + jobs = fetch_run_jobs(run_id) - + for job in jobs: name = job.get("name", "") status = job.get("status", "") conclusion = job.get("conclusion", "") - + # 跳过非CI核心job（如AI Code Review、Preview等） skip_prefixes = ("AI Code Review", "Preview", "PR Automation", "Auto") if any(name.startswith(p) for p in skip_prefixes): continue - + stats = job_stats[name] stats["total"] += 1 - + if conclusion == "success": stats["success"] += 1 consecutive[name]["current"] = 0 elif conclusion == "failure": stats["failure"] += 1 - stats["failures"].append({ - "run_id": run_id, - "time": run_started, - "event": event, - }) + stats["failures"].append( + { + "run_id": run_id, + "time": run_started, + "event": event, + } + ) consecutive[name]["current"] += 1 if consecutive[name]["current"] > consecutive[name]["max"]: consecutive[name]["max"] = consecutive[name]["current"] @@ -165,7 +178,7 @@ def analyze_failures(runs): elif conclusion == "cancelled": stats["cancelled"] += 1 # cancelled不算失败也不打断 - + # 计算失败率 for name, stats in job_stats.items(): total_actual = stats["total"] - stats["skipped"] - stats["cancelled"] @@ -175,14 +188,14 @@ def analyze_failures(runs): ) else: stats["failure_rate"] = 0.0 - + return dict(job_stats), dict(consecutive) def find_high_failures(job_stats, consecutive): """ 找出高风险job - + 告警级别： - critical: 连续失败 >= CONSECUTIVE_FAIL_THRESHOLD，或 失败率>=50%且失败次数>=5 - warning: 失败率>=FAIL_RATE_THRESHOLD且失败次数>=FAIL_THRESHOLD @@ -191,13 +204,13 @@ def find_high_failures(job_stats, consecutive): critical = [] warning = [] info = [] - + for name, stats in job_stats.items(): fail_count = stats["failure"] + stats["error"] rate = stats["failure_rate"] streak = consecutive.get(name, {}).get("current", 0) max_streak = consecutive.get(name, {}).get("max", 0) - + issue = { "name": name, "fail_count": fail_count, @@ -207,19 +220,19 @@ def find_high_failures(job_stats, consecutive): "max_streak": max_streak, "recent_failures": stats["failures"][-5:], # 最近5次 } - + if streak >= CONSECUTIVE_FAIL_THRESHOLD or (rate >= 50 and fail_count >= 5): critical.append(issue) elif rate >= FAIL_RATE_THRESHOLD and fail_count >= FAIL_THRESHOLD: warning.append(issue) elif fail_count >= 2: info.append(issue) - + # 按失败次数倒序 critical.sort(key=lambda x: x["fail_count"], reverse=True) warning.sort(key=lambda x: x["fail_count"], reverse=True) info.sort(key=lambda x: x["fail_count"], reverse=True) - + return critical, warning, info @@ -230,51 +243,67 @@ def generate_report(critical, warning, info, days, total_runs): lines.append("") lines.append(f"**统计周期**: 最近{days}天") lines.append(f"**扫描Runs**: {total_runs}个") - lines.append(f"**生成时间**: {datetime.now(timezone.utc).strftime('%Y-%m-%d %H:%M UTC')}") + lines.append( + f"**生成时间**: {datetime.now(timezone.utc).strftime('%Y-%m-%d %H:%M UTC')}" + ) lines.append("") - + lines.append(f"## 概览") lines.append("") lines.append(f"| 级别 | 数量 |") lines.append(f"|------|------|") - lines.append(f"| 🔴 严重 (连续失败≥{CONSECUTIVE_FAIL_THRESHOLD}次 或 失败率≥50%) | {len(critical)} |") - lines.append(f"| 🟡 警告 (失败率≥{FAIL_RATE_THRESHOLD}% 且 失败≥{FAIL_THRESHOLD}次) | {len(warning)} |") + lines.append( + f"| 🔴 严重 (连续失败≥{CONSECUTIVE_FAIL_THRESHOLD}次 或 失败率≥50%) | {len(critical)} |" + ) + lines.append( + f"| 🟡 警告 (失败率≥{FAIL_RATE_THRESHOLD}% 且 失败≥{FAIL_THRESHOLD}次) | {len(warning)} |" + ) lines.append(f"| 🔵 关注 (失败≥2次) | {len(info)} |") lines.append("") - + if critical: lines.append("## 🔴 严重问题") lines.append("") for item in critical: lines.append(f"### {item['name']}") lines.append("") - lines.append(f"- 失败次数: **{item['fail_count']}** / {item['total']} 次运行") + lines.append( + f"- 失败次数: **{item['fail_count']}** / {item['total']} 次运行" + ) lines.append(f"- 失败率: **{item['failure_rate']}%**") - lines.append(f"- 当前连续失败: **{item['current_streak']}** 次 (历史最高: {item['max_streak']} 次)") + lines.append( + f"- 当前连续失败: **{item['current_streak']}** 次 (历史最高: {item['max_streak']} 次)" + ) lines.append("") - if item['recent_failures']: + if item["recent_failures"]: lines.append("最近失败:") lines.append("") - for f in item['recent_failures']: - lines.append(f"- [{f['time'][:16]}] run #{f['run_id']} ({f['event']})") + for f in item["recent_failures"]: + lines.append( + f"- [{f['time'][:16]}] run #{f['run_id']} ({f['event']})" + ) lines.append("") - + if warning: lines.append("## 🟡 警告") lines.append("") for item in warning: - lines.append(f"- **{item['name']}**: {item['fail_count']}次失败 / {item['total']}次运行 ({item['failure_rate']}%)") + lines.append( + f"- **{item['name']}**: {item['fail_count']}次失败 / {item['total']}次运行 ({item['failure_rate']}%)" + ) lines.append("") - + if info: lines.append("## 🔵 关注列表") lines.append("") lines.append("| Job名称 | 失败次数 | 总次数 | 失败率 | 当前连续 |") lines.append("|---------|----------|--------|--------|----------|") for item in info[:20]: # 最多显示20个 - lines.append(f"| {item['name']} | {item['fail_count']} | {item['total']} | {item['failure_rate']}% | {item['current_streak']} |") + lines.append( + f"| {item['name']} | {item['fail_count']} | {item['total']} | {item['failure_rate']}% | {item['current_streak']} |" + ) lines.append("") - + return "\n".join(lines) @@ -283,17 +312,17 @@ def send_feishu_notification(critical, warning, info, days): if not WEBHOOK: print(" ⚠️ 未配置WEBHOOK，跳过飞书通知") return False - + total_issues = len(critical) + len(warning) + len(info) if total_issues == 0: print(" ✅ 无异常，不发送通知") return True - + level = "🔴 严重告警" if critical else "🟡 警告" if warning else "🔵 关注" - + title = f"CI重复失败检测 - {level}" text = f"统计周期: 最近{days}天\n\n" - + if critical: text += "【严重问题】\n" for item in critical[:5]: @@ -302,7 +331,7 @@ def send_feishu_notification(critical, warning, info, days): if len(critical) > 5: text += f" ...还有{len(critical)-5}个\n" text += "\n" - + if warning: text += "【警告】\n" for item in warning[:5]: @@ -310,25 +339,22 @@ def send_feishu_notification(critical, warning, info, days): if len(warning) > 5: text += f" ...还有{len(warning)-5}个\n" text += "\n" - + if info and not critical and not warning: text += "【关注列表】\n" for item in info[:10]: text += f"• {item['name']}: {item['fail_count']}次失败\n" text += "\n" - + text += f"共发现 {total_issues} 个异常job" - - payload = { - "msg_type": "text", - "content": { - "text": f"{title}\n\n{text}" - } - } - + + payload = {"msg_type": "text", "content": {"text": f"{title}\n\n{text}"}} + data = json.dumps(payload).encode() - req = urllib.request.Request(WEBHOOK, data=data, headers={"Content-Type": "application/json"}) - + 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()) @@ -348,56 +374,62 @@ def main(): print(f"统计周期: 最近{DAYS}天") print(f"仓库: {REPO}") print() - + print("1. 获取最近的Runs...") runs = fetch_recent_runs(days=DAYS) print(f" 找到 {len(runs)} 个runs") - + if not runs: print("⚠️ 没有找到runs，退出") return - + print() print("2. 分析job失败情况（可能需要点时间）...") job_stats, consecutive = analyze_failures(runs) print(f" 共统计 {len(job_stats)} 个job") - + print() print("3. 识别高风险job...") critical, warning, info = find_high_failures(job_stats, consecutive) print(f" 🔴 严重: {len(critical)}") print(f" 🟡 警告: {len(warning)}") print(f" 🔵 关注: {len(info)}") - + print() print("4. 生成报告...") report = generate_report(critical, warning, info, DAYS, len(runs)) - + # 保存报告 - report_path = os.environ.get("REPORT_PATH", f"/tmp/ci_failure_report_{int(time.time())}.md") + report_path = os.environ.get( + "REPORT_PATH", f"/tmp/ci_failure_report_{int(time.time())}.md" + ) with open(report_path, "w") as f: f.write(report) print(f" 报告已保存: {report_path}") - + # 打印摘要 print() print("=== 摘要 ===") if critical: print("🔴 严重问题:") for item in critical[:5]: - print(f" {item['name']}: {item['fail_count']}次失败, {item['failure_rate']}%, 连续{item['current_streak']}次") + print( + f" {item['name']}: {item['fail_count']}次失败, {item['failure_rate']}%, 连续{item['current_streak']}次" + ) if warning: print("🟡 警告:") for item in warning[:5]: - print(f" {item['name']}: {item['fail_count']}次失败, {item['failure_rate']}%") - + print( + f" {item['name']}: {item['fail_count']}次失败, {item['failure_rate']}%" + ) + print() print("5. 发送飞书通知...") send_feishu_notification(critical, warning, info, DAYS) - + print() print("✅ 检测完成") - + # 有严重问题时退出码非零，方便workflow标记 if critical: sys.exit(2) @@ -406,4 +438,4 @@ def main(): if __name__ == "__main__": - main() \ No newline at end of file + main()