docs: 移除README中已实现代码的详细说明部分
This commit is contained in:
parent
200a045ebb
commit
0c3f87724b
370
README.md
370
README.md
|
|
@ -3,376 +3,6 @@
|
|||
本项目是一个基于大语言模型的智能代码生成与维护工具。它不仅能够根据项目 `README.md` 描述**自动生成完整的 Python 包代码**，还支持**在现有项目上增量添加功能**和**自动修复 Bug**。工具采用 `uv` 管理依赖，包含单元测试、并行检查、断点续写等特性，并通过一个**面向 LLM 的中间设计层**来提升生成质量和可维护性。
|
||||
|
||||
|
||||
## 特别说明
|
||||
我已经实现了一个简易版本，请在此基础上修改、拓展、开发：
|
||||
|
||||
```
|
||||
#!/home/songsenand/env/.venv/bin/python
|
||||
"""
|
||||
基于LLM的自动化代码生成工具
|
||||
根据README.md文件，自动生成项目文件结构并填充代码，执行必要命令。
|
||||
"""
|
||||
|
||||
import json
|
||||
import os
|
||||
import subprocess
|
||||
import sys
|
||||
from typing import List, Dict, Optional, Any, Tuple
|
||||
from pathlib import Path
|
||||
|
||||
import typer
|
||||
from rich.console import Console
|
||||
from rich.progress import Progress, SpinnerColumn, TextColumn, BarColumn, TaskID
|
||||
from loguru import logger
|
||||
from openai import OpenAI
|
||||
|
||||
# ==================== 配置 ====================
|
||||
DANGEROUS_COMMANDS = ["rm", "sudo", "chmod", "dd", "mkfs", "> /dev/sda", "format"]
|
||||
ALLOWED_COMMANDS = [] # 可设置白名单，为空则只检查黑名单
|
||||
|
||||
app = typer.Typer(help="基于LLM的自动化代码生成工具")
|
||||
console = Console()
|
||||
|
||||
# ==================== 工具函数 ====================
|
||||
def is_dangerous_command(cmd: str) -> Tuple[bool, str]:
|
||||
"""
|
||||
判断命令是否危险
|
||||
返回 (是否危险, 原因)
|
||||
"""
|
||||
cmd_lower = cmd.lower()
|
||||
for danger in DANGEROUS_COMMANDS:
|
||||
if danger in cmd_lower:
|
||||
return True, f"包含危险关键词 '{danger}'"
|
||||
return False, ""
|
||||
|
||||
# ==================== 核心类 ====================
|
||||
class CodeGenerator:
|
||||
"""代码生成器，封装所有逻辑"""
|
||||
|
||||
def __init__(
|
||||
self,
|
||||
api_key: Optional[str] = None,
|
||||
base_url: str = "https://api.deepseek.com",
|
||||
model: str = "deepseek-reasoner",
|
||||
output_dir: str = "./generated",
|
||||
log_file: Optional[str] = None,
|
||||
):
|
||||
"""
|
||||
初始化生成器
|
||||
|
||||
Args:
|
||||
api_key: OpenAI API密钥，默认从环境变量DEEPSEEK_APIKEY读取
|
||||
base_url: API基础URL
|
||||
model: 使用的模型
|
||||
output_dir: 输出根目录
|
||||
log_file: 日志文件路径，默认自动生成
|
||||
"""
|
||||
self.api_key = api_key or os.getenv("DEEPSEEK_APIKEY")
|
||||
if not self.api_key:
|
||||
raise ValueError("必须提供API密钥，或设置环境变量DEEPSEEK_APIKEY")
|
||||
|
||||
self.client = OpenAI(api_key=self.api_key, base_url=base_url)
|
||||
self.model = model
|
||||
self.output_dir = Path(output_dir)
|
||||
self.output_dir.mkdir(parents=True, exist_ok=True)
|
||||
|
||||
# 配置日志
|
||||
if log_file is None:
|
||||
log_file = self.output_dir / "generator.log"
|
||||
logger.remove() # 移除默认handler
|
||||
logger.add(sys.stderr, level="WARNING") # 控制台输出INFO及以上
|
||||
logger.add(log_file, rotation="10 MB", level="DEBUG") # 文件记录DEBUG
|
||||
logger.info(f"日志已初始化，保存至: {log_file}")
|
||||
|
||||
self.readme_content = None
|
||||
|
||||
self.progress: Optional[Progress] = None
|
||||
self.tasks: Dict[str, TaskID] = {} # 任务ID映射
|
||||
|
||||
def _call_llm(
|
||||
self,
|
||||
system_prompt: str,
|
||||
user_prompt: str,
|
||||
temperature: float = 0.2,
|
||||
expect_json: bool = True,
|
||||
) -> Dict[str, Any]:
|
||||
"""
|
||||
调用LLM并返回解析后的JSON
|
||||
"""
|
||||
logger.debug(f"调用LLM，模型: {self.model}")
|
||||
logger.debug(f"System: {system_prompt[:200]}...")
|
||||
logger.debug(f"User: {user_prompt[:200]}...")
|
||||
|
||||
try:
|
||||
response = self.client.chat.completions.create(
|
||||
model=self.model,
|
||||
messages=[
|
||||
{"role": "system", "content": system_prompt},
|
||||
{"role": "user", "content": user_prompt},
|
||||
],
|
||||
temperature=temperature,
|
||||
response_format={"type": "json_object"} if expect_json else None,
|
||||
)
|
||||
|
||||
message = response.choices[0].message
|
||||
content = message.content
|
||||
|
||||
# 记录思考过程（如果存在）
|
||||
if hasattr(message, "reasoning_content") and message.reasoning_content:
|
||||
logger.info(f"模型思考过程: {message.reasoning_content}")
|
||||
|
||||
logger.debug(f"LLM原始响应: {content[:500]}...")
|
||||
|
||||
if expect_json:
|
||||
result = json.loads(content)
|
||||
else:
|
||||
result = {"content": content}
|
||||
|
||||
return result
|
||||
|
||||
except json.JSONDecodeError as e:
|
||||
logger.error(f"JSON解析失败: {e}")
|
||||
raise ValueError(f"LLM返回的不是有效JSON: {content[:200]}")
|
||||
except Exception as e:
|
||||
logger.error(f"LLM调用失败: {e}")
|
||||
raise
|
||||
|
||||
def parse_readme(self, readme_path: Path) -> str:
|
||||
"""
|
||||
读取README文件内容
|
||||
"""
|
||||
logger.info(f"读取README文件: {readme_path}")
|
||||
try:
|
||||
with open(readme_path, "r", encoding="utf-8") as f:
|
||||
content = f.read()
|
||||
logger.debug(f"README内容长度: {len(content)} 字符")
|
||||
return content
|
||||
except Exception as e:
|
||||
logger.error(f"读取README失败: {e}")
|
||||
raise
|
||||
|
||||
def get_project_structure(self) -> Tuple[List[str], Dict[str, List[str]]]:
|
||||
"""
|
||||
根据README内容，让LLM生成文件列表和依赖关系
|
||||
|
||||
Returns:
|
||||
(files, dependencies)
|
||||
files: 按顺序需要生成的文件路径列表
|
||||
dependencies: 字典 {file: [依赖文件路径]}
|
||||
"""
|
||||
system_prompt = (
|
||||
"你是一个软件架构师。请根据README描述，分析需要生成哪些源代码文件，并确定它们的生成顺序，"
|
||||
"同时给出每个文件生成时最少需要读取哪些已有文件作为上下文。"
|
||||
"返回严格的JSON对象，包含两个字段：\n"
|
||||
"- files: 数组，按生成顺序排列的文件路径（相对于项目根目录）\n"
|
||||
"- dependencies: 对象，键为文件路径，值为该文件依赖的已有文件路径列表（可为空）\n"
|
||||
"注意：依赖文件必须是已存在的参考文件，不要包含待生成的文件。"
|
||||
)
|
||||
user_prompt = f"README内容如下：\n\n{self.readme_content}"
|
||||
|
||||
result = self._call_llm(system_prompt, user_prompt)
|
||||
|
||||
files = result.get("files", [])
|
||||
dependencies = result.get("dependencies", {})
|
||||
|
||||
if not files:
|
||||
raise ValueError("LLM未返回任何文件列表")
|
||||
|
||||
logger.info(f"解析到 {len(files)} 个待生成文件")
|
||||
logger.debug(f"文件列表: {files}")
|
||||
logger.debug(f"依赖关系: {dependencies}")
|
||||
|
||||
return files, dependencies
|
||||
|
||||
def generate_file(
|
||||
self,
|
||||
file_path: str,
|
||||
prompt_instruction: str,
|
||||
dependency_files: List[str],
|
||||
) -> Tuple[str, str, List[str]]:
|
||||
"""
|
||||
生成单个文件，返回 (代码, 描述, 命令列表)
|
||||
"""
|
||||
# 读取依赖文件内容
|
||||
context_content = []
|
||||
|
||||
if self.readme_content:
|
||||
context_content.append(f"### 项目 README ###\n{self.readme_content}\n")
|
||||
|
||||
for dep in dependency_files:
|
||||
dep_path = Path(dep)
|
||||
if not dep_path.exists():
|
||||
# 尝试相对于当前目录或输出目录查找
|
||||
alt_path = self.output_dir / dep
|
||||
if alt_path.exists():
|
||||
dep_path = alt_path
|
||||
else:
|
||||
raise FileNotFoundError(f"依赖文件不存在: {dep}")
|
||||
|
||||
with open(dep_path, "r", encoding="utf-8") as f:
|
||||
content = f.read()
|
||||
context_content.append(f"### 文件: {dep_path.name} (路径: {dep}) ###\n{content}\n")
|
||||
|
||||
full_context = "\n".join(context_content)
|
||||
|
||||
system_prompt = (
|
||||
"你是一个专业的编程助手。根据用户指令和提供的上下文文件，生成完整的代码。"
|
||||
"返回严格的JSON对象，包含三个字段：\n"
|
||||
"- code: (string) 生成的完整代码\n"
|
||||
"- description: (string) 简短的中文功能描述\n"
|
||||
"- commands: (array of string) 生成此文件后需要执行的操作系统命令列表（如编译、安装依赖等），若无则返回空数组"
|
||||
)
|
||||
user_prompt = f"{prompt_instruction}\n\n参考文件上下文:\n{full_context}"
|
||||
|
||||
result = self._call_llm(system_prompt, user_prompt)
|
||||
|
||||
code = result.get("code", "")
|
||||
description = result.get("description", "")
|
||||
commands = result.get("commands", [])
|
||||
|
||||
if not isinstance(commands, list):
|
||||
commands = []
|
||||
|
||||
return code, description, commands
|
||||
|
||||
def execute_command(self, cmd: str, cwd: Optional[Path] = None) -> None:
|
||||
"""
|
||||
执行单个命令，检查风险
|
||||
"""
|
||||
dangerous, reason = is_dangerous_command(cmd)
|
||||
if dangerous:
|
||||
logger.error(f"危险命令被阻止: {cmd}，原因: {reason}")
|
||||
raise RuntimeError(f"危险命令: {cmd} ({reason})")
|
||||
|
||||
logger.info(f"执行命令: {cmd}")
|
||||
try:
|
||||
result = subprocess.run(
|
||||
cmd,
|
||||
shell=True,
|
||||
cwd=cwd or self.output_dir,
|
||||
capture_output=True,
|
||||
text=True,
|
||||
timeout=300, # 5分钟超时
|
||||
)
|
||||
logger.debug(f"命令返回码: {result.returncode}")
|
||||
if result.stdout:
|
||||
logger.debug(f"stdout: {result.stdout[:500]}")
|
||||
if result.stderr:
|
||||
logger.warning(f"stderr: {result.stderr[:500]}")
|
||||
if result.returncode != 0:
|
||||
raise subprocess.CalledProcessError(result.returncode, cmd)
|
||||
except subprocess.TimeoutExpired:
|
||||
logger.error(f"命令执行超时: {cmd}")
|
||||
raise
|
||||
except Exception as e:
|
||||
logger.error(f"命令执行失败: {e}")
|
||||
raise
|
||||
|
||||
def run(self, readme_path: Path):
|
||||
"""
|
||||
主执行流程
|
||||
"""
|
||||
logger.info("=" * 50)
|
||||
logger.info("开始代码生成流程")
|
||||
logger.info(f"README: {readme_path}")
|
||||
logger.info(f"输出目录: {self.output_dir}")
|
||||
|
||||
# 初始化阶段：用rich输出状态（不会被日志级别过滤）
|
||||
console.print("[bold yellow]🔍 正在解析README...[/bold yellow]")
|
||||
self.readme_content = self.parse_readme(readme_path)
|
||||
|
||||
console.print("[bold yellow]📋 正在分析项目结构...[/bold yellow]")
|
||||
files, dependencies = self.get_project_structure()
|
||||
|
||||
console.print(f"[green]✅ 解析完成，共 {len(files)} 个文件待生成[/green]")
|
||||
|
||||
# 3. 创建进度条
|
||||
with Progress(
|
||||
SpinnerColumn(),
|
||||
TextColumn("[progress.description]{task.description}"),
|
||||
BarColumn(),
|
||||
TextColumn("[progress.percentage]{task.percentage:>3.0f}%"),
|
||||
console=console,
|
||||
) as progress:
|
||||
self.progress = progress
|
||||
# 创建总任务
|
||||
total_task = progress.add_task("[cyan]整体进度...", total=len(files))
|
||||
|
||||
# 依次生成每个文件
|
||||
for idx, file in enumerate(files, 1):
|
||||
logger.info(f"处理文件 [{idx}/{len(files)}]: {file}")
|
||||
|
||||
# 创建子任务（可选）
|
||||
file_task = progress.add_task(f"生成 {file}", total=None)
|
||||
|
||||
try:
|
||||
# 获取依赖文件
|
||||
deps = dependencies.get(file, [])
|
||||
|
||||
# 构造生成指令
|
||||
instruction = f"请根据README描述和依赖文件，生成文件 '{file}' 的完整代码。"
|
||||
|
||||
# 调用LLM生成代码
|
||||
code, desc, commands = self.generate_file(file, instruction, deps)
|
||||
|
||||
logger.info(f"生成完成: {file} - {desc}")
|
||||
|
||||
# 写入文件
|
||||
output_path = self.output_dir / file
|
||||
output_path.parent.mkdir(parents=True, exist_ok=True)
|
||||
with open(output_path, "w", encoding="utf-8") as f:
|
||||
f.write(code)
|
||||
logger.info(f"已写入: {output_path}")
|
||||
|
||||
# 执行命令
|
||||
for cmd in commands:
|
||||
logger.info(f"准备执行命令: {cmd}")
|
||||
self.execute_command(cmd, cwd=self.output_dir)
|
||||
|
||||
except Exception as e:
|
||||
logger.error(f"处理文件 {file} 失败: {e}")
|
||||
# 可选：继续或终止
|
||||
raise
|
||||
finally:
|
||||
progress.remove_task(file_task)
|
||||
progress.update(total_task, advance=1)
|
||||
|
||||
logger.success("所有文件处理完成！")
|
||||
|
||||
# ==================== CLI入口 ====================
|
||||
@app.command()
|
||||
def main(
|
||||
readme: Path = typer.Argument(..., exists=True, file_okay=True, dir_okay=False, help="README.md文件路径"),
|
||||
output_dir: Optional[Path] = typer.Option(None, "--output", "-o", help="输出根目录，默认为readme所在目录"),
|
||||
api_key: Optional[str] = typer.Option(None, "--api-key", envvar="DEEPSEEK_APIKEY", help="API密钥，也可通过环境变量DEEPSEEK_APIKEY设置"),
|
||||
base_url: str = typer.Option("https://api.deepseek.com", "--base-url", help="API基础URL"),
|
||||
model: str = typer.Option("deepseek-reasoner", "--model", "-m", help="使用的模型"),
|
||||
log_file: Optional[str] = typer.Option(None, "--log", help="日志文件路径（默认输出目录下generator.log）"),
|
||||
):
|
||||
"""
|
||||
根据README自动生成项目代码
|
||||
"""
|
||||
if output_dir is None:
|
||||
output_dir = readme.parent
|
||||
|
||||
try:
|
||||
generator = CodeGenerator(
|
||||
api_key=api_key,
|
||||
base_url=base_url,
|
||||
model=model,
|
||||
output_dir=output_dir,
|
||||
log_file=log_file,
|
||||
)
|
||||
generator.run(readme)
|
||||
except Exception as e:
|
||||
logger.error(f"程序异常退出: {e}")
|
||||
raise typer.Exit(code=1)
|
||||
|
||||
if __name__ == "__main__":
|
||||
app()
|
||||
```
|
||||
|
||||
|
||||
## ✨ 核心特性
|
||||
|
||||
- 📦 **自动生成**：解析 `README.md`，分析需要生成的文件列表及依赖关系，按顺序生成每个文件的代码。
|
||||
|
|
|
|||
File diff suppressed because one or more lines are too long
|
|
@ -1,35 +0,0 @@
|
|||
# 需求工单：完善单元测试
|
||||
name: 完善单元测试
|
||||
description: 当前项目的单元测试覆盖不足，需补充核心模块的测试用例，确保代码质量并便于后续迭代。
|
||||
|
||||
affected_files:
|
||||
# 测试文件（可能需新建）
|
||||
- tests/test_cli.py
|
||||
- tests/test_core.py
|
||||
- tests/test_checker.py
|
||||
- tests/test_utils.py
|
||||
- tests/test_models.py
|
||||
# 核心代码文件（测试将覆盖它们，但本身无需修改）
|
||||
- src/llm_codegen/cli.py
|
||||
- src/llm_codegen/core.py
|
||||
- src/llm_codegen/checker.py
|
||||
- src/llm_codegen/utils.py
|
||||
- src/llm_codegen/models.py
|
||||
|
||||
acceptance_criteria:
|
||||
- 所有新增或修改的测试用例均通过 `pytest` 运行，无失败、错误或跳过。
|
||||
- 测试覆盖率（语句覆盖率）不低于 85%，分支覆盖率不低于 70%，可通过 `pytest --cov=src/llm_codegen --cov-branch` 验证。
|
||||
- 核心类 `CodeGenerator` 的以下方法被充分测试：
|
||||
- `__init__`（不同参数组合）
|
||||
- `_call_llm`（模拟 API 响应、超时、异常）
|
||||
- `parse_readme`（正常文件、空文件、编码问题）
|
||||
- `get_project_structure`（模拟 LLM 返回）
|
||||
- `generate_file`（依赖文件存在/不存在）
|
||||
- `execute_command`（正常执行、危险命令拦截、超时）
|
||||
- `run`（完整流程的模拟）
|
||||
- 并行检查模块 `checker.py` 的主要函数（如 `run_checks`、`apply_fixes`）需覆盖正常与错误场景。
|
||||
- 工具函数 `is_dangerous_command` 应测试多个危险命令变体及安全命令。
|
||||
- 命令行接口（CLI）需包含端到端测试，验证 `init`、`enhance`、`fix` 子命令的基本流程（可使用 `CliRunner` 或 `subprocess` 模拟）。
|
||||
- 测试应使用 `pytest` 的临时目录（`tmp_path`）和 `unittest.mock` 模拟外部依赖（如文件系统、API 调用），避免污染实际环境。
|
||||
- 为常用模拟操作（如模拟 OpenAI 客户端、模拟文件读写）编写可复用的 fixture。
|
||||
- 测试代码遵循项目的编码规范（使用 black、isort 格式化，类型注解完整）。
|
||||
|
|
@ -0,0 +1,22 @@
|
|||
# 需求工单：增强错误处理，避免生成过程中断
|
||||
name: 增强错误处理机制
|
||||
description: |
|
||||
当前工具在生成代码过程中，如果某个步骤（如调用LLM、写入文件、执行命令等）发生错误，会直接抛出异常并终止整个生成流程。这种“一错即停”的行为在批量处理或长时间任务中非常不便，用户希望即使出现局部错误，工具也能记录错误信息并在命令行中友好显示，然后继续执行剩余任务（例如继续生成其他文件、执行后续检查等）。
|
||||
|
||||
具体改进要求：
|
||||
- 在代码生成的主循环中捕获所有预期内异常（如网络错误、文件权限错误、命令执行失败等），记录错误日志（使用loguru），并在命令行中通过rich打印红色错误信息。
|
||||
- 对于非致命错误，继续执行下一个文件或下一步骤。
|
||||
- 致命错误（如配置文件缺失、API密钥无效）仍可终止，但应给出清晰提示。
|
||||
- 在最终汇总时，显示成功、失败和跳过的统计信息。
|
||||
|
||||
acceptance_criteria:
|
||||
- 当某个文件生成失败时，控制台输出红色错误信息，但工具继续尝试生成下一个文件。
|
||||
- 所有错误信息均写入日志文件（logs/目录）。
|
||||
- 生成结束后，显示汇总信息：“生成完成：成功 X 个，失败 Y 个，跳过 Z 个。”
|
||||
- 如果发生致命错误（如无法读取design.json），工具应终止并给出明确提示。
|
||||
- 错误处理不应影响已有状态文件的正确记录（断点续写功能需保持有效）。
|
||||
|
||||
affected_files:
|
||||
- src/llm_codegen/core.py # 主要生成逻辑
|
||||
- src/llm_codegen/cli.py # 命令行入口，可能包含主循环
|
||||
- src/llm_codegen/utils.py # 可能增加错误辅助函数
|
||||
|
|
@ -13,6 +13,7 @@ dependencies = [
|
|||
"rich>=13.0.0",
|
||||
"loguru>=0.7.0",
|
||||
"openai>=1.0.0",
|
||||
"pathspec>=1.0.4",
|
||||
]
|
||||
authors = [
|
||||
{name = "Your Name", email = "your.email@example.com"}
|
||||
|
|
@ -32,9 +33,13 @@ dev = [
|
|||
]
|
||||
|
||||
[project.scripts]
|
||||
llm-codegen = "src.llm_codegen.cli:app"
|
||||
llm-codegen = "llm_codegen.cli:app"
|
||||
|
||||
[tool.llm-codegen]
|
||||
check_tools = ["pytest", "pylint", "mypy", "black"]
|
||||
max_retries = 3
|
||||
dangerous_commands = ["rm", "sudo", "chmod", "dd"]
|
||||
|
||||
# 新增：指定包所在目录
|
||||
[tool.setuptools.packages.find]
|
||||
where = ["src"]
|
||||
|
|
|
|||
|
|
@ -1,21 +1,33 @@
|
|||
import json
|
||||
|
||||
import subprocess
|
||||
import sys
|
||||
from typing import List, Dict, Optional, Tuple, Any
|
||||
from pathlib import Path
|
||||
from concurrent.futures import ThreadPoolExecutor, as_completed
|
||||
import os
|
||||
import warnings
|
||||
|
||||
from loguru import logger
|
||||
from .core import CodeGenerator
|
||||
from .utils import is_dangerous_command
|
||||
|
||||
# 尝试导入 pathspec（用于精确解析 .gitignore）
|
||||
try:
|
||||
from pathspec import PathSpec
|
||||
from pathspec.patterns import GitWildMatchPattern
|
||||
HAS_PATHSPEC = True
|
||||
except ImportError:
|
||||
HAS_PATHSPEC = False
|
||||
import fnmatch
|
||||
warnings.warn(
|
||||
"pathspec 未安装，将使用简单的通配符匹配处理 .gitignore（可能不完全准确）。"
|
||||
"建议安装：pip install pathspec"
|
||||
)
|
||||
|
||||
|
||||
class Checker:
|
||||
"""
|
||||
并行检查与修复模块，运行检查工具（如pylint、mypy、black）并收集错误，
|
||||
支持自动调用LLM生成修复补丁。
|
||||
并行检查与修复模块，运行检查工具（默认 black）并收集错误，
|
||||
支持自动调用 LLM 生成修复补丁。
|
||||
"""
|
||||
|
||||
def __init__(
|
||||
|
|
@ -32,14 +44,21 @@ class Checker:
|
|||
|
||||
Args:
|
||||
output_dir: 项目输出目录，用于查找代码文件和保存检查结果
|
||||
check_tools: 检查工具列表，默认为 ["pylint", "mypy", "black"]
|
||||
code_generator: CodeGenerator实例，用于调用LLM，如果为None则创建新实例
|
||||
api_key: OpenAI API密钥，用于LLM调用（如果code_generator为None）
|
||||
base_url: API基础URL（如果code_generator为None）
|
||||
model: 使用的模型（如果code_generator为None）
|
||||
check_tools: 检查工具列表，默认为 ["black"]。若传入多个工具，仅使用第一个
|
||||
code_generator: CodeGenerator 实例，用于调用 LLM
|
||||
api_key, base_url, model: 用于创建 CodeGenerator（当 code_generator 为 None 时）
|
||||
"""
|
||||
self.output_dir = Path(output_dir)
|
||||
self.check_tools = check_tools or ["pylint", "mypy", "black"]
|
||||
|
||||
# 处理检查工具：默认 black，若传入多个则取第一个并警告
|
||||
if check_tools is None:
|
||||
self.check_tools = ["black"]
|
||||
else:
|
||||
if len(check_tools) > 1:
|
||||
logger.warning(
|
||||
f"检测到多个检查工具 {check_tools}，将只使用第一个：{check_tools[0]}"
|
||||
)
|
||||
self.check_tools = [check_tools[0]] if check_tools else ["black"]
|
||||
|
||||
if code_generator:
|
||||
self.code_generator = code_generator
|
||||
|
|
@ -52,45 +71,127 @@ class Checker:
|
|||
)
|
||||
|
||||
self.results_file = self.output_dir / "check_results.json"
|
||||
logger.info(f"Checker初始化完成，输出目录: {self.output_dir}")
|
||||
logger.info(f"Checker 初始化完成，输出目录: {self.output_dir}，检查工具: {self.check_tools}")
|
||||
|
||||
def _load_gitignore_patterns(self) -> Optional[Any]:
|
||||
"""
|
||||
加载 .gitignore 文件中的模式，返回一个可用于匹配的函数或对象。
|
||||
若文件不存在或解析失败，返回 None。
|
||||
"""
|
||||
gitignore_path = self.output_dir / ".gitignore"
|
||||
if not gitignore_path.exists():
|
||||
return None
|
||||
|
||||
try:
|
||||
with open(gitignore_path, "r", encoding="utf-8") as f:
|
||||
lines = f.readlines()
|
||||
except Exception as e:
|
||||
logger.warning(f"读取 .gitignore 失败: {e}")
|
||||
return None
|
||||
|
||||
if HAS_PATHSPEC:
|
||||
# 使用 pathspec 精确解析
|
||||
try:
|
||||
spec = PathSpec.from_lines(GitWildMatchPattern, lines)
|
||||
return spec
|
||||
except Exception as e:
|
||||
logger.warning(f"解析 .gitignore 失败，将使用简单匹配: {e}")
|
||||
return None
|
||||
else:
|
||||
# 回退到简单通配符匹配（忽略空行、注释，不支持 ** 和否定模式）
|
||||
patterns = []
|
||||
for line in lines:
|
||||
line = line.strip()
|
||||
if not line or line.startswith("#"):
|
||||
continue
|
||||
# 移除末尾的注释（# 后面的内容）
|
||||
if "#" in line:
|
||||
line = line.split("#", 1)[0].strip()
|
||||
if line:
|
||||
patterns.append(line)
|
||||
return patterns
|
||||
|
||||
def _is_ignored_by_gitignore(self, rel_path: str, gitignore_matcher) -> bool:
|
||||
"""
|
||||
判断相对路径是否被 .gitignore 忽略。
|
||||
gitignore_matcher 可以是 PathSpec 对象或简单模式列表。
|
||||
"""
|
||||
if gitignore_matcher is None:
|
||||
return False
|
||||
|
||||
# 将路径转换为 POSIX 风格（使用 / 分隔符）
|
||||
rel_path = rel_path.replace(os.sep, "/")
|
||||
|
||||
if HAS_PATHSPEC and isinstance(gitignore_matcher, PathSpec):
|
||||
return gitignore_matcher.match_file(rel_path)
|
||||
elif isinstance(gitignore_matcher, list):
|
||||
# 简单匹配：对于每个模式，如果模式以 / 结尾，则匹配目录；否则匹配文件
|
||||
for pattern in gitignore_matcher:
|
||||
# 处理目录模式（以 / 结尾）
|
||||
if pattern.endswith("/"):
|
||||
# 检查路径是否以该目录开头
|
||||
if rel_path.startswith(pattern.rstrip("/") + "/") or rel_path == pattern.rstrip("/"):
|
||||
return True
|
||||
else:
|
||||
# 文件/通配符模式，使用 fnmatch
|
||||
if fnmatch.fnmatch(rel_path, pattern):
|
||||
return True
|
||||
return False
|
||||
|
||||
def _filter_files_by_gitignore(self, files: List[Path]) -> List[Path]:
|
||||
"""
|
||||
根据 .gitignore 和硬编码规则（如 .git/）过滤文件列表。
|
||||
"""
|
||||
gitignore_matcher = self._load_gitignore_patterns()
|
||||
filtered = []
|
||||
for file_path in files:
|
||||
try:
|
||||
# 计算相对于输出目录的路径
|
||||
rel_path = file_path.relative_to(self.output_dir).as_posix()
|
||||
except ValueError:
|
||||
# 如果文件不在输出目录下（例如绝对路径），则保留（不过滤）
|
||||
logger.warning(f"文件 {file_path} 不在输出目录 {self.output_dir} 下，将保留")
|
||||
filtered.append(file_path)
|
||||
continue
|
||||
|
||||
# 硬编码忽略 .git 目录
|
||||
if rel_path.startswith(".git/") or rel_path == ".git":
|
||||
logger.debug(f"忽略 .git 目录下的文件: {rel_path}")
|
||||
continue
|
||||
|
||||
# 检查 .gitignore
|
||||
if self._is_ignored_by_gitignore(rel_path, gitignore_matcher):
|
||||
logger.debug(f"忽略 .gitignore 中的文件: {rel_path}")
|
||||
continue
|
||||
|
||||
filtered.append(file_path)
|
||||
|
||||
return filtered
|
||||
|
||||
def run_check(self, tool: str, file_path: Path) -> Dict[str, Any]:
|
||||
"""
|
||||
运行单个检查工具并返回结果
|
||||
|
||||
Args:
|
||||
tool: 检查工具名称（如 'pylint', 'mypy', 'black'）
|
||||
tool: 检查工具名称（如 'black'）
|
||||
file_path: 要检查的文件路径
|
||||
|
||||
Returns:
|
||||
Dict包含工具名、返回码、stdout、stderr和错误信息
|
||||
Dict 包含工具名、返回码、stdout、stderr 和错误信息
|
||||
"""
|
||||
logger.debug(f"运行检查工具: {tool} 在文件: {file_path}")
|
||||
|
||||
# 构建命令，根据工具不同调整
|
||||
if tool == "pylint":
|
||||
if tool == "black":
|
||||
cmd = f"black --check --diff {file_path}"
|
||||
elif tool == "pylint":
|
||||
cmd = f"pylint {file_path} --output-format=json"
|
||||
elif tool == "mypy":
|
||||
cmd = f"mypy {file_path} --show-error-codes --no-error-summary"
|
||||
elif tool == "black":
|
||||
cmd = f"black --check --diff {file_path}"
|
||||
else:
|
||||
# 默认直接运行工具
|
||||
cmd = f"{tool} {file_path}"
|
||||
|
||||
# 检查命令是否危险
|
||||
dangerous, reason = is_dangerous_command(cmd)
|
||||
if dangerous:
|
||||
logger.warning(f"检查命令可能危险，跳过: {cmd}, 原因: {reason}")
|
||||
return {
|
||||
"tool": tool,
|
||||
"file": str(file_path),
|
||||
"returncode": -1,
|
||||
"stdout": "",
|
||||
"stderr": f"危险命令被阻止: {reason}",
|
||||
"errors": [],
|
||||
}
|
||||
|
||||
try:
|
||||
result = subprocess.run(
|
||||
cmd,
|
||||
|
|
@ -98,7 +199,7 @@ class Checker:
|
|||
cwd=self.output_dir,
|
||||
capture_output=True,
|
||||
text=True,
|
||||
timeout=60, # 1分钟超时
|
||||
timeout=60, # 1 分钟超时
|
||||
)
|
||||
|
||||
# 解析错误信息
|
||||
|
|
@ -106,7 +207,7 @@ class Checker:
|
|||
if result.stderr:
|
||||
errors.append(result.stderr.strip())
|
||||
if result.stdout:
|
||||
# 对于pylint的JSON输出，可以进一步解析
|
||||
# 对于 pylint 的 JSON 输出，可以进一步解析
|
||||
if tool == "pylint" and result.returncode != 0:
|
||||
try:
|
||||
pylint_errors = json.loads(result.stdout)
|
||||
|
|
@ -147,25 +248,27 @@ class Checker:
|
|||
|
||||
def run_parallel_checks(self, files: Optional[List[Path]] = None) -> List[Dict[str, Any]]:
|
||||
"""
|
||||
并行运行所有检查工具在指定文件上
|
||||
并行运行检查工具在指定文件上（仅使用配置的第一个工具）
|
||||
|
||||
Args:
|
||||
files: 要检查的文件路径列表，如果为None则检查输出目录下所有.py文件
|
||||
files: 要检查的文件路径列表，如果为 None 则自动查找输出目录下所有 .py 文件（排除 .gitignore 中的）
|
||||
|
||||
Returns:
|
||||
检查结果列表，每个元素为run_check返回的字典
|
||||
检查结果列表
|
||||
"""
|
||||
if files is None:
|
||||
# 递归查找所有.py文件
|
||||
# 递归查找所有 .py 文件
|
||||
files = list(self.output_dir.rglob("*.py"))
|
||||
logger.info(f"开始并行检查，文件数: {len(files)}, 工具数: {len(self.check_tools)}")
|
||||
# 过滤 .gitignore 中的文件
|
||||
files = self._filter_files_by_gitignore(files)
|
||||
|
||||
# 只使用第一个工具（已在 __init__ 中保证 self.check_tools 只有一个元素）
|
||||
tool = self.check_tools[0]
|
||||
logger.info(f"开始并行检查，文件数: {len(files)}，工具: {tool}")
|
||||
|
||||
all_results = []
|
||||
with ThreadPoolExecutor(max_workers=min(4, len(self.check_tools) * len(files))) as executor:
|
||||
futures = []
|
||||
for tool in self.check_tools:
|
||||
for file_path in files:
|
||||
futures.append(executor.submit(self.run_check, tool, file_path))
|
||||
with ThreadPoolExecutor(max_workers=min(4, len(files))) as executor:
|
||||
futures = [executor.submit(self.run_check, tool, file_path) for file_path in files]
|
||||
|
||||
for future in as_completed(futures):
|
||||
try:
|
||||
|
|
@ -180,7 +283,7 @@ class Checker:
|
|||
return all_results
|
||||
|
||||
def save_results(self, results: List[Dict[str, Any]]) -> None:
|
||||
"""保存检查结果到JSON文件"""
|
||||
"""保存检查结果到 JSON 文件"""
|
||||
try:
|
||||
with open(self.results_file, "w", encoding="utf-8") as f:
|
||||
json.dump(results, f, indent=2, ensure_ascii=False)
|
||||
|
|
@ -193,7 +296,7 @@ class Checker:
|
|||
从检查结果中收集所有错误
|
||||
|
||||
Args:
|
||||
results: 检查结果列表，如果为None则从文件加载
|
||||
results: 检查结果列表，如果为 None 则从文件加载
|
||||
|
||||
Returns:
|
||||
错误列表，每个错误包含文件、工具和错误信息
|
||||
|
|
@ -225,11 +328,11 @@ class Checker:
|
|||
|
||||
def auto_fix(self, errors: List[Dict[str, Any]], context_files: Optional[List[str]] = None) -> bool:
|
||||
"""
|
||||
自动调用LLM生成修复补丁并应用
|
||||
自动调用 LLM 生成修复补丁并应用
|
||||
|
||||
Args:
|
||||
errors: 错误列表，来自collect_errors
|
||||
context_files: 上下文文件路径列表，用于LLM生成修复
|
||||
errors: 错误列表，来自 collect_errors
|
||||
context_files: 上下文文件路径列表，用于 LLM 生成修复
|
||||
|
||||
Returns:
|
||||
bool: 修复是否成功（至少修复了一个错误）
|
||||
|
|
@ -240,16 +343,16 @@ class Checker:
|
|||
|
||||
logger.info(f"开始自动修复 {len(errors)} 个错误")
|
||||
|
||||
# 准备上下文：包括README、design.json和相关代码文件
|
||||
# 准备上下文：包括 README、design.json 和相关代码文件
|
||||
context_content = []
|
||||
|
||||
# 添加README（如果存在）
|
||||
# 添加 README（如果存在）
|
||||
readme_path = self.output_dir / "README.md"
|
||||
if readme_path.exists():
|
||||
with open(readme_path, "r", encoding="utf-8") as f:
|
||||
context_content.append(f"### 项目 README ###\n{f.read()}\n")
|
||||
|
||||
# 添加design.json（如果存在）
|
||||
# 添加 design.json（如果存在）
|
||||
design_path = self.output_dir / "design.json"
|
||||
if design_path.exists():
|
||||
with open(design_path, "r", encoding="utf-8") as f:
|
||||
|
|
@ -272,11 +375,11 @@ class Checker:
|
|||
|
||||
full_context = "\n".join(context_content)
|
||||
|
||||
# 调用LLM生成修复
|
||||
# 调用 LLM 生成修复
|
||||
system_prompt = (
|
||||
"你是一个专业的编程助手，擅长修复代码错误。根据提供的上下文（包括项目README、设计文件、相关代码和检查错误），"
|
||||
"生成修复补丁代码。返回严格的JSON对象，包含两个字段：\n"
|
||||
"- patches: 数组，每个元素是一个对象，包含'file'（文件路径）和'code'（修复后的完整代码或差异）\n"
|
||||
"你是一个专业的编程助手，擅长修复代码错误。根据提供的上下文（包括项目 README、设计文件、相关代码和检查错误），"
|
||||
"生成修复补丁代码。返回严格的 JSON 对象，包含两个字段：\n"
|
||||
"- patches: 数组，每个元素是一个对象，包含 'file'（文件路径）和 'code'（修复后的完整代码或差异）\n"
|
||||
"- description: 简短的中文修复描述\n"
|
||||
"注意：只修复提到的错误，保持代码风格一致。"
|
||||
)
|
||||
|
|
@ -286,7 +389,7 @@ class Checker:
|
|||
result = self.code_generator._call_llm(system_prompt, user_prompt, temperature=0.1)
|
||||
patches = result.get("patches", [])
|
||||
description = result.get("description", "无描述")
|
||||
logger.info(f"LLM生成修复补丁: {description}, 补丁数: {len(patches)}")
|
||||
logger.info(f"LLM 生成修复补丁: {description}, 补丁数: {len(patches)}")
|
||||
|
||||
# 应用补丁
|
||||
success_count = 0
|
||||
|
|
@ -310,7 +413,7 @@ class Checker:
|
|||
logger.info(f"自动修复完成，成功修复 {success_count}/{len(patches)} 个补丁")
|
||||
return success_count > 0
|
||||
except Exception as e:
|
||||
logger.error(f"调用LLM生成修复失败: {e}")
|
||||
logger.error(f"调用 LLM 生成修复失败: {e}")
|
||||
return False
|
||||
|
||||
def run_full_check_and_fix(self, max_retries: int = 3) -> bool:
|
||||
|
|
|
|||
|
|
@ -1,12 +1,12 @@
|
|||
#!/usr/bin/env python3
|
||||
"""
|
||||
LLM 代码生成工具的命令行接口
|
||||
支持 init、enhance、fix 三种操作模式，使用 typer 构建 CLI。
|
||||
支持 init、enhance、fix、check 四种操作模式，使用 typer 构建 CLI。
|
||||
"""
|
||||
|
||||
import sys
|
||||
from pathlib import Path
|
||||
from typing import Optional
|
||||
import sys
|
||||
|
||||
import typer
|
||||
from rich.console import Console
|
||||
|
|
@ -19,6 +19,19 @@ app = typer.Typer(help="基于LLM的自动化代码生成与维护工具")
|
|||
console = Console()
|
||||
|
||||
|
||||
def init_logging(output_dir: Path, log_file: Optional[str] = None, command_name: str = "cli") -> str:
|
||||
"""初始化日志配置到logs/目录"""
|
||||
log_dir = output_dir / "logs"
|
||||
log_dir.mkdir(parents=True, exist_ok=True)
|
||||
if log_file is None:
|
||||
log_file = str(log_dir / f"{command_name}.log")
|
||||
logger.remove() # 移除默认handler
|
||||
logger.add(sys.stderr, level="WARNING") # 控制台输出WARNING及以上
|
||||
logger.add(log_file, rotation="10 MB", level="DEBUG") # 文件记录DEBUG
|
||||
logger.info(f"日志已初始化到: {log_file}")
|
||||
return log_file
|
||||
|
||||
|
||||
@app.command()
|
||||
def init(
|
||||
readme: Path = typer.Argument(..., exists=True, file_okay=True, dir_okay=False, help="README.md 文件路径"),
|
||||
|
|
@ -28,21 +41,25 @@ def init(
|
|||
model: str = typer.Option("deepseek-reasoner", "--model", "-m", help="使用的模型"),
|
||||
log_file: Optional[str] = typer.Option(None, "--log", help="日志文件路径"),
|
||||
):
|
||||
"""
|
||||
初始化项目：根据 README.md 自动生成完整的代码。
|
||||
"""
|
||||
"""初始化项目：根据 README.md 自动生成完整的代码。"""
|
||||
if output_dir is None:
|
||||
output_dir = Path.cwd()
|
||||
|
||||
# 初始化日志配置
|
||||
log_file_path = init_logging(output_dir, log_file, command_name="init")
|
||||
|
||||
# 处理致命错误：检查README文件存在性（已由typer处理），其他错误在try块中捕获
|
||||
try:
|
||||
generator = CodeGenerator(
|
||||
api_key=api_key,
|
||||
base_url=base_url,
|
||||
model=model,
|
||||
output_dir=str(output_dir),
|
||||
log_file=log_file,
|
||||
log_file=log_file_path,
|
||||
)
|
||||
generator.run(readme)
|
||||
# 调用core.CodeGenerator.run并显示最终统计信息（假设从日志或生成器状态获取）
|
||||
console.print("[green]生成完成。成功处理文件，详情请查看日志。[/green]")
|
||||
except Exception as e:
|
||||
logger.error(f"初始化失败: {e}")
|
||||
raise typer.Exit(code=1)
|
||||
|
|
@ -57,12 +74,19 @@ def enhance(
|
|||
model: str = typer.Option("deepseek-reasoner", "--model", "-m", help="使用的模型"),
|
||||
log_file: Optional[str] = typer.Option(None, "--log", help="日志文件路径"),
|
||||
):
|
||||
"""
|
||||
增强项目：根据需求工单添加新功能。
|
||||
"""
|
||||
"""增强项目：根据需求工单添加新功能。"""
|
||||
if output_dir is None:
|
||||
output_dir = Path.cwd()
|
||||
|
||||
# 初始化日志配置
|
||||
log_file_path = init_logging(output_dir, log_file, command_name="enhance")
|
||||
|
||||
# 处理致命错误：检查design.json是否存在
|
||||
design_path = output_dir / "design.json"
|
||||
if not design_path.exists():
|
||||
logger.error(f"design.json 不存在于 {output_dir}，请先运行 init 命令初始化项目。")
|
||||
raise typer.Exit(code=1)
|
||||
|
||||
# 读取工单文件
|
||||
try:
|
||||
with open(issue_file, "r", encoding="utf-8") as f:
|
||||
|
|
@ -71,31 +95,19 @@ def enhance(
|
|||
logger.error(f"读取工单文件失败: {e}")
|
||||
raise typer.Exit(code=1)
|
||||
|
||||
# 检查 design.json 是否存在
|
||||
design_path = output_dir / "design.json"
|
||||
if not design_path.exists():
|
||||
logger.error(f"design.json 不存在于 {output_dir}，请先运行 init 命令初始化项目。")
|
||||
raise typer.Exit(code=1)
|
||||
|
||||
try:
|
||||
generator = CodeGenerator(
|
||||
api_key=api_key,
|
||||
base_url=base_url,
|
||||
model=model,
|
||||
output_dir=str(output_dir),
|
||||
log_file=log_file,
|
||||
log_file=log_file_path,
|
||||
)
|
||||
# 简化增强逻辑：基于工单内容调用 LLM 生成代码变更
|
||||
logger.info(f"处理增强工单: {issue_file}")
|
||||
console.print(f"[yellow]注意：增强功能为简化实现，基于工单内容生成变更。工单内容预览: {issue_content[:100]}...[/yellow]")
|
||||
# 实际应用中，这里应解析工单并调用 generator 或类似方法生成代码
|
||||
# 示例：生成一个占位文件或调用检查器
|
||||
checker = Checker(output_dir=output_dir, code_generator=generator)
|
||||
success = checker.run_full_check_and_fix()
|
||||
success = generator.process_issue(issue_content, issue_type="enhance")
|
||||
if not success:
|
||||
logger.error("增强过程中检查失败")
|
||||
logger.error("增强处理失败")
|
||||
raise typer.Exit(code=1)
|
||||
console.print("[green]增强处理完成，请检查生成的代码和日志。[/green]")
|
||||
console.print("[green]增强处理完成。成功处理文件，详情请查看日志。[/green]")
|
||||
except Exception as e:
|
||||
logger.error(f"增强失败: {e}")
|
||||
raise typer.Exit(code=1)
|
||||
|
|
@ -110,12 +122,19 @@ def fix(
|
|||
model: str = typer.Option("deepseek-reasoner", "--model", "-m", help="使用的模型"),
|
||||
log_file: Optional[str] = typer.Option(None, "--log", help="日志文件路径"),
|
||||
):
|
||||
"""
|
||||
修复项目：根据Bug工单自动修复 Bug。
|
||||
"""
|
||||
"""修复项目：根据Bug工单自动修复 Bug。"""
|
||||
if output_dir is None:
|
||||
output_dir = Path.cwd()
|
||||
|
||||
# 初始化日志配置
|
||||
log_file_path = init_logging(output_dir, log_file, command_name="fix")
|
||||
|
||||
# 处理致命错误：检查design.json是否存在
|
||||
design_path = output_dir / "design.json"
|
||||
if not design_path.exists():
|
||||
logger.error(f"design.json 不存在于 {output_dir}，请确保项目已初始化。")
|
||||
raise typer.Exit(code=1)
|
||||
|
||||
# 读取工单文件
|
||||
try:
|
||||
with open(issue_file, "r", encoding="utf-8") as f:
|
||||
|
|
@ -124,11 +143,39 @@ def fix(
|
|||
logger.error(f"读取工单文件失败: {e}")
|
||||
raise typer.Exit(code=1)
|
||||
|
||||
# 检查 design.json 是否存在
|
||||
design_path = output_dir / "design.json"
|
||||
if not design_path.exists():
|
||||
logger.error(f"design.json 不存在于 {output_dir}，请确保项目已初始化。")
|
||||
try:
|
||||
generator = CodeGenerator(
|
||||
api_key=api_key,
|
||||
base_url=base_url,
|
||||
model=model,
|
||||
output_dir=str(output_dir),
|
||||
log_file=log_file_path,
|
||||
)
|
||||
success = generator.process_issue(issue_content, issue_type="fix")
|
||||
if not success:
|
||||
logger.error("修复处理失败")
|
||||
raise typer.Exit(code=1)
|
||||
console.print("[green]修复处理完成。成功处理文件，详情请查看日志。[/green]")
|
||||
except Exception as e:
|
||||
logger.error(f"修复失败: {e}")
|
||||
raise typer.Exit(code=1)
|
||||
|
||||
|
||||
@app.command()
|
||||
def check(
|
||||
output_dir: Optional[Path] = typer.Option(None, "--output", "-o", help="项目根目录，默认为当前目录"),
|
||||
api_key: Optional[str] = typer.Option(None, "--api-key", envvar="DEEPSEEK_APIKEY", help="API密钥"),
|
||||
base_url: str = typer.Option("https://api.deepseek.com", "--base-url", help="API基础URL"),
|
||||
model: str = typer.Option("deepseek-reasoner", "--model", "-m", help="使用的模型"),
|
||||
log_file: Optional[str] = typer.Option(None, "--log", help="日志文件路径"),
|
||||
max_retries: int = typer.Option(3, "--max-retries", help="最大修复重试次数"),
|
||||
):
|
||||
"""运行代码检查和自动修复（不依赖于工单）"""
|
||||
if output_dir is None:
|
||||
output_dir = Path.cwd()
|
||||
|
||||
# 初始化日志配置
|
||||
log_file_path = init_logging(output_dir, log_file, command_name="check")
|
||||
|
||||
try:
|
||||
generator = CodeGenerator(
|
||||
|
|
@ -136,19 +183,16 @@ def fix(
|
|||
base_url=base_url,
|
||||
model=model,
|
||||
output_dir=str(output_dir),
|
||||
log_file=log_file,
|
||||
log_file=log_file_path,
|
||||
)
|
||||
# 简化修复逻辑：基于工单内容调用检查器进行修复
|
||||
logger.info(f"处理Bug工单: {issue_file}")
|
||||
console.print(f"[yellow]注意：修复功能为简化实现，基于工单内容调用检查器。工单内容预览: {issue_content[:100]}...[/yellow]")
|
||||
checker = Checker(output_dir=output_dir, code_generator=generator)
|
||||
success = checker.run_full_check_and_fix()
|
||||
success = checker.run_full_check_and_fix(max_retries=max_retries)
|
||||
if not success:
|
||||
logger.error("修复过程中检查失败")
|
||||
logger.error("检查修复失败")
|
||||
raise typer.Exit(code=1)
|
||||
console.print("[green]修复处理完成，请检查修复后的代码和日志。[/green]")
|
||||
console.print("[green]检查与修复完成。详情请查看日志。[/green]")
|
||||
except Exception as e:
|
||||
logger.error(f"修复失败: {e}")
|
||||
logger.error(f"检查失败: {e}")
|
||||
raise typer.Exit(code=1)
|
||||
|
||||
|
||||
|
|
|
|||
|
|
@ -45,6 +45,7 @@ class CodeGenerator:
|
|||
self.output_dir = Path(output_dir)
|
||||
self.output_dir.mkdir(parents=True, exist_ok=True)
|
||||
self.state_file = self.output_dir / ".llm_generator_state.json"
|
||||
self.console = Console() # 添加console实例用于rich打印
|
||||
|
||||
# 配置日志
|
||||
if log_file is None:
|
||||
|
|
@ -103,9 +104,11 @@ class CodeGenerator:
|
|||
|
||||
except json.JSONDecodeError as e:
|
||||
logger.error(f"JSON解析失败: {e}")
|
||||
self.console.print(f"[bold red]❌ JSON解析失败: {e}[/bold red]")
|
||||
raise ValueError(f"LLM返回的不是有效JSON: {content[:200]}")
|
||||
except Exception as e:
|
||||
logger.error(f"LLM调用失败: {e}")
|
||||
self.console.print(f"[bold red]❌ LLM调用失败: {e}[/bold red]")
|
||||
raise
|
||||
|
||||
def parse_readme(self, readme_path: Path) -> str:
|
||||
|
|
@ -120,6 +123,7 @@ class CodeGenerator:
|
|||
return content
|
||||
except Exception as e:
|
||||
logger.error(f"读取README失败: {e}")
|
||||
self.console.print(f"[bold red]❌ 读取README失败: {e}[/bold red]")
|
||||
raise
|
||||
|
||||
def generate_design_json(self) -> DesignModel:
|
||||
|
|
@ -140,7 +144,7 @@ class CodeGenerator:
|
|||
# 写入design.json文件
|
||||
design_path = self.output_dir / "design.json"
|
||||
with open(design_path, "w", encoding="utf-8") as f:
|
||||
json.dump(design.dict(), f, indent=2, ensure_ascii=False)
|
||||
json.dump(design.model_dump(), f, indent=2, ensure_ascii=False)
|
||||
logger.info(f"已生成design.json: {design_path}")
|
||||
|
||||
return design
|
||||
|
|
@ -156,6 +160,7 @@ class CodeGenerator:
|
|||
return self.state
|
||||
except Exception as e:
|
||||
logger.error(f"加载状态失败: {e}")
|
||||
self.console.print(f"[bold red]❌ 加载状态失败: {e}[/bold red]")
|
||||
return None
|
||||
return None
|
||||
|
||||
|
|
@ -170,7 +175,7 @@ class CodeGenerator:
|
|||
readme_path=self.readme_content[:100] if self.readme_content else ""
|
||||
)
|
||||
with open(self.state_file, "w", encoding="utf-8") as f:
|
||||
json.dump(state.dict(), f, indent=2, ensure_ascii=False)
|
||||
json.dump(state.model_dump(), f, indent=2, ensure_ascii=False)
|
||||
logger.debug(f"状态已保存: {self.state_file}")
|
||||
|
||||
def get_project_structure(self) -> Tuple[List[str], Dict[str, List[str]]]:
|
||||
|
|
@ -199,52 +204,91 @@ class CodeGenerator:
|
|||
file_path: str,
|
||||
prompt_instruction: str,
|
||||
dependency_files: List[str],
|
||||
existing_content: Optional[str] = None,
|
||||
) -> Tuple[str, str, List[str]]:
|
||||
"""
|
||||
生成单个文件，返回 (代码, 描述, 命令列表)
|
||||
|
||||
Args:
|
||||
file_path: 目标文件路径
|
||||
prompt_instruction: 生成指令
|
||||
dependency_files: 依赖文件列表（用于上下文）
|
||||
existing_content: 文件现有内容（若为修改模式）
|
||||
"""
|
||||
# 读取依赖文件内容
|
||||
# 收集上下文内容
|
||||
context_content = []
|
||||
|
||||
if self.readme_content:
|
||||
context_content.append(f"### 项目 README ###\n{self.readme_content}\n")
|
||||
|
||||
# 添加design.json上下文
|
||||
# 添加 design.json 上下文
|
||||
design_path = self.output_dir / "design.json"
|
||||
if design_path.exists():
|
||||
try:
|
||||
with open(design_path, "r", encoding="utf-8") as f:
|
||||
design_content = f.read()
|
||||
context_content.append(f"### 设计文件: design.json ###\n{design_content}\n")
|
||||
except Exception as e:
|
||||
logger.error(f"读取design.json失败: {e}")
|
||||
self.console.print(f"[bold red]❌ 读取design.json失败: {e}[/bold red]")
|
||||
# 如果design.json读取失败，可能无法继续，但保持上下文为空或部分
|
||||
|
||||
# 添加依赖文件内容（仅读取存在的文件）
|
||||
for dep in dependency_files:
|
||||
dep_path = Path(dep)
|
||||
if not dep_path.exists():
|
||||
# 尝试相对于当前目录或输出目录查找
|
||||
alt_path = self.output_dir / dep
|
||||
if alt_path.exists():
|
||||
dep_path = alt_path
|
||||
else:
|
||||
raise FileNotFoundError(f"依赖文件不存在: {dep}")
|
||||
logger.warning(f"依赖文件不存在，已跳过: {dep}")
|
||||
self.console.print(f"[yellow]⚠ 依赖文件不存在，已跳过: {dep}[/yellow]")
|
||||
continue
|
||||
|
||||
try:
|
||||
with open(dep_path, "r", encoding="utf-8") as f:
|
||||
content = f.read()
|
||||
context_content.append(f"### 文件: {dep_path.name} (路径: {dep}) ###\n{content}\n")
|
||||
except Exception as e:
|
||||
logger.error(f"读取依赖文件 {dep} 失败: {e}")
|
||||
self.console.print(f"[bold red]❌ 读取依赖文件 {dep} 失败: {e}[/bold red]")
|
||||
# 跳过此依赖文件
|
||||
|
||||
# 如果有现有内容，也加入上下文
|
||||
if existing_content is not None:
|
||||
context_content.append(f"### 当前文件内容 ({file_path}) ###\n{existing_content}\n")
|
||||
|
||||
full_context = "\n".join(context_content)
|
||||
|
||||
# 根据是否有现有内容调整系统提示
|
||||
if existing_content is not None:
|
||||
system_prompt = (
|
||||
"你是一个专业的编程助手。根据用户指令和提供的上下文文件，**修改**现有的代码文件。"
|
||||
"返回严格的 JSON 对象，包含三个字段：\n"
|
||||
"- code: (string) 修改后的完整代码\n"
|
||||
"- description: (string) 简短的中文修改描述\n"
|
||||
"- commands: (array of string) 修改此文件后需要执行的操作系统命令列表（如编译、安装依赖等），若无则返回空数组"
|
||||
)
|
||||
else:
|
||||
system_prompt = (
|
||||
"你是一个专业的编程助手。根据用户指令和提供的上下文文件，生成完整的代码。"
|
||||
"返回严格的JSON对象，包含三个字段：\n"
|
||||
"返回严格的 JSON 对象，包含三个字段：\n"
|
||||
"- code: (string) 生成的完整代码\n"
|
||||
"- description: (string) 简短的中文功能描述\n"
|
||||
"- commands: (array of string) 生成此文件后需要执行的操作系统命令列表（如编译、安装依赖等），若无则返回空数组"
|
||||
)
|
||||
|
||||
user_prompt = f"{prompt_instruction}\n\n参考文件上下文:\n{full_context}"
|
||||
|
||||
try:
|
||||
result = self._call_llm(system_prompt, user_prompt)
|
||||
llm_response = LLMResponse(**result)
|
||||
|
||||
return llm_response.code, llm_response.description, llm_response.commands
|
||||
except Exception as e:
|
||||
logger.error(f"生成文件 {file_path} 时调用LLM失败: {e}")
|
||||
self.console.print(f"[bold red]❌ 生成文件 {file_path} 时调用LLM失败: {e}[/bold red]")
|
||||
# 返回默认值以便继续
|
||||
return "# 生成失败，请检查日志", "生成失败，发生错误", []
|
||||
|
||||
def execute_command(self, cmd: str, cwd: Optional[Path] = None) -> bool:
|
||||
"""
|
||||
|
|
@ -256,6 +300,7 @@ class CodeGenerator:
|
|||
dangerous, reason = is_dangerous_command(cmd)
|
||||
if dangerous:
|
||||
logger.error(f"危险命令被阻止: {cmd}，原因: {reason}")
|
||||
self.console.print(f"[bold red]❌ 危险命令被阻止: {cmd}，原因: {reason}[/bold red]")
|
||||
return False
|
||||
|
||||
logger.info(f"执行命令: {cmd}")
|
||||
|
|
@ -275,52 +320,85 @@ class CodeGenerator:
|
|||
logger.warning(f"stderr: {result.stderr[:500]}")
|
||||
if result.returncode != 0:
|
||||
logger.error(f"命令执行失败，返回码: {result.returncode}")
|
||||
self.console.print(f"[bold red]❌ 命令执行失败，返回码: {result.returncode}[/bold red]")
|
||||
return False
|
||||
return True
|
||||
except subprocess.TimeoutExpired:
|
||||
logger.error(f"命令执行超时: {cmd}")
|
||||
self.console.print(f"[bold red]❌ 命令执行超时: {cmd}[/bold red]")
|
||||
return False
|
||||
except Exception as e:
|
||||
logger.error(f"命令执行失败: {e}")
|
||||
self.console.print(f"[bold red]❌ 命令执行失败: {e}[/bold red]")
|
||||
return False
|
||||
|
||||
def run(self, readme_path: Path):
|
||||
"""
|
||||
主执行流程，支持设计层生成和断点续写
|
||||
"""
|
||||
console = Console()
|
||||
logger.info("=" * 50)
|
||||
logger.info("开始代码生成流程")
|
||||
logger.info(f"README: {readme_path}")
|
||||
logger.info(f"输出目录: {self.output_dir}")
|
||||
|
||||
# 解析README
|
||||
console.print("[bold yellow]🔍 正在解析README...[/bold yellow]")
|
||||
self.console.print("[bold yellow]🔍 正在解析README...[/bold yellow]")
|
||||
try:
|
||||
self.readme_content = self.parse_readme(readme_path)
|
||||
except Exception as e:
|
||||
logger.error(f"解析README失败，无法继续: {e}")
|
||||
self.console.print(f"[bold red]❌ 解析README失败，无法继续: {e}[/bold red]")
|
||||
return # 致命错误，退出
|
||||
|
||||
# 加载状态
|
||||
state = self.load_state()
|
||||
if state:
|
||||
console.print(f"[green]✅ 检测到断点状态，从文件索引 {state.current_file_index} 继续[/green]")
|
||||
self.console.print(f"[green]✅ 检测到断点状态，从文件索引 {state.current_file_index} 继续[/green]")
|
||||
self.state = state
|
||||
# 从状态恢复设计，假设design.json已存在
|
||||
design_path = self.output_dir / "design.json"
|
||||
if design_path.exists():
|
||||
try:
|
||||
with open(design_path, "r", encoding="utf-8") as f:
|
||||
design_data = json.load(f)
|
||||
self.design = DesignModel(**design_data)
|
||||
else:
|
||||
console.print("[bold yellow]⚠ design.json不存在，重新生成...[/bold yellow]")
|
||||
except Exception as e:
|
||||
logger.error(f"加载design.json失败: {e}")
|
||||
self.console.print(f"[bold red]❌ 加载design.json失败: {e}[/bold red]")
|
||||
self.console.print("[bold yellow]⚠ design.json损坏，重新生成...[/bold yellow]")
|
||||
try:
|
||||
self.design = self.generate_design_json()
|
||||
except Exception as e2:
|
||||
logger.error(f"重新生成design.json失败: {e2}")
|
||||
self.console.print(f"[bold red]❌ 重新生成design.json失败: {e2}[/bold red]")
|
||||
return
|
||||
else:
|
||||
console.print("[bold yellow]📋 正在生成设计文件...[/bold yellow]")
|
||||
self.console.print("[bold yellow]⚠ design.json不存在，重新生成...[/bold yellow]")
|
||||
try:
|
||||
self.design = self.generate_design_json()
|
||||
except Exception as e:
|
||||
logger.error(f"生成design.json失败: {e}")
|
||||
self.console.print(f"[bold red]❌ 生成design.json失败: {e}[/bold red]")
|
||||
return
|
||||
else:
|
||||
self.console.print("[bold yellow]📋 正在生成设计文件...[/bold yellow]")
|
||||
try:
|
||||
self.design = self.generate_design_json()
|
||||
self.state = None
|
||||
except Exception as e:
|
||||
logger.error(f"生成design.json失败: {e}")
|
||||
self.console.print(f"[bold red]❌ 生成design.json失败: {e}[/bold red]")
|
||||
return
|
||||
|
||||
# 获取项目结构
|
||||
console.print("[bold yellow]📋 正在分析项目结构...[/bold yellow]")
|
||||
self.console.print("[bold yellow]📋 正在分析项目结构...[/bold yellow]")
|
||||
try:
|
||||
files, dependencies = self.get_project_structure()
|
||||
console.print(f"[green]✅ 解析完成，共 {len(files)} 个文件待生成[/green]")
|
||||
except Exception as e:
|
||||
logger.error(f"获取项目结构失败: {e}")
|
||||
self.console.print(f"[bold red]❌ 获取项目结构失败: {e}[/bold red]")
|
||||
return
|
||||
self.console.print(f"[green]✅ 解析完成，共 {len(files)} 个文件待生成[/green]")
|
||||
|
||||
# 断点续写：确定起始索引
|
||||
start_index = self.state.current_file_index if self.state else 0
|
||||
|
|
@ -332,7 +410,7 @@ class CodeGenerator:
|
|||
TextColumn("[progress.description]{task.description}"),
|
||||
BarColumn(),
|
||||
TextColumn("[progress.percentage]{task.percentage:>3.0f}%"),
|
||||
console=console,
|
||||
console=self.console,
|
||||
) as progress:
|
||||
self.progress = progress
|
||||
total_task = progress.add_task("[cyan]整体进度...", total=len(files))
|
||||
|
|
@ -354,10 +432,16 @@ class CodeGenerator:
|
|||
# 写入文件
|
||||
output_path = self.output_dir / file
|
||||
output_path.parent.mkdir(parents=True, exist_ok=True)
|
||||
try:
|
||||
with open(output_path, "w", encoding="utf-8") as f:
|
||||
f.write(code)
|
||||
logger.info(f"已写入: {output_path}")
|
||||
generated_files.append(file)
|
||||
except Exception as e:
|
||||
logger.error(f"写入文件 {file} 失败: {e}")
|
||||
self.console.print(f"[bold red]❌ 写入文件 {file} 失败: {e}[/bold red]")
|
||||
# 跳过命令执行
|
||||
commands = []
|
||||
|
||||
# 执行命令
|
||||
for cmd in commands:
|
||||
|
|
@ -368,9 +452,10 @@ class CodeGenerator:
|
|||
|
||||
except Exception as e:
|
||||
logger.error(f"处理文件 {file} 失败: {e}")
|
||||
# 保存状态以便断点续写
|
||||
self.console.print(f"[bold red]❌ 处理文件 {file} 时发生错误: {e}[/bold red]")
|
||||
# 不抛出异常，继续执行下一个文件
|
||||
# 保存状态
|
||||
self.save_state(idx, generated_files, dependencies)
|
||||
raise
|
||||
finally:
|
||||
progress.remove_task(file_task)
|
||||
progress.update(total_task, advance=1)
|
||||
|
|
@ -380,5 +465,230 @@ class CodeGenerator:
|
|||
logger.success("所有文件处理完成！")
|
||||
# 清理状态文件
|
||||
if self.state_file.exists():
|
||||
try:
|
||||
self.state_file.unlink()
|
||||
logger.info("状态文件已清理")
|
||||
except Exception as e:
|
||||
logger.error(f"清理状态文件失败: {e}")
|
||||
self.console.print(f"[bold red]❌ 清理状态文件失败: {e}[/bold red]")
|
||||
|
||||
|
||||
def process_issue(self, issue_content: str, issue_type: str) -> bool:
|
||||
"""
|
||||
处理需求增强或 Bug 修复工单
|
||||
|
||||
Args:
|
||||
issue_content: 工单文件内容（文本）
|
||||
issue_type: 'enhance' 或 'fix'
|
||||
|
||||
Returns:
|
||||
bool: 处理是否成功
|
||||
"""
|
||||
logger.info(f"开始处理 {issue_type} 工单")
|
||||
self.console.print(f"[bold yellow]📋 正在分析 {issue_type} 工单...[/bold yellow]")
|
||||
|
||||
# 加载现有 design.json
|
||||
design_path = self.output_dir / "design.json"
|
||||
if not design_path.exists():
|
||||
logger.error(f"design.json 不存在于 {self.output_dir}，请先运行 init 命令初始化项目。")
|
||||
self.console.print(f"[bold red]❌ design.json 不存在于 {self.output_dir}，请先运行 init 命令初始化项目。[/bold red]")
|
||||
return False
|
||||
|
||||
try:
|
||||
with open(design_path, "r", encoding="utf-8") as f:
|
||||
design_data = json.load(f)
|
||||
self.design = DesignModel(**design_data)
|
||||
except Exception as e:
|
||||
logger.error(f"加载design.json失败: {e}")
|
||||
self.console.print(f"[bold red]❌ 加载design.json失败: {e}[/bold red]")
|
||||
return False
|
||||
|
||||
# 加载 README 内容（如果存在）
|
||||
readme_path = self.output_dir / "README.md"
|
||||
if readme_path.exists():
|
||||
try:
|
||||
with open(readme_path, "r", encoding="utf-8") as f:
|
||||
self.readme_content = f.read()
|
||||
except Exception as e:
|
||||
logger.error(f"读取README.md失败: {e}")
|
||||
self.console.print(f"[bold red]❌ 读取README.md失败: {e}[/bold red]")
|
||||
self.readme_content = ""
|
||||
else:
|
||||
self.readme_content = ""
|
||||
|
||||
# 步骤1: 分析工单，生成变更计划
|
||||
try:
|
||||
change_plan = self._analyze_issue(issue_content, issue_type)
|
||||
except Exception as e:
|
||||
logger.error(f"分析工单失败: {e}")
|
||||
self.console.print(f"[bold red]❌ 分析工单失败: {e}[/bold red]")
|
||||
return False
|
||||
if not change_plan:
|
||||
logger.error("无法生成变更计划")
|
||||
self.console.print(f"[bold red]❌ 无法生成变更计划[/bold red]")
|
||||
return False
|
||||
|
||||
affected_files = change_plan.get("affected_files", [])
|
||||
if not affected_files:
|
||||
logger.warning("工单分析结果未指定任何受影响文件")
|
||||
self.console.print(f"[yellow]⚠ 工单分析结果未指定任何受影响文件[/yellow]")
|
||||
return True # 无变更
|
||||
|
||||
self.console.print(f"[green]✅ 分析完成，将处理 {len(affected_files)} 个文件[/green]")
|
||||
|
||||
# 步骤2: 逐个处理文件
|
||||
generated_files = []
|
||||
for file_info in affected_files:
|
||||
file_path = file_info["path"]
|
||||
action = file_info.get("action", "modify") # modify 或 create
|
||||
description = file_info.get("description", "")
|
||||
dependencies = file_info.get("dependencies", [])
|
||||
|
||||
logger.info(f"处理文件: {file_path} (操作: {action})")
|
||||
|
||||
# 读取现有内容（如果是修改）
|
||||
existing = None
|
||||
full_path = self.output_dir / file_path
|
||||
if action == "modify" and full_path.exists():
|
||||
try:
|
||||
with open(full_path, "r", encoding="utf-8") as f:
|
||||
existing = f.read()
|
||||
except Exception as e:
|
||||
logger.error(f"读取文件 {file_path} 失败: {e}")
|
||||
self.console.print(f"[bold red]❌ 读取文件 {file_path} 失败: {e}[/bold red]")
|
||||
existing = None # 如果读取失败，按新文件处理
|
||||
elif action == "create" and full_path.exists():
|
||||
logger.warning(f"文件 {file_path} 已存在，将覆盖")
|
||||
self.console.print(f"[yellow]⚠ 文件 {file_path} 已存在，将覆盖[/yellow]")
|
||||
existing = None # 创建模式，即使存在也按新文件处理
|
||||
|
||||
# 收集实际存在的依赖文件
|
||||
dep_paths = []
|
||||
missing_deps = []
|
||||
for dep in dependencies:
|
||||
dep_full = self.output_dir / dep
|
||||
if dep_full.exists():
|
||||
dep_paths.append(dep)
|
||||
else:
|
||||
missing_deps.append(dep)
|
||||
if missing_deps:
|
||||
logger.warning(f"依赖文件缺失，将不使用这些文件作为上下文: {missing_deps}")
|
||||
self.console.print(f"[yellow]⚠ 依赖文件缺失，将不使用这些文件作为上下文: {missing_deps}[/yellow]")
|
||||
|
||||
# 构建生成指令
|
||||
instruction = f"请根据工单描述{'修改' if action == 'modify' else '生成'}文件 '{file_path}'。\n"
|
||||
instruction += f"工单内容摘要：{description}\n"
|
||||
if action == "modify":
|
||||
instruction += "请在现有代码基础上进行修改，保持原有风格和功能不变。"
|
||||
else:
|
||||
instruction += "请生成完整的代码文件。"
|
||||
|
||||
# 调用 generate_file
|
||||
try:
|
||||
code, desc, commands = self.generate_file(
|
||||
file_path,
|
||||
instruction,
|
||||
dep_paths,
|
||||
existing_content=existing,
|
||||
)
|
||||
logger.info(f"生成完成: {file_path} - {desc}")
|
||||
|
||||
# 写入文件
|
||||
full_path.parent.mkdir(parents=True, exist_ok=True)
|
||||
try:
|
||||
with open(full_path, "w", encoding="utf-8") as f:
|
||||
f.write(code)
|
||||
logger.info(f"已写入: {full_path}")
|
||||
generated_files.append(file_path)
|
||||
except Exception as e:
|
||||
logger.error(f"写入文件 {file_path} 失败: {e}")
|
||||
self.console.print(f"[bold red]❌ 写入文件 {file_path} 失败: {e}[/bold red]")
|
||||
# 跳过命令执行
|
||||
commands = []
|
||||
|
||||
# 执行关联命令
|
||||
for cmd in commands:
|
||||
logger.info(f"准备执行命令: {cmd}")
|
||||
success = self.execute_command(cmd, cwd=self.output_dir)
|
||||
if not success:
|
||||
logger.warning(f"命令执行失败，但继续处理: {cmd}")
|
||||
|
||||
except Exception as e:
|
||||
logger.error(f"处理文件 {file_path} 失败: {e}")
|
||||
self.console.print(f"[bold red]❌ 处理文件 {file_path} 失败: {e}[/bold red]")
|
||||
# 继续处理其他文件
|
||||
continue
|
||||
|
||||
# 步骤3: 更新 design.json
|
||||
if generated_files:
|
||||
try:
|
||||
self._update_design(generated_files, change_plan.get("design_updates", {}))
|
||||
self.console.print("[green]✅ design.json 已更新[/green]")
|
||||
except Exception as e:
|
||||
logger.error(f"更新design.json失败: {e}")
|
||||
self.console.print(f"[bold red]❌ 更新design.json失败: {e}[/bold red]")
|
||||
|
||||
self.console.print(f"[bold green]🎉 {issue_type} 处理完成！[/bold green]")
|
||||
return True
|
||||
|
||||
|
||||
def _analyze_issue(self, issue_content: str, issue_type: str) -> Dict[str, Any]:
|
||||
"""
|
||||
调用 LLM 分析工单，返回结构化变更计划
|
||||
"""
|
||||
system_prompt = (
|
||||
"你是一个软件架构师。根据用户提供的工单内容和现有项目设计文件（design.json），"
|
||||
"分析需要进行的代码变更。返回严格的 JSON 对象，包含以下字段：\n"
|
||||
"- affected_files: 数组，每个元素为一个对象，包含：\n"
|
||||
" - path: 文件路径（相对于项目根目录）\n"
|
||||
" - action: 'create' 或 'modify'\n"
|
||||
" - description: 对此文件变更的简短描述\n"
|
||||
" - dependencies: 此文件可能依赖的其他文件路径列表（可选）\n"
|
||||
"- design_updates: 对象，描述对 design.json 的更新，例如新增的文件条目、修改的摘要等（可选）\n"
|
||||
"注意：仅返回 JSON，不要包含其他文本。"
|
||||
)
|
||||
|
||||
# 将现有 design.json 内容作为上下文的一部分
|
||||
design_str = json.dumps(self.design.dict(), indent=2, ensure_ascii=False)
|
||||
user_prompt = (
|
||||
f"工单类型: {issue_type}\n"
|
||||
f"工单内容:\n{issue_content}\n\n"
|
||||
f"现有设计文件 (design.json):\n{design_str}"
|
||||
)
|
||||
|
||||
result = self._call_llm(system_prompt, user_prompt, temperature=0.2)
|
||||
return result
|
||||
|
||||
def _update_design(self, generated_files: List[str], design_updates: Dict[str, Any]):
|
||||
"""
|
||||
根据生成的变更更新 design.json
|
||||
注意：假设 self.design.files 是 List[dict]（无 FileModel），直接操作字典
|
||||
"""
|
||||
updated = False
|
||||
|
||||
# 处理新增文件
|
||||
for file_path in generated_files:
|
||||
# 检查文件是否已在 design.files 中
|
||||
exists = any(f.get("path") == file_path for f in self.design.files)
|
||||
if not exists:
|
||||
# 创建新文件条目（字典）
|
||||
new_file = {
|
||||
"path": file_path,
|
||||
"summary": design_updates.get(file_path, {}).get("summary", "自动生成的新文件"),
|
||||
"dependencies": design_updates.get(file_path, {}).get("dependencies", []),
|
||||
"functions": [],
|
||||
"classes": [],
|
||||
}
|
||||
self.design.files.append(new_file)
|
||||
updated = True
|
||||
logger.info(f"已将新文件 {file_path} 添加到 design.json")
|
||||
|
||||
# 如果 design_updates 中提供了具体的更新信息，可以进一步处理（例如修改现有文件的摘要）
|
||||
# 这里可根据实际需求扩展，当前仅处理新增文件
|
||||
|
||||
if updated:
|
||||
# 保存更新后的 design.json
|
||||
design_path = self.output_dir / "design.json"
|
||||
with open(design_path, "w", encoding="utf-8") as f:
|
||||
json.dump(self.design.dict(), f, indent=2, ensure_ascii=False)
|
||||
logger.info("design.json 已更新")
|
||||
|
|
|
|||
|
|
@ -1,6 +1,7 @@
|
|||
from typing import Tuple
|
||||
import os
|
||||
from pathlib import Path
|
||||
from loguru import logger # 添加导入
|
||||
|
||||
# 危险命令列表，可配置
|
||||
DANGEROUS_COMMANDS = ["rm", "sudo", "chmod", "dd", "mkfs", "> /dev/sda", "format"]
|
||||
|
|
@ -84,3 +85,38 @@ def safe_join(base_path: str, *paths: str) -> str:
|
|||
if not full_path.startswith(base_abs):
|
||||
raise ValueError(f"路径拼接越界: {full_path} 不在 {base_abs} 下")
|
||||
return full_path
|
||||
|
||||
|
||||
def log_error(error: Exception, message: str = None, is_fatal: bool = False) -> None:
|
||||
"""
|
||||
记录和显示错误
|
||||
|
||||
Args:
|
||||
error: 异常对象
|
||||
message: 可选的自定义错误消息，如果为 None 则使用 error 的字符串表示
|
||||
is_fatal: 指示错误是否致命
|
||||
|
||||
Returns:
|
||||
None
|
||||
"""
|
||||
if message is None:
|
||||
message = str(error)
|
||||
log_msg = f"错误: {message}"
|
||||
if is_fatal:
|
||||
logger.critical(log_msg)
|
||||
else:
|
||||
logger.error(log_msg)
|
||||
|
||||
|
||||
def is_fatal_error(error: Exception) -> bool:
|
||||
"""
|
||||
判断错误类型是否为致命错误
|
||||
|
||||
Args:
|
||||
error: 异常对象
|
||||
|
||||
Returns:
|
||||
bool: 如果是致命错误返回 True，否则返回 False
|
||||
"""
|
||||
fatal_exceptions = (SystemExit, KeyboardInterrupt, MemoryError, OSError)
|
||||
return isinstance(error, fatal_exceptions)
|
||||
|
|
|
|||
|
|
@ -36,7 +36,6 @@ def checker(fake_code_generator, tmp_path):
|
|||
output_dir.mkdir()
|
||||
return Checker(
|
||||
output_dir=output_dir,
|
||||
check_tools=["pylint", "mypy", "black"],
|
||||
code_generator=fake_code_generator,
|
||||
)
|
||||
|
||||
|
|
@ -48,7 +47,7 @@ class TestChecker:
|
|||
def test_init(self, checker, tmp_path):
|
||||
"""测试初始化方法"""
|
||||
assert checker.output_dir == tmp_path / "test_output"
|
||||
assert checker.check_tools == ["pylint", "mypy", "black"]
|
||||
assert checker.check_tools == ["black"]
|
||||
assert checker.results_file == checker.output_dir / "check_results.json"
|
||||
assert isinstance(checker.code_generator, FakeCodeGenerator)
|
||||
|
||||
|
|
@ -56,11 +55,6 @@ class TestChecker:
|
|||
"""测试 run_check 方法成功运行检查工具"""
|
||||
file_path = Path("test_file.py")
|
||||
|
||||
# 模拟危险检测返回安全
|
||||
def fake_dangerous(cmd):
|
||||
return (False, "")
|
||||
monkeypatch.setattr("src.llm_codegen.checker.is_dangerous_command", fake_dangerous)
|
||||
|
||||
# 模拟 subprocess.run 返回成功
|
||||
def fake_run(cmd, *args, **kwargs):
|
||||
return subprocess.CompletedProcess(
|
||||
|
|
@ -78,29 +72,10 @@ class TestChecker:
|
|||
assert result["returncode"] == 0
|
||||
assert result["errors"] == []
|
||||
|
||||
def test_run_check_dangerous_command(self, checker, monkeypatch):
|
||||
"""测试 run_check 处理危险命令"""
|
||||
file_path = Path("test_file.py")
|
||||
|
||||
# 替换 is_dangerous_command 返回危险
|
||||
def fake_dangerous(cmd):
|
||||
return (True, "包含危险关键词 'rm'")
|
||||
monkeypatch.setattr("src.llm_codegen.checker.is_dangerous_command", fake_dangerous)
|
||||
|
||||
result = checker.run_check("rm -rf /", file_path)
|
||||
|
||||
assert result["returncode"] == -1
|
||||
assert "危险命令被阻止" in result["stderr"]
|
||||
|
||||
def test_run_check_timeout(self, checker, monkeypatch):
|
||||
"""测试 run_check 处理超时"""
|
||||
file_path = Path("test_file.py")
|
||||
|
||||
# 模拟危险检测返回安全
|
||||
def fake_dangerous(cmd):
|
||||
return (False, "")
|
||||
monkeypatch.setattr("src.llm_codegen.checker.is_dangerous_command", fake_dangerous)
|
||||
|
||||
# 让 subprocess.run 抛出超时异常
|
||||
def fake_run_timeout(*args, **kwargs):
|
||||
raise subprocess.TimeoutExpired(cmd="pylint", timeout=60)
|
||||
|
|
@ -131,9 +106,9 @@ class TestChecker:
|
|||
|
||||
results = checker.run_parallel_checks([test_file])
|
||||
|
||||
assert len(results) == 3
|
||||
assert len(results) == 1
|
||||
assert all(r["returncode"] == 0 for r in results)
|
||||
assert call_count == 3
|
||||
assert call_count == 1
|
||||
|
||||
def test_save_results(self, checker, tmp_path):
|
||||
"""测试保存检查结果"""
|
||||
|
|
|
|||
|
|
@ -11,21 +11,18 @@ runner = CliRunner()
|
|||
|
||||
def test_cli_init_success():
|
||||
"""测试 init 命令成功执行"""
|
||||
from src.llm_codegen.cli import app # 假设从项目根目录运行测试
|
||||
from src.llm_codegen.cli import app
|
||||
|
||||
# 模拟 CodeGenerator 和其方法，避免实际调用 API
|
||||
with patch('src.llm_codegen.cli.CodeGenerator') as mock_generator:
|
||||
mock_instance = Mock()
|
||||
mock_instance.run = Mock()
|
||||
mock_generator.return_value = mock_instance
|
||||
|
||||
# 创建一个虚拟的 README 文件用于测试
|
||||
test_readme = Path("test_readme.md")
|
||||
test_readme.write_text("# Test Project\n\nA test project for CLI.")
|
||||
|
||||
result = runner.invoke(app, ["init", str(test_readme), "--output", "./test_output"])
|
||||
|
||||
# 清理
|
||||
test_readme.unlink()
|
||||
|
||||
assert result.exit_code == 0
|
||||
|
|
@ -34,61 +31,64 @@ def test_cli_init_success():
|
|||
mock_instance.run.assert_called_once_with(test_readme)
|
||||
|
||||
|
||||
|
||||
def test_cli_init_failure_no_readme():
|
||||
"""测试 init 命令当 README 不存在时失败"""
|
||||
from src.llm_codegen.cli import app
|
||||
|
||||
result = runner.invoke(app, ["init", "nonexistent.md"])
|
||||
|
||||
assert result.exit_code != 0 # 应该退出码非零
|
||||
assert result.exit_code != 0
|
||||
|
||||
|
||||
|
||||
def test_cli_enhance_success():
|
||||
"""测试 enhance 命令成功执行（简化版，基于工单）"""
|
||||
"""测试 enhance 命令成功执行（基于新实现，使用 process_issue）"""
|
||||
from src.llm_codegen.cli import app
|
||||
|
||||
# 模拟依赖文件和环境
|
||||
# 模拟 CodeGenerator 和 Path.exists（仅使 design.json 存在判断通过）
|
||||
with patch('src.llm_codegen.cli.CodeGenerator') as mock_generator, \
|
||||
patch('src.llm_codegen.cli.Checker') as mock_checker, \
|
||||
patch('pathlib.Path.exists') as mock_exists:
|
||||
|
||||
mock_exists.return_value = True # 模拟 design.json 存在
|
||||
mock_instance = Mock()
|
||||
mock_instance.run_full_check_and_fix = Mock(return_value=True)
|
||||
mock_checker.return_value = mock_instance
|
||||
mock_generator.return_value = Mock()
|
||||
# 使 design.json 存在检查返回 True
|
||||
mock_exists.return_value = True
|
||||
|
||||
# 创建一个虚拟的工单文件
|
||||
mock_instance = Mock()
|
||||
mock_instance.process_issue = Mock(return_value=True)
|
||||
mock_generator.return_value = mock_instance
|
||||
|
||||
# 创建临时工单文件
|
||||
test_issue = Path("test_feature.issue")
|
||||
test_issue.write_text("name: Add feature\ndescription: Test feature")
|
||||
issue_content = "name: Add feature\ndescription: Test feature"
|
||||
test_issue.write_text(issue_content)
|
||||
|
||||
result = runner.invoke(app, ["enhance", str(test_issue), "--output", "./test_output"])
|
||||
|
||||
# 清理
|
||||
test_issue.unlink()
|
||||
|
||||
assert result.exit_code == 0
|
||||
assert "增强失败" not in result.stdout
|
||||
mock_checker.assert_called_once()
|
||||
mock_instance.run_full_check_and_fix.assert_called_once()
|
||||
mock_generator.assert_called_once()
|
||||
mock_instance.process_issue.assert_called_once_with(issue_content, issue_type="enhance")
|
||||
|
||||
|
||||
|
||||
def test_cli_fix_success():
|
||||
"""测试 fix 命令成功执行（简化版，基于工单）"""
|
||||
"""测试 fix 命令成功执行（基于新实现，使用 process_issue）"""
|
||||
from src.llm_codegen.cli import app
|
||||
|
||||
with patch('src.llm_codegen.cli.CodeGenerator') as mock_generator, \
|
||||
patch('src.llm_codegen.cli.Checker') as mock_checker, \
|
||||
patch('pathlib.Path.exists') as mock_exists:
|
||||
|
||||
mock_exists.return_value = True
|
||||
|
||||
mock_instance = Mock()
|
||||
mock_instance.run_full_check_and_fix = Mock(return_value=True)
|
||||
mock_checker.return_value = mock_instance
|
||||
mock_generator.return_value = Mock()
|
||||
mock_instance.process_issue = Mock(return_value=True)
|
||||
mock_generator.return_value = mock_instance
|
||||
|
||||
test_issue = Path("test_bug.issue")
|
||||
test_issue.write_text("name: Fix bug\ndescription: Test bug")
|
||||
issue_content = "name: Fix bug\ndescription: Test bug"
|
||||
test_issue.write_text(issue_content)
|
||||
|
||||
result = runner.invoke(app, ["fix", str(test_issue), "--output", "./test_output"])
|
||||
|
||||
|
|
@ -96,8 +96,8 @@ def test_cli_fix_success():
|
|||
|
||||
assert result.exit_code == 0
|
||||
assert "修复失败" not in result.stdout
|
||||
mock_checker.assert_called_once()
|
||||
mock_instance.run_full_check_and_fix.assert_called_once()
|
||||
mock_generator.assert_called_once()
|
||||
mock_instance.process_issue.assert_called_once_with(issue_content, issue_type="fix")
|
||||
|
||||
|
||||
def test_cli_help():
|
||||
|
|
@ -108,7 +108,6 @@ def test_cli_help():
|
|||
assert result.exit_code == 0
|
||||
assert "基于LLM的自动化代码生成与维护工具" in result.stdout
|
||||
|
||||
# 测试子命令帮助
|
||||
result = runner.invoke(app, ["init", "--help"])
|
||||
assert result.exit_code == 0
|
||||
assert "README.md 文件路径" in result.stdout
|
||||
|
|
@ -119,7 +118,7 @@ def test_cli_enhance_no_design():
|
|||
from src.llm_codegen.cli import app
|
||||
|
||||
with patch('pathlib.Path.exists') as mock_exists:
|
||||
mock_exists.return_value = False # 模拟 design.json 不存在
|
||||
mock_exists.return_value = False # design.json 不存在
|
||||
|
||||
test_issue = Path("test_feature.issue")
|
||||
test_issue.write_text("name: Test")
|
||||
|
|
@ -147,6 +146,27 @@ def test_cli_fix_no_design():
|
|||
|
||||
assert result.exit_code != 0
|
||||
|
||||
def test_cli_check_success():
|
||||
"""测试 check 命令成功执行"""
|
||||
from src.llm_codegen.cli import app
|
||||
|
||||
with patch('src.llm_codegen.cli.CodeGenerator') as mock_generator, \
|
||||
patch('src.llm_codegen.cli.Checker') as mock_checker:
|
||||
|
||||
mock_gen_instance = Mock()
|
||||
mock_generator.return_value = mock_gen_instance
|
||||
|
||||
mock_check_instance = Mock()
|
||||
mock_check_instance.run_full_check_and_fix = Mock(return_value=True)
|
||||
mock_checker.return_value = mock_check_instance
|
||||
|
||||
result = runner.invoke(app, ["check", "--output", "./test_output"])
|
||||
|
||||
assert result.exit_code == 0
|
||||
assert "检查与修复完成" in result.stdout
|
||||
mock_checker.assert_called_once()
|
||||
mock_check_instance.run_full_check_and_fix.assert_called_once_with(max_retries=3)
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
pytest.main([__file__])
|
||||
|
|
|
|||
Loading…
Reference in New Issue