# -*- 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")))