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/complete
        [>] — 단계 6 완료 · 프로젝트 DONE(요구 300 B1 · 본문 없음)
    POST /api/projects/{id}/estimation/reset
        [초기화](요구 298-2 C1 · C3 · C4 · C5 · 301 · 본문 없음) — 사용자 값 전부(내역 · 자재 ·
        원가 · 공기) · 단가 책 · `estimation` 설정 구획 지움 · 서고 사본 정리 · 견적서 파일 남김 →
        답 = GET 꼴(기준 = 지금 B08 저장본 + 지금 M01 단가) · 단계 = [저장]과 같음
    POST /api/projects/{id}/estimation/prices   {user?}
        [업데이트] — 원본이 바뀐 구조물 사본 재적용(703-3 · `B09_Estimation_Refresh`) →
        수요 단가 전부를 지금 M01 값으로 새로 굳힘 · user = 화면 캐시(296 · 지운 줄 · 그 사본은
        수요에서 뺌 · 저장은 안 함) · M01 원본이 지워진 줄 = 굳힌 값 그대로 · 덮개 그대로 ·
        답 notes 첫 줄 = 바뀐 수
    POST /api/projects/{id}/estimation/prices/preview   {user?}
        [업데이트] 미리 보기(703-3) — 바뀌는 단가 줄 · 다시 가져올 사본 · 지킬 고친 칸 ·
        노임 판(쓰지 않음)
    POST /api/projects/{id}/estimation/calc   {user}
        조작 중 서버 셈(702-3 · 단가연결 즉시 금액) — 화면 캐시로 GET 꼴 셈 · 저장 안 함
    POST /api/projects/{id}/estimation/logic-preview   {code, given, rate, picks}
        「운반 단가」 모달 계산 절차(253-3) — 그 로직 한 산근을 프로젝트 단가로 풀어 글 줄 · 단가 ·
        저장 안 함(`render_sangeun_text` 를 `Sangeun.Solver` 로) · `blocks` = 칸 나눔(277-2)
    POST /api/projects/{id}/estimation/library   {kind, layer, owner, key}
        설계내역서 연결 고르기(204-2) — 서고 구조물조합 · 사전조합을 프로젝트 사본으로
        (있으면 그대로) → {폴더, 만듦, key = 대가 블록 id, payload} ·
        층 읽기 권한 = `M02_Layer_Access` · 줄 연결은 화면 캐시 →
        [저장] 때 안 쓰는 사본 지움(`Bill_Library.prune`) ·
        운반공 줄 = {given, own, dist} 같이 → 거리 값 사본(263-2 · 폴더 `~값 해시`)
    POST /api/projects/{id}/estimation/library-inputs   {kind, layer, owner, key, lib?}
        「운반 단가」 모달(263-2) — 담은 로직 줄 입력 · 값(읽기만 · 권한 = library 길과 같음)
    POST /api/projects/{id}/estimation/daega/personal   {cover, name?, overwrite?, user?}
        「내 자료집에 저장」(703-2) — 펼친 일위대가 → 개인 층 사전조합
        (`B09_Estimation_Daega_Personal`) · user = 화면 캐시 덮개 ·
        같은 이름 = 409 {같은이름} → 덮어쓰기(overwrite = 키) · 새 이름
    GET  /api/projects/{id}/estimation/past
        재료 창고 「지난 공사」(703-1) — 같은 회사 다른 공사의 일위대가 목록(`B09_Estimation_Past`)
    POST /api/projects/{id}/estimation/past   {project_id, cover}
        지난 공사 일위대가 → 이 공사 서고 사본(덮개째) → {폴더, 만듦, key, cover, 덮개, payload} ·
        덮개 = 그 공사 저장 칸(화면 캐시 cover 가 비었을 때만 얹음 · 같은 꼴 블록이 먼저 있으면 빈 것) ·
        같은 회사만(남의 회사 403)
경쟁사 내역 줄 불러오기(707-1) = `B09_Estimation_Router_Import`.
운반 사본 연결 = 셈 때마다 거리로 사본 다시 고름(`Bill_Library.resolve_hauls` · `fix_keys`).

