zyc
aab0312cec
All checks were successful
Build and Deploy Log Center / build-and-deploy (push) Successful in 1m35s
168 lines
4.4 KiB
Python
168 lines
4.4 KiB
Python
"""
|
||
Claude Service - 调用 Claude Code CLI
|
||
"""
|
||
import subprocess
|
||
from typing import Optional
|
||
from loguru import logger
|
||
|
||
from ..config import settings
|
||
from ..models import Bug
|
||
|
||
|
||
class ClaudeService:
|
||
"""通过 CLI 调用 Claude Code"""
|
||
|
||
def __init__(self):
|
||
self.cli_path = settings.claude_cli_path
|
||
self.timeout = settings.claude_timeout
|
||
|
||
def execute_prompt(
|
||
self,
|
||
prompt: str,
|
||
cwd: str,
|
||
tools: str = "Edit,Read,Write",
|
||
) -> tuple[bool, str]:
|
||
"""
|
||
执行 Claude CLI 命令
|
||
|
||
Args:
|
||
prompt: 提示词
|
||
cwd: 工作目录
|
||
tools: 可用工具
|
||
|
||
Returns:
|
||
(成功与否, 输出内容)
|
||
"""
|
||
try:
|
||
cmd = [
|
||
self.cli_path,
|
||
"-p", prompt,
|
||
"--tools", tools,
|
||
"--dangerously-skip-permissions",
|
||
]
|
||
|
||
logger.debug(f"执行 Claude CLI: {' '.join(cmd[:3])}...")
|
||
|
||
result = subprocess.run(
|
||
cmd,
|
||
cwd=cwd,
|
||
capture_output=True,
|
||
text=True,
|
||
timeout=self.timeout,
|
||
)
|
||
|
||
output = result.stdout + result.stderr
|
||
|
||
if result.returncode == 0:
|
||
logger.info("Claude CLI 执行成功")
|
||
return True, output.strip()
|
||
else:
|
||
logger.error(f"Claude CLI 执行失败: {output}")
|
||
return False, output.strip()
|
||
|
||
except subprocess.TimeoutExpired:
|
||
logger.error(f"Claude CLI 超时 ({self.timeout}秒)")
|
||
return False, "执行超时"
|
||
except FileNotFoundError:
|
||
logger.error(f"Claude CLI 未找到: {self.cli_path}")
|
||
return False, "Claude CLI 未安装"
|
||
except Exception as e:
|
||
logger.error(f"Claude CLI 执行异常: {e}")
|
||
return False, str(e)
|
||
|
||
def batch_fix_bugs(
|
||
self,
|
||
bugs: list[Bug],
|
||
project_path: str,
|
||
) -> tuple[bool, str]:
|
||
"""
|
||
批量修复多个 Bug
|
||
|
||
Args:
|
||
bugs: Bug 列表
|
||
project_path: 项目路径
|
||
|
||
Returns:
|
||
(成功与否, Claude 输出)
|
||
"""
|
||
if not bugs:
|
||
return False, "没有需要修复的 Bug"
|
||
|
||
# 构造批量修复 Prompt
|
||
prompt_parts = [
|
||
f"请修复以下 {len(bugs)} 个 Bug：",
|
||
"",
|
||
]
|
||
|
||
for bug in bugs:
|
||
prompt_parts.append(bug.format_for_prompt())
|
||
prompt_parts.append("")
|
||
|
||
prompt_parts.extend([
|
||
"## 修复要求",
|
||
"1. 依次修复上述所有 Bug",
|
||
"2. 每个 Bug 只做最小必要的改动",
|
||
"3. 确保不破坏现有功能",
|
||
"4. 修复完成后简要说明每个 Bug 的修复方式",
|
||
"",
|
||
"请开始修复。",
|
||
])
|
||
|
||
prompt = "\n".join(prompt_parts)
|
||
|
||
logger.info(f"开始批量修复 {len(bugs)} 个 Bug...")
|
||
return self.execute_prompt(prompt, project_path)
|
||
|
||
def analyze_bug(self, bug: Bug, project_path: str) -> tuple[bool, str]:
|
||
"""
|
||
分析单个 Bug（不修复）
|
||
|
||
Args:
|
||
bug: Bug 对象
|
||
project_path: 项目路径
|
||
|
||
Returns:
|
||
(成功与否, 分析结果)
|
||
"""
|
||
prompt = f"""
|
||
请分析以下 Bug，但不要修改任何代码：
|
||
|
||
{bug.format_for_prompt()}
|
||
|
||
请分析：
|
||
1. 错误的根本原因
|
||
2. 建议的修复方案
|
||
3. 可能影响的其他文件
|
||
"""
|
||
|
||
return self.execute_prompt(prompt, project_path, tools="Read")
|
||
|
||
def review_changes(self, diff: str, project_path: str) -> tuple[bool, str]:
|
||
"""
|
||
审核代码变更
|
||
|
||
Args:
|
||
diff: Git diff 内容
|
||
project_path: 项目路径
|
||
|
||
Returns:
|
||
(是否通过, 审核意见)
|
||
"""
|
||
prompt = f"""
|
||
请审核以下代码变更是否安全：
|
||
|
||
```diff
|
||
{diff}
|
||
```
|
||
|
||
审核要点：
|
||
1. 是否修复了目标问题
|
||
2. 是否引入了新的 Bug
|
||
3. 是否破坏了原有业务逻辑
|
||
4. 是否有安全风险
|
||
|
||
如果审核通过，回复 "APPROVED"。否则说明问题。
|
||
"""
|
||
|
||
return self.execute_prompt(prompt, project_path, tools="Read")
|