xiaoxia-saas/scripts/ci_code_review.py

#!/usr/bin/env python3
"""
CI Code Review Script
- 从 Gitea 获取 PR diff
- 调用 LLM 进行代码审查
- 将审查结果写回 PR 评论
"""

import argparse
import json
import logging
import os
import sys
from typing import Optional, Tuple

import requests

# ============== 日志配置 ==============
logging.basicConfig(
    level=logging.INFO,
    format="[%(asctime)s] [%(levelname)s] %(message)s",
    datefmt="%Y-%m-%d %H:%M:%S",
)
logger = logging.getLogger("ci_code_review")


# ============== 常量配置 ==============
# diff 最大字符数（超过则截断）
MAX_DIFF_CHARS = int(os.getenv("MAX_DIFF_CHARS", "30000"))
# LLM 调用超时时间（秒）
LLM_TIMEOUT = int(os.getenv("LLM_TIMEOUT", "120"))
# Gitea API 超时时间（秒）
GITEA_TIMEOUT = int(os.getenv("GITEA_TIMEOUT", "30"))
# 最大重试次数
MAX_RETRIES = int(os.getenv("MAX_RETRIES", "2"))
# LLM 提供商: openai (OpenAI兼容) / coze (扣子原生Bot API)
LLM_PROVIDER = os.getenv("LLM_PROVIDER", "coze").lower()


# ============== 工具函数 ==============
def truncate_diff(diff_text: str, max_chars: int) -> Tuple[str, bool]:
    """
    截断过大的 diff 内容，避免超出 LLM 上下文限制。
    优先保留文件头和前面的变更，末尾加提示。
    """
    if len(diff_text) <= max_chars:
        return diff_text, False

    # 找到一个合适的截断位置（尽量在文件边界）
    truncated = diff_text[:max_chars]
    # 尝试在最后一个 "diff --git" 处截断，避免截断到一半
    last_file_boundary = truncated.rfind("\ndiff --git ")
    if last_file_boundary > max_chars // 2:
        truncated = truncated[:last_file_boundary]

    truncated += (
        f"\n\n... [DIFF TRUNCATED] 原始 diff 共 {len(diff_text)} 字符，"
        f"已截断至 {len(truncated)} 字符，仅审查前半部分。\n"
    )
    return truncated, True


def get_env_or_fail(name: str) -> str:
    """从环境变量获取值，不存在则报错退出。"""
    value = os.getenv(name)
    if not value:
        logger.error(f"环境变量 {name} 未设置")
        sys.exit(1)
    return value


