xiaoxia-saas/scripts/ci/pr_auto_scan.py

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

新增：AI审查联动 - AI代码审查发现严重问题时，不自动审批
"""

import argparse
import json
import re
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:
        body = e.read().decode()
        if body:
            try:
                return json.loads(body), e.code
            except json.JSONDecodeError:
                return {"error": body}, e.code
        return {"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 not in ("success", "skipped"):
            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 get_ai_review_result(token, repo, pr_number):
    """
    检查AI代码审查结果，返回 (has_critical, review_body)
    has_critical: 是否有严重问题（需修改的问题 > 0）
    review_body: 最新的AI审查评论文本
    """
    # AI审查评论标记
    AI_REVIEW_MARKER = "AI_CODE_REVIEW_AUTO_COMMENT"

    comments, code = api_request(token, repo, f"issues/{pr_number}/comments")
    if code != 200:
        return False, None

    # 找最新的AI审查评论
    ai_comments = [c for c in comments if isinstance(c, dict) and AI_REVIEW_MARKER in c.get("body", "")]

    if not ai_comments:
        return False, None

    # 按时间排序，取最新的
    latest = max(ai_comments, key=lambda c: c.get("created_at", ""))
    body = latest.get("body", "")

    # 解析严重问题数量
    # 匹配 "严重问题数量：X 个" 或 "需修改的问题（严重）" 下的列表
    critical_count = 0

    # 方式1：直接匹配数字
    match = re.search(r"严重问题数量[：:]\s*(\d+)\s*个", body)
    if match:
        critical_count = int(match.group(1))
    else:
        # 方式2：数 "需修改的问题" 章节下的条目数
        critical_section = re.search(
            r"###\s*[❌⚠️].*?(?:需修改|问题).*?\n(.*?)(?=\n###|\Z)",
            body,
            re.DOTALL,
        )
        if critical_section:
            section_text = critical_section.group(1)
            # 数编号条目 1. 2. 3.
            items = re.findall(r"^\d+\.\s+\*\*", section_text, re.MULTILINE)
            critical_count = len(items)

    has_critical = critical_count > 0
    return has_critical, body


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

    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": reason},
    )

    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": reason},
        )
        if code3 in (200, 201):
            return True, "审批提交成功(备用端点)"
        return False, f"审批提交失败: HTTP {code2}/{code3}"


def add_pr_label(token, repo, pr_number, label):
    """给PR添加标签"""
    data, code = api_request(
        token,
        repo,
        f"issues/{pr_number}/labels",
        method="POST",
        data={"labels": [label]},
    )
    return code in (200, 201)


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数")
    parser.add_argument("--skip-ai-review", action="store_true", help="跳过AI审查检查（强制审批）")

    args = parser.parse_args()

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

    # required contexts（与分支保护一致）
    REQUIRED_CONTEXTS_FULL = [
        # 统一使用CI Gate作为合并门禁（与pr-automation和分支保护保持一致）
        # CI Gate内部已包含: 代码质量/类型检查/迁移检查/单测/集成测试/前端Lint/前端单测/构建/AI审查
        "CI/CD Pipeline / CI Gate (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 = [
        # 纯前端PR也用CI Gate统一判断，内部自动跳过后端相关检查
        "CI/CD Pipeline / CI Gate (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
    ai_blocked_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)

        # === AI审查检查 ===
        ai_has_critical = False
        if not args.skip_ai_review and all_ok and not failed and args.approve:
            ai_has_critical, ai_body = get_ai_review_result(args.token, args.repo, pr_num)
            if ai_has_critical:
                print("  ⚠️  AI审查发现严重问题，阻止自动审批")
                ai_blocked_count += 1
                # 给PR打标签便于人工识别
                if not dry_run:
                    add_pr_label(args.token, args.repo, pr_num, "ai-review/需修改")

        # === 自动审批 ===
        if args.approve and all_ok and not failed and not ai_has_critical:
            if has_approval(args.token, args.repo, pr_num):
                print("  ✅ 已有审批，跳过")
            else:
                if dry_run:
                    print("  🎯 [DRY-RUN] 将自动审批")
                else:
                    print("  🎯 执行自动审批...")
                    ok, msg = approve_pr(args.token, args.repo, pr_num)
                    if ok:
                        print(f"  ✅ 审批成功: {msg}")
                        approved_count += 1
                    else:
                        print(f"  ❌ 审批失败: {msg}")
        elif ai_has_critical:
            print("  🚫 AI审查阻止审批（人工可手动审批覆盖）")
        elif failed:
            print("  ❌ CI有失败项，跳过审批")
        elif pending:
            print("  ⏳ 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("  🎯 [DRY-RUN] 将自动合并")
                else:
                    print("  🎯 执行自动合并...")
                    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("  ⏳ 合并条件未满足: CI运行中")
            elif merge_failed:
                print("  ❌ 合并条件未满足: CI有失败")
            elif not approved:
                print("  ⏳ 合并条件未满足: 无审批")

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


if __name__ == "__main__":
    main()