xiaoxia-saas/scripts/ci/ci_dashboard.py

#!/usr/bin/env python3
"""
CI 可观测性看板 - 从 Gitea Actions API 拉取数据并生成 Markdown 日报

用法:
  python3 scripts/ci/ci_dashboard.py --days 7
  python3 scripts/ci/ci_dashboard.py --days 30 --output ci_report.md
  python3 scripts/ci/ci_dashboard.py --workflow ci-cd.yml --days 7

环境变量:
  GITEA_URL      Gitea 地址 (默认 https://git.xiaoxiajianji.com)
  GITEA_REPO     仓库 (默认 xiaoxia/xiaoxia-saas)
  GITEA_TOKEN    API Token (优先) 或 GITEA_USERNAME + GITEA_PASSWORD
"""

import argparse
import base64
import json
import math
import os
import statistics
import sys
import urllib.error
import urllib.request
from collections import defaultdict
from datetime import datetime, timedelta, timezone

# ── 配置 ──────────────────────────────────────────────
DEFAULT_GITEA_URL = "https://git.xiaoxiajianji.com"
DEFAULT_REPO = "xiaoxia/xiaoxia-saas"
DEFAULT_DAYS = 7
PAGE_LIMIT = 50  # 每页数量，最大50


# ── API 封装 ─────────────────────────────────────────
class GiteaActions:
    def __init__(self, base_url, repo, token=None, username=None, password=None):
        self.base_url = base_url.rstrip("/")
        self.repo = repo
        self.token = token
        self.username = username
        self.password = password
        self.api_base = f"{self.base_url}/api/v1/repos/{self.repo}/actions"

    def _request(self, path):
        url = f"{self.api_base}/{path}"
        req = urllib.request.Request(url)
        if self.token:
            req.add_header("Authorization", f"token {self.token}")
        elif self.username and self.password:
            auth = base64.b64encode(f"{self.username}:{self.password}".encode()).decode()
            req.add_header("Authorization", f"Basic {auth}")
        try:
            with urllib.request.urlopen(req, timeout=30) as resp:
                return json.loads(resp.read().decode())
        except urllib.error.HTTPError as e:
            print(f"[WARN] HTTP {e.code}: {url}", file=sys.stderr)
            return None
        except Exception as e:
            print(f"[WARN] 请求失败 {url}: {e}", file=sys.stderr)
            return None

    def list_runs(self, status=None, branch=None, event=None, page=1, limit=PAGE_LIMIT):
        """获取 workflow runs 列表"""
        params = []
        if status:
            params.append(f"status={status}")
        if branch:
            params.append(f"branch={branch}")
        if event:
            params.append(f"event={event}")
        params.append(f"page={page}")
        params.append(f"limit={limit}")
        path = f"runs?{'&'.join(params)}"
        data = self._request(path)
        if not data:
            return [], 0
        runs = data.get("workflow_runs", [])
        total = data.get("total_count", 0)
        return runs, total

    def get_run_jobs(self, run_id):
        """获取 run 的所有 job"""
        data = self._request(f"runs/{run_id}/jobs")
        if not data:
            return []
        return data.get("jobs", [])

    def list_workflows(self):
        """获取所有 workflow"""
        data = self._request("workflows")
        if not data:
            return []
        return data.get("workflows", [])


# ── 工具函数 ─────────────────────────────────────────
def parse_datetime(s):
    """解析 ISO 格式时间字符串"""
    if not s or s.startswith("1970") or s.startswith("0001"):
        return None
    try:
        if s.endswith("Z"):
            s = s[:-1] + "+00:00"
        return datetime.fromisoformat(s)
    except Exception:
        return None


def to_shanghai(dt):
    """转换为上海时区"""
    if dt is None:
        return None
    if dt.tzinfo is None:
        dt = dt.replace(tzinfo=timezone.utc)
    return dt.astimezone(timezone(timedelta(hours=8)))


def duration_seconds(start_str, end_str):
    """计算耗时（秒）"""
    start = parse_datetime(start_str)
    end = parse_datetime(end_str)
    if not start or not end:
        return None
    return (end - start).total_seconds()


def fmt_duration(seconds):
    """格式化耗时显示"""
    if seconds is None:
        return "N/A"
    seconds = int(seconds)
    if seconds < 60:
        return f"{seconds}s"
    mins, secs = divmod(seconds, 60)
    if mins < 60:
        return f"{mins}m{secs:02d}s"
    hours, mins = divmod(mins, 60)
    return f"{hours}h{mins:02d}m"


