#!/usr/bin/env python3 """CI中自动修复代码格式（Python: black + isort | Frontend: prettier），并推送回原分支。 - PR事件：所有PR只要Code Quality因格式问题失败，自动修复并push回源分支 - Push事件（develop/main）：自动修复并push回原分支，保持主干格式永远正确 - 防循环：修复commit带 [skip ci-format-check] 标记，检测到该标记则跳过修复 - 只修格式（black/isort/prettier），ruff逻辑类错误不动 当code quality检查因格式问题失败时触发。 """ import json import os import subprocess import sys import time import urllib.request def run(cmd, check=True, capture=True, cwd=None): """运行shell命令""" result = subprocess.run(cmd, shell=True, capture_output=capture, text=True, cwd=cwd) if check and result.returncode != 0: print(f"命令失败: {cmd}", file=sys.stderr) if result.stderr: print(result.stderr, file=sys.stderr) sys.exit(1) return result def ensure_git_repo(api_url, repo, token, pr_number): """确保当前目录是git仓库，并切换到PR源分支。 checkout脚本用tarball方式下载代码（PR merge后的commit），没有.git目录。 这里自动初始化git仓库，fetch PR源分支并强制checkout， 使工作区变为PR源分支的代码，确保后续格式化修复基于源分支。 """ if os.path.exists(".git"): return print("检测到tarball checkout（无.git目录），自动初始化git仓库...") # 构造带认证的远端URL server_url = api_url.rsplit("/api/v1", 1)[0] remote_url = f"{server_url.replace('https://', f'https://x-access-token:{token}@')}/{repo}.git" # 获取PR的源分支 pr_api_url = f"{api_url}/repos/{repo}/pulls/{pr_number}" req_obj = urllib.request.Request(pr_api_url, headers={"Authorization": f"token {token}"}) with urllib.request.urlopen(req_obj) as resp: pr = json.loads(resp.read()) head_branch = pr["head"]["ref"] print(f"PR源分支: {head_branch}") # 初始化git run("git init -q") run(f"git remote add origin {remote_url}") run('git config user.name "CI Bot"') run('git config user.email "ci-bot@xiaoxiajianji.com"') # fetch源分支（浅克隆，只要最新commit） print("fetch源分支...") run(f"git fetch --depth=1 origin {head_branch}") # 强制checkout到源分支（覆盖tarball内容） # tarball是merge后的commit，源分支才是我们要修改并推送的目标 print("切换到源分支...") run(f"git checkout -f -B {head_branch} FETCH_HEAD") result = run("git status --porcelain") if result.stdout.strip(): n = len(result.stdout.strip().splitlines()) print(f"⚠️ 工作区有 {n} 个未追踪文件") else: print("✅ git仓库就绪，工作区clean") return head_branch def ensure_git_repo_for_push(api_url, repo, token, branch_name): """push事件下确保git仓库可用，并切换到目标分支。 checkout脚本用tarball方式下载代码，没有.git目录。 这里自动初始化git仓库，fetch目标分支并checkout。 """ if os.path.exists(".git"): # 已有git，确认在正确分支 result = run("git rev-parse --abbrev-ref HEAD", check=False) if result.stdout.strip() == branch_name: return # 不在目标分支，切换 run(f"git checkout {branch_name}", check=False) return print(f"检测到tarball checkout（无.git目录），初始化git仓库（push模式，分支: {branch_name}）...") server_url = api_url.rsplit("/api/v1", 1)[0] remote_url = f"{server_url.replace('https://', f'https://x-access-token:{token}@')}/{repo}.git" run("git init -q") run(f"git remote add origin {remote_url}") run('git config user.name "CI Bot"') run('git config user.email "ci-bot@xiaoxiajianji.com"') print(f"fetch {branch_name} 分支...") run(f"git fetch --depth=1 origin {branch_name}") print(f"切换到 {branch_name} 分支...") run(f"git checkout -f -B {branch_name} FETCH_HEAD") result = run("git status --porcelain") if result.stdout.strip(): n = len(result.stdout.strip().splitlines()) print(f"⚠️ 工作区有 {n} 个未追踪文件") else: print("✅ git仓库就绪，工作区clean") def get_changed_files(pr_number, api_url, token): """获取PR中变更的文件列表""" url = f"{api_url}/pulls/{pr_number}/files?limit=100" req = urllib.request.Request(url, headers={"Authorization": f"token {token}"}) with urllib.request.urlopen(req) as resp: files = json.loads(resp.read()) return [f["filename"] for f in files if f["status"] != "removed"] def get_pr_head_branch(pr_number, api_url, token): """获取PR的来源分支名""" url = f"{api_url}/pulls/{pr_number}" req = urllib.request.Request(url, headers={"Authorization": f"token {token}"}) with urllib.request.urlopen(req) as resp: pr = json.loads(resp.read()) return pr["head"]["ref"] def fix_python(target_py_files, scan_mode): """修复 Python 文件格式 (black + isort)""" if not target_py_files: print("没有需要修复的 Python 文件，跳过") return target_str = " ".join(target_py_files) print() print("--- black 格式化 ---") result = run(f"python3 -m black {target_str}", check=False) print(result.stdout[-500:] if result.stdout else "") if result.returncode != 0: print("black执行失败，但继续尝试isort", file=sys.stderr) print() print("--- isort 排序 ---") result = run(f"python3 -m isort {target_str}", check=False) print(result.stdout[-500:] if result.stdout else "") if result.returncode != 0: print("isort执行失败", file=sys.stderr) def fix_frontend(target_fe_files, scan_mode, repo_root): """修复前端文件格式 (prettier)""" if not target_fe_files: print("没有需要修复的前端文件，跳过") return # 检查 prettier 是否可用 web_dir = os.path.join(repo_root, "apps", "web") prettier_bin = os.path.join(web_dir, "node_modules", ".bin", "prettier") if not os.path.exists(prettier_bin): print() print("--- 安装前端依赖 (prettier) ---") result = run("npm install --no-audit --no-fund --prefer-offline", check=False, cwd=web_dir) if result.returncode != 0: print("npm install 失败，跳过 prettier 修复", file=sys.stderr) return print("依赖安装完成") if not os.path.exists(prettier_bin): print("prettier 仍不可用，跳过", file=sys.stderr) return print() print("--- prettier 格式化 ---") if scan_mode == "incremental": # 增量模式：只格式化变更的前端文件 target_str = " ".join(target_fe_files) cmd = f"{prettier_bin} --write {target_str}" else: # 全量模式：格式化整个前端目录 cmd = f"{prettier_bin} --write ." result = run(cmd, check=False, cwd=web_dir if scan_mode != "incremental" else repo_root) print(result.stdout[-800:] if result.stdout else "") if result.stderr: print(result.stderr[-500:], file=sys.stderr) def main(): event_name = os.environ.get("GITHUB_EVENT_NAME", "") github_ref = os.environ.get("GITHUB_REF", "") api_url = os.environ.get("GITHUB_API_URL", "") repo = os.environ.get("GITHUB_REPOSITORY", "") token = os.environ.get("REVIEW_TOKEN", "") or os.environ.get("GITHUB_TOKEN", "") scan_mode = os.environ.get("SCAN_MODE", "full") changed_files_env = os.environ.get("CHANGED_FILES", "") if not token: print("缺少REVIEW_TOKEN或GITHUB_TOKEN，无法推送修复", file=sys.stderr) sys.exit(1) repo_root = os.getcwd() # ====== Push事件处理（develop/main等受保护分支） ====== if event_name == "push": # 从 refs/heads/xxx 提取分支名 if not github_ref.startswith("refs/heads/"): print(f"push事件但refs格式异常: {github_ref}，跳过") return branch_name = github_ref.replace("refs/heads/", "") # 只在受保护分支（develop/main）上自动修复并推送 protected_branches = {"develop", "main", "master"} if branch_name not in protected_branches: print(f"push事件，分支 {branch_name} 不是受保护分支，跳过自动修复") return print("=== Push事件：检测到格式问题，自动修复并推送回原分支 ===") print(f"分支: {branch_name}") print(f"扫描模式: {scan_mode}") # 初始化git仓库 ensure_git_repo_for_push(api_url, repo, token, branch_name) head_branch = branch_name fix_mode = "auto_fix_and_push" # ====== PR事件处理 ====== elif event_name == "pull_request": pr_number = github_ref.split("/")[2] if github_ref.startswith("refs/pull/") else "" if not pr_number: print("无法获取PR号，跳过自动修复") return # 获取PR信息 pr_info_url = f"{api_url}/repos/{repo}/pulls/{pr_number}" req_pr = urllib.request.Request(pr_info_url, headers={"Authorization": f"token {token}"}) with urllib.request.urlopen(req_pr) as resp: pr_info = json.loads(resp.read()) pr_author = pr_info.get("user", {}).get("login", "") print(f"PR作者: {pr_author}") # 防循环检测：检查最新commit是否已经是格式修复commit # 修复commit message 带 [skip ci-format-check] 标记，检测到则跳过 head_branch_tmp = pr_info.get("head", {}).get("ref", "") skip_marker = "[skip ci-format-check]" try: commits_url = f"{api_url}/repos/{repo}/pulls/{pr_number}/commits?limit=3" req_commits = urllib.request.Request(commits_url, headers={"Authorization": f"token {token}"}) with urllib.request.urlopen(req_commits) as resp_commits: commits = json.loads(resp_commits.read()) latest_msg = commits[0].get("commit", {}).get("message", "") if commits else "" if skip_marker in latest_msg: print(f"检测到最新commit包含 {skip_marker} 标记，跳过格式修复（防循环）") print("本次格式检查失败是格式修复commit触发的CI回跑，属正常现象") sys.exit(0) except Exception as e: print(f"⚠️ 防循环检测失败，继续执行: {e}") # 所有PR都自动修复格式（不再区分人/Agent） print("检测到格式问题，将自动修复并推送回分支") fix_mode = "auto_fix_and_push" print("=== 检测到代码格式问题，尝试自动修复 ===") print(f"PR #{pr_number}") print(f"扫描模式: {scan_mode}") # 确保git仓库可用（tarball checkout模式下自动初始化） head_branch = ensure_git_repo(api_url, repo, token, pr_number) # ====== 其他事件跳过 ====== else: print(f"事件 {event_name} 不支持自动修复，跳过") return # 前端文件扩展名 fe_extensions = ( ".ts", ".tsx", ".js", ".jsx", ".css", ".scss", ".less", ".json", ".html", ".md", ".yaml", ".yml", ) py_extensions = (".py",) # 确定要修复的文件范围 if scan_mode == "incremental" and changed_files_env: all_changed = changed_files_env.split() target_py_files = [f for f in all_changed if f.endswith(py_extensions)] target_fe_files = [f for f in all_changed if f.endswith(fe_extensions)] print(f"增量模式: {len(target_py_files)} 个Python文件, {len(target_fe_files)} 个前端文件") else: target_py_files = ["alembic", "apps", "packages", "tests", "scripts"] target_fe_files = ["apps/web"] print("全量模式，修复所有文件") # Python 格式化 fix_python(target_py_files, scan_mode) # 前端格式化 if scan_mode != "incremental": fix_frontend(["apps/web"], scan_mode, repo_root) else: fix_frontend(target_fe_files, scan_mode, repo_root) # 检查是否有改动 result = run("git status --porcelain") if not result.stdout.strip(): print() print("没有需要提交的格式改动") return print() print("变更文件:") for line in result.stdout.strip().split("\n"): print(f" {line}") # 提交修复 run("git add -A") run('git commit -m "style: auto-format with black + isort + prettier [skip ci-format-check]"') # 推送（head_branch已从ensure_git_repo获取） print(f"\nPR来源分支: {head_branch}") print("推送格式修复到远端...") # 推送前先 rebase 拉取远端最新，避免快进冲突 # 最多重试 3 次：rebase → push，失败则重新拉取再试 max_retries = 3 push_success = False last_error = "" for attempt in range(1, max_retries + 1): print(f" 尝试 {attempt}/{max_retries}: 拉取最新代码并推送...") # 先拉取远端最新 commit 并 rebase fetch_result = run(f"git fetch origin {head_branch}", check=False) if fetch_result.returncode != 0: last_error = f"git fetch 失败: {fetch_result.stderr.strip()}" print(f" {last_error}") time.sleep(2) continue rebase_result = run(f"git rebase origin/{head_branch}", check=False) if rebase_result.returncode != 0: last_error = f"git rebase 失败，中止并重置: {rebase_result.stderr.strip()[:200]}" print(f" {last_error}") run("git rebase --abort", check=False) # rebase 失败通常是冲突，重试没用，直接跳出 break # 推送 push_result = run(f'git push origin "HEAD:{head_branch}"', check=False) if push_result.returncode == 0: push_success = True break last_error = push_result.stderr.strip() or push_result.stdout.strip() print(f" push 失败: {last_error[:200]}") time.sleep(3) if not push_success: print(f"\n❌ 推送失败（已重试 {max_retries} 次）", file=sys.stderr) print(f"最后错误: {last_error}", file=sys.stderr) sys.exit(1) print() print("✅ 格式已自动修复并推送回分支") print("新的commit会重新触发CI检查") if __name__ == "__main__": main()