AIDeveloper-PC/requirements_generator/core_utils.py

# core_utils.py - 跨 handler 共用的核心工具函数

import json
import os
from datetime import datetime
from typing import List, Optional

from rich.console import Console

import config
from constants import CHG_DELETE
from database.db_manager import DBManager
from database.models import ChangeHistory, FunctionalRequirement, Project
from core.code_generator import CodeGenerator
from core.requirement_analyzer import RequirementAnalyzer
from utils.output_writer import (
    patch_signatures_with_url,
    write_function_signatures_json,
    write_project_readme,
)

console = Console()
db = DBManager()


# ══════════════════════════════════════════════════════
#  变更历史
# ══════════════════════════════════════════════════════

def log_change(
        project_id: int,
        change_type: str,
        summary: str,
        module: str = "",
        req_id: int = None,
        status: str = "pending",
) -> ChangeHistory:
    """
    记录变更历史。
    将 change_type / module / req_id / summary 序列化为 JSON 写入 changes 字段。
    """
    changes_payload = {
        "change_type": change_type,
        "module": module,
        "req_id": req_id,
        "summary": summary,
        "logged_at": datetime.utcnow().isoformat(),
    }
    changes_text = json.dumps(changes_payload, ensure_ascii=False, indent=2)

    history = ChangeHistory(project_id=project_id,
                            changes=changes_text,
                            status=status)
    # ✅ 只传模型真实存在的字段
    db.create_change_history(history)
    return history


def get_change_histories(project_id: int, status: str = None) -> list[dict]:
    """
    获取变更历史列表，并将 changes 字段反序列化为结构化 dict。

    Returns:
        每条记录为 dict，包含：
          id, project_id, change_time, status,
          change_type, module, req_id, summary, logged_at
    """
    histories = db.list_change_histories(project_id, status=status)
    result = []
    for h in histories:
        try:
            payload = json.loads(h.changes)
        except (json.JSONDecodeError, TypeError):
            # 兼容旧格式纯文本
            payload = {"summary": h.changes, "change_type": "unknown",
                       "module": "", "req_id": None, "logged_at": ""}
        result.append({
            "id": h.id,
            "project_id": h.project_id,
            "change_time": h.change_time,
            "status": h.status,
            **payload,  # change_type / module / req_id / summary / logged_at
        })
    return result


# ══════════════════════════════════════════════════════
#  需求过滤
# ══════════════════════════════════════════════════════

def active_reqs(
        reqs: List[FunctionalRequirement],
) -> List[FunctionalRequirement]:
    """过滤掉已软删除（status='deleted'）的需求。"""
    return [r for r in reqs if r.status != "deleted"]


# ══════════════════════════════════════════════════════
#  语言扩展名
# ══════════════════════════════════════════════════════

def get_ext(language: str) -> str:
    ext_map = {
        "python": ".py",
        "javascript": ".js",
        "typescript": ".ts",
        "java": ".java",
        "go": ".go",
        "rust": ".rs",
    }
    return ext_map.get((language or "python").lower(), ".txt")


# ══════════════════════════════════════════════════════
#  LLM 批量生成：签名
# ══════════════════════════════════════════════════════

def generate_signatures(
        analyzer: RequirementAnalyzer,
        func_reqs: List[FunctionalRequirement],
        knowledge_text: str = "",
) -> List[dict]:
    """
    调用 analyzer 批量生成函数签名，
    逐条打印进度（成功 / 降级）。
    """

    def on_progress(i, t, req, sig, err):
        status = "[red]✗ 降级[/red]" if err else "[green]✓[/green]"
        console.print(
            f"  {status} [{i}/{t}] REQ.{req.index_no:02d} "
            f"[cyan]{req.function_name}[/cyan]"
            + (f" | [red]{err}[/red]" if err else "")
        )

    return analyzer.build_function_signatures_batch(
        func_reqs=func_reqs,
        knowledge=knowledge_text,
        on_progress=on_progress,
    )


