Compare commits
1 Commits
| Author | SHA1 | Date | |
|---|---|---|---|
| 7fb3de7be7 |
@@ -96,7 +96,7 @@ def build_feishu_card(data: dict) -> dict:
|
||||
|
||||
# 失败详情(最多显示5条)
|
||||
fail_detail_lines = []
|
||||
for i, run in enumerate(failed_runs[:5]):
|
||||
for _i, run in enumerate(failed_runs[:5]):
|
||||
run_id = run["id"]
|
||||
title = run.get("title", "")[:35]
|
||||
branch = run.get("branch", "")
|
||||
|
||||
@@ -122,8 +122,8 @@ def analyze_failures(runs):
|
||||
|
||||
for run in sorted_runs:
|
||||
run_id = run.get("id")
|
||||
run_status = run.get("status", "")
|
||||
run_conclusion = run.get("conclusion", "")
|
||||
run.get("status", "")
|
||||
run.get("conclusion", "")
|
||||
run_started = run.get("started_at", run.get("created_at", ""))
|
||||
event = run.get("event", "")
|
||||
|
||||
@@ -135,7 +135,7 @@ def analyze_failures(runs):
|
||||
|
||||
for job in jobs:
|
||||
name = job.get("name", "")
|
||||
status = job.get("status", "")
|
||||
job.get("status", "")
|
||||
conclusion = job.get("conclusion", "")
|
||||
|
||||
# 跳过非CI核心job(如AI Code Review、Preview等)
|
||||
@@ -176,7 +176,7 @@ def analyze_failures(runs):
|
||||
# cancelled不算失败也不打断
|
||||
|
||||
# 计算失败率
|
||||
for name, stats in job_stats.items():
|
||||
for _name, stats in job_stats.items():
|
||||
total_actual = stats["total"] - stats["skipped"] - stats["cancelled"]
|
||||
if total_actual > 0:
|
||||
stats["failure_rate"] = round((stats["failure"] + stats["error"]) / total_actual * 100, 1)
|
||||
@@ -240,10 +240,10 @@ def generate_report(critical, warning, info, days, total_runs):
|
||||
lines.append(f"**生成时间**: {datetime.now(timezone.utc).strftime('%Y-%m-%d %H:%M UTC')}")
|
||||
lines.append("")
|
||||
|
||||
lines.append(f"## 概览")
|
||||
lines.append("## 概览")
|
||||
lines.append("")
|
||||
lines.append(f"| 级别 | 数量 |")
|
||||
lines.append(f"|------|------|")
|
||||
lines.append("| 级别 | 数量 |")
|
||||
lines.append("|------|------|")
|
||||
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)} |")
|
||||
@@ -350,7 +350,7 @@ def send_feishu_notification(critical, warning, info, days):
|
||||
|
||||
|
||||
def main():
|
||||
print(f"=== CI重复失败检测 ===")
|
||||
print("=== CI重复失败检测 ===")
|
||||
print(f"统计周期: 最近{DAYS}天")
|
||||
print(f"仓库: {REPO}")
|
||||
print()
|
||||
|
||||
@@ -8,9 +8,7 @@ PR自动扫描器:扫描所有open PR,对CI全绿的进行自动审批/合
|
||||
|
||||
import argparse
|
||||
import json
|
||||
import os
|
||||
import re
|
||||
import sys
|
||||
import time
|
||||
import urllib.error
|
||||
import urllib.request
|
||||
|
||||
@@ -44,7 +44,7 @@ def api_get(path):
|
||||
time.sleep(2**attempt)
|
||||
continue
|
||||
raise
|
||||
except Exception as e:
|
||||
except Exception:
|
||||
if attempt < 2:
|
||||
time.sleep(2**attempt)
|
||||
continue
|
||||
@@ -100,7 +100,7 @@ def send_alert(pr_num, pr_title, pr_url, head_sha, commit_age_min):
|
||||
print(" ⚠️ 未配置CI_NOTIFY_WEBHOOK,跳过告警")
|
||||
return
|
||||
|
||||
gitea_url = get_env("GITEA_URL", "https://git.xiaoxiajianji.com")
|
||||
get_env("GITEA_URL", "https://git.xiaoxiajianji.com")
|
||||
|
||||
content = {
|
||||
"msg_type": "interactive",
|
||||
@@ -185,7 +185,7 @@ def main():
|
||||
# 解析updated_at(ISO格式)
|
||||
try:
|
||||
# 2026-07-17T09:22:43+08:00
|
||||
from datetime import datetime, timedelta, timezone
|
||||
from datetime import datetime
|
||||
|
||||
# 简化处理:直接用字符串解析
|
||||
ts_str = updated_at.replace("Z", "+00:00")
|
||||
@@ -202,7 +202,7 @@ def main():
|
||||
|
||||
# 少于2分钟的跳过,给CI一点启动时间
|
||||
if age_min < 2:
|
||||
print(f" ⏳ 刚更新,等待CI启动...")
|
||||
print(" ⏳ 刚更新,等待CI启动...")
|
||||
continue
|
||||
|
||||
# 获取commit状态
|
||||
|
||||
@@ -10,8 +10,6 @@ import sys
|
||||
|
||||
sys.stdout = io.TextIOWrapper(sys.stdout.buffer, encoding="utf-8")
|
||||
|
||||
import json
|
||||
from datetime import datetime, timedelta
|
||||
|
||||
import requests
|
||||
|
||||
|
||||
@@ -10,8 +10,6 @@ import sys
|
||||
|
||||
sys.stdout = io.TextIOWrapper(sys.stdout.buffer, encoding="utf-8")
|
||||
|
||||
import json
|
||||
from datetime import datetime, timedelta
|
||||
|
||||
import requests
|
||||
|
||||
|
||||
@@ -35,7 +35,6 @@ def init_database():
|
||||
""",
|
||||
("认证与账号体系", "Phase 4", "2026-06-10", "2026-06-17", "completed", "JWT 登录、注册、密码管理"),
|
||||
)
|
||||
milestone1_id = cursor.lastrowid
|
||||
|
||||
auth_tasks = [
|
||||
("JWT 工具类实现", "sign/verify/refresh Token 功能", "completed", "high"),
|
||||
|
||||
@@ -44,7 +44,7 @@ def main() -> None:
|
||||
owner_headers = _register_login(owner, "owner")
|
||||
intruder_headers = _register_login(intruder, "intruder")
|
||||
|
||||
workspace = _json_or_raise(
|
||||
_json_or_raise(
|
||||
"owner_workspace",
|
||||
owner.post(f"{BASE_URL}/workspaces", json={"name": "Boundary Workspace"}, headers=owner_headers, timeout=30),
|
||||
)
|
||||
|
||||
@@ -47,7 +47,7 @@ def main() -> None:
|
||||
)
|
||||
headers = {"Authorization": f"Bearer {login['access_token']}"}
|
||||
|
||||
workspace = _json_or_raise(
|
||||
_json_or_raise(
|
||||
"workspace",
|
||||
session.post(f"{BASE_URL}/workspaces", json={"name": "Upload Smoke Workspace"}, headers=headers, timeout=30),
|
||||
)
|
||||
@@ -75,7 +75,7 @@ def main() -> None:
|
||||
timeout=30,
|
||||
),
|
||||
)
|
||||
library_id = library["id"]
|
||||
library["id"]
|
||||
|
||||
upload = _json_or_raise(
|
||||
"upload",
|
||||
|
||||
Reference in New Issue
Block a user