@@ -0,0 +1,627 @@
#!/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 ( " \n diff --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 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 ( )