itdrui.de
7b60b00c8b
Core & Architektur: - Neues Verzeichnis addon/core/ mit router.py, trakt.py, metadata.py, gui.py, playstate.py, plugin_manager.py, updater.py - Tests-Verzeichnis hinzugefügt (24 Tests, pytest + Coverage) Trakt-Integration: - OAuth Device Flow, Scrobbling, Watchlist, History, Calendar - Upcoming Episodes, Weiterschauen (Continue Watching) - Watched-Status in Episodenlisten - _trakt_find_in_plugins() mit 5-Min-Cache Serienstream-Suche: - API-Ergebnisse werden immer mit Katalog-Cache ergänzt (serverseitiges 10-Treffer-Limit) - Katalog-Cache wird beim Addon-Start im Daemon-Thread vorgewärmt - Notification nach Cache-Load via xbmc.executebuiltin() (thread-sicher) Bugfixes (Code-Review): - Race Condition auf _TRAKT_WATCHED_CACHE: _TRAKT_WATCHED_CACHE_LOCK hinzugefügt - GUI-Dialog aus Daemon-Thread: xbmcgui -> xbmc.executebuiltin() - ValueError in Trakt-Watchlist-Routen abgesichert - Token expires_at==0 Check korrigiert - get_setting_bool() Kontrollfluss in gui.py bereinigt - topstreamfilm_plugin: try-finally um xbmcvfs.File.close() Cleanup: - default.py.bak und refactor_router.py entfernt - .gitignore: /tests/ Eintrag entfernt - Type-Hints vereinheitlicht (Dict/List/Tuple -> dict/list/tuple)
59 lines
2.2 KiB
Python
59 lines
2.2 KiB
Python
from __future__ import annotations
|
|
import sys
|
|
from typing import Any, Callable, Dict, Optional
|
|
from urllib.parse import parse_qs
|
|
|
|
|
|
class Router:
|
|
"""A simple router for Kodi add-ons."""
|
|
|
|
def __init__(self) -> None:
|
|
self._routes: Dict[str, Callable[[Dict[str, str]], Any]] = {}
|
|
self._fallback: Optional[Callable[[Dict[str, str]], Any]] = None
|
|
|
|
def route(self, action: str) -> Callable[[Callable[[Dict[str, str]], Any]], Callable[[Dict[str, str]], Any]]:
|
|
"""Decorator to register a function for a specific action."""
|
|
def decorator(handler: Callable[[Dict[str, str]], Any]) -> Callable[[Dict[str, str]], Any]:
|
|
self._routes[action] = handler
|
|
return handler
|
|
return decorator
|
|
|
|
def fallback(self) -> Callable[[Callable[[Dict[str, str]], Any]], Callable[[Dict[str, str]], Any]]:
|
|
"""Decorator to register the fallback (default) handler."""
|
|
def decorator(handler: Callable[[Dict[str, str]], Any]) -> Callable[[Dict[str, str]], Any]:
|
|
self._fallback = handler
|
|
return handler
|
|
return decorator
|
|
|
|
def dispatch(self, action: Optional[str] = None, params: Optional[Dict[str, str]] = None) -> Any:
|
|
"""Dispatch the request to the registered handler."""
|
|
if params is None:
|
|
params = {}
|
|
|
|
handler = self._routes.get(action) if action else self._fallback
|
|
if not handler:
|
|
handler = self._fallback
|
|
|
|
if handler:
|
|
return handler(params)
|
|
|
|
raise KeyError(f"No route or fallback defined for action: {action}")
|
|
|
|
|
|
def parse_params(argv: Optional[list[str]] = None) -> dict[str, str]:
|
|
"""Parst Kodi-Plugin-Parameter aus `sys.argv[2]` oder der übergebenen Liste."""
|
|
if argv is None:
|
|
argv = sys.argv
|
|
if len(argv) <= 2 or not argv[2]:
|
|
return {}
|
|
raw_params = parse_qs(argv[2].lstrip("?"), keep_blank_values=True)
|
|
return {key: values[0] for key, values in raw_params.items()}
|
|
|
|
|
|
def parse_positive_int(value: str, *, default: int = 1) -> int:
|
|
try:
|
|
parsed = int(value)
|
|
return parsed if parsed > 0 else default
|
|
except (ValueError, TypeError):
|
|
return default
|