Trixy7/plugins/tv_program/epg/xmltv_parser.py

# -*- coding: utf-8 -*-
"""
XMLTV-Parser fuer EPG-Daten.

Parst XMLTV-XML-Dateien speichereffizient via iterparse und wandelt
Sendungsdaten in interne Programm-Dicts um.
"""

from __future__ import annotations

import io
import re
from datetime import datetime, timedelta, timezone
from pathlib import Path
from xml.etree.ElementTree import iterparse

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

from plugins.tv_program.epg.channels import ChannelMapper


# Regex fuer XMLTV-Zeitformat: "YYYYMMDDHHmmss +HHMM"
_XMLTV_TIME_RE = re.compile(
    r"(\d{4})(\d{2})(\d{2})(\d{2})(\d{2})(\d{2})\s*([+-]\d{4})"
)

# Genre-Kategorien fuer Typ-Erkennung
_FILM_GENRES = {"spielfilm", "film", "kinofilm", "fernsehfilm", "tv-film", "movie"}
_SERIE_GENRES = {"serie", "series", "soap", "telenovela", "sitcom", "dramedy"}
_SHOW_GENRES = {"show", "unterhaltung", "entertainment", "gameshow", "talkshow", "reality"}
_DOKU_GENRES = {"dokumentation", "doku", "documentary", "reportage", "wissen"}
_SPORT_GENRES = {"sport", "fussball", "football", "tennis", "formel 1", "motorsport"}
_NEWS_GENRES = {"nachrichten", "news", "magazin", "journal"}
_KIDS_GENRES = {"kinder", "zeichentrick", "animation", "anime", "children"}


