Trixy7/trixy_core/online/cache.py

# -*- coding: utf-8 -*-
"""
Offline-Response-Cache.

Speichert HTTP-Responses auf Festplatte, damit sie bei Offline-Zustand
als Fallback zurueckgegeben werden koennen.

Cache-Schluessel: SHA256(method + url + body + sorted_params)
Speicherort: {cache_dir}/online_cache/{hash}.json
"""

from __future__ import annotations

import hashlib
import json
import os
import time
from dataclasses import dataclass, field
from pathlib import Path
from typing import Any

from trixy_core.utils.debug import pdebug


@dataclass
class CacheEntry:
    """Ein Cache-Eintrag."""
    url: str
    method: str
    status_code: int
    headers: dict[str, str]
    body: str | bytes
    cached_at: float          # time.time()
    ttl_seconds: float = 0    # 0 = kein Ablauf
    is_binary: bool = False

    @property
    def age_seconds(self) -> float:
        """Alter des Eintrags in Sekunden."""
        return time.time() - self.cached_at

    @property
    def is_expired(self) -> bool:
        """Prueft ob der Eintrag abgelaufen ist."""
        if self.ttl_seconds <= 0:
            return False
        return self.age_seconds > self.ttl_seconds

    def to_dict(self) -> dict[str, Any]:
        """Serialisiert den Eintrag."""
        body = self.body
        if isinstance(body, bytes):
            body = body.hex()
        return {
            "url": self.url,
            "method": self.method,
            "status_code": self.status_code,
            "headers": self.headers,
            "body": body,
            "cached_at": self.cached_at,
            "ttl_seconds": self.ttl_seconds,
            "is_binary": self.is_binary,
        }

    @classmethod
    def from_dict(cls, data: dict[str, Any]) -> CacheEntry:
        """Deserialisiert einen Eintrag."""
        body = data.get("body", "")
        is_binary = data.get("is_binary", False)
        if is_binary and isinstance(body, str):
            body = bytes.fromhex(body)
        return cls(
            url=data.get("url", ""),
            method=data.get("method", "GET"),
            status_code=data.get("status_code", 200),
            headers=data.get("headers", {}),
            body=body,
            cached_at=data.get("cached_at", 0),
            ttl_seconds=data.get("ttl_seconds", 0),
            is_binary=is_binary,
        )


class ResponseCache:
    """
    Persistenter HTTP-Response-Cache fuer Offline-Betrieb.

    Speichert Responses als JSON-Dateien auf der Festplatte.
    Bei erneutem Request im Offline-Modus wird der Cache zurueckgegeben.

    Beispiel:
        ```python
        cache = ResponseCache("/home/pi/trixy/source/cache")

        # Beim Speichern (nach erfolgreichem Request)
        cache.put("https://api.weather.com/...", "GET", response_body, headers)

        # Beim Abrufen (wenn offline)
        entry = cache.get("https://api.weather.com/...", "GET")
        if entry:
            return entry.body  # Gecachte Wetterdaten
        ```
    """

    def __init__(self, cache_dir: str) -> None:
        self._cache_dir = Path(cache_dir) / "online_cache"
        self._cache_dir.mkdir(parents=True, exist_ok=True)

    @staticmethod
    def _make_key(
        url: str,
        method: str = "GET",
        body: str | bytes = "",
        params: dict[str, str] | None = None,
    ) -> str:
        """Erzeugt einen Cache-Schluessel."""
        parts = [method.upper(), url]
        if body:
            parts.append(str(body))
        if params:
            parts.append(str(sorted(params.items())))
        raw = "|".join(parts)
        return hashlib.sha256(raw.encode()).hexdigest()

    def _entry_path(self, key: str) -> Path:
        """Pfad zur Cache-Datei."""
        return self._cache_dir / f"{key}.json"

    def get(
        self,
        url: str,
        method: str = "GET",
        body: str | bytes = "",
        params: dict[str, str] | None = None,
        ignore_ttl: bool = False,
    ) -> CacheEntry | None:
        """
        Liest einen Cache-Eintrag.

        Args:
            url: Request-URL
            method: HTTP-Methode
            body: Request-Body (fuer POST)
            params: Query-Parameter
            ignore_ttl: TTL ignorieren (z.B. im Offline-Modus)

        Returns:
            CacheEntry oder None wenn nicht gefunden/abgelaufen
        """
        key = self._make_key(url, method, body, params)
        path = self._entry_path(key)

        if not path.is_file():
            return None

        try:
            with open(path) as f:
                data = json.load(f)
            entry = CacheEntry.from_dict(data)

            # TTL pruefen (ausser ignore_ttl ist gesetzt)
            if not ignore_ttl and entry.is_expired:
                pdebug(f"Cache abgelaufen: {url} (Alter: {entry.age_seconds:.0f}s)")
                return None

            pdebug(f"Cache-Treffer: {url} (Alter: {entry.age_seconds:.0f}s)")
            return entry

        except (json.JSONDecodeError, OSError, KeyError):
            return None

    def put(
        self,
        url: str,
        method: str,
        status_code: int,
        headers: dict[str, str],
        body: str | bytes,
        params: dict[str, str] | None = None,
        ttl_seconds: float = 0,
    ) -> None:
        """
        Speichert eine Response im Cache.

        Args:
            url: Request-URL
            method: HTTP-Methode
            status_code: HTTP Status-Code
            headers: Response-Headers
            body: Response-Body
            params: Query-Parameter (fuer Cache-Key)
            ttl_seconds: Ablaufzeit (0 = kein Ablauf)
        """
        key = self._make_key(url, method, "", params)
        entry = CacheEntry(
            url=url,
            method=method,
            status_code=status_code,
            headers=headers,
            body=body,
            cached_at=time.time(),
            ttl_seconds=ttl_seconds,
            is_binary=isinstance(body, bytes),
        )

        path = self._entry_path(key)
        try:
            with open(path, "w") as f:
                json.dump(entry.to_dict(), f, ensure_ascii=False)
        except OSError as e:
            pdebug(f"Cache-Schreibfehler: {e}")

    def remove(
        self,
        url: str,
        method: str = "GET",
        params: dict[str, str] | None = None,
    ) -> bool:
        """Entfernt einen Cache-Eintrag."""
        key = self._make_key(url, method, "", params)
        path = self._entry_path(key)
        if path.is_file():
            path.unlink()
            return True
        return False

    def clear(self) -> int:
        """Leert den gesamten Cache. Gibt Anzahl entfernter Eintraege zurueck."""
        count = 0
        for f in self._cache_dir.glob("*.json"):
            f.unlink()
            count += 1
        return count

    def clear_expired(self) -> int:
        """Entfernt abgelaufene Eintraege. Gibt Anzahl entfernter Eintraege zurueck."""
        count = 0
        for f in self._cache_dir.glob("*.json"):
            try:
                with open(f) as fh:
                    data = json.load(fh)
                entry = CacheEntry.from_dict(data)
                if entry.is_expired:
                    f.unlink()
                    count += 1
            except (json.JSONDecodeError, OSError):
                f.unlink()
                count += 1
        return count

    @property
    def size(self) -> int:
        """Anzahl der Cache-Eintraege."""
        return len(list(self._cache_dir.glob("*.json")))