Aislo/M02_MasterTemplete/M02_Item_Source.py

"""집계 항목 원본 — 층(시스템 · 회사 · 개인)에서 항목 내용을 모음 (PLAN 21-1 · 25-1b · 25-2b).

모으는 것 = `{파일 이름: 내용}`:
    `집계.json` SS 줄 + 열 정의 · `도면.json` SD · `면적.json` SA · `재료.json` SM ·
    `자재조합.json` UM 줄 · `일위대가.json` UA 줄 + 담은 로직 줄(M01 시스템) ·
    `입력연결.json` 층의 `inputs/<열id>.json`(없으면 빈 연결) ·
    `단가.json` 계산에 닿는 마스터 줄 전부
`원본판` = 모은 내용의 해시 — 지금 층 · 마스터에서 다시 모은 판과 대어 「마스터가 바뀜」 을 앎.
프로젝트가 고치는 값(`입력값.json` · `고른자재.json`)은 원본이 아니라 여기서 안 모음.
원본 층 자리 = `M02_Template_Layers` 경로 · 일위대가 · 단가는 M01 시스템 것.
"""

from __future__ import annotations

import hashlib
import json
from pathlib import Path
from typing import Any

from M02_MasterTemplete import M02_Template_Layers as layers

TABLE_NAME = "구조물집계표"
UM_FILE = "자재조합.json"
#: 열 id → 스냅샷 파일(구조물 도면 · 면적산출 · 재료산출)
PART_FILES = (("structure", "도면.json"), ("area", "면적.json"), ("material", "재료.json"))
LINK_FILE = "입력연결.json"
PRICE_FILE = "단가.json"
LABOR_FILE = "일위대가.json"


def layer_root(item: dict[str, Any]) -> Path:
    """항목 원본 층 뿌리 — 시스템 · 회사 · 개인."""
    if item["layer"] == "system":
        return layers.system_dir()
    if item["layer"] == "company":
        return layers.company_dir(item["owner"])
    company, user = str(item["owner"]).split("/")
    return layers.personal_dir(company, user)


def read_json(path: Path) -> Any:
    try:
        return json.loads(path.read_text(encoding="utf-8"))
    except (OSError, ValueError):
        return None


def row_of(document: Any, key: str) -> dict[str, Any] | None:
    rows = document.get("줄") if isinstance(document, dict) else None
    for row in rows or []:
        # 옛 키로도 — 회사 · 개인 층 SS 키는 새 모양(SC · SP)으로 옮기며 `옛키` 에 남김(PLAN 26-4a)
        if isinstance(row, dict) and (row.get("키") == key or key in (row.get("옛키") or [])):
            return row
    return None


def row_path(row: dict[str, Any]) -> list[str]:
    """SS 줄의 경로(26-1) — `경로` 칸이 있으면 그것 · 옛 줄은 종류 · 분류1~3(빈 분류는 건너뜀)."""
    if isinstance(row.get("경로"), list):
        return [str(p) for p in row["경로"] if str(p or "").strip()]
    return [str(row[k]) for k in ("종류", "분류1", "분류2", "분류3") if row.get(k)]


def item_path(item: dict[str, Any]) -> list[str]:
    """층 SS 줄의 경로 — 없으면 LookupError."""
    row = row_of(read_json(layer_root(item) / layers.COMBO_FILE), str(item.get("key")))
    if row is None:
        raise LookupError(f"집계 항목 {item.get('key')} 이 {item.get('layer')} 층에 없음")
    return row_path(row) or [str(row["키"])]


def item_name(item: dict[str, Any]) -> str:
    """층 SS 줄의 보이는 이름(경로를 이은 글) — 없으면 LookupError."""
    return " ".join(item_path(item))


def version_of(content: Any) -> str:
    raw = json.dumps(content, ensure_ascii=False, sort_keys=True).encode("utf-8")
    return hashlib.sha256(raw).hexdigest()[:16]


def labor_rows(ua_key: str) -> dict[str, Any] | None:
    """조합 줄(마스터 UA · 회사 UC · 개인 UP) + 담은 로직 줄(M01 시스템 · 없는 로직은 빼고 키만
    `빠진로직`)."""
    from M01_MasterData import M01_MasterData_Store as m01
    from M02_MasterTemplete import M02_MasterTemplete_Combo_Store as combos

    ua = combos.find(ua_key)
    if ua is None:
        return None
    wanted = [str(entry.get("로직")) for entry in ua.get("담은로직") or [] if entry.get("로직")]
    found: dict[str, dict[str, Any]] = {}
    raw: dict[str, Any] = {}  # M01 적재본은 수가 Decimal — 줄은 원본 파일 JSON 그대로 담음
    for key in wanted:
        try:
            file = m01.logic(key)["file"]
        except m01.StoreError:
            continue
        if file not in raw:
            raw[file] = read_json(Path(m01.FOLDER) / file)
        row = row_of(raw[file], key)
        if row is not None:
            found[key] = row
    return {
        "줄": ua,
        "로직": [found[key] for key in wanted if key in found],
        "빠진로직": [key for key in wanted if key not in found],
    }


def input_links(root: Path, column_id: str) -> dict[str, Any]:
    """층의 항목 연결 `inputs/<열id>.json`(sub4 `_Inputs.read_file`) · 없으면 빈 연결."""
    from M02_MasterTemplete import M02_MasterTemplete_Inputs as inputs
    from M02_MasterTemplete import M02_MasterTemplete_Store as store

    if not column_id:
        return {"판": None, "줄": []}
    with store.at(root):
        found = inputs.read_file(column_id)
    if isinstance(found, dict) and isinstance(found.get("줄"), list):
        return found
    return {"판": None, "줄": []}


def build_source(item: dict[str, Any]) -> dict[str, Any]:
    """원본 층 · 마스터에서 항목 내용을 모음 — SS 줄이 없으면 LookupError."""
    from M02_MasterTemplete import M02_Item_Prices as prices

    root = layer_root(item)
    row = row_of(read_json(root / layers.COMBO_FILE), item["key"])
    if row is None:
        raise LookupError(f"집계 항목 {item['key']} 이 {item['layer']} 층에 없음")
    column_id = str(row.get("열") or "")
    table = read_json(root / "table" / f"{TABLE_NAME}.json") or {}
    column = next((c for c in table.get("열") or [] if c.get("id") == column_id), None)
    files: dict[str, Any] = {"집계.json": {"줄": row, "열": column}}
    for kind, name in PART_FILES:
        part = read_json(root / kind / f"{column_id}.json") if column_id else None
        if part is not None:
            files[name] = part
    if row.get("자재조합"):
        um = row_of(read_json(root / UM_FILE), str(row["자재조합"]))
        if um is not None:
            files[UM_FILE] = um
    labor = labor_rows(str(row["일위대가"])) if row.get("일위대가") else None
    if labor is not None:
        files[LABOR_FILE] = labor
    files[LINK_FILE] = input_links(root, column_id)
    logic_keys = [str(one.get("키")) for one in (labor or {}).get("로직", [])]
    files[PRICE_FILE] = prices.to_json(prices.collect(logic_keys)) if logic_keys else {}
    return files