def percentile(sorted_values, p):
    """计算百分位数"""
    if not sorted_values:
        return None
    k = (len(sorted_values) - 1) * (p / 100)
    f = math.floor(k)
    c = math.ceil(k)
    if f == c:
        return sorted_values[int(k)]
    return sorted_values[f] * (c - k) + sorted_values[c] * (k - f)


def classify_failure(job_name, step_name=None):
    """根据失败的 job/step 名称分类失败原因"""
    name = f"{job_name} {step_name or ''}".lower()
    if any(k in name for k in ["lint", "ruff", "flake8", "eslint", "prettier", "black", "mypy"]):
        return "代码质量 / Lint"
    if any(k in name for k in ["unit test", "pytest", "vitest", "jest"]):
        return "单元测试失败"
    if any(k in name for k in ["integration", "e2e"]):
        return "集成测试 / E2E"
    if any(k in name for k in ["build", "compile", "docker", "image"]):
        return "构建失败"
    if any(k in name for k in ["deploy", "preview", "release"]):
        return "部署失败"
    if any(k in name for k in ["setup", "checkout", "cache", "install", "deps"]):
        return "环境 / 依赖"
    if any(k in name for k in ["migrate", "migration", "schema"]):
        return "数据库迁移"
    return "其他"


# ── 数据收集 ─────────────────────────────────────────
def fetch_runs_in_range(ga, start_date, end_date, workflow_filter=None):
    """拉取指定日期范围内的所有 completed runs"""
    all_runs = []
    page = 1
    print(f"[INFO] 拉取 {start_date} ~ {end_date} 的 CI runs...", file=sys.stderr)

    while True:
        runs, total = ga.list_runs(status="completed", page=page, limit=PAGE_LIMIT)
        if not runs:
            break

        if workflow_filter:
            runs = [r for r in runs if workflow_filter in r.get("path", "")]

        in_range = []
        out_range_old = False

        for run in runs:
            started = to_shanghai(parse_datetime(run.get("started_at")))
            if not started:
                continue
            run_date = started.date()
            if start_date <= run_date <= end_date:
                in_range.append(run)
            elif run_date < start_date:
                out_range_old = True

        all_runs.extend(in_range)
        print(
            f"[INFO] 第 {page} 页: {len(runs)} 条, 范围内 {len(in_range)} 条, 累计 {len(all_runs)} 条", file=sys.stderr
        )

        if out_range_old or len(runs) < PAGE_LIMIT:
            break

        page += 1
        if page > 100:
            print("[WARN] 超过100页，停止拉取", file=sys.stderr)
            break

    print(f"[INFO] 共获取 {len(all_runs)} 条 run 数据", file=sys.stderr)
    return all_runs


