xiaoxia-saas/scripts/ci/auto_fix_formatting.py

#!/usr/bin/env python3
"""CI中自动修复代码格式（Python: black + isort | Frontend: prettier），并推送回PR分支。

只在PR事件中执行，避免直接修改主干。
当code quality检查因格式问题失败时触发，自动修复后push回原分支。
"""

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仓库，保持在 merge commit 状态。

    checkout脚本用tarball方式下载代码（PR merge后的commit），没有.git目录。
    这里自动初始化git仓库，但不切换分支，保持在merge commit状态，
    使格式化检查与 Frontend Lint 的范围完全一致。
    """
    if os.path.exists(".git"):
        return

    print("检测到tarball checkout（无.git目录），自动初始化git仓库...")
    print("保持在 merge commit 状态（与 Frontend Lint 一致）")

    # 构造带认证的远端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的源分支（仅用于后续推送，不checkout）
    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（不fetch/checkout，保持tarball内容即merge commit状态）
    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"')

    # 将当前目录（tarball解压的merge commit）作为初始commit
    run("git add -A")
    run('git commit -q -m "CI merge commit snapshot"')

    result = run("git status --porcelain")
    if result.stdout.strip():
        n = len(result.stdout.strip().splitlines())
        print(f"⚠️ 工作区有 {n} 个未追踪文件")
    else:
        print("✅ git仓库就绪（merge commit 状态）")

    return head_branch


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():
    # 只在PR事件中执行
    event_name = os.environ.get("GITHUB_EVENT_NAME", "")
    if event_name != "pull_request":
        print("非PR事件，跳过自动修复")
        return

    github_ref = os.environ.get("GITHUB_REF", "")
    pr_number = github_ref.split("/")[2] if github_ref.startswith("refs/pull/") else ""
    if not pr_number:
        print("无法获取PR号，跳过自动修复")
        return

    api_url = os.environ.get("GITHUB_API_URL", "")
    repo = os.environ.get("GITHUB_REPOSITORY", "")
    token = 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("缺少GITHUB_TOKEN，无法推送修复", file=sys.stderr)
        sys.exit(1)

    repo_root = os.getcwd()

    # 确保git仓库可用（tarball checkout模式下自动初始化）
    # 返回PR源分支名，供后续推送使用
    head_branch = ensure_git_repo(api_url, repo, token, pr_number)

    print("=== 检测到代码格式问题，尝试自动修复 ===")
    print(f"PR #{pr_number}")
    print(f"扫描模式: {scan_mode}")

    # 前端文件扩展名
    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}")

    # 提交修复（在 merge commit 状态）
    run("git add -A")
    run('git commit -m "style: auto-format with black + isort + prettier"')

    # 记录格式修复的 commit hash
    format_commit = run("git rev-parse HEAD").stdout.strip()

    # 推送：将格式化变更推回 PR 源分支
    print(f"\nPR来源分支: {head_branch}")

    # 切换到源分支，cherry-pick 格式化修复
    print(f"fetch 源分支 {head_branch}...")
    run(f"git fetch --depth=1 origin {head_branch}")
    run(f"git checkout -f -B {head_branch} FETCH_HEAD")

    print(f"cherry-pick 格式化修复到源分支...")
    cherry_result = run(f"git cherry-pick {format_commit}", check=False)
    if cherry_result.returncode != 0:
        print(f"cherry-pick 冲突（源分支可能已有不同格式），中止cherry-pick")
        run("git cherry-pick --abort", check=False)
        print("没有需要推送的格式化改动")
        return

    # 推送（保留原有的重试逻辑）
    print("推送格式修复到远端...")

    max_retries = 3
    push_success = False
    last_error = ""

    for attempt in range(1, max_retries + 1):
        print(f"  尝试 {attempt}/{max_retries}: 推送...")

        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)
            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()