nirvana-7777/script.service.ultimate
1
0
Fork 0
mirror of https://github.com/nirvana-7777/script.service.ultimate.git synced 2026-10-05 23:42:57 +02:00
Code Issues Packages Projects Releases Wiki Activity

magentaeu: migrate to new structure

Browse Source
This commit is contained in:
Nirvana
2026-10-03 15:43:24 +02:00
parent ecf4cd7e39
commit f946539652
4 changed files with 275 additions and 64 deletions
Download Patch File Download Diff File Expand all files Collapse all files
lib/streaming_providers/providers/magentaeu/epg_manager.py
+83 -13
View File
@@ -23,6 +23,21 @@ Design notes
cached in-memory via ``_ProgramDetailsCache`` to avoid hammering the API.
* Credit labels are localised per country because the bifrost API returns
role names in the content language, sometimes ALL-CAPS (HR, ME).
Migration note
--------------
This manager now subclasses ``base.managers.EpgManager``. Two method names
changed to match the ABC:
get_channel_epg -> get_epg
get_channel_epg_batch -> get_epg_grid
Everything else is unchanged. The provider's own ``get_epg`` /
``get_epg_grid`` / ``get_program_details`` still work; they now call the
renamed methods.
The provider's ``epg_window`` property delegates to ``self.epg_window``
on this class, which was previously read from the provider.
"""
from __future__ import annotations
@@ -32,6 +47,7 @@ from datetime import datetime, timedelta, timezone
from zoneinfo import ZoneInfo
from typing import Any, Dict, List, Optional, Set, Tuple
from ...base.managers import EpgManager
from ...base.utils.logger import logger
from ...base.models.epg_models import EPGEntry, EPGProgramDetails, EPGFlags
from .constants import (
@@ -143,7 +159,7 @@ _ROLE_MAPS: Dict[str, Dict[str, str]] = {
# MagentaEUEpgManager
# ---------------------------------------------------------------------------
class MagentaEUEpgManager:
class MagentaEUEpgManager(EpgManager):
"""
Fetches and normalises EPG data from the Magenta EU bifrost API.
@@ -174,9 +190,11 @@ class MagentaEUEpgManager:
def __init__(
self,
*,
country: str,
http_manager: Any,
authenticator: Any,
auth: Any,
config: Any,
cache: Optional[_ProgramDetailsCache] = None,
fetch_details: bool = False,
) -> None:
@@ -185,8 +203,13 @@ class MagentaEUEpgManager:
----------
country: Two-letter country code (at / pl / hr / me / hu).
http_manager: Provider's shared HTTPManager instance.
authenticator: MagentaAuthenticator — used only to read device_id /
session_id from the current token (no auth calls made).
auth: MagentaAuthenticator — used only to read device_id /
session_id from the current token and to obtain
guest-session ids via get_guest_session_ids().
No auth calls are made from here directly.
config: The provider's config object (COUNTRY_CONFIG entry
or a wrapper). The ABC requires it for consistency;
this manager does not read it.
cache: Optional shared _ProgramDetailsCache. A default in-memory
instance is created if omitted.
fetch_details: If True, fetch full programme details (description, credits,
@@ -195,9 +218,20 @@ class MagentaEUEpgManager:
if country not in SUPPORTED_COUNTRIES:
raise ValueError(f"MagentaEUEpgManager: unsupported country '{country}'")
# Forward ONLY the four required collaborators. Extra state goes on
# self below. A typo at a call site becomes an immediate TypeError.
super().__init__(
http_manager=http_manager,
auth=auth,
country=country,
config=config,
)
# Preserve internal aliases so every existing method body below
# continues to read the same names.
self._country = country
self._http = http_manager
self._auth = authenticator
self._auth = auth
self._cache = cache or _ProgramDetailsCache()
self._fetch_details = fetch_details
self._role_map = _ROLE_MAPS.get(country, _ROLES_DE)
@@ -211,22 +245,44 @@ class MagentaEUEpgManager:
f"fetch_details={fetch_details}"
)
# ------------------------------------------------------------------
# ABC capability
# ------------------------------------------------------------------
@property
def epg_window(self) -> Tuple[int, int]:
"""
Return the EPG window as (past_days, future_days).
Magenta EU provides 7 days each direction for every country.
The provider's own epg_window property delegates here.
"""
return 7, 7
# ------------------------------------------------------------------
# Public API
# ------------------------------------------------------------------
def get_channel_epg(
def get_epg(
self,
channel_id: str,
start_time: Optional[datetime] = None,
end_time: Optional[datetime] = None,
**_kwargs: Any,
) -> List[EPGEntry]:
"""
Return all EPG entries for a single channel within the window.
(Method body is identical to the previous get_channel_epg; only the
name changed to match the ABC.)
"""
date_from, date_to = self._resolve_window(start_time, end_time)
# Collect the calendar dates spanned by the window in UTC
dates: List[datetime] = []
current = date_from.astimezone(timezone.utc).replace(hour=0, minute=0, second=0, microsecond=0)
current = date_from.astimezone(timezone.utc).replace(
hour=0, minute=0, second=0, microsecond=0
)
utc_to = date_to.astimezone(timezone.utc)
while current.date() <= utc_to.date():
dates.append(current)
@@ -260,13 +316,19 @@ class MagentaEUEpgManager:
)
return programmes
def get_channel_epg_batch(
def get_epg_grid(
self,
channel_ids: List[str],
start_time: Optional[datetime] = None,
end_time: Optional[datetime] = None,
**_kwargs: Any,
) -> Dict[str, List[EPGEntry]]:
"""
Batch EPG for multiple channels.
(Method body is identical to the previous get_channel_epg_batch;
only the name changed to match the ABC.)
"""
if not channel_ids:
return {}
@@ -274,7 +336,9 @@ class MagentaEUEpgManager:
# Collect the calendar dates spanned by the window in UTC
dates: List[datetime] = []
current = date_from.astimezone(timezone.utc).replace(hour=0, minute=0, second=0, microsecond=0)
current = date_from.astimezone(timezone.utc).replace(
hour=0, minute=0, second=0, microsecond=0
)
utc_to = date_to.astimezone(timezone.utc)
while current.date() <= utc_to.date():
dates.append(current)
@@ -323,7 +387,9 @@ class MagentaEUEpgManager:
return result
def get_program_details(self, program_id: str) -> Optional[EPGProgramDetails]:
def get_program_details(
self, program_id: str, **_kwargs: Any
) -> Optional[EPGProgramDetails]:
"""
Fetch detailed metadata for a single programme.
@@ -450,8 +516,12 @@ class MagentaEUEpgManager:
return dt.replace(tzinfo=timezone.utc)
return dt
def _fetch_day_schedules(self, date: datetime, start_time: Optional[datetime] = None,
end_time: Optional[datetime] = None) -> Dict[str, Any]:
def _fetch_day_schedules(
self,
date: datetime,
start_time: Optional[datetime] = None,
end_time: Optional[datetime] = None,
) -> Dict[str, Any]:
"""Fetch only the 3-hour blocks that overlap with the requested time window.
`date` and the window bounds are UTC-aware (per _resolve_window). The
@@ -826,7 +896,7 @@ class MagentaEUEpgManager:
first_aired=None, # Not available from bifrost API
imdb_number=None, # Not available from bifrost API
series_link=None, # Not available from bifrost API
flags=flags, # Could be set based on programme properties if needed
flags=flags, # Could be set based on programme properties if needed
)
# ------------------------------------------------------------------
lib/streaming_providers/providers/magentaeu/provider.py
+54 -13
View File
@@ -25,6 +25,7 @@ from .vod_errors import VodCatchupRequiredError, VodNotFoundError
from .constants import (
API_ENDPOINTS,
CONTENT_TYPE_LIVE,
COUNTRY_CONFIG,
DEFAULT_COUNTRY,
DEFAULT_MAX_RETRIES,
DEFAULT_REQUEST_TIMEOUT,
@@ -99,11 +100,21 @@ class MagentaEUProvider(StreamingProvider):
proxy_config=self.http_manager.config.proxy_config,
)
# EPG manager — owns all schedule fetch/parse logic
# EPG manager — owns all schedule fetch/parse logic.
#
# Migration: constructor now takes the four ABC-required keyword
# collaborators (http_manager, auth, country, config). The
# `config` here is the COUNTRY_CONFIG entry for this country --
# the ABC requires a config object but Magenta's managers do not
# read it (they use country-derived URLs directly via helpers in
# constants.py). Passing the COUNTRY_CONFIG entry keeps the
# shape uniform and lets a future refactor to a proper config
# class happen without touching the manager.
self.epg_manager = MagentaEUEpgManager(
country=country,
http_manager=self.http_manager,
authenticator=self.authenticator,
auth=self.authenticator,
config=COUNTRY_CONFIG.get(country, COUNTRY_CONFIG[DEFAULT_COUNTRY]),
)
# VOD manager — lazy, same reasoning as epg_manager but VOD is
@@ -143,7 +154,14 @@ class MagentaEUProvider(StreamingProvider):
@property
def epg_window(self) -> Tuple[int, int]:
return 7, 7
"""
Return the EPG window as (past_days, future_days).
Delegates to the EPG manager's own property so there is one
source of truth. Kept here because external callers and the
operations layer already read it from the provider.
"""
return self.epg_manager.epg_window
@property
def catchup_window(self) -> int:
@@ -158,10 +176,16 @@ class MagentaEUProvider(StreamingProvider):
if self._vod_manager is None:
with self._vod_manager_lock:
if self._vod_manager is None: # re-check inside the lock
# Migration: constructor now takes auth= and config=
# instead of authenticator=. The config argument
# follows the same pattern as epg_manager above.
self._vod_manager = MagentaEUVodManager(
country=self.country,
http_manager=self.http_manager,
authenticator=self.authenticator,
auth=self.authenticator,
config=COUNTRY_CONFIG.get(
self.country, COUNTRY_CONFIG[DEFAULT_COUNTRY]
),
)
return self._vod_manager
@@ -170,7 +194,16 @@ class MagentaEUProvider(StreamingProvider):
return True
def get_vod_category(self, content_id: str = "", **kwargs) -> List:
return self.vod_manager.get_category_children(content_id)
"""
Return the children of a VOD node.
Migration: the manager's ABC method now returns a VodPage; the
provider keeps returning the entries list for backward
compatibility with existing callers. Migrating callers to
VodPage is a separate step and can be done incrementally --
for now, `.entries` preserves the old shape exactly.
"""
return self.vod_manager.get_vod_category(content_id, **kwargs).entries
def search_vod(
self,
@@ -330,6 +363,10 @@ class MagentaEUProvider(StreamingProvider):
directly (the manager constructs these internally; no dict conversion
happens at this layer).
Migration: the manager's method was renamed from get_channel_epg
to get_epg to match the ABC. Call shape and semantics are
unchanged.
Parameters
----------
channel_id: Station ID (theplatform Station URI) — same value stored
@@ -339,8 +376,8 @@ class MagentaEUProvider(StreamingProvider):
"""
if not self._ensure_channels_cache():
return []
return self.epg_manager.get_channel_epg(
channel_id=channel_id,
return self.epg_manager.get_epg(
channel_id,
start_time=kwargs.get("start_time"),
end_time=kwargs.get("end_time"),
)
@@ -355,9 +392,13 @@ class MagentaEUProvider(StreamingProvider):
"""
Get EPG data for multiple channels efficiently as EPGEntry objects.
Uses get_channel_epg_batch() which fetches schedule data once per
calendar day and extracts all channels in a single pass (8*D HTTP
requests rather than 8*D*N).
Uses get_epg_grid() on the manager, which fetches schedule data
once per calendar day and extracts all channels in a single pass
(8*D HTTP requests rather than 8*D*N).
Migration: the manager's method was renamed from
get_channel_epg_batch to get_epg_grid to match the ABC. Call
shape and semantics are unchanged.
Note on wall-clock cost: each calendar day in the window requires 8
sequential HTTP requests (3-hour blocks) with a 1-second sleep between
@@ -383,8 +424,8 @@ class MagentaEUProvider(StreamingProvider):
if channel_ids is None:
channel_ids = [channel.channel_id for channel in self._channels_cache]
return self.epg_manager.get_channel_epg_batch(
channel_ids=channel_ids,
return self.epg_manager.get_epg_grid(
channel_ids,
start_time=start_time,
end_time=end_time,
)
@@ -401,7 +442,7 @@ class MagentaEUProvider(StreamingProvider):
"""
if not self._ensure_channels_cache():
return None
return self.epg_manager.get_program_details(program_id)
return self.epg_manager.get_program_details(program_id, **kwargs)
def enrich_channel_data(
self, channel: StreamingChannel, **kwargs
lib/streaming_providers/providers/magentaeu/vod_errors.py
+53 -32
View File
@@ -3,6 +3,10 @@
"""
Typed exceptions for the MagentaEU VOD path.
Rebased on the shared hierarchy in base.errors. The class names are
unchanged so existing callers (except VodError: ...) keep working; the
only change is what these classes inherit from.
Rationale: the bifrost API returns 4xx for several distinct conditions --
expired token, geo-block, entitlement denial, content removed -- and they
need to be handled differently. `raise_for_status()` alone collapses them
@@ -11,39 +15,56 @@ to distinguish "refresh and retry" from "tell the user they can't watch
this here" from "this title is gone" from "this is actually catch-up,
not VOD -- hand it to the channel/catchup pathway instead".
Hierarchy:
Hierarchy (now rebased on base.errors):
VodError (base, catch-all)
├── VodAuthError (401 -- token expired/invalid)
├── VodGeoBlockError (403 -- not available in your region)
├── VodEntitlementError (403 -- not in your subscription)
│ └── VodAccountVodDisabledError (account-level VOD gate is off)
├── VodNotFoundError (404 -- content removed)
├── VodRateLimitError (429)
├── VodServerError (5xx -- retryable)
├── VodCatchupRequiredError (item has no watch/trailer action but
│ DOES have schedules/catchup_schedules
│ -- it's a linear-catchup item, not a
│ playable VOD asset; see docstring)
└── VodNotImplementedError (feature captured-but-not-yet-mapped)
VodError (base, catch-all -- subclass of ProviderError)
├── VodAuthError (401 -- token expired/invalid)
├── VodGeoBlockError (403 -- not available in your region)
├── VodEntitlementError (403 -- not in your subscription)
│ └── VodAccountVodDisabledError (account-level VOD gate is off)
├── VodNotFoundError (404 -- content removed)
├── VodBadRequestError (400 -- routing signal)
├── VodRateLimitError (429)
├── VodServerError (5xx -- retryable)
├── VodCatchupRequiredError (item has no watch/trailer action but
│ DOES have schedules/catchup_schedules
│ -- it's a linear-catchup item, not a
│ playable VOD asset)
└── VodNotImplementedError (feature captured-but-not-yet-mapped)
Callers can now catch either the local name (VodAuthError) or the shared
base (AuthError) depending on their intent.
"""
from __future__ import annotations
from typing import Any, Dict, List, Optional
class VodError(Exception):
"""Base class for all MagentaEU VOD errors."""
def __init__(self, message: str, *, status: Optional[int] = None,
url: Optional[str] = None) -> None:
super().__init__(message)
self.status = status
self.url = url
from ...base.errors import (
AuthError,
BadRequestError,
CatchupRequiredError,
EntitlementError,
GeoBlockError,
NotFoundError,
NotImplementedYetError,
ProviderError,
RateLimitError,
ServerError,
)
class VodAuthError(VodError):
class VodError(ProviderError):
"""
Base class for all MagentaEU VOD errors.
Kept as a distinct name so existing `except VodError:` sites continue
to work; it is now a subclass of the shared ProviderError, so callers
that prefer the shared name (`except ProviderError:`) also catch it.
"""
class VodAuthError(AuthError, VodError):
"""
Raised on 401 or on a locally-detected expired Bff_token.
@@ -55,11 +76,11 @@ class VodAuthError(VodError):
"""
class VodGeoBlockError(VodError):
class VodGeoBlockError(GeoBlockError, VodError):
"""Content is geo-blocked for the current network egress."""
class VodEntitlementError(VodError):
class VodEntitlementError(EntitlementError, VodError):
"""
The subscriber is authenticated but not entitled to this content.
@@ -84,11 +105,11 @@ class VodAccountVodDisabledError(VodEntitlementError):
"""
class VodNotFoundError(VodError):
class VodNotFoundError(NotFoundError, VodError):
"""Content has been removed or never existed."""
class VodBadRequestError(VodError):
class VodBadRequestError(BadRequestError, VodError):
"""
400 -- malformed request for the endpoint called.
@@ -100,15 +121,15 @@ class VodBadRequestError(VodError):
"""
class VodRateLimitError(VodError):
class VodRateLimitError(RateLimitError, VodError):
"""429 -- caller should back off."""
class VodServerError(VodError):
class VodServerError(ServerError, VodError):
"""5xx -- retryable."""
class VodCatchupRequiredError(VodError):
class VodCatchupRequiredError(CatchupRequiredError, VodError):
"""
This "VOD" list entry is actually a catch-up item from a linear
channel, not a TVOD/SVOD asset.
@@ -150,7 +171,7 @@ class VodCatchupRequiredError(VodError):
self.catchup_schedules = catchup_schedules or []
class VodNotImplementedError(VodError):
class VodNotImplementedError(NotImplementedYetError, VodError):
"""
Raised by methods whose endpoint shape has not been captured yet.
lib/streaming_providers/providers/magentaeu/vod_manager.py
+85 -6
View File
@@ -75,6 +75,13 @@ exists to build it from.
Everything else (browse, search, pagination) is unchanged from the
original proposal, which matched the capture well.
Migration note
--------------
This manager now subclasses ``base.managers.VodManager``. Three new
methods (get_vod_category, get_vod_manifest, get_vod_drm) wrap the
existing ones (get_category_children, get_manifest, get_drm) to satisfy
the ABC contract. None of the existing methods changed.
"""
from __future__ import annotations
@@ -83,9 +90,11 @@ import time
from dataclasses import dataclass
from typing import Any, Dict, List, Optional, Tuple, Union
from ...base.managers import VodManager
from ...base.models import ContentType, DRMConfig, StreamingMode
from ...base.models.vod import VodCategory, VodItem
from ...base.utils.logger import logger
from ...base.vod import VodPage
from ..lib_theplatform import (
build_licence_url,
@@ -123,7 +132,7 @@ from .vod_errors import (
# ---------------------------------------------------------------------------
@dataclass
class VodPage:
class VodPage_Internal:
"""A page of VOD results (see get_component_assets_page)."""
entries: List[Union[VodCategory, VodItem]]
next_offset: Optional[int]
@@ -161,7 +170,7 @@ class ResolvedPlayback:
# Manager
# ---------------------------------------------------------------------------
class MagentaEUVodManager:
class MagentaEUVodManager(VodManager):
"""VOD browse / search / playback-info manager for MagentaEU."""
DEFAULT_ASSET_PAGE_SIZE = 20
@@ -176,13 +185,36 @@ class MagentaEUVodManager:
def __init__(
self,
*,
country: str,
http_manager,
authenticator: MagentaAuthenticator,
auth: MagentaAuthenticator,
config: Any,
) -> None:
"""
Args:
country: Two-letter country code.
http_manager: Provider's shared HTTPManager instance.
auth: MagentaAuthenticator -- provides the access token
and per-request auth headers.
config: The provider's config object (COUNTRY_CONFIG
entry or a wrapper). Required by the ABC; this
manager does not read it.
"""
# Forward ONLY the four required collaborators. Extra state
# (country-derived URLs, playback cache) goes on self below.
super().__init__(
http_manager=http_manager,
auth=auth,
country=country,
config=config,
)
# Preserve internal aliases so every existing method body below
# continues to read the same names.
self._country = country
self._http = http_manager
self._auth = authenticator
self._auth = auth
self._bifrost_url = get_bifrost_url(country)
self._natco_key = get_natco_key(country)
@@ -195,6 +227,53 @@ class MagentaEUVodManager:
logger.info(f"[MagentaEUVodManager/{country}] initialised")
# ==================================================================
# ABC methods (base.managers.VodManager)
# ==================================================================
#
# These three wrappers satisfy the ABC. Each delegates to an existing
# method that already implements the behavior under a different name.
# No existing method changes.
def get_vod_category(
self,
content_id: str = "",
cursor: Optional[str] = None,
page_size: int = 24,
**kw,
) -> VodPage:
"""
ABC entry point. Delegates to the existing get_category_children
and wraps the result in a VodPage.
Magenta's VOD catalogue does not paginate -- the API returns the
full list for a node in one call -- so next_cursor is always None
and total is always None.
"""
entries = self.get_category_children(content_id)
return VodPage(entries=entries, next_cursor=None, total=None)
def get_vod_manifest(self, content_id: str, **kw) -> Optional[str]:
"""
ABC entry point. Delegates to the existing get_manifest().
The existing method implements the full flow: resolve playback
info via playinfo/media, cache the result, return the manifest
URL. This wrapper exists to satisfy the ABC name.
"""
return self.get_manifest(content_id, **kw)
def get_vod_drm(self, content_id: str, **kw) -> List[DRMConfig]:
"""
ABC entry point. Delegates to the existing get_drm().
Same reasoning as get_vod_manifest. Because this method is
overridden (the ABC's default returns []), the provider's
implements_drm property correctly reports True for Magenta
without any additional flag.
"""
return self.get_drm(content_id, **kw)
# ==================================================================
# Public API -- VOD enablement
# ==================================================================
@@ -314,7 +393,7 @@ class MagentaEUVodManager:
component_id: str,
offset: int = 0,
page_size: Optional[int] = None,
) -> VodPage:
) -> VodPage_Internal:
size = page_size or self.DEFAULT_ASSET_PAGE_SIZE
data = self._request(
@@ -344,7 +423,7 @@ class MagentaEUVodManager:
f"[{self._country}] component {component_id} offset={offset}: "
f"{len(entries)} items, next={normalised_next}"
)
return VodPage(entries=entries, next_offset=normalised_next)
return VodPage_Internal(entries=entries, next_offset=normalised_next)
# ==================================================================
# Public API -- details