nirvana-7777/script.service.ultimate
1
0
Fork 0
mirror of https://github.com/nirvana-7777/script.service.ultimate.git synced 2026-09-25 02:22:44 +02:00
Code Issues Packages Projects Releases Wiki Activity

Fix HRTi VOD

Browse Source
This commit is contained in:
Nirvana
2026-05-05 12:44:26 +02:00
parent cb158c37bb
commit 0cdc3d5408
3 changed files with 257 additions and 368 deletions
Download Patch File Download Diff File Expand all files Collapse all files
lib/streaming_providers/providers/hrti/constants.py
-2
View File
@@ -35,7 +35,6 @@ class HRTiDefaults:
"catalogue_structure": f"{BASE_URL}/api/api/ott/GetCatalogueStructure",
"catalogue": f"{BASE_URL}/api/api/ott/GetCatalogue",
"vod_details": f"{BASE_URL}/api/api/ott/GetVodDetails",
"seasons": f"{BASE_URL}/api/api/ott/GetSeasons",
"episodes": f"{BASE_URL}/api/api/ott/GetSeries",
"watch_later": f"{BASE_URL}/api/api/ott/GetWatchLater",
"editors_choice": f"{BASE_URL}/api/api/ott/GetEditorsChoice",
@@ -130,7 +129,6 @@ class HRTiConfig:
"catalogue_structure": f"{base_api_url}/GetCatalogueStructure",
"catalogue": f"{base_api_url}/GetCatalogue",
"vod_details": f"{base_api_url}/GetVodDetails",
"seasons": f"{base_api_url}/GetSeasons",
"episodes": f"{base_api_url}/GetSeries",
"watch_later": f"{base_api_url}/GetWatchLater",
"editors_choice": f"{base_api_url}/GetEditorsChoice",
lib/streaming_providers/providers/hrti/provider.py
+2 -2
View File
@@ -16,9 +16,9 @@ from .auth import HRTiAuthenticator
from .constants import HRTiConfig, HRTiDefaults
from .vod_manager import HRTiVodManager
# Navigation-only prefixes — these content_ids are category/series/season containers.
# 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:", "season:", "special:")
_NAV_PREFIXES = ("catalogue:", "series:", "special:")
# Playable VOD prefix — "details:{ref_id}" is an explicit single-item request.
_DETAILS_PREFIX = "details:"
lib/streaming_providers/providers/hrti/vod_manager.py
+255 -364
View File
@@ -2,45 +2,72 @@
"""
HRTi VOD Manager - Handles all VOD catalogue operations.
This module contains all VOD-specific logic including:
- Catalogue structure navigation
- Series/season/episode hierarchy
- Special collections (Watch Later, Editor's Choice)
- VOD item parsing and conversion to VodCategory/VodItem
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)
Field mapping notes (from real API responses):
NOT available: GetSeasons, GetEpisodes — these do not exist on this API.
GetCatalogue item Type values:
Type 2 → playable VOD (VodData sub-object)
Type 3 → series (SeriesData sub-object)
Type 5 → episode in a series listing (EpisodeData sub-object)
absent → navigable folder
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.
GetCatalogue vs GetVodDetails field names:
Listing: Title, PosterLandscape, VodData.Duration (ms), VodData.ProductionYear,
VodData.ContentRating, SeriesData.SeriesReferenceId, EpisodeData.*
Details: Title, PosterLandscape, DurationInFrames (frames at ~25fps → /25 for sec),
ProductionYear, ContentRating, SeriesReferenceId, SeasonNr, EpisodeNr,
FileName (streaming URL), SVODVideostores, IsSVOD, Actors, Directors,
Writers, Editors, Producers, ReferenceID (capital ID in details)
GetSeries: Result.Items list; each item uses EpisodeData sub-object
Type-3 series items in GetCatalogue listings link to GetSeries via:
content_id = "series:{SeriesReferenceId}:{SeasonReferenceId}"
The default SeasonReferenceId is "{SeriesReferenceId}_S01".
FileName URL parsing (e.g. https://cdn1oiv.akamaized.net/hrtvodorigin/REF.smil/manifest.mpd):
path[0] = origin (e.g. "hrtvodorigin")
path[1] = ref.smil (e.g. "1870699590_delin.smil")
ContentReferenceId = path[1] without ".smil"
ContentDrmId = "{origin}_{ref}.smil"
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
AuthorizeSession ContentType:
SVODVideostores non-empty → "svod", pass VideostoreReferenceIds
SVODVideostores empty → "vod", VideostoreReferenceIds = null
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
Pagination: NumberOfItems (not TotalCount), no HasMore field.
Token: _current_token.access_token (no get_access_token() method exists).
Pagination: NumberOfItems field (not TotalCount); no HasMore field.
"""
import json
from typing import Dict, List, Optional, Union
from urllib.parse import urlparse
from ...base.models.vod import VodCategory, VodItem
from ...base.utils.logger import logger
@@ -49,41 +76,24 @@ from .constants import HRTiDefaults
# GetCatalogue / GetSeries Type discriminator values
_TYPE_VOD = 2
_TYPE_SERIES = 3
_TYPE_EPISODE = 5 # episode within a series listing (GetSeries Result.Items)
_TYPE_EPISODE = 5
# Frames-per-second assumed for DurationInFrames → seconds conversion
# DurationInFrames in GetVodDetails is at ~25 fps
_ASSUMED_FPS = 25
class HRTiVodManager:
"""
Manages HRTi VOD operations.
Content ID encoding:
- "" → root catalogue structure
- "catalogue:{ref_id}" → items in category (paginated)
- "series:{series_ref_id}" → seasons of a series
- "season:{series_id}:{season_id}" → episodes of a season
- "details:{ref_id}" → single VOD item details
- "special:watch_later" → user's watch later list
- "special:editors_choice" → editor's picks
"""
"""Manages HRTi VOD catalogue operations."""
def __init__(self, provider):
"""
Initialize VOD manager.
Args:
provider: HRTiProvider instance (for http_manager, authenticator, config)
"""
self.provider = provider
self._config = provider.hrti_config
self._http_manager = provider.http_manager
self._authenticator = provider.authenticator
# --------------------------------------------------------------------------
# -------------------------------------------------------------------------
# Public API (called by provider.get_vod_category)
# --------------------------------------------------------------------------
# -------------------------------------------------------------------------
def get_category(
self,
@@ -92,16 +102,13 @@ class HRTiVodManager:
page_size: int = 24,
) -> Union[List, Dict]:
"""
Get VOD category children.
Return children for a VOD navigation node.
Args:
content_id: Opaque node identifier (see class docstring)
cursor: Pagination cursor (page number as string, 1-indexed)
page_size: Items per page
Returns:
Dict with ``entries``, ``next_cursor``, ``total`` for paginated endpoints;
List of VodCategory/VodItem for non-paginated endpoints.
Routing by content_id prefix:
"" → GetCatalogueStructure (root, top-level categories)
"catalogue:{ref_id}" → GetCatalogue items (paginated)
"series:{sid}:{ssid}" → GetSeries episodes
"special:*" → special collections
"""
if not content_id:
return self._fetch_catalogue_structure()
@@ -111,23 +118,13 @@ class HRTiVodManager:
return self._fetch_catalogue_items(ref_id, cursor, page_size)
if content_id.startswith("series:"):
series_id = content_id[len("series:"):]
return self._fetch_series_seasons(series_id)
if content_id.startswith("season:"):
# Format: season:{series_id}:{season_id}
parts = content_id[len("season:"):].split(":", 1)
# Format: series:{series_ref_id}:{season_ref_id}
parts = content_id[len("series:"):].split(":", 1)
if len(parts) == 2:
series_id, season_id = parts
return self._fetch_season_episodes(series_id, season_id)
logger.warning(f"Malformed season content_id: {content_id}")
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.startswith("details:"):
ref_id = content_id[len("details:"):]
item = self._fetch_vod_details(ref_id)
return [item] if item else []
if content_id == "special:watch_later":
return self._fetch_watch_later()
@@ -137,9 +134,9 @@ class HRTiVodManager:
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."""
@@ -154,7 +151,6 @@ class HRTiVodManager:
"Accept": "application/json",
"Content-Type": "application/json",
}
token = (
self._authenticator._current_token.access_token
if self._authenticator._current_token
@@ -162,7 +158,6 @@ class HRTiVodManager:
)
if token:
headers["authorization"] = f"Client {token}"
return headers
def _post(
@@ -172,56 +167,55 @@ class HRTiVodManager:
referer_path: str = "/videostore",
) -> Optional[Union[Dict, List]]:
"""
Make an authenticated POST to a named HRTi endpoint.
POST to a named endpoint and return the Result field.
Returns:
The ``Result`` field from the response JSON, or None on any error.
Result may be a dict (GetCatalogue) or a list (GetCatalogueStructure,
GetSeasons, GetEpisodes).
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:
response = self._http_manager.post(
resp = self._http_manager.post(
url,
operation="api",
headers=self._get_headers(referer_path),
data=json.dumps(payload),
)
response.raise_for_status()
data = response.json()
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')}"
)
return None
result = data.get("Result")
if result is None:
logger.debug(f"No Result in response for {endpoint_key}")
return result
return data.get("Result")
except Exception as e:
logger.error(f"Error calling {endpoint_key}: {e}")
return None
# --------------------------------------------------------------------------
# Root catalogue structure (GetCatalogueStructure)
# --------------------------------------------------------------------------
# -------------------------------------------------------------------------
# GetCatalogueStructure — root navigation tree
# -------------------------------------------------------------------------
def _fetch_catalogue_structure(self) -> List[VodCategory]:
"""
Fetch top-level categories from GetCatalogueStructure.
Fetch top-level navigation categories from GetCatalogueStructure.
The Result is a list of folder nodes. The field names here ARE
``Name`` and ``ReferenceId`` (confirmed from the root-level curl output).
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")
@@ -231,44 +225,45 @@ class HRTiVodManager:
categories = []
for node in result:
category = self._parse_catalogue_node(node)
if category:
categories.append(category)
cat = self._parse_structure_node(node)
if cat:
categories.append(cat)
logger.info(f"Retrieved {len(categories)} root VOD categories from HRTi")
return categories
def _parse_catalogue_node(self, node: Dict) -> Optional[VodCategory]:
def _parse_structure_node(self, node: Dict) -> Optional[VodCategory]:
"""
Parse a GetCatalogueStructure node.
Parse one node from GetCatalogueStructure.
These nodes use ``Name`` / ``ReferenceId`` / ``ChildCount`` / ``Icon``
(distinct from catalogue item nodes which use ``Title`` etc.).
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("Skipping catalogue node — missing ReferenceId or 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,
logo_url=node.get("Icon"),
description=node.get("Description"),
child_count=node.get("ChildCount"),
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 catalogue node: {e}")
logger.error(f"Error parsing structure node: {e}")
return None
# --------------------------------------------------------------------------
# Catalogue items (GetCatalogue)
# --------------------------------------------------------------------------
# -------------------------------------------------------------------------
# GetCatalogue — paginated items for a catalogue node
# -------------------------------------------------------------------------
def _fetch_catalogue_items(
self,
@@ -277,17 +272,10 @@ class HRTiVodManager:
page_size: int = 24,
) -> Dict:
"""
Fetch paginated items from GetCatalogue.
Fetch paginated items from GetCatalogue for a given ReferenceId.
The Result is a dict:
{
"Name": "...",
"NumberOfItems": 22,
"Items": [ ... ]
}
There is no ``HasMore`` or ``TotalCount`` field — pagination is derived
from ``NumberOfItems``.
Response: { "Name": "...", "NumberOfItems": N, "Items": [...] }
Pagination is derived from NumberOfItems — no HasMore field exists.
"""
page = 1
if cursor:
@@ -297,22 +285,18 @@ class HRTiVodManager:
logger.warning(f"Invalid cursor '{cursor}', using page 1")
if page > HRTiDefaults.MAX_VOD_PAGES:
logger.warning(f"Page {page} exceeds max {HRTiDefaults.MAX_VOD_PAGES}")
return {"entries": [], "next_cursor": None, "total": None}
payload = {
"ReferenceId": ref_id,
"ItemsPerPage": page_size,
"PageNumber": page,
}
result = self._post("catalogue", payload, "/videostore")
result = self._post(
"catalogue",
{"ReferenceId": ref_id, "ItemsPerPage": page_size, "PageNumber": page},
)
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") # actual field name
total_count = result.get("NumberOfItems")
entries = []
for item in raw_items:
@@ -320,118 +304,95 @@ class HRTiVodManager:
if parsed:
entries.append(parsed)
# has_more: total known and we haven't exhausted it yet
has_more = (
total_count is not None
and page * page_size < total_count
and page < HRTiDefaults.MAX_VOD_PAGES
)
next_cursor = str(page + 1) if has_more else None
logger.debug(
f"catalogue {ref_id} page {page}: "
f"{len(entries)} entries, total {total_count}, next_cursor {next_cursor}"
f"catalogue {ref_id} page {page}: {len(entries)} entries, "
f"total {total_count}, next_cursor {page + 1 if has_more else None}"
)
return {"entries": entries, "next_cursor": next_cursor, "total": total_count}
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]]:
"""
Parse one item from a GetCatalogue ``Items`` list.
Discrimination is done via the ``Type`` field:
Type 2 → playable VOD (VodData present)
Type 3 → series (SeriesData present)
other → treat as navigable sub-category
Field names in catalogue items:
Title ← display name
ReferenceId ← stable id
PosterLandscape / PosterPortrait ← artwork
VodData.Duration ← ms
VodData.ProductionYear ← int
VodData.ContentRating ← string
SeriesData.SeriesReferenceId
"""
"""Dispatch catalogue item parsing by Type field."""
try:
item_type = item.get("Type")
if item_type == _TYPE_SERIES:
return self._parse_series_item(item)
if item_type == _TYPE_VOD:
return self._parse_vod_catalogue_item(item)
# Anything without a known Type is a navigable folder
return self._parse_folder_item(item)
if item_type == _TYPE_SERIES:
return self._parse_series_catalogue_item(item)
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}")
return None
def _parse_folder_item(self, item: Dict) -> Optional[VodCategory]:
"""Parse a navigable folder node found inside a GetCatalogue result."""
def _parse_folder_catalogue_item(self, item: Dict) -> Optional[VodCategory]:
"""No Type — navigable sub-folder node."""
ref_id = item.get("ReferenceId")
name = (item.get("Title") or item.get("Name") or "").strip()
if not ref_id or not name:
logger.debug("Skipping folder item — missing ReferenceId or title")
return None
return VodCategory(
name=name,
content_id=f"catalogue:{ref_id}",
provider=self.provider.provider_name,
logo_url=item.get("PosterLandscape") or item.get("PosterPortrait"),
description=item.get("Description"),
child_count=item.get("ChildCount"), # usually absent; that's fine
description=None,
child_count=item.get("ChildCount"),
)
def _parse_series_item(self, item: Dict) -> Optional[VodCategory]:
"""Parse a Type-3 series item into a drillable VodCategory."""
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.
"""
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:
logger.debug("Skipping series item — missing SeriesReferenceId or Title")
return None
season_count = (
series_data.get("LastSeasonNumber") # best proxy available
)
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}",
content_id=f"series:{series_ref_id}:{default_season_ref}",
provider=self.provider.provider_name,
logo_url=item.get("PosterLandscape") or item.get("PosterPortrait"),
description=item.get("Description"),
description=None,
child_count=season_count,
)
def _parse_vod_catalogue_item(self, item: Dict) -> Optional[VodItem]:
"""
Parse a Type-2 (playable) item from a GetCatalogue listing.
Type-2 playable item from a GetCatalogue listing.
Metadata here is sparse — no StreamingURL, no genres list, etc.
The ``content_id`` is the raw ReferenceId so callers can pass it to
``get_manifest()`` / ``get_drm()`` directly.
manifest_script is None — FileName only available via GetVodDetails.
content_id is the raw ReferenceId (no prefix) so provider routing works.
"""
ref_id = item.get("ReferenceId")
title = (item.get("Title") or "").strip()
if not ref_id or not title:
logger.debug("Skipping VOD catalogue item — missing ReferenceId or Title")
return None
vod_data = item.get("VodData") or {}
# Duration in the catalogue listing is milliseconds
duration_ms = vod_data.get("Duration")
duration_sec = int(duration_ms / 1000) if duration_ms else None
# Genre comes as a comma-separated category string in catalogue listings
category_str = item.get("VodCategoryNames") or ""
genres = [g.strip() for g in category_str.split(",") if g.strip()] or None
@@ -440,7 +401,7 @@ class HRTiVodManager:
content_id=ref_id,
provider=self.provider.provider_name,
logo_url=item.get("PosterLandscape") or item.get("PosterPortrait"),
description=item.get("Description"),
description=None,
long_description=None,
original_title=None,
duration_seconds=duration_sec,
@@ -457,91 +418,16 @@ class HRTiVodManager:
trailer_url=None,
is_highlight=False,
content_type="MOVIE",
manifest_script=None, # not provided in listings — fetched on demand
manifest_script=None,
)
# --------------------------------------------------------------------------
# Series, Seasons, Episodes
# --------------------------------------------------------------------------
def _fetch_series_seasons(self, series_ref_id: str) -> List[VodCategory]:
def _parse_episode_catalogue_item(self, item: Dict) -> Optional[VodItem]:
"""
Get seasons for a series from GetSeasons.
Type-5 episode item from GetCatalogue or GetSeries.
Result is a list of season objects.
"""
payload = {"SeriesReferenceId": series_ref_id}
result = self._post("seasons", payload, "/videostore")
if not result or not isinstance(result, list):
logger.warning(f"No seasons returned for series {series_ref_id}")
return []
seasons = []
for season in result:
season_ref_id = season.get("ReferenceId")
season_name = (season.get("Name") or "").strip()
if not season_name:
season_num = season.get("SeasonNumber")
season_name = f"Season {season_num}" if season_num else "Unknown Season"
seasons.append(VodCategory(
name=season_name,
content_id=f"season:{series_ref_id}:{season_ref_id}",
provider=self.provider.provider_name,
logo_url=season.get("PosterLandscape") or season.get("Icon"),
description=season.get("Description"),
child_count=season.get("EpisodeCount"),
))
logger.info(f"Retrieved {len(seasons)} seasons for series {series_ref_id}")
return seasons
def _fetch_season_episodes(
self,
series_ref_id: str,
season_ref_id: str,
) -> List[VodItem]:
"""
Get episodes for a season from GetSeries.
The endpoint is GetSeries (not GetEpisodes — that endpoint does not exist).
Result is a dict: { "Items": [...], "LastWatchedEpisodeReferenceId": ..., ... }
Each item uses EpisodeData sub-object (not VodData) and Type 5.
"""
payload = {
"SeriesReferenceId": series_ref_id,
"SeasonReferenceId": season_ref_id,
}
result = self._post("episodes", payload, "/videostore")
# Result is a dict with an Items list (not a bare list)
if not result or not isinstance(result, dict):
logger.warning(f"No episodes returned for season {season_ref_id}")
return []
raw_items = result.get("Items") or []
episodes = []
for ep in raw_items:
vod_item = self._parse_series_episode_item(ep, series_ref_id)
if vod_item:
episodes.append(vod_item)
logger.info(
f"Retrieved {len(episodes)} episodes for season {season_ref_id} "
f"of series {series_ref_id}"
)
return episodes
def _parse_series_episode_item(self, item: Dict, series_ref_id: str) -> Optional[VodItem]:
"""
Parse a Type-5 episode from a GetSeries Result.Items list.
Schema differs from both catalogue items and vod_details:
ReferenceId, Title, PosterLandscape, PosterPortrait, VodCategoryNames
EpisodeData.EpisodeNr, EpisodeData.SeasonNr, EpisodeData.Duration (ms),
EpisodeData.ContentRating
SeriesData.SeriesReferenceId, SeriesData.SeriesName
EpisodeData sub-object: EpisodeNr, SeasonNr, Duration (ms), ContentRating.
SeriesData sub-object: SeriesReferenceId, SeriesName.
content_id is the raw ReferenceId.
"""
ref_id = item.get("ReferenceId")
title = (item.get("Title") or "").strip()
@@ -550,10 +436,8 @@ class HRTiVodManager:
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
@@ -574,79 +458,101 @@ class HRTiVodManager:
director=None,
season_number=ep_data.get("SeasonNr"),
episode_number=ep_data.get("EpisodeNr"),
series_id=series_data.get("SeriesReferenceId") or series_ref_id,
series_id=series_data.get("SeriesReferenceId"),
series_title=series_data.get("SeriesName"),
trailer_url=None,
is_highlight=False,
content_type="SERIES",
manifest_script=None, # only available via GetVodDetails
manifest_script=None,
)
# --------------------------------------------------------------------------
# VOD item details (GetVodDetails)
# --------------------------------------------------------------------------
# -------------------------------------------------------------------------
# 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.
"""
result = self._post(
"episodes",
{
"SeriesReferenceId": series_ref_id,
"SeasonReferenceId": season_ref_id,
"PageSize": 100,
"PageNumber": 1,
},
)
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.
This is also the only place StreamingURL is available.
"""
result = self._post("vod_details", {"ReferenceId": ref_id}, "/videostore")
"""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)
def _parse_vod_details_item(self, item: Dict) -> Optional[VodItem]:
"""
Parse a rich VOD item as returned by GetVodDetails.
Parse a rich VOD item from GetVodDetails.
Real field names (verified from API):
ReferenceID (capital ID!), Title, OriginalTitle, Description,
FileName (streaming URL — NOT StreamingURL),
DurationInFrames (frames, divide by _ASSUMED_FPS for seconds),
ProductionYear, ContentRating, SeriesReferenceId, SeriesName,
SeasonNr, EpisodeNr, TrailerUrl, Type ("vod" / "episode"),
PosterLandscape, PosterPortrait,
Actors / Directors / Writers / Editors / Producers (strings or null),
SVODVideostores (list of store ids, governs auth type),
IsSVOD (bool — but SVODVideostores non-empty is the reliable signal).
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
"""
try:
# Details uses "ReferenceID" (capital ID), listings use "ReferenceId"
ref_id = item.get("ReferenceID") or item.get("ReferenceId")
title = (item.get("Title") or "").strip()
if not ref_id or not title:
logger.debug("Skipping VOD details item — missing ReferenceId or Title")
return None
# DurationInFrames at ~25fps → seconds
frames = item.get("DurationInFrames")
duration_sec = int(frames / _ASSUMED_FPS) if frames else None
# Actors field is a string (comma-separated or null), not a list
actors_str = item.get("Actors")
cast = (
[a.strip() for a in actors_str.split(",") if a.strip()]
if actors_str else None
)
# Directors, Writers also strings
directors_str = item.get("Directors") or ""
genre_from_category = item.get("AssetCategory") or ""
# Type field: "vod" or "episode"
item_type_str = (item.get("Type") or "").lower()
content_type = "SERIES" if item_type_str == "episode" else "MOVIE"
# Streaming URL is in FileName, not StreamingURL
filename = item.get("FileName") or ""
# Store SVOD data on the item for use by get_vod_session_data
genre = item.get("AssetCategory") or None
svod_videostores = item.get("SVODVideostores") or []
vod_item = VodItem(
@@ -660,10 +566,10 @@ class HRTiVodManager:
duration_seconds=duration_sec,
release_year=item.get("ProductionYear"),
rating=item.get("ContentRating"),
genre=genre_from_category or None,
genres=[genre_from_category] if genre_from_category else None,
genre=genre,
genres=[genre] if genre else None,
cast=cast,
director=directors_str or None,
director=item.get("Directors") or None,
season_number=item.get("SeasonNr"),
episode_number=item.get("EpisodeNr"),
series_id=item.get("SeriesReferenceId"),
@@ -671,13 +577,9 @@ class HRTiVodManager:
trailer_url=item.get("TrailerUrl"),
is_highlight=False,
content_type=content_type,
manifest_script=filename or None,
manifest_script=item.get("FileName") or None,
)
# Stash SVOD stores so get_vod_session_data can read them without
# a second GetVodDetails call
vod_item._svod_videostores = svod_videostores
logger.debug(f"Parsed VOD details: {title} ({ref_id}), svod={bool(svod_videostores)}")
return vod_item
@@ -685,68 +587,57 @@ class HRTiVodManager:
logger.error(f"Error parsing VOD details item: {e}")
return None
# --------------------------------------------------------------------------
# -------------------------------------------------------------------------
# Special collections
# --------------------------------------------------------------------------
# -------------------------------------------------------------------------
def _fetch_watch_later(self) -> List[VodItem]:
"""Get user's Watch Later list (requires user auth)."""
"""User's Watch Later list (requires user auth)."""
result = self._post("watch_later", {}, "/watch_later")
if not result or not isinstance(result, list):
logger.debug("No watch later items returned (may require user auth)")
return []
items = [self._parse_vod_details_item(i) for i in result]
items = [i for i in items if i]
logger.info(f"Retrieved {len(items)} items from watch later")
return items
return [i for i in items if i]
def _fetch_editors_choice(self) -> List[VodItem]:
"""Get Editor's Choice list."""
"""Editor's Choice list."""
result = self._post("editors_choice", {}, "/editors_choice")
if not result or not isinstance(result, list):
logger.debug("No editor's choice items returned")
return []
items = [self._parse_vod_details_item(i) for i in result]
items = [i for i in items if i]
logger.info(f"Retrieved {len(items)} items from editor's choice")
return items
return [i for i in items if i]
# --------------------------------------------------------------------------
# Provider-facing helpers
# --------------------------------------------------------------------------
# -------------------------------------------------------------------------
# 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.
The URL comes from the ``FileName`` field in GetVodDetails
(stored as ``manifest_script`` on the VodItem).
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/StreamingURL found for VOD item {content_id}")
logger.warning(f"No FileName found for VOD item {content_id}")
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 GetVodDetails:
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" (ref with extension)
→ ContentReferenceId = "REF" (stem without .smil)
→ ContentDrmId = "hrtvodorigin_REF.smil" (origin_ref.smil)
path[0] = "hrtvodorigin" (origin)
path[1] = "REF.smil"
SVODVideostores non-empty → ContentType="svod", pass VideostoreReferenceIds
SVODVideostores empty → ContentType="vod", VideostoreReferenceIds=null
ContentReferenceId = "REF" (stem without .smil)
ContentDrmId = "hrtvodorigin_REF.smil"
ContentType = "svod" if SVODVideostores non-empty else "vod"
VideostoreReferenceIds = SVODVideostores or null
"""
try:
details = self._fetch_vod_details(content_id)
@@ -754,22 +645,22 @@ class HRTiVodManager:
logger.error(f"Cannot authorize VOD session — no FileName for {content_id}")
return None
filename = details.manifest_script # e.g. https://.../hrtvodorigin/REF.smil/manifest.mpd
filename = details.manifest_script
svod_stores = getattr(details, "_svod_videostores", []) or []
# Parse ContentReferenceId and ContentDrmId from FileName path
from urllib.parse import urlparse
path_parts = urlparse(filename).path.strip("/").split("/")
# path_parts: ["hrtvodorigin", "REF.smil", "manifest.mpd"]
if len(path_parts) < 2:
logger.error(f"Cannot parse FileName path for {content_id}: {filename}")
return None
origin = path_parts[0] # "hrtvodorigin"
ref_with_ext = path_parts[1] # "REF.smil"
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}" # "hrtvodorigin_REF.smil"
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