Aislo/B08_Quantity/B08_Quantity_Router_Material.py

"""B08 자재 집계표 셋 길(PLAN 121-1) — 총괄자재 · 콘크리트소요자재 · 자재수량 집계표.

    GET /api/projects/{id}/quantity/materials
        구조물 집계표(98 `structure_summary` 그대로 부름) + 저장된 사용자 값 → 세 표
    PUT /api/projects/{id}/quantity/materials   {overrides}
        [저장] — 사용자 값을 프로젝트 설정 quantity › material_sheets 에 · 서버가 정본으로 다시 셈

계산 = `B08_Quantity_Material_Calc.py`(화면 짝 `.ts` 와 같은 식) · 총괄 모래 · 자갈 · 시멘트 할증 =
좌측 패널 값(`surcharge_pct` · 저장은 옵션 길 PUT …/quantity/options)
기본값은 파일에 안 씀(읽을 때 채움).
단위수량 기본(143-4) = 연결된 사본 셈(`M02_Item_Copy.calc` = 층 셈 한 벌)의 일위대가 자재 ·
로직 재료(`snapshot_units` · 서버 단독 — 브라우저는 받은 값으로 짝 계산) · 참조 줄(승격 · 사전조합)
안 재료도 셈 · 기계 주연료 · 할증(ℓ)은 뺌(원가 몫 · 160-3).
"""

from __future__ import annotations

import asyncio
import re
from decimal import Decimal
from pathlib import Path
from typing import Any
from uuid import UUID

from fastapi import APIRouter, Body

from B08_Quantity.B08_Quantity_Material_Calc import (
    DEDUCT_KEY,
    PANEL_KEY,
    _num,
    clean_overrides,
    compute,
    panel_surcharge,
    round6,
)
from B08_Quantity.B08_Quantity_Router_Hydro import _project_root
from B08_Quantity.B08_Quantity_Router_Summary import _combo, structure_summary
from common_util.common_util_project_settings import quantity_settings, save_section
from common_util.common_util_summary_item import GROUP_JOIN

router = APIRouter(prefix="/api/projects", tags=["B08 Material"])

SETTINGS_KEY = "material_sheets"
#: 기계 주연료 · 할증 · 잡유(ℓ) — 기계경비(원가) 몫 · 자재 표에서 뺌(160-3).
#   ponytail: 이름 글 맞춤 — 새 연료 이름이 나오면 여기에 더함
FUEL = re.compile(r"연료|경유|휘발유|중유|등유|잡유|gasoline", re.IGNORECASE)


def _fuel(name: str, unit: str) -> bool:
    return unit in ("ℓ", "L") and FUEL.search(name) is not None


def _qty(value: Any) -> float | None:
    return _num(float(value) if isinstance(value, Decimal) else value)


def _ref_mats(whole: Any, read: Any, ref: dict, trail: tuple[str, ...] = ()) -> list[tuple]:
    """참조 문서 1 단위 재료 `[(재료 줄, 수량)]` — 층 셈 펼침(`M02_Labor_Ref.Expand`)과 같은 길:
    승격 = 원 로직 재료 ÷ 배 · 사전조합 = 안 줄 재료 × 손 수량(빈칸 1) · 로직 줄은 알약 배로 나눔 ·
    순환 · 깊이 넘음 · 없는 문서 = 빈 것."""
    from M01_MasterData import M01_MasterData_Store_Combo as combo_store
    from M02_MasterTemplete import M02_Labor_Doc as labor
    from M02_MasterTemplete import M02_Labor_Ref as refs

    rid = refs.ref_id(ref)
    doc = read(ref)
    if rid in trail or len(trail) >= refs.DEPTH or not isinstance(doc, dict):
        return []

    def run(one: dict, times: Any) -> list[tuple]:
        got = labor.run_line(
            whole,
            str(one.get("로직") or ""),
            one.get("입력값"),
            one.get("켜쌓기율"),
            one.get("고름"),
        )
        div = _qty(times) or 1.0
        return [(m, q / div) for m in got.get("재료") or [] if (q := _qty(m.get("수량")))]

    lines = [one for one in doc.get("담은로직") or [] if isinstance(one, dict)]
    if refs.kind_of(ref["키"]) == "promote":
        return run(lines[0], doc.get("배")) if lines else []
    kept = doc.get("알약")
    pills = kept if isinstance(kept, list) else combo_store.combo_pills({"담은로직": lines})["물량"]
    by_logic = {code: pill for pill in pills for code in pill.get("로직") or []}
    out: list[tuple] = []
    for one in lines:
        if isinstance(one.get("참조"), dict):
            sub = _ref_mats(whole, read, one["참조"], (*trail, rid))
        elif one.get("로직"):
            sub = run(one, (by_logic.get(one["로직"]) or {}).get("배"))
        else:
            continue
        hand = _qty((one.get("수량") or {}).get("손"))
        out += [(m, q * (1.0 if hand is None else hand)) for m, q in sub]
    return out