# ============== Gitea API 相关 ==============
class GiteaClient:
    """Gitea API 客户端"""

    def __init__(self, base_url: str, token: str, repo: str):
        # 确保 base_url 以 / 结尾
        self.base_url = base_url.rstrip("/") + "/"
        self.token = token
        self.repo = repo  # 格式: owner/repo
        self.session = requests.Session()
        self.session.headers.update(
            {
                "Authorization": f"token {token}",
                "Accept": "application/json",
                "Content-Type": "application/json",
            }
        )

    def _api_url(self, path: str) -> str:
        """拼接 API 路径"""
        return f"{self.base_url}api/v1/repos/{self.repo}/{path.lstrip('/')}"

    def get_pr_diff(self, pr_number: int) -> str:
        """
        获取 PR 的 diff 内容。
        Gitea API: GET /repos/{owner}/{repo}/pulls/{index}.diff
        """
        url = self._api_url(f"pulls/{pr_number}.diff")
        logger.info(f"获取 PR #{pr_number} diff: {url}")

        resp = self.session.get(
            url,
            timeout=GITEA_TIMEOUT,
            headers={
                "Accept": "text/plain",
            },
        )
        if resp.status_code != 200:
            logger.error(f"获取 diff 失败: HTTP {resp.status_code} - {resp.text[:200]}")
            raise RuntimeError(f"Failed to get PR diff: HTTP {resp.status_code}")

        diff_text = resp.text
        logger.info(f"获取到 diff，共 {len(diff_text)} 字符")
        return diff_text

    def get_pr_files(self, pr_number: int) -> list:
        """
        获取 PR 修改的文件列表。
        Gitea API: GET /repos/{owner}/{repo}/pulls/{index}/files
        """
        url = self._api_url(f"pulls/{pr_number}/files")
        logger.info(f"获取 PR #{pr_number} 文件列表")

        resp = self.session.get(url, timeout=GITEA_TIMEOUT)
        if resp.status_code != 200:
            logger.warning(f"获取文件列表失败: HTTP {resp.status_code}")
            return []

        files = resp.json()
        logger.info(f"PR 修改了 {len(files)} 个文件")
        return files

    def post_pr_comment(self, pr_number: int, body: str) -> bool:
        """
        在 PR 上发布评论。
        Gitea API: POST /repos/{owner}/{repo}/issues/{index}/comments
        （Gitea 中 PR 评论走 issues 接口）
        """
        url = self._api_url(f"issues/{pr_number}/comments")
        logger.info(f"发布 PR 评论: {url}")

        payload = {"body": body}
        resp = self.session.post(
            url,
            data=json.dumps(payload),
            timeout=GITEA_TIMEOUT,
        )
        if resp.status_code not in (200, 201):
            logger.error(f"发布评论失败: HTTP {resp.status_code} - {resp.text[:200]}")
            return False

        logger.info(f"评论发布成功，评论 ID: {resp.json().get('id', 'unknown')}")
        return True

    def get_existing_review_comments(self, pr_number: int, marker: str) -> list:
        """
        获取 PR 上已有的 AI 审查评论 ID 列表（带标识 marker）。
        """
        url = self._api_url(f"issues/{pr_number}/comments")
        resp = self.session.get(url, timeout=GITEA_TIMEOUT)
        if resp.status_code != 200:
            logger.warning(f"获取评论列表失败: HTTP {resp.status_code}")
            return []

        comments = resp.json()
        review_comment_ids = []
        for c in comments:
            body = c.get("body", "")
            if marker in body:
                review_comment_ids.append(c.get("id"))
        logger.info(f"找到 {len(review_comment_ids)} 条旧的 AI 审查评论")
        return review_comment_ids

    def delete_pr_comment(self, pr_number: int, comment_id: int) -> bool:
        """
        删除 PR 上的指定评论。
        """
        url = self._api_url(f"issues/comments/{comment_id}")
        resp = self.session.delete(url, timeout=GITEA_TIMEOUT)
        if resp.status_code not in (200, 204):
            logger.warning(f"删除评论 {comment_id} 失败: HTTP {resp.status_code}")
            return False
        return True


def call_llm_openai(
    prompt: str,
    llm_base_url: str,
    llm_api_key: str,
    llm_model: str,
) -> Optional[str]:
    """OpenAI 兼容模式调用"""
    base_url = llm_base_url.rstrip("/") + "/"
    api_url = f"{base_url}chat/completions"

    headers = {
        "Authorization": f"Bearer {llm_api_key}",
        "Content-Type": "application/json",
    }

    payload = {
        "model": llm_model,
        "messages": [
            {
                "role": "system",
                "content": "你是一位严谨的资深代码审查专家，擅长发现代码中的逻辑错误、安全隐患和性能问题。",
            },
            {
                "role": "user",
                "content": prompt,
            },
        ],
        "temperature": 0.3,
        "max_tokens": 2048,
    }

    logger.info(f"调用 LLM (OpenAI兼容): {api_url}, model={llm_model}")

    last_error = None
    for attempt in range(MAX_RETRIES + 1):
        try:
            resp = requests.post(
                api_url,
                headers=headers,
                json=payload,
                timeout=LLM_TIMEOUT,
            )
            if resp.status_code != 200:
                logger.warning(f"LLM 调用失败 (第 {attempt + 1} 次): " f"HTTP {resp.status_code} - {resp.text[:200]}")
                last_error = f"HTTP {resp.status_code}"
                continue

            data = resp.json()
            choices = data.get("choices", [])
            if not choices:
                logger.warning(f"LLM 返回空结果 (第 {attempt + 1} 次)")
                last_error = "empty choices"
                continue

            content = choices[0].get("message", {}).get("content", "")
            if not content.strip():
                logger.warning(f"LLM 返回空内容 (第 {attempt + 1} 次)")
                last_error = "empty content"
                continue

            logger.info(f"LLM 审查完成，结果长度: {len(content)} 字符")
            return content

        except requests.Timeout:
            logger.warning(f"LLM 调用超时 (第 {attempt + 1} 次)")
            last_error = "timeout"
        except requests.RequestException as e:
            logger.warning(f"LLM 调用异常 (第 {attempt + 1} 次): {e}")
            last_error = str(e)

    logger.error(f"LLM 调用最终失败: {last_error}")
    return None


