diff --git a/.gitea/workflows/code-review.yml b/.gitea/workflows/code-review.yml
index 07daa8bd5..9c691adba 100644
--- a/.gitea/workflows/code-review.yml
+++ b/.gitea/workflows/code-review.yml
@@ -28,6 +28,14 @@ jobs:
 
       - name: Install dependencies
         run: |
+          # 确保 python3-pip 可用（兼容不同基础镜像）
+          if ! python3 -m pip --version >/dev/null 2>&1; then
+            apt-get update -qq && apt-get install -y -qq python3-pip python3-venv >/dev/null 2>&1
+          fi
+          # 部分镜像 ensurepip 方式兜底
+          if ! python3 -m pip --version >/dev/null 2>&1; then
+            python3 -m ensurepip --upgrade 2>/dev/null || curl -sS https://bootstrap.pypa.io/get-pip.py | python3
+          fi
           python3 -m pip install --upgrade pip
           python3 -m pip install requests
 
diff --git a/scripts/ci/ci_health_check.py b/scripts/ci/ci_health_check.py
new file mode 100644
index 000000000..cc7523e60
--- /dev/null
+++ b/scripts/ci/ci_health_check.py
@@ -0,0 +1,298 @@
+#!/usr/bin/env python3
+"""
+CI 健康度快速检查脚本
+- 统计最近 N 条 run 的成功率（按 workflow 分类）
+- 列出失败的 run 和失败的 job/step
+- 区分基础设施问题 vs 业务代码问题
+- 输出简洁的健康度报告
+
+用法:
+  python3 scripts/ci/ci_health_check.py [--limit 20] [--workflow ci-pipeline.yml] [--json]
+
+环境变量:
+  GITEA_TOKEN   API token（必需）
+  GITEA_API_URL Gitea API 地址，默认 https://git.xiaoxiajianji.com/api/v1
+  GITEA_REPO    仓库，默认 xiaoxia/xiaoxia-saas
+"""
+
+import argparse
+import json
+import os
+import sys
+import urllib.request
+from datetime import datetime, timedelta, timezone
+
+# ---- 基础设施问题关键词（命中即判定为基础设施问题）----
+INFRA_KEYWORDS = [
+    # 网络/连接
+    "Couldn't connect to server",
+    "Connection refused",
+    "Connection reset",
+    "Connection timed out",
+    "Failed to connect to",
+    "network is unreachable",
+    "TLS handshake timeout",
+    "SSL certificate problem",
+    # 容器/Runner
+    "No such container",
+    "container already exists",
+    "docker: not found",
+    "no space left on device",
+    "out of memory",
+    "OOMKilled",
+    "pull access denied",
+    "manifest unknown",
+    "Error response from daemon",
+    "runner",
+    "runner is not online",
+    "no matching runners",
+    # Checkout/Git
+    "Could not resolve host",
+    "fatal: unable to access",
+    "The remote end hung up unexpectedly",
+    "early EOF",
+    "index-pack failed",
+    "git fetch",
+    "checkout failed",
+    "ETXTBSY",
+    "text file busy",
+    # 镜像/环境
+    "No module named pip",
+    "pip: not found",
+    "command not found: python",
+    "python3: not found",
+    "node: not found",
+    "npm: not found",
+    "exec format error",
+    "standard_init_linux.go",
+    # 系统/资源
+    "Input/output error",
+    "device or resource busy",
+    "No space left on device",
+    "Disk full",
+    # 鉴权/配置
+    "401 Unauthorized",
+    "403 Forbidden",
+    "404 Not Found",
+    "identity_sign: private key",
+    "Permission denied",
+]
+
+
+def api_get(path: str) -> dict:
+    base = os.environ.get("GITEA_API_URL", "https://git.xiaoxiajianji.com/api/v1")
+    repo = os.environ.get("GITEA_REPO", "xiaoxia/xiaoxia-saas")
+    token = os.environ.get("GITEA_TOKEN", "")
+    url = f"{base}/repos/{repo}/{path}"
+    req = urllib.request.Request(url, headers={"Authorization": f"token {token}"})
+    with urllib.request.urlopen(req, timeout=30) as resp:
+        return json.loads(resp.read().decode())
+
+
+def get_run_jobs(run_id: int) -> list:
+    return api_get(f"actions/runs/{run_id}/jobs").get("jobs", [])
+
+
+def get_job_log(job_id: int) -> str:
+    try:
+        return api_get(f"actions/jobs/{job_id}/logs")
+    except Exception:
+        return ""
+
+
+def classify_failure(job: dict) -> str:
+    """判断失败原因类型: infra / business / unknown"""
+    name = job.get("name", "")
+    # 仅根据 job 名称做初步分类（更精确需读日志，但代价高）
+    infra_jobs = ["Checkout", "Build", "Deploy", "Cleanup"]
+    business_jobs = [
+        "Unit Tests",
+        "Integration Tests",
+        "Frontend Lint",
+        "Frontend Unit Tests",
+        "Staging E2E",
+        "E2E",
+        "Validate Code Quality",
+    ]
+    name_lower = name.lower()
+    if (
+        any(k.lower() in name_lower for k in infra_jobs)
+        and "Test" not in name
+        and "Lint" not in name
+        and "Validate" not in name
+    ):
+        return "infra"
+    if any(k.lower() in name_lower for k in business_jobs):
+        return "business"
+    return "unknown"
+
+
+def analyze_with_log(job_id: int) -> str:
+    """通过日志关键词精确分类"""
+    log = get_job_log(job_id)
+    log_lower = log.lower()
+    for kw in INFRA_KEYWORDS:
+        if kw.lower() in log_lower:
+            return "infra"
+    return "business"
+
+
+def fmt_time(t: str) -> str:
+    if not t or t.startswith("1970"):
+        return "-"
+    try:
+        dt = datetime.fromisoformat(t.replace("Z", "+00:00"))
+        bj = dt.astimezone(timezone(timedelta(hours=8)))
+        return bj.strftime("%m-%d %H:%M")
+    except Exception:
+        return t[:16]
+
+
+def main():
+    parser = argparse.ArgumentParser(description="CI 健康度快速检查")
+    parser.add_argument("--limit", type=int, default=20, help="最近多少条 run")
+    parser.add_argument("--workflow", type=str, default="", help="只看某个 workflow")
+    parser.add_argument("--json", action="store_true", help="JSON 输出")
+    parser.add_argument("--deep", action="store_true", help="深度检查（读日志，较慢）")
+    args = parser.parse_args()
+
+    if not os.environ.get("GITEA_TOKEN"):
+        print("错误: 请设置 GITEA_TOKEN 环境变量", file=sys.stderr)
+        sys.exit(1)
+
+    # 1. 获取最近 run
+    runs = api_get(f"actions/runs?limit={args.limit}").get("workflow_runs", [])
+    if args.workflow:
+        runs = [r for r in runs if args.workflow in r.get("path", "")]
+
+    if not runs:
+        print("没有找到匹配的 run")
+        return
+
+    # 按 workflow 分组统计
+    wf_stats = {}
+    failed_runs = []
+
+    for r in runs:
+        path = r.get("path", "unknown")
+        # 提取 workflow 文件名，兼容各种 path 格式
+        if ".yml" in path or ".yaml" in path:
+            # ci-pipeline.yml@refs/heads/develop -> ci-pipeline.yml
+            wf = path.split("@")[0].split("/")[-1]
+        else:
+            wf = path.split("/")[-1] if "/" in path else path
+        if wf not in wf_stats:
+            wf_stats[wf] = {"total": 0, "success": 0, "failure": 0, "cancelled": 0, "others": 0}
+        wf_stats[wf]["total"] += 1
+        status = r.get("status", "")
+        conc = r.get("conclusion", "")
+        if status != "completed":
+            wf_stats[wf]["others"] += 1
+            continue
+        if conc == "success":
+            wf_stats[wf]["success"] += 1
+        elif conc == "failure":
+            wf_stats[wf]["failure"] += 1
+            failed_runs.append(r)
+        elif conc == "cancelled":
+            wf_stats[wf]["cancelled"] += 1
+        else:
+            wf_stats[wf]["others"] += 1
+
+    # 2. 失败 run 详情
+    failed_details = []
+    for r in failed_runs[:10]:  # 最多看10个失败的
+        jobs = get_run_jobs(r["id"])
+        failed_jobs = [j for j in jobs if j.get("conclusion") == "failure"]
+        job_infos = []
+        for j in failed_jobs:
+            cat = classify_failure(j)
+            if args.deep and cat == "unknown":
+                cat = analyze_with_log(j["id"])
+            # 找失败的 step
+            failed_steps = []
+            for step in j.get("steps", []):
+                if step.get("conclusion") == "failure":
+                    failed_steps.append(step.get("name", "?"))
+            job_infos.append(
+                {
+                    "name": j.get("name", ""),
+                    "category": cat,
+                    "failed_steps": failed_steps,
+                    "runner": j.get("runner_name", ""),
+                }
+            )
+        failed_details.append(
+            {
+                "id": r["id"],
+                "title": r.get("display_title", ""),
+                "branch": r.get("head_branch", ""),
+                "time": fmt_time(r.get("updated_at", "")),
+                "jobs": job_infos,
+            }
+        )
+
+    # 3. 输出
+    if args.json:
+        result = {"workflows": wf_stats, "failed_runs": failed_details}
+        print(json.dumps(result, ensure_ascii=False, indent=2))
+        return
+
+    # 文本报告
+    print("=" * 60)
+    print("  CI 健康度报告")
+    print("=" * 60)
+    print(f"统计范围: 最近 {len(runs)} 条 run")
+    print(f"时间: {datetime.now(timezone(timedelta(hours=8))).strftime('%Y-%m-%d %H:%M:%S')}")
+    print()
+
+    print("📊 各 Workflow 成功率:")
+    print("-" * 60)
+    for wf, s in sorted(wf_stats.items()):
+        total = s["total"]
+        succ = s["success"]
+        rate = (succ / total * 100) if total > 0 else 0
+        bar = "█" * int(rate / 5) + "░" * (20 - int(rate / 5))
+        icon = "🟢" if rate >= 90 else ("🟡" if rate >= 70 else "🔴")
+        print(f"  {icon} {wf:35s} {rate:5.1f}%  {bar}  ({succ}/{total})")
+        if s["failure"]:
+            print(f"      失败: {s['failure']}  取消: {s['cancelled']}  进行中: {s['others']}")
+
+    if failed_details:
+        print()
+        print("❌ 失败详情:")
+        print("-" * 60)
+        for d in failed_details:
+            print(f"  #{d['id']} [{d['time']}] {d['title'][:45]}")
+            print(f"      分支: {d['branch']}")
+            for j in d["jobs"]:
+                cat_icon = "🏗️" if j["category"] == "infra" else ("🐛" if j["category"] == "business" else "❓")
+                steps = ", ".join(j["failed_steps"][:3]) if j["failed_steps"] else "未知"
+                print(f"      {cat_icon} {j['name'][:30]:30s} 失败步骤: {steps}")
+                if j["runner"]:
+                    print(f"          runner: {j['runner']}")
+    else:
+        print()
+        print("✅ 最近没有失败的 run")
+
+    # 总结
+    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())
+    infra_fail = sum(1 for d in failed_details for j in d["jobs"] if j["category"] == "infra")
+    biz_fail = sum(1 for d in failed_details for j in d["jobs"] if j["category"] == "business")
+    rate_all = (succ_all / total_all * 100) if total_all > 0 else 0
+    print()
+    print("=" * 60)
+    print(f"  总结: 总成功率 {rate_all:.1f}% ({succ_all}/{total_all})")
+    if fail_all > 0:
+        print(f"  失败job分类: 基础设施 {infra_fail} 个 | 业务代码 {biz_fail} 个")
+        if infra_fail > biz_fail:
+            print("  ⚠️ 主要是基础设施问题，建议优先排查 CI 环境")
+        else:
+            print("  💡 主要是业务代码问题，建议关注业务侧修复")
+    print("=" * 60)
+
+
+if __name__ == "__main__":
+    main()