style: black格式化ci_trigger_monitor.py

scripts/ci_trigger_monitor.py

@@ -30,23 +30,23 @@ def api_get(path):
     token = get_env("GITEA_API_TOKEN")
     base_url = get_env("GITEA_URL", "https://git.xiaoxiajianji.com")
     repo = get_env("GITEA_REPO", "xiaoxia/xiaoxia-saas")
-    
+
     url = f"{base_url}/api/v1/repos/{repo}{path}"
     req = urllib.request.Request(url)
     req.add_header("Authorization", f"token {token}")
-    
+
     for attempt in range(3):
         try:
             with urllib.request.urlopen(req, timeout=30) as resp:
                 return json.loads(resp.read())
         except urllib.error.HTTPError as e:
             if e.code >= 500 and attempt < 2:
-                time.sleep(2 ** attempt)
+                time.sleep(2**attempt)
                 continue
             raise
         except Exception as e:
             if attempt < 2:
-                time.sleep(2 ** attempt)
+                time.sleep(2**attempt)
                 continue
             raise
 
@@ -77,18 +77,18 @@ def get_commit_status(sha):
 
 def has_ci_started(statuses):
     """判断是否有CI job已经启动（pending/running/success/failure都算启动了）"""
-    pr_statuses = [s for s in statuses if 'pull_request' in s.get('context', '')]
+    pr_statuses = [s for s in statuses if "pull_request" in s.get("context", "")]
     if not pr_statuses:
         return False
     # 只要有非pending且非空的状态，就算启动了
     for s in pr_statuses:
-        if s.get('status') in ['success', 'failure', 'running']:
+        if s.get("status") in ["success", "failure", "running"]:
             return True
-        if s.get('status') == 'pending' and 'Has started running' in s.get('description', ''):
+        if s.get("status") == "pending" and "Has started running" in s.get("description", ""):
             return True
     # 全是"Blocked by required conditions"的pending也算（说明CI系统收到了事件）
     for s in pr_statuses:
-        if 'Blocked' in s.get('description', ''):
+        if "Blocked" in s.get("description", ""):
             return True
     return False
 
@@ -99,26 +99,23 @@ def send_alert(pr_num, pr_title, pr_url, head_sha, commit_age_min):
     if not webhook:
         print("  ⚠️  未配置CI_NOTIFY_WEBHOOK，跳过告警")
         return
-    
+
     gitea_url = get_env("GITEA_URL", "https://git.xiaoxiajianji.com")
-    
+
     content = {
         "msg_type": "interactive",
         "card": {
             "header": {
-                "title": {
-                    "tag": "plain_text",
-                    "content": f"⚠️ CI告警 - PR#{pr_num} CI未触发"
-                },
-                "template": "red"
+                "title": {"tag": "plain_text", "content": f"⚠️ CI告警 - PR#{pr_num} CI未触发"},
+                "template": "red",
             },
             "elements": [
                 {
                     "tag": "div",
                     "text": {
                         "tag": "lark_md",
-                        "content": f"**PR**: [{pr_title}]({pr_url})\n**最新commit**: `{head_sha[:12]}`\n**已等待**: {commit_age_min:.0f} 分钟仍无CI启动\n**可能原因**: Gitea Actions事件丢失 / Webhook失败 / Runner资源不足"
-                    }
+                        "content": f"**PR**: [{pr_title}]({pr_url})\n**最新commit**: `{head_sha[:12]}`\n**已等待**: {commit_age_min:.0f} 分钟仍无CI启动\n**可能原因**: Gitea Actions事件丢失 / Webhook失败 / Runner资源不足",
+                    },
                 },
                 {
                     "tag": "action",
@@ -127,29 +124,29 @@ def send_alert(pr_num, pr_title, pr_url, head_sha, commit_age_min):
                             "tag": "button",
                             "text": {"tag": "plain_text", "content": "查看PR"},
                             "url": pr_url,
-                            "type": "primary"
+                            "type": "primary",
                         },
                         {
                             "tag": "button",
                             "text": {"tag": "plain_text", "content": "查看Actions"},
                             "url": f"{pr_url}/files",
-                            "type": "default"
-                        }
-                    ]
+                            "type": "default",
+                        },
+                    ],
                 },
                 {
                     "tag": "note",
                     "elements": [
                         {"tag": "plain_text", "content": f"CI触发监控 | 检测时间: {time.strftime('%Y-%m-%d %H:%M:%S')}"}
-                    ]
-                }
-            ]
-        }
+                    ],
+                },
+            ],
+        },
     }
