Aislo/B09_Estimation/B09_Estimation_Router.py

"""B09 원가계산 길(PLAN 144 · 145) — 결과는 저장하지 않고 GET 때 서버가 셈(B08 꼴).

    GET  /api/projects/{id}/estimation
        입력(굳힌 단가 · 수요 · B08 수량) + 사용자 값 → 표 전부
    PUT  /api/projects/{id}/estimation   {user}
        [저장] — 사용자 값 · 책에 없는 새 단가 줄만 굳힘 → 서버가 정본으로 다시 셈
    POST /api/projects/{id}/estimation/prices
        [단가 갱신] — 수요 단가 전부를 지금 M01 값으로 새로 굳힘
    POST /api/projects/{id}/estimation/logic-preview   {code, given, rate, picks}
        「운반 단가」 모달 계산 절차(253-3) — 그 로직 한 산근을 프로젝트 단가로 풀어 글 줄 · 단가 ·
        저장 안 함(`render_sangeun_text` 를 `Sangeun.Solver` 로)
    POST /api/projects/{id}/estimation/library   {kind, layer, owner, key}
        설계내역서 연결 고르기(204-2) — 서고 구조물조합 · 사전조합을 프로젝트 사본으로
        (있으면 그대로) → {폴더, 만듦, key = 대가 블록 id, payload} ·
        층 읽기 권한 = `M02_Layer_Access` · 줄 연결은 화면 캐시 → [저장] 때 안 쓰는 사본 지움(`Bill_Library.prune`)

저장 자리 = 프로젝트 안 `B09_Estimation/v1/`(`B09_Estimation_Store`) ·
단가 = `B09_Estimation_Prices` ·
수요 = `B09_Estimation_Linked`(연결된 사본) · 셈 = `B09_Estimation_Calc_*`(화면 짝과 같은 식) ·
산근 = `B09_Estimation_Sangeun`(서버 단독 · 145-1 · 화면은 받은 그대로).
"""

from __future__ import annotations

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

from fastapi import APIRouter, Body, Depends, HTTPException
from pydantic import BaseModel, Field

from B08_Quantity.B08_Quantity_Haul import haul_rows
from B08_Quantity.B08_Quantity_Milestone_Calc import compute_milestone
from B08_Quantity.B08_Quantity_Router_Earthwork import earthwork_table
from B08_Quantity.B08_Quantity_Router_Hydro import _project_root
from B08_Quantity.B08_Quantity_Router_Material import material_sheets
from B08_Quantity.B08_Quantity_Router_Milestone import read_data as read_milestone
from B08_Quantity.B08_Quantity_Router_Milestone import site_info
from B09_Estimation import B09_Estimation_Bill_Library as library
from B09_Estimation import B09_Estimation_Prices as prices
from B09_Estimation import B09_Estimation_Sangeun as sangeun
from B09_Estimation import B09_Estimation_Store as store
from B09_Estimation.B09_Estimation_Bill_Items import bill_items, hopyo_prices
from B09_Estimation.B09_Estimation_Calc_Base import compute_base
from B09_Estimation.B09_Estimation_Calc_Bill import compute_estimation
from B09_Estimation.B09_Estimation_Calc_Labor import compute_labor, daega_price
from B09_Estimation.B09_Estimation_Calc_Total import compute_total
from B09_Estimation.B09_Estimation_Labor_Source import daega_inputs
from B09_Estimation.B09_Estimation_Linked import linked_resource_keys
from B09_Estimation.B09_Estimation_Rates import rate_candidates, rate_table
from B09_Estimation.B09_Estimation_Router_Quotes import router as quotes_router
from B09_Estimation.B09_Estimation_Summary import compute_summary, hopyo_lines, used_materials
from common_util.common_util_auth import verify_session
from M02_MasterTemplete import M02_Layer_Access as access
from M02_MasterTemplete.M02_Item_Snapshot import importer

try:  # 252 공기 초안(draft 폴더 · 지우면 공기 탭 원천 없이)
    from B09_Estimation.draft import B09_Draft_Schedule_Source as schedule_draft
except ImportError:  # pragma: no cover
    schedule_draft = None

router = APIRouter(prefix="/api/projects", tags=["B09 Estimation"])


async def _b08(project_id: UUID, notes: list[str]) -> dict[str, Any]:
    """B08 수량 원천 — 못 읽으면 빈칸 + 알림(B09 는 B08 를 고치지 않음)."""
    out: dict[str, Any] = {}
    for key, name, read in (
        ("materials", "자재 집계표", material_sheets),
        ("earthwork", "토공집계표", earthwork_table),
    ):
        try:
            out[key] = await read(project_id)
        except Exception as error:  # noqa: BLE001 — 원천 하나가 막혀도 나머지 표는 그림
            notes.append(f"B08 {name} 못 읽음 — {getattr(error, 'detail', error)}")
            out[key] = None
    # 구조물 집계표 = 자재 집계표가 이미 부른 값(열 · 계 · 조합) — 두 번 안 부름
    out["summary"] = (out["materials"] or {}).get("source")
    # 총괄 · 콘크리트 표 = B08 답의 `sheets` 한 층 아래(재료비목록표 · 운반 수량이 읽음 · 167-2)
    out["materials"] = (out["materials"] or {}).get("sheets")
    return out


