Aislo/B09_Estimation/draft/B09_Draft_Print_Drawings_Source.py

"""인쇄 미리보기 도면(PLAN 302-2) — 목록 · 도면 모으기.

차례 = B07 도면번호(B07 목록 그대로) →
구조물 조합 도면(도번 차례 · 도번마다 한 번 · `도면` 다음 `더한장`).
B07 도면 = B07 단건 조회 그대로(확정본 = 저장 JSON · 미확정 = 서버 조립) + 미확정 횡단 장 구조물 =
B07 화면과 같은 TS 를 Node 로(`B09_Draft_Print_Drawings_Node.ts` · CLAUDE.md 5장 ②).
구조물 도면 = 프로젝트 사본 `도면.json` + 얹는 도각(M02 `structure_frame` · 장 도번).
저장 · 확정 상태는 손대지 않음 — 매번 읽어 만듦.
"""

from __future__ import annotations

import asyncio
import logging
import subprocess
from pathlib import Path
from typing import Any
from uuid import UUID

from B07_DesignDetail.B07_DesignDetail_Engine_Template import (
    entities_bbox,
    template_entities,
    use_project_templates,
    use_title_fields,
)
from B09_Estimation.draft.B09_Draft_Print_Drawings_Pdf import paper_name
from common_util.common_util_node_bundle import ROOT, bundle_stale, node_env, run_bundle_json

logger = logging.getLogger(__name__)

B07_PREFIX = "b07:"
STRUCTURE_PREFIX = "structure:"
_ENTRY = Path(__file__).with_name("B09_Draft_Print_Drawings_Node.ts")
# 번들은 git 밖(tmp) — package.json 을 안 건드리려고 vite 를 바로 부름
BUNDLE = ROOT / "tmp" / "print_drawings_node" / "B09_Draft_Print_Drawings_Node.js"
_BUILD = [
    "node",
    "./config/node_modules/vite/bin/vite.js",
    "build",
    "--configLoader",
    "runner",
    "--ssr",
    "../B09_Estimation/draft/B09_Draft_Print_Drawings_Node.ts",
    "--outDir",
    "../tmp/print_drawings_node",
]


async def project_root(project_id: UUID) -> Path:
    from B03_FileInput.B03_FileInput_Repository import get_project_storage_relative_path
    from common_util.common_util_storage import resolve_stored_project_path
    from config.config_db import get_db_pool

    async with get_db_pool().acquire() as connection:
        stored = await get_project_storage_relative_path(connection, project_id)
    return Path(resolve_stored_project_path(stored)).resolve()


async def title_fields(project_id: UUID) -> dict[str, str]:
    from B07_DesignDetail.B07_DesignDetail_Router import title_block_fields

    return await title_block_fields(project_id)


async def b07_list(project_id: UUID) -> tuple[int | None, list[dict[str, Any]]]:
    """(노선 id, B07 목록) — 노선 미확정 · 자료 없음이면 (None, [])."""
    from B07_DesignDetail.B07_DesignDetail_Router import get_design_drawing_list

    response = await get_design_drawing_list(project_id)
    if not hasattr(response, "drawings"):  # JSONResponse = 404 · 409 · 500
        return None, []
    return response.route_id, [item.model_dump() for item in response.drawings]


async def b07_drawing(project_id: UUID, drawing_id: str) -> dict[str, Any]:
    from B07_DesignDetail.B07_DesignDetail_Router import get_design_drawing

    response = await get_design_drawing(project_id, drawing_id)
    if not hasattr(response, "drawing"):
        raise ValueError(f"B07 도면 「{drawing_id}」 을 읽지 못했습니다.")
    return response.model_dump()


async def section_detail(project_id: UUID, route_id: int) -> dict[str, Any] | None:
    from B06_Section.B06_Section_Router import get_section_detail

    response = await get_section_detail(project_id, route_id)
    dump = getattr(response, "model_dump", None)
    return dump(mode="json") if dump else None


def _bundle_ready() -> bool:
    stale = bundle_stale(BUNDLE) or _ENTRY.stat().st_mtime > BUNDLE.stat().st_mtime
    if not stale:
        return True
    result = subprocess.run(  # noqa: S603 — 고정 명령
        _BUILD, cwd=str(ROOT), env=node_env(), capture_output=True, text=True, timeout=300
    )
    if result.returncode != 0:
        logger.error("인쇄 도면 Node 번들 빌드 실패:\n%s", result.stderr)
    return result.returncode == 0


def place_structures(detail: dict[str, Any], drawings: dict[str, dict[str, Any]]) -> bool:
    """미확정 횡단 장에 구조물을 얹음(제자리) — 실패면 거짓(서버 그림만 나감)."""
    if not drawings or not _bundle_ready():
        return False
    output = run_bundle_json(BUNDLE, "", {"detail": detail, "drawings": drawings})
    placed = (output or {}).get("drawings") if isinstance(output, dict) else None
    if not isinstance(placed, dict):
        return False
    for key, drawing in placed.items():
        if key in drawings and isinstance(drawing, dict):
            drawings[key].clear()
            drawings[key].update(drawing)
    return True


