Aislo/M02_MasterTemplete/M02_Hopyo_Store.py

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

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

from __future__ import annotations

import hashlib
import json
import re
import threading
from datetime import datetime, timedelta, timezone
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 = "격자 없음 — 다시 올리면 보임"
RELOAD = "읽기 꼴이 바뀐 옛 불러오기 — 엑셀을 다시 올림"
EXPIRE = timedelta(days=1)  # 작업 중 HP 보관 — 올린 뒤 하루
BOOK = 3  # 읽기 꼴(Book) 판 — 읽기 · 정규화 출력 꼴을 바꾸면 올림
_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 stale(doc: dict[str, Any], now: datetime | None = None) -> bool:
    """하루 지난 HP · 다른 판 HP — 올린때 기준 · 못 읽는 때 = 지난 것."""
    if doc.get("판") != BOOK:
        return True
    try:
        when = datetime.fromisoformat(str(doc.get("올린때") or "").replace("Z", "+00:00"))
    except ValueError:
        return True
    return (now or datetime.now(timezone.utc)) - when > EXPIRE


def rows(folder: str | Path) -> list[dict[str, Any]]:
    """층 실무자료 목록(키 차례) — 작업 중(지문 있음 · 하루 안)만 · 옛 HP · 지난 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("지문") and not stale(doc):
            out.append({**brief(doc), "지문": doc["지문"]})
    return out


def done(folder: str | Path) -> dict[str, Any]:
    """키 번 기록 `{번, 지문}`(지문 = 743 전 끝낸 기록 · 안 읽음 · 정리 `Forget.tidy`)."""
    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, old: bool = False) -> dict[str, Any]:
    """HP — 다른 판 = 409 「다시 올림」(`old` = 끝내기 · 정리만 옛 판도 읽음)."""
    doc = _load(_path(folder, key))
    if not isinstance(doc, dict):
        raise store.StoreError(404, f"없는 실무자료 「{key}」")
    if not old and doc.get("판") != BOOK:
        raise store.StoreError(409, RELOAD)
    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,
            "판": BOOK,
        }
        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) -> None:
    """HP · 격자 지움(끝낸 기록 없음 · 같은 엑셀 다시 올림 됨 · 743-1)."""
    path = _path(folder, key)
    with _LOCK:
        if not isinstance(_load(path), dict):
            raise store.StoreError(404, f"없는 실무자료 「{key}」")
        path.unlink()
        _grid_path(folder, key).unlink(missing_ok=True)
        (_dir(folder) / "산근" / f"{key}.json").unlink(missing_ok=True)  # 793-3b 산근 시트


# ── 맞춤 기억 ─────────────────────────────────────────────────────────
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