""" Local web server for managing Stremio addon sources from a phone/browser. Serves a web UI and exposes small JSON endpoints for CRUD operations on the Stremio custom addons stored in the Kodi addon cache. """ import ipaddress import json import os import socket import threading from datetime import timedelta from http.server import BaseHTTPRequestHandler, HTTPServer from urllib.parse import urlsplit, urlunsplit import requests from lib.api.stremio.addon_manager import build_addon_instance_key from lib.clients.stremio.constants import ( STREMIO_ADDONS_CATALOGS_KEY, STREMIO_ADDONS_KEY, STREMIO_TV_ADDONS_KEY, STREMIO_USER_ADDONS, decode_selected_ids, encode_selected_ids, ) from lib.db.cached import cache from lib.utils.general.utils import USER_AGENT_HEADER from lib.utils.kodi.utils import kodilog _CACHE_EXPIRY = timedelta(days=365 * 20) _WEB_DIR = os.path.join( os.path.dirname(os.path.dirname(os.path.dirname(os.path.abspath(__file__)))), "resources", "web", ) # --------------------------------------------------------------------------- # Sync helpers (no xbmcgui imports — safe for HTTP thread) # --------------------------------------------------------------------------- def _get_user_addons(): """Return list of custom addon dicts from cache.""" return [a for a in (cache.get(STREMIO_USER_ADDONS) or []) if a.get("transportName") == "custom"] def _get_selected_ids(): """Return dict of selected addon IDs per category.""" result = {} for key, label in [ (STREMIO_ADDONS_KEY, "stream"), (STREMIO_ADDONS_CATALOGS_KEY, "catalog"), (STREMIO_TV_ADDONS_KEY, "tv"), ]: result[label] = decode_selected_ids(cache.get(key)) return result def _addon_capabilities(manifest): """Determine stream/catalog/tv capabilities from a manifest dict.""" def supports_searchable_stream(prefixes): normalized = {str(prefix).rstrip(":") for prefix in (prefixes or []) if prefix} return bool(normalized.intersection({"tt", "tmdb"})) resources = manifest.get("resources", []) types = manifest.get("types", []) manifest_prefixes = manifest.get("idPrefixes", []) is_stream = False is_catalog = False is_tv = False for res in resources: if isinstance(res, dict): res_name = res.get("name") res_types = res.get("types", types) res_prefixes = res.get("idPrefixes", manifest_prefixes) if res_name == "stream": # Stremio addon providing streams if ( any(t in res_types for t in ("movie", "series", "anime")) and supports_searchable_stream(res_prefixes) ) or supports_searchable_stream(res_prefixes): is_stream = True if any(t in res_types for t in ("tv", "channel")): is_tv = True if res_name == "catalog": is_catalog = True elif isinstance(res, str): if res == "stream": if any( t in types for t in ("movie", "series", "anime") ) and supports_searchable_stream(manifest_prefixes): is_stream = True if any(t in types for t in ("tv", "channel")): is_tv = True if res == "catalog": is_catalog = True return {"stream": is_stream, "catalog": is_catalog, "tv": is_tv} def _normalize_manifest_url(url): normalized = url.strip() if not normalized: return "" if normalized.startswith("stremio://"): normalized = normalized.replace("stremio://", "https://", 1) parts = urlsplit(normalized) path = parts.path or "" if not path.endswith("/manifest.json"): if path.endswith("/"): path = f"{path}manifest.json" else: path = f"{path}/manifest.json" return urlunsplit((parts.scheme, parts.netloc, path, parts.query, parts.fragment)) def _is_local_or_private_host(url): host = (urlsplit(url).hostname or "").lower() if not host: return False if host == "localhost": return True try: ip = ipaddress.ip_address(host) return ip.is_private or ip.is_loopback except ValueError: return False def _fetch_manifest(url, initial_url): def try_fetch(target_url): try: kodilog(f"[WebServer] Fetching Stremio manifest: {target_url}") resp = requests.get(target_url, headers=USER_AGENT_HEADER, timeout=10) resp.raise_for_status() return resp, resp.json(), None except Exception as ex: return None, None, str(ex) resp, manifest, error = try_fetch(url) if error and url.startswith("https://"): if initial_url.startswith("stremio://") or _is_local_or_private_host(url): http_url = url.replace("https://", "http://", 1) kodilog(f"[WebServer] HTTPS failed, trying HTTP fallback: {http_url}") r2, m2, e2 = try_fetch(http_url) if not e2: return r2, m2, None return resp, manifest, error def _sync_add_addon(url): """Fetch manifest from URL, validate, store in cache. Returns (addon_dict, error).""" url = (url or "").strip() if not url: return None, "Empty URL provided." initial_url = url url = _normalize_manifest_url(url) resp, manifest, error = _fetch_manifest(url, initial_url) if error or not resp or not manifest: kodilog(f"[WebServer] Failed to add addon: {error}") return None, f"Failed to fetch manifest: {error or 'unknown error'}" addon_key = build_addon_instance_key({"manifest": manifest, "transportUrl": resp.url}) # Check duplicate user_addons = list(cache.get(STREMIO_USER_ADDONS) or []) if any(build_addon_instance_key(a) == addon_key for a in user_addons): return None, "This addon is already added." caps = _addon_capabilities(manifest) # Auto-select into categories if caps["stream"]: _add_to_selection(STREMIO_ADDONS_KEY, addon_key) if caps["catalog"]: _add_to_selection(STREMIO_ADDONS_CATALOGS_KEY, addon_key) if caps["tv"]: _add_to_selection(STREMIO_TV_ADDONS_KEY, addon_key) custom_addon = { "manifest": manifest, "transportUrl": resp.url, "transportName": "custom", } user_addons.append(custom_addon) cache.set(STREMIO_USER_ADDONS, user_addons, _CACHE_EXPIRY) return custom_addon, None def _sync_remove_addon(addon_id): """Remove addon by manifest ID from all caches.""" user_addons = list(cache.get(STREMIO_USER_ADDONS) or []) new_addons = [a for a in user_addons if build_addon_instance_key(a) != addon_id] cache.set(STREMIO_USER_ADDONS, new_addons, _CACHE_EXPIRY) for cache_key in [ STREMIO_ADDONS_KEY, STREMIO_ADDONS_CATALOGS_KEY, STREMIO_TV_ADDONS_KEY, ]: selected = cache.get(cache_key) if selected: keys = decode_selected_ids(selected) keys = [k for k in keys if k != addon_id] cache.set(cache_key, encode_selected_ids(keys), _CACHE_EXPIRY) def _sync_toggle_addon(addon_id, category, enabled): """Add or remove addon_id from a selection category.""" key_map = { "stream": STREMIO_ADDONS_KEY, "catalog": STREMIO_ADDONS_CATALOGS_KEY, "tv": STREMIO_TV_ADDONS_KEY, } cache_key = key_map.get(category) if not cache_key: return keys = list(decode_selected_ids(cache.get(cache_key))) if enabled and addon_id not in keys: keys.append(addon_id) elif not enabled and addon_id in keys: keys.remove(addon_id) cache.set(cache_key, encode_selected_ids(keys), _CACHE_EXPIRY) def _add_to_selection(cache_key, addon_id): keys = list(decode_selected_ids(cache.get(cache_key))) if addon_id not in keys: keys.append(addon_id) cache.set(cache_key, encode_selected_ids(keys), _CACHE_EXPIRY) def _validate_manifest(url): """Fetch and validate a manifest URL. Returns (manifest_dict, error).""" url = (url or "").strip() if not url: return None, "Empty URL" initial_url = url url = _normalize_manifest_url(url) kodilog(f"[WebServer] Validating Stremio manifest: {url}") _, manifest, error = _fetch_manifest(url, initial_url) if error or not manifest: kodilog(f"[WebServer] Manifest validation failed: {error}") return None, str(error or "unknown error") if not manifest.get("id") and not manifest.get("name"): return None, "Manifest missing 'id' and 'name'." return manifest, None # --------------------------------------------------------------------------- # HTTP request handler # --------------------------------------------------------------------------- class StremioRequestHandler(BaseHTTPRequestHandler): """Handles web UI + JSON API requests.""" def log_message(self, format, *args): kodilog(f"[WebServer] {format % args}") def _send_json(self, data, status=200): body = json.dumps(data).encode("utf-8") self.send_response(status) self.send_header("Content-Type", "application/json") self.send_header("Content-Length", str(len(body))) self.send_header("Access-Control-Allow-Origin", "*") self.end_headers() self.wfile.write(body) def _send_html(self, html_content): body = html_content.encode("utf-8") self.send_response(200) self.send_header("Content-Type", "text/html; charset=utf-8") self.send_header("Content-Length", str(len(body))) self.end_headers() self.wfile.write(body) def _read_body(self): length = int(self.headers.get("Content-Length", 0)) return json.loads(self.rfile.read(length)) if length else {} # --- Routes --- def do_GET(self): if self.path == "/" or self.path == "": self._serve_index() elif self.path == "/api/addons": self._api_get_addons() else: self.send_error(404) def do_POST(self): if self.path == "/api/addons": self._api_add_addon() elif self.path == "/api/addons/validate": self._api_validate() else: self.send_error(404) def do_DELETE(self): if self.path.startswith("/api/addons/"): addon_id = self.path[len("/api/addons/") :] self._api_remove_addon(addon_id) else: self.send_error(404) def do_PUT(self): if self.path.startswith("/api/addons/") and self.path.endswith("/toggle"): addon_id = self.path[len("/api/addons/") : -len("/toggle")] self._api_toggle_addon(addon_id) else: self.send_error(404) def do_OPTIONS(self): self.send_response(200) self.send_header("Access-Control-Allow-Origin", "*") self.send_header("Access-Control-Allow-Methods", "GET, POST, PUT, DELETE, OPTIONS") self.send_header("Access-Control-Allow-Headers", "Content-Type") self.end_headers() # --- Handlers --- def _serve_index(self): index_path = os.path.join(_WEB_DIR, "index.html") try: with open(index_path, encoding="utf-8") as f: self._send_html(f.read()) except FileNotFoundError: self.send_error(500, "Web UI not found") def _api_get_addons(self): addons = _get_user_addons() selected = _get_selected_ids() result = [] for a in addons: manifest = a.get("manifest", {}) transport_url = a.get("transportUrl", "") addon_key = build_addon_instance_key(a) caps = _addon_capabilities(manifest) logo = manifest.get("logo", "") if logo and transport_url and not logo.startswith(("http://", "https://")): from urllib.parse import urljoin logo = urljoin(transport_url, logo) result.append( { "id": addon_key, "name": manifest.get("name", "Unknown"), "description": manifest.get("description", ""), "logo": logo, "version": manifest.get("version", ""), "types": manifest.get("types", []), "transportUrl": transport_url, "capabilities": caps, "selected": { "stream": addon_key in selected.get("stream", []), "catalog": addon_key in selected.get("catalog", []), "tv": addon_key in selected.get("tv", []), }, } ) self._send_json({"addons": result}) def _api_add_addon(self): try: data = self._read_body() except Exception: self._send_json({"error": "Invalid JSON"}, 400) return url = data.get("url", "").strip() if not url: self._send_json({"error": "URL is required"}, 400) return addon, error = _sync_add_addon(url) if error or not addon: self._send_json({"error": error}, 400) return manifest = addon.get("manifest", {}) transport_url = addon.get("transportUrl", "") addon_key = build_addon_instance_key(addon) caps = _addon_capabilities(manifest) logo = manifest.get("logo", "") if logo and transport_url and not logo.startswith(("http://", "https://")): from urllib.parse import urljoin logo = urljoin(transport_url, logo) self._send_json( { "success": True, "addon": { "id": addon_key, "name": manifest.get("name", "Unknown"), "description": manifest.get("description", ""), "logo": logo, "version": manifest.get("version", ""), "types": manifest.get("types", []), "transportUrl": addon.get("transportUrl", ""), "capabilities": caps, "selected": caps, }, } ) def _api_remove_addon(self, addon_id): from urllib.parse import unquote addon_id = unquote(addon_id) _sync_remove_addon(addon_id) self._send_json({"success": True}) def _api_toggle_addon(self, addon_id): from urllib.parse import unquote addon_id = unquote(addon_id) try: data = self._read_body() except Exception: self._send_json({"error": "Invalid JSON"}, 400) return category = data.get("category") enabled = data.get("enabled", False) if category not in ("stream", "catalog", "tv"): self._send_json({"error": "Invalid category"}, 400) return _sync_toggle_addon(addon_id, category, enabled) self._send_json({"success": True}) def _api_validate(self): try: data = self._read_body() except Exception: self._send_json({"error": "Invalid JSON"}, 400) return url = data.get("url", "").strip() if not url: self._send_json({"error": "URL is required"}, 400) return manifest, error = _validate_manifest(url) if error: self._send_json({"error": error}, 400) return self._send_json({"valid": True, "manifest": manifest}) # --------------------------------------------------------------------------- # Server lifecycle # --------------------------------------------------------------------------- class StremioWebServer: """Threaded HTTP server that starts/stops on demand.""" def __init__(self, port=8081): self.port = port self._server = None self._thread = None def start(self): if self._server: return try: self._server = HTTPServer(("0.0.0.0", self.port), StremioRequestHandler) self._thread = threading.Thread(target=self._server.serve_forever, daemon=True) self._thread.start() kodilog(f"[WebServer] Started on port {self.port}") except Exception as e: kodilog(f"[WebServer] Failed to start: {e}") self._server = None def stop(self): if self._server: self._server.shutdown() self._server.server_close() self._server = None self._thread = None kodilog("[WebServer] Stopped") @property def is_running(self): return self._server is not None def get_local_ip(): """Get the machine's LAN IP address.""" try: s = socket.socket(socket.AF_INET, socket.SOCK_DGRAM) s.connect(("8.8.8.8", 80)) ip = s.getsockname()[0] s.close() return ip except Exception: return "127.0.0.1"