Aislo/B09_Estimation/B09_Estimation_Router_Import.py

"""B09 경쟁사 결과물 불러오기 길 — 내역 줄(707-1) · 요율(707-2) · 기초 단가표(707-3) · 저장 안 함
(화면 캐시로 → [저장] 때 서버 정본).

    POST /api/projects/{id}/estimation/import/bill   (multipart: file | file_key · user?)
        읽기 + 미리 보기 — 파일 = 프로젝트 `B09_Estimation/v1/import/{표지}/`
        (file_key = 표지 · 같은 파일 다시 씀) → {file_key, 파일, 꼴, 시트, 묶음, 항목, 합계,
        조달수수료율, 합계줄, 율줄, 같은이름, 남는자동, 이음} ·
        user = 화면 캐시(JSON 글 · 없으면 저장 값) · DRM = 400 「읽을 수 없음」
    POST /api/projects/{id}/estimation/import/bill/apply   {file_key, mode, attach, drop, user?}
        짜기(`B09_Estimation_Bill_Import.apply`) →
        {user(새 캐시), 알림, 어긋남, 엑셀합계, payload(그 캐시로 셈)} ·
        이음 = 읽을 수 있는 층(개인 → 회사 → 시스템)에서 같은 파일 이름으로 들여온 실무자료 사전조합
        (본뜬 「HP-… 제N호표」) → 서고 사본(층 쓰기 0 · 프로젝트 사본만) · 대가 N 줄 연결.
    POST /api/projects/{id}/estimation/import/rates    (multipart: file | file_key · user?)
        원가계산서 요율(707-2) → {file_key, name, kind, rows, rates, notes, check, patch{estimate}}
    POST /api/projects/{id}/estimation/import/prices   (multipart: file | file_key · user?)
        노임 · 자재 · 기계 단가표(707-3) → 이 공사 단가 책에 이음 →
        {file_key, name, sheets, counts, linked, missing,
        patch{labor_cover, machine_cover, materials}}
    file_key = 세 길 같은 표지(`_keep` · `_saved`) — 같은 파일을 다시 올리지 않음 ·
    .estx(EST 원본 · 707-5 `B09_Estimation_Estx`) · .STC(STmate 원본 · 707-6 `B09_Estimation_Stc`) =
        이름 끝으로 갈림 — 내역 줄 · 단가표 같은 그릇 · 요율 = 400 「같은 공사 엑셀에서」 ·
        미리 보기 `호표` = 대가 · 산근 목록 수 + 코드로 가리킨 줄(이음 후보만) ·
        `알림` = 안 읽은 몫(다공구 · 변경 차수)
    화면은 `patch` 를 캐시에 얹고 [저장] 때 서버가 정본으로 다시 셈.
`B09_Estimation_Router` 가 끼워 넣음(같은 머리 · 같은 권한).
"""

from __future__ import annotations

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

from fastapi import APIRouter, Depends, File, Form, HTTPException, UploadFile
from pydantic import BaseModel, Field

from B08_Quantity.B08_Quantity_Router_Hydro import _project_root
from B09_Estimation import B09_Estimation_Bill_Import as importer
from B09_Estimation import B09_Estimation_Bill_Library as library
from B09_Estimation import B09_Estimation_Estx as estx
from B09_Estimation import B09_Estimation_Import_Prices as import_prices
from B09_Estimation import B09_Estimation_Import_Rates as import_rates
from B09_Estimation import B09_Estimation_Prices as prices
from B09_Estimation import B09_Estimation_Stc as stc
from B09_Estimation import B09_Estimation_Store as store
from B09_Estimation.B09_Estimation_Bill_Excel import BillExcelError, read_bill
from B09_Estimation.B09_Estimation_Calc_Base import compute_base
from B09_Estimation.B09_Estimation_Labor_Source import daega_inputs
from B09_Estimation.B09_Estimation_Rates import rate_table
from common_util.common_util_auth import verify_session
from M02_MasterTemplete import M02_Hopyo_Build as hopyo_build
from M02_MasterTemplete import M02_Labor_Ref as refs
from M02_MasterTemplete import M02_Layer_Access as access
from M02_MasterTemplete.M02_Hopyo_Excel import ExcelError
from M02_MasterTemplete.M02_Item_Snapshot import importer as who_of

router = APIRouter()
ROOT = "/{project_id}/estimation/import"
PATH = ROOT + "/bill"
IMPORT = "import"
MAX_BYTES = 50 * 1024 * 1024
_TAG = re.compile(r"^[0-9a-f]{8}$")
_MARK = re.compile(r"^HP-\d{6} 제(\d+)호표$")


def _saved(root: Path, key: str) -> Path:
    """file_key → 둔 파일 — 없으면 404."""
    folder = store.folder(root) / IMPORT / key
    files = (
        [f for f in folder.iterdir() if f.is_file()] if _TAG.match(key) and folder.is_dir() else []
    )
    if not files:
        raise HTTPException(status_code=404, detail="불러온 파일이 없음 — 다시 고름")
    return files[0]


