captcha_solver/utils/cache.py

"""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()