fix(ci): auto-approve/merge改为短作业+5分钟定时扫描，防止长轮询占垮runner

scripts/ci/pr_auto_scan.py

@@ -0,0 +1,318 @@
+#!/usr/bin/env python3
+"""
+PR自动扫描器：扫描所有open PR，对CI全绿的进行自动审批/合并
+作为短作业模式的兜底机制，每5分钟运行一次
+"""
+import argparse
+import json
+import sys
+import urllib.request
+import urllib.error
+import time
+import os
+
+
+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()