nirvana-7777/script.service.ultimate
1
0
Fork 0
mirror of https://github.com/nirvana-7777/script.service.ultimate.git synced 2026-10-06 07:52:26 +02:00
Code Issues Packages Projects Releases Wiki Activity

Simpli - corrections

Browse Source
This commit is contained in:
Nirvana
2026-10-04 08:49:40 +02:00
parent 1d3288b0af
commit 55fbb2cfd0
9 changed files with 1143 additions and 297 deletions
Download Patch File Download Diff File Expand all files Collapse all files
lib/streaming_providers/providers/simplitv/auth.py
+181 -51
View File
@@ -2,8 +2,7 @@
"""
simpliTV authentication.
Two deliberate deviations from the AuthProtocol's *implied* shape (the
protocol only requires the three method names, not their semantics):
Two deliberate deviations from the AuthProtocol's *implied* shape:
1. The simpliTV token is passed as a *query parameter* (GET) or *body
field* (POST), never as an Authorization header. build_headers()
@@ -12,19 +11,35 @@ protocol only requires the three method names, not their semantics):
query parameter `tokenValue` instead of `token`.
2. A second credential, the *device key*, is required by
AcquireContent. It lives on the server: GetDevices returns the
account's registered device, and a device is only registered when
the list is empty. Nothing about it is persisted locally, so a
restart or invalidate() can never leak devices.
AcquireContent. It is discovered from the server (GetDevices) and
registered once if the account has none. Sessions are not
persisted: the token is cheap to re-acquire.
Sessions are not persisted: the token is cheap to re-acquire, and the
device key is re-read from the server.
Device key lifecycle
--------------------
On first use:
1. GetDevices. If the account already has devices, reuse the first
one's key. This is the common case on a re-install.
2. If GetDevices is empty, generate a fresh key and RegisterDevice.
3. GetDevices again and use the first entry. If the server does not
hand a device back, raise -- do not proceed with an unverified
key. The read-back, not the RegisterDevice response, is the
authoritative check (the RegisterDevice response shape is not
verified).
The key is cached in-process only. The settings_manager accessor names
are host-specific and are not guessed. A process restart therefore hits
GetDevices again, which returns the already-registered device.
Error handling: network / JSON failures in the device calls are wrapped
in ServerError by transport_errors(); typed provider errors (AuthError
etc.) pass through unchanged.
"""
import secrets
import string
import threading
import time
from datetime import datetime, timezone
from typing import Any, Dict, Optional
from urllib.parse import parse_qsl, urlencode, urlparse, urlunparse
@@ -33,20 +48,73 @@ from ...base.errors import AuthError, CredentialsError
from ...base.utils.logger import logger
from .constants import SimpliTVConfig, SimpliTVDefaults
from .helpers import parse_iso, transport_errors
def _generate_device_key() -> str:
"""32 lowercase-alnum chars (the addon's format), from a CSPRNG."""
alphabet = string.ascii_lowercase + string.digits
return "".join(secrets.choice(alphabet) for _ in range(32))
"""
Return a 10-digit numeric device key.
The browser log shows the format the server accepts:
"1421859737". Earlier versions of this provider used a 32-char
lowercase-alnum string (ported from the addon); that shape is not
what the API advertises.
"""
return f"{secrets.randbelow(10**10):010d}"
def _mask(key: Optional[str]) -> str:
"""Short tag for logs; never the full key."""
if not key or len(key) < 6:
return "***"
return f"{key[:2]}...{key[-2:]}"
def _parse_expiry(iso: Optional[str]) -> Optional[int]:
"""
Parse `tokenExpirationTime` to seconds from now. Returns None on
missing / malformed input or a time already in the past.
"""
dt = parse_iso(iso)
if dt is None:
return None
delta = (dt - datetime.now(timezone.utc)).total_seconds()
if delta <= 0:
return None
return int(delta)
def _extract_device_key(entry: Any) -> str:
"""
Return the device key from one GetDevices entry.
The response shape is unverified (the browser capture never calls
GetDevices). "key" is the addon's field name; "deviceKey" is the
field the RegisterDevice payload uses. Either is accepted; if
neither is present (or the entry is not an object), raise rather
than silently return empty.
"""
if not isinstance(entry, dict):
raise AuthError(
f"simpliTV: GetDevices entry is not an object: "
f"{type(entry).__name__}"
)
for field in ("key", "deviceKey"):
value = entry.get(field)
if value:
return str(value)
raise AuthError(
f"simpliTV: GetDevices entry has no device key field "
f"(keys: {sorted(entry.keys())!r})"
)
class SimpliTVAuth:
"""
Authenticator for simpliTV.
Not a subclass of any base class -- matches AuthProtocol by shape for
the three shared methods, plus simpliTV-specific helpers
Not a subclass of any base class -- matches AuthProtocol by shape
for the three shared methods, plus simpliTV-specific helpers
(with_token, auth_body) and the device-key accessor.
Thread-safe: the backend serves requests from several threads, and
@@ -93,8 +161,8 @@ class SimpliTVAuth:
"""
Return *base* headers for the simpliTV API.
The token is deliberately NOT placed here -- see with_token() and
auth_body(). The `token` argument is accepted for protocol
The token is deliberately NOT placed here -- see with_token()
and auth_body(). The `token` argument is accepted for protocol
compatibility and ignored.
"""
return self.config.get_api_headers()
@@ -120,7 +188,15 @@ class SimpliTVAuth:
*,
param: str = SimpliTVDefaults.TOKEN_PARAM,
) -> str:
"""Append the token (as `param`) and any extra params to `url`."""
"""
Append the token (as `param`) and any extra params to `url`.
NOTE: this parses and re-encodes the existing query string, so
the pre-encoded `$headers` blob is re-encoded. The key/value
pairs are identical, but the bytes differ from the browser
capture. If the server ever proves sensitive to that, build the
query string by hand instead.
"""
token = self.get_access_token()
parsed = urlparse(url)
query = dict(parse_qsl(parsed.query))
@@ -139,23 +215,37 @@ class SimpliTVAuth:
"""
Return the account's device key.
Reads GetDevices first and uses the registered device; registers
a new one only if the account has none.
Order of preference:
1. In-process cache.
2. GetDevices -- reuse the first registered device.
3. RegisterDevice with a fresh key, then GetDevices again.
"""
with self._lock:
if self._device_key:
return self._device_key
devices = self._list_devices()
if not devices:
self._register_device()
devices = self._list_devices()
if devices:
self._device_key = _extract_device_key(devices[0])
logger.debug(
f"simpliTV[{self.country}]: reusing device "
f"{_mask(self._device_key)}"
)
return self._device_key
self._register_device(_generate_device_key())
devices = self._list_devices()
if not devices:
raise AuthError(
"simpliTV: device registration returned no devices"
"simpliTV: device registration did not result in a "
"registered device (GetDevices is still empty)"
)
self._device_key = devices[0]["key"]
logger.debug(f"simpliTV[{self.country}]: device key resolved")
self._device_key = _extract_device_key(devices[0])
logger.debug(
f"simpliTV[{self.country}]: registered device "
f"{_mask(self._device_key)}"
)
return self._device_key
# ------------------------------------------------------------------
@@ -166,25 +256,32 @@ class SimpliTVAuth:
creds = self._resolve_credentials()
logger.debug(f"simpliTV[{self.country}]: logging in")
# The browser sends lowercase `login`/`password` and does NOT
# send LongExpiration.
payload = {
"Login": creds["username"],
"Password": creds["password"],
"LongExpiration": "true",
"platformCodename": self.config.platform_codename,
"login": creds["username"],
"password": creds["password"],
}
resp = self.http_manager.post(
self.config.authenticate_url(),
json=payload,
headers=self.build_headers(),
)
token_value = resp.json().get("token")
data = resp.json()
token_value = data.get("token")
if not token_value:
raise AuthError("simpliTV: no token in authenticate response")
expires_in = (
_parse_expiry(data.get("tokenExpirationTime"))
or SimpliTVDefaults.TOKEN_LIFETIME_SECONDS
)
return BaseAuthToken(
access_token=token_value,
token_type="token",
expires_in=SimpliTVDefaults.TOKEN_LIFETIME_SECONDS,
expires_in=expires_in,
issued_at=time.time(),
)
@@ -206,39 +303,72 @@ class SimpliTVAuth:
"""
Seam for credentials held by the host's settings_manager.
TODO(host): the template expects a fallback to stored
credentials, but the settings_manager accessor name is not
known here and is deliberately not guessed. Implement against
the real base API; return None when nothing is stored.
TODO(host): implement against the real base API; return None
when nothing is stored.
"""
return None
# ------------------------------------------------------------------
# Device list / registration
# ------------------------------------------------------------------
def _list_devices(self) -> list:
"""
GET /v1/Devices/GetDevices. Returns the raw device list (may be
empty; that is not an error).
with_token() runs outside transport_errors so a login failure
keeps its own type.
"""
url = self.with_token(
self.config.get_devices_url(),
{"platformCodename": self.config.platform_codename},
)
resp = self.http_manager.get(url, headers=self.build_headers())
return resp.json().get("devices") or []
with transport_errors("GetDevices"):
resp = self.http_manager.get(
url, headers=self.config.get_api_headers()
)
devices = resp.json().get("devices")
return devices if isinstance(devices, list) else []
def _register_device(self) -> None:
logger.info(
f"simpliTV[{self.country}]: no registered device on the "
f"account, registering one"
def _register_device(self, device_key: str) -> None:
"""
Register a device key with the server.
Payload matches the browser capture for RegisterDevice. The
response shape is not verified, so only an explicit
`result.success == false` is treated as a refusal; anything
else is accepted and the caller's GetDevices read-back decides
whether registration actually happened.
"""
logger.debug(
f"simpliTV[{self.country}]: registering device "
f"{_mask(device_key)}"
)
payload = {
"deviceKey": _generate_device_key(),
"deviceName": SimpliTVDefaults.DEVICE_NAME,
"generalDeviceType": SimpliTVDefaults.DEVICE_GENERAL_TYPE,
"operatingSystem": SimpliTVDefaults.DEVICE_OS,
"platformCodename": self.config.platform_codename,
"pushToken": "",
"userAgent": self.config.user_agent,
"deviceKey": device_key,
"userToken": self.get_access_token(),
"pushToken": "",
"generalDeviceType": SimpliTVDefaults.DEVICE_GENERAL_TYPE,
"deviceName": SimpliTVDefaults.DEVICE_NAME,
"browserVersion": SimpliTVDefaults.DEVICE_BROWSER_VERSION,
"userAgent": self.config.user_agent,
"operatingSystem": SimpliTVDefaults.DEVICE_OS,
"versionOs": SimpliTVDefaults.DEVICE_OS_VERSION,
}
self.http_manager.post(
self.config.register_device_url(),
json=payload,
headers=self.build_headers(),
)
with transport_errors("RegisterDevice"):
resp = self.http_manager.post(
self.config.register_device_url(),
json=payload,
headers=self.build_headers(),
)
try:
data = resp.json()
except ValueError:
return # empty / non-JSON body: rely on the read-back
result = data.get("result") if isinstance(data, dict) else None
if isinstance(result, dict) and result.get("success") is False:
raise AuthError(
f"simpliTV: RegisterDevice refused (response: {data!r})"
)
lib/streaming_providers/providers/simplitv/catchup_manager.py
+64 -11
View File
@@ -11,13 +11,28 @@ and this manager mirrors both:
caller supplies epg_id (the programme codename, an EPG broadcast_id,
or a prog: id).
2. Restart from the beginning. The addon plays the *live* manifest, which
carries a 3 hour DVR window, and seeks
(window - age - margin) seconds into it. A manifest URL cannot express
a seek, so this is exposed separately as get_restart_manifest(),
returning (url, seek_seconds). get_catchup_manifest never answers a
restart request with a bare live URL: the ABC can only hand back a
URL, and the caller would silently play live.
2. Restart from the beginning. The addon plays the *live* manifest,
which carries a DVR window, and seeks
(window - age - margin) seconds into it. A manifest URL cannot
express a seek, so this is exposed separately as
get_restart_manifest(), returning (url, seek_seconds).
get_catchup_manifest never answers a restart request with a bare live
URL: the ABC can only hand back a URL, and the caller would silently
play live.
DVR window
----------
The window is per-channel and advertised in the AcquireContent
response as AdditionalInfo.Epg_TimeshiftSeconds (the browser log shows
7200, 10800 and 14400 across channels). get_restart_manifest reads
that value from the channel manager's AcquireContent response. If the
field is missing, the smallest observed window (2h) is used, so a seek
can never land outside the real window.
The ABC exposes catchup_window_hours as a single integer, which cannot
express a per-channel value. It returns the same smallest observed
window as the conservative answer: a host that trusts it will only
offer replays the shortest channel allows.
Contract
--------
@@ -31,7 +46,7 @@ from typing import List, Optional, Tuple
from ...base.errors import BadRequestError, NotFoundError
from ...base.managers import CatchupManager
from ...base.models import DRMConfig
from ...base.models.drm_config import DRMConfig
from .channel_manager import (
parse_catchup_id,
@@ -41,6 +56,11 @@ from .channel_manager import (
from .constants import SimpliTVDefaults
# Smallest timeshift window observed across channels (7200s = 2h). Used
# as catchup_window_hours and as the per-channel fallback.
_MIN_TIMESHIFT_HOURS = 2
class SimpliTVCatchupManager(CatchupManager):
"""Catchup for simpliTV."""
@@ -67,7 +87,15 @@ class SimpliTVCatchupManager(CatchupManager):
@property
def catchup_window_hours(self) -> int:
return SimpliTVDefaults.CATCHUP_WINDOW_HOURS
"""
Conservative global window, in hours.
The API advertises per-channel windows (2h, 3h, 4h); the ABC
only exposes a single integer, so the smallest observed value
is returned. get_restart_manifest uses the real per-channel
value.
"""
return _MIN_TIMESHIFT_HOURS
# ------------------------------------------------------------------
# Abstract method
@@ -118,6 +146,10 @@ class SimpliTVCatchupManager(CatchupManager):
start_time is not given. None when the start is in the future,
outside the window (less the safety margin), or the channel has
no manifest.
The window is read from AcquireContent's
AdditionalInfo.Epg_TimeshiftSeconds for the channel, with the
2h minimum as fallback.
"""
codename, embedded = _channel_and_ts(content_id)
if start_time <= 0 and embedded:
@@ -125,7 +157,12 @@ class SimpliTVCatchupManager(CatchupManager):
if start_time <= 0 or self._channels is None:
return None
window = SimpliTVDefaults.CATCHUP_WINDOW_HOURS * 3600
try:
acquire = self._channels.acquire_content(codename)
except NotFoundError:
return None
window = _timeshift_seconds(acquire)
margin = SimpliTVDefaults.CATCHUP_SEEK_MARGIN_SECONDS
age = int(time.time()) - start_time
if age < 0 or age >= window - margin:
@@ -156,7 +193,7 @@ class SimpliTVCatchupManager(CatchupManager):
"""
DRM for catchup: the programme's own (epg_id) when replaying,
otherwise the channel's live DRM (restart plays the live
stream). Scheme selection stays in the channel manager.
stream).
"""
codename, _ = _channel_and_ts(content_id)
if self._channels is None:
@@ -180,3 +217,19 @@ def _channel_and_ts(content_id: str) -> Tuple[str, Optional[int]]:
raise BadRequestError(
f"simpliTV: not a catchup/live id: {content_id!r}"
)
def _timeshift_seconds(acquire: dict) -> int:
"""
Read AdditionalInfo.Epg_TimeshiftSeconds from an AcquireContent
response. Accepts an int or a numeric string; a missing or
non-positive value falls back to the 2h minimum.
"""
info = acquire.get("AdditionalInfo") or {}
try:
seconds = int(info.get("Epg_TimeshiftSeconds"))
except (TypeError, ValueError):
seconds = 0
if seconds > 0:
return seconds
return _MIN_TIMESHIFT_HOURS * 3600
lib/streaming_providers/providers/simplitv/channel_manager.py
+158 -68
View File
@@ -21,16 +21,26 @@ Manifest and DRM are folded into this manager because both arrive in one
AcquireContent response (README: DRM shares state with the manifest step,
so it folds). The response is cached briefly so the manifest and DRM
steps of one playback share a single request.
Format selection
----------------
The AcquireContent response carries DASH (Type 9) and/or HLS (Type 2)
entries. Protected content prefers DASH (inputstream.adaptive needs it
for Widevine/PlayReady; PSSH extraction is more reliable). Unprotected
content prefers HLS, which is what the CDN has always served the addon.
prefer_dash() runs only when a protected asset offers no DASH entry.
"""
import threading
import time
from typing import Any, Dict, List, Optional
from typing import Any, Dict, List, Optional, Tuple
from ...base.errors import BadRequestError, NotFoundError
from ...base.managers import ChannelManager
from ...base.models import Channel
from ...base.models.drm_models import DRMConfig, DRMSystem, LicenseConfig
from ...base.models.drm_config import DRMConfig
from ...base.models.drm_systems import DRMSystem
from ...base.models.license_config import LicenseConfig
from ...base.utils.logger import logger
from .constants import SimpliTVDefaults
@@ -86,13 +96,13 @@ class SimpliTVChannelManager(ChannelManager):
"""
Return the channel list.
The tile endpoint yields only codenames (the existing addon reads
nothing else from it, and only from the first group), so names
and logos are derived locally -- see logos.py. EPG metadata is
not fetched here; that keeps this call independent of EPG
availability.
The tile endpoint yields only codenames, so names and logos are
derived locally -- see logos.py. EPG metadata is not fetched
here; that keeps this call independent of EPG availability.
NOTE: token is in the body (auth_body), not a header. The
`$headers` query string is baked into channel_tiles_url().
"""
# NOTE: token is in the body (auth_body), not a header.
body = self.auth.auth_body({
"isParentalControlEnabled": "false",
"platformCodename": self.config.platform_codename,
@@ -135,13 +145,14 @@ class SimpliTVChannelManager(ChannelManager):
"""
Return the manifest for a live:, rec: or prog: content id.
DRM-protected streams are rewritten to DASH (what
inputstream.adaptive needs); unprotected streams keep the HLS URL
the API returned.
Protected content prefers DASH, unprotected content prefers
HLS. If a protected asset has no DASH entry, prefer_dash() is
applied to the URL that was selected.
The returned URL does NOT contain the API token (it was only used
for the AcquireContent request), and manifest/segment requests
need just a User-Agent -- see get_channel_manifest_headers().
The returned URL does NOT contain the API token (it was only
used for the AcquireContent request), and manifest/segment
requests need just a User-Agent -- see
SimpliTVConfig.get_stream_headers().
"""
codename = codename_from_playable_id(content_id)
@@ -150,30 +161,37 @@ class SimpliTVChannelManager(ChannelManager):
acquire = self.acquire_content(codename, refresh=True)
try:
manifest_url = acquire["MediaFiles"][0]["Formats"][0]["Url"]
formats = acquire["MediaFiles"][0]["Formats"]
except (KeyError, IndexError, TypeError):
raise NotFoundError(
f"simpliTV: no manifest in AcquireContent for {codename!r}"
f"simpliTV: no media files in AcquireContent for "
f"{codename!r}"
)
if acquire.get("DrmInfo"):
protected = bool(acquire.get("DrmInfo"))
manifest_url, fmt_type = _select_format(formats, protected=protected)
if not manifest_url:
raise NotFoundError(
f"simpliTV: no manifest URL in AcquireContent for "
f"{codename!r}"
)
# Protected asset without a DASH entry: rewrite to the DASH
# equivalent. Never applied to unprotected content.
if protected and fmt_type != SimpliTVDefaults.FORMAT_TYPE_DASH:
manifest_url = prefer_dash(manifest_url)
return manifest_url
# ------------------------------------------------------------------
# Player request headers
# ------------------------------------------------------------------
# The base layer's method names for these hooks were not available
# when this was written; the names below follow the review of the
# package. If the base calls different names, rename accordingly --
# until then these are simply unused.
def get_channel_manifest_headers(self, content_id: str, **kw) -> dict:
"""User-Agent only -- see SimpliTVConfig.get_stream_headers()."""
"""User-Agent + Origin -- see SimpliTVConfig.get_stream_headers()."""
return self.config.get_stream_headers()
def get_segment_headers(self, content_id: str, **kw) -> dict:
"""User-Agent only: the CDN does not authenticate segments."""
"""User-Agent + Origin: the CDN does not authenticate segments."""
return self.config.get_stream_headers()
# ------------------------------------------------------------------
@@ -184,56 +202,90 @@ class SimpliTVChannelManager(ChannelManager):
self, content_id: str, **kw
) -> List[DRMConfig]:
"""
Return DRM for a live:, rec: or prog: content id.
Return DRM configs for a live:, rec: or prog: content id.
Both Widevine (priority 1) and PlayReady (priority 2) are
returned when the AcquireContent response advertises them.
inputstream.adaptive selects whichever CDM the platform
actually supports, so the priority ordering is safe on a
PlayReady-only device. FairPlay is ignored: Kodi cannot play
it.
Reads the response cached by get_channel_manifest when it is
still fresh, otherwise fetches it.
still fresh, otherwise fetches it. Unprotected content returns
[] (per the None-vs-exception rule).
"""
codename = codename_from_playable_id(content_id)
acquire = self.acquire_content(codename)
drm_info = acquire.get("DrmInfo") or []
if not drm_info:
# Unprotected content is not a failure: [] per the
# None-vs-exception rule.
return []
wanted = (
SimpliTVDefaults.DRM_INDEX_PLAYREADY
if self.config.prefer_playready
else SimpliTVDefaults.DRM_INDEX_WIDEVINE
)
if wanted < len(drm_info):
index = wanted
else:
# Requested scheme isn't offered; some assets are
# single-scheme, so use what is there.
index = 0
system = (
DRMSystem.PLAYREADY
if index == SimpliTVDefaults.DRM_INDEX_PLAYREADY
else DRMSystem.WIDEVINE
)
# Index by DrmSystem string, not position: the server may add
# or reorder systems.
by_system: Dict[str, dict] = {}
for entry in drm_info:
name = entry.get("DrmSystem")
if name and name not in by_system:
by_system[name] = entry
entry = drm_info[index]
license_url = entry.get("LicenseServerUrl")
if not license_url:
return []
configs: List[DRMConfig] = []
widevine = by_system.get(SimpliTVDefaults.DRM_SYSTEM_WIDEVINE)
if widevine and widevine.get("LicenseServerUrl"):
configs.append(self._build_drm_config(
system=DRMSystem.WIDEVINE,
priority=SimpliTVDefaults.DRM_PRIORITY_WIDEVINE,
entry=widevine,
))
playready = by_system.get(SimpliTVDefaults.DRM_SYSTEM_PLAYREADY)
if playready and playready.get("LicenseServerUrl"):
configs.append(self._build_drm_config(
system=DRMSystem.PLAYREADY,
priority=SimpliTVDefaults.DRM_PRIORITY_PLAYREADY,
entry=playready,
))
return configs
def _build_drm_config(
self,
*,
system: DRMSystem,
priority: int,
entry: dict,
) -> DRMConfig:
"""
Build one DRMConfig from a single DrmInfo entry.
The challenge custom data is passed *raw* (base64) in the
`drmchallengecustomdata` header, matching the browser capture.
It is deliberately not URL-quoted here.
UNVERIFIED: whether the host's LicenseConfig / inputstream
.adaptive serialisation of req_headers preserves `+`, `/` and
`=` unchanged cannot be determined from this package. Check on
a real device (a mangled value shows up as a licence-server
4xx); if it is altered, the quoting belongs in LicenseConfig,
not here.
The licence body is the raw CDM challenge, injected by ISA via
the {CHA-RAW} placeholder (see SimpliTVDefaults.REQ_DATA_CHA_RAW).
"""
challenge = entry.get("DrmChallengeCustomData")
return [
DRMConfig(
system=system,
priority=1,
license=LicenseConfig(
server_url=license_url,
req_headers=self.config.get_license_headers(challenge),
# Body = the raw CDM challenge (the addon's R{SSM}).
req_data="{CHA-RAW}",
use_http_get_request=False,
),
)
]
return DRMConfig(
system=system,
priority=priority,
license=LicenseConfig(
server_url=entry["LicenseServerUrl"],
req_headers=self.config.get_license_headers(challenge),
req_data=SimpliTVDefaults.REQ_DATA_CHA_RAW,
use_http_get_request=False,
),
)
# ------------------------------------------------------------------
# AcquireContent (shared with SimpliTVCatchupManager via this class)
@@ -245,6 +297,11 @@ class SimpliTVChannelManager(ChannelManager):
"""
GET /Player/AcquireContent for a codename.
The request carries the token, the device key, the platform
codename, and a millisecond cache-buster `t=` -- all in the
query string. There is NO `$headers` parameter on this
endpoint.
Responses are cached for PLAYBACK_CACHE_TTL seconds so the
manifest and DRM steps of one playback make one request;
refresh=True bypasses the cache (and repopulates it).
@@ -257,15 +314,13 @@ class SimpliTVChannelManager(ChannelManager):
logger.debug(
f"simpliTV: AcquireContent {codename!r} (refresh={refresh})"
)
# NOTE: token is in the URL (via with_token), not a header. It
# authenticates THIS request only; the manifest URL returned in
# the response does not carry it.
url = self.auth.with_token(
self.config.acquire_content_url(),
{
"platformCodename": self.config.platform_codename,
"deviceKey": self.auth.get_device_key(),
"codename": codename,
"t": int(time.time() * 1000),
},
)
with transport_errors(f"AcquireContent for {codename!r}"):
@@ -312,6 +367,41 @@ class SimpliTVChannelManager(ChannelManager):
)
# ----------------------------------------------------------------------
# AcquireContent format selection
# ----------------------------------------------------------------------
def _select_format(
formats: List[dict], *, protected: bool
) -> Tuple[Optional[str], Optional[int]]:
"""
Return (manifest URL, format Type) from a Formats array.
Protected content prefers DASH (Type 9), then HLS (Type 2).
Unprotected content prefers HLS, then DASH. The first entry with a
URL is a final fallback (its Type is returned as-is, possibly
None). Array order is never assumed.
"""
if protected:
order = (
SimpliTVDefaults.FORMAT_TYPE_DASH,
SimpliTVDefaults.FORMAT_TYPE_HLS,
)
else:
order = (
SimpliTVDefaults.FORMAT_TYPE_HLS,
SimpliTVDefaults.FORMAT_TYPE_DASH,
)
for wanted in order:
for fmt in formats:
if fmt.get("Type") == wanted and fmt.get("Url"):
return fmt["Url"], wanted
for fmt in formats:
if fmt.get("Url"):
return fmt["Url"], fmt.get("Type")
return None, None
# ----------------------------------------------------------------------
# Content-id grammar -- module scope so every manager can import the
# parsers without circular dependencies. All raise BadRequestError on
@@ -352,8 +442,8 @@ def parse_catchup_id(content_id: str):
"""
Return (channel codename, start_ts) from catchup:<channel>@<unix_ts>.
The @<ts> suffix is required -- a bare catchup:<channel> is a grammar
error, not a live id in disguise.
The @<ts> suffix is required -- a bare catchup:<channel> is a
grammar error, not a live id in disguise.
"""
rest = _strip_prefix(
content_id, SimpliTVDefaults.CATCHUP_PREFIX, "catchup"
@@ -392,8 +482,8 @@ def codename_from_playable_id(content_id: str) -> str:
def programme_codename_from_epg_id(epg_id: str) -> str:
"""
Programme codename from whatever the host passes as epg_id: an EPG
broadcast_id (simplitv:<channel>:<start>:<programme>), a prog: id, or
the bare programme codename.
broadcast_id (simplitv:<channel>:<start>:<programme>), a prog: id,
or the bare programme codename.
"""
if not epg_id:
raise BadRequestError("simpliTV: empty epg_id")
@@ -406,4 +496,4 @@ def programme_codename_from_epg_id(epg_id: str) -> str:
return parts[3]
if epg_id.startswith(SimpliTVDefaults.PROGRAMME_PREFIX):
return parse_programme_id(epg_id)
return epg_id
return epg_id
lib/streaming_providers/providers/simplitv/constants.py
+154 -65
View File
@@ -13,8 +13,13 @@ class SimpliTVDefaults:
PROVIDER_NAME = "simplitv"
PROVIDER_LOGO = "https://files.app.simplitv.at/files/orf1-hd-bunt.png"
BASE_URL = "https://api.app.simplitv.at"
WEBSITE = "https://streaming.simplitv.at"
# --- Endpoints / hosts ---------------------------------------------
# Confirmed against the browser capture: the API host is
# api.app.austrostream.at (multi-tenant; simpliTV is the
# X-Tenant-Codename: simplitv tenant). The web app itself lives on
# streaming.simpli.at, NOT streaming.simplitv.at.
BASE_URL = "https://api.app.austrostream.at"
WEBSITE = "https://streaming.simpli.at"
# --- Endpoint paths ------------------------------------------------
PATH_AUTHENTICATE = "/v1/OrsUser/Authenticate"
@@ -22,29 +27,48 @@ class SimpliTVDefaults:
PATH_REGISTER_DEVICE = "/v1/Devices/RegisterDevice"
PATH_CHANNEL_TILES = "/v1/EpgTile/FilterChannelTiles"
PATH_PROGRAM_TILES = "/v1/EpgTile/FilterProgramTiles"
PATH_EPG_AVAILABLE_DAYS = "/v1/EpgTile/GetAvailableDays"
PATH_TILE_DETAILS = "/v2/Tile/GetTiles"
PATH_ACQUIRE_CONTENT = "/v1/Player/AcquireContent"
PATH_GET_RECORDINGS = "/v2/Pvr/GetRecordings"
PATH_SCHEDULE_RECORDING = "/v1/Pvr/ScheduleRecording"
PATH_DELETE_RECORDING = "/v1/Pvr/DeleteRecording"
PATH_GET_USER_PRODUCTS = "/v1/IpottTransaction/GetUserProducts"
PATH_GET_ALL_REMINDERS = "/v1/Reminder/GetAllReminders"
# --- Token transport -----------------------------------------------
# Most endpoints take `token`; GetRecordings (v2) takes `tokenValue`.
TOKEN_PARAM = "token"
TOKEN_PARAM_RECORDINGS = "tokenValue"
# The existing addon sends the headers a second time as a `$headers`
# query parameter on the two anonymous EPG endpoints. Mirrored here
# because it is known to work.
EPG_HEADERS_QUERY = (
# The API expects a `$headers` query parameter carrying a JSON blob
# of the request headers. Two shapes appear in the browser capture:
#
# * WITH Content-Type — Authenticate, FilterChannelTiles,
# GetAvailableDays, RegisterDevice
# * WITHOUT Content-Type — GetUserProducts, GetRecordings,
# GetAllReminders, FilterProgramTiles
#
# Pre-encoded (percent-escaped) so they can be appended verbatim.
EPG_HEADERS_QUERY_WITH_CT = (
"$headers=%7B%22Content-Type%22:%22application%2Fjson%3B"
"charset%3Dutf-8%22,%22X-Api-Date-Format%22:%22iso%22,"
"%22X-Api-Resource-Language-Context%22:%22de%22,"
"%22X-Api-Camel-Case%22:true%7D"
)
EPG_HEADERS_QUERY_NO_CT = (
"$headers=%7B%22X-Api-Date-Format%22:%22iso%22,"
"%22X-Api-Resource-Language-Context%22:%22de%22,"
"%22X-Api-Camel-Case%22:true%7D"
)
# --- Platform / device identity ------------------------------------
PLATFORM_CODENAME = "www"
TENANT_CODENAME = "simplitv"
RESOURCE_LANGUAGE_CONTEXT = "de"
DEVICE_NAME = "Firefox"
DEVICE_BROWSER_VERSION = "110.0"
DEVICE_GENERAL_TYPE = "2"
DEVICE_OS = "Windows"
DEVICE_OS_VERSION = "10"
@@ -60,14 +84,34 @@ class SimpliTVDefaults:
TIMEOUT = 30
# --- DRM -----------------------------------------------------------
# DrmInfo indices in the AcquireContent response:
# [0] = PlayReady, [1] = Widevine.
DRM_INDEX_PLAYREADY = 0
DRM_INDEX_WIDEVINE = 1
# DrmInfo is returned as a list of objects with a `DrmSystem` string
# (PlayReady / Widevine / FairPlay). Kodi cannot play FairPlay, and
# Widevine is more broadly available than PlayReady, so we emit:
#
# Widevine priority 1
# PlayReady priority 2
#
# inputstream.adaptive picks the lowest-numbered system whose CDM is
# actually present on the platform, so this ordering is safe even on
# a PlayReady-only device.
DRM_SYSTEM_WIDEVINE = "Widevine"
DRM_SYSTEM_PLAYREADY = "PlayReady"
DRM_PRIORITY_WIDEVINE = 1
DRM_PRIORITY_PLAYREADY = 2
# ISA placeholder for the raw CDM challenge in the license body.
# LicenseConfig auto-base64-encodes plain strings, so "{CHA-RAW}"
# is stored (and sent to ISA) as its base64 form; ISA decodes it
# back to the literal placeholder and substitutes the challenge.
REQ_DATA_CHA_RAW = "{CHA-RAW}"
# MediaFiles[].Formats[].Type values in AcquireContent responses.
FORMAT_TYPE_HLS = 2
FORMAT_TYPE_DASH = 9
# --- Auth ----------------------------------------------------------
# simpliTV does not return expires_in; refresh well inside any
# plausible server-side expiry.
# Fallback only: the response now carries `tokenExpirationTime`,
# which is preferred when present.
TOKEN_LIFETIME_SECONDS = 12 * 3600
# --- Caches --------------------------------------------------------
@@ -77,35 +121,27 @@ class SimpliTVDefaults:
EPG_CACHE_MAX = 64
# --- EPG -----------------------------------------------------------
# The server advertises its actual window via GetAvailableDays;
# these are fallbacks used only when that call fails.
EPG_PAST_DAYS = 7
EPG_FUTURE_DAYS = 7
# The API is queried in day-sized windows, like the existing addon.
EPG_FUTURE_DAYS = 14
EPG_CHUNK_HOURS = 24
# --- Content-id grammar -------------------------------------------
# live:<channel codename> -> SimpliTVChannelManager
# rec:<programme codename> -> SimpliTVChannelManager (same
# AcquireContent call as live)
# prog:<programme codename> -> SimpliTVChannelManager (EPG
# programme replay; also the id
# form scheduling a recording
# needs)
# catchup:<channel>@<unix_ts> -> SimpliTVCatchupManager
# (restart-from-beginning)
LIVE_PREFIX = "live:"
RECORDING_PREFIX = "rec:"
PROGRAMME_PREFIX = "prog:"
CATCHUP_PREFIX = "catchup:"
CATCHUP_TS_SEPARATOR = "@"
# EPG broadcast ids: simplitv:<channel codename>:<start>:<programme
# codename>. The last segment is what catchup takes as epg_id.
BROADCAST_ID_PREFIX = "simplitv:"
# --- Catchup -------------------------------------------------------
# "Restart" plays the live manifest, which carries a 3 hour DVR
# window; the player seeks (window - age - margin) seconds into it.
CATCHUP_WINDOW_HOURS = 3
# --- Catchup / timeshift ------------------------------------------
# The DVR window is per-channel (AdditionalInfo.Epg_TimeshiftSeconds
# in AcquireContent; 7200, 10800 and 14400 in the browser capture).
# SimpliTVCatchupManager reads the per-channel value and falls back
# to its own _MIN_TIMESHIFT_HOURS when the field is missing -- there
# is deliberately no global "catchup window" constant, because the
# API has none.
CATCHUP_SEEK_MARGIN_SECONDS = 35
@@ -114,10 +150,8 @@ class SimpliTVConfig:
Per-instance configuration.
Attributes relied on by provider.py / managers:
user_agent -- passed to _setup_http_manager.
timeout -- passed to _setup_http_manager.
prefer_playready -- selects PlayReady (True) or Widevine (False)
in get_channel_drm. Defaults to Widevine.
user_agent -- passed to _setup_http_manager.
timeout -- passed to _setup_http_manager.
"""
def __init__(self, config_dict: Optional[dict] = None):
@@ -125,7 +159,6 @@ class SimpliTVConfig:
self.base_url = config.get("base_url", SimpliTVDefaults.BASE_URL)
self.user_agent = config.get("user_agent", SimpliTVDefaults.USER_AGENT)
self.timeout = config.get("timeout", SimpliTVDefaults.TIMEOUT)
self.prefer_playready = bool(config.get("prefer_playready", False))
self.platform_codename = config.get(
"platform_codename", SimpliTVDefaults.PLATFORM_CODENAME
)
@@ -136,53 +169,70 @@ class SimpliTVConfig:
"""
Headers for the JSON API endpoints.
These are *base* headers only -- the simpliTV token is not a
header (see auth.py). Callers put the token in the query string
(auth.with_token) or the body (auth.auth_body).
The simpliTV token is NOT a header (see auth.py): it is passed
as a URL query parameter or body field. These are the base
headers the browser sends alongside every API call.
"""
return {
"User-Agent": self.user_agent,
"Content-type": "application/json;charset=utf-8",
"Accept": "application/json, text/plain, */*",
"Content-Type": "application/json;charset=utf-8",
"X-Api-Date-Format": SimpliTVDefaults.HEADER_API_DATE_FORMAT,
"X-Api-Camel-Case": SimpliTVDefaults.HEADER_API_CAMEL_CASE,
"X-Api-Resource-Language-Context":
SimpliTVDefaults.RESOURCE_LANGUAGE_CONTEXT,
"X-Tenant-Codename": SimpliTVDefaults.TENANT_CODENAME,
"Origin": SimpliTVDefaults.WEBSITE,
"Referer": f"{SimpliTVDefaults.WEBSITE}/",
}
def get_manifest_headers(self) -> dict:
"""Headers for /Player/AcquireContent (same UA as the API)."""
"""Headers for /Player/AcquireContent (UA + JSON accept)."""
return {
"User-Agent": self.user_agent,
"Accept": "application/json",
"X-Tenant-Codename": SimpliTVDefaults.TENANT_CODENAME,
"Origin": SimpliTVDefaults.WEBSITE,
}
def get_stream_headers(self) -> dict:
"""
Headers for manifest and segment requests made by the player.
The addon gives inputstream the User-Agent and nothing else: the
token is NOT part of the manifest URL (it only authenticates the
AcquireContent request) and the CDN does not authenticate
segments. Do not derive these from build_headers(), which carries
JSON / X-Api-* / Referer headers the CDN never saw from the addon.
The CDN does not authenticate: the addon gives inputstream only
the User-Agent. The token in the AcquireContent URL
authenticates the API call, not the CDN fetch. Origin is added
to match the browser capture; the CDN sends
access-control-allow-origin: * so it is not required.
"""
return {"User-Agent": self.user_agent}
return {
"User-Agent": self.user_agent,
"Origin": SimpliTVDefaults.WEBSITE,
}
def get_license_headers(self, challenge: Optional[str] = None) -> dict:
"""
Headers for the DRM licence request.
The challenge custom data travels as a *header*, URL-quoted; the
body is the raw CDM challenge (see channel_manager.get_channel_drm).
"""
from urllib.parse import quote
The challenge custom data travels as a *header*, sent raw: the
value is already base64 and must not be URL-quoted. The request
body is the raw CDM challenge, injected by ISA via the
{CHA-RAW} placeholder.
The dict is passed to LicenseConfig, which urlencodes it for
the inputstream.adaptive DRM JSON. ISA decodes the
percent-encoding before sending the header, so the DRM server
sees the raw base64 -- matching the browser. If a real device
ever rejects the licence, the first thing to check is whether
ISA truly decodes req_headers values.
"""
headers = {
"User-Agent": self.user_agent,
"Referer": f"{SimpliTVDefaults.WEBSITE}/",
"Origin": SimpliTVDefaults.WEBSITE,
"Content-Type": "application/octet-stream",
}
if challenge:
headers["drmchallengecustomdata"] = quote(challenge)
headers["drmchallengecustomdata"] = challenge
return headers
# ----- URL builders -----------------------------------------------
@@ -190,38 +240,77 @@ class SimpliTVConfig:
def _url(self, path: str) -> str:
return f"{self.base_url}{path}"
def _url_with_headers(self, path: str, *, include_ct: bool = True) -> str:
"""Append the pre-encoded `$headers=` query string to `path`."""
q = (
SimpliTVDefaults.EPG_HEADERS_QUERY_WITH_CT
if include_ct
else SimpliTVDefaults.EPG_HEADERS_QUERY_NO_CT
)
return f"{self._url(path)}?{q}"
def authenticate_url(self) -> str:
return self._url(SimpliTVDefaults.PATH_AUTHENTICATE)
return self._url_with_headers(
SimpliTVDefaults.PATH_AUTHENTICATE, include_ct=True
)
def get_devices_url(self) -> str:
return self._url(SimpliTVDefaults.PATH_GET_DEVICES)
return self._url_with_headers(
SimpliTVDefaults.PATH_GET_DEVICES, include_ct=False
)
def register_device_url(self) -> str:
return self._url(SimpliTVDefaults.PATH_REGISTER_DEVICE)
return self._url_with_headers(
SimpliTVDefaults.PATH_REGISTER_DEVICE, include_ct=True
)
def channel_tiles_url(self) -> str:
return self._url(SimpliTVDefaults.PATH_CHANNEL_TILES)
return self._url_with_headers(
SimpliTVDefaults.PATH_CHANNEL_TILES, include_ct=True
)
def program_tiles_url(self) -> str:
return (
f"{self._url(SimpliTVDefaults.PATH_PROGRAM_TILES)}"
f"?{SimpliTVDefaults.EPG_HEADERS_QUERY}"
# FilterProgramTiles: no Content-Type in its $headers blob.
return self._url_with_headers(
SimpliTVDefaults.PATH_PROGRAM_TILES, include_ct=False
)
def epg_available_days_url(self) -> str:
return self._url_with_headers(
SimpliTVDefaults.PATH_EPG_AVAILABLE_DAYS, include_ct=True
)
def tile_details_url(self) -> str:
return (
f"{self._url(SimpliTVDefaults.PATH_TILE_DETAILS)}"
f"?{SimpliTVDefaults.EPG_HEADERS_QUERY}"
return self._url_with_headers(
SimpliTVDefaults.PATH_TILE_DETAILS, include_ct=False
)
def acquire_content_url(self) -> str:
# AcquireContent does NOT use $headers; it carries the token
# and deviceKey directly in the query string.
return self._url(SimpliTVDefaults.PATH_ACQUIRE_CONTENT)
def get_recordings_url(self) -> str:
return self._url(SimpliTVDefaults.PATH_GET_RECORDINGS)
return self._url_with_headers(
SimpliTVDefaults.PATH_GET_RECORDINGS, include_ct=False
)
def schedule_recording_url(self) -> str:
return self._url(SimpliTVDefaults.PATH_SCHEDULE_RECORDING)
return self._url_with_headers(
SimpliTVDefaults.PATH_SCHEDULE_RECORDING, include_ct=True
)
def delete_recording_url(self) -> str:
return self._url(SimpliTVDefaults.PATH_DELETE_RECORDING)
return self._url_with_headers(
SimpliTVDefaults.PATH_DELETE_RECORDING, include_ct=True
)
def user_products_url(self) -> str:
return self._url_with_headers(
SimpliTVDefaults.PATH_GET_USER_PRODUCTS, include_ct=False
)
def reminders_url(self) -> str:
return self._url_with_headers(
SimpliTVDefaults.PATH_GET_ALL_REMINDERS, include_ct=False
)
lib/streaming_providers/providers/simplitv/epg_manager.py
+292 -81
View File
@@ -2,27 +2,38 @@
"""
simpliTV EPG manager.
Two endpoints, both anonymous (no token -- the existing addon sends
none):
Endpoints:
* /v1/EpgTile/FilterProgramTiles -- every channel's programmes in a
[from, to] window. The response's `programs` is a dict:
{channel codename: [programme, ...]}.
* /v2/Tile/GetTiles -- tile details for a programme id (cast, genres,
images).
* GET /v1/EpgTile/FilterProgramTiles -- every channel's programmes in
a [from, to] window. The response's `programs` is a dict:
{channel codename: [programme, ...]}. Programme tiles carry
`from` / `to` and have NO title, description or images.
One request therefore serves the whole grid, so get_epg_grid is
overridden natively and get_epg is just a one-channel view of the same
data. Windows are fetched in day-sized chunks (as the addon does) and
cached briefly, so per-channel callers don't re-download the grid.
* POST /v2/Tile/GetTiles -- tile details by id. Called in batches to
enrich the programmes FilterProgramTiles returned. Only a compact
summary (title, description, icon) is cached per id, and the cache
is size-bounded. Ids the server did not return are cached as empty
so they are not re-requested on every call. The summaries used for
one request are collected locally, so cache eviction can never
blank titles in the response being built.
Window: -7..+7 days around "now", matching the addon's TV-Guide.
* POST /v1/EpgTile/GetAvailableDays -- the server's EPG window. Used
only to expose epg_window; the fetch windows themselves follow the
caller's [start, end].
If GetTiles fails, the grid is still returned: titles fall back to the
programme codename (and a warning is logged) instead of failing the
whole EPG request.
Windows are fetched in day-sized chunks (as the addon does) and cached
briefly, so per-channel callers don't re-download the grid.
"""
import threading
import time
from datetime import datetime, timedelta, timezone
from typing import Any, Dict, Iterator, List, Optional, Tuple
from urllib.parse import quote
from ...base.errors import ServerError
from ...base.managers import EpgManager
@@ -30,7 +41,13 @@ from ...base.models.epg_models import EPGEntry, EPGProgramDetails
from ...base.utils.logger import logger
from .constants import SimpliTVDefaults
from .helpers import transport_errors
from .helpers import parse_iso, transport_errors
# GetTiles ids per request.
_TILE_BATCH = 500
# Upper bound on cached tile summaries; the oldest half is dropped when
# exceeded.
_TILE_CACHE_MAX = 50000
class SimpliTVEpgManager(EpgManager):
@@ -47,14 +64,21 @@ class SimpliTVEpgManager(EpgManager):
self._window_cache: Dict[
Tuple[str, str], Tuple[float, Dict[str, List[dict]]]
] = {}
# programme id -> {"title", "description", "icon"} ({} = unknown)
self._tile_cache: Dict[str, dict] = {}
self._available_window: Optional[Tuple[int, int]] = None
self._available_window_expires: float = 0.0
self._cache_lock = threading.Lock()
@property
def epg_window(self) -> Tuple[int, int]:
return (
SimpliTVDefaults.EPG_PAST_DAYS,
SimpliTVDefaults.EPG_FUTURE_DAYS,
)
"""
Server-advertised EPG window as (past_days, future_days).
Informational: it does NOT bound the fetch windows, which
follow the caller's [start, end].
"""
return self._fetch_available_window()
# ------------------------------------------------------------------
# get_epg / get_epg_grid
@@ -74,8 +98,8 @@ class SimpliTVEpgManager(EpgManager):
suffix); EPG is keyed on the channel codename.
"""
codename = _strip_live_prefix(channel_id)
programs = self._fetch_programs(start_time, end_time)
return _entries_for(codename, programs.get(codename, []))
programs, tiles = self._fetch_programs(start_time, end_time)
return _entries_for(codename, programs.get(codename, []), tiles)
def get_epg_grid(
self,
@@ -85,12 +109,12 @@ class SimpliTVEpgManager(EpgManager):
**kw,
) -> Dict[str, List[EPGEntry]]:
"""Native batch: one fetch serves every requested channel."""
programs = self._fetch_programs(start_time, end_time)
programs, tiles = self._fetch_programs(start_time, end_time)
grid: Dict[str, List[EPGEntry]] = {}
for channel_id in channel_ids:
codename = _strip_live_prefix(channel_id)
grid[channel_id] = _entries_for(
codename, programs.get(codename, [])
codename, programs.get(codename, []), tiles
)
return grid
@@ -104,22 +128,68 @@ class SimpliTVEpgManager(EpgManager):
"""
Extended details for one tile id (EPGEntry.program_id).
Always fetches the full tile (cast, subtitle, year are not in
the cached summary), so every call makes one small request.
Returns None when the tile is not found.
"""
body = {
"platformCodename": self.config.platform_codename,
"requestedTiles": [{"id": program_id}],
}
with transport_errors(f"GetTiles for {program_id!r}"):
tile = self._fetch_tiles([program_id]).get(program_id)
if not tile:
return None
return _tile_to_program_details(tile)
# ------------------------------------------------------------------
# Available-days window
# ------------------------------------------------------------------
def _fetch_available_window(self) -> Tuple[int, int]:
"""
Return (past_days, future_days) advertised by the server.
Cached for EPG_CACHE_TTL seconds; falls back to the constants
on any error or unusable answer, logged at info because being
permanently on the fallback is worth noticing.
"""
now = time.monotonic()
with self._cache_lock:
if (
self._available_window is not None
and self._available_window_expires > now
):
return self._available_window
try:
resp = self.http_manager.post(
self.config.tile_details_url(),
json=body,
self.config.epg_available_days_url(),
json={},
headers=self.config.get_api_headers(),
)
tiles = resp.json().get("tiles") or []
if not tiles:
return None
return _tile_to_program_details(tiles[0])
data = resp.json()
window = _days_from_window(data.get("from"), data.get("to"))
if window is not None:
with self._cache_lock:
self._available_window = window
self._available_window_expires = (
now + SimpliTVDefaults.EPG_CACHE_TTL
)
return window
logger.info(
"simpliTV: GetAvailableDays returned no usable from/to; "
"using defaults for epg_window"
)
except Exception as e:
logger.info(
f"simpliTV: GetAvailableDays failed ({e}); using "
f"defaults for epg_window"
)
fallback = (
SimpliTVDefaults.EPG_PAST_DAYS,
SimpliTVDefaults.EPG_FUTURE_DAYS,
)
with self._cache_lock:
self._available_window = fallback
self._available_window_expires = now + 60
return fallback
# ------------------------------------------------------------------
# Fetching
@@ -127,8 +197,17 @@ class SimpliTVEpgManager(EpgManager):
def _fetch_programs(
self, start_time: Optional[datetime], end_time: Optional[datetime]
) -> Dict[str, List[dict]]:
"""Merge day-sized windows into {codename: [programme]}."""
) -> Tuple[Dict[str, List[dict]], Dict[str, dict]]:
"""
Merge day-sized windows into {codename: [programme]} and
return it together with {programme id: tile summary} for every
programme in it.
The summaries are gathered into a local dict (from the cache
for known ids, from GetTiles for the rest) rather than re-read
from the shared cache afterwards, so eviction during a large
fetch cannot drop titles from this response.
"""
start, end = _normalise_window(start_time, end_time)
merged: Dict[str, Dict[str, dict]] = {}
for w_start, w_end in _chunks(start, end):
@@ -139,10 +218,43 @@ class SimpliTVEpgManager(EpgManager):
# A programme straddling a boundary appears in both
# windows; the id dedupes it.
key = programme.get("id") or (
f"{programme.get('codename')}:{programme.get('start')}"
f"{programme.get('codename')}:"
f"{programme.get('from')}"
)
bucket[key] = programme
return {c: list(b.values()) for c, b in merged.items()}
all_ids = list(dict.fromkeys(
programme["id"]
for bucket in merged.values()
for programme in bucket.values()
if programme.get("id")
))
tiles: Dict[str, dict] = {}
missing: List[str] = []
with self._cache_lock:
for pid in all_ids:
if pid in self._tile_cache:
tiles[pid] = self._tile_cache[pid]
else:
missing.append(pid)
for i in range(0, len(missing), _TILE_BATCH):
batch = missing[i:i + _TILE_BATCH]
try:
fresh = self._fetch_tiles(batch)
except Exception as e:
# Degrade to codename titles rather than fail the EPG.
logger.warning(
f"simpliTV: GetTiles failed ({e}); programme titles "
f"fall back to codenames"
)
break
for pid in batch:
tile = fresh.get(pid)
tiles[pid] = _summarise_tile(tile) if tile else {}
return {c: list(b.values()) for c, b in merged.items()}, tiles
def _fetch_window(
self, from_iso: str, to_iso: str
@@ -154,18 +266,20 @@ class SimpliTVEpgManager(EpgManager):
if hit and hit[0] > now:
return hit[1]
body = {
"platformCodename": self.config.platform_codename,
"from": from_iso,
"to": to_iso,
}
# FilterProgramTiles is a GET with query-string parameters.
# `from` and `to` MUST be percent-encoded (colons, dots).
url = (
f"{self.config.program_tiles_url()}"
f"&platformCodename={quote(self.config.platform_codename)}"
f"&from={quote(from_iso)}"
f"&to={quote(to_iso)}"
)
with transport_errors(f"FilterProgramTiles {from_iso}..{to_iso}"):
resp = self.http_manager.post(
self.config.program_tiles_url(),
json=body,
headers=self.config.get_api_headers(),
resp = self.http_manager.get(
url, headers=self.config.get_api_headers()
)
programs = resp.json().get("programs")
if not isinstance(programs, dict):
raise ServerError(
"simpliTV: FilterProgramTiles `programs` is not a "
@@ -185,6 +299,55 @@ class SimpliTVEpgManager(EpgManager):
)
return programs
# ------------------------------------------------------------------
# Tile details (batched, cached as compact summaries)
# ------------------------------------------------------------------
def _fetch_tiles(self, program_ids: List[str]) -> Dict[str, dict]:
"""
POST /v2/Tile/GetTiles for the given ids.
Returns {id: full tile} for the ids the server returned, and
records a compact summary for every requested id in the cache
(an empty summary for ids the server did not return). Raises
ServerError on transport failure; callers decide how to
degrade.
"""
if not program_ids:
return {}
body = {
"platformCodename": self.config.platform_codename,
"requestedTiles": [{"id": pid} for pid in program_ids],
}
with transport_errors(f"GetTiles for {len(program_ids)} tiles"):
resp = self.http_manager.post(
self.config.tile_details_url(),
json=body,
headers=self.config.get_api_headers(),
)
tiles = resp.json().get("tiles") or []
fresh: Dict[str, dict] = {}
for tile in tiles:
if not isinstance(tile, dict):
continue
tid = tile.get("id")
if tid:
fresh[tid] = tile
with self._cache_lock:
for pid in program_ids:
tile = fresh.get(pid)
self._tile_cache[pid] = (
_summarise_tile(tile) if tile else {}
)
if len(self._tile_cache) > _TILE_CACHE_MAX:
# dicts keep insertion order: drop the oldest half.
for old in list(self._tile_cache)[: len(self._tile_cache) // 2]:
del self._tile_cache[old]
return fresh
# ----------------------------------------------------------------------
# Module-private helpers
@@ -232,29 +395,65 @@ def _iso(dt: datetime) -> str:
def _parse_iso_to_unix(s: Optional[str]) -> Optional[int]:
"""
ISO-8601 (Z or offset) -> Unix seconds, or None if missing/malformed.
"""ISO-8601 -> Unix seconds, or None if missing / malformed."""
dt = parse_iso(s)
return int(dt.timestamp()) if dt is not None else None
A timestamp without offset is read as UTC (as the addon does), never
as the server's local time.
def _days_from_window(
from_iso: Optional[str], to_iso: Optional[str]
) -> Optional[Tuple[int, int]]:
"""
if not s:
Convert GetAvailableDays's from/to to (past_days, future_days)
relative to now. None if either bound is missing or unparseable.
"""
frm = parse_iso(from_iso)
to = parse_iso(to_iso)
if frm is None or to is None:
return None
try:
dt = datetime.fromisoformat(s.replace("Z", "+00:00"))
except (ValueError, AttributeError, TypeError):
now = datetime.now(timezone.utc)
past = max(0, int((now - frm).total_seconds() // 86400))
future = max(0, int((to - now).total_seconds() // 86400))
return past, future
def _first_image_url(images: Any) -> Optional[str]:
"""
URL of the first image, tolerating an unverified response shape:
a list of {"url": ...} objects, a list of plain strings, or
anything else (-> None).
"""
if not isinstance(images, list) or not images:
return None
if dt.tzinfo is None:
dt = dt.replace(tzinfo=timezone.utc)
return int(dt.timestamp())
first = images[0]
if isinstance(first, dict):
return first.get("url") or None
if isinstance(first, str):
return first or None
return None
def _summarise_tile(tile: Dict[str, Any]) -> Dict[str, Any]:
"""Compact per-programme summary kept in the tile cache."""
return {
"title": tile.get("title") or "",
"description": tile.get("description") or "",
"icon": _first_image_url(tile.get("images")),
}
def _entries_for(
channel_codename: str, programmes: List[dict]
channel_codename: str,
programmes: List[dict],
tiles: Dict[str, dict],
) -> List[EPGEntry]:
entries = []
for programme in programmes:
entry = _programme_to_entry(channel_codename, programme)
entry = _programme_to_entry(
channel_codename,
programme,
tiles.get(programme.get("id") or "") or {},
)
if entry is not None:
entries.append(entry)
entries.sort(key=lambda e: e.start)
@@ -262,18 +461,25 @@ def _entries_for(
def _programme_to_entry(
channel_codename: str, programme: Dict[str, Any]
channel_codename: str,
programme: Dict[str, Any],
tile: Dict[str, Any],
) -> Optional[EPGEntry]:
"""
Map a programme tile to EPGEntry; None if it has no usable times
(an entry without times would collide on its broadcast id).
Map a FilterProgramTiles programme (plus its tile summary, if any)
to EPGEntry.
broadcast_id embeds the programme codename as its last segment --
that is what the catchup manager takes as epg_id for replay.
program_id is the tile id GetTiles needs.
FilterProgramTiles emits `from` / `to` (not `start` / `stop`) and
carries no title, description or images; those come from `tile`.
When no title is available the programme codename is used, so the
grid is never blank.
"""
start = _parse_iso_to_unix(programme.get("start"))
end = _parse_iso_to_unix(programme.get("stop"))
start = _parse_iso_to_unix(
programme.get("from") or programme.get("start")
)
end = _parse_iso_to_unix(
programme.get("to") or programme.get("stop")
)
if start is None or end is None:
logger.debug(
f"simpliTV: skipping programme without times: "
@@ -281,19 +487,23 @@ def _programme_to_entry(
)
return None
images = programme.get("images") or []
icon = images[0].get("url") if images else None
title = (
tile.get("title")
or programme.get("title")
or programme.get("codename")
or "(no title)"
)
return EPGEntry(
broadcast_id=(
f"{SimpliTVDefaults.BROADCAST_ID_PREFIX}{channel_codename}:"
f"{start}:{programme.get('codename', '')}"
),
title=programme.get("title", ""),
description=programme.get("description", ""),
title=title,
description=tile.get("description") or "",
start=start,
end=end,
icon=icon,
icon=tile.get("icon"),
program_id=programme.get("id"),
)
@@ -302,21 +512,22 @@ def _tile_to_program_details(tile: Dict[str, Any]) -> EPGProgramDetails:
"""
Map a /v2/Tile/GetTiles tile to the shared EPGProgramDetails shape.
Season/episode numbers, countries and genres are on the tile but not
in the shared shape; extend EPGProgramDetails first (template rule)
rather than inventing kwargs here.
Season/episode numbers, countries and genres are on the tile but
not in the shared shape; extend EPGProgramDetails first (template
rule) rather than inventing kwargs here.
"""
images = tile.get("images") or []
icon = images[0].get("url") if images else None
people = tile.get("people") or []
cast = [p.get("fullName", "") for p in people if p.get("fullName")]
people = tile.get("people")
cast = [
p["fullName"]
for p in (people if isinstance(people, list) else [])
if isinstance(p, dict) and p.get("fullName")
]
return EPGProgramDetails(
program_id=tile.get("id", ""),
description=tile.get("description", ""),
episode_name=tile.get("subTitle", ""),
year=tile.get("date") or None,
icon=icon,
icon=_first_image_url(tile.get("images")),
cast=cast,
)
)
lib/streaming_providers/providers/simplitv/helpers.py
+52 -6
View File
@@ -6,7 +6,10 @@ Public names on purpose: the template README requires that a helper
used by several modules is public (no leading underscore).
"""
import re
from contextlib import contextmanager
from datetime import datetime, timezone
from typing import Optional
from ...base.errors import ProviderError, ServerError
@@ -15,12 +18,21 @@ def prefer_dash(url: str) -> str:
"""
Rewrite an HLS manifest URL to its DASH equivalent.
Use ONLY for DRM-protected content: inputstream.adaptive needs DASH
there. Unprotected streams stay on the HLS URL the API returned (the
existing addon plays those through ffmpegdirect). The replacement
list is exactly the addon's forward direction; the "nodrm" pair
exists only in the addon's reverse (DASH -> HLS) direction and must
not be applied here.
Fallback only, and only for DRM-protected content. The
AcquireContent response normally carries both DASH (Type 9) and HLS
(Type 2) entries, and the channel manager picks the DASH one
directly when the content is protected; this function is called
only when the protected asset is HLS-only.
Never applied to unprotected content: the CDN has been serving the
HLS URL to the addon for those since day one, and swapping it out
has not been validated. Restart relies on a DVR seek offset from
the window start, and DASH time-shift semantics can differ from
HLS.
The replacement list is exactly the addon's forward direction; the
"nodrm" pair exists only in the addon's reverse (DASH -> HLS)
direction and must not be applied here.
"""
for old, new in (
(".m3u8", ".mpd"),
@@ -49,3 +61,37 @@ def transport_errors(what: str):
raise
except Exception as e:
raise ServerError(f"simpliTV: {what} failed: {e}") from e
# Fractional seconds followed by end-of-string, Z, or a UTC offset.
_FRACTION = re.compile(r"\.(\d+)(?=$|[Zz]|[+-]\d{2}(?::?\d{2})?$)")
_OFFSET_NO_COLON = re.compile(r"([+-]\d{2})(\d{2})$")
def parse_iso(value: Optional[str]) -> Optional[datetime]:
"""
Parse an ISO-8601 timestamp to a timezone-aware datetime, or None
if missing / malformed.
datetime.fromisoformat on Python < 3.11 accepts only 3 or 6
fractional digits and rejects "Z" and "+HHMM". .NET-style APIs
emit 7 digits, so the string is normalised first: fraction cut or
padded to 6 digits, "Z" -> "+00:00", "+HHMM" -> "+HH:MM". A
timestamp without an offset is read as UTC, never as local time.
"""
if not value or not isinstance(value, str):
return None
s = value.strip()
if not s:
return None
s = _FRACTION.sub(lambda m: "." + m.group(1)[:6].ljust(6, "0"), s)
if s[-1] in "Zz":
s = s[:-1] + "+00:00"
s = _OFFSET_NO_COLON.sub(r"\1:\2", s)
try:
dt = datetime.fromisoformat(s)
except (ValueError, TypeError):
return None
if dt.tzinfo is None:
dt = dt.replace(tzinfo=timezone.utc)
return dt
lib/streaming_providers/providers/simplitv/logos.py
+225 -9
View File
@@ -4,12 +4,151 @@ simpliTV channel names and logos.
The channel-tile endpoint only yields a codename, so display names and
logos are derived locally. Ported from the existing addon's
channelname() / logomapper() so behaviour is identical.
channelname() / logomapper() so behaviour is identical, then extended
for the codenames observed in the browser capture.
Lookup order:
1. _CODENAME_ALIASES -- explicit codename -> display-name table for
the codenames whose derivation does not match the spelling used
as a LOGO_MAP key.
2. channel_name_from_codename -- general derivation (dashes to
spaces, uppercase, strip trailing HD/FHD).
3. LOGO_MAP -- display name -> logo URL.
4. PROVIDER_LOGO fallback.
"""
from .constants import SimpliTVDefaults
# Codenames whose local derivation does not match the display name
# used as a LOGO_MAP key. Anything not in this table falls back to
# channel_name_from_codename().
_CODENAME_ALIASES = {
"atvhd": "ATV",
"atv-ii-fhd": "ATV II",
"atv-2": "ATV 2",
"n-tvhd": "N TV",
"tele-5-hd": "TELE 5",
"puls4austriahd": "PULS4",
"puls24hd": "PULS24",
"puls24": "PULS24",
"pu4": "PU4",
"prosiebenaustriahd": "PROSIEBEN AUSTRIA",
"sat1austriahd": "SAT.1 AUSTRIA",
"sat1goldaustriahd": "SAT.1 GOLD AUSTRIA",
"rtlaustriahd": "RTL AUSTRIA",
"rtlzweiaustriahd": "RTL ZWEI AUSTRIA",
"rtlup-austria-hd": "RTLUP AUSTRIA",
"rtl-crime-hd": "RTL CRIME",
"rtl-living-hd": "RTL LIVING",
"rtl-passion-hd": "RTL PASSION",
"voxaustriahd": "VOX AUSTRIA",
"kabeleinsaustriahd": "KABEL EINS AUSTRIA",
"kabeleinsdokuhd": "KABELEINS DOKU",
"prosiebenmaxxaustriahd": "PROSIEBEN MAXX AUSTRIA",
"sixxaustriahd": "SIXX AUSTRIA",
"nitro-austria-hd": "NITRO AUSTRIA",
"dmaxaustriahd": "DMAX AUSTRIA",
"tlcaustriahd": "TLC AUSTRIA",
"geo-television-hd": "GEO TELEVISION",
"hgtv-hd": "HGTV",
"servustv-austria": "SERVUS TV AUSTRIA",
"orf-sport-plus": "ORF SPORT PLUS",
"orf-iii": "ORF III",
"orf2-europe": "ORF 2 EUROPE",
"orf1": "ORF 1",
"orf2": "ORF 2",
"orf2no": "ORF 2 NO",
"orf2oo": "ORF 2 OOE",
"orf2st": "ORF 2 STEIERMARK",
"orf2t": "ORF 2 TIROL",
"orf2b": "ORF 2 BURGENLAND",
"orf2k": "ORF 2 KAERNTEN",
"orf2s": "ORF 2 SALZBURG",
"orf2v": "ORF 2 VORARLBERG",
"orf-kids": "ORF KIDS",
"zdf-hd": "ZDF",
"zdfneo": "ZDF NEO",
"zdfinfo-hd": "ZDF INFO",
"ard-hd": "ARD",
"ard-alpha-hd": "ARD ALPHA",
"arte": "ARTE",
"3sat": "3SAT",
"kika-hd": "KIKA",
"phoenix": "PHOENIX",
"one": "ONE",
"tagesschau-24-hd": "TAGESSCHAU 24",
"hr-fernsehen-hd": "HR FERNSEHEN",
"mdr-sachsen-hd": "MDR SACHSEN",
"ndr": "NDR",
"rbb-berlin-hd": "RBB BERLIN",
"radio-bremen-hd": "RADIO BREMEN",
"sr-fernsehen-hd": "SR FERNSEHEN",
"swr-bw-hd": "SWR BW",
"wdr-koeln-hd": "WDR KOELN",
"br-hd": "BFS",
"cnn": "CNN",
"welt": "WELT",
"euronews-german-sd": "EURONEWS ENGLISH",
"al-jazeera-english-hd": "AL JAZEERA ENGLISH",
"bloomberg-europe": "BLOOMBERG EUROPE",
"tvp-world": "TVP WORLD",
"nickelodeon": "NICKELODEON",
"ric": "RIC",
"ric-today": "RIC TODAY",
"fix-und-foxi": "FIX UND FOXI",
"superrtl-austria-hd": "SUPERRTL AUSTRIA",
"k-tv": "K-T-V",
"kronehit-tv": "KRONEHIT TV",
"schau-tv": "SCHAU TV",
"r9-oesterreich-hd": "R9 OESTERREICH",
"sonnenklartv-hd": "SONNENKLARTV",
"lt1-hd": "LT1",
"tv1-ooe": "TV1 OOE",
"laendletv": "LAENDLETV",
"tiroltv": "TIROLTV",
"dorftv": "DORFTV",
"n1-noe-tv": "N1 NOE TV",
"w24": "W24",
"kt1": "KT1",
"myzentv": "MYZENTV",
"museum-tv": "MUSEUM TV",
"kaminfeuer": "KAMINFEUER",
"shoplc": "SHOP LC",
"pegasus-hls": "PEGASUS HLS",
"motorvision": "MOTORVISION",
"more-than-sports": "MORE THAN SPORTS",
"wedotv": "WEDOTV",
"wedo-sports": "WEDO SPORTS",
"wedo-big-stories": "WEDO BIG STORIES",
"wedo-true-stories": "WEDO TRUE STORIES",
"bibeltv-hd": "BIBELTV",
"hopetvhd": "HOPETV",
"kronehit-tv": "KRONEHIT TV",
"schlager-deluxe": "SCHLAGER DELUXE",
"melodietv": "MELODIETV",
"deutsches-musikfernsehen": "DEUTSCHES MUSIKFERNSEHEN",
"mtvaustriahd": "MTV AUSTRIA",
"deluxemusicaustriahd": "DELUXEMUSIC AUSTRIA",
"starbaradies-tv": "STARPARADISES TV",
"lilo-tv": "LILO TV",
"playboy-tv": "PLAYBOY TV",
"hustler-tv": "HUSTLER TV",
"dorcel-tv": "DORCEL TV",
"radio-bremen-hd": "RADIO BREMEN",
"ard-alpha-hd": "ARD ALPHA",
"bfs": "BFS",
"mtvaustriahd": "MTV AUSTRIA",
"k-tv": "K-TV",
"k-tv": "K-T-V",
}
LOGO_MAP = {
# ------------------------------------------------------------------
# Original addon entries, preserved verbatim.
# ------------------------------------------------------------------
"ORF1": "https://files.app.simplitv.at/files/orf1-hd-bunt.png",
"ORF2NO": "https://files.app.simplitv.at/files/orf-no-e-hd-bunt.png",
"ATV": "https://files.app.simplitv.at/files/atv-hd-bunt.png",
@@ -88,7 +227,7 @@ LOGO_MAP = {
"SWR BW": "https://files.app.simplitv.at/files/062-swtbwhd-366x71.png",
"SR FERNSEHEN": "https://files.app.simplitv.at/files/067-srfernsehenhd-1024x627.png",
"NDR": "https://files.app.simplitv.at/files/ndr-hd-bunt.png",
"WDR KÖLN": "https://files.app.simplitv.at/files/063-wdrhd-1024x341.png",
"WDR KOELN": "https://files.app.simplitv.at/files/063-wdrhd-1024x341.png",
"MDR SACHSEN": "https://files.app.simplitv.at/files/mdr-rgb.png",
"HR FERNSEHEN": "https://files.app.simplitv.at/files/065-hr-fernsehenhd-170x75.png",
"RBB BERLIN": "https://files.app.simplitv.at/files/rbb-rgb.png",
@@ -112,19 +251,96 @@ LOGO_MAP = {
"PLAYBOY TV": "https://files.app.simplitv.at/files/dorcel-2022-playboytveurope-logo-noir-transparent.png",
"HUSTLER TV": "https://files.app.simplitv.at/files/hustlertv-light.jpg",
"DORCEL TV": "https://files.app.simplitv.at/files/2022-dorceltv-black-rvb.jpg",
# ------------------------------------------------------------------
# Extensions from the browser capture.
# Names here are the alias-table targets; do not rename without
# updating _CODENAME_ALIASES.
# ------------------------------------------------------------------
"ATV II": "https://files.app.simplitv.at/files/atv2-bunt.png",
"PULS4": "https://files.app.simplitv.at/files/puls4-hd-bunt.png",
"PROSIEBEN AUSTRIA": "https://files.app.simplitv.at/files/pro7-hd-bunt.png",
"SAT.1 AUSTRIA": "https://files.app.simplitv.at/files/sat1-hd-bunt.png",
"SAT.1 GOLD AUSTRIA": "https://files.app.simplitv.at/files/sat1-gold-hd-bunt.png",
"RTL AUSTRIA": "https://files.app.simplitv.at/files/rtl-austria-hd-bunt.png",
"RTL ZWEI AUSTRIA": "https://files.app.simplitv.at/files/rtl-zwei-hd-bunt.png",
"RTLUP AUSTRIA": "https://files.app.simplitv.at/files/rtluphd-kopie.png",
"RTL CRIME": "https://files.app.simplitv.at/files/rtl-crime-hd-bunt.png",
"RTL LIVING": "https://files.app.simplitv.at/files/rtl-living-hd-bunt.png",
"RTL PASSION": "https://files.app.simplitv.at/files/rtl-passion-hd-bunt.png",
"VOX AUSTRIA": "https://files.app.simplitv.at/files/vox-hd-bunt.png",
"KABEL EINS AUSTRIA": "https://files.app.simplitv.at/files/kabel1-hd-austria-bunt.png",
"KABELEINS DOKU": "https://files.app.simplitv.at/files/kabel1-doku-austria-bunt.png",
"PROSIEBEN MAXX AUSTRIA": "https://files.app.simplitv.at/files/pro7-maxx-austria-bunt.png",
"SIXX AUSTRIA": "https://files.app.simplitv.at/files/sixx-austria-hd-bunt.png",
"DMAX AUSTRIA": "https://files.app.simplitv.at/files/dmax-hd-bunt.png",
"TLC AUSTRIA": "https://files.app.simplitv.at/files/tlc-hd-bunt.png",
"GEO TELEVISION": "https://files.app.simplitv.at/files/geo-television-hd-bunt.png",
"ZDF NEO": "https://files.app.simplitv.at/files/zdf-neo-hd-bunt.png",
"ZDF INFO": "https://files.app.simplitv.at/files/zdf-info-hd-bunt.png",
"ORF 1": "https://files.app.simplitv.at/files/orf1-hd-bunt.png",
"ORF 2": "https://files.app.simplitv.at/files/orf2-hd-bunt.png",
"ORF 2 NO": "https://files.app.simplitv.at/files/orf-no-e-hd-bunt.png",
"ORF 2 OOE": "https://files.app.simplitv.at/files/orf-oo-e-hd-bunt.png",
"ORF 2 STEIERMARK": "https://files.app.simplitv.at/files/orf-steiermark-hd-bunt.png",
"ORF 2 TIROL": "https://files.app.simplitv.at/files/orf-tirol-hd-bunt.png",
"ORF 2 BURGENLAND": "https://files.app.simplitv.at/files/orf-hd-burgenland-bunt.png",
"ORF 2 KAERNTEN": "https://files.app.simplitv.at/files/orf-hd-ka-ernten-bunt.png",
"ORF 2 SALZBURG": "https://files.app.simplitv.at/files/orf-salzburg-hd-bunt.png",
"ORF 2 VORARLBERG": "https://files.app.simplitv.at/files/orf-vorarlberg-hd-bunt.png",
"ORF 2 EUROPE": "https://files.app.simplitv.at/files/orf2-europe-hd-bunt.png",
"ORF KIDS": "https://files.app.simplitv.at/files/orf-kids-bunt.png",
"RIC TODAY": "https://files.app.simplitv.at/files/ric-today-bunt.png",
"MUSEUM TV": "https://files.app.simplitv.at/files/museum-tv-logo.png",
"KAMINFEUER": "https://files.app.simplitv.at/files/kaminfeuer-bunt.png",
"SHOP LC": "https://files.app.simplitv.at/files/shoplc-bunt.png",
"PEGASUS HLS": "https://files.app.simplitv.at/files/pegasus-hls.png",
"MOTORVISION": "https://files.app.simplitv.at/files/motorvision-hd-bunt.png",
"MORE THAN SPORTS": "https://files.app.simplitv.at/files/more-than-sports-bunt.png",
"WEDOTV": "https://files.app.simplitv.at/files/wedotv-bunt.png",
"WEDO SPORTS": "https://files.app.simplitv.at/files/wedo-sports-bunt.png",
"WEDO BIG STORIES": "https://files.app.simplitv.at/files/wedo-big-stories-bunt.png",
"WEDO TRUE STORIES": "https://files.app.simplitv.at/files/wedo-true-stories-bunt.png",
"HOPETV": "https://files.app.simplitv.at/files/hopetv-bunt.png",
"KRONEHIT TV": "https://files.app.simplitv.at/files/kronehit-tv-logo.png",
"MYZENTV": "https://files.app.simplitv.at/files/museum-tv-logo.png",
"K-T-V": "https://files.app.simplitv.at/files/k-tv-bunt.png",
"K-TV": "https://files.app.simplitv.at/files/k-tv-bunt.png",
"DELUXEMUSIC AUSTRIA": "https://files.app.simplitv.at/files/deluxe-music-hd-bunt.png",
"STARPARADISES TV": "https://files.app.simplitv.at/files/starparadies.png",
}
def channel_name_from_codename(codename: str) -> str:
"""Display name for a codename (mirrors the addon's channelname())."""
name = codename.replace("-", " ").upper()
if name.endswith(" HD"):
name = name[:-3]
if name.endswith("HD"):
name = name[:-2]
"""
Display name for a codename.
Prefers the explicit alias table; otherwise normalises dashes and
underscores to spaces, uppercases, and strips a trailing HD / FHD
quality suffix.
"""
if not codename:
return ""
lower = codename.lower()
if lower in _CODENAME_ALIASES:
return _CODENAME_ALIASES[lower]
name = codename.replace("-", " ").replace("_", " ").upper()
# Strip a trailing quality marker. FHD must be checked before HD.
for suffix in (" FHD", "FHD", " HD", "HD"):
if name.endswith(suffix) and len(name) > len(suffix):
name = name[: -len(suffix)].rstrip()
break
# Collapse any double spaces left behind.
while " " in name:
name = name.replace(" ", " ")
return name
def logo_for_name(name: str) -> str:
"""Logo URL for a display name; provider logo when unknown."""
return LOGO_MAP.get(name) or SimpliTVDefaults.PROVIDER_LOGO
return LOGO_MAP.get(name) or SimpliTVDefaults.PROVIDER_LOGO
lib/streaming_providers/providers/simplitv/models.py
+4 -3
View File
@@ -6,8 +6,9 @@ A custom Channel subclass carries extra provider metadata (codename,
logo, recording id/status). Base fields are preserved; extras are
appended.
No custom AuthToken subclass -- the token is a plain opaque string and
the device key is read from the server (see auth.py).
No custom AuthToken subclass: the token is a plain opaque string, and
the device key is discovered via GetDevices (registered only if the
account has none) -- see auth.py for the two-step flow.
"""
from dataclasses import dataclass
@@ -56,4 +57,4 @@ class SimpliTVChannel(Channel):
result["IsCatchupEnabled"] = self.is_catchup_enabled
result["RecordingId"] = self.recording_id
result["RecordingStatus"] = self.recording_status
return result
return result
lib/streaming_providers/providers/simplitv/provider.py
+13 -3
View File
@@ -36,6 +36,15 @@ raise BadRequestError and are not swallowed.
Restart-from-beginning cannot be expressed as a manifest URL (it needs a
player-side seek), so it has its own method: get_restart().
Catchup windows
---------------
The DVR window is per-channel (AdditionalInfo.Epg_TimeshiftSeconds in
the AcquireContent response: 2h, 3h or 4h in the browser capture).
SimpliTVCatchupManager reads the per-channel value inside
get_restart_manifest. Its catchup_window_hours property returns the
conservative minimum (2h) because the ABC can only express a single
integer.
"""
from typing import Any, Callable, ClassVar, Dict, List, Optional, Tuple
@@ -289,8 +298,9 @@ class SimpliTVProvider(StreamingProvider):
Restart a programme from its beginning.
Returns (live manifest URL, seek_seconds from the start of the
3h DVR window) or None when outside the window. The caller must
seek; the URL alone plays live.
channel's DVR window) or None when the programme began outside
that window. The window is per-channel (see "Catchup windows"
above). The caller must seek; the URL alone plays live.
"""
if self.catchup is None:
return None
@@ -346,4 +356,4 @@ class SimpliTVProvider(StreamingProvider):
def get_channels(self, **kw):
if self.channels is None:
return []
return self.channels.get_channels(**kw)
return self.channels.get_channels(**kw)