def _keep(root: Path, name: str, data: bytes) -> str:
    key = importer.tag_of(data)
    folder = store.folder(root) / IMPORT / key
    folder.mkdir(parents=True, exist_ok=True)
    for old in folder.iterdir():
        old.unlink()
    (folder / (store.safe_name(name) or "내역.xlsx")).write_bytes(data)
    return key


_RAW = {".estx": estx, ".stc": stc}


def _raw(name: str) -> Any:
    """원본 파일 읽기 모듈(.estx · .STC) — 엑셀이면 None."""
    return _RAW.get(Path(name).suffix.lower())


def _parse(name: str, data: bytes) -> dict[str, Any]:
    try:
        return (raw.read_bill if (raw := _raw(name)) else read_bill)(name, data)
    except BillExcelError as error:
        raise HTTPException(status_code=400, detail=str(error)) from error


def _hopyo(name: str, data: bytes, parsed: dict[str, Any]) -> dict[str, int] | None:
    """원본 호표 목록 수 · 내역 줄이 코드로 가리킨 수(이음 후보 — 구성 줄은 `UNKNOWN`)."""
    if not (raw := _raw(name)):
        return None
    got = raw.read_hopyo(name, data)
    codes = {(k, x["코드"]) for k, v in got.items() for x in v}
    hits = [h for line in parsed["줄"] if (h := line.get("호표"))]
    return {**{k: len(v) for k, v in got.items()},
            "가리킴": sum((h["종류"], h["코드"]) in codes for h in hits)}  # fmt: skip


def _draft(raw: Any) -> dict[str, Any] | None:
    if raw in (None, ""):
        return None
    try:
        user = json.loads(raw) if isinstance(raw, str) else raw
    except ValueError as error:
        raise HTTPException(status_code=400, detail="user 가 JSON 이 아님") from error
    return store.clean_user(user) if isinstance(user, dict) else None


def _found(session: dict[str, Any], name: str, numbers: set[int]) -> dict[int, tuple]:
    """대가 번 → (층, 주인, 사전조합 키) — 같은 파일 이름 실무자료 본뜬 · 앞 층이 이김."""
    out: dict[int, tuple] = {}
    for layer in ("personal", "company", "system"):
        try:
            got = access.grant(session, layer)
        except HTTPException:  # 회사 없음 등 — 그 층 건너뜀
            continue
        for key, origin, _raw in hopyo_build._origins(refs.folder_of(got.layer, got.owner)):
            hit = _MARK.match(str(origin.get("키") or ""))
            no = int(hit.group(1)) if hit else None
            if origin.get("파일") == name and no in numbers and no not in out:
                out[no] = (got.layer, got.owner, key)
    return out


def _links(root: Path, found: dict[int, tuple], who: Any) -> dict[int, dict[str, Any]]:
    folders = {}
    for no, (layer, owner, key) in found.items():
        try:
            folders[no] = library.import_copy(root, "precombo", layer, owner, key, who)["폴더"]
        except (LookupError, ValueError):  # 원본 없음 · 모양 틀림 = 고정 단가로
            continue
    libs = (daega_inputs(root).get("libs") or {}) if folders else {}
    return {
        no: {"tab": "dg", "key": libs[name], "lib": name}
        for no, name in folders.items()
        if libs.get(name)
    }


def _numbers(parsed: dict[str, Any]) -> set[int]:
    return {
        int(h["번"])
        for line in parsed["줄"]
        if (h := line.get("호표") or {}).get("종류") == "대가" and "번" in h
    }


@router.post(PATH)
async def import_bill(
    project_id: UUID,
    file: UploadFile | None = File(None),
    file_key: str | None = Form(None),
    user: str | None = Form(None),
    session: dict[str, Any] = Depends(verify_session),
) -> dict[str, Any]:
    from B09_Estimation.B09_Estimation_Router import payload  # 서로 부름 — 부를 때만

    root = await _project_root(project_id)
    if file is not None:
        data = await file.read(MAX_BYTES + 1)
        if len(data) > MAX_BYTES:
            raise HTTPException(status_code=400, detail="파일이 너무 큼(50MB 넘음)")
        parsed = await asyncio.to_thread(_parse, file.filename or "", data)  # 읽힌 파일만 둠
        file_key = await asyncio.to_thread(_keep, root, file.filename or "", data)
    if not file_key:
        raise HTTPException(status_code=400, detail="파일을 고름")
    path = _saved(root, file_key)
    data = path.read_bytes()
    if file is None:
        parsed = await asyncio.to_thread(_parse, path.name, data)
    hopyo = await asyncio.to_thread(_hopyo, path.name, data, parsed)
    rows = (await payload(project_id, root, _draft(user), True))["sheets"]["bill"]["rows"]
    found = await asyncio.to_thread(_found, session, path.name, _numbers(parsed))
    return {
        "file_key": file_key,
        "파일": path.name,
        **importer.preview(parsed, rows),
        "이음": len(found),
        "호표": hopyo,
        "알림": parsed.get("알림") or [],
    }


