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

movetv: Add catchup

Browse Source
This commit is contained in:
Nirvana
2026-07-25 07:53:13 +02:00
parent e7efe70582
commit efcd9484ca
3 changed files with 318 additions and 471 deletions
Download Patch File Download Diff File Expand all files Collapse all files
lib/streaming_providers/providers/movetv/constants.py
+6
View File
@@ -33,6 +33,7 @@ class MoveTVConfig:
# EPG
PATH_EPG_ALL: str = "/api/v2/content/epg/all"
PATH_EPG_SOURCE: str = "/api/v2/content/epg/source/get" # Catchup manifest
# VOD — catalogue / filtering
PATH_VOD_FILTERS: str = "/api/v2/content/vod/filters"
@@ -151,6 +152,11 @@ class MoveTVConfig:
"""Full URL for fetching EPG data for a single channel."""
return f"{cls.API_BASE_URL}{cls.PATH_EPG_ALL}"
@classmethod
def epg_source_url(cls) -> str:
"""Full URL for fetching catchup manifest for an EPG entry."""
return f"{cls.API_BASE_URL}{cls.PATH_EPG_SOURCE}"
@classmethod
def vod_filters_url(cls) -> str:
return f"{cls.API_BASE_URL}{cls.PATH_VOD_FILTERS}"
lib/streaming_providers/providers/movetv/epg_manager.py
+8 -8
View File
@@ -125,18 +125,18 @@ class MoveTvEpgManager:
The provider's numeric content ID for the channel (as a string).
backwards:
Hours of past programming to include (API default: 2).
Ignored when start_time/end_time are provided.
Ignored when both start_time and end_time are provided.
forwards:
Hours of future programming to include (API default: 2).
Ignored when start_time/end_time are provided.
Ignored when both start_time and end_time are provided.
start_time:
Inclusive window start (timezone-aware or naive-local datetime).
When provided together with end_time, takes precedence over
backwards/forwards.
When provided, anchors the API query. If end_time is omitted,
uses `backwards` and `forwards` around this anchor.
end_time:
Inclusive window end (timezone-aware or naive-local datetime).
When provided together with start_time, takes precedence over
backwards/forwards.
backwards/forwards and calculates a symmetric window.
Returns
-------
@@ -303,7 +303,7 @@ class MoveTvEpgManager:
Only start_time given:
anchor = start_time
backwards = 0
backwards = backwards (caller's value, default 2)
forwards = forwards (caller's value, default 2)
Only end_time given:
@@ -366,9 +366,9 @@ class MoveTvEpgManager:
anchor_utc = _to_utc(start_time)
logger.debug(
f"MoveTV EPG: start_time only — anchor={anchor_utc.isoformat()} "
f"backwards=0 forwards={forwards}h"
f"backwards={backwards}h forwards={forwards}h"
)
return anchor_utc, 0, forwards
return anchor_utc, backwards, forwards
if end_time is not None:
anchor_utc = _to_utc(end_time)
lib/streaming_providers/providers/movetv/provider.py
+304 -463
View File
@@ -1,7 +1,10 @@
# streaming_providers/providers/movetv/provider.py
import re
import time
import threading
import requests
from collections import OrderedDict
from datetime import datetime, timezone
from typing import ClassVar, Dict, List, Optional, Any, Tuple
from ...base.models import DRMConfig, StreamingChannel
@@ -21,26 +24,38 @@ from .vod_manager import (
from .epg_manager import MoveTvEpgManager
class _SourceCache:
"""Thread-safe LRU-like cache for source URLs and Play-Auth headers."""
def __init__(self, max_size: int = 100):
self._cache: OrderedDict[str, tuple] = OrderedDict()
self._lock = threading.Lock()
self._max_size = max_size
def get(self, key: str) -> Optional[tuple]:
with self._lock:
if key in self._cache:
self._cache.move_to_end(key)
return self._cache[key]
return None
def set(self, key: str, value: tuple) -> None:
with self._lock:
if key in self._cache:
self._cache.move_to_end(key)
self._cache[key] = value
if len(self._cache) > self._max_size:
self._cache.popitem(last=False)
def pop(self, key: str, default: Any = None) -> Any:
with self._lock:
return self._cache.pop(key, default)
class MoveTVChannel(StreamingChannel):
"""
Extends StreamingChannel with move.tv-specific fields.
content_id stores the *contentId* so that get_epg(content_id) can pass
it directly to the EPG endpoint without any mapping lookup.
Extra fields
------------
live_id : int – the liveId used by the live-source / manifest
endpoint (required for get_manifest mapping)
catchup_hours : int – how many hours of catch-up are available (0 = none)
stream_uid : str – the streamUid used by the CDN (e.g. "rts1")
play_auth_header: str – the X-Play-Auth value delivered with the manifest
source response; callers must inject this when
requesting the actual .mpd / .m3u8 from the CDN
"""
# How many seconds before the stated expiry we treat the token as stale.
# 60 s gives enough runway to start playback before the CDN rejects it.
_PLAY_AUTH_EARLY_EXPIRY_BUFFER: int = 60
def __init__(self, *args, live_id: int = 0, catchup_hours: int = 0,
@@ -51,17 +66,10 @@ class MoveTVChannel(StreamingChannel):
self.catchup_hours: int = catchup_hours
self.stream_uid: str = stream_uid
self.play_auth_header: str = play_auth_header
# Unix timestamp after which play_auth_header must be treated as stale.
# 0.0 means "no expiry parsed / never cached".
self.play_auth_expires_at: float = play_auth_expires_at
def is_play_auth_valid(self) -> bool:
"""Return True only when a header is cached and has not yet expired."""
if not self.play_auth_header:
return False
if self.play_auth_expires_at == 0.0:
# No expiry was parsed — treat as already invalid so a fresh fetch
# is triggered.
if not self.play_auth_header or self.play_auth_expires_at == 0.0:
return False
return time.time() < (self.play_auth_expires_at - self._PLAY_AUTH_EARLY_EXPIRY_BUFFER)
@@ -78,50 +86,22 @@ class MoveTVChannel(StreamingChannel):
class MoveTVProvider(StreamingProvider):
"""
Streaming provider for move.tv (MTS-SI platform).
Authentication
--------------
Token-based: login returns an auth_token used as X-Auth-Token on all
subsequent requests. No anonymous / client-credentials fallback exists.
Channel list
------------
GET /api/v2/content/live/all – only subscribed channels are kept.
catchup_hours comes from catchup.duration (already in hours).
content_id convention
---------------------
contentId is stored as content_id so get_epg(content_id) can pass it
directly to the EPG endpoint with no mapping lookup. liveId is kept in
live_id and looked up when get_manifest() needs to call the source endpoint.
Manifest
--------
POST /api/v2/content/live/source/get – requires customer_id,
customer_profile_id, liveId (= content_id) and dtype. Returns a
content_url (.mpd) and an X-Play-Auth protection header. This provider
returns the content_url only; the caller injects X-Play-Auth.
DRM
---
The AES-128 / token-based protection is handled via X-Play-Auth (see
above). Widevine/PlayReady are not active in observed traffic; get_drm()
returns an empty list.
"""
PROVIDER_LABEL: ClassVar[str] = "move.tv"
SUPPORTED_AUTH_TYPES: ClassVar[List[str]] = ["user_credentials"]
_PLAY_AUTH_EXPIRY_RE = re.compile(r"expires-(\d+)")
def __init__(
self,
country: str = "SI",
config: Optional[Dict] = None,
proxy_config: Optional[ProxyConfig] = None,
settings_manager=None,
self,
country: str = "SI",
config: Optional[Dict] = None,
proxy_config: Optional[ProxyConfig] = None,
settings_manager=None,
):
super().__init__(country)
# HTTP manager (shared with authenticator)
self.http_manager = self._setup_http_manager(
provider_name="movetv",
proxy_config=proxy_config,
@@ -129,7 +109,6 @@ class MoveTVProvider(StreamingProvider):
timeout=MoveTVConfig.TIMEOUT,
)
# Authenticator
self.authenticator = MoveTVAuthenticator(
proxy_config=proxy_config,
http_manager=self.http_manager,
@@ -139,17 +118,11 @@ class MoveTVProvider(StreamingProvider):
self._vod = MoveTvVodManager(self)
self._epg = MoveTvEpgManager(self.authenticator)
# Share the same http_manager session with the authenticator
self.http_manager = self._share_http_manager_with_authenticator(self.authenticator)
# Standalone play-auth cache: content_id -> (header_value, expires_at)
# Used by _store_play_auth / _get_play_auth_header so they never depend
# on self.channels being populated (get_channels may not have been called).
self._play_auth_cache: Dict[str, tuple] = {}
self._manifest_url_cache: Dict[str, str] = {}
self._last_content_id: Optional[str] = None
# Unified thread-safe cache for both Live and Catchup sources
self._source_cache = _SourceCache(max_size=100)
# Attempt authentication at startup; non-fatal if it fails
try:
self.authenticator.authenticate()
logger.info("move.tv: Authentication successful during initialisation")
@@ -174,7 +147,6 @@ class MoveTVProvider(StreamingProvider):
@property
def uses_dynamic_manifests(self) -> bool:
# Manifests are fetched per-play via the source endpoint
return True
@property
@@ -183,35 +155,155 @@ class MoveTVProvider(StreamingProvider):
@property
def requires_manifest_context(self) -> bool:
# Manifests need http_manager context
return True
@property
def catchup_window(self) -> int:
return 168 # 7 days
@property
def supports_catchup(self) -> bool:
return True
# ------------------------------------------------------------------
# Header helpers
# Internal Helpers
# ------------------------------------------------------------------
def _authenticated_headers(self) -> Dict[str, str]:
"""Return API headers with the current X-Auth-Token injected."""
auth_token = self.authenticator.get_auth_token()
return MoveTVConfig.get_api_headers(auth_token=auth_token)
def _channel_by_id(self, content_id: str) -> Optional[MoveTVChannel]:
for ch in (self.channels or []):
if isinstance(ch, MoveTVChannel) and ch.content_id == content_id:
return ch
return None
def _get_channel_with_rebuild(self, content_id: str) -> Optional[MoveTVChannel]:
channel = self._channel_by_id(content_id)
if channel is None:
logger.info(f"move.tv: Channel not found for {content_id}; rebuilding cache...")
self.get_channels()
channel = self._channel_by_id(content_id)
return channel
def _fetch_source_endpoint(
self, url: str, payload: Dict[str, Any], content_id: str, cache_key: str, operation: str = "manifest"
) -> Optional[Dict[str, Any]]:
"""
Unified POST method for fetching live or catchup source manifests.
Handles caching, 401 retries, and Play-Auth extraction.
"""
# 1. Check Cache
cached = self._source_cache.get(cache_key)
if cached:
content_url, header_value, expires_at = cached
if expires_at and time.time() < (expires_at - MoveTVChannel._PLAY_AUTH_EARLY_EXPIRY_BUFFER):
logger.debug(f"move.tv: Cache hit for {operation} (key={cache_key})")
return {"content_url": content_url, "protection": {"headerValue": header_value}}
else:
logger.debug(f"move.tv: Cache expired for {operation} (key={cache_key})")
# 2. Get Session
session = self.authenticator.get_session_info()
if not session:
logger.info(f"move.tv: No session info; re-authenticating before {operation} fetch")
self.authenticator.authenticate(force_refresh=True)
session = self.authenticator.get_session_info()
if not session:
logger.error(f"move.tv: Unable to obtain session info for {operation} fetch")
return None
# 3. Prepare Payload
payload["customerId"] = session["customer_id"]
payload["customerProfileId"] = session["customer_profile_id"]
payload["appVersion"] = MoveTVConfig.APP_VERSION
payload["dtype"] = MoveTVConfig.DTYPE_DASH
headers = MoveTVConfig.get_api_headers(auth_token=session["auth_token"])
# 4. Execute POST with 401 Retry
try:
response = self.http_manager.post(url, operation=operation, json=payload, headers=headers)
is_401 = response.status_code == 401
except requests.HTTPError as exc:
if exc.response is not None and exc.response.status_code == 401:
is_401 = True
response = exc.response
else:
logger.error(f"move.tv: HTTP error during {operation} fetch: {exc}")
raise
if is_401:
logger.info(f"move.tv: 401 on {operation} fetch — attempting token refresh …")
token = self.authenticator.get_current_token()
refreshed = self.authenticator.refresh_token(token) if token else None
if not refreshed:
logger.info("move.tv: Refresh failed, falling back to full login")
self.authenticator.authenticate(force_refresh=True)
session = self.authenticator.get_session_info()
if not session:
logger.error("move.tv: Unable to obtain session after token refresh")
return None
payload["customerId"] = session["customer_id"]
payload["customerProfileId"] = session["customer_profile_id"]
headers = MoveTVConfig.get_api_headers(auth_token=session["auth_token"])
try:
response = self.http_manager.post(url, operation=operation, json=payload, headers=headers)
except requests.HTTPError as exc:
logger.error(f"move.tv: HTTP error during {operation} fetch retry: {exc}")
raise
# 5. Parse Response
try:
response.raise_for_status()
data = response.json()
except Exception as exc:
logger.error(f"move.tv: Failed to parse {operation} response: {exc}")
return None
if not data.get("success"):
logger.warning(f"move.tv: {operation} source endpoint returned success=false")
return None
# 6. Cache and Return
content_url = data.get("content_url")
protection = data.get("protection", {})
header_value = protection.get("headerValue", "")
expires_at = 0.0
if header_value:
match = self._PLAY_AUTH_EXPIRY_RE.search(header_value)
if match:
expires_at = float(match.group(1))
else:
logger.warning(f"move.tv: Could not parse expiry from X-Play-Auth for {cache_key}")
self._source_cache.set(cache_key, (content_url, header_value, expires_at))
logger.info(f"move.tv: Fetched and cached {operation} for {cache_key}: {content_url}")
# Update channel object if it exists in memory
channel = self._channel_by_id(content_id)
if isinstance(channel, MoveTVChannel):
channel.play_auth_header = header_value
channel.play_auth_expires_at = expires_at
return data
# ------------------------------------------------------------------
# get_channels
# ------------------------------------------------------------------
def get_channels(self, **kwargs) -> List[MoveTVChannel]:
"""
Fetch the live channel list and return only subscribed channels.
POST /api/v2/content/live/all with customerId and appVersion.
Only subscribed channels are kept; liveId is used as content_id.
"""
try:
channels = self._fetch_channels()
self.channels = channels # type: ignore[assignment]
self.channels = channels
logger.info(f"move.tv: Loaded {len(channels)} subscribed channels")
return channels
except requests.HTTPError as exc:
if exc.response is not None and exc.response.status_code == 401:
logger.info("move.tv: 401 on channel fetch — attempting token refresh …")
@@ -219,7 +311,7 @@ class MoveTVProvider(StreamingProvider):
token = self.authenticator.get_current_token()
if token and self.authenticator.refresh_token(token):
channels = self._fetch_channels()
self.channels = channels # type: ignore[assignment]
self.channels = channels
return channels
except Exception:
pass
@@ -229,28 +321,19 @@ class MoveTVProvider(StreamingProvider):
try:
self.authenticator.authenticate(force_refresh=True)
channels = self._fetch_channels()
self.channels = channels # type: ignore[assignment]
self.channels = channels
return channels
except Exception as retry_exc:
logger.error(f"move.tv: Channel fetch retry failed: {retry_exc}")
return []
except requests.RequestException as exc:
logger.error(f"move.tv: HTTP error fetching channels: {exc}")
try:
logger.info("move.tv: Retrying channel fetch after token refresh …")
self.authenticator.invalidate_token()
channels = self._fetch_channels()
self.channels = channels # type: ignore[assignment]
return channels
except Exception as retry_exc:
logger.error(f"move.tv: Channel fetch retry failed: {retry_exc}")
return []
return []
except Exception as exc:
logger.error(f"move.tv: Unexpected error fetching channels: {exc}")
return []
def _fetch_channels(self) -> List[MoveTVChannel]:
"""POST the channel list endpoint and parse the response."""
session = self.authenticator.get_session_info()
if not session:
self.authenticator.authenticate(force_refresh=True)
@@ -284,15 +367,7 @@ class MoveTVProvider(StreamingProvider):
return channels
def _parse_channel_item(self, item: Dict[str, Any]) -> Optional[MoveTVChannel]:
"""
Parse a single item from the /api/v2/content/live/all content array.
Returns None for:
- unsubscribed channels
- items missing required identifiers
"""
try:
# Drop unsubscribed channels
if not item.get("subscribed", False):
return None
@@ -301,24 +376,17 @@ class MoveTVProvider(StreamingProvider):
name = item.get("contentName", "")
stream_uid = item.get("streamUid", "")
# Both identifiers are required
if not live_id or not catalog_id or not name:
logger.debug(f"move.tv: Skipping channel item with missing ids/name: {item}")
return None
# Logo
picture = item.get("picture", {})
logo_url = MoveTVConfig.build_logo_url(picture.get("icon"))
# Catch-up duration — the API field is already in hours
catchup: Dict = item.get("catchup", {})
catchup_hours: int = int(catchup.get("duration", 0)) if catchup else 0
# Audio-only channels
is_audio = bool(item.get("audioOnly", False))
# contentId stored as content_id so get_epg() needs no mapping;
# liveId stored in live_id for use by get_manifest()
channel = MoveTVChannel(
name=name,
content_id=str(catalog_id),
@@ -332,12 +400,10 @@ class MoveTVProvider(StreamingProvider):
is_radio=is_audio,
language="sr",
country=self.country,
# move.tv specifics
live_id=int(live_id),
catchup_hours=catchup_hours,
stream_uid=stream_uid,
)
channel.channel_number = item.get("contentPosition")
return channel
@@ -346,397 +412,172 @@ class MoveTVProvider(StreamingProvider):
return None
# ------------------------------------------------------------------
# get_manifest
# Live Manifest
# ------------------------------------------------------------------
def get_manifest_with_headers(self, content_id: str, **kwargs) -> Tuple[Optional[str], Dict[str, str]]:
"""Unified method to fetch both URL and headers in one go, avoiding redundant API calls."""
try:
channel = self._get_channel_with_rebuild(content_id)
if not channel:
return None, {"User-Agent": MoveTVConfig.USER_AGENT}
payload = {"liveId": channel.live_id}
source_data = self._fetch_source_endpoint(
MoveTVConfig.live_source_url(),
payload=payload,
content_id=content_id,
cache_key=content_id,
operation="live_manifest"
)
if source_data:
content_url = source_data.get("content_url")
header_value = source_data.get("protection", {}).get("headerValue", "")
return content_url, {"User-Agent": MoveTVConfig.USER_AGENT, "X-Play-Auth": header_value}
return None, {"User-Agent": MoveTVConfig.USER_AGENT}
except Exception as exc:
logger.error(f"move.tv: Unexpected error fetching live manifest: {exc}")
return None, {"User-Agent": MoveTVConfig.USER_AGENT}
def get_manifest(self, content_id: str, **kwargs) -> Optional[str]:
if content_id != self._last_content_id:
logger.debug(
f"move.tv: Channel switch detected "
f"({self._last_content_id!r} → {content_id!r}), "
f"invalidating play-auth and manifest URL cache"
)
self._play_auth_cache.pop(content_id, None)
self._manifest_url_cache.pop(content_id, None)
self._last_content_id = content_id
channel = self._channel_by_id(content_id)
if channel is None:
logger.info(
f"move.tv: Channel not found in cache for content_id={content_id!r}; "
"attempting to rebuild channel cache before manifest fetch"
)
self.get_channels()
channel = self._channel_by_id(content_id)
if channel is None:
logger.error(
f"move.tv: Channel still not found after cache rebuild for "
f"content_id={content_id!r}; cannot resolve liveId for manifest fetch"
)
return None
live_id = channel.live_id
# If we have both a valid play-auth token and a cached manifest URL,
# skip the source endpoint entirely — the MPD TTL is server-driven at 0
# but the token (~24h) and URL are both stable.
cached_token = self._play_auth_cache.get(content_id)
cached_url = self._manifest_url_cache.get(content_id)
if cached_token and cached_url:
header_value, expires_at = cached_token
buffer = MoveTVChannel._PLAY_AUTH_EARLY_EXPIRY_BUFFER
if expires_at and time.time() < (expires_at - buffer):
logger.debug(
f"move.tv: Reusing cached manifest URL and X-Play-Auth for "
f"content_id={content_id} (token expires in "
f"{expires_at - time.time():.0f}s)"
)
return cached_url
# Cache miss or expired token — hit the source endpoint.
source_data = self._fetch_live_source(live_id)
if source_data is None:
return None
self._store_play_auth(content_id, source_data)
content_url: Optional[str] = source_data.get("content_url")
if not content_url:
logger.warning(
f"move.tv: No content_url in source response for liveId={live_id}"
)
return None
self._manifest_url_cache[content_id] = content_url
logger.info(f"move.tv: Manifest URL for liveId={live_id}: {content_url}")
return content_url
def _get_play_auth_header(self, content_id: str) -> Optional[str]:
"""
Return the X-Play-Auth header value for a channel.
Checks the standalone _play_auth_cache first (populated by every
_store_play_auth call, regardless of whether self.channels exists).
A cached value is only returned when it has not expired (with a
60-second early-expiry buffer). A stale or absent entry triggers a
fresh _fetch_live_source call.
"""
cached = self._play_auth_cache.get(content_id)
if cached:
header_value, expires_at = cached
buffer = MoveTVChannel._PLAY_AUTH_EARLY_EXPIRY_BUFFER
if expires_at and time.time() < (expires_at - buffer):
return header_value
# Cache miss or expired — fetch a fresh source and re-cache.
try:
channel = self._channel_by_id(content_id)
if channel is None:
logger.info(
f"move.tv: Channel not found for content_id={content_id!r}; "
"attempting to rebuild channel cache before play-auth fetch"
)
self.get_channels()
channel = self._channel_by_id(content_id)
if channel is None:
logger.error(
f"move.tv: Channel still not found after cache rebuild for "
f"content_id={content_id!r}; cannot resolve liveId for play-auth fetch"
)
return None
live_id = channel.live_id
source_data = self._fetch_live_source(live_id)
if source_data:
self._store_play_auth(content_id, source_data)
cached = self._play_auth_cache.get(content_id)
return cached[0] if cached else None
except requests.HTTPError as exc:
if exc.response is not None and exc.response.status_code == 401:
logger.info(
f"move.tv: 401 fetching play-auth for liveId={content_id} "
"— attempting token refresh …"
)
try:
token = self.authenticator.get_current_token()
refreshed = self.authenticator.refresh_token(token) if token else None
if not refreshed:
logger.info(
"move.tv: Refresh failed, falling back to full login "
"for play-auth fetch"
)
self.authenticator.authenticate(force_refresh=True)
# Retry once with the new token.
live_id = channel.live_id
source_data = self._fetch_live_source(live_id)
if source_data:
self._store_play_auth(content_id, source_data)
cached = self._play_auth_cache.get(content_id)
return cached[0] if cached else None
except Exception as retry_exc:
logger.error(
f"move.tv: Failed to fetch play auth header after re-auth: {retry_exc}"
)
else:
logger.error(f"move.tv: Failed to fetch play auth header: {exc}")
except Exception as exc:
logger.error(f"move.tv: Failed to fetch play auth header: {exc}")
return None
url, _ = self.get_manifest_with_headers(content_id, **kwargs)
return url
def get_manifest_headers(self, content_id: str, **kwargs) -> Dict[str, str]:
headers: Dict[str, str] = {
"User-Agent": MoveTVConfig.USER_AGENT,
"X-Play-Auth": self._get_play_auth_header(content_id),
}
logger.debug(f"move.tv: Manifest headers: {headers}")
_, headers = self.get_manifest_with_headers(content_id, **kwargs)
return headers
# ------------------------------------------------------------------
# get_drm (stub – AES-128/token auth, no active Widevine/PlayReady)
# ------------------------------------------------------------------
def get_drm(self, content_id: str, **kwargs) -> List[DRMConfig]:
"""
move.tv uses AES-128 token-based stream protection via X-Play-Auth.
No Widevine or PlayReady DRM is active in observed traffic.
"""
return []
# ------------------------------------------------------------------
# Internal helpers
# Catchup Manifest
# ------------------------------------------------------------------
def _fetch_live_source(self, live_id: int) -> Optional[Dict[str, Any]]:
"""
POST /api/v2/content/live/source/get and return the parsed JSON.
Requires an active authenticated session; the customer_id,
customer_profile_id and device_id are read from the auth token via
get_session_info().
"""
session = self.authenticator.get_session_info()
if not session:
# Force re-authentication and retry once
logger.info("move.tv: No session info; re-authenticating before manifest fetch")
self.authenticator.authenticate(force_refresh=True)
session = self.authenticator.get_session_info()
if not session:
logger.error("move.tv: Unable to obtain session info for manifest fetch")
return None
payload = {
"customerId": session["customer_id"],
"customerProfileId": session["customer_profile_id"],
"liveId": live_id,
"dtype": MoveTVConfig.DTYPE_DASH,
"appVersion": MoveTVConfig.APP_VERSION,
}
headers = MoveTVConfig.get_api_headers(auth_token=session["auth_token"])
logger.debug(f"move.tv: POST {MoveTVConfig.live_source_url()} liveId={live_id}")
def get_catchup_manifest_with_headers(
self, content_id: str, start_time: int, end_time: int, epg_id: Optional[str] = None, **kwargs
) -> Tuple[Optional[str], Dict[str, str]]:
"""Unified method to fetch both catchup URL and headers in one go."""
try:
response = self.http_manager.post(
MoveTVConfig.live_source_url(),
operation="manifest",
json=payload,
headers=headers,
)
# http_manager returned a response object — check status before raise_for_status().
is_401 = response.status_code == 401
except requests.HTTPError as exc:
# http_manager raised before returning (e.g. it calls raise_for_status internally).
# Extract the 401 from the exception's attached response instead.
if exc.response is not None and exc.response.status_code == 401:
is_401 = True
response = exc.response
else:
raise
# 1. Resolve EPG ID
resolved_epg_id = epg_id
if not resolved_epg_id:
# A fixed 6-hour backward window anchored at start_time covers long-running
# programs (like films or sports) without pulling massive EPG ranges.
epg_entries = self._epg.get_channel_epg(
channel_id=content_id,
backwards=6,
forwards=2,
start_time=datetime.fromtimestamp(start_time, tz=timezone.utc),
)
# On 401, attempt a token refresh and retry once before giving up.
if is_401:
logger.info(f"move.tv: 401 on manifest fetch for liveId={live_id} — attempting token refresh …")
token = self.authenticator.get_current_token()
refreshed = self.authenticator.refresh_token(token) if token else None
if not refreshed:
logger.info("move.tv: Refresh failed, falling back to full login for manifest fetch")
self.authenticator.authenticate(force_refresh=True)
# Rebuild session and headers with the new token.
session = self.authenticator.get_session_info()
if not session:
logger.error("move.tv: Unable to obtain session after token refresh for manifest fetch")
return None
payload["customerId"] = session["customer_id"]
payload["customerProfileId"] = session["customer_profile_id"]
headers = MoveTVConfig.get_api_headers(auth_token=session["auth_token"])
response = self.http_manager.post(
MoveTVConfig.live_source_url(),
operation="manifest",
json=payload,
headers=headers,
if not epg_entries:
return None, {"User-Agent": MoveTVConfig.USER_AGENT}
target_entry = None
closest_entry = None
closest_diff = float('inf')
for entry in epg_entries:
if entry.start <= start_time < entry.end:
target_entry = entry
break
diff = abs(entry.start - start_time)
if diff < closest_diff:
closest_diff = diff
closest_entry = entry
if target_entry:
resolved_epg_id = str(target_entry.program_id)
elif closest_entry and closest_diff < 3600:
resolved_epg_id = str(closest_entry.program_id)
else:
logger.warning(f"move.tv: No suitable EPG entry found for {content_id} at {start_time}")
return None, {"User-Agent": MoveTVConfig.USER_AGENT}
# 2. Fetch Source
cache_key = f"{content_id}:{resolved_epg_id}"
channel = self._get_channel_with_rebuild(content_id)
if not channel:
return None, {"User-Agent": MoveTVConfig.USER_AGENT}
try:
epg_id_int = int(resolved_epg_id)
content_id_int = int(content_id)
except ValueError:
logger.error(f"move.tv: Invalid ID format. epg_id={resolved_epg_id}, content_id={content_id}")
return None, {"User-Agent": MoveTVConfig.USER_AGENT}
payload = {"epgId": epg_id_int, "contentId": content_id_int}
source_data = self._fetch_source_endpoint(
MoveTVConfig.epg_source_url(),
payload=payload,
content_id=content_id,
cache_key=cache_key,
operation="catchup_manifest"
)
response.raise_for_status()
data = response.json()
if source_data:
content_url = source_data.get("content_url")
header_value = source_data.get("protection", {}).get("headerValue", "")
return content_url, {"User-Agent": MoveTVConfig.USER_AGENT, "X-Play-Auth": header_value}
if not data.get("success"):
logger.warning(
f"move.tv: live source endpoint returned success=false for liveId={live_id}"
)
return None
return None, {"User-Agent": MoveTVConfig.USER_AGENT}
return data
except Exception as exc:
logger.error(f"move.tv: Unexpected error in get_catchup_manifest_with_headers: {exc}")
return None, {"User-Agent": MoveTVConfig.USER_AGENT}
def _channel_by_id(self, content_id: str) -> Optional[MoveTVChannel]:
"""Return the cached MoveTVChannel whose content_id matches, or None."""
for ch in (self.channels or []):
if isinstance(ch, MoveTVChannel) and ch.content_id == content_id:
return ch
return None
def get_catchup_manifest(self, content_id: str, start_time: int, end_time: int, epg_id: Optional[str] = None,
**kwargs) -> Optional[str]:
url, _ = self.get_catchup_manifest_with_headers(content_id, start_time, end_time, epg_id, **kwargs)
return url
# Regex to extract the Unix expiry from the X-Play-Auth token value.
# The token format embeds it as "expires-<unix_ts>" e.g.:
# hash-…-expires-1774465909-cusid-…
_PLAY_AUTH_EXPIRY_RE = re.compile(r"expires-(\d+)")
def get_catchup_manifest_headers(self, content_id: str, start_time: int, end_time: int,
epg_id: Optional[str] = None, **kwargs) -> Dict[str, str]:
_, headers = self.get_catchup_manifest_with_headers(content_id, start_time, end_time, epg_id, **kwargs)
return headers
def _store_play_auth(self, content_id: str, source_data: Dict[str, Any]) -> None:
"""
Cache the X-Play-Auth header value and its expiry.
def get_catchup_drm(
self, content_id: str, start_time: int, end_time: int, epg_id: Optional[str] = None,
drm_variant: Optional[str] = None, **kwargs
) -> List[DRMConfig]:
return []
Written to both the standalone _play_auth_cache dict (always) and to
the matching MoveTVChannel object (when the channel list has been
loaded). Using the dict means the header is available even when
get_channels() has not yet been called.
"""
protection: Dict = source_data.get("protection", {})
header_value: str = protection.get("headerValue", "")
if not header_value:
return
# Parse the expiry timestamp from the token itself.
expires_at: float = 0.0
match = self._PLAY_AUTH_EXPIRY_RE.search(header_value)
if match:
expires_at = float(match.group(1))
else:
logger.warning(
f"move.tv: Could not parse expiry from X-Play-Auth for "
f"liveId={content_id}; token will not be reused."
)
# Always populate the standalone cache so _get_play_auth_header works
# regardless of whether self.channels has been populated.
self._play_auth_cache[content_id] = (header_value, expires_at)
# Also mirror onto the channel object when available.
def get_catchup_window_for_channel(self, content_id: str) -> int:
channel = self._channel_by_id(content_id)
if channel:
channel.play_auth_header = header_value
channel.play_auth_expires_at = expires_at
if channel and hasattr(channel, 'catchup_hours'):
return channel.catchup_hours
return self.catchup_window
logger.debug(
f"move.tv: Cached X-Play-Auth for liveId={content_id} "
f"expires_at={expires_at}: {header_value[:60]}…"
)
# ------------------------------------------------------------------ #
# VOD — catalogue #
# ------------------------------------------------------------------ #
# ------------------------------------------------------------------
# VOD & EPG Delegations
# ------------------------------------------------------------------
def get_vod_filters(self) -> VodFilters:
"""Return available VOD content types, categories, catalogs, tags, and sort options."""
return self._vod.get_vod_filters()
def get_vod_items(
self,
page: int = 1,
sort: str = "newest",
tag_id=None,
category_id=None,
catalog_id=None,
content_type_id=None,
search_query=None,
) -> VodPage:
"""Return a paginated VOD catalogue with optional filtering."""
return self._vod.get_vod_items(
page=page,
sort=sort,
tag_id=tag_id,
category_id=category_id,
catalog_id=catalog_id,
content_type_id=content_type_id,
search_query=search_query,
)
def get_vod_items(self, page: int = 1, sort: str = "newest", tag_id=None, category_id=None, catalog_id=None,
content_type_id=None, search_query=None) -> VodPage:
return self._vod.get_vod_items(page=page, sort=sort, tag_id=tag_id, category_id=category_id,
catalog_id=catalog_id, content_type_id=content_type_id,
search_query=search_query)
def get_all_vod_items(
self,
sort: str = "newest",
tag_id=None,
category_id=None,
catalog_id=None,
content_type_id=None,
max_pages=None,
):
"""Fetch every VOD page and return a flat list of VodItems."""
return self._vod.get_all_vod_items(
sort=sort,
tag_id=tag_id,
category_id=category_id,
catalog_id=catalog_id,
content_type_id=content_type_id,
max_pages=max_pages,
)
# ------------------------------------------------------------------ #
# VOD — homepage / page layout #
# ------------------------------------------------------------------ #
def get_all_vod_items(self, sort: str = "newest", tag_id=None, category_id=None, catalog_id=None,
content_type_id=None, max_pages=None):
return self._vod.get_all_vod_items(sort=sort, tag_id=tag_id, category_id=category_id, catalog_id=catalog_id,
content_type_id=content_type_id, max_pages=max_pages)
def get_page_components(self, page_id: int):
"""
Return the ordered list of component descriptors for a UI page.
Pass the returned component IDs to get_component_items() to load
the actual content cards.
"""
return self._vod.get_page_components(page_id)
def get_component_items(self, component_id: int):
"""
Return the content cards for a single carousel / banner component.
Richer than get_vod_items() cards: includes description, age rating,
release year, duration, and per-item subscription status.
"""
return self._vod.get_component_items(component_id)
def get_epg(
self,
channel_id: str,
backwards: int = 2,
forwards: int = 2,
**kwargs,
) -> List[EPGEntry]:
"""
Return the EPG schedule for *channel_id*.
Parameters
----------
channel_id:
The channel's contentId (stored as ``MoveTVChannel.content_id``).
Passed directly to the EPG API with no mapping required.
backwards:
Hours of past programming to include (default 2).
forwards:
Hours of future programming to include (default 2).
Returns
-------
List of EPGEntry objects from ``MoveTvEpgManager`` (see
epg_manager.py for the field mapping and known lossy fields).
"""
# content_id IS the contentId the EPG endpoint expects — no mapping needed.
def get_epg(self, channel_id: str, backwards: int = 2, forwards: int = 2, **kwargs) -> List[EPGEntry]:
return self._epg.get_channel_epg(
channel_id,
backwards=kwargs.get("backwards", backwards),