def enrich_with_jobs(ga, runs, max_failures=50):
    """为 runs 补充 job 详情（失败原因分析 + runner 统计）

    失败 run 按时间倒序取最近 N 个（避免 API 调用过多），
    成功 run 采样用于 runner 分布统计。
    """
    # 失败 run 取最近 N 个
    failure_runs = [r for r in runs if r.get("conclusion") != "success"]
    failure_runs = failure_runs[:max_failures]  # 已经是时间倒序
    print(f"[INFO] 为最近 {len(failure_runs)} 个失败 run 拉取 job 详情...", file=sys.stderr)

    for i, run in enumerate(failure_runs):
        jobs = ga.get_run_jobs(run["id"])
        run["_jobs"] = jobs
        if (i + 1) % 10 == 0:
            print(f"[INFO] 已处理 {i+1}/{len(failure_runs)}", file=sys.stderr)

    # 成功 run 采样用于 runner 分布
    success_runs = [r for r in runs if r.get("conclusion") == "success"]
    sample_size = min(50, len(success_runs))
    if sample_size > 0:
        sampled = success_runs[:: max(1, len(success_runs) // sample_size)]
        print(f"[INFO] 采样 {len(sampled)} 个成功 run 用于 runner 统计...", file=sys.stderr)
        for run in sampled:
            if "_jobs" not in run:
                jobs = ga.get_run_jobs(run["id"])
                run["_jobs"] = jobs

    return runs


# ── 统计分析 ─────────────────────────────────────────
def analyze_runs(runs):
    """对 runs 做全面统计分析"""
    if not runs:
        return {}

    # 基础统计
    total = len(runs)
    success = sum(1 for r in runs if r.get("conclusion") == "success")
    failure = sum(1 for r in runs if r.get("conclusion") == "failure")
    cancelled = sum(1 for r in runs if r.get("conclusion") == "cancelled")
    other = total - success - failure - cancelled
    success_rate = (success / total * 100) if total > 0 else 0

    # 耗时统计
    durations = []
    for r in runs:
        d = duration_seconds(r.get("started_at"), r.get("completed_at"))
        if d and d > 0:
            durations.append(d)
    durations.sort()

    avg_dur = statistics.mean(durations) if durations else None
    median_dur = percentile(durations, 50)
    p95_dur = percentile(durations, 95)

    # 按日期统计
    daily_stats = defaultdict(lambda: {"total": 0, "success": 0, "failure": 0, "durations": []})
    for r in runs:
        started = to_shanghai(parse_datetime(r.get("started_at")))
        if not started:
            continue
        day = started.date().isoformat()
        daily_stats[day]["total"] += 1
        if r.get("conclusion") == "success":
            daily_stats[day]["success"] += 1
        elif r.get("conclusion") == "failure":
            daily_stats[day]["failure"] += 1
        d = duration_seconds(r.get("started_at"), r.get("completed_at"))
        if d and d > 0:
            daily_stats[day]["durations"].append(d)

    # 按 workflow 统计
    wf_stats = defaultdict(lambda: {"total": 0, "success": 0, "failure": 0, "durations": []})
    for r in runs:
        path = r.get("path", "")
        wf_name = path.split("@")[0] if "@" in path else path
        wf_stats[wf_name]["total"] += 1
        if r.get("conclusion") == "success":
            wf_stats[wf_name]["success"] += 1
        elif r.get("conclusion") == "failure":
            wf_stats[wf_name]["failure"] += 1
        d = duration_seconds(r.get("started_at"), r.get("completed_at"))
        if d and d > 0:
            wf_stats[wf_name]["durations"].append(d)

    # 按触发事件统计
    event_stats = defaultdict(lambda: {"total": 0, "success": 0, "failure": 0})
    for r in runs:
        evt = r.get("event", "unknown")
        event_stats[evt]["total"] += 1
        if r.get("conclusion") == "success":
            event_stats[evt]["success"] += 1
        elif r.get("conclusion") == "failure":
            event_stats[evt]["failure"] += 1

    # 失败原因 + runner + job 耗时（需要 _jobs 数据）
    failure_categories = defaultdict(int)
    failed_jobs_by_name = defaultdict(int)
    runner_stats = defaultdict(lambda: {"jobs": 0, "success": 0, "failure": 0, "durations": []})
    job_time_stats = defaultdict(list)

    for r in runs:
        jobs = r.get("_jobs", [])
        if not jobs:
            continue
        for job in jobs:
            runner = job.get("runner_name", "unknown")
            conclusion = job.get("conclusion", "unknown")
            runner_stats[runner]["jobs"] += 1
            if conclusion == "success":
                runner_stats[runner]["success"] += 1
            elif conclusion == "failure":
                runner_stats[runner]["failure"] += 1

            jd = duration_seconds(job.get("started_at"), job.get("completed_at"))
            if jd and jd > 0:
                runner_stats[runner]["durations"].append(jd)
                job_time_stats[job.get("name", "unknown")].append(jd)

            if conclusion == "failure":
                failed_jobs_by_name[job.get("name", "unknown")] += 1
                failed_step = None
                for step in job.get("steps", []):
                    if step.get("conclusion") == "failure":
                        failed_step = step.get("name")
                        break
                category = classify_failure(job.get("name", ""), failed_step)
                failure_categories[category] += 1

    return {
        "total": total,
        "success": success,
        "failure": failure,
        "cancelled": cancelled,
        "other": other,
        "success_rate": success_rate,
        "avg_duration": avg_dur,
        "median_duration": median_dur,
        "p95_duration": p95_dur,
        "durations": durations,
        "daily_stats": dict(sorted(daily_stats.items())),
        "workflow_stats": dict(wf_stats),
        "event_stats": dict(event_stats),
        "failure_categories": dict(failure_categories),
        "failed_jobs_top": dict(sorted(failed_jobs_by_name.items(), key=lambda x: -x[1])[:15]),
        "runner_stats": dict(runner_stats),
        "job_time_stats": dict(job_time_stats),
    }


# ── 报表生成 ─────────────────────────────────────────
def generate_markdown(stats, start_date, end_date, repo):
    """生成 Markdown 格式的日报"""
    lines = []

    lines.append("# CI 运行状态看板")
    lines.append("")
    lines.append(f"> 统计周期: **{start_date} ~ {end_date}**")
    lines.append(f"> 仓库: `{repo}`")
    lines.append(f"> 生成时间: {datetime.now().strftime('%Y-%m-%d %H:%M:%S')}")
    lines.append("")

    # 概览
    lines.append("## 📊 整体概览")
    lines.append("")
    lines.append("| 指标 | 数值 |")
    lines.append("|------|------|")
    lines.append(f"| 总构建次数 | **{stats['total']}** |")
    lines.append(f"| ✅ 成功 | {stats['success']} |")
    lines.append(f"| ❌ 失败 | {stats['failure']} |")
    lines.append(f"| ⏹️ 取消 | {stats['cancelled']} |")
    lines.append(f"| 📈 成功率 | **{stats['success_rate']:.1f}%** |")
    lines.append(f"| ⏱️ 平均耗时 | {fmt_duration(stats['avg_duration'])} |")
    lines.append(f"| ⏱️ P50 耗时 | {fmt_duration(stats['median_duration'])} |")
    lines.append(f"| ⏱️ P95 耗时 | {fmt_duration(stats['p95_duration'])} |")
    lines.append("")

    # 每日趋势
    lines.append("## 📈 每日趋势")
    lines.append("")
    lines.append("| 日期 | 总次数 | 成功 | 失败 | 成功率 | 平均耗时 | P95 耗时 |")
    lines.append("|------|--------|------|------|--------|----------|----------|")
    for day, s in stats["daily_stats"].items():
        rate = (s["success"] / s["total"] * 100) if s["total"] > 0 else 0
        durations = sorted(s["durations"])
        avg = statistics.mean(durations) if durations else None
        p95 = percentile(durations, 95) if durations else None
        lines.append(
            f"| {day} | {s['total']} | {s['success']} | {s['failure']} | {rate:.1f}% | {fmt_duration(avg)} | {fmt_duration(p95)} |"
        )
    lines.append("")

    # 成功率趋势图
    lines.append("### 成功率趋势图")
    lines.append("")
    lines.append("```")
    max_bar = 40
    days = list(stats["daily_stats"].keys())
    if len(days) > 14:
        days = days[-14:]
    for day in days:
        s = stats["daily_stats"][day]
        rate = (s["success"] / s["total"] * 100) if s["total"] > 0 else 0
        bar_len = int(rate / 100 * max_bar)
        bar = "█" * bar_len + "░" * (max_bar - bar_len)
        lines.append(f"{day} {bar} {rate:5.1f}% ({s['total']}次)")
    lines.append("```")
    lines.append("")

    # 按 Workflow 统计
    lines.append("## 🧩 各 Workflow 统计")
    lines.append("")
    wf_sorted = sorted(stats["workflow_stats"].items(), key=lambda x: -x[1]["total"])
    lines.append("| Workflow | 次数 | 成功 | 失败 | 成功率 | 平均耗时 | P95 耗时 |")
    lines.append("|----------|------|------|------|--------|----------|----------|")
    for wf, s in wf_sorted:
        rate = (s["success"] / s["total"] * 100) if s["total"] > 0 else 0
        durations = sorted(s["durations"])
        avg = statistics.mean(durations) if durations else None
        p95 = percentile(durations, 95) if durations else None
        wf_short = wf.split("/")[-1] if "/" in wf else wf
        lines.append(
            f"| `{wf_short}` | {s['total']} | {s['success']} | {s['failure']} | {rate:.1f}% | {fmt_duration(avg)} | {fmt_duration(p95)} |"
        )
    lines.append("")

    # 失败原因分析
    if stats["failure_categories"]:
        lines.append("## ❌ 失败原因分析")
        lines.append("")
        lines.append("> ⚠️ 基于最近 N 个失败 run 采样分析，用于趋势参考")
        lines.append("")
        lines.append("### 按分类统计")
        lines.append("")
        total_failures = sum(stats["failure_categories"].values())
        fc_sorted = sorted(stats["failure_categories"].items(), key=lambda x: -x[1])
        lines.append("| 分类 | 次数 | 占比 |")
        lines.append("|------|------|------|")
        for cat, cnt in fc_sorted:
            pct = (cnt / total_failures * 100) if total_failures > 0 else 0
            lines.append(f"| {cat} | {cnt} | {pct:.1f}% |")
        lines.append("")

        lines.append("### Top 失败 Job")
        lines.append("")
        lines.append("| Job 名称 | 失败次数 |")
        lines.append("|----------|----------|")
        for job, cnt in stats["failed_jobs_top"].items():
            lines.append(f"| `{job}` | {cnt} |")
        lines.append("")

    # Runner 利用率
    if stats["runner_stats"]:
        lines.append("## 🏃 Runner 利用率")
        lines.append("")
        runner_sorted = sorted(stats["runner_stats"].items(), key=lambda x: -x[1]["jobs"])
        lines.append("| Runner | Job 数 | 成功 | 失败 | 成功率 | 平均耗时 |")
        lines.append("|--------|--------|------|------|--------|----------|")
        for runner, s in runner_sorted:
            rate = (s["success"] / s["jobs"] * 100) if s["jobs"] > 0 else 0
            avg = statistics.mean(s["durations"]) if s["durations"] else None
            lines.append(
                f"| `{runner}` | {s['jobs']} | {s['success']} | {s['failure']} | {rate:.1f}% | {fmt_duration(avg)} |"
            )
        lines.append("")

    # Job 耗时排行
    if stats["job_time_stats"]:
        lines.append("## ⏱️ Job 耗时排行 (Top 20 by P95)")
        lines.append("")
        job_stats = []
        for name, durs in stats["job_time_stats"].items():
            if not durs:
                continue
            durs_sorted = sorted(durs)
            job_stats.append(
                {
                    "name": name,
                    "count": len(durs_sorted),
                    "avg": statistics.mean(durs_sorted),
                    "p50": percentile(durs_sorted, 50),
                    "p95": percentile(durs_sorted, 95),
                }
            )
        job_stats.sort(key=lambda x: -x["p95"])
        top_n = min(20, len(job_stats))
        lines.append("| Job 名称 | 次数 | 平均 | P50 | P95 |")
        lines.append("|----------|------|------|-----|-----|")
        for j in job_stats[:top_n]:
            lines.append(
                f"| `{j['name']}` | {j['count']} | {fmt_duration(j['avg'])} | {fmt_duration(j['p50'])} | {fmt_duration(j['p95'])} |"
            )
        lines.append("")

    # 触发事件分布
    lines.append("## 📋 触发事件分布")
    lines.append("")
    evt_sorted = sorted(stats["event_stats"].items(), key=lambda x: -x[1]["total"])
    lines.append("| 事件类型 | 次数 | 成功 | 失败 | 成功率 |")
    lines.append("|----------|------|------|------|--------|")
    for evt, s in evt_sorted:
        rate = (s["success"] / s["total"] * 100) if s["total"] > 0 else 0
        lines.append(f"| `{evt}` | {s['total']} | {s['success']} | {s['failure']} | {rate:.1f}% |")
    lines.append("")

    return "\n".join(lines)


# ── 主函数 ───────────────────────────────────────────
def main():
    parser = argparse.ArgumentParser(description="CI 可观测性看板 - 生成 Gitea Actions 运行状态报表")
    parser.add_argument("--days", type=int, default=DEFAULT_DAYS, help=f"统计最近 N 天 (默认 {DEFAULT_DAYS})")
    parser.add_argument("--output", "-o", type=str, help="输出文件路径 (默认输出到 stdout)")
    parser.add_argument("--workflow", type=str, help="只统计指定 workflow (如 ci-cd.yml)")
    parser.add_argument("--gitea-url", type=str, default=os.environ.get("GITEA_URL", DEFAULT_GITEA_URL))
    parser.add_argument("--repo", type=str, default=os.environ.get("GITEA_REPO", DEFAULT_REPO))
    parser.add_argument("--token", type=str, default=os.environ.get("GITEA_TOKEN"))
    parser.add_argument("--username", type=str, default=os.environ.get("GITEA_USERNAME"))
    parser.add_argument("--password", type=str, default=os.environ.get("GITEA_PASSWORD"))
    parser.add_argument("--no-job-detail", action="store_true", help="不拉取 job 详情")
    parser.add_argument("--max-failures", type=int, default=50, help="最多分析多少个失败 run 的 job 详情 (默认 50)")
    args = parser.parse_args()

    ga = GiteaActions(
        base_url=args.gitea_url,
        repo=args.repo,
        token=args.token,
        username=args.username,
        password=args.password,
    )

    end_date = datetime.now().date()
    start_date = end_date - timedelta(days=args.days - 1)

    runs = fetch_runs_in_range(ga, start_date, end_date, args.workflow)
    if not runs:
        print("[ERROR] 未获取到任何数据", file=sys.stderr)
        sys.exit(1)

    if not args.no_job_detail:
        runs = enrich_with_jobs(ga, runs, max_failures=args.max_failures)

    stats = analyze_runs(runs)
    md = generate_markdown(stats, start_date, end_date, args.repo)

    if args.output:
        with open(args.output, "w", encoding="utf-8") as f:
            f.write(md)
        print(f"[INFO] 报表已保存到 {args.output}", file=sys.stderr)
    else:
        print(md)


if __name__ == "__main__":
    main()