"""M02 일위대가 맞춤 — 우리 돈 로직 지문 · 마스터 판 열쇠 · 디스크 캐시 · 엔진 셈 캐시
(PLAN 338-1 · 설계 2-3).

지문 = 정적(호표 인력 → 직종 · 「로직(기계경비 …)」 → 장비 갈래 · 규격 · 재료 이름 · 덧줄 % ·
이름 · 고르기 글 조건 낱말 · 기계경비 로직) + 변형(고르기 입력 곱 64 이하 모두 · 넘으면 하나씩 ·
지역 · 조종원직종 · 보정작업 · 기종 목록 칸 뺌)마다 단위당 L(직종 인) · X(장비 시간) · M(자재) ·
S(딸린 로직) · mix(인력비율 · 기계비율).
캐시 = `tmp/m02_hopyo/우리지문_<판>.json`(환경 `AISLO_HOPYO_CACHE` 로 바꿈) · 판 = 마스터 첫 층 JSON
sha256 묶음 · 처음 셈은 뒤 스레드(≈ 2 분) · 다음은 읽기만.
엔진 = `Labor_Doc.run_line` 과 같은 길(자동값 + 입력 → `mf.run`) · 셈 잠금 하나(뒤 셈 · 요청 겹침) ·
메모리 캐시. 원본(`resources/master_data`) 읽기만. 규칙은 336 실험(C4 `s2_ours_fp` · K5 `run`)
그대로.
"""

from __future__ import annotations

import hashlib
import itertools
import json
import math
import os
import re
import threading
import time
from decimal import Decimal
from pathlib import Path
from typing import Any

from common_util.common_util_json import atomic_write_json
from M01_MasterData import M01_MasterData_Store as m01
from M01_MasterData import M01_MasterData_Store_Cache as m01_cache
from M02_MasterTemplete import M02_Hopyo_Match as hopyo_match
from M02_MasterTemplete.M02_Hopyo_Words import conds, mtypes, specnums

FP_VERSION = 1  # 지문 꼴이 바뀌면 올림 — 옛 캐시 버림
ROOT = Path(__file__).resolve().parents[1]
_PER = re.compile(r"원\s*/\s*(\d[\d,]*\.?\d*|\.\d+)?")
_SUB = re.compile(r"^\s*로직\(\s*(\w+)\s*(?:,(.*))?\)\s*$")
EQ_CODE = re.compile(r"^\d{4}-\d{4}$")
_PCT = re.compile(r"\*\s*([\d.]+)\s*/\s*100")
SKIP_IN = ("지역", "조종원직종", "보정작업")
FIXED_MACH = {
    "GC000267": {"굴착기"},
    "GC000270": {"굴착기", "집게"},
    "GC000271": {"굴착기", "브레이커"},
    "GF000145": {"덤프트럭"},
}
RUN_CAP = 50_000  # ponytail: 엔진 셈 캐시 칸 수 — 넘으면 통째 비움(LRU 는 필요해지면)
_ENGINE = threading.RLock()  # ponytail: 셈 잠금 하나 — 엔진을 뒤 · 요청 스레드가 같이 부를 때
_STATE: dict[str, Any] = {"판": None, "자국": None, "표": None, "스레드": None}
_RUN: dict[tuple, Any] = {}
_LOCK = threading.Lock()
_NO_PRICE = {str(i): "-" for i in range(400)}  # 고른 자재 = 없는 키 → 값 없음 · 찾기 안 함
WHOLE_TTL = 2.0  # ponytail: 마스터 색인을 2 초 붙듦 — 줄마다 파일 자국을 다시 안 봄


def folder() -> Path:
    return Path(os.environ.get("AISLO_HOPYO_CACHE") or ROOT / "tmp" / "m02_hopyo")


# ── 마스터 판 열쇠 ─────────────────────────────────────────────────────
def version() -> str:
    """마스터 판 = 첫 층 *.json(밑줄 뺌) 이름 + sha256 16 묶음의 sha256 앞 16 — 자국이 같으면 다시
    안 셈."""
    stamps = m01_cache._stamps(Path(m01.FOLDER))
    if _STATE["자국"] != stamps:
        parts = [
            f"{name}:{m01.version_of((Path(m01.FOLDER) / name).read_bytes())}" for name in stamps
        ]
        _STATE.update(자국=stamps, 판=hashlib.sha256("\n".join(parts).encode()).hexdigest()[:16])
    return _STATE["판"]


