#!/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()