ci: Agent提交前自动格式化 + AI审查与auto-approve联动

P0-2: 新增 scripts/format.sh 一键格式化脚本（black + isort + eslint）
P0-3: pr_auto_scan.py 新增AI审查联动，发现严重问题时阻止自动审批
  - get_ai_review_result() 解析AI审查评论中的严重问题数量
  - 有严重问题时不自动审批，打 ai-review/需修改 标签
  - 新增 --skip-ai-review 参数可强制跳过AI检查
  - 人工仍可手动审批覆盖，不加硬门禁

scripts/ci/pr_auto_scan.py

@@ -2,11 +2,14 @@
 """
 PR自动扫描器：扫描所有open PR，对CI全绿的进行自动审批/合并
 作为短作业模式的兜底机制，每5分钟运行一次
+
+新增：AI审查联动 - AI代码审查发现严重问题时，不自动审批
 """
 
 import argparse
 import json
 import os
+import re
 import sys
 import time
 import urllib.error
@@ -110,7 +113,58 @@ def has_approval(token, repo, pr_number):
     return any(r.get("state") == "APPROVED" for r in reviews if isinstance(r, dict))
 
 
-def approve_pr(token, repo, pr_number):
+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(
@@ -118,7 +172,7 @@ def approve_pr(token, repo, pr_number):
         repo,
         f"pulls/{pr_number}/reviews",
         method="POST",
-        data={"event": "PENDING", "body": "CI全绿，自动审批通过。"},
+        data={"event": "PENDING", "body": reason},
     )
 
     if code not in (200, 201):
@@ -137,7 +191,7 @@ def approve_pr(token, repo, pr_number):
         repo,
         f"pulls/{pr_number}/reviews/{review_id}/events",
         method="POST",
-        data={"event": "APPROVED", "body": "CI全绿，自动审批通过。"},
+        data={"event": "APPROVED", "body": reason},
     )
 
     if code2 in (200, 201):
@@ -149,13 +203,25 @@ def approve_pr(token, repo, pr_number):
             repo,
             f"pulls/{pr_number}/reviews/{review_id}",
             method="POST",
-            data={"event": "APPROVED", "body": "CI全绿，自动审批通过。"},
+            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）"""
     # 等待几秒让状态同步
@@ -200,6 +266,7 @@ 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数")
+    parser.add_argument("--skip-ai-review", action="store_true", help="跳过AI审查检查（强制审批）")
 
     args = parser.parse_args()
 
@@ -233,6 +300,7 @@ def main():
     approved_count = 0
     merged_count = 0
     skipped_count = 0
+    ai_blocked_count = 0
 
     for pr in prs[: args.max_prs]:
         pr_num = pr["number"]
@@ -269,8 +337,19 @@ def main():
         # 检查审批用的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(f"  ⚠️  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:
+        if args.approve and all_ok and not failed and not ai_has_critical:
             if has_approval(args.token, args.repo, pr_num):
                 print(f"  ✅ 已有审批，跳过")
             else:
@@ -284,6 +363,8 @@ def main():
                         approved_count += 1
                     else:
                         print(f"  ❌ 审批失败: {msg}")
+        elif ai_has_critical:
+            print(f"  🚫 AI审查阻止审批（人工可手动审批覆盖）")
         elif failed:
             print(f"  ❌ CI有失败项，跳过审批")
         elif pending:
@@ -321,9 +402,11 @@ def main():
     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(f"  模式: {'DRY-RUN' if dry_run else '正式执行'}")
 
 
 if __name__ == "__main__":
     main()
+

scripts/format.sh

@@ -0,0 +1,33 @@
+#!/usr/bin/env bash
+# Agent提交前自动格式化脚本
+# 用法：./scripts/format.sh [path1 path2 ...]
+# 不传参数则格式化所有后端代码
+set -euo pipefail
+
+cd "$(dirname "$0")/.."
+
+echo "=== 代码格式化 ==="
+
+# 后端：black + isort（顺序：先isort后black，与pyproject.toml配置一致）
+if command -v black &>/dev/null && command -v isort &>/dev/null; then
+    TARGETS="${@:-alembic apps packages tests scripts}"
+    echo "后端格式化: $TARGETS"
+    python3 -m isort $TARGETS
+    python3 -m black $TARGETS
+    echo "✅ 后端格式化完成"
+else
+    echo "⚠️  未安装black/isort，跳过后端格式化"
+fi
+
+# 前端：prettier + eslint --fix（如果有前端改动）
+if [ -d "apps/web" ] && command -v npx &>/dev/null; then
+    if [ "$#" -eq 0 ] || echo "$@" | grep -q "apps/web"; then
+        echo "前端格式化: apps/web"
+        (cd apps/web && npx eslint src --ext .ts,.tsx --fix 2>/dev/null || true)
+        (cd apps/web && npx prettier --write "src/**/*.{ts,tsx,css,json}" 2>/dev/null || true)
+        echo "✅ 前端格式化完成"
+    fi
+fi
+
+echo "=== 格式化全部完成 ==="
+