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 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 -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 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}")

    # 提交修复
    run("git add -A")
    run('git commit -m "style: auto-format with black + isort + prettier [ci skip]"')

    # 推送（head_branch已从ensure_git_repo获取）
    print(f"\nPR来源分支: {head_branch}")

    run(f'git push origin "HEAD:{head_branch}"')

    print()
    print("✅ 格式已自动修复并推送回分支")
    print("新的commit会重新触发CI检查")


if __name__ == "__main__":
    main()