"""Filesystem cache for rerank results, keyed by SHA256(query + candidate_ids).""" from __future__ import annotations import hashlib import json import time from pathlib import Path from typing import Any, Dict, List, Optional DEFAULT_CACHE_DIR = Path.home() / ".cache" / "notion-search" DEFAULT_TTL_SECONDS = 86400 # 24 hours def _cache_key(query: str, candidate_ids: List[str]) -> str: payload = f"{query}|{','.join(sorted(candidate_ids))}" return hashlib.sha256(payload.encode("utf-8")).hexdigest() def cache_get( query: str, candidate_ids: List[str], *, cache_dir: Path = DEFAULT_CACHE_DIR, ttl_seconds: int = DEFAULT_TTL_SECONDS, ) -> Optional[List[Dict[str, Any]]]: """Return cached results if file exists and is fresh, else None. Corrupted cache files are silently treated as misses (and removed). `ttl_seconds <= 0` always returns None (deterministic "always expired"). """ if ttl_seconds <= 0: return None key = _cache_key(query, candidate_ids) path = cache_dir / f"{key}.json" if not path.exists(): return None try: data = json.loads(path.read_text(encoding="utf-8")) except (json.JSONDecodeError, OSError): try: path.unlink() except OSError: pass return None cached_at = data.get("cached_at_epoch", 0) if time.time() - cached_at > ttl_seconds: return None return data.get("results") def cache_put( query: str, candidate_ids: List[str], results: List[Dict[str, Any]], *, cache_dir: Path = DEFAULT_CACHE_DIR, ) -> None: """Write results to cache. Creates cache_dir if missing.""" cache_dir.mkdir(parents=True, exist_ok=True) key = _cache_key(query, candidate_ids) path = cache_dir / f"{key}.json" payload = { "query": query, "candidate_ids": list(candidate_ids), "results": results, "cached_at_epoch": time.time(), } path.write_text(json.dumps(payload, ensure_ascii=False, indent=2), encoding="utf-8")