diff --git a/.idea/script.service.ultimate.iml b/.idea/script.service.ultimate.iml
index 979d44d..6767ad0 100644
--- a/.idea/script.service.ultimate.iml
+++ b/.idea/script.service.ultimate.iml
@@ -13,6 +13,8 @@
-
+
+
+
\ No newline at end of file
diff --git a/lib/streaming_providers/providers/hrti/__init__.py b/lib/streaming_providers/providers/hrti/__init__.py
index c294164..014029e 100644
--- a/lib/streaming_providers/providers/hrti/__init__.py
+++ b/lib/streaming_providers/providers/hrti/__init__.py
@@ -1,11 +1,11 @@
-# [file name]: __init__.py
-# [file content begin]
-# lib/streaming_providers/providers/hrti/__init__.py
+# streaming_providers/providers/hrti/__init__.py
"""
-HRTi streaming provider module
+HRTi (Croatian Radio Television Internet) provider.
+
+Public API:
+ from streaming_providers.providers.hrti import HRTiProvider
"""
from .provider import HRTiProvider
-__all__ = ["HRTiProvider"]
-# [file content end]
+__all__ = ["HRTiProvider"]
\ No newline at end of file
diff --git a/lib/streaming_providers/providers/hrti/api.py b/lib/streaming_providers/providers/hrti/api.py
new file mode 100644
index 0000000..65b5cd4
--- /dev/null
+++ b/lib/streaming_providers/providers/hrti/api.py
@@ -0,0 +1,116 @@
+# streaming_providers/providers/hrti/api.py
+"""
+Shared API-call helper for the HRTi managers (one place for the error
+contract instead of a try/except in every method).
+
+ * endpoint URL from the lazily loaded dynamic config
+ * authorization/device headers from the session (a fresh token per request)
+ * 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``)
+"""
+
+import json
+from typing import Any, Dict, Optional
+
+from ...base.errors import (
+ AuthError,
+ ProviderError,
+ ServerError,
+ TransportError,
+ default_error_for_status_heuristic,
+)
+from ...base.utils.logger import logger
+from .constants import HRTiConfig
+
+
+def _snippet(response: Any) -> str:
+ try:
+ return (response.text or "")[:200]
+ except Exception: # response object without text
+ return ""
+
+
+def typed_error_for(exc: BaseException, url: Optional[str] = None) -> ProviderError:
+ """Typed error for an exception raised by the HTTP layer: an exception that
+ carries a response is mapped by its status, anything else is a transport
+ fault."""
+ failed = getattr(exc, "response", None)
+ status = getattr(failed, "status_code", None)
+ if isinstance(status, int):
+ return default_error_for_status_heuristic(
+ status, f"HRTi HTTP {status}: {exc}", url=url, body_snippet=_snippet(failed)
+ )
+ return TransportError(f"HRTi request failed: {exc}", url=url)
+
+
+def _post_json(http_manager: Any, url: str, headers: Dict[str, str], payload: Dict) -> Dict:
+ try:
+ response = http_manager.post(
+ url, operation="api", headers=headers, data=json.dumps(payload)
+ )
+ except ProviderError:
+ raise
+ except Exception as exc: # HTTP manager raising for 4xx/5xx or transport faults
+ raise typed_error_for(exc, url) from exc
+
+ status = getattr(response, "status_code", None)
+ if isinstance(status, int) and status >= 400:
+ raise default_error_for_status_heuristic(
+ status, f"HRTi HTTP {status}", url=url, body_snippet=_snippet(response)
+ )
+ try:
+ data = response.json()
+ except ValueError as exc:
+ raise ServerError("HRTi: response is not valid JSON", url=url) from exc
+ if not isinstance(data, dict):
+ raise ServerError("HRTi: unexpected response shape", url=url)
+ return data
+
+
+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."""
+ url = config.endpoint(endpoint_key)
+ data: Dict = {}
+ 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.
+ headers = config.api_headers(
+ token=session.get_access_token(),
+ device_id=session.device_id,
+ ip_address=session.ip_address,
+ referer_path=referer_path,
+ )
+ try:
+ data = _post_json(http_manager, url, headers, payload)
+ break
+ except AuthError:
+ if attempt == 1:
+ logger.info(f"HRTi: {endpoint_key} answered 401, refreshing the session once")
+ session.refresh()
+ continue
+ raise
+
+ code = data.get("ErrorCode", 0)
+ if code not in (0, None):
+ description = data.get("ErrorDescription")
+ if required:
+ raise ServerError(
+ f"HRTi API error for {endpoint_key}: [{code}] {description}",
+ url=url,
+ code=str(code),
+ )
+ logger.debug(f"HRTi optional endpoint {endpoint_key}: [{code}] {description}")
+ return None
+ return data.get("Result")
diff --git a/lib/streaming_providers/providers/hrti/auth.py b/lib/streaming_providers/providers/hrti/auth.py
index d2ea802..f21c4fe 100644
--- a/lib/streaming_providers/providers/hrti/auth.py
+++ b/lib/streaming_providers/providers/hrti/auth.py
@@ -7,54 +7,37 @@ from datetime import datetime, timedelta
from typing import Any, Dict, Optional
from ...base.auth.base_auth import BaseAuthenticator, BaseAuthToken, TokenAuthLevel
-from ...base.models.proxy_models import ProxyConfig
+from ...base.errors import AuthError, ProviderError, ServerError
from ...base.utils.logger import logger
-from .constants import HRTiConfig
+from .api import typed_error_for
+from .constants import HRTiConfig, HRTiDefaults
from .models import HRTiAuthToken, HRTiCredentials
class HRTiAuthenticator(BaseAuthenticator):
- def __init__(
- self,
- credentials=None,
- config_dir=None,
- proxy_config: Optional[ProxyConfig] = None,
- http_manager=None,
- ):
-
- # Initialize configuration FIRST
- self._config = HRTiConfig()
-
- # Get proxy_config if not provided
- if proxy_config is None:
- from ...base.network import ProxyConfigManager
-
- proxy_mgr = ProxyConfigManager(config_dir)
- proxy_config = proxy_mgr.get_proxy_config("hrti")
-
- # Store HTTP manager and proxy config locally (like MagentaEU)
- self._http_manager = http_manager
- self._proxy_config = proxy_config
-
- # Require http_manager like MagentaEU does
+ def __init__(self, *, http_manager, config: HRTiConfig, credentials=None, config_dir=None):
+ """
+ ``config`` is the ONE HRTiConfig of the provider (shared with session
+ and managers; never build a second one here). Storage keys stay FLAT
+ (country=None): credentials, persisted session and device id of
+ existing installations are stored without a country.
+ """
if http_manager is None:
raise ValueError("http_manager is required for HRTiAuthenticator")
+ self._config = config
+ self._http_manager = http_manager
- # Initialize HRTi-specific properties BEFORE calling parent init
+ # HRTi-specific state BEFORE the parent init (it loads the session)
self._ip_address = None
self._device_id = None
- self._user_id = None # Initialize _user_id here
+ self._user_id = None
- # Call parent init WITHOUT proxy_config and http_manager (like MagentaEU)
super().__init__(
- provider_name="hrti",
+ provider_name=HRTiDefaults.PROVIDER_NAME,
credentials=credentials,
- country=None, # Changed from 'HR' to None for flat credential structure
+ country=None,
config_dir=config_dir,
- # NO proxy_config or http_manager passed to parent
)
-
- # Load or initialize device ID
self._initialize_device()
@property
@@ -80,7 +63,7 @@ class HRTiAuthenticator(BaseAuthenticator):
@property
def auth_endpoint(self) -> str:
"""HRTi authentication endpoint"""
- return self.config.api_endpoints.get("grant_access", "")
+ return self.config.endpoint("grant_access")
def _get_auth_headers(self) -> Dict[str, str]:
"""Get headers specifically for authentication endpoint"""
@@ -176,7 +159,7 @@ class HRTiAuthenticator(BaseAuthenticator):
token = HRTiAuthToken(
access_token=access_token,
token_type="Client",
- expires_in=86400, # 24 hours default
+ expires_in=HRTiDefaults.TOKEN_LIFETIME_SECONDS, # GrantAccess carries no lifetime
issued_at=time.time(),
user_id=self._user_id or "", # Ensure user_id is never None
valid_from=result.get("ValidFrom", ""),
@@ -194,7 +177,10 @@ class HRTiAuthenticator(BaseAuthenticator):
def get_fallback_credentials(self):
"""Get fallback credentials (anonymous access)"""
- return HRTiCredentials(username="anonymoushrt", password="an0nPasshrt")
+ return HRTiCredentials(
+ username=HRTiDefaults.ANONYMOUS_USERNAME,
+ password=HRTiDefaults.ANONYMOUS_PASSWORD,
+ )
def _perform_authentication(self) -> HRTiAuthToken:
"""Perform HRTi custom authentication flow"""
@@ -250,24 +236,18 @@ class HRTiAuthenticator(BaseAuthenticator):
except Exception as e:
logger.debug(f"Could not read configured IP from settings: {e}")
- # Priority 2: Fall back to API fetch if no configured IP
+ # Priority 2: ask the API (endpoint resolved lazily from the dynamic config)
try:
logger.debug("No configured IP found, fetching from API...")
- get_ip_endpoint = self.config.api_endpoints.get("get_ip", "")
- if not get_ip_endpoint:
- logger.warning("No get_ip endpoint configured, using fallback")
- self._ip_address = "0.0.0.0"
- return self._ip_address
-
- response = self.http_manager.get(get_ip_endpoint, operation="api")
+ response = self.http_manager.get(self.config.endpoint("get_ip"), operation="api")
response.raise_for_status()
self._ip_address = response.text.strip().strip('"') # Remove quotes if present
logger.info(f"Retrieved IP address from API: {self._ip_address}")
return self._ip_address
except Exception as e:
+ # Not cached: the next call may succeed
logger.error(f"Error getting IP address from API: {e}")
- self._ip_address = "0.0.0.0" # Fallback
- return self._ip_address
+ return "0.0.0.0"
def _get_configured_ip_from_settings(self) -> Optional[str]:
"""
@@ -297,26 +277,9 @@ class HRTiAuthenticator(BaseAuthenticator):
return None
def _get_environment_config(self):
- """Get HRTi environment configuration"""
- try:
- # Get env config
- env_response = self.http_manager.get(self.config.env_endpoint, operation="api")
- env_response.raise_for_status()
- env_data = env_response.json()
-
- # Get main config
- config_response = self.http_manager.get(self.config.config_endpoint, operation="api")
- config_response.raise_for_status()
- config_data = config_response.json()
-
- # Update config with retrieved values
- self.config.update_from_api(env_data, config_data)
- logger.debug("HRTi environment configuration loaded")
-
- except Exception as e:
- logger.warning(f"Error loading HRTi environment config: {e}")
- # Ensure API endpoints are built even if config fetch fails
- self.config._build_fallback_endpoints()
+ """Dynamic endpoints (env.json / config.production.json), loaded once by
+ the shared config -- also used by the managers, independent of login."""
+ self.config.ensure_loaded()
def _perform_grant_access(self) -> Dict[str, Any]:
"""Perform grant access authentication with proper headers and debugging"""
@@ -346,9 +309,7 @@ class HRTiAuthenticator(BaseAuthenticator):
safe_payload["Password"] = "***" if safe_payload["Password"] else ""
logger.debug(f"HRTi Auth Payload: {safe_payload}")
- auth_endpoint = self.auth_endpoint
- if not auth_endpoint:
- raise Exception("No grant_access endpoint configured")
+ auth_endpoint = self.auth_endpoint # ConfigurationError when missing
response = self.http_manager.post(
auth_endpoint,
@@ -362,7 +323,10 @@ class HRTiAuthenticator(BaseAuthenticator):
result = response.json()
if "Result" not in result:
- raise Exception("No result in grant access response")
+ raise AuthError(
+ "HRTi: GrantAccess response has no Result "
+ f"(ErrorCode={result.get('ErrorCode')}: {result.get('ErrorDescription')})"
+ )
customer_id = result.get("Result", {}).get("Customer", {}).get("CustomerId", "unknown")
logger.debug(f"Grant access successful - user: {customer_id}")
@@ -373,13 +337,15 @@ class HRTiAuthenticator(BaseAuthenticator):
# Only fallback to anonymous if we're not already using it
if (
not isinstance(self.credentials, HRTiCredentials)
- or self.credentials.username != "anonymoushrt"
+ or self.credentials.username != HRTiDefaults.ANONYMOUS_USERNAME
):
logger.info("Falling back to anonymous credentials")
self.credentials = self.get_fallback_credentials()
return self._perform_grant_access()
else:
- raise e
+ if isinstance(e, ProviderError):
+ raise
+ raise ServerError(f"HRTi login failed: {e}") from e
def _register_device(self):
"""Register device with HRTi using API headers with proper authorization"""
@@ -412,7 +378,7 @@ class HRTiAuthenticator(BaseAuthenticator):
logger.debug(f"HRTi Device Registration Payload: {payload}")
- register_endpoint = self.config.api_endpoints.get("register_device", "")
+ register_endpoint = self.config.endpoint("register_device")
if not register_endpoint:
logger.warning("No register_device endpoint configured")
return
@@ -443,7 +409,7 @@ class HRTiAuthenticator(BaseAuthenticator):
headers["referer"] = f"{self.config.base_website}/"
# Get content ratings
- content_endpoint = self.config.api_endpoints.get("content_ratings", "")
+ content_endpoint = self.config.endpoint("content_ratings")
if content_endpoint:
content_response = self.http_manager.post(
content_endpoint,
@@ -454,7 +420,7 @@ class HRTiAuthenticator(BaseAuthenticator):
content_response.raise_for_status()
# Get profiles
- profiles_endpoint = self.config.api_endpoints.get("profiles", "")
+ profiles_endpoint = self.config.endpoint("profiles")
if profiles_endpoint:
profiles_response = self.http_manager.post(
profiles_endpoint,
@@ -500,27 +466,62 @@ class HRTiAuthenticator(BaseAuthenticator):
return self._device_id
def get_ip_address(self) -> str:
- """Get IP address"""
- if not self._ip_address:
- self._get_ip_address()
- return self._ip_address
+ """Get IP address (cached; "0.0.0.0" fallback is not cached)."""
+ return self._get_ip_address()
def _load_session(self) -> None:
- """Override to ensure _user_id is set when loading session"""
+ """
+ Restore the persisted token. The stored dict is FLAT (``to_dict``), so
+ it is rebuilt with HRTiAuthToken.from_dict. It must NOT go through
+ _create_token_from_response (the base default), which expects a
+ GrantAccess response and used to produce a token with an EMPTY
+ access_token and a fresh issued_at -- "valid" for 24 h and never
+ refreshed.
+ """
+ self._current_token = None
try:
- # Call parent method first
- super()._load_session()
-
- # If we have a current token and it's an HRTiAuthToken, set _user_id
- if self._current_token and isinstance(self._current_token, HRTiAuthToken):
- self._user_id = self._current_token.user_id
- logger.debug(f"Set _user_id from loaded session: {self._user_id}")
-
+ token_data = self.settings_manager.load_token_data(self.provider_name, self.country)
+ if token_data and token_data.get("access_token"):
+ token = HRTiAuthToken.from_dict(token_data)
+ token.auth_level = self._classify_token(token)
+ self._current_token = token
+ self._user_id = token.user_id or self._user_id or ""
+ logger.info("Restored persisted HRTi session")
+ else:
+ logger.info("No existing HRTi session found")
except Exception as e:
- logger.error(f"Error in HRTi session loading: {e}")
- # Ensure _user_id is at least an empty string
+ logger.error(f"Error loading HRTi session: {e}")
+ self._current_token = None
self._user_id = self._user_id or ""
+ # ------------------------------------------------------------------
+ # State read by HRTiSession
+ # ------------------------------------------------------------------
+ def get_current_token(self) -> Optional[BaseAuthToken]:
+ return self._current_token
+
+ def get_user_id(self) -> str:
+ return self._user_id or ""
+
+ def has_user_credentials(self) -> bool:
+ """True when real (non-anonymous) user credentials are stored or set."""
+ from ...base.auth.credentials import UserPasswordCredentials
+
+ try:
+ stored = self.settings_manager.get_provider_credentials(
+ self.provider_name, self.country
+ )
+ except TypeError:
+ stored = self.settings_manager.get_provider_credentials(self.provider_name)
+ for creds in (stored, self.credentials):
+ if (
+ isinstance(creds, UserPasswordCredentials)
+ and creds.validate()
+ and creds.username != HRTiDefaults.ANONYMOUS_USERNAME
+ ):
+ return True
+ return False
+
def authorize_session(
self,
content_type: str,
@@ -530,69 +531,61 @@ class HRTiAuthenticator(BaseAuthenticator):
channel_id: str = None,
start_time: str = None,
end_time: str = None,
- ) -> Optional[Dict[str, Any]]:
- """Authorize a playback session"""
+ ) -> Dict[str, Any]:
+ """Authorize a playback session. Raises typed errors (AuthError on 401,
+ ServerError otherwise); returns the raw ``Result`` dict -- the caller
+ checks ``Authorized``."""
+ bearer_token = self._current_token.access_token if self._current_token else ""
+ headers = self._get_api_headers(bearer_token)
+
+ # Set referer based on content type
+ if channel_id is None:
+ headers["referer"] = f"{self.config.base_website}/videostore"
+ elif content_type == "tlive":
+ headers["referer"] = f"{self.config.base_website}/live/tv?channel={channel_id}"
+ elif content_type == "rlive":
+ headers["referer"] = f"{self.config.base_website}/live/radio"
+ else:
+ headers["referer"] = f"{self.config.base_website}/live/"
+
+ payload = {
+ "ContentType": content_type,
+ "ContentReferenceId": content_ref_id,
+ "ContentDrmId": content_drm_id or f"{content_ref_id}_drm",
+ "VideostoreReferenceIds": video_store_ids,
+ "ChannelReferenceId": channel_id,
+ "StartTime": start_time,
+ "EndTime": end_time,
+ }
+ logger.debug(
+ f"Authorizing session - type: {content_type}, ref: {content_ref_id}, drm: {content_drm_id}"
+ )
+
+ auth_endpoint = self.config.endpoint("authorize_session")
try:
- bearer_token = self._current_token.access_token if self._current_token else ""
- headers = self._get_api_headers(bearer_token)
-
- # Set referer based on content type
- if channel_id is None:
- headers["referer"] = f"{self.config.base_website}/videostore"
- else:
- if content_type == "tlive":
- headers["referer"] = f"{self.config.base_website}/live/tv?channel={channel_id}"
- elif content_type == "rlive":
- headers["referer"] = f"{self.config.base_website}/live/radio"
- else:
- headers["referer"] = f"{self.config.base_website}/live/"
-
- payload = {
- "ContentType": content_type,
- "ContentReferenceId": content_ref_id,
- "ContentDrmId": content_drm_id or f"{content_ref_id}_drm",
- "VideostoreReferenceIds": video_store_ids,
- "ChannelReferenceId": channel_id,
- "StartTime": start_time,
- "EndTime": end_time,
- }
-
- logger.debug(
- f"Authorizing session - type: {content_type}, ref: {content_ref_id}, drm: {content_drm_id}"
- )
-
- auth_endpoint = self.config.api_endpoints.get("authorize_session", "")
- if not auth_endpoint:
- logger.error("No authorize_session endpoint configured")
- return None
-
response = self.http_manager.post(
- auth_endpoint,
- operation="api",
- headers=headers,
- data=json.dumps(payload),
+ auth_endpoint, operation="api", headers=headers, data=json.dumps(payload)
)
response.raise_for_status()
-
result = response.json()
- session_result = result.get("Result")
- if session_result:
- authorized = session_result.get("Authorized", False)
- session_id = session_result.get("SessionId") or session_result.get("DrmId")
- logger.debug(
- f"Session authorization result - authorized: {authorized}, session_id: {session_id}"
- )
- return session_result
- else:
- logger.warning(
- f"No result in session authorization response "
- f"(ErrorCode={result.get('ErrorCode')}: {result.get('ErrorDescription')})"
- )
- return None
+ except ProviderError:
+ raise
+ except Exception as exc:
+ raise typed_error_for(exc, auth_endpoint) from exc
- except Exception as e:
- logger.error(f"HRTi session authorization failed: {e}")
- return None
+ session_result = result.get("Result")
+ if not session_result:
+ raise ServerError(
+ "HRTi: no result in AuthorizeSession response "
+ f"(ErrorCode={result.get('ErrorCode')}: {result.get('ErrorDescription')})",
+ url=auth_endpoint,
+ code=str(result.get("ErrorCode")),
+ )
+ logger.debug(
+ f"Session authorization result - authorized: {session_result.get('Authorized', False)}, "
+ f"session_id: {session_result.get('SessionId') or session_result.get('DrmId')}"
+ )
+ return session_result
def report_session_event(self, session_id: str, channel_id: str = None) -> bool:
"""Report session event (like play start)"""
@@ -608,7 +601,7 @@ class HRTiAuthenticator(BaseAuthenticator):
payload = {"SessionEventId": 1, "SessionId": session_id} # 1 = play start
- report_endpoint = self.config.api_endpoints.get("report_session", "")
+ report_endpoint = self.config.endpoint("report_session")
if not report_endpoint:
logger.warning("No report_session endpoint configured")
return False
@@ -628,33 +621,6 @@ class HRTiAuthenticator(BaseAuthenticator):
logger.warning(f"Session event reporting failed: {e}")
return False
- def get_license_data(self, session_id: str) -> str:
- """Generate license data for DRM - returns base64 encoded string"""
- try:
- # Use merchant from config (now dynamically loaded)
- merchant = self.config.merchant or "aviion2"
-
- drm_license = {
- "userId": self._user_id or "",
- "sessionId": session_id,
- "merchant": merchant,
- }
-
- logger.debug(
- f"Creating license data - userId: {self._user_id}, sessionId: {session_id}, merchant: {merchant}"
- )
-
- # Encode to JSON then to base64
- license_bytes = json.dumps(drm_license).encode("utf-8")
- license_b64 = base64.b64encode(license_bytes).decode("utf-8")
-
- logger.debug(f"License data (base64, first 30 chars): {license_b64[:30]}...")
- return license_b64
-
- except Exception as e:
- logger.error(f"Error generating license data: {e}")
- return ""
-
@staticmethod
def get_time_offset(hours_offset: int) -> int:
"""Get timestamp with offset in milliseconds"""
@@ -669,7 +635,7 @@ class HRTiAuthenticator(BaseAuthenticator):
# Check if it's an HRTiAuthToken and has user_id attribute
if isinstance(token, HRTiAuthToken):
# Check if it's an anonymous token by user_id
- if token.user_id == "anonymoushrt":
+ if token.user_id == HRTiDefaults.ANONYMOUS_USERNAME:
return TokenAuthLevel.ANONYMOUS
# Check if we have user credentials available
@@ -695,18 +661,18 @@ class HRTiAuthenticator(BaseAuthenticator):
)
# If we have user credentials and token has a real user_id, it's user authenticated
- if has_user_creds and token.user_id and token.user_id != "anonymoushrt":
+ if has_user_creds and token.user_id and token.user_id != HRTiDefaults.ANONYMOUS_USERNAME:
return TokenAuthLevel.USER_AUTHENTICATED
# Check if using anonymous credentials
- if hasattr(self.credentials, "username") and self.credentials.username == "anonymoushrt":
+ if hasattr(self.credentials, "username") and self.credentials.username == HRTiDefaults.ANONYMOUS_USERNAME:
return TokenAuthLevel.ANONYMOUS
# If we have user credentials, consider it user authenticated
if (
hasattr(self.credentials, "username")
and self.credentials.username
- and self.credentials.username != "anonymoushrt"
+ and self.credentials.username != HRTiDefaults.ANONYMOUS_USERNAME
):
return TokenAuthLevel.USER_AUTHENTICATED
diff --git a/lib/streaming_providers/providers/hrti/channel_manager.py b/lib/streaming_providers/providers/hrti/channel_manager.py
new file mode 100644
index 0000000..68d1b2b
--- /dev/null
+++ b/lib/streaming_providers/providers/hrti/channel_manager.py
@@ -0,0 +1,160 @@
+# streaming_providers/providers/hrti/channel_manager.py
+"""
+HRTi live channel manager (live TV + radio).
+
+Playback flow (unchanged from v1, now one shared authorised session):
+ AuthorizeSession(rlive|tlive, channel id, drm id derived from the stream URL)
+ -> report the session start -> manifest = the channel's StreamingURL
+ -> DRM from the same session (DrmId -> DRMToday upfront token).
+Manifest and DRM share ONE authorised session through the provider-owned
+playout cache (5 s TTL), so a zap costs one AuthorizeSession, not two.
+
+Return conventions (ChannelManager ABC):
+ * handles_content_id is ids.is_live (digits only); anything else is not ours
+ * unknown live id -> NotFoundError; session not authorised -> EntitlementError
+ * transport / auth / server failures propagate typed (base.errors)
+"""
+
+import threading
+import time
+from typing import Any, Dict, List, Optional, Tuple
+
+from ...base.errors import EntitlementError, NotFoundError
+from ...base.managers import ChannelManager
+from ...base.models import Channel, DRMConfig
+from ...base.utils.logger import logger
+from . import ids
+from .api import api_post
+from .constants import HRTiDefaults
+from .drm import build_widevine, derive_live_drm_id
+from .models import LiveChannelInfo
+
+_LIVE_REFERER = "/"
+
+
+class HRTiChannelManager(ChannelManager):
+ def __init__(
+ self,
+ *,
+ http_manager: Any,
+ auth: Any,
+ country: str,
+ config: Any,
+ playout_cache: Optional[Dict[str, Tuple[Dict, float]]] = None,
+ ):
+ super().__init__(
+ http_manager=http_manager, auth=auth, country=country, config=config
+ )
+ # Provider-owned dict, borrowed by reference (cleared on credential change).
+ self._playout_cache = playout_cache if playout_cache is not None else {}
+ # channel id -> API record (stream URL, radio flag); filled by get_channels()
+ self._records: Dict[str, LiveChannelInfo] = {}
+ self._records_lock = threading.Lock()
+
+ # ------------------------------------------------------------------
+ # Routing
+ # ------------------------------------------------------------------
+ def handles_content_id(self, content_id: str) -> bool:
+ return ids.is_live(content_id)
+
+ # ------------------------------------------------------------------
+ # Channels
+ # ------------------------------------------------------------------
+ def get_channels(self, **kw: Any) -> List[Channel]:
+ result = api_post(
+ http_manager=self.http_manager,
+ session=self.auth,
+ config=self.config,
+ endpoint_key="channels",
+ payload={},
+ referer_path=_LIVE_REFERER,
+ )
+ if not result:
+ logger.warning("HRTi: no channels in the GetChannels response")
+ return []
+
+ records: Dict[str, LiveChannelInfo] = {}
+ channels: List[Channel] = []
+ for raw in result:
+ try:
+ record = LiveChannelInfo.from_api_response(raw)
+ channel = record.to_channel(HRTiDefaults.PROVIDER_NAME, self.country.upper())
+ except (KeyError, TypeError, AttributeError, ValueError) as exc:
+ logger.warning(f"HRTi: skipping malformed channel entry: {exc!r}")
+ continue
+ records[record.content_id] = record
+ channels.append(channel)
+
+ with self._records_lock:
+ self._records = records
+ logger.info(f"HRTi: {len(channels)} channels loaded")
+ return channels
+
+ def clear_caches(self) -> None:
+ """Called by the provider when the credentials / session change."""
+ self._playout_cache.clear()
+ with self._records_lock:
+ self._records = {}
+
+ def _record_for(self, content_id: str) -> LiveChannelInfo:
+ with self._records_lock:
+ record = self._records.get(content_id)
+ if record is None:
+ self.get_channels() # fills self._records
+ with self._records_lock:
+ record = self._records.get(content_id)
+ if record is None:
+ raise NotFoundError(f"HRTi: channel {content_id} is not in the channel list")
+ return record
+
+ # ------------------------------------------------------------------
+ # Authorised playback session (shared by manifest and DRM)
+ # ------------------------------------------------------------------
+ def _session(self, content_id: str) -> Dict[str, Any]:
+ now = time.time()
+ cached = self._playout_cache.get(content_id)
+ if cached and (now - cached[1]) < HRTiDefaults.PLAYOUT_CACHE_TTL:
+ return cached[0]
+
+ record = self._record_for(content_id)
+ content_type = "rlive" if record.is_radio else "tlive"
+ session_data = self.auth.authorize_session(
+ content_type=content_type,
+ content_ref_id=content_id,
+ content_drm_id=derive_live_drm_id(record.stream_url),
+ video_store_ids=None,
+ channel_id=content_id,
+ start_time=None,
+ end_time=None,
+ )
+ if not session_data.get("Authorized", False):
+ raise EntitlementError(
+ f"HRTi: session for channel {content_id} was not authorised"
+ )
+ session_id = session_data.get("SessionId")
+ if session_id:
+ self.auth.report_session_event(session_id, content_id) # best effort
+ self._playout_cache[content_id] = (session_data, now)
+ return session_data
+
+ # ------------------------------------------------------------------
+ # Manifest / DRM
+ # ------------------------------------------------------------------
+ def get_channel_manifest(self, content_id: str, **kw: Any) -> Optional[str]:
+ self._session(content_id) # authorise first (v1 behaviour)
+ record = self._record_for(content_id)
+ if not record.stream_url:
+ logger.warning(f"HRTi: no streaming URL for channel {content_id}")
+ return None
+ return record.stream_url
+
+ def get_channel_drm(self, content_id: str, **kw: Any) -> List[DRMConfig]:
+ # The backend passes drm_variant / preferred_* kwargs: accepted, unused.
+ session_data = self._session(content_id)
+ return build_widevine(
+ self.config,
+ self.auth.user_id,
+ session_data,
+ content_id,
+ referer_path=_LIVE_REFERER,
+ )
diff --git a/lib/streaming_providers/providers/hrti/constants.py b/lib/streaming_providers/providers/hrti/constants.py
index d663fc2..2d7ec44 100644
--- a/lib/streaming_providers/providers/hrti/constants.py
+++ b/lib/streaming_providers/providers/hrti/constants.py
@@ -1,26 +1,47 @@
# streaming_providers/providers/hrti/constants.py
"""
-HRTi provider constants and default configurations
+HRTi constants -- the single source of truth for URLs, header values,
+identifiers and defaults. Managers, session, drm.py and auth.py read from here.
+
+Identity (ONE id everywhere): directory == registry key == provider_name ==
+PROVIDER_NAME == credentials / session key == "hrti". PROVIDER_LABEL ("HRTi")
+is only the UI label.
+
+Dynamic endpoints
+-----------------
+The API hosts are NOT hardcoded: ``env.json`` and ``config.production.json``
+on hrti.hrt.hr name them (``apiUrl``, ``webApiUrl``, merchant, ...). ONE
+HRTiConfig instance is shared by authenticator, session and managers and
+loads them lazily (``ensure_loaded``) -- independent of the login, so it also
+works after a restart with a persisted session. If the bootstrap fails the
+defaults below are used (they equal the live values).
"""
+import threading
+from typing import Any, Dict, Optional
+
+from ...base.errors import ConfigurationError
+from ...base.utils.logger import logger
+
class HRTiDefaults:
- """Default values for HRTi provider"""
-
- # Provider information
+ PROVIDER_NAME = "hrti"
+ PROVIDER_LABEL = "HRTi"
PROVIDER_LOGO = "https://upload.wikimedia.org/wikipedia/en/thumb/9/9e/Logo_of_the_HRT.svg/960px-Logo_of_the_HRT.svg.png"
- PROVIDER_NAME = "HRTi"
+ SUPPORTED_COUNTRIES = ("HR",)
- # Website and base URLs
+ # Website / bootstrap
BASE_WEBSITE = "https://hrti.hrt.hr"
BASE_URL = "https://hrti.hrt.hr"
-
- # Configuration endpoints
ENV_ENDPOINT = f"{BASE_URL}/assets/config/env.json"
CONFIG_ENDPOINT = f"{BASE_URL}/assets/config/config.production.json"
- # API endpoint path templates (will be combined with dynamic base URLs)
- # These are the path parts only - base URLs come from API config
+ # Fallbacks, used only when the bootstrap above fails
+ FALLBACK_HSAPI_URL = "https://hsapi.aviion.tv/client.svc/json"
+ FALLBACK_WEB_API_URL = "api/api/ott"
+ FALLBACK_MERCHANT = "aviion2"
+
+ # Endpoint path templates; base URLs come from the bootstrap
API_ENDPOINT_PATHS = {
"get_ip": "/getIPAddress",
"grant_access": "/GrantAccess",
@@ -38,186 +59,196 @@ class HRTiDefaults:
"watch_later": "/GetWatchLater",
"editors_choice": "/GetEditorsChoice",
}
+ # 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",
+ )
+ HSAPI_ENDPOINTS = ("register_device", "content_ratings", "profiles")
- # Device information (static defaults, may be overridden by config)
- DEVICE_REFERENCE_ID = "6" # String '6' as required by headers
+ # Device information (overridable through the config dict)
+ DEVICE_REFERENCE_ID = "6" # string '6' as required by the headers
OPERATOR_REFERENCE_ID = "hrt"
CONNECTION_TYPE = "LAN/WiFi"
- APPLICATION_VERSION = "5.97.6"
+ APPLICATION_VERSION = "5.97.6" # replaced by env.json's applicationVersion
OS_VERSION = "Linux"
- CLIENT_TYPE = "Chrome 142"
- # User Agent
- USER_AGENT = "Mozilla/5.0 (X11; Linux x86_64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/142.0.0.0 Safari/537.36"
+ # One browser version -> User-Agent AND client type (no drift between them)
+ CHROME_MAJOR = 142
+ CLIENT_TYPE = f"Chrome {CHROME_MAJOR}"
+ USER_AGENT = (
+ "Mozilla/5.0 (X11; Linux x86_64) AppleWebKit/537.36 (KHTML, like Gecko) "
+ f"Chrome/{CHROME_MAJOR}.0.0.0 Safari/537.36"
+ )
- # HTTP settings
+ # Anonymous (client-credentials) access -- the deliberate fallback
+ ANONYMOUS_USERNAME = "anonymoushrt"
+ ANONYMOUS_PASSWORD = "an0nPasshrt"
+
+ # Auth / session
+ TOKEN_LIFETIME_SECONDS = 86400 # GrantAccess carries no lifetime
+ SESSION_BACKOFF_BASE = 2.0 # transient login failures: base * 2**(n-1) s
+ SESSION_BACKOFF_CAP = 60.0
+
+ # HTTP / VOD
DEFAULT_TIMEOUT = 30
-
- # VOD settings
VOD_ITEMS_PER_PAGE = 24
MAX_VOD_PAGES = 100
+ SERIES_PAGE_SIZE = 100
+
+ # DRM (DRMToday, token travels in this header; priority was the lib default)
+ DRM_AUTH_HEADER = "dt-custom-data"
+ WIDEVINE_PRIORITY = 3
+
+ # Caches
+ PLAYOUT_CACHE_TTL = 5.0 # manifest + DRM share one authorised session
class HRTiConfig:
- """Configuration class for HRTi provider"""
+ """Per-instance configuration. ONE instance, shared by every layer."""
- def __init__(self, config_dict: dict = None):
- """Initialize with optional configuration overrides"""
+ def __init__(self, config_dict: Optional[dict] = None):
config = config_dict or {}
-
self.logo = config.get("logo", HRTiDefaults.PROVIDER_LOGO)
-
- # Website and base URLs
self.base_website = config.get("base_website", HRTiDefaults.BASE_WEBSITE)
self.base_url = config.get("base_url", HRTiDefaults.BASE_URL)
- # Dynamic URLs - will be populated from API
- self.hsapi_base_url = config.get("hsapi_base_url", None)
- self.web_api_url = config.get("web_api_url", None)
+ # Dynamic values, filled by the bootstrap
+ self.hsapi_base_url = config.get("hsapi_base_url")
+ self.web_api_url = config.get("web_api_url")
+ self.merchant = config.get("merchant")
+ self.player_license_key = config.get("player_license_key")
+ self.selfcare_url = None
+ self.homepage_url = None
- # Configuration endpoints
self.env_endpoint = config.get("env_endpoint", HRTiDefaults.ENV_ENDPOINT)
self.config_endpoint = config.get("config_endpoint", HRTiDefaults.CONFIG_ENDPOINT)
-
- # API endpoints - start with paths only, build full URLs after API call
self.api_endpoint_paths = config.get(
"api_endpoint_paths", HRTiDefaults.API_ENDPOINT_PATHS.copy()
)
- self.api_endpoints = {} # Will be populated in update_from_api
+ self.api_endpoints: Dict[str, str] = {}
- # Device configuration
- self.device_reference_id = config.get(
- "device_reference_id", HRTiDefaults.DEVICE_REFERENCE_ID
- )
- self.operator_reference_id = config.get(
- "operator_reference_id", HRTiDefaults.OPERATOR_REFERENCE_ID
- )
- self.merchant = config.get("merchant", None) # Will be set from API
- self.player_license_key = config.get("player_license_key", None) # Will be set from API
+ self.device_reference_id = config.get("device_reference_id", HRTiDefaults.DEVICE_REFERENCE_ID)
+ self.operator_reference_id = config.get("operator_reference_id", HRTiDefaults.OPERATOR_REFERENCE_ID)
self.connection_type = config.get("connection_type", HRTiDefaults.CONNECTION_TYPE)
- self.application_version = config.get(
- "application_version", HRTiDefaults.APPLICATION_VERSION
- )
+ self.application_version = config.get("application_version", HRTiDefaults.APPLICATION_VERSION)
self.os_version = config.get("os_version", HRTiDefaults.OS_VERSION)
self.client_type = config.get("client_type", HRTiDefaults.CLIENT_TYPE)
-
- # HTTP settings
self.user_agent = config.get("user_agent", HRTiDefaults.USER_AGENT)
self.timeout = config.get("timeout", HRTiDefaults.DEFAULT_TIMEOUT)
-
- # VOD settings
self.vod_items_per_page = config.get("vod_items_per_page", HRTiDefaults.VOD_ITEMS_PER_PAGE)
- def _build_api_endpoints(self):
- """Build full API endpoints from dynamic base URLs and static paths"""
- endpoints = {}
+ self._http: Any = None
+ self._loaded = False
+ self.loaded_from_fallback = False
+ self._lock = threading.RLock()
- # Build endpoints using webApiUrl (for OTT endpoints)
- if self.web_api_url:
- base_api_url = f"{self.base_url}/{self.web_api_url}"
- ott_endpoints = [
- "get_ip", "grant_access", "channels", "programme",
- "authorize_session", "report_session", "catalogue_structure",
- "catalogue", "vod_details", "episodes", "watch_later",
- "editors_choice"
- ]
- for key in ott_endpoints:
- if key in self.api_endpoint_paths:
- endpoints[key] = f"{base_api_url}{self.api_endpoint_paths[key]}"
+ # ------------------------------------------------------------------
+ # Lazy dynamic endpoints
+ # ------------------------------------------------------------------
+ def attach_http(self, http_manager: Any) -> None:
+ """Give the config the HTTP manager used for the bootstrap fetch."""
+ self._http = http_manager
- # Build endpoints using hsapi_base_url (for HSAPI endpoints)
- if self.hsapi_base_url:
- hsapi_endpoints = ["register_device", "content_ratings", "profiles"]
- for key in hsapi_endpoints:
- if key in self.api_endpoint_paths:
- endpoints[key] = f"{self.hsapi_base_url}{self.api_endpoint_paths[key]}"
+ @property
+ def is_loaded(self) -> bool:
+ return self._loaded
- return endpoints
+ def ensure_loaded(self) -> None:
+ """Fetch env.json + config.production.json once; fall back to defaults."""
+ if self._loaded:
+ return
+ with self._lock:
+ if self._loaded:
+ return
+ if self._http is None:
+ raise ConfigurationError(
+ "HRTi: HTTP manager not attached to the config (attach_http)"
+ )
+ try:
+ env_resp = self._http.get(self.env_endpoint, operation="api")
+ env_resp.raise_for_status()
+ cfg_resp = self._http.get(self.config_endpoint, operation="api")
+ cfg_resp.raise_for_status()
+ self.update_from_api(env_resp.json(), cfg_resp.json())
+ self.loaded_from_fallback = False
+ logger.debug("HRTi environment configuration loaded")
+ except Exception as exc: # deliberate: bootstrap failure -> documented defaults
+ logger.warning(f"HRTi: environment config unavailable ({exc}); using defaults")
+ self._build_fallback_endpoints()
+ self.loaded_from_fallback = True
+ self._loaded = True
- def update_from_api(self, env_data: dict, config_data: dict):
- """Update configuration from API responses"""
- try:
- # Update from env data
- if "applicationVersion" in env_data:
- self.application_version = env_data["applicationVersion"]
+ def endpoint(self, key: str) -> str:
+ """Full URL of a named endpoint (loads the dynamic config on first use)."""
+ self.ensure_loaded()
+ url = self.api_endpoints.get(key)
+ if not url:
+ raise ConfigurationError(f"HRTi: no endpoint configured for '{key}'")
+ return url
- # Update from config data
- if "apiUrl" in config_data:
- self.hsapi_base_url = config_data["apiUrl"]
-
- if "webApiUrl" in config_data:
- self.web_api_url = config_data["webApiUrl"]
-
- # Update operator-specific settings
- if "operators" in config_data and config_data["operators"]:
- operator = config_data["operators"][0]
- if "playerMerchant" in operator:
- self.merchant = operator["playerMerchant"]
- if "playerLicenseKey" in operator:
- self.player_license_key = operator["playerLicenseKey"]
- if "selfcareUrl" in operator:
- self.selfcare_url = operator["selfcareUrl"]
- if "homepageUrl" in operator:
- self.homepage_url = operator["homepageUrl"]
-
- # Build API endpoints from dynamic URLs
- self.api_endpoints = self._build_api_endpoints()
-
- # Log what we got from API (debug)
- import logging
- logging.debug(f"HRTi config updated from API - hsapi_url: {self.hsapi_base_url}, "
- f"web_api_url: {self.web_api_url}, merchant: {self.merchant}")
-
- except Exception as e:
- import logging
- logging.debug(f"Error updating HRTi config from API: {e}")
- # Fallback to build endpoints with defaults if API update fails
- self._build_fallback_endpoints()
-
- def _build_fallback_endpoints(self):
- """Build fallback API endpoints if dynamic config fails"""
- # Use default hsapi URL if not set
- if not self.hsapi_base_url:
- self.hsapi_base_url = "https://hsapi.aviion.tv/client.svc/json"
-
- # Use default web API URL if not set
- if not self.web_api_url:
- self.web_api_url = "api/api/ott"
-
- # Use default merchant if not set
- if not self.merchant:
- self.merchant = "aviion2"
+ def get_merchant(self) -> str:
+ self.ensure_loaded()
+ return self.merchant or HRTiDefaults.FALLBACK_MERCHANT
+ def update_from_api(self, env_data: dict, config_data: dict) -> None:
+ if "applicationVersion" in env_data:
+ self.application_version = env_data["applicationVersion"]
+ if "apiUrl" in config_data:
+ self.hsapi_base_url = config_data["apiUrl"]
+ if "webApiUrl" in config_data:
+ self.web_api_url = config_data["webApiUrl"]
+ operators = config_data.get("operators") or []
+ if operators:
+ operator = operators[0]
+ self.merchant = operator.get("playerMerchant", self.merchant)
+ self.player_license_key = operator.get("playerLicenseKey", self.player_license_key)
+ self.selfcare_url = operator.get("selfcareUrl", self.selfcare_url)
+ self.homepage_url = operator.get("homepageUrl", self.homepage_url)
self.api_endpoints = self._build_api_endpoints()
- def get_base_headers(self) -> dict:
- """Get base HTTP headers"""
- return {
+ def _build_fallback_endpoints(self) -> None:
+ self.hsapi_base_url = self.hsapi_base_url or HRTiDefaults.FALLBACK_HSAPI_URL
+ self.web_api_url = self.web_api_url or HRTiDefaults.FALLBACK_WEB_API_URL
+ self.merchant = self.merchant or HRTiDefaults.FALLBACK_MERCHANT
+ self.api_endpoints = self._build_api_endpoints()
+
+ def _build_api_endpoints(self) -> Dict[str, str]:
+ endpoints: Dict[str, str] = {}
+ if self.web_api_url:
+ base = f"{self.base_url}/{self.web_api_url}"
+ for key in HRTiDefaults.WEB_API_ENDPOINTS:
+ if key in self.api_endpoint_paths:
+ endpoints[key] = f"{base}{self.api_endpoint_paths[key]}"
+ if self.hsapi_base_url:
+ for key in HRTiDefaults.HSAPI_ENDPOINTS:
+ if key in self.api_endpoint_paths:
+ endpoints[key] = f"{self.hsapi_base_url}{self.api_endpoint_paths[key]}"
+ return endpoints
+
+ # ------------------------------------------------------------------
+ # Header builders
+ # ------------------------------------------------------------------
+ def api_headers(
+ self,
+ token: Optional[str] = None,
+ device_id: Optional[str] = None,
+ ip_address: Optional[str] = None,
+ referer_path: str = "/",
+ ) -> Dict[str, str]:
+ """Headers of the manager API calls (NOT for CDN / manifest requests)."""
+ headers = {
+ "deviceid": device_id or "",
+ "devicetypeid": self.device_reference_id,
+ "ipaddress": ip_address or "",
+ "operatorreferenceid": self.operator_reference_id,
+ "origin": self.base_website,
+ "referer": f"{self.base_website}{referer_path}",
"User-Agent": self.user_agent,
"Accept": "application/json",
"Content-Type": "application/json",
}
-
- def get_auth_headers(
- self, device_id: str = None, ip_address: str = None, token: str = None
- ) -> dict:
- """Get authenticated headers for API requests"""
- headers = self.get_base_headers()
-
- if device_id:
- headers["deviceid"] = device_id
- if ip_address:
- headers["ipaddress"] = ip_address
if token:
headers["authorization"] = f"Client {token}"
-
- headers.update(
- {
- "devicetypeid": self.device_reference_id,
- "operatorreferenceid": self.operator_reference_id,
- "origin": self.base_website,
- "referer": self.base_website,
- }
- )
-
- return headers
\ No newline at end of file
+ return headers
diff --git a/lib/streaming_providers/providers/hrti/drm.py b/lib/streaming_providers/providers/hrti/drm.py
new file mode 100644
index 0000000..2882818
--- /dev/null
+++ b/lib/streaming_providers/providers/hrti/drm.py
@@ -0,0 +1,81 @@
+# streaming_providers/providers/hrti/drm.py
+"""
+Shared DRM building for the live and the VOD manager (one place to fix a
+header / license-data bug).
+
+HRTi licences go to DRMToday. The "upfront token" is NOT a bearer token: it is
+base64(JSON{userId, sessionId, merchant}) bound to an authorised playback
+session (AuthorizeSession -> DrmId). Sent in the ``dt-custom-data`` header.
+
+Known limitation (README §11): the DRMConfig lives as long as inputstream.
+adaptive caches it; a fresh zap re-invokes get_drm and authorises again.
+"""
+
+import base64
+import json
+from typing import Any, Dict, List, NamedTuple, Optional
+from urllib.parse import urlparse
+
+from ...base.errors import ServerError
+from ...base.models import DRMConfig
+from ...base.utils.logger import logger
+from ..lib_drmtoday import create_drmtoday_widevine_config
+from .constants import HRTiConfig, HRTiDefaults
+
+
+class VodSessionParams(NamedTuple):
+ content_ref_id: str
+ content_drm_id: str
+
+
+def derive_live_drm_id(streaming_url: str) -> Optional[str]:
+ """First two path segments joined by '_' (/cdn1oiv/hrtliveorigin/... ->
+ "cdn1oiv_hrtliveorigin"); None when the path is too short."""
+ parts = urlparse(streaming_url).path.strip("/").split("/")
+ if len(parts) >= 2:
+ return f"{parts[0]}_{parts[1]}"
+ return None
+
+
+def derive_vod_session_params(file_name: str) -> VodSessionParams:
+ """
+ FileName = https://cdn1oiv.akamaized.net/hrtvodorigin/REF.smil/manifest.mpd
+ ContentReferenceId = "REF" (stem without .smil)
+ ContentDrmId = "hrtvodorigin_REF.smil"
+ """
+ parts = urlparse(file_name).path.strip("/").split("/")
+ if len(parts) < 2:
+ raise ServerError(f"HRTi: cannot derive session parameters from FileName {file_name!r}")
+ origin, ref_with_ext = parts[0], parts[1]
+ ref_id = ref_with_ext[: -len(".smil")] if ref_with_ext.endswith(".smil") else ref_with_ext
+ return VodSessionParams(content_ref_id=ref_id, content_drm_id=f"{origin}_{ref_with_ext}")
+
+
+def build_license_data(user_id: str, session_id: str, merchant: str) -> str:
+ payload = {"userId": user_id or "", "sessionId": session_id, "merchant": merchant}
+ return base64.b64encode(json.dumps(payload).encode("utf-8")).decode("utf-8")
+
+
+def build_widevine(
+ config: HRTiConfig,
+ user_id: str,
+ session_data: Dict[str, Any],
+ content_id: str,
+ referer_path: str,
+) -> List[DRMConfig]:
+ """Widevine config from an AUTHORISED session. [] when the session has no
+ DrmId (logged; the stream cannot be licensed)."""
+ drm_id = session_data.get("DrmId")
+ if not drm_id:
+ logger.error(f"HRTi: no DrmId in session data for {content_id}")
+ return []
+ license_data = build_license_data(user_id, drm_id, config.get_merchant())
+ drm_config = create_drmtoday_widevine_config(
+ upfront_token=license_data,
+ origin=config.base_website,
+ referer=f"{config.base_website}{referer_path}",
+ user_agent=config.user_agent,
+ priority=HRTiDefaults.WIDEVINE_PRIORITY,
+ auth_header_name=HRTiDefaults.DRM_AUTH_HEADER,
+ )
+ return [drm_config]
diff --git a/lib/streaming_providers/providers/hrti/ids.py b/lib/streaming_providers/providers/hrti/ids.py
new file mode 100644
index 0000000..ce6535e
--- /dev/null
+++ b/lib/streaming_providers/providers/hrti/ids.py
@@ -0,0 +1,46 @@
+# streaming_providers/providers/hrti/ids.py
+"""
+The ONE place that knows the HRTi content-id grammar. Both managers'
+handles_content_id() call these predicates, so they are complementary by
+construction (README §7). Pure functions, no I/O.
+
+Grammar
+-------
+ "" VOD root (navigation, not playable)
+ "catalogue_[" VOD folder (navigation)
+ "series_--" VOD series season (navigation)
+ "special_watch_later" VOD special list (navigation)
+ "special_editors_choice" VOD special list (navigation)
+ "details_][" one playable VOD item (explicit)
+ "][" not purely digits one playable VOD item (UUID or "1870699590_delin")
+ "" live channel / radio (e.g. "40013")
+
+Navigation ids are claimed by NEITHER manager for playback: they are browsed
+through get_vod_category, which the backend calls directly.
+"""
+
+NAV_PREFIXES = ("catalogue_", "series_", "special_")
+DETAILS_PREFIX = "details_"
+
+
+def is_nav(content_id: str) -> bool:
+ return any(content_id.startswith(p) for p in NAV_PREFIXES)
+
+
+def is_live(content_id: str) -> bool:
+ return bool(content_id) and content_id.isdigit()
+
+
+def is_vod(content_id: str) -> bool:
+ if not content_id or is_nav(content_id):
+ return False
+ if content_id.startswith(DETAILS_PREFIX):
+ return True
+ return not content_id.isdigit()
+
+
+def vod_ref(content_id: str) -> str:
+ """The bare ReferenceId of a playable VOD id (strips ``details_``)."""
+ if content_id.startswith(DETAILS_PREFIX):
+ return content_id[len(DETAILS_PREFIX):]
+ return content_id
diff --git a/lib/streaming_providers/providers/hrti/models.py b/lib/streaming_providers/providers/hrti/models.py
index 43707b3..6fc42ac 100644
--- a/lib/streaming_providers/providers/hrti/models.py
+++ b/lib/streaming_providers/providers/hrti/models.py
@@ -1,26 +1,28 @@
-# [file name]: models.py
-# [file content begin]
# streaming_providers/providers/hrti/models.py
-from dataclasses import dataclass
-from typing import Any, Dict, Optional
+"""
+Provider-local models: credentials, auth token, and strict parsers for the
+live-channel and VOD-details responses. Parsing is STRICT (KeyError /
+ValueError on missing required fields); the managers parse entry by entry and
+skip malformed entries, so one bad entry cannot take a whole list down.
+"""
+
+from dataclasses import dataclass, field
+from typing import Any, Dict, List, Optional
from ...base.auth.base_auth import BaseAuthToken
from ...base.auth.credentials import UserPasswordCredentials
+from ...base.models import Channel
from .constants import HRTiDefaults
@dataclass
class HRTiCredentials(UserPasswordCredentials):
- """
- HRTi specific username/password credentials
- """
+ """HRTi username/password credentials."""
def __init__(self, username: str, password: str):
super().__init__(username=username, password=password)
- # credential_type is not a property in base class, so we don't set it
def to_auth_payload(self) -> Dict[str, Any]:
- """Convert to authentication payload for HRTi"""
return {
"Username": self.username,
"Password": self.password,
@@ -28,14 +30,11 @@ class HRTiCredentials(UserPasswordCredentials):
}
def validate(self) -> bool:
- """Validate HRTi credentials"""
return bool(self.username and self.password)
class HRTiAuthToken(BaseAuthToken):
- """
- HRTi specific authentication token
- """
+ """HRTi authentication token (persisted flat via to_dict / from_dict)."""
def __init__(
self,
@@ -48,7 +47,6 @@ class HRTiAuthToken(BaseAuthToken):
valid_to: str = "",
refresh_token: Optional[str] = None,
):
-
super().__init__(
access_token=access_token,
token_type=token_type,
@@ -56,198 +54,101 @@ class HRTiAuthToken(BaseAuthToken):
issued_at=issued_at,
refresh_token=refresh_token,
)
-
- # HRTi specific fields
self.user_id = user_id
self.valid_from = valid_from
self.valid_to = valid_to
- # Don't set credential_type - it's not in base class
def to_dict(self) -> Dict[str, Any]:
- """Convert token to dictionary"""
base_dict = {
"access_token": self.access_token,
"token_type": self.token_type,
"expires_in": self.expires_in,
"issued_at": self.issued_at,
"refresh_token": self.refresh_token or "",
+ "user_id": self.user_id,
+ "valid_from": self.valid_from,
+ "valid_to": self.valid_to,
}
-
- # Add HRTi specific fields
- base_dict.update(
- {
- "user_id": self.user_id,
- "valid_from": self.valid_from,
- "valid_to": self.valid_to,
- }
- )
-
- # Add auth_level if available
if hasattr(self, "auth_level") and self.auth_level:
base_dict["auth_level"] = self.auth_level.value
-
return base_dict
@classmethod
def from_dict(cls, data: Dict[str, Any]) -> "HRTiAuthToken":
- """Create token from dictionary"""
return cls(
access_token=data["access_token"],
token_type=data.get("token_type", "Client"),
- expires_in=data.get("expires_in", 86400),
+ expires_in=data.get("expires_in", HRTiDefaults.TOKEN_LIFETIME_SECONDS),
issued_at=data.get("issued_at", 0),
user_id=data.get("user_id", ""),
valid_from=data.get("valid_from", ""),
valid_to=data.get("valid_to", ""),
- refresh_token=data.get("refresh_token"),
+ refresh_token=data.get("refresh_token") or None,
)
- def is_valid(self) -> bool:
- """Check if token is still valid"""
- import time
-
- if not self.access_token:
- return False
-
- # Basic expiration check
- current_time = time.time()
- buffer_time = 300 # 5 minutes buffer
-
- return current_time < (self.issued_at + self.expires_in - buffer_time)
-
@dataclass
-class HRTiChannel:
- """
- HRTi channel information
- """
+class LiveChannelInfo:
+ """One entry of GetChannels. The record stays in the ChannelManager."""
- id: str
+ content_id: str
name: str
- reference_id: str
- streaming_url: str
- icon_url: str
+ stream_url: str = ""
is_radio: bool = False
- description: Optional[str] = None
- sort_order: int = 0
-
- def __post_init__(self):
- """Validate channel data after initialization"""
- if not self.id or not self.name or not self.reference_id:
- raise ValueError("Channel must have id, name, and reference_id")
+ icon_url: str = ""
@classmethod
- def from_api_response(cls, data: Dict[str, Any]) -> "HRTiChannel":
- """Create channel from API response data"""
+ def from_api_response(cls, data: Dict[str, Any]) -> "LiveChannelInfo":
+ # The API spells it 'ReferenceID' (capital ID) here.
+ name = data["Name"]
+ content_id = data["ReferenceID"]
+ if not name or not content_id:
+ raise ValueError("channel entry without name or ReferenceID")
return cls(
- id=data.get("ReferenceId", ""),
- name=data.get("Name", ""),
- reference_id=data.get("ReferenceId", ""),
- streaming_url=data.get("StreamingURL", ""),
- icon_url=data.get("Icon", ""),
- is_radio=data.get("Radio", False),
- description=data.get("Description"),
- sort_order=data.get("SortOrder", 0),
+ content_id=str(content_id),
+ name=name,
+ stream_url=data.get("StreamingURL") or "",
+ is_radio=bool(data.get("Radio", False)),
+ icon_url=data.get("Icon") or "",
)
- def to_dict(self) -> Dict[str, Any]:
- """Convert channel to dictionary"""
- return {
- "id": self.id,
- "name": self.name,
- "reference_id": self.reference_id,
- "streaming_url": self.streaming_url,
- "icon_url": self.icon_url,
- "is_radio": self.is_radio,
- "description": self.description,
- "sort_order": self.sort_order,
- }
+ def to_channel(self, provider_name: str, country: str) -> Channel:
+ """Same fields as v1: a session-manifest channel whose manifest_script
+ carries the stream URL; radio keeps content_type "AUDIO" (the clients
+ and the v1 code distinguish radio by it)."""
+ return Channel(
+ name=self.name,
+ content_id=self.content_id,
+ provider=provider_name,
+ logo_url=self.icon_url,
+ mode="live",
+ session_manifest=True,
+ manifest=None,
+ manifest_script=self.stream_url,
+ content_type="AUDIO" if self.is_radio else "LIVE",
+ country=country,
+ is_radio=self.is_radio,
+ language="hr",
+ use_cdm=True,
+ cdm_type="widevine",
+ )
@dataclass
-class HRTiEpgEntry:
- """
- HRTi EPG (Electronic Program Guide) entry
- """
+class VodDetails:
+ """Playback-relevant part of GetVodDetails."""
- reference_id: str
- title: str
- description_short: str
- description_long: str
- start_time: str
- end_time: str
- image_url: str
- channel_reference_id: str
- content_rating: Optional[str] = None
- episode_number: Optional[str] = None
- season_number: Optional[str] = None
+ ref_id: str
+ file_name: str = "" # the DASH manifest URL
+ svod_stores: List[str] = field(default_factory=list)
@classmethod
- def from_api_response(cls, data: Dict[str, Any], channel_id: str) -> "HRTiEpgEntry":
- """Create EPG entry from API response"""
+ def from_api_response(cls, data: Dict[str, Any]) -> "VodDetails":
+ ref_id = data.get("ReferenceID") or data.get("ReferenceId")
+ if not ref_id:
+ raise ValueError("GetVodDetails response without ReferenceID")
return cls(
- reference_id=data.get("ReferenceId", ""),
- title=data.get("Title", ""),
- description_short=data.get("DescriptionShort", ""),
- description_long=data.get("DescriptionLong", ""),
- start_time=data.get("TimeStart", ""),
- end_time=data.get("TimeEnd", ""),
- image_url=data.get("ImagePath", ""),
- channel_reference_id=channel_id,
- content_rating=data.get("ContentRating"),
- episode_number=data.get("EpisodeNr"),
- season_number=data.get("SeasonNr"),
+ ref_id=str(ref_id),
+ file_name=data.get("FileName") or "",
+ svod_stores=list(data.get("SVODVideostores") or []),
)
-
- def to_dict(self) -> Dict[str, Any]:
- """Convert EPG entry to dictionary"""
- return {
- "reference_id": self.reference_id,
- "title": self.title,
- "description_short": self.description_short,
- "description_long": self.description_long,
- "start_time": self.start_time,
- "end_time": self.end_time,
- "image_url": self.image_url,
- "channel_reference_id": self.channel_reference_id,
- "content_rating": self.content_rating,
- "episode_number": self.episode_number,
- "season_number": self.season_number,
- }
-
-
-@dataclass
-class HRTiSession:
- """
- HRTi playback session information
- """
-
- session_id: str
- authorized: bool
- drm_id: str
- channel_reference_id: str
- content_reference_id: str
-
- @classmethod
- def from_api_response(cls, data: Dict[str, Any]) -> "HRTiSession":
- """Create session from API response"""
- return cls(
- session_id=data.get("SessionId", ""),
- authorized=data.get("Authorized", False),
- drm_id=data.get("DrmId", ""),
- channel_reference_id=data.get("ChannelReferenceId", ""),
- content_reference_id=data.get("ContentReferenceId", ""),
- )
-
- def to_dict(self) -> Dict[str, Any]:
- """Convert session to dictionary"""
- return {
- "session_id": self.session_id,
- "authorized": self.authorized,
- "drm_id": self.drm_id,
- "channel_reference_id": self.channel_reference_id,
- "content_reference_id": self.content_reference_id,
- }
-
-
-# [file content end]
diff --git a/lib/streaming_providers/providers/hrti/provider.py b/lib/streaming_providers/providers/hrti/provider.py
index b9eafb7..c0404d1 100644
--- a/lib/streaming_providers/providers/hrti/provider.py
+++ b/lib/streaming_providers/providers/hrti/provider.py
@@ -1,44 +1,59 @@
# streaming_providers/providers/hrti/provider.py
-import json
-import datetime
-import traceback
-from urllib.parse import urlparse
-from typing import ClassVar, Dict, List, Optional, Union
+"""
+HRTi provider -- orchestrator only.
-import requests
+Owns the SHARED resources (ONE config, http_manager, authenticator + session,
+caches) and builds the managers; everything generic (capability flags,
+content-id routing, manifest / DRM delegation, legacy surface) comes from
+ManagedProvider.
-from ...base.models import DRMConfig, StreamingChannel, Event
+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.
+"""
+
+from typing import ClassVar, Dict, List, Optional
+
+from ...base.managed_provider import ManagedProvider
from ...base.models.proxy_models import ProxyConfig
-from ...base.provider import AuthType, StreamingProvider
-from ...base.utils import logger
-from ..lib_drmtoday import create_drmtoday_widevine_config
from .auth import HRTiAuthenticator
+from .channel_manager import HRTiChannelManager
from .constants import HRTiConfig, HRTiDefaults
+from .session import HRTiSession
from .vod_manager import HRTiVodManager
-# Navigation-only prefixes — these content_ids are category/series containers.
-# They are never directly playable and must never be sent to get_manifest/get_drm.
-_NAV_PREFIXES = ("catalogue_", "series_", "special_")
-
-# Playable VOD prefix — "details_{ref_id}" is an explicit single-item request.
-_DETAILS_PREFIX = "details_"
+# Only the provider is public (discovery takes the FIRST StreamingProvider
+# subclass in the package namespace -- never export ManagedProvider).
+__all__ = ["HRTiProvider"]
-class HRTiProvider(StreamingProvider):
- """
- HRTi (Croatian Radio Television Internet) provider implementation.
- """
-
- # ============================================================================
- # STATIC METADATA
- # ============================================================================
- PROVIDER_LABEL: ClassVar[str] = "HRTi"
- SUPPORTED_AUTH_TYPES: ClassVar[List[str]] = [
- "client_credentials",
- "user_credentials",
- ]
+class HRTiProvider(ManagedProvider):
+ PROVIDER_LABEL: ClassVar[str] = HRTiDefaults.PROVIDER_LABEL
PROVIDER_LOGO: ClassVar[str] = HRTiDefaults.PROVIDER_LOGO
- SUPPORTED_COUNTRIES: ClassVar[List[str]] = ["HR"]
+ SUPPORTED_AUTH_TYPES: ClassVar[List[str]] = ["client_credentials", "user_credentials"]
+ SUPPORTED_COUNTRIES: ClassVar[List[str]] = list(HRTiDefaults.SUPPORTED_COUNTRIES)
+
+ # Manifest and DRM share one authorised session -> DRM lives in the managers.
+ DRM_IN_MANAGERS: ClassVar[bool] = True
+ # v1 had no manager header hooks and sent {} for manifest/segments; the DRM
+ # config carries Origin/Referer/UA itself. Decided: keep {}.
+ HEADERS_FROM_MANAGERS: ClassVar[bool] = False
+
+ @property
+ def provider_name(self) -> str:
+ """Must equal the directory name / registry key."""
+ return HRTiDefaults.PROVIDER_NAME
+
+ @property
+ def uses_dynamic_manifests(self) -> bool:
+ # v1 value (True) differs from the legacy default (False): keep.
+ return True
+
+ @property
+ def provider_logo(self) -> str:
+ # v1: the logo can be overridden through the config dict.
+ return self.hrti_config.logo
def __init__(
self,
@@ -46,551 +61,65 @@ class HRTiProvider(StreamingProvider):
config: Optional[Dict] = None,
proxy_config: Optional[ProxyConfig] = None,
):
- super().__init__(country)
+ super().__init__(country=country)
+ # Registry passes UPPERCASE for single-country providers; the proxy
+ # lookup keys by lowercase. NOTE: the authenticator keeps its FLAT
+ # storage keys (country=None) -- existing credentials/session/device
+ # id are stored without a country and must stay reachable.
+ self.country = self.country.lower()
- # Initialize configuration with overrides
+ # 1. ONE config object, shared by authenticator, session and managers.
self.hrti_config = HRTiConfig(config)
- self.channels_cache = None
- # Short-lived session cache: {channel_id: session_data}.
- # Populated by _get_live_manifest so that a subsequent _get_live_drm
- # call for the same channel (same playback request) can reuse the
- # already-authorized session instead of making a second round trip.
- self._session_cache: Dict[str, dict] = {}
-
- # Setup HTTP manager using abstraction
+ # 2. HTTP manager (proxy: ctor arg -> ProxyConfigManager -> global).
self.http_manager = self._setup_http_manager(
- provider_name="hrti",
+ provider_name=HRTiDefaults.PROVIDER_NAME,
proxy_config=proxy_config,
user_agent=self.hrti_config.user_agent,
timeout=self.hrti_config.timeout,
)
+ # The dynamic endpoints (env.json / config.production.json) are fetched
+ # lazily through this manager on first use -- not in __init__.
+ self.hrti_config.attach_http(self.http_manager)
- # Initialize authenticator and share HTTP manager
+ # 3. Auth: stateful authenticator (kept: the legacy auth mixin reads
+ # self.authenticator) + the session adapter the managers use.
self.authenticator = HRTiAuthenticator(
- proxy_config=proxy_config, http_manager=self.http_manager
+ http_manager=self.http_manager, config=self.hrti_config
+ )
+ self.auth = HRTiSession(authenticator=self.authenticator, config=self.hrti_config)
+
+ # 4. Provider-owned caches, borrowed by the managers by reference.
+ self._playout_cache: Dict = {}
+ self.auth.on_invalidate(self._clear_caches)
+
+ # 5. Managers. self.channels is the ChannelManager (never a list).
+ self._init_managers()
+
+ def _clear_caches(self) -> None:
+ self._playout_cache.clear()
+ if self.channels is not None:
+ self.channels.clear_caches()
+
+ # ------------------------------------------------------------------
+ # Manager factories (None = capability not offered)
+ # ------------------------------------------------------------------
+ def _build_channels(self):
+ return HRTiChannelManager(
+ http_manager=self.http_manager,
+ auth=self.auth,
+ country=self.country,
+ config=self.hrti_config,
+ playout_cache=self._playout_cache,
)
- # Share HTTP manager
- self.http_manager = self._share_http_manager_with_authenticator(self.authenticator)
-
- # Initialize VOD manager
- self._vod_manager = HRTiVodManager(self)
-
- try:
- self.authenticator.get_bearer_token()
- logger.debug("HRTi authentication successful during initialization")
- except Exception as e:
- logger.warning(f"HRTi could not authenticate during initialization: {e}")
-
- # ============================================================================
- # Provider properties
- # ============================================================================
-
- @property
- def provider_name(self) -> str:
- return "hrti"
-
- @property
- def provider_label(self) -> str:
- return self.get_static_label(self.country)
-
- @property
- def provider_logo(self) -> str:
- return self.hrti_config.logo or self.PROVIDER_LOGO
-
- @property
- def uses_dynamic_manifests(self) -> bool:
- return True
-
- @property
- def implements_epg(self) -> bool:
- return False
-
- @property
- def implements_vod(self) -> bool:
- return True
-
- @property
- def supported_auth_types(self) -> List[str]:
- return self.SUPPORTED_AUTH_TYPES
-
- # ============================================================================
- # Internal helpers
- # ============================================================================
-
- def _get_hrti_authenticated_headers(self) -> Dict[str, str]:
- """Build authenticated headers for HRTi API requests."""
- return self._build_provider_headers(
- auth_type=AuthType.CLIENT,
- token_key="authorization", # HRTi uses lowercase
- provider_headers={
- "deviceid": self.authenticator.get_device_id(),
- "devicetypeid": self.hrti_config.device_reference_id,
- "ipaddress": self.authenticator.get_ip_address(),
- "operatorreferenceid": self.hrti_config.operator_reference_id,
- "origin": self.hrti_config.base_website,
- "referer": f"{self.hrti_config.base_website}/login",
- },
+ def _build_vod(self):
+ return HRTiVodManager(
+ http_manager=self.http_manager,
+ auth=self.auth,
+ country=self.country,
+ config=self.hrti_config,
+ playout_cache=self._playout_cache,
)
- @staticmethod
- def _is_nav_node(content_id: str) -> bool:
- """Return True for navigation-only ids (category/series/season/special).
- These are never playable and must not be passed to get_manifest/get_drm."""
- return any(content_id.startswith(p) for p in _NAV_PREFIXES)
-
- @staticmethod
- def _is_playable_vod(content_id: str) -> bool:
- """Return True for ids that represent a playable VOD item.
-
- Two forms are playable:
- - ``details:{ref_id}`` — explicit single-item request
- - bare ref-id (no prefix) — raw ReferenceId from catalogue listing
- (UUIDs like "85b273d1-..." or legacy ids like "1870699590_delin")
-
- Live channel ids are also bare numeric strings (e.g. "40013") so this
- method alone does not distinguish VOD from live — callers must first
- exclude nav nodes, then check against the live channel list.
- """
- if any(content_id.startswith(p) for p in _NAV_PREFIXES):
- return False
- # "details_" prefix is an explicit VOD request
- if content_id.startswith(_DETAILS_PREFIX):
- return True
- # Bare id — could be VOD or live; live ids are short numerics.
- # We treat any non-numeric bare id as VOD; live channel lookup
- # will confirm or deny for numeric ids.
- return not content_id.isdigit()
-
- @staticmethod
- def _derive_content_drm_id(streaming_url: str) -> Optional[str]:
- """
- Derive the content DRM ID from a streaming URL.
-
- HRTi expects the first two path segments joined by '_'.
- Example: /cdn1oiv/hrtliveorigin/... → "cdn1oiv_hrtliveorigin"
- """
- parts = urlparse(streaming_url).path.strip("/").split("/")
- if len(parts) >= 2:
- return f"{parts[0]}_{parts[1]}"
- return None
-
- def _find_channel(self, content_id: str) -> Optional[StreamingChannel]:
- """Return the cached channel for *content_id*, fetching if needed."""
- channels = self.channels if (hasattr(self, "channels") and self.channels) else self.get_channels()
- for ch in channels:
- # StreamingChannel is constructed with content_id= kwarg; some base-class
- # versions also alias it as channel_id — check both to be safe.
- ch_id = getattr(ch, "content_id", None) or getattr(ch, "channel_id", None)
- if ch_id == content_id:
- return ch
- return None
-
- def _authorize_live_session(self, content_id: str) -> Optional[dict]:
- """
- Authorize a live playback session for *content_id*.
-
- Looks up the channel to determine the correct content_type
- (``rlive`` for radio, ``tlive`` for TV) and derives the DRM ID
- from the streaming URL. Reports the session-start event on
- success and stores the result in ``_session_cache``.
-
- Returns the raw session dict, or None on failure.
- """
- channel = self._find_channel(content_id)
- if not channel:
- logger.error(f"Channel {content_id} not found for session authorization")
- return None
-
- content_type = "rlive" if channel.content_type == "AUDIO" else "tlive"
- content_drm_id = self._derive_content_drm_id(channel.manifest_script or "")
-
- logger.debug(
- f"Authorizing session — channel: {content_id}, "
- f"content_type: {content_type}, drm_id: {content_drm_id}"
- )
-
- session_data = self.authenticator.authorize_session(
- content_type=content_type,
- content_ref_id=content_id,
- content_drm_id=content_drm_id,
- video_store_ids=None,
- channel_id=content_id,
- start_time=None,
- end_time=None,
- )
-
- if not session_data or not session_data.get("Authorized", False):
- logger.error(f"Session authorization failed for channel {content_id}")
- return None
-
- logger.debug(f"Session authorized for channel {content_id}")
-
- session_id = session_data.get("SessionId")
- if session_id:
- self.authenticator.report_session_event(session_id, content_id)
-
- # Cache so a same-request get_drm() call can skip a second auth round trip
- self._session_cache[content_id] = session_data
- return session_data
-
- def _build_widevine_drm_config(
- self,
- session_data: dict,
- content_id: str,
- referer_path: str,
- ) -> List[DRMConfig]:
- """
- Build a Widevine DRMConfig list from an already-authorized session.
-
- Args:
- session_data: Authorized session dict containing at least ``DrmId``.
- content_id: Used only for log messages.
- referer_path: Appended to base_website for the Referer header
- (e.g. ``"/"`` for live, ``"/videostore"`` for VOD).
- """
- drm_id = session_data.get("DrmId")
- if not drm_id:
- logger.error(f"No DrmId in session data for {content_id}")
- return []
-
- license_data = self.authenticator.get_license_data(drm_id)
- if not license_data:
- logger.error(f"Failed to generate license data for {content_id}")
- return []
-
- drm_config = create_drmtoday_widevine_config(
- upfront_token=license_data,
- origin=self.hrti_config.base_website,
- referer=f"{self.hrti_config.base_website}{referer_path}",
- user_agent=self.hrti_config.user_agent,
- auth_header_name="dt-custom-data",
- )
-
- if not drm_config:
- logger.error(f"create_drmtoday_widevine_config returned nothing for {content_id}")
- return []
-
- logger.debug(f"Created Widevine DRM config for {content_id}")
- return [drm_config]
-
- # ============================================================================
- # Channel fetching
- # ============================================================================
-
- def _fetch_channels_once(self) -> List[StreamingChannel]:
- """Single attempt to fetch and parse the channel list."""
- headers = self._get_hrti_authenticated_headers()
- logger.debug(
- "Fetching HRTi channels with authorization: "
- f"{'Client ...' if 'authorization' in headers else 'NO AUTHORIZATION'}"
- )
-
- channels_endpoint = self.hrti_config.api_endpoints.get("channels", "")
- if not channels_endpoint:
- logger.error("No channels endpoint configured")
- return []
-
- response = self.http_manager.post(
- channels_endpoint,
- operation="api",
- headers=headers,
- data=json.dumps({}),
- )
- response.raise_for_status()
-
- channels_data = response.json()
- if not channels_data.get("Result"):
- logger.warning("No channels found in HRTi response")
- logger.debug(f"HRTi channels response: {channels_data}")
- return []
-
- channels = []
- for raw in channels_data["Result"]:
- ch = self._parse_channel_data(raw)
- if ch:
- channels.append(ch)
-
- self.channels = channels
- return channels
-
- def get_channels(self, **kwargs) -> List[StreamingChannel]:
- """Fetch channels from HRTi API, with one auth-refresh retry on failure."""
- try:
- channels = self._fetch_channels_once()
- logger.info(f"Successfully fetched {len(channels)} channels from HRTi")
- return channels
- except requests.RequestException as e:
- logger.error(f"Error fetching HRTi channels: {e}")
- logger.info("Attempting to refresh authentication and retry...")
- self.authenticator.invalidate_token()
- try:
- channels = self._fetch_channels_once()
- logger.info(f"Successfully fetched {len(channels)} channels from HRTi on retry")
- return channels
- except Exception as retry_e:
- logger.error(f"Retry failed: {retry_e}")
- return []
- except Exception as e:
- logger.error(f"Error parsing HRTi channels: {e}")
- return []
-
- def get_events(
- self,
- start_time: Optional[datetime.datetime] = None,
- end_time: Optional[datetime.datetime] = None,
- **kwargs,
- ) -> List[Event]:
- return []
-
- def _parse_channel_data(self, channel_data: Dict) -> Optional[StreamingChannel]:
- """Parse a raw HRTi channel dict into a StreamingChannel."""
- try:
- name = channel_data.get("Name", "")
- # API returns 'ReferenceID' (capital ID), not 'ReferenceId'
- channel_id = channel_data.get("ReferenceID", "")
- streaming_url = channel_data.get("StreamingURL", "")
- is_radio = channel_data.get("Radio", False)
- icon_url = channel_data.get("Icon", "")
-
- if not name or not channel_id:
- logger.debug(f"Skipping channel — missing name or ID: {channel_data}")
- return None
-
- channel = StreamingChannel(
- name=name,
- content_id=channel_id,
- provider=self.provider_name,
- logo_url=icon_url,
- mode="live",
- session_manifest=True,
- manifest=None,
- manifest_script=streaming_url,
- content_type="AUDIO" if is_radio else "LIVE",
- country=self.country,
- is_radio=is_radio,
- language="hr",
- )
- channel.use_cdm = True
- channel.cdm_type = "widevine"
-
- logger.debug(f"Parsed HRTi channel: {name} ({channel_id}) — radio: {is_radio}")
- return channel
-
- except Exception as e:
- logger.error(f"Error parsing channel {channel_data}: {e}")
- return None
-
- # ============================================================================
- # VOD
- # ============================================================================
-
- def get_vod_category(
- self,
- content_id: str = "",
- cursor: Optional[str] = None,
- page_size: int = 24,
- **kwargs,
- ) -> Union[List, Dict]:
- """
- Get VOD category children. Delegates to HRTiVodManager.
-
- Args:
- content_id: Opaque node identifier (see HRTiVodManager docstring)
- cursor: Pagination cursor
- page_size: Items per page
-
- Returns:
- Dict with ``entries``, ``next_cursor``, ``total`` keys.
- """
- result = self._vod_manager.get_category(content_id, cursor, page_size)
-
- if isinstance(result, dict):
- return result
- if isinstance(result, list):
- return {"entries": result, "next_cursor": None, "total": None}
- return {"entries": [], "next_cursor": None, "total": None}
-
- # ============================================================================
- # Manifest
- # ============================================================================
-
- def get_manifest(self, content_id: str, **kwargs) -> Optional[str]:
- """
- Return the playback manifest URL for *content_id*.
-
- Routing:
- - Navigation nodes (catalogue:/series:/season:/special:) → error, not playable
- - details_{ref} or bare non-numeric id → VOD
- - bare numeric id → live channel
- """
- if self._is_nav_node(content_id):
- logger.error(
- f"get_manifest called with navigation node '{content_id}' — "
- f"this is a category/series/season container, not a playable item"
- )
- return None
-
- # Strip "details_" prefix before passing to VOD manager
- vod_ref = content_id[len(_DETAILS_PREFIX):] if content_id.startswith(_DETAILS_PREFIX) else content_id
-
- if self._is_playable_vod(content_id):
- return self._vod_manager.get_vod_streaming_url(vod_ref)
-
- return self._get_live_manifest(content_id, **kwargs)
-
- def _get_live_manifest(self, content_id: str, **kwargs) -> Optional[str]:
- """Authorize a session and return the streaming URL for a live channel."""
- try:
- session_data = self._authorize_live_session(content_id)
- if not session_data:
- return None
-
- channel = self._find_channel(content_id)
- if channel and channel.manifest_script:
- return channel.manifest_script
-
- logger.warning(f"No streaming URL found for channel {content_id}")
- return None
-
- except Exception as e:
- logger.error(f"Error getting manifest for channel {content_id}: {e}")
- return None
-
- # ============================================================================
- # DRM
- # ============================================================================
-
- def get_drm(self, content_id: str, **kwargs) -> List[DRMConfig]:
- """
- Return Widevine DRM config(s) for *content_id*.
-
- Routing mirrors get_manifest() — nav nodes are rejected, playable VOD
- goes to _get_vod_drm, live channels go to _get_live_drm.
- """
- if self._is_nav_node(content_id):
- logger.error(
- f"get_drm called with navigation node '{content_id}' — "
- f"this is a category/series/season container, not a playable item"
- )
- return []
-
- vod_ref = content_id[len(_DETAILS_PREFIX):] if content_id.startswith(_DETAILS_PREFIX) else content_id
-
- if self._is_playable_vod(content_id):
- return self._get_vod_drm(vod_ref, **kwargs)
- return self._get_live_drm(content_id, **kwargs)
-
- def _get_live_drm(self, content_id: str, **kwargs) -> List[DRMConfig]:
- """Get Widevine DRM config for a live channel."""
- try:
- # Prefer session data passed explicitly by the caller, then the
- # cache populated by _get_live_manifest, then authorize fresh.
- session_data = (
- kwargs.get("session_data")
- or self._session_cache.get(content_id)
- or self._authorize_live_session(content_id)
- )
-
- if not session_data:
- return []
-
- return self._build_widevine_drm_config(
- session_data=session_data,
- content_id=content_id,
- referer_path="/",
- )
-
- except Exception as e:
- logger.error(f"Error getting DRM config for channel {content_id}: {e}")
- logger.error(traceback.format_exc())
- return []
-
- def _get_vod_drm(self, content_id: str, **kwargs) -> List[DRMConfig]:
- """Get Widevine DRM config for a VOD item."""
- try:
- session_data = self._vod_manager.get_vod_session_data(content_id)
-
- if not session_data or not session_data.get("Authorized", False):
- logger.error(f"Failed to authorize session for VOD DRM: {content_id}")
- return []
-
- return self._build_widevine_drm_config(
- session_data=session_data,
- content_id=content_id,
- referer_path="/videostore",
- )
-
- except Exception as e:
- logger.error(f"Error getting DRM for VOD {content_id}: {e}")
- logger.error(traceback.format_exc())
- return []
-
- # ============================================================================
- # EPG and ancillary methods
- # ============================================================================
-
- def get_epg(self, channel_id: str, **kwargs) -> List[Dict]:
- """Get EPG data for a channel."""
- try:
- headers = self._get_hrti_authenticated_headers()
- start_time = self.authenticator.get_time_offset(-4)
- end_time = self.authenticator.get_time_offset(4)
-
- payload = {
- "ChannelReferenceIds": [channel_id],
- "StartTime": f"/Date({start_time})/",
- "EndTime": f"/Date({end_time})/",
- }
-
- programme_endpoint = self.hrti_config.api_endpoints.get("programme", "")
- if not programme_endpoint:
- logger.error("No programme endpoint configured")
- return []
-
- response = self.http_manager.post(
- programme_endpoint,
- operation="api",
- headers=headers,
- data=json.dumps(payload),
- )
- response.raise_for_status()
-
- epg_data = response.json()
- if "Result" in epg_data:
- return [
- {
- "title": entry.get("Title", ""),
- "description": entry.get("Description", ""),
- "start": entry.get("StartTime", ""),
- "end": entry.get("EndTime", ""),
- "genre": entry.get("Genre", ""),
- }
- for entry in epg_data["Result"]
- ]
- return []
-
- except Exception as e:
- logger.error(f"Error getting EPG data for channel {channel_id}: {e}")
- return []
-
- def get_license_url(self, channel: StreamingChannel, **kwargs) -> Optional[str]:
- """Return the Widevine license server URL for *channel*."""
- drm_configs = self.get_drm(channel.channel_id, **kwargs)
- if drm_configs:
- return drm_configs[0].license.server_url
- return None
-
- @property
- def catchup_window(self) -> int:
- """HRTi does not support live-channel catchup."""
- return 0
-
- def get_epg_xmltv(self, **kwargs) -> Optional[str]:
- """HRTi does not provide XMLTV natively."""
- return None
-
- def get_dynamic_manifest_params(self, channel: StreamingChannel, **kwargs) -> Optional[str]:
- """Session authorization is handled inside get_manifest(); nothing extra needed here."""
- return None
\ No newline at end of file
+ # _build_epg(): step 2 (HRTiEpgManager, native batch GetProgramme).
diff --git a/lib/streaming_providers/providers/hrti/session.py b/lib/streaming_providers/providers/hrti/session.py
new file mode 100644
index 0000000..da71d5a
--- /dev/null
+++ b/lib/streaming_providers/providers/hrti/session.py
@@ -0,0 +1,193 @@
+# streaming_providers/providers/hrti/session.py
+"""
+HRTiSession -- VARIANT B auth adapter in front of the stateful HRTiAuthenticator.
+
+Managers see ``auth`` (= this session) satisfying AuthProtocol
+(get_access_token / build_headers / invalidate) plus the auth-derived state
+they need (device_id, ip_address, user_id) and the two session calls of the
+playback flow (authorize_session, report_session_event). They never touch the
+raw authenticator. The provider keeps BOTH objects (``self.authenticator``
+for the legacy auth mixin, ``self.auth`` for the managers).
+
+Behaviour
+---------
+* LAZY: nothing here makes a network call at construction.
+* ``get_access_token`` / ``require`` raise typed errors; ``ensure`` never
+ raises (opportunistic pre-login, settings UI); ``last_error`` keeps the
+ ORIGINAL exception.
+* Retry policy: PERMANENT failures (AuthError incl. CredentialsError,
+ ConfigurationError) are not retried until ``reset()`` / ``invalidate()``;
+ TRANSIENT ones (network, 5xx, rate limit, anything untyped -> ServerError)
+ back off exponentially, base * 2**(n-1) s, capped. Injectable clock.
+* The anonymous fallback of the authenticator is deliberate and unchanged:
+ 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.
+"""
+
+import threading
+import time
+from typing import Any, Callable, Dict, List, Optional
+
+from ...base.errors import AuthError, ConfigurationError, ProviderError, ServerError
+from ...base.utils.logger import logger
+from .constants import HRTiConfig, HRTiDefaults
+
+_PERMANENT = (AuthError, ConfigurationError)
+
+
+class HRTiSession:
+ def __init__(
+ self,
+ *,
+ authenticator: Any,
+ config: HRTiConfig,
+ clock: Callable[[], float] = time.monotonic,
+ backoff_base: float = HRTiDefaults.SESSION_BACKOFF_BASE,
+ backoff_cap: float = HRTiDefaults.SESSION_BACKOFF_CAP,
+ ):
+ self._authenticator = authenticator
+ self._config = config
+ self._clock = clock
+ self._backoff_base = backoff_base
+ self._backoff_cap = backoff_cap
+ self._lock = threading.RLock()
+ self._callbacks: List[Callable[[], None]] = []
+ self._last_error: Optional[BaseException] = None
+ self._typed_error: Optional[ProviderError] = None
+ self._permanent = False
+ self._failures = 0
+ self._retry_at = 0.0
+
+ # ------------------------------------------------------------------
+ # AuthProtocol
+ # ------------------------------------------------------------------
+ def get_access_token(self, force_refresh: bool = False) -> str:
+ with self._lock:
+ self._raise_if_blocked()
+ try:
+ token = self._authenticator.get_bearer_token(force_refresh=force_refresh)
+ except Exception as exc:
+ typed = self._record_failure(exc)
+ if typed is exc:
+ raise
+ raise typed from exc
+ if not token:
+ error = AuthError("HRTi: login returned an empty access token")
+ raise self._record_failure(error) from None
+ self._clear_failure()
+ return token
+
+ def build_headers(self) -> Dict[str, str]:
+ """API headers incl. the token (not for CDN requests)."""
+ return self._config.api_headers(
+ token=self.get_access_token(),
+ device_id=self.device_id,
+ ip_address=self.ip_address,
+ )
+
+ def invalidate(self) -> None:
+ """Drop the session (memory + persisted) and everything derived from it."""
+ with self._lock:
+ self._authenticator.invalidate_token()
+ self._clear_failure()
+ for callback in list(self._callbacks):
+ try:
+ callback()
+ except Exception as exc: # a cache callback must not break the recovery
+ logger.warning(f"HRTi: on_invalidate callback failed: {exc!r}")
+
+ # ------------------------------------------------------------------
+ # Session API for the UI and the 401 recovery
+ # ------------------------------------------------------------------
+ def refresh(self) -> str:
+ """401 recovery: invalidate, then one forced fresh login."""
+ self.invalidate()
+ return self.get_access_token(force_refresh=True)
+
+ def require(self) -> str:
+ return self.get_access_token()
+
+ def ensure(self) -> bool:
+ try:
+ self.get_access_token()
+ return True
+ except Exception:
+ return False
+
+ def reset(self) -> None:
+ """Forget failures (credentials changed): the next call logs in again."""
+ with self._lock:
+ self._clear_failure()
+
+ def on_invalidate(self, callback: Callable[[], None]) -> None:
+ self._callbacks.append(callback)
+
+ @property
+ def last_error(self) -> Optional[BaseException]:
+ return self._last_error
+
+ # ------------------------------------------------------------------
+ # Auth-derived state read by the managers
+ # ------------------------------------------------------------------
+ @property
+ def device_id(self) -> str:
+ return self._authenticator.get_device_id()
+
+ @property
+ def ip_address(self) -> str:
+ return self._authenticator.get_ip_address()
+
+ @property
+ def user_id(self) -> str:
+ return self._authenticator.get_user_id() or ""
+
+ def has_user_credentials(self) -> bool:
+ return self._authenticator.has_user_credentials()
+
+ # ------------------------------------------------------------------
+ # Playback-session calls (authenticator owns the protocol details)
+ # ------------------------------------------------------------------
+ def authorize_session(self, **kw: Any) -> Dict[str, Any]:
+ 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)
+
+ def report_session_event(self, session_id: str, channel_id: Optional[str] = None) -> bool:
+ return self._authenticator.report_session_event(session_id, channel_id)
+
+ # ------------------------------------------------------------------
+ # Failure bookkeeping
+ # ------------------------------------------------------------------
+ def _raise_if_blocked(self) -> None:
+ if self._typed_error is None:
+ return
+ 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}")
+ self._last_error = exc
+ self._typed_error = typed
+ if isinstance(typed, _PERMANENT):
+ self._permanent = True
+ else:
+ self._failures += 1
+ 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"{'permanent' if self._permanent else f'retry in {self._retry_at - self._clock():.0f}s'}): {typed}"
+ )
+ return typed
+
+ def _clear_failure(self) -> None:
+ self._last_error = None
+ self._typed_error = None
+ self._permanent = False
+ self._failures = 0
+ self._retry_at = 0.0
diff --git a/lib/streaming_providers/providers/hrti/vod_manager.py b/lib/streaming_providers/providers/hrti/vod_manager.py
index e136ce8..38f7f40 100644
--- a/lib/streaming_providers/providers/hrti/vod_manager.py
+++ b/lib/streaming_providers/providers/hrti/vod_manager.py
@@ -1,327 +1,181 @@
# streaming_providers/providers/hrti/vod_manager.py
"""
-HRTi VOD Manager - Handles all VOD catalogue operations.
+HRTi VOD manager -- catalogue navigation and VOD playback.
-Endpoints actually available (confirmed from API):
- GetCatalogueStructure → full nested tree (Children arrays), root nav
- GetCatalogue → paginated items for a catalogue node
- GetSeries → episodes for a series season
- GetVodDetails → full metadata + FileName for a single item
- GetEditorsChoice → editor's picks
- GetWatchLater → user watch later (requires user auth)
+Endpoints (confirmed from the API): GetCatalogueStructure (root tree),
+GetCatalogue (paged items), GetSeries (episodes of a season), GetVodDetails
+(metadata + FileName), GetEditorsChoice, GetWatchLater (user auth).
+There is NO GetSeasons / GetEpisodes.
-NOT available: GetSeasons, GetEpisodes — these do not exist on this API.
+content_id grammar (see ids.py):
+ "" root -> GetCatalogueStructure
+ "catalogue_][" folder items -> GetCatalogue (paged)
+ "series_--" episodes -> GetSeries
+ "special_watch_later" / "special_editors_choice"
+ "details_][" | "][" one playable item (VodManager.handles_content_id)
-Navigation model
-----------------
- GetCatalogueStructure returns the full tree with Children nesting.
- Every navigable node at every level uses ``catalogue:{ReferenceId}``.
- There is no separate seasons API — seasons appear as Children of a
- series node inside GetCatalogueStructure and are therefore already
- handled by the catalogue_ routing.
+Return conventions
+------------------
+Navigation always returns a VodPage. next_cursor (the next page number) is set
+ONLY for GetCatalogue, which honours PageNumber/ItemsPerPage; every other
+endpoint returns the full list (next_cursor None). Failures raise typed
+errors; GetWatchLater / GetEditorsChoice are optional lists: an API error
+there yields an empty page.
- Type-3 series items in GetCatalogue listings link to GetSeries via:
- content_id = "series_{SeriesReferenceId}--{SeasonReferenceId}"
- The default SeasonReferenceId is "{SeriesReferenceId}_S01".
-
-Content ID scheme
------------------
- "" root (GetCatalogueStructure)
- "catalogue_{ref_id}" items (GetCatalogue, paginated)
- "series_{series_ref}--{season_ref}" episodes (GetSeries)
- "special_watch_later" GetWatchLater
- "special_editors_choice" GetEditorsChoice
-
-Field mapping (verified from real API responses)
--------------------------------------------------
- GetCatalogueStructure nodes:
- Name, ReferenceId, Children (list), PosterLandscape, ParentReferenceId
-
- GetCatalogue items:
- Type 2 → playable VOD (VodData.Duration ms, VodData.ProductionYear,
- VodData.ContentRating)
- Type 3 → series (SeriesData.SeriesReferenceId, SeriesData.SeriesName,
- SeriesData.LastSeasonNumber)
- Type 5 → episode (EpisodeData.EpisodeNr, EpisodeData.SeasonNr,
- EpisodeData.Duration ms, EpisodeData.ContentRating)
- other → navigable folder
-
- GetVodDetails:
- ReferenceID (capital), Title, FileName (streaming URL — NOT StreamingURL),
- DurationInFrames (frames / 25 ≈ seconds), ProductionYear, ContentRating,
- AssetCategory, Directors, Actors (plain strings, not lists),
- SeriesReferenceId, SeriesName, SeasonNr, EpisodeNr,
- SVODVideostores (list → "svod" auth when non-empty),
- PosterLandscape, PosterPortrait
-
- AuthorizeSession parameters derived from FileName:
- FileName = https://cdn1oiv.akamaized.net/hrtvodorigin/REF.smil/manifest.mpd
- ContentReferenceId = path[1] without ".smil" e.g. "1870699590_delin"
- ContentDrmId = "{path[0]}_{path[1]}" e.g. "hrtvodorigin_1870699590_delin.smil"
- ContentType = "svod" if SVODVideostores else "vod"
- VideostoreReferenceIds = SVODVideostores or null
-
- Token: _current_token.access_token (no get_access_token() method exists).
- Pagination: NumberOfItems field (not TotalCount); no HasMore field.
+Manifest and DRM share one GetVodDetails through the provider-owned playout
+cache (5 s TTL); the AuthorizeSession of the DRM call is cached the same way.
"""
-import json
-from typing import Dict, List, Optional, Union
-from urllib.parse import urlparse
+import time
+from typing import Any, Dict, List, Optional, Tuple, Union
+from ...base.errors import BadRequestError, EntitlementError, NotFoundError
+from ...base.managers import VodManager
+from ...base.models import DRMConfig
from ...base.models.vod import VodCategory, VodItem
from ...base.utils.logger import logger
+from ...base.vod import VodPage
+from . import ids
+from .api import api_post
from .constants import HRTiDefaults
+from .drm import build_widevine, derive_vod_session_params
+from .models import VodDetails
-# GetCatalogue / GetSeries Type discriminator values
+# GetCatalogue / GetSeries "Type" discriminator values
_TYPE_VOD = 2
_TYPE_SERIES = 3
_TYPE_EPISODE = 5
-# DurationInFrames in GetVodDetails is at ~25 fps
+# DurationInFrames in GetVodDetails is ~25 fps
_ASSUMED_FPS = 25
+_VOD_REFERER = "/videostore"
+_PROVIDER = HRTiDefaults.PROVIDER_NAME
+_PARSE_ERRORS = (KeyError, TypeError, AttributeError, ValueError)
-class HRTiVodManager:
- """Manages HRTi VOD catalogue operations."""
- def __init__(self, provider):
- self.provider = provider
- self._config = provider.hrti_config
- self._http_manager = provider.http_manager
- self._authenticator = provider.authenticator
+class HRTiVodManager(VodManager):
+ def __init__(
+ self,
+ *,
+ http_manager: Any,
+ auth: Any,
+ country: str,
+ config: Any,
+ playout_cache: Optional[Dict[str, Tuple[Any, float]]] = None,
+ ):
+ super().__init__(
+ http_manager=http_manager, auth=auth, country=country, config=config
+ )
+ # Provider-owned dict, borrowed by reference. Keys: "vod:][" ->
+ # (VodDetails, ts) and "vodsess:][" -> (session dict, ts).
+ self._playout_cache = playout_cache if playout_cache is not None else {}
- # -------------------------------------------------------------------------
- # Public API (called by provider.get_vod_category)
- # -------------------------------------------------------------------------
+ # ------------------------------------------------------------------
+ # Routing
+ # ------------------------------------------------------------------
+ def handles_content_id(self, content_id: str) -> bool:
+ return ids.is_vod(content_id)
- def get_category(
+ # ------------------------------------------------------------------
+ # Navigation
+ # ------------------------------------------------------------------
+ def get_vod_category(
self,
content_id: str = "",
cursor: Optional[str] = None,
- page_size: int = 24,
- ) -> Union[List, Dict]:
- """
- Return children for a VOD navigation node.
-
- Routing by content_id prefix:
- "" → GetCatalogueStructure (root, top-level categories)
- "catalogue_{ref_id}" → GetCatalogue items (paginated)
- "series_{sid}--{ssid}" → GetSeries episodes
- "special_*" → special collections
- """
+ page_size: int = HRTiDefaults.VOD_ITEMS_PER_PAGE,
+ **kw: Any,
+ ) -> VodPage:
if not content_id:
return self._fetch_catalogue_structure()
-
if content_id.startswith("catalogue_"):
- ref_id = content_id[len("catalogue_"):]
- return self._fetch_catalogue_items(ref_id, cursor, page_size)
-
+ return self._fetch_catalogue_items(content_id[len("catalogue_"):], cursor, page_size)
if content_id.startswith("series_"):
- # Format: series:{series_ref_id}:{season_ref_id}
parts = content_id[len("series_"):].split("--", 1)
- if len(parts) == 2:
- return self._fetch_series_episodes(parts[0], parts[1])
- logger.warning(f"Malformed series content_id (expected series_SID--SSID): {content_id}")
- return []
-
- if content_id == "special_watch_later":
- return self._fetch_watch_later()
-
- if content_id == "special_editors_choice":
- return self._fetch_editors_choice()
-
- logger.warning(f"Unknown VOD content_id format: {content_id}")
- return []
-
- # -------------------------------------------------------------------------
- # HTTP helpers
- # -------------------------------------------------------------------------
-
- def _get_headers(self, referer_path: str = "/videostore") -> Dict[str, str]:
- """Build authenticated headers for VOD 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 ""
- )
- if token:
- headers["authorization"] = f"Client {token}"
- return headers
-
- def _post(
- self,
- endpoint_key: str,
- payload: Dict,
- referer_path: str = "/videostore",
- ) -> Optional[Union[Dict, List]]:
- """
- POST to a named endpoint and return the Result field.
-
- Returns None on HTTP error, API error (ErrorCode != 0), or missing Result.
- Result type varies by endpoint:
- GetCatalogueStructure → list
- GetCatalogue → dict with Items, NumberOfItems, Name
- GetSeries → dict with Items, LastWatchedEpisodeReferenceId
- GetVodDetails → dict (single item)
- """
- url = self._config.api_endpoints.get(endpoint_key)
- if not url:
- logger.error(f"Unknown endpoint key: {endpoint_key}")
- return None
- try:
- resp = self._http_manager.post(
- url,
- operation="api",
- headers=self._get_headers(referer_path),
- data=json.dumps(payload),
- )
- resp.raise_for_status()
- data = resp.json()
- if data.get("ErrorCode", 0) != 0:
- logger.warning(
- f"HRTi API error for {endpoint_key}: "
- f"[{data.get('ErrorCode')}] {data.get('ErrorDescription')}"
+ if len(parts) != 2 or not all(parts):
+ raise BadRequestError(
+ f"HRTi: malformed series id (expected series_SID--SSID): {content_id}"
)
- return None
- return data.get("Result")
- except Exception as e:
- logger.error(f"Error calling {endpoint_key}: {e}")
- return None
+ return self._fetch_series_episodes(parts[0], parts[1])
+ if content_id == "special_watch_later":
+ return self._fetch_special("watch_later", "/watch_later")
+ if content_id == "special_editors_choice":
+ return self._fetch_special("editors_choice", "/editors_choice")
+ raise BadRequestError(f"HRTi: unknown VOD content_id format: {content_id}")
- # -------------------------------------------------------------------------
- # GetCatalogueStructure — root navigation tree
- # -------------------------------------------------------------------------
+ def _post(self, endpoint_key: str, payload: Dict, referer_path: str = _VOD_REFERER,
+ required: bool = True) -> Optional[Union[Dict, List]]:
+ return api_post(
+ http_manager=self.http_manager,
+ session=self.auth,
+ config=self.config,
+ endpoint_key=endpoint_key,
+ payload=payload,
+ referer_path=referer_path,
+ required=required,
+ )
- def _fetch_catalogue_structure(self) -> List[VodCategory]:
- """
- Fetch top-level navigation categories from GetCatalogueStructure.
-
- The API returns a flat list of root nodes, each with a Children array
- containing the full subtree. We parse only the top level here —
- children are lazy-loaded when the user navigates via their
- catalogue:{ReferenceId} content_id.
-
- child_count is set from len(Children) since the API provides no
- separate count field. A non-zero child_count signals the UI that
- this node is drillable.
- """
- result = self._post("catalogue_structure", {}, "/videostore")
-
- if not result or not isinstance(result, list):
- logger.warning("No catalogue structure returned from HRTi")
- return []
-
- categories = []
- for node in result:
- cat = self._parse_structure_node(node)
- if cat:
- categories.append(cat)
-
- logger.info(f"Retrieved {len(categories)} root VOD categories from HRTi")
- return categories
+ # --- GetCatalogueStructure: root tree -------------------------------
+ def _fetch_catalogue_structure(self) -> VodPage:
+ result = self._post("catalogue_structure", {})
+ if not isinstance(result, list):
+ logger.warning("HRTi: no catalogue structure returned")
+ return VodPage()
+ categories = [c for c in (self._parse_structure_node(n) for n in result) if c]
+ logger.info(f"HRTi: {len(categories)} root VOD categories")
+ return VodPage(entries=categories)
def _parse_structure_node(self, node: Dict) -> Optional[VodCategory]:
- """
- Parse one node from GetCatalogueStructure.
-
- Fields: Name, ReferenceId, Children (list), PosterLandscape,
- ParentReferenceId (ignored — content_id encodes the path).
- """
try:
ref_id = node.get("ReferenceId")
name = (node.get("Name") or "").strip()
if not ref_id or not name:
- logger.debug(f"Skipping structure node — missing id or name")
return None
-
children = node.get("Children") or []
return VodCategory(
name=name,
content_id=f"catalogue_{ref_id}",
- provider=self.provider.provider_name,
+ provider=_PROVIDER,
logo_url=node.get("PosterLandscape"),
description=None,
- # child_count from Children length; None if empty (may still have
- # items via GetCatalogue even with no sub-categories)
child_count=len(children) if children else None,
)
- except Exception as e:
- logger.error(f"Error parsing structure node: {e}")
+ except _PARSE_ERRORS as exc:
+ logger.warning(f"HRTi: skipping malformed structure node: {exc!r}")
return None
- # -------------------------------------------------------------------------
- # GetCatalogue — paginated items for a catalogue node
- # -------------------------------------------------------------------------
-
- def _fetch_catalogue_items(
- self,
- ref_id: str,
- cursor: Optional[str] = None,
- page_size: int = 24,
- ) -> Dict:
- """
- Fetch paginated items from GetCatalogue for a given ReferenceId.
-
- Response: { "Name": "...", "NumberOfItems": N, "Items": [...] }
- Pagination is derived from NumberOfItems — no HasMore field exists.
- """
+ # --- GetCatalogue: paged items ---------------------------------------
+ def _fetch_catalogue_items(self, ref_id: str, cursor: Optional[str], page_size: int) -> VodPage:
page = 1
if cursor:
try:
page = max(1, int(cursor))
except ValueError:
- logger.warning(f"Invalid cursor '{cursor}', using page 1")
-
+ logger.warning(f"HRTi: invalid cursor '{cursor}', using page 1")
if page > HRTiDefaults.MAX_VOD_PAGES:
- return {"entries": [], "next_cursor": None, "total": None}
+ return VodPage()
result = self._post(
- "catalogue",
- {"ReferenceId": ref_id, "ItemsPerPage": page_size, "PageNumber": page},
+ "catalogue", {"ReferenceId": ref_id, "ItemsPerPage": page_size, "PageNumber": page}
)
+ if not isinstance(result, dict):
+ return VodPage()
- if not result or not isinstance(result, dict):
- return {"entries": [], "next_cursor": None, "total": None}
-
- raw_items = result.get("Items") or []
- total_count = result.get("NumberOfItems")
-
- entries = []
- for item in raw_items:
- parsed = self._parse_catalogue_item(item)
- if parsed:
- entries.append(parsed)
-
+ entries = [e for e in (self._parse_catalogue_item(i) for i in result.get("Items") or []) if e]
+ total = result.get("NumberOfItems")
has_more = (
- total_count is not None
- and page * page_size < total_count
+ total is not None
+ and page * page_size < total
and page < HRTiDefaults.MAX_VOD_PAGES
)
-
- logger.debug(
- f"catalogue {ref_id} page {page}: {len(entries)} entries, "
- f"total {total_count}, next_cursor {page + 1 if has_more else None}"
+ return VodPage(
+ entries=entries,
+ next_cursor=str(page + 1) if has_more else None,
+ total=total,
)
- return {
- "entries": entries,
- "next_cursor": str(page + 1) if has_more else None,
- "total": total_count,
- }
def _parse_catalogue_item(self, item: Dict) -> Optional[Union[VodCategory, VodItem]]:
- """Dispatch catalogue item parsing by Type field."""
try:
item_type = item.get("Type")
if item_type == _TYPE_VOD:
@@ -331,12 +185,12 @@ class HRTiVodManager:
if item_type == _TYPE_EPISODE:
return self._parse_episode_catalogue_item(item)
return self._parse_folder_catalogue_item(item)
- except Exception as e:
- logger.error(f"Error parsing catalogue item: {e}")
+ except _PARSE_ERRORS as exc:
+ logger.warning(f"HRTi: skipping malformed catalogue item: {exc!r}")
return None
def _parse_folder_catalogue_item(self, item: Dict) -> Optional[VodCategory]:
- """No Type — navigable sub-folder node."""
+ """No Type: a navigable sub-folder."""
ref_id = item.get("ReferenceId")
name = (item.get("Title") or item.get("Name") or "").strip()
if not ref_id or not name:
@@ -344,67 +198,48 @@ class HRTiVodManager:
return VodCategory(
name=name,
content_id=f"catalogue_{ref_id}",
- provider=self.provider.provider_name,
+ provider=_PROVIDER,
logo_url=item.get("PosterLandscape") or item.get("PosterPortrait"),
description=None,
child_count=item.get("ChildCount"),
)
def _parse_series_catalogue_item(self, item: Dict) -> Optional[VodCategory]:
- """
- Type-3 series item.
-
- Navigation goes to GetSeries which requires SeriesReferenceId and
- SeasonReferenceId. We default to the _S01 suffix since there is no
- GetSeasons endpoint. Multi-season series have their seasons listed as
- Children in GetCatalogueStructure, so they are browsed via catalogue:
- nodes and never reach this path.
- """
+ """Type 3: navigation goes to GetSeries (needs series + season ids);
+ there is no GetSeasons, so the default season is ``_S01``."""
series_data = item.get("SeriesData") or {}
series_ref_id = series_data.get("SeriesReferenceId") or item.get("ReferenceId")
name = (item.get("Title") or "").strip()
if not series_ref_id or not name:
return None
-
- default_season_ref = f"{series_ref_id}_S01"
- season_count = series_data.get("LastSeasonNumber")
-
return VodCategory(
name=name,
- content_id=f"series_{series_ref_id}--{default_season_ref}",
- provider=self.provider.provider_name,
+ content_id=f"series_{series_ref_id}--{series_ref_id}_S01",
+ provider=_PROVIDER,
logo_url=item.get("PosterLandscape") or item.get("PosterPortrait"),
description=None,
- child_count=season_count,
+ child_count=series_data.get("LastSeasonNumber"),
)
def _parse_vod_catalogue_item(self, item: Dict) -> Optional[VodItem]:
- """
- Type-2 playable item from a GetCatalogue listing.
-
- manifest_script is None — FileName only available via GetVodDetails.
- content_id is the raw ReferenceId (no prefix) so provider routing works.
- """
+ """Type 2: playable item. manifest_script stays None (FileName only
+ comes from GetVodDetails); content_id is the raw ReferenceId."""
ref_id = item.get("ReferenceId")
title = (item.get("Title") or "").strip()
if not ref_id or not title:
return None
-
vod_data = item.get("VodData") or {}
duration_ms = vod_data.get("Duration")
- duration_sec = int(duration_ms / 1000) if duration_ms else None
- category_str = item.get("VodCategoryNames") or ""
- genres = [g.strip() for g in category_str.split(",") if g.strip()] or None
-
+ genres = [g.strip() for g in (item.get("VodCategoryNames") or "").split(",") if g.strip()] or None
return VodItem(
name=title,
content_id=ref_id,
- provider=self.provider.provider_name,
+ provider=_PROVIDER,
logo_url=item.get("PosterLandscape") or item.get("PosterPortrait"),
description=None,
long_description=None,
original_title=None,
- duration_seconds=duration_sec,
+ duration_seconds=int(duration_ms / 1000) if duration_ms else None,
release_year=vod_data.get("ProductionYear"),
rating=vod_data.get("ContentRating"),
genre=genres[0] if genres else None,
@@ -422,34 +257,24 @@ class HRTiVodManager:
)
def _parse_episode_catalogue_item(self, item: Dict) -> Optional[VodItem]:
- """
- Type-5 episode item from GetCatalogue or GetSeries.
-
- EpisodeData sub-object: EpisodeNr, SeasonNr, Duration (ms), ContentRating.
- SeriesData sub-object: SeriesReferenceId, SeriesName.
- content_id is the raw ReferenceId.
- """
+ """Type 5: episode (EpisodeData / SeriesData sub-objects)."""
ref_id = item.get("ReferenceId")
title = (item.get("Title") or "").strip()
if not ref_id or not title:
return None
-
ep_data = item.get("EpisodeData") or {}
series_data = item.get("SeriesData") or {}
duration_ms = ep_data.get("Duration")
- duration_sec = int(duration_ms / 1000) if duration_ms else None
- category_str = item.get("VodCategoryNames") or ""
- genres = [g.strip() for g in category_str.split(",") if g.strip()] or None
-
+ genres = [g.strip() for g in (item.get("VodCategoryNames") or "").split(",") if g.strip()] or None
return VodItem(
name=title,
content_id=ref_id,
- provider=self.provider.provider_name,
+ provider=_PROVIDER,
logo_url=item.get("PosterLandscape") or item.get("PosterPortrait"),
description=None,
long_description=None,
original_title=None,
- duration_seconds=duration_sec,
+ duration_seconds=int(duration_ms / 1000) if duration_ms else None,
release_year=None,
rating=ep_data.get("ContentRating"),
genre=genres[0] if genres else None,
@@ -466,109 +291,57 @@ class HRTiVodManager:
manifest_script=None,
)
- # -------------------------------------------------------------------------
- # GetSeries — episodes for a specific series season
- # -------------------------------------------------------------------------
-
- def _fetch_series_episodes(
- self,
- series_ref_id: str,
- season_ref_id: str,
- ) -> Dict:
- """
- Fetch all episodes from GetSeries for a given series + season.
-
- Payload: SeriesReferenceId, SeasonReferenceId, PageSize, PageNumber.
- Response: { "Items": [...], "LastWatchedEpisodeReferenceId": ..., ... }
- Each item uses Type 5 / EpisodeData schema.
- """
+ # --- GetSeries: episodes of one season --------------------------------
+ def _fetch_series_episodes(self, series_ref_id: str, season_ref_id: str) -> VodPage:
result = self._post(
"episodes",
{
"SeriesReferenceId": series_ref_id,
"SeasonReferenceId": season_ref_id,
- "PageSize": 100,
+ "PageSize": HRTiDefaults.SERIES_PAGE_SIZE,
"PageNumber": 1,
},
)
+ if not isinstance(result, dict):
+ logger.warning(f"HRTi: no episodes for series {series_ref_id} season {season_ref_id}")
+ return VodPage()
+ episodes = [e for e in (self._parse_episode_catalogue_item(i) for i in result.get("Items") or []) if e]
+ return VodPage(entries=episodes, total=len(episodes))
- if not result or not isinstance(result, dict):
- logger.warning(
- f"No episodes returned for series {series_ref_id} "
- f"season {season_ref_id}"
- )
- return {"entries": [], "next_cursor": None, "total": None}
-
- raw_items = result.get("Items") or []
- episodes = []
- for ep in raw_items:
- parsed = self._parse_episode_catalogue_item(ep)
- if parsed:
- episodes.append(parsed)
-
- logger.info(
- f"Retrieved {len(episodes)} episodes for series {series_ref_id} "
- f"season {season_ref_id}"
- )
- return {"entries": episodes, "next_cursor": None, "total": len(episodes)}
-
- # -------------------------------------------------------------------------
- # GetVodDetails — full metadata for one item
- # -------------------------------------------------------------------------
-
- def _fetch_vod_details(self, ref_id: str) -> Optional[VodItem]:
- """Fetch full metadata for a single VOD item from GetVodDetails."""
- result = self._post("vod_details", {"ReferenceId": ref_id})
- if not result:
- logger.warning(f"No details returned for VOD item {ref_id}")
- return None
- return self._parse_vod_details_item(result)
+ # --- Special lists (optional endpoints) --------------------------------
+ def _fetch_special(self, endpoint_key: str, referer_path: str) -> VodPage:
+ result = self._post(endpoint_key, {}, referer_path, required=False)
+ if not isinstance(result, list):
+ return VodPage()
+ items = [i for i in (self._parse_vod_details_item(r) for r in result) if i]
+ return VodPage(entries=items)
def _parse_vod_details_item(self, item: Dict) -> Optional[VodItem]:
- """
- Parse a rich VOD item from GetVodDetails.
-
- Key differences from catalogue listing schema:
- - ReferenceID (capital ID, not ReferenceId)
- - FileName (streaming URL, not StreamingURL)
- - DurationInFrames (divide by 25 for seconds)
- - Directors / Actors are plain strings, not lists
- - SVODVideostores list controls auth type
- """
+ """Rich item as returned by GetVodDetails / the special lists: ReferenceID
+ (capital ID), FileName (the stream URL), DurationInFrames (/25 = s),
+ Directors / Actors are plain strings."""
try:
ref_id = item.get("ReferenceID") or item.get("ReferenceId")
title = (item.get("Title") or "").strip()
if not ref_id or not title:
return None
-
frames = item.get("DurationInFrames")
- duration_sec = int(frames / _ASSUMED_FPS) if frames else None
-
- actors_str = item.get("Actors")
- cast = (
- [a.strip() for a in actors_str.split(",") if a.strip()]
- if actors_str else None
- )
-
- item_type_str = (item.get("Type") or "").lower()
- content_type = "SERIES" if item_type_str == "episode" else "MOVIE"
+ actors = item.get("Actors")
genre = item.get("AssetCategory") or None
- svod_videostores = item.get("SVODVideostores") or []
-
- vod_item = VodItem(
+ return VodItem(
name=title,
content_id=ref_id,
- provider=self.provider.provider_name,
+ provider=_PROVIDER,
logo_url=item.get("PosterLandscape") or item.get("PosterPortrait"),
description=item.get("Description"),
long_description=None,
original_title=item.get("OriginalTitle"),
- duration_seconds=duration_sec,
+ duration_seconds=int(frames / _ASSUMED_FPS) if frames else None,
release_year=item.get("ProductionYear"),
rating=item.get("ContentRating"),
genre=genre,
genres=[genre] if genre else None,
- cast=cast,
+ cast=[a.strip() for a in actors.split(",") if a.strip()] if actors else None,
director=item.get("Directors") or None,
season_number=item.get("SeasonNr"),
episode_number=item.get("EpisodeNr"),
@@ -576,110 +349,72 @@ class HRTiVodManager:
series_title=item.get("SeriesName"),
trailer_url=item.get("TrailerUrl"),
is_highlight=False,
- content_type=content_type,
+ content_type="SERIES" if (item.get("Type") or "").lower() == "episode" else "MOVIE",
manifest_script=item.get("FileName") or None,
)
- vod_item._svod_videostores = svod_videostores
- logger.debug(f"Parsed VOD details: {title} ({ref_id}), svod={bool(svod_videostores)}")
- return vod_item
-
- except Exception as e:
- logger.error(f"Error parsing VOD details item: {e}")
+ except _PARSE_ERRORS as exc:
+ logger.warning(f"HRTi: skipping malformed VOD details item: {exc!r}")
return None
- # -------------------------------------------------------------------------
- # Special collections
- # -------------------------------------------------------------------------
-
- def _fetch_watch_later(self) -> List[VodItem]:
- """User's Watch Later list (requires user auth)."""
- result = self._post("watch_later", {}, "/watch_later")
- if not result or not isinstance(result, list):
- return []
- items = [self._parse_vod_details_item(i) for i in result]
- return [i for i in items if i]
-
- def _fetch_editors_choice(self) -> List[VodItem]:
- """Editor's Choice list."""
- result = self._post("editors_choice", {}, "/editors_choice")
- if not result or not isinstance(result, list):
- return []
- items = [self._parse_vod_details_item(i) for i in result]
- return [i for i in items if i]
-
- # -------------------------------------------------------------------------
- # Provider-facing helpers (called by provider.py)
- # -------------------------------------------------------------------------
-
- def get_vod_streaming_url(self, content_id: str) -> Optional[str]:
- """
- Return the DASH manifest URL for a playable VOD item.
-
- Always calls GetVodDetails — FileName is only available there,
- never in catalogue listings.
- """
- details = self._fetch_vod_details(content_id)
- if details and details.manifest_script:
- return details.manifest_script
- logger.warning(f"No FileName found for VOD item {content_id}")
+ # ------------------------------------------------------------------
+ # Playback: details + authorised session, cached per request
+ # ------------------------------------------------------------------
+ def _cached(self, key: str) -> Optional[Any]:
+ entry = self._playout_cache.get(key)
+ if entry and (time.time() - entry[1]) < HRTiDefaults.PLAYOUT_CACHE_TTL:
+ return entry[0]
return None
- def get_vod_session_data(self, content_id: str) -> Optional[Dict]:
- """
- Authorize a VOD playback session and return the session dict.
-
- Derives all AuthorizeSession parameters from the FileName in GetVodDetails:
-
- FileName = https://cdn1oiv.akamaized.net/hrtvodorigin/REF.smil/manifest.mpd
- path[0] = "hrtvodorigin" (origin)
- path[1] = "REF.smil"
-
- ContentReferenceId = "REF" (stem without .smil)
- ContentDrmId = "hrtvodorigin_REF.smil"
- ContentType = "svod" if SVODVideostores non-empty else "vod"
- VideostoreReferenceIds = SVODVideostores or null
- """
+ def _details(self, ref_id: str) -> VodDetails:
+ cached = self._cached(f"vod:{ref_id}")
+ if cached is not None:
+ return cached
+ result = self._post("vod_details", {"ReferenceId": ref_id})
+ if not isinstance(result, dict) or not result:
+ raise NotFoundError(f"HRTi: VOD item {ref_id} not found")
try:
- details = self._fetch_vod_details(content_id)
- if not details or not details.manifest_script:
- logger.error(f"Cannot authorize VOD session — no FileName for {content_id}")
- return None
+ details = VodDetails.from_api_response(result)
+ except (KeyError, TypeError, AttributeError, ValueError) as exc:
+ raise NotFoundError(f"HRTi: VOD item {ref_id} has no usable details: {exc}") from exc
+ self._playout_cache[f"vod:{ref_id}"] = (details, time.time())
+ return details
- filename = details.manifest_script
- svod_stores = getattr(details, "_svod_videostores", []) or []
+ def _session(self, ref_id: str) -> Dict[str, Any]:
+ cached = self._cached(f"vodsess:{ref_id}")
+ if cached is not None:
+ return cached
+ details = self._details(ref_id)
+ if not details.file_name:
+ raise EntitlementError(f"HRTi: no FileName (not entitled?) for VOD item {ref_id}")
+ params = derive_vod_session_params(details.file_name)
+ stores = details.svod_stores
+ session_data = self.auth.authorize_session(
+ content_type="svod" if stores else "vod",
+ content_ref_id=params.content_ref_id,
+ content_drm_id=params.content_drm_id,
+ video_store_ids=stores or None,
+ channel_id=None,
+ start_time=None,
+ end_time=None,
+ )
+ if not session_data.get("Authorized", False):
+ raise EntitlementError(f"HRTi: VOD session for {ref_id} was not authorised")
+ self._playout_cache[f"vodsess:{ref_id}"] = (session_data, time.time())
+ return session_data
- path_parts = urlparse(filename).path.strip("/").split("/")
- if len(path_parts) < 2:
- logger.error(f"Cannot parse FileName path for {content_id}: {filename}")
- return None
+ def get_vod_manifest(self, content_id: str, **kw: Any) -> Optional[str]:
+ details = self._details(ids.vod_ref(content_id))
+ if not details.file_name:
+ logger.warning(f"HRTi: no FileName for VOD item {content_id}")
+ return None
+ return details.file_name
- origin = path_parts[0]
- ref_with_ext = path_parts[1]
- ref_id = (
- ref_with_ext.removesuffix(".smil")
- if ref_with_ext.endswith(".smil")
- else ref_with_ext
- )
- content_drm_id = f"{origin}_{ref_with_ext}"
- content_type = "svod" if svod_stores else "vod"
- video_store_ids = svod_stores if svod_stores else None
-
- logger.debug(
- f"Authorizing VOD session — content_id: {content_id}, "
- f"ref: {ref_id}, drm: {content_drm_id}, "
- f"type: {content_type}, stores: {video_store_ids}"
- )
-
- return self._authenticator.authorize_session(
- content_type=content_type,
- content_ref_id=ref_id,
- content_drm_id=content_drm_id,
- video_store_ids=video_store_ids,
- channel_id=None,
- start_time=None,
- end_time=None,
- )
-
- except Exception as e:
- logger.error(f"Error getting VOD session data for {content_id}: {e}")
- return None
\ No newline at end of file
+ def get_vod_drm(self, content_id: str, **kw: Any) -> List[DRMConfig]:
+ ref_id = ids.vod_ref(content_id)
+ return build_widevine(
+ self.config,
+ self.auth.user_id,
+ self._session(ref_id),
+ ref_id,
+ referer_path=_VOD_REFERER,
+ )
]