def sheet_number(number: str, index: int) -> str:
    """장 도번 — 1장 = 도번 · n 째 장 = 도번-n(M02 `sheetNumber` 와 같음)."""
    return "" if not number else number if index == 0 else f"{number}-{index + 1}"


def structure_sheets(root: Path) -> list[dict[str, Any]]:
    """구조물 조합 도면 장 — 도번 차례 · 같은 도번은 한 번 · {id, label, number, drawing}."""
    from M02_MasterTemplete.M02_Item_Snapshot import linked_snapshots, structures_dir
    from M02_MasterTemplete.M02_Item_Source import read_json

    docs: dict[str, tuple[str, dict[str, Any]]] = {}
    for owner, copy in linked_snapshots(root).items():
        if "도면" not in copy["부분"]:
            continue
        doc = read_json(structures_dir(root) / copy["폴더"] / "도면.json")
        if not isinstance(doc, dict) or not isinstance(doc.get("도면"), dict):
            continue
        number = str(doc.get("도번") or "")
        name = str((copy.get("항목") or {}).get("name") or copy["폴더"])
        docs.setdefault(number or f"~{owner}", (name, doc))
    sheets = []
    for key in sorted(docs):
        name, doc = docs[key]
        number = "" if key.startswith("~") else key
        pages = [{"이름": "", "도면": doc["도면"]}, *(doc.get("더한장") or [])]
        for index, page in enumerate(pages):
            if not isinstance(page, dict) or not isinstance(page.get("도면"), dict):
                continue
            label = f"{name} · {page['이름']}" if page.get("이름") else name
            numbered = sheet_number(number, index) or f"{key.lstrip('~')}-{index + 1}"
            sheets.append(
                {
                    "id": f"{STRUCTURE_PREFIX}{numbered}",
                    "label": label,
                    "number": sheet_number(number, index),
                    "drawing": page["도면"],
                }
            )
    return sheets


def with_frame(drawing: dict[str, Any], number: str) -> dict[str, Any]:
    """구조물 도면 + 얹는 도각(M02 화면과 같은 `structure_frame` · 표제란 = 프로젝트 값)."""
    from M02_MasterTemplete.M02_MasterTemplete_Router_Drawing import structure_frame

    frame = structure_frame(number)
    layers = [*(drawing.get("layers") or []), *([frame["layer"]] if frame["layer"] else [])]
    return {
        **drawing,
        "entities": [*frame["entities"], *drawing.get("entities", [])],
        "layers": layers,
    }


async def drawing_items(project_id: UUID) -> tuple[list[dict[str, Any]], dict[str, Any]]:
    """(목록, 만들 때 쓸 문맥).

    목록 칸 = {id, source, kind, label, number, confirmed, paper, images_skipped}.
    """
    root = await project_root(project_id)
    use_project_templates(root)
    template = entities_bbox(template_entities())
    paper = paper_name(template[2] - template[0], template[3] - template[1]) if template else None
    route_id, b07 = await b07_list(project_id)
    items = [
        {
            "id": f"{B07_PREFIX}{item['id']}",
            "source": "b07",
            "kind": item["kind"],
            "label": item["label"],
            "number": str(number),
            "confirmed": bool(item.get("confirmed")),
            "paper": paper,
            "images_skipped": None,  # PDF 만들 때 셈 — 응답 머리 X-Aislo-Skipped-Images
        }
        for number, item in enumerate(b07, start=1)
    ]
    sheets = await asyncio.to_thread(structure_sheets, root)
    items += [
        {
            "id": sheet["id"],
            "source": "structure",
            "kind": "structure",
            "label": sheet["label"],
            "number": sheet["number"],
            "confirmed": True,
            "paper": paper,
            "images_skipped": None,
        }
        for sheet in sheets
    ]
    return items, {"root": root, "route_id": route_id, "sheets": {s["id"]: s for s in sheets}}


async def drawings_in_order(
    project_id: UUID, items: list[dict[str, Any]], context: dict[str, Any]
) -> list[dict[str, Any]]:
    """고른 목록 차례대로 CAD 도면 — 미확정 횡단 장은 구조물을 얹어서."""
    base_fields = await title_fields(project_id)
    drawings: list[dict[str, Any]] = []
    unconfirmed_cross: dict[str, dict[str, Any]] = {}
    for item in items:
        if item["source"] == "b07":
            response = await b07_drawing(project_id, item["id"][len(B07_PREFIX) :])
            drawing = response["drawing"]
            if item["kind"] == "cross" and not response.get("confirmed"):
                unconfirmed_cross[item["id"]] = drawing
        else:
            sheet = context["sheets"][item["id"]]
            use_project_templates(context["root"])
            use_title_fields(base_fields)
            drawing = with_frame(sheet["drawing"], sheet["number"])
        drawings.append(drawing)
    if unconfirmed_cross and context.get("route_id") is not None:
        detail = await section_detail(project_id, int(context["route_id"]))
        if detail is None or not await asyncio.to_thread(
            place_structures, detail, unconfirmed_cross
        ):
            logger.warning(
                "인쇄 도면 — 미확정 횡단 장 구조물을 못 얹음(서버 그림만): %s", project_id
            )
    return drawings