From d327136be556eb47cf85b2ecf18f304c81ceaa6a Mon Sep 17 00:00:00 2001 From: Nirvana Date: Fri, 16 Oct 2026 11:03:55 +0200 Subject: [PATCH] Fix hrti --- lib/streaming_providers/providers/hrti/api.py | 112 ++- .../providers/hrti/auth.py | 198 ++++- .../providers/hrti/constants.py | 7 +- .../providers/hrti/epg_manager.py | 755 ++++++++++-------- .../providers/hrti/provider.py | 24 +- .../providers/hrti/session.py | 90 ++- 6 files changed, 807 insertions(+), 379 deletions(-) diff --git a/lib/streaming_providers/providers/hrti/api.py b/lib/streaming_providers/providers/hrti/api.py index 65b5cd4..0e21829 100644 --- a/lib/streaming_providers/providers/hrti/api.py +++ b/lib/streaming_providers/providers/hrti/api.py @@ -8,12 +8,22 @@ contract instead of a try/except in every method). * transport / HTTP-status errors become typed errors (base.errors); works whether the HTTP manager raises or returns the response * HTTP 401 / AuthError -> ONE session refresh + exactly one retry - * API body ``ErrorCode != 0``: raises ServerError for required endpoints, - returns None for optional ones (``required=False``) + * body ErrorCode == 4 (the server no longer knows this device session -- + typically a device id loaded from persisted storage that the server has + since expired) -> ONE device re-registration through the session + ONE + retry; a request that STILL answers ErrorCode 4 raises AuthError, also + for optional endpoints (a broken session is not "nothing there") + * other API body ``ErrorCode != 0``: raises ServerError for required + endpoints, returns None for optional ones (``required=False``) + +Worst case per call: 4 requests, 2 session refreshes, 1 device +re-registration. The state change lives in the session +(reregister_device), which locks and re-checks -- concurrent callers that +all fail with the same stale device id must not stampede RegisterDevice. """ import json -from typing import Any, Dict, Optional +from typing import Any, Dict, Optional, Tuple from ...base.errors import ( AuthError, @@ -25,6 +35,12 @@ from ...base.errors import ( from ...base.utils.logger import logger from .constants import HRTiConfig +#: Body-level error code: the server no longer knows this device session. +#: Observed with a device id loaded from persisted storage after the +#: server-side registration had expired. Recoverable exactly once per call +#: (see api_post); a surviving ErrorCode 4 is an AuthError. +DEVICE_SESSION_INVALID_CODE = 4 + def _snippet(response: Any) -> str: try: @@ -70,25 +86,34 @@ def _post_json(http_manager: Any, url: str, headers: Dict[str, str], payload: Di return data -def api_post( +def _post_with_auth_recovery( *, http_manager: Any, session: Any, config: HRTiConfig, + url: str, endpoint_key: str, payload: Dict, - referer_path: str = "/", - required: bool = True, -) -> Optional[Any]: - """POST to a named endpoint and return the ``Result`` field.""" - url = config.endpoint(endpoint_key) + referer_path: str, +) -> Tuple[Dict, Optional[str]]: + """POST + parse with ONE session refresh on HTTP 401 / AuthError. + + Returns ``(parsed body, device id used for the request that produced + it)``. The device id is captured at header-build time (token first, then + device id -- v1 evaluation order) so the ErrorCode-4 recovery can hand + the session the id that actually failed (concurrency re-check in + HRTiSession.reregister_device). + """ data: Dict = {} + device_id: Optional[str] = None for attempt in (1, 2): # Token and device state are read OUTSIDE the try: a login failure is # already a typed error and must not trigger the 401 recovery. + token = session.get_access_token() + device_id = session.device_id headers = config.api_headers( - token=session.get_access_token(), - device_id=session.device_id, + token=token, + device_id=device_id, ip_address=session.ip_address, referer_path=referer_path, ) @@ -101,6 +126,69 @@ def api_post( session.refresh() continue raise + return data, device_id + + +def api_post( + *, + http_manager: Any, + session: Any, + config: HRTiConfig, + endpoint_key: str, + payload: Dict, + referer_path: str = "/", + required: bool = True, +) -> Optional[Any]: + """POST to a named endpoint and return the ``Result`` field. + + Recovery ladder, each step at most ONCE per call: + 1. HTTP 401 / AuthError -> session.refresh() + one retry + 2. body ErrorCode == 4 -> session.reregister_device() + one retry + (the retry runs through step 1 again, so a token that died together + with the old device registration is refreshed there) + + A failure DURING recovery (re-registration or the retry request) raises + its typed error and propagates. ErrorCode 4 that survives the + re-registration raises AuthError -- for optional endpoints too: with the + old code a stale device id silently emptied EPG and every other optional + endpoint. All other ErrorCode != 0 follow the required/optional rule. + """ + url = config.endpoint(endpoint_key) + + data, device_id_used = _post_with_auth_recovery( + http_manager=http_manager, + session=session, + config=config, + url=url, + endpoint_key=endpoint_key, + payload=payload, + referer_path=referer_path, + ) + + if data.get("ErrorCode") == DEVICE_SESSION_INVALID_CODE: + logger.warning( + f"HRTi: {endpoint_key} answered ErrorCode {DEVICE_SESSION_INVALID_CODE} " + f"(invalid device/session) for device id {device_id_used!r}; " + f"re-registering the device once" + ) + session.reregister_device(failed_device_id=device_id_used) + data, _ = _post_with_auth_recovery( + http_manager=http_manager, + session=session, + config=config, + url=url, + endpoint_key=endpoint_key, + payload=payload, + referer_path=referer_path, + ) + if data.get("ErrorCode") == DEVICE_SESSION_INVALID_CODE: + # NOTE: constructor kwargs of base.errors are not fully verified + # (README §8); add url= here too if AuthError supports it + # (ServerError / TransportError do). + raise AuthError( + "HRTi: device session still invalid (ErrorCode " + f"{DEVICE_SESSION_INVALID_CODE}) after re-registration" + ) code = data.get("ErrorCode", 0) if code not in (0, None): @@ -113,4 +201,4 @@ def api_post( ) logger.debug(f"HRTi optional endpoint {endpoint_key}: [{code}] {description}") return None - return data.get("Result") + return data.get("Result") \ No newline at end of file diff --git a/lib/streaming_providers/providers/hrti/auth.py b/lib/streaming_providers/providers/hrti/auth.py index f21c4fe..6fd5199 100644 --- a/lib/streaming_providers/providers/hrti/auth.py +++ b/lib/streaming_providers/providers/hrti/auth.py @@ -1,5 +1,23 @@ # streaming_providers/providers/hrti/auth.py -import base64 +""" +HRTiAuthenticator -- stateful login flow (Variant B raw authenticator). + +Managers never call this class; they go through HRTiSession (the AuthProtocol +adapter). The provider keeps both objects: ``self.authenticator`` for the +legacy auth mixin, ``self.auth`` (the session) for the managers. + +Device identity +--------------- +The device id (a UUID) is CLIENT identity, not account state: it survives +invalidate() / refresh() and is deliberately NOT cleared with the token. It +is loaded from the settings manager on startup (flat keys, country=None) and +persisted whenever a fresh one is generated or the ErrorCode-4 recovery +re-registers it (register_device -> _persist_device_id). The server-side +registration of that id can expire while the stored id lives on -- that is +ErrorCode 4, recovered reactively by api.api_post (ONE re-registration + ONE +retry, then AuthError). +""" + import json import time import uuid @@ -9,7 +27,7 @@ from typing import Any, Dict, Optional from ...base.auth.base_auth import BaseAuthenticator, BaseAuthToken, TokenAuthLevel from ...base.errors import AuthError, ProviderError, ServerError from ...base.utils.logger import logger -from .api import typed_error_for +from .api import DEVICE_SESSION_INVALID_CODE, typed_error_for from .constants import HRTiConfig, HRTiDefaults from .models import HRTiAuthToken, HRTiCredentials @@ -115,7 +133,7 @@ class HRTiAuthenticator(BaseAuthenticator): # Add authorization header with Client prefix if bearer_token: headers["authorization"] = f"Client {bearer_token}" - logger.debug(f"Added authorization header with token: {bearer_token[:20]}...") + logger.debug("Added authorization header to API headers") return headers @@ -146,8 +164,6 @@ class HRTiAuthenticator(BaseAuthenticator): # Extract token access_token = result.get("Token", "") logger.debug(f"Extracted access_token: {'present' if access_token else 'MISSING'}") - if access_token: - logger.debug(f"Token length: {len(access_token)}, starts with: {access_token[:20]}...") # Store user ID if available if "Customer" in result: @@ -201,9 +217,7 @@ class HRTiAuthenticator(BaseAuthenticator): # Debug: Verify token is available logger.debug(f"Token created - access_token present: {bool(token.access_token)}") - if token.access_token: - logger.debug(f"Token value: {token.access_token[:20]}...") - else: + if not token.access_token: logger.warning("Token created but access_token is empty!") # Step 5: Register device (now with token available in self._current_token) @@ -347,13 +361,37 @@ class HRTiAuthenticator(BaseAuthenticator): raise raise ServerError(f"HRTi login failed: {e}") from e - def _register_device(self): - """Register device with HRTi using API headers with proper authorization""" + def _register_device(self, strict: bool = False) -> bool: + """ + Register the device with HRTi (RegisterDevice, HSAPI endpoint). + + strict=False (login flow, step 5 of _perform_authentication): best + effort -- every failure is logged and swallowed, a missing token is + a warning; registration must never abort the login. Returns + True/False (the login flow ignores it). + + strict=True (ErrorCode-4 recovery, called via register_device()): + failures RAISE typed errors -- AuthError when there is no token to + register with, ServerError for transport/HTTP/body-ErrorCode + failures, ConfigurationError when the endpoint is not configured. + The recovery ladder in api.api_post must be able to tell a FAILED + re-registration from a successful one; swallowing here would make + it retry the original request, get ErrorCode 4 again and raise a + misleading AuthError while the real cause is only a warning in + the log. + """ + url: Optional[str] = None try: bearer_token = self._current_token.access_token if self._current_token else "" if not bearer_token: + if strict: + raise AuthError("HRTi: no token available for device registration") logger.warning("No bearer token available for device registration") - return + return False + + # ConfigurationError when missing (endpoint() raises, never returns + # a falsy URL -- the old `if not register_endpoint` check was dead) + url = self.config.endpoint("register_device") # Use API headers with proper authorization and referer headers = self._get_api_headers(bearer_token) @@ -378,13 +416,8 @@ class HRTiAuthenticator(BaseAuthenticator): logger.debug(f"HRTi Device Registration Payload: {payload}") - register_endpoint = self.config.endpoint("register_device") - if not register_endpoint: - logger.warning("No register_device endpoint configured") - return - response = self.http_manager.post( - register_endpoint, + url, operation="api", headers=headers, data=json.dumps(payload), @@ -394,10 +427,36 @@ class HRTiAuthenticator(BaseAuthenticator): response.raise_for_status() result = response.json() + error_code = result.get("ErrorCode") if isinstance(result, dict) else None + if error_code not in (0, None): + if strict: + raise ServerError( + f"HRTi device registration failed: [{error_code}] " + f"{result.get('ErrorDescription')}", + url=url, + code=str(error_code), + ) + logger.warning( + f"HRTi device registration error: [{error_code}] " + f"{result.get('ErrorDescription')}" + ) + return False logger.debug(f"HRTi device registration successful: {result}") + return True - except Exception as e: + except ProviderError as e: + # Typed error (incl. ConfigurationError from endpoint() and the + # ServerError/AuthError raised above): fatal in the recovery + # path, logged and swallowed in the login flow (v1 behaviour). + if strict: + raise logger.warning(f"HRTi device registration failed: {e}") + return False + except Exception as e: + if strict: + raise typed_error_for(e, url) from e + logger.warning(f"HRTi device registration failed: {e}") + return False def _get_initial_data(self): """Get initial content rating and profiles using API headers with proper authorization""" @@ -441,16 +500,23 @@ class HRTiAuthenticator(BaseAuthenticator): self._device_id = self.settings_manager.get_device_id(self.provider_name, self.country) if not self._device_id: self._device_id = str(uuid.uuid4()) + # A freshly generated id MUST survive the restart, otherwise + # the device identity churns on every start and the server + # sees a brand-new device each time. No-op + warning when the + # settings manager has no setter (see _persist_device_id). + self._persist_device_id() logger.debug(f"Generated new device ID: {self._device_id}") # Ensure it's a valid UUID format if not self._validate_uuid(self._device_id): logger.warning(f"Invalid device ID format, generating new one: {self._device_id}") self._device_id = str(uuid.uuid4()) + self._persist_device_id() except Exception as e: logger.error(f"Error initializing device ID: {e}") self._device_id = str(uuid.uuid4()) + self._persist_device_id() @staticmethod def _validate_uuid(uuid_string): @@ -575,11 +641,23 @@ class HRTiAuthenticator(BaseAuthenticator): session_result = result.get("Result") if not session_result: + error_code = result.get("ErrorCode") + if error_code == DEVICE_SESSION_INVALID_CODE: + # Stale device session: AuthError (NOT ServerError) so that + # HRTiSession.authorize_session's existing recovery ladder + # (refresh = invalidate + forced login, which re-registers the + # device inside _perform_authentication) heals it with + # exactly one retry. Assumption to verify on device: + # ErrorCode 4 means the same here as on the data endpoints. + raise AuthError( + "HRTi: AuthorizeSession rejected the device session " + f"(ErrorCode {error_code}: {result.get('ErrorDescription')})" + ) raise ServerError( "HRTi: no result in AuthorizeSession response " - f"(ErrorCode={result.get('ErrorCode')}: {result.get('ErrorDescription')})", + f"(ErrorCode={error_code}: {result.get('ErrorDescription')})", url=auth_endpoint, - code=str(result.get("ErrorCode")), + code=str(error_code), ) logger.debug( f"Session authorization result - authorized: {session_result.get('Authorized', False)}, " @@ -793,4 +871,82 @@ class HRTiAuthenticator(BaseAuthenticator): return self._perform_authentication() except Exception as e: logger.error(f"HRTi token refresh failed: {e}") - return None \ No newline at end of file + return None + + # ------------------------------------------------------------------ + # ErrorCode-4 recovery (called through HRTiSession.reregister_device) + # ------------------------------------------------------------------ + def register_device(self, force: bool = False) -> None: + """ + Public, targeted device re-registration -- the ErrorCode-4 recovery + path. Called ONLY through HRTiSession.reregister_device (which locks + and dedupes concurrent callers); managers and api.py never call this + directly. + + Unlike the best-effort registration inside the login flow + (_register_device(strict=False) swallows every failure), this RAISES + typed errors (AuthError / ServerError / ConfigurationError) -- see + _register_device(strict=True). + + The device id is NOT rotated: RegisterDevice (re-)creates the + server-side record for the DeviceSerial supplied, so the SAME id is + sent; generating a fresh one would churn the device identity and any + server-side per-device state. ``force`` is therefore accepted for the + session contract but has no shortcut to skip today -- registration + in the login flow is unconditional anyway (the web client + re-registers on every login too). + + Persists the device id after success (write-through, see + _persist_device_id): the original bug was a stored id whose + server-side registration had expired; the reactive recovery fixes + the session, the write-through keeps memory and storage consistent. + """ + logger.debug(f"HRTi: targeted device re-registration (force={force})") + if self._register_device(strict=True): + self._persist_device_id() + + def _persist_device_id(self) -> None: + """ + Write the CURRENT device id back to the storage it is loaded from on + startup (settings_manager.get_device_id(provider_name, country)). + + Defensive about the save API (its name is not pinned anywhere in + this file): tries the natural setter names, each with the signature + MIRRORING the getter -- (provider_name, country, device_id), country + being None for the flat HRTi keys. No other signature is guessed: a + 2-arg call against a setter with an optional country parameter would + silently bind the device id to the country slot. + + Never raises. If no setter exists, logs ONE warning naming the + expected API -- then the id lives only in memory for this process + and the reactive ErrorCode-4 recovery in api.api_post self-heals + after every restart instead of never. If get_device_id turns out to + be get-or-create, this whole method is a harmless no-op except in + the freshly-generated branches of _initialize_device. + """ + device_id = self._device_id + if not device_id: + return + settings_manager = getattr(self, "settings_manager", None) + if settings_manager is None: + return + for name in ("set_device_id", "save_device_id", "store_device_id"): + setter = getattr(settings_manager, name, None) + if not callable(setter): + continue + try: + setter(self.provider_name, self.country, device_id) + logger.debug(f"HRTi: device id persisted via {name}") + return + except TypeError: + continue # wrong name/signature -- try the next candidate + except Exception as exc: + logger.warning(f"HRTi: persisting the device id failed ({name}): {exc!r}") + return + logger.warning( + "HRTi: settings manager exposes no device-id setter " + "(tried set_device_id/save_device_id/store_device_id with the " + "get_device_id-mirroring signature (provider, country, device_id)) " + "-- the current device id is not persisted; ErrorCode-4 recovery " + "stays reactive. Add the setter to make it stick across restarts." + ) \ No newline at end of file diff --git a/lib/streaming_providers/providers/hrti/constants.py b/lib/streaming_providers/providers/hrti/constants.py index 2d7ec44..0cf7ba4 100644 --- a/lib/streaming_providers/providers/hrti/constants.py +++ b/lib/streaming_providers/providers/hrti/constants.py @@ -47,6 +47,7 @@ class HRTiDefaults: "grant_access": "/GrantAccess", "channels": "/GetChannels", "programme": "/GetProgramme", + "epg_details": "/GetEpgDetails", "authorize_session": "/AuthorizeSession", "report_session": "/ReportSessionEvent", "register_device": "/RegisterDevice", @@ -61,9 +62,9 @@ class HRTiDefaults: } # served by the web API (webApiUrl) / by the HSAPI (apiUrl) WEB_API_ENDPOINTS = ( - "get_ip", "grant_access", "channels", "programme", "authorize_session", - "report_session", "catalogue_structure", "catalogue", "vod_details", - "episodes", "watch_later", "editors_choice", + "get_ip", "grant_access", "channels", "programme", "epg_details", + "authorize_session", "report_session", "catalogue_structure", + "catalogue", "vod_details", "episodes", "watch_later", "editors_choice", ) HSAPI_ENDPOINTS = ("register_device", "content_ratings", "profiles") diff --git a/lib/streaming_providers/providers/hrti/epg_manager.py b/lib/streaming_providers/providers/hrti/epg_manager.py index 61e2b26..a250b43 100644 --- a/lib/streaming_providers/providers/hrti/epg_manager.py +++ b/lib/streaming_providers/providers/hrti/epg_manager.py @@ -1,384 +1,479 @@ # streaming_providers/providers/hrti/epg_manager.py """ -HRTi EPG Manager - Handles Electronic Program Guide operations. +HRTi EPG Manager - Electronic Program Guide operations (v2). + +Native endpoints (both web API, both POST): + GetProgramme one call covers 1..N channels - the web client sends all + ~23 ChannelReferenceIds in a single request, so + get_epg_grid() overrides the ABC's per-channel loop with + ONE native batch call. + GetEpgDetails per-programme enrichment (description, categories, + credits, production countries) -> EPGProgramDetails; also + carries the catchup / start-over stream URLs (FileName*) + which the catchup manager reads through get_epg_details(). + +Collaborators: exactly the four of the manager contract. ``auth`` is the HRTi +session (AuthProtocol plus device_id / ip_address / refresh - see api.api_post); +``config`` is the shared HRTiConfig with the lazily loaded endpoints. + +Error contract (README §8): + * transport / HTTP errors raise typed errors (via api.api_post); a 401 is + answered there with ONE session refresh and exactly ONE retry + * ErrorCode != 0 on either endpoint -> [] / None with a warning: for EPG + "nothing there" is a valid answer and must not take the backend down + (v1 warned and returned empty results as well) + * a malformed single programme is skipped with a warning and never fails + the whole channel + +``limit`` / ``country`` arrive in **kw from the backend and are ignored on +purpose: the API has no server-side paging and HRTi is single-country ("HR"). """ -import json -from datetime import datetime, timedelta -from typing import Dict, List, Optional, Any, Tuple +from __future__ import annotations -from ...base.models.epg_models import EPGEntry +import re +from datetime import datetime, timedelta, timezone +from typing import Any, Dict, List, Optional, Tuple + +from ...base.managers.epg import EpgManager +from ...base.models.epg_models import ( + EPGEntry, + EPGFlags, + EPGGenre, + EPGProgramDetails, +) from ...base.utils.logger import logger +from .api import api_post +from .constants import HRTiConfig, HRTiDefaults -class HRTiEPGManager: - """Manages HRTi EPG operations.""" +class HRTiEpgManager(EpgManager): + """Manages HRTi EPG operations (GetProgramme / GetEpgDetails).""" - # Default EPG window: 2 days past, 7 days future - DEFAULT_PAST_DAYS = 2 - DEFAULT_FUTURE_DAYS = 7 + # EPG window (v1 values, kept): 2 days back, 7 days forward + PAST_DAYS = 2 + FUTURE_DAYS = 7 - def __init__(self, provider): - self.provider = provider - self._config = provider.hrti_config - self._http_manager = provider.http_manager - self._authenticator = provider.authenticator - self._channels_cache = None + # referer the web client uses on both EPG endpoints (v1 used "/tv") + EPG_REFERER_PATH = "/live/programme" - def _get_headers(self, referer_path: str = "/tv") -> Dict[str, str]: - """Build authenticated headers for EPG API calls.""" - headers = { - "deviceid": self._authenticator.get_device_id(), - "devicetypeid": self._config.device_reference_id, - "ipaddress": self._authenticator.get_ip_address(), - "operatorreferenceid": self._config.operator_reference_id, - "origin": self._config.base_website, - "referer": f"{self._config.base_website}{referer_path}", - "User-Agent": self._config.user_agent, - "Accept": "application/json", - "Content-Type": "application/json", - } - token = ( - self._authenticator._current_token.access_token - if self._authenticator._current_token - else "" + # WCF JSON date format "/Date()/" + _DATE_RE = re.compile(r"/Date\((\d+)\)/") + + # Size guard for the programme -> channel hint map. Entries are two short + # strings; cleared wholesale when exceeded - it is only a hint that every + # grid fetch re-populates. + _CHANNEL_HINT_CAP = 20000 + + # HRTi category label -> DVB-SI genre (ETSI EN 300 468); labels are + # upper-cased before the lookup. Same map as v1. + _CATEGORY_GENRES = { + "VIJESTI": EPGGenre.NEWSCURRENTAFFAIRS, + "SPORT": EPGGenre.SPORTS, + "GLAZBA": EPGGenre.MUSICBALLETDANCE, + "IGRANI PROGRAM": EPGGenre.MOVIEDRAMA, + "UMJETNOST I KULTURA": EPGGenre.ARTSCULTURE, + "ZABAVA": EPGGenre.SHOW, + "RELIGIJA": EPGGenre.SOCIALPOLITICALECONOMICS, + "OBRAZOVANJE": EPGGenre.EDUCATIONALSCIENCE, + "ZNANOST": EPGGenre.EDUCATIONALSCIENCE, + "DRUGI INFORMATIVNI SADRŽAJI": EPGGenre.NEWSCURRENTAFFAIRS, + "PROMOCIJA": EPGGenre.SHOW, + "KOMERCIJALNI PROGRAMI": EPGGenre.SHOW, + } + + def __init__(self, *, http_manager: Any, auth: Any, country: str, + config: HRTiConfig) -> None: + super().__init__( + http_manager=http_manager, auth=auth, country=country, config=config ) - if token: - headers["authorization"] = f"Client {token}" - return headers + # EPG ReferenceID -> channel ReferenceID. GetEpgDetails wants BOTH ids + # while the ABC hands get_program_details() only the programme id, so + # the mapping is remembered while parsing grids (and from the + # ChannelReferenceID of every details answer). Advisory only: a lost + # update under concurrent grid fetches costs nothing but the empty + # ChannelReferenceId fallback. + self._channel_by_epg_id: Dict[str, str] = {} - def get_epg_for_channel( - self, - channel_id: str, - start_time: Optional[datetime] = None, - end_time: Optional[datetime] = None, + # ------------------------------------------------------------------ + # Capability (ABC) + # ------------------------------------------------------------------ + + @property + def epg_window(self) -> Tuple[int, int]: + """(past_days, future_days) - v1 window, kept unchanged.""" + return self.PAST_DAYS, self.FUTURE_DAYS + + # handles_channel_id(): the ABC default (True) is correct - every HRTi + # channel, TV and radio alike, carries an EpgList. + + # ------------------------------------------------------------------ + # Schedule (ABC) + # ------------------------------------------------------------------ + + def get_epg( + self, + channel_id: str, + start_time: Optional[datetime] = None, + end_time: Optional[datetime] = None, + **kw: Any, ) -> List[EPGEntry]: """ - Get EPG data for a single channel. + EPG for one channel (one GetProgramme call with a single id). - Args: - channel_id: Channel reference ID (e.g., "40013") - start_time: Start time for EPG query (default: now - DEFAULT_PAST_DAYS days) - end_time: End time for EPG query (default: now + DEFAULT_FUTURE_DAYS days) + ``channel_id`` is the Channel.content_id == the HRTi channel + ReferenceID (e.g. "40013"). ``start_time``/``end_time`` are the + timezone-aware UTC datetimes the backend hands down, already clamped + to epg_window; called directly without times, the default window + applies (2 days back / 7 days forward). - Returns: - List of EPGEntry objects + **kw: ``limit`` and ``country`` arrive from the backend and are + ignored on purpose (no server-side paging, single country). + + Returns [] when the channel has no programmes in the window ("nothing + there" is not an error); transport/auth failures raise typed errors. """ - # Set default time range if not provided - now = datetime.now() - if start_time is None: - start_time = now - timedelta(days=self.DEFAULT_PAST_DAYS) - if end_time is None: - end_time = now + timedelta(days=self.DEFAULT_FUTURE_DAYS) + grid = self._fetch_programmes([channel_id], start_time, end_time) + return grid.get(str(channel_id), []) - # Convert to milliseconds timestamp for API - start_ts = int(start_time.timestamp() * 1000) - end_ts = int(end_time.timestamp() * 1000) - - try: - headers = self._get_headers() - payload = { - "ChannelReferenceIds": [channel_id], - "StartTime": f"/Date({start_ts})/", - "EndTime": f"/Date({end_ts})/", - } - - logger.debug(f"Fetching EPG for channel {channel_id}: {start_time} to {end_time}") - - response = self._http_manager.post( - self._config.api_endpoints["programme"], - operation="api", - headers=headers, - data=json.dumps(payload), - ) - response.raise_for_status() - - data = response.json() - if data.get("ErrorCode", 0) != 0: - logger.warning(f"EPG API error: {data.get('ErrorDescription')}") - return [] - - result = data.get("Result", []) - if not result: - return [] - - # Find the channel in the result - channel_data = None - for item in result: - if item.get("ReferenceID") == channel_id: - channel_data = item - break - - if not channel_data: - logger.debug(f"No EPG data found for channel {channel_id}") - return [] - - epg_list = channel_data.get("EpgList", []) - - # Convert to EPGEntry objects - entries = [] - for epg_item in epg_list: - entry = self._parse_epg_item(epg_item, channel_id) - if entry: - entries.append(entry) - - logger.debug(f"Retrieved {len(entries)} EPG entries for channel {channel_id}") - return entries - - except Exception as e: - logger.error(f"Error fetching EPG for channel {channel_id}: {e}") - return [] - - def get_epg_for_all_channels( - self, - start_time: Optional[datetime] = None, - end_time: Optional[datetime] = None, + def get_epg_grid( + self, + channel_ids: List[str], + start_time: Optional[datetime] = None, + end_time: Optional[datetime] = None, + **kw: Any, ) -> Dict[str, List[EPGEntry]]: """ - Get EPG data for all channels in a single API call. + Native batch override: GetProgramme takes 1..N ChannelReferenceIds in + one POST (the web client sends all ~23 at once), so the grid is ONE + request, not the ABC's per-channel loop. - Note: HRTi API supports fetching EPG for all channels at once, - which is more efficient than per-channel calls. + Every requested channel id is present in the result (with [] when the + API returns nothing for it). An empty ``channel_ids`` returns {} - the + provider layer fills the ids from get_channels() when the backend + passes none. - Returns: - Dictionary mapping channel_id -> list of EPGEntry objects + One request for all channels also means one failure for all channels: + there is no per-channel isolation here (ISOLATE_GRID_FAILURES only + affects the default loop, which we do not use). """ - # Set default time range - now = datetime.now() + if not channel_ids: + return {} + return self._fetch_programmes(list(channel_ids), start_time, end_time) + + # ------------------------------------------------------------------ + # Programme details (ABC + HRTi-specific raw access) + # ------------------------------------------------------------------ + + def get_program_details( + self, program_id: str, **kw: Any + ) -> Optional[EPGProgramDetails]: + """ + Enrichment for one programme (GetEpgDetails) -> EPGProgramDetails. + + ``program_id`` is the EPG ReferenceID - the value this manager puts + into EPGEntry.program_id while parsing grids. The API also wants the + channel id: it is remembered from grid parses (and from every details + answer); an unknown programme falls back to an empty + ChannelReferenceId (the v1 request shape), and a failing lookup + simply yields None. + """ + channel_id = self._channel_by_epg_id.get(program_id, "") + raw = self.get_epg_details(channel_id, program_id) + if not raw: + return None + return self._parse_program_details(raw, program_id) + + def get_epg_details( + self, channel_id: str, reference_id: str + ) -> Optional[Dict[str, Any]]: + """ + Raw GetEpgDetails Result for one programme. + + HRTi-specific surface next to the ABC (NOT a delegated capability): + the catchup manager reads FileName / FileNameStartOver (and the + StartOver / PauseLiveTV / Npvr flags) from here - fields with no place + on EPGProgramDetails. Intended flow for EPG-based catchup: the + catchup manager takes this manager as a keyword-only ``epg=`` extra, + asks get_epg() for the programme at start_time, takes its + EPGEntry.program_id and calls this method. + + ``channel_id`` may be empty when unknown (see get_program_details). + Returns None when the API has no details for the programme + (ErrorCode != 0 - an optional endpoint). + """ + payload = { + "ChannelReferenceId": channel_id or "", + "ReferenceId": reference_id, + } + return api_post( + http_manager=self.http_manager, + session=self.auth, + config=self.config, + endpoint_key="epg_details", + payload=payload, + referer_path=self.EPG_REFERER_PATH, + required=False, + ) + + # ------------------------------------------------------------------ + # Core fetch + # ------------------------------------------------------------------ + + def _fetch_programmes( + self, + channel_ids: List[str], + start_time: Optional[datetime], + end_time: Optional[datetime], + ) -> Dict[str, List[EPGEntry]]: + """ONE GetProgramme call; every requested id is present in the dict.""" + ids = [str(cid) for cid in channel_ids if cid] + start_time, end_time = self._resolve_range(start_time, end_time) + payload = { + "ChannelReferenceIds": ids, + "StartTime": f"/Date({int(start_time.timestamp() * 1000)})/", + "EndTime": f"/Date({int(end_time.timestamp() * 1000)})/", + } + + logger.debug( + f"HRTi EPG: GetProgramme for {len(ids)} channel(s), " + f"{start_time.isoformat()} .. {end_time.isoformat()}" + ) + + # required=False: ErrorCode != 0 (or a missing Result) is "no EPG for + # this request", not a failure - v1 warned and returned [] as well. + result = api_post( + http_manager=self.http_manager, + session=self.auth, + config=self.config, + endpoint_key="programme", + payload=payload, + referer_path=self.EPG_REFERER_PATH, + required=False, + ) + + grid: Dict[str, List[EPGEntry]] = {cid: [] for cid in ids} + if result is None: + logger.warning( + "HRTi EPG: GetProgramme returned no result (ErrorCode != 0?)" + ) + return grid + if not result: + return grid + + total = 0 + for channel_data in result: + channel_id = channel_data.get("ReferenceID") + if not channel_id: + continue + entries = [] + for item in channel_data.get("EpgList") or []: + entry = self._parse_epg_item(item, str(channel_id)) + if entry is not None: + entries.append(entry) + entries.sort(key=lambda e: e.start) + grid[str(channel_id)] = entries + total += len(entries) + + logger.debug(f"HRTi EPG: {total} programme(s) on {len(result)} channel(s)") + return grid + + def _resolve_range( + self, start_time: Optional[datetime], end_time: Optional[datetime] + ) -> Tuple[datetime, datetime]: + """ + Default window when called without times (the backend always hands + clamped, timezone-aware datetimes - defaults only matter for direct + calls). UTC "now" keeps the window correct regardless of the host's + local timezone. + """ + now = datetime.now(timezone.utc) if start_time is None: - start_time = now - timedelta(days=self.DEFAULT_PAST_DAYS) + start_time = now - timedelta(days=self.PAST_DAYS) if end_time is None: - end_time = now + timedelta(days=self.DEFAULT_FUTURE_DAYS) + end_time = now + timedelta(days=self.FUTURE_DAYS) + return start_time, end_time - start_ts = int(start_time.timestamp() * 1000) - end_ts = int(end_time.timestamp() * 1000) + # ------------------------------------------------------------------ + # Parsing + # ------------------------------------------------------------------ - # Get all channel IDs from provider - channels = self._get_all_channels() - if not channels: - return {} - - channel_ids = [ch.content_id for ch in channels] - - try: - headers = self._get_headers() - payload = { - "ChannelReferenceIds": channel_ids, - "StartTime": f"/Date({start_ts})/", - "EndTime": f"/Date({end_ts})/", - } - - logger.debug(f"Fetching EPG for {len(channel_ids)} channels") - - response = self._http_manager.post( - self._config.api_endpoints["programme"], - operation="api", - headers=headers, - data=json.dumps(payload), - ) - response.raise_for_status() - - data = response.json() - if data.get("ErrorCode", 0) != 0: - logger.warning(f"EPG API error: {data.get('ErrorDescription')}") - return {} - - result = data.get("Result", []) - - # Parse results per channel - epg_by_channel = {} - for channel_data in result: - channel_id = channel_data.get("ReferenceID") - if not channel_id: - continue - - epg_list = channel_data.get("EpgList", []) - entries = [] - for epg_item in epg_list: - entry = self._parse_epg_item(epg_item, channel_id) - if entry: - entries.append(entry) - - if entries: - epg_by_channel[channel_id] = entries - - logger.debug(f"Retrieved EPG for {len(epg_by_channel)} channels") - return epg_by_channel - - except Exception as e: - logger.error(f"Error fetching EPG for all channels: {e}") - return {} - - def get_epg_details(self, reference_id: str) -> Optional[Dict[str, Any]]: + def _parse_epg_item( + self, item: Dict[str, Any], channel_id: str + ) -> Optional[EPGEntry]: """ - Get detailed information for a specific EPG entry. + One EpgList item -> EPGEntry (or None, skipped with a warning). - This is useful for catch-up functionality as it provides the - FileName (streaming URL) for the recorded program. + Observed shape (GetProgramme): + {"AdultContent": false, "CategoryReferenceId": "GLAZBA", + "ContentRating": "", "ImagePath": "https://...", + "ReferenceID": "2230531584", + "TimeEnd": "/Date(1792094400000)/", "TimeEndUnixEpoch": 1792094400, + "TimeStart": "/Date(1792087200000)/", "TimeStartUnixEpoch": 1792087200, + "Title": "Mozart za spavalice"} - Args: - reference_id: EPG entry reference ID - - Returns: - Dictionary with EPG details including FileName + Programme ids are per broadcast slot (channel + time), so ReferenceID + becomes EPGEntry.program_id - the id get_program_details() expects. + Grid items carry no description; that arrives via details (and + EPGEntry.merge_details / merge_content on the caller side). """ - try: - headers = self._get_headers("/tv") - payload = { - "ChannelReferenceId": "", # Not required, but API expects it - "ReferenceId": reference_id, - } + reference_id = item.get("ReferenceID") + if reference_id is not None: + self._remember_channel(str(reference_id), channel_id) - response = self._http_manager.post( - self._config.api_endpoints["epg_details"], - operation="api", - headers=headers, - data=json.dumps(payload), + title = str(item.get("Title") or "").strip() + start = self._epoch_seconds(item.get("TimeStartUnixEpoch"), item.get("TimeStart")) + end = self._epoch_seconds(item.get("TimeEndUnixEpoch"), item.get("TimeEnd")) + + if not title: + logger.warning(f"HRTi EPG: skipping programme without title on channel {channel_id}") + return None + if not start or not end or end <= start: + logger.warning( + f"HRTi EPG: skipping programme '{title}' with invalid times ({reference_id})" ) - response.raise_for_status() - - data = response.json() - if data.get("ErrorCode", 0) != 0: - logger.warning(f"EPG details API error: {data.get('ErrorDescription')}") - return None - - return data.get("Result") - - except Exception as e: - logger.error(f"Error fetching EPG details for {reference_id}: {e}") return None - def _parse_epg_item(self, item: Dict[str, Any], channel_id: str) -> Optional[EPGEntry]: - """ - Parse a single EPG item from API response into EPGEntry. + category = str(item.get("CategoryReferenceId") or "").strip() or None - Expected item format from GetProgramme: - { - "AdultContent": false, - "CategoryReferenceId": "VIJESTI", - "ContentRating": "", - "ImagePath": "https://...", - "ReferenceID": "2059396656", - "TimeEnd": "/Date(1777725000000)/", - "TimeEndUnixEpoch": 1777725000, - "TimeStart": "/Date(1777723200000)/", - "TimeStartUnixEpoch": 1777723200, - "Title": "Vijesti" - } - """ try: - # Extract timestamps (prefer Unix epoch, fallback to parsing /Date() format) - start = item.get("TimeStartUnixEpoch") - if start is None: - start_str = item.get("TimeStart", "") - start = self._parse_date_string(start_str) - - end = item.get("TimeEndUnixEpoch") - if end is None: - end_str = item.get("TimeEnd", "") - end = self._parse_date_string(end_str) - - if not start or not end: - logger.warning(f"Missing timestamps for EPG item: {item.get('ReferenceID')}") - return None - - title = item.get("Title", "").strip() - if not title: - return None - - # Generate broadcast ID (32-bit unique identifier) - broadcast_id = EPGEntry.encode_broadcast_id( - self.provider.provider_name, - channel_id, - start - ) - - # Get genre type from CategoryReferenceId - genre_type = self._map_category_to_genre(item.get("CategoryReferenceId", "")) - return EPGEntry( - broadcast_id=broadcast_id, + broadcast_id=EPGEntry.encode_broadcast_id( + HRTiDefaults.PROVIDER_NAME, channel_id, start + ), title=title, start=start, end=end, - description=item.get("DescriptionShort") or item.get("DescriptionLong"), - icon=item.get("ImagePath"), - genre=genre_type, - genre_description=item.get("CategoryReferenceId"), + program_id=str(reference_id) if reference_id is not None else None, + description=item.get("DescriptionShort") + or item.get("DescriptionLong") + or None, + icon=item.get("ImagePath") or None, + genre=self._map_category_to_genre(category), + genre_description=category, + genres=[category] if category else None, parental_rating=self._parse_parental_rating(item.get("ContentRating")), - flags=EPGFlags.IS_SERIES if item.get("IsSeries", False) else EPGFlags.UNDEFINED, + flags=EPGFlags.IS_SERIES if item.get("IsSeries") else EPGFlags.UNDEFINED, + ) + except ValueError as exc: # EPGEntry post-init validation + logger.warning( + f"HRTi EPG: skipping invalid programme '{title}' on {channel_id}: {exc}" ) - - except Exception as e: - logger.error(f"Error parsing EPG item: {e}") return None + def _parse_program_details( + self, raw: Dict[str, Any], program_id: str + ) -> EPGProgramDetails: + """ + Map a GetEpgDetails Result onto EPGProgramDetails. + + Not mapped (no field on the model): OriginalTitle, ChannelName / + ChannelIcon (channel manager data), ThumbnailsDefinitionURL, + EpgRatingAverage, Price / IsPPVEvent, EnabledStreaming and the + FileName* stream URLs (catchup manager, via get_epg_details()). + """ + # self-healing hint map: details answers carry the channel id + channel_ref = raw.get("ChannelReferenceID") + if channel_ref: + self._remember_channel(program_id, str(channel_ref)) + + genres: List[str] = [] + for category in [raw.get("Category")] + list(raw.get("AdditionalCategories") or []): + if category and category not in genres: + genres.append(category) + + credits = self._parse_credits(raw.get("Credits")) + + # "Duration" is minutes (verified against the schedule: Duration == 11 + # for an 11-minute event) -> seconds, the unit the rest of the + # codebase uses. Re-check on a longer programme during the device test. + duration = None + if isinstance(raw.get("Duration"), (int, float)): + duration = int(raw["Duration"]) * 60 + + countries = [c for c in (raw.get("ProductionCountries") or []) if c] + + return EPGProgramDetails( + program_id=str(raw.get("ReferenceID") or program_id), + description=raw.get("DescriptionLong") or raw.get("DescriptionShort") or None, + icon=raw.get("ImagePath") or None, + genres=genres or None, + parental_rating=self._parse_parental_rating(raw.get("ContentRating")), + duration=duration, + country_of_origin=countries or None, + **credits, + ) + @staticmethod - def _parse_date_string(date_str: str) -> Optional[int]: - """Parse /Date(timestamp)/ format to Unix timestamp.""" - if not date_str: - return None + def _parse_credits(credits: Any) -> Dict[str, List[str]]: + """ + Best-effort split of the Credits list into cast / directors / writers. - import re - match = re.search(r'/Date\((\d+)\)/', date_str) - if match: - # Convert milliseconds to seconds - return int(match.group(1)) // 1000 + Every observed response carries an empty list, so the item shape is + INFERRED (name + role fields with Croatian or English role labels, or + plain strings -> cast). Unknown shapes are skipped; credits must never + fail the details parse. Verify on a programme with credits during the + device test. + """ + split: Dict[str, List[str]] = {} + if not isinstance(credits, list): + return split + for credit in credits: + if isinstance(credit, str): + name, role = credit.strip(), "" + elif isinstance(credit, dict): + name = str(credit.get("Name") or credit.get("name") or "").strip() + role = str( + credit.get("Role") or credit.get("role") or credit.get("RoleName") or "" + ).strip().lower() + else: + continue + if not name: + continue + if "redat" in role or "direct" in role: # redatelj / director + split.setdefault("directors", []).append(name) + elif "scenar" in role or "writ" in role or "autor" in role: + split.setdefault("writers", []).append(name) # scenarist / writer + else: # glumac / rest + split.setdefault("cast", []).append(name) + return split + @classmethod + def _epoch_seconds(cls, epoch: Any, date_string: Any) -> Optional[int]: + """ + Unix SECONDS from the API's double encoding: the *UnixEpoch field + (already seconds) with the /Date(milliseconds)/ string as fallback. + The conversion lives here because EPGEntry's coercion understands + ints and ISO strings but not the WCF /Date()/ format. + """ + if isinstance(epoch, (int, float)) and epoch > 0: + return int(epoch) + if isinstance(date_string, str): + match = cls._DATE_RE.search(date_string) + if match: + return int(match.group(1)) // 1000 return None - @staticmethod - def _map_category_to_genre(category: str) -> int: - """ - Map HRTi category to EPG genre type. - Based on ETSI EN 300 468 DVB-SI standard. - """ - category_map = { - "VIJESTI": EPGGenre.NEWSCURRENTAFFAIRS, # News/Current Affairs - "SPORT": EPGGenre.SPORTS, # Sports - "GLAZBA": EPGGenre.MUSICBALLETDANCE, # Music/Ballet/Dance - "IGRANI PROGRAM": EPGGenre.MOVIEDRAMA, # Movie/Drama - "UMJETNOST I KULTURA": EPGGenre.ARTSCULTURE, # Arts/Culture - "ZABAVA": EPGGenre.SHOW, # Show/Game Show - "RELIGIJA": EPGGenre.SOCIALPOLITICALECONOMICS, # Religion - "OBRAZOVANJE": EPGGenre.EDUCATIONALSCIENCE, # Educational - "ZNANOST": EPGGenre.EDUCATIONALSCIENCE, # Science - "DRUGI INFORMATIVNI SADRŽAJI": EPGGenre.NEWSCURRENTAFFAIRS, - "PROMOCIJA": EPGGenre.SHOW, - "KOMERCIJALNI PROGRAMI": EPGGenre.SHOW, - } - - return category_map.get(category, EPGGenre.UNDEFINED) - - @staticmethod - def _parse_parental_rating(rating_str: str) -> Optional[int]: - """Parse parental rating string to integer.""" - if not rating_str: + @classmethod + def _map_category_to_genre(cls, category: Optional[str]) -> Optional[int]: + """HRTi category label -> DVB-SI genre type (ETSI EN 300 468).""" + if not category: return None + return cls._CATEGORY_GENRES.get(category.strip().upper(), EPGGenre.UNDEFINED) - # Try to extract number from strings like "12", "FSK 12", "12+" - import re - match = re.search(r'(\d+)', rating_str) - if match: - return int(match.group(1)) + @staticmethod + def _parse_parental_rating(rating: Any) -> Optional[int]: + """Numeric age out of ContentRating ('', '12', 'FSK 12', '12+', 12).""" + if rating is None: + return None + if isinstance(rating, (int, float)): + value = int(rating) + return value if value > 0 else None + match = re.search(r"(\d+)", str(rating)) + return int(match.group(1)) if match else None - return None - - def _get_all_channels(self): - """Get all channels from provider (with caching).""" - if self._channels_cache is None: - self._channels_cache = self.provider.get_channels() - return self._channels_cache - - def get_epg_window(self) -> Tuple[int, int]: - """ - Return the EPG window as (past_days, future_days). - - Returns: - Tuple of (past_days, future_days) - """ - return self.DEFAULT_PAST_DAYS, self.DEFAULT_FUTURE_DAYS - - -# Import EPG models at the bottom to avoid circular imports -from ...base.models.epg_models import EPGEntry, EPGGenre, EPGFlags \ No newline at end of file + def _remember_channel(self, epg_reference_id: str, channel_id: str) -> None: + """Programme -> channel hint for GetEpgDetails (advisory, see __init__).""" + if len(self._channel_by_epg_id) >= self._CHANNEL_HINT_CAP: + self._channel_by_epg_id.clear() + self._channel_by_epg_id[epg_reference_id] = channel_id \ No newline at end of file diff --git a/lib/streaming_providers/providers/hrti/provider.py b/lib/streaming_providers/providers/hrti/provider.py index c0404d1..21ba766 100644 --- a/lib/streaming_providers/providers/hrti/provider.py +++ b/lib/streaming_providers/providers/hrti/provider.py @@ -7,10 +7,11 @@ caches) and builds the managers; everything generic (capability flags, content-id routing, manifest / DRM delegation, legacy surface) comes from ManagedProvider. -Capabilities in this step: live channels + radio, VOD. EPG is deliberately -NOT registered yet (v1 had implements_epg False; HRTiEpgManager follows as a -second step and needs a real GetProgramme sample). The unwired -epg_manager.py stays untouched in the package until then. +Capabilities: live channels + radio, VOD, EPG (native batch GetProgramme, +per-programme GetEpgDetails enrichment; epg_window = (2, 7), v1 values +kept). A future catchup manager would take the EPG manager as a +keyword-only ``epg=`` extra (_init_managers builds epg before catchup, so +``self.epg`` is available in _build_catchup). """ from typing import ClassVar, Dict, List, Optional @@ -20,6 +21,7 @@ from ...base.models.proxy_models import ProxyConfig from .auth import HRTiAuthenticator from .channel_manager import HRTiChannelManager from .constants import HRTiConfig, HRTiDefaults +from .epg_manager import HRTiEpgManager from .session import HRTiSession from .vod_manager import HRTiVodManager @@ -100,6 +102,8 @@ class HRTiProvider(ManagedProvider): self._playout_cache.clear() if self.channels is not None: self.channels.clear_caches() + # NOTE: the EPG manager's programme->channel hint map is schedule + # data, not account-specific state -- deliberately NOT cleared here. # ------------------------------------------------------------------ # Manager factories (None = capability not offered) @@ -122,4 +126,14 @@ class HRTiProvider(ManagedProvider): playout_cache=self._playout_cache, ) - # _build_epg(): step 2 (HRTiEpgManager, native batch GetProgramme). + def _build_epg(self): + # Exactly the four collaborators: no playout_cache (EPG plays + # nothing), no channels= (the provider layer fills channel_ids from + # get_channels() when the backend passes none; the programme->channel + # hint map inside the manager is its own state). + return HRTiEpgManager( + http_manager=self.http_manager, + auth=self.auth, + country=self.country, + config=self.hrti_config, + ) \ No newline at end of file diff --git a/lib/streaming_providers/providers/hrti/session.py b/lib/streaming_providers/providers/hrti/session.py index da71d5a..5998f5e 100644 --- a/lib/streaming_providers/providers/hrti/session.py +++ b/lib/streaming_providers/providers/hrti/session.py @@ -23,7 +23,15 @@ Behaviour rejected USER credentials degrade to an anonymous token (they do not raise). * ``invalidate()`` clears memory AND persisted storage (authenticator) and runs the registered callbacks (the provider clears its caches there). - ``refresh()`` is the 401 recovery: invalidate + one forced fresh login. + ``refresh()`` is the 401 recovery: invalidate + one forced fresh login + (which also re-registers the device -- step 5 of the login flow). +* ErrorCode-4 recovery (stale device session, called from api.api_post + ONLY): ``reregister_device`` re-registers the device ONCE under the + session lock -- deduped by a short TIME window (registration does NOT + rotate the device id, so an id comparison alone cannot dedupe) and by the + failed-id comparison (kept for the day registration rotates) -- so + concurrent requests failing together do not stampede RegisterDevice; + failures feed the same permanent/backoff bookkeeping as login failures. """ import threading @@ -38,6 +46,12 @@ _PERMANENT = (AuthError, ConfigurationError) class HRTiSession: + #: Seconds after a successful re-registration during which further + #: reregister_device() calls are deduped. Registration does not rotate + #: the device id, so concurrent callers that failed together can simply + #: retry with the (now re-registered) same id. + REREGISTER_DEDUPE_SECONDS = 5.0 + def __init__( self, *, @@ -59,6 +73,7 @@ class HRTiSession: self._permanent = False self._failures = 0 self._retry_at = 0.0 + self._last_reregister_at = float("-inf") # ------------------------------------------------------------------ # AuthProtocol @@ -102,7 +117,15 @@ class HRTiSession: # Session API for the UI and the 401 recovery # ------------------------------------------------------------------ def refresh(self) -> str: - """401 recovery: invalidate, then one forced fresh login.""" + """401 recovery: invalidate, then one forced fresh login. + + Note: the forced login re-registers the device as a side effect + (step 5 of the login flow) -- so a full refresh also heals a stale + device session, at the cost of discarding a possibly still-valid + token. The ErrorCode-4 ladder in api.api_post therefore prefers the + lighter, targeted reregister_device() and only falls back to this + when the token is genuinely dead (HTTP 401). + """ self.invalidate() return self.get_access_token(force_refresh=True) @@ -153,9 +176,9 @@ class HRTiSession: self.get_access_token() # typed login errors, backoff, lazy login try: return self._authenticator.authorize_session(**kw) - except AuthError: # stale token: one refresh, one retry - self.refresh() - return self._authenticator.authorize_session(**kw) + except AuthError: # stale token OR stale device session + self.refresh() # (ErrorCode 4 -> AuthError there): + return self._authenticator.authorize_session(**kw) # one retry def report_session_event(self, session_id: str, channel_id: Optional[str] = None) -> bool: return self._authenticator.report_session_event(session_id, channel_id) @@ -169,8 +192,8 @@ class HRTiSession: if self._permanent or self._clock() < self._retry_at: raise self._typed_error - def _record_failure(self, exc: BaseException) -> ProviderError: - typed = exc if isinstance(exc, ProviderError) else ServerError(f"HRTi login failed: {exc}") + def _record_failure(self, exc: BaseException, what: str = "login") -> ProviderError: + typed = exc if isinstance(exc, ProviderError) else ServerError(f"HRTi {what} failed: {exc}") self._last_error = exc self._typed_error = typed if isinstance(typed, _PERMANENT): @@ -180,7 +203,7 @@ class HRTiSession: delay = min(self._backoff_cap, self._backoff_base * (2 ** (self._failures - 1))) self._retry_at = self._clock() + delay logger.warning( - f"HRTi: login failed ({type(typed).__name__}, " + f"HRTi: {what} failed ({type(typed).__name__}, " f"{'permanent' if self._permanent else f'retry in {self._retry_at - self._clock():.0f}s'}): {typed}" ) return typed @@ -191,3 +214,54 @@ class HRTiSession: self._permanent = False self._failures = 0 self._retry_at = 0.0 + + # ------------------------------------------------------------------ + # ErrorCode-4 recovery (called from api.api_post ONLY) + # ------------------------------------------------------------------ + def reregister_device(self, failed_device_id: Optional[str] = None) -> None: + """ + ErrorCode-4 recovery: re-register the device with the backend. + + Called from api.api_post only (managers never call this). The session + is the ONE place allowed to touch the raw authenticator. + + Concurrency / stampede control, two guards: + * TIME (the effective one): registration does NOT rotate the device + id -- the authenticator re-registers the SAME uuid, RegisterDevice + (re-)creates the server-side record for the DeviceSerial supplied + -- so an id comparison alone cannot dedupe. A successful + re-registration therefore blocks further calls for + REREGISTER_DEDUPE_SECONDS; concurrent callers that failed together + simply retry with the (re-registered) same id. + * ID (kept for the day registration rotates): if the id HAS moved on + since the caller's request went out, this is a no-op. + + A FAILED re-registration feeds the shared permanent/backoff + bookkeeping (a fresh failure blocks the next concurrent attempt via + _raise_if_blocked instead of hammering RegisterDevice); a success + clears it. + + Raises typed errors (AuthError / ServerError / ConfigurationError); + never fails silently. + """ + with self._lock: + # Normally a no-op here: api_post calls get_access_token() right + # before the request, and every success clears the bookkeeping. + # It guards the failure case -- a re-registration that just + # failed must block the next concurrent attempt. + self._raise_if_blocked() + now = self._clock() + if now < self._last_reregister_at + self.REREGISTER_DEDUPE_SECONDS: + return # another caller re-registered moments ago + current = self._authenticator.get_device_id() + if failed_device_id is not None and current != failed_device_id: + return # id already rotated by another caller + try: + self._authenticator.register_device(force=True) + except Exception as exc: + typed = self._record_failure(exc, what="device registration") + if typed is exc: + raise + raise typed from exc + self._last_reregister_at = now + self._clear_failure() \ No newline at end of file