Trixy7/plugins/tv_program/epg/cache.py

# -*- 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),
        }