# utils/output_writer.py - 代码文件 & JSON 输出 & 项目信息渲染 import os import json from pathlib import Path from typing import Dict, List, Any import config VALID_TYPES = { "integer", "string", "boolean", "float", "list", "dict", "object", "void", "any", } VALID_INOUT = {"in", "out", "inout"} # ══════════════════════════════════════════════════════ # 目录管理 # ══════════════════════════════════════════════════════ def build_project_output_dir(project_name: str) -> str: safe = "".join(c if c.isalnum() or c in "-_" else "_" for c in project_name) return os.path.join(config.OUTPUT_BASE_DIR, safe) def ensure_project_dir(project_name: str) -> str: """确保项目根输出目录存在,并创建 __init__.py""" output_dir = build_project_output_dir(project_name) os.makedirs(output_dir, exist_ok=True) init_file = os.path.join(output_dir, "__init__.py") if not os.path.exists(init_file): Path(init_file).write_text("# Auto-generated project package\n", encoding="utf-8") return output_dir def ensure_module_dir(output_dir: str, module: str) -> str: """确保模块子目录存在,并创建 __init__.py""" module_dir = os.path.join(output_dir, module) os.makedirs(module_dir, exist_ok=True) init_file = os.path.join(module_dir, "__init__.py") if not os.path.exists(init_file): Path(init_file).write_text( f"# Auto-generated module package: {module}\n", encoding="utf-8" ) return module_dir # ══════════════════════════════════════════════════════ # README # ══════════════════════════════════════════════════════ def write_project_readme( output_dir: str, project_name: str, project_description: str, requirements_summary: str, modules: List[str] = None, ) -> str: """生成项目 README.md""" module_section = "" if modules: module_list = "\n".join(f"- `{m}/`" for m in sorted(set(modules))) module_section = f"\n## 功能模块\n\n{module_list}\n" content = f"""# {project_name} > Auto-generated by Requirement Analyzer {project_description or ""} {module_section} ## 功能需求列表 {requirements_summary} """ path = os.path.join(output_dir, "README.md") Path(path).write_text(content, encoding="utf-8") return path # ══════════════════════════════════════════════════════ # 函数签名 JSON 导出 # ══════════════════════════════════════════════════════ def build_signatures_document( project_name: str, project_description: str, signatures: List[dict], ) -> dict: return { "project": project_name, "description": project_description or "", "functions": signatures, } def patch_signatures_with_url( signatures: List[dict], func_name_to_url: Dict[str, str], ) -> List[dict]: for sig in signatures: url = func_name_to_url.get(sig.get("name", ""), "") _insert_field_after(sig, after_key="type", new_key="url", new_value=url) return signatures def _insert_field_after(d: dict, after_key: str, new_key: str, new_value) -> None: if new_key in d: d[new_key] = new_value return items = list(d.items()) insert_pos = next((i + 1 for i, (k, _) in enumerate(items) if k == after_key), len(items)) items.insert(insert_pos, (new_key, new_value)) d.clear() d.update(items) def write_function_signatures_json( output_dir: str, signatures: List[dict], project_name: str, project_description: str, file_name: str = "function_signatures.json", ) -> str: os.makedirs(output_dir, exist_ok=True) document = build_signatures_document(project_name, project_description, signatures) file_path = os.path.join(output_dir, file_name) with open(file_path, "w", encoding="utf-8") as f: json.dump(document, f, ensure_ascii=False, indent=2) return file_path # ══════════════════════════════════════════════════════ # 签名结构校验 # ══════════════════════════════════════════════════════ def validate_signature_schema(signature: dict) -> List[str]: errors: List[str] = [] for key in ("name", "requirement_id", "description", "type", "parameters"): if key not in signature: errors.append(f"缺少顶层字段: '{key}'") if "url" in signature: if not isinstance(signature["url"], str): errors.append("'url' 字段必须是字符串类型") elif signature["url"] == "": errors.append("'url' 字段不能为空字符串") params = signature.get("parameters", {}) if isinstance(params, dict): for pname, pdef in params.items(): if not isinstance(pdef, dict): errors.append(f"参数 '{pname}' 定义必须是 dict") continue if "type" not in pdef: errors.append(f"参数 '{pname}' 缺少 'type'") else: parts = [p.strip() for p in pdef["type"].split("|")] if not all(p in VALID_TYPES for p in parts): errors.append(f"参数 '{pname}' type='{pdef['type']}' 含不合法类型") if "inout" not in pdef: errors.append(f"参数 '{pname}' 缺少 'inout'") elif pdef["inout"] not in VALID_INOUT: errors.append(f"参数 '{pname}' inout='{pdef['inout']}' 应为 in/out/inout") if "required" not in pdef: errors.append(f"参数 '{pname}' 缺少 'required'") elif not isinstance(pdef["required"], bool): errors.append(f"参数 '{pname}' 'required' 应为布尔值") ret = signature.get("return") if ret is None: errors.append("缺少 'return' 字段") elif isinstance(ret, dict): ret_type = ret.get("type") if not ret_type: errors.append("'return' 缺少 'type'") elif ret_type not in VALID_TYPES: errors.append(f"'return.type'='{ret_type}' 不合法") is_void = (ret_type == "void") for sub_key in ("on_success", "on_failure"): sub = ret.get(sub_key) if is_void: if sub is not None: errors.append(f"void 函数 'return.{sub_key}' 应为 null") else: if sub is None: errors.append(f"非 void 函数缺少 'return.{sub_key}'") elif isinstance(sub, dict): if not sub.get("value"): errors.append(f"'return.{sub_key}.value' 不能为空") if not sub.get("description"): errors.append(f"'return.{sub_key}.description' 不能为空") return errors def validate_all_signatures(signatures: List[dict]) -> Dict[str, List[str]]: return { sig.get("name", f"unknown_{i}"): errs for i, sig in enumerate(signatures) if (errs := validate_signature_schema(sig)) } # ══════════════════════════════════════════════════════ # 项目信息渲染(需求-模块-代码关系树) # ══════════════════════════════════════════════════════ def render_project_info(full_info: Dict[str, Any], console) -> None: """ 使用 rich 将项目完整信息(需求-模块-代码关系)渲染到终端。 Args: full_info: DBManager.get_project_full_info() 返回的字典 console: rich.console.Console 实例 """ from rich.table import Table from rich.panel import Panel from rich.tree import Tree from rich.text import Text project = full_info["project"] stats = full_info["stats"] modules = full_info["modules"] raw_reqs = full_info["raw_requirements"] # ── 项目基本信息 ────────────────────────────────── info_table = Table(show_header=False, box=None, padding=(0, 2)) info_table.add_column("key", style="bold cyan", width=14) info_table.add_column("value", style="white") info_table.add_row("项目 ID", str(project.id)) info_table.add_row("项目名称", project.name) info_table.add_row("目标语言", project.language) info_table.add_row("描述", project.description or "(无)") info_table.add_row("输出目录", project.output_dir or "(未生成)") info_table.add_row("创建时间", str(project.created_at)[:19]) info_table.add_row("更新时间", str(project.updated_at)[:19]) console.print(Panel(info_table, title="[bold cyan]📁 项目信息[/bold cyan]", border_style="cyan")) # ── 统计摘要 ────────────────────────────────────── stat_table = Table(show_header=False, box=None, padding=(0, 3)) stat_table.add_column("k", style="bold yellow", width=16) stat_table.add_column("v", style="white") stat_table.add_row("原始需求数", str(stats["raw_req_count"])) stat_table.add_row("功能需求数", str(stats["func_req_count"])) stat_table.add_row("已生成代码", f"{stats['generated_count']} / {stats['func_req_count']}") stat_table.add_row("功能模块数", str(stats["module_count"])) stat_table.add_row("代码文件数", str(stats["code_file_count"])) console.print(Panel(stat_table, title="[bold yellow]📊 统计摘要[/bold yellow]", border_style="yellow")) # ── 原始需求列表 ────────────────────────────────── if raw_reqs: raw_table = Table(title="📝 原始需求", show_lines=True) raw_table.add_column("ID", style="dim", width=5) raw_table.add_column("来源", style="cyan", width=8) raw_table.add_column("文件名", width=20) raw_table.add_column("内容摘要", width=55) raw_table.add_column("创建时间", width=20) for rr in raw_reqs: raw_table.add_row( str(rr.id), rr.source_type, rr.source_name or "-", (rr.content[:80] + "...") if len(rr.content) > 80 else rr.content, str(rr.created_at)[:19], ) console.print(raw_table) # ── 需求-模块-代码 关系树 ───────────────────────── if not modules: console.print("[dim]暂无功能需求[/dim]") return priority_color = {"high": "red", "medium": "yellow", "low": "green"} status_icon = {"generated": "✅", "pending": "⏳", "failed": "❌"} root = Tree( f"[bold cyan]🗂 {project.name}[/bold cyan] " f"[dim]({stats['func_req_count']} 需求 · " f"{stats['module_count']} 模块 · " f"{stats['code_file_count']} 代码文件)[/dim]" ) for module_name in sorted(modules.keys()): mod_data = modules[module_name] req_list = mod_data["requirements"] mod_count = len(req_list) gen_count = sum(1 for r in req_list if r["req"].status == "generated") mod_branch = root.add( f"[magenta bold]📦 {module_name}[/magenta bold] " f"[dim]{gen_count}/{mod_count} 已生成[/dim]" ) for item in req_list: req = item["req"] code_files = item["code_files"] p_color = priority_color.get(req.priority, "white") s_icon = status_icon.get(req.status, "❓") req_label = ( f"{s_icon} [bold]{req.title}[/bold] " f"[{p_color}][{req.priority}][/{p_color}] " f"[dim]REQ.{req.index_no:02d} · ID={req.id}[/dim]" ) req_branch = mod_branch.add(req_label) # 需求详情子节点 req_branch.add( f"[dim]函数名: [/dim][code]{req.function_name}[/code]" ) req_branch.add( f"[dim]描述: [/dim]{req.description[:80]}" + ("..." if len(req.description) > 80 else "") ) # 代码文件子节点 if code_files: code_branch = req_branch.add("[green]📄 代码文件[/green]") for cf in code_files: exists = os.path.exists(cf.file_path) icon = "🟢" if exists else "🔴" cf_text = Text() cf_text.append(f"{icon} ", style="") cf_text.append(cf.file_name, style="bold green" if exists else "bold red") cf_text.append(f" [{cf.language}]", style="dim") cf_text.append(f" {cf.file_path}", style="dim italic") code_branch.add(cf_text) else: req_branch.add("[dim]📄 暂无代码文件[/dim]") console.print(Panel(root, title="[bold green]🔗 需求 · 模块 · 代码 关系树[/bold green]", border_style="green", padding=(1, 2))) def render_project_list(projects: list, stats_map: Dict[int, Dict], console) -> None: """ 渲染所有项目列表(含统计列)。 Args: projects: Project 对象列表 stats_map: {project_id: stats_dict} console: rich.console.Console 实例 """ from rich.table import Table if not projects: console.print("[dim]暂无项目,请先运行 `python main.py run` 创建项目。[/dim]") return table = Table(title=f"📋 项目列表(共 {len(projects)} 个)", show_lines=True) table.add_column("ID", style="cyan bold", width=5) table.add_column("项目名称", style="bold", width=22) table.add_column("语言", style="magenta", width=8) table.add_column("功能需求", style="yellow", width=8) table.add_column("已生成", style="green", width=8) table.add_column("模块数", width=6) table.add_column("代码文件", width=8) table.add_column("描述", width=28) table.add_column("创建时间", width=20) for p in projects: st = stats_map.get(p.id, {}) func_total = st.get("func_req_count", 0) gen_count = st.get("generated_count", 0) gen_cell = ( f"[green]{gen_count}[/green]" if gen_count == func_total and func_total > 0 else f"[yellow]{gen_count}[/yellow]" ) table.add_row( str(p.id), p.name, p.language, str(func_total), gen_cell, str(st.get("module_count", 0)), str(st.get("code_file_count", 0)), (p.description[:28] + "...") if p.description and len(p.description) > 28 else (p.description or "[dim](无)[/dim]"), str(p.created_at)[:19], ) console.print(table)