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 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":
        # 增量模式：只格式化变更的前端文件
        # 转换为相对于 apps/web 的路径或用绝对路径
        target_str = " ".join(target_fe_files)
        # 从项目根目录运行，prettier 会找配置文件
        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()

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

    # 前端文件扩展名
    fe_extensions = (
        ".ts", ".tsx", ".js", ".jsx",
        ".css", ".scss", ".less",
        ".json", ".html", ".md",
        ".yaml", ".yml",
    )
    # Python 文件扩展名
    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"]
        # 全量模式下 prettier 在前端目录内部运行，无需传文件列表
        target_fe_files = ["apps/web"]  # 标记为有前端文件需要处理
        print("全量模式，修复所有文件")

    # Python 格式化
    fix_python(target_py_files, scan_mode)

    # 前端格式化
    # 全量模式下直接传 web 目录标记
    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}")

    # 配置git
    run('git config user.name "CI Bot"')
    run('git config user.email "ci-bot@xiaoxiajianji.com"')

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

    # 获取来源分支并推送
    head_branch = get_pr_head_branch(pr_number, f"{api_url}/repos/{repo}", token)
    print(f"\nPR来源分支: {head_branch}")

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

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


if __name__ == "__main__":
    main()