pipeline/cache.py

"""Tiny keyed cache under DATA_DIR for JSON-serializable pipeline artifacts.

Used to avoid re-running search / ranking / extraction for identical inputs. Image
and docx binaries are written directly under OUT_DIR by their own modules; this
handles the lightweight JSON steps.
"""
from __future__ import annotations

import hashlib
import json
from typing import Any, Optional

from . import config


def run_key(*parts: str) -> str:
    """Stable short hash identifying a run from its inputs."""
    joined = " ".join(p or "" for p in parts)
    h = hashlib.sha256(joined.encode("utf-8"))
    return h.hexdigest()[:16]


def _path(key: str, name: str):
    return config.CACHE_DIR / f"{key}.{name}.json"


def get(key: str, name: str) -> Optional[Any]:
    p = _path(key, name)
    if p.exists():
        try:
            return json.loads(p.read_text(encoding="utf-8"))
        except Exception:
            return None
    return None


def put(key: str, name: str, value: Any) -> None:
    p = _path(key, name)
    try:
        p.write_text(json.dumps(value, ensure_ascii=False, indent=2), encoding="utf-8")
    except Exception:
        # Cache is best-effort; never fail the pipeline over it.
        pass