批量 PR Review —— refs/pull/*/head + 并行 Codex，批量审查

# 查看所有 PR 引用
git ls-remote https://github.com/org/repo.git | grep refs/pull

# 输出示例：
# abc123  refs/pull/1/head      # PR #1 的最新提交
# def456  refs/pull/1/merge     # PR #1 的合并提交
# ghi789  refs/pull/2/head      # PR #2 的最新提交
# jkl012  refs/pull/3/head      # PR #3 的最新提交

refs/pull/{N}/head   → PR 分支的最新提交（作者提交的代码）
refs/pull/{N}/merge  → 与目标分支的预合并提交（GitHub 计算的合并结果）

┌──────────────────────────────────────────────┐
│           /head vs /merge 对比                │
├──────────────────────────────────────────────┤
│ /head                                        │
│ - 包含作者提交的实际代码                       │
│ - 反映 PR 的真实变更                           │
│ - 适用于代码审查                               │
│                                              │
│ /merge                                       │
│ - GitHub 自动计算的合并结果                    │
│ - 用于检查合并是否冲突                         │
│ - 适用于合并前验证                             │
│ - 如果 PR 未开启自动合并，可能不存在            │
└──────────────────────────────────────────────┘

# 方法 1：使用 git fetch 获取特定 PR
git fetch origin refs/pull/123/head:pr-123
git checkout pr-123

# 方法 2：直接查看 diff
git fetch origin refs/pull/123/head
git diff HEAD..FETCH_HEAD

# 方法 3：使用 GitHub API
curl -s https://api.github.com/repos/org/repo/pulls/123/files

# 方法 4：使用 gh CLI
gh pr view 123 --json files,title,body
gh pr diff 123

# 使用 gh CLI 获取所有开放 PR
gh pr list --state open --json number,title,author,labels

# 获取前 20 个 PR
gh pr list --state open --limit 20 --json number,title,author

# 获取特定作者的所有 PR
gh pr list --state open --author username --json number,title

# 获取带有特定 label 的 PR
gh pr list --state open --label "needs-review" --json number,title

┌─────────────────────────────────────────────────────────┐
│                    批量 PR 审查系统                       │
├─────────────────────────────────────────────────────────┤
│                                                         │
│  ┌──────────┐    ┌───────────┐    ┌──────────────┐      │
│  │  PR 发现  │───→│ 任务分发  │───→│ Codex 集群   │      │
│  │          │    │           │    │              │      │
│  │ gh pr    │    │ 按优先级  │    │ 并行 Codex   │      │
│  │ list     │    │ 排序      │    │ 实例池       │      │
│  └──────────┘    └───────────┘    └──────┬───────┘      │
│                                         │               │
│                                    ┌────▼───────┐       │
│                                    │ 结果聚合   │       │
│                                    │            │       │
│                                    │ 报告生成   │       │
│                                    │ 优先级排序 │       │
│                                    └────┬───────┘       │
│                                         │               │
│                                    ┌────▼───────┐       │
│                                    │ 人工审核   │       │
│                                    │            │       │
│                                    │ 只看红色   │       │
│                                    │ 标记的 PR  │       │
│                                    └────────────┘       │
└─────────────────────────────────────────────────────────┘

1. 发现 PR  →  gh pr list 获取所有待审查 PR
2. 过滤     →  按 label/author/size 过滤
3. 排序     →  按优先级排序（安全 > Bug > Feature）
4. 分发     →  为每个 PR 创建临时工作目录
5. 审查     →  并行启动 Codex 实例审查每个 PR
6. 聚合     →  收集所有审查结果
7. 报告     →  生成汇总报告
8. 清理     →  删除临时文件

import json
import os
import time
from datetime import datetime
from concurrent.futures import ThreadPoolExecutor
from dataclasses import dataclass, field
from typing import List, Optional

@dataclass
class PRInfo:
    """PR 信息"""
    number: int
    title: str
    author: str
    labels: List[str] = field(default_factory=list)
    body: str = ""
    files: List[dict] = field(default_factory=list)
    review_status: str = "pending"
    review_result: Optional[dict] = None

@dataclass
class BatchReviewConfig:
    """批量审查配置"""
    repo: str
    base_dir: str = "/opt/data/batch-pr-reviews"
    max_parallel: int = 3
    timeout_per_pr: int = 600
    pr_limit: int = 20
    label_filter: Optional[str] = None
    author_filter: Optional[str] = None


class BatchPRReviewer:
    """批量 PR 审查器"""

    def __init__(self, config: BatchReviewConfig):
        self.config = config
        self.base_dir = config.base_dir
        self.timestamp = datetime.now().strftime("%Y%m%d_%H%M%S")
        self.session_dir = f"{self.base_dir}/session-{self.timestamp}"
        self.review_prompt = self._build_review_prompt()

    def _build_review_prompt(self) -> str:
        """构建审查 prompt"""
        return """你是资深代码审查专家。请对以下 PR 进行审查。

审查重点：
1. 安全问题：SQL 注入、XSS、CSRF、命令注入、路径遍历、敏感数据泄露
2. 逻辑错误：边界条件、空值处理、类型错误
3. 代码质量：可读性、复杂度、重复代码、命名
4. 最佳实践：错误处理、资源管理、API 设计
5. 性能：N+1 查询、不必要的循环、内存泄漏

输出格式：
## 审查结果

### 严重程度
[🔴 严重 / 🟡 警告 / 🟢 通过]

### 问题列表
[具体问题描述，包含文件路径和行号]

### 修复建议
[具体的修复建议]

### 合并建议
[可以合并 / 需要修改 / 需要讨论]"""

    def discover_prs(self) -> List[PRInfo]:
        """发现待审查的 PR"""
        print(f"📋 正在获取 PR 列表...")

        # 构建 gh pr list 命令
        cmd_parts = [
            f"gh pr list",
            f"--state open",
            f"--limit {self.config.pr_limit}",
            f"--json number,title,author,labels,body,files"
        ]

        if self.config.label_filter:
            cmd_parts.append(f"--label {self.config.label_filter}")

        cmd = " ".join(cmd_parts)
        result = terminal(command=cmd, timeout=60)

        if result.exit_code != 0:
            print(f"❌ 获取 PR 列表失败: {result.output}")
            return []

        pr_data = json.loads(result.output)
        prs = []

        for data in pr_data:
            pr = PRInfo(
                number=data["number"],
                title=data["title"],
                author=data["author"]["login"],
                labels=[l["name"] for l in data.get("labels", [])],
                body=data.get("body", ""),
                files=data.get("files", [])
            )
            prs.append(pr)

        print(f"✅ 发现 {len(prs)} 个待审查 PR")
        return prs

    def prioritize_prs(self, prs: List[PRInfo]) -> List[PRInfo]:
        """按优先级排序 PR"""

        def priority_score(pr: PRInfo) -> int:
            score = 0

            # 安全相关 PR 优先级最高
            security_labels = {"security", "vulnerability", "cve", "hotfix"}
            if any(l.lower() in security_labels for l in pr.labels):
                score += 100

            # Bug 修复次之
            bug_labels = {"bug", "fix", "bugfix"}
            if any(l.lower() in bug_labels for l in pr.labels):
                score += 50

            # 紧急标签
            if any("urgent" in l.lower() or "critical" in l.lower() for l in pr.labels):
                score += 75

            # 文件数少的优先（快速审查）
            score += max(0, 20 - len(pr.files))

            return score

        sorted_prs = sorted(prs, key=priority_score, reverse=True)

        print("\n📊 PR 优先级排序:")
        for i, pr in enumerate(sorted_prs):
            score = priority_score(pr)
            label_str = ", ".join(pr.labels) if pr.labels else "无标签"
            print(f"  {i+1}. PR #{pr.number}: {pr.title[:50]} (优先级: {score}, 标签: {label_str})")

        return sorted_prs

    def review_single_pr(self, pr: PRInfo) -> dict:
        """审查单个 PR"""

        pr_dir = f"{self.session_dir}/pr-{pr.number}"
        print(f"\n{'='*50}")
        print(f"🔍 开始审查 PR #{pr.number}: {pr.title}")
        print(f"{'='*50}")

        # 创建 PR 工作目录
        terminal(command=f"mkdir -p {pr_dir}", timeout=10)

        # 使用 gh pr checkout 检出 PR（一步克隆+检出）
        clone_checkout = terminal(
            command=f"cd {self.session_dir} && gh pr checkout {pr.number} --clone pr-{pr.number}",
            timeout=180
        )

        if clone_checkout.exit_code != 0:
            print(f"❌ PR #{pr.number} 检出失败: {clone_checkout.output}")
            return {
                "pr_number": pr.number,
                "status": "failed",
                "error": clone_checkout.output
            }

        # 获取 PR diff
        diff = terminal(
            command=f"cd {pr_dir} && git diff --stat HEAD~1..HEAD",
            timeout=30
        )
        print(f"变更: {diff.output.strip()}")

        # 构建定制化的 prompt
        prompt = self.review_prompt + f"""

PR 信息：
- 标题: {pr.title}
- 作者: {pr.author}
- 标签: {', '.join(pr.labels)}
- 描述: {pr.body[:300] if pr.body else '无'}"""

        # 启动 Codex 审查
        session = terminal(
            command=f"cd {pr_dir} && codex exec -p \"{prompt}\"",
            background=True,
            pty=True,
            timeout=self.config.timeout_per_pr,
            notify_on_complete=True
        )

        print(f"🤖 Codex 审查已启动 (Session: {session.session_id})")

        # 等待审查完成
        result = process(
            action="wait",
            session_id=session.session_id,
            timeout=self.config.timeout_per_pr
        )

        # 获取审查输出
        log = process(
            action="log",
            session_id=session.session_id,
            limit=500
        )

        # 保存审查报告
        report_path = f"{pr_dir}/review-report.md"
        with open(report_path, 'w') as f:
            f.write(f"# PR #{pr.number} 审查报告\n\n")
            f.write(f"- 标题: {pr.title}\n")
            f.write(f"- 作者: {pr.author}\n")
            f.write(f"- 时间: {datetime.now().isoformat()}\n\n")
            f.write(log)

        print(f"📝 审查报告已保存: {report_path}")

        return {
            "pr_number": pr.number,
            "status": "completed",
            "exit_code": result.exit_code,
            "report_path": report_path,
            "review_output": log
        }

    def run_batch_review(self) -> dict:
        """执行批量审查"""

        # 创建会话目录
        terminal(command=f"mkdir -p {self.session_dir}", timeout=10)

        # 发现 PR
        prs = self.discover_prs()
        if not prs:
            print("没有需要审查的 PR")
            return {"total": 0, "results": []}

        # 排序
        sorted_prs = self.prioritize_prs(prs)

        # 并行审查
        print(f"\n🚀 启动并行审查（最多 {self.config.max_parallel} 个并发）...")

        results = []
        active_sessions = {}

        # 启动第一批
        prs_to_review = list(sorted_prs)
        while prs_to_review or active_sessions:
            # 填充到最大并发数
            while len(active_sessions) < self.config.max_parallel and prs_to_review:
                pr = prs_to_review.pop(0)
                session = terminal(
                    command=f"cd {self.session_dir} && gh pr checkout {pr.number} --clone pr-{pr.number} && cd pr-{pr.number} && codex exec -p \"{self.review_prompt} PR信息: 标题={pr.title}, 作者={pr.author}, 标签={', '.join(pr.labels)}\"",
                    background=True,
                    pty=True,
                    timeout=self.config.timeout_per_pr,
                    notify_on_complete=True
                )
                active_sessions[session.session_id] = pr
                print(f"  🚀 PR #{pr.number} -> Session {session.session_id}")

            if not active_sessions:
                break

            # 等待任意一个完成
            time.sleep(10)

            for sid, pr in list(active_sessions.items()):
                status = process(action="poll", session_id=sid)

                if status.status != "running":
                    # 获取结果
                    log = process(action="log", session_id=sid, limit=500)

                    pr_dir = f"{self.session_dir}/pr-{pr.number}"
                    report_path = f"{pr_dir}/review-report.md"

                    with open(report_path, 'w') as f:
                        f.write(f"# PR #{pr.number} 审查报告\n\n")
                        f.write(f"- 标题: {pr.title}\n")
                        f.write(f"- 作者: {pr.author}\n\n")
                        f.write(log)

                    results.append({
                        "pr_number": pr.number,
                        "title": pr.title,
                        "status": "completed",
                        "report_path": report_path,
                        "output_preview": log[:500]
                    })

                    print(f"  ✅ PR #{pr.number} 审查完成")
                    del active_sessions[sid]

        # 生成汇总报告
        summary = self.generate_summary(results)

        return {
            "total": len(sorted_prs),
            "completed": len(results),
            "results": results,
            "summary_path": summary
        }

    def generate_summary(self, results: List[dict]) -> str:
        """生成汇总报告"""

        summary_path = f"{self.session_dir}/BATCH-REVIEW-SUMMARY.md"

        with open(summary_path, 'w') as f:
            f.write(f"# 批量 PR 审查汇总报告\n\n")
            f.write(f"生成时间: {datetime.now().isoformat()}\n")
            f.write(f"审查数量: {len(results)}\n\n")

            f.write("## 审查结果概览\n\n")
            f.write("| PR # | 标题 | 状态 | 报告 |\n")
            f.write("|------|------|------|------|\n")

            for r in results:
                f.write(f"| #{r['pr_number']} | {r['title'][:40]} | {r['status']} | [报告]({r['report_path']}) |\n")

            f.write(f"\n## 详细审查结果\n\n")

            for r in results:
                f.write(f"---\n\n")
                f.write(f"### PR #{r['pr_number']}: {r['title']}\n\n")

                if r.get("output_preview"):
                    f.write(f"{r['output_preview']}\n\n")

        print(f"\n📊 汇总报告已生成: {summary_path}")
        return summary_path

    def cleanup(self):
        """清理临时文件"""
        print(f"\n🗑️ 清理临时文件...")
        # 保留报告，只删除克隆的仓库
        # terminal(command=f"find {self.session_dir} -name '.git' -type d -exec rm -rf {{}} \\;")
        print(f"📁 审查报告保留在: {self.session_dir}")

# 配置
config = BatchReviewConfig(
    repo="org/my-project",
    base_dir="/opt/data/batch-pr-reviews",
    max_parallel=3,          # 同时审查 3 个 PR
    timeout_per_pr=600,      # 每个 PR 最多 10 分钟
    pr_limit=20,             # 最多审查 20 个 PR
    label_filter="needs-review"  # 只审查标记为 needs-review 的 PR
)

# 创建审查器
reviewer = BatchPRReviewer(config)

# 执行批量审查
result = reviewer.run_batch_review()

# 查看汇总
print(f"\n{'='*60}")
print(f"📊 批量审查完成")
print(f"{'='*60}")
print(f"总 PR 数: {result['total']}")
print(f"已完成: {result['completed']}")
print(f"汇总报告: {result['summary_path']}")

def quick_diff_review(pr_number: int) -> str:
    """快速获取 PR diff 并进行审查（不克隆仓库）"""

    review_dir = f"/opt/data/quick-review-{pr_number}"
    terminal(command=f"mkdir -p {review_dir}", timeout=10)

    # 直接获取 PR diff
    diff = terminal(
        command=f"gh pr diff {pr_number}",
        timeout=60
    )

    # 保存 diff 到文件
    diff_path = f"{review_dir}/pr-{pr_number}.diff"
    with open(diff_path, 'w') as f:
        f.write(diff.output)

    # 让 Codex 基于 diff 文件进行审查
    prompt = f"""请审查以下 PR 的 diff 内容。

审查重点：
1. 安全问题
2. 逻辑错误
3. 代码质量
4. 潜在 Bug

diff 文件路径: {diff_path}
请读取 diff 文件并进行审查。"""

    session = terminal(
        command=f"cd {review_dir} && codex exec -p \"{prompt}\"",
        background=True,
        pty=True,
        timeout=300,
        notify_on_complete=True
    )

    result = process(action="wait", session_id=session.session_id, timeout=300)
    log = process(action="log", session_id=session.session_id)

    return log

def classify_review(review_output: str) -> str:
    """根据审查输出自动分类"""

    output_lower = review_output.lower()

    # 检查严重程度标记
    if "🔴" in review_output or "严重" in output_lower or "critical" in output_lower:
        return "🔴 需要立即关注"

    if "🟡" in review_output or "警告" in output_lower or "warning" in output_lower:
        return "🟡 建议修改后合并"

    if "🟢" in review_output or "通过" in output_lower or "approved" in output_lower:
        return "🟢 可以合并"

    # 基于关键词判断
    security_keywords = ["注入", "xss", "csrf", "泄露", "injection", "vulnerability"]
    if any(kw in output_lower for kw in security_keywords):
        return "🔴 存在安全问题"

    if "需要修改" in output_lower or "需要修复" in output_lower or "request changes" in output_lower:
        return "🟡 需要修改"

    if "可以合并" in output_lower or "approve" in output_lower:
        return "🟢 可以合并"

    return "⚪ 需要人工判断"

def auto_comment_batch(results: List[dict]):
    """批量发布审查评论到各个 PR"""

    for r in results:
        pr_number = r["pr_number"]
        report_path = r["report_path"]

        # 生成简洁的评论
        comment_cmd = f"""gh pr comment {pr_number} --body "$(cat << 'EOF'
# 🤖 AI 代码审查报告

> 此审查由 Codex CLI 自动生成，仅供参考。

## 审查结果
{classify_review(open(report_path).read())}

详细报告请查看完整输出。

---
*AI Reviewer | {datetime.now().strftime('%Y-%m-%d %H:%M')}*
EOF
)"""

        result = terminal(command=comment_cmd, timeout=30)
        if result.exit_code == 0:
            print(f"💬 评论已发布到 PR #{pr_number}")
        else:
            print(f"❌ 评论发布失败 PR #{pr_number}: {result.output}")

#!/usr/bin/env python3
"""
批量 PR 审查工具
用法: python batch_review.py --limit 10 --parallel 3 --label needs-review
"""

import argparse
import json
import sys
from datetime import datetime

def main():
    parser = argparse.ArgumentParser(description="批量 PR 审查工具")
    parser.add_argument("--limit", type=int, default=20, help="最大审查 PR 数量")
    parser.add_argument("--parallel", type=int, default=3, help="最大并行数")
    parser.add_argument("--timeout", type=int, default=600, help="每个 PR 超时（秒）")
    parser.add_argument("--label", type=str, default=None, help="按标签过滤")
    parser.add_argument("--author", type=str, default=None, help="按作者过滤")
    parser.add_argument("--comment", action="store_true", help="自动发布评论")
    args = parser.parse_args()

    print(f"🤖 批量 PR 审查工具")
    print(f"{'='*40}")
    print(f"PR 限制: {args.limit}")
    print(f"并行数: {args.parallel}")
    print(f"超时: {args.timeout}s")
    print(f"标签过滤: {args.label or '无'}")
    print(f"自动评论: {'是' if args.comment else '否'}")
    print(f"{'='*40}\n")

    # 配置
    config = BatchReviewConfig(
        repo="org/my-project",
        max_parallel=args.parallel,
        timeout_per_pr=args.timeout,
        pr_limit=args.limit,
        label_filter=args.label,
        author_filter=args.author
    )

    # 执行
    reviewer = BatchPRReviewer(config)
    result = reviewer.run_batch_review()

    # 输出汇总
    print(f"\n{'='*40}")
    print(f"📊 审查汇总")
    print(f"{'='*40}")
    print(f"总 PR 数: {result['total']}")
    print(f"已完成: {result['completed']}")
    print(f"汇总报告: {result['summary_path']}")

    # 自动评论
    if args.comment and result['completed'] > 0:
        print(f"\n💬 正在发布评论...")
        auto_comment_batch(result['results'])

    print(f"\n✅ 批量审查完成！")

if __name__ == "__main__":
    main()

┌──────────────────────────────────────────────────┐
│            并发数调优建议                          │
├──────────────────────────────────────────────────┤
│                                                  │
│ 1-2 个实例：                                      │
│   - 适合资源受限的环境                            │
│   - 适合大型 PR（复杂代码库）                     │
│                                                  │
│ 3-5 个实例：                                      │
│   - 推荐的默认值                                  │
│   - 平衡速度和资源消耗                            │
│   - 适合大多数场景                                │
│                                                  │
│ 5+ 个实例：                                       │
│   - 需要充足的 CPU 和内存                         │
│   - 注意 GitHub API 速率限制                     │
│   - 注意 Codex API 速率限制                       │
│                                                  │
│ 监控指标：                                        │
│   - CPU 使用率                                    │
│   - 内存使用                                      │
│   - 磁盘 I/O                                     │
│   - API 调用频率                                  │
└──────────────────────────────────────────────────┘

def handle_rate_limit():
    """处理 GitHub API 和 Codex API 的速率限制"""

    # 检查 GitHub API 速率限制
    rate_check = terminal(
        command="gh api rate_limit",
        timeout=30
    )

    # 如果接近限制，暂停
    if rate_check.exit_code == 0:
        rate_data = json.loads(rate_check.output)
        remaining = rate_data.get("resources", {}).get("core", {}).get("remaining", 0)

        if remaining < 10:
            print("⚠️ GitHub API 速率限制即将到达，等待重置...")
            time.sleep(60)

def get_already_reviewed(session_dir: str) -> set:
    """获取已经审查过的 PR 编号"""
    reviewed = set()

    import glob
    for report in glob.glob(f"{session_dir}/pr-*/review-report.md"):
        pr_num = int(report.split("pr-")[1].split("/")[0])
        reviewed.add(pr_num)

    return reviewed

# 在 discover_prs 后过滤
already_reviewed = get_already_reviewed(self.session_dir)
prs_to_review = [pr for pr in prs if pr.number not in already_reviewed]
print(f"跳过 {len(already_reviewed)} 个已审查的 PR")