def path_of(ver: str | None = None) -> Path:
    return folder() / f"우리지문_{ver or version()}.json"


# ── 엔진 한 번 ─────────────────────────────────────────────────────────
def whole() -> Any:
    now = time.monotonic()
    if _STATE.get("색인") is None or now - _STATE.get("색인때", 0) > WHOLE_TTL:
        _STATE.update(색인=m01.loaded()[1], 색인때=now)
    return _STATE["색인"]


def is_mach_logic(row: dict[str, Any]) -> bool:
    return "시간당" in str(row.get("이름") or "")


def eq_of(code: Any) -> tuple[set[str], set[float]]:
    one = whole().numbers.get("EQ", {}).get(str(code)) or {}
    if not one:
        return set(), set()
    return mtypes(f"{one.get('상세구분')} {one.get('이름')}"), specnums(one.get("규격"))


def times_of(row: dict[str, Any]) -> float:
    per = _PER.findall(str(row.get("결과단위") or ""))
    return float(per[0].replace(",", "")) if per and per[0] else 1.0


def _args(text: str | None) -> dict[str, str]:
    out = {}
    for part in re.split(r",(?![^()]*\))", text or ""):
        if "=" in part:
            k, v = part.split("=", 1)
            out[k.strip()] = v.strip()
    return out


def _labor_names() -> dict[str, str]:
    w = whole()
    if _STATE.get("LB판") is not w:
        lb = {str(r.get("키")): str(r.get("이름")) for r in w.numbers.get("LB", {}).values()}
        _STATE.update(LB판=w, LB=lb)
    return _STATE["LB"]


def _engine(code: str, given: dict[str, Any]) -> dict[str, Any]:
    """{L, X, M, S, mix} — 줄 수량 ÷ 결과단위 수(단위당)."""
    w = whole()
    row = w.logic(code)
    times = times_of(row)
    by_name: dict[Any, dict] = {}
    for h in row.get("호표") or []:
        by_name.setdefault(h.get("이름"), h)
    auto = {k: v for k, v in m01.mau.values(w, row).items() if v is not None}
    with _ENGINE, m01.cm.mf.mp.choosing(_NO_PRICE):  # 품만 씀 — 자재 값 찾기 건너뜀(수량 같음)
        got = m01.cm.mf.run(w, code, {**auto, **m01._dec(given)})
    mid, lb = got.get("중간") or {}, _labor_names()
    L: dict[str, float] = {}
    X: list[list] = []
    M: dict[str, float] = {}
    S: dict[str, float] = {}
    for line in got.get("줄") or []:
        h = by_name.get(line.get("이름"))
        if h is None:
            continue
        try:
            q = float(Decimal(str(line.get("수량") or 0))) / times
        except ArithmeticError:
            continue
        kind = h.get("종류")
        if kind == "인력":
            t = lb.get(str(h.get("요소")), str(h.get("이름")))
            L[t] = L.get(t, 0.0) + q
        elif kind == "로직":
            m = _SUB.match(str(h.get("요소") or ""))
            sub = m.group(1) if m else ""
            sub_row = w.logic(sub) if sub else {}
            if sub_row and is_mach_logic(sub_row):
                ts = set(FIXED_MACH.get(sub, set())) | mtypes(h.get("이름"))
                sp = specnums(h.get("이름"))
                for v in _args(m.group(2)).values():
                    lit = v.strip("'\"")
                    if not EQ_CODE.match(lit):
                        lit = str(mid.get(lit) or given.get(lit) or auto.get(lit) or "")
                    if EQ_CODE.match(lit):
                        a, b = eq_of(lit)
                        ts |= a
                        sp |= b
                if q:
                    X.append([sorted(ts), sorted(sp), q])
            elif q:
                S[sub] = S.get(sub, 0.0) + q
        elif kind == "재료" and q:
            M[str(h.get("이름"))] = M.get(str(h.get("이름")), 0.0) + q
    mix = {}
    for k in ("인력비율", "기계비율"):
        try:
            if mid.get(k) not in (None, ""):
                mix[k] = float(Decimal(str(mid.get(k))))
        except ArithmeticError:
            pass
    return {"L": L, "X": X, "M": M, "S": S, "mix": mix}


