Aislo/M02_MasterTemplete/M02_Hopyo_Store.py

"""M02 일위대가 실무자료 저장(PLAN 324-1) — 층 폴더 `실무자료/<키>.json` ·
맞춤 기억 `실무자료/맞춤.json`.

파일 = 읽은 결과만(원본 엑셀 안 둠) `{키, 파일, 올린때, 올린이, 시트, 호표, 검산}` · 키 `HP-######`
(층마다 다음 번호) · 맞춤 기억 = `{줄: {줄 열쇠: {로직?|자재?|빼기?, 입력값, 환산, 확정}}}`.
층 종류(`layers.KINDS`) 밖 폴더 — 양식 목록 · 복사 · 프로젝트 씨앗에 안 섞임 · 권한은 길 몫.
"""

from __future__ import annotations

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

from common_util.common_util_json import atomic_write_json
from M02_MasterTemplete import M02_MasterTemplete_Store as store

DIRNAME = "실무자료"
MEMORY = "맞춤"
_KEY = re.compile(r"^HP-(\d{6})$")
# ponytail: 프로세스 안 잠금 — 서버를 여럿 띄우면 파일 잠금으로
_LOCK = threading.Lock()


def _dir(folder: str | Path) -> Path:
    return Path(folder) / DIRNAME


def _path(folder: str | Path, key: str) -> Path:
    if not _KEY.match(str(key)):
        raise store.StoreError(404, f"없는 실무자료 「{key}」")
    return _dir(folder) / f"{key}.json"


def _load(path: Path) -> Any:
    try:
        return json.loads(path.read_bytes())
    except (OSError, ValueError):
        return None


def brief(doc: dict[str, Any]) -> dict[str, Any]:
    """목록 한 줄 — `{키, 파일, 올린때, 올린이, 시트, 호표수, 어긋남}`."""
    keep = ("키", "파일", "올린때", "올린이", "시트")
    return {
        **{k: doc.get(k) for k in keep},
        "호표수": len(doc.get("호표") or []),
        "어긋남": len((doc.get("검산") or {}).get("어긋남") or []),
    }


def rows(folder: str | Path) -> list[dict[str, Any]]:
    """층 실무자료 목록(키 차례)."""
    out = []
    for path in sorted(_dir(folder).glob("HP-*.json")):
        doc = _load(path)
        if isinstance(doc, dict) and _KEY.match(path.stem):
            out.append(brief(doc))
    return out


def read(folder: str | Path, key: str) -> dict[str, Any]:
    doc = _load(_path(folder, key))
    if not isinstance(doc, dict):
        raise store.StoreError(404, f"없는 실무자료 「{key}」")
    return doc


def save(folder: str | Path, name: str, parsed: dict[str, Any], who: Any) -> dict[str, Any]:
    """읽은 결과 → 새 키로 씀 → 목록 한 줄."""
    with _LOCK:
        have = [
            int(m.group(1)) for p in _dir(folder).glob("HP-*.json") if (m := _KEY.match(p.stem))
        ]
        key = f"HP-{max(have, default=0) + 1:06d}"
        doc = {
            "키": key,
            "파일": name,
            "올린때": datetime.now().isoformat(timespec="seconds"),
            "올린이": who,
            **parsed,
        }
        atomic_write_json(_path(folder, key), doc)
    return brief(doc)


def delete(folder: str | Path, key: str) -> None:
    path = _path(folder, key)
    with _LOCK:
        if not path.is_file():
            raise store.StoreError(404, f"없는 실무자료 「{key}」")
        path.unlink()


# ── 맞춤 기억 ─────────────────────────────────────────────────────────
def memory(folder: str | Path) -> dict[str, dict[str, Any]]:
    """층 맞춤 기억 `{줄 열쇠: 확정}` — 없으면 빈 것."""
    doc = _load(_dir(folder) / f"{MEMORY}.json")
    got = doc.get("줄") if isinstance(doc, dict) else None
    return got if isinstance(got, dict) else {}


def remember(folder: str | Path, entries: dict[str, dict[str, Any] | None]) -> dict[str, Any]:
    """기억 더함 · 바꿈(값 None = 지움) → 새 기억 전부."""
    with _LOCK:
        got = memory(folder)
        for key, value in entries.items():
            if value is None:
                got.pop(key, None)
            else:
                got[key] = value
        atomic_write_json(_dir(folder) / f"{MEMORY}.json", {"줄": got})
    return got