Aislo/M02_MasterTemplete/M02_Hopyo_Store.py

"""M02 일위대가 실무자료 저장(PLAN 324-1) — 층 폴더 `실무자료/<키>.json` ·
맞춤 기억 `실무자료/맞춤.json` · 엑셀꼴 격자 `실무자료/격자/<키>.json`(329-1 · 파일과 같이 지움).

파일 = 읽은 결과만(원본 엑셀 안 둠) `{키, 파일, 올린때, 올린이, 시트, 호표, 검산}` · 키 `HP-######`
(층마다 다음 번호) · 맞춤 기억 = `{줄: {줄 열쇠: {로직?|자재?|빼기?, 입력값, 환산, 확정}}}`.
층 종류(`layers.KINDS`) 밖 폴더 — 양식 목록 · 복사 · 프로젝트 씨앗에 안 섞임 · 권한은 길 몫.
한시 불러오기(718-1) — 파일에 `지문`(올린 엑셀 sha256) · 목록 = 지문 있는 것(작업 중)만 · 옛 HP 는
파일 그대로 목록에서만 숨김 · 끝낸 기록 `실무자료/불러옴.json`
`{번, 지문 {지문: {파일, 키, 끝낸때, 만든}}}` · 번 = 낸 키 가장 큰 번(지운 뒤에도 키를 다시 안 씀).
"""

from __future__ import annotations

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

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

DIRNAME = "실무자료"
MEMORY = "맞춤"
GRID = "격자"
DONE = "불러옴"
NO_GRID = "격자 없음 — 다시 올리면 보임"
_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 _grid_path(folder: str | Path, key: str) -> Path:
    return _dir(folder) / GRID / _path(folder, key).name


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 fingerprint(data: bytes) -> str:
    """올린 엑셀 지문 — 같은 파일 다시 올림을 알아봄."""
    return hashlib.sha256(data).hexdigest()


def rows(folder: str | Path) -> list[dict[str, Any]]:
    """층 실무자료 목록(키 차례) — 작업 중(지문 있음)만 · 옛 HP 는 숨김."""
    out = []
    for path in sorted(_dir(folder).glob("HP-*.json")):
        doc = _load(path)
        if isinstance(doc, dict) and _KEY.match(path.stem) and doc.get("지문"):
            out.append({**brief(doc), "지문": doc["지문"]})
    return out


def done(folder: str | Path) -> dict[str, Any]:
    """끝낸 기록 `{번, 지문 {지문: {파일, 키, 끝낸때, 만든}}}`."""
    doc = _load(_dir(folder) / f"{DONE}.json")
    doc = doc if isinstance(doc, dict) else {}
    marks = doc.get("지문")
    return {"번": int(doc.get("번") or 0), "지문": marks if isinstance(marks, dict) else {}}


def _put_done(folder: str | Path, got: dict[str, Any]) -> None:
    atomic_write_json(_dir(folder) / f"{DONE}.json", got)


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,
    grid: dict[str, Any] | None = None,
    mark: str | None = None,
) -> dict[str, Any]:
    """읽은 결과(+ 격자) → 새 키로 씀 → 목록 한 줄 · `mark` = 올린 엑셀 지문."""
    with _LOCK:
        have = [
            int(m.group(1)) for p in _dir(folder).glob("HP-*.json") if (m := _KEY.match(p.stem))
        ]
        log = done(folder)
        log["번"] = max(*have, log["번"], 0) + 1
        key = f"HP-{log['번']:06d}"
        doc = {
            "키": key,
            "파일": name,
            "올린때": utc_now_text(),
            "올린이": who,
            **({"지문": mark} if mark else {}),
            **parsed,
        }
        if grid is not None:
            book = {**grid["책"], "열": key}
            atomic_write_json(_grid_path(folder, key), {**grid, "책": book})
        atomic_write_json(_path(folder, key), doc)
        _put_done(folder, log)
    return {**brief(doc), **({"지문": mark} if mark else {})}


def grid(folder: str | Path, key: str) -> dict[str, Any]:
    """엑셀꼴 격자 `{책, 호표, 단산}` — 없으면(옛 파일) 404."""
    doc = _load(_grid_path(folder, key))
    if not isinstance(doc, dict):
        raise store.StoreError(404, NO_GRID)
    return doc


def delete(folder: str | Path, key: str, made: list[str] | None = None) -> None:
    """HP · 격자 지움 · 만든 것(`made` SC 키)이 있으면 지문을 끝낸 기록에 남김(다시 올림 막음)."""
    path = _path(folder, key)
    with _LOCK:
        doc = _load(path)
        if not isinstance(doc, dict):
            raise store.StoreError(404, f"없는 실무자료 「{key}」")
        if made and doc.get("지문"):
            log = done(folder)
            log["지문"][doc["지문"]] = {
                "파일": doc.get("파일"),
                "키": key,
                "끝낸때": utc_now_text(),
                "만든": made,
            }
            _put_done(folder, log)
        path.unlink()
        _grid_path(folder, key).unlink(missing_ok=True)


# ── 맞춤 기억 ─────────────────────────────────────────────────────────
def memory(folder: str | Path, name: str = MEMORY) -> dict[str, dict[str, Any]]:
    """층 맞춤 기억 `{줄 열쇠: 확정}` — 없으면 빈 것 ·
    `name` = 기억 파일(내역 잇기 707-4 = 「내역맞춤」)."""
    doc = _load(_dir(folder) / f"{name}.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], name: str = MEMORY
) -> dict[str, Any]:
    """기억 더함 · 바꿈(값 None = 지움) → 새 기억 전부."""
    with _LOCK:
        got = memory(folder, name)
        for key, value in entries.items():
            if value is None:
                got.pop(key, None)
            else:
                got[key] = value
        atomic_write_json(_dir(folder) / f"{name}.json", {"줄": got})
    return got