"""파생 저장본 — 입력 지문 열쇠 + 결과 한 쌍(343 · 341 부록 4 2-1). 자리 = 프로젝트 `_derived/<종류>/<열쇠>.json` · 작업본에서 다시 셀 수 있는 값만(정본 아님) · 통째로 지워도 다시 셈으로 돌아옴 · 종류마다 최근 `KEEP` 벌(꺼내면 최근으로). 결과 옆 열쇠(`<이름>.key.json` — 결과가 곧 산출물인 도엽 3D 면 등)는 `key_matches` · `file_stamp`. """ from __future__ import annotations import hashlib import json from functools import lru_cache from pathlib import Path from typing import Any from common_util.common_util_json import atomic_write_json DERIVED_DIR = "_derived" KEEP = 2 def derived_key(*parts: Any) -> str: """입력 지문 — 작은 입력은 내용 그대로 · 파일은 `file_stamp` 를 넣는다.""" body = json.dumps(parts, sort_keys=True, ensure_ascii=False, default=str) return hashlib.sha1(body.encode("utf-8")).hexdigest() def file_stamp(path: Path | None) -> str | None: """파일 지문 = 수정시각 ns : 크기. 없으면 None.""" try: stat = path.stat() if path is not None else None except OSError: return None return f"{stat.st_mtime_ns}:{stat.st_size}" if stat else None def file_digest(path: Path | None) -> str | None: """파일 내용 지문(blake2b) — 옮겨 심기 · 동기화 · 다시 짓기로 시각만 바뀌면 같은 값(763-2). 한 프로세스 안에서는 경로 · 수정시각 · 크기로 기억(다시 안 읽음). 없으면 None.""" try: stat = path.stat() if path is not None else None except OSError: return None if stat is None: return None return _digest(str(Path(path).resolve()), stat.st_mtime_ns, stat.st_size) # ponytail: 지형 파일 수백 개 전제 · 프로젝트가 크게 늘면 maxsize 를 셈. @lru_cache(maxsize=512) def _digest(path: str, _mtime_ns: int, _size: int) -> str: digest = hashlib.blake2b(digest_size=16) with open(path, "rb") as handle: for block in iter(lambda: handle.read(1 << 20), b""): digest.update(block) return digest.hexdigest() def key_matches(key_path: Path, key: Any) -> bool: """결과 옆 열쇠 파일이 지금 입력과 같은지.""" try: return json.loads(key_path.read_text(encoding="utf-8")) == key except (OSError, ValueError): return False def derived_dir(project_root: Path, kind: str) -> Path: return Path(project_root) / DERIVED_DIR / kind def read_derived(project_root: Path, kind: str, key: str) -> dict[str, Any] | None: """저장본 `{"value": …}` — 없거나 못 읽으면 None(값이 null 이어도 맞음과 구별).""" path = derived_dir(project_root, kind) / f"{key}.json" try: data = json.loads(path.read_text(encoding="utf-8")) except (OSError, ValueError): return None if not isinstance(data, dict) or data.get("key") != key or "value" not in data: return None try: path.touch() # 최근에 쓴 것 = 정리 때 남김 except OSError: pass return data def write_derived(project_root: Path, kind: str, key: str, value: Any, keep: int = KEEP) -> None: """원자 쓰기 뒤 같은 종류 옛 열쇠를 정리한다.""" atomic_write_json(derived_dir(project_root, kind) / f"{key}.json", {"key": key, "value": value}) prune_derived(project_root, kind, keep) def prune_derived(project_root: Path, kind: str, keep: int = KEEP) -> int: """종류 하나를 최근 `keep` 벌만 남기고 지운다. 지운 수.""" folder = derived_dir(project_root, kind) files = sorted( ((path.stat().st_mtime_ns, path) for path in folder.glob("*.json")), reverse=True ) for _, path in files[keep:]: path.unlink(missing_ok=True) return max(0, len(files) - keep)