167 lines
7.2 KiB
Python
167 lines
7.2 KiB
Python
from typing import List
|
||
|
||
from pathlib import Path
|
||
from .core import CodeGenerator
|
||
from .models import OutputFormat
|
||
from .design_manager import DesignManager
|
||
|
||
|
||
class FixGenerator(CodeGenerator):
|
||
"""处理 Bug 修复逻辑的生成器类,继承自 BaseGenerator。"""
|
||
|
||
def __init__(self, **kwargs):
|
||
"""初始化 FixGenerator,继承基类参数。"""
|
||
super().__init__(**kwargs)
|
||
|
||
def process_fix(
|
||
self,
|
||
bug_issue_path: Path,
|
||
output_format: OutputFormat = OutputFormat.FULL,
|
||
) -> (bool, List[Path]):
|
||
"""
|
||
处理 fix 命令逻辑:读取 Bug 工单,分析变更,生成并应用修复代码。
|
||
|
||
Args:
|
||
bug_issue_path: Bug 工单文件路径(如 bug.issue)
|
||
output_format: 输出格式,默认为 FULL,支持 DIFF 用于差异生成
|
||
|
||
Returns:
|
||
bool: 修复是否成功
|
||
"""
|
||
# 读取 Bug 工单文件内容
|
||
try:
|
||
with open(bug_issue_path, "r", encoding="utf-8") as f:
|
||
issue_content = f.read()
|
||
except Exception as e:
|
||
self.logger.error(f"读取 Bug 工单文件失败: {e}")
|
||
self.console.print(f"[bold red]❌ 读取 Bug 工单文件失败: {e}[/bold red]")
|
||
return (False, [])
|
||
|
||
# 添加 README 哈希检查逻辑
|
||
try:
|
||
design_file_path = self.output_dir / "design.json"
|
||
if design_file_path.exists():
|
||
design_manager = DesignManager(design_file_path)
|
||
design = design_manager.load_design()
|
||
readme_path = None
|
||
if design.readme_path:
|
||
readme_path = Path(design.readme_path)
|
||
else:
|
||
readme_path = self.output_dir / "README.md"
|
||
|
||
if readme_path and readme_path.exists():
|
||
if not design_manager.validate_readme_hash(design, readme_path):
|
||
warning_msg = "README 哈希不一致,当前内容可能与设计不符"
|
||
self.logger.warning(warning_msg)
|
||
self.console.print(f"[bold yellow]⚠ 警告: {warning_msg}[/bold yellow]")
|
||
else:
|
||
self.logger.info("README 哈希检查通过")
|
||
else:
|
||
self.logger.warning("README 文件不存在,无法进行哈希检查")
|
||
else:
|
||
self.logger.warning("design.json 文件不存在,无法检查 README 哈希")
|
||
except Exception as e:
|
||
self.logger.warning(f"检查 README 哈希时出错: {e}")
|
||
self.console.print(f"[bold yellow]⚠ 警告: 检查 README 哈希时出错: {e}[/bold yellow]")
|
||
|
||
# 调用 LLM 分析工单,获取变更计划
|
||
try:
|
||
analysis = self._analyze_issue(issue_content, "bug")
|
||
except Exception as e:
|
||
self.logger.error(f"分析工单失败: {e}")
|
||
self.console.print(f"[bold red]❌ 分析工单失败: {e}[/bold red]")
|
||
return (False, [])
|
||
|
||
affected_files = analysis.get("affected_files", [])
|
||
|
||
successful_files = []
|
||
for file_info in affected_files:
|
||
file_path = file_info.get("path")
|
||
action = file_info.get("action") # 'create' 或 'modify'
|
||
# description = file_info.get("description", "")
|
||
dependencies = file_info.get("dependencies", [])
|
||
|
||
if action == "modify":
|
||
# 修改现有文件:读取当前内容
|
||
full_path = self.output_dir / file_path
|
||
if not full_path.exists():
|
||
self.logger.error(f"文件不存在,无法修改: {file_path}")
|
||
self.console.print(f"[bold red]❌ 文件不存在,无法修改: {file_path}[/bold red]")
|
||
continue
|
||
|
||
try:
|
||
with open(full_path, "r", encoding="utf-8") as f:
|
||
existing_content = f.read()
|
||
except Exception as e:
|
||
self.logger.error(f"读取文件 {file_path} 失败: {e}")
|
||
self.console.print(f"[bold red]❌ 读取文件 {file_path} 失败: {e}[/bold red]")
|
||
continue
|
||
|
||
# 生成修复代码
|
||
instruction = (
|
||
f"根据 Bug 工单修复文件 '{file_path}'..."
|
||
)
|
||
code, desc, commands = self.generate_file(
|
||
file_path=file_path,
|
||
prompt_instruction=instruction,
|
||
dependency_files=dependencies,
|
||
existing_content=existing_content,
|
||
output_format=output_format,
|
||
)
|
||
|
||
# 写入修复后的内容
|
||
try:
|
||
with open(full_path, "w", encoding="utf-8") as f:
|
||
f.write(code)
|
||
self.logger.info(f"已修复文件: {file_path} - {desc}")
|
||
successful_files.append(file_path)
|
||
except Exception as e:
|
||
self.logger.error(f"写入文件 {file_path} 失败: {e}")
|
||
self.console.print(f"[bold red]❌ 写入文件 {file_path} 失败: {e}[/bold red]")
|
||
continue
|
||
|
||
# 执行相关命令
|
||
for cmd in commands:
|
||
self.execute_command(cmd, cwd=self.output_dir)
|
||
|
||
# 更新 design.json 中的文件条目
|
||
self.update_file_entry(file_path, code)
|
||
|
||
elif action == "create":
|
||
# 创建新文件:无需现有内容
|
||
instruction = (
|
||
f"根据 Bug 工单创建文件 '{file_path}'..."
|
||
)
|
||
code, desc, commands = self.generate_file(
|
||
file_path=file_path,
|
||
prompt_instruction=instruction,
|
||
dependency_files=dependencies,
|
||
existing_content=None,
|
||
output_format=output_format,
|
||
)
|
||
|
||
# 写入新文件
|
||
full_path = self.output_dir / file_path
|
||
full_path.parent.mkdir(parents=True, exist_ok=True)
|
||
try:
|
||
with open(full_path, "w", encoding="utf-8") as f:
|
||
f.write(code)
|
||
self.logger.info(f"已创建文件: {file_path} - {desc}")
|
||
successful_files.append(file_path)
|
||
except Exception as e:
|
||
self.logger.error(f"创建文件 {file_path} 失败: {e}")
|
||
self.console.print(f"[bold red]❌ 创建文件 {file_path} 失败: {e}[/bold red]")
|
||
continue
|
||
|
||
for cmd in commands:
|
||
self.execute_command(cmd, cwd=self.output_dir)
|
||
|
||
if successful_files:
|
||
self.logger.info(f"修复成功,处理了 {len(successful_files)} 个文件")
|
||
self.console.print(f"[green]✅ 修复成功,处理了 {len(successful_files)} 个文件[/green]")
|
||
return (True, successful_files)
|
||
else:
|
||
self.logger.error("修复失败,没有文件被成功处理")
|
||
self.console.print("[bold red]❌ 修复失败,没有文件被成功处理[/bold red]")
|
||
return (False, [])
|