378 lines
16 KiB
Python
378 lines
16 KiB
Python
#!/usr/bin/env python3
|
||
"""
|
||
LLM 代码生成工具的命令行接口
|
||
支持 init、enhance、fix、check、design 五种操作模式，使用 typer 构建 CLI。
|
||
"""
|
||
|
||
from pathlib import Path
|
||
from typing import Optional, List
|
||
import sys
|
||
import traceback
|
||
|
||
import typer
|
||
from rich.console import Console
|
||
from rich.progress import Progress, SpinnerColumn, TextColumn, BarColumn
|
||
from loguru import logger
|
||
|
||
from .init_generator import InitGenerator
|
||
from .enhance_generator import EnhanceGenerator
|
||
from .fix_generator import FixGenerator
|
||
from .design_generator import DesignGenerator # 新增导入DesignGenerator
|
||
|
||
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 文件路径"
|
||
),
|
||
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_concurrency: int = typer.Option(
|
||
4, "--max-concurrency", help="并发生成的最大工作线程数，默认4"
|
||
),
|
||
):
|
||
"""初始化项目：根据 README.md 自动生成完整的代码。"""
|
||
if output_dir is None:
|
||
output_dir = Path.cwd()
|
||
|
||
# 初始化日志配置
|
||
init_logging(output_dir, log_file, command_name="init")
|
||
|
||
# 处理致命错误：检查README文件存在性（已由typer处理），其他错误在try块中捕获
|
||
try:
|
||
with Progress(
|
||
SpinnerColumn(),
|
||
TextColumn("[progress.description]{task.description}"),
|
||
BarColumn(),
|
||
console=console,
|
||
) as progress:
|
||
task_id = progress.add_task("正在初始化项目...", total=1) # 修改：设置总任务数为1以控制进度显示
|
||
generator = InitGenerator(
|
||
api_key=api_key,
|
||
base_url=base_url,
|
||
model=model,
|
||
output_dir=str(output_dir),
|
||
max_concurrency=max_concurrency,
|
||
)
|
||
generator.run(readme)
|
||
progress.update(task_id, completed=1, description="初始化完成") # 修改：更新完成状态
|
||
# 调用core.CodeGenerator.run并显示最终统计信息（假设从日志或生成器状态获取）
|
||
console.print("[green]生成完成。成功处理文件，详情请查看日志。[/green]")
|
||
except Exception as e:
|
||
logger.error(f"初始化失败: {e}")
|
||
raise typer.Exit(code=1)
|
||
|
||
|
||
@app.command()
|
||
def enhance(
|
||
issue_file: Path = typer.Argument(
|
||
...,
|
||
exists=True,
|
||
file_okay=True,
|
||
dir_okay=False,
|
||
help="需求工单文件路径（如 feature.issue）",
|
||
),
|
||
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_concurrency: int = typer.Option(
|
||
4, "--max-concurrency", help="并发生成的最大工作线程数，默认4"
|
||
),
|
||
):
|
||
"""增强项目：根据需求工单添加新功能。"""
|
||
if output_dir is None:
|
||
output_dir = Path.cwd()
|
||
|
||
# 初始化日志配置
|
||
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)
|
||
|
||
with Progress(
|
||
SpinnerColumn(),
|
||
TextColumn("[progress.description]{task.description}"),
|
||
BarColumn(),
|
||
console=console,
|
||
) as progress:
|
||
task_id = progress.add_task("正在增强项目...", total=1) # 修改：设置总任务数为1以控制进度显示
|
||
generator = EnhanceGenerator(
|
||
api_key=api_key,
|
||
base_url=base_url,
|
||
model=model,
|
||
output_dir=str(output_dir),
|
||
max_concurrency=max_concurrency,
|
||
)
|
||
success, affected_files = generator.process_enhance(issue_file, output_format="full")
|
||
if success:
|
||
progress.update(task_id, completed=1, description="增强处理完成") # 修改：成功时更新完成状态
|
||
generator = DesignGenerator(
|
||
api_key=api_key,
|
||
base_url=base_url,
|
||
model=model,
|
||
output_dir=str(output_dir),
|
||
max_concurrency=max_concurrency,
|
||
)
|
||
for file_path in affected_files:
|
||
generator.update_single_file_design(file_path)
|
||
else:
|
||
progress.update(task_id, description="增强处理失败") # 修改：失败时更新描述
|
||
if not success:
|
||
logger.error("增强处理失败")
|
||
raise typer.Exit(code=1)
|
||
console.print("[green]增强处理完成。成功处理文件，详情请查看日志。[/green]")
|
||
|
||
|
||
@app.command()
|
||
def fix(
|
||
issue_file: Path = typer.Argument(
|
||
...,
|
||
exists=True,
|
||
file_okay=True,
|
||
dir_okay=False,
|
||
help="Bug工单文件路径（如 bug.issue）",
|
||
),
|
||
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-chat", "--model", "-m", help="使用的模型"),
|
||
log_file: Optional[str] = typer.Option(None, "--log", help="日志文件路径"),
|
||
max_concurrency: int = typer.Option(
|
||
4, "--max-concurrency", help="并发生成的最大工作线程数，默认4"
|
||
),
|
||
):
|
||
"""修复项目：根据Bug工单自动修复 Bug。"""
|
||
if output_dir is None:
|
||
output_dir = Path.cwd()
|
||
|
||
# 初始化日志配置
|
||
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 Progress(
|
||
SpinnerColumn(),
|
||
TextColumn("[progress.description]{task.description}"),
|
||
BarColumn(),
|
||
console=console
|
||
) as progress:
|
||
task_id = progress.add_task("正在修复项目...", total=1) # 修改：设置总任务数为1以控制进度显示
|
||
generator = FixGenerator(
|
||
api_key=api_key,
|
||
base_url=base_url,
|
||
model=model,
|
||
output_dir=str(output_dir),
|
||
max_concurrency=max_concurrency,
|
||
)
|
||
(success, affected_files) = generator.process_fix(issue_file, output_format="full")
|
||
if success:
|
||
progress.update(task_id, completed=1, description="修复处理完成") # 修改：成功时更新完成状态
|
||
generator = DesignGenerator(
|
||
api_key=api_key,
|
||
base_url=base_url,
|
||
model=model,
|
||
output_dir=str(output_dir),
|
||
max_concurrency=max_concurrency,
|
||
)
|
||
for file_path in affected_files:
|
||
generator.update_single_file_design(file_path)
|
||
else:
|
||
progress.update(task_id, description="修复处理失败") # 修改：失败时更新描述
|
||
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 design(
|
||
file: Optional[Path] = typer.Option(
|
||
None,
|
||
"--file",
|
||
"-f",
|
||
help="README文件路径，用于生成design.json；如果与--source同时使用，则生成design.json后从源代码刷新",
|
||
exists=True,
|
||
file_okay=True,
|
||
dir_okay=False,
|
||
),
|
||
source: Optional[Path] = typer.Option(
|
||
None,
|
||
"--source",
|
||
help="源代码目录路径，用于从源代码刷新design.json；必须为目录",
|
||
exists=True,
|
||
file_okay=False,
|
||
dir_okay=True,
|
||
),
|
||
single_files: Optional[List[Path]] = typer.Option(
|
||
None,
|
||
"--single-file",
|
||
help="指定单个或多个文件路径来更新design.json中的条目，可以是相对或绝对路径；支持多次使用以指定多个文件，保持向后兼容全量生成行为",
|
||
exists=True,
|
||
file_okay=True,
|
||
dir_okay=False,
|
||
),
|
||
output_dir: Optional[Path] = typer.Option(
|
||
None, "--output", "-o", help="输出目录，design.json将保存在此，默认为当前目录"
|
||
),
|
||
force: bool = typer.Option(False, "--force", help="强制覆盖已存在的design.json，或强制从源代码刷新"),
|
||
dry_run: bool = typer.Option(False, "--dry-run", 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-chat", "--model", "-m", help="使用的模型"),
|
||
log_file: Optional[str] = typer.Option(None, "--log", help="日志文件路径"),
|
||
max_concurrency: int = typer.Option(
|
||
4, "--max-concurrency", help="并发生成的最大工作线程数，默认4"
|
||
),
|
||
):
|
||
"""生成或更新design.json：支持从README生成、从源代码刷新，并集成新的设计生成逻辑，新增单个或多个文件更新功能。"""
|
||
if output_dir is None:
|
||
output_dir = Path.cwd()
|
||
|
||
# 初始化日志配置
|
||
init_logging(output_dir, log_file, command_name="design")
|
||
|
||
# 检查是否提供了至少一个操作参数
|
||
if file is None and source is None and not single_files:
|
||
logger.error("必须提供 --file、--source 或 --single-file 参数之一来执行操作。")
|
||
raise typer.Exit(code=1)
|
||
|
||
# 初始化DesignGenerator以集成新的设计生成逻辑
|
||
try:
|
||
with Progress(
|
||
SpinnerColumn(),
|
||
TextColumn("[progress.description]{task.description}"),
|
||
BarColumn(),
|
||
console=console,
|
||
) as progress:
|
||
task_id = progress.add_task("正在处理design命令...", total=1)
|
||
generator = DesignGenerator(
|
||
api_key=api_key,
|
||
base_url=base_url,
|
||
model=model,
|
||
output_dir=str(output_dir),
|
||
max_concurrency=max_concurrency,
|
||
)
|
||
|
||
design_path = output_dir / "design.json"
|
||
|
||
# 处理--dry-run选项
|
||
if dry_run:
|
||
console.print("[yellow]模拟运行模式：不会实际写入文件或执行命令。[/yellow]")
|
||
if single_files:
|
||
for f in single_files:
|
||
console.print(f"[blue]模拟：将更新文件 {f} 在design.json中的条目[/blue]")
|
||
if file is not None:
|
||
logger.info(f"模拟：将从README文件 {file} 生成design.json")
|
||
# 在dry-run模式下，仅模拟解析README
|
||
content = generator.parse_readme(file)
|
||
console.print(f"[blue]模拟解析README内容完成，长度: {len(content)} 字符[/blue]")
|
||
if source is not None:
|
||
logger.info(f"模拟：将从源代码目录 {source} 刷新design.json")
|
||
# 模拟分析源代码
|
||
design_info = generator.analyze_source_files(source)
|
||
console.print(f"[blue]模拟分析完成，共分析 {len(design_info['files'])} 个文件[/blue]")
|
||
progress.update(task_id, completed=1, description="模拟运行完成")
|
||
console.print("[green]✅ 模拟运行完成，无实际文件操作。[/green]")
|
||
return
|
||
|
||
# 实际运行逻辑
|
||
if single_files:
|
||
# 处理单个或多个文件更新
|
||
logger.info(f"开始处理文件更新，共 {len(single_files)} 个文件")
|
||
for f in single_files:
|
||
success = generator.update_single_file_design(f)
|
||
if not success:
|
||
logger.error(f"更新文件 {f} 失败")
|
||
raise typer.Exit(code=1)
|
||
console.print("[green]✅ 所有指定文件已更新design.json[/green]")
|
||
progress.update(task_id, completed=1, description="文件更新完成")
|
||
return # 处理完文件更新后退出，不执行其他全量生成逻辑
|
||
|
||
# 全量生成行为（保持向后兼容）
|
||
if file is not None:
|
||
# 生成design.json从README
|
||
if not force and design_path.exists():
|
||
logger.error(
|
||
f"design.json 已存在于 {design_path}。使用 --force 参数以强制覆盖。"
|
||
)
|
||
raise typer.Exit(code=1)
|
||
generator.run(readme_path=file)
|
||
logger.info(f"已从README生成design.json: {design_path}")
|
||
|
||
if source is not None:
|
||
# 从源代码刷新design.json
|
||
success = generator.refresh_design_from_source(source)
|
||
if not success:
|
||
logger.error("从源代码刷新design.json失败")
|
||
raise typer.Exit(code=1)
|
||
logger.info(f"已从源代码刷新design.json: {design_path}")
|
||
|
||
progress.update(task_id, completed=1, description="design命令处理完成")
|
||
console.print(f"[green]✅ design.json 已处理完成，路径: {design_path}[/green]")
|
||
except Exception as e:
|
||
error_message = traceback.format_exc()
|
||
logger.error(f"处理design命令失败: {e}, 堆栈跟踪: {error_message}")
|
||
raise typer.Exit(code=1)
|
||
|
||
|
||
if __name__ == "__main__":
|
||
app()
|