xiaoxia-saas/scripts/ci/ci_repeated_failure_detector.py

#!/usr/bin/env python3
"""
CI重复失败检测脚本
- 扫描最近N天的CI失败
- 按job名称分组统计失败率
- 识别高失败率job（系统性故障）
- 飞书通知告警
"""

import json
import os
import sys
import time
import urllib.error
import urllib.request
from collections import defaultdict
from datetime import datetime, timedelta, timezone


def get_env(name, default=None, required=False):
    val = os.environ.get(name, default)
    if required and not val:
        print(f"❌ 缺少环境变量: {name}")
        sys.exit(1)
    return val


GITEA_URL = get_env("GITEA_URL", "https://git.xiaoxiajianji.com")
GITEA_TOKEN = get_env("GITEA_API_TOKEN", required=False) or get_env("GITHUB_TOKEN", "")
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"))  # 连续失败阈值
WEBHOOK = get_env("CI_NOTIFY_WEBHOOK", "")


def api_get(path):
    """调用Gitea API"""
    url = f"{GITEA_URL}/api/v1{path}"
    req = urllib.request.Request(url)
    if GITEA_TOKEN:
        req.add_header("Authorization", f"token {GITEA_TOKEN}")
    try:
        with urllib.request.urlopen(req, timeout=30) as resp:
            return json.loads(resp.read())
    except urllib.error.HTTPError as e:
        print(f"  HTTP {e.code}: {path}")
        return None
    except Exception as e:
        print(f"  错误: {e}")
        return None


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:
            started = r.get("started_at", r.get("created_at", ""))
            if started and started >= since:
                all_runs.append(r)
            else:
                oldest = started

        if oldest and oldest < since:
            break

        if len(runs) < per_page:
            break

    return all_runs


def fetch_run_jobs(run_id):
    """获取run的所有jobs"""
    path = f"/repos/{REPO}/actions/runs/{run_id}/jobs"
    data = api_get(path)
    if not data:
        return []
    return data.get("jobs", [])


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": [],
        }
    )

    # 按时间正序排列（旧→新）用于连续失败计算
    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,
                    }
                )
                consecutive[name]["current"] += 1
                if consecutive[name]["current"] > consecutive[name]["max"]:
                    consecutive[name]["max"] = consecutive[name]["current"]
                consecutive[name]["last_run"] = run_id
            elif conclusion == "error":
                stats["error"] += 1
                # error也算失败的一种
                consecutive[name]["current"] += 1
                if consecutive[name]["current"] > consecutive[name]["max"]:
                    consecutive[name]["max"] = consecutive[name]["current"]
            elif conclusion == "skipped":
                stats["skipped"] += 1
                # skipped不算也不打断连续失败
            elif conclusion == "cancelled":
                stats["cancelled"] += 1
                # cancelled不算失败也不打断

    # 计算失败率
    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)
        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
    - info: 失败次数>=2
    """
    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,
            "total": stats["total"],
            "failure_rate": rate,
            "current_streak": streak,
            "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


def generate_report(critical, warning, info, days, total_runs):
    """生成Markdown报告"""
    lines = []
    lines.append("# CI重复失败检测报告")
    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("")

    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"| 🔵 关注 (失败≥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['failure_rate']}%**")
            lines.append(f"- 当前连续失败: **{item['current_streak']}** 次 (历史最高: {item['max_streak']} 次)")
            lines.append("")
            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']})")
                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("")

    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("")

    return "\n".join(lines)


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]:
            text += f"• {item['name']}\n"
            text += f"  失败 {item['fail_count']}/{item['total']} ({item['failure_rate']}%) 连续{item['current_streak']}次\n"
        if len(critical) > 5:
            text += f"  ...还有{len(critical)-5}个\n"
        text += "\n"

    if warning:
        text += "【警告】\n"
        for item in warning[:5]:
            text += f"• {item['name']}: {item['fail_count']}次失败 ({item['failure_rate']}%)\n"
        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}"}}

    data = json.dumps(payload).encode()
    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())
            if result.get("code") == 0 or result.get("StatusCode") == 0:
                print("  ✅ 飞书通知已发送")
                return True
            else:
                print(f"  ⚠️  飞书返回: {result}")
                return False
    except Exception as e:
        print(f"  ❌ 飞书通知失败: {e}")
        return False


def main():
    print(f"=== CI重复失败检测 ===")
    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")
    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']}次"
            )
    if warning:
        print("🟡 警告:")
        for item in warning[:5]:
            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)
    elif warning:
        sys.exit(1)


if __name__ == "__main__":
    main()