')
+ html_parts.append(' ")
+ html_parts.append('
')
+ html_parts.append('
')
+ html_parts.append('
总成功率
')
+ html_parts.append(f'
{success_rate}%
')
+ html_parts.append("
")
+ html_parts.append('
')
+ html_parts.append('
总 Run 数
')
+ html_parts.append(f'
{total_runs}次
')
+ html_parts.append("
")
+ html_parts.append('
')
+ html_parts.append('
平均耗时
')
+ html_parts.append(f'
{avg_dur_min}分钟
')
+ html_parts.append("
")
+ html_parts.append('
')
+ html_parts.append('
基础设施故障率
')
+ html_parts.append(f'
{infra_fail_rate}%
')
+ html_parts.append("
")
+ html_parts.append("
")
+ html_parts.append('
')
+ html_parts.append('
')
+ html_parts.append("
📈 CI 成功率趋势
")
+ html_parts.append('
')
+ html_parts.append("
")
+ html_parts.append("
")
+ html_parts.append('
')
+ html_parts.append('
')
+ html_parts.append("
⏱️ 各 Workflow 平均耗时
")
+ html_parts.append('
')
+ html_parts.append("
")
+ html_parts.append('
')
+ html_parts.append("
❌ 失败原因分布
")
+ html_parts.append('
')
+ html_parts.append("
")
+ html_parts.append("
")
+ html_parts.append('
')
+ html_parts.append('
')
+ html_parts.append("
📋 各 Job 成功率排行(最低 15 名)
")
+ html_parts.append('
')
+ html_parts.append("
")
+ html_parts.append("
")
+ html_parts.append('
')
+ html_parts.append('
')
+ html_parts.append("
📊 每日 Run 数量趋势
")
+ html_parts.append('
')
+ html_parts.append("
")
+ html_parts.append("
")
+ html_parts.append(' ")
+ html_parts.append("
")
+ html_parts.append(" ")
+ html_parts.append("")
+ html_parts.append("")
+
+ return "\n".join(html_parts)
+
+
+# ── 主函数 ───────────────────────────────────────────
+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)")
+
+ # HTML 输出相关参数
+ parser.add_argument("--html", action="store_true", help="生成 HTML 可视化看板")
+ parser.add_argument("--html-output", type=str, help="HTML 输出文件路径 (默认 ci_dashboard.html)")
+
+ 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)
+
+ # HTML 模式
+ if args.html:
+ html = generate_html(stats, start_date, end_date, args.repo)
+ html_output = args.html_output or args.output or "ci_dashboard.html"
+ with open(html_output, "w", encoding="utf-8") as f:
+ f.write(html)
+ print(f"[INFO] HTML 看板已保存到 {html_output}", file=sys.stderr)
+ return
+
+ # 默认 Markdown 模式(向后兼容)
+ 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()
diff --git a/scripts/ci/ci_env.sh b/scripts/ci/ci_env.sh
new file mode 100755
index 000000000..516acc952
--- /dev/null
+++ b/scripts/ci/ci_env.sh
@@ -0,0 +1,14 @@
+#!/bin/bash
+# CI共享环境变量与常量定义
+# 所有CI脚本source此文件获取统一的配置,避免硬编码分散
+
+# === 共享常驻PG实例(CI_USE_SHARED_PG=true时使用)===
+export CI_SHARED_PG_PORT="${CI_SHARED_PG_PORT:-5433}"
+export CI_SHARED_PG_USER="${CI_SHARED_PG_USER:-postgres}"
+export CI_SHARED_PG_PASSWORD="${CI_SHARED_PG_PASSWORD:-ci_pg_2026!}"
+
+# === 本地PG默认端口(CI_USE_SHARED_PG=false时容器映射或本地PG)===
+export CI_LOCAL_PG_PORT="${CI_LOCAL_PG_PORT:-5432}"
+
+# === 默认数据库名 ===
+export CI_DEFAULT_DB="${CI_DEFAULT_DB:-xiaoxia_saas}"
diff --git a/scripts/ci/ci_failure_diagnosis.py b/scripts/ci/ci_failure_diagnosis.py
new file mode 100644
index 000000000..5244484b2
--- /dev/null
+++ b/scripts/ci/ci_failure_diagnosis.py
@@ -0,0 +1,447 @@
+#!/usr/bin/env python3
+"""CI失败诊断增强脚本:自动分类失败原因 + 提取关键错误 + 给出修复建议。
+# Trigger CI after auto-format fix
+
+支持的失败类型:
+1. Lint/格式问题 (ruff/black/eslint/prettier)
+2. 单元测试失败
+3. Docker构建失败
+4. 依赖安装失败 (pip/npm)
+5. 超时
+6. 缓存问题
+7. 数据库/迁移问题
+8. 网络问题
+9. 其他
+
+用法:
+ python3 scripts/ci/ci_failure_diagnosis.py [--job-name "Job Name"] [--log-file /path/to/log]
+
+如果不传--log-file,会尝试从Gitea API获取失败job的日志。
+"""
+
+import json
+import os
+import re
+import sys
+import urllib.request
+from dataclasses import dataclass, field
+from typing import List, Optional
+
+
+@dataclass
+class FailureDiagnosis:
+ """失败诊断结果"""
+
+ category: str # 失败分类
+ category_cn: str # 中文分类名
+ severity: str # 严重程度: high / medium / low
+ summary: str # 一句话摘要
+ error_lines: List[str] = field(default_factory=list) # 关键错误行
+ suggestions: List[str] = field(default_factory=list) # 修复建议
+ auto_fixable: bool = False # 是否可以自动修复
+ related_docs: str = "" # 相关文档链接
+
+
+# ============================================================
+# 失败模式定义
+# ============================================================
+
+FAILURE_PATTERNS = [
+ # ===== Lint / 格式问题 =====
+ {
+ "pattern": r"(ruff|black|isort)\b.*(error|failed|Error)",
+ "category": "lint_python",
+ "category_cn": "Python代码质量检查",
+ "severity": "low",
+ "summary_contains": ["ruff", "black", "isort"],
+ "suggestions": [
+ "本地运行 `black . && isort . && ruff check --fix .` 自动修复",
+ "使用 `scripts/agent-commit.sh` 提交(自动格式化)",
+ "如确认无误,可加 `# noqa: xxx` 忽略特定规则",
+ ],
+ "auto_fixable": True,
+ },
+ {
+ "pattern": r"ESLint|prettier|eslint",
+ "category": "lint_frontend",
+ "category_cn": "前端代码检查",
+ "severity": "low",
+ "summary_contains": ["eslint", "prettier"],
+ "suggestions": [
+ "本地运行 `cd apps/web && npm run lint:fix` 自动修复",
+ "Prettier问题: `cd apps/web && npx prettier --write .`",
+ ],
+ "auto_fixable": True,
+ },
+ {
+ "pattern": r"F\d{3}|E\d{3}|W\d{3}.*ruff|ruff.*F\d{3}",
+ "category": "lint_python",
+ "category_cn": "Python代码质量检查",
+ "severity": "low",
+ "suggestions": [
+ "F401: 删除未使用的import",
+ "F841: 删除未使用的变量或加下划线前缀",
+ "E501: 行超长,加 `# noqa: E501`",
+ "F811: 删重复import",
+ "运行 `ruff check --fix .` 自动修复大部分问题",
+ ],
+ "auto_fixable": True,
+ },
+ # ===== 单元测试失败 =====
+ {
+ "pattern": r"FAILED|assert.*Error|AssertionError",
+ "category": "unit_test",
+ "category_cn": "单元测试失败",
+ "severity": "high",
+ "suggestions": [
+ "检查相关测试文件,确认是代码问题还是测试用例问题",
+ "本地运行对应测试:`pytest path/to/test.py -v`",
+ "如测试依赖外部服务,检查mock是否正确",
+ ],
+ "auto_fixable": False,
+ },
+ {
+ "pattern": r"pytest.*failed|\d+ failed.*\d+ passed",
+ "category": "unit_test",
+ "category_cn": "单元测试失败",
+ "severity": "high",
+ "suggestions": [
+ "查看上方日志中的FAILED测试用例",
+ "检查失败断言的期望值 vs 实际值",
+ "新代码影响了现有测试行为,确认是预期内变更吗?",
+ ],
+ "auto_fixable": False,
+ },
+ # ===== Docker 构建失败 =====
+ {
+ "pattern": r"Dockerfile.*not found|docker build.*failed|ERROR: failed to solve",
+ "category": "docker_build",
+ "category_cn": "Docker构建失败",
+ "severity": "high",
+ "suggestions": [
+ "检查Dockerfile语法是否正确",
+ "检查引用的基础镜像是否存在",
+ "本地运行 `docker build -f path/to/Dockerfile .` 复现",
+ ],
+ "auto_fixable": False,
+ },
+ {
+ "pattern": r"manifest.*not found|no such image|image.*not found",
+ "category": "docker_build",
+ "category_cn": "镜像不存在",
+ "severity": "medium",
+ "suggestions": [
+ "检查基础镜像名称和tag是否正确",
+ "确认镜像仓库可访问,登录是否有效",
+ "如为新基础镜像,需先手动构建一次基础镜像",
+ ],
+ "auto_fixable": False,
+ },
+ {
+ "pattern": r"ETXTBSY|text file busy",
+ "category": "docker_build",
+ "category_cn": "文件锁冲突(ETXTBSY)",
+ "severity": "low",
+ "summary": "esbuild并发构建冲突,重试即可",
+ "suggestions": ["偶发问题,点击Rerun重新运行即可", "如频繁出现,检查是否有多个job并发写入同一文件"],
+ "auto_fixable": True,
+ },
+ # ===== 依赖安装失败 =====
+ {
+ "pattern": r"pip install.*error|Could not find a version|No matching distribution",
+ "category": "dependency",
+ "category_cn": "pip依赖安装失败",
+ "severity": "medium",
+ "suggestions": [
+ "检查requirements.txt中的版本号是否正确",
+ "如为新版本刚发布,可能源还没同步,稍后重试",
+ "检查网络连接,可尝试切换pip镜像源",
+ ],
+ "auto_fixable": False,
+ },
+ {
+ "pattern": r"npm.*ERR|npm install.*failed|E404|ECONNREFUSED.*npm",
+ "category": "dependency",
+ "category_cn": "npm依赖安装失败",
+ "severity": "medium",
+ "suggestions": [
+ "检查package.json中的版本号是否存在",
+ "网络问题:检查npm registry是否可访问",
+ "国内网络建议配置npmmirror镜像源",
+ ],
+ "auto_fixable": False,
+ },
+ {
+ "pattern": r"Connection refused|timed out|network.*unreachable",
+ "category": "network",
+ "category_cn": "网络问题",
+ "severity": "medium",
+ "summary": "网络连接失败,可能是源站问题或DNS问题",
+ "suggestions": [
+ "点击Rerun重试,网络问题通常是临时的",
+ "如持续失败,检查对应服务是否正常",
+ "检查Runner网络配置",
+ ],
+ "auto_fixable": True,
+ },
+ # ===== 超时 =====
+ {
+ "pattern": r"timeout|timed out|exceeded.*time limit|job.*cancelled.*timeout",
+ "category": "timeout",
+ "category_cn": "执行超时",
+ "severity": "medium",
+ "suggestions": [
+ "如首次出现:重试一次,可能是临时性能波动",
+ "频繁出现:检查构建是否变慢了,最近是否加了新依赖",
+ "可适当增加timeout-minutes配置",
+ ],
+ "auto_fixable": False,
+ },
+ # ===== 数据库/迁移 =====
+ {
+ "pattern": r"alembic.*error|migration.*failed|relation.*does not exist|column.*does not exist",
+ "category": "migration",
+ "category_cn": "数据库迁移失败",
+ "severity": "high",
+ "suggestions": [
+ "检查迁移脚本是否正确,down_revision是否对",
+ "确认数据库中是否有脏数据或残留表",
+ "迁移脚本合并冲突时,重新生成迁移文件",
+ ],
+ "auto_fixable": False,
+ },
+ # ===== 缓存问题 =====
+ {
+ "pattern": r"cache.*corrupt|cache.*invalid|snapshot.*not found|failed to compute cache key",
+ "category": "cache",
+ "category_cn": "缓存损坏",
+ "severity": "low",
+ "suggestions": ["构建系统会自动清理损坏缓存并重试,通常无需干预", "如持续失败,手动清理Runner上的缓存目录"],
+ "auto_fixable": True,
+ },
+ # ===== Checkout 失败 =====
+ {
+ "pattern": r"Could not resolve host|fatal:.*repository|SSL.*problem",
+ "category": "checkout",
+ "category_cn": "代码拉取失败",
+ "severity": "low",
+ "suggestions": ["临时网络问题,点击Rerun重试", "如持续失败,检查Gitea服务状态"],
+ "auto_fixable": True,
+ },
+]
+
+
+def analyze_log(log_text: str, job_name: str = "") -> FailureDiagnosis:
+ """分析日志,返回诊断结果"""
+
+ lines = log_text.strip().split("\n")
+
+ # 收集所有匹配的模式
+ matched = []
+ error_lines = []
+
+ for line in lines:
+ line_stripped = line.strip()
+ # 收集ERROR/FAILED/Failed等错误行(最多20行)
+ if re.search(r"(ERROR|FAILED|Error|error:|FAIL:|Traceback)", line_stripped):
+ if len(error_lines) < 20:
+ error_lines.append(line_stripped)
+
+ for pattern_info in FAILURE_PATTERNS:
+ if re.search(pattern_info["pattern"], line_stripped, re.IGNORECASE):
+ matched.append(pattern_info)
+ break # 一行只匹配一个模式
+
+ if not matched:
+ # 未识别的失败类型
+ return FailureDiagnosis(
+ category="unknown",
+ category_cn="未知错误",
+ severity="medium",
+ summary="未识别的失败类型,需要人工查看日志",
+ error_lines=error_lines[:10],
+ suggestions=[
+ "点击'查看失败日志'查看完整日志",
+ "如为偶发问题,可先重试一次",
+ "常见原因:环境问题、配置问题、新增逻辑引入的bug",
+ ],
+ auto_fixable=False,
+ )
+
+ # 选最严重、最具体的那个
+ severity_order = {"high": 3, "medium": 2, "low": 1}
+ matched.sort(key=lambda x: severity_order.get(x["severity"], 0), reverse=True)
+ best_match = matched[0]
+
+ # 生成摘要
+ if "summary" in best_match:
+ summary = best_match["summary"]
+ else:
+ summary = f"{best_match['category_cn']}检查失败"
+ if job_name:
+ summary = f"[{job_name}] {summary}"
+
+ # 从error_lines中过滤出与该分类相关的
+ relevant_errors = error_lines[:10]
+
+ return FailureDiagnosis(
+ category=best_match["category"],
+ category_cn=best_match["category_cn"],
+ severity=best_match["severity"],
+ summary=summary,
+ error_lines=relevant_errors,
+ suggestions=best_match["suggestions"],
+ auto_fixable=best_match.get("auto_fixable", False),
+ )
+
+
+def fetch_failed_job_log(run_id: str, job_id: str, token: str, repo: str) -> Optional[str]:
+ """从Gitea API获取失败job的日志"""
+ api_base = f"https://git.xiaoxiajianji.com/api/v1/repos/{repo}"
+
+ # 尝试获取job的日志
+ url = f"{api_base}/actions/runs/{run_id}/jobs/{job_id}/log"
+ req = urllib.request.Request(url)
+ req.add_header("Authorization", f"token {token}")
+
+ try:
+ with urllib.request.urlopen(req, timeout=15) as resp:
+ return resp.read().decode("utf-8", errors="replace")
+ except Exception as e:
+ print(f"获取日志失败: {e}", file=sys.stderr)
+ return None
+
+
+def format_diagnosis_markdown(d: FailureDiagnosis, job_name: str = "", run_url: str = "") -> str:
+ """将诊断结果格式化为飞书卡片markdown"""
+
+ severity_emoji = {"high": "🔴", "medium": "🟡", "low": "🟢"}
+ emoji = severity_emoji.get(d.severity, "⚪")
+
+ lines = []
+ lines.append(f"**分类**: {emoji} {d.category_cn}")
+ lines.append(f"**问题**: {d.summary}")
+
+ if d.error_lines:
+ lines.append("")
+ lines.append("**关键错误行**:")
+ for err in d.error_lines[:5]:
+ # 截断过长的行
+ if len(err) > 150:
+ err = err[:147] + "..."
+ lines.append(f" `{err}`")
+
+ lines.append("")
+ lines.append("**修复建议**:")
+ for i, s in enumerate(d.suggestions[:5], 1):
+ lines.append(f" {i}. {s}")
+
+ if d.auto_fixable:
+ lines.append("")
+ lines.append("💡 **可自动修复**:如格式问题,可尝试点击Rerun让auto-fix自动处理")
+
+ if run_url:
+ lines.append("")
+ lines.append(f"[查看完整日志]({run_url})")
+
+ return "\n".join(lines)
+
+
+def main():
+ job_name = os.environ.get("FAILED_JOB", "")
+ run_id = os.environ.get("GITHUB_RUN_ID", "")
+ repo = os.environ.get("GITHUB_REPOSITORY", "xiaoxia/xiaoxia-saas")
+ token = os.environ.get("GITHUB_TOKEN", "")
+
+ # 1. 尝试获取日志
+ log_text = ""
+
+ # 优先从环境变量或文件读取
+ log_file = os.environ.get("CI_LOG_FILE", "")
+ if log_file and os.path.exists(log_file):
+ with open(log_file) as f:
+ log_text = f.read()
+ elif run_id and token:
+ # 尝试从API获取(需要job_id,这里简化处理)
+ pass
+
+ # 如果没有日志,用job_name做粗略分类
+ if not log_text:
+ # 基于job名做初始判断
+ if any(k in job_name.lower() for k in ["validate", "lint", "quality"]):
+ d = FailureDiagnosis(
+ category="lint_general",
+ category_cn="代码质量检查",
+ severity="low",
+ summary=f"{job_name} 检查失败(日志不可用,基于job名初步诊断)",
+ suggestions=["点击查看日志获取具体错误信息", "格式类问题通常可自动修复"],
+ auto_fixable=True,
+ )
+ elif "build" in job_name.lower():
+ d = FailureDiagnosis(
+ category="build_general",
+ category_cn="构建失败",
+ severity="high",
+ summary=f"{job_name} 构建失败(日志不可用)",
+ suggestions=["点击查看日志获取具体构建错误", "常见原因:Dockerfile错误、依赖安装失败、网络问题"],
+ auto_fixable=False,
+ )
+ elif "test" in job_name.lower():
+ d = FailureDiagnosis(
+ category="test_general",
+ category_cn="测试失败",
+ severity="high",
+ summary=f"{job_name} 测试失败(日志不可用)",
+ suggestions=["点击查看日志获取具体失败的测试用例", "检查最近代码改动是否影响了测试"],
+ auto_fixable=False,
+ )
+ elif "deploy" in job_name.lower():
+ d = FailureDiagnosis(
+ category="deploy_general",
+ category_cn="部署失败",
+ severity="high",
+ summary=f"{job_name} 部署失败(日志不可用)",
+ suggestions=["检查目标服务器状态和网络", "检查镜像是否正确推送", "查看服务器上的容器日志"],
+ auto_fixable=False,
+ )
+ else:
+ d = FailureDiagnosis(
+ category="unknown",
+ category_cn="未知错误",
+ severity="medium",
+ summary=f"{job_name} 失败",
+ suggestions=["点击查看日志获取详细信息"],
+ auto_fixable=False,
+ )
+ else:
+ d = analyze_log(log_text, job_name)
+
+ # 输出诊断结果
+ run_url = f"https://git.xiaoxiajianji.com/{repo}/actions/runs/{run_id}" if run_id else ""
+
+ print("=" * 60)
+ print(" CI 失败诊断报告")
+ print("=" * 60)
+ print()
+ print(format_diagnosis_markdown(d, job_name, run_url))
+ print()
+ print("=" * 60)
+
+ # 将诊断结果写入文件(供通知脚本读取)
+ output_file = os.environ.get("DIAGNOSIS_OUTPUT", "/tmp/ci_diagnosis.json")
+ result = {
+ "category": d.category,
+ "category_cn": d.category_cn,
+ "severity": d.severity,
+ "summary": d.summary,
+ "error_lines": d.error_lines,
+ "suggestions": d.suggestions,
+ "auto_fixable": d.auto_fixable,
+ }
+ with open(output_file, "w") as f:
+ json.dump(result, f, ensure_ascii=False, indent=2)
+ print(f"\n诊断结果已保存到: {output_file}")
+
+
+if __name__ == "__main__":
+ main()
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()
diff --git a/scripts/ci/ci_health_report.py b/scripts/ci/ci_health_report.py
new file mode 100644
index 000000000..0ccf0e446
--- /dev/null
+++ b/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())
diff --git a/scripts/ci/ci_repeated_failure_detector.py b/scripts/ci/ci_repeated_failure_detector.py
new file mode 100644
index 000000000..b5dc95d34
--- /dev/null
+++ b/scripts/ci/ci_repeated_failure_detector.py
@@ -0,0 +1,417 @@
+#!/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.get("status", "")
+ 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", "")
+ 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("## 概览")
+ lines.append("")
+ 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)} |")
+ 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("=== 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()
diff --git a/scripts/ci/ci_trace_report.py b/scripts/ci/ci_trace_report.py
new file mode 100755
index 000000000..eb53f1422
--- /dev/null
+++ b/scripts/ci/ci_trace_report.py
@@ -0,0 +1,375 @@
+#!/usr/bin/env python3
+"""
+CI Trace Report Script - Reports CI Trace data to AgentLoop from Gitea Actions workflows.
+
+Usage in CI workflow jobs:
+ - At start: python3 scripts/ci/ci_trace_report.py --status running
+ - At end: python3 scripts/ci/ci_trace_report.py --status ok --start-time $CI_TRACE_START_TIME
+
+Environment variables (built-in Gitea Actions):
+ GITEA_REPOSITORY / GITHUB_REPOSITORY - repository (owner/repo)
+ GITEA_WORKFLOW / GITHUB_WORKFLOW - workflow name
+ GITEA_JOB / GITHUB_JOB - job ID
+ GITEA_SHA / GITHUB_SHA - commit SHA
+ GITEA_REF_NAME / GITHUB_REF_NAME - branch name
+ GITEA_RUN_ID / GITHUB_RUN_ID - run ID
+ GITEA_ACTOR / GITHUB_ACTOR - trigger actor
+ GITEA_EVENT_NAME / GITHUB_EVENT_NAME - event type
+ PR_NUMBER / GITEA_PR_NUMBER - PR number (if PR triggered)
+
+AgentLoop configuration (injected via Secrets):
+ AGENTLOOP_LICENSE_KEY - LicenseKey (required)
+ AGENTLOOP_ENDPOINT - Trace endpoint (optional, has default)
+ AGENTLOOP_PROJECT - SLS Project name (optional)
+ AGENTLOOP_WORKSPACE - CMS Workspace name (optional)
+"""
+
+import argparse
+import json
+import os
+import sys
+import time
+import urllib.error
+import urllib.request
+import uuid
+
+# ========== Default Configuration ==========
+DEFAULT_ENDPOINT = "https://proj-xtrace-495e81719a1fd9a2c5fd671eefafbe-cn-hangzhou.cn-hangzhou.log.aliyuncs.com/apm/trace/opentelemetry/v1/traces"
+DEFAULT_PROJECT = "proj-xtrace-495e81719a1fd9a2c5fd671eefafbe-cn-hangzhou"
+DEFAULT_WORKSPACE = "agentloop-13b8d6efb7fde6e9b193eb982ade68e2"
+
+
+# ========== OTLP Protobuf Manual Encoding ==========
+
+
+def _encode_varint(value):
+ result = bytearray()
+ while value > 0x7F:
+ result.append((value & 0x7F) | 0x80)
+ value >>= 7
+ result.append(value & 0x7F)
+ return bytes(result)
+
+
+def _encode_tag(field_number, wire_type):
+ return _encode_varint((field_number << 3) | wire_type)
+
+
+def _encode_string_field(field_number, value):
+ value_bytes = value.encode("utf-8")
+ return _encode_tag(field_number, 2) + _encode_varint(len(value_bytes)) + value_bytes
+
+
+def _encode_bytes_field(field_number, value_bytes):
+ return _encode_tag(field_number, 2) + _encode_varint(len(value_bytes)) + value_bytes
+
+
+def _encode_int_field(field_number, value):
+ return _encode_tag(field_number, 0) + _encode_varint(value & 0xFFFFFFFFFFFFFFFF)
+
+
+def _encode_message_field(field_number, message_bytes):
+ return _encode_tag(field_number, 2) + _encode_varint(len(message_bytes)) + message_bytes
+
+
+def _encode_key_value(key, value_str):
+ any_value = _encode_string_field(1, value_str)
+ return _encode_string_field(1, key) + _encode_message_field(2, any_value)
+
+
+def _encode_status(status_code, status_msg=""):
+ data = _encode_int_field(1, status_code)
+ if status_msg:
+ data += _encode_string_field(2, status_msg)
+ return data
+
+
+def _encode_span(
+ trace_id_bytes,
+ span_id_bytes,
+ parent_span_id_bytes,
+ name,
+ start_time_unix_nano,
+ end_time_unix_nano,
+ span_kind,
+ attributes,
+ status_code,
+ status_msg="",
+):
+ data = b""
+ data += _encode_bytes_field(1, trace_id_bytes)
+ data += _encode_bytes_field(2, span_id_bytes)
+ if parent_span_id_bytes:
+ data += _encode_bytes_field(3, parent_span_id_bytes)
+ data += _encode_string_field(4, name)
+ data += _encode_int_field(5, span_kind)
+ data += _encode_int_field(6, start_time_unix_nano)
+ data += _encode_int_field(7, end_time_unix_nano)
+ for key, value in attributes.items():
+ kv = _encode_key_value(key, str(value))
+ data += _encode_message_field(9, kv)
+ status = _encode_status(status_code, status_msg)
+ data += _encode_message_field(12, status)
+ return data
+
+
+def _encode_resource_spans(service_name, scope_spans_bytes):
+ svc_kv = _encode_key_value("service.name", service_name)
+ resource = _encode_message_field(1, svc_kv)
+ data = _encode_message_field(1, resource)
+ data += _encode_message_field(2, scope_spans_bytes)
+ return data
+
+
+def _encode_scope_spans(scope_name, spans_bytes_list):
+ scope = _encode_string_field(1, scope_name)
+ data = _encode_message_field(1, scope)
+ for span_bytes in spans_bytes_list:
+ data += _encode_message_field(2, span_bytes)
+ return data
+
+
+def _encode_traces_data(resource_spans_bytes_list):
+ data = b""
+ for rs_bytes in resource_spans_bytes_list:
+ data += _encode_message_field(1, rs_bytes)
+ return data
+
+
+# ========== Helper Functions ==========
+
+
+def _gen_trace_id():
+ return uuid.uuid4().bytes
+
+
+def _gen_span_id():
+ return uuid.uuid4().bytes[:8]
+
+
+def _env(name, default=""):
+ """Get env var with GITEA_/GITHUB_ prefix fallback."""
+ val = os.getenv(name, "")
+ if val:
+ return val
+ if name.startswith("GITEA_"):
+ alt = "GITHUB_" + name[6:]
+ return os.getenv(alt, default)
+ if name.startswith("GITHUB_"):
+ alt = "GITEA_" + name[7:]
+ return os.getenv(alt, default)
+ return default
+
+
+def _get_pr_number():
+ """Get PR number from environment or event file."""
+ pr = os.getenv("PR_NUMBER", "") or os.getenv("GITEA_PR_NUMBER", "")
+ if pr:
+ return pr
+
+ event_path = os.getenv("GITHUB_EVENT_PATH", "") or os.getenv("GITEA_EVENT_PATH", "")
+ if event_path and os.path.isfile(event_path):
+ try:
+ with open(event_path, "r") as f:
+ event = json.load(f)
+ if "pull_request" in event and "number" in event["pull_request"]:
+ return str(event["pull_request"]["number"])
+ except Exception:
+ pass
+
+ return ""
+
+
+def _get_ci_attributes():
+ """Collect attributes from CI environment variables."""
+ attrs = {
+ "ci.repo": _env("GITEA_REPOSITORY") or _env("GITHUB_REPOSITORY") or "unknown",
+ "ci.workflow": _env("GITEA_WORKFLOW") or _env("GITHUB_WORKFLOW") or "unknown",
+ "ci.job": _env("GITEA_JOB") or _env("GITHUB_JOB") or "unknown",
+ "ci.commit_sha": _env("GITEA_SHA") or _env("GITHUB_SHA") or "unknown",
+ "ci.branch": _env("GITEA_REF_NAME") or _env("GITHUB_REF_NAME") or "unknown",
+ "ci.run_id": _env("GITEA_RUN_ID") or _env("GITHUB_RUN_ID") or "unknown",
+ "ci.actor": _env("GITEA_ACTOR") or _env("GITHUB_ACTOR") or "unknown",
+ "ci.event": _env("GITEA_EVENT_NAME") or _env("GITHUB_EVENT_NAME") or "unknown",
+ }
+ pr = _get_pr_number()
+ if pr:
+ attrs["ci.pr_number"] = pr
+ return attrs
+
+
+# ========== Trace Building & Reporting ==========
+
+
+def build_trace(service_name, trace_name, status, duration_ms, attributes=None):
+ """Build an OTLP trace payload (protobuf bytes). No external dependencies."""
+ trace_id = _gen_trace_id()
+ end_time = int(time.time() * 1e9)
+ start_time = end_time - int(duration_ms * 1e6)
+ status_code = 1 if status in ("ok", "running") else 2
+ status_msg = "" if status in ("ok", "running") else "Job failed"
+
+ main_attrs = {
+ "agent.trace_name": trace_name,
+ "agent.service": service_name,
+ "ci.trace_status": status,
+ }
+ if attributes:
+ main_attrs.update(attributes)
+
+ main_span = _encode_span(
+ trace_id_bytes=trace_id,
+ span_id_bytes=_gen_span_id(),
+ parent_span_id_bytes=b"",
+ name=trace_name,
+ start_time_unix_nano=start_time,
+ end_time_unix_nano=end_time,
+ span_kind=1,
+ attributes=main_attrs,
+ status_code=status_code,
+ status_msg=status_msg,
+ )
+
+ scope_spans = _encode_scope_spans("ci-trace", [main_span])
+ resource_spans = _encode_resource_spans(service_name, scope_spans)
+ return _encode_traces_data([resource_spans])
+
+
+def report_ci_trace(
+ service_name,
+ trace_name,
+ status="ok",
+ duration_ms=1000,
+ endpoint=None,
+ license_key=None,
+ project=None,
+ workspace=None,
+ extra_attributes=None,
+):
+ """
+ Report CI Trace data. Returns (success: bool, message: str).
+ Never raises exceptions; returns False on failure.
+ """
+ try:
+ endpoint = endpoint or os.getenv("AGENTLOOP_ENDPOINT", DEFAULT_ENDPOINT)
+ license_key = license_key or os.getenv("AGENTLOOP_LICENSE_KEY", "")
+ project = project or os.getenv("AGENTLOOP_PROJECT", DEFAULT_PROJECT)
+ workspace = workspace or os.getenv("AGENTLOOP_WORKSPACE", DEFAULT_WORKSPACE)
+
+ if not license_key:
+ return False, "[Trace] skipped: AGENTLOOP_LICENSE_KEY not configured"
+
+ attrs = _get_ci_attributes()
+ if extra_attributes:
+ attrs.update(extra_attributes)
+
+ payload = build_trace(
+ service_name=service_name,
+ trace_name=trace_name,
+ status=status,
+ duration_ms=duration_ms,
+ attributes=attrs,
+ )
+
+ headers = {
+ "Content-Type": "application/x-protobuf",
+ "x-arms-license-key": license_key,
+ "x-arms-project": project,
+ "x-cms-workspace": workspace,
+ }
+
+ req = urllib.request.Request(endpoint, data=payload, headers=headers, method="POST")
+ try:
+ with urllib.request.urlopen(req, timeout=10) as resp:
+ status_code = resp.status
+ resp_body = resp.read().decode("utf-8", errors="replace")
+ except urllib.error.HTTPError as e:
+ status_code = e.code
+ resp_body = e.read().decode("utf-8", errors="replace")
+
+ if status_code in (200, 202):
+ return True, (f"[Trace] success: {service_name} / {trace_name} " f"({status}, {duration_ms}ms)")
+ else:
+ return False, (f"[Trace] failed: HTTP {status_code} - {resp_body[:200]}")
+ except Exception as e:
+ return False, f"[Trace] error: {type(e).__name__}: {str(e)}"
+
+
+def main():
+ parser = argparse.ArgumentParser(description="CI AgentLoop Trace Reporter")
+ parser.add_argument(
+ "--service",
+ dest="service_name",
+ default=os.getenv("TRACE_SERVICE", ""),
+ help="Service name (also via TRACE_SERVICE env)",
+ )
+ parser.add_argument(
+ "--name",
+ dest="trace_name",
+ default=os.getenv("TRACE_NAME", ""),
+ help="Trace name (also via TRACE_NAME env)",
+ )
+ parser.add_argument(
+ "--status",
+ default=os.getenv("TRACE_STATUS", "ok"),
+ choices=["ok", "error", "running"],
+ help="Status: ok / error / running (default ok)",
+ )
+ parser.add_argument(
+ "--start-time",
+ dest="start_time",
+ default=os.getenv("TRACE_START_TIME", ""),
+ help="Start timestamp (seconds) for duration calculation",
+ )
+ parser.add_argument(
+ "--duration-ms",
+ dest="duration_ms",
+ type=int,
+ default=0,
+ help="Direct duration in ms; takes precedence over --start-time",
+ )
+ parser.add_argument("--attrs", default="", help="Extra attributes (JSON string)")
+
+ args = parser.parse_args()
+
+ if not args.service_name:
+ print("[Trace] skipped: no service specified (--service or TRACE_SERVICE)")
+ sys.exit(0)
+
+ duration_ms = args.duration_ms
+ if duration_ms <= 0 and args.start_time:
+ try:
+ start_ts = float(args.start_time)
+ duration_ms = int((time.time() - start_ts) * 1000)
+ except (ValueError, TypeError):
+ duration_ms = 1000
+ if duration_ms <= 0:
+ duration_ms = 1000
+
+ extra_attrs = {}
+ if args.attrs:
+ try:
+ extra_attrs = json.loads(args.attrs)
+ except json.JSONDecodeError:
+ pass
+
+ trace_name = args.trace_name
+ if not trace_name:
+ wf = _env("GITEA_WORKFLOW") or _env("GITHUB_WORKFLOW") or "CI"
+ job = _env("GITEA_JOB") or _env("GITHUB_JOB") or "job"
+ trace_name = f"{wf} / {job}"
+
+ success, msg = report_ci_trace(
+ service_name=args.service_name,
+ trace_name=trace_name,
+ status=args.status,
+ duration_ms=duration_ms,
+ extra_attributes=extra_attrs,
+ )
+
+ print(msg)
+ sys.exit(0)
+
+
+if __name__ == "__main__":
+ main()
diff --git a/scripts/ci/docker_build_only.sh b/scripts/ci/docker_build_only.sh
new file mode 100755
index 000000000..d0aaa18d7
--- /dev/null
+++ b/scripts/ci/docker_build_only.sh
@@ -0,0 +1,43 @@
+#!/bin/bash
+# PR构建专用:只构建不输出,验证Dockerfile能否正常构建
+# 无本地缓存(12个runner不共享,反而添乱),只用ACR远程缓存
+set -eu
+
+NO_CACHE_FLAG=""
+if [ "$1" = "--no-cache" ]; then
+ NO_CACHE_FLAG="--no-cache"
+ shift
+fi
+
+DOCKERFILE="$1"
+IMAGE_TAG="$2"
+CACHE_REF="$3"
+shift 3
+BUILD_ARGS=""
+for arg in "$@"; do
+ BUILD_ARGS="$BUILD_ARGS --build-arg $arg"
+done
+
+BUILDER_NAME="ci-pr-builder-${GITHUB_RUN_ID:-local}"
+if ! docker buildx inspect "$BUILDER_NAME" > /dev/null 2>&1; then
+ docker buildx create --use --name "$BUILDER_NAME" --driver docker-container
+else
+ docker buildx use "$BUILDER_NAME"
+fi
+docker buildx inspect --bootstrap
+
+echo "=== PR Build: build only, no output, remote cache only ==="
+echo "Dockerfile: ${DOCKERFILE}"
+echo "Image tag: ${IMAGE_TAG}"
+echo ""
+
+docker buildx build \
+ $NO_CACHE_FLAG \
+ $BUILD_ARGS \
+ --cache-from "type=registry,ref=${CACHE_REF}" \
+ -f "${DOCKERFILE}" \
+ -t "${IMAGE_TAG}" \
+ .
+
+echo ""
+echo "PR build OK (build only, no output): ${IMAGE_TAG}"
diff --git a/scripts/ci/docker_build_push.sh b/scripts/ci/docker_build_push.sh
new file mode 100755
index 000000000..f3ba4bb80
--- /dev/null
+++ b/scripts/ci/docker_build_push.sh
@@ -0,0 +1,106 @@
+#!/bin/bash
+# 通用Docker镜像构建+推送脚本(local cache为主 + registry cache共享)
+# 用法: docker_build_push.sh [--no-cache]