From 1fb228bd464da1969eb9245aeea34b88e6cb97c9 Mon Sep 17 00:00:00 2001 From: Nirvana Date: Wed, 14 Oct 2026 09:59:46 +0200 Subject: [PATCH] Migrate movetv --- .../base/managed_provider.py | 25 +- lib/streaming_providers/base/managers/epg.py | 36 +- .../providers/_template_v2/MIGRATION_BRIEF.md | 15 +- .../providers/_template_v2/README.md | 156 ++++ .../providers/movetv/__init__.py | 22 +- .../providers/movetv/api.py | 187 +++++ .../providers/movetv/auth.py | 53 +- .../providers/movetv/catchup_manager.py | 182 +++++ .../providers/movetv/channel_manager.py | 186 +++++ .../providers/movetv/constants.py | 47 +- .../providers/movetv/epg_manager.py | 251 ++---- .../providers/movetv/ids.py | 55 ++ .../providers/movetv/provider.py | 768 ++++-------------- .../providers/movetv/session.py | 177 ++++ .../providers/movetv/vod_manager.py | 431 ++++------ 15 files changed, 1470 insertions(+), 1121 deletions(-) create mode 100644 lib/streaming_providers/providers/movetv/api.py create mode 100644 lib/streaming_providers/providers/movetv/catchup_manager.py create mode 100644 lib/streaming_providers/providers/movetv/channel_manager.py create mode 100644 lib/streaming_providers/providers/movetv/ids.py create mode 100644 lib/streaming_providers/providers/movetv/session.py diff --git a/lib/streaming_providers/base/managed_provider.py b/lib/streaming_providers/base/managed_provider.py index a28697a..6ed1c3f 100644 --- a/lib/streaming_providers/base/managed_provider.py +++ b/lib/streaming_providers/base/managed_provider.py @@ -105,6 +105,15 @@ class ManagedProvider(StreamingProvider): # -- only for providers whose catchup is encrypted exactly like live. CATCHUP_DRM_FROM_LIVE: ClassVar[bool] = False + # Opt-in (default False = the historical behaviour, unchanged). + # False: get_manifest_headers() ignores catchup arguments; only + # get_segment_headers() routes a start_time kwarg to the catchup + # manager. A provider that needs more overrides get_manifest_headers. + # True: a get_manifest_headers() call carrying start_time is routed to + # CatchupManager.get_catchup_manifest_headers(), like the segment + # call. Only effective with HEADERS_FROM_MANAGERS = True. + CATCHUP_HEADERS_ON_MANIFEST_CALL: ClassVar[bool] = False + ROUTE_CACHE_SIZE: ClassVar[int] = 512 # NOTE: `self.channels` holds the ChannelManager here (template @@ -353,6 +362,20 @@ class ManagedProvider(StreamingProvider): def get_manifest_headers(self, content_id: str, **kw: Any) -> Dict[str, str]: if not self.HEADERS_FROM_MANAGERS: return super().get_manifest_headers(content_id, **kw) + if ( + self.CATCHUP_HEADERS_ON_MANIFEST_CALL + and self.catchup is not None + and kw.get("start_time") is not None + ): + rest = dict(kw) + start = rest.pop("start_time") + return self.catchup.get_catchup_manifest_headers( + content_id, + start, + end_time=rest.pop("end_time", None), + epg_id=rest.pop("epg_id", None), + **rest, + ) name, mgr = self._routed_manager(content_id) if name == "channels": return mgr.get_channel_manifest_headers(content_id, **kw) @@ -640,4 +663,4 @@ class ManagedProvider(StreamingProvider): def delete_bookmark(self, content_id: str, **kw: Any) -> None: if self.bookmarks is None: return super().delete_bookmark(content_id, **kw) - return self.bookmarks.delete_bookmark(content_id, **kw) + return self.bookmarks.delete_bookmark(content_id, **kw) \ No newline at end of file diff --git a/lib/streaming_providers/base/managers/epg.py b/lib/streaming_providers/base/managers/epg.py index 8b34a4c..f8fe2f2 100644 --- a/lib/streaming_providers/base/managers/epg.py +++ b/lib/streaming_providers/base/managers/epg.py @@ -24,15 +24,26 @@ from __future__ import annotations from abc import abstractmethod from datetime import datetime -from typing import Any, Dict, List, Optional, Tuple +from typing import Any, ClassVar, Dict, List, Optional, Tuple +from ..errors import AuthError, GeoBlockError, ProviderError, RateLimitError from ..models.epg_models import EPGEntry, EPGProgramDetails +from ..utils.logger import logger from ._base import ManagerBase class EpgManager(ManagerBase): """Abstract base for provider EPG managers.""" + # Opt-in (default False = the historical behaviour, unchanged). + # False: get_epg_grid() lets any exception of ONE channel abort the grid. + # True: AuthError / RateLimitError / GeoBlockError still abort (they hit + # every channel alike); any other ProviderError of one channel yields + # [] for that channel plus a warning, so one bad channel cannot take + # the whole grid down. Only affects the DEFAULT get_epg_grid; a + # manager that overrides get_epg_grid is not touched. + ISOLATE_GRID_FAILURES: ClassVar[bool] = False + # ------------------------------------------------------------------ # Capability # ------------------------------------------------------------------ @@ -89,15 +100,32 @@ class EpgManager(ManagerBase): Default: loops get_epg per channel, skipping channels for which handles_channel_id() is False (returns [] for those without calling get_epg). Override when the provider has a native batch endpoint. + + With ISOLATE_GRID_FAILURES = True a failure of one channel (other than + auth / rate-limit / geo) yields [] for that channel instead of + aborting the grid. Default: False (unchanged behaviour). """ result: Dict[str, List[EPGEntry]] = {} for cid in channel_ids: if not self.handles_channel_id(cid): result[cid] = [] continue - result[cid] = self.get_epg( - cid, start_time=start_time, end_time=end_time, **kw - ) + if not self.ISOLATE_GRID_FAILURES: + result[cid] = self.get_epg( + cid, start_time=start_time, end_time=end_time, **kw + ) + continue + try: + result[cid] = self.get_epg( + cid, start_time=start_time, end_time=end_time, **kw + ) + except (AuthError, RateLimitError, GeoBlockError): + raise + except ProviderError as exc: + logger.warning( + f"{self.__class__.__name__}: channel {cid} skipped in EPG grid: {exc}" + ) + result[cid] = [] return result def get_program_details( diff --git a/lib/streaming_providers/providers/_template_v2/MIGRATION_BRIEF.md b/lib/streaming_providers/providers/_template_v2/MIGRATION_BRIEF.md index d8e7362..1ce4cc4 100644 --- a/lib/streaming_providers/providers/_template_v2/MIGRATION_BRIEF.md +++ b/lib/streaming_providers/providers/_template_v2/MIGRATION_BRIEF.md @@ -17,6 +17,7 @@ python -m streaming_providers.base.testing.migration_pack --json before.js | Directory / registry key / `provider_name` / credentials key | ❓ all equal? (§4) | | Class name | | | `SUPPORTED_COUNTRIES` (pack §1) | `[]` / one entry / several | +| **Country the registry really passes** vs. the constructor default vs. the provider's real country | ❓ three values, they differ more often than not (movetv: `"DE"` / `"SI"` / `"RS"`); stored credentials, sessions and proxy settings are keyed by lower-case country (README §4) | | How the registry calls the constructor (pack §1, source of `create_instance`) | ❓ `cls(country=…)`? `config`? extras? | | Constructor signature the migrated class must keep | | | Extra `ctor_kwargs` the contract test needs | | @@ -26,6 +27,8 @@ python -m streaming_providers.base.testing.migration_pack --json before.js ## 2. Capabilities (what v1 really has) channels ☐ vod ☐ epg ☐ recordings ☐ favorites ☐ bookmarks ☐ catchup ☐ drm ☐ timers ☐(not delegated) events ☐(not delegated) +**Completeness check of the supplied v1 files (README §12.0b):** `grep -n "unchanged\|^\s*pass$"` — any placeholder body? ❓ (a stubbed capability stays unregistered until the body exists) + For each: does a v1 manager class already exist? Does it subclass the ABC? Does it follow the 4-collaborator constructor, or take the provider (`__init__(self, provider)`)? ❓ @@ -38,6 +41,15 @@ constructor, or take the provider (`__init__(self, provider)`)? ❓ | Shared state: entitlement / profile / … (§5.1) | session / dedicated collaborator / cache / helper | every back-reference in pack §4 mapped to a row | | Credentials API (`set_user_credentials`, `get_auth_details`, …) | keep / drop | callers found (pack §5) **and** does the v1 surface (`before.json`) define them? Dropping a method v1 has is a REMOVED line | | Token buffer(s) in the authenticator (grep `expires_in -`) | keep (reason: ___) / remove (device test: ___) | never remove by reflex (§9) | +| **Token validity model** | expiry-based / server-validated (a network call inside `authenticate()`?) | server-validated → validate once + 401 recovery (§9); count the calls per request today | +| **`success=false` / empty answers, per endpoint** | required (raise) / optional (empty or `None`) | rule in §8; one row per endpoint | +| **Ids that cannot be told apart** (live vs VOD both numeric) | who owns numeric ids; which manager claims nothing | §7 | +| **Channel subclass / API-internal ids** | move to a record in the ChannelManager; extra `to_dict()` keys: ___ | external users of the removed keys (pack §5, addon repos) | +| **Types named like base types** (`VodPage`, …) | rename to: ___ | §12.2 | +| **Legacy properties not delegated** (`uses_dynamic_manifests`, `requires_manifest_context`, `supports_catchup`, label/logo/auth types) | keep (v1 value ≠ default) / delete | check the legacy mixin default | +| **Capabilities that must stay unregistered** (stub bodies) | list + the flag that enables them | §12.2 | +| **Opt-in base flags needed** (`CATCHUP_HEADERS_ON_MANIFEST_CALL`, `EpgManager.ISOLATE_GRID_FAILURES`) | yes / no | §6.3, §6.6 — or a local override until the base change is merged | +| **`operation=` values passed to the HTTP manager** | keep v1 strings / use `"api"` | §8 | | Provider-specific exceptions callers rely on | are they `ProviderError` subclasses? if not: make them so or `passthrough=` in `transport_errors` | `issubclass(X, ProviderError)` | | `get_events` / `enrich_channel_data` / `get_program_details` | real implementation → keep on provider; inert stub → delete | §12.2 | | Dead v1 methods (`enrich_channel_data`, `populate_streaming_data`, …) | no caller → delete; caller → move to provider unchanged | grep result (pack §5): ___ | @@ -57,12 +69,13 @@ constructor, or take the provider (`__init__(self, provider)`)? ❓ | | | | ## 5. External users (pack §5) +Pack and `before.json` written **outside** the scanned tree? ☐ Kodi / PVR addon repos passed with `--extra-root`? ☐ Files uploaded as a zip (no same-named overwrites)? ☐ | Where | What it uses | Action (delegate / call-site fix / none) | |---|---|---| | | | | ## 6. Behaviour changes accepted (§12.5) -☐ typed errors instead of `[]`/`None` ☐ headers from manager hooks ☐ capabilities now reachable (list which) ☐ eager login removed ☐ token buffer changed (reason recorded) ☐ other: ______ +☐ country / storage keys changed (credentials re-entered or key migrated) ☐ per-call token validation replaced ☐ `success=false` rule applied ☐ per-channel catchup window enforced ☐ typed errors instead of `[]`/`None` ☐ headers from manager hooks ☐ capabilities now reachable (list which) ☐ eager login removed ☐ token buffer changed (reason recorded) ☐ other: ______ ## 7. Done when Definition of done (README §18); `--compare before.json` reviewed — expected: REMOVED = deleted boilerplate/stubs/delegated flags, ADDED = only `_build_*`, `DRM_IN_MANAGERS`, `HEADERS_FROM_MANAGERS` (anything else is unexplained); tests parse and run, **at least one test fails on the v1 shape**; device checklist (§12.7) signed off. diff --git a/lib/streaming_providers/providers/_template_v2/README.md b/lib/streaming_providers/providers/_template_v2/README.md index a49ab72..2350859 100644 --- a/lib/streaming_providers/providers/_template_v2/README.md +++ b/lib/streaming_providers/providers/_template_v2/README.md @@ -206,6 +206,19 @@ directory name == AVAILABLE_PROVIDERS key == provider_name == PROVIDER_NAME as Allente does. * `get_all_possible_instances()` hard-codes `"DE"` for single-country providers (known issue, §17) — do not rely on it for AT/SE/… providers. +* **Take the country from the registry, not from the constructor default.** Pack §1 + shows what the registry really passes (movetv: ctor default `"SI"`, registry + `"DE"`, correct value `"RS"`). Ask for the provider's real country before + writing `SUPPORTED_COUNTRIES`. +* **Changing the country (or adding the lower-casing) moves storage keys.** + CredentialManager / SessionManager / ProxyConfigManager key by lower-case country: + stored credentials, sessions and proxy settings of a provider that used to run + under another country (`de`) are **not found** under the new one (`rs`). Record it + in the brief (§12.5) and either migrate the stored key or tell users to re-enter + the credentials. Lower-casing *before* `_setup_http_manager` can also make a stored + proxy apply that v1 never found. +* `Content.country` / `Channel.country` are **upper-case** (`"RS"`), the manager's + `country` collaborator is lower-case: use `self.country.upper()` for models. ### How the registry constructs providers — check, don't assume @@ -453,6 +466,20 @@ class YourManager(ChannelManager): # or Vod/Epg/Recordings/Favorite end)`) and, when the backend passes no ids, builds them from `get_channels()`. * `get_program_details(program_id) -> Optional[EPGProgramDetails]`. * `handles_channel_id(channel_id)`: override to skip channels without EPG. +* **Grid failure isolation.** The default `get_epg_grid` does not isolate a failing + channel: one `ServerError` aborts the grid. Set the class attribute + **`ISOLATE_GRID_FAILURES = True`** on your EpgManager (opt-in, default `False` = + unchanged): `AuthError` / `RateLimitError` / `GeoBlockError` still abort (they hit + every channel alike), any other `ProviderError` of ONE channel yields `[]` plus a + warning. A manager that overrides `get_epg_grid` is not affected. (movetv carries + its own override with the same semantics until the base change is merged.) +* **EPG-based catchup** (the catchup source is requested per programme): the catchup + manager needs the EPG manager as an explicit keyword-only extra (`epg=`) and asks + it for the programme at `start_time`. A provider-native window in `**kw` + (`backwards=`/`forwards=` hours, movetv) is fine; ignore unknown kwargs such as + `limit=` knowingly and say so in the docstring. +* A failed request RAISES; "no programmes" (`success=false`, empty list) is `[]` + (decision rule in §8). ### 6.4 `RecordingsManager` @@ -519,6 +546,20 @@ class YourManager(ChannelManager): # or Vod/Epg/Recordings/Favorite * `catchup:@` style ids are provider grammar: parse them in the provider's `get_manifest`/`get_drm` and hand the manager plain `(content_id, start_time)`; a malformed id raises `BadRequestError`. +* **`ManagedProvider` routes catchup arguments asymmetrically**: `get_segment_headers` + sends a `start_time` kwarg to the catchup manager, `get_manifest_headers` does not. + A provider whose manifest-header call carries `start_time` sets + **`CATCHUP_HEADERS_ON_MANIFEST_CALL = True`** (opt-in, default `False` = unchanged; + needs `HEADERS_FROM_MANAGERS = True`): `get_manifest_headers(id, start_time=…, + end_time=…, epg_id=…)` is then routed to `get_catchup_manifest_headers` like the + segment call. Until that base change is merged, use a provider-level + `get_manifest_headers` override that calls the catchup manager and otherwise + `super()` (movetv). +* **Catchup headers never fall back to the live headers** (wrong token / Play-Auth): + raise `NotFoundError` when no catchup source can be resolved. The segment-header + call is swallowed by the DRM pipeline (§8), so test the manifest-header path. +* Catchup that needs other managers takes them as keyword-only extras + (`channels=`, `epg=`, a shared source cache) — still no provider reference (§5.1). ### 6.7 `DRM` @@ -557,6 +598,14 @@ non-`NotFoundError` stops the routing. Put the grammar in **one pure `ids.py`** complementary by construction, and unit-test a sample of ids of each kind: exactly one manager must claim each. +**Ids that cannot be told apart** (live and VOD ids are both plain numerics): the +grammar cannot separate them, so decide who *owns* numeric ids. If one side has no +playback (movetv: VOD is browse-only), that manager returns `False` from +`handles_content_id` for **every** id and `None` from its manifest method; the other +claims the numerics. State this in `ids.py` and in the brief, and add the test with +sample ids of both kinds (§15). Browse-node ids (`category__`) are not +playable ids and must not match the live predicate. + Id grammar guidance: * Ids are opaque to the base. Use short **prefixes** when several kinds coexist @@ -626,6 +675,18 @@ KeyError` / `except RuntimeError` in the operations layer keep working. `AuthError`/`RateLimitError`** into whatever the handler does (usually `[]` or `ServerError`) and the caller can no longer back off or show the real reason. Catch `ProviderError` first and re-raise, or use `transport_errors`. +* **`success=false` / empty answers — decide per endpoint and write it down.** There + is no global rule; the one that worked for movetv: an endpoint whose result the + caller cannot do without (channel list, live source) **raises** `ServerError`; an + endpoint where "nothing there" is a valid answer (EPG, catalogue/browse, a catchup + source that may have expired) returns `[]` / `None` with a warning. Record the + choice per endpoint in the brief; it is the single most common silent behaviour + change of a migration. +* **`operation=` on `http_manager.post/get`** selects the proxy/timeout scope (`"api"`, + `"auth"`, `"manifest"`, `"license"` in the template). v1 providers pass ad-hoc + names (`"live_manifest"`, `"token_validate"`) and omit it elsewhere (default). + The accepted values are **not verified**: keep v1's strings where v1 passed one; + use `"api"` where v1 passed nothing and note the change in §12.5. ### Backend behaviours you must know @@ -723,6 +784,27 @@ must never call the raw authenticator — they use the adapter. object (`config=`). Do not build a second config inside it (two config constructions in one provider drift apart). +### Server-validated tokens (no usable expiry) — an accepted exception to "single truth" + +Some authenticators cannot trust a token's expiry: movetv's login response has no +lifetime (v1 invented 365 days) and v1 asked the server to validate the token inside +**every** `authenticate()` call (one network POST per VOD/EPG/catchup request, up to +three per call). A session adapter for such a provider: + +* validates **once per process** (the first `authenticate()` after start, which also + validates a persisted token) and then trusts the token; +* on HTTP **401** runs `recover()`: refresh the token, else a full login with + `force_refresh=True`, then **exactly one** retry of the request (a shared + `api_post` helper does this for every manager; a second 401 raises `AuthError`); +* does not remove the validation silently: record it in the brief (token decision + row) and in §12.5, and test: first call validates, later calls do not, 401 → one + refresh and one retry, second 401 → `AuthError`, 5xx → no retry. + +The template adapter assumes `current_token`, `has_user_credentials()` and +`invalidate_token()` on the authenticator; **check them in `BaseAuthenticator`** +before wrapping, or use the provider authenticator's own getters (movetv: +`get_current_token()`). + ### Credentials API on the provider (OPTIONAL — only if v1 has it) **A migration preserves the v1 public surface; it never adds to it.** Of the two @@ -904,6 +986,18 @@ python -m streaming_providers.base.testing.migration_pack --compare before Every `REMOVED`/`CHANGED` line must be explained (deleted boilerplate is fine; a removed public method that something calls is not), every `ADDED` line justified. +**Input hygiene (each cost a round trip once):** + +* **The supplied v1 files must be complete.** Search them for placeholder bodies + before anything else: `grep -n "unchanged\|^\s*pass$" `. A `pass # ... unchanged ...` + body (movetv `vod_manager.py`) cannot be migrated or rewritten; the migrator needs + the real file (git history) or an explicit decision to ship it disabled. +* **Write `pack.txt` / `before.json` OUTSIDE the scanned tree** (e.g. `/tmp`): a pack + inside `--repo` is scanned and reports itself as an "external user". +* **Never upload same-named files separately** (`provider.py`, `auth.py`, + `constants.py`, `__init__.py`, `channel.py`, `vod.py` of several packages overwrite + each other). Zip the directories or keep relative paths. + ### 12.1 Inventory (10 minutes, saves hours) ```bash @@ -955,6 +1049,11 @@ Fill this table for your provider: | `__init__(…, **kwargs)` | drop unless the registry needs it | | `catchup_window_hours` = minimum of per-channel windows | **maximum**, plus `catchup_window_for_channel` (§6.6) | | ad-hoc `get_restart()` etc. | **keep** (provider-level, outside the ABC) | +| `uses_dynamic_manifests`, `requires_manifest_context`, `supports_catchup`, `provider_label/_logo`, `supported_auth_types` properties | not delegated by `ManagedProvider`. Keep the two behaviour flags if v1 returns non-default values (check what the legacy mixin returns); `supports_catchup` and the label/logo/auth-type properties duplicate the ClassVars/derived flags → delete. List each in the `--compare` review | +| `class XChannel(Channel)` with API-internal ids (`live_id`, stream uid, token fields) | **no Channel subclass**: keep the records in the ChannelManager (`content_id → record`); the extra keys disappear from `Channel.to_dict()` — grep external users (also the Kodi/PVR repos) before accepting that | +| provider-local result/model types named like base types (`VodPage`) | **rename** (`VodListing`); the base name is what `get_vod_category` returns | +| a provider package `__init__` exporting auth/config/models | **reduce to `__all__ = ["XProvider"]`**; grep external imports first | +| stub bodies (`pass`, `# TODO`) in a v1 manager | raise `NotImplementedYetError` (typed) and keep the capability **unregistered** (`_build_*` returns `None`) until the body exists; never register a manager that crashes | ### 12.3 Step by step @@ -995,6 +1094,14 @@ catchup window, the EPG flag network call). | Managers read a **fresh token per request** from `self.auth` | a `provider.bearer_token` snapshot is stale by the first request | — | | Eager login removed from `__init__` | network at registry-construction time; first call is now slower | restore only as the documented opportunistic login (stored credentials) | | A second token buffer removed / kept | **needs a recorded reason either way** (§9) | — | +| Per-call server token validation replaced by validate-once + 401 recovery | one network POST per request is not acceptable (§9 server-validated tokens) | validate in `_ready()` | +| Country / `SUPPORTED_COUNTRIES` changed, or lower-cased before `_setup_http_manager` | wrong or inconsistent country; storage keys move | re-enter credentials, or migrate the stored key (§4) | +| `success=false` now raises for required endpoints, stays empty for optional ones | per-endpoint rule (§8) | wrap the call | +| Per-channel catchup window enforced (outside → `None`; 0 h → `None`) | the backend only knows the global maximum (§6.6) | return the manifest regardless | +| `get_catchup_drm` returns `[]` → `ManagedProvider` raises `NotImplementedError` (PSSH from the manifest) | legacy contract (Trap 9); a v1 `[]` meant "clear" | set `CATCHUP_DRM_FROM_LIVE = True` if catchup is encrypted like live | +| Channel fields removed from `to_dict()` (no Channel subclass) | internals stay in the manager | — | +| One failing channel no longer empties the EPG grid (if you override it, §6.3) | resilience | keep the default | +| Capability exists in the code but is deliberately **not registered** (stub parsers) | a registered capability that crashes is worse than a missing one | flip the provider flag when the body exists | ### 12.6 External references (needs the whole repo — the migration pack section 5 does this) @@ -1070,6 +1177,9 @@ providers can be mixed freely. * `Channel.catchup_hours` is the per-channel window; leave `0` unless catchup is really offered (do not conflate with "start-over"). * `provider` on models is the **registry key** (`PROVIDER_NAME`). +* **Do not subclass `Channel`** to carry API-internal ids (a v1 habit); they vanish + from `to_dict()` in v2. Keep a small record per channel in the manager. Remember + `Channel.country` is upper-case (§4). * **Time:** the backend hands **timezone-aware UTC** `datetime`s to EPG; EPG and catchup use **unix seconds (int)** at the manager boundary. Never mix naive and aware datetimes (that was a real `TypeError` in `PricePoint.is_active`). Parse @@ -1270,6 +1380,23 @@ Things that already cost time. Each is covered by a test or the contract unless truncation (§6.2). 46. `User-Agent` and `sec-ch-ua*` derived from different constants (§10). 47. Two copies of the DRM config builder in live and VOD managers (§11). +48. Placeholder bodies (`pass # ... unchanged ...`) in the supplied v1 file: the pack's + line numbers match the stub, so the repo may be stubbed too — confirm before you + migrate (§12.0b). +49. A provider-local type named like a base type (`VodPage`) — imports resolve to the + wrong one (§12.2). +50. Changing the country or the lower-casing without migrating stored credentials, + sessions and proxy keys (§4). The constructor default is not what the registry passes. +51. Subclassing `Channel` for API-internal ids; they silently leave `to_dict()` (§13). +52. `ManagedProvider.get_manifest_headers` does not route `start_time` to catchup (only + `get_segment_headers` does) — v1 providers that did need a provider-level override (§6.6). +53. A hidden network call inside `authenticate()` (server-side token validation) paid on + every request (§9). +54. `success=false` treated one way everywhere: required endpoints must raise, "no data" + endpoints return empty (§8). +55. A registered capability whose parsers/bodies are stubs — hide it instead (§12.2). +56. Same-named files uploaded separately overwrite each other; packs written inside the + scanned tree report themselves as external users (§12.0b). --- @@ -1310,6 +1437,29 @@ Known gaps in the base, so you don't hunt for them: * `optional/session_adapter.py` updated to the rules in §9. * `tests/conftest.py` with `FakeHttp`, `FakeAuthenticator`, fake clock, `env` (§15). +**Base changes — every one must be additive with an unchanged default and pass the +contract test for ALL existing providers (Allente, simpli, Magenta2, Joyn, movetv) +before it is merged. Two are implemented as patches (below); the rest are proposals:** + +* **Implemented as patches, opt-in, defaults unchanged** (`managed_provider.py`, + `managers/epg.py`; default behaviour verified identical, flags tested): + `ManagedProvider.CATCHUP_HEADERS_ON_MANIFEST_CALL` (§6.6) and + `EpgManager.ISOLATE_GRID_FAILURES` (§6.3). Before merging run the contract test for + every provider and + `grep -rn "ISOLATE_GRID_FAILURES\|CATCHUP_HEADERS_ON_MANIFEST_CALL" providers/` + (a same-named attribute in a provider would shadow the new one). +* **Deliberately NOT added:** an `EpgManager.get_entry_at()` helper. Only movetv + needs "the programme at a timestamp"; its matching rules (containment, tolerance) + are provider-specific. Add it when a second provider needs the same rules. +* One shared session adapter with `recover()` and `on_invalidate` (backwards + compatible with the template adapter's public methods) plus the shared `api_post` + helper (401 → one recovery → one retry, both HTTP-manager error shapes) — as a new + module; no provider is forced to switch. +* Document the accepted `operation=` values after reading `HTTPManager` (§8). +* `migration_pack`: placeholder-body detection (`pass # ...`), output-inside-tree + guard, `--bundle` that keeps relative paths, and rows for the actual constructor + country vs. the declared one. + --- ## 18. Definition of done @@ -1342,6 +1492,12 @@ Known gaps in the base, so you don't hunt for them: - [ ] Retry loops: only `ServerError`/untyped retried, other `ProviderError`s propagate unchanged (test per manager) - [ ] Authenticator raises typed `AuthError` for rejected credentials/MFA; managers use `require()`/`get_access_token()`, never `ensure()` - [ ] Tests parse and run; at least one test fails on the v1 shape; credential-change cache-clearing test exists +- [ ] `SUPPORTED_COUNTRIES` and the constructor country match what the registry passes; moved storage keys handled or documented +- [ ] No placeholder/stub body is registered as a capability +- [ ] `success=false` / empty-answer decision recorded per endpoint +- [ ] No `Channel` subclass; internals live in the manager; external users of removed `to_dict()` keys checked +- [ ] No provider-local type shares a name with a base type +- [ ] Token validation model recorded (expiry-based vs. server-validated) and tested --- diff --git a/lib/streaming_providers/providers/movetv/__init__.py b/lib/streaming_providers/providers/movetv/__init__.py index a87c9b4..24e53cf 100644 --- a/lib/streaming_providers/providers/movetv/__init__.py +++ b/lib/streaming_providers/providers/movetv/__init__.py @@ -1,12 +1,14 @@ # streaming_providers/providers/movetv/__init__.py -from .provider import MoveTVProvider, MoveTVChannel -from .auth import MoveTVAuthenticator, MoveTVAuthToken -from .constants import MoveTVConfig +""" +move.tv (MTS-SI) streaming provider. -__all__ = [ - "MoveTVProvider", - "MoveTVChannel", - "MoveTVAuthenticator", - "MoveTVAuthToken", - "MoveTVConfig", -] \ No newline at end of file +Public API: + from streaming_providers.providers.movetv import MoveTVProvider + +Only the provider is public: discovery takes the FIRST StreamingProvider +subclass found in this namespace (README §5.7). +""" + +from .provider import MoveTVProvider + +__all__ = ["MoveTVProvider"] \ No newline at end of file diff --git a/lib/streaming_providers/providers/movetv/api.py b/lib/streaming_providers/providers/movetv/api.py new file mode 100644 index 0000000..ffb632e --- /dev/null +++ b/lib/streaming_providers/providers/movetv/api.py @@ -0,0 +1,187 @@ +# streaming_providers/providers/movetv/api.py +""" +Shared request helpers of the move.tv provider (module functions, no state +except SourceCache): one authenticated POST with ONE 401 recovery, and the +live/catchup "source" fetch with its Play-Auth cache. + +Error handling (README §8): + * typed ProviderErrors propagate unchanged and are never retried; + * the HTTP manager's 4xx/5xx behaviour is not verified, so BOTH a returned + error response and an exception carrying `.response` are handled; + * everything unexpected becomes ServerError. +""" + +import re +import threading +import time +from collections import OrderedDict +from typing import Any, Callable, Dict, Optional, Tuple + +from ...base.errors import ProviderError, ServerError, default_error_for_status_heuristic +from ...base.utils.logger import logger +from .constants import MoveTVConfig + +# (content_url, play_auth_header_value, expires_at_epoch_seconds) +SourceInfo = Tuple[str, str, float] + +_PLAY_AUTH_EXPIRY_RE = re.compile(MoveTVConfig.PLAY_AUTH_EXPIRY_PATTERN) + + +class SourceCache: + """Thread-safe LRU cache for source URLs + Play-Auth values. + + Keyed by content_id (live), ":" (catchup) and by the + content_url itself (so segment/manifest header lookups can find the entry + from the URL alone). + """ + + def __init__(self, max_size: int = MoveTVConfig.SOURCE_CACHE_SIZE): + self._cache: "OrderedDict[str, SourceInfo]" = OrderedDict() + self._lock = threading.Lock() + self._max_size = max_size + + def get(self, key: str) -> Optional[SourceInfo]: + 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: SourceInfo) -> 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 clear(self) -> None: + with self._lock: + self._cache.clear() + + +def is_fresh(entry: Optional[SourceInfo]) -> bool: + """A cached source is usable while its Play-Auth expiry is > buffer away. + An entry without a parsed expiry (0.0) is never reused (as in v1).""" + if not entry: + return False + expires_at = entry[2] + return bool(expires_at) and time.time() < (expires_at - MoveTVConfig.PLAY_AUTH_EARLY_EXPIRY_SECONDS) + + +def parse_play_auth_expiry(header_value: str) -> float: + if not header_value: + return 0.0 + match = _PLAY_AUTH_EXPIRY_RE.search(header_value) + if match: + return float(match.group(1)) + logger.warning("move.tv: could not parse expiry from X-Play-Auth") + return 0.0 + + +def _status_of_response(response: Any) -> Optional[int]: + status = getattr(response, "status_code", None) + return status if isinstance(status, int) else None + + +def _status_of_exception(exc: BaseException) -> Optional[int]: + return _status_of_response(getattr(exc, "response", None)) + + +def api_post( + http_manager: Any, + session: Any, + url: str, + build_payload: Callable[[Dict[str, Any]], Dict[str, Any]], + *, + what: str, + operation: str = "api", + need_profile: bool = True, +) -> Dict[str, Any]: + """ + POST an authenticated JSON request and return the parsed body (a dict). + + `build_payload(session_info)` is called for every attempt, so ids that + change after a token refresh are picked up. A 401 triggers exactly ONE + `session.recover()` (refresh token, else full login) and one retry; a second + 401 raises AuthError. Nothing else is retried. + """ + response: Any = None + for attempt in (1, 2): + info = session.info(need_profile=need_profile) # raises typed AuthError & co. + headers = MoveTVConfig.get_api_headers(info["auth_token"]) + try: + response = http_manager.post( + url, operation=operation, json=build_payload(info), headers=headers + ) + status = _status_of_response(response) + except ProviderError: + raise + except Exception as exc: # HTTP manager may raise on 4xx/5xx + status = _status_of_exception(exc) + if status is None: + raise ServerError(f"move.tv: {what} failed: {exc}", url=url) from exc + response = exc.response # type: ignore[attr-defined] + if status == 401 and attempt == 1: + logger.info(f"move.tv: 401 on {what} -- recovering the session") + session.recover() + continue + if status is not None and status >= 400: + raise default_error_for_status_heuristic( + status, f"move.tv: {what} failed (HTTP {status})", url=url + ) + break + try: + data = response.json() + except Exception as exc: + raise ServerError(f"move.tv: {what}: invalid JSON response: {exc}", url=url) from exc + if not isinstance(data, dict): + raise ServerError(f"move.tv: {what}: unexpected response body", url=url) + return data + + +def fetch_source( + *, + http_manager: Any, + session: Any, + cache: SourceCache, + url: str, + build_payload: Callable[[], Dict[str, Any]], + cache_key: str, + operation: str, + what: str, +) -> Optional[SourceInfo]: + """ + Fetch a live or catchup source (manifest URL + Play-Auth header value). + + Returns None when the API answers success=false or without a URL ("no + such source"); typed errors propagate. The result is cached under + `cache_key` and under the URL itself. + """ + cached = cache.get(cache_key) + if is_fresh(cached): + logger.debug(f"move.tv: cache hit for {what} (key={cache_key})") + return cached + + def payload(info: Dict[str, Any]) -> Dict[str, Any]: + body = dict(build_payload()) + body["customerId"] = info["customer_id"] + body["customerProfileId"] = info["customer_profile_id"] + body["appVersion"] = MoveTVConfig.APP_VERSION + body["dtype"] = MoveTVConfig.DTYPE_DASH + return body + + data = api_post(http_manager, session, url, payload, what=what, operation=operation) + if not data.get("success"): + logger.warning(f"move.tv: {what}: source endpoint returned success=false") + return None + content_url = data.get("content_url") + if not content_url: + logger.warning(f"move.tv: {what}: response without content_url") + return None + header_value = (data.get("protection") or {}).get("headerValue", "") or "" + entry: SourceInfo = (content_url, header_value, parse_play_auth_expiry(header_value)) + cache.set(cache_key, entry) + cache.set(content_url, entry) + logger.info(f"move.tv: fetched {what}: {content_url}") + return entry \ No newline at end of file diff --git a/lib/streaming_providers/providers/movetv/auth.py b/lib/streaming_providers/providers/movetv/auth.py index 02479c7..a9960ae 100644 --- a/lib/streaming_providers/providers/movetv/auth.py +++ b/lib/streaming_providers/providers/movetv/auth.py @@ -5,6 +5,13 @@ from typing import Any, Dict, Optional, cast from ...base.auth.base_auth import BaseAuthenticator, BaseAuthToken, TokenAuthLevel from ...base.auth.credentials import UserPasswordCredentials +from ...base.errors import ( + AuthError, + CredentialsError, + ProviderError, + ServerError, + default_error_for_status_heuristic, +) from ...base.utils.logger import logger from .constants import MoveTVConfig @@ -94,7 +101,7 @@ class MoveTVAuthenticator(BaseAuthenticator): def _build_auth_payload(self) -> Dict[str, Any]: """Build the login payload from current credentials.""" if not self.credentials or not isinstance(self.credentials, UserPasswordCredentials): - raise ValueError("move.tv requires username/password credentials") + raise AuthError("move.tv requires username/password credentials") uid = self.get_device_id() return { "username": self.credentials.username, @@ -189,18 +196,39 @@ class MoveTVAuthenticator(BaseAuthenticator): logger.debug(f"move.tv: Authenticating with slot UID: {uid}") - response = self.http_manager.post( - self.auth_endpoint, - json=payload, - headers=self._get_auth_headers(), - operation="auth", - ) - response.raise_for_status() - data = response.json() + # Typed errors (README §8/§9): 401 -> AuthError (permanent in the + # session), 403/429/5xx keep their own type (transient). The HTTP + # manager may return an error response OR raise with `.response`. + try: + response = self.http_manager.post( + self.auth_endpoint, + json=payload, + headers=self._get_auth_headers(), + operation="auth", + ) + status = getattr(response, "status_code", None) + except ProviderError: + raise + except Exception as exc: + status = getattr(getattr(exc, "response", None), "status_code", None) + if not isinstance(status, int): + raise ServerError(f"move.tv: login request failed: {exc}") from exc + raise default_error_for_status_heuristic( + status, f"move.tv: login failed (HTTP {status})", url=self.auth_endpoint + ) from exc + if isinstance(status, int) and status >= 400: + raise default_error_for_status_heuristic( + status, f"move.tv: login failed (HTTP {status})", url=self.auth_endpoint + ) + try: + data = response.json() + except Exception as exc: + raise ServerError(f"move.tv: login: invalid JSON response: {exc}") from exc if not data.get("success"): message = data.get("message", "Unknown login error") - raise RuntimeError(f"Move.tv auth failed: {message}") + # Rejected credentials are PERMANENT: no retry until they change. + raise CredentialsError(f"Move.tv auth failed: {message}") return self._create_token_from_response({**data, "uid": uid, "device_id": uid}) @@ -388,7 +416,10 @@ class MoveTVAuthenticator(BaseAuthenticator): def _full_login(self) -> MoveTVAuthToken: """Ensure credentials, run the login request, persist the session.""" if not self._ensure_credentials(): - raise Exception("No valid credentials available for move.tv") + raise AuthError( + "move.tv: no valid credentials available. " + "Set username and password in the addon settings." + ) token = self._perform_authentication() self._current_token = token diff --git a/lib/streaming_providers/providers/movetv/catchup_manager.py b/lib/streaming_providers/providers/movetv/catchup_manager.py new file mode 100644 index 0000000..fabf227 --- /dev/null +++ b/lib/streaming_providers/providers/movetv/catchup_manager.py @@ -0,0 +1,182 @@ +# streaming_providers/providers/movetv/catchup_manager.py +""" +move.tv catchup manager. + +The catchup source is requested per programme ("epgId"), so the manager +resolves the programme from the EPG (`epg_id` if the caller has it, else the +programme at `start_time`). It needs three collaborators besides the four +standard ones: the channel manager (channel records / per-channel window), the +EPG manager and the provider-owned SourceCache. + + * `end_time` is accepted and ignored (the API never takes an end bound). + * get_catchup_manifest returns None for "no catchup" (unknown channel, outside + the channel's window, no programme found, source refused); it NEVER falls + back to the live URL. + * catchup streams are clear: get_catchup_drm keeps the ABC default `[]`, which + ManagedProvider turns into NotImplementedError ("extract the PSSH from the + manifest") -- the legacy contract. +""" + +import threading +import time +from datetime import datetime, timezone +from typing import Any, Dict, List, Optional, Tuple + +from ...base.errors import NotFoundError +from ...base.managers import CatchupManager +from ...base.models.epg_models import EPGEntry +from ...base.utils.logger import logger +from .api import SourceCache, SourceInfo, api_post, fetch_source, is_fresh # noqa: F401 +from .channel_manager import MoveTvChannelManager +from .constants import MoveTVConfig +from .ids import parse_live_id + +_PROVIDER = MoveTVConfig.PROVIDER_LABEL + + +class MoveTvCatchupManager(CatchupManager): + def __init__( + self, + *, + http_manager: Any, + auth: Any, + country: str, + config: Any, + channels: MoveTvChannelManager, + epg: Any, + source_cache: SourceCache, + ): + super().__init__(http_manager=http_manager, auth=auth, country=country, config=config) + self._channels = channels + self._epg = epg + self._source_cache = source_cache + # EPG lookups used to resolve epg_id (rapid manifest polls): content_id -> (entries, expires_at) + self._epg_lookup: Dict[str, Tuple[List[EPGEntry], float]] = {} + self._lock = threading.Lock() + + # ------------------------------------------------------------------ + # Capability + # ------------------------------------------------------------------ + @property + def catchup_window_hours(self) -> int: + """Provider-wide MAXIMUM (constant; must cover every channel).""" + return self.config.CATCHUP_WINDOW_HOURS + + def catchup_window_for_channel(self, content_id: str) -> int: + record = self._channels.record(content_id) + return record.catchup_hours if record is not None else self.catchup_window_hours + + def clear(self) -> None: + with self._lock: + self._epg_lookup = {} + + # ------------------------------------------------------------------ + # EPG resolution + # ------------------------------------------------------------------ + def _epg_entries(self, content_id: str, start_time: int) -> List[EPGEntry]: + now = time.time() + with self._lock: + cached = self._epg_lookup.get(content_id) + if cached: + entries, expires_at = cached + if now < expires_at and entries: + min_start = min(e.start for e in entries) + max_end = max(e.end for e in entries) + tolerance = self.config.CATCHUP_EPG_TOLERANCE_SECONDS + if min_start - tolerance <= start_time <= max_end + tolerance: + return entries + entries = self._epg.get_epg( + content_id, + start_time=datetime.fromtimestamp(start_time, tz=timezone.utc), + backwards=self.config.CATCHUP_EPG_BACK_HOURS, + forwards=self.config.CATCHUP_EPG_FORWARD_HOURS, + ) + with self._lock: + self._epg_lookup[content_id] = (entries, now + self.config.EPG_LOOKUP_CACHE_TTL) + return entries + + def _resolve_epg_id(self, content_id: str, start_time: int) -> Optional[str]: + entries = self._epg_entries(content_id, start_time) + if not entries: + return None + closest: Optional[EPGEntry] = None + closest_diff = float("inf") + for entry in entries: + if entry.start <= start_time < entry.end: + return str(entry.program_id) + diff = abs(entry.start - start_time) + if diff < closest_diff: + closest, closest_diff = entry, diff + if closest is not None and closest_diff < self.config.CATCHUP_EPG_TOLERANCE_SECONDS: + return str(closest.program_id) + logger.warning(f"{_PROVIDER}: no suitable EPG entry for {content_id} at {start_time}") + return None + + # ------------------------------------------------------------------ + # Source + # ------------------------------------------------------------------ + def _catchup_source( + self, content_id: str, start_time: int, epg_id: Optional[str] + ) -> Optional[SourceInfo]: + channel_id = parse_live_id(content_id) # BadRequestError if malformed + record = self._channels.record(content_id, rebuild=True) + if record is None: + return None + # Per-channel window, enforced here (the backend only knows the global one). + if record.catchup_hours <= 0 or time.time() - start_time > record.catchup_hours * 3600: + return None + + resolved = epg_id or self._resolve_epg_id(content_id, start_time) + if not resolved: + return None + try: + epg_id_int = int(resolved) + except ValueError: + logger.error(f"{_PROVIDER}: invalid epg id {resolved!r} for {content_id}") + return None + + return fetch_source( + http_manager=self.http_manager, + session=self.auth, + cache=self._source_cache, + url=self.config.epg_source_url(), + build_payload=lambda: {"epgId": epg_id_int, "contentId": channel_id}, + cache_key=f"{content_id}:{resolved}", + operation="catchup_manifest", + what=f"catchup source for {content_id}:{resolved}", + ) + + # ------------------------------------------------------------------ + # Manifest / headers + # ------------------------------------------------------------------ + def get_catchup_manifest( + self, + content_id: str, + start_time: int, + end_time: Optional[int] = None, + epg_id: Optional[str] = None, + **kw: Any, + ) -> Optional[str]: + source = self._catchup_source(content_id, int(start_time), epg_id) + return source[0] if source else None + + def get_catchup_manifest_headers( + self, + content_id: str, + start_time: int, + end_time: Optional[int] = None, + epg_id: Optional[str] = None, + **kw: Any, + ) -> Dict[str, str]: + url = kw.get("url") or kw.get("manifest_url") or kw.get("segment_url") + if url: + cached = self._source_cache.get(url) + if is_fresh(cached): + return self.config.stream_headers(cached[1]) + source = self._catchup_source(content_id, int(start_time), epg_id) + if source is None: + # Never fall back to live headers (wrong token); fail loudly instead. + raise NotFoundError( + f"{_PROVIDER}: no catchup source for {content_id} at {start_time}" + ) + return self.config.stream_headers(source[1]) \ No newline at end of file diff --git a/lib/streaming_providers/providers/movetv/channel_manager.py b/lib/streaming_providers/providers/movetv/channel_manager.py new file mode 100644 index 0000000..318ab55 --- /dev/null +++ b/lib/streaming_providers/providers/movetv/channel_manager.py @@ -0,0 +1,186 @@ +# streaming_providers/providers/movetv/channel_manager.py +""" +move.tv channel manager (live channels, live manifest, stream headers). + + * get_channels() -> subscribed channels (one malformed entry + is skipped, the list survives) + * get_channel_manifest(id) -> live manifest URL + None = not mine (non-numeric id, or a + channel the account does not have) + * get_channel_manifest_headers -> CDN headers: User-Agent + X-Play-Auth + (the per-stream value that comes with + the manifest URL; never the API token) + +The manager owns what v1 kept on the provider: the channel records +(`live_id`, catchup hours, stream uid -- fields of the old MoveTVChannel +subclass) and uses the provider-owned SourceCache shared with the catchup +manager. No DRM: `get_channel_drm` keeps the ABC default `[]`. +""" + +import threading +from dataclasses import dataclass +from typing import Any, Dict, List, Optional + +from ...base.errors import NotFoundError, ServerError +from ...base.managers import ChannelManager +from ...base.models import Channel +from ...base.utils.logger import logger +from .api import SourceCache, SourceInfo, api_post, fetch_source, is_fresh +from .constants import MoveTVConfig +from .ids import is_live_id + +_PROVIDER = MoveTVConfig.PROVIDER_LABEL + + +@dataclass(frozen=True) +class ChannelRecord: + """What the API needs about a channel beyond the Channel model.""" + live_id: int + catchup_hours: int + stream_uid: str + + +class MoveTvChannelManager(ChannelManager): + def __init__( + self, + *, + http_manager: Any, + auth: Any, + country: str, + config: Any, + source_cache: Optional[SourceCache] = None, + ): + super().__init__(http_manager=http_manager, auth=auth, country=country, config=config) + # Provider-owned, borrowed by reference (shared with the catchup manager). + self._source_cache = source_cache if source_cache is not None else SourceCache() + self._records: Dict[str, ChannelRecord] = {} + self._lock = threading.Lock() + + # ------------------------------------------------------------------ + # Routing + # ------------------------------------------------------------------ + def handles_content_id(self, content_id: str) -> bool: + return is_live_id(content_id) + + # ------------------------------------------------------------------ + # Channels + # ------------------------------------------------------------------ + def get_channels(self, **kw: Any) -> List[Channel]: + data = api_post( + self.http_manager, self.auth, self.config.channels_url(), + lambda info: { + "customerId": info["customer_id"], + "customerProfileId": info["customer_profile_id"], + "lang": self.config.DEFAULT_LANG, + "appVersion": self.config.APP_VERSION, + }, + what="channel list", operation="api", + ) + if not data.get("success"): + raise ServerError(f"{_PROVIDER}: channels endpoint returned success=false") + + channels: List[Channel] = [] + records: Dict[str, ChannelRecord] = {} + for item in data.get("content") or []: + parsed = self._parse_item(item) + if parsed is None: + continue + channel, record = parsed + channels.append(channel) + records[channel.content_id] = record + with self._lock: + self._records = records + logger.info(f"{_PROVIDER}: loaded {len(channels)} subscribed channels") + return channels + + def _parse_item(self, item: Dict[str, Any]): + try: + if not item.get("subscribed", False): + return None + catalog_id = item.get("contentId") + live_id = item.get("liveId") + name = item.get("contentName", "") + if not live_id or not catalog_id or not name: + return None + + picture = item.get("picture") or {} + catchup = item.get("catchup") or {} + catchup_hours = int(catchup.get("duration", 0)) if catchup else 0 + is_audio = bool(item.get("audioOnly", False)) + + common = dict( + logo_url=self.config.build_logo_url(picture.get("icon")), + session_manifest=True, + language="sr", + country=self.country.upper(), # Content.country is upper-case + catchup_hours=catchup_hours, + ) + factory = Channel.create_radio_channel if is_audio else Channel.create_live_channel + channel = factory(name, str(catalog_id), self.config.PROVIDER_NAME, **common) + channel.channel_number = item.get("contentPosition") + record = ChannelRecord( + live_id=int(live_id), + catchup_hours=catchup_hours, + stream_uid=item.get("streamUid", "") or "", + ) + return channel, record + except (KeyError, TypeError, ValueError, AttributeError) as exc: + logger.warning(f"{_PROVIDER}: skipping malformed channel entry: {exc!r} -- {item}") + return None + + # ------------------------------------------------------------------ + # Records (also used by the catchup manager) + # ------------------------------------------------------------------ + def record(self, content_id: str, *, rebuild: bool = False) -> Optional[ChannelRecord]: + with self._lock: + found = self._records.get(content_id) + if found is None and rebuild: + logger.info(f"{_PROVIDER}: channel {content_id} unknown; reloading the channel list") + self.get_channels() + with self._lock: + found = self._records.get(content_id) + return found + + def clear(self) -> None: + """Account changed: forget channel records (provider callback).""" + with self._lock: + self._records = {} + + # ------------------------------------------------------------------ + # Live source (manifest + Play-Auth in ONE request, cached) + # ------------------------------------------------------------------ + def _live_source(self, content_id: str) -> Optional[SourceInfo]: + if not is_live_id(content_id): + return None + record = self.record(content_id, rebuild=True) + if record is None: + return None # not one of this account's channels + source = fetch_source( + http_manager=self.http_manager, + session=self.auth, + cache=self._source_cache, + url=self.config.live_source_url(), + build_payload=lambda: {"liveId": record.live_id}, + cache_key=content_id, + operation="live_manifest", + what=f"live source for channel {content_id}", + ) + if source is None: + raise ServerError(f"{_PROVIDER}: no live source for channel {content_id}") + return source + + def get_channel_manifest(self, content_id: str, **kw: Any) -> Optional[str]: + source = self._live_source(content_id) + return source[0] if source else None + + def get_channel_manifest_headers(self, content_id: str, **kw: Any) -> Dict[str, str]: + # Proxy/segment requests may hand the URL back: look it up first. + url = kw.get("url") or kw.get("manifest_url") or kw.get("segment_url") + if url: + cached = self._source_cache.get(url) + if is_fresh(cached): + return self.config.stream_headers(cached[1]) + source = self._live_source(content_id) + if source is None: + raise NotFoundError(f"{_PROVIDER}: no live source for channel {content_id}") + return self.config.stream_headers(source[1]) \ No newline at end of file diff --git a/lib/streaming_providers/providers/movetv/constants.py b/lib/streaming_providers/providers/movetv/constants.py index da1da96..15ddbe5 100644 --- a/lib/streaming_providers/providers/movetv/constants.py +++ b/lib/streaming_providers/providers/movetv/constants.py @@ -7,9 +7,19 @@ class MoveTVConfig: Central configuration for the move.tv / MTS-SI provider. All endpoint paths and fixed request parameters live here so that - provider.py, auth.py, and vod_manager.py never contain raw strings - or magic numbers. + provider.py, auth.py, and the managers never contain raw strings + or magic numbers. The class itself is the shared config object (static, + passed to auth and managers as `config=`). + + Identity rule: directory == registry key == provider_name == PROVIDER_NAME + == credentials key. Never rename after release. """ + PROVIDER_NAME: str = "movetv" + PROVIDER_LABEL: str = "move.tv" + # SUPPORTED_COUNTRIES semantics (README §Countries): exactly one entry -> + # the registry pins the construction country to it (UPPER-case, verbatim). + SUPPORTED_COUNTRIES: tuple = ("RS",) # Serbia + DEFAULT_COUNTRY: str = "RS" PROVIDER_LOGO: str = "https://m.media-amazon.com/images/I/41NLX+Ej5+L.png" # ------------------------------------------------------------------------- # Base URLs @@ -66,6 +76,30 @@ class MoveTVConfig: ) TIMEOUT: int = 30 + # ------------------------------------------------------------------------- + # Catchup / EPG / caches + # ------------------------------------------------------------------------- + # Provider-wide catchup window = the MAXIMUM over all channels (hours). + # Must be >= every channel's catchup.duration; the per-channel value is + # enforced by the catchup manager. + CATCHUP_WINDOW_HOURS: int = 168 # 7 days + EPG_WINDOW_DAYS: tuple = (3, 3) # (past_days, future_days) + EPG_DEFAULT_BACKWARDS_HOURS: int = 2 + EPG_DEFAULT_FORWARDS_HOURS: int = 2 + # Catchup resolves the programme from the EPG: a fixed 6 h backward window + # anchored at start_time covers long programmes (films, sports). + CATCHUP_EPG_BACK_HOURS: int = 6 + CATCHUP_EPG_FORWARD_HOURS: int = 2 + CATCHUP_EPG_TOLERANCE_SECONDS: int = 3600 + EPG_LOOKUP_CACHE_TTL: float = 60.0 # rapid manifest polls + SOURCE_CACHE_SIZE: int = 200 + PLAY_AUTH_EARLY_EXPIRY_SECONDS: int = 60 + PLAY_AUTH_EXPIRY_PATTERN: str = r"expires-(\d+)" + HEADER_PLAY_AUTH: str = "X-Play-Auth" + + # VOD: browse tag per content-type NAME (as in v1) + TAG_ID_BY_CONTENT_TYPE_NAME: Dict[str, int] = {"Film": 1, "Serija": 2} + # ------------------------------------------------------------------------- # Token sentinel: the API uses the literal string "null" for unauthenticated # requests instead of omitting the header entirely. @@ -123,6 +157,15 @@ class MoveTVConfig: """Headers for authenticated JSON API calls (channels, manifest source, VOD).""" return cls.get_base_headers(auth_token) + @classmethod + def stream_headers(cls, play_auth: Optional[str] = None) -> Dict[str, str]: + """Headers for manifest / segment fetches by the player (CDN): the + User-Agent and the per-stream X-Play-Auth value, never the API token.""" + headers = {"User-Agent": cls.USER_AGENT} + if play_auth is not None: + headers[cls.HEADER_PLAY_AUTH] = play_auth + return headers + # ------------------------------------------------------------------------- # Full endpoint URL helpers # ------------------------------------------------------------------------- diff --git a/lib/streaming_providers/providers/movetv/epg_manager.py b/lib/streaming_providers/providers/movetv/epg_manager.py index aa4cd0a..f9905f6 100644 --- a/lib/streaming_providers/providers/movetv/epg_manager.py +++ b/lib/streaming_providers/providers/movetv/epg_manager.py @@ -4,20 +4,17 @@ 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. +* An EpgManager subclass (README §6.3): the four standard collaborators, the + shared HTTP manager and the session (`auth`); no reference to the provider or + the raw authenticator. +* Requests go through `api_post` (one 401 recovery, typed errors): a failed + request RAISES instead of returning [] (v1 behaviour). "No programmes" + (success=false, empty content) is still a valid empty result. +* get_epg accepts the generic form (start_time/end_time datetimes, as passed by + EPGOperations) and the native form (backwards=/forwards= hours in **kw). + EPGOperations also passes `limit=`; like v1 it is ignored. +* The v1 helper `get_channels_epg` (no callers) is gone; `get_epg_grid` is the + batch entry point. """ from __future__ import annotations @@ -26,12 +23,16 @@ import math from datetime import datetime, timezone from typing import Any, Dict, List, Optional, Tuple -from ...base.utils.logger import logger +from ...base.errors import AuthError, GeoBlockError, ProviderError, RateLimitError +from ...base.managers import EpgManager from ...base.models.epg_models import EPGEntry, EPGFlags +from ...base.utils.logger import logger +from .api import api_post from .constants import MoveTVConfig +from .ids import is_live_id, parse_live_id -class MoveTvEpgManager: +class MoveTvEpgManager(EpgManager): """ Fetches and normalises EPG data from the MoveTV API into EPGEntry objects (the same typed contract used by EPGOperations and the other @@ -74,209 +75,101 @@ class MoveTvEpgManager: relying on this contract. """ - 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 + # Capability # ------------------------------------------------------------------ + @property + def epg_window(self) -> Tuple[int, int]: + """(past_days, future_days) -- same as v1.""" + return self.config.EPG_WINDOW_DAYS - def get_channel_epg( + def handles_channel_id(self, channel_id: str) -> bool: + return is_live_id(channel_id) + + # ------------------------------------------------------------------ + # EPG + # ------------------------------------------------------------------ + def get_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, + **kw: Any, ) -> List[EPGEntry]: """ - Fetch the EPG schedule for a single live channel. + Fetch the EPG schedule for ONE live channel. - Accepts two calling conventions — use whichever the caller has: + Generic form: start_time/end_time (aware or naive-local datetimes) are + translated to the API's backwards/forwards hour offsets around the + window midpoint. Native form: backwards=/forwards= (hours, default 2) + in **kw, anchored at start_time (or now). Other kwargs (limit=, ...) + are ignored. - Generic (EPGOperations contract): - get_channel_epg(channel_id, start_time=, end_time=) - 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 both start_time and end_time are provided. - forwards: - Hours of future programming to include (API default: 2). - Ignored when both start_time and end_time are provided. - start_time: - Inclusive window start (timezone-aware or naive-local datetime). - 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 and calculates a symmetric window. - - Returns - ------- - List of EPGEntry objects (see ``_parse_items``), or an empty list - on any error or for items that fail EPGEntry validation. + Returns [] when the API has no data; raises typed errors on failures. """ - # ---------------------------------------------------------------- - # Step 1 — resolve the time anchor and API window parameters - # ---------------------------------------------------------------- + backwards = int(kw.get("backwards", self.config.EPG_DEFAULT_BACKWARDS_HOURS)) + forwards = int(kw.get("forwards", self.config.EPG_DEFAULT_FORWARDS_HOURS)) + content_id = parse_live_id(channel_id) # BadRequestError if malformed 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() - except Exception as exc: - logger.error( - f"MoveTV EPG: Request failed for channel {channel_id} — {exc}" - ) - return [] - - # Decode JSON separately so a missing / malformed body doesn't - # collapse into a silent "Request failed" that hides the real cause. - try: - data: Dict[str, Any] = response.json() - except Exception as exc: - logger.error( - f"MoveTV EPG: Failed to decode JSON response for channel " - f"{channel_id} — {exc} " - f"(HTTP {response.status_code}, body={response.text[:200]!r})" - ) - return [] - - # The API may return a 200 with success=false, or (rarely) a null body - # parsed as None. Guard both cases before calling .get() on data. - if data is None: - logger.error( - f"MoveTV EPG: API returned null/empty JSON body for channel {channel_id}" - ) - return [] - + data = api_post( + self.http_manager, self.auth, self.config.epg_all_url(), + lambda info: { + "customerId": info["customer_id"], + "partnerId": self.config.PARTNER_ID, + "contentId": content_id, + "time": now_ms, + "backwards": backwards, + "forwards": forwards, + "appVersion": self.config.APP_VERSION, + }, + what=f"EPG for channel {channel_id}", operation="api", need_profile=False, + ) if not data.get("success"): logger.warning( - f"MoveTV EPG: API returned success=false for channel " - f"{channel_id}: {data}" + f"MoveTV EPG: API returned success=false for channel {channel_id}: {data}" ) return [] - items = data.get("content", []) - logger.debug( - f"MoveTV EPG: Received {len(items)} programme(s) for channel {channel_id}" - ) + items = data.get("content") or [] + logger.debug(f"MoveTV EPG: Received {len(items)} programme(s) for channel {channel_id}") return self._parse_items(items, channel_id) - def get_channels_epg( + def get_epg_grid( self, channel_ids: List[str], - backwards: int = 2, - forwards: int = 2, start_time: Optional[datetime] = None, end_time: Optional[datetime] = None, + **kw: Any, ) -> Dict[str, List[EPGEntry]]: - """ - Convenience wrapper: fetch EPG for multiple channels. + """Batch EPG: one request per channel (no native batch endpoint). - Returns a dict keyed by channel_id, each value a list of EPGEntry - objects. Missing / failed channels are present with an empty list - so callers don't have to guard KeyError. - """ + Auth / rate-limit / geo errors abort the whole grid (they hit every + channel alike); any other failure of ONE channel yields [] for it, so + one bad channel cannot take the grid down.""" result: Dict[str, List[EPGEntry]] = {} for cid in channel_ids: - result[cid] = self.get_channel_epg( - cid, - backwards=backwards, - forwards=forwards, - start_time=start_time, - end_time=end_time, - ) + if not self.handles_channel_id(cid): + result[cid] = [] + continue + try: + result[cid] = self.get_epg(cid, start_time=start_time, end_time=end_time, **kw) + except (AuthError, RateLimitError, GeoBlockError): + raise + except ProviderError as exc: + logger.warning(f"MoveTV EPG: channel {cid} skipped in grid: {exc}") + result[cid] = [] return result - # ------------------------------------------------------------------ - # Internal helpers - # ------------------------------------------------------------------ - @staticmethod def _resolve_window( start_time: Optional[datetime], diff --git a/lib/streaming_providers/providers/movetv/ids.py b/lib/streaming_providers/providers/movetv/ids.py new file mode 100644 index 0000000..0a8f4d1 --- /dev/null +++ b/lib/streaming_providers/providers/movetv/ids.py @@ -0,0 +1,55 @@ +# streaming_providers/providers/movetv/ids.py +""" +Id grammar of the move.tv provider -- the SINGLE owner (pure functions, no I/O). + +live : the numeric catalogue id of a channel, e.g. "1234" (str.isdigit()) +VOD : browse nodes only (no VOD playback exists, README §7): + "" root -> content types + "content_type_" content type -> categories + "category__" category -> items (paged) + VOD item ids are numeric too, but are never routed: the VOD manager + claims no ids (handles_content_id is False), so a numeric id is always + a live id for the router. +""" + +import re +from typing import Tuple, Union + +from ...base.errors import BadRequestError + +CONTENT_TYPE_PREFIX = "content_type_" +CATEGORY_PREFIX = "category_" + +_CONTENT_TYPE_RE = re.compile(r"^content_type_(\d+)$") +_CATEGORY_RE = re.compile(r"^category_(\d+)_(\d+)$") + + +def is_live_id(content_id: object) -> bool: + return isinstance(content_id, str) and content_id.isascii() and content_id.isdigit() + + +def parse_live_id(content_id: str) -> int: + if not is_live_id(content_id): + raise BadRequestError(f"move.tv: not a live channel id: {content_id!r}") + return int(content_id) + + +def content_type_node(content_type_id: int) -> str: + return f"{CONTENT_TYPE_PREFIX}{int(content_type_id)}" + + +def category_node(content_type_id: int, category_id: Union[int, str]) -> str: + return f"{CATEGORY_PREFIX}{int(content_type_id)}_{int(category_id)}" + + +def parse_vod_node(content_id: str) -> Tuple: + """-> ("root",) | ("content_type", ct) | ("category", ct, cat); BadRequestError otherwise.""" + if not content_id or content_id == "root": + return ("root",) + m = _CONTENT_TYPE_RE.match(content_id) + if m: + return "content_type", int(m.group(1)) + m = _CATEGORY_RE.match(content_id) + if m: + return "category", int(m.group(1)), int(m.group(2)) + raise BadRequestError(f"move.tv: unknown VOD node id: {content_id!r}") \ No newline at end of file diff --git a/lib/streaming_providers/providers/movetv/provider.py b/lib/streaming_providers/providers/movetv/provider.py index 98838aa..6ac0a21 100644 --- a/lib/streaming_providers/providers/movetv/provider.py +++ b/lib/streaming_providers/providers/movetv/provider.py @@ -1,673 +1,191 @@ # 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 +""" +move.tv (MTS-SI) provider -- orchestrator only. -from ...base.models import DRMConfig, StreamingChannel +The provider owns the SHARED resources (config, http_manager, authenticator, +session, SourceCache) and BUILDS the managers. Capability flags, content-id +routing, manifest / header / EPG / catchup delegation are inherited from +ManagedProvider. + +Managers: channels, epg, catchup (+ vod, see VOD_BROWSE_ENABLED). +No recordings / favorites / bookmarks / DRM (all clear streams). +""" + +from typing import ClassVar, Dict, List, Optional + +from ...base.errors import BadRequestError +from ...base.managed_provider import ManagedProvider from ...base.models.proxy_models import ProxyConfig -from ...base.models.epg_models import EPGEntry -from ...base.provider import StreamingProvider -from ...base.utils.logger import logger from .auth import MoveTVAuthenticator +from .catchup_manager import MoveTvCatchupManager +from .api import SourceCache +from .channel_manager import MoveTvChannelManager from .constants import MoveTVConfig - -from .vod_manager import ( - MoveTvVodManager, - VodFilters, - VodPage, -) - from .epg_manager import MoveTvEpgManager +from .session import MoveTvSession +from .vod_manager import MoveTvVodManager + +# Only the provider is public. Provider discovery takes the FIRST +# StreamingProvider subclass found in the package namespace. +__all__ = ["MoveTVProvider"] -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. - """ - _PLAY_AUTH_EARLY_EXPIRY_BUFFER: int = 60 - - def __init__(self, *args, live_id: int = 0, catchup_hours: int = 0, - stream_uid: str = "", play_auth_header: str = "", - play_auth_expires_at: float = 0.0, **kwargs): - super().__init__(*args, **kwargs) - self.live_id: int = live_id - self.catchup_hours: int = catchup_hours - self.stream_uid: str = stream_uid - self.play_auth_header: str = play_auth_header - self.play_auth_expires_at: float = play_auth_expires_at - - def is_play_auth_valid(self) -> bool: - 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) - - def to_dict(self) -> Dict: - result = super().to_dict() - result["LiveId"] = self.live_id - result["CatchupHours"] = self.catchup_hours - result["StreamUid"] = self.stream_uid - result["PlayAuthHeader"] = self.play_auth_header - result["PlayAuthExpiresAt"] = self.play_auth_expires_at - return result - - -class MoveTVProvider(StreamingProvider): - """ - Streaming provider for move.tv (MTS-SI platform). - """ - - PROVIDER_LABEL: ClassVar[str] = "move.tv" +class MoveTVProvider(ManagedProvider): + PROVIDER_LABEL: ClassVar[str] = MoveTVConfig.PROVIDER_LABEL + PROVIDER_LOGO: ClassVar[str] = MoveTVConfig.PROVIDER_LOGO SUPPORTED_AUTH_TYPES: ClassVar[List[str]] = ["user_credentials"] + SUPPORTED_COUNTRIES: ClassVar[List[str]] = list(MoveTVConfig.SUPPORTED_COUNTRIES) # Serbia: ["RS"] - _PLAY_AUTH_EXPIRY_RE = re.compile(r"expires-(\d+)") + # No DRM at all (get_drm was always []). + DRM_IN_MANAGERS: ClassVar[bool] = False + # The manifest request needs the per-stream X-Play-Auth header that comes with + # the manifest URL: the managers' header hooks provide it (v1 overrode the + # provider methods, so this is the same behaviour, now in the managers). + HEADERS_FROM_MANAGERS: ClassVar[bool] = True - def __init__( - self, - country: str = "SI", - config: Optional[Dict] = None, - proxy_config: Optional[ProxyConfig] = None, - settings_manager=None, - ): - super().__init__(country) - - self.http_manager = self._setup_http_manager( - provider_name="movetv", - proxy_config=proxy_config, - user_agent=MoveTVConfig.USER_AGENT, - timeout=MoveTVConfig.TIMEOUT, - ) - - self.authenticator = MoveTVAuthenticator( - proxy_config=proxy_config, - http_manager=self.http_manager, - settings_manager=settings_manager, - ) - - self._vod = MoveTvVodManager(self) - self._epg = MoveTvEpgManager(self.authenticator) - - self.http_manager = self._share_http_manager_with_authenticator(self.authenticator) - - # Unified thread-safe LRU cache for both Live and Catchup sources. - # Keyed by content_id (live) or content_id:epg_id (catchup). - # Also stores entries keyed by the content_url itself so segment/manifest - # header requests can look them up safely. - self._source_cache = _SourceCache(max_size=200) - - # EPG response cache to prevent hammering the API on rapid manifest polls - self._epg_cache: Dict[str, Tuple[List[EPGEntry], float]] = {} - self._epg_lock = threading.Lock() - - try: - self.authenticator.authenticate() - logger.info("move.tv: Authentication successful during initialisation") - except Exception as exc: - logger.warning(f"move.tv: Could not authenticate during initialisation: {exc}") - - # ------------------------------------------------------------------ - # StreamingProvider identity properties - # ------------------------------------------------------------------ + # The four VOD parsers are placeholders in the v1 source that was supplied + # for the migration (vod_manager.py). While False, the VOD manager is not + # registered as a capability (v1 never exposed VOD browsing through the + # backend either); the provider-specific get_vod_* methods below still work + # once the parsers are restored. Flip to True afterwards. + VOD_BROWSE_ENABLED: ClassVar[bool] = False @property def provider_name(self) -> str: - return "movetv" - - @property - def provider_label(self) -> str: - return self.PROVIDER_LABEL - - @property - def provider_logo(self) -> str: - return MoveTVConfig.PROVIDER_LOGO + """Must equal the directory name / registry key.""" + return MoveTVConfig.PROVIDER_NAME + # Not delegated by ManagedProvider (v1 values kept). @property def uses_dynamic_manifests(self) -> bool: return True - @property - def supported_auth_types(self) -> List[str]: - return self.SUPPORTED_AUTH_TYPES - @property def requires_manifest_context(self) -> bool: return True - @property - def catchup_window(self) -> int: - return 168 # 7 days + def __init__( + self, + country: str = MoveTVConfig.DEFAULT_COUNTRY, + config: Optional[Dict] = None, # accepted for compatibility, unused (as in v1) + proxy_config: Optional[ProxyConfig] = None, + settings_manager=None, + ): + # The registry constructs providers with `country` only (pinned to "RS"). + super().__init__(country=country) + # Credentials / sessions / proxy settings are keyed by lower-case country. + self.country = self.country.lower() - @property - def supports_catchup(self) -> bool: - return True + # 1. ONE static config (class), shared by every layer. + self.provider_config = MoveTVConfig - # ------------------------------------------------------------------ - # Internal Helpers - # ------------------------------------------------------------------ - - def _authenticated_headers(self) -> Dict[str, str]: - 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}") - - # Cache by the primary key (e.g. content_id or content_id:epg_id) - self._source_cache.set(cache_key, (content_url, header_value, expires_at)) - - # CRITICAL: Also cache by the content_url itself. This allows stateless - # proxy/segment header lookups if they pass the URL back to get_manifest_headers. - if content_url: - self._source_cache.set(content_url, (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]: - try: - channels = self._fetch_channels() - 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 …") - try: - token = self.authenticator.get_current_token() - if token and self.authenticator.refresh_token(token): - channels = self._fetch_channels() - self.channels = channels - return channels - except Exception: - pass - logger.info("move.tv: Refresh failed, retrying channel fetch after full login …") - else: - logger.error(f"move.tv: HTTP error fetching channels: {exc}") - try: - self.authenticator.authenticate(force_refresh=True) - channels = self._fetch_channels() - 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}") - return [] - except Exception as exc: - logger.error(f"move.tv: Unexpected error fetching channels: {exc}") - return [] - - def _fetch_channels(self) -> List[MoveTVChannel]: - session = self.authenticator.get_session_info() - if not session: - self.authenticator.authenticate(force_refresh=True) - session = self.authenticator.get_session_info() - if not session: - raise RuntimeError("move.tv: Unable to obtain session info for channel fetch") - - payload = { - "customerId": session["customer_id"], - "customerProfileId": session["customer_profile_id"], - "lang": MoveTVConfig.DEFAULT_LANG, - "appVersion": MoveTVConfig.APP_VERSION, - } - response = self.http_manager.post( - MoveTVConfig.channels_url(), - operation="api", - json=payload, - headers=self._authenticated_headers(), + # 2. HTTP manager (proxy: ctor arg -> ProxyConfigManager -> global). + self.http_manager = self._setup_http_manager( + provider_name=MoveTVConfig.PROVIDER_NAME, + proxy_config=proxy_config, + user_agent=MoveTVConfig.USER_AGENT, + timeout=MoveTVConfig.TIMEOUT, ) - response.raise_for_status() - data = response.json() - if not data.get("success"): - raise RuntimeError("move.tv: channels endpoint returned success=false") + # 3. Auth: the stateful authenticator (persisted session) behind a + # session adapter. LAZY: no network call in __init__ (v1 logged in here). + self.authenticator = MoveTVAuthenticator( + proxy_config=proxy_config, + http_manager=self.http_manager, + settings_manager=settings_manager, + ) + self.auth = MoveTvSession(authenticator=self.authenticator, config=self.provider_config) + self.auth.on_invalidate(self._clear_account_caches) - channels: List[MoveTVChannel] = [] - for item in data.get("content", []): - channel = self._parse_channel_item(item) - if channel: - channels.append(channel) - return channels + # 4. Provider-owned cache, borrowed by the channel and catchup managers. + self._source_cache = SourceCache() - def _parse_channel_item(self, item: Dict[str, Any]) -> Optional[MoveTVChannel]: - try: - if not item.get("subscribed", False): - return None - - catalog_id = item.get("contentId") - live_id = item.get("liveId") - name = item.get("contentName", "") - stream_uid = item.get("streamUid", "") - - if not live_id or not catalog_id or not name: - return None - - picture = item.get("picture", {}) - logo_url = MoveTVConfig.build_logo_url(picture.get("icon")) - - catchup: Dict = item.get("catchup", {}) - catchup_hours: int = int(catchup.get("duration", 0)) if catchup else 0 - - is_audio = bool(item.get("audioOnly", False)) - - channel = MoveTVChannel( - name=name, - content_id=str(catalog_id), - provider=self.provider_name, - logo_url=logo_url, - mode="live", - session_manifest=True, - manifest=None, - content_type="RADIO" if is_audio else "LIVE", - quality="AUDIO" if is_audio else None, - is_radio=is_audio, - language="sr", - country=self.country, - live_id=int(live_id), - catchup_hours=catchup_hours, - stream_uid=stream_uid, - ) - channel.channel_number = item.get("contentPosition") - return channel - - except Exception as exc: - logger.warning(f"move.tv: Error parsing channel item: {exc} — {item}") - return None + # 5. Managers. NOTE: self.channels is the ChannelManager (shadows the + # legacy list). + self._init_managers() # ------------------------------------------------------------------ - # Live Manifest + # Manager factories # ------------------------------------------------------------------ + def _build_channels(self): + return MoveTvChannelManager( + http_manager=self.http_manager, auth=self.auth, country=self.country, + config=self.provider_config, source_cache=self._source_cache, + ) - 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} + def _build_vod(self): + self._vod_catalogue = MoveTvVodManager( + http_manager=self.http_manager, auth=self.auth, country=self.country, + config=self.provider_config, + ) + return self._vod_catalogue if self.VOD_BROWSE_ENABLED else None - 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" - ) + def _build_epg(self): + return MoveTvEpgManager( + http_manager=self.http_manager, auth=self.auth, country=self.country, + config=self.provider_config, + ) - 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} + def _build_catchup(self): + # Build order: channels, vod, epg, ..., catchup -> both exist here. + return MoveTvCatchupManager( + http_manager=self.http_manager, auth=self.auth, country=self.country, + config=self.provider_config, channels=self.channels, epg=self.epg, + source_cache=self._source_cache, + ) - 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} + # ------------------------------------------------------------------ + # Credential change / logout: drop everything account-specific + # ------------------------------------------------------------------ + def _clear_account_caches(self) -> None: + self._source_cache.clear() + if self.channels is not None: + self.channels.clear() + if self.catchup is not None: + self.catchup.clear() - def get_manifest(self, content_id: str, **kwargs) -> Optional[str]: - url, _ = self.get_manifest_with_headers(content_id, **kwargs) - return url - - def get_manifest_headers(self, content_id: str, **kwargs) -> Dict[str, str]: - # 1. Check if URL is provided in kwargs (common for segment/manifest proxy requests) - url = kwargs.get("url") or kwargs.get("manifest_url") or kwargs.get("segment_url") - if url: - cached = self._source_cache.get(url) - if cached: - content_url, header_value, expires_at = cached - if expires_at and time.time() < (expires_at - MoveTVChannel._PLAY_AUTH_EARLY_EXPIRY_BUFFER): - return {"User-Agent": MoveTVConfig.USER_AGENT, "X-Play-Auth": header_value} - - start_time = kwargs.get("start_time") - end_time = kwargs.get("end_time") - epg_id = kwargs.get("epg_id") - - # 2. If start_time and end_time are present, this is a catchup manifest/segment request - if start_time is not None and end_time is not None: + # ------------------------------------------------------------------ + # Provider-specific grammar: catchup arguments on the manifest-headers call + # ------------------------------------------------------------------ + def get_manifest_headers(self, content_id: str, **kw) -> Dict[str, str]: + """v1 treated a headers request carrying start_time as a catchup request. + (ManagedProvider routes only get_segment_headers that way.)""" + if self.catchup is not None and kw.get("start_time") is not None: + rest = dict(kw) try: - # FIX: Strip start_time, end_time, and epg_id from kwargs before forwarding - # to prevent Python TypeError (multiple values for argument). - forward_kwargs = { - k: v for k, v in kwargs.items() - if k not in ("start_time", "end_time", "epg_id") - } - - if epg_id: - logger.debug(f"move.tv: Fetching catchup headers with provided epg_id={epg_id}") - else: - logger.debug(f"move.tv: epg_id missing in kwargs, will re-resolve for catchup headers") - - return self.get_catchup_manifest_headers( - content_id, int(start_time), int(end_time), epg_id, **forward_kwargs - ) - except ValueError: - logger.warning(f"move.tv: ValueError parsing catchup times for {content_id}") - except Exception as exc: - # Don't silently swallow TypeErrors or other exceptions here - logger.error(f"move.tv: Error getting catchup manifest headers: {exc}") - - # SAFETY NET: If catchup resolution failed for any reason, do NOT fall back - # to live headers. Return an empty auth header so the CDN rejects it - # loudly (401/403) rather than silently applying the wrong token. - logger.error(f"move.tv: Catchup header resolution failed for {content_id}. Returning safe empty headers.") - return {"User-Agent": MoveTVConfig.USER_AGENT} - - # 3. Fallback to live manifest headers (only if NOT a catchup request) - _, headers = self.get_manifest_with_headers(content_id, **kwargs) - return headers - - def get_segment_headers(self, content_id: str, **kwargs) -> Dict[str, str]: - return self.get_manifest_headers(content_id, **kwargs) - - def get_drm(self, content_id: str, **kwargs) -> List[DRMConfig]: - return [] - - # ------------------------------------------------------------------ - # Catchup Manifest - # ------------------------------------------------------------------ - - 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: - # 1. Resolve EPG ID - resolved_epg_id = epg_id - if not resolved_epg_id: - # Check EPG cache first to avoid hammering the API on repeated manifest polls - epg_entries = None - with self._epg_lock: - cached_epg = self._epg_cache.get(content_id) - if cached_epg: - entries, expires_at = cached_epg - if time.time() < expires_at and entries: - # Validate that start_time actually falls within the cached entries' covered range - min_start = min(e.start for e in entries) - max_end = max(e.end for e in entries) - # Add a 1-hour buffer to handle edge cases gracefully - if min_start - 3600 <= start_time <= max_end + 3600: - epg_entries = entries - logger.debug( - f"move.tv: EPG cache hit for {content_id} covering [{min_start}, {max_end}]") - - if epg_entries is None: - logger.debug(f"move.tv: EPG cache miss for {content_id} at {start_time}") - # 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), - ) - # Cache the EPG response for 60 seconds to handle rapid manifest polls - if epg_entries is not None: - with self._epg_lock: - self._epg_cache[content_id] = (epg_entries, time.time() + 60) - - 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" + start = int(rest.pop("start_time")) + except (TypeError, ValueError): + raise BadRequestError(f"move.tv: invalid start_time {kw.get('start_time')!r}") + return self.catchup.get_catchup_manifest_headers( + content_id, start, + end_time=rest.pop("end_time", None), + epg_id=rest.pop("epg_id", None), + **rest, ) - - 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 in get_catchup_manifest_with_headers: {exc}") - return None, {"User-Agent": MoveTVConfig.USER_AGENT} - - 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 - - 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 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 [] - - def get_catchup_window_for_channel(self, content_id: str) -> int: - channel = self._channel_by_id(content_id) - if channel and hasattr(channel, 'catchup_hours'): - return channel.catchup_hours - return self.catchup_window + return super().get_manifest_headers(content_id, **kw) # ------------------------------------------------------------------ - # VOD & EPG Delegations + # Provider-specific VOD catalogue (v1 surface, delegated 1:1) # ------------------------------------------------------------------ + def get_vod_filters(self): + return self._vod_catalogue.get_vod_filters() - def get_vod_filters(self) -> VodFilters: - 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): + return self._vod_catalogue.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): - 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_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_catalogue.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 self._vod.get_page_components(page_id) + return self._vod_catalogue.get_page_components(page_id) def get_component_items(self, component_id: int): - 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 self._epg.get_channel_epg( - channel_id, - backwards=kwargs.get("backwards", backwards), - forwards=kwargs.get("forwards", forwards), - **{k: v for k, v in kwargs.items() if k not in ("backwards", "forwards")}, - ) - - @property - def epg_window(self) -> Tuple[int, int]: - return 3, 3 \ No newline at end of file + return self._vod_catalogue.get_component_items(component_id) \ No newline at end of file diff --git a/lib/streaming_providers/providers/movetv/session.py b/lib/streaming_providers/providers/movetv/session.py new file mode 100644 index 0000000..95fe0db --- /dev/null +++ b/lib/streaming_providers/providers/movetv/session.py @@ -0,0 +1,177 @@ +# streaming_providers/providers/movetv/session.py +""" +AUTH VARIANT B -- AuthProtocol adapter in front of MoveTVAuthenticator +(README §9). Managers talk to this object; the raw authenticator stays on the +provider (`self.authenticator`). + + managers --(AuthProtocol)--> MoveTvSession --> MoveTVAuthenticator + +Differences to the v1 usage (each is a recorded behaviour change): + * the token is validated against the server ONCE per process (first login of + the stored session) and afterwards only when the server answers 401 + (`recover()`), instead of a validate POST in every authenticate() call; + * rejected credentials / missing credentials are PERMANENT (typed AuthError): + no retry until reset() -- no login loop, no account lock; + * any other typed ProviderError keeps its type and backs off (transient); + untyped errors become ServerError; + * `ensure()` is for the settings UI only; managers use require()/info(). + +Error contract: ensure() -> bool, never raises. require()/info()/recover() +raise typed errors. last_error keeps the ORIGINAL exception. +""" + +import threading +import time +from typing import Any, Callable, Dict, List, Optional, Tuple, Type + +from ...base.errors import AuthError, ProviderError, ServerError +from ...base.utils.logger import logger + +BACKOFF_BASE_SECONDS = 30 +BACKOFF_MAX_SECONDS = 900 + + +class MoveTvSession: + PERMANENT_ERRORS: Tuple[Type[Exception], ...] = (AuthError,) + + def __init__(self, *, authenticator: Any, config: Any, + clock: Callable[[], float] = time.monotonic): + self.authenticator = authenticator + self.config = config + self._clock = clock + self.last_error: Optional[Exception] = None + self._failure: Optional[Tuple[Type[ProviderError], str]] = None + self._logged_in = False + self._permanent = False + self._failures = 0 + self._retry_at = 0.0 + self._lock = threading.RLock() + self._on_invalidate: List[Callable[[], None]] = [] + + # ---- state --------------------------------------------------------- + def on_invalidate(self, callback: Callable[[], None]) -> None: + """Register a callback run by invalidate() (the provider clears its + account-specific caches there).""" + self._on_invalidate.append(callback) + + def reset(self) -> None: + """Credentials changed: forget everything, including the backoff.""" + self._logged_in = False + self.last_error = None + self._failure = None + self._permanent = False + self._failures = 0 + self._retry_at = 0.0 + + def _token(self) -> Any: + return self.authenticator.get_current_token() + + def _ready(self) -> bool: + token = self._token() + return bool( + self._logged_in and token is not None + and getattr(token, "access_token", None) and not token.is_expired + ) + + # ---- failure bookkeeping --------------------------------------------- + def _typed(self, exc: Exception) -> Tuple[Type[ProviderError], bool]: + if isinstance(exc, ProviderError): + return type(exc), isinstance(exc, self.PERMANENT_ERRORS) + return ServerError, False + + def _record_failure(self, exc: Exception) -> None: + kind, permanent = self._typed(exc) + self.last_error = exc + self._failure = (kind, f"authentication failed -- {exc}") + self._logged_in = False + self._permanent = permanent + if not permanent: + self._failures += 1 + delay = min(BACKOFF_BASE_SECONDS * 2 ** (self._failures - 1), BACKOFF_MAX_SECONDS) + self._retry_at = self._clock() + delay + logger.error(f"move.tv: {self._failure[1]}") + + def _record_success(self) -> None: + self._logged_in = True + self.last_error, self._failure = None, None + self._permanent = False + self._failures, self._retry_at = 0, 0.0 + + def _raise_failure(self) -> None: + kind, message = self._failure or (AuthError, "not authenticated") + try: + error: ProviderError = kind(message) + except TypeError: # exotic constructor shape + error = ServerError(message) + raise error from self.last_error + + # ---- gate ---------------------------------------------------------- + def ensure(self) -> bool: + if self._ready(): + return True + with self._lock: + if self._ready(): + return True + if self._permanent: + return False # until reset() + if self._clock() < self._retry_at: + return False # still backing off + return self._login() + + def require(self) -> None: + if self.ensure(): + return + self._raise_failure() + + def _login(self) -> bool: + try: + token = self.authenticator.authenticate() + if token is None: + raise AuthError("move.tv: authentication returned no token") + except Exception as exc: + self._record_failure(exc) + return False + self._record_success() + return True + + def recover(self) -> None: + """The server rejected the token (401): refresh it, else log in again. + Raises typed errors; a failed full login is permanent (AuthError).""" + with self._lock: + try: + token = self._token() + renewed = self.authenticator.refresh_token(token) if token else None + if renewed is None: + renewed = self.authenticator.authenticate(force_refresh=True) + if renewed is None: + raise AuthError("move.tv: re-authentication returned no token") + except Exception as exc: + self._record_failure(exc) + self._raise_failure() + self._record_success() + + def info(self, *, need_profile: bool = True) -> Dict[str, Any]: + """Session data the API payloads need (customer ids, token).""" + self.require() + info = self.authenticator.get_session_info() + if (not info or not info.get("customer_id") or not info.get("auth_token") + or (need_profile and not info.get("customer_profile_id"))): + raise AuthError("move.tv: incomplete session data") + return info + + # ---- AuthProtocol -------------------------------------------------- + def get_access_token(self) -> str: + self.require() + return self._token().access_token + + def build_headers(self) -> Dict[str, str]: + return self.config.get_api_headers(self.get_access_token()) + + def invalidate(self) -> None: + self.authenticator.invalidate_token() # memory AND persisted storage + self.reset() + for callback in list(self._on_invalidate): + try: + callback() + except Exception as exc: # a cache-clearing bug must not block logout + logger.warning(f"move.tv: on_invalidate callback failed: {exc}") \ No newline at end of file diff --git a/lib/streaming_providers/providers/movetv/vod_manager.py b/lib/streaming_providers/providers/movetv/vod_manager.py index 8350597..b480042 100644 --- a/lib/streaming_providers/providers/movetv/vod_manager.py +++ b/lib/streaming_providers/providers/movetv/vod_manager.py @@ -1,48 +1,52 @@ # streaming_providers/providers/movetv/vod_manager.py """ -MoveTV VOD Manager +move.tv VOD manager (browse only -- the API has no logged VOD playback yet). -Handles all VOD-related API interactions for the MoveTV / MTS provider: - - get_vod_filters() → available content types, categories, catalogs, tags, sort options - - get_vod_items() → paginated catalogue listing with filtering / sorting - - get_all_vod_items() → transparent pagination helper - - get_page_components() → homepage / page layout (ordered list of component descriptors) - - get_component_items() → items inside a single carousel / banner component - - get_vod_item() → (stub — implement when detail endpoint is logged) - - get_vod_category() → (stub — implement when browse endpoint is logged) + Catalogue (provider-specific, delegated 1:1 by the provider as in v1): + get_vod_filters() -> VodFilters + get_vod_items(...) -> VodListing (page / has_next_page) + get_all_vod_items(...) -> List[VodItem] (pages transparently) + get_page_components() -> homepage layout descriptors + get_component_items() -> cards of one carousel -The manager is intentionally decoupled from the provider class so that -provider.py only needs a single delegation call per public VOD method. + VodManager contract (README §6.2): + get_vod_category(content_id, cursor, page_size) -> VodPage + "" root -> content types ("content_type_") + "content_type_" -> categories ("category__") + "category__" -> items; cursor = next page number, only while + the API says has_next_page (the endpoint + honours `page`); the page size is fixed by + the API, page_size is ignored + get_vod_manifest -> None, handles_content_id -> False: there is no VOD + playback, so no id is ever routed here (ids.py). -Minimal provider.py wiring: - from .vod_manager import MoveTvVodManager - - class MoveTvProvider(StreamingProvider): - def __init__(self, ...): - ... - self._vod = MoveTvVodManager(self) - - # --- VOD catalogue --- - def get_vod_filters(self): return self._vod.get_vod_filters() - def get_vod_items(self, **kw): return self._vod.get_vod_items(**kw) - def get_all_vod_items(self, **kw): return self._vod.get_all_vod_items(**kw) - - # --- Homepage / pages --- - def get_page_components(self, page_id): return self._vod.get_page_components(page_id) - def get_component_items(self, cid): return self._vod.get_component_items(cid) +!! The four parsers below (_parse_filters, _parse_vod_page, +!! _parse_page_components, _parse_component_items) were `pass` in the v1 file +!! that was supplied for the migration. They now raise NotImplementedYetError +!! instead of returning None. Paste the real bodies back in before enabling +!! VOD (MoveTVProvider.VOD_BROWSE_ENABLED). """ from __future__ import annotations from dataclasses import dataclass, field -from typing import TYPE_CHECKING, Any, Dict, List, Optional, Union +from typing import Any, Dict, List, Optional -from ...base.utils.logger import logger +from ...base.errors import BadRequestError, NotImplementedYetError +from ...base.managers import VodManager from ...base.models.vod import VodCategory, VodItem +from ...base.utils.logger import logger +from ...base.vod import VodPage +from .api import api_post from .constants import MoveTVConfig +from .ids import category_node, content_type_node, parse_vod_node -if TYPE_CHECKING: - from .provider import MoveTVProvider +_NAME = MoveTVConfig.PROVIDER_NAME + +_MISSING_PARSER = ( + "move.tv: {name} has no implementation (the file supplied for the " + "migration contained only a placeholder); restore the v1 parser body." +) # --------------------------------------------------------------------------- @@ -118,8 +122,11 @@ class VodFilters: @dataclass -class VodPage: - """Paginated result of POST /api/v2/content/vod/get/all.""" +class VodListing: + """Paginated result of POST /api/v2/content/vod/get/all. + + (v1 name: VodPage -- renamed, it collided with the base VodPage that + get_vod_category returns.)""" items: List[VodItem] current_page: int has_next_page: bool @@ -312,122 +319,51 @@ class ComponentItem: # Manager # --------------------------------------------------------------------------- -class MoveTvVodManager: - """ - Handles all VOD API calls for the MoveTV provider. +class MoveTvVodManager(VodManager): + """VOD browse for move.tv. The four standard collaborators only: no + reference to the provider (v1 took `provider`).""" - Accesses the parent provider only through well-defined attributes: - provider.http_manager -- shared HTTPManager instance - provider.authenticator -- for authentication and session info - provider.provider_name -- used in log messages - """ + # ------------------------------------------------------------------ + # Routing / playback (none) + # ------------------------------------------------------------------ + def handles_content_id(self, content_id: str) -> bool: + return False # no VOD playback: never routed - def __init__(self, provider: "MoveTVProvider") -> None: - self._provider = provider + def get_vod_manifest(self, content_id: str, **kw: Any) -> Optional[str]: + return None # ------------------------------------------------------------------ # Internal helpers # ------------------------------------------------------------------ - - @property - def _http(self): - """Shortcut to the provider's HTTP manager (asserted to exist).""" - mgr = self._provider.http_manager - if mgr is None: - raise RuntimeError( - "MoveTvVodManager: HTTP manager is not initialised. " - "Call provider.authenticate() / provider.setup() first." - ) - 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.""" - 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. - - Fetches customer_profile_id fresh each time to ensure it's always - current (supports token refresh scenarios). - """ - session = self._ensure_session() - return { - "customerProfileId": session["customer_profile_id"], - "lang": MoveTVConfig.DEFAULT_LANG, - "appVersion": MoveTVConfig.APP_VERSION, - } - - def _post(self, url: str, payload: Dict[str, Any]) -> Dict: - """ - Execute a POST request and return the parsed JSON body. - Raises on network / JSON errors after logging them. - """ - try: - response = self._http.post(url, json=payload, headers=self._auth_headers()) - return response.json() - except Exception as exc: - logger.error(f"{self._provider.provider_name}: POST {url} failed -- {exc}") - raise + def _post(self, url: str, extra: Dict[str, Any], *, what: str) -> Dict: + """POST with the common VOD fields (customerProfileId, lang, appVersion), + re-read from the session on every attempt (token refresh safe).""" + return api_post( + self.http_manager, self.auth, url, + lambda info: { + "customerProfileId": info["customer_profile_id"], + "lang": self.config.DEFAULT_LANG, + "appVersion": self.config.APP_VERSION, + **extra, + }, + what=what, operation="api", + ) # ------------------------------------------------------------------ - # Parsing helpers (keep these exactly as they were) + # Parsing helpers -- BODIES MISSING IN THE SUPPLIED v1 FILE (see docstring) # ------------------------------------------------------------------ - def _parse_filters(self, data: Dict) -> VodFilters: - # ... unchanged ... - pass + raise NotImplementedYetError(_MISSING_PARSER.format(name="_parse_filters")) - def _parse_vod_page(self, data: Dict, fallback_page: int) -> VodPage: - # ... unchanged ... - pass + def _parse_vod_page(self, data: Dict, fallback_page: int) -> VodListing: + raise NotImplementedYetError(_MISSING_PARSER.format(name="_parse_vod_page")) @staticmethod def _parse_page_components(data: Dict) -> List[PageComponent]: - # ... unchanged ... - pass + raise NotImplementedYetError(_MISSING_PARSER.format(name="_parse_page_components")) def _parse_component_items(self, data: Dict) -> List[ComponentItem]: - # ... unchanged ... - pass + raise NotImplementedYetError(_MISSING_PARSER.format(name="_parse_component_items")) # ------------------------------------------------------------------ # Public API -- VOD catalogue @@ -445,17 +381,17 @@ class MoveTvVodManager: VodFilters with content_types, categories, catalogs, tags, sort_types. """ url = MoveTVConfig.vod_filters_url() - logger.debug(f"{self._provider.provider_name}: Fetching VOD filters") + logger.debug(f"{_NAME}: Fetching VOD filters") - data = self._post(url, self._base_payload()) + data = self._post(url, {}, what="VOD filters") if not data.get("success"): - logger.warning(f"{self._provider.provider_name}: VOD filters -- success=false") + logger.warning(f"{_NAME}: VOD filters -- success=false") return VodFilters() filters = self._parse_filters(data) logger.info( - f"{self._provider.provider_name}: VOD filters -- " + f"{_NAME}: VOD filters -- " f"{len(filters.content_types)} content types, " f"{len(filters.categories)} categories, " f"{len(filters.catalogs)} catalogs, " @@ -472,7 +408,7 @@ class MoveTvVodManager: catalog_id: Optional[int] = None, content_type_id: Optional[int] = None, search_query: Optional[str] = None, - ) -> VodPage: + ) -> VodListing: """ Fetch a paginated list of VOD items with optional filtering. @@ -498,10 +434,10 @@ class MoveTvVodManager: search_query: Free-text search string. Returns: - VodPage(items, current_page, has_next_page). + VodListing(items, current_page, has_next_page). """ url = MoveTVConfig.vod_get_all_url() - payload: Dict[str, Any] = {**self._base_payload(), "page": page, "sort": sort} + payload: Dict[str, Any] = {"page": page, "sort": sort} # Only include filter keys that were explicitly provided -- # omitting them returns all content. @@ -517,20 +453,20 @@ class MoveTvVodManager: payload["search"] = search_query logger.debug( - f"{self._provider.provider_name}: VOD get/all " + f"{_NAME}: VOD get/all " f"page={page} sort={sort} " f"tag={tag_id} cat={category_id} catalog={catalog_id}" ) - data = self._post(url, payload) + data = self._post(url, payload, what=f"VOD page {page}") if not data.get("success"): - logger.warning(f"{self._provider.provider_name}: VOD get/all -- success=false") - return VodPage(items=[], current_page=page, has_next_page=False) + logger.warning(f"{_NAME}: VOD get/all -- success=false") + return VodListing(items=[], current_page=page, has_next_page=False) vod_page = self._parse_vod_page(data, fallback_page=page) logger.info( - f"{self._provider.provider_name}: VOD page {vod_page.current_page} -- " + f"{_NAME}: VOD page {vod_page.current_page} -- " f"{len(vod_page.items)} items, next={vod_page.has_next_page}" ) return vod_page @@ -576,13 +512,13 @@ class MoveTvVodManager: break if max_pages is not None and page >= max_pages: logger.debug( - f"{self._provider.provider_name}: VOD pagination capped at {max_pages} pages" + f"{_NAME}: VOD pagination capped at {max_pages} pages" ) break page += 1 logger.info( - f"{self._provider.provider_name}: VOD full fetch -- " + f"{_NAME}: VOD full fetch -- " f"{len(all_items)} items across {page} page(s)" ) return all_items @@ -612,23 +548,23 @@ class MoveTvVodManager: all components for the requested page arrive in a single response. """ url = MoveTVConfig.page_get_url() - payload = {**self._base_payload(), "page": page_id} + payload = {"page": page_id} logger.debug( - f"{self._provider.provider_name}: Fetching page components page_id={page_id}" + f"{_NAME}: Fetching page components page_id={page_id}" ) - data = self._post(url, payload) + data = self._post(url, payload, what=f"page {page_id}") if not data.get("success"): logger.warning( - f"{self._provider.provider_name}: page/get page_id={page_id} -- success=false" + f"{_NAME}: page/get page_id={page_id} -- success=false" ) return [] components = self._parse_page_components(data) logger.info( - f"{self._provider.provider_name}: Page {page_id} -- {len(components)} components" + f"{_NAME}: Page {page_id} -- {len(components)} components" ) return components @@ -651,171 +587,90 @@ class MoveTvVodManager: Use ComponentItem.as_vod_item() to convert non-live cards to VodItem. """ url = MoveTVConfig.component_get_url() - payload = {**self._base_payload(), "componentId": component_id} + payload = {"componentId": component_id} logger.debug( - f"{self._provider.provider_name}: Fetching component items " + f"{_NAME}: Fetching component items " f"component_id={component_id}" ) - data = self._post(url, payload) + data = self._post(url, payload, what=f"component {component_id}") if not data.get("success"): logger.warning( - f"{self._provider.provider_name}: component/get " + f"{_NAME}: component/get " f"id={component_id} -- success=false" ) return [] items = self._parse_component_items(data) logger.info( - f"{self._provider.provider_name}: Component {component_id} -- {len(items)} items" + f"{_NAME}: Component {component_id} -- {len(items)} items" ) return items # ------------------------------------------------------------------ - # Stubs for endpoints not yet logged + # VodManager: browse tree # ------------------------------------------------------------------ - - def get_vod_item(self, content_id: str, **kwargs) -> Optional[VodItem]: - """ - Fetch full metadata for a single VOD item (manifest, cast, DRM config, ...). - - Not yet implemented -- log the browser request when clicking a title - to play it, then implement here. Expected pattern: - POST /api/v2/content/vod/get/{content_id} - or /api/v2/content/vod/details { contentId, ... } - """ - logger.warning( - f"{self._provider.provider_name}: get_vod_item({content_id}) -- " - "not yet implemented; log the detail/play endpoint to add support." - ) - return None - def get_vod_category( - self, path_ids: List[str], **kwargs - ) -> List[Union[VodCategory, VodItem]]: - """ - Fetch children of a VOD category node. + self, + content_id: str = "", + cursor: Optional[str] = None, + page_size: int = 24, + **kw: Any, + ) -> VodPage: + node = parse_vod_node(content_id) # BadRequestError on a foreign id - Path structure: - [] or ["root"] → Return content types (Film, Serija) - ["root", "Film"] → Return categories for movies - ["root", "Serija"] → Return categories for series - ["root", "Film", "123"] → Return VOD items for that category - ["root", "Serija", "456"] → Return VOD items for that series category - """ - - # Handle root level - return content types - if not path_ids or path_ids[0] == "root": - # Fetch filters to get content types + if node[0] == "root": filters = self.get_vod_filters() - - categories = [] - for content_type in filters.content_types: - # Map content type ID to a display name - type_name = content_type.name # "Film" or "Serija" - - categories.append( - VodCategory( - name=type_name, - content_id=f"content_type_{content_type.content_type_id}", - provider=self._provider.provider_name, - # Store the content_type_id in the path for next level - # The UI will pass this as path_ids[1] - ) + entries = [ + VodCategory( + name=ct.name, + content_id=content_type_node(ct.content_type_id), + provider=_NAME, ) + for ct in filters.content_types + ] + logger.info(f"{_NAME}: VOD root -> {len(entries)} content types") + return VodPage(entries=entries) - logger.info( - f"{self._provider.provider_name}: VOD root -> " - f"{len(categories)} content types" - ) - return categories + content_type_id = node[1] + filters = self.get_vod_filters() + content_type = next( + (ct for ct in filters.content_types if ct.content_type_id == content_type_id), None + ) + if content_type is None: + logger.warning(f"{_NAME}: unknown content type {content_type_id}") + return VodPage() - # Handle content type level (e.g., ["root", "Film"] or ["root", "Serija"]) - if len(path_ids) == 2 and path_ids[0] == "root": - content_type_name = path_ids[1] - - # Fetch filters to get categories for this content type - filters = self.get_vod_filters() - - # Find the content type ID - content_type_id = None - for ct in filters.content_types: - if ct.name == content_type_name: - content_type_id = ct.content_type_id - break - - if not content_type_id: - logger.warning( - f"{self._provider.provider_name}: Unknown content type '{content_type_name}'" - ) - return [] - - # Return categories/genres for this content type - categories = [] + if node[0] == "content_type": + entries = [] for category in filters.categories: - # Create a category that can be browsed further - categories.append( + if not str(category.content_id).isdigit(): + logger.warning(f"{_NAME}: skipping category with non-numeric id: {category}") + continue + entries.append( VodCategory( name=category.name, - content_id=str(category.content_id), - provider=self._provider.provider_name, - # The path will be ["root", content_type_name, category_id] + content_id=category_node(content_type_id, category.content_id), + provider=_NAME, ) ) + logger.info(f"{_NAME}: VOD content type '{content_type.name}' -> {len(entries)} categories") + return VodPage(entries=entries) - logger.info( - f"{self._provider.provider_name}: VOD content type '{content_type_name}' -> " - f"{len(categories)} categories" - ) - return categories - - # Handle category level - return VOD items - if len(path_ids) >= 3 and path_ids[0] == "root": - content_type_name = path_ids[1] - category_id = path_ids[2] - - # Fetch filters to get content type ID - filters = self.get_vod_filters() - content_type_id = None - for ct in filters.content_types: - if ct.name == content_type_name: - content_type_id = ct.content_type_id - break - - if not content_type_id: - return [] - - # Map content type name to tag ID (optional filtering) - tag_map = { - "Film": 1, # FILM tag - "Serija": 2, # SERIJA tag - } - tag_id = tag_map.get(content_type_name) - - # Fetch VOD items for this category - vod_page = self.get_vod_items( - page=1, - sort="newest", - tag_id=tag_id, - category_id=int(category_id) if category_id != "root" else None, - content_type_id=content_type_id, - ) - - # Convert VodItems to the expected return format - # (VodOperations likely expects VodItem or similar) - items = [] - for item in vod_page.items: - items.append(item) - - logger.info( - f"{self._provider.provider_name}: VOD category '{category_id}' -> " - f"{len(items)} items" - ) - return items - - logger.warning( - f"{self._provider.provider_name}: Unhandled VOD path: {path_ids}" + # node[0] == "category": items, paged by the API's `page` + try: + page = int(cursor) if cursor else 1 + except ValueError: + raise BadRequestError(f"move.tv: invalid VOD cursor {cursor!r}") + listing = self.get_vod_items( + page=page, + sort="newest", + tag_id=self.config.TAG_ID_BY_CONTENT_TYPE_NAME.get(content_type.name), + category_id=node[2], + content_type_id=content_type_id, ) - return [] \ No newline at end of file + next_cursor = str(listing.current_page + 1) if listing.has_next_page else None + logger.info(f"{_NAME}: VOD category {node[2]} page {page} -> {len(listing.items)} items") + return VodPage(entries=list(listing.items), next_cursor=next_cursor) \ No newline at end of file