From 2c5281e6d902229816dcc7a289a1984fba69daf0 Mon Sep 17 00:00:00 2001 From: xiaoxia Date: Tue, 28 Jul 2026 08:20:14 +0800 Subject: [PATCH 1/4] =?UTF-8?q?fix(ci):=20=E9=98=B2=E5=BE=AA=E7=8E=AF?= =?UTF-8?q?=E6=A3=80=E6=B5=8B=E5=8F=96=E6=9C=80=E6=96=B0commit=20-=20API?= =?UTF-8?q?=E8=BF=94=E5=9B=9E=E5=80=92=E5=BA=8F=E7=94=A8[0]=E8=80=8C?= =?UTF-8?q?=E9=9D=9E[-1]?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Gitea PR commits API返回倒序（最新在前），commits[-1]取到最早的commit， 导致防循环检测失效。修复为commits[0]取最新commit。 --- scripts/ci/auto_fix_formatting.py | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/scripts/ci/auto_fix_formatting.py b/scripts/ci/auto_fix_formatting.py index 1d927af58..bc32c5308 100755 --- a/scripts/ci/auto_fix_formatting.py +++ b/scripts/ci/auto_fix_formatting.py @@ -258,7 +258,7 @@ def main(): req_commits = urllib.request.Request(commits_url, headers={"Authorization": f"token {token}"}) with urllib.request.urlopen(req_commits) as resp_commits: commits = json.loads(resp_commits.read()) - latest_msg = commits[-1].get("commit", {}).get("message", "") if commits else "" + latest_msg = commits[0].get("commit", {}).get("message", "") if commits else "" if skip_marker in latest_msg: print(f"检测到最新commit包含 {skip_marker} 标记，跳过格式修复（防循环）") print("本次格式检查失败是格式修复commit触发的CI回跑，属正常现象") -- 2.54.0 From 09c878621d079fa0aa6c40814064bd45d73181ff Mon Sep 17 00:00:00 2001 From: xiaoxia Date: Tue, 28 Jul 2026 08:58:33 +0800 Subject: [PATCH 2/4] =?UTF-8?q?fix(ci):=20AI=20Code=20Review=20fail-open?= =?UTF-8?q?=20-=20LLM=E5=A4=B1=E8=B4=A5/=E5=BC=82=E5=B8=B8=E6=97=B6?= =?UTF-8?q?=E4=B8=8D=E9=98=BB=E5=A1=9E=E5=90=88=E5=B9=B6?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 修复两处违反fail-open原则的bug： 1. LLM调用失败返回None时 exit 1 → exit 0 2. 未捕获异常时 exit 1 → exit 0 fail-open原则：AI审查是辅助手段，不能因自身故障阻塞正常开发流程。 --- scripts/ci_code_review.py | 780 -------------------------------------- 1 file changed, 780 deletions(-) diff --git a/scripts/ci_code_review.py b/scripts/ci_code_review.py index 1ec5e8330..e69de29bb 100644 --- a/scripts/ci_code_review.py +++ b/scripts/ci_code_review.py @@ -1,780 +0,0 @@ -#!/usr/bin/env python3 -""" -CI Code Review Script -- 从 Gitea 获取 PR diff -- 调用 LLM 进行代码审查 -- 将审查结果写回 PR 评论 -""" - -import argparse -import json -import logging -import os -import re -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 create_commit_status( - self, sha: str, state: str, context: str, description: str = "", target_url: str = "" - ) -> bool: - """ - 给指定 commit 打 status。 - state: pending / success / failure / error / warning - Gitea API: POST /repos/{owner}/{repo}/statuses/{sha} - """ - url = self._api_url(f"statuses/{sha}") - logger.info(f"设置 commit status: sha={sha[:12]}..., state={state}, context={context}") - - payload = { - "state": state, - "context": context, - "description": description[:200] if description else "", - } - if target_url: - payload["target_url"] = target_url - - resp = self.session.post( - url, - data=json.dumps(payload), - timeout=GITEA_TIMEOUT, - ) - if resp.status_code not in (200, 201): - logger.error(f"设置 status 失败: HTTP {resp.status_code} - {resp.text[:200]}") - return False - - logger.info(f"Status 设置成功: {context} = {state}") - 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} -``` - -## 审查要求 -请从以下维度进行审查，重点关注**阻塞级问题**： - -### 问题分级标准 -- **🔴 阻塞级（BLOCKER）**：必须修复，否则不允许合并。包括： - 1. **明显逻辑bug**：条件判断错误、死循环、返回值错误、空指针/None引用未处理、边界条件遗漏导致功能异常 - 2. **安全漏洞**：SQL注入、XSS、命令注入、敏感信息明文存储/泄露、权限绕过、认证缺失 - 3. **语法错误**：代码存在语法层面的错误，无法运行 - 4. **数据损坏风险**：可能导致数据丢失、数据不一致、脏数据写入的问题 - -- **💡 建议级（SUGGESTION）**：不阻塞合并，仅供参考改进。包括： - 1. 命名不规范、代码风格问题 - 2. 最佳实践建议、设计模式优化 - 3. 格式问题（缩进、空行、import顺序等） - 4. 代码可读性改进、注释补充 - 5. 非关键路径的轻微性能优化建议 - 6. 重复代码、过长函数等代码质量问题 - -1. **逻辑正确性**：是否有明显的逻辑错误、边界条件遗漏、空指针/None引用风险 -2. **异常处理**：异常捕获是否合理，是否有裸except，错误处理是否完善 -3. **参数校验**：函数入参、返回值是否有必要的校验 -4. **代码质量**：是否有重复代码、命名不清晰、过于复杂的函数 -5. **性能问题**：是否有明显的性能隐患（如循环内重复计算、不必要的数据库查询） -6. **安全问题**：是否有注入风险、敏感信息泄露、权限控制问题 - -## 输出格式 -请使用以下格式输出，语言为中文。**必须严格按照格式输出，尤其是【阻塞级判定】部分**： - -### 【阻塞级判定】 -- 是否存在阻塞级问题：（是 / 否） -- 阻塞级问题数量：X 个 - -### 📊 审查概览 -- 整体评价：（通过 / 有建议 / 需修改） -- 建议级问题数量：X 个 - -### 🔴 阻塞级问题（必须修复） -（如果没有阻塞级问题，写"无"） -1. **[文件: 行号] 问题标题** - - 问题类型：（逻辑bug / 安全漏洞 / 语法错误 / 数据损坏风险） - - 问题描述：... - - 修改建议：... - -### 💡 改进建议（不阻塞合并） -（如果没有建议，写"无"） -1. **[文件: 行号] 建议标题** - - 具体内容：... - -### ✅ 良好实践 -（可选，列出值得肯定的地方） - -请务必基于代码实际内容审查，不要编造不存在的问题。如果代码质量良好，直接给出通过结论即可。 -**重要：【阻塞级判定】必须准确，只有确实存在严重问题时才写"是"。** -""" - return prompt - - -def parse_blocker_result(review_text: str) -> Tuple[bool, int]: - """ - 从审查结果中解析是否存在阻塞级问题。 - 返回 (has_blocker, blocker_count) - """ - # 先找【阻塞级判定】部分的明确标记 - pattern = r"【阻塞级判定】[\s\S]*?是否存在阻塞级问题[：:]\s*(是|否)" - match = re.search(pattern, review_text) - if match: - has_blocker = match.group(1) == "是" - else: - # fallback 1: 找"阻塞级问题数量" - count_pattern = r"阻塞级问题数量[：:]\s*(\d+)" - count_match = re.search(count_pattern, review_text) - if count_match: - has_blocker = int(count_match.group(1)) > 0 - else: - # fallback 2: 检查是否有"阻塞级问题"section且内容不是"无" - has_blocker = False - blocker_section = re.search(r"### 🔴 阻塞级问题[\s\S]*?(?=### |\Z)", review_text) - if blocker_section: - section_text = blocker_section.group(0) - # 如果有编号列表项，说明有问题 - if re.search(r"\d+\.\s*\*\*", section_text): - has_blocker = True - - # 提取数量 - count_pattern = r"阻塞级问题数量[：:]\s*(\d+)" - count_match = re.search(count_pattern, review_text) - blocker_count = int(count_match.group(1)) if count_match else (1 if has_blocker else 0) - - logger.info(f"阻塞级问题解析: 存在={has_blocker}, 数量={blocker_count}") - return has_blocker, blocker_count - - -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 = "" - full_comment = f"""{review_result} - ---- -🤖 由 AI 代码审查机器人自动生成 | {timestamp} | 模型: {llm_model} - -{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. 解析阻塞级问题，用退出码决定 job 状态 - # 有阻塞级问题 → exit 1 → job失败 → Gitea自动打failure status → 门禁拦截 - # 无阻塞级问题 → exit 0 → job成功 → Gitea自动打success status - # LLM调用失败等异常 → exit 0 → fail-open，不阻塞正常开发 - has_blocker, blocker_count = parse_blocker_result(review_result) - - if has_blocker: - logger.error(f"检测到 {blocker_count} 个阻塞级问题，审查不通过") - logger.info("代码审查完成（失败）") - sys.exit(1) - else: - logger.info("无阻塞级问题，审查通过") - logger.info("代码审查完成（通过）") - sys.exit(0) - - except Exception as e: - logger.exception(f"审查脚本发生未预期的异常: {e}") - sys.exit(1) - - -if __name__ == "__main__": - main() -- 2.54.0 From 424c7362757a61f090b4a315e5d90721e62feb52 Mon Sep 17 00:00:00 2001 From: xiaoxia Date: Tue, 28 Jul 2026 08:59:08 +0800 Subject: [PATCH 3/4] =?UTF-8?q?fix(ci):=20AI=20Code=20Review=20fail-open?= =?UTF-8?q?=20-=20LLM=E5=A4=B1=E8=B4=A5/=E5=BC=82=E5=B8=B8=E6=97=B6?= =?UTF-8?q?=E4=B8=8D=E9=98=BB=E5=A1=9E=E5=90=88=E5=B9=B6?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 修复两处违反fail-open原则的bug： 1. LLM调用失败返回None时 exit 1 → exit 0 2. 未捕获异常时 exit 1 → exit 0 fail-open原则：AI审查是辅助手段，不能因自身故障阻塞正常开发。 -- 2.54.0 From 71d062ba63671f8d29f4cd79ad09760bbc66c714 Mon Sep 17 00:00:00 2001 From: xiaoxia Date: Tue, 28 Jul 2026 08:59:40 +0800 Subject: [PATCH 4/4] =?UTF-8?q?fix(ci):=20AI=20Code=20Review=20fail-open?= =?UTF-8?q?=20-=20LLM=E5=A4=B1=E8=B4=A5/=E5=BC=82=E5=B8=B8=E6=97=B6?= =?UTF-8?q?=E4=B8=8D=E9=98=BB=E5=A1=9E=E5=90=88=E5=B9=B6?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 修复两处违反fail-open原则的bug： 1. LLM调用失败返回None时 exit 1 → exit 0 2. 未捕获异常时 exit 1 → exit 0 fail-open原则：AI审查是辅助手段，不能因自身故障阻塞正常开发。 --- scripts/ci_code_review.py | 780 ++++++++++++++++++++++++++++++++++++++ 1 file changed, 780 insertions(+) diff --git a/scripts/ci_code_review.py b/scripts/ci_code_review.py index e69de29bb..c67bf88d9 100644 --- a/scripts/ci_code_review.py +++ b/scripts/ci_code_review.py @@ -0,0 +1,780 @@ +#!/usr/bin/env python3 +""" +CI Code Review Script +- 从 Gitea 获取 PR diff +- 调用 LLM 进行代码审查 +- 将审查结果写回 PR 评论 +""" + +import argparse +import json +import logging +import os +import re +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 create_commit_status( + self, sha: str, state: str, context: str, description: str = "", target_url: str = "" + ) -> bool: + """ + 给指定 commit 打 status。 + state: pending / success / failure / error / warning + Gitea API: POST /repos/{owner}/{repo}/statuses/{sha} + """ + url = self._api_url(f"statuses/{sha}") + logger.info(f"设置 commit status: sha={sha[:12]}..., state={state}, context={context}") + + payload = { + "state": state, + "context": context, + "description": description[:200] if description else "", + } + if target_url: + payload["target_url"] = target_url + + resp = self.session.post( + url, + data=json.dumps(payload), + timeout=GITEA_TIMEOUT, + ) + if resp.status_code not in (200, 201): + logger.error(f"设置 status 失败: HTTP {resp.status_code} - {resp.text[:200]}") + return False + + logger.info(f"Status 设置成功: {context} = {state}") + 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} +``` + +## 审查要求 +请从以下维度进行审查，重点关注**阻塞级问题**： + +### 问题分级标准 +- **🔴 阻塞级（BLOCKER）**：必须修复，否则不允许合并。包括： + 1. **明显逻辑bug**：条件判断错误、死循环、返回值错误、空指针/None引用未处理、边界条件遗漏导致功能异常 + 2. **安全漏洞**：SQL注入、XSS、命令注入、敏感信息明文存储/泄露、权限绕过、认证缺失 + 3. **语法错误**：代码存在语法层面的错误，无法运行 + 4. **数据损坏风险**：可能导致数据丢失、数据不一致、脏数据写入的问题 + +- **💡 建议级（SUGGESTION）**：不阻塞合并，仅供参考改进。包括： + 1. 命名不规范、代码风格问题 + 2. 最佳实践建议、设计模式优化 + 3. 格式问题（缩进、空行、import顺序等） + 4. 代码可读性改进、注释补充 + 5. 非关键路径的轻微性能优化建议 + 6. 重复代码、过长函数等代码质量问题 + +1. **逻辑正确性**：是否有明显的逻辑错误、边界条件遗漏、空指针/None引用风险 +2. **异常处理**：异常捕获是否合理，是否有裸except，错误处理是否完善 +3. **参数校验**：函数入参、返回值是否有必要的校验 +4. **代码质量**：是否有重复代码、命名不清晰、过于复杂的函数 +5. **性能问题**：是否有明显的性能隐患（如循环内重复计算、不必要的数据库查询） +6. **安全问题**：是否有注入风险、敏感信息泄露、权限控制问题 + +## 输出格式 +请使用以下格式输出，语言为中文。**必须严格按照格式输出，尤其是【阻塞级判定】部分**： + +### 【阻塞级判定】 +- 是否存在阻塞级问题：（是 / 否） +- 阻塞级问题数量：X 个 + +### 📊 审查概览 +- 整体评价：（通过 / 有建议 / 需修改） +- 建议级问题数量：X 个 + +### 🔴 阻塞级问题（必须修复） +（如果没有阻塞级问题，写"无"） +1. **[文件: 行号] 问题标题** + - 问题类型：（逻辑bug / 安全漏洞 / 语法错误 / 数据损坏风险） + - 问题描述：... + - 修改建议：... + +### 💡 改进建议（不阻塞合并） +（如果没有建议，写"无"） +1. **[文件: 行号] 建议标题** + - 具体内容：... + +### ✅ 良好实践 +（可选，列出值得肯定的地方） + +请务必基于代码实际内容审查，不要编造不存在的问题。如果代码质量良好，直接给出通过结论即可。 +**重要：【阻塞级判定】必须准确，只有确实存在严重问题时才写"是"。** +""" + return prompt + + +def parse_blocker_result(review_text: str) -> Tuple[bool, int]: + """ + 从审查结果中解析是否存在阻塞级问题。 + 返回 (has_blocker, blocker_count) + """ + # 先找【阻塞级判定】部分的明确标记 + pattern = r"【阻塞级判定】[\s\S]*?是否存在阻塞级问题[：:]\s*(是|否)" + match = re.search(pattern, review_text) + if match: + has_blocker = match.group(1) == "是" + else: + # fallback 1: 找"阻塞级问题数量" + count_pattern = r"阻塞级问题数量[：:]\s*(\d+)" + count_match = re.search(count_pattern, review_text) + if count_match: + has_blocker = int(count_match.group(1)) > 0 + else: + # fallback 2: 检查是否有"阻塞级问题"section且内容不是"无" + has_blocker = False + blocker_section = re.search(r"### 🔴 阻塞级问题[\s\S]*?(?=### |\Z)", review_text) + if blocker_section: + section_text = blocker_section.group(0) + # 如果有编号列表项，说明有问题 + if re.search(r"\d+\.\s*\*\*", section_text): + has_blocker = True + + # 提取数量 + count_pattern = r"阻塞级问题数量[：:]\s*(\d+)" + count_match = re.search(count_pattern, review_text) + blocker_count = int(count_match.group(1)) if count_match else (1 if has_blocker else 0) + + logger.info(f"阻塞级问题解析: 存在={has_blocker}, 数量={blocker_count}") + return has_blocker, blocker_count + + +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(0) # fail-open: LLM调用失败不阻塞合并 + + # 7. 加上审查时间和标识（便于识别是自动审查） + from datetime import datetime + + timestamp = datetime.now().strftime("%Y-%m-%d %H:%M:%S") + marker = "" + full_comment = f"""{review_result} + +--- +🤖 由 AI 代码审查机器人自动生成 | {timestamp} | 模型: {llm_model} + +{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. 解析阻塞级问题，用退出码决定 job 状态 + # 有阻塞级问题 → exit 1 → job失败 → Gitea自动打failure status → 门禁拦截 + # 无阻塞级问题 → exit 0 → job成功 → Gitea自动打success status + # LLM调用失败等异常 → exit 0 → fail-open，不阻塞正常开发 + has_blocker, blocker_count = parse_blocker_result(review_result) + + if has_blocker: + logger.error(f"检测到 {blocker_count} 个阻塞级问题，审查不通过") + logger.info("代码审查完成（失败）") + sys.exit(1) + else: + logger.info("无阻塞级问题，审查通过") + logger.info("代码审查完成（通过）") + sys.exit(0) + + except Exception as e: + logger.exception(f"审查脚本发生未预期的异常: {e}") + sys.exit(0) # fail-open: 异常不阻塞正常开发 + + +if __name__ == "__main__": + main() -- 2.54.0