Aislo/B09_Estimation/B09_Estimation_Store.py

"""B09 원가 저장 자리(PLAN 144-1) — 프로젝트 안 `B09_Estimation/v1/`.

    price_book.json   굳힌 단가 한 벌 {"made": 시각, "rows": {키: 입력 줄},
                      "driver_factor": …, "fx": […]}
                      — [단가 갱신] 때만 M01 로 새로 ·
                      [저장] 때 책에 없는 새 줄만 더함(있는 줄 안 바꿈)
    user_values.json  사용자 값 {"materials": {자재 키: {applied?, supply?}}, "estimate": {…},
                      "bill": {"qty": {줄 키: 수}, "rows": [손 줄]}}(166-2 설계내역서 손질)
                      — 고친 칸만(키가 있으면 그 값 · null = 빈칸 · 없으면 기본)

결과(표)는 저장하지 않음 — GET 때 서버가 셈(B08 꼴).
"""

from __future__ import annotations

import json
import math
import os
from pathlib import Path
from typing import Any

from B09_Estimation.B09_Estimation_Calc_Base import SUPPLY
from B09_Estimation.B09_Estimation_Calc_Bill import SECTIONS

FOLDER = ("B09_Estimation", "v1")
BOOK = "price_book.json"
USER = "user_values.json"

#: 원가계산 사용자 입력 칸 — 요율 고르는 값 · 손입력(엑셀 총괄 · 원가계산서 · 내역 손입력 칸)
ESTIMATE_FIELDS = (
    "estimated_price",  # 추정가격(원) — 요율 구간 고름
    "duration_days",  # 공사기간(일)
    "work_type",  # 공종(글)
    "safety_base",  # 산업안전보건관리비 기초액(원)
    "profit_adjust",  # 이윤 보정(원 · 빼는 값)
    "vat_adjust",  # 부가가치세 조정(원)
    "indirect_material",  # 간접재료비(원)
    "scrap",  # 작업설 부산물(원 · 빼는 값)
    "procurement_fee_pct",  # 조달수수료율(%)
)


def folder(project_root: str | Path) -> Path:
    return Path(project_root).joinpath(*FOLDER)


def _read(path: Path) -> dict[str, Any] | None:
    try:
        data = json.loads(path.read_text(encoding="utf-8"))
    except (OSError, ValueError):
        return None
    return data if isinstance(data, dict) else None


def _write(path: Path, data: dict[str, Any]) -> None:
    path.parent.mkdir(parents=True, exist_ok=True)
    temp = path.with_suffix(".tmp")
    temp.write_text(json.dumps(data, ensure_ascii=False, indent=1), encoding="utf-8")
    os.replace(temp, path)


def read_book(project_root: str | Path) -> dict[str, Any] | None:
    book = _read(folder(project_root) / BOOK)
    if book is None or not isinstance(book.get("rows"), dict):
        return None
    return book


def write_book(project_root: str | Path, book: dict[str, Any]) -> None:
    _write(folder(project_root) / BOOK, book)


def _num(value: Any) -> float | None:
    if isinstance(value, bool) or not isinstance(value, (int, float)):
        return None
    return float(value) if math.isfinite(value) else None


def _text(value: Any) -> str:
    return str(value)[:200] if isinstance(value, (str, int, float)) else ""


def _clean_ref(ref: Any) -> dict[str, Any] | None:
    """손 줄 연결 — dg · sg = 호표 키 · logic = M01 로직(입력값 · 켜쌓기율 · 고름)."""
    if not isinstance(ref, dict):
        return None
    tab = ref.get("tab")
    if tab in ("dg", "sg") and isinstance(ref.get("key"), str) and ref["key"]:
        return {"tab": tab, "key": ref["key"]}
    if tab == "logic" and isinstance(ref.get("code"), str) and ref["code"]:
        given = ref.get("given") if isinstance(ref.get("given"), dict) else {}
        picks = ref.get("picks") if isinstance(ref.get("picks"), dict) else None
        return {"tab": "logic", "code": ref["code"], "name": _text(ref.get("name")),
                "given": {str(k): v for k, v in given.items() if isinstance(v, (str, int, float))},
                "rate": _num(ref.get("rate")), "picks": picks}  # fmt: skip
    return None


def clean_bill(raw: Any) -> dict[str, Any]:
    """설계내역서 손질(166-2) — 수량 덮어쓰기(수만) · 손 줄(`hand:` 키 · 아는 공종만)."""
    raw = raw if isinstance(raw, dict) else {}
    qty = raw.get("qty") if isinstance(raw.get("qty"), dict) else {}
    codes = {code for code, _name, _parent in SECTIONS}
    rows = []
    for row in raw.get("rows") if isinstance(raw.get("rows"), list) else []:
        if not isinstance(row, dict) or not str(row.get("key") or "").startswith("hand:"):
            continue
        if row.get("section") not in codes:
            continue
        rows.append(
            {
                "key": str(row["key"]),
                "section": row["section"],
                "after": _text(row.get("after")),
                **{k: _text(row.get(k)) for k in ("name", "spec", "unit")},
                "qty": _num(row.get("qty")),
                "ref": _clean_ref(row.get("ref")),
            }
        )
    out: dict[str, Any] = {}
    kept = {str(k): _num(v) for k, v in qty.items() if _num(v) is not None}
    if kept:
        out["qty"] = kept
    if rows:
        out["rows"] = rows
    return out


def clean_user(raw: Any) -> dict[str, Any]:
    """사용자 값 다듬기 — 모르는 칸 버림 · 수 칸은 수 · null(빈칸)만."""
    raw = raw if isinstance(raw, dict) else {}
    materials: dict[str, dict[str, Any]] = {}
    for key, cell in (raw.get("materials") or {}).items():
        if not isinstance(cell, dict):
            continue
        out: dict[str, Any] = {}
        if "applied" in cell:
            out["applied"] = _num(cell["applied"])
        if cell.get("supply") in SUPPLY:
            out["supply"] = cell["supply"]
        if out:
            materials[str(key)] = out
    estimate: dict[str, Any] = {}
    given = raw.get("estimate") if isinstance(raw.get("estimate"), dict) else {}
    for field in ESTIMATE_FIELDS:
        if field not in given:
            continue
        value = given[field]
        estimate[field] = (str(value) if value else None) if field == "work_type" else _num(value)
    rates = given.get("rates")
    if isinstance(rates, dict):
        estimate["rates"] = {str(k): _num(v) for k, v in rates.items()}
    bill = clean_bill(raw.get("bill"))
    return {"materials": materials, "estimate": estimate, **({"bill": bill} if bill else {})}


def read_user(project_root: str | Path) -> dict[str, Any]:
    return clean_user(_read(folder(project_root) / USER))


def write_user(project_root: str | Path, user: dict[str, Any]) -> dict[str, Any]:
    clean = clean_user(user)
    _write(folder(project_root) / USER, clean)
    return clean