You cannot select more than 25 topics
Topics must start with a letter or number, can include dashes ('-') and can be up to 35 characters long.
78 lines
2.4 KiB
Python
78 lines
2.4 KiB
Python
"""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")
|