src/data/cache_manager.py

"""Simple disk-based caching with TTL support."""

import hashlib
import json
import logging
import os
import pickle
import time
from typing import Any, Optional

logger = logging.getLogger(__name__)


class CacheManager:
    """File-based cache with TTL."""

    def __init__(self, cache_dir: str = "data/cache", default_ttl: int = 3600):
        self.cache_dir = cache_dir
        self.default_ttl = default_ttl
        os.makedirs(cache_dir, exist_ok=True)

    def _key_to_path(self, key: str) -> str:
        hashed = hashlib.md5(key.encode()).hexdigest()
        return os.path.join(self.cache_dir, f"{hashed}.pkl")

    def get(self, key: str, ttl: Optional[int] = None) -> Optional[Any]:
        """Get cached value if it exists and hasn't expired."""
        path = self._key_to_path(key)
        if not os.path.exists(path):
            return None

        ttl = ttl or self.default_ttl
        age = time.time() - os.path.getmtime(path)
        if age > ttl:
            os.remove(path)
            return None

        try:
            with open(path, "rb") as f:
                return pickle.load(f)
        except Exception:
            return None

    def set(self, key: str, value: Any) -> None:
        """Cache a value to disk."""
        path = self._key_to_path(key)
        try:
            with open(path, "wb") as f:
                pickle.dump(value, f)
        except Exception as e:
            logger.error(f"Cache write failed for {key}: {e}")

    def clear(self) -> None:
        """Clear all cached data."""
        for f in os.listdir(self.cache_dir):
            path = os.path.join(self.cache_dir, f)
            if f.endswith(".pkl"):
                os.remove(path)