Trixy7/plugins/tv_program/epg/collector.py

# -*- coding: utf-8 -*-
"""
EPG-Daten-Collector.

Laedt XMLTV-Daten von konfigurierten Quellen via aiohttp herunter,
cached die Rohdaten und dedupliziert die geparsten Sendungen.
"""

from __future__ import annotations

import asyncio
import gzip
import hashlib
from pathlib import Path

from trixy_core.utils.debug import pinfo, pdebug, perror, pwarn

from plugins.tv_program.epg.channels import ChannelMapper
from plugins.tv_program.epg.xmltv_parser import XMLTVParser

# Optionale aiohttp-Abhaengigkeit
_AIOHTTP_AVAILABLE = False
try:
    import aiohttp

    _AIOHTTP_AVAILABLE = True
except ImportError:
    pass

# Download-Timeout in Sekunden
_FETCH_TIMEOUT = 60


class EPGCollector:
    """
    Sammelt EPG-Daten von konfigurierten XMLTV-Quellen.

    Laedt XMLTV-Dateien asynchron herunter, parst sie und
    dedupliziert die Ergebnisse. Rohdaten werden lokal gecached.
    """

    def __init__(
        self,
        sources: list[dict],
        cache_dir: Path,
        channel_mapper: ChannelMapper,
    ) -> None:
        """
        Args:
            sources: Liste von Quellen-Dicts mit mindestens "url" und "name".
                     Optional "enabled" (default True).
            cache_dir: Verzeichnis fuer den Download-Cache.
            channel_mapper: Mapper zur XMLTV-ID-Aufloesung.
        """
        self._sources = sources
        self._cache_dir = cache_dir
        self._raw_dir = cache_dir / "raw"
        self._mapper = channel_mapper
        self._parser = XMLTVParser(channel_mapper)

        # Cache-Verzeichnisse erstellen
        self._raw_dir.mkdir(parents=True, exist_ok=True)

    async def fetch_all(self) -> list[dict]:
        """
        Laedt alle aktivierten Quellen herunter und gibt deduplizierte
        Programm-Dicts zurueck.

        Returns:
            Zusammengefuehrte, deduplizierte Liste aller Sendungen.
        """
        if not _AIOHTTP_AVAILABLE:
            perror("EPGCollector: aiohttp nicht installiert, Download nicht moeglich")
            return []

        all_programs: list[dict] = []
        enabled_sources = [
            s for s in self._sources if s.get("enabled", True)
        ]

        if not enabled_sources:
            pwarn("EPGCollector: Keine aktiven EPG-Quellen konfiguriert")
            return []

        pinfo(f"EPGCollector: Lade {len(enabled_sources)} EPG-Quelle(n)...")

        for source in enabled_sources:
            name = source.get("name", "unbekannt")
            try:
                raw_data = await self._fetch_source(source)
                if raw_data:
                    # XML-Parsen in Thread auslagern (CPU-intensiv, ~500ms fuer 11 MB)
                    # damit der Event-Loop nicht blockiert wird
                    programs = await asyncio.to_thread(self._parser.parse_bytes, raw_data)
                    pinfo(f"EPGCollector: {len(programs)} Sendungen von '{name}' geparst")
                    all_programs.extend(programs)
                else:
                    pwarn(f"EPGCollector: Keine Daten von '{name}' erhalten")
            except Exception as e:
                perror(f"EPGCollector: Fehler bei Quelle '{name}': {e}")

        # Deduplizieren
        deduplicated = self._deduplicate(all_programs)
        pinfo(
            f"EPGCollector: {len(deduplicated)} Sendungen nach Deduplizierung "
            f"(von {len(all_programs)} gesamt)"
        )

        return deduplicated

    async def _fetch_source(self, source: dict) -> bytes:
        """
        Laedt eine einzelne XMLTV-Quelle herunter.

        Speichert die Rohdaten zusaetzlich im Cache-Verzeichnis
        fuer Debugging-Zwecke.

        Args:
            source: Quellen-Dict mit "url" und "name".

        Returns:
            Rohe XML-Bytes der heruntergeladenen Datei.

        Raises:
            aiohttp.ClientError: Bei Netzwerkfehlern.
            asyncio.TimeoutError: Bei Timeout.
        """
        url = source["url"]
        name = source.get("name", "unbekannt")

        pdebug(f"EPGCollector: Lade '{name}' von {url}")

        timeout = aiohttp.ClientTimeout(total=_FETCH_TIMEOUT)
        async with aiohttp.ClientSession(timeout=timeout) as session:
            async with session.get(url) as response:
                response.raise_for_status()
                data = await response.read()

        # gzip-komprimierte Dateien entpacken
        if url.endswith(".gz"):
            try:
                data = gzip.decompress(data)
                pdebug(f"EPGCollector: gzip entpackt: {len(data)} Bytes")
            except (gzip.BadGzipFile, OSError) as e:
                perror(f"EPGCollector: gzip-Entpacken fehlgeschlagen: {e}")
                return b""

        # Rohdaten im Cache speichern
        cache_file = self._raw_dir / f"{name}.xml"
        try:
            cache_file.write_bytes(data)
            pdebug(f"EPGCollector: Rohdaten gespeichert: {cache_file}")
        except OSError as e:
            pwarn(f"EPGCollector: Konnte Rohdaten nicht cachen: {e}")

        pdebug(f"EPGCollector: {len(data)} Bytes von '{name}' heruntergeladen")
        return data

    @staticmethod
    def _deduplicate(programs: list[dict]) -> list[dict]:
        """
        Entfernt Duplikate basierend auf Programm-ID (Hash aus Sender + Startzeit).

        Bei Duplikaten wird der Eintrag mit mehr Informationen bevorzugt.

        Args:
            programs: Liste aller geparsten Sendungen.

        Returns:
            Deduplizierte Liste.
        """
        seen: dict[str, dict] = {}

        for program in programs:
            prog_id = EPGCollector._generate_id(program)

            if prog_id not in seen:
                seen[prog_id] = program
            else:
                # Eintrag mit mehr Feldern bevorzugen
                existing = seen[prog_id]
                if len(program) > len(existing):
                    seen[prog_id] = program

        return list(seen.values())

    @staticmethod
    def _generate_id(program: dict) -> str:
        """
        Generiert eine eindeutige ID fuer eine Sendung.

        Basiert auf MD5-Hash von Sender-ID und Startzeit.

        Args:
            program: Programm-Dict mit "ch" und "st" Feldern.

        Returns:
            12-stelliger Hex-Hash als Programm-ID.
        """
        key = f"{program.get('ch', '')}_{program.get('s', '')}"
        return hashlib.md5(key.encode()).hexdigest()[:12]