567 lines
20 KiB
Python
567 lines
20 KiB
Python
#!/usr/bin/env python3
|
||
"""
|
||
CI Code Review Script
|
||
- 从 Gitea 获取 PR diff
|
||
- 调用 LLM 进行代码审查
|
||
- 将审查结果写回 PR 评论
|
||
"""
|
||
|
||
import os
|
||
import sys
|
||
import json
|
||
import logging
|
||
import argparse
|
||
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 上已有的审查评论（带标识），用于后续更新或删除旧评论。
|
||
"""
|
||
url = self._api_url(f"issues/{pr_number}/comments")
|
||
resp = self.session.get(url, timeout=GITEA_TIMEOUT)
|
||
if resp.status_code != 200:
|
||
return []
|
||
|
||
comments = resp.json()
|
||
return [c for c in comments if marker in c.get("body", "")]
|
||
|
||
|
||
# ============== LLM 调用 ==============
|
||
def build_review_prompt(diff_text: str, pr_number: int, file_list: list) -> str:
|
||
"""构建代码审查的 Prompt"""
|
||
file_names = [f.get("filename", "") for f in file_list] if file_list else []
|
||
files_summary = ", ".join(file_names[:10]) if file_names else "未知"
|
||
if len(file_names) > 10:
|
||
files_summary += f" 等 {len(file_names)} 个文件"
|
||
|
||
prompt = f"""你是一位资深代码审查专家，请对以下 Pull Request 的代码变更进行严格审查。
|
||
|
||
**PR 信息：**
|
||
- PR 编号：#{pr_number}
|
||
- 修改文件：{files_summary}
|
||
|
||
**审查重点：**
|
||
1. **严重问题**：逻辑错误、潜在 Bug、安全漏洞、数据不一致、空指针、资源泄漏、并发问题等
|
||
2. **代码质量**：边界条件处理、错误处理是否完善、异常场景覆盖
|
||
3. **性能隐患**：明显的性能问题、低效算法、不必要的重复计算
|
||
4. **最佳实践**：代码规范、可读性、可维护性、命名是否清晰
|
||
|
||
**审查原则：**
|
||
- 只针对变更的代码（diff）进行审查，不要审查未改动的代码
|
||
- 严重问题必须指出具体文件名和大致行号（根据 diff 中的行号推断）
|
||
- 给出明确、可操作的建议，不要空泛
|
||
- 如果代码质量很好、没有明显问题，也请如实说明
|
||
- 用中文回复
|
||
|
||
**输出格式要求（严格遵守，不要输出格式以外的内容）：**
|
||
|
||
## 代码审查结果 - PR #{pr_number}
|
||
|
||
### ⚠️ 问题（N个需要修改）
|
||
1. **文件名 第X行**：问题描述（说明原因和可能的影响）
|
||
2. **文件名 第X行**：问题描述
|
||
|
||
### 💡 建议（N个可选）
|
||
1. 建议描述（可选优化、代码风格等）
|
||
|
||
---
|
||
✅ 格式检查通过 | ❌ 逻辑审查需修改 | ⚠️ 建议关注性能
|
||
|
||
**说明：** 底部的三个状态标签，根据审查结果勾选或取消对应标记（用 ✅/❌/⚠️ 表示）：
|
||
- 格式检查：代码格式、命名规范等是否达标
|
||
- 逻辑审查：是否存在必须修改的逻辑问题
|
||
- 性能：是否存在需要关注的性能问题
|
||
|
||
**以下是代码 diff 内容：**
|
||
|
||
```diff
|
||
{diff_text}
|
||
```
|
||
"""
|
||
return prompt
|
||
|
||
|
||
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 调用"""
|
||
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()
|
||
|
||
# 解析 Coze 返回格式，兼容多种可能的返回结构
|
||
content = None
|
||
# 方式1: data.messages 数组
|
||
messages = data.get("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
|
||
|
||
# 方式2: 直接 content 字段
|
||
if not content:
|
||
content = data.get("data", {}).get("content") or data.get("content")
|
||
|
||
# 方式3: choices 格式（兼容）
|
||
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 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)}")
|
||
# 审查失败不阻断 CI，返回 0
|
||
logger.info("审查脚本因配置缺失而跳过，退出码 0")
|
||
sys.exit(0)
|
||
|
||
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}")
|
||
logger.info("审查脚本异常退出，退出码 0（不阻断 CI）")
|
||
sys.exit(0)
|
||
|
||
# 3. 过滤掉不需要审查的文件（如 lock 文件、生成的文件等）
|
||
skip_extensions = (".lock", ".sum", ".min.js", ".min.css", ".map", ".png", ".jpg", ".jpeg", ".gif", ".svg", ".ico", ".woff", ".woff2", ".ttf", ".eot")
|
||
if file_list:
|
||
skipped = [f.get("filename") for f in file_list
|
||
if f.get("filename", "").endswith(skip_extensions)
|
||
or f.get("status") == "removed"]
|
||
if skipped:
|
||
logger.info(f"跳过 {len(skipped)} 个非文本/已删除文件: {', '.join(skipped[:5])}...")
|
||
|
||
# 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 审查失败，跳过发布评论")
|
||
logger.info("审查脚本异常退出，退出码 0（不阻断 CI）")
|
||
sys.exit(0)
|
||
|
||
# 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:
|
||
success = gitea.post_pr_comment(pr_number, full_comment)
|
||
if not success:
|
||
logger.warning("评论发布失败，但不影响 CI 通过")
|
||
|
||
# 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}")
|
||
# 任何异常都不阻断 CI
|
||
logger.info("审查脚本异常退出，退出码 0（不阻断 CI）")
|
||
sys.exit(0)
|
||
|
||
|
||
if __name__ == "__main__":
|
||
main()
|