"""LRU+TTL cache for captcha answers, keyed by perceptual hash.""" from __future__ import annotations import time from collections import OrderedDict from dataclasses import dataclass from threading import Lock from typing import Optional @dataclass class _Entry: answer: str solver: str confidence: float expires_at: float hits: int = 0 class SolveCache: """Thread-safe in-memory LRU cache with TTL. Keys are perceptual hashes (hex strings). Values are the solved answer plus the solver that produced it. Entries expire after `ttl_seconds` and are evicted LRU when over capacity. """ def __init__(self, ttl_seconds: int = 3600, max_entries: int = 10_000): self.ttl = ttl_seconds self.max = max_entries self._store: OrderedDict[str, _Entry] = OrderedDict() self._lock = Lock() self.hits = 0 self.misses = 0 def get(self, key: str) -> Optional[_Entry]: now = time.time() with self._lock: entry = self._store.get(key) if entry is None: self.misses += 1 return None if entry.expires_at < now: del self._store[key] self.misses += 1 return None entry.hits += 1 self.hits += 1 self._store.move_to_end(key) return entry def set(self, key: str, answer: str, solver: str, confidence: float) -> None: now = time.time() with self._lock: self._store[key] = _Entry( answer=answer, solver=solver, confidence=confidence, expires_at=now + self.ttl, ) self._store.move_to_end(key) while len(self._store) > self.max: self._store.popitem(last=False) def stats(self) -> dict: with self._lock: total = self.hits + self.misses return { "size": len(self._store), "max": self.max, "hits": self.hits, "misses": self.misses, "hit_rate": (self.hits / total) if total else 0.0, } def clear(self) -> None: with self._lock: self._store.clear()