.gitea/workflows/ci-health-daily.yml

@@ -0,0 +1,81 @@
+name: CI Health Daily Report
+
+on:
+  schedule:
+    - cron: '0 1 * * *'  # UTC 01:00 = 北京时间 09:00
+  workflow_dispatch:
+
+permissions:
+  contents: read
+
+jobs:
+  ci-health-report:
+    name: CI健康度每日巡检
+    runs-on: saas
+    timeout-minutes: 10
+
+    steps:
+    - name: Checkout code
+      shell: sh
+      env:
+        GITHUB_TOKEN: ${{ github.token }}
+      run: |
+        set -eu
+        python3 - <<'PY'
+        import io, os, tarfile, time, urllib.request, urllib.error
+        url = f"{os.environ['GITHUB_API_URL']}/repos/{os.environ['GITHUB_REPOSITORY']}/archive/{os.environ['GITHUB_SHA']}.tar.gz"
+        request = urllib.request.Request(url, headers={"Authorization": f"token {os.environ['GITHUB_TOKEN']}"})
+        last_err = None
+        for attempt in range(5):
+            try:
+                with urllib.request.urlopen(request, timeout=120) as response:
+                    archive = response.read()
+                break
+            except urllib.error.HTTPError as e:
+                last_err = e
+                if e.code >= 500 and attempt < 4:
+                    wait = 2 ** attempt
+                    print(f"Checkout HTTP {e.code}, retrying in {wait}s (attempt {attempt+1}/5)...")
+                    time.sleep(wait)
+                    continue
+                raise
+            except Exception as e:
+                last_err = e
+                if attempt < 4:
+                    wait = 2 ** attempt
+                    print(f"Checkout error: {e}, retrying in {wait}s (attempt {attempt+1}/5)...")
+                    time.sleep(wait)
+                    continue
+                raise
+        else:
+            raise last_err
+        with tarfile.open(fileobj=io.BytesIO(archive), mode='r:gz') as tar:
+            root_prefix = tar.getmembers()[0].name.split('/', 1)[0] + '/'
+            for member in tar.getmembers():
+                name = member.name
+                if name == root_prefix[:-1]:
+                    continue
+                if name.startswith(root_prefix):
+                    member.name = name[len(root_prefix):]
+                if member.name:
+                    tar.extract(member, '.')
+        PY
+
+    - name: Run CI health check and report
+      shell: sh
+      env:
+        GITEA_TOKEN: ${{ github.token }}
+        CI_NOTIFY_WEBHOOK: ${{ secrets.CI_NOTIFY_WEBHOOK }}
+      run: |
+        set +e
+        echo "=== CI健康度每日巡检 ==="
+        echo "时间: $(date '+%Y-%m-%d %H:%M:%S')"
+        echo ""
+
+        python3 scripts/ci/ci_health_report.py --limit 30
+        EXIT_CODE=$?
+
+        echo ""
+        echo "巡检完成 (exit code: $EXIT_CODE)"
+        # 永远成功，不影响CI状态（通知失败不应该标红）
+        exit 0

scripts/ci/ci_health_report.py