def _mach_view(code: str, row: dict, given: dict, v: dict) -> dict:
    """「기계 시간당 기계경비」 = 고른 기계(EQ 칸)의 갈래 · 규격 1 시간(조종원은 기계경비 속)."""
    auto = m01.mau.values(whole(), row)
    X = []
    for s in row.get("입력") or []:
        opts = s.get("고르기") or []
        if opts and all(isinstance(o, str) and EQ_CODE.match(o) for o in opts):
            ts, sp = eq_of(given.get(s["이름"], auto.get(s["이름"])))
            X.append([sorted(ts | FIXED_MACH.get(code, set())), sorted(sp), 1.0])
    if not X and FIXED_MACH.get(code):
        X = [[sorted(FIXED_MACH[code]), [], 1.0]]
    return {**v, "L": {}, "X": X}


def run(code: str, given: dict[str, Any] | None = None) -> dict[str, Any] | None:
    """엔진 한 번 → {L, X, M, S, mix} · 못 돌면 None(메모리 캐시)."""
    given = dict(given or {})
    key = (code, json.dumps(given, sort_keys=True, default=str, ensure_ascii=False))
    if key not in _RUN:
        try:
            v = _engine(code, given)
            row = whole().logic(code)
            v = _mach_view(code, row, given, v) if is_mach_logic(row) else v
        except Exception:  # noqa: BLE001 — 못 도는 조합은 없음으로(엔진 예외가 여럿)
            v = None
        if len(_RUN) >= RUN_CAP:
            _RUN.clear()
        _RUN[key] = v
    return _RUN[key]


# ── 지문 ───────────────────────────────────────────────────────────────
def _static(code: str, row: dict[str, Any], unit: str) -> dict[str, Any]:
    picks = [s for s in row.get("입력") or [] if s.get("고르기")]
    labels, opt_mach = [], []
    for s in picks:
        names = s.get("고르기이름") or {}
        for v in s.get("고르기") or []:
            if isinstance(v, str) and EQ_CODE.match(v):
                ts, sp = eq_of(v)
                if ts:
                    opt_mach.append([sorted(ts), sorted(sp)])
            else:
                labels.append(str(names.get(v) or v))
    lb = _labor_names()
    trades, mach, mats = set(), [], []
    for h in row.get("호표") or []:
        kind = h.get("종류")
        if kind == "인력":
            trades.add(lb.get(str(h.get("요소")), str(h.get("이름"))))
        elif kind == "로직":
            m = _SUB.match(str(h.get("요소") or ""))
            ts = set(FIXED_MACH.get(m.group(1) if m else "", set())) | mtypes(h.get("이름"))
            for v in _args(m.group(2) if m else "").values():
                if EQ_CODE.match(v.strip("'\"")):
                    ts |= eq_of(v.strip("'\""))[0]
            if ts:
                mach.append([sorted(ts), sorted(specnums(h.get("이름")))])
        elif kind == "재료":
            el = h.get("요소")
            words = (
                (el.get("검색어") or el.get("이름") or el.get("상세구분"))
                if isinstance(el, dict)
                else ""
            )
            mats.append(f"{h.get('이름')} {words or ''}".strip())
    pct = [
        [str(d.get("이름")), float(v)]
        for d in row.get("덧줄") or []
        for v in _PCT.findall(str(d.get("식") or ""))
    ]
    if is_mach_logic(row):
        mach.append([sorted(FIXED_MACH.get(code, set())), []])
    pick_text = " ".join(labels + [str(s.get("이름")) for s in picks])
    return {
        "로직": code,
        "이름": str(row.get("이름") or ""),
        "단위": unit,
        "직종": sorted(trades),
        "장비": mach,
        "장비고름": opt_mach,
        "자재": mats,
        "덧줄율": pct,
        "이름조건": {k: sorted(v) for k, v in conds(row.get("이름")).items()},
        "고름조건": {k: sorted(v) for k, v in conds(pick_text).items()},
        "기계경비": is_mach_logic(row),
    }


