# -*- coding: utf-8 -*- """ EPG-Cache mit In-Memory-Indizes. Verwaltet Sendungsdaten als JSON-Datei und bietet schnelle Abfragen ueber vorberechnete Indizes (nach Sender, Titel, Serien-ID, Tags). """ from __future__ import annotations import bisect import json import re from datetime import date, datetime, timedelta from pathlib import Path from trixy_core.utils.debug import pinfo, pdebug, perror, pwarn # Artikel, die bei der Titel-Normalisierung entfernt werden _STRIP_ARTICLES = {"der", "die", "das", "ein", "eine", "the", "a", "an"} # Regex fuer Serien-ID-Generierung: nur alphanumerisch + Bindestriche _SLUG_RE = re.compile(r"[^a-z0-9]+") class EPGCache: """ Cache fuer EPG-Sendungsdaten mit JSON-Persistenz und In-Memory-Indizes. Unterstuetzt Abfragen nach Sender, Titel, Serien-ID, Tags und Genre. Indizes werden beim Laden oder nach Aenderungen automatisch aufgebaut. """ FILE_NAME = "epg.json" def __init__(self, cache_dir: Path) -> None: """ Args: cache_dir: Verzeichnis fuer die Cache-Datei. """ self._cache_dir = cache_dir self._file_path = cache_dir / self.FILE_NAME # Hauptdaten: {program_id: program_dict} self._programs: dict[str, dict] = {} # Enrichment-Daten: {series_id: enrichment_dict} self._enrichments: dict[str, dict] = {} # Erinnerungen: [{title, sid, active, satellite_id, ...}] self._reminders: list[dict] = [] # Indizes (werden bei _rebuild_indices aufgebaut) self._by_channel: dict[str, list[dict]] = {} self._by_title: dict[str, list[dict]] = {} self._by_series_id: dict[str, list[dict]] = {} self._by_tag: dict[str, list[dict]] = {} # Cache-Verzeichnis erstellen self._cache_dir.mkdir(parents=True, exist_ok=True) # --- JSON I/O --- def load(self) -> bool: """ Laedt den Cache aus der JSON-Datei und baut die Indizes auf. Toleriert fehlende oder beschaedigte Dateien. Returns: True, wenn erfolgreich geladen, False bei Fehler oder fehlender Datei. """ if not self._file_path.exists(): pdebug("EPGCache: Keine Cache-Datei vorhanden, starte leer") return False try: raw = self._file_path.read_text(encoding="utf-8") data = json.loads(raw) self._programs = data.get("programs", {}) self._enrichments = data.get("enrichments", {}) self._reminders = data.get("reminders", []) self._rebuild_indices() pinfo( f"EPGCache: {len(self._programs)} Sendungen, " f"{len(self._enrichments)} Enrichments geladen" ) return True except (json.JSONDecodeError, KeyError, TypeError) as e: perror(f"EPGCache: Fehler beim Laden der Cache-Datei: {e}") self._programs = {} self._enrichments = {} self._reminders = [] return False def save(self) -> None: """ Speichert den Cache atomar in die JSON-Datei. Schreibt zuerst in eine temporaere Datei und benennt dann um, um Datenverlust bei Absturz zu vermeiden. """ self._save_sync() async def save_async(self) -> None: """Async-Speicherung — blockiert den Event-Loop nicht.""" import asyncio await asyncio.to_thread(self._save_sync) def _save_sync(self) -> None: """Synchrone Speicherung (kann auch aus Thread aufgerufen werden).""" data = { "programs": self._programs, "enrichments": self._enrichments, "reminders": self._reminders, } tmp_path = self._file_path.with_suffix(".tmp") try: tmp_path.write_text( json.dumps(data, ensure_ascii=False, separators=(",", ":")), encoding="utf-8", ) tmp_path.replace(self._file_path) pdebug(f"EPGCache: {len(self._programs)} Sendungen gespeichert") except OSError as e: perror(f"EPGCache: Fehler beim Speichern: {e}") # --- Index-Verwaltung --- def _rebuild_indices(self) -> None: """ Baut alle Indizes aus den aktuellen Programmdaten neu auf. Indizes: - _by_channel: Sendungen pro Sender, sortiert nach Startzeit - _by_title: Sendungen pro normalisiertem Titel - _by_series_id: Sendungen pro Serien-ID - _by_tag: Sendungen pro Tag (aus Enrichment-Daten) """ self._by_channel = {} self._by_title = {} self._by_series_id = {} self._by_tag = {} for prog_id, program in self._programs.items(): # Sender-Index channel = program.get("ch", "") if channel: self._by_channel.setdefault(channel, []).append(program) # Titel-Index (normalisiert) title = program.get("t", "") if title: norm_title = self._normalize_title(title) self._by_title.setdefault(norm_title, []).append(program) # Serien-ID-Index series_id = self._generate_series_id(title) if series_id: self._by_series_id.setdefault(series_id, []).append(program) # Sender-Listen nach Startzeit sortieren (fuer binaere Suche) for channel in self._by_channel: self._by_channel[channel].sort(key=lambda p: p.get("s", "")) # Tag-Index aus Enrichment-Daten aufbauen for series_id, enrichment in self._enrichments.items(): tags = enrichment.get("tags", []) programs = self._by_series_id.get(series_id, []) for tag in tags: tag_lower = tag.lower() self._by_tag.setdefault(tag_lower, []).extend(programs) pdebug( f"EPGCache: Indizes aufgebaut - " f"{len(self._by_channel)} Sender, " f"{len(self._by_title)} Titel, " f"{len(self._by_series_id)} Serien, " f"{len(self._by_tag)} Tags" ) # --- Daten-Mutation --- def update_programs(self, programs: list[dict]) -> int: """ Fuegt neue Sendungen hinzu oder aktualisiert bestehende. Args: programs: Liste von Programm-Dicts. Returns: Anzahl der neu hinzugefuegten Sendungen. """ new_count = 0 for program in programs: prog_id = self._make_program_id(program) if prog_id not in self._programs: new_count += 1 self._programs[prog_id] = program if new_count > 0: self._rebuild_indices() pdebug(f"EPGCache: {new_count} neue Sendungen hinzugefuegt") return new_count def update_enrichment(self, series_id: str, data: dict) -> None: """ Aktualisiert Enrichment-Daten fuer eine Serie. Args: series_id: Die Serien-ID. data: Enrichment-Dict (z.B. mit Tags, Beschreibung, Streaming-Info). """ self._enrichments[series_id] = data # Tag-Index neu aufbauen, da sich Tags geaendert haben koennten self._rebuild_tag_index() pdebug(f"EPGCache: Enrichment fuer '{series_id}' aktualisiert") def prune_old(self, retention_days: int = 14) -> int: """ Entfernt Sendungen, die aelter als retention_days sind. Args: retention_days: Anzahl Tage, die behalten werden. Returns: Anzahl der entfernten Sendungen. """ cutoff = datetime.now().astimezone() - timedelta(days=retention_days) cutoff_iso = cutoff.isoformat() to_remove = [ prog_id for prog_id, program in self._programs.items() if program.get("e", program.get("s", "")) < cutoff_iso ] for prog_id in to_remove: del self._programs[prog_id] if to_remove: self._rebuild_indices() pinfo(f"EPGCache: {len(to_remove)} alte Sendungen entfernt") return len(to_remove) # --- Abfragen --- def get_program_at(self, channel_id: str, time: datetime) -> dict | None: """ Findet die Sendung, die zu einem bestimmten Zeitpunkt auf einem Sender laeuft. Nutzt binaere Suche auf dem Sender-Index. Args: channel_id: Interne Sender-ID. time: Zeitpunkt. Returns: Programm-Dict oder None. """ programs = self._by_channel.get(channel_id, []) if not programs: return None time_iso = time.isoformat() # Binaere Suche: finde Position, wo time einsortiert wuerde idx = bisect.bisect_right( programs, time_iso, key=lambda p: p.get("s", "") ) # Die Sendung vor dieser Position koennte laufen if idx > 0: candidate = programs[idx - 1] start = candidate.get("s", "") end = candidate.get("e", "") # Pruefen, ob die Sendung noch laeuft if start <= time_iso and (not end or end > time_iso): return candidate return None def get_programs_on_channel(self, channel_id: str, target_date: date) -> list[dict]: """ Gibt alle Sendungen eines Senders an einem bestimmten Tag zurueck. Args: channel_id: Interne Sender-ID. target_date: Datum. Returns: Liste der Sendungen, sortiert nach Startzeit. """ programs = self._by_channel.get(channel_id, []) if not programs: return [] # Tagesgrenzen als ISO-Strings (naive Filterung) day_start = datetime(target_date.year, target_date.month, target_date.day).isoformat() day_end = datetime( target_date.year, target_date.month, target_date.day, 23, 59, 59 ).isoformat() # Binaere Suche fuer den Tagesanfang start_idx = bisect.bisect_left( programs, day_start, key=lambda p: p.get("s", "") ) result: list[dict] = [] for i in range(start_idx, len(programs)): prog = programs[i] prog_start = prog.get("s", "") if prog_start > day_end: break result.append(prog) return result def get_primetime(self, channel_id: str, target_date: date) -> dict | None: """ Gibt die Primetime-Sendung (20:15) eines Senders zurueck. Args: channel_id: Interne Sender-ID. target_date: Datum. Returns: Programm-Dict oder None. """ primetime = datetime( target_date.year, target_date.month, target_date.day, 20, 15 ).astimezone() return self.get_program_at(channel_id, primetime) def search_title(self, query: str, limit: int = 10) -> list[dict]: """ Sucht Sendungen nach Titel (unscharfe Suche). Zukuenftige Sendungen werden bevorzugt zurueckgegeben. Args: query: Suchbegriff. limit: Maximale Anzahl Ergebnisse. Returns: Liste passender Sendungen. """ norm_query = self._normalize_title(query) if not norm_query: return [] now_iso = datetime.now().astimezone().isoformat() matches: list[dict] = [] for norm_title, programs in self._by_title.items(): if norm_query in norm_title or norm_title in norm_query: matches.extend(programs) # Zukuenftige Sendungen zuerst, dann nach Startzeit sortieren matches.sort( key=lambda p: ( 0 if p.get("s", "") >= now_iso else 1, p.get("s", ""), ) ) return matches[:limit] def search_by_tag(self, tag: str) -> list[dict]: """ Sucht Sendungen nach Tag (aus Enrichment-Daten). Args: tag: Der gesuchte Tag. Returns: Liste passender Sendungen. """ return list(self._by_tag.get(tag.lower(), [])) def search_by_genre(self, genre: str) -> list[dict]: """ Sucht Sendungen nach Genre. Args: genre: Der gesuchte Genre-String. Returns: Liste passender Sendungen. """ genre_lower = genre.lower() results: list[dict] = [] for program in self._programs.values(): categories = program.get("g", []) if any(genre_lower in cat.lower() for cat in categories): results.append(program) # Nach Startzeit sortieren results.sort(key=lambda p: p.get("s", "")) return results def get_next_airing(self, series_id: str) -> dict | None: """ Findet die naechste Ausstrahlung einer Serie. Args: series_id: Die Serien-ID. Returns: Programm-Dict oder None. """ programs = self._by_series_id.get(series_id, []) if not programs: return None now_iso = datetime.now().astimezone().isoformat() for program in sorted(programs, key=lambda p: p.get("s", "")): if program.get("s", "") >= now_iso: return program return None def get_all_airings(self, series_id: str) -> list[dict]: """ Gibt alle Ausstrahlungen einer Serie zurueck. Args: series_id: Die Serien-ID. Returns: Liste aller Ausstrahlungen, sortiert nach Startzeit. """ programs = self._by_series_id.get(series_id, []) return sorted(programs, key=lambda p: p.get("s", "")) def get_streaming_info(self, series_id: str) -> list[str]: """ Gibt Streaming-Anbieter fuer eine Serie zurueck. Args: series_id: Die Serien-ID. Returns: Liste der Streaming-Anbieter. """ enrichment = self._enrichments.get(series_id) if enrichment: return enrichment.get("streaming", []) return [] def get_enrichment(self, series_id: str) -> dict | None: """ Gibt die Enrichment-Daten fuer eine Serie zurueck. Args: series_id: Die Serien-ID. Returns: Enrichment-Dict oder None. """ return self._enrichments.get(series_id) def get_similar_shows(self, series_id: str) -> list[str]: """ Gibt aehnliche Sendungen basierend auf Enrichment-Daten zurueck. Args: series_id: Die Serien-ID. Returns: Liste von aehnlichen Serien-IDs. """ enrichment = self._enrichments.get(series_id) if enrichment: return enrichment.get("similar", []) return [] # --- Hilfsfunktionen --- def _rebuild_tag_index(self) -> None: """Baut nur den Tag-Index neu auf (nach Enrichment-Aenderungen).""" self._by_tag = {} for series_id, enrichment in self._enrichments.items(): tags = enrichment.get("tags", []) programs = self._by_series_id.get(series_id, []) for tag in tags: tag_lower = tag.lower() self._by_tag.setdefault(tag_lower, []).extend(programs) @staticmethod def _make_program_id(program: dict) -> str: """Generiert eine eindeutige ID fuer eine Sendung (MD5 aus ch+st).""" import hashlib key = f"{program.get('ch', '')}_{program.get('s', '')}" return hashlib.md5(key.encode()).hexdigest()[:12] @staticmethod def _normalize_title(title: str) -> str: """ Normalisiert einen Sendungstitel fuer die Suche. Konvertiert zu Kleinbuchstaben und entfernt fuehrende Artikel. Args: title: Der Originaltitel. Returns: Normalisierter Titel. """ normalized = title.strip().lower() # Fuehrende Artikel entfernen parts = normalized.split() if parts and parts[0] in _STRIP_ARTICLES: normalized = " ".join(parts[1:]) return normalized @staticmethod def _generate_series_id(title: str) -> str: """ Generiert eine Serien-ID aus dem Titel. Slug-Format: Kleinbuchstaben, nicht-alphanumerische Zeichen durch Bindestriche ersetzt. Args: title: Der Sendungstitel. Returns: Serien-ID als Slug. """ if not title: return "" slug = _SLUG_RE.sub("-", title.strip().lower()) # Fuehrende und abschliessende Bindestriche entfernen return slug.strip("-") def get_all_programs(self) -> list[dict]: """Gibt alle Programme als Liste zurueck.""" return list(self._programs.values()) def get_program_by_series_id(self, series_id: str) -> dict | None: """Gibt das erste Programm fuer eine Serien-ID zurueck.""" programs = self._by_series_id.get(series_id, []) return programs[0] if programs else None @property def reminders(self) -> list[dict]: """Gibt die Erinnerungsliste zurueck (direkte Referenz fuer Modifikation).""" return self._reminders def get_stats(self) -> dict: """ Gibt Statistiken ueber den Cache-Inhalt zurueck. Returns: Dict mit Anzahlen fuer Programme, Sender, Enrichments, etc. """ return { "program_count": len(self._programs), "channel_count": len(self._by_channel), "enriched_count": len(self._enrichments), "reminder_count": len(self._reminders), "title_count": len(self._by_title), "series_count": len(self._by_series_id), "tag_count": len(self._by_tag), }