AIDeveloper-PC/requirements_generator/utils/output_writer.py

# utils/output_writer.py - 代码文件 & JSON 输出工具
import os
import json
from pathlib import Path
from typing import Dict, List

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:
    """
    构建顶层签名文档结构::

        {
          "project":     "<name>",
          "description": "<description>",
          "functions":   [ ... ]
        }
    """
    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]:
    """
    将代码文件路径回写到签名的 "url" 字段（紧跟 "type" 之后）。

    Args:
        signatures:       签名列表（in-place 修改）
        func_name_to_url: {函数名: 文件绝对路径}

    Returns:
        修改后的签名列表
    """
    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:
    """在有序 dict 中将 new_key 插入到 after_key 之后"""
    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:
    """将签名列表导出为 JSON 文件"""
    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))
    }