diff --git a/Dockerfile b/Dockerfile index f7ccd8d..200b586 100644 --- a/Dockerfile +++ b/Dockerfile @@ -108,9 +108,11 @@ RUN if [ ! -f /config/config.json ]; then \ # Expose port EXPOSE ${ULTIMATE_PORT} -# Health check -HEALTHCHECK --interval=30s --timeout=10s --start-period=10s --retries=3 \ - CMD curl -f http://localhost:${ULTIMATE_PORT}/api/providers || exit 1 +# Health check: /health is a pure liveness probe (no provider/session/disk +# access). /api/providers does real work, so a busy or slow request made the +# probe time out and the container look dead. +HEALTHCHECK --interval=30s --timeout=5s --start-period=30s --retries=3 \ + CMD curl -fsS http://localhost:${ULTIMATE_PORT}/health || exit 1 # Labels LABEL maintainer="nirvana-7777" \ diff --git a/lib/streaming_providers/base/auth/session_manager.py b/lib/streaming_providers/base/auth/session_manager.py index 27fa5e9..f995e16 100644 --- a/lib/streaming_providers/base/auth/session_manager.py +++ b/lib/streaming_providers/base/auth/session_manager.py @@ -1,4 +1,5 @@ # streaming_providers/base/auth/session_manager.py +import functools import time import uuid from typing import Any, Dict, Optional @@ -14,6 +15,18 @@ from .base_auth import BaseAuthToken _SENSITIVE_KEY_PARTS = ("token", "secret", "password", "jwt", "serial", "device_number") +def _locked(method): + """Run a read-modify-write method while holding session.json's process-wide + lock, so concurrent request threads cannot overwrite each other's changes.""" + + @functools.wraps(method) + def wrapper(self, *args, **kwargs): + with self._file_lock: + return method(self, *args, **kwargs) + + return wrapper + + class SessionManager: """ Manages persistent session data including tokens and device IDs @@ -41,6 +54,9 @@ class SessionManager: # Ensure base directory exists self.vfs.mkdirs("") + # Shared by every SessionManager/VFS instance that targets this file + self._file_lock = self.vfs.file_lock(self.session_file) + logger.debug(f"SessionManager initialized with VFS base: {self.vfs.base_path}") logger.debug(f"Session file: {self.vfs.join_path(self.session_file)}") @@ -107,6 +123,7 @@ class SessionManager: logger.error(f"Error loading session for {provider}{country_str}: {e}") return None + @_locked def save_session( self, provider: str, session_data: Dict[str, Any], country: Optional[str] = None ) -> bool: @@ -210,6 +227,7 @@ class SessionManager: logger.error(f"Full traceback: {traceback.format_exc()}") return False + @_locked def save_scoped_token( self, provider: str, @@ -432,6 +450,7 @@ class SessionManager: # No refresh token or no expiration info return False + @_locked def save_token( self, provider: str, token: BaseAuthToken, country: Optional[str] = None ) -> bool: @@ -527,6 +546,7 @@ class SessionManager: return device_id + @_locked def clear_session(self, provider: str, country: Optional[str] = None) -> bool: """Clear session data for a provider and optional country""" country_str = f" (country: {country})" if country else "" @@ -569,6 +589,7 @@ class SessionManager: logger.error(f"Error clearing session for {provider}{country_str}: {e}") return False + @_locked def clear_scoped_token(self, provider: str, scope: str, country: Optional[str] = None) -> bool: """ Clear token for a specific scope @@ -601,6 +622,7 @@ class SessionManager: logger.error(f"Error clearing scoped token for {provider}{country_str}/{scope}: {e}") return False + @_locked def clear_token(self, provider: str, country: Optional[str] = None) -> bool: """Clear only token data but keep other session data (legacy compatibility)""" country_str = f" (country: {country})" if country else "" diff --git a/lib/streaming_providers/base/errors.py b/lib/streaming_providers/base/errors.py index 1ae6069..9ace821 100644 --- a/lib/streaming_providers/base/errors.py +++ b/lib/streaming_providers/base/errors.py @@ -147,6 +147,40 @@ class SessionExpiredError(AuthError): """Server-side session is gone; a full re-login is required.""" +class AuthPendingError(AuthError): + """ + An interactive login (QR / device code) has been started and is waiting + for the user. Not a failure: the caller should show the code and retry. + + Subclasses AuthError so generic "needs login" handling still applies; + HTTP layers should check for it first and answer 202 with the payload. + + Attributes: + login_code: short code for manual entry + qr_target_url: URL to encode in a QR code (may be None) + remaining_seconds: time left before the code expires + retry_after: suggested seconds before the caller retries + """ + + def __init__( + self, + message: str = "Interactive login pending", + *, + login_code: Optional[str] = None, + qr_target_url: Optional[str] = None, + remaining_seconds: Optional[int] = None, + retry_after: Optional[int] = None, + status: Optional[int] = None, + url: Optional[str] = None, + code: Optional[str] = None, + ) -> None: + super().__init__(message, status=status, url=url, code=code) + self.login_code = login_code + self.qr_target_url = qr_target_url + self.remaining_seconds = remaining_seconds + self.retry_after = retry_after + + # --------------------------------------------------------------------------- # Access control # --------------------------------------------------------------------------- diff --git a/lib/streaming_providers/base/http_errors.py b/lib/streaming_providers/base/http_errors.py new file mode 100644 index 0000000..46c6915 --- /dev/null +++ b/lib/streaming_providers/base/http_errors.py @@ -0,0 +1,103 @@ +# streaming_providers/base/http_errors.py +""" +Transport-agnostic mapping from the shared provider error hierarchy +(base/errors.py) to an HTTP status and a JSON-serialisable body. + +Kept free of any web-framework import so it can be unit-tested and reused; +the route layer applies the result to its own response object. + +Rules: + * Errors that tell the CLIENT something actionable (auth, entitlement, + not found, bad request, geo-block, rate limit, pending login) echo the + provider's message. + * Upstream/server-side failures and unknown exceptions return a generic + message. Their detail (which can contain URLs with tokens) is for the + log only - see `log_detail`. +""" + +from __future__ import annotations + +from typing import Any, Dict, Optional, Tuple + +from .errors import ( + AccountRestrictedError, + AuthError, + AuthPendingError, + BadRequestError, + CatchupRequiredError, + ConfigurationError, + EntitlementError, + GeoBlockError, + NotFoundError, + NotImplementedYetError, + OperationFailedError, + PlaybackRestrictedError, + ProviderError, + RateLimitError, + ServerError, +) + +# Most specific first: the first isinstance() match wins. +_STATUS_BY_TYPE: Tuple[Tuple[type, int, bool], ...] = ( + # (exception type, HTTP status, echo provider message to client?) + (AuthPendingError, 202, True), + (AuthError, 401, True), + (GeoBlockError, 451, True), + (AccountRestrictedError, 403, True), + (EntitlementError, 403, True), + (PlaybackRestrictedError, 403, True), + (NotFoundError, 404, True), + (BadRequestError, 400, True), + (RateLimitError, 429, True), + (CatchupRequiredError, 409, True), + (NotImplementedYetError, 501, True), # also UnsupportedOperationError + (ConfigurationError, 503, True), + (ServerError, 502, False), # also TransportError + (OperationFailedError, 502, False), + (ProviderError, 502, False), +) + +GENERIC_UPSTREAM_MESSAGE = "Upstream provider error" +GENERIC_INTERNAL_MESSAGE = "Internal server error" + + +def error_to_http(exc: BaseException) -> Tuple[int, Dict[str, Any], Dict[str, str]]: + """ + Returns (status, body, headers) for an exception. + + A pending login is a normal outcome, so it maps to 202 with everything a + client needs to display the code, plus a Retry-After header. + """ + headers: Dict[str, str] = {} + + if isinstance(exc, AuthPendingError): + body: Dict[str, Any] = { + "status": "login_pending", + "login_code": exc.login_code, + "qr_target_url": exc.qr_target_url, + "remaining_seconds": exc.remaining_seconds, + } + if exc.retry_after: + headers["Retry-After"] = str(int(exc.retry_after)) + return 202, body, headers + + for exc_type, status, echo in _STATUS_BY_TYPE: + if isinstance(exc, exc_type): + message = str(exc) if echo else GENERIC_UPSTREAM_MESSAGE + body = {"error": message, "error_type": type(exc).__name__} + code = getattr(exc, "code", None) + if code is not None: + body["code"] = code + if isinstance(exc, RateLimitError) and exc.retry_after: + headers["Retry-After"] = str(int(exc.retry_after)) + return status, body, headers + + return 500, {"error": GENERIC_INTERNAL_MESSAGE}, headers + + +def log_detail(exc: BaseException, context: str) -> Optional[str]: + """Message for the server log, or None if nothing should be logged + (a pending login is not an error).""" + if isinstance(exc, AuthPendingError): + return None + return f"{context}: {type(exc).__name__}: {exc}" \ No newline at end of file diff --git a/lib/streaming_providers/base/settings/provider_settings_store.py b/lib/streaming_providers/base/settings/provider_settings_store.py index 8cad7ca..79f45f4 100644 --- a/lib/streaming_providers/base/settings/provider_settings_store.py +++ b/lib/streaming_providers/base/settings/provider_settings_store.py @@ -32,23 +32,25 @@ class ProviderSettingsStore: return country_data.get(key, default) def set(self, provider: str, country: str, key: str, value: Any) -> bool: - data = self.vfs.read_json(self.FILE) or {} - provider_data = data.get(provider) - if not isinstance(provider_data, dict): - provider_data = data[provider] = {} - country_data = provider_data.get(country) - if not isinstance(country_data, dict): - country_data = provider_data[country] = {} - country_data[key] = value - ok = self.vfs.write_json(self.FILE, data) + with self.vfs.file_lock(self.FILE): + data = self.vfs.read_json(self.FILE) or {} + provider_data = data.get(provider) + if not isinstance(provider_data, dict): + provider_data = data[provider] = {} + country_data = provider_data.get(country) + if not isinstance(country_data, dict): + country_data = provider_data[country] = {} + country_data[key] = value + ok = self.vfs.write_json(self.FILE, data) if not ok: logger.error(f"Failed to write {self.FILE}") return ok def clear(self, provider: str, country: str, key: str) -> bool: - data = self.vfs.read_json(self.FILE) or {} - country_data = (data.get(provider) or {}).get(country) - if isinstance(country_data, dict) and key in country_data: - del country_data[key] - return self.vfs.write_json(self.FILE, data) - return True \ No newline at end of file + with self.vfs.file_lock(self.FILE): + data = self.vfs.read_json(self.FILE) or {} + country_data = (data.get(provider) or {}).get(country) + if isinstance(country_data, dict) and key in country_data: + del country_data[key] + return self.vfs.write_json(self.FILE, data) + return True \ No newline at end of file diff --git a/lib/streaming_providers/base/utils/vfs.py b/lib/streaming_providers/base/utils/vfs.py index 1ee758c..63a8413 100644 --- a/lib/streaming_providers/base/utils/vfs.py +++ b/lib/streaming_providers/base/utils/vfs.py @@ -6,6 +6,7 @@ Provides transparent file operations for both Kodi and regular Python environmen import json import os +import tempfile import threading from typing import Any, Dict, List, Optional, Tuple @@ -16,6 +17,23 @@ from .environment import get_environment_manager, is_kodi_environment from .logger import logger +# One re-entrant lock per absolute file path, shared by ALL VFS instances in +# the process (SessionManager, the provider's own SessionManager and the +# authenticator's settings manager each own a separate VFS pointing at the +# same session.json, so a per-instance lock would not protect anything). +_file_locks: Dict[str, "threading.RLock"] = {} +_file_locks_guard = threading.Lock() + + +def _lock_for_path(abs_path: str) -> "threading.RLock": + key = os.path.normcase(os.path.abspath(abs_path)) if "://" not in abs_path else abs_path + lock = _file_locks.get(key) + if lock is None: + with _file_locks_guard: + lock = _file_locks.setdefault(key, threading.RLock()) + return lock + + class VFS: """ Virtual File System abstraction layer @@ -151,6 +169,21 @@ class VFS: logger.error(f"Error creating directory {dirpath}: {e}") return False + def file_lock(self, filepath: str) -> "threading.RLock": + """ + Process-wide re-entrant lock for one file. Hold it around a whole + read-modify-write sequence so concurrent request threads cannot lose + each other's updates: + + with vfs.file_lock("session.json"): + data = vfs.read_json("session.json") or {} + data["x"] = 1 + vfs.write_json("session.json", data) + """ + if not os.path.isabs(filepath): + filepath = self.join_path(filepath) + return _lock_for_path(filepath) + def read_text(self, filepath: str, encoding: str = "utf-8") -> Optional[str]: """ Read text content from file @@ -166,22 +199,23 @@ class VFS: if not os.path.isabs(filepath): filepath = self.join_path(filepath) - if is_kodi_environment(): - import xbmcvfs + with _lock_for_path(filepath): + if is_kodi_environment(): + import xbmcvfs - if not xbmcvfs.exists(filepath): - return None - with xbmcvfs.File(filepath, "r") as f: - content = f.read() - return content if content else None - else: - import pathlib + if not xbmcvfs.exists(filepath): + return None + with xbmcvfs.File(filepath, "r") as f: + content = f.read() + return content if content else None + else: + import pathlib - path = pathlib.Path(filepath) - if not path.exists(): - return None - with open(path, "r", encoding=encoding) as f: - return f.read() + path = pathlib.Path(filepath) + if not path.exists(): + return None + with open(path, "r", encoding=encoding) as f: + return f.read() except Exception as e: logger.error(f"Error reading file {filepath}: {e}") return None @@ -202,32 +236,63 @@ class VFS: if not os.path.isabs(filepath): filepath = self.join_path(filepath) - if is_kodi_environment(): - import xbmcvfs + with _lock_for_path(filepath): + if is_kodi_environment(): + import xbmcvfs - # Ensure directory exists - dir_path = "/".join(filepath.split("/")[:-1]) - if dir_path and not xbmcvfs.exists(dir_path): - xbmcvfs.mkdirs(dir_path) + # Ensure directory exists + dir_path = "/".join(filepath.split("/")[:-1]) + if dir_path and not xbmcvfs.exists(dir_path): + xbmcvfs.mkdirs(dir_path) - with xbmcvfs.File(filepath, "w") as f: - bytes_written = f.write(content) - logger.debug( - f"Kodi file write: {bytes_written} bytes to {filepath}" - ) - return bytes_written > 0 - else: - import pathlib + # xbmcvfs has no atomic replace; the lock still keeps + # in-process readers from seeing a half-written file. + with xbmcvfs.File(filepath, "w") as f: + bytes_written = f.write(content) + logger.debug( + f"Kodi file write: {bytes_written} bytes to {filepath}" + ) + return bytes_written > 0 + else: + import pathlib - path = pathlib.Path(filepath) - path.parent.mkdir(parents=True, exist_ok=True) - with open(path, "w", encoding=encoding) as f: - f.write(content) - return True + path = pathlib.Path(filepath) + path.parent.mkdir(parents=True, exist_ok=True) + self._atomic_write(path, content.encode(encoding)) + return True except Exception as e: logger.error(f"Error writing file {filepath}: {e}") return False + @staticmethod + def _atomic_write(path, data: bytes) -> None: + """ + Write to a temp file in the same directory, fsync, then os.replace(). + A reader (or a crash/restart mid-write) sees either the complete old + file or the complete new one, never a truncated session.json. + """ + fd, tmp_name = tempfile.mkstemp( + dir=str(path.parent), prefix=f".{path.name}.", suffix=".tmp" + ) + try: + with os.fdopen(fd, "wb") as f: + f.write(data) + f.flush() + os.fsync(f.fileno()) + try: + # keep the original file's permissions if it already exists + if path.exists(): + os.chmod(tmp_name, path.stat().st_mode & 0o7777) + except OSError: + pass + os.replace(tmp_name, path) + except BaseException: + try: + os.unlink(tmp_name) + except OSError: + pass + raise + def write_binary(self, filepath: str, data: bytes) -> bool: """ Write binary content to file diff --git a/lib/streaming_providers/providers/magenta2/auth.py b/lib/streaming_providers/providers/magenta2/auth.py index 396fbcf..998b005 100644 --- a/lib/streaming_providers/providers/magenta2/auth.py +++ b/lib/streaming_providers/providers/magenta2/auth.py @@ -21,6 +21,7 @@ from typing import Any, Dict, Optional from ...base.auth.base_auth import BaseAuthenticator, BaseAuthToken, TokenAuthLevel from ...base.auth.base_oauth2_auth import OIDCConfiguration from ...base.auth.credentials import ClientCredentials +from ...base.errors import AuthPendingError from ...base.utils.logger import logger from .constants import ( DEFAULT_COUNTRY, @@ -41,6 +42,10 @@ from .taa_client import TaaClient from .token_flow_manager import TokenFlowManager from .token_utils import JWTParser, PersonaTokenComposer +# Suggested client retry interval while an interactive login is pending +# (matches the backchannel polling interval). +REMOTE_LOGIN_RETRY_AFTER_SECONDS = 10 + @dataclass class Magenta2Credentials(ClientCredentials): @@ -717,6 +722,7 @@ class Magenta2Authenticator(BaseAuthenticator): backchannel_start_url=backchannel_start_url, qr_code_url_template=qr_code_url_template, provider_name="MagentaTV", + platform=self.platform, ) logger.info( @@ -779,6 +785,7 @@ class Magenta2Authenticator(BaseAuthenticator): Base64-encoded persona token Raises: + AuthPendingError: an interactive (QR/code) login is in progress Exception: If persona token cannot be obtained """ if not self.token_flow_manager: @@ -787,6 +794,9 @@ class Magenta2Authenticator(BaseAuthenticator): persona_result = self.token_flow_manager.get_persona_token(force_refresh=force_refresh) if not persona_result.success: + pending = self.get_pending_login_error() + if pending is not None: + raise pending raise Exception(f"Failed to get persona token: {persona_result.error}") return persona_result.persona_token @@ -1113,12 +1123,46 @@ class Magenta2Authenticator(BaseAuthenticator): if not self._sam3_client: return None + # Non-blocking: never hold the calling HTTP request for the whole + # login window. Starting is idempotent while a login is pending, so + # repeated requests reuse the same code instead of issuing new ones. try: - return self._sam3_client.remote_login(scope="tvhubs offline_access") + status = self._sam3_client.start_remote_login_background( + scope="tvhubs offline_access" + ) + if status is None: + return None + if status.state == "completed": + # The user finished since the last request: hand the token to + # TokenFlowManager, which continues the normal chain. + return self._sam3_client.take_remote_login_token() + # Still pending: report "no token yet"; get_persona_token() turns + # this into AuthPendingError for the caller. + return None except Exception as e: logger.error(f"Remote login flow failed: {e}") return None + def get_pending_login_error(self) -> Optional[AuthPendingError]: + """ + AuthPendingError describing the running interactive login, or None. + + Public so AuthBridge (which calls TokenFlowManager directly, not + get_persona_token()) can raise it without touching private state. + """ + if not self._sam3_client: + return None + status = self._sam3_client.get_background_login_status() + if status is None or status.state != "pending": + return None + return AuthPendingError( + "Interactive login pending: enter the code to continue", + login_code=status.login_code, + qr_target_url=status.qr_target_url, + remaining_seconds=status.remaining_seconds, + retry_after=REMOTE_LOGIN_RETRY_AFTER_SECONDS, + ) + # ======================================================================== # TAA Validation and Debugging # ======================================================================== diff --git a/lib/streaming_providers/providers/magenta2/auth_bridge.py b/lib/streaming_providers/providers/magenta2/auth_bridge.py index c599e72..e67bfe3 100644 --- a/lib/streaming_providers/providers/magenta2/auth_bridge.py +++ b/lib/streaming_providers/providers/magenta2/auth_bridge.py @@ -92,8 +92,10 @@ class AuthBridge: Return a valid persona token, using an in-memory TTL cache. Raises: - RuntimeError: If TokenFlowManager is not available. - Exception: If the underlying token flow fails. + AuthPendingError: An interactive (QR/code) login is waiting for + the user; callers should surface the code. + RuntimeError: If TokenFlowManager is not available. + Exception: If the underlying token flow fails. """ if not force_refresh and self._persona_cache and self._persona_cache.success: expires_at: float = self._persona_cache.expires_at @@ -111,6 +113,13 @@ class AuthBridge: persona_result: PersonaResult = tfm.get_persona_token(force_refresh=force_refresh) if not persona_result.success: + # A login that is merely waiting for the user is not a failure: + # raise it typed so the HTTP layer can answer 202 with the code + # instead of a generic 500. + pending_check = getattr(self._authenticator, "get_pending_login_error", None) + pending = pending_check() if callable(pending_check) else None + if pending is not None: + raise pending raise Exception(f"Failed to get persona token: {persona_result.error}") self._persona_cache = persona_result diff --git a/lib/streaming_providers/providers/magenta2/channel_manager.py b/lib/streaming_providers/providers/magenta2/channel_manager.py index 658aa8a..c78d4b2 100644 --- a/lib/streaming_providers/providers/magenta2/channel_manager.py +++ b/lib/streaming_providers/providers/magenta2/channel_manager.py @@ -9,6 +9,7 @@ import re import time from typing import Callable, Dict, List, Optional +from ...base.errors import AuthPendingError, ProviderError from ...base.models import StreamingChannel from ...base.utils.logger import logger from ..lib_theplatform import paginate_feed, PaginationError @@ -290,8 +291,13 @@ class ChannelManager: self._cached_channels = channels return channels + except ProviderError: + # Typed errors (AuthPendingError, AuthError, RateLimitError, ...) + # must reach the route unchanged so it can pick the right status; + # wrapping them in a bare Exception turned every one into a 500. + raise except Exception as e: - raise Exception(f"Error fetching channels from Magenta2 API: {e}") + raise Exception(f"Error fetching channels from Magenta2 API: {e}") from e def get_playback_id_for_station(self, station_id: str) -> Optional[str]: """Get the playback ID (opaque PID) for a station ID.""" @@ -450,6 +456,8 @@ class ChannelManager: except Magenta2PlaybackRestrictedException: raise + except ProviderError: + raise except KeyError as e: logger.error(f"No entitlement token in response for {content_id}: {e}") raise Exception(f"No entitlement token in response for {content_id}: {e}") @@ -551,6 +559,11 @@ class ChannelManager: logger.warning(f"Playback restricted for {channel.name}: {e}") is_restricted = True + except AuthPendingError: + # Retrying 3x per channel while waiting for the user would + # just burn time; surface it immediately. + raise + except Exception as e: retries += 1 if retries < max_retries: diff --git a/lib/streaming_providers/providers/magenta2/sam3_client.py b/lib/streaming_providers/providers/magenta2/sam3_client.py index cc526a0..778bed7 100644 --- a/lib/streaming_providers/providers/magenta2/sam3_client.py +++ b/lib/streaming_providers/providers/magenta2/sam3_client.py @@ -6,7 +6,7 @@ from urllib.parse import parse_qs, urlparse from ...base.network import HTTPManager from ...base.utils.logger import logger -from .constants import DEFAULT_REQUEST_TIMEOUT, GRANT_TYPES, SSO_USER_AGENT +from .constants import DEFAULT_PLATFORM, DEFAULT_REQUEST_TIMEOUT, GRANT_TYPES, SSO_USER_AGENT @dataclass @@ -52,9 +52,14 @@ class Sam3Client: backchannel_start_url: str = None, qr_code_url_template: str = None, provider_name: str = "MagentaTV", + platform: str = DEFAULT_PLATFORM, ): """ Initialize SAM3 client with all required endpoints + + platform: emulated client platform; forwarded to the remote login + handler so the backchannel/token requests present the same client + identity as the rest of the flow. """ self.http_manager = http_manager self.session_id = session_id @@ -87,6 +92,7 @@ class Sam3Client: self._remote_login_handler: Optional["RemoteLoginHandler"] = None self.provider_name = provider_name + self.platform = platform logger.debug( f"SAM3 client initialized - " @@ -467,6 +473,7 @@ class Sam3Client: token_endpoint=self._get_token_endpoint(), qr_code_url_template=self.qr_code_url_template, provider_name=self.provider_name, # ← add this + platform=self.platform, ) logger.info( @@ -503,6 +510,37 @@ class Sam3Client: return token_data + # --- Non-blocking remote login -------------------------------------- + # remote_login() blocks for up to ~5 minutes, which is unusable inside an + # HTTP request. These three methods drive the handler's background mode. + + def start_remote_login_background(self, scope: str = "tvhubs offline_access"): + """ + Start a remote login without blocking, or join the one already + running (same login code). Returns the handler's BackgroundLoginStatus, + or None if remote login is unavailable. + """ + handler = self._get_remote_login_handler() + if not handler: + logger.error("Remote login not available") + return None + return handler.start_background_login(scope) + + def get_background_login_status(self): + """Current BackgroundLoginStatus, or None if remote login is unavailable. + Never creates a handler or starts a login.""" + handler = self._remote_login_handler + return handler.get_background_status() if handler else None + + def take_remote_login_token(self) -> Optional[Dict[str, Any]]: + """Return the token data of a finished background login (once).""" + handler = self._remote_login_handler + token_data = handler.take_background_token() if handler else None + if token_data and "refresh_token" in token_data: + self.refresh_token = token_data["refresh_token"] + logger.info("✓ Remote login successful, refresh token obtained") + return token_data + def get_remote_login_status(self) -> Optional[Dict[str, Any]]: """Get current remote login session status""" handler = self._get_remote_login_handler() @@ -770,4 +808,4 @@ class Sam3Client: f"oauth_token={bool(self.oauth_token_endpoint)}, " f"line_auth={bool(self.line_auth_endpoint)}, " f"backchannel={bool(self.backchannel_start_url)}" - ) + ) \ No newline at end of file diff --git a/routes/providers.py b/routes/providers.py index 7037f77..b15ad52 100644 --- a/routes/providers.py +++ b/routes/providers.py @@ -3,10 +3,62 @@ Provider-related route handlers """ +import functools +from typing import Optional + from bottle import request, response +from streaming_providers.base.errors import ProviderError +from streaming_providers.base.http_errors import error_to_http, log_detail from streaming_providers.base.utils import logger +def handle_provider_errors(context: str, *, value_error_status: Optional[int] = None): + """ + Route decorator: translate the shared provider error hierarchy into a + proper HTTP status (see base/http_errors.py) instead of a blanket 500 + that echoes str(exception). + + A pending interactive login becomes 202 with the login code in the body. + Unknown exceptions are logged with a traceback and answered with a + generic 500 so internals don't leak. + + value_error_status: ChannelOperations signals "provider not found or + disabled" with a plain ValueError (the stream routes already treat it as + 404). Pass e.g. 404 to map ValueError that way for a route; the default + leaves ValueError as an unexpected error (500). + """ + + def decorator(func): + @functools.wraps(func) + def wrapper(*args, **kwargs): + try: + return func(*args, **kwargs) + except Exception as exc: # noqa: BLE001 - this IS the boundary + if ( + value_error_status is not None + and isinstance(exc, ValueError) + and not isinstance(exc, ProviderError) + ): + logger.warning(f"{context}: {exc}") + response.status = value_error_status + return {"error": str(exc)} + status, body, headers = error_to_http(exc) + detail = log_detail(exc, context) + if detail is not None: + if status >= 500: + logger.error(detail, exc_info=True) + else: + logger.warning(detail) + response.status = status + for name, value in headers.items(): + response.set_header(name, value) + return body + + return wrapper + + return decorator + + def setup_provider_routes(app, manager, service): """Setup provider-related routes""" @@ -93,50 +145,48 @@ def setup_provider_routes(app, manager, service): return {"error": str(api_err)} @app.route("/api/providers//channels") + @handle_provider_errors( + "GET /api/providers//channels", value_error_status=404 + ) def get_channels(provider): - try: - channels = manager.get_channels( - provider_name=provider, - fetch_manifests=request.query.get("fetch_manifests", "false").lower() - == "true", - country=request.query.get("country"), - ) + channels = manager.get_channels( + provider_name=provider, + fetch_manifests=request.query.get("fetch_manifests", "false").lower() + == "true", + country=request.query.get("country"), + ) - # Get provider instance to check catchup support - provider_instance = manager.get_provider(provider) - provider_catchup_hours = getattr( - provider_instance, "catchup_window", 0 - ) # CHANGE - epg_window = getattr(provider_instance, "epg_window", (0, 0)) + # Get provider instance to check catchup support + provider_instance = manager.get_provider(provider) + provider_catchup_hours = getattr( + provider_instance, "catchup_window", 0 + ) # CHANGE + epg_window = getattr(provider_instance, "epg_window", (0, 0)) - # Build channel list with catchup info - channels_data = [] - for c in channels: - channel_dict = c.to_dict() + # Build channel list with catchup info + channels_data = [] + for c in channels: + channel_dict = c.to_dict() - # Add catchup hours - use channel-specific if available, else provider default - if hasattr(c, "catchup_hours"): # CHANGE - channel_dict["CatchupHours"] = c.catchup_hours # CHANGE - else: - channel_dict["CatchupHours"] = provider_catchup_hours # CHANGE + # Add catchup hours - use channel-specific if available, else provider default + if hasattr(c, "catchup_hours"): # CHANGE + channel_dict["CatchupHours"] = c.catchup_hours # CHANGE + else: + channel_dict["CatchupHours"] = provider_catchup_hours # CHANGE - channels_data.append(channel_dict) + channels_data.append(channel_dict) - return { - "provider": provider, - "country": provider_instance.country if provider_instance else "DE", - "catchup_window_hours": provider_catchup_hours, # CHANGE - "epg_window": { - "past_days": epg_window[0], - "future_days": epg_window[1], - "implements_epg": epg_window != (0, 0), - }, - "channels": channels_data, - } - except Exception as api_err: - logger.error(f"API Error in /api/providers/{provider}: {str(api_err)}") - response.status = 500 - return {"error": str(api_err)} + return { + "provider": provider, + "country": provider_instance.country if provider_instance else "DE", + "catchup_window_hours": provider_catchup_hours, # CHANGE + "epg_window": { + "past_days": epg_window[0], + "future_days": epg_window[1], + "implements_epg": epg_window != (0, 0), + }, + "channels": channels_data, + } @app.route("/api/providers//auth/status") def get_provider_auth_status(provider):