저장 자리 = 프로젝트 안 `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
import copy
import importlib
import logging
import time
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_Daega_Personal as personal
from B09_Estimation import B09_Estimation_Past as past
from B09_Estimation import B09_Estimation_Prices as prices
from B09_Estimation import B09_Estimation_Refresh as refresh
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_Import import router as import_router
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 common_util.common_util_project_settings import reset_section
from common_util.common_util_storage import resolve_stored_project_path
from common_util.common_util_workflow_state import mark_stage
from M02_MasterTemplete import M02_Layer_Access as access
from M02_MasterTemplete.M02_Item_Snapshot import importer
from M02_MasterTemplete.M02_MasterTemplete_Router_Items import _call, _root
from M02_MasterTemplete.M02_MasterTemplete_Router_Layers import _company_projects, _project

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"])


# 702-3 조작 중 셈 = 마지막 GET · [저장] 이 읽은 B08 원천 다시 씀(B08 은 B09 조작으로 안 바뀜)
# ponytail: 시간 창만 봄 — 창 안에 남이 B08 을 고치면 미리보기만 옛 수량 · [저장] 은 늘 새로 읽음
B08_REUSE_S = 120.0
_b08_seen: dict[UUID, tuple[float, dict[str, Any], list[str]]] = {}


async def _b08_for(project_id: UUID, notes: list[str], reuse: bool) -> dict[str, Any]:
    """B08 원천 — reuse = 시간 창 안 마지막 값(사본) · 아니면 새로 읽고 남김."""
    seen = _b08_seen.get(project_id)
    if reuse and seen and time.monotonic() - seen[0] < B08_REUSE_S:
        notes.extend(seen[2])
        return copy.deepcopy(seen[1])
    mine: list[str] = []
    out = await _b08(project_id, mine)
    _b08_seen[project_id] = (time.monotonic(), copy.deepcopy(out), mine)
    notes.extend(mine)
    return out


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, user.get("daega_cover")
    )
    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, draft: dict[str, Any] | None = None) -> dict[str, Any]:
    """셈 · 수요 · 사본 정리용 사용자 값 — 저장 값(draft = 화면 캐시 · 296) + 거리이정표 운반 연결 ·
    거리(화면엔 안 보냄) · 운반 사본 연결 = 지금 거리의 사본(263-2)."""
    user = {**(store.read_user(root) if draft is None else draft), **haul_user(haul_routes(root))}
    return library.resolve_hauls(root, user)


async def _demand(root: Path, draft: dict[str, Any] | None = None) -> list[str]:
    """단가 수요 키 — 연결된 사본 + 설계내역서 손 줄 로직(166-2) +
    거리이정표 운반 연결(251-5) · draft = 화면 캐시(296 · 서고 사본 = 그 줄이 잇는 것만)."""
    user = await asyncio.to_thread(calc_user, root, draft)  # 운반 값 사본 먼저(263-2)
    keep = None if draft is None else library.used_folders(user)
    keys = await asyncio.to_thread(linked_resource_keys, root, keep)
    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, draft: dict[str, Any] | None = None, reuse_b08: bool = False
) -> dict[str, Any]:
    notes: list[str] = []
    b08 = await _b08_for(project_id, notes, reuse_b08)
    keys = await _demand(root, draft)
    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)
    # 702-3 draft = 화면 캐시로 셈(저장 안 함) · 없으면 저장 값
    user = await asyncio.to_thread(store.read_user, root) if draft is None else draft
    calc = await asyncio.to_thread(library.resolve_hauls, root, {**user, **haul_user(hauls)})
    inputs["bill"] = bill_items({**b08, **calc, "haul_routes": hauls})
    inputs["hopyo"] = hopyo_prices(root)
    inputs["daega"] = await asyncio.to_thread(daega_inputs, root)
    library.fix_keys(calc, inputs["daega"].get("libs") or {})  # 263-2 바뀐 사본 = 그 블록
    sheets, summary = await asyncio.to_thread(compute_twice, root, inputs, calc)
    if lost := len(sheets.get("daega_orphans") or []):  # 702-2 데이터 점검 알림
        notes.append(
            f"짝 잃은 고친 칸 {lost} — 일위대가 줄이 바뀌어 붙을 줄 없음 · 고친 값은 남겨 둠"
        )
    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 = await asyncio.to_thread(calc_user, root)  # 263-2 운반 값 사본 = 지금 거리
    if (rows := calc["bill"].get("rows")) != ((user.get("bill") or {}).get("rows") or []):
        # 바뀐 사본 연결 = 저장 값에도(응답 user → 화면 캐시 · 다음 [저장]이 옛 lib 안 씀)
        bill = {**user["bill"], "rows": rows}
        await asyncio.to_thread(store.write_user, root, {**user, "bill": bill})
    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)
    await mark_stage(str(project_id), 6, "save")  # [저장](B1)
    return await payload(project_id, root)


@router.post("/{project_id}/estimation/complete")
async def estimation_complete(project_id: UUID) -> dict[str, Any]:
    """B09 [>] — 단계 6 완료(요구 300 B1) · 빈 줄 있으면 409(701-3 확정 = 빈 줄 0 · 서버 정본으로 셈)."""
    root = await _project_root(project_id)  # 없는 프로젝트는 404
    sheets = (await payload(project_id, root))["sheets"]
    blank = (((sheets.get("bill") or {}).get("totals") or {}).get("direct") or {}).get("blank")
    if blank:
        raise HTTPException(409, f"빈 줄 {blank} · 미완성 — 확정은 빈 줄 0 일 때만")
    await mark_stage(str(project_id), 6, "next")
    return {"status": "success", "project_id": str(project_id)}


def reset_files(root: Path) -> None:
    """[초기화] 파일 몫 — 사용자 값 · 단가 책 지움 · 운반 연결(B08 몫)이 안 잇는 서고 사본 지움."""
    (store.folder(root) / store.USER).unlink(missing_ok=True)
    (store.folder(root) / store.BOOK).unlink(missing_ok=True)
    library.prune(root, calc_user(root))
    reset_section(root, "estimation")


@router.post("/{project_id}/estimation/reset")
async def estimation_reset(project_id: UUID) -> dict[str, Any]:
    root = await _project_root(project_id)
    await asyncio.to_thread(reset_files, root)
    await mark_stage(str(project_id), 6, "save")  # [초기화] = 작업본 바꿈 = [저장]과 같음
    return await payload(project_id, root)


@router.post("/{project_id}/estimation/prices")
async def refresh_prices(
    project_id: UUID,
    body: dict[str, Any] | None = Body(None),
    session: dict[str, Any] = Depends(verify_session),
) -> dict[str, Any]:
    """[업데이트] 한 곳(703-3) — 원본이 바뀐 구조물 사본 재적용 → 단가 책 다시 굳힘 ·
    덮개 그대로."""
    root = await _project_root(project_id)
    raw = (body or {}).get("user")
    draft = store.clean_user(raw) if isinstance(raw, dict) else None  # 296 화면 캐시(저장 안 함)
    again = await asyncio.to_thread(refresh.reapply_changed, root, importer(session))
    keys = await _demand(root, draft)
    old = await asyncio.to_thread(store.read_book, root)
    book = await asyncio.to_thread(prices.refresh_book, keys, old)
    await asyncio.to_thread(store.write_book, root, book)
    out = await payload(project_id, root, draft)
    diff = prices.book_diff(old, book)
    cells = refresh.cover_cells((draft or out.get("user") or {}).get("daega_cover"))
    out["notes"] = [
        f"업데이트 — 단가 줄 {diff['new'] + diff['changed']} 바뀜 · 구조물 사본 "
        f"{len(again['다시'])} 다시 가져옴 · 고친 칸 {cells} 그대로",
        *(f"재적용 못 함 {why}" for why in again["막힘"]),
        *(out.get("notes") or []),
    ]
    return out


@router.post("/{project_id}/estimation/prices/preview")
async def refresh_preview(
    project_id: UUID, body: dict[str, Any] | None = Body(None)
) -> dict[str, Any]:
    """[업데이트] 미리 보기(703-3) — 바뀌는 단가 줄 · 다시 가져올 사본 · 지킬 고친 칸 · 노임 판."""
    root = await _project_root(project_id)
    raw = (body or {}).get("user")
    draft = store.clean_user(raw) if isinstance(raw, dict) else None
    keys = await _demand(root, draft)
    book = await asyncio.to_thread(store.read_book, root)
    user = draft if draft is not None else await asyncio.to_thread(store.read_user, root)
    return await asyncio.to_thread(refresh.preview, root, book, keys, user.get("daega_cover"))


@router.post("/{project_id}/estimation/calc")
async def calc_estimation(project_id: UUID, body: dict[str, Any] = Body(...)) -> dict[str, Any]:
    """702-3 조작 중 서버 셈 — 화면 캐시(user)로 GET 꼴 셈 · 저장 안 함 · 단가 = 굳힌 책
    (없는 줄 = 지금 M01 · [저장] 때 책에 굳힘 → 같은 값)."""
    root = await _project_root(project_id)
    return await payload(project_id, root, store.clean_user(body.get("user")), reuse_b08=True)


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)
    solver = sangeun.Solver(base)
    block = solver.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"],
        # 277-2 칸 나눔 — 맨 위 + 하위 산근 블록(차례) · 셈 그대로 · 화면이 머리 · 줄 · 계를 칸으로
        "blocks": [_block_shown(b) for b in sorted(solver.blocks.values(), key=lambda b: b["no"])],
    }


SHOWN_LINE = ("name", "spec", "qty", "unit", "kind", *sangeun.PARTS, "why", "text")


def _block_shown(block: dict[str, Any]) -> dict[str, Any]:
    """미리보기 칸 하나 — Solver 블록에서 화면이 쓰는 것만(머리 · 호표 줄 · 계 · 출처)."""
    return {
        **{k: block.get(k) for k in ("no", "logic", "name", "unit", "source", "head", "foot")},
        "price": block.get("sum"),
        "parts": {p: block.get(p) for p in sangeun.PARTS},
        "why": block.get("why"),
        "lines": [{k: ln.get(k) for k in SHOWN_LINE} for ln in block.get("lines") or []],
    }


@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
    # 263-2 운반공 줄 — 손 값 {줄 id: {입력: 값}} · 덮은 칸 · 그 행 거리 · 있는 사본
    given: dict[str, Any] | None = None
    own: list[str] | None = None
    dist: dict[str, Any] | None = None
    lib: str | None = None


@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),
            {"given": body.given, "own": body.own, "dist": body.dist, "lib": body.lib},
        )
    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}


@router.post("/{project_id}/estimation/library-inputs")
async def library_inputs(
    project_id: UUID, body: LibraryBody, session: dict[str, Any] = Depends(verify_session)
) -> dict[str, Any]:
    got = access.grant(session, body.layer, body.owner)
    root = await _project_root(project_id)
    try:
        lines = await asyncio.to_thread(
            library.library_inputs, root, body.kind, got.layer, got.owner, body.key, body.lib
        )
    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
    return {"lines": lines}


class PersonalBody(BaseModel):
    cover: str = Field(min_length=1, max_length=200)
    name: str | None = Field(None, max_length=200)
    overwrite: str | None = Field(None, max_length=20)
    user: dict[str, Any] | None = None


@router.post("/{project_id}/estimation/daega/personal")
async def save_personal(
    project_id: UUID, body: PersonalBody, session: dict[str, Any] = Depends(verify_session)
) -> dict[str, Any]:
    root = await _project_root(project_id)
    raw = (
        body.user if isinstance(body.user, dict) else await asyncio.to_thread(store.read_user, root)
    )
    cover = store.clean_user(raw).get("daega_cover") or {}
    try:
        made = await asyncio.to_thread(personal.build, root, body.cover, cover, body.name)
    except LookupError as error:
        raise HTTPException(status_code=404, detail=str(error)) from error
    if not made["문서"]["이름"]:
        raise HTTPException(status_code=400, detail="이름이 빔")
    folder = await asyncio.to_thread(_root, session, "personal", None, True)
    got = await asyncio.to_thread(_call, folder, personal.write, made["문서"], body.overwrite)
    return {
        "키": got["이름"],
        "이름": made["문서"]["이름"],
        "담음": made["담음"],
        "안담음": made["안담음"],
    }


def _past_rows(projects: list[dict[str, Any]], here: str) -> list[dict[str, Any]]:
    out = []
    for other in projects:
        if other["project_id"] == here or not other.get("storage_path"):
            continue
        try:
            rows = past.rows_of(Path(resolve_stored_project_path(other["storage_path"])))
        except (ValueError, OSError):  # 못 읽는 공사 = 건너뜀
            continue
        ref = {"project_id": other["project_id"], "project": other["name"]}
        out += [{**ref, **row} for row in rows]
    return out


@router.get("/{project_id}/estimation/past")
async def past_list(
    project_id: UUID, session: dict[str, Any] = Depends(verify_session)
) -> dict[str, Any]:
    here = await _project(session, str(project_id))
    projects = await _company_projects(int(here["company_id"]))
    return {"rows": await asyncio.to_thread(_past_rows, projects, str(project_id))}


class PastBody(BaseModel):
    project_id: str = Field(min_length=1, max_length=64)
    cover: str = Field(min_length=3, max_length=200)


@router.post("/{project_id}/estimation/past")
async def past_import(
    project_id: UUID, body: PastBody, session: dict[str, Any] = Depends(verify_session)
) -> dict[str, Any]:
    here = await _project(session, str(project_id))
    if body.project_id == str(project_id):
        raise HTTPException(status_code=400, detail="이 공사 것은 「이 공사」 탭에서 고름")
    other = await _project(session, body.project_id)
    if other["company_id"] != here["company_id"]:
        raise HTTPException(status_code=403, detail="같은 회사 공사에서만 가져옴")
    ref = {"project_id": body.project_id, "프로젝트": other.get("name")}
    try:
        made = await asyncio.to_thread(
            past.import_past, here["root"], other["root"], ref, body.cover, 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, here["root"])
    daega = out["inputs"].get("daega") or {}
    key = (daega.get("libs") or {}).get(made["폴더"])
    # 같은 꼴 블록이 이 공사에 먼저 있으면 그 블록(다른 cover) — 덮개 안 옮김(그 블록 덮개 그대로)
    cover = next((b["cover"] for b in daega.get("blocks") or [] if b["id"] == key), None)
    if cover != made.pop("덮개자리"):
        made["덮개"] = {}
    return {**made, "key": key, "cover": cover, "payload": out}


# 166-4 견적서 파일 길 · 707-1 경쟁사 내역 줄 불러오기 길 — 같은 머리 · 같은 권한
router.include_router(quotes_router)
router.include_router(import_router)


# 302 초안 길(draft/*_Router.py 의 router · 폴더째 지우면 길도 없음) — 같은 머리 · 같은 권한
def include_drafts(target: APIRouter, folder: Path, package: str) -> list[str]:
    """folder 의 *_Router.py 마다 router 를 끼움 → 끼운 파일 이름(못 읽은 초안은 건너뜀).
    초안이 같은 머리(/api/projects)를 스스로 달았으면 떼고 끼움 — 머리가 두 번 붙지 않게."""
    done = []
    for path in sorted(folder.glob("*_Router.py")):
        try:
            draft = importlib.import_module(f"{package}.{path.stem}").router
            if target.prefix and draft.prefix.startswith(target.prefix):
                # 길은 만들 때 머리가 박힘 — 사본 길에서 떼어 냄(원본은 앱에 바로 붙일 때 그대로)
                shim, cut = copy.copy(draft), len(target.prefix)
                shim.routes = [copy.copy(route) for route in draft.routes]
                for route in shim.routes:
                    route.path = route.path[cut:]
                draft = shim
            target.include_router(draft)
            done.append(path.stem)
        except (ImportError, AttributeError) as error:  # 한 초안이 막혀도 B09 는 뜸
            logging.getLogger(__name__).warning("초안 길 %s 못 읽음 — %s", path.name, error)
    return done


include_drafts(router, Path(__file__).parent / "draft", "B09_Estimation.draft")