@@ -0,0 +1,251 @@
+#!/usr/bin/env python3
+"""
+CI健康度每日巡检报告脚本
+- 调用ci_health_check.py获取数据
+- 有失败时生成飞书卡片通知并发送
+- 无失败时静默退出（不打扰）
+- 用于每日定时巡检
+
+用法:
+  python3 scripts/ci/ci_health_report.py [--limit 30] [--dry-run]
+
+环境变量:
+  GITEA_TOKEN         API token（必需）
+  CI_NOTIFY_WEBHOOK   飞书webhook地址（必需，用于发报告）
+  GITEA_API_URL       Gitea API 地址
+  GITEA_REPO          仓库
+"""
+
+import argparse
+import json
+import os
+import subprocess
+import sys
+import urllib.request
+from datetime import datetime, timedelta, timezone
+
+
+def run_health_check(limit: int) -> dict:
+    """调用ci_health_check.py获取JSON结果"""
+    script_dir = os.path.dirname(os.path.abspath(__file__))
+    cmd = [
+        sys.executable,
+        os.path.join(script_dir, "ci_health_check.py"),
+        "--json",
+        "--limit",
+        str(limit),
+    ]
+    env = os.environ.copy()
+    # 确保GITEA_TOKEN传递
+    if not env.get("GITEA_TOKEN") and env.get("GITHUB_TOKEN"):
+        env["GITEA_TOKEN"] = env["GITHUB_TOKEN"]
+
+    result = subprocess.run(cmd, capture_output=True, text=True, env=env)
+    if result.returncode != 0:
+        print(f"health check failed: {result.stderr}")
+        return {"workflows": {}, "failed_runs": []}
+    try:
+        return json.loads(result.stdout)
+    except json.JSONDecodeError:
+        print(f"failed to parse health check output: {result.stdout[:200]}")
+        return {"workflows": {}, "failed_runs": []}
+
+
+def build_feishu_card(data: dict) -> dict:
+    """构建飞书卡片消息"""
+    wf_stats = data.get("workflows", {})
+    failed_runs = data.get("failed_runs", [])
+
+    # 统计数据
+    total_all = sum(s["total"] for s in wf_stats.values())
+    succ_all = sum(s["success"] for s in wf_stats.values())
+    fail_all = sum(s["failure"] for s in wf_stats.values())
+    rate_all = (succ_all / total_all * 100) if total_all > 0 else 0
+
+    # 失败分类
+    infra_fail = 0
+    biz_fail = 0
+    unknown_fail = 0
+    for run in failed_runs:
+        for job in run.get("jobs", []):
+            cat = job.get("category", "unknown")
+            if cat == "infra":
+                infra_fail += 1
+            elif cat == "business":
+                biz_fail += 1
+            else:
+                unknown_fail += 1
+
+    now = datetime.now(timezone(timedelta(hours=8))).strftime("%Y-%m-%d %H:%M")
+
+    # 各workflow成功率行
+    wf_lines = []
+    for wf, s in sorted(wf_stats.items()):
+        total = s["total"]
+        succ = s["success"]
+        fail = s["failure"]
+        rate = (succ / total * 100) if total > 0 else 0
+        icon = "🟢" if rate >= 90 else ("🟡" if rate >= 70 else "🔴")
+        wf_name = (
+            wf.replace("ci-pipeline.yml", "CI Pipeline")
+            .replace("code-review.yml", "Code Review")
+            .replace("daily-check.yml", "Daily Check")
+            .replace("preview-deploy.yml", "Preview Deploy")
+        )
+        wf_lines.append(f"{icon} **{wf_name}**: {rate:.0f}% ({succ}/{total}，失败{fail})")
+
+    # 失败详情（最多显示5条）
+    fail_detail_lines = []
+    for i, run in enumerate(failed_runs[:5]):
+        run_id = run["id"]
+        title = run.get("title", "")[:35]
+        branch = run.get("branch", "")
+        jobs_str = ", ".join(j["name"][:15] for j in run.get("jobs", [])[:3])
+        fail_detail_lines.append(f"• **#{run_id}** {title}\n   分支: {branch} | 失败: {jobs_str}")
+
+    if len(failed_runs) > 5:
+        fail_detail_lines.append(f"... 还有 {len(failed_runs) - 5} 条失败记录")
+
+    # 整体状态
+    if fail_all == 0:
+        status_text = "✅ 全部通过"
+        status_color = "green"
+    elif infra_fail > biz_fail:
+        status_text = "⚠️ 基础设施问题为主"
+        status_color = "yellow"
+    else:
+        status_text = "🔴 存在业务失败"
+        status_color = "red"
+
+    card = {
+        "config": {"wide_screen_mode": True},
+        "header": {
+            "title": {"tag": "plain_text", "content": f"CI告警 - 每日健康度巡检 ({now})"},
+            "template": status_color,
+        },
+        "elements": [
+            {
+                "tag": "div",
+                "text": {
+                    "tag": "lark_md",
+                    "content": f"**统计范围**: 最近 {total_all} 条 run\n**整体状态**: {status_text}\n**总成功率**: {rate_all:.1f}% ({succ_all}/{total_all})",
+                },
+            },
+            {"tag": "hr"},
+            {
+                "tag": "div",
+                "text": {
+                    "tag": "lark_md",
+                    "content": "**📊 各Workflow成功率**\n" + "\n".join(wf_lines) if wf_lines else "暂无数据",
+                },
+            },
+        ],
+    }
+
+    # 失败分类统计
+    if fail_all > 0:
+        card["elements"].append({"tag": "hr"})
+        card["elements"].append(
+            {
+                "tag": "div",
+                "text": {
+                    "tag": "lark_md",
+                    "content": f"**失败原因分类**\n🏗️ 基础设施: {infra_fail} 个\n🐛 业务代码: {biz_fail} 个\n❓ 待确认: {unknown_fail} 个",
+                },
+            }
+        )
+
+    # 失败详情
+    if fail_detail_lines:
+        card["elements"].append({"tag": "hr"})
+        card["elements"].append(
+            {
+                "tag": "div",
+                "text": {
+                    "tag": "lark_md",
+                    "content": "**❌ 失败详情**\n" + "\n\n".join(fail_detail_lines),
+                },
+            }
+        )
+
+    # 查看更多
+    card["elements"].append({"tag": "hr"})
+    base_url = os.environ.get("GITEA_BASE_URL", "https://git.xiaoxiajianji.com")
+    repo = os.environ.get("GITEA_REPO", "xiaoxia/xiaoxia-saas")
+    card["elements"].append(
+        {
+            "tag": "action",
+            "actions": [
+                {
+                    "tag": "button",
+                    "text": {"tag": "plain_text", "content": "查看CI面板"},
+                    "type": "primary",
+                    "url": f"{base_url}/{repo}/actions",
+                }
+            ],
+        }
+    )
+
+    return {"msg_type": "interactive", "card": card}
+
+
+def send_feishu(webhook: str, payload: dict) -> bool:
+    """发送飞书webhook"""
+    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().decode())
+            return result.get("code", -1) == 0 or result.get("StatusCode", -1) == 0
+    except Exception as e:
+        print(f"send feishu failed: {e}")
+        return False
+
+
+def main():
+    parser = argparse.ArgumentParser(description="CI健康度每日巡检报告")
+    parser.add_argument("--limit", type=int, default=30, help="统计最近N条run")
+    parser.add_argument("--dry-run", action="store_true", help="只打印不发送")
+    parser.add_argument("--always-notify", action="store_true", help="即使全部通过也发送通知")
+    args = parser.parse_args()
+
+    webhook = os.environ.get("CI_NOTIFY_WEBHOOK", "")
+    if not webhook and not args.dry_run:
+        print("未配置 CI_NOTIFY_WEBHOOK，跳过通知")
+        # 还是执行健康检查输出到日志，方便排查
+        data = run_health_check(args.limit)
+        print(f"health check done: {len(data.get('failed_runs', []))} failed")
+        return 0
+
+    # 执行健康检查
+    data = run_health_check(args.limit)
+    failed_count = len(data.get("failed_runs", []))
+
+    # 无失败且不强制通知 → 静默退出
+    if failed_count == 0 and not args.always_notify:
+        print("✅ 全部通过，静默退出")
+        return 0
+
+    # 构建并发送卡片
+    card = build_feishu_card(data)
+
+    if args.dry_run:
+        print(json.dumps(card, ensure_ascii=False, indent=2))
+        return 0
+
+    success = send_feishu(webhook, card)
+    if success:
+        print(f"📤 已发送健康度报告，失败 {failed_count} 条")
+    else:
+        print("❌ 发送飞书通知失败")
+
+    # 通知失败不阻断流程
+    return 0
+
+
+if __name__ == "__main__":
+    sys.exit(main())