From b1913f48e21f0f8479a3a5cf290786f321cd85fd Mon Sep 17 00:00:00 2001 From: xiaoxia Date: Wed, 22 Jul 2026 20:31:49 +0800 Subject: [PATCH] =?UTF-8?q?fix(ci):=20=E6=A0=BC=E5=BC=8F=E5=8C=96pr=5Fauto?= =?UTF-8?q?=5Fscan.py=E4=BF=AE=E5=A4=8Ddevelop=E5=88=86=E6=94=AFCode=20Qua?= =?UTF-8?q?lity=E9=97=A8=E7=A6=81?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- scripts/ci/pr_auto_scan.py | 127 ++++++++++++++++++++----------------- 1 file changed, 69 insertions(+), 58 deletions(-) diff --git a/scripts/ci/pr_auto_scan.py b/scripts/ci/pr_auto_scan.py index 483f7819a..9ddfc936f 100644 --- a/scripts/ci/pr_auto_scan.py +++ b/scripts/ci/pr_auto_scan.py @@ -3,31 +3,30 @@ PR自动扫描器：扫描所有open PR，对CI全绿的进行自动审批/合并 作为短作业模式的兜底机制，每5分钟运行一次 """ + import argparse import json -import sys -import urllib.request -import urllib.error -import time import os +import sys +import time +import urllib.error +import urllib.request -def api_request(token, repo, endpoint, method='GET', data=None): +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" - } + 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 @@ -35,13 +34,12 @@ def api_request(token, repo, endpoint, method='GET', data=None): return json.loads(e.read().decode()) if e.read() else {"error": str(e)}, e.code -def get_open_prs(token, repo, base='develop'): +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}") + 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) @@ -63,11 +61,11 @@ 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": @@ -76,7 +74,7 @@ def check_required_contexts(token, repo, sha, contexts): any_pending = True if state in ("failure", "error"): any_failed = True - + return all_success, any_pending, any_failed, statuses @@ -85,8 +83,7 @@ def get_pr_files(token, repo, pr_number): files = [] page = 1 while True: - data, code = api_request(token, repo, - f"pulls/{pr_number}/files?per_page=300&page={page}") + 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) @@ -116,34 +113,44 @@ def has_approval(token, repo, pr_number): 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全绿，自动审批通过。"}) - + 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, + data2, code2 = api_request( + token, + repo, f"pulls/{pr_number}/reviews/{review_id}/events", method="POST", - data={"event": "APPROVED", "body": "CI全绿，自动审批通过。"}) - + data={"event": "APPROVED", "body": "CI全绿，自动审批通过。"}, + ) + if code2 in (200, 201): return True, "审批提交成功" else: # 尝试另一个端点 - data3, code3 = api_request(token, repo, + data3, code3 = api_request( + token, + repo, f"pulls/{pr_number}/reviews/{review_id}", method="POST", - data={"event": "APPROVED", "body": "CI全绿，自动审批通过。"}) + data={"event": "APPROVED", "body": "CI全绿，自动审批通过。"}, + ) if code3 in (200, 201): return True, "审批提交成功(备用端点)" return False, f"审批提交失败: HTTP {code2}/{code3}" @@ -153,25 +160,29 @@ 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", + 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 - }) - + "force_merge": False, + }, + ) + if code == 200: return True, "合并成功" elif code == 405: @@ -189,11 +200,11 @@ def main(): 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)", @@ -213,39 +224,39 @@ def main(): 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]: + + 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 @@ -254,11 +265,10 @@ def main(): 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) - + 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): @@ -278,16 +288,17 @@ def main(): 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) - + 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] 将自动合并") @@ -305,7 +316,7 @@ def main(): print(f" ❌ 合并条件未满足: CI有失败") elif not approved: print(f" ⏳ 合并条件未满足: 无审批") - + print(f"\n=== 扫描结果 ===") print(f" 处理PR数: {min(len(prs), args.max_prs)}") print(f" 自动审批: {approved_count} 个") -- 2.54.0