xiaoxia-saas/scripts/ci/pr_auto_scan.py

#!/usr/bin/env python3
"""
PR自动扫描器：扫描所有open PR，对CI全绿的进行自动审批/合并
作为短作业模式的兜底机制，每5分钟运行一次
"""

import argparse
import json
import os
import sys
import time
import urllib.error
import urllib.request


def api_request(token, repo, endpoint, method="GET", data=None):
    """Gitea API请求"""
    url = f"https://git.xiaoxiajianji.com/api/v1/repos/{repo}/{endpoint}"
    headers = {"Authorization": f"token {token}", "Content-Type": "application/json"}
    body = json.dumps(data).encode() if data else None
    req = urllib.request.Request(url, data=body, headers=headers, method=method)

    # 跳过SSL验证
    import ssl

    ctx = ssl.create_default_context()
    ctx.check_hostname = False
    ctx.verify_mode = ssl.CERT_NONE

    try:
        resp = urllib.request.urlopen(req, context=ctx)
        return json.loads(resp.read().decode()), resp.status
    except urllib.error.HTTPError as e:
        return json.loads(e.read().decode()) if e.read() else {"error": str(e)}, e.code


def get_open_prs(token, repo, base="develop"):
    """获取所有open的PR"""
    prs = []
    page = 1
    while True:
        data, code = api_request(token, repo, f"pulls?state=open&base={base}&sort=recentupdate&per_page=50&page={page}")
        if code != 200 or not isinstance(data, list) or len(data) == 0:
            break
        prs.extend(data)
        if len(data) < 50:
            break
        page += 1
    return prs


def get_commit_status(token, repo, sha):
    """获取commit的CI状态汇总"""
    data, code = api_request(token, repo, f"commits/{sha}/status")
    if code != 200:
        return {}, "error"
    return data, data.get("state", "unknown")


def check_required_contexts(token, repo, sha, contexts):
    """检查指定的context是否都通过"""
    data, _ = get_commit_status(token, repo, sha)
    statuses = {s["context"]: s["status"] for s in data.get("statuses", [])}

    all_success = True
    any_pending = False
    any_failed = False

    for ctx in contexts:
        state = statuses.get(ctx, "pending")
        if state != "success":
            all_success = False
        if state == "pending":
            any_pending = True
        if state in ("failure", "error"):
            any_failed = True

    return all_success, any_pending, any_failed, statuses


def get_pr_files(token, repo, pr_number):
    """获取PR变更文件"""
    files = []
    page = 1
    while True:
        data, code = api_request(token, repo, f"pulls/{pr_number}/files?per_page=300&page={page}")
        if code != 200 or not isinstance(data, list) or len(data) == 0:
            break
        files.extend(data)
        if len(data) < 300:
            break
        page += 1
    return [f["filename"] for f in files]


def is_frontend_only(files):
    """判断是否纯前端改动"""
    if not files:
        return False
    frontend_count = sum(1 for f in files if f.startswith("apps/web/"))
    backend_count = len(files) - frontend_count
    return backend_count == 0 and frontend_count > 0


def has_approval(token, repo, pr_number):
    """检查PR是否已有审批"""
    reviews, code = api_request(token, repo, f"pulls/{pr_number}/reviews")
    if code != 200:
        return False
    return any(r.get("state") == "APPROVED" for r in reviews if isinstance(r, dict))


def approve_pr(token, repo, pr_number):
    """审批PR"""
    # 创建review
    data, code = api_request(
        token,
        repo,
        f"pulls/{pr_number}/reviews",
        method="POST",
        data={"event": "PENDING", "body": "CI全绿，自动审批通过。"},
    )

    if code not in (200, 201):
        return False, f"创建review失败: HTTP {code}"

    review_id = data.get("id")
    if data.get("state") == "APPROVED":
        return True, "直接创建APPROVED成功"

    if not review_id:
        return False, "未获取到review ID"

    # submit为APPROVED
    data2, code2 = api_request(
        token,
        repo,
        f"pulls/{pr_number}/reviews/{review_id}/events",
        method="POST",
        data={"event": "APPROVED", "body": "CI全绿，自动审批通过。"},
    )

    if code2 in (200, 201):
        return True, "审批提交成功"
    else:
        # 尝试另一个端点
        data3, code3 = api_request(
            token,
            repo,
            f"pulls/{pr_number}/reviews/{review_id}",
            method="POST",
            data={"event": "APPROVED", "body": "CI全绿，自动审批通过。"},
        )
        if code3 in (200, 201):
            return True, "审批提交成功(备用端点)"
        return False, f"审批提交失败: HTTP {code2}/{code3}"


def merge_pr(token, repo, pr_number):
    """合并PR（squash merge）"""
    # 等待几秒让状态同步
    time.sleep(30)

    # 检查PR状态
    pr_data, code = api_request(token, repo, f"pulls/{pr_number}")
    if code != 200:
        return False, f"获取PR状态失败: HTTP {code}"
    if pr_data.get("state") != "open":
        return False, f"PR状态不是open: {pr_data.get('state')}"

    # 执行squash merge
    data, code = api_request(
        token,
        repo,
        f"pulls/{pr_number}/merge",
        method="POST",
        data={
            "do": "squash",
            "merge_title_field": "",
            "merge_message_field": "",
            "delete_branch_after_merge": True,
            "force_merge": False,
        },
    )

    if code == 200:
        return True, "合并成功"
    elif code == 405:
        return False, "合并返回405（门禁未满足或冲突）"
    else:
        return False, f"合并失败: HTTP {code}"