def compute_sheets(root: Path, inputs: dict[str, Any], user: dict[str, Any]) -> dict[str, Any]:
    """짝 셈 한 벌 — 바탕(144) → 산근(145-1 · 서버 단독) → 대가(145-2) → 내역(145-3) → 총괄(145-4).

    화면 짝 = `B09_Estimation_UI_Shell.computeSheets`(산근만 서버 답 그대로)."""
    base = compute_base(inputs, user)
    sg = sangeun.build(root, base, user)  # 145-1 — 굳힌 단가로 풀이(+ 166-2 손 줄 로직)
    by_key = {row["key"]: row for row in sg["sg_list"]}
    labor = compute_labor(inputs.get("daega"), {**base, "gaps": []}, by_key)
    books = {"daega": labor["daega"], "sg_list": sg["sg_list"], "sg_hand": sg["sg_hand"]}
    sheets = compute_estimation(inputs, user, lambda owner: daega_price(labor, owner), books)
    sheets.update(sg)
    sheets.update({k: v for k, v in labor.items() if k != "gaps"})
    sheets["gaps"] = [*sheets["gaps"], *sg["sg_gaps"], *labor["gaps"]]
    # 145-4 총괄 · 원가계산서 = 설계내역서 비목 합(145-3 `bill.sums`) + 요율(사용자 ?? 자동 166-1) · 손입력
    sheets["total"] = compute_total(
        (sheets.get("bill") or {}).get("sums"), user.get("estimate"), rate_table()
    )
    return sheets


def compute_twice(root: Path, inputs: dict[str, Any], user: dict[str, Any]) -> tuple[dict, dict]:
    """183-1 두 번 셈 — 1차 셈으로 쓰는 자재만 남기고(`used_materials`) 재료비목록표 수량 =
    1차 역전개 총수량(`material_qty` = 재료비수량금액집계표 수량) → 2차 셈 · 집계표.
    `inputs` 를 고쳐 둠(화면 짝이 거른 입력 · 수량을 그대로 받음)."""
    first = compute_sheets(root, inputs, user)
    summary = compute_summary(first, hopyo_lines(first))
    keep = used_materials(first, inputs, user)
    inputs["materials"] = [m for m in inputs.get("materials") or [] if m["key"] in keep]
    inputs["material_qty"] = {row["key"]: row["qty"] for row in summary["sum_material"]}
    sheets = compute_sheets(root, inputs, user)
    return sheets, compute_summary(sheets, hopyo_lines(sheets))


def haul_routes(root: Path) -> list[dict[str, Any]]:
    """거리이정표 행 → 운반공 줄(253-1 · `haul_rows` · 거리 L1 · L2 포함) — 못 읽으면 빈 목록."""
    try:
        data = read_milestone(root)
        mid = site_info(root, data["site"]["mid_station_m"])["mid_default_m"]
        return haul_rows(data, compute_milestone(data, mid))
    except Exception:  # noqa: BLE001 — 거리이정표 없음 = 운반 연결 없음
        return []


def haul_user(rows: list[dict[str, Any]]) -> dict[str, Any]:
    """셈 덧값 — `haul_refs{줄 키: ref}`(251-5) · `haul_dist{줄 키: {L1, L2}}`(253-2 거리 채움)."""
    return {
        "haul_refs": {r["key"]: r["ref"] for r in rows if r.get("ref")},
        "haul_dist": {r["key"]: {"L1": r["L1"], "L2": r["L2"]} for r in rows},
    }


def calc_user(root: Path) -> dict[str, Any]:
    """셈 · 수요 · 사본 정리용 사용자 값 — 저장 값 + 거리이정표 운반 연결 · 거리(화면엔 안 보냄)."""
    return {**store.read_user(root), **haul_user(haul_routes(root))}


async def _demand(root: Path) -> list[str]:
    """단가 수요 키 — 연결된 사본 + 설계내역서 손 줄 로직(166-2) +
    거리이정표 운반 연결(251-5)."""
    keys = await asyncio.to_thread(linked_resource_keys, root)
    user = await asyncio.to_thread(calc_user, root)
    hand = await asyncio.to_thread(sangeun.hand_resource_keys, user)
    return [*keys, *(k for k in hand if k not in keys)]