class ApplyBody(BaseModel):
    file_key: str = Field(min_length=8, max_length=8)
    mode: str = "take"
    attach: bool = True
    drop: bool = False
    user: dict[str, Any] | None = None


@router.post(PATH + "/apply")
async def apply_bill(
    project_id: UUID, body: ApplyBody, session: dict[str, Any] = Depends(verify_session)
) -> dict[str, Any]:
    from B09_Estimation.B09_Estimation_Router import payload  # 서로 부름 — 부를 때만

    if body.mode not in importer.MODES:
        raise HTTPException(status_code=400, detail="수량 갈래는 take · price 중 하나")
    root = await _project_root(project_id)
    path = _saved(root, body.file_key)
    parsed = await asyncio.to_thread(_parse, path.name, path.read_bytes())
    draft = _draft(body.user) or await asyncio.to_thread(store.read_user, root)
    rows = (await payload(project_id, root, draft, True))["sheets"]["bill"]["rows"]
    found = await asyncio.to_thread(_found, session, path.name, _numbers(parsed))
    links = await asyncio.to_thread(_links, root, found, who_of(session))
    made, report = importer.apply(
        parsed, draft, rows, body.file_key, body.mode, body.attach, body.drop, links
    )
    made = store.clean_user(made)
    out = await payload(project_id, root, made, True)
    lines = out["sheets"]["bill"]["rows"]
    return {
        "user": made,
        "알림": {k: v for k, v in report.items() if k != "줄"},
        "어긋남": importer.differences(parsed, report, lines),
        "엑셀합계": parsed["합계"],
        "payload": out,
    }


async def _file(
    file: UploadFile | None, file_key: str | None, root: Path
) -> tuple[str, str, bytes]:
    """올린 파일 · 표지 → (표지, 이름, 내용) — 올린 파일은 내역 줄 길과 같은 자리에 둠."""
    if file is not None:
        data = await file.read(MAX_BYTES + 1)
        if len(data) > MAX_BYTES:
            raise HTTPException(status_code=400, detail="파일이 너무 큼(50MB 넘음)")
        key = await asyncio.to_thread(_keep, root, file.filename or "", data)
        return key, store.safe_name(file.filename or "") or "내역.xlsx", data
    if not file_key:
        raise HTTPException(status_code=400, detail="file 또는 file_key 가 필요함")
    path = _saved(root, file_key)
    return file_key, path.name, path.read_bytes()


@router.post(ROOT + "/rates")
async def import_rate_sheet(
    project_id: UUID,
    file: UploadFile | None = File(None),
    file_key: str | None = Form(None),
    user: str | None = Form(None),
) -> dict[str, Any]:
    root = await _project_root(project_id)
    key, name, data = await _file(file, file_key, root)
    draft = _draft(user)
    if _raw(name):
        raise HTTPException(
            400, f"원가계산서 요율은 {Path(name).suffix} 에서 못 읽음 — 같은 공사 엑셀에서 불러옴"
        )
    mine = (draft if draft is not None else await asyncio.to_thread(store.read_user, root))[
        "estimate"
    ]
    try:
        read = await asyncio.to_thread(import_rates.read_rates, name, data)
    except ExcelError as error:
        raise HTTPException(400, str(error)) from error
    check = await asyncio.to_thread(import_rates.check, read, mine, rate_table())
    estimate = {**mine, **read["estimate"], "rates": {**(mine.get("rates") or {}), **read["rates"]}}
    keep = ("kind", "sheet", "rows", "rates", "notes")
    return {"file_key": key, "name": name, **{k: read[k] for k in keep}, "check": check,
            "patch": {"estimate": estimate}}  # fmt: skip


@router.post(ROOT + "/prices")
async def import_price_sheets(
    project_id: UUID,
    file: UploadFile | None = File(None),
    file_key: str | None = Form(None),
    user: str | None = Form(None),
) -> dict[str, Any]:
    from B09_Estimation import B09_Estimation_Router as main  # 서로 부름 — 부를 때만

    root = await _project_root(project_id)
    key, name, data = await _file(file, file_key, root)
    draft = _draft(user)
    mine = draft if draft is not None else await asyncio.to_thread(store.read_user, root)
    try:
        reader = raw.read_prices if (raw := _raw(name)) else import_prices.read_prices
        read = await asyncio.to_thread(reader, name, data)
    except (ExcelError, BillExcelError) as error:
        raise HTTPException(400, str(error)) from error
    keys = await main._demand(root)  # 화면 단가 목록(GET)과 같은 수요 — 사본 전부
    b08 = await main._b08_for(project_id, [], True)
    book = await asyncio.to_thread(store.read_book, root)
    inputs = await asyncio.to_thread(prices.build_inputs, book, keys, b08["materials"])
    base = compute_base(inputs, {})  # 덮개 뺀 이 공사 단가 책
    out = import_prices.propose(read, base, mine)
    sheets = {k: read[f"{k}_sheet"] for k in import_prices.KINDS}
    return {"file_key": key, "name": name, "sheets": sheets, **out}