def _folder_units(root: str | Path, folder: str, meta: dict[str, dict]) -> dict[str, float]:
    """사본 하나 = 집계 항목 1 단위의 자재 `{키: 수량}` — 일위대가 자재(키 = MT 코드) + 로직 재료
    (키 = 고른 품목 MT 코드 · 없으면 이름|단위 · 재료 수량 × 줄 수량 ÷ 배) + 참조 줄 안 재료
    (160-3) · 줄 수량 빈칸 · 0 · 파형강관 · 기계 연료는 안 담음 · 셈 못 하면 빈 것."""
    from M02_MasterTemplete import M02_Item_Copy as copies
    from M02_MasterTemplete import M02_Labor_Ref as refs
    from M02_MasterTemplete.M02_Item_Prices import load_prices
    from M02_MasterTemplete.M02_Item_Source import LABOR_FILE, PRICE_FILE, read_json

    try:
        got = copies.calc(root, folder)
        bundle = read_json(copies.folder_of(root, folder) / PRICE_FILE) or {}
        labor = read_json(copies.folder_of(root, folder) / LABOR_FILE) or {}
    except (LookupError, ValueError):
        return {}
    rows = {
        row.get("키"): row
        for data in bundle.values()
        if isinstance(data, dict)
        for row in data.get("줄") or data.get("표") or []
        if isinstance(row, dict)
    }
    out: dict[str, float] = {}

    def put(key: str, name: str, spec: str, unit: str, qty: float) -> None:
        if not qty or name.startswith("파형강관"):  # 0 · 관 몫(관매설 → 파형강관 1 · 관경 열)
            return
        meta.setdefault(key, {"key": key, "name": name, "spec": spec, "unit": unit})
        out[key] = out.get(key, 0.0) + qty

    def logic_mat(mat: dict, qty: float) -> None:
        name, unit = str(mat.get("이름") or ""), str(mat.get("단위") or "")
        if _fuel(name, unit):
            return
        goods = mat.get("품목")
        code = str(goods.get("키") or "") if isinstance(goods, dict) else ""
        if row := rows.get(code) if code else None:  # 고른 품목 = 이름 · 규격 맞춤(160-3)
            put(code, str(row.get("이름") or name), str(row.get("규격") or ""), unit, qty)
        else:
            put(f"{name}|{unit}", name, "", unit, qty)

    frozen = (labor.get("참조사본") if isinstance(labor, dict) else None) or {}
    whole = None  # 사본 단가 — 참조 줄이 있을 때만 읽음

    def read(ref: dict) -> Any:
        return frozen.get(refs.ref_id(ref))

    for line in got.get("줄") or []:
        qty, times = _num(line.get("수량")), _num(line.get("배")) or 1.0
        if qty is None:
            continue
        if isinstance(line.get("참조"), dict):  # 참조 줄 = 배 1
            whole = load_prices(bundle) if whole is None else whole
            for mat, per in _ref_mats(whole, read, line["참조"]):
                logic_mat(mat, per * qty)
            continue
        for mat in line.get("재료") or []:
            if (per := _num(mat.get("수량"))) is not None:
                logic_mat(mat, per * qty / times)
    for mat in got.get("자재") or []:
        code, qty = str(mat.get("코드") or ""), _num(mat.get("수량"))
        if code and qty is not None:
            row = rows.get(code) or {}
            put(
                code,
                str(row.get("이름") or code),
                str(row.get("규격") or ""),
                str(row.get("단위") or ""),
                qty,
            )
    return out


