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℃ 이상) —
    고르기(264) = 참고 설계서 셋(거창 25 · 울진 24 · 울진 소광 25 「공사기간산정표」) · 0에서 직접 ·
    달마다 사용자 덮개가 이김
사용자 값 = user["schedule"] — 다듬기 `clean_schedule`(모르는 칸 버림 · 수 칸은 수 · 고르기 칸은 목록 안)
"""

from __future__ import annotations

import calendar
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 HOLIDAY_KINDS, 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],
}
#: 기후 자료 고르기(264) — 첫 줄 = 기본 · rain_note · heat_note = 공기집계 비고
CLIMATES = {
    "geochang25": {
        **CLIMATE,
        "name": "거창 25 참고",
        "rain_note": "일강수량 5mm 이상",
        "heat_note": "33℃ 이상",
    },
    "uljin24": {
        "name": "울진 24 참고",
        "label": "경북 울진(참고 설계서 24 · 강수 = 일강수량 5mm 이상 · 혹서 없음)",
        "rain": [0.6, 1.3, 2.9, 4.3, 3.2, 3.9, 7.4, 8.3, 4, 2.8, 2.6, 1.4],
        "heat": [0] * 12,
        "rain_note": "일강수량 5mm 이상",
        "heat_note": "혹서 없음(참고 설계서)",
    },
    "sogwang25": {
        "name": "울진 소광 25 참고",
        "label": "경북 울진 소광(참고 설계서 25 · 강수 = 일강수량 10mm 이상 · 혹서 없음)",
        "rain": [2.1, 2, 3.3, 4, 2.1, 3.7, 6.3, 7.9, 5, 4.2, 3.6, 1.6],
        "heat": [0] * 12,
        "rain_note": "일강수량 10mm 이상",
        "heat_note": "혹서 없음(참고 설계서)",
    },
    "zero": {
        "name": "0에서 직접",
        "label": "직접 입력(달마다 강수 · 혹서 칸)",
        "rain": [0] * 12,
        "heat": [0] * 12,
        "rain_note": "직접 입력",
        "heat_note": "직접 입력",
    },
}
START_MONTH = 3
NUMBERS = ("hours", "prep_docs", "prep_setup", "etc", "tidy", "rate_haul", "rate_base")
POSITIVE = ("units", "crew")
#: 고르기 칸 — 기본값이 첫째
CHOICES = {
    "climate_set": tuple(CLIMATES),
    "rest": ("sun", "satsun"),
    "machine_basis": ("max", "sum"),
    "c_round": ("none", "round1"),
    "last_off": ("ratio", "all"),
}


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())
        # 264 주 장비 기준 「장비 시간 합」 = 차례로 일하는 장비 · n = 장비 수
        spent = [hr for hr in hours.values() if hr > 0]
        out[row["key"]] = {
            "machine": machine,
            "labor": labor if labor > 0 else None,
            "machine_sum": {"hr": sum(spent), "n": len(spent)} if machine 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,
        "climates": CLIMATES,
        "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 <= (100 if field.startswith("rate") else 1000):
            out[field] = v
    for field in POSITIVE:
        if (v := _positive(raw.get(field))) is not None and v <= 1e4:
            out[field] = v
    for field, allowed in CHOICES.items():
        if raw.get(field) in allowed:
            out[field] = raw[field]
    off = raw.get("holiday_off") if isinstance(raw.get("holiday_off"), list) else []
    kinds = [k for k in HOLIDAY_KINDS if k in off]
    if kinds:
        out["holiday_off"] = kinds
    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)}
        day = num(start.get("day"))
        if day and 1 < day <= calendar.monthrange(int(year), int(month))[1]:
            out["start"]["day"] = int(day)
    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