xiaoxia-saas/scripts/ci/auto_fix_formatting.py

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

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

import json
import os
import subprocess
import sys
import urllib.request


def run(cmd, check=True, capture=True):
    """运行shell命令"""
    result = subprocess.run(cmd, shell=True, capture_output=capture, text=True)
    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_py_files(pr_number, api_url, token):
    """获取PR中变更的Python文件列表"""
    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["filename"].endswith(".py") and 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 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_py_files = os.environ.get("CHANGED_PY_FILES", "")

    if not token:
        print("缺少GITHUB_TOKEN，无法推送修复", file=sys.stderr)
        sys.exit(1)

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

    # 确定要修复的文件范围
    if scan_mode == "incremental" and changed_py_files:
        target_files = changed_py_files.split()
        print(f"增量模式，修复 {len(target_files)} 个变更的Python文件")
    else:
        target_files = ["alembic", "apps", "packages", "tests", "scripts"]
        print("全量模式，修复所有Python文件")

    # 自动修复
    target_str = " ".join(target_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)

    # 检查是否有改动
    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 [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()