""" 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, allowed_tools: Optional[str] = None, ) -> tuple[bool, str]: """ 执行 Claude CLI 命令 Args: prompt: 提示词 cwd: 工作目录 allowed_tools: 可选，限制可用工具（逗号分隔）。 为 None 时配合 --dangerously-skip-permissions 允许所有工具。 Returns: (成功与否, 输出内容) """ try: cmd = [ self.cli_path, "-p", prompt, "--dangerously-skip-permissions", ] if allowed_tools: cmd.extend(["--allowedTools", allowed_tools]) logger.debug(f"执行 Claude CLI (cwd={cwd})") logger.debug(f"Prompt 长度: {len(prompt)} 字符") 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 执行成功") logger.debug(f"输出前 500 字符: {output[:500]}") return True, output.strip() else: logger.error(f"Claude CLI 执行失败 (code={result.returncode}): {output[:500]}") 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" bug_ids = "_".join(str(b.id) for b in bugs) test_file = f"repair_test_bug_{bug_ids}.py" # 构造批量修复 Prompt prompt_parts = [ f"你是一个自动化 Bug 修复代理。请直接修复以下 {len(bugs)} 个 Bug。", "", "重要：你必须使用工具（Read、Edit、Write 等）直接修改源代码文件来修复 Bug。", "不要只是分析问题或给出建议，你需要实际定位文件并编辑代码。", "", ] for bug in bugs: prompt_parts.append(bug.format_for_prompt()) prompt_parts.append("") prompt_parts.extend([ "## 修复要求", "1. 先用 Grep/Glob 定位相关源代码文件", "2. 用 Read 读取文件内容，理解上下文", "3. 用 Edit 或 Write 直接修改代码来修复 Bug", "4. 每个 Bug 只做最小必要的改动", "5. 确保不破坏现有功能", "", "## 测试用例要求（必须严格遵守）", "", "**如果是代码逻辑 Bug（运行时错误、TypeError、AttributeError 等）：**", f"- 修复完成后，你必须针对本次修复编写测试用例", f"- 将测试用例保存到项目根目录下的 `{test_file}` 文件中", "- 测试文件必须是可独立运行的（包含必要的 import）", "- 对于 Django 项目，测试类应继承 `django.test.TestCase`", "- **不要运行测试，只写测试文件。测试会由系统自动运行。**", "", "**如果是 CI/CD 构建或部署 Bug（Docker build 失败、依赖错误、语法错误等）：**", f"- 编写一个简单的验证脚本保存到 `{test_file}`", "- 脚本内容：验证项目能正常加载（如 import 检查、`manage.py check` 等）", "- **不要运行脚本，只写文件。**", "", "最后简要说明每个 Bug 的修复方式。", "", "请立即开始修复，直接编辑文件。", ]) prompt = "\n".join(prompt_parts) logger.info(f"开始批量修复 {len(bugs)} 个 Bug...") return self.execute_prompt(prompt, project_path) def retry_fix_bugs( self, bugs: list[Bug], project_path: str, previous_diff: str, previous_test_output: str, round_num: int, ) -> tuple[bool, str]: """ 重试修复 Bug，带上次失败的上下文 Args: bugs: Bug 列表 project_path: 项目路径 previous_diff: 上轮修复产生的 diff previous_test_output: 上轮测试输出 round_num: 当前轮次 Returns: (成功与否, Claude 输出) """ bug_ids = "_".join(str(b.id) for b in bugs) test_file = f"repair_test_bug_{bug_ids}.py" prompt_parts = [ f"你是一个自动化 Bug 修复代理。这是第 {round_num} 次修复尝试。", "", "## 上次修复尝试失败", "", "上次你进行了如下代码修改：", "```diff", previous_diff[:3000], "```", "", "但是测试未通过，测试输出如下：", "```", previous_test_output[:3000], "```", "", "请分析上次修复失败的原因，避免同样的错误，重新修复以下 Bug：", "", ] for bug in bugs: prompt_parts.append(bug.format_for_prompt()) prompt_parts.append("") prompt_parts.extend([ "## 修复要求", "1. 先分析上次测试失败的原因", "2. 用 Grep/Glob 定位相关源代码文件", "3. 用 Read 读取文件内容，理解上下文", "4. 用 Edit 或 Write 直接修改代码来修复 Bug", "5. 每个 Bug 只做最小必要的改动", "6. 确保不破坏现有功能", "", "## 测试用例要求（必须严格遵守）", "", "**如果是代码逻辑 Bug（运行时错误、TypeError、AttributeError 等）：**", f"- 修复完成后，你必须针对本次修复编写测试用例", f"- 将测试用例保存到项目根目录下的 `{test_file}` 文件中", "- 测试文件必须是可独立运行的（包含必要的 import）", "- 对于 Django 项目，测试类应继承 `django.test.TestCase`", "- **不要运行测试，只写测试文件。测试会由系统自动运行。**", "", "**如果是 CI/CD 构建或部署 Bug（Docker build 失败、依赖错误、语法错误等）：**", f"- 编写一个简单的验证脚本保存到 `{test_file}`", "- 脚本内容：验证项目能正常加载（如 import 检查、`manage.py check` 等）", "- **不要运行脚本，只写文件。**", "", "最后简要说明每个 Bug 的修复方式和与上次的区别。", "", "请立即开始修复，直接编辑文件。", ]) prompt = "\n".join(prompt_parts) logger.info(f"开始第 {round_num} 轮修复 {len(bugs)} 个 Bug...") return self.execute_prompt(prompt, project_path) def triage_bug(self, bug: Bug, project_path: str) -> tuple[bool, str]: """ 分诊 Bug：判断是否为可修复的代码缺陷。 输出中包含 VERDICT:FIX 或 VERDICT:CANNOT_REPRODUCE。 """ prompt = f"""你是一个 Bug 分诊专家。请分析以下 Bug，判断它是否是一个需要修复的**代码缺陷**。 {bug.format_for_prompt()} ## 判断规则 ### 运行时错误（runtime） 属于 **无法复现 / 不需要修复** 的情况（CANNOT_REPRODUCE）： 1. JWT Token 过期、认证失败 — 正常认证流程，不是代码 Bug 2. HTTP 405 Method Not Allowed — 客户端请求了错误的方法 3. 第三方库内部错误且 file_path 指向 site-packages / sdk — 非项目代码 4. 瞬态网络错误、加载中断（如 PlatformException: Loading interrupted） 5. 客户端传参错误导致的验证失败 6. 错误堆栈中没有项目代码帧（全在框架/三方库中） 属于 **需要修复** 的情况（FIX）： 1. 堆栈中有项目代码（apps/ 或 lib/ 开头）且错误原因明确 2. 数据库约束错误（IntegrityError）由项目代码逻辑引起 3. TypeError / AttributeError 出现在项目视图或模型中 ### CI/CD 构建错误（CICDFailure） 属于 **需要修复** 的情况（FIX）： 1. Docker build 失败且日志中包含代码编译/语法错误（SyntaxError、ImportError、ModuleNotFoundError 等） 2. npm/pip install 成功但 build 阶段因代码问题失败 3. Dockerfile 中 RUN 命令因项目代码问题报错 属于 **无法复现 / 不需要修复** 的情况（CANNOT_REPRODUCE）： 1. Docker 镜像仓库认证失败、推送失败 — 基础设施问题 2. kubectl 部署失败、K8s 资源不足 — 运维问题 3. 网络超时、镜像拉取失败 — 瞬态错误 4. 日志中没有具体代码错误，只有通用构建/部署失败信息 请先用 Grep/Read 查看相关源文件确认当前代码状态，然后给出判断。 **最后一行必须输出以下格式之一（只输出一个）：** VERDICT:FIX VERDICT:CANNOT_REPRODUCE """ return self.execute_prompt(prompt, project_path, allowed_tools="Read,Grep,Glob") 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, allowed_tools="Read,Grep,Glob") 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, allowed_tools="Read,Grep,Glob")