async def payload(project_id: UUID, root: Path) -> dict[str, Any]:
    notes: list[str] = []
    b08 = await _b08(project_id, notes)
    keys = await _demand(root)
    if not keys:
        notes.append(
            "연결된 구조물 조합 사본 없음 — 노무 · 경비 · 중기 단가 줄은 사본이 연결되면 생김"
        )
    book = await asyncio.to_thread(store.read_book, root)
    inputs = await asyncio.to_thread(prices.build_inputs, book, keys, b08["materials"])
    hauls = await asyncio.to_thread(haul_routes, root)
    inputs["bill"] = bill_items({**b08, **haul_user(hauls), "haul_routes": hauls})
    inputs["hopyo"] = hopyo_prices(root)
    inputs["daega"] = await asyncio.to_thread(daega_inputs, root)
    user = await asyncio.to_thread(store.read_user, root)
    calc = {**user, **haul_user(hauls)}
    sheets, summary = await asyncio.to_thread(compute_twice, root, inputs, calc)
    if schedule_draft:
        schedule_draft.attach(inputs, sheets, user)
    return {
        "inputs": inputs,
        "user": user,
        "book": {"made": book.get("made")} if book else None,
        # 203-1 [업데이트] 배지 까닭(지금 M01 과 다른 것 한 줄 · 같으면 None)
        "stale": await asyncio.to_thread(prices.book_stale, book, keys),
        "sheets": sheets,
        "rate_candidates": rate_candidates(),
        "rate_table": rate_table(),
        "notes": notes,
        # 집계표 넷 = 서버 단독(145-5) — 짝 셈 결과를 역전개 · 조작 중엔 안 바뀜
        "summary": summary,
    }


@router.get("/{project_id}/estimation")
async def estimation(project_id: UUID) -> dict[str, Any]:
    root = await _project_root(project_id)
    return await payload(project_id, root)


@router.put("/{project_id}/estimation")
async def save_estimation(project_id: UUID, body: dict[str, Any] = Body(...)) -> dict[str, Any]:
    root = await _project_root(project_id)
    user = await asyncio.to_thread(store.write_user, root, body.get("user"))
    calc = {**user, **haul_user(await asyncio.to_thread(haul_routes, root))}
    await asyncio.to_thread(library.prune, root, calc)  # 204-2 줄이 안 잇는 서고 사본
    keys = await _demand(root)
    book = await asyncio.to_thread(store.read_book, root)
    book = await asyncio.to_thread(prices.extend_book, book, keys)
    await asyncio.to_thread(store.write_book, root, book)
    return await payload(project_id, root)


@router.post("/{project_id}/estimation/prices")
async def refresh_prices(project_id: UUID) -> dict[str, Any]:
    root = await _project_root(project_id)
    keys = await _demand(root)
    book = await asyncio.to_thread(prices.refresh_book, keys)
    await asyncio.to_thread(store.write_book, root, book)
    return await payload(project_id, root)


class PreviewBody(BaseModel):
    code: str = Field(min_length=1, max_length=40)
    given: dict[str, Any] = Field(default_factory=dict)
    rate: float | None = None
    picks: dict[str, Any] | None = None


def logic_preview(root: Path, body: PreviewBody) -> dict[str, Any]:
    """로직 한 산근 미리보기 — 단가 = 굳힌 책(없는 줄 = 지금 M01) · 사용자 대비표 고름."""
    src = {"code": body.code, "given": sangeun._clean_given(body.given), "rate": body.rate,
           "picks": body.picks}  # fmt: skip
    hands = [("preview", "미리보기", src)]
    keys = sangeun.hand_resource_keys({"haul_refs": {"preview": {"tab": "logic", **src}}})
    user = store.read_user(root)
    base = compute_base(prices.build_inputs(store.read_book(root), keys), user)
    block = sangeun.Solver(base).add(sangeun.hand_whole(hands), src, None)
    text = [
        *block["head"],
        *(t for ln in block["lines"] for t in ln.get("text") or []),
        block["foot"],
    ]
    return {
        "name": block["name"],
        "lines": [t for t in text if t],
        "unit": block["unit"],
        "price": block["sum"],
        "parts": {p: block[p] for p in sangeun.PARTS},
        "why": block["why"],
    }


@router.post("/{project_id}/estimation/logic-preview")
async def preview_logic(project_id: UUID, body: PreviewBody) -> dict[str, Any]:
    root = await _project_root(project_id)
    return await asyncio.to_thread(logic_preview, root, body)


class LibraryBody(BaseModel):
    kind: str
    layer: str
    owner: str | None = None
    key: str


@router.post("/{project_id}/estimation/library")
async def import_library(
    project_id: UUID, body: LibraryBody, session: dict[str, Any] = Depends(verify_session)
) -> dict[str, Any]:
    got = access.grant(session, body.layer, body.owner)  # 못 읽는 층 = 403 · 회사 없음 = 409
    root = await _project_root(project_id)
    try:
        made = await asyncio.to_thread(
            library.import_copy, root, body.kind, got.layer, got.owner, body.key, importer(session)
        )
    except LookupError as error:
        raise HTTPException(status_code=404, detail=str(error)) from error
    except ValueError as error:
        raise HTTPException(status_code=400, detail=str(error)) from error
    out = await payload(project_id, root)
    key = ((out["inputs"].get("daega") or {}).get("libs") or {}).get(made["폴더"])
    return {**made, "key": key, "payload": out}


# 166-4 견적서 파일 길 — 같은 머리 · 같은 권한
router.include_router(quotes_router)