Upload captcha_solver/utils/cache.py with huggingface_hub

captcha_solver/utils/cache.py

@@ -0,0 +1,79 @@
+"""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()