"""Local exhibit text cache with atomic write and checksum.""" from __future__ import annotations import json from pathlib import Path from libs.common.config import get_settings from libs.common.ids import sha256_checksum def _exhibit_dir(accession_no: str) -> Path: settings = get_settings() return Path(settings.exhibit_cache_dir) / accession_no def exhibit_path(accession_no: str, exhibit_type: str) -> Path: """Return canonical path for cached exhibit text.""" safe_type = exhibit_type.replace("/", "_").replace(" ", "_") return _exhibit_dir(accession_no) / f"{safe_type}.txt" def sidecar_path(accession_no: str, exhibit_type: str) -> Path: p = exhibit_path(accession_no, exhibit_type) return p.with_suffix(".meta.json") def write_exhibit(accession_no: str, exhibit_type: str, content: str) -> str: """Atomically write exhibit text; return sha256 checksum. Raises FileExistsError if file already exists (idempotency guard). Use exists_exhibit() to check first. """ path = exhibit_path(accession_no, exhibit_type) path.parent.mkdir(parents=True, exist_ok=True) checksum = sha256_checksum(content.encode("utf-8")) # Atomic write via temp file tmp = path.with_suffix(".tmp") try: tmp.write_text(content, encoding="utf-8") tmp.rename(path) except Exception: tmp.unlink(missing_ok=True) raise # Write sidecar metadata meta = { "accession_no": accession_no, "exhibit_type": exhibit_type, "content_hash": checksum, "size_bytes": len(content.encode("utf-8")), } sidecar_path(accession_no, exhibit_type).write_text( json.dumps(meta, indent=2), encoding="utf-8" ) return checksum def read_exhibit(accession_no: str, exhibit_type: str) -> str: """Read cached exhibit text. Raises FileNotFoundError if missing.""" path = exhibit_path(accession_no, exhibit_type) return path.read_text(encoding="utf-8") def exists_exhibit(accession_no: str, exhibit_type: str) -> bool: return exhibit_path(accession_no, exhibit_type).exists() def get_checksum(accession_no: str, exhibit_type: str) -> str | None: """Return stored checksum from sidecar, or None if missing.""" sp = sidecar_path(accession_no, exhibit_type) if not sp.exists(): return None meta = json.loads(sp.read_text(encoding="utf-8")) return meta.get("content_hash")