Aislo/B09_Estimation/draft/B09_Draft_Schedule_Source.py

"""B09 공기 초안 서버 원천(PLAN 252-2 · 사용자 161번 — draft 폴더째 지우면 사라짐).

본 코드 훅 = `B09_Estimation_Router.payload` · `B09_Estimation_Store.clean_user` 가 이 모듈을
    가져오지 못하면 건너뜀(공기 값 · 탭 없이 그대로).
    inputs["schedule"] = 원천(줄마다 단위당 자원 · 공휴일 표 · 기후 기본값 · 착공 기본값)
    sheets["schedule"] = 서버 정본 셈(`…_Calc.compute_schedule` · 화면은 TS 짝으로 즉시)
단위당 자원 = 설계내역서 항목 줄의 호표 1 단위 역전개(`B09_Estimation_Summary.expand`) —
    주 장비 = 단위당 시간이 가장 긴 장비 · 인력 = 단위당 인 합(직종 무관)
공휴일 = `B09_Draft_Schedule_Holidays.json`(2024~2040 · 법정 · 대체 · 확정 선거일 — 공개 `holidays`
    목록 0.106 에서 뽑음 · 2027 이후 선거일은 날짜가 안 정해져 뺌 · 달마다 사용자 덮개)
기후 기본값 = 참고 설계서(25 거창) 월별 강수일(일강수량 5mm 이상) · 혹서일(33℃ 이상) —
    지역마다 사용자 고침
사용자 값 = user["schedule"] — 다듬기 `clean_schedule`(모르는 칸 버림 · 수 칸은 수)
"""

from __future__ import annotations

import datetime
import json
from pathlib import Path
from typing import Any

from B09_Estimation.B09_Estimation_Calc_Base import num
from B09_Estimation.B09_Estimation_Summary import expand, hopyo_lines
from B09_Estimation.draft.B09_Draft_Schedule_Calc import compute_schedule

HOLIDAYS = Path(__file__).with_name("B09_Draft_Schedule_Holidays.json")
CLIMATE = {
    "label": "경남 거창(참고 설계서 · 강수 = 일강수량 5mm 이상 · 혹서 = 33℃ 이상)",
    "rain": [1.2, 1.4, 3.6, 4.5, 3.6, 4.4, 7.7, 6.9, 5.5, 3.1, 2.2, 1.2],
    "heat": [0, 0, 0, 0, 0.5, 1.2, 5.9, 8.4, 0, 0, 0, 0],
}
START_MONTH = 3
NUMBERS = ("hours", "prep_docs", "prep_setup", "etc", "tidy")


def unit_resources(sheets: dict[str, Any]) -> dict[str, dict[str, Any]]:
    """설계내역서 항목 줄 → {줄 키: {machine: {key, name, hr} | None, labor: 인 | None}}(1 단위)."""
    lines_of = hopyo_lines(sheets)
    machines = {row["key"]: row for row in sheets.get("machine") or []}
    out: dict[str, dict[str, Any]] = {}
    for row in (sheets.get("bill") or {}).get("rows") or []:
        key = str((row.get("ref") or {}).get("key") or "")
        if row.get("kind") != "item" or key not in lines_of:
            continue
        used: dict[str, dict[str, float]] = {}
        expand(lines_of, key, 1.0, used)
        machine = None
        hours = used.get("machine") or {}
        if hours:
            mkey, hr = max(hours.items(), key=lambda kv: kv[1])
            if hr > 0:
                src = machines.get(mkey) or {}
                name = " ".join(t for t in (src.get("name"), src.get("spec")) if t) or mkey
                machine = {"key": mkey, "name": name, "hr": hr}
        labor = sum((used.get("labor") or {}).values())
        out[row["key"]] = {"machine": machine, "labor": labor if labor > 0 else None}
    return out


def holiday_table() -> dict[str, list[list[str]]]:
    try:
        return json.loads(HOLIDAYS.read_text(encoding="utf-8"))
    except (OSError, ValueError):
        return {}


def schedule_inputs(sheets: dict[str, Any], today: datetime.date | None = None) -> dict[str, Any]:
    today = today or datetime.date.today()
    return {
        "units": unit_resources(sheets),
        "holidays": holiday_table(),
        "climate": CLIMATE,
        "start": {"year": today.year + 1, "month": START_MONTH},
    }


def attach(inputs: dict[str, Any], sheets: dict[str, Any], user: dict[str, Any]) -> None:
    """payload 훅 — 원천 · 서버 정본 셈을 끼움."""
    inputs["schedule"] = schedule_inputs(sheets)
    bill = (sheets.get("bill") or {}).get("rows") or []
    sheets["schedule"] = compute_schedule(inputs["schedule"], bill, user.get("schedule") or {})


def _positive(value: Any) -> float | None:
    v = num(value)
    return v if v is not None and v > 0 else None


def _twelve(values: Any) -> list[float | None] | None:
    if not isinstance(values, list) or len(values) != 12:
        return None
    return [v if (v := num(x)) is not None and v >= 0 else None for x in values]


def clean_schedule(raw: Any) -> dict[str, Any]:
    """user["schedule"] 다듬기 — 모르는 칸 버림 · 수 칸은 0 이상 수 · 빈 것 = 칸 없음."""
    if not isinstance(raw, dict):
        return {}
    out: dict[str, Any] = {}
    for field in NUMBERS:
        v = num(raw.get(field))
        if v is not None and 0 <= v <= 1000:
            out[field] = v
    start = raw.get("start") if isinstance(raw.get("start"), dict) else {}
    year, month = num(start.get("year")), num(start.get("month"))
    if year and month and 2000 <= year <= 2100 and 1 <= month <= 12:
        out["start"] = {"year": int(year), "month": int(month)}
    climate = raw.get("climate") if isinstance(raw.get("climate"), dict) else {}
    kept = {k: v for k in ("rain", "heat") if (v := _twelve(climate.get(k)))}
    if kept:
        out["climate"] = kept
    holiday = raw.get("holiday") if isinstance(raw.get("holiday"), dict) else {}
    marks = {
        str(k): v
        for k, x in holiday.items()
        if len(str(k)) == 7 and (v := num(x)) is not None and 0 <= v <= 31
    }
    if marks:
        out["holiday"] = marks
    rows = {}
    for key, cell in (raw.get("rows") or {}).items() if isinstance(raw.get("rows"), dict) else []:
        if not isinstance(cell, dict) or not str(key) or len(str(key)) > 200:
            continue
        one = {}
        rate = num(cell.get("rate"))
        if rate is not None and 0 <= rate <= 100:
            one["rate"] = rate
        for field in ("units", "crew", "daily"):
            if (v := _positive(cell.get(field))) is not None:
                one[field] = v
        if one:
            rows[str(key)] = one
    if rows:
        out["rows"] = rows
    return out