def call_llm_coze(
    prompt: str,
    llm_base_url: str,
    llm_api_key: str,
    llm_model: str,
    coze_bot_id: str,
) -> Optional[str]:
    """扣子(Coze)原生 Bot API 调用（支持异步轮询）"""
    import time

    base_url = llm_base_url.rstrip("/") + "/"
    api_url = f"{base_url}v3/chat"

    headers = {
        "Authorization": f"Bearer {llm_api_key}",
        "Content-Type": "application/json",
    }

    payload = {
        "bot_id": coze_bot_id,
        "user_id": "ci-code-review-bot",
        "stream": False,
        "additional_messages": [
            {
                "role": "user",
                "content": prompt,
                "content_type": "text",
            }
        ],
    }

    logger.info(f"调用 LLM (Coze): {api_url}, bot_id={coze_bot_id}")

    last_error = None
    for attempt in range(MAX_RETRIES + 1):
        try:
            resp = requests.post(
                api_url,
                headers=headers,
                json=payload,
                timeout=LLM_TIMEOUT,
            )
            if resp.status_code != 200:
                logger.warning(f"Coze 调用失败 (第 {attempt + 1} 次): " f"HTTP {resp.status_code} - {resp.text[:300]}")
                last_error = f"HTTP {resp.status_code}"
                continue

            data = resp.json()
            chat_data = data.get("data", {})
            chat_id = chat_data.get("id", "")
            conversation_id = chat_data.get("conversation_id", "")
            status = chat_data.get("status", "")

            # Coze v3 API 异步：先返回 in_progress，需要轮询
            if status == "in_progress" and conversation_id and chat_id:
                logger.info(f"Coze 异步处理中，开始轮询... (chat_id={chat_id[:12]}...)")
                # 轮询 message 列表接口（GET + query参数），最多等 LLM_TIMEOUT 秒
                poll_url = f"{base_url}v3/chat/message/list"
                poll_start = time.time()
                poll_interval = 3  # 每3秒轮询一次

                while time.time() - poll_start < LLM_TIMEOUT:
                    time.sleep(poll_interval)
                    poll_params = {
                        "chat_id": chat_id,
                        "conversation_id": conversation_id,
                    }
                    poll_resp = requests.get(
                        poll_url,
                        headers=headers,
                        params=poll_params,
                        timeout=GITEA_TIMEOUT,
                    )
                    if poll_resp.status_code != 200:
                        logger.debug(f"轮询返回 HTTP {poll_resp.status_code}: {poll_resp.text[:100]}")
                        continue

                    poll_data = poll_resp.json()
                    if poll_data.get("code", 0) != 0:
                        logger.debug(f"轮询返回错误: {poll_data.get('msg', '')}")
                        continue

                    messages = poll_data.get("data", []) or []

                    # 找assistant的answer消息
                    content = None
                    for msg in messages:
                        if msg.get("role") == "assistant" and msg.get("type") == "answer":
                            content = msg.get("content", "")
                            break

                    if content and content.strip():
                        logger.info(f"Coze 审查完成，结果长度: {len(content)} 字符")
                        return content

                logger.warning(f"Coze 轮询超时 ({LLM_TIMEOUT}s)，未拿到结果")
                last_error = "poll timeout"
                continue

            # 同步返回的情况（兼容）
            content = None
            messages = chat_data.get("messages", []) or data.get("messages", [])
            for msg in messages:
                if msg.get("role") == "assistant" and msg.get("type") == "answer":
                    content = msg.get("content", "")
                    break

            if not content:
                content = chat_data.get("content") or data.get("content")

            if not content:
                choices = data.get("choices", [])
                if choices:
                    content = choices[0].get("message", {}).get("content", "")

            if not content or not content.strip():
                logger.warning(f"Coze 返回空内容 (第 {attempt + 1} 次): {str(data)[:200]}")
                last_error = "empty content"
                continue

            logger.info(f"Coze 审查完成，结果长度: {len(content)} 字符")
            return content

        except requests.Timeout:
            logger.warning(f"Coze 调用超时 (第 {attempt + 1} 次)")
            last_error = "timeout"
        except requests.RequestException as e:
            logger.warning(f"Coze 调用异常 (第 {attempt + 1} 次): {e}")
            last_error = str(e)

    logger.error(f"Coze 调用最终失败: {last_error}")
    return None


