# -*- coding: utf-8 -*- """ Streaming-Plugin fuer Trixy. Beantwortet Fragen zum Streaming-Katalog per Sprache: - "Welche Horrorfilme gibt es auf Netflix?" - "Gibt es Naruto auf Crunchyroll?" - "Wo kann ich Breaking Bad schauen?" - "Was ist neu auf Disney+?" - "Empfiehl mir einen guten Anime auf Crunchyroll" Datenquelle: TMDB Watch Providers API (kostenloser Key). Unterstuetzte Dienste: Netflix, Amazon Prime, Crunchyroll, Disney+, WOW. """ from __future__ import annotations import asyncio import json as _json import re from datetime import datetime from pathlib import Path from typing import Any, 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 if TYPE_CHECKING: from plugins.streaming.catalog.cache import StreamingCache from plugins.streaming.catalog.tmdb_discover import TMDBDiscoverClient from plugins.streaming.catalog.sync import CatalogSync # ====================================================================== # Service-Name Mapping (gesprochen → service_id) # ====================================================================== _SERVICE_ALIASES: dict[str, str] = { "netflix": "netflix", "amazon prime": "amazon_prime", "amazon": "amazon_prime", "prime": "amazon_prime", "prime video": "amazon_prime", "crunchyroll": "crunchyroll", "disney+": "disney_plus", "disney plus": "disney_plus", "disney": "disney_plus", "wow": "wow", "sky": "wow", "sky ticket": "wow", } # Dub/Sub-Erkennung _DUB_KEYWORDS: list[str] = [ "deutsche synchronisation", "deutscher syncro", "deutscher synchro", "auf deutsch", "deutsch vertont", "german dub", "deutsche vertonung", "synchronisiert", "syncro", "synchro", ] _SUB_KEYWORDS: list[str] = [ "untertitel", "untertiteln", "subs", "subtitles", "sub", "mit sub", "mit subs", "mit untertitel", "japanisch mit", "ov mit", "original mit", ] class StreamingPlugin(TrixyPlugin): """Streaming-Dienst Katalog-Suche per Sprache.""" NAME = "streaming" VERSION = "1.0.0" DESCRIPTION = "Streaming-Dienst Katalog-Suche (Netflix, Crunchyroll, Prime, Disney+, WOW)" AUTHOR = "Trixy" def __init__(self, application: Any, plugin_path: Path, config: dict[str, Any] | None = None) -> None: super().__init__(application, plugin_path, config) self._registry = IntentRegistry.get_instance() self._cache: StreamingCache | None = None self._tmdb_client: TMDBDiscoverClient | None = None self._catalog_sync: CatalogSync | None = None self._sync_task: asyncio.Task | None = None self._weekly_job_id: str = "" self._daily_job_id: str = "" self._watchlist_job_id: str = "" self._stt_aliases: dict[str, str] = {} # ====================================================================== # Lifecycle # ====================================================================== async def on_load(self) -> None: """Initialisiert Cache, TMDB-Client und startet Background-Sync.""" pinfo(f"StreamingPlugin: Lade v{self.VERSION}") # 1. Cache-Verzeichnis erstellen cache_dir = Path(self.get_config_value("cache_dir", "cache/streaming")) cache_dir.mkdir(parents=True, exist_ok=True) # 2. StreamingCache initialisieren und laden from plugins.streaming.catalog.cache import StreamingCache self._cache = StreamingCache(cache_dir) loaded = self._cache.load() if loaded: stats = self._cache.get_stats() pinfo( f"StreamingPlugin: Cache geladen — " f"{stats.get('total_titles', 0)} Titel, " f"{stats.get('service_count', 0)} Dienste" ) else: pdebug("StreamingPlugin: Kein Cache vorhanden — erster Sync noetig") # 3. TMDB-Client initialisieren api_key = self.get_config_value("tmdb_api_key", "") if not api_key: pwarn( "StreamingPlugin: Kein TMDB API Key konfiguriert — " "bitte in config.json unter 'tmdb_api_key' eintragen" ) else: from plugins.streaming.catalog.tmdb_discover import TMDBDiscoverClient self._tmdb_client = TMDBDiscoverClient( api_key=api_key, region=self.get_config_value("region", "DE"), language=self.get_config_value("language", "de-DE"), ) # 4. CatalogSync initialisieren if self._tmdb_client and self._cache: from plugins.streaming.catalog.sync import CatalogSync services_config = self.get_config_value("services", {}) self._catalog_sync = CatalogSync( tmdb_client=self._tmdb_client, cache=self._cache, services_config=services_config, max_titles_per_service=self.get_config_value("sync.max_titles_per_service", 500), ) # 5. STT-Aliases laden (Vosk-Fehlerkennungen fuer Streaming-Begriffe) self._load_stt_aliases() # 6. Background-Sync starten (nicht blockierend) if self._catalog_sync: self._sync_task = asyncio.create_task(self._initial_sync()) # 7. Scheduler-Jobs registrieren (woechentlich + taeglich) self._register_scheduler_jobs() pinfo(f"StreamingPlugin: Geladen v{self.VERSION}") async def on_unload(self) -> None: """Stoppt Background-Tasks, Scheduler-Jobs und deregistriert Intents.""" # Sync-Task abbrechen if self._sync_task and not self._sync_task.done(): self._sync_task.cancel() try: await self._sync_task except asyncio.CancelledError: pass # Scheduler-Jobs entfernen self._remove_scheduler_jobs() # Intents deregistrieren self._registry.unregister_plugin(self.NAME) pdebug("StreamingPlugin: Entladen") # ====================================================================== # Background-Sync # ====================================================================== async def _initial_sync(self) -> None: """Initialer Sync beim Server-Start (einmalig, non-blocking).""" await asyncio.sleep(10) # Server fertig starten lassen try: await self._run_sync(full=True) except asyncio.CancelledError: raise except Exception as e: perror(f"StreamingPlugin: Initialer Sync-Fehler: {e}") async def _run_sync(self, full: bool = False) -> None: """ Fuehrt einen Katalog-Sync durch. Args: full: True fuer vollstaendigen Sync, False fuer Quick-Update """ if not self._catalog_sync: return sync_type = "Voll-Sync" if full else "Quick-Update" pinfo(f"StreamingPlugin: Starte {sync_type}...") try: result = await self._catalog_sync.sync(full=full) if self._cache: self._cache.save() pinfo( f"StreamingPlugin: {sync_type} fertig — " f"{result.get('new_titles', 0)} neue Titel, " f"{result.get('updated_titles', 0)} aktualisiert, " f"{result.get('services_synced', 0)} Dienste" ) except Exception as e: perror(f"StreamingPlugin: {sync_type} fehlgeschlagen: {e}") # ====================================================================== # Scheduler-Jobs # ====================================================================== def _register_scheduler_jobs(self) -> None: """ Registriert Scheduler-Jobs fuer Katalog-Sync. - Woechentlicher Voll-Sync (Sonntag um konfigurierte Uhrzeit) - Taegliches Quick-Update (wenn aktiviert) """ scheduler = self.application.services.get_service("SchedulerService") if not scheduler: pdebug("StreamingPlugin: 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: Woechentlicher Voll-Sync sync_hour = self.get_config_value("sync.sync_hour", 4) sync_day = self.get_config_value("sync.sync_day", "sunday") day_map = { "monday": 1, "tuesday": 2, "wednesday": 3, "thursday": 4, "friday": 5, "saturday": 6, "sunday": 0, } cron_day = day_map.get(sync_day.lower(), 0) weekly_job = scheduler.create_job( trigger=CronTrigger(f"0 {sync_hour} * * {cron_day}"), actions=CallbackAction(self._scheduled_full_sync), job_id="streaming_weekly_sync", save=False, ) self._weekly_job_id = weekly_job.job_id pdebug( f"StreamingPlugin: Woechentlicher Sync registriert " f"({sync_day} um {sync_hour:02d}:00)" ) # Job 2: Taegliches Quick-Update if self.get_config_value("sync.daily_update_enabled", True): daily_job = scheduler.create_job( trigger=IntervalTrigger(hours=24), actions=CallbackAction(self._scheduled_daily_update), job_id="streaming_daily_update", save=False, ) self._daily_job_id = daily_job.job_id pdebug("StreamingPlugin: Taegliches Quick-Update registriert (alle 24h)") # Job 3: Watchlist-Check (taeglich) watchlist_job = scheduler.create_job( trigger=IntervalTrigger(hours=12), actions=CallbackAction(self._check_watchlist_updates), job_id="streaming_watchlist_check", save=False, ) self._watchlist_job_id = watchlist_job.job_id pdebug("StreamingPlugin: Watchlist-Check registriert (alle 12h)") except Exception as e: perror(f"StreamingPlugin: 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._weekly_job_id, self._daily_job_id, self._watchlist_job_id): if job_id: try: scheduler.remove_job(job_id) except Exception: pass async def _scheduled_full_sync(self, *args: Any, **kwargs: Any) -> None: """Vom Scheduler aufgerufener woechentlicher Voll-Sync.""" pinfo("StreamingPlugin: Geplanter Voll-Sync gestartet") try: await self._run_sync(full=True) except Exception as e: perror(f"StreamingPlugin: Geplanter Voll-Sync fehlgeschlagen: {e}") async def _scheduled_daily_update(self, *args: Any, **kwargs: Any) -> None: """Vom Scheduler aufgerufenes taegliches Quick-Update.""" pinfo("StreamingPlugin: Taegliches Quick-Update gestartet") try: await self._run_sync(full=False) except Exception as e: perror(f"StreamingPlugin: Taegliches Quick-Update fehlgeschlagen: {e}") # ====================================================================== # STT-Korrektur (Plugin-spezifisch) # ====================================================================== def _load_stt_aliases(self) -> None: """ Laedt STT-Aliases aus stt_aliases.json. Diese Korrekturen werden NUR auf Slots in Streaming-Intents angewendet, nicht global auf alle STT-Ergebnisse. """ 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"StreamingPlugin: {len(self._stt_aliases)} STT-Aliases geladen") except Exception as e: pdebug(f"StreamingPlugin: STT-Aliases Fehler: {e}") def _correct_query(self, text: str) -> str: """ Korrigiert bekannte Vosk-Fehlerkennungen in Slot-Werten. Wird auf query/service Slots angewendet NACHDEM der Intent erkannt wurde — nur in Streaming-Plugin Handlern. Beispiele: "kronchi roll" -> "crunchyroll" "disney plus" -> "disney+" "nach utopie" -> "naruto" """ 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) corrected = corrected.strip() if corrected != text.lower().strip(): pdebug(f"StreamingPlugin: STT-Korrektur '{text}' -> '{corrected}'") return corrected # ====================================================================== # Service-Aufloesung # ====================================================================== def _resolve_service(self, text: str) -> str | None: """ Loest einen gesprochenen Dienstnamen zu einer service_id auf. Args: text: Gesprochener Dienstname (z.B. "Netflix", "Prime", "Sky") Returns: Service-ID (z.B. "netflix", "amazon_prime") oder None """ if not text: return None corrected = self._correct_query(text) normalized = corrected.lower().strip().rstrip("?!. ") # Direkte Zuordnung if normalized in _SERVICE_ALIASES: return _SERVICE_ALIASES[normalized] # Teilstring-Match (z.B. "auf amazon prime video" → "amazon_prime") for alias, service_id in sorted( _SERVICE_ALIASES.items(), key=lambda x: len(x[0]), reverse=True, ): if alias in normalized: return service_id return None def _get_service_display_name(self, service_id: str) -> str: """Gibt den Anzeigenamen eines Streaming-Dienstes zurueck.""" names: dict[str, str] = { "netflix": "Netflix", "amazon_prime": "Amazon Prime", "crunchyroll": "Crunchyroll", "disney_plus": "Disney+", "wow": "WOW", } return names.get(service_id, service_id) # ====================================================================== # Sprach-Erkennung (Dub/Sub) # ====================================================================== @staticmethod def _wants_dub(text: str) -> bool: """Erkennt ob der Nutzer nach deutscher Synchronisation fragt.""" lower = text.lower() return any(kw in lower for kw in _DUB_KEYWORDS) @staticmethod def _wants_sub(text: str) -> bool: """Erkennt ob der Nutzer nach Untertiteln fragt.""" lower = text.lower() return any(kw in lower for kw in _SUB_KEYWORDS) # ====================================================================== # Formatierung # ====================================================================== def _format_title_for_tts(self, title_data: dict[str, Any]) -> str: """ Formatiert einen Titel fuer Sprachausgabe (TTS). Args: title_data: Titel-Daten aus dem Cache Returns: Formatierter String fuer TTS """ name = title_data.get("title", "Unbekannt") media_type = title_data.get("type", "") year = title_data.get("year", "") rating = title_data.get("rating", 0) parts = [name] if year: parts.append(f"von {year}") if media_type == "movie": parts.append("Film") elif media_type == "tv": seasons = title_data.get("seasons", 0) if seasons: parts.append(f"Serie mit {seasons} Staffeln" if seasons > 1 else "Serie mit 1 Staffel") else: parts.append("Serie") if rating and rating > 0: parts.append(f"Bewertung {rating:.1f}") return ", ".join(parts) def _format_title_list(self, titles: list[dict[str, Any]], max_count: int = 5) -> str: """ Formatiert eine Liste von Titeln fuer Sprachausgabe. Args: titles: Liste von Titel-Daten max_count: Maximale Anzahl zu nennender Titel Returns: Formatierter TTS-String """ if not titles: return "Ich habe leider keine passenden Titel gefunden." shown = titles[:max_count] lines = [self._format_title_for_tts(t) for t in shown] result = ". ".join(lines) remaining = len(titles) - max_count if remaining > 0: result += f". Und {remaining} weitere." return result def _format_services_list(self, services: list[str]) -> str: """Formatiert eine Liste von Diensten fuer TTS.""" if not services: return "keinem bekannten Dienst" names = [self._get_service_display_name(s) for s in services] if len(names) == 1: return names[0] return ", ".join(names[:-1]) + f" und {names[-1]}" # ====================================================================== # Hilfsmethoden # ====================================================================== @staticmethod def _no_data() -> IntentResult: """Antwort wenn Cache noch nicht geladen ist.""" return IntentResult.success_with_response( "Die Streaming-Daten werden noch geladen. Bitte versuche es gleich nochmal." ) def _build_context( self, data: IntentReceivedData, title_data: dict[str, Any] | None = None, **extra: Any, ) -> dict[str, Any]: """ Baut Follow-Up-Kontext aus aktueller Abfrage. Der Kontext wird bei follow_up() als data mitgegeben und beim naechsten Turn an den streaming_follow_up Handler weitergereicht. """ ctx: dict[str, Any] = { "ctx_intent": data.intent, "ctx_service": extra.get("service_id", ""), "ctx_genre": extra.get("genre", ""), } if title_data: ctx["ctx_title"] = title_data.get("title", "") ctx["ctx_title_id"] = title_data.get("id", "") ctx["ctx_type"] = title_data.get("type", "") return ctx def _respond_with_followup(self, text: str, ctx: dict[str, Any]) -> IntentResult: """Antwortet und haelt die Konversation offen fuer Nachfragen.""" return IntentResult.follow_up( response_text=text, follow_up_intent="streaming_follow_up", **ctx, ) def _extract_two_services(self, text: str) -> tuple[str | None, str | None]: """ Extrahiert zwei Dienstnamen aus einem Vergleichs-Text. Beispiel: "Netflix oder Crunchyroll" → ("netflix", "crunchyroll") """ lower = text.lower() found: list[str] = [] # Laengere Aliase zuerst pruefen for alias, service_id in sorted( _SERVICE_ALIASES.items(), key=lambda x: len(x[0]), reverse=True, ): if alias in lower and service_id not in found: found.append(service_id) if len(found) >= 2: break if len(found) >= 2: return found[0], found[1] elif len(found) == 1: return found[0], None return None, None # ====================================================================== # Intent-Handler # ====================================================================== @intent("streaming_search", description="Streaming-Katalog nach Genre durchsuchen") @pattern("welche {genre} gibt es auf {service}") @pattern("was fuer {genre} hat {service}") @pattern("{genre} auf {service}") @example("Welche Horrorfilme gibt es auf Netflix?") @example("Was fuer Animes hat Crunchyroll?") async def handle_search( self, data: IntentReceivedData, genre: str = "", service: str = "", ) -> IntentResult: """Welche [Genre]-Filme/Serien gibt es auf [Dienst]?""" if not self._cache: return self._no_data() genre_raw = self._correct_query(data.get_slot("genre", genre)) service_raw = data.get_slot("service", service) service_id = self._resolve_service(service_raw) if not service_id: return IntentResult.success_with_response( f"Ich kenne den Streaming-Dienst '{service_raw}' leider nicht." ) results = self._cache.search_on_service(service_id, genre=genre_raw) service_name = self._get_service_display_name(service_id) if not results: return IntentResult.success_with_response( f"Ich habe keine {genre_raw}-Titel auf {service_name} gefunden." ) response = f"Auf {service_name} gibt es folgende {genre_raw}-Titel: {self._format_title_list(results)}" ctx = self._build_context( data, title_data=results[0] if results else None, service_id=service_id, genre=genre_raw, ) return self._respond_with_followup(response, ctx) @intent("streaming_find", description="Bestimmten Titel auf Streaming-Dienst suchen") @pattern("gibt es {query} auf {service}") @pattern("laeuft {query} auf {service}") @pattern("kann ich {query} auf {service} schauen") @example("Gibt es Naruto auf Crunchyroll?") @example("Laeuft Breaking Bad auf Netflix?") async def handle_find( self, data: IntentReceivedData, query: str = "", service: str = "", ) -> IntentResult: """Gibt es [Titel] auf [Dienst]?""" if not self._cache: return self._no_data() query_raw = self._correct_query(data.get_slot("query", query)) service_raw = data.get_slot("service", service) service_id = self._resolve_service(service_raw) if service_raw else None title_data = self._cache.find_title(query_raw) if not title_data: return IntentResult.success_with_response( f"Ich habe '{query_raw}' leider nicht in meinem Katalog gefunden." ) title_name = title_data.get("title", query_raw) available_services = title_data.get("services", []) # Spezifischer Dienst angefragt? if service_id: service_name = self._get_service_display_name(service_id) if service_id in available_services: response = f"Ja, {title_name} ist auf {service_name} verfuegbar." else: if available_services: alt = self._format_services_list(available_services) response = ( f"Nein, {title_name} ist nicht auf {service_name} verfuegbar. " f"Aber du kannst es auf {alt} schauen." ) else: response = f"Nein, {title_name} ist aktuell nicht auf {service_name} verfuegbar." else: # Kein spezifischer Dienst → alle Dienste auflisten if available_services: svc_list = self._format_services_list(available_services) response = f"{title_name} ist auf {svc_list} verfuegbar." else: response = f"{title_name} ist aktuell auf keinem der unterstuetzten Dienste verfuegbar." ctx = self._build_context(data, title_data=title_data, service_id=service_id or "") return self._respond_with_followup(response, ctx) @intent("streaming_where", description="Auf welchen Diensten ist ein Titel verfuegbar") @pattern("wo kann ich {query} schauen") @pattern("wo kann ich {query} streamen") @pattern("auf welchem streamingdienst gibt es {query}") @example("Wo kann ich Breaking Bad schauen?") @example("Auf welchem Streamingdienst gibt es Naruto?") async def handle_where( self, data: IntentReceivedData, query: str = "", ) -> IntentResult: """Wo kann ich [Titel] schauen?""" if not self._cache: return self._no_data() query_raw = self._correct_query(data.get_slot("query", query)) title_data = self._cache.find_title(query_raw) if not title_data: return IntentResult.success_with_response( f"Ich habe '{query_raw}' leider nicht in meinem Katalog gefunden." ) title_name = title_data.get("title", query_raw) available_services = title_data.get("services", []) if available_services: svc_list = self._format_services_list(available_services) response = f"{title_name} kannst du auf {svc_list} schauen." else: response = ( f"{title_name} ist aktuell auf keinem der unterstuetzten " f"Streaming-Dienste verfuegbar." ) ctx = self._build_context(data, title_data=title_data) return self._respond_with_followup(response, ctx) @intent("streaming_new", description="Neue Titel auf einem Streaming-Dienst") @pattern("was ist neu auf {service}") @pattern("was gibt es neues auf {service}") @pattern("neue filme auf {service}") @example("Was ist neu auf Netflix?") @example("Neue Serien auf Crunchyroll") async def handle_new( self, data: IntentReceivedData, service: str = "", ) -> IntentResult: """Was ist neu auf [Dienst]?""" if not self._cache: return self._no_data() service_raw = data.get_slot("service", service) service_id = self._resolve_service(service_raw) if not service_id: return IntentResult.success_with_response( f"Ich kenne den Streaming-Dienst '{service_raw}' leider nicht." ) results = self._cache.get_recent(service_id) service_name = self._get_service_display_name(service_id) if not results: return IntentResult.success_with_response( f"Ich habe keine neuen Titel auf {service_name} gefunden." ) response = f"Neu auf {service_name}: {self._format_title_list(results)}" ctx = self._build_context( data, title_data=results[0] if results else None, service_id=service_id, ) return self._respond_with_followup(response, ctx) @intent("streaming_genre", description="Streaming-Katalog nach Genre filtern") @pattern("welche {genre} gibt es auf {service}") @pattern("welche {genre} filme hat {service}") @pattern("anime serien auf {service}") @example("Welche Animes gibt es auf Crunchyroll?") @example("Welche Horror Filme hat Netflix?") async def handle_genre( self, data: IntentReceivedData, genre: str = "", service: str = "", ) -> IntentResult: """Welche [Genre] gibt es auf [Dienst]?""" if not self._cache: return self._no_data() genre_raw = self._correct_query(data.get_slot("genre", genre)) service_raw = data.get_slot("service", service) service_id = self._resolve_service(service_raw) if not service_id: return IntentResult.success_with_response( f"Ich kenne den Streaming-Dienst '{service_raw}' leider nicht." ) results = self._cache.get_by_genre(genre_raw, service_id) service_name = self._get_service_display_name(service_id) if not results: return IntentResult.success_with_response( f"Ich habe keine {genre_raw}-Titel auf {service_name} gefunden." ) response = ( f"Auf {service_name} gibt es {len(results)} {genre_raw}-Titel. " f"Hier sind einige: {self._format_title_list(results)}" ) ctx = self._build_context( data, title_data=results[0] if results else None, service_id=service_id, genre=genre_raw, ) return self._respond_with_followup(response, ctx) @intent("streaming_language", description="Titel mit bestimmter Sprache suchen") @pattern("gibt es {query} mit deutscher synchronisation") @pattern("gibt es {query} auf deutsch auf {service}") @pattern("gibt es {query} mit untertiteln auf {service}") @example("Gibt es Naruto mit deutscher Syncro auf Crunchyroll?") @example("Gibt es One Piece mit Untertiteln?") async def handle_language( self, data: IntentReceivedData, query: str = "", service: str = "", ) -> IntentResult: """Gibt es [Titel] auf deutsch / mit Untertiteln?""" if not self._cache: return self._no_data() query_raw = self._correct_query(data.get_slot("query", query)) service_raw = data.get_slot("service", service) service_id = self._resolve_service(service_raw) if service_raw else None original_text = (data.original_text or "").lower() title_data = self._cache.find_title(query_raw) if not title_data: return IntentResult.success_with_response( f"Ich habe '{query_raw}' leider nicht in meinem Katalog gefunden." ) title_name = title_data.get("title", query_raw) wants_dub = self._wants_dub(original_text) wants_sub = self._wants_sub(original_text) # Sprach-Informationen aus den Titel-Daten extrahieren has_german_dub = title_data.get("has_german_dub", False) has_german_sub = title_data.get("has_german_sub", False) parts: list[str] = [] if wants_dub: if has_german_dub: parts.append(f"{title_name} ist mit deutscher Synchronisation verfuegbar") else: parts.append(f"{title_name} ist leider nicht mit deutscher Synchronisation verfuegbar") if wants_sub: if has_german_sub: parts.append(f"{title_name} ist mit deutschen Untertiteln verfuegbar") else: parts.append(f"{title_name} ist leider nicht mit deutschen Untertiteln verfuegbar") if not wants_dub and not wants_sub: # Allgemeine Sprachinfo lang_info: list[str] = [] if has_german_dub: lang_info.append("deutsche Synchronisation") if has_german_sub: lang_info.append("deutsche Untertitel") if lang_info: parts.append(f"{title_name} bietet {' und '.join(lang_info)}") else: parts.append( f"Zu {title_name} habe ich leider keine Sprachinformationen" ) # Dienst-spezifisch? if service_id: service_name = self._get_service_display_name(service_id) available = title_data.get("services", []) if service_id in available: parts.append(f"auf {service_name}") else: parts.append(f"aber nicht auf {service_name}") response = ". ".join(parts) + "." ctx = self._build_context(data, title_data=title_data, service_id=service_id or "") return self._respond_with_followup(response, ctx) @intent("streaming_compare", description="Titel auf zwei Diensten vergleichen") @pattern("ist {query} auf {service} oder {service} besser") @pattern("vergleiche {query} auf {service} und {service}") @example("Ist Naruto auf Netflix oder Crunchyroll besser?") @example("Vergleiche One Piece auf Netflix und Crunchyroll") async def handle_compare( self, data: IntentReceivedData, query: str = "", ) -> IntentResult: """Ist [Titel] auf [Dienst1] oder [Dienst2] besser?""" if not self._cache: return self._no_data() query_raw = self._correct_query(data.get_slot("query", query)) original_text = data.original_text or "" # Zwei Dienste aus dem Text extrahieren service_a, service_b = self._extract_two_services(original_text) if not service_a or not service_b: return IntentResult.success_with_response( "Ich konnte nicht erkennen welche zwei Streaming-Dienste du vergleichen moechtest." ) title_data = self._cache.find_title(query_raw) if not title_data: return IntentResult.success_with_response( f"Ich habe '{query_raw}' leider nicht in meinem Katalog gefunden." ) title_name = title_data.get("title", query_raw) available = title_data.get("services", []) name_a = self._get_service_display_name(service_a) name_b = self._get_service_display_name(service_b) on_a = service_a in available on_b = service_b in available if on_a and on_b: # Beide haben den Titel → Sub/Dub Vergleich dub_info_a = title_data.get(f"dub_{service_a}", False) dub_info_b = title_data.get(f"dub_{service_b}", False) sub_info_a = title_data.get(f"sub_{service_a}", False) sub_info_b = title_data.get(f"sub_{service_b}", False) parts: list[str] = [f"{title_name} ist sowohl auf {name_a} als auch auf {name_b} verfuegbar"] diff_parts: list[str] = [] if dub_info_a and not dub_info_b: diff_parts.append(f"Auf {name_a} gibt es deutsche Synchronisation, auf {name_b} nicht") elif dub_info_b and not dub_info_a: diff_parts.append(f"Auf {name_b} gibt es deutsche Synchronisation, auf {name_a} nicht") if sub_info_a and not sub_info_b: diff_parts.append(f"Auf {name_a} gibt es deutsche Untertitel, auf {name_b} nicht") elif sub_info_b and not sub_info_a: diff_parts.append(f"Auf {name_b} gibt es deutsche Untertitel, auf {name_a} nicht") if diff_parts: parts.extend(diff_parts) else: parts.append("Ich sehe keine grossen Unterschiede zwischen den beiden Anbietern") response = ". ".join(parts) + "." elif on_a and not on_b: response = f"{title_name} ist nur auf {name_a} verfuegbar, nicht auf {name_b}." elif on_b and not on_a: response = f"{title_name} ist nur auf {name_b} verfuegbar, nicht auf {name_a}." else: response = f"{title_name} ist weder auf {name_a} noch auf {name_b} verfuegbar." ctx = self._build_context(data, title_data=title_data, service_id=service_a) return self._respond_with_followup(response, ctx) @intent("streaming_recommend", description="Empfehlungen fuer Genre und Dienst") @pattern("empfiehl mir einen guten {genre} auf {service}") @pattern("empfehle mir etwas auf {service}") @pattern("was empfiehlst du mir auf {service}") @example("Empfiehl mir einen guten Anime auf Crunchyroll") @example("Was soll ich auf Netflix schauen?") async def handle_recommend( self, data: IntentReceivedData, genre: str = "", service: str = "", ) -> IntentResult: """Empfiehl mir [Genre] auf [Dienst].""" if not self._cache: return self._no_data() genre_raw = self._correct_query(data.get_slot("genre", genre)) service_raw = data.get_slot("service", service) service_id = self._resolve_service(service_raw) if not service_id: return IntentResult.success_with_response( f"Ich kenne den Streaming-Dienst '{service_raw}' leider nicht." ) results = self._cache.get_recommendations( genre=genre_raw if genre_raw else None, service_id=service_id, min_rating=7.0, ) service_name = self._get_service_display_name(service_id) if not results: genre_hint = f" im Genre {genre_raw}" if genre_raw else "" return IntentResult.success_with_response( f"Ich habe leider keine passenden Empfehlungen{genre_hint} auf {service_name} gefunden." ) genre_hint = f" {genre_raw}" if genre_raw else "" response = ( f"Hier sind meine{genre_hint}-Empfehlungen auf {service_name}: " f"{self._format_title_list(results)}" ) ctx = self._build_context( data, title_data=results[0] if results else None, service_id=service_id, genre=genre_raw, ) return self._respond_with_followup(response, ctx) # ====================================================================== # Staffeln / Episoden # ====================================================================== @intent("streaming_seasons", description="Staffel-/Episodenanzahl abfragen") @pattern("(wie viele|wieviele) staffeln hat {query} [auf] {service}") @pattern("(wie viele|wieviele) (folgen|episoden) hat {query} [auf] {service}") @example("Wie viele Staffeln hat Detektiv Conan auf Crunchyroll?") @example("Wie viele Episoden hat Naruto?") async def handle_seasons( self, data: IntentReceivedData, query: str = "", service: str = "", ) -> IntentResult: """Staffel- und Episodenanzahl von TMDB abrufen.""" 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 Serie meinst du?") title = self._cache.find_title(query) if not title: return IntentResult.success_with_response( f"Ich habe '{query}' nicht im Streaming-Katalog gefunden." ) tmdb_id = title.get("tmdb_id") title_name = title.get("t", query) # TMDB Serien-Details abrufen (Staffeln/Episoden) if tmdb_id and title.get("ty") == "serie" and self._discover: try: details = await self._discover._get(f"/tv/{tmdb_id}") if details: seasons = details.get("number_of_seasons", 0) episodes = details.get("number_of_episodes", 0) text = data.original_text.lower() if data.original_text else "" wants_episodes = any(w in text for w in ["folge", "episode", "episoden"]) if wants_episodes: response = f"{title_name} hat {episodes} Episoden in {seasons} Staffeln." else: response = f"{title_name} hat {seasons} Staffeln mit insgesamt {episodes} Episoden." # Pruefen ob der Nutzer eine bestimmte Staffel meint import re season_match = re.search(r"(\d+)\.\s*staffel|staffel\s*(\d+)", text) if season_match: season_num = int(season_match.group(1) or season_match.group(2)) season_details = await self._discover._get(f"/tv/{tmdb_id}/season/{season_num}") if season_details: ep_count = len(season_details.get("episodes", [])) season_name = season_details.get("name", f"Staffel {season_num}") response = f"{title_name}, {season_name} hat {ep_count} Episoden." else: response = f"Staffel {season_num} von {title_name} wurde nicht gefunden." ctx = self._build_context(data, title_name=title_name, title_slug=self._cache._generate_slug(title_name)) return self._respond_with_followup(response, ctx) except Exception as e: pdebug(f"StreamingPlugin: TMDB Staffel-Abfrage fehlgeschlagen: {e}") return IntentResult.success_with_response( f"Staffel-Informationen fuer '{title_name}' sind leider nicht verfuegbar." ) # ====================================================================== # Watchlist # ====================================================================== @intent("streaming_watchlist_add", description="Serie auf Watchlist setzen") @pattern("(merke|merk) [dir] {query} [vor]") @pattern("(markiere|markier) {query} [auf der watchlist]") @pattern("setze {query} auf [die] (watchlist|merkliste)") @pattern("benachrichtige mich bei {query}") @example("Merke dir Detektiv Conan vor") @example("Benachrichtige mich wenn es neue Folgen von Naruto gibt") async def handle_watchlist_add( self, data: IntentReceivedData, query: str = "", ) -> IntentResult: """Fuegt eine Serie zur Watchlist hinzu.""" 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 Serie soll ich vormerken?") slug = self._cache._generate_slug(query) title = self._cache.find_title(query) title_name = title.get("t", query) if title else query # Duplikat-Check watchlist = self._cache._data.get("watchlist", []) for entry in watchlist: if entry.get("slug") == slug and entry.get("active"): return IntentResult.success_with_response( f"{title_name} ist bereits auf deiner Watchlist." ) # Aktuelle Staffel-/Episoden-Info merken fuer Aenderungs-Erkennung current_seasons = 0 current_episodes = 0 current_dub: list[str] = [] if title and title.get("tmdb_id") and self._discover: try: details = await self._discover._get(f"/tv/{title['tmdb_id']}") if details: current_seasons = details.get("number_of_seasons", 0) current_episodes = details.get("number_of_episodes", 0) except Exception: pass # Dub-Status merken (fuer Syncro-Benachrichtigung) for svc_data in title.get("on", {}).values(): for lang in svc_data.get("dub", []): if lang not in current_dub: current_dub.append(lang) watchlist.append({ "slug": slug, "title": title_name, "active": True, "satellite_id": data.satellite_id, "added_at": datetime.now().astimezone().isoformat(), "last_seasons": current_seasons, "last_episodes": current_episodes, "last_dub": current_dub, "last_checked": datetime.now().astimezone().isoformat(), }) self._cache._data["watchlist"] = watchlist await self._cache.save_async() response = f"{title_name} wurde zu deiner Watchlist hinzugefuegt." if current_seasons > 0: response += f" Aktuell: {current_seasons} Staffeln, {current_episodes} Episoden." response += " Ich benachrichtige dich bei neuen Folgen oder Synchronisationen." return IntentResult.success_with_response(response) @intent("streaming_watchlist_list", description="Watchlist anzeigen") @pattern("[meine] [streaming] (watchlist|merkliste)") @pattern("welche serien habe ich [markiert|vorgemerkt]") @example("Meine Watchlist") @example("Welche Serien habe ich markiert?") async def handle_watchlist_list(self, data: IntentReceivedData) -> IntentResult: """Listet alle Serien auf der Watchlist.""" if not self._cache: return self._no_data() watchlist = [ w for w in self._cache._data.get("watchlist", []) if w.get("active") ] if not watchlist: return IntentResult.success_with_response( "Deine Streaming-Watchlist ist leer." ) titles = [w["title"] for w in watchlist] return IntentResult.success_with_response( f"Du hast {len(titles)} Serien auf deiner Watchlist: {', '.join(titles)}." ) @intent("streaming_watchlist_remove", description="Serie von Watchlist entfernen") @pattern("(loesche|entferne) {query} [von] [der] (watchlist|merkliste)") @pattern("{query} [von der] (watchlist|merkliste) (loeschen|entfernen|streichen)") @example("Loesche Naruto von meiner Watchlist") @example("Naruto von der Watchlist streichen") async def handle_watchlist_remove( self, data: IntentReceivedData, query: str = "", ) -> IntentResult: """Entfernt eine Serie von der Watchlist.""" 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 Serie?") slug = self._cache._generate_slug(query) watchlist = self._cache._data.get("watchlist", []) found = False for entry in watchlist: if entry.get("slug") == slug and entry.get("active"): entry["active"] = False found = True if found: await self._cache.save_async() return IntentResult.success_with_response( f"{query} wurde von deiner Watchlist entfernt." ) return IntentResult.success_with_response( f"{query} ist nicht auf deiner Watchlist." ) # ====================================================================== # Watchlist-Check (Scheduler-Job) # ====================================================================== async def _check_watchlist_updates(self, *args, **kwargs) -> None: """ Prueft Watchlist-Serien auf neue Episoden und Synchronisationen. Wird taeglich vom Scheduler aufgerufen. Vergleicht aktuelle TMDB-Daten mit dem letzten bekannten Stand und benachrichtigt den Nutzer per TTS wenn sich etwas geaendert hat. """ if not self._cache or not self._discover: return watchlist = [ w for w in self._cache._data.get("watchlist", []) if w.get("active") ] if not watchlist: return pdebug(f"StreamingPlugin: Pruefe {len(watchlist)} Watchlist-Eintraege...") for entry in watchlist: slug = entry.get("slug", "") title_name = entry.get("title", slug) title = self._cache.find_title(title_name) if not title or not title.get("tmdb_id"): continue tmdb_id = title["tmdb_id"] notifications: list[str] = [] # Staffeln/Episoden pruefen try: details = await self._discover._get(f"/tv/{tmdb_id}") if details: new_seasons = details.get("number_of_seasons", 0) new_episodes = details.get("number_of_episodes", 0) old_seasons = entry.get("last_seasons", 0) old_episodes = entry.get("last_episodes", 0) if new_seasons > old_seasons: diff = new_seasons - old_seasons notifications.append( f"{diff} neue Staffel{'n' if diff > 1 else ''}" ) elif new_episodes > old_episodes: diff = new_episodes - old_episodes notifications.append( f"{diff} neue Episode{'n' if diff > 1 else ''}" ) entry["last_seasons"] = new_seasons entry["last_episodes"] = new_episodes except Exception: pass # Dub-Status pruefen (neue Synchronisationen) current_dub: list[str] = [] for svc_data in title.get("on", {}).values(): for lang in svc_data.get("dub", []): if lang not in current_dub: current_dub.append(lang) old_dub = entry.get("last_dub", []) new_dub_langs = [l for l in current_dub if l not in old_dub] if new_dub_langs: lang_names = {"de": "Deutsch", "en": "Englisch", "ja": "Japanisch"} lang_str = ", ".join(lang_names.get(l, l) for l in new_dub_langs) notifications.append(f"neue Synchronisation: {lang_str}") entry["last_dub"] = current_dub entry["last_checked"] = datetime.now().astimezone().isoformat() # Benachrichtigung senden if notifications: text = f"Neuigkeiten fuer {title_name}: {', '.join(notifications)}." events = getattr(self.application, "events", None) if events: await events.emit("tts_request", { "request_id": f"streaming-watchlist-{slug}", "satellite_id": entry.get("satellite_id", ""), "text": text, }) pinfo(f"StreamingPlugin: Watchlist-Update — {text}") await self._cache.save_async() # ====================================================================== # Follow-Up Handler # ====================================================================== @intent("streaming_follow_up", description="Streaming-Nachfrage in laufender Konversation") async def handle_follow_up(self, data: IntentReceivedData) -> IntentResult: """ Verarbeitet Nachfragen im Konversations-Kontext. Erkennt Muster wie: - "Und auf Netflix?" → gleicher Titel, anderer Dienst - "Gibt es das auch auf deutsch?" → Sprachinfo zum letzten Titel - "Erzaehl mir mehr" → Details zum letzten Titel """ if not self._cache: return self._no_data() text = (data.original_text or "").strip().lower() ctx_title = data.get_slot("ctx_title", "") ctx_service = data.get_slot("ctx_service", "") ctx_genre = data.get_slot("ctx_genre", "") ctx_intent = data.get_slot("ctx_intent", "") # --- "Und auf Netflix?" → Dienst-Wechsel --- new_service = self._extract_service_from_followup(text) if new_service and ctx_title: title_data = self._cache.find_title(ctx_title) if title_data: service_name = self._get_service_display_name(new_service) available = title_data.get("services", []) title_name = title_data.get("title", ctx_title) if new_service in available: response = f"Ja, {title_name} ist auch auf {service_name} verfuegbar." else: response = f"Nein, {title_name} ist nicht auf {service_name} verfuegbar." ctx = self._build_context(data, title_data=title_data, service_id=new_service) return self._respond_with_followup(response, ctx) # --- "Gibt es das auch auf deutsch?" / "mit Untertiteln?" --- if ctx_title and (self._wants_dub(text) or self._wants_sub(text)): title_data = self._cache.find_title(ctx_title) if title_data: data.slots["query"] = ctx_title return await self.handle_language(data, query=ctx_title) # --- "Erzaehl mir mehr" / "Worum geht es?" --- if self._matches_info(text) and ctx_title: title_data = self._cache.find_title(ctx_title) if title_data: overview = title_data.get("overview", "") title_name = title_data.get("title", ctx_title) if overview: response = f"{title_name}: {overview}" else: response = f"Zu {title_name} habe ich leider keine Beschreibung." ctx = self._build_context(data, title_data=title_data, service_id=ctx_service) return self._respond_with_followup(response, ctx) # --- Nichts erkannt → Konversation beenden --- return IntentResult.success_with_response( "Das habe ich leider nicht verstanden." ) # ====================================================================== # Follow-Up Hilfsmethoden # ====================================================================== def _extract_service_from_followup(self, text: str) -> str | None: """Extrahiert einen Dienstnamen aus Follow-Up-Text.""" # "und auf Netflix?" / "auf Amazon Prime?" / "und Crunchyroll?" cleaned = text for prefix in ["und auf ", "auf ", "und ", "gibt es das auch auf ", "auch auf "]: if cleaned.startswith(prefix): cleaned = cleaned[len(prefix):] break cleaned = cleaned.rstrip("?!. ") if cleaned: return self._resolve_service(cleaned) return None @staticmethod def _matches_info(text: str) -> bool: """Erkennt Info-Absicht in Follow-Up.""" keywords = [ "erzaehl", "erzaehl mir", "mehr darueber", "mehr dazu", "mehr infos", "mehr informationen", "worum geht", "was ist das", "beschreibung", "info", ] return any(kw in text for kw in keywords)