def combos(row: dict[str, Any]) -> list[dict[str, Any]]:
    """변형 입력 — 고르기 곱 64 이하 모두 · 넘으면 하나씩(기본 + 칸마다)."""
    picks = []
    for s in row.get("입력") or []:
        opts = s.get("고르기") or []
        if not opts or any(w in str(s.get("이름")) for w in SKIP_IN):
            continue
        if any(isinstance(v, str) and EQ_CODE.match(v) for v in opts):
            continue
        picks.append((str(s["이름"]), list(opts)))
    if not picks:
        return [{}]
    if math.prod(len(o) for _, o in picks) <= 64:
        names = [n for n, _ in picks]
        return [dict(zip(names, vals)) for vals in itertools.product(*[o for _, o in picks])]
    return [{}] + [{n: v} for n, opts in picks for v in opts]


def _sig(v: dict) -> str:
    r = lambda x: float(f"{x:.4g}")  # noqa: E731
    return json.dumps(
        [
            sorted((k, r(q)) for k, q in v["L"].items()),
            sorted((tuple(a), tuple(b), r(q)) for a, b, q in v["X"]),
            sorted((k, r(q)) for k, q in v["M"].items()),
        ],
        ensure_ascii=False,
    )


def fingerprint(code: str, unit: str) -> dict[str, Any]:
    """로직 하나 → 정적 지문 + 변형(서로 다른 품 꼴만)."""
    row = whole().logic(code)
    fp = _static(code, row, unit)
    seen, variants, errors = set(), [], 0
    for given in combos(row) if not fp["기계경비"] else [{}]:
        try:
            v = _engine(code, given)  # 기계경비 로직도 엔진 그대로(맞춤이 고르기 장비로 봄)
        except Exception:  # noqa: BLE001 — 못 도는 조합은 셈만
            errors += 1
            continue
        if (k := _sig(v)) not in seen:
            seen.add(k)
            variants.append({"in": {n: str(x) for n, x in given.items()}, **v})
    return {**fp, "갈래수": len(variants), "오류": errors, "변형": variants}


# ── 표 · 캐시 · 데움 ───────────────────────────────────────────────────
def _read(ver: str) -> list[dict] | None:
    try:
        doc = json.loads(path_of(ver).read_bytes())
    except (OSError, ValueError):
        return None
    ok = isinstance(doc, dict) and doc.get("판") == ver and doc.get("지문판") == FP_VERSION
    return doc.get("표") if ok else None


def build(ver: str | None = None) -> list[dict]:
    """1,438 돈 로직 지문을 셈 → 디스크 캐시(같은 판이면 덮음)."""
    ver = ver or version()
    table = [fingerprint(one["로직"], one["단위"]) for one in hopyo_match.targets()["로직"]]
    atomic_write_json(path_of(ver), {"판": ver, "지문판": FP_VERSION, "표": table})
    with _LOCK:
        _STATE.update(표=table, 표판=ver)
    return table


def table() -> list[dict] | None:
    """우리 지문(판이 같은 것) · 준비 전 None — 디스크에 있으면 읽어 둠."""
    ver = version()
    with _LOCK:
        if _STATE.get("표판") != ver:
            _STATE.update(표=_read(ver), 표판=ver)
        return _STATE["표"]


def ready() -> bool:
    return table() is not None


def warm(build_missing: bool = False) -> threading.Thread | None:
    """디스크 캐시 읽기(뒤) · `build_missing` 이면 없을 때 뒤 스레드로 셈 — 셈 중이면 그 스레드."""
    running = _STATE.get("스레드")
    if running is not None and running.is_alive():
        return running

    def work() -> None:
        try:
            if table() is None and build_missing:
                build(version())
        except Exception:  # noqa: BLE001 — 마스터가 깨져도 서버를 막지 않음(보기에서 「준비 중」)
            pass

    thread = threading.Thread(target=work, name="m02-hopyo-ours", daemon=True)
    _STATE["스레드"] = thread
    thread.start()
    return thread