def build_review_prompt(diff_text: str, pr_number: int, file_list: list) -> str:
    """
    构建代码审查的 Prompt。
    包含：PR 基本信息、修改文件列表、diff 内容、审查要求。
    """
    # 提取文件名列表
    file_names = [f.get("filename", "") for f in file_list] if file_list else []
    file_list_str = "\n".join(f"  - {fn}" for fn in file_names) if file_names else "  （未获取到文件列表）"

    prompt = f"""请作为资深代码审查专家，对以下 Pull Request 的代码变更进行严格审查。

## PR 基本信息
- PR 编号: #{pr_number}
- 修改文件数: {len(file_list) if file_list else '未知'}

## 修改文件列表
{file_list_str}

## 代码变更（diff）
```diff
{diff_text}
```

## 审查要求
请从以下维度进行审查，重点关注严重问题：

1. **逻辑正确性**：是否有明显的逻辑错误、边界条件遗漏、空指针/None引用风险
2. **异常处理**：异常捕获是否合理，是否有裸except，错误处理是否完善
3. **参数校验**：函数入参、返回值是否有必要的校验
4. **代码质量**：是否有重复代码、命名不清晰、过于复杂的函数
5. **性能问题**：是否有明显的性能隐患（如循环内重复计算、不必要的数据库查询）
6. **安全问题**：是否有注入风险、敏感信息泄露、权限控制问题

## 输出格式
请使用以下格式输出，语言为中文：

### 📊 审查概览
- 整体评价：（通过 / 有建议 / 需修改）
- 严重问题数量：X 个
- 一般建议数量：X 个

### ❌ 需修改的问题（严重）
（如果没有严重问题，写"无"）
1. **[文件: 行号] 问题标题**
   - 问题描述：...
   - 修改建议：...

### 💡 改进建议（一般）
（如果没有建议，写"无"）
1. **[文件: 行号] 建议标题**
   - 具体内容：...

### ✅ 良好实践
（可选，列出值得肯定的地方）

请务必基于代码实际内容审查，不要编造不存在的问题。如果代码质量良好，直接给出通过结论即可。
"""
    return prompt


def call_llm_for_review(
    diff_text: str,
    pr_number: int,
    file_list: list,
    llm_base_url: str,
    llm_api_key: str,
    llm_model: str,
    coze_bot_id: str = "",
) -> Optional[str]:
    """
    调用 LLM 进行代码审查，返回审查结果文本。
    失败时返回 None。
    根据 LLM_PROVIDER 环境变量选择调用方式。
    """
    prompt = build_review_prompt(diff_text, pr_number, file_list)
    logger.info(f"Prompt 长度: {len(prompt)} 字符")

    provider = LLM_PROVIDER

    if provider == "coze":
        return call_llm_coze(prompt, llm_base_url, llm_api_key, llm_model, coze_bot_id)
    else:
        # 默认 OpenAI 兼容
        return call_llm_openai(prompt, llm_base_url, llm_api_key, llm_model)


