style: black format ci_repeated_failure_detector.py

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()
+    main()