class XMLTVParser:
    """
    Parser fuer XMLTV-Daten (Standard-EPG-Format).

    Nutzt iterparse fuer speichereffizientes Parsen grosser XML-Dateien.
    Konvertiert nur Sendungen von Sendern, die im ChannelMapper registriert sind.
    """

    def __init__(self, channel_mapper: ChannelMapper) -> None:
        """
        Args:
            channel_mapper: Mapper zur Aufloesung von XMLTV-Sender-IDs.
        """
        self._mapper = channel_mapper
        # Mapping: XMLTV channel-id (z.B. "76740") → interner channel_id (z.B. "rtl")
        # Wird beim Parsen aus <channel> Elementen aufgebaut
        self._xmltv_id_map: dict[str, str] = {}

    def parse_file(self, path: Path) -> list[dict]:
        """
        Parst eine XMLTV-Datei und gibt eine Liste von Programm-Dicts zurueck.

        Args:
            path: Pfad zur XMLTV-XML-Datei.

        Returns:
            Liste der geparsten Sendungen.
        """
        if not path.exists():
            perror(f"XMLTVParser: Datei nicht gefunden: {path}")
            return []

        pdebug(f"XMLTVParser: Parse Datei {path}")
        programs: list[dict] = []

        self._xmltv_id_map = {}
        try:
            for event, elem in iterparse(str(path), events=("end",)):
                if elem.tag == "channel":
                    self._map_channel_element(elem)
                    elem.clear()
                elif elem.tag == "programme":
                    program = self._parse_programme(elem)
                    if program is not None:
                        programs.append(program)
                    elem.clear()
        except Exception as e:
            perror(f"XMLTVParser: Fehler beim Parsen von {path}: {e}")

        pdebug(f"XMLTVParser: {len(programs)} Sendungen aus {path.name} geparst")
        return programs

    def parse_bytes(self, data: bytes) -> list[dict]:
        """
        Parst XMLTV-Daten aus einem Byte-Buffer.

        Args:
            data: Rohe XML-Bytes.

        Returns:
            Liste der geparsten Sendungen.
        """
        programs: list[dict] = []
        self._xmltv_id_map = {}

        try:
            stream = io.BytesIO(data)
            for event, elem in iterparse(stream, events=("end",)):
                if elem.tag == "channel":
                    # Channel-Element: XMLTV-ID → Display-Name → interner channel_id
                    self._map_channel_element(elem)
                    elem.clear()
                elif elem.tag == "programme":
                    program = self._parse_programme(elem)
                    if program is not None:
                        programs.append(program)
                    elem.clear()
        except Exception as e:
            perror(f"XMLTVParser: Fehler beim Parsen der Byte-Daten: {e}")

        pdebug(
            f"XMLTVParser: {len(programs)} Sendungen aus Byte-Daten geparst "
            f"({len(self._xmltv_id_map)} Sender zugeordnet)"
        )
        return programs

    def _map_channel_element(self, elem) -> None:
        """
        Mappt ein <channel>-Element auf einen internen Sender.

        XMLTV-Quellen nutzen unterschiedliche ID-Formate:
        - Domain-basiert: "DasErste.de", "RTL.de"
        - Numerisch: "76740", "76762"

        In beiden Faellen versuchen wir den Display-Name
        ueber den ChannelMapper aufzuloesen.
        """
        xmltv_id = elem.get("id", "")
        if not xmltv_id:
            return

        # Zuerst direkte XMLTV-ID-Aufloesung versuchen (Domain-Format)
        channel_id = self._mapper.resolve_xmltv_id(xmltv_id)
        if channel_id:
            self._xmltv_id_map[xmltv_id] = channel_id
            return

        # Fallback: Display-Name ueber resolve() matchen
        display_name_elem = elem.find("display-name")
        if display_name_elem is not None and display_name_elem.text:
            display_name = display_name_elem.text.strip()
            channel_id = self._mapper.resolve(display_name)
            if channel_id:
                self._xmltv_id_map[xmltv_id] = channel_id
                pdebug(f"XMLTVParser: '{display_name}' (ID={xmltv_id}) → {channel_id}")

    def _parse_programme(self, elem) -> dict | None:
        """
        Parst ein einzelnes <programme>-Element in ein Programm-Dict.

        Args:
            elem: Das XML-Element <programme>.

        Returns:
            Programm-Dict oder None, wenn der Sender nicht bekannt ist.
        """
        # XMLTV-Sender-ID pruefen (nutzt vorab aufgebaute Map)
        xmltv_channel = elem.get("channel", "")
        channel_id = self._xmltv_id_map.get(xmltv_channel)
        if channel_id is None:
            # Fallback: direkte XMLTV-ID-Aufloesung (Domain-Format)
            channel_id = self._mapper.resolve_xmltv_id(xmltv_channel)
        if channel_id is None:
            return None

        # Start- und Endzeit parsen
        start_str = elem.get("start", "")
        stop_str = elem.get("stop", "")

        if not start_str:
            return None

        try:
            start_iso = self._parse_xmltv_time(start_str)
        except ValueError:
            pwarn(f"XMLTVParser: Ungueltiges Startzeit-Format: {start_str}")
            return None

        stop_iso = ""
        if stop_str:
            try:
                stop_iso = self._parse_xmltv_time(stop_str)
            except ValueError:
                pass

        # Titel (Pflichtfeld)
        title_elem = elem.find("title")
        if title_elem is None or not title_elem.text:
            return None
        title = title_elem.text.strip()

        # Optionale Felder
        subtitle = ""
        sub_elem = elem.find("sub-title")
        if sub_elem is not None and sub_elem.text:
            subtitle = sub_elem.text.strip()

        description = ""
        desc_elem = elem.find("desc")
        if desc_elem is not None and desc_elem.text:
            description = desc_elem.text.strip()

        # Kategorien sammeln
        categories: list[str] = []
        for cat_elem in elem.findall("category"):
            if cat_elem.text:
                categories.append(cat_elem.text.strip())

        # Episoden-Informationen (xmltv_ns: 0-basiert!)
        season: int | None = None
        episode: int | None = None
        ep_num_elem = elem.find("episode-num")
        if ep_num_elem is not None and ep_num_elem.text:
            system = ep_num_elem.get("system", "")
            if system == "xmltv_ns":
                season, episode = self._parse_episode_ns(ep_num_elem.text)

        # Wiederholung oder Premiere
        is_premiere = elem.find("previously-shown") is None
        # Wenn <premiere> vorhanden, ist es definitiv eine Premiere
        if elem.find("premiere") is not None:
            is_premiere = True

        # Altersfreigabe
        rating = ""
        rating_elem = elem.find("rating")
        if rating_elem is not None:
            value_elem = rating_elem.find("value")
            if value_elem is not None and value_elem.text:
                rating = value_elem.text.strip()

        # Produktionsjahr
        year = ""
        date_elem = elem.find("date")
        if date_elem is not None and date_elem.text:
            year = date_elem.text.strip()

        # Typ anhand der Kategorien und des Titels ermitteln
        program_type = self._detect_type(categories, title)

        # Programm-Dict zusammenbauen
        program: dict = {
            "ch": channel_id,
            "t": title,
            "s": start_iso,
            "e": stop_iso,
            "pr": is_premiere,
            "ty": program_type,
        }

        # Optionale Felder nur setzen, wenn vorhanden
        if subtitle:
            program["et"] = subtitle
        if description:
            program["d"] = description
        if categories:
            program["g"] = categories
        if season is not None:
            program["sn"] = season
        if episode is not None:
            program["ep"] = episode
        if rating:
            program["rt"] = rating
        if year:
            program["yr"] = year

        # Sprach-Info: Deutsches TV ist fast immer synchronisiert
        # Wird spaeter durch Enrichment (AniList/Streaming) ergaenzt
        program["lang"] = "de"    # Audio-Sprache (de = deutsche Synchro)
        # "sub" wird nur gesetzt wenn bekannt (z.B. bei OmU-Sendungen)

        return program

    @staticmethod
    def _parse_xmltv_time(time_str: str) -> str:
        """
        Konvertiert XMLTV-Zeitformat in ISO 8601.

        XMLTV-Format: "YYYYMMDDHHmmss +HHMM"
        Ergebnis: ISO 8601 String mit Zeitzone, z.B. "2026-04-11T20:15:00+02:00"

        Args:
            time_str: Zeitstring im XMLTV-Format.

        Returns:
            ISO 8601 Zeitstring.

        Raises:
            ValueError: Bei ungueltigem Format.
        """
        match = _XMLTV_TIME_RE.match(time_str.strip())
        if not match:
            raise ValueError(f"Ungueltiges XMLTV-Zeitformat: {time_str}")

        year, month, day = int(match.group(1)), int(match.group(2)), int(match.group(3))
        hour, minute, second = int(match.group(4)), int(match.group(5)), int(match.group(6))
        tz_str = match.group(7)

        # Zeitzone parsen: "+0200" -> timedelta(hours=2)
        tz_sign = 1 if tz_str[0] == "+" else -1
        tz_hours = int(tz_str[1:3])
        tz_minutes = int(tz_str[3:5])
        tz_offset = timezone(timedelta(hours=tz_sign * tz_hours, minutes=tz_sign * tz_minutes))

        dt = datetime(year, month, day, hour, minute, second, tzinfo=tz_offset)
        return dt.isoformat()

    @staticmethod
    def _detect_type(categories: list[str], title: str) -> str:
        """
        Erkennt den Sendungstyp anhand der Kategorien und des Titels.

        Args:
            categories: Liste der XMLTV-Kategorien.
            title: Titel der Sendung.

        Returns:
            Typ-String: "film", "serie", "show", "doku", "sport",
                         "news", "kids" oder "other".
        """
        # Kategorien normalisieren
        normalized_cats = {cat.lower() for cat in categories}

        # Priorisierte Erkennung
        if normalized_cats & _FILM_GENRES:
            return "film"
        if normalized_cats & _SERIE_GENRES:
            return "serie"
        if normalized_cats & _KIDS_GENRES:
            return "kids"
        if normalized_cats & _SPORT_GENRES:
            return "sport"
        if normalized_cats & _DOKU_GENRES:
            return "doku"
        if normalized_cats & _NEWS_GENRES:
            return "news"
        if normalized_cats & _SHOW_GENRES:
            return "show"

        # Heuristik: Anime oft als Kategorie erwaehnt
        if "anime" in normalized_cats:
            return "kids"

        return "other"

    @staticmethod
    def _parse_episode_ns(text: str) -> tuple[int | None, int | None]:
        """
        Parst xmltv_ns Episodennummerierung.

        Format: "season.episode." (0-basiert) -> +1 fuer menschliche Nummerierung.

        Args:
            text: Der xmltv_ns-String, z.B. "2.15."

        Returns:
            Tuple (season, episode), jeweils 1-basiert oder None.
        """
        parts = text.strip().split(".")
        season: int | None = None
        episode: int | None = None

        if len(parts) >= 1 and parts[0].strip():
            try:
                season = int(parts[0].strip()) + 1
            except ValueError:
                pass

        if len(parts) >= 2 and parts[1].strip():
            try:
                episode = int(parts[1].strip()) + 1
            except ValueError:
                pass

        return season, episode