# -*- coding: utf-8 -*- """ TV-Programm Plugin fuer Trixy. Beantwortet Fragen zum TV-Programm per Sprache: - "Was kommt um 20:15 auf RTL2?" - "Wann kommt die naechste Folge Naruto?" - "Gibt es diese Woche Ecchi-Animes?" - "Wo kann ich Naruto streamen?" Datenquellen: - EPG: XMLTV via iptv-org/epg (hoerzu.de) - Anime-Tags: AniList GraphQL (kostenlos, kein Key) - Filme/Serien: TMDB API (kostenloser Key) - Fallback: TVDB API (kostenloser Key) Sync laeuft im Hintergrund, resume-faehig bei Neustarts. """ from __future__ import annotations import asyncio import re from datetime import datetime, date, timedelta from pathlib import Path from typing import TYPE_CHECKING from trixy_core.plugins import TrixyPlugin from trixy_core.nlp.intent_registry import IntentRegistry from trixy_core.nlp.decorators import intent, pattern, example from trixy_core.nlp.handler import IntentReceivedData, IntentResult from trixy_core.utils.debug import pinfo, pdebug, perror, pwarn from plugins.tv_program.epg.channels import ChannelMapper from plugins.tv_program.epg.collector import EPGCollector from plugins.tv_program.epg.cache import EPGCache from plugins.tv_program.sync.sync_state import SyncState from plugins.tv_program.enrichment.enricher import Enricher class TVProgramPlugin(TrixyPlugin): """TV-Programm Abfragen per Sprache.""" NAME = "tv_program" VERSION = "1.1.0" DESCRIPTION = "TV-Programm, Anime-Tags, Streaming-Infos per Sprache" AUTHOR = "Trixy" def __init__(self, application, plugin_path, config=None) -> None: super().__init__(application, plugin_path, config) self._registry = IntentRegistry.get_instance() self._cache: EPGCache | None = None self._channel_mapper: ChannelMapper | None = None self._sync_state: SyncState | None = None self._enricher: Enricher | None = None self._sync_task: asyncio.Task | None = None self._reminder_task: asyncio.Task | None = None self._epg_job_id: str = "" self._enrichment_job_id: str = "" self._stt_aliases: dict[str, str] = {} # Vosk-Fehlerkennungen korrigieren # ====================================================================== # Lifecycle # ====================================================================== async def on_load(self) -> None: """Initialisiert Cache, Channels und startet Background-Sync.""" pinfo(f"TVProgramPlugin: Lade v{self.VERSION}") cache_dir = Path(self.get_config_value("cache_dir", "cache/tv_program")) cache_dir.mkdir(parents=True, exist_ok=True) # Channel-Mapper (Whitelist hat Vorrang, Blacklist schliesst aus) whitelist = self.get_config_value("channels.whitelist", []) blacklist = self.get_config_value("channels.blacklist", []) self._channel_mapper = ChannelMapper( channel_filter=whitelist or None, blacklist=blacklist or None, custom_aliases=self.get_config_value("channels.custom_aliases", {}), ) # EPG-Cache laden (schnell, ~50ms fuer 3MB JSON) self._cache = EPGCache(cache_dir) loaded = self._cache.load() stats = self._cache.get_stats() if loaded: pinfo( f"TVProgramPlugin: Cache geladen — " f"{stats['program_count']} Sendungen, " f"{stats['channel_count']} Sender, " f"{stats['enriched_count']} angereichert" ) else: pdebug("TVProgramPlugin: Kein Cache vorhanden — erster Sync noetig") # Sync-State self._sync_state = SyncState(cache_dir) self._sync_state.load() # Enricher self._enricher = Enricher( config=self.get_config_value("enrichment", {}), cache=self._cache, sync_state=self._sync_state, ) # Initialen Sync starten (NICHT blockierend) self._sync_task = asyncio.create_task(self._initial_sync()) # Scheduler-Jobs registrieren self._register_scheduler_jobs() # Reminder-Task starten if self.get_config_value("reminders.enabled", True): self._reminder_task = asyncio.create_task(self._reminder_check_loop()) # STT-Aliases laden (Vosk-Fehlerkennungen fuer Anime/TV-Begriffe) self._load_stt_aliases() # Slot-Listen fuer Intent-Training registrieren self._register_slot_lists() pinfo(f"TVProgramPlugin: Geladen v{self.VERSION}") async def on_unload(self) -> None: """Stoppt Background-Tasks, Scheduler-Jobs und deregistriert Intents.""" for task in (self._sync_task, self._reminder_task): if task and not task.done(): task.cancel() try: await task except asyncio.CancelledError: pass # Scheduler-Jobs entfernen self._remove_scheduler_jobs() self._registry.unregister_plugin(self.NAME) pdebug("TVProgramPlugin: Entladen") # ====================================================================== # Background-Sync (Resume-Faehig) # ====================================================================== async def _initial_sync(self) -> None: """Initialer Sync beim Server-Start (einmalig, non-blocking).""" await asyncio.sleep(5) # Server fertig starten lassen try: await self._run_sync() except asyncio.CancelledError: raise except Exception as e: perror(f"TVProgramPlugin: Sync-Fehler: {e}") # ====================================================================== # Scheduler-Jobs # ====================================================================== def _register_scheduler_jobs(self) -> None: """ Registriert Scheduler-Jobs fuer EPG-Fetch und Enrichment. - EPG-Fetch: Taeglich um konfigurierte Uhrzeit (CronTrigger) - Enrichment: Stuendlich wenn Queue > 0 (IntervalTrigger) """ scheduler = self.application.services.get_service("SchedulerService") if not scheduler: pdebug("TVProgramPlugin: SchedulerService nicht verfuegbar — keine Jobs registriert") return try: from trixy_core.scheduler.trigger.cron import CronTrigger from trixy_core.scheduler.trigger.interval import IntervalTrigger from trixy_core.scheduler.action.callback import CallbackAction # Job 1: Taeglicher EPG-Fetch sync_hour = self.get_config_value("epg.sync_hour", 6) sync_minute = self.get_config_value("epg.sync_minute", 0) epg_job = scheduler.create_job( trigger=CronTrigger(f"{sync_minute} {sync_hour} * * *"), actions=CallbackAction(self._scheduled_epg_sync), job_id="tv_program_epg_sync", save=False, ) self._epg_job_id = epg_job.job_id pdebug( f"TVProgramPlugin: EPG-Sync Job registriert " f"(taeglich um {sync_hour:02d}:{sync_minute:02d})" ) # Job 2: Stuendliches Enrichment (fuer Rueckstand-Aufholung) enrichment_job = scheduler.create_job( trigger=IntervalTrigger(hours=1), actions=CallbackAction(self._scheduled_enrichment), job_id="tv_program_enrichment", save=False, ) self._enrichment_job_id = enrichment_job.job_id pdebug("TVProgramPlugin: Enrichment Job registriert (stuendlich)") except Exception as e: perror(f"TVProgramPlugin: Scheduler-Jobs konnten nicht registriert werden: {e}") def _remove_scheduler_jobs(self) -> None: """Entfernt die registrierten Scheduler-Jobs.""" scheduler = self.application.services.get_service("SchedulerService") if not scheduler: return for job_id in (self._epg_job_id, self._enrichment_job_id): if job_id: try: scheduler.remove_job(job_id) except Exception: pass async def _scheduled_epg_sync(self, *args, **kwargs) -> None: """Vom Scheduler aufgerufener EPG-Sync (taeglich).""" pinfo("TVProgramPlugin: Geplanter EPG-Sync gestartet") try: await self._run_sync() except Exception as e: perror(f"TVProgramPlugin: Geplanter Sync fehlgeschlagen: {e}") async def _scheduled_enrichment(self, *args, **kwargs) -> None: """ Vom Scheduler aufgerufenes Enrichment (stuendlich). Ueberspringt wenn Queue leer — kein unnuetiger Overhead. """ if not self._sync_state or not self._sync_state.needs_enrichment(): return # Queue leer — nichts zu tun queue_len = len(self._sync_state.enrichment_queue) pinfo(f"TVProgramPlugin: Geplantes Enrichment ({queue_len} ausstehend)") try: enriched = await self._enricher.enrich_pending() if enriched > 0: await self._cache.save_async() self._sync_state.mark_enrichment_done() remaining = len(self._sync_state.enrichment_queue) pinfo( f"TVProgramPlugin: Enrichment fertig — " f"{enriched} angereichert, {remaining} verbleibend" ) except Exception as e: perror(f"TVProgramPlugin: Geplantes Enrichment fehlgeschlagen: {e}") async def _run_sync(self) -> None: """ Vollstaendiger Sync-Zyklus mit Resume. 1. EPG-Daten frisch genug? → Sonst XMLTV downloaden + parsen 2. Enrichment-Queue abarbeiten (resume-faehig) """ # Schritt 1: EPG-Fetch noetig? if self._sync_state.needs_epg_fetch(): pinfo("TVProgramPlugin: Starte EPG-Sync...") collector = EPGCollector( sources=self.get_config_value("epg.sources", []), cache_dir=Path(self.get_config_value("cache_dir", "cache/tv_program")), channel_mapper=self._channel_mapper, ) programs = await collector.fetch_all() if programs: new_count = self._cache.update_programs(programs) pruned = self._cache.prune_old( self.get_config_value("epg.cleanup_after_days", 7) ) await self._cache.save_async() self._sync_state.mark_epg_fetched( retention_days=self.get_config_value("epg.retention_days", 14), min_interval_hours=self.get_config_value("epg.min_sync_interval_hours", 20), ) pinfo( f"TVProgramPlugin: EPG-Sync fertig — " f"{new_count} neu, {pruned} entfernt, " f"{len(programs)} gesamt" ) # Enrichment-Queue fuer neue Titel bauen if self.get_config_value("enrichment.enabled", True): self._enricher.build_enrichment_queue(programs) self._sync_state.save() else: pwarn("TVProgramPlugin: Keine EPG-Daten empfangen") else: pdebug("TVProgramPlugin: EPG-Daten noch aktuell, Sync uebersprungen") # Schritt 2: Initiales Enrichment (beim Start einmal, danach via Scheduler) if self._sync_state.needs_enrichment(): queue_len = len(self._sync_state.enrichment_queue) pinfo(f"TVProgramPlugin: Starte initiales Enrichment ({queue_len} ausstehend)...") enriched = await self._enricher.enrich_pending() if enriched > 0: await self._cache.save_async() self._sync_state.mark_enrichment_done() remaining = len(self._sync_state.enrichment_queue) pinfo( f"TVProgramPlugin: Enrichment fertig — " f"{enriched} angereichert, {remaining} verbleibend" ) if remaining > 0: pinfo( f"TVProgramPlugin: Restliche {remaining} Titel werden " f"stuendlich via Scheduler nachgeladen" ) # ====================================================================== # STT-Korrektur (Plugin-spezifisch) # ====================================================================== def _load_stt_aliases(self) -> None: """ Laedt STT-Aliases aus stt_aliases.json. Diese Korrekturen werden NUR auf Slots in TV-Intents angewendet, nicht global auf alle STT-Ergebnisse. So wird "nach utopie" nur in TV-Kontexten zu "naruto" korrigiert. """ import json as _json alias_file = Path(self.plugin_path) / "stt_aliases.json" if not alias_file.is_file(): return try: with open(alias_file, encoding="utf-8") as f: data = _json.load(f) for key, value in data.items(): if not key.startswith("_"): # _comment ueberspringen self._stt_aliases[key.lower()] = value if self._stt_aliases: pdebug(f"TVProgramPlugin: {len(self._stt_aliases)} STT-Aliases geladen") except Exception as e: pdebug(f"TVProgramPlugin: STT-Aliases Fehler: {e}") def _correct_query(self, text: str) -> str: """ Korrigiert bekannte Vosk-Fehlerkennungen in Slot-Werten. Wird auf query/channel/genre Slots angewendet NACHDEM der Intent erkannt wurde — nur in TV-Plugin Handlern. Beispiele: "nach utopie" → "naruto" "etc anime" → "ecchi anime" "etchi" → "ecchi" """ if not self._stt_aliases or not text: return text result = text.lower() # Mehrwort-Phrasen zuerst (laengere zuerst) for phrase, replacement in sorted( self._stt_aliases.items(), key=lambda x: len(x[0]), reverse=True, ): if " " in phrase and phrase in result: result = result.replace(phrase, replacement) # Einzelwoerter words = result.split() for i, word in enumerate(words): if word in self._stt_aliases: words[i] = self._stt_aliases[word] corrected = " ".join(words) # Fuellwoerter am Anfang entfernen die nicht zum Sendungstitel gehoeren filler_prefixes = [ "die nächste folge von ", "die naechste folge von ", "der nächste folge von ", "der naechste folge von ", "nächste folge von ", "naechste folge von ", "die nächste folge ", "die naechste folge ", "der nächste folge ", "der naechste folge ", "nächste folge ", "naechste folge ", "der nächste ", "die nächste ", "das nächste ", "der naechste ", "die naechste ", "das naechste ", "den nächsten ", "die nächsten ", "den naechsten ", "nächste ", "naechste ", "nächsten ", "naechsten ", "folge von ", "folge ", ] for prefix in filler_prefixes: if corrected.startswith(prefix): corrected = corrected[len(prefix):] break corrected = corrected.strip() if corrected != text.lower().strip(): pdebug(f"TVProgramPlugin: STT-Korrektur '{text}' → '{corrected}'") return corrected # ====================================================================== # Slot-Listen fuer Intent-Training # ====================================================================== def _register_slot_lists(self) -> None: """ Registriert Slot-Listen fuer das Intent-Training. Der IntentTrainer laedt slot_lists.json automatisch aus jedem Plugin-Verzeichnis (trixy_core/trainer Erweiterung). Hier aktualisieren wir die JSON-Datei dynamisch mit den aktuellen Sendernamen aus CHANNELS — so sind die Slot-Listen immer synchron mit der Sender-Definition. Ergebnis: Der Trainer generiert Trainingsbeispiele wie: "was kommt auf ProSieben" → tv_whats_on, channel="ProSieben" "gibt es ecchi animes" → tv_genre_search, genre="Ecchi" """ import json as _json slot_file = Path(self.plugin_path) / "slot_lists.json" if not slot_file.is_file(): return try: # Aktuelle Sendernamen aus CHANNELS extrahieren from plugins.tv_program.epg.channels import CHANNELS channel_values = set() for info in CHANNELS.values(): channel_values.add(info["name"]) for alias in info.get("aliases", []): channel_values.add(alias) # slot_lists.json aktualisieren (channel-Werte synchronisieren) with open(slot_file, "r", encoding="utf-8") as f: data = _json.load(f) if "channel" in data: existing = set(data["channel"].get("values", [])) if channel_values != existing: data["channel"]["values"] = sorted(channel_values) with open(slot_file, "w", encoding="utf-8") as f: _json.dump(data, f, indent=4, ensure_ascii=False) pdebug( f"TVProgramPlugin: slot_lists.json aktualisiert — " f"{len(channel_values)} Sender-Aliase" ) except Exception as e: pdebug(f"TVProgramPlugin: Slot-Listen Aktualisierung fehlgeschlagen: {e}") # ====================================================================== # Konversations-Kontext # ====================================================================== def _build_context( self, data: IntentReceivedData, program: dict | None = None, **extra, ) -> dict: """ Baut Follow-Up-Kontext aus aktueller Abfrage. Der Kontext wird bei follow_up() als `data` mitgegeben und beim naechsten Turn als Slots an den tv_follow_up Handler weitergereicht. So weiss der Follow-Up-Handler was zuletzt gefragt wurde. """ ctx = { "ctx_time": extra.get("time_iso", ""), "ctx_channel": extra.get("channel_id", ""), "ctx_date": extra.get("date_iso", ""), "ctx_intent": data.intent, } if program: ctx["ctx_program"] = program.get("t", "") ctx["ctx_program_sid"] = program.get("sid", "") ctx["ctx_program_ch"] = program.get("ch", "") ctx["ctx_program_start"] = program.get("s", "") return ctx def _respond_with_followup(self, text: str, ctx: dict) -> IntentResult: """Antwortet und haelt die Konversation offen fuer Nachfragen.""" return IntentResult.follow_up( response_text=text, follow_up_intent="tv_follow_up", **ctx, ) # ====================================================================== # Follow-Up Handler # ====================================================================== @intent("tv_follow_up", description="TV-Programm Nachfrage in laufender Konversation") async def handle_follow_up(self, data: IntentReceivedData) -> IntentResult: """ Verarbeitet Nachfragen im Konversations-Kontext. Erkennt Muster wie: - "Und auf RTL2?" → gleiche Zeit, anderer Sender - "Und morgen?" → gleicher Sender, anderer Tag - "Und um 21 Uhr?" → gleicher Sender, andere Zeit - "Erinnere mich daran" → Reminder fuer letztes Programm - "Wo kann ich das streamen?" → Streaming fuer letztes Programm - "Was ist das?" / "Erzaehl mir mehr" → Info zum letzten Programm """ if not self._cache: return self._no_data() text = (data.original_text or "").strip().lower() ctx_time = data.get_slot("ctx_time", "") ctx_channel = data.get_slot("ctx_channel", "") ctx_date = data.get_slot("ctx_date", "") ctx_program = data.get_slot("ctx_program", "") ctx_program_sid = data.get_slot("ctx_program_sid", "") ctx_program_ch = data.get_slot("ctx_program_ch", "") ctx_intent = data.get_slot("ctx_intent", "") # --- "Erinnere mich daran" / "Merke dir das" --- if self._matches_reminder(text): if ctx_program: data.slots["query"] = ctx_program return await self.handle_set_reminder(data, query=ctx_program) return IntentResult.success_with_response( "Ich weiss leider nicht mehr an was ich dich erinnern soll." ) # --- "Wo kann ich das streamen?" / "Kann ich das streamen?" --- if self._matches_streaming(text): query = ctx_program or "" if query: data.slots["query"] = query return await self.handle_streaming(data, query=query) return IntentResult.success_with_response( "Welche Sendung meinst du?" ) # --- "Was ist das?" / "Erzaehl mir mehr" / "Info" --- if self._matches_info(text): query = ctx_program or "" if query: data.slots["query"] = query return await self.handle_show_info(data, query=query) return IntentResult.success_with_response("Worueber moechtest du mehr erfahren?") # --- "Und auf RTL2?" / "Und auf ProSieben?" → Sender-Wechsel --- new_channel = self._extract_channel_from_text(text) if new_channel: target_time = self._time_from_context(ctx_time, ctx_date) program = self._cache.get_program_at(new_channel, target_time) ctx = self._build_context( data, program, channel_id=new_channel, time_iso=ctx_time, date_iso=ctx_date, ) if not program: ch_name = self._channel_mapper.get_display_name(new_channel) return self._respond_with_followup( f"Um {target_time.strftime('%H:%M')} laeuft nichts auf {ch_name}.", ctx, ) return self._respond_with_followup(self._format_program(program), ctx) # --- "Und morgen?" / "Und uebermorgen?" → Datums-Wechsel --- new_date = self._extract_date_from_text(text) if new_date and ctx_channel: target_time = self._time_from_context(ctx_time, new_date.isoformat()) target_time = target_time.replace( year=new_date.year, month=new_date.month, day=new_date.day ) program = self._cache.get_program_at(ctx_channel, target_time) ctx = self._build_context( data, program, channel_id=ctx_channel, time_iso=ctx_time, date_iso=new_date.isoformat(), ) if not program: ch_name = self._channel_mapper.get_display_name(ctx_channel) date_label = self._format_date_tts(target_time.isoformat()) return self._respond_with_followup( f"{date_label} um {target_time.strftime('%H:%M')} laeuft nichts auf {ch_name}.", ctx, ) return self._respond_with_followup(self._format_program(program), ctx) # --- "Und um 21 Uhr?" → Zeit-Wechsel --- new_time = self._extract_time_from_text(text) if new_time is not None and ctx_channel: target_date = date.fromisoformat(ctx_date) if ctx_date else date.today() target_time = datetime( target_date.year, target_date.month, target_date.day, new_time[0], new_time[1], ) program = self._cache.get_program_at(ctx_channel, target_time) ctx = self._build_context( data, program, channel_id=ctx_channel, time_iso=target_time.isoformat(), date_iso=ctx_date, ) if not program: ch_name = self._channel_mapper.get_display_name(ctx_channel) return self._respond_with_followup( f"Um {target_time.strftime('%H:%M')} laeuft nichts auf {ch_name}.", ctx, ) return self._respond_with_followup(self._format_program(program), ctx) # --- Nichts erkannt → Konversation beenden --- return IntentResult.success_with_response( "Das habe ich leider nicht verstanden." ) # ====================================================================== # Follow-Up Text-Erkennung # ====================================================================== @staticmethod def _matches_reminder(text: str) -> bool: """Erkennt Erinnerungs-Absicht in Follow-Up.""" keywords = [ "erinner", "erinnerung", "merke", "merk", "vormerken", "benachrichtig", "daran erinnern", "nicht vergessen", ] return any(kw in text for kw in keywords) @staticmethod def _matches_streaming(text: str) -> bool: """Erkennt Streaming-Absicht in Follow-Up.""" keywords = [ "streamen", "stream", "streaming", "online schauen", "online gucken", "online sehen", "netflix", "amazon", "crunchyroll", "disney", ] return any(kw in text for kw in keywords) @staticmethod def _matches_info(text: str) -> bool: """Erkennt Info-Absicht in Follow-Up.""" keywords = [ "was ist das", "worum geht", "erzaehl", "erzaehl mir", "mehr darueber", "mehr dazu", "info", "beschreibung", "wovon handelt", "genre", ] return any(kw in text for kw in keywords) def _extract_channel_from_text(self, text: str) -> str | None: """Versucht einen Sender-Namen aus dem Follow-Up-Text zu extrahieren.""" # "und auf RTL2?" → entferne Fuellwoerter, versuche Channel-Resolve cleaned = text for prefix in ["und auf ", "auf ", "und ", "was kommt auf ", "was laeuft auf "]: if cleaned.startswith(prefix): cleaned = cleaned[len(prefix):] break # Fragezeichen und Satzzeichen entfernen cleaned = cleaned.rstrip("?!. ") if cleaned: return self._channel_mapper.resolve(cleaned) return None @staticmethod def _extract_date_from_text(text: str) -> date | None: """Erkennt Datumsbezug im Follow-Up-Text.""" if "morgen" in text: return date.today() + timedelta(days=1) if "uebermorgen" in text or "übermorgen" in text: return date.today() + timedelta(days=2) # Wochentage tage = { "montag": 0, "dienstag": 1, "mittwoch": 2, "donnerstag": 3, "freitag": 4, "samstag": 5, "sonntag": 6, } for tag_name, weekday in tage.items(): if tag_name in text: today = date.today() days_ahead = weekday - today.weekday() if days_ahead <= 0: days_ahead += 7 return today + timedelta(days=days_ahead) return None @staticmethod def _extract_time_from_text(text: str) -> tuple[int, int] | None: """Erkennt Zeitangabe im Follow-Up-Text.""" # "und um 21 Uhr?" / "um 21:30" / "um 22 Uhr 15" cleaned = text.lower().replace("uhr", "").strip() match = re.search(r"(\d{1,2})\s*[:\s]\s*(\d{2})", cleaned) if match: h, m = int(match.group(1)), int(match.group(2)) if 0 <= h <= 23 and 0 <= m <= 59: return (h, m) match = re.search(r"(\d{1,2})", cleaned) if match: h = int(match.group(1)) if 0 <= h <= 23: return (h, 0) return None def _time_from_context(self, ctx_time: str, ctx_date: str) -> datetime: """Rekonstruiert datetime aus Kontext-Strings.""" now = datetime.now().astimezone() if ctx_time: try: return datetime.fromisoformat(ctx_time) except ValueError: pass if ctx_date: try: d = date.fromisoformat(ctx_date) return datetime(d.year, d.month, d.day, now.hour, now.minute) except ValueError: pass return now # ====================================================================== # Intent-Handler (mit Follow-Up-Kontext) # ====================================================================== @intent("tv_whats_on", description="TV-Programm fuer Sender und Uhrzeit") @pattern("was kommt [heute|morgen] [um] {time} [uhr] auf {channel}") @pattern("was laeuft [heute|morgen] [um] {time} [uhr] [auf] {channel}") @pattern("programm [von|auf] {channel} [heute|morgen]") @example("Was kommt heute um 20 Uhr auf RTL2?") @example("Was laeuft auf Pro7?") async def handle_whats_on( self, data: IntentReceivedData, channel: str = "", time: str = "", ) -> IntentResult: """Was kommt um X auf Sender Y?""" if not self._cache: return self._no_data() channel_name = self._correct_query(data.get_slot("channel", "")) channel_id = self._channel_mapper.resolve(channel_name) if channel_name else None if not channel_id: return IntentResult.failure( "Sender nicht erkannt", response_text=f"Ich kenne den Sender '{channel_name}' leider nicht.", ) # Wurde eine explizite Uhrzeit genannt? has_explicit_time = bool(data.get_slot("time", "")) target_time = self._parse_time(data) program = self._cache.get_program_at(channel_id, target_time) ctx = self._build_context( data, program, channel_id=channel_id, time_iso=target_time.isoformat(), date_iso=target_time.date().isoformat(), ) display_name = self._channel_mapper.get_display_name(channel_id) if not program: if has_explicit_time: return self._respond_with_followup( f"Um {self._format_time_tts(target_time.isoformat())} " f"laeuft nichts auf {display_name}.", ctx, ) return self._respond_with_followup( f"Gerade laeuft nichts auf {display_name}.", ctx, ) if has_explicit_time: return self._respond_with_followup( self._format_program(program), ctx, ) # Ohne explizite Zeit: "Gerade laeuft X auf Y" statt "Um 02:47 kommt X" return self._respond_with_followup( self._format_program_now(program, display_name), ctx, ) @intent("tv_primetime", description="Primetime-Programm aller Sender") @pattern("was kommt [heute|morgen] (abend|abends|im fernsehen)") @pattern("fernsehprogramm [heute|morgen]") @example("Was kommt heute Abend?") @example("Fernsehprogramm heute") async def handle_primetime(self, data: IntentReceivedData) -> IntentResult: """Was kommt heute Abend auf den Top-Sendern?""" if not self._cache: return self._no_data() target_date = self._parse_date(data) top_channels = ["ard", "zdf", "rtl", "sat1", "pro7", "vox"] highlights = [] last_program = None for ch_id in top_channels: program = self._cache.get_primetime(ch_id, target_date) if program: name = self._channel_mapper.get_display_name(ch_id) highlights.append(f"Auf {name}: {program['t']}") last_program = program if not highlights: return IntentResult.success_with_response( "Ich habe kein Primetime-Programm gefunden." ) date_label = "Heute" if target_date == date.today() else "Morgen" ctx = self._build_context( data, last_program, time_iso=datetime(target_date.year, target_date.month, target_date.day, 20, 15).isoformat(), date_iso=target_date.isoformat(), ) return self._respond_with_followup( f"{date_label} Abend um 20 Uhr 15. {'. '.join(highlights)}.", ctx, ) @intent("tv_when_next", description="Naechste Ausstrahlung einer Sendung") @pattern("wann kommt [die naechste folge] {query}") @pattern("wann laeuft {query} [wieder]") @pattern("naechste folge [von] {query}") @example("Wann kommt die naechste Folge Naruto?") @example("Wann laeuft Tatort wieder?") async def handle_when_next( self, data: IntentReceivedData, query: str = "", ) -> IntentResult: """Wann kommt Sendung X als naechstes?""" if not self._cache: return self._no_data() query = self._correct_query(data.get_slot("query", "")) if not query: return IntentResult.failure( "Kein Suchbegriff", response_text="Welche Sendung meinst du?", ) results = self._cache.search_title(query, limit=1) if not results: return IntentResult.success_with_response( f"Ich habe '{query}' nicht im Programm gefunden." ) program = results[0] channel_name = self._channel_mapper.get_display_name(program["ch"]) date_label = self._format_date_tts(program["s"]) time_label = self._format_time_tts(program["s"]) response = f"{program['t']} kommt {date_label} um {time_label} auf {channel_name}." if program.get("sn") and program.get("ep"): response += f" Staffel {program['sn']}, Folge {program['ep']}." ctx = self._build_context( data, program, channel_id=program["ch"], time_iso=program["s"], date_iso=program["s"][:10], ) return self._respond_with_followup(response, ctx) @intent("tv_which_channel", description="Sender einer Sendung finden") @pattern("auf welchem (sender|kanal) kommt {query}") @pattern("wo kommt {query}") @pattern("wo laeuft {query}") @example("Auf welchem Sender kommt DSDS?") @example("Wo laeuft Naruto?") async def handle_which_channel( self, data: IntentReceivedData, query: str = "", ) -> IntentResult: """Auf welchem Sender laeuft X?""" if not self._cache: return self._no_data() query = self._correct_query(data.get_slot("query", "")) if not query: return IntentResult.failure("Kein Suchbegriff", response_text="Welche Sendung meinst du?") results = self._cache.search_title(query, limit=10) if not results: return IntentResult.success_with_response( f"Ich habe '{query}' nicht im Programm gefunden." ) # Eindeutige Sender sammeln channels = list(dict.fromkeys( self._channel_mapper.get_display_name(p["ch"]) for p in results )) first = results[0] ctx = self._build_context( data, first, channel_id=first["ch"], time_iso=first.get("s", ""), date_iso=first.get("s", "")[:10] if first.get("s") else "", ) if len(channels) == 1: return self._respond_with_followup(f"{query} laeuft auf {channels[0]}.", ctx) return self._respond_with_followup( f"{query} laeuft auf {', '.join(channels[:5])}.", ctx, ) @intent("tv_genre_search", description="Sendungen nach Genre/Tag suchen") @pattern("gibt es [heute|morgen|diese woche] {genre} [im fernsehen]") @pattern("welche {genre} (kommen|laufen) [heute|morgen|diese woche]") @example("Gibt es diese Woche Ecchi-Animes?") @example("Welche Animes kommen heute?") async def handle_genre_search( self, data: IntentReceivedData, genre: str = "", ) -> IntentResult: """Sendungen nach Genre oder Tag suchen.""" if not self._cache: return self._no_data() genre_text = self._correct_query(data.get_slot("genre", "")) if not genre_text: return IntentResult.failure("Kein Genre", response_text="Welches Genre meinst du?") # "Ecchi-Animes" → tag="Ecchi", "Action Filme" → tag="Action" tag = self._extract_genre_tag(genre_text) results = self._cache.search_by_tag(tag) if not results: # Fallback: Genre-Suche results = self._cache.search_by_genre(tag) if not results: return IntentResult.success_with_response( f"Ich habe keine {genre_text} im Programm gefunden." ) # Auf Zeitraum filtern und Top 5 ausgeben now = datetime.now().astimezone() future = [r for r in results if datetime.fromisoformat(r["s"]) > now] future.sort(key=lambda r: r["s"]) if not future: return IntentResult.success_with_response( f"Aktuell sind keine {genre_text} geplant." ) items = [] for p in future[:5]: ch = self._channel_mapper.get_display_name(p["ch"]) items.append(f"{p['t']} — {self._format_date_tts(p['s'])} um {self._format_time_tts(p['s'])} auf {ch}") last = future[0] ctx = self._build_context( data, last, channel_id=last["ch"], time_iso=last.get("s", ""), date_iso=last.get("s", "")[:10] if last.get("s") else "", ) return self._respond_with_followup( f"{len(future)} {genre_text} gefunden. {'. '.join(items)}.", ctx, ) @intent("tv_streaming", description="Streaming-Verfuegbarkeit abfragen") @pattern("wo kann ich {query} (streamen|schauen|sehen)") @pattern("auf welchem streamingdienst gibt es {query}") @example("Wo kann ich Naruto streamen?") @example("Auf welchem Streamingdienst gibt es Breaking Bad?") async def handle_streaming( self, data: IntentReceivedData, query: str = "", ) -> IntentResult: """Wo kann man X streamen?""" if not self._cache: return self._no_data() query = self._correct_query(data.get_slot("query", "")) if not query: return IntentResult.failure("Kein Suchbegriff", response_text="Welche Sendung meinst du?") series_id = EPGCache._generate_series_id(query) services = self._cache.get_streaming_info(series_id) # On-Demand Enrichment falls noch nicht vorhanden if not services and self._enricher: enrichment = await self._enricher.enrich_on_demand(query) if enrichment: services = enrichment.get("str", []) if not services: return IntentResult.success_with_response( f"Ich habe keine Streaming-Informationen fuer {query}." ) ctx = self._build_context(data) ctx["ctx_program"] = query ctx["ctx_program_sid"] = series_id return self._respond_with_followup( f"{query} ist verfuegbar auf {', '.join(services)}.", ctx, ) @intent("tv_show_info", description="Informationen zu einer Sendung") @pattern("(was ist|worum geht es bei) {query}") @pattern("info [zu|ueber] {query}") @example("Was ist Naruto?") @example("Worum geht es bei Breaking Bad?") async def handle_show_info( self, data: IntentReceivedData, query: str = "", ) -> IntentResult: """Info ueber eine Sendung.""" if not self._cache: return self._no_data() query = self._correct_query(data.get_slot("query", "")) if not query: return IntentResult.failure("Kein Suchbegriff", response_text="Ueber welche Sendung?") series_id = EPGCache._generate_series_id(query) enrichment = self._cache.get_enrichment(series_id) # On-Demand Enrichment if not enrichment and self._enricher: enrichment = await self._enricher.enrich_on_demand(query) if not enrichment: return IntentResult.success_with_response( f"Ich habe keine Informationen zu {query}." ) parts = [enrichment.get("t", query)] genres = enrichment.get("g", []) if genres: parts.append(f"Genre: {', '.join(genres)}") desc = enrichment.get("d", "") if desc: # Max 2 Saetze fuer TTS sentences = desc.split(". ") short = ". ".join(sentences[:2]) if not short.endswith("."): short += "." parts.append(short) rating = enrichment.get("rt") if rating: parts.append(f"Bewertung: {rating} von 10") streaming = enrichment.get("str", []) if streaming: parts.append(f"Verfuegbar auf: {', '.join(streaming)}") ctx = self._build_context(data) ctx["ctx_program"] = enrichment.get("t", query) ctx["ctx_program_sid"] = series_id return self._respond_with_followup(". ".join(parts) + ".", ctx) @intent("tv_language_search", description="Anime/Serien nach Sprache suchen (Syncro/Sub)") @pattern("welche animes kommen [heute|morgen] mit deutscher (syncro|synchro|synchronisation)") @pattern("wann kommt {query} (auf deutsch|mit deutscher syncro)") @pattern("wann kommt {query} mit (sub|subs|untertitel)") @pattern("gibt es {query} (auf deutsch|mit untertitel)") @example("Welche Animes kommen heute mit deutscher Syncro?") @example("Wann kommt Naruto auf deutsch?") async def handle_language_search( self, data: IntentReceivedData, query: str = "", ) -> IntentResult: """Sucht nach Sendungen mit bestimmter Sprache (Syncro/Sub).""" if not self._cache: return self._no_data() text = (data.original_text or "").lower() query = self._correct_query(data.get_slot("query", "")) # Was sucht der Nutzer? Syncro (deutsch audio) oder Sub (OmU)? wants_dub = any(w in text for w in [ "syncro", "synchro", "synchronisation", "auf deutsch", "deutscher", "deutsche", "deutsch", ]) wants_sub = any(w in text for w in [ "sub", "subs", "untertitel", "subtitles", "omu", "originalsprache", "japanisch", ]) # Im deutschen TV ist fast alles synchronisiert # Spaeter bei Streaming-Integration: sub/dub Info aus Enrichment if wants_dub and not wants_sub: # Deutsche Syncro gesucht if query: # Spezifischer Titel results = self._cache.search_title(query, limit=1) if results: program = results[0] ch = self._channel_mapper.get_display_name(program["ch"]) date_l = self._format_date_tts(program["s"]) time_l = self._format_time_tts(program["s"]) # Enrichment pruefen fuer Originalsprache sid = program.get("sid", self._cache._generate_series_id(program["t"])) enrichment = self._cache.get_enrichment(sid) orig = "" if enrichment and enrichment.get("orig_lang"): lang_names = {"ja": "Japanisch", "ko": "Koreanisch", "zh": "Chinesisch"} orig = lang_names.get(enrichment["orig_lang"], "") response = ( f"{program['t']} kommt {date_l} um {time_l} auf {ch} " f"mit deutscher Synchronisation." ) if orig: response += f" Originalsprache: {orig}." ctx = self._build_context(data, program, channel_id=program["ch"], time_iso=program["s"], date_iso=program["s"][:10]) return self._respond_with_followup(response, ctx) else: return IntentResult.success_with_response( f"Ich habe '{query}' nicht im Programm gefunden." ) else: # Alle Anime mit deutscher Syncro (= alle Anime im deutschen TV) tag = "anime" results = self._cache.search_by_genre(tag) now = datetime.now().astimezone() future = [r for r in results if datetime.fromisoformat(r["s"]) > now] future.sort(key=lambda r: r["s"]) if not future: return IntentResult.success_with_response( "Aktuell sind keine Animes mit deutscher Syncro geplant." ) items = [] for p in future[:5]: ch = self._channel_mapper.get_display_name(p["ch"]) items.append(f"{p['t']} — {self._format_date_tts(p['s'])} um {self._format_time_tts(p['s'])} auf {ch}") return IntentResult.success_with_response( f"Im deutschen TV laufen Animes immer mit deutscher Synchronisation. " f"{len(future)} Animes geplant. {'. '.join(items)}." ) elif wants_sub and not wants_dub: # Untertitel / OmU gesucht return IntentResult.success_with_response( "Im linearen TV gibt es leider selten Untertitel-Versionen. " "Auf Streaming-Diensten wie Crunchyroll oder Netflix sind " "die meisten Animes mit japanischem Originalton und deutschen " "Untertiteln verfuegbar." ) else: # Beides oder unklar if query: results = self._cache.search_title(query, limit=1) if results: program = results[0] ch = self._channel_mapper.get_display_name(program["ch"]) return IntentResult.success_with_response( f"{program['t']} laeuft auf {ch} mit deutscher Synchronisation. " f"Fuer die Originalversion mit Untertiteln schau auf " f"Crunchyroll oder Netflix." ) return IntentResult.success_with_response( "Im deutschen Fernsehen laufen Animes mit deutscher Synchronisation. " "Fuer japanischen Originalton mit Untertiteln empfehle ich " "Streaming-Dienste wie Crunchyroll oder Netflix." ) @intent("tv_set_reminder", description="TV-Erinnerung setzen") @pattern("(erinnere|erinner) [mich] [wenn] {query} (kommt|laeuft)") @pattern("erinnerung [fuer] {query}") @pattern("(merke|merk) [dir] {query} [vor]") @example("Erinnere mich wenn Naruto kommt") @example("Merke dir DSDS vor") async def handle_set_reminder( self, data: IntentReceivedData, query: str = "", ) -> IntentResult: """Setzt eine TV-Erinnerung.""" if not self._cache: return self._no_data() query = self._correct_query(data.get_slot("query", "")) if not query: return IntentResult.failure("Kein Suchbegriff", response_text="An was soll ich dich erinnern?") series_id = EPGCache._generate_series_id(query) # Duplikat-Check reminders = self._cache.reminders for r in reminders: if r.get("sid") == series_id and r.get("active"): return IntentResult.success_with_response( f"Erinnerung fuer {query} ist bereits aktiv." ) # Erinnerung speichern reminders.append({ "title": query, "sid": series_id, "active": True, "satellite_id": data.satellite_id, "created_at": datetime.now().astimezone().isoformat(), }) await self._cache.save_async() # Naechste Ausstrahlung suchen next_airing = self._cache.get_next_airing(series_id) if not next_airing: results = self._cache.search_title(query, limit=1) if results: next_airing = results[0] if next_airing: ch = self._channel_mapper.get_display_name(next_airing["ch"]) return IntentResult.success_with_response( f"Erinnerung fuer {query} gespeichert. " f"Naechste Ausstrahlung: {self._format_date_tts(next_airing['s'])} " f"um {self._format_time_tts(next_airing['s'])} auf {ch}." ) return IntentResult.success_with_response( f"Erinnerung fuer {query} gespeichert." ) @intent("tv_list_reminders", description="TV-Erinnerungen auflisten") @pattern("welche tv erinnerungen [habe ich]") @pattern("[meine] tv (erinnerungen|merkliste)") @example("Welche TV Erinnerungen habe ich?") @example("Meine TV Merkliste") async def handle_list_reminders(self, data: IntentReceivedData) -> IntentResult: """Listet alle aktiven TV-Erinnerungen.""" if not self._cache: return self._no_data() reminders = [ r for r in self._cache.reminders if r.get("active") ] if not reminders: return IntentResult.success_with_response( "Du hast keine aktiven TV-Erinnerungen." ) titles = [r["title"] for r in reminders] return IntentResult.success_with_response( f"Du hast {len(titles)} TV-Erinnerungen: {', '.join(titles)}." ) @intent("tv_delete_reminder", description="TV-Erinnerung loeschen") @pattern("(loesche|entferne) [die] tv erinnerung [fuer] {query}") @pattern("{query} [von der] merkliste (loeschen|entfernen|streichen)") @example("Loesche die TV Erinnerung fuer Naruto") @example("Naruto von der Merkliste streichen") async def handle_delete_reminder( self, data: IntentReceivedData, query: str = "", ) -> IntentResult: """Loescht eine TV-Erinnerung.""" if not self._cache: return self._no_data() query = self._correct_query(data.get_slot("query", "")) if not query: return IntentResult.failure("Kein Suchbegriff", response_text="Welche Erinnerung?") series_id = EPGCache._generate_series_id(query) reminders = self._cache.reminders found = False for r in reminders: if r.get("sid") == series_id and r.get("active"): r["active"] = False found = True if found: await self._cache.save_async() return IntentResult.success_with_response( f"Erinnerung fuer {query} geloescht." ) return IntentResult.success_with_response( f"Keine aktive Erinnerung fuer {query} gefunden." ) # ====================================================================== # Reminder-System # ====================================================================== async def _reminder_check_loop(self) -> None: """Prueft alle 60s ob eine erinnerte Sendung gleich anfaengt.""" interval = self.get_config_value("reminders.check_interval_seconds", 60) notify_before = self.get_config_value("reminders.notify_minutes_before", 5) while True: try: await asyncio.sleep(interval) await self._check_reminders(notify_before) except asyncio.CancelledError: break except Exception as e: perror(f"TVProgramPlugin: Reminder-Fehler: {e}") async def _check_reminders(self, notify_before_minutes: int) -> None: """Prueft Erinnerungen und sendet TTS-Benachrichtigungen.""" if not self._cache: return now = datetime.now().astimezone() threshold = now + timedelta(minutes=notify_before_minutes) for reminder in self._cache.reminders: if not reminder.get("active"): continue sid = reminder.get("sid", "") next_airing = self._cache.get_next_airing(sid) if not next_airing: continue start = datetime.fromisoformat(next_airing["s"]) # Schon benachrichtigt fuer diese Ausstrahlung? if reminder.get("last_notified") == next_airing["s"]: continue if now <= start <= threshold: channel = self._channel_mapper.get_display_name(next_airing["ch"]) text = ( f"{reminder['title']} kommt in {notify_before_minutes} Minuten " f"auf {channel}." ) events = getattr(self.application, "events", None) if events: await events.emit("tts_request", { "request_id": f"tv-reminder-{sid}", "satellite_id": reminder.get("satellite_id", ""), "text": text, }) # Markieren um Doppel-Benachrichtigung zu vermeiden reminder["last_notified"] = next_airing["s"] await self._cache.save_async() pdebug(f"TVProgramPlugin: Reminder gesendet — {reminder['title']}") # ====================================================================== # Hilfsmethoden # ====================================================================== def _no_data(self) -> IntentResult: """Standard-Antwort wenn kein Cache vorhanden.""" return IntentResult.failure( "Keine Daten", response_text="Die TV-Programm-Daten werden noch geladen. Bitte versuche es gleich nochmal.", ) def _parse_time(self, data: IntentReceivedData) -> datetime: """Parst Uhrzeit aus Slot oder nutzt aktuelle Zeit.""" time_str = data.get_slot("time", "") target_date = self._parse_date(data) now = datetime.now().astimezone() if not time_str: # Kein expliziter Zeitpunkt → jetzt return now # "20:15", "20 Uhr 15", "20 Uhr", "halb neun", etc. time_str = time_str.lower().replace("uhr", "").strip() # "20:15" oder "20 15" match = re.match(r"(\d{1,2})\s*[:\s]\s*(\d{2})", time_str) if match: h, m = int(match.group(1)), int(match.group(2)) return datetime(target_date.year, target_date.month, target_date.day, h, m) # "20" (nur Stunde) match = re.match(r"(\d{1,2})$", time_str.strip()) if match: h = int(match.group(1)) if 0 <= h <= 23: return datetime(target_date.year, target_date.month, target_date.day, h, 0) return now def _parse_date(self, data: IntentReceivedData) -> date: """Erkennt 'heute'/'morgen' aus dem gesamten Intent-Text.""" text = self._correct_query(getattr(data, "original_text", "") or "") text_lower = text.lower() if "morgen" in text_lower: return date.today() + timedelta(days=1) return date.today() def _format_program_now(self, program: dict, channel_name: str) -> str: """Formatiert ein aktuell laufendes Programm fuer TTS.""" title = program["t"] genre = program.get("g", "") parts = [f"Gerade laeuft '{title}' auf {channel_name}"] if genre: parts.append(f"Genre: {genre}") if program.get("sn") and program.get("ep"): parts.append(f"Staffel {program['sn']}, Folge {program['ep']}") # Restdauer berechnen try: end = datetime.fromisoformat(program["e"]) remaining = int((end - datetime.now().astimezone()).total_seconds() / 60) if remaining > 0: parts.append(f"Noch {remaining} Minuten") except (ValueError, KeyError): pass return ". ".join(parts) + "." def _format_program(self, program: dict) -> str: """Formatiert ein Programm fuer TTS-Ausgabe.""" ch = self._channel_mapper.get_display_name(program["ch"]) time_label = self._format_time_tts(program["s"]) title = program["t"] genre = program.get("g", "") parts = [f"Um {time_label} kommt '{title}' auf {ch}"] if genre: parts.append(f"Genre: {genre}") if program.get("sn") and program.get("ep"): parts.append(f"Staffel {program['sn']}, Folge {program['ep']}") if program.get("et"): parts.append(f"Titel: {program['et']}") # Dauer berechnen try: start = datetime.fromisoformat(program["s"]) end = datetime.fromisoformat(program["e"]) duration = int((end - start).total_seconds() / 60) if duration > 0: parts.append(f"Dauer: {duration} Minuten") except (ValueError, KeyError): pass return ". ".join(parts) + "." @staticmethod def _format_date_tts(iso_str: str) -> str: """ISO-Datum → 'heute' / 'morgen' / 'am Freitag' / '12. April'.""" try: dt = datetime.fromisoformat(iso_str) today = date.today() delta = (dt.date() - today).days if delta == 0: return "heute" elif delta == 1: return "morgen" elif 2 <= delta <= 6: tage = ["Montag", "Dienstag", "Mittwoch", "Donnerstag", "Freitag", "Samstag", "Sonntag"] return f"am {tage[dt.weekday()]}" else: monate = ["Januar", "Februar", "Maerz", "April", "Mai", "Juni", "Juli", "August", "September", "Oktober", "November", "Dezember"] return f"am {dt.day}. {monate[dt.month - 1]}" except (ValueError, IndexError): return "" @staticmethod def _format_time_tts(iso_str: str) -> str: """ISO-Zeit → '20 Uhr 15' / '20 Uhr'.""" try: dt = datetime.fromisoformat(iso_str) if dt.minute == 0: return f"{dt.hour} Uhr" return f"{dt.hour} Uhr {dt.minute}" except ValueError: return "" @staticmethod def _extract_genre_tag(genre_text: str) -> str: """ Extrahiert den eigentlichen Tag/Genre aus gesprochenem Text. 'Ecchi-Animes' → 'Ecchi' 'Horrorfilme' → 'Horror' 'Action Filme' → 'Action' 'Shounen Animes' → 'Shounen' 'Dokumentationen' → 'Dokumentation' 'Krimis' → 'Krimi' 'Komödien' → 'Komoedie' """ text = genre_text.strip() lower = text.lower() # Zusammengeschriebene Suffixe (Horrorfilme, Actionserien, etc.) for suffix in ["filme", "film", "serien", "serie", "sendungen", "sendung"]: if lower.endswith(suffix) and len(lower) > len(suffix) + 2: text = text[:len(text) - len(suffix)] lower = text.lower() break # Getrennte Suffixe mit Bindestrich oder Leerzeichen for suffix in ["-animes", "-anime", " animes", " anime", " filme", " film", " serien", " serie", " sendungen", " sendung", " dokus", " doku"]: if lower.endswith(suffix): text = text[:len(text) - len(suffix)] lower = text.lower() break # Deutsche Pluralformen → Singular plural_map = { "krimis": "Krimi", "komödien": "Komoedie", "komoedien": "Komoedie", "dokumentationen": "Dokumentation", "dokus": "Doku", "thriller": "Thriller", # bleibt gleich "animes": "Anime", "cartoons": "Cartoon", } if lower in plural_map: return plural_map[lower] # Generischer Plural: trailing 's' entfernen if lower.endswith("s") and len(text) > 3: text = text[:-1] return text.strip()