xiaoxia/xiaoxia-saas
1
0
Fork 0
Code Issues 11 Pull Requests 3 Actions Packages Projects Releases 7 Wiki Activity

fix(ci): 格式化pr_auto_scan.py修复develop分支Code Quality门禁 #741

Merged
xiaoxia merged 1 commits from fix/ci-pr-auto-scan-black-format into develop 2026-07-22 20:52:24 +08:00
Conversation 3 Commits 1 Files Changed 1
+69 -58
1 changed files with 69 additions and 58 deletions
Download Patch File Download Diff File Expand all files Collapse all files
scripts/ci/pr_auto_scan.py
+69 -58
View File
@@ -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} 个")