nirvana-7777/script.service.ultimate
1
0
Fork 0
mirror of https://github.com/nirvana-7777/script.service.ultimate.git synced 2026-10-13 19:32:43 +02:00
Code Issues Packages Projects Releases Wiki Activity

joyn vod

Browse Source
This commit is contained in:
Nirvana
2026-10-12 09:55:01 +02:00
parent 82b084569e
commit 099b359acd
7 changed files with 1097 additions and 180 deletions
Download Patch File Download Diff File Expand all files Collapse all files
lib/streaming_providers/providers/joyn/channel_manager.py
+36 -25
View File
@@ -10,17 +10,19 @@ two declared extras:
get_channel_manifest and get_channel_drm share ONE
playlist call per zap
Headers come from `self.config`; the bearer comes from
`self.auth.get_access_token()`. Never through the provider.
Headers come from `self.config`. The channel manager's GraphQL header builder
tolerates an absent session (see `_graphql_headers`): the LiveChannelsAndEpg
operation is served anonymously (joyn-user-state=A_A) as well as authenticated
(R_A), and there is no reason a broken login should hide the channel list. The
VOD manager uses a stricter builder for user-scoped queries; the two
deliberately differ.
Routing: handles_content_id() is the cheap pure pre-filter (README §7). It is the
mirror image of JoynVodManager.handles_content_id — both call ids.is_vod_id() —
so a VOD id is never handed to this manager first.
"""
import json
import time
import urllib.parse
from typing import Any, Dict, List, Optional, Tuple
from ...base.errors import BadRequestError
@@ -37,9 +39,6 @@ from .constants import (
GRAPHQL_LIVE_CHANNELS_FILTER,
GRAPHQL_MAX_RESULTS,
GRAPHQL_OFFSET,
GRAPHQL_PERSISTED_QUERY_VERSION,
GRAPHQL_QUERY_HASHES,
JOYN_GRAPHQL_ENDPOINTS,
JOYN_STREAMING_ENDPOINTS,
MODE_LIVE,
MODE_VOD,
@@ -49,6 +48,7 @@ from .constants import (
from .drm import build_widevine_config
from .ids import is_live_id, is_vod_id
from .models import JoynChannel, JoynPlayout
from .graphql import build_url
from .signing import ( # noqa: F401 (re-exported: external callers imported these from here)
build_signature,
create_video_payload,
@@ -88,13 +88,34 @@ class JoynChannelManager(ChannelManager):
# ------------------------------------------------------------------
def _graphql_headers(self) -> Dict[str, str]:
# get_access_token() RAISES a typed error if no session can be had
# (README §8). An anonymous token is a normal, valid result when no
# credentials are stored; a permanent failure (MFA, rejected credentials)
# must surface here, not silently degrade to an anonymous channel list
# that then fails at playback.
token = self.auth.get_access_token()
return self.config.graphql_headers(token=token, authenticated=True)
"""
GraphQL headers for the channel listing.
Deliberately tolerant of an absent session:
* LiveChannelsAndEpg is served to anonymous callers
(joyn-user-state=A_A). It is a public catalogue, not a
user-scoped resource.
* A permanent login failure (MFA, rejected credentials) must NOT
stop the channel list: there is no reason a user who cannot
log in should be unable to see what is on. The failure
surfaces on the first authenticated call (VOD, entitlement,
playback), where the user can act on it.
* A *transient* login failure (network, WAF) also does not block
the listing: `ensure()` returns False and the request goes out
anonymously, and the session's backoff keeps the next attempt
from hammering the login endpoint.
The VOD manager's `_get_graphql_headers(authenticated=True)` uses
the opposite policy — a user-scoped query is meaningless without a
session and raises typed errors. The two policies are intentional.
"""
token = None
if self.auth.ensure():
token = self.auth.get_access_token()
return self.config.graphql_headers(
token=token, authenticated=token is not None
)
def get_channel_manifest_headers(self, content_id: str, **kw: Any) -> Dict[str, str]:
# CDN headers: no Authorization. The manifest URL carries its own
@@ -192,17 +213,7 @@ class JoynChannelManager(ChannelManager):
"from": current_time,
"to": end_time,
}
variables_encoded = urllib.parse.quote(json.dumps(variables))
extensions_encoded = urllib.parse.quote(json.dumps({
"persistedQuery": {
"version": GRAPHQL_PERSISTED_QUERY_VERSION,
"sha256Hash": GRAPHQL_QUERY_HASHES["LIVE_CHANNELS"],
}
}))
url = (
f"{JOYN_GRAPHQL_ENDPOINTS['LIVE_CHANNELS']}"
f"&variables={variables_encoded}&extensions={extensions_encoded}"
)
url = build_url("LIVE_CHANNELS", variables)
response = self.http_manager.get(
url, operation="api", headers=headers,
timeout=DEFAULT_REQUEST_TIMEOUT,
lib/streaming_providers/providers/joyn/constants.py
+56 -9
View File
@@ -92,19 +92,66 @@ JOYN_API_BASE_HEADERS = {
JOYN_GRAPHQL_BASE_URL = "https://api.joyn.de/graphql"
# GraphQL persisted query hashes
# Persisted-query registry. Every GraphQL operation the provider uses lives
# here, hash and operation name side by side, so a reader sees both at once
# and they cannot drift apart. Callers name an operation by its key; the
# URL builder (graphql.py) reads both dicts.
#
# Adding a new operation:
# 1. Capture the request in a browser (DevTools → Network → the GraphQL call)
# 2. Copy the sha256Hash from the `extensions` query parameter here
# 3. Copy the operationName (or the operationName query parameter) here
# 4. Note where the hash came from, in a comment, so a future maintainer
# can tell whether it is a captured value or a legacy const.py import.
GRAPHQL_QUERY_HASHES = {
"LIVE_PLAYER": "52b37a3cf5bc75e56026aed7b0d234874eeabd2eccd369d0cd3d3a6ea15ef566",
"LIVE_CHANNELS": "b7703103ddd0516be6b49ed66186092a6c6f6d815ccc502a9f50800a8cc18dd2",
# --- Live TV ---
"LIVE_PLAYER": "52b37a3cf5bc75e56026aed7b0d234874eeabd2eccd369d0cd3d3a6ea15ef566",
"LIVE_CHANNELS": "b7703103ddd0516be6b49ed66186092a6c6f6d815ccc502a9f50800a8cc18dd2",
"LIVESTREAM_BY_BRAND": "333399353bead5c153684c1c326504e82e7494ae2e89ca1d11d4ad76a16a839e",
"LIVESTREAM_OVERVIEW": "5a094812b90747ed18d2eac87f8b29a82e403a02f329d03478f7b2a5cf6ca463",
"PLAYABLE_LIVESTREAM": "2057248fde6948eebc02b03bb4244a1361707c96ed8bab25027d1f3a80ea0cb8",
# --- VOD ---
"NAVIGATION": "818622ff5afe143664241034ba0c537650ec9a2eeae4d568830b47d8de605b7e",
"LANDING_PAGE": "5838ac6b328cda91a70da264d5be09d4735cf95de8d9ee00cec60e77fcc3df08",
"LANDING_BLOCKS": "1655591f83b0dc1508ad4d52c5f37f72d410f48ad08c3e5f2de8622f86a21c68",
"GET_ME_STATE": "55ebb3812b45628017ee6c7f36f0b88a94e9778b9f11ad8a6fc05849182c07ec",
"LIVE_LANE": "51659c62d4e4a6628d1e512190a3b0659486478b12be494875bef5a83dcb79ed",
"HERO_RESUME": "d3b7e480f593ba4866598f8cfe95185b3e400fcff112c026dc4e1b5ad4b0d537",
"COLLECTION_QUERY": "bdf4e08de65351750eefb2165a58af50c9e4b3526b78cd75e2066df2bc7ec8d8",
"PAGE_OVERVIEW_GENRE": "37ba6d0dde470df3f8999d49bcd24bc5c72b8e7192768026d82447c664c6ab7f",
"SEASON": "ee2396bb1b7c9f800e5cefd0b341271b7213fceb4ebe18d5a30dab41d703009f",
"MOVIE_DETAIL": "9ae6bcd8c45a5e350438d1cc415a022fe053e938c93438509f60ae3abb425fa7",
"PLAYABLE_ASSET": "e2db6e6f9090f14848d3989920a1342f6813099c65ee7faef1e334f23e390970",
# Recovered from the legacy addon's const.py (SEARCH.HASH) — not captured
# independently from live traffic like the hashes above, so keep an eye on
# this if Joyn ever rotates persisted-query hashes.
"SEARCH": "bb2bab6cbe17321d7eddd5006e7f40765faedd79790b193a59d83f4640694856",
}
# GraphQL endpoints with full URLs
JOYN_GRAPHQL_ENDPOINTS = {
"LIVE_PLAYER": f'{JOYN_GRAPHQL_BASE_URL}?operationName=PageLivePlayerClientSide&enable_user_location=true&watch_assistant_variant=true&extensions=%7B%22persistedQuery%22%3A%7B%22version%22%3A1%2C%22sha256Hash%22%3A%22{GRAPHQL_QUERY_HASHES["LIVE_PLAYER"]}%22%7D%7D',
"LIVE_CHANNELS": f"{JOYN_GRAPHQL_BASE_URL}?operationName=LiveChannelsAndEpg&enable_user_location=true&watch_assistant_variant=true",
GRAPHQL_OPERATIONS = {
"LIVE_PLAYER": "PageLivePlayerClientSide",
"LIVE_CHANNELS": "LiveChannelsAndEpg",
"LIVESTREAM_BY_BRAND": "LivestreamIdByBrand",
"LIVESTREAM_OVERVIEW": "LivestreamOverviewByBrand",
"PLAYABLE_LIVESTREAM": "PlayableLivestream",
"NAVIGATION": "Navigation",
"LANDING_PAGE": "LandingPageClient",
"LANDING_BLOCKS": "LandingBlocks",
"GET_ME_STATE": "GetMeState",
"LIVE_LANE": "LiveLane",
"HERO_RESUME": "HeroLandingResumePositionsWithToken",
"COLLECTION_QUERY": "PageOverviewCollectionQuery",
"PAGE_OVERVIEW_GENRE": "PageOverviewGenre",
"SEASON": "Season",
"MOVIE_DETAIL": "PageMovieDetailStatic",
"PLAYABLE_ASSET": "PlayableAssetWithToken",
# Legacy const.py names this operation "SearchQ", not "Search".
"SEARCH": "SearchQ",
}
# Base GraphQL headers
# Base GraphQL headers (the shared header set without the per-call additions
# built by JoynConfig.graphql_headers()).
JOYN_GRAPHQL_BASE_HEADERS = {
"X-Api-Key": "4f0fd9f18abbe3cf0e87fdb556bc39c8",
"Accept": "application/json",
@@ -112,7 +159,7 @@ JOYN_GRAPHQL_BASE_HEADERS = {
"User-Agent": JOYN_USER_AGENT,
}
# GraphQL query defaults
# GraphQL query defaults (used by the channel and EPG managers)
GRAPHQL_PERSISTED_QUERY_VERSION = 1
GRAPHQL_LIVE_CHANNELS_FILTER = "DEFAULT"
GRAPHQL_MAX_RESULTS = 5000
lib/streaming_providers/providers/joyn/epg_manager.py
+347
View File
@@ -0,0 +1,347 @@
# streaming_providers/providers/joyn/epg_manager.py
# -*- coding: utf-8 -*-
"""
Joyn EPG Manager.
Fetches the channel list + EPG in one operation (`LiveChannelsAndEpg` — the
same one the channel manager uses), parses the `epgEvents` into EPGEntry
objects, and answers the ABC methods (get_epg, get_epg_grid,
get_program_details).
Why reuse `LiveChannelsAndEpg` and not a dedicated EPG operation:
* It is the only EPG operation we have a captured hash for. The other
two operations in the captures (LivestreamIdByBrand, LivestreamOverview)
return EPG ids *without* the program fields (title, start, end), so they
cannot produce EPGEntry by themselves — they would need a third "get
EpgEntry by id" query that we do not have a hash for.
* It is what the reference client itself uses to render the EPG grid.
* It works anonymously (A_A) and authenticated (R_A); the manager uses
`auth.ensure()` so it degrades gracefully when no session is available
(see the channel_manager docstring for the same reasoning).
epg_window
----------
The captured response for a channel spans roughly one day of events
regardless of the requested `from`/`to` window: the server pads a short
window with more events. We declare `(0, 1)` — today plus tomorrow — which
is what the response reliably covers. The window is a declaration of what
we can deliver, not an aspiration.
program_id / epg_event_id
-------------------------
Joyn sends exactly one identifier per event (`program.id`, e.g.
"stv-25_58385168525_1791700800000_..."), and it encodes both the channel
and the broadcast slot in the string itself. There is no separate
"listing guid" as Magenta2 has. EPGEntry has two slots for providers that
distinguish them; Joyn does not, so:
* program_id = the Joyn program.id (this is what the provider's own
operations accept back)
* epg_event_id = None (see the EPGEntry docstring: "leave
as None for providers where no such distinction exists")
"""
import time
from datetime import datetime
from typing import Any, Dict, List, Optional, Tuple
from ...base.managers import EpgManager
from ...base.models.epg_models import (
EPGEntry,
EPGFlags,
EPGProgramDetails,
)
from ...base.utils.logger import logger
from ...base.utils.transport import transport_errors
from .constants import (
DEFAULT_LIVESTREAM_TYPES,
DEFAULT_REQUEST_TIMEOUT,
GRAPHQL_LIVE_CHANNELS_FILTER,
GRAPHQL_MAX_RESULTS,
GRAPHQL_OFFSET,
MODE_LIVE,
PROVIDER_NAME,
)
from .graphql import build_url
_PROVIDER = "Joyn"
# How far the cached channel+EPG listing is reused before we refetch.
# The listing's own server-side cache is ~50s; a slightly longer local TTL
# avoids hammering the API on a grid scroll without giving stale data.
_EPG_CACHE_TTL = 120.0
class JoynEpgManager(EpgManager):
def __init__(
self,
*,
http_manager: Any,
auth: Any,
country: str,
config: Any,
):
super().__init__(
http_manager=http_manager, auth=auth, country=country, config=config
)
# {channel_id: [EPGEntry, ...]} plus the timestamp of the fetch
# that produced it. Populated lazily by _all_epg().
self._epg_by_channel: Optional[Dict[str, List[EPGEntry]]] = None
self._epg_fetched_at: float = 0.0
# ------------------------------------------------------------------
# Capability / routing
# ------------------------------------------------------------------
@property
def epg_window(self) -> Tuple[int, int]:
# Past 0 days (Joyn's EPG is not historical), future 1 day (what
# LiveChannelsAndEpg reliably returns per channel).
return 0, 1
def handles_channel_id(self, channel_id: str) -> bool:
# Every channel that appears in get_channels can appear in the EPG.
# The ABC default is `True`; overriding to document the intent.
return True
# ------------------------------------------------------------------
# Public API — ABC methods
# ------------------------------------------------------------------
def get_epg(
self,
channel_id: str,
start_time: Optional[datetime] = None,
end_time: Optional[datetime] = None,
**kw: Any,
) -> List[EPGEntry]:
events = self._all_epg().get(channel_id, [])
return self._filter(events, start_time, end_time, kw.get("limit"))
def get_epg_grid(
self,
channel_ids: List[str],
start_time: Optional[datetime] = None,
end_time: Optional[datetime] = None,
**kw: Any,
) -> Dict[str, List[EPGEntry]]:
"""
Batch override. LiveChannelsAndEpg returns every channel in one
response, so this is one query for all requested channels, not N
queries via the ABC default (which would loop get_epg).
"""
all_events = self._all_epg()
limit = kw.get("limit")
return {
cid: self._filter(all_events.get(cid, []), start_time, end_time, limit)
for cid in channel_ids
}
def get_program_details(
self, program_id: str, **kw: Any
) -> Optional[EPGProgramDetails]:
"""
Cache-search over the last fetched EPG. Joyn gives us one program id
per event and no dedicated "get details by id" operation that we have
a hash for, so the only source of truth available is the schedule we
already parsed.
Returns None if the program id is not in the current cache — the
ABC permits this (the caller treats it as "no enrichment available").
"""
for events in self._all_epg().values():
for entry in events:
if entry.program_id == program_id:
return self._details_from_entry(entry)
return None
# ------------------------------------------------------------------
# Fetch and parse
# ------------------------------------------------------------------
def _all_epg(self) -> Dict[str, List[EPGEntry]]:
"""
Fetch the channel list with EPG (LiveChannelsAndEpg) and parse it
into {channel_id: [EPGEntry]}. Cached for _EPG_CACHE_TTL seconds;
a second call within the TTL reuses the previous result.
"""
now = time.time()
if self._epg_by_channel is not None and (now - self._epg_fetched_at) < _EPG_CACHE_TTL:
return self._epg_by_channel
streams = self._fetch_live_streams()
by_channel: Dict[str, List[EPGEntry]] = {}
for stream in streams:
cid = stream.get("id")
if not cid:
continue
is_live = stream.get("type") == "LINEAR"
events: List[EPGEntry] = []
for ev in stream.get("epgEvents") or []:
entry = self._parse_event(cid, ev, is_live)
if entry is not None:
events.append(entry)
# Sort by start; the API is usually already sorted but we do not
# rely on it.
events.sort(key=lambda e: e.start)
by_channel[cid] = events
self._epg_by_channel = by_channel
self._epg_fetched_at = now
logger.info(
f"{_PROVIDER}: EPG fetched for {len(by_channel)} channels "
f"({sum(len(v) for v in by_channel.values())} events)"
)
return by_channel
def _fetch_live_streams(self) -> List[Dict[str, Any]]:
"""
One GraphQL call, same operation the channel manager uses. The
window matches the channel manager's (now .. now + 3h) — the
server pads it, and the parsed events carry their own start/end,
so a wider `to` would not give us more events per channel.
"""
now_ts = int(time.time())
variables = {
"liveStreamGroupFilter": GRAPHQL_LIVE_CHANNELS_FILTER,
"first": GRAPHQL_MAX_RESULTS,
"offset": GRAPHQL_OFFSET,
"livestreamTypes": DEFAULT_LIVESTREAM_TYPES,
"from": now_ts,
"to": now_ts + 3 * 3600,
}
url = build_url("LIVE_CHANNELS", variables)
# EPG is served anonymously; use ensure() so a broken or absent
# session does not block a public channel listing.
token = None
if self.auth.ensure():
token = self.auth.get_access_token()
headers = self.config.graphql_headers(
token=token, authenticated=token is not None
)
with transport_errors("epg_fetch", _PROVIDER):
response = self.http_manager.get(
url,
operation="api",
headers=headers,
timeout=DEFAULT_REQUEST_TIMEOUT,
)
response.raise_for_status()
data = response.json()
return (data.get("data") or {}).get("liveStreams") or []
# ------------------------------------------------------------------
# Parsing
# ------------------------------------------------------------------
def _parse_event(
self, channel_id: str, ev: Dict[str, Any], is_live_stream: bool
) -> Optional[EPGEntry]:
"""
Turn one `epgEvents[]` item into an EPGEntry.
Joyn's event wrapper sometimes carries start/end on the wrapper and
sometimes on the embedded `program` (it varies by __typename), so
both are checked. A missing required field returns None and the
event is skipped — one bad event must not fail the whole listing
(README §12.5).
"""
program = ev.get("program") or {}
start = program.get("startDate") or ev.get("startDate")
end = program.get("endDate") or ev.get("endDate")
program_id = program.get("id")
title = program.get("title")
if not (program_id and title and start and end):
return None
try:
start_i = int(start)
end_i = int(end)
except (TypeError, ValueError):
return None
if end_i <= start_i:
return None
# Flags from __typename and the enclosing stream type.
flags = EPGFlags.UNDEFINED
if is_live_stream:
flags |= EPGFlags.IS_LIVE
if program.get("__typename") == "Episode" and program.get("series"):
flags |= EPGFlags.IS_SERIES
# `secondaryTitle` is the XMLTV-style sub-title; empty in most
# captured entries but populated for some. `series.title` is a
# fallback for episodes without a secondary title.
episode_name = program.get("secondaryTitle") or None
if not episode_name and isinstance(program.get("series"), dict):
episode_name = program["series"].get("title") or None
# `genres` — a list of dicts, mapped to a list of name strings.
genres_raw = program.get("genres") or []
genres = [g.get("name") for g in genres_raw if g.get("name")] or None
# The poster image URL, if any.
icon = (program.get("posterImage") or {}).get("url")
return EPGEntry(
broadcast_id=EPGEntry.encode_broadcast_id(
PROVIDER_NAME, channel_id, start_i
),
title=str(title),
start=start_i,
end=end_i,
program_id=str(program_id),
# epg_event_id intentionally None — see the module docstring.
episode_name=episode_name,
icon=icon,
genres=genres,
flags=flags or None,
)
def _details_from_entry(self, entry: EPGEntry) -> EPGProgramDetails:
"""
Build an EPGProgramDetails from an EPGEntry we already have.
The dedicated detail endpoint (if any) is not captured, so this is
the only information available; it fills the shared fields
(description, episode_name, icon, genres, flags-adjacent data) and
leaves the rest as None. A future capture of a "details by id"
operation would replace this with a real fetch.
"""
return EPGProgramDetails(
program_id=entry.program_id or "",
description=entry.description,
episode_name=entry.episode_name,
icon=entry.icon,
genres=entry.genres,
season_number=entry.season_number,
episode_number=entry.episode_number,
parental_rating=entry.parental_rating,
)
# ------------------------------------------------------------------
# Filtering helpers
# ------------------------------------------------------------------
@staticmethod
def _filter(
events: List[EPGEntry],
start_time: Optional[datetime],
end_time: Optional[datetime],
limit: Optional[int],
) -> List[EPGEntry]:
"""
Window filter. `start_time` / `end_time` are timezone-aware UTC
datetimes (per the EpgManager contract); EPGEntry.start / .end
are unix seconds. An overlap test is correct here — a program that
starts before the window but is still airing counts as inside.
"""
result = events
if start_time is not None or end_time is not None:
start_ts = int(start_time.timestamp()) if start_time is not None else 0
end_ts = int(end_time.timestamp()) if end_time is not None else 1 << 62
result = [e for e in result if e.end > start_ts and e.start < end_ts]
if limit is not None and limit > 0:
result = result[:limit]
return result
lib/streaming_providers/providers/joyn/graphql.py
+56
View File
@@ -0,0 +1,56 @@
# streaming_providers/providers/joyn/graphql.py
# -*- coding: utf-8 -*-
"""
Joyn GraphQL URL builder.
The single place that knows how to turn a persisted-query key plus variables
into an API URL. The hashes and operation names themselves live in
constants.py (GRAPHQL_QUERY_HASHES / GRAPHQL_OPERATIONS); this module is
only the assembler.
Why separate from constants.py: constants is pure data, this is the one bit
of logic over it. Having a single builder means every manager builds the
URL the same way, and if the API ever changes the query-string shape (e.g.
adds a mandatory parameter) there is exactly one place to change.
"""
import json
import urllib.parse
from typing import Any, Dict, Optional
from .constants import (
GRAPHQL_OPERATIONS,
GRAPHQL_QUERY_HASHES,
JOYN_GRAPHQL_BASE_URL,
)
def build_url(operation_key: str, variables: Optional[Dict[str, Any]] = None) -> str:
"""
Build a persisted-query URL.
`operation_key` is a key in GRAPHQL_OPERATIONS / GRAPHQL_QUERY_HASHES
(e.g. "LIVE_CHANNELS"). Both the operationName and the sha256Hash come
from the constants, so they cannot drift.
Raises KeyError if operation_key is unknown — that's a programmer error,
not a runtime condition, so it fails loudly.
"""
operation_name = GRAPHQL_OPERATIONS[operation_key]
query_hash = GRAPHQL_QUERY_HASHES[operation_key]
params = {
"operationName": operation_name,
"enable_user_location": "true",
"watch_assistant_variant": "true",
}
if variables:
params["variables"] = json.dumps(variables, separators=(",", ":"))
params["extensions"] = json.dumps(
{"persistedQuery": {"version": 1, "sha256Hash": query_hash}},
separators=(",", ":"),
)
query_string = "&".join(
f"{k}={urllib.parse.quote(v)}" for k, v in params.items()
)
return f"{JOYN_GRAPHQL_BASE_URL}?{query_string}"
lib/streaming_providers/providers/joyn/provider.py
+10 -1
View File
@@ -43,6 +43,7 @@ from .constants import JOYN_LOGO, PROVIDER_NAME, SUPPORTED_COUNTRIES
from .entitlement import JoynEntitlement
from .session import JoynSession
from .vod_manager import JoynVodManager
from .epg_manager import JoynEpgManager
# Only the provider is public. Discovery takes the FIRST StreamingProvider
# subclass in the package namespace; ManagedProvider must never appear here.
@@ -176,7 +177,15 @@ class JoynProvider(ManagedProvider):
entitlement=self._entitlement,
)
# No _build_epg, no _build_catchup: Joyn does not have them. A provider
def _build_epg(self):
return JoynEpgManager(
http_manager=self.http_manager,
auth=self.auth,
country=self.country,
config=self.provider_config,
)
# No _build_catchup: Joyn does not have it. A provider
# without a capability simply has no manager.
# ------------------------------------------------------------------
lib/streaming_providers/providers/joyn/tests/test_joyn_provider.py
+573 -31
View File
@@ -1,43 +1,585 @@
def test_contract(http):
errors, warnings = check_provider(JoynProvider, "joyn")
# streaming_providers/providers/joyn/tests/test_joyn_provider.py
# -*- coding: utf-8 -*-
"""
Joyn provider tests — offline (fake HTTP, fake authenticator), README §15 pattern.
Run against the REAL package: pytest streaming_providers/providers/joyn/tests
The one test that needs `base.testing.provider_contract` is skipped when that
module is not importable (so the file also runs in a stripped-down sandbox).
"""
import threading
import time
import types
import pytest
from streaming_providers.base.errors import (
AuthError,
BadRequestError,
EntitlementError,
NotFoundError,
RateLimitError,
ServerError,
)
from streaming_providers.base.vod import VodPage
from streaming_providers.providers.joyn import provider as provider_module
from streaming_providers.providers.joyn import vod_manager as vod_module
from streaming_providers.providers.joyn.constants import (
ERROR_CODES,
JOYN_BROWSER_MAJOR,
JOYN_SEC_CH_UA,
JOYN_SEC_CH_UA_PLATFORM,
JOYN_STREAMING_ENDPOINTS,
JOYN_USER_AGENT,
)
from streaming_providers.providers.joyn.entitlement import JoynEntitlement # noqa: F401
from streaming_providers.providers.joyn.ids import is_live_id, is_vod_id
from streaming_providers.providers.joyn.models import (
JoynAuthError,
JoynEntitlementError,
JoynMfaRequiredException,
)
from streaming_providers.providers.joyn.provider import JoynProvider
from streaming_providers.providers.joyn.session import (
BACKOFF_BASE_SECONDS,
JoynSession,
)
# ============================================================================
# Fakes
# ============================================================================
class FakeResponse:
def __init__(self, status=200, body=None):
self.status_code = status
self._body = {} if body is None else body
self.text = ""
def json(self):
if isinstance(self._body, Exception):
raise self._body
return self._body
def raise_for_status(self):
if self.status_code >= 400:
raise RuntimeError(f"HTTP {self.status_code}")
class FakeHttpError(Exception):
"""What an HTTP manager that raises on 4xx looks like: it carries .response."""
def __init__(self, response):
super().__init__(f"HTTP {response.status_code}")
self.response = response
class FakeHttp:
CHANNELS = {"data": {"liveStreams": [
{"id": "sat1-de", "title": "SAT.1", "type": "LINEAR", "quality": "HD",
"logo": {"url": "https://logo.example/sat1.png"}},
{"title": "entry without id"}, # malformed -> skipped
"garbage", # malformed -> skipped
{"id": "prosieben-de", "title": "ProSieben", "type": "LINEAR"},
]}}
def __init__(self):
self.calls = [] # (method, url, headers)
self.entitlement = {} # content_id -> (status, body)
self.raise_on_4xx = False # emulate an HTTP manager that raises
self.license_url = "https://license.example/wv"
# -- helpers ---------------------------------------------------------
def urls(self, needle):
return [u for _, u, _ in self.calls if needle in u]
# -- HTTPManager surface --------------------------------------------
def post(self, url, operation=None, headers=None, json_data=None, data=None,
timeout=None, **kw):
self.calls.append(("POST", url, dict(headers or {})))
if url == JOYN_STREAMING_ENDPOINTS["ENTITLEMENT"]:
cid = json_data["content_id"]
status, body = self.entitlement.get(
cid, (200, {"entitlement_token": f"ent-{cid}"})
)
response = FakeResponse(status, body)
if status >= 400 and self.raise_on_4xx:
raise FakeHttpError(response)
return response
if "/playlist" in url:
ident = url.split("/playlist")[0].rsplit("/", 1)[-1]
return FakeResponse(200, {
"manifestUrl": f"https://cdn.example/{ident}.mpd",
"licenseUrl": self.license_url,
"certificateUrl": "https://license.example/cert",
"streamingFormat": "dash",
})
raise AssertionError(f"unexpected POST {url}")
def get(self, url, operation=None, headers=None, timeout=None, **kw):
self.calls.append(("GET", url, dict(headers or {})))
if "LiveChannelsAndEpg" in url:
return FakeResponse(200, self.CHANNELS)
return FakeResponse(200, {"data": {}}) # navigation/asset: empty -> fallbacks
class FakeToken:
def __init__(self, access_token):
self.access_token = access_token
self.is_expired = False
class FakeAuthenticator:
"""Stands in for JoynAuthenticator: no 7pass flow, no network, no disk."""
def __init__(self, **ctor_kwargs):
self.ctor_kwargs = ctor_kwargs
self.current_token = None
self.fail = None
self.attempts = 0
self.logins = 0
self.invalidated = 0
self.force_flags = []
self.delay = 0.0
def get_bearer_token(self, force_refresh=False, force_upgrade=False):
self.force_flags.append(force_refresh)
self.attempts += 1
if self.delay:
time.sleep(self.delay)
if self.fail is not None:
raise self.fail
if self.current_token is None or self.current_token.is_expired:
self.logins += 1
self.current_token = FakeToken(f"tok-{self.logins}")
return self.current_token.access_token
def invalidate_token(self):
self.current_token = None
self.invalidated += 1
@pytest.fixture
def env(monkeypatch):
http = FakeHttp()
holder = types.SimpleNamespace(http=http, auth=None)
def make_authenticator(**kw):
holder.auth = FakeAuthenticator(**kw)
return holder.auth
monkeypatch.setattr(JoynProvider, "_setup_http_manager", lambda self, **kw: http)
monkeypatch.setattr(provider_module, "JoynAuthenticator", make_authenticator)
monkeypatch.setattr(vod_module.time, "sleep", lambda s: holder.__dict__
.setdefault("sleeps", []).append(s))
return holder
@pytest.fixture
def make(env):
def _make(country="DE"):
return JoynProvider(country)
return _make
# ============================================================================
# Contract, construction, identity
# ============================================================================
def test_contract(env):
mod = pytest.importorskip("streaming_providers.base.testing.provider_contract")
errors, warnings = mod.check_provider(JoynProvider, "joyn")
assert errors == [], (errors, warnings)
def test_no_network_in_init(http):
JoynProvider("DE")
assert http.calls == [] # the old eager login is gone
def test_joyn_headers_do_not_leak_into_cdn(http):
p = JoynProvider("DE")
h = p.get_manifest_headers("sat1-de")
assert "Authorization" not in h
assert "joyn-client-version" not in h
def test_no_network_in_init(env, make):
make()
assert env.http.calls == []
assert env.auth.attempts == 0 # the old eager login is gone
def test_segment_headers_equal_manifest_headers(http):
p = JoynProvider("DE")
def test_country_is_normalised_and_unsupported_rejected(env, make):
assert make("DE").country == "de"
with pytest.raises(NotImplementedError):
make("US")
def test_identity_and_exports(env, make):
import streaming_providers.providers.joyn as pkg
assert make().provider_name == "joyn"
assert pkg.__all__ == ["JoynProvider"]
assert not hasattr(pkg, "ManagedProvider")
def test_one_shared_config_reaches_the_authenticator(env, make):
p = make("AT")
assert env.auth.ctor_kwargs["config"] is p.provider_config
assert p.provider_config.country == "at"
assert p.provider_config.distribution_tenant == "JOYN_AT"
def test_client_hints_match_user_agent():
assert f"Chrome/{JOYN_BROWSER_MAJOR}." in JOYN_USER_AGENT
assert JOYN_SEC_CH_UA.count(JOYN_BROWSER_MAJOR) == 2
assert "Windows" in JOYN_USER_AGENT and JOYN_SEC_CH_UA_PLATFORM == '"Windows"'
# ============================================================================
# Id grammar and routing
# ============================================================================
@pytest.mark.parametrize("cid", [
"a_abc123", "b_movie", "c_season", "d_episode", "block-12",
"/serien", "/serien/genre/krimi", "foo:bar",
])
def test_vod_ids(cid):
assert is_vod_id(cid) and not is_live_id(cid)
@pytest.mark.parametrize("cid", ["sat1-de", "sat1-de-hd", "prosieben-de"])
def test_live_ids(cid):
assert is_live_id(cid) and not is_vod_id(cid)
@pytest.mark.parametrize("cid", ["", None])
def test_empty_id_is_neither(cid):
assert not is_vod_id(cid) and not is_live_id(cid)
def test_handles_content_id_is_pure_and_complementary(env, make):
p = make()
for cid in ["sat1-de", "sat1-de-hd", "a_abc123", "b_x", "c_x", "d_x",
"block-1", "/serien/x", "x:y"]:
assert p.channels.handles_content_id(cid) != p.vod.handles_content_id(cid), cid
assert env.http.calls == [] # no I/O to decide ownership
def test_vod_id_reaches_the_vod_manager_not_the_channel_manager(env, make):
p = make()
manifest = p.get_manifest(content_id="a_abc123")
assert manifest == "https://cdn.example/a_abc123.mpd"
assert env.http.urls("/asset/a_abc123/playlist")
assert not env.http.urls("/channel/")
def test_channel_manager_refuses_a_vod_id_it_is_handed_directly(env, make):
p = make()
with pytest.raises(BadRequestError):
p.channels.get_channel_manifest("a_abc123")
# ============================================================================
# Playout: manifest + DRM share one call; -hd probing; cache
# ============================================================================
def test_live_manifest_and_drm_share_one_playlist_call(env, make):
p = make()
assert p.get_manifest(content_id="sat1-de") == "https://cdn.example/sat1-de-hd.mpd"
drm = p.get_drm(content_id="sat1-de", drm_variant="auto")
assert len(drm) == 1 and drm[0].priority == 1
assert len(env.http.urls("/channel/")) == 1
def test_hd_variant_is_probed_and_resolution_is_cached(env, make):
env.http.entitlement["sat1-de-hd"] = (
400, [{"code": ERROR_CODES["NOT_FOUND"], "msg": "no hd"}])
p = make()
assert p.get_manifest(content_id="sat1-de") == "https://cdn.example/sat1-de.mpd"
assert p._entitlement._resolved_channel_variants["sat1-de"] == "sat1-de"
def test_expired_playout_entries_are_evicted_on_insert(env, make):
p = make()
p._playout_cache["old:default"] = (object(), 0.0)
p.get_manifest(content_id="sat1-de")
assert "old:default" not in p._playout_cache
# ============================================================================
# Entitlement error mapping
# ============================================================================
def test_playback_restricted_becomes_typed_entitlement_error(env, make):
body = (400, [{"code": ERROR_CODES["PLAYBACK_RESTRICTED"], "msg": "nope"}])
env.http.entitlement.update({"sat1-de": body, "sat1-de-hd": body})
with pytest.raises(EntitlementError):
make().get_manifest(content_id="sat1-de")
def test_400_is_detected_when_the_http_manager_raises_instead_of_returning(env, make):
env.http.raise_on_4xx = True
body = (400, [{"code": ERROR_CODES["BUSINESS_MODEL_NOT_SUITABLE"], "msg": "plus"}])
env.http.entitlement.update({"sat1-de": body, "sat1-de-hd": body})
with pytest.raises(EntitlementError) as info:
make().get_manifest(content_id="sat1-de")
assert "Subscription required" in str(info.value)
def test_hd_probe_works_when_the_http_manager_raises_on_400(env, make):
env.http.raise_on_4xx = True
env.http.entitlement["sat1-de-hd"] = (
400, [{"code": ERROR_CODES["NOT_FOUND"], "msg": "no hd"}])
assert make().get_manifest(content_id="sat1-de") == "https://cdn.example/sat1-de.mpd"
def test_malformed_400_body_is_an_entitlement_error_not_a_crash(env, make):
for body in ({"unexpected": 1}, [], ["x"], ValueError("not json")):
env.http.entitlement.update({"sat1-de": (400, body), "sat1-de-hd": (400, body)})
with pytest.raises(JoynEntitlementError):
make().get_manifest(content_id="sat1-de")
# ============================================================================
# Headers
# ============================================================================
def test_cdn_headers_carry_no_api_headers_or_bearer(env, make):
p = make()
p.get_manifest(content_id="sat1-de")
headers = p.get_manifest_headers("sat1-de")
assert "Authorization" not in headers and "joyn-client-version" not in headers
assert set(headers) == {"User-Agent", "Origin"}
def test_segment_headers_equal_manifest_headers(env, make):
p = make()
p.get_manifest(content_id="sat1-de")
assert p.get_segment_headers("sat1-de") == p.get_manifest_headers("sat1-de")
def test_entitlement_400_playback_restricted_becomes_typed_error(http):
http.entitlement_response = 400, [{"code": "ENT_RVOD_Playback_Restricted", "msg": "nope"}]
p = JoynProvider("DE")
with pytest.raises(EntitlementError):
p.get_manifest("sat1-de")
def test_mfa_raises_auth_error_not_anonymous_fallback(http):
http.login_redirect_to_mfa = True
p = JoynProvider("DE")
p.set_user_credentials(...) if that exists # it doesn't, so:
# ============================================================================
# Channels
# ============================================================================
def test_malformed_channel_entries_are_skipped_not_fatal(env, make):
channels = make().channels.get_channels()
assert [c.content_id for c in channels] == ["sat1-de", "prosieben-de"]
def test_channel_list_uses_a_bearer_and_user_state(env, make):
make().channels.get_channels()
headers = [h for m, u, h in env.http.calls if m == "GET"][0]
assert headers["Authorization"] == "Bearer tok-1"
# ============================================================================
# Auth failures are typed, loud and not retried
# ============================================================================
def test_mfa_raises_auth_error_not_an_anonymous_channel_list(env, make):
p = make()
env.auth.fail = JoynMfaRequiredException("mfa on")
with pytest.raises(AuthError):
p.channels.get_channels()
assert env.http.calls == [] # no silent anonymous request
def test_get_vod_category_returns_vodpage(http):
from ...base.vod import VodPage
p = JoynProvider("DE")
page = p.vod.get_vod_category("")
def test_mfa_is_not_retried(env, make):
p = make()
env.auth.fail = JoynMfaRequiredException("mfa on")
for _ in range(3):
with pytest.raises(AuthError):
p.channels.get_channels()
assert env.auth.attempts == 1
def test_rejected_credentials_are_permanent(env, make):
p = make()
env.auth.fail = JoynAuthError("rejected")
for _ in range(3):
with pytest.raises(AuthError):
p.channels.get_channels()
assert env.auth.attempts == 1
# ============================================================================
# Session (fake clock, no provider)
# ============================================================================
class Clock:
def __init__(self):
self.now = 0.0
def __call__(self):
return self.now
@pytest.fixture
def session_parts():
auth = FakeAuthenticator()
clock = Clock()
callbacks = []
session = JoynSession(authenticator=auth, config=types.SimpleNamespace(
api_headers=lambda t: {"Authorization": f"Bearer {t}"}),
clock=clock, on_invalidate=lambda: callbacks.append(1))
return types.SimpleNamespace(auth=auth, clock=clock, session=session, callbacks=callbacks)
def test_first_login_does_not_force_a_refresh(session_parts):
assert session_parts.session.ensure() is True
assert session_parts.auth.force_flags == [False] # persisted token path stays usable
def test_transient_failure_backs_off_exponentially_then_recovers(session_parts):
s, a, c = session_parts.session, session_parts.auth, session_parts.clock
a.fail = ConnectionError("net down")
assert s.ensure() is False and a.attempts == 1
assert s.ensure() is False and a.attempts == 1 # inside the backoff window
c.now = BACKOFF_BASE_SECONDS
assert s.ensure() is False and a.attempts == 2 # window over, second failure
c.now = BACKOFF_BASE_SECONDS + 2 * BACKOFF_BASE_SECONDS - 1
assert s.ensure() is False and a.attempts == 2 # doubled window not over yet
a.fail = None
c.now = BACKOFF_BASE_SECONDS * 3
assert s.ensure() is True and a.attempts == 3
def test_provider_error_type_is_preserved(session_parts):
session_parts.auth.fail = RateLimitError("slow down")
with pytest.raises(RateLimitError):
session_parts.session.require()
def test_untyped_exception_becomes_server_error_and_last_error_keeps_the_original(session_parts):
original = ConnectionError("boom")
session_parts.auth.fail = original
with pytest.raises(ServerError) as info:
session_parts.session.require()
assert session_parts.session.last_error is original
assert info.value.__cause__ is original
def test_ensure_never_raises(session_parts):
session_parts.auth.fail = RuntimeError("anything")
assert session_parts.session.ensure() is False
def test_one_login_for_concurrent_callers(session_parts):
session_parts.auth.delay = 0.05
results = []
threads = [threading.Thread(target=lambda: results.append(session_parts.session.ensure()))
for _ in range(8)]
for t in threads:
t.start()
for t in threads:
t.join()
assert results == [True] * 8
assert session_parts.auth.logins == 1
def test_invalidate_resets_backoff_and_runs_callbacks(session_parts):
s, a = session_parts.session, session_parts.auth
a.fail = JoynMfaRequiredException("mfa")
assert s.ensure() is False and s.ensure() is False and a.attempts == 1
a.fail = None
s.invalidate()
assert a.invalidated == 1 and session_parts.callbacks == [1]
assert s.ensure() is True
def test_a_broken_invalidate_callback_does_not_block_invalidation(session_parts):
def boom():
raise RuntimeError("cache hook failed")
session_parts.session._on_invalidate.insert(0, boom)
session_parts.session.invalidate()
assert session_parts.callbacks == [1]
# ============================================================================
# Credential change clears account-specific caches
# ============================================================================
def test_credentials_change_clears_playout_entitlement_and_vod_caches(env, make):
p = make()
p.get_manifest(content_id="sat1-de")
p.vod._cache["video_id:b_x"] = {"timestamp": time.time(), "data": "a_x"}
assert p._playout_cache and p._entitlement._resolved_channel_variants and p.vod._cache
p.auth.invalidate()
assert p._playout_cache == {}
assert p._entitlement._resolved_channel_variants == {}
assert p.vod._cache == {}
assert env.auth.invalidated == 1
# ============================================================================
# VOD
# ============================================================================
def test_get_vod_category_returns_a_vodpage(env, make):
page = make().vod.get_vod_category("")
assert isinstance(page, VodPage)
assert page.has_more is False and len(page.entries) >= 2 # static fallback + genres
def test_handles_content_id_is_pure(http):
p = JoynProvider("DE")
# No HTTP call should be made for either of these
assert p.vod.handles_content_id("a_abc123") is True
assert p.vod.handles_content_id("sat1-de") is False
assert http.calls == []
@pytest.mark.parametrize("cid,pageable", [
("block-1", True), ("/serien/genre/krimi", True),
("c_season", False), ("/serien/some-show", False), ("/filme/some-movie", False),
("/neu-beliebt", False), ("/serien/genre", False), ("", False),
])
def test_only_endpoints_that_honour_offset_are_pageable(env, make, cid, pageable):
assert make().vod._is_pageable(cid) is pageable
def test_full_pageable_page_yields_a_next_cursor(env, make, monkeypatch):
p = make()
monkeypatch.setattr(p.vod, "_get_vod_category_entries", lambda **kw: list(range(24)))
assert p.vod.get_vod_category("block-1", page_size=24).next_cursor == "24"
assert p.vod.get_vod_category("block-1", cursor="24", page_size=24).next_cursor == "48"
monkeypatch.setattr(p.vod, "_get_vod_category_entries", lambda **kw: list(range(5)))
assert p.vod.get_vod_category("block-1", page_size=24).next_cursor is None
def test_non_pageable_endpoints_never_claim_more_pages(env, make, monkeypatch):
p = make()
monkeypatch.setattr(p.vod, "_get_vod_category_entries", lambda **kw: list(range(40)))
for cid in ("/serien/some-show", "c_season", "/neu-beliebt"):
assert p.vod.get_vod_category(cid, page_size=24).next_cursor is None
def test_non_pageable_branches_keep_the_helpers_default_size(env, make, monkeypatch):
p = make()
seen = {}
monkeypatch.setattr(p.vod, "_get_series_items_fallback",
lambda path, authenticated=True, **kw: seen.update(kw) or [])
monkeypatch.setattr(p.vod, "get_collection_items",
lambda **kw: seen.update(block=kw) or [])
p.vod._get_vod_category_entries(content_id="/serien/x", cursor=None,
page_size=10, authenticated=True)
assert "first" not in seen # helper default (v1 behaviour) applies
p.vod._get_vod_category_entries(content_id="block-1", cursor="20",
page_size=10, authenticated=True)
assert seen["block"]["first"] == 10 and seen["block"]["offset"] == 20
def test_unresolvable_vod_id_is_not_found_not_not_mine(env, make):
with pytest.raises(NotFoundError):
make().vod.get_vod_manifest("b_unknown")
def test_vod_typed_errors_propagate_unchanged_and_are_not_retried(env, make):
p = make()
env.auth.fail = RateLimitError("slow")
with pytest.raises(RateLimitError):
p.vod.get_vod_manifest("a_abc")
assert env.__dict__.get("sleeps", []) == []
def test_vod_entitlement_error_is_not_retried(env, make):
env.http.entitlement["a_abc"] = (
400, [{"code": ERROR_CODES["PLAYBACK_RESTRICTED"], "msg": "no"}])
with pytest.raises(EntitlementError):
make().vod.get_vod_manifest("a_abc")
assert len(env.http.urls("entitlement")) == 1
def test_vod_server_errors_are_retried_then_raised(env, make):
env.http.entitlement["a_abc"] = (503, {})
with pytest.raises(ServerError):
make().vod.get_vod_manifest("a_abc", max_retries=3)
assert len(env.http.urls("entitlement")) == 3
assert len(env.__dict__["sleeps"]) == 2
def test_get_vod_drm_returns_the_shared_widevine_config(env, make):
drm = make().vod.get_vod_drm("a_abc")
assert len(drm) == 1 and drm[0].license.server_url == env.http.license_url
lib/streaming_providers/providers/joyn/vod_manager.py
+19 -114
View File
@@ -35,13 +35,16 @@ only a result or a typed failure is valid. Typed `ProviderError`s (AuthError,
RateLimitError, GeoBlockError, EntitlementError, NotFoundError, ...) ALWAYS
propagate unchanged; only untyped exceptions are wrapped in `ServerError`.
Only `ServerError`s are retried.
GraphQL URLs: every operation is built through `graphql.build_url(key, ...)`.
The hashes and operation names live in `constants.GRAPHQL_QUERY_HASHES` /
`constants.GRAPHQL_OPERATIONS`; there is no local copy in this file.
"""
import functools
import json
import re
import time
import urllib.parse
from typing import Any, Dict, List, Optional, Union
from ...base.errors import NotFoundError, ProviderError, ServerError
@@ -55,11 +58,11 @@ from ...base.vod import VodPage
from .drm import build_widevine_config
from .ids import is_vod_id
from .signing import build_signature, create_video_payload, video_config_fingerprint
from .graphql import build_url
from .constants import (
CONTENT_TYPE_VOD,
DEFAULT_MAX_RETRIES,
DEFAULT_REQUEST_TIMEOUT,
JOYN_GRAPHQL_BASE_URL,
JOYN_STREAMING_ENDPOINTS,
JOYN_USER_AGENT,
)
@@ -69,44 +72,6 @@ from .models import (
SubscriptionRequiredException,
)
# ============================================================================
# GraphQL query hashes (VOD-specific; live hashes live in constants.py)
# ============================================================================
VOD_GRAPHQL_HASHES = {
"NAVIGATION": "818622ff5afe143664241034ba0c537650ec9a2eeae4d568830b47d8de605b7e",
"LANDING_PAGE": "5838ac6b328cda91a70da264d5be09d4735cf95de8d9ee00cec60e77fcc3df08",
"LANDING_BLOCKS": "1655591f83b0dc1508ad4d52c5f37f72d410f48ad08c3e5f2de8622f86a21c68",
"GET_ME_STATE": "55ebb3812b45628017ee6c7f36f0b88a94e9778b9f11ad8a6fc05849182c07ec",
"LIVE_LANE": "51659c62d4e4a6628d1e512190a3b0659486478b12be494875bef5a83dcb79ed",
"HERO_RESUME": "d3b7e480f593ba4866598f8cfe95185b3e400fcff112c026dc4e1b5ad4b0d537",
"COLLECTION_QUERY": "bdf4e08de65351750eefb2165a58af50c9e4b3526b78cd75e2066df2bc7ec8d8",
"PAGE_OVERVIEW_GENRE": "37ba6d0dde470df3f8999d49bcd24bc5c72b8e7192768026d82447c664c6ab7f",
"SEASON": "ee2396bb1b7c9f800e5cefd0b341271b7213fceb4ebe18d5a30dab41d703009f",
"MOVIE_DETAIL": "9ae6bcd8c45a5e350438d1cc415a022fe053e938c93438509f60ae3abb425fa7",
"PLAYABLE_ASSET": "e2db6e6f9090f14848d3989920a1342f6813099c65ee7faef1e334f23e390970",
# Recovered from the legacy addon's const.py (SEARCH.HASH) — not captured
# independently from live traffic like the hashes above, so keep an eye on
# this if Joyn ever rotates persisted-query hashes.
"SEARCH": "bb2bab6cbe17321d7eddd5006e7f40765faedd79790b193a59d83f4640694856",
}
GRAPHQL_OPERATIONS = {
"NAVIGATION": "Navigation",
"LANDING_PAGE": "LandingPageClient",
"LANDING_BLOCKS": "LandingBlocks",
"GET_ME_STATE": "GetMeState",
"LIVE_LANE": "LiveLane",
"HERO_RESUME": "HeroLandingResumePositionsWithToken",
"COLLECTION_QUERY": "PageOverviewCollectionQuery",
"PAGE_OVERVIEW_GENRE": "PageOverviewGenre",
"SEASON": "Season",
"MOVIE_DETAIL": "PageMovieDetailStatic",
"PLAYABLE_ASSET": "PlayableAssetWithToken",
# Legacy const.py names this operation "SearchQ", not "Search".
"SEARCH": "SearchQ",
}
def ttl_cache(ttl_seconds: int = 300):
"""
@@ -196,9 +161,6 @@ class JoynVodManager(VodManager):
self._cache: Dict[str, Dict[str, Any]] = {}
self._cache_ttl: int = 300
self._query_hashes = VOD_GRAPHQL_HASHES
self._operations = GRAPHQL_OPERATIONS
self._user_state: Optional[Dict[str, Any]] = None
self._has_plus = False
@@ -245,37 +207,19 @@ class JoynVodManager(VodManager):
search, season under an account)" — we then ask the session for a
token and set joyn-user-state=code=R_A. When False, the query is
anonymous and the reference client sends A_A.
Unlike the channel/EPG managers' header builder (which uses ensure()
and degrades to anonymous), this one RAISES on a permanent login
failure: a user-scoped query is meaningless without a session, and
the failure must surface so the user can act on it (README §8).
"""
token: Optional[str] = None
if authenticated:
# RAISES a typed error if no session can be had (README §8) instead of
# silently degrading to an anonymous request.
token = self.auth.get_access_token()
return self.config.graphql_headers(
token=token, authenticated=token is not None
)
@staticmethod
def _build_graphql_url(
operation_name: str,
query_hash: str,
variables: Optional[Dict] = None,
) -> str:
base_url = JOYN_GRAPHQL_BASE_URL
params = {
"operationName": operation_name,
"enable_user_location": "true",
"watch_assistant_variant": "true",
}
if variables:
params["variables"] = json.dumps(variables, separators=(",", ":"))
extensions = {"persistedQuery": {"version": 1, "sha256Hash": query_hash}}
params["extensions"] = json.dumps(extensions, separators=(",", ":"))
query_string = "&".join(
f"{k}={urllib.parse.quote(v)}" for k, v in params.items()
)
return f"{base_url}?{query_string}"
# ========================================================================
# PATH / ID CLASSIFICATION HELPERS
# ========================================================================
@@ -306,10 +250,7 @@ class JoynVodManager(VodManager):
def get_navigation(self) -> Dict[str, Any]:
with transport_errors("vod_navigation", "Joyn"):
url = self._build_graphql_url(
operation_name=self._operations["NAVIGATION"],
query_hash=self._query_hashes["NAVIGATION"],
)
url = build_url("NAVIGATION")
headers = self._get_graphql_headers(authenticated=False)
response = self.http_manager.get(
url,
@@ -423,11 +364,7 @@ class JoynVodManager(VodManager):
if not path.startswith("/"):
path = f"/{path}"
variables = {"first": first, "path": path, "offset": offset}
url = self._build_graphql_url(
operation_name=self._operations["PAGE_OVERVIEW_GENRE"],
query_hash=self._query_hashes["PAGE_OVERVIEW_GENRE"],
variables=variables,
)
url = build_url("PAGE_OVERVIEW_GENRE", variables)
headers = self._get_graphql_headers(authenticated=authenticated)
response = self.http_manager.get(
url, operation="vod_genre_page", headers=headers,
@@ -538,11 +475,7 @@ class JoynVodManager(VodManager):
if not path.startswith("/"):
path = f"/{path}"
variables = {"path": path}
url = self._build_graphql_url(
operation_name=self._operations["MOVIE_DETAIL"],
query_hash=self._query_hashes["MOVIE_DETAIL"],
variables=variables,
)
url = build_url("MOVIE_DETAIL", variables)
headers = self._get_graphql_headers(authenticated=authenticated)
response = self.http_manager.get(
url, operation="vod_movie_detail", headers=headers,
@@ -586,11 +519,7 @@ class JoynVodManager(VodManager):
"id": season_id, "first": first,
"licenseFilter": license_filter, "offset": offset,
}
url = self._build_graphql_url(
operation_name=self._operations["SEASON"],
query_hash=self._query_hashes["SEASON"],
variables=variables,
)
url = build_url("SEASON", variables)
headers = self._get_graphql_headers(authenticated=authenticated)
response = self.http_manager.get(
url, operation="vod_season", headers=headers,
@@ -649,11 +578,7 @@ class JoynVodManager(VodManager):
"first": first, "offset": offset, "blockId": block_id,
"hasToken": has_token,
}
url = self._build_graphql_url(
operation_name=self._operations["COLLECTION_QUERY"],
query_hash=self._query_hashes["COLLECTION_QUERY"],
variables=variables,
)
url = build_url("COLLECTION_QUERY", variables)
headers = self._get_graphql_headers(authenticated=authenticated)
response = self.http_manager.get(
url, operation="vod_collection", headers=headers,
@@ -703,11 +628,7 @@ class JoynVodManager(VodManager):
if not path.startswith("/"):
path = f"/{path}"
variables = {"path": path, "variation": variation}
url = self._build_graphql_url(
operation_name=self._operations["LANDING_PAGE"],
query_hash=self._query_hashes["LANDING_PAGE"],
variables=variables,
)
url = build_url("LANDING_PAGE", variables)
headers = self._get_graphql_headers(authenticated=authenticated)
response = self.http_manager.get(
url, operation="vod_landing_page", headers=headers,
@@ -772,11 +693,7 @@ class JoynVodManager(VodManager):
# force_refresh=True bypasses the cache for one call; the decorator
# strips it before the call reaches this function.
with transport_errors("vod_user_state", "Joyn"):
url = self._build_graphql_url(
operation_name=self._operations["GET_ME_STATE"],
query_hash=self._query_hashes["GET_ME_STATE"],
variables={},
)
url = build_url("GET_ME_STATE")
headers = self._get_graphql_headers(authenticated=True)
response = self.http_manager.get(
url, operation="vod_user_state", headers=headers,
@@ -961,10 +878,6 @@ class JoynVodManager(VodManager):
Returned as the opaque `next_cursor` string; callers pass it back as
`cursor`.
"""
query_hash = self._query_hashes.get("SEARCH", "")
if not query_hash:
raise ServerError("Joyn SEARCH persisted query hash is not configured.")
offset = 0
if cursor:
try:
@@ -979,11 +892,7 @@ class JoynVodManager(VodManager):
"first": page_size,
"offset": offset,
}
url = self._build_graphql_url(
operation_name=self._operations["SEARCH"],
query_hash=query_hash,
variables=variables,
)
url = build_url("SEARCH", variables)
headers = self._get_graphql_headers(authenticated=True)
response = self.http_manager.get(
url, operation="vod_search", headers=headers,
@@ -1220,11 +1129,7 @@ class JoynVodManager(VodManager):
Returns the `asset` object including its `video.id` (a_…).
"""
with transport_errors(f"vod_playable_asset {asset_id}", "Joyn"):
url = self._build_graphql_url(
operation_name=self._operations["PLAYABLE_ASSET"],
query_hash=self._query_hashes["PLAYABLE_ASSET"],
variables={"id": asset_id},
)
url = build_url("PLAYABLE_ASSET", {"id": asset_id})
headers = self._get_graphql_headers(authenticated=authenticated)
response = self.http_manager.get(
url, operation="vod_playable_asset", headers=headers,