def main():
    parser = argparse.ArgumentParser(description="PR自动扫描器")
    parser.add_argument("--token", required=True, help="Gitea API token")
    parser.add_argument("--repo", default="xiaoxia/xiaoxia-saas", help="仓库")
    parser.add_argument("--base", default="develop", help="目标分支")
    parser.add_argument("--approve", action="store_true", help="执行自动审批")
    parser.add_argument("--merge", action="store_true", help="执行自动合并")
    parser.add_argument("--dry-run", default="false", help="试运行模式")
    parser.add_argument("--max-prs", type=int, default=20, help="最多处理的PR数")

    args = parser.parse_args()

    dry_run = args.dry_run.lower() == "true"

    # required contexts（与分支保护一致）
    REQUIRED_CONTEXTS_FULL = [
        "CI/CD Pipeline / Validate - Code Quality (pull_request)",
        "CI/CD Pipeline / Validate - Type Check (mypy) (pull_request)",
        "CI/CD Pipeline / Validate - Migration (alembic) (pull_request)",
        "CI/CD Pipeline / Frontend Lint (pull_request)",
        "CI/CD Pipeline / PR Build API Image (pull_request)",
        "CI/CD Pipeline / PR Build Worker Image (pull_request)",
        "CI/CD Pipeline / PR Build Web Image (pull_request)",
    ]
    REQUIRED_CONTEXTS_APPROVE = [
        "CI/CD Pipeline / Validate - Code Quality (pull_request)",
        "CI/CD Pipeline / Validate - Type Check (mypy) (pull_request)",
        "CI/CD Pipeline / Validate - Migration (alembic) (pull_request)",
        "CI/CD Pipeline / Frontend Lint (pull_request)",
    ]
    FRONTEND_ONLY_CONTEXT = [
        "CI/CD Pipeline / Frontend Lint (pull_request)",
    ]

    # 获取所有open PR
    print(f"获取 {args.base} 分支的open PR...")
    prs = get_open_prs(args.token, args.repo, args.base)
    print(f"找到 {len(prs)} 个open PR")

    approved_count = 0
    merged_count = 0
    skipped_count = 0

    for pr in prs[: args.max_prs]:
        pr_num = pr["number"]
        pr_title = pr["title"]
        head_sha = pr["head"]["sha"]
        base_ref = pr.get("base", {}).get("ref", "")

        # 跳过draft
        if pr.get("draft"):
            print(f"\n⏭️  #{pr_num} {pr_title[:50]} - draft，跳过")
            skipped_count += 1
            continue

        # 跳过目标分支不对的
        if base_ref != args.base:
            skipped_count += 1
            continue

        print(f"\n--- #{pr_num} {pr_title[:60]} ---")

        # 判断是否纯前端
        files = get_pr_files(args.token, args.repo, pr_num)
        frontend_only = is_frontend_only(files)

        if frontend_only:
            approve_contexts = FRONTEND_ONLY_CONTEXT
            merge_contexts = FRONTEND_ONLY_CONTEXT
            print(f"  类型: 纯前端改动 ({len(files)}个文件)")
        else:
            approve_contexts = REQUIRED_CONTEXTS_APPROVE
            merge_contexts = REQUIRED_CONTEXTS_FULL
            print(f"  类型: 全栈/后端改动 ({len(files)}个文件)")

        # 检查审批用的CI状态
        all_ok, pending, failed, _ = check_required_contexts(args.token, args.repo, head_sha, approve_contexts)

        # === 自动审批 ===
        if args.approve and all_ok and not failed:
            if has_approval(args.token, args.repo, pr_num):
                print(f"  ✅ 已有审批，跳过")
            else:
                if dry_run:
                    print(f"  🎯 [DRY-RUN] 将自动审批")
                else:
                    print(f"  🎯 执行自动审批...")
                    ok, msg = approve_pr(args.token, args.repo, pr_num)
                    if ok:
                        print(f"  ✅ 审批成功: {msg}")
                        approved_count += 1
                    else:
                        print(f"  ❌ 审批失败: {msg}")
        elif failed:
            print(f"  ❌ CI有失败项，跳过审批")
        elif pending:
            print(f"  ⏳ CI仍在运行，跳过")

        # === 自动合并 ===
        if args.merge:
            # 检查合并用的CI状态
            merge_ok, merge_pending, merge_failed, _ = check_required_contexts(
                args.token, args.repo, head_sha, merge_contexts
            )

            # 检查审批
            approved = has_approval(args.token, args.repo, pr_num)

            if merge_ok and approved and not merge_failed:
                if dry_run:
                    print(f"  🎯 [DRY-RUN] 将自动合并")
                else:
                    print(f"  🎯 执行自动合并...")
                    ok, msg = merge_pr(args.token, args.repo, pr_num)
                    if ok:
                        print(f"  ✅ 合并成功: {msg}")
                        merged_count += 1
                    else:
                        print(f"  ⚠️  合并失败: {msg}")
            elif merge_pending:
                print(f"  ⏳ 合并条件未满足: CI运行中")
            elif merge_failed:
                print(f"  ❌ 合并条件未满足: CI有失败")
            elif not approved:
                print(f"  ⏳ 合并条件未满足: 无审批")

    print(f"\n=== 扫描结果 ===")
    print(f"  处理PR数: {min(len(prs), args.max_prs)}")
    print(f"  自动审批: {approved_count} 个")
    print(f"  自动合并: {merged_count} 个")
    print(f"  跳过: {skipped_count} 个")
    print(f"  模式: {'DRY-RUN' if dry_run else '正式执行'}")


if __name__ == "__main__":
    main()