def snapshot_units(root: str | Path, summary: dict[str, Any]) -> dict[str, Any]:
    """연결된 사본 자재 → 집계표 열마다 단위수량 기본(143-4) — 칸마다(측점 · 시설) 그 칸을 낸
    조합의 사본 자재 × 칸 값을 모아 열 계로 나눔(시설마다 사본이 달라도 계가 맞음) · 사본 없으면
    안 담음."""
    folders: dict[tuple[str, str | None], list[str]] = {}
    for owner_id, entry in (summary.get("linked") or {}).items():
        owner = owner_id.partition(GROUP_JOIN)[0]
        folders.setdefault((owner, _combo(entry.get("항목"))), []).append(entry["폴더"])
    meta: dict[str, dict] = {}
    per_folder: dict[str, dict[str, float]] = {}
    sums: dict[str, dict[str, float]] = {}
    for cid, value, owner, combos in summary.get("sources") or []:
        for folder in (f for combo in combos for f in folders.get((owner, combo), [])):
            if folder not in per_folder:
                per_folder[folder] = _folder_units(root, folder, meta)
            cell = sums.setdefault(cid, {})
            for key, qty in per_folder[folder].items():
                cell[key] = cell.get(key, 0.0) + value * qty
    total = summary.get("total") or {}
    units = {
        cid: {key: round6(qty / total[cid]) for key, qty in cells.items()}
        for cid, cells in sums.items()
        if cells and total.get(cid)
    }
    used = {key for cells in units.values() for key in cells}
    return {"materials": [m for key, m in meta.items() if key in used], "units": units}


async def _sheets(project_id: UUID, root: str | Path, overrides: dict[str, Any]) -> dict[str, Any]:
    summary = await structure_summary(project_id)
    source = {
        "columns": summary["columns"],
        "total": summary["total"],
        "snapshot": await asyncio.to_thread(snapshot_units, root, summary),
    }
    panel = panel_surcharge(quantity_settings(root).get(PANEL_KEY))
    sheets = compute(source, overrides, panel)
    notes = list(summary.get("notes") or [])
    if lost := sheets["quantity"]["unmatched"]:
        names = {m["key"]: m for m in sheets["quantity"]["materials"]}
        listed = " · ".join(
            " ".join(filter(None, (names[k]["name"], names[k]["spec"], names[k]["unit"])))
            for k in lost
        )
        notes.append(f"2공구 열 맞춤 없음 {len(lost)}종(총괄 끝 줄 그대로) — {listed}")
    return {
        "source": source,
        "overrides": overrides,
        "surcharge_pct": panel,
        # 구조물 집계표 알림(「집수정 모양 기록 없음 n곳」 등) — 총괄 · 콘크리트 위 빈 까닭(131)
        "notes": notes,
        "sheets": sheets,
    }


async def picked_deduction(project_id: UUID, root: str | Path) -> float | None:
    """채집석 공제 ㎥(160-5) — 저장본 기준 총괄 「채집석 공제」 줄 값 · 토공 잔토처리에서 뺌."""
    overrides = clean_overrides(quantity_settings(root).get(SETTINGS_KEY))
    rows = (await _sheets(project_id, root, overrides))["sheets"]["material"]["rows"]
    return next((row["amount"] for row in rows if row["key"] == DEDUCT_KEY), None)


@router.get("/{project_id}/quantity/materials")
async def material_sheets(project_id: UUID) -> dict[str, Any]:
    root = await _project_root(project_id)
    overrides = clean_overrides(quantity_settings(root).get(SETTINGS_KEY))
    return await _sheets(project_id, root, overrides)


@router.put("/{project_id}/quantity/materials")
async def save_material_sheets(
    project_id: UUID, body: dict[str, Any] = Body(...)
) -> dict[str, Any]:
    root = await _project_root(project_id)
    overrides = clean_overrides(body.get("overrides"))
    save_section(root, "quantity", {SETTINGS_KEY: overrides}, replace_keys=(SETTINGS_KEY,))
    return await _sheets(project_id, root, overrides)