-    
+
     try:
         data = json.dumps(content).encode()
-        req = urllib.request.Request(webhook, data=data, method='POST')
+        req = urllib.request.Request(webhook, data=data, method="POST")
         req.add_header("Content-Type", "application/json")
         with urllib.request.urlopen(req, timeout=10) as resp:
             resp.read()
@@ -160,86 +157,81 @@ def send_alert(pr_num, pr_title, pr_url, head_sha, commit_age_min):
 
 def main():
     stale_threshold = int(get_env("STALE_THRESHOLD_MIN", "5"))
-    
+
     print("=" * 60)
     print(f"CI触发监控 - 检测时间: {time.strftime('%Y-%m-%d %H:%M:%S')}")
     print(f"告警阈值: {stale_threshold}分钟无CI启动")
     print("=" * 60)
-    
+
     # 获取open PR列表
     try:
         prs = get_open_prs()
     except Exception as e:
         print(f"❌ 获取PR列表失败: {e}")
         sys.exit(0)  # 告警脚本不阻断CI
-    
+
     print(f"\n共 {len(prs)} 个open PR\n")
-    
+
     stale_prs = []
     now = time.time()
-    
+
     for pr in prs:
-        pr_num = pr['number']
-        pr_title = pr['title']
-        pr_url = pr['html_url']
-        head_sha = pr['head']['sha']
-        updated_at = pr['updated_at']
-        
+        pr_num = pr["number"]
+        pr_title = pr["title"]
+        pr_url = pr["html_url"]
+        head_sha = pr["head"]["sha"]
+        updated_at = pr["updated_at"]
+
         # 解析updated_at（ISO格式）
         try:
             # 2026-07-17T09:22:43+08:00
             from datetime import datetime, timezone, timedelta
+
             # 简化处理：直接用字符串解析
-            ts_str = updated_at.replace('Z', '+00:00')
+            ts_str = updated_at.replace("Z", "+00:00")
             # 手动解析
             dt = datetime.fromisoformat(ts_str)
             commit_time = dt.timestamp()
         except Exception as e:
             print(f"  ⚠️  PR#{pr_num} 时间解析失败: {e}")
             continue
-        
+
         age_min = (now - commit_time) / 60
-        
+
         print(f"PR#{pr_num:3d} | {pr_title[:45]:45s} | 更新于 {age_min:.0f}min前")
-        
+
         # 少于2分钟的跳过，给CI一点启动时间
         if age_min < 2:
             print(f"  ⏳ 刚更新，等待CI启动...")
             continue
-        
+
         # 获取commit状态
         status = get_commit_status(head_sha)
-        statuses = status.get('statuses', [])
-        
+        statuses = status.get("statuses", [])
+
         if has_ci_started(statuses):
             print(f"  ✅ CI已启动 (state={status.get('state')})")
             continue
-        
+
         # CI未启动，判断是否超过阈值
         if age_min >= stale_threshold:
             print(f"  🚨 CI未触发！已等待 {age_min:.0f} 分钟")
-            stale_prs.append({
-                'num': pr_num,
-                'title': pr_title,
-                'url': pr_url,
-                'sha': head_sha,
-                'age_min': age_min
-            })
+            stale_prs.append({"num": pr_num, "title": pr_title, "url": pr_url, "sha": head_sha, "age_min": age_min})
         else:
             print(f"  ⏳ CI尚未启动 ({age_min:.0f}min < {stale_threshold}min阈值)")
-    
+
     # 发送告警
     print(f"\n{'=' * 60}")
     print(f"检测结果: {len(stale_prs)} 个PR CI未触发超过阈值")
-    
+
     if stale_prs:
         print("\n告警列表:")
         for pr in stale_prs:
             print(f"  - PR#{pr['num']}: {pr['title'][:40]} ({pr['age_min']:.0f}min)")
-            send_alert(pr['num'], pr['title'], pr['url'], pr['sha'], pr['age_min'])
+            send_alert(pr["num"], pr["title"], pr["url"], pr["sha"], pr["age_min"])
     else:
         print("✅ 所有PR CI触发正常")
-    
+
     print("=" * 60)