File size: 2,412 Bytes
fafbad3 | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 | """Local file-based cache for abstracts."""
import hashlib
import json
from datetime import datetime, timedelta
from pathlib import Path
from .models import CacheEntry
class CacheManager:
"""Persists abstracts on disk with TTL expiry."""
def __init__(self, cache_dir: Path, enabled: bool = True, ttl_hours: int = 168):
self.cache_dir = Path(cache_dir)
self.enabled = enabled
self.ttl = timedelta(hours=ttl_hours)
self.hits = 0
self.misses = 0
if self.enabled:
self.cache_dir.mkdir(parents=True, exist_ok=True)
def get(self, key: str) -> str | None:
if not self.enabled:
return None
cache_path = self._cache_path(key)
if not cache_path.exists():
self.misses += 1
return None
with open(cache_path, encoding="utf-8") as fh:
entry = CacheEntry(**json.load(fh))
if entry.expires_at and datetime.now() > entry.expires_at:
cache_path.unlink()
self.misses += 1
return None
entry.access_count += 1
entry.last_accessed = datetime.now()
self._write_entry(cache_path, entry)
self.hits += 1
return entry.value
def set(self, key: str, value: str) -> None:
if not self.enabled:
return
entry = CacheEntry(
key=key,
value=value,
expires_at=datetime.now() + self.ttl if self.ttl else None,
)
self._write_entry(self._cache_path(key), entry)
def clear(self) -> int:
if not self.cache_dir.exists():
return 0
count = sum(1 for f in self.cache_dir.glob("*.json") if f.unlink() is None)
self.hits = self.misses = 0
return count
def get_stats(self) -> dict:
total = self.hits + self.misses
return {
"enabled": self.enabled,
"hits": self.hits,
"misses": self.misses,
"hit_rate": self.hits / total if total else 0,
"cache_dir": str(self.cache_dir),
}
def _cache_path(self, key: str) -> Path:
return self.cache_dir / f"{hashlib.md5(key.encode()).hexdigest()}.json"
def _write_entry(self, path: Path, entry: CacheEntry) -> None:
with open(path, "w", encoding="utf-8") as fh:
json.dump(entry.model_dump(), fh, default=str, indent=2)
|