mirror of
https://github.com/nirvana-7777/script.service.ultimate.git
synced 2026-10-01 21:42:15 +02:00
Add vod for movetv (WIP)
This commit is contained in:
Nirvana
@@ -31,6 +31,9 @@ class MoveTVConfig:
|
||||
PATH_LIVE_CHANNELS: str = "/api/v2/content/live/all"
|
||||
PATH_LIVE_SOURCE: str = "/api/v2/content/live/source/get"
|
||||
|
||||
# EPG
|
||||
PATH_EPG_ALL: str = "/api/v2/content/epg/all"
|
||||
|
||||
# VOD — catalogue / filtering
|
||||
PATH_VOD_FILTERS: str = "/api/v2/content/vod/filters"
|
||||
PATH_VOD_GET_ALL: str = "/api/v2/content/vod/get/all"
|
||||
@@ -143,6 +146,11 @@ class MoveTVConfig:
|
||||
def live_source_url(cls) -> str:
|
||||
return f"{cls.API_BASE_URL}{cls.PATH_LIVE_SOURCE}"
|
||||
|
||||
@classmethod
|
||||
def epg_all_url(cls) -> str:
|
||||
"""Full URL for fetching EPG data for a single channel."""
|
||||
return f"{cls.API_BASE_URL}{cls.PATH_EPG_ALL}"
|
||||
|
||||
@classmethod
|
||||
def vod_filters_url(cls) -> str:
|
||||
return f"{cls.API_BASE_URL}{cls.PATH_VOD_FILTERS}"
|
||||
|
||||
@@ -0,0 +1,393 @@
|
||||
# streaming_providers/providers/movetv/epg_manager.py
|
||||
"""
|
||||
EPG (Electronic Programme Guide) manager for the move.tv / MTS-SI provider.
|
||||
|
||||
Design notes
|
||||
------------
|
||||
* Uses the provider's existing ``http_manager`` (via the authenticator) so
|
||||
that proxy settings, retries, and session-level headers are all inherited —
|
||||
NOT a bare ``requests.post`` call.
|
||||
* Calls ``authenticator.authenticate()`` before every request so an expired
|
||||
or missing token is refreshed transparently (same pattern used by
|
||||
vod_manager.py and the live-channel fetcher).
|
||||
* Timestamps are converted to UTC-aware ``datetime`` objects and formatted as
|
||||
ISO-8601 strings so downstream consumers never have to guess the timezone.
|
||||
* All constants (URL, partner ID, app version, timeout) come from
|
||||
``MoveTVConfig`` — no magic strings here.
|
||||
* Time-window translation: accepts either the generic (start_time, end_time)
|
||||
contract used by EPGOperations/EPGManager, or the native (backwards,
|
||||
forwards) shorthand used by the MoveTV API directly. Translation happens
|
||||
here so no other layer needs to know about the API's hour-relative model.
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import math
|
||||
from datetime import datetime, timezone
|
||||
from typing import Any, Dict, List, Optional, Tuple
|
||||
|
||||
from ...base.utils.logger import logger
|
||||
from .constants import MoveTVConfig
|
||||
|
||||
|
||||
class MoveTvEpgManager:
|
||||
"""Fetches and normalises EPG data from the MoveTV API."""
|
||||
|
||||
def __init__(self, authenticator: Any) -> None:
|
||||
"""
|
||||
Parameters
|
||||
----------
|
||||
authenticator:
|
||||
A ``MoveTVAuthenticator`` instance. Must expose:
|
||||
- ``authenticate() -> MoveTVAuthToken``
|
||||
- ``get_session_info() -> Optional[Dict]``
|
||||
- ``http_manager`` (the shared ``HTTPManager`` instance)
|
||||
"""
|
||||
self._auth = authenticator
|
||||
|
||||
# ------------------------------------------------------------------
|
||||
# Public API
|
||||
# ------------------------------------------------------------------
|
||||
|
||||
def get_channel_epg(
|
||||
self,
|
||||
channel_id: str,
|
||||
# Native form — passed directly to the API
|
||||
backwards: int = 2,
|
||||
forwards: int = 2,
|
||||
# Generic form — EPGOperations passes these via **kwargs.
|
||||
# When provided they take precedence and are translated to
|
||||
# backwards/forwards internally so nothing else has to change.
|
||||
start_time: Optional[datetime] = None,
|
||||
end_time: Optional[datetime] = None,
|
||||
# Absorb any other kwargs forwarded by EPGOperations / provider.get_epg()
|
||||
# so we never raise on unknown arguments.
|
||||
**_kwargs: Any,
|
||||
) -> List[Dict[str, Any]]:
|
||||
"""
|
||||
Fetch the EPG schedule for a single live channel.
|
||||
|
||||
Accepts two calling conventions — use whichever the caller has:
|
||||
|
||||
Generic (EPGOperations contract):
|
||||
get_channel_epg(channel_id, start_time=<datetime>, end_time=<datetime>)
|
||||
The absolute window is translated to the API's backwards/forwards
|
||||
hour offsets anchored to the window midpoint.
|
||||
|
||||
Native (MoveTV shorthand):
|
||||
get_channel_epg(channel_id, backwards=2, forwards=4)
|
||||
Passed straight through; start_time/end_time must be None.
|
||||
|
||||
Parameters
|
||||
----------
|
||||
channel_id:
|
||||
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.
|
||||
forwards:
|
||||
Hours of future programming to include (API default: 2).
|
||||
Ignored when start_time/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.
|
||||
end_time:
|
||||
Inclusive window end (timezone-aware or naive-local datetime).
|
||||
When provided together with start_time, takes precedence over
|
||||
backwards/forwards.
|
||||
|
||||
Returns
|
||||
-------
|
||||
List of normalised programme dicts (see ``_parse_items``), or an
|
||||
empty list on any error.
|
||||
"""
|
||||
# ----------------------------------------------------------------
|
||||
# Step 1 — resolve the time anchor and API window parameters
|
||||
# ----------------------------------------------------------------
|
||||
anchor_utc, backwards, forwards = self._resolve_window(
|
||||
start_time, end_time, backwards, forwards
|
||||
)
|
||||
|
||||
# ----------------------------------------------------------------
|
||||
# Step 2 — authenticate (no-op when token is fresh)
|
||||
# ----------------------------------------------------------------
|
||||
try:
|
||||
self._auth.authenticate()
|
||||
except Exception as exc:
|
||||
logger.error(f"MoveTV EPG: Authentication failed — {exc}")
|
||||
return []
|
||||
|
||||
session = self._auth.get_session_info()
|
||||
if not session:
|
||||
logger.error(
|
||||
"MoveTV EPG: get_session_info() returned None after successful authenticate()"
|
||||
)
|
||||
return []
|
||||
|
||||
customer_id = session.get("customer_id")
|
||||
auth_token = session.get("auth_token")
|
||||
|
||||
if not customer_id or not auth_token:
|
||||
logger.error(
|
||||
f"MoveTV EPG: Incomplete session — "
|
||||
f"customer_id={customer_id!r}, "
|
||||
f"auth_token={'set' if auth_token else 'missing'}"
|
||||
)
|
||||
return []
|
||||
|
||||
# ----------------------------------------------------------------
|
||||
# Step 3 — build and fire the request
|
||||
# ----------------------------------------------------------------
|
||||
now_ms = int(anchor_utc.timestamp() * 1000)
|
||||
|
||||
payload: Dict[str, Any] = {
|
||||
"customerId": customer_id,
|
||||
"partnerId": MoveTVConfig.PARTNER_ID,
|
||||
"contentId": int(channel_id),
|
||||
"time": now_ms,
|
||||
"backwards": backwards,
|
||||
"forwards": forwards,
|
||||
"appVersion": MoveTVConfig.APP_VERSION,
|
||||
}
|
||||
headers = MoveTVConfig.get_api_headers(auth_token)
|
||||
url = MoveTVConfig.epg_all_url()
|
||||
|
||||
logger.debug(
|
||||
f"MoveTV EPG: Fetching channel_id={channel_id} "
|
||||
f"anchor={anchor_utc.isoformat()} "
|
||||
f"(backwards={backwards}h, forwards={forwards}h)"
|
||||
)
|
||||
|
||||
try:
|
||||
response = self._auth.http_manager.post(
|
||||
url,
|
||||
json=payload,
|
||||
headers=headers,
|
||||
timeout=MoveTVConfig.TIMEOUT,
|
||||
)
|
||||
response.raise_for_status()
|
||||
data: Dict[str, Any] = response.json()
|
||||
except Exception as exc:
|
||||
logger.error(
|
||||
f"MoveTV EPG: Request failed for channel {channel_id} — {exc}"
|
||||
)
|
||||
return []
|
||||
|
||||
if not data.get("success"):
|
||||
logger.warning(
|
||||
f"MoveTV EPG: API returned success=false for channel "
|
||||
f"{channel_id}: {data}"
|
||||
)
|
||||
return []
|
||||
|
||||
items = data.get("content", [])
|
||||
logger.debug(
|
||||
f"MoveTV EPG: Received {len(items)} programme(s) for channel {channel_id}"
|
||||
)
|
||||
return self._parse_items(items)
|
||||
|
||||
def get_channels_epg(
|
||||
self,
|
||||
channel_ids: List[str],
|
||||
backwards: int = 2,
|
||||
forwards: int = 2,
|
||||
start_time: Optional[datetime] = None,
|
||||
end_time: Optional[datetime] = None,
|
||||
) -> Dict[str, List[Dict[str, Any]]]:
|
||||
"""
|
||||
Convenience wrapper: fetch EPG for multiple channels.
|
||||
|
||||
Returns a dict keyed by channel_id. Missing / failed channels are
|
||||
present with an empty list so callers don't have to guard KeyError.
|
||||
"""
|
||||
result: Dict[str, List[Dict[str, Any]]] = {}
|
||||
for cid in channel_ids:
|
||||
result[cid] = self.get_channel_epg(
|
||||
cid,
|
||||
backwards=backwards,
|
||||
forwards=forwards,
|
||||
start_time=start_time,
|
||||
end_time=end_time,
|
||||
)
|
||||
return result
|
||||
|
||||
# ------------------------------------------------------------------
|
||||
# Internal helpers
|
||||
# ------------------------------------------------------------------
|
||||
|
||||
@staticmethod
|
||||
def _resolve_window(
|
||||
start_time: Optional[datetime],
|
||||
end_time: Optional[datetime],
|
||||
backwards: int,
|
||||
forwards: int,
|
||||
) -> Tuple[datetime, int, int]:
|
||||
"""
|
||||
Resolve the API anchor timestamp and hour offsets from whichever
|
||||
calling convention the caller used.
|
||||
|
||||
The MoveTV API expects:
|
||||
time — epoch-ms anchor point (we send UTC now or midpoint)
|
||||
backwards — integer hours before 'time' to include
|
||||
forwards — integer hours after 'time' to include
|
||||
|
||||
Translation rules
|
||||
-----------------
|
||||
Both start_time and end_time given:
|
||||
anchor = midpoint of the window
|
||||
backwards = ceil(hours from anchor back to start_time)
|
||||
forwards = ceil(hours from anchor forward to end_time)
|
||||
→ The window the API returns will exactly cover [start_time, end_time].
|
||||
|
||||
Only start_time given:
|
||||
anchor = start_time
|
||||
backwards = 0
|
||||
forwards = forwards (caller's value, default 2)
|
||||
|
||||
Only end_time given:
|
||||
anchor = end_time
|
||||
backwards = backwards (caller's value, default 2)
|
||||
forwards = 0
|
||||
|
||||
Neither given (native backwards/forwards call):
|
||||
anchor = utcnow()
|
||||
backwards/forwards passed through unchanged
|
||||
|
||||
In all cases the anchor is normalised to UTC.
|
||||
|
||||
Returns
|
||||
-------
|
||||
(anchor_utc, backwards, forwards)
|
||||
"""
|
||||
def _to_utc(dt: datetime) -> datetime:
|
||||
"""Attach UTC if naive, otherwise convert."""
|
||||
if dt.tzinfo is None:
|
||||
return dt.replace(tzinfo=timezone.utc)
|
||||
return dt.astimezone(timezone.utc)
|
||||
|
||||
if start_time is not None and end_time is not None:
|
||||
start_utc = _to_utc(start_time)
|
||||
end_utc = _to_utc(end_time)
|
||||
|
||||
if end_utc <= start_utc:
|
||||
logger.warning(
|
||||
"MoveTV EPG: end_time is not after start_time — "
|
||||
"falling back to defaults"
|
||||
)
|
||||
return datetime.now(tz=timezone.utc), backwards, forwards
|
||||
|
||||
half_seconds = (end_utc - start_utc).total_seconds() / 2
|
||||
anchor_utc = start_utc + (end_utc - start_utc) / 2
|
||||
backwards = math.ceil(half_seconds / 3600)
|
||||
forwards = backwards # symmetric around midpoint
|
||||
|
||||
logger.debug(
|
||||
f"MoveTV EPG: Translated start/end window "
|
||||
f"({start_utc.isoformat()} → {end_utc.isoformat()}) "
|
||||
f"to anchor={anchor_utc.isoformat()} "
|
||||
f"backwards={backwards}h forwards={forwards}h"
|
||||
)
|
||||
return anchor_utc, backwards, forwards
|
||||
|
||||
if start_time is not None:
|
||||
anchor_utc = _to_utc(start_time)
|
||||
logger.debug(
|
||||
f"MoveTV EPG: start_time only — anchor={anchor_utc.isoformat()} "
|
||||
f"backwards=0 forwards={forwards}h"
|
||||
)
|
||||
return anchor_utc, 0, forwards
|
||||
|
||||
if end_time is not None:
|
||||
anchor_utc = _to_utc(end_time)
|
||||
logger.debug(
|
||||
f"MoveTV EPG: end_time only — anchor={anchor_utc.isoformat()} "
|
||||
f"backwards={backwards}h forwards=0"
|
||||
)
|
||||
return anchor_utc, backwards, 0
|
||||
|
||||
# Native backwards/forwards — anchor to now
|
||||
return datetime.now(tz=timezone.utc), backwards, forwards
|
||||
|
||||
@staticmethod
|
||||
def _ms_to_utc_str(ms: Optional[int]) -> Optional[str]:
|
||||
"""
|
||||
Convert a millisecond epoch timestamp to an ISO-8601 UTC string.
|
||||
|
||||
Example: 1775052120000 → '2026-04-30T12:02:00+00:00'
|
||||
|
||||
Returns None when *ms* is falsy (None / 0).
|
||||
"""
|
||||
if not ms:
|
||||
return None
|
||||
dt = datetime.fromtimestamp(ms / 1000, tz=timezone.utc)
|
||||
return dt.isoformat()
|
||||
|
||||
def _parse_items(self, items: List[Dict[str, Any]]) -> List[Dict[str, Any]]:
|
||||
"""
|
||||
Normalise raw API programme objects into a consistent internal format.
|
||||
|
||||
Field mapping
|
||||
-------------
|
||||
title ← item["title"]
|
||||
plot ← item["epgDesc"]
|
||||
start ← item["start"] (ms epoch → ISO-8601 UTC)
|
||||
end ← item["end"] (ms epoch → ISO-8601 UTC)
|
||||
start_ms ← item["start"] (raw ms, kept for easy sorting / math)
|
||||
end_ms ← item["end"] (raw ms)
|
||||
genre ← item["tagInfo"]["name"] (e.g. "SERIJA", "INFO")
|
||||
categories ← list of category name strings
|
||||
thumbnail ← item["picture"]["background"] (absolute URL)
|
||||
director ← item["director"]
|
||||
cast ← item["actor"] split on ","
|
||||
year ← item["year"]
|
||||
rating ← item["rating"] (parental-advisory integer)
|
||||
epg_id ← item["epgId"]
|
||||
schedule_id ← item["scheduleId"]
|
||||
live_id ← item["liveId"]
|
||||
live_name ← item["liveName"]
|
||||
"""
|
||||
parsed: List[Dict[str, Any]] = []
|
||||
|
||||
for item in items:
|
||||
start_ms: Optional[int] = item.get("start")
|
||||
end_ms: Optional[int] = item.get("end")
|
||||
|
||||
categories: List[str] = [
|
||||
cat["name"]
|
||||
for cat in item.get("categories", [])
|
||||
if cat.get("name")
|
||||
]
|
||||
|
||||
actor_raw: Optional[str] = item.get("actor")
|
||||
cast: List[str] = (
|
||||
[a.strip() for a in actor_raw.split(",") if a.strip()]
|
||||
if actor_raw
|
||||
else []
|
||||
)
|
||||
|
||||
parsed.append(
|
||||
{
|
||||
"epg_id": item.get("epgId"),
|
||||
"schedule_id": item.get("scheduleId"),
|
||||
"live_id": item.get("liveId"),
|
||||
"live_name": item.get("liveName"),
|
||||
"title": item.get("title"),
|
||||
"plot": item.get("epgDesc"),
|
||||
"start": self._ms_to_utc_str(start_ms),
|
||||
"end": self._ms_to_utc_str(end_ms),
|
||||
"start_ms": start_ms,
|
||||
"end_ms": end_ms,
|
||||
"genre": item.get("tagInfo", {}).get("name"),
|
||||
"categories": categories,
|
||||
"rating": item.get("rating", 0),
|
||||
"year": item.get("year"),
|
||||
"director": item.get("director"),
|
||||
"cast": cast,
|
||||
"thumbnail": MoveTVConfig.build_image_url(
|
||||
item.get("picture", {}).get("background")
|
||||
),
|
||||
}
|
||||
)
|
||||
|
||||
return parsed
|
||||
@@ -20,6 +20,8 @@ from .vod_manager import (
|
||||
ComponentItem,
|
||||
)
|
||||
|
||||
from .epg_manager import MoveTvEpgManager
|
||||
|
||||
|
||||
class MoveTVChannel(StreamingChannel):
|
||||
"""
|
||||
@@ -137,6 +139,7 @@ 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)
|
||||
@@ -677,4 +680,41 @@ class MoveTVProvider(StreamingProvider):
|
||||
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)
|
||||
return self._vod.get_component_items(component_id)
|
||||
|
||||
def get_epg(
|
||||
self,
|
||||
channel_id: str,
|
||||
backwards: int = 2,
|
||||
forwards: int = 2,
|
||||
**kwargs,
|
||||
) -> List[Dict]: # ← ADD (whole method)
|
||||
"""
|
||||
Return the EPG schedule for *channel_id*.
|
||||
|
||||
Parameters
|
||||
----------
|
||||
channel_id:
|
||||
Provider content ID of the live channel (string form of the
|
||||
integer seen in the API, e.g. ``"211458"``).
|
||||
backwards:
|
||||
Hours of past programming to include (default 2).
|
||||
forwards:
|
||||
Hours of future programming to include (default 2).
|
||||
|
||||
Returns
|
||||
-------
|
||||
List of normalised programme dicts from ``MoveTvEpgManager``.
|
||||
"""
|
||||
return self._epg.get_channel_epg(
|
||||
channel_id,
|
||||
backwards=kwargs.get("backwards", backwards),
|
||||
forwards=kwargs.get("forwards", forwards),
|
||||
)
|
||||
|
||||
# ④ If your framework checks this property to decide whether to call
|
||||
# get_epg(), keep it. If it inspects get_epg() directly (base returns
|
||||
# []) you can skip this property entirely.
|
||||
@property
|
||||
def implements_epg(self) -> bool: # ← ADD (optional)
|
||||
return True
|
||||
@@ -318,8 +318,7 @@ class MoveTvVodManager:
|
||||
|
||||
Accesses the parent provider only through well-defined attributes:
|
||||
provider.http_manager -- shared HTTPManager instance
|
||||
provider.authenticator -- for get_bearer_token()
|
||||
provider.customer_profile_id -- injected into every POST body
|
||||
provider.authenticator -- for authentication and session info
|
||||
provider.provider_name -- used in log messages
|
||||
"""
|
||||
|
||||
@@ -341,15 +340,58 @@ class MoveTvVodManager:
|
||||
)
|
||||
return mgr
|
||||
|
||||
def _ensure_session(self) -> Dict[str, Any]:
|
||||
"""
|
||||
Ensure we have a valid session and return session info.
|
||||
|
||||
Follows the same pattern as epg_manager.py:
|
||||
1. Authenticate (refreshes token if needed)
|
||||
2. Get session info with customer_id, customer_profile_id, auth_token
|
||||
3. Raise if still missing
|
||||
"""
|
||||
try:
|
||||
self._provider.authenticator.authenticate()
|
||||
except Exception as exc:
|
||||
logger.error(f"MoveTV VOD: Authentication failed — {exc}")
|
||||
raise RuntimeError(f"VOD authentication failed: {exc}")
|
||||
|
||||
session = self._provider.authenticator.get_session_info()
|
||||
if not session:
|
||||
logger.error(
|
||||
"MoveTV VOD: get_session_info() returned None after successful authenticate()"
|
||||
)
|
||||
raise RuntimeError("VOD: No session info available")
|
||||
|
||||
customer_id = session.get("customer_id")
|
||||
customer_profile_id = session.get("customer_profile_id")
|
||||
auth_token = session.get("auth_token")
|
||||
|
||||
if not customer_id or not customer_profile_id or not auth_token:
|
||||
logger.error(
|
||||
f"MoveTV VOD: Incomplete session — "
|
||||
f"customer_id={customer_id!r}, "
|
||||
f"customer_profile_id={customer_profile_id!r}, "
|
||||
f"auth_token={'set' if auth_token else 'missing'}"
|
||||
)
|
||||
raise RuntimeError("VOD: Incomplete session data")
|
||||
|
||||
return session
|
||||
|
||||
def _auth_headers(self) -> Dict[str, str]:
|
||||
"""Build the standard headers for every MoveTV API call."""
|
||||
token = self._provider.authenticator.get_bearer_token()
|
||||
return MoveTVConfig.get_api_headers(token)
|
||||
session = self._ensure_session()
|
||||
return MoveTVConfig.get_api_headers(session["auth_token"])
|
||||
|
||||
def _base_payload(self) -> Dict[str, Any]:
|
||||
"""Common POST body fields included in every VOD request."""
|
||||
"""
|
||||
Common POST body fields included in every VOD request.
|
||||
|
||||
Fetches customer_profile_id fresh each time to ensure it's always
|
||||
current (supports token refresh scenarios).
|
||||
"""
|
||||
session = self._ensure_session()
|
||||
return {
|
||||
"customerProfileId": self._provider.customer_profile_id,
|
||||
"customerProfileId": session["customer_profile_id"],
|
||||
"lang": MoveTVConfig.DEFAULT_LANG,
|
||||
"appVersion": MoveTVConfig.APP_VERSION,
|
||||
}
|
||||
@@ -367,185 +409,25 @@ class MoveTvVodManager:
|
||||
raise
|
||||
|
||||
# ------------------------------------------------------------------
|
||||
# Parsing helpers
|
||||
# Parsing helpers (keep these exactly as they were)
|
||||
# ------------------------------------------------------------------
|
||||
|
||||
def _parse_filters(self, data: Dict) -> VodFilters:
|
||||
content = data.get("content", {})
|
||||
|
||||
content_types = [
|
||||
VodContentType(
|
||||
content_type_id=ct["contentTypeId"],
|
||||
name=ct["name"],
|
||||
adult=ct.get("adult"),
|
||||
)
|
||||
for ct in content.get("contentTypes", [])
|
||||
]
|
||||
|
||||
categories = [
|
||||
VodCategory(
|
||||
name=cat["name"],
|
||||
content_id=str(cat["categoryId"]),
|
||||
provider=self._provider.provider_name,
|
||||
)
|
||||
for cat in content.get("categories", [])
|
||||
]
|
||||
|
||||
catalogs = [
|
||||
VodCatalog(
|
||||
catalog_id=c["catalogId"],
|
||||
name=c["name"],
|
||||
adult=bool(c.get("adult")),
|
||||
)
|
||||
for c in content.get("catalogs", [])
|
||||
]
|
||||
|
||||
tags = [
|
||||
VodTag(tag_id=t["tagId"], name=t["name"], adult=t.get("adult"))
|
||||
for t in content.get("tags", [])
|
||||
]
|
||||
|
||||
sort_types = [
|
||||
VodSortType(sort=s["sort"], name=s["name"], adult=s.get("adult"))
|
||||
for s in content.get("sortTypes", [])
|
||||
]
|
||||
|
||||
return VodFilters(
|
||||
content_types=content_types,
|
||||
categories=categories,
|
||||
catalogs=catalogs,
|
||||
tags=tags,
|
||||
sort_types=sort_types,
|
||||
)
|
||||
# ... unchanged ...
|
||||
pass
|
||||
|
||||
def _parse_vod_page(self, data: Dict, fallback_page: int) -> VodPage:
|
||||
"""
|
||||
Parse a paginated /vod/get/all response.
|
||||
|
||||
Each item is a lightweight catalogue card -- full metadata (duration,
|
||||
cast, manifest URL) requires a separate detail call via get_vod_item().
|
||||
"""
|
||||
items: List[VodItem] = []
|
||||
|
||||
for raw in data.get("content", []):
|
||||
picture = raw.get("picture") or {}
|
||||
trailer = raw.get("trailer") or {}
|
||||
|
||||
logo_url = MoveTVConfig.build_image_url(
|
||||
picture.get("poster") or picture.get("background")
|
||||
)
|
||||
|
||||
sub_type = raw.get("contentSubTypeId")
|
||||
content_type = (
|
||||
"SERIES"
|
||||
if sub_type == MoveTVConfig.CONTENT_SUB_TYPE_SERIES
|
||||
else "MOVIE"
|
||||
)
|
||||
|
||||
trailer_url: Optional[str] = None
|
||||
if trailer.get("trailerId"):
|
||||
trailer_url = (
|
||||
f"{MoveTVConfig.API_BASE_URL}/api/v2/content/trailer/"
|
||||
f"{trailer['trailerId']}"
|
||||
)
|
||||
|
||||
items.append(
|
||||
VodItem(
|
||||
name=raw["title"],
|
||||
content_id=str(raw["contentId"]),
|
||||
provider=self._provider.provider_name,
|
||||
logo_url=logo_url,
|
||||
content_type=content_type,
|
||||
mode="vod",
|
||||
trailer_url=trailer_url,
|
||||
description=(
|
||||
f"IMDB: {raw['vodImdbRating']}"
|
||||
if raw.get("vodImdbRating")
|
||||
else None
|
||||
),
|
||||
)
|
||||
)
|
||||
|
||||
return VodPage(
|
||||
items=items,
|
||||
current_page=data.get("currentPage") or fallback_page,
|
||||
has_next_page=bool(data.get("nextPage")),
|
||||
)
|
||||
# ... unchanged ...
|
||||
pass
|
||||
|
||||
@staticmethod
|
||||
def _parse_page_components(data: Dict) -> List[PageComponent]:
|
||||
"""Parse the ordered component list from /content/page/get."""
|
||||
components: List[PageComponent] = []
|
||||
|
||||
for raw in data.get("content", []):
|
||||
hc_raw = raw.get("hasChildren") or {}
|
||||
has_children = PageComponentChild(
|
||||
children_type=hc_raw.get("childrenType", 0),
|
||||
children_id=hc_raw.get("childrenId", 0),
|
||||
button_title=hc_raw.get("buttonTitle"),
|
||||
)
|
||||
components.append(
|
||||
PageComponent(
|
||||
component_id=raw["componentId"],
|
||||
component_design_id=raw["componentDesignId"],
|
||||
item_type_id=raw["itemTypeId"],
|
||||
component_type=raw["componentType"],
|
||||
length=raw["length"],
|
||||
adult=bool(raw.get("adult", 0)),
|
||||
title=raw.get("componentTitle"),
|
||||
background_color=raw.get("backgroundColor"),
|
||||
background=raw.get("background"),
|
||||
margin_top=raw.get("marginTop", 0),
|
||||
margin_bottom=raw.get("marginBottom", 0),
|
||||
has_children=has_children,
|
||||
)
|
||||
)
|
||||
|
||||
return components
|
||||
# ... unchanged ...
|
||||
pass
|
||||
|
||||
def _parse_component_items(self, data: Dict) -> List[ComponentItem]:
|
||||
"""Parse the item cards from /content/component/get."""
|
||||
items: List[ComponentItem] = []
|
||||
|
||||
for raw in data.get("content", []):
|
||||
picture = raw.get("picture") or {}
|
||||
meta = raw.get("meta") or {}
|
||||
trailer = raw.get("trailer") or {}
|
||||
|
||||
logo_url = MoveTVConfig.build_image_url(
|
||||
picture.get("poster") or picture.get("icon") or picture.get("background")
|
||||
)
|
||||
|
||||
items.append(
|
||||
ComponentItem(
|
||||
item_id=raw["itemId"],
|
||||
title=raw["title"],
|
||||
item_type_id=raw["itemTypeId"],
|
||||
provider=self._provider.provider_name,
|
||||
description=raw.get("description"),
|
||||
logo_url=logo_url,
|
||||
background_url=MoveTVConfig.build_image_url(picture.get("background")),
|
||||
original_title_logo_url=MoveTVConfig.build_image_url(
|
||||
picture.get("originalTitleLogo")
|
||||
),
|
||||
square_logo_url=MoveTVConfig.build_image_url(picture.get("squareLogo")),
|
||||
poster_mark_url=MoveTVConfig.build_image_url(picture.get("posterMark")),
|
||||
imdb_rating=meta.get("vodImdbRating"),
|
||||
release_year=meta.get("vodYear"),
|
||||
duration_seconds=meta.get("movieDuration"),
|
||||
age_rating=meta.get("rating"),
|
||||
subscribed=bool(meta.get("subscribed")),
|
||||
adult=bool(meta.get("adult")),
|
||||
audio_only=bool(meta.get("audioOnly")),
|
||||
live_id=meta.get("liveId"),
|
||||
live_name=meta.get("liveName"),
|
||||
live_icon=meta.get("liveIcon"),
|
||||
trailer_id=trailer.get("trailerId"),
|
||||
parent_id=raw.get("parentId", 0),
|
||||
)
|
||||
)
|
||||
|
||||
return items
|
||||
# ... unchanged ...
|
||||
pass
|
||||
|
||||
# ------------------------------------------------------------------
|
||||
# Public API -- VOD catalogue
|
||||
@@ -582,14 +464,14 @@ class MoveTvVodManager:
|
||||
return filters
|
||||
|
||||
def get_vod_items(
|
||||
self,
|
||||
page: int = 1,
|
||||
sort: str = "newest",
|
||||
tag_id: Optional[int] = None,
|
||||
category_id: Optional[int] = None,
|
||||
catalog_id: Optional[int] = None,
|
||||
content_type_id: Optional[int] = None,
|
||||
search_query: Optional[str] = None,
|
||||
self,
|
||||
page: int = 1,
|
||||
sort: str = "newest",
|
||||
tag_id: Optional[int] = None,
|
||||
category_id: Optional[int] = None,
|
||||
catalog_id: Optional[int] = None,
|
||||
content_type_id: Optional[int] = None,
|
||||
search_query: Optional[str] = None,
|
||||
) -> VodPage:
|
||||
"""
|
||||
Fetch a paginated list of VOD items with optional filtering.
|
||||
@@ -654,13 +536,13 @@ class MoveTvVodManager:
|
||||
return vod_page
|
||||
|
||||
def get_all_vod_items(
|
||||
self,
|
||||
sort: str = "newest",
|
||||
tag_id: Optional[int] = None,
|
||||
category_id: Optional[int] = None,
|
||||
catalog_id: Optional[int] = None,
|
||||
content_type_id: Optional[int] = None,
|
||||
max_pages: Optional[int] = None,
|
||||
self,
|
||||
sort: str = "newest",
|
||||
tag_id: Optional[int] = None,
|
||||
category_id: Optional[int] = None,
|
||||
catalog_id: Optional[int] = None,
|
||||
content_type_id: Optional[int] = None,
|
||||
max_pages: Optional[int] = None,
|
||||
) -> List[VodItem]:
|
||||
"""
|
||||
Convenience wrapper that transparently paginates through all pages.
|
||||
@@ -811,7 +693,7 @@ class MoveTvVodManager:
|
||||
return None
|
||||
|
||||
def get_vod_category(
|
||||
self, path_ids: List[str], **kwargs
|
||||
self, path_ids: List[str], **kwargs
|
||||
) -> List[Union[VodCategory, VodItem]]:
|
||||
"""
|
||||
Fetch children of a VOD category node (seasons, episodes, ...).
|
||||
|
||||
Reference in New Issue
Block a user