From 8ca2ffe272e7453ff0e88a22e0aced357b858792 Mon Sep 17 00:00:00 2001 From: xiaoxia Date: Thu, 23 Jul 2026 17:53:30 +0800 Subject: [PATCH] =?UTF-8?q?fix(ci):=20pr=5Fauto=5Fscan.py=20black=20120?= =?UTF-8?q?=E5=88=97=E6=A0=BC=E5=BC=8F=E5=8C=96?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 修复Code Quality job因脚本自身格式问题失败的问题 --- scripts/ci/pr_auto_scan.py | 408 +++++++++++++++++++++++++++++++++++++ 1 file changed, 408 insertions(+) create mode 100644 scripts/ci/pr_auto_scan.py diff --git a/scripts/ci/pr_auto_scan.py b/scripts/ci/pr_auto_scan.py new file mode 100644 index 000000000..f17753d27 --- /dev/null +++ b/scripts/ci/pr_auto_scan.py @@ -0,0 +1,408 @@ +#!/usr/bin/env python3 +""" +PR自动扫描器：扫描所有open PR，对CI全绿的进行自动审批/合并 +作为短作业模式的兜底机制，每5分钟运行一次 + +新增：AI审查联动 - AI代码审查发现严重问题时，不自动审批 +""" + +import argparse +import json +import os +import re +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 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/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 + 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("re", "") + + # 跳过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()