# ============== 主流程 ==============
def main():
    parser = argparse.ArgumentParser(description="CI AI 代码审查脚本")
    parser.add_argument("--pr", type=int, help="PR 编号（也可通过 PR_NUMBER 环境变量）")
    parser.add_argument("--repo", type=str, help="仓库名 owner/repo（也可通过 REPO_NAME 环境变量）")
    parser.add_argument("--gitea-url", type=str, help="Gitea 地址（也可通过 GITEA_API_URL 环境变量）")
    parser.add_argument("--gitea-token", type=str, help="Gitea Token（也可通过 GITEA_TOKEN 环境变量）")
    parser.add_argument("--dry-run", action="store_true", help="只输出审查结果，不发表评论")
    args = parser.parse_args()

    # 读取配置
    gitea_url = args.gitea_url or os.getenv("GITEA_API_URL") or os.getenv("GITEA_SERVER_URL")
    gitea_token = args.gitea_token or os.getenv("GITEA_TOKEN")
    repo_name = args.repo or os.getenv("REPO_NAME") or os.getenv("GITEA_REPO")
    pr_number = args.pr or int(os.getenv("PR_NUMBER") or os.getenv("GITEA_PR_NUMBER") or 0)

    llm_base_url = os.getenv("LLM_BASE_URL")
    llm_api_key = os.getenv("LLM_API_KEY")
    llm_model = os.getenv("LLM_MODEL", "")
    coze_bot_id = os.getenv("COZE_BOT_ID", os.getenv("COZE_BOTID", ""))

    # 根据 provider 设置默认值
    provider = LLM_PROVIDER
    if provider == "coze":
        # 扣子模式：默认国内站，key 兼容多种环境变量名
        if not llm_base_url:
            llm_base_url = "https://api.coze.cn"
        if not llm_api_key:
            llm_api_key = os.getenv("COZE_API_KEY", "") or os.getenv("COZE_PAT", "")
    else:
        # OpenAI兼容模式：默认模型
        if not llm_model:
            llm_model = "gpt-4o-mini"

    # 必要参数校验
    missing = []
    if not gitea_url:
        missing.append("GITEA_API_URL")
    if not gitea_token:
        missing.append("GITEA_TOKEN")
    if not repo_name:
        missing.append("REPO_NAME")
    if not pr_number:
        missing.append("PR_NUMBER")
    if not llm_base_url:
        missing.append("LLM_BASE_URL")
    if not llm_api_key:
        missing.append("LLM_API_KEY")
    if provider == "coze" and not coze_bot_id:
        missing.append("COZE_BOT_ID (扣子模式需要)")

    if missing:
        logger.error(f"缺少必要配置: {', '.join(missing)}")
        sys.exit(1)

    logger.info(f"开始审查 PR #{pr_number}，仓库: {repo_name}")
    logger.info(f"Gitea: {gitea_url}")
    logger.info(f"LLM: {llm_base_url} (model={llm_model})")

    try:
        # 1. 初始化 Gitea 客户端
        gitea = GiteaClient(gitea_url, gitea_token, repo_name)

        # 2. 获取 PR diff 和文件列表
        try:
            diff_text = gitea.get_pr_diff(pr_number)
            file_list = gitea.get_pr_files(pr_number)
        except Exception as e:
            logger.error(f"获取 PR 信息失败: {e}")
            sys.exit(1)

        # 3. 过滤掉不需要审查的文件（如 lock 文件、生成的文件、二进制文件等）
        skip_extensions = (
            ".lock",
            ".sum",
            ".min.js",
            ".min.css",
            ".map",
            ".png",
            ".jpg",
            ".jpeg",
            ".gif",
            ".svg",
            ".ico",
            ".woff",
            ".woff2",
            ".ttf",
            ".eot",
        )
        skipped_files = []
        if file_list:
            skipped_files = [
                f.get("filename")
                for f in file_list
                if f.get("filename", "").endswith(skip_extensions) or f.get("status") == "removed"
            ]
            if skipped_files:
                logger.info(f"跳过 {len(skipped_files)} 个非文本/已删除文件: {', '.join(skipped_files[:5])}...")

        # 实际从 diff 中移除跳过的文件（按文件边界切割）
        if skipped_files:
            diff_lines = diff_text.split("\n")
            filtered_lines = []
            current_file = None
            skip_current = False
            i = 0
            while i < len(diff_lines):
                line = diff_lines[i]
                # 检测新文件开始: diff --git a/xxx b/xxx
                if line.startswith("diff --git "):
                    # 提取文件名
                    parts = line.split(" ")
                    if len(parts) >= 4:
                        # b/ 后面的是目标文件名
                        current_file = parts[3][2:] if parts[3].startswith("b/") else parts[3]
                        skip_current = any(current_file == sf for sf in skipped_files) or any(
                            current_file.endswith(ext) for ext in skip_extensions
                        )
                    else:
                        skip_current = False
                if not skip_current:
                    filtered_lines.append(line)
                i += 1
            original_len = len(diff_text)
            diff_text = "\n".join(filtered_lines)
            logger.info(f"Diff 过滤后: {original_len} -> {len(diff_text)} 字符 (减少 {original_len - len(diff_text)})")

        # 4. 截断过大的 diff
        diff_text, was_truncated = truncate_diff(diff_text, MAX_DIFF_CHARS)
        if was_truncated:
            logger.warning(f"Diff 过大，已截断至 {len(diff_text)} 字符")

        # 5. 如果 diff 为空，直接跳过
        if not diff_text.strip():
            logger.info("Diff 为空，无需审查")
            sys.exit(0)

        # 6. 调用 LLM 审查
        review_result = call_llm_for_review(
            diff_text=diff_text,
            pr_number=pr_number,
            file_list=file_list,
            llm_base_url=llm_base_url,
            llm_api_key=llm_api_key,
            llm_model=llm_model,
            coze_bot_id=coze_bot_id,
        )

        if not review_result:
            logger.error("LLM 审查失败")
            sys.exit(1)

        # 7. 加上审查时间和标识（便于识别是自动审查）
        from datetime import datetime

        timestamp = datetime.now().strftime("%Y-%m-%d %H:%M:%S")
        marker = "<!-- AI_CODE_REVIEW_AUTO_COMMENT -->"
        full_comment = f"""{review_result}

---
<sub>🤖 由 AI 代码审查机器人自动生成 | {timestamp} | 模型: {llm_model}</sub>

{marker}
"""

        # 8. 输出审查结果到日志
        logger.info("=" * 60)
        logger.info("审查结果:")
        for line in review_result.split("\n")[:30]:
            logger.info(line)
        if len(review_result.split("\n")) > 30:
            logger.info(f"... 共 {len(review_result.split(chr(10)))} 行")
        logger.info("=" * 60)

        # 9. 发布评论（先删除旧的审查评论，避免刷屏）
        if args.dry_run:
            logger.info("--dry-run 模式，跳过发布评论")
            print(full_comment)
        else:
            # 去重：删除之前的 AI 审查评论
            old_comments = gitea.get_existing_review_comments(pr_number, marker)
            if old_comments:
                logger.info(f"找到 {len(old_comments)} 条旧的 AI 审查评论，先删除")
                for cid in old_comments:
                    gitea.delete_pr_comment(pr_number, cid)
            # 发布新评论
            success = gitea.post_pr_comment(pr_number, full_comment)
            if not success:
                logger.error("评论发布失败")
                sys.exit(1)

        # 10. 判断是否有严重问题（可选阻断）
        # 目前只做建议，不阻断合并，始终返回 0
        has_critical = "问题" in review_result and ("❌" in review_result or "需修改" in review_result)
        if has_critical:
            logger.warning("检测到需修改的问题，但当前配置为仅建议，不阻断合并")

        logger.info("代码审查完成")
        sys.exit(0)

    except Exception as e:
        logger.exception(f"审查脚本发生未预期的异常: {e}")
        sys.exit(1)


if __name__ == "__main__":
    main()