#!/usr/bin/env python3 """ LLM 代码生成工具的命令行接口 支持 init、enhance、fix、check 四种操作模式，使用 typer 构建 CLI。 """ from pathlib import Path from typing import Optional import sys import typer from rich.console import Console from rich.progress import Progress, SpinnerColumn, TextColumn, BarColumn from loguru import logger from .core import CodeGenerator from .checker import Checker 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() # 初始化日志配置 log_file_path = 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=None) generator = CodeGenerator( api_key=api_key, base_url=base_url, model=model, output_dir=str(output_dir), log_file=log_file_path, max_concurrency=max_concurrency, ) generator.run(readme) progress.update(task_id, 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() # 初始化日志配置 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: issue_content = f.read() except Exception as e: logger.error(f"读取工单文件失败: {e}") 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=None) generator = CodeGenerator( api_key=api_key, base_url=base_url, model=model, output_dir=str(output_dir), log_file=log_file_path, max_concurrency=max_concurrency, ) success = generator.process_issue(issue_content, issue_type="enhance") 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) """ with Progress( SpinnerColumn(), TextColumn("[progress.description]{task.description}"), BarColumn(), console=console ) as progress: task_id = progress.add_task("正在增强项目...", total=None) generator = CodeGenerator( api_key=api_key, base_url=base_url, model=model, output_dir=str(output_dir), log_file=log_file_path, max_concurrency=max_concurrency, ) success = generator.process_issue(issue_content, issue_type="enhance") 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-reasoner", "--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() # 初始化日志配置 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: issue_content = f.read() except Exception as e: logger.error(f"读取工单文件失败: {e}") 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=None) generator = CodeGenerator( api_key=api_key, base_url=base_url, model=model, output_dir=str(output_dir), log_file=log_file_path, max_concurrency=max_concurrency, ) success = generator.process_issue(issue_content, issue_type="fix") 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 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="最大修复重试次数"), max_concurrency: int = typer.Option(4, "--max-concurrency", help="并发生成的最大工作线程数，默认4"), ): """运行代码检查和自动修复（不依赖于工单）""" if output_dir is None: output_dir = Path.cwd() # 初始化日志配置 log_file_path = init_logging(output_dir, log_file, command_name="check") try: generator = CodeGenerator( api_key=api_key, base_url=base_url, model=model, output_dir=str(output_dir), log_file=log_file_path, max_concurrency=max_concurrency, ) checker = Checker(output_dir=output_dir, code_generator=generator) success = checker.run_full_check_and_fix(max_retries=max_retries) 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) if __name__ == "__main__": app()