# ══════════════════════════════════════════════════════
#  LLM 批量生成：代码
# ══════════════════════════════════════════════════════

def generate_code(
        generator: CodeGenerator,
        project: Project,
        func_reqs: List[FunctionalRequirement],
        output_dir: str,
        knowledge_text: str = "",
        signatures: List[dict] = None,
) -> list:
    """
    调用 generator 批量生成代码文件，
    逐条打印进度（成功 / 失败）。
    """

    def on_progress(i, t, req, code_file, err):
        if err:
            console.print(
                f"  [red]✗[/red] [{i}/{t}] {req.function_name}"
                f" | [red]{err}[/red]"
            )
        else:
            console.print(
                f"  [green]✓[/green] [{i}/{t}] {req.function_name}"
                f" → [dim]{code_file.file_path}[/dim]"
            )

    return generator.generate_batch(
        func_reqs=func_reqs,
        output_dir=output_dir,
        language=project.language,
        knowledge=knowledge_text,
        signatures=signatures,
        on_progress=on_progress,
    )


# ══════════════════════════════════════════════════════
#  README 写入
# ══════════════════════════════════════════════════════

def write_readme(
        project: Project,
        func_reqs: List[FunctionalRequirement],
        output_dir: str,
) -> str:
    """生成项目 README.md 并写入输出目录，返回文件路径。"""
    req_lines = "\n".join(
        f"{i + 1}. **{r.title}** (`{r.function_name}`) — {r.description}"
        for i, r in enumerate(func_reqs)
    )
    modules = list({r.module for r in func_reqs if r.module})
    path = write_project_readme(
        output_dir=output_dir,
        project_name=project.name,
        project_description=project.description or "",
        requirements_summary=req_lines,
        modules=modules,
    )
    console.print(f"[green]✓ README 已生成: {path}[/green]")
    return path


# ══════════════════════════════════════════════════════
#  签名 URL 回填 & 保存
# ══════════════════════════════════════════════════════

def patch_and_save_signatures(
        project: Project,
        signatures: List[dict],
        code_files: list,
        output_dir: str,
        file_name: str = "function_signatures.json",
) -> str:
    """将代码文件路径回填到签名 URL 字段，并重新写入 JSON 文件。"""
    ext = get_ext(project.language)
    func_name_to_url = {
        cf.file_name.replace(ext, ""): cf.file_path
        for cf in code_files
    }
    patch_signatures_with_url(signatures, func_name_to_url)
    path = write_function_signatures_json(
        output_dir=output_dir,
        signatures=signatures,
        project_name=project.name,
        project_description=project.description or "",
        file_name=file_name,
    )
    console.print(f"[green]✓ 签名 URL 已回填: {path}[/green]")
    return path


# ══════════════════════════════════════════════════════
#  主签名 JSON 合并
# ══════════════════════════════════════════════════════

def merge_signatures_to_main(
        project: Project,
        new_sigs: List[dict],
        output_dir: str,
        main_file: str = "function_signatures.json",
) -> None:
    """
    将 new_sigs 合并（upsert by name）到主签名 JSON 文件中。
    若主文件不存在则新建。
    """
    main_path = os.path.join(output_dir, main_file)
    if os.path.exists(main_path):
        with open(main_path, "r", encoding="utf-8") as f:
            main_doc = json.load(f)
        existing_sigs = main_doc.get("functions", [])
    else:
        existing_sigs = []
        main_doc = {
            "project": project.name,
            "description": project.description or "",
            "functions": [],
        }

    sig_map = {s["name"]: s for s in existing_sigs}
    for sig in new_sigs:
        sig_map[sig["name"]] = sig
    main_doc["functions"] = list(sig_map.values())

    with open(main_path, "w", encoding="utf-8") as f:
        json.dump(main_doc, f, ensure_ascii=False, indent=2)
    console.print(
        f"[green]✓ 主签名 JSON 已更新: {main_path}"
        f"（共 {len(main_doc['functions'])} 条）[/green]"
    )