Aislo/B08_Quantity/B08_Quantity_Summary_Sheet.py

"""B08 구조물 집계표 모음 상자 — 열 · 측점 줄 · 계 · 사본 주인별 값 · 칸 출처.

`B08_Quantity_Router_Summary.py` 가 700줄을 넘어 뗌(10-9 통합) — 셈 규칙은 그쪽 그대로.
"""

from __future__ import annotations

from typing import Any

from common_util.common_util_summary_item import GROUP_JOIN


class _Copy(str):
    """잇기 조합 id + 그 사본 주인 id(`linked_snapshots` 키 · `{구조물 id}__{그룹}`) — 145-3."""

    copy: str


def _linked(combo: str | None, copy: str) -> str | None:
    """잇기 조합에 사본 주인 id 를 붙임 — 본체 조합(그냥 글)은 시설 id 가 주인."""
    if combo is None:
        return None
    out = _Copy(combo)
    out.copy = copy
    return out


class _Sheet:
    """열 · 측점 줄 · 계를 모음 — 같은 측점(1 cm)의 같은 열은 더함."""

    def __init__(self) -> None:
        self.columns: dict[str, dict[str, Any]] = {}
        self.rows: dict[int, dict[str, Any]] = {}
        self.owner = ""  # 지금 셈하는 시설 — 사본 없어 안 센 칸의 시설을 모음(143-1)
        self.kind = ""  # 그 시설 종류 — 같은 측점 박스 + 관은 주인 id 가 같음(154-1)
        self.unlinked: set[tuple[str, str]] = set()
        self.sources: list[list[Any]] = []  # 칸마다 [열, 값, 시설, 조합] — 자재 재연결(143-4)

    def add(
        self,
        chainage: float,
        key: str,
        sort: tuple[Any, ...],
        group: list[str],
        label: str,
        value: float | None,
        combos: list[str | None],
    ) -> None:
        if not any(combos):  # 그 칸을 낸 조합 사본이 없음 — 안 셈(143-1)
            self.unlinked.add((self.kind, self.owner))
            return
        cid = f"pipe:{key}"
        column = self.columns.setdefault(
            cid,
            {"id": cid, "group": group, "label": label, "combos": [], "owners": {}, "_sort": sort},
        )
        for combo in combos:
            if combo and combo not in column["combos"]:
                column["combos"].append(str(combo))
        row = self.rows.setdefault(round(chainage * 100), {"chainage_m": chainage, "values": {}})
        if value is not None:  # 열은 세움 — 놓였는데 값이 빈칸인 것이 보여야 함
            row["values"][cid] = round((row["values"].get(cid) or 0.0) + value, 3)
            # 사본 주인별 값(145-3 · 설계내역서가 대가 호표에 이음) — 잇기 사본이 먼저 · 없으면 본체
            owner = next((c.copy for c in combos if isinstance(c, _Copy)), self.owner)
            owners = column["owners"]
            owners[owner] = round(owners.get(owner, 0.0) + value, 3)
            self.sources.append([cid, value, self.owner, [c for c in combos if c]])

    def result(self) -> dict[str, Any]:
        columns = sorted(self.columns.values(), key=lambda column: column["_sort"])
        rows = [self.rows[key] for key in sorted(self.rows)]
        total: dict[str, float] = {}
        for row in rows:
            for cid, value in row["values"].items():
                total[cid] = round(total.get(cid, 0.0) + value, 3)
        return {
            "columns": [{k: v for k, v in c.items() if k != "_sort"} for c in columns],
            "rows": rows,
            "total": total,
            "sources": self.sources,
        }


#: 옵션 그룹 정의 밖 필수 컨테이너 잇기(`design_structure_ford.ts` FORD_SPEC `link` · 제목)
FIXED_SLOTS = {
    "ford_bridge": (("gford", "세월교"), ("gfordin", "유입 구조"), ("gfordout", "유출 구조"))
}


def unlinked_facilities(
    items: list[dict[str, Any]],
    linked: dict[str, Any],
    missing: set[tuple[str, str]],
    type_names: dict[str, str],
    stored: dict[tuple[str, str], dict[str, Any]],
) -> list[dict[str, Any]]:
    """미연결 시설(셈한 칸 기준 `missing` · 154-1) — 시설마다 칸 전부 + 연결 여부(154-2 바로 연결).

    `{id, type_id, name, facility(관 지점만), chainage_m, slots[{copy, group, item, linked}]}` ·
    본체 칸 `group` = None · `stored` = 시설마다 저장된 `group_items`(필수 컨테이너 잇기 칸)."""
    out: dict[tuple[str, str], dict[str, Any]] = {}
    for item in items:
        owner = item["복사본"].partition(GROUP_JOIN)[0]
        key = (item["type_id"], owner)
        if key not in missing:
            continue
        entry = out.setdefault(
            key,
            {
                "id": owner,
                "type_id": item["type_id"],
                "name": type_names.get(item["type_id"], item["type_id"]),
                "facility": item["type_id"] if owner.startswith("pipe@") else None,
                "chainage_m": float(item["chainage_m"]),
                "slots": [],
            },
        )
        entry["slots"].append(
            {
                "copy": item["복사본"],
                "group": item.get("그룹") if item["복사본"] != owner else None,
                "item": item.get("summary_item"),
                "linked": item["복사본"] in linked,
            }
        )
    for (type_id, owner), entry in out.items():
        have = {slot["copy"] for slot in entry["slots"]}
        links = stored.get((type_id, owner)) or {}
        for gid, title in FIXED_SLOTS.get(type_id, ()):
            copy = f"{owner}{GROUP_JOIN}{gid}"
            if copy not in have:
                entry["slots"].append(
                    {"copy": copy, "group": title, "item": links.get(gid), "linked": copy in linked}
                )
    return sorted(out.values(), key=lambda entry: (entry["chainage_m"], entry["id"]))