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仓库，并切换到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 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", "")

    # 获取PR作者信息，判断是人还是Agent提交的
    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}")

    # 判断是否为Agent提交的PR
    # Agent账号：actions, auto-approve-bot 等bot用户
    # 人提交的PR（如xiaoxia）：只诊断不自动修
    agent_authors = {"actions", "auto-approve-bot", "gitea-actions"}
    is_agent_pr = pr_author in agent_authors or "bot" in pr_author.lower()

    if is_agent_pr:
        print(f"检测到Agent提交的PR（作者: {pr_author}），将自动修复并推送")
        fix_mode = "auto_fix_and_push"
    else:
        print(f"检测到人提交的PR（作者: {pr_author}），仅诊断不自动修改")
        print("（如需自动修复，请用Agent账号提交PR，或手动运行格式化脚本）")
        fix_mode = "diagnose_only"
    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

    # 诊断模式：只报告问题，不修改不推送
    if fix_mode == "diagnose_only":
        print()
        print("=" * 50)
        print("📋 格式问题诊断报告（人提交的PR，仅诊断不自动修复）")
        print("=" * 50)
        print()
        print("以下文件存在格式问题，建议手动修复：")
        for line in result.stdout.strip().split("\n"):
            print(f"  {line}")
        print()
        print("修复方式：")
        print("  后端(Python): 运行 black + isort")
        print("  前端: 运行 prettier --write")
        print("  或使用 scripts/agent-commit.sh 提交（自动格式化）")
        print()
        print("=" * 50)
        # 以非0状态码退出，让CI继续报失败（因为问题没修）
        sys.exit(1)

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