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

Remove livgolf

Browse Source
This commit is contained in:
Nirvana
2026-06-05 13:16:28 +02:00
parent b16c839370
commit decb217ac8
6 changed files with 0 additions and 1087 deletions
Download Patch File Download Diff File Expand all files Collapse all files
lib/streaming_providers/providers/livgolf/__init__.py
-25
View File
@@ -1,25 +0,0 @@
# streaming_providers/providers/livgolf/__init__.py
from .auth import LivGolfAuthToken, LivGolfAuthenticator
from .constants import (
API_ENDPOINTS,
DEFAULT_CHAMPION_ID,
PROVIDER_LABEL,
PROVIDER_LOGO,
PROVIDER_NAME,
)
from .channel_manager import LivGolfChannelManager
from .provider import LivGolfProvider
__all__ = [
"LivGolfProvider",
"LivGolfAuthenticator",
"LivGolfAuthToken",
"LivGolfChannelManager",
"PROVIDER_NAME",
"PROVIDER_LABEL",
"PROVIDER_LOGO",
"API_ENDPOINTS",
"DEFAULT_CHAMPION_ID",
]
__version__ = "1.0.0"
lib/streaming_providers/providers/livgolf/auth.py
-275
View File
@@ -1,275 +0,0 @@
# streaming_providers/providers/livgolf/auth.py
# -*- coding: utf-8 -*-
"""
Anonymous authentication for the LIV Golf / ViewLift API.
The flow is intentionally simple:
1. Generate a stable ``browser-<uuid4>`` device ID (persisted in the token).
2. POST to the anonymous-token endpoint — no credentials required.
3. The response contains a JWT ``authorizationToken`` valid for ~1 year.
4. On every subsequent request supply it as the ``Authorization`` header.
Token persistence reuses the BaseAuthenticator / BaseAuthToken machinery so
that the token survives process restarts (stored via settings_manager).
"""
import json
import time
import uuid
import base64
from dataclasses import dataclass, field
from typing import Any, Dict, Optional
from ...base.auth.base_auth import BaseAuthenticator, BaseAuthToken, TokenAuthLevel
from ...base.auth.credentials import ClientCredentials
from ...base.models.proxy_models import ProxyConfig
from ...base.utils.logger import logger
from .constants import (
API_ENDPOINTS,
DEFAULT_REQUEST_TIMEOUT,
PLATFORM,
PROVIDER_NAME,
SITE,
get_base_headers,
)
# ---------------------------------------------------------------------------
# JWT helpers (no external dependency — same pattern as magentaeu)
# ---------------------------------------------------------------------------
def _base64url_decode(s: str) -> bytes:
padding = "=" * (4 - len(s) % 4)
return base64.urlsafe_b64decode(s + padding)
def _decode_jwt_payload(token: str) -> Dict[str, Any]:
"""Decode JWT payload without signature verification."""
try:
_, payload_b64, _ = token.split(".")
return json.loads(_base64url_decode(payload_b64).decode("utf-8"))
except Exception as exc:
raise ValueError(f"Cannot decode JWT: {exc}") from exc
def _token_expires_at(token: str) -> float:
"""Return the ``exp`` claim as a Unix timestamp, or 0 on failure."""
try:
return float(_decode_jwt_payload(token).get("exp", 0))
except Exception:
return 0.0
# ---------------------------------------------------------------------------
# Token dataclass
# ---------------------------------------------------------------------------
@dataclass
class LivGolfAuthToken(BaseAuthToken):
"""
Thin wrapper around BaseAuthToken that adds the ``device_id`` used when
requesting the anonymous token (needed to stay consistent across refreshes).
"""
device_id: str = field(default="")
# ------------------------------------------------------------------
# Serialisation — must round-trip through BaseAuthenticator's
# settings_manager.save_session / load_session machinery.
# ------------------------------------------------------------------
def to_dict(self) -> Dict[str, Any]:
return {
"access_token": self.access_token,
"token_type": self.token_type,
"expires_in": self.expires_in,
"issued_at": self.issued_at,
"auth_level": (
self.auth_level.value if self.auth_level else TokenAuthLevel.CLIENT_CREDENTIALS.value
),
"credential_type": self.credential_type or "client_credentials",
"device_id": self.device_id,
}
# ---------------------------------------------------------------------------
# Authenticator
# ---------------------------------------------------------------------------
class LivGolfAuthenticator(BaseAuthenticator):
"""
Anonymous-only authenticator for the LIV Golf ViewLift API.
No username / password are required or supported. The token is a long-lived
JWT (~1 year) tied to a generated ``browser-<uuid>`` device ID.
"""
def __init__(
self,
config_dir: Optional[str] = None,
http_manager=None,
proxy_config: Optional[ProxyConfig] = None,
) -> None:
if http_manager is None:
raise ValueError("http_manager is required for LivGolfAuthenticator")
self._http_manager = http_manager
self._proxy_config = proxy_config
# BaseAuthenticator.__init__ loads any persisted session automatically.
super().__init__(
provider_name=PROVIDER_NAME,
config_dir=config_dir,
# No credentials needed for anonymous auth
credentials=None,
# No Kodi integration required
enable_kodi_integration=False,
)
# Ensure we always have a device_id — either restored from disk or freshly generated.
if not self._current_token or not isinstance(self._current_token, LivGolfAuthToken):
device_id = f"browser-{uuid.uuid4()}"
self._current_token = LivGolfAuthToken(
access_token="",
token_type="Bearer",
expires_in=0,
issued_at=time.time(),
device_id=device_id,
)
logger.debug(f"[LivGolfAuthenticator] Generated new device_id: {device_id}")
else:
logger.debug(
f"[LivGolfAuthenticator] Restored device_id: {self._current_token.device_id}"
)
# ------------------------------------------------------------------
# BaseAuthenticator abstract-method implementations
# ------------------------------------------------------------------
@property
def auth_endpoint(self) -> str:
"""Anonymous token endpoint — device_id filled in at call time."""
return API_ENDPOINTS["ANONYMOUS_TOKEN"]
def _get_auth_headers(self) -> Dict[str, str]:
return get_base_headers()
def _build_auth_payload(self) -> Dict[str, Any]:
# Anonymous token is obtained via GET — no POST body needed.
return {}
def get_fallback_credentials(self):
# Anonymous authenticator - treat as client credentials
return ClientCredentials(client_id=PROVIDER_NAME, client_secret="anonymous")
def _perform_authentication(self) -> BaseAuthToken:
"""Fetch a fresh anonymous token from the ViewLift identity endpoint."""
device_id = self._device_id
url = API_ENDPOINTS["ANONYMOUS_TOKEN"].format(
site=SITE,
platform=PLATFORM,
device_id=device_id,
)
logger.info(f"[LivGolfAuthenticator] Requesting anonymous token (device_id={device_id})")
response = self._http_manager.get(
url,
operation="auth_anonymous",
headers=get_base_headers(),
timeout=DEFAULT_REQUEST_TIMEOUT,
)
response.raise_for_status()
data = response.json()
raw_token = data.get("authorizationToken")
if not raw_token:
raise ValueError(
f"Anonymous token response missing 'authorizationToken'. "
f"Keys received: {list(data.keys())}"
)
exp = _token_expires_at(raw_token)
issued = time.time()
expires_in = max(0, int(exp - issued)) if exp else 86400 * 365
token = LivGolfAuthToken(
access_token=raw_token,
token_type="Bearer",
expires_in=expires_in,
issued_at=issued,
device_id=device_id,
auth_level=TokenAuthLevel.CLIENT_CREDENTIALS,
)
logger.info(
f"[LivGolfAuthenticator] Anonymous token obtained "
f"(expires_in={expires_in}s, device_id={device_id})"
)
return token
def _refresh_token(self) -> Optional[BaseAuthToken]:
"""
ViewLift anonymous tokens are very long-lived (~1 year), but when they
do expire we simply request a new one — same device_id, new JWT.
"""
logger.info("[LivGolfAuthenticator] Refreshing anonymous token")
try:
return self._perform_authentication()
except Exception as exc:
logger.warning(f"[LivGolfAuthenticator] Token refresh failed: {exc}")
return None
def _classify_token(self, token: BaseAuthToken) -> TokenAuthLevel:
return TokenAuthLevel.CLIENT_CREDENTIALS
def _create_token_from_response(self, response_data: Dict[str, Any]) -> BaseAuthToken:
"""Restore a token from the persisted session dict."""
raw_token = response_data.get("access_token", "")
device_id = response_data.get("device_id") or f"browser-{uuid.uuid4()}"
issued = response_data.get("issued_at", time.time())
expires_in = response_data.get("expires_in", 0)
token = LivGolfAuthToken(
access_token=raw_token,
token_type=response_data.get("token_type", "Bearer"),
expires_in=expires_in,
issued_at=issued,
device_id=device_id,
auth_level=TokenAuthLevel.CLIENT_CREDENTIALS,
)
return token
# ------------------------------------------------------------------
# Public helpers
# ------------------------------------------------------------------
@property
def _device_id(self) -> str:
"""Return the stable device ID, regardless of token state."""
if self._current_token and isinstance(self._current_token, LivGolfAuthToken):
return self._current_token.device_id or f"browser-{uuid.uuid4()}"
return f"browser-{uuid.uuid4()}"
def get_authorization_header(self, force_refresh: bool = False) -> str:
"""
Return a ready-to-use ``Authorization`` header value (the raw JWT,
*without* a ``Bearer `` prefix — the LIV Golf API uses the raw token).
"""
token = self.get_bearer_token(force_refresh=force_refresh)
# get_bearer_token may prepend "Bearer " — strip it for this API
if token and token.startswith("Bearer "):
return token[len("Bearer "):]
return token or ""
def is_token_expired(self) -> bool:
"""True if the current token is missing or within the expiry margin."""
if not self._current_token or not self._current_token.access_token:
return True
exp = _token_expires_at(self._current_token.access_token)
if exp == 0:
return False # Cannot determine expiry — assume valid
from .constants import TOKEN_EXPIRY_MARGIN_SECONDS
return time.time() >= (exp - TOKEN_EXPIRY_MARGIN_SECONDS)
lib/streaming_providers/providers/livgolf/channel_manager.py
-391
View File
@@ -1,391 +0,0 @@
# streaming_providers/providers/livgolf/channel_manager.py
# -*- coding: utf-8 -*-
"""
Channel manager for the LIV Golf provider.
Responsibilities
----------------
* Discover the best CDN edge region from /mobii/regions.
* Fetch team-camera and group-camera stream lists for a champion (tournament).
* Rewrite manifest URLs to use the preferred regional CDN.
* Return normalised ``StreamingChannel`` objects ready for the provider.
Design notes
------------
* Region selection is cached for the lifetime of the process — CDN topology
does not change during a session.
* Stream lists are fetched fresh on every ``get_channels()`` call so that live
tournament URLs (which rotate) are always current.
* No EPG support — this provider is channels-only.
"""
from __future__ import annotations
import re
from typing import Any, Dict, List, Optional
from ...base.models import StreamingChannel
from ...base.utils.logger import logger
from .constants import (
API_ENDPOINTS,
CONTENT_TYPE_LIVE,
DEFAULT_CHAMPION_ID,
DEFAULT_REQUEST_TIMEOUT,
FALLBACK_CDN_BASE,
PROVIDER_NAME,
REGION_PREFERENCE_ORDER,
STREAMING_FORMAT_DASH,
get_authenticated_headers,
)
# ---------------------------------------------------------------------------
# Internal helpers
# ---------------------------------------------------------------------------
# Matches the CDN host in a mobii.com manifest URL, e.g.:
# https://gcp-edge-us-c.mobii.com/api/video/...
_CDN_HOST_RE = re.compile(r"https://([^/]+\.mobii\.com)/")
def _rewrite_cdn(url: str, preferred_base: str) -> str:
"""
Replace the CDN host in a mobii.com manifest URL with ``preferred_base``.
``preferred_base`` is the full origin, e.g. ``https://gcp-edge-eu-w.mobii.com``.
Returns the original URL unchanged if it does not match the expected pattern.
"""
if not url:
return url
return _CDN_HOST_RE.sub(preferred_base.rstrip("/") + "/", url, count=1)
# ---------------------------------------------------------------------------
# LivGolfChannelManager
# ---------------------------------------------------------------------------
class LivGolfChannelManager:
"""
Fetches LIV Golf live camera streams and normalises them into ``StreamingChannel`` objects.
Parameters
----------
http_manager:
The provider's shared HTTPManager instance.
authenticator:
``LivGolfAuthenticator`` — used to obtain the current anonymous token.
"""
def __init__(self, http_manager: Any, authenticator: Any) -> None:
self._http = http_manager
self._auth = authenticator
# Cached preferred CDN base URL (None = not yet resolved)
self._preferred_cdn: Optional[str] = None
# Cache for channels keyed by content_id (video_id)
self._channels_cache: Dict[str, StreamingChannel] = {}
logger.info("[LivGolfChannelManager] Initialised")
# ------------------------------------------------------------------
# Public API
# ------------------------------------------------------------------
def get_channels(self, champion_id: str = DEFAULT_CHAMPION_ID) -> List[StreamingChannel]:
"""
Return all available live streams for *champion_id* as ``StreamingChannel`` objects.
Both team-camera and group-camera feeds are fetched and merged.
Each stream becomes one Channel with a DASH manifest URL rewritten to
the closest CDN region.
Parameters
----------
champion_id:
The LIV Golf tournament/champion identifier (default: ``"59"``).
"""
authorization = self._auth.get_authorization_header()
if not authorization:
logger.error("[LivGolfChannelManager] No authorization token available")
return []
preferred_cdn = self._get_preferred_cdn(authorization)
headers = get_authenticated_headers(authorization)
team_streams = self._fetch_streams(
API_ENDPOINTS["TEAM_STREAMS"].format(champion_id=champion_id),
headers,
stream_kind="team",
)
group_streams = self._fetch_streams(
API_ENDPOINTS["GROUP_STREAMS"].format(champion_id=champion_id),
headers,
stream_kind="group",
)
channels: List[StreamingChannel] = []
new_cache: Dict[str, StreamingChannel] = {}
for stream in team_streams:
channel = self._build_channel(stream, preferred_cdn, stream_kind="team")
if channel:
channels.append(channel)
new_cache[channel.content_id] = channel
for stream in group_streams:
channel = self._build_channel(stream, preferred_cdn, stream_kind="group")
if channel:
channels.append(channel)
new_cache[channel.content_id] = channel
# Update cache with fresh results
self._channels_cache = new_cache
logger.info(
f"[LivGolfChannelManager] champion={champion_id}: "
f"{len(team_streams)} team + {len(group_streams)} group streams → "
f"{len(channels)} channels"
)
return channels
def get_channel(
self, content_id: str, champion_id: str = DEFAULT_CHAMPION_ID
) -> Optional[StreamingChannel]:
"""
Return a single StreamingChannel by its content_id.
If the channel is not in the cache, it triggers a fresh get_channels() call
to discover it.
"""
channel = self._channels_cache.get(content_id)
if channel:
return channel
logger.info(
f"[LivGolfChannelManager] Channel '{content_id}' not in cache — "
"triggering fresh fetch"
)
self.get_channels(champion_id=champion_id)
return self._channels_cache.get(content_id)
# ------------------------------------------------------------------
# Region / CDN resolution
# ------------------------------------------------------------------
def _get_preferred_cdn(self, authorization: str) -> str:
"""
Return the base URL for the closest CDN region.
Result is cached after the first successful call. Falls back to
``FALLBACK_CDN_BASE`` if the regions endpoint is unreachable or returns
unexpected data.
"""
if self._preferred_cdn:
return self._preferred_cdn
self._preferred_cdn = self._resolve_preferred_cdn(authorization)
logger.info(f"[LivGolfChannelManager] Preferred CDN: {self._preferred_cdn}")
return self._preferred_cdn
def _resolve_preferred_cdn(self, authorization: str) -> str:
"""
Fetch /mobii/regions and select the best regional base URL according
to ``REGION_PREFERENCE_ORDER``.
"""
try:
headers = get_authenticated_headers(authorization)
response = self._http.get(
API_ENDPOINTS["REGIONS"],
operation="regions",
headers=headers,
timeout=DEFAULT_REQUEST_TIMEOUT,
)
response.raise_for_status()
data = response.json()
except Exception as exc:
logger.warning(
f"[LivGolfChannelManager] Regions fetch failed, using fallback: {exc}"
)
return FALLBACK_CDN_BASE
# Build a lookup: abbreviation → default uri
region_map: Dict[str, str] = {}
for region in data.get("uris", []):
abbrev = region.get("abbreviation", "")
for uri_entry in region.get("uris", []):
if uri_entry.get("isDefault"):
base = uri_entry.get("uri", "").rstrip("/")
if base:
region_map[abbrev] = base
break
if not region_map:
logger.warning("[LivGolfChannelManager] Empty region map, using fallback CDN")
return FALLBACK_CDN_BASE
# Walk the preference list and return the first available region
for preferred in REGION_PREFERENCE_ORDER:
if preferred in region_map:
return region_map[preferred]
# Fall back to the first returned region
first_base = next(iter(region_map.values()))
logger.warning(
f"[LivGolfChannelManager] No preferred region matched — "
f"using first available: {first_base}"
)
return first_base
# ------------------------------------------------------------------
# Stream fetching
# ------------------------------------------------------------------
def _fetch_streams(
self,
url: str,
headers: Dict[str, str],
stream_kind: str,
) -> List[Dict[str, Any]]:
"""
Fetch a team or group stream list from the API.
Returns the list of stream dicts from the response, or an empty list
on any error.
"""
try:
response = self._http.get(
url,
operation=f"streams_{stream_kind}",
headers=headers,
timeout=DEFAULT_REQUEST_TIMEOUT,
)
response.raise_for_status()
data = response.json()
streams = data.get("streams", [])
logger.debug(
f"[LivGolfChannelManager] Fetched {len(streams)} {stream_kind} streams"
)
return streams
except Exception as exc:
logger.warning(
f"[LivGolfChannelManager] Failed to fetch {stream_kind} streams: {exc}"
)
return []
# ------------------------------------------------------------------
# Channel construction
# ------------------------------------------------------------------
def _build_channel(
self,
stream: Dict[str, Any],
preferred_cdn: str,
stream_kind: str,
) -> Optional[StreamingChannel]:
"""
Convert a single stream dict into a ``StreamingChannel``.
Stream dicts have at least: ``id``, ``name``, ``dashUrl``, ``hlsUrl``.
Team streams additionally carry ``teamId`` and ``livTeamId``.
"""
video_id = stream.get("id")
raw_name = stream.get("name", "")
if not video_id or not raw_name:
logger.debug(
f"[LivGolfChannelManager] Skipping stream with missing id or name: {stream}"
)
return None
dash_url = stream.get("dashUrl", "")
if not dash_url:
logger.debug(
f"[LivGolfChannelManager] Skipping stream '{raw_name}' — no DASH URL"
)
return None
# Rewrite the CDN host to the preferred region
manifest = _rewrite_cdn(dash_url, preferred_cdn)
# Build a human-readable name.
# Raw names look like "Team_01" or "Group_06" — normalise them.
label = self._format_stream_name(raw_name, stream, stream_kind)
# Content ID doubles as the ViewLift video id — uniquely identifies the feed.
content_id = video_id
# team_id is useful as an external reference; omit if not present.
team_id = stream.get("teamId")
liv_team_id = stream.get("livTeamId")
# Build an optional metadata note carried in manifest_script (same pattern
# as magentaeu which stuffs chno/epgid/media there).
meta_parts = [f"kind={stream_kind}", f"vid={video_id}"]
if team_id is not None:
meta_parts.append(f"team={team_id}")
if liv_team_id is not None:
meta_parts.append(f"liv_team={liv_team_id}")
manifest_script = " ".join(meta_parts)
try:
channel = StreamingChannel(
name=label,
content_id=content_id,
provider=PROVIDER_NAME,
manifest=manifest,
manifest_script=manifest_script,
# LIV Golf streams are DRM-free
use_cdm=False,
cdm_type=None,
cdm=None,
cdm_mode="",
streaming_format=STREAMING_FORMAT_DASH,
content_type=CONTENT_TYPE_LIVE,
mode="live",
session_manifest=False,
video="best",
on_demand=False,
)
return channel
except Exception as exc:
logger.warning(
f"[LivGolfChannelManager] Failed to construct Channel for '{raw_name}': {exc}"
)
return None
# ------------------------------------------------------------------
# Naming helpers
# ------------------------------------------------------------------
@staticmethod
def _format_stream_name(
raw_name: str,
stream: Dict[str, Any],
stream_kind: str,
) -> str:
"""
Turn a raw API name like ``"Team_06"`` or ``"Group_03"`` into a
presentable label like ``"LIV Golf – Team Feed 6"`` or
Found Presentable Label ``"LIV Golf – Group Feed 3"``.
For team streams the team number is replaced by the teamId where
available, since the numeric suffix is just an ordering index.
"""
try:
# Extract the numeric suffix (e.g. "06" → 6)
suffix = int(raw_name.split("_")[-1])
except (ValueError, IndexError):
suffix = None
kind_label = "Team" if stream_kind == "team" else "Group"
if stream_kind == "team":
team_id = stream.get("teamId")
index = team_id if team_id is not None else suffix
else:
index = suffix
if index is not None:
return f"LIV Golf – {kind_label} Feed {index}"
return f"LIV Golf – {kind_label} Feed ({raw_name})"
lib/streaming_providers/providers/livgolf/constants.py
-137
View File
@@ -1,137 +0,0 @@
# streaming_providers/providers/livgolf/constants.py
# -*- coding: utf-8 -*-
# ============================================================================
# LIV Golf Configuration
# ============================================================================
from typing import Dict, List
# ============================================================================
# Provider Identity
# ============================================================================
PROVIDER_NAME = "livgolf"
PROVIDER_LABEL = "LIV Golf"
PROVIDER_LOGO = "https://fr.wikipedia.org/wiki/Fichier:LIVGOLF_logo_v2.png"
# ============================================================================
# Application / Device Configuration
# ============================================================================
SITE = "liv-golf"
PLATFORM = "web_browser"
DEVICE_TYPE = "web_browser"
CONTENT_CONSUMPTION = "web"
BROWSER = "Chrome"
BROWSER_VERSION = "147"
OS = "Linux"
USER_AGENT = (
f"Mozilla/5.0 (X11; {OS} x86_64) AppleWebKit/537.36 "
f"(KHTML, like Gecko) {BROWSER}/{BROWSER_VERSION}.0.0.0 Safari/537.36"
)
ORIGIN = "https://www.livgolf.com"
REFERER = "https://www.livgolf.com/"
# Static API key — present in all browser requests as x-api-key header
API_KEY = "79613f5e-52ac-45e7-a8a0-99ea2beb3540"
# ============================================================================
# API Endpoints
# ============================================================================
API_BASE_URL = "https://liv-golf.api.viewlift.com"
API_ENDPOINTS = {
# Anonymous token — no credentials required
"ANONYMOUS_TOKEN": (
API_BASE_URL
+ "/identity/anonymous-token"
+ "?site={site}&platform={platform}&deviceId={device_id}"
),
# Regional CDN clusters — used to pick the closest edge node
"REGIONS": API_BASE_URL + "/v3/content/champions/mobii/regions",
# Team-camera streams for a given champion (tournament) ID
"TEAM_STREAMS": API_BASE_URL + "/v3/content/champions/{champion_id}/team/streams",
# Group-camera streams for a given champion (tournament) ID
"GROUP_STREAMS": API_BASE_URL + "/v3/content/champions/{champion_id}/group/streams",
# Entitlement / stream URL check for a single video id
"ENTITLEMENT": (
API_BASE_URL
+ "/v3/entitlement/video/status"
+ "?id={video_id}&deviceType={device_type}&contentConsumption={content_consumption}&ssaiDisable=false"
),
}
# Champion ID used when no override is given.
# 59 is the current live tournament champion seen in the captured traffic.
DEFAULT_CHAMPION_ID = "59"
# ============================================================================
# Streaming Configuration
# ============================================================================
STREAMING_FORMAT_DASH = "dash"
CONTENT_TYPE_LIVE = "LIVE"
# ============================================================================
# Regional CDN — preference order for closest-edge selection.
#
# The /mobii/regions endpoint returns a list of regions. We rank them by the
# abbreviation prefix so that, e.g., a European client prefers
# gcp-edge-eu-w > gcp-edge-uk-s > gcp-edge-us-* etc.
# The list below is ordered from "most preferred for EU" to "last resort".
# ============================================================================
REGION_PREFERENCE_ORDER: List[str] = [
"gcp-edge-eu-w", # GCP Europe West (Zurich) — closest for most EU users
"gcp-edge-uk-s", # GCP Europe West (London)
"gcp-edge-me-c", # GCP Middle East (Doha)
"gcp-edge-za-n", # GCP South Africa (Johannesburg)
"gcp-edge-a-m", # GCP Asia South (Mumbai)
"gcp-edge-au-s", # GCP Australia South East
"gcp-edge-sa-e", # GCP South America East
"gcp-edge-us-e", # GCP East US
"gcp-edge-us-c", # GCP Central US — default fallback seen in captures
]
# Fallback CDN base URL when region discovery fails entirely
FALLBACK_CDN_BASE = "https://gcp-edge-eu-w.mobii.com"
# ============================================================================
# Request Configuration
# ============================================================================
DEFAULT_REQUEST_TIMEOUT = 30
DEFAULT_MAX_RETRIES = 3
# Token TTL margin — refresh the anonymous token this many seconds before it expires
TOKEN_EXPIRY_MARGIN_SECONDS = 300
# ============================================================================
# Headers helpers
# ============================================================================
def get_base_headers() -> Dict[str, str]:
"""Minimal headers required for all LIV Golf API calls."""
return {
"User-Agent": USER_AGENT,
"Accept": "application/json, text/plain, */*",
"Origin": ORIGIN,
"Referer": REFERER,
}
def get_authenticated_headers(authorization: str) -> Dict[str, str]:
"""Headers that include the anonymous JWT token and the static API key."""
headers = get_base_headers()
headers.update(
{
"Authorization": authorization,
"x-api-key": API_KEY,
}
)
return headers
lib/streaming_providers/providers/livgolf/provider.py
-248
View File
@@ -1,248 +0,0 @@
# streaming_providers/providers/livgolf/provider.py
# -*- coding: utf-8 -*-
"""
LIV Golf streaming provider.
Supported features
------------------
* Channels (live team and group camera feeds) — no authentication required.
* No EPG, no events, no catch-up.
Authentication
--------------
Anonymous JWT token via the ViewLift identity endpoint. The token is
long-lived (~1 year) and is persisted between sessions by the base
authenticator's settings_manager.
"""
from typing import ClassVar, Dict, List, Optional, Tuple
from ...base.models import Event, StreamingChannel
from ...base.models.proxy_models import ProxyConfig
from ...base.provider import StreamingProvider
from ...base.utils.logger import logger
from .auth import LivGolfAuthenticator
from .constants import (
DEFAULT_CHAMPION_ID,
DEFAULT_MAX_RETRIES,
DEFAULT_REQUEST_TIMEOUT,
PROVIDER_LABEL,
PROVIDER_LOGO,
PROVIDER_NAME,
USER_AGENT,
)
from .channel_manager import LivGolfChannelManager
class LivGolfProvider(StreamingProvider):
"""
StreamingProvider implementation for LIV Golf.
Only ``get_channels()`` is meaningful — ``get_events()``, ``get_epg()``,
``get_manifest()``, ``get_catchup_manifest()``, and ``get_drm()`` all
return empty / None, consistent with the base contract.
"""
PROVIDER_LOGO: ClassVar[str] = PROVIDER_LOGO
SUPPORTED_COUNTRIES: ClassVar[List[str]] = ["global"]
def __init__(
self,
country: str = "global",
config_dir: Optional[str] = None,
proxy_config: Optional[ProxyConfig] = None,
proxy_url: Optional[str] = None,
# Allow callers to pin a specific tournament; defaults to the live one.
champion_id: str = DEFAULT_CHAMPION_ID,
) -> None:
logger.info(f"[LivGolfProvider] __init__ START (country={country})")
# StreamingProvider.__init__ expects a country; LIV Golf is global.
super().__init__(country=country)
self._champion_id = champion_id
self.http_manager = self._setup_http_manager(
provider_name=PROVIDER_NAME,
proxy_config=proxy_config,
proxy_url=proxy_url,
config_dir=config_dir,
user_agent=USER_AGENT,
timeout=DEFAULT_REQUEST_TIMEOUT,
max_retries=DEFAULT_MAX_RETRIES,
)
self.authenticator = LivGolfAuthenticator(
config_dir=config_dir,
http_manager=self.http_manager,
proxy_config=self.http_manager.config.proxy_config,
)
self.channel_manager = LivGolfChannelManager(
http_manager=self.http_manager,
authenticator=self.authenticator,
)
logger.info("[LivGolfProvider] __init__ COMPLETE")
# ------------------------------------------------------------------
# StreamingProvider identity properties
# ------------------------------------------------------------------
@property
def provider_name(self) -> str:
return PROVIDER_NAME
@property
def provider_label(self) -> str:
return PROVIDER_LABEL
@property
def provider_logo(self) -> str:
return PROVIDER_LOGO
@property
def uses_dynamic_manifests(self) -> bool:
# Manifest URLs are stable for the duration of a tournament round.
return False
@property
def epg_window(self) -> Tuple[int, int]:
# No EPG support.
return 0, 0
@property
def catchup_window(self) -> int:
return 0
@property
def supported_auth_types(self) -> List[str]:
# Anonymous only — no user credentials accepted.
return ["anonymous"]
# ------------------------------------------------------------------
# Authentication
# ------------------------------------------------------------------
def authenticate(self, **kwargs) -> str:
"""
Obtain / refresh the anonymous token and return it as a Bearer string.
"""
logger.info("[LivGolfProvider] authenticate() called")
force_refresh = kwargs.get("force_refresh", False)
token = self.authenticator.get_bearer_token(force_refresh=force_refresh)
logger.info("[LivGolfProvider] authenticate() complete")
return token or ""
def refresh_authentication(self) -> str:
return self.authenticate(force_refresh=True)
# ------------------------------------------------------------------
# Channels — the sole data surface of this provider
# ------------------------------------------------------------------
def get_channels(self, **kwargs) -> List[StreamingChannel]:
"""
Return all live LIV Golf camera feeds as ``StreamingChannel`` objects.
Both team-camera streams and group-camera streams are included.
Keyword Arguments
-----------------
champion_id : str, optional
Override the tournament champion ID (default: provider-level setting,
itself defaulting to ``DEFAULT_CHAMPION_ID``).
"""
champion_id = kwargs.get("champion_id", self._champion_id)
logger.info(f"[LivGolfProvider] get_channels(champion_id={champion_id})")
try:
# Ensure we have a valid anonymous token before delegating.
if self.authenticator.is_token_expired():
logger.info("[LivGolfProvider] Token expired — refreshing before get_channels")
self.authenticate(force_refresh=True)
channels = self.channel_manager.get_channels(champion_id=champion_id)
logger.info(f"[LivGolfProvider] Returning {len(channels)} channels")
return channels
except Exception as exc:
logger.error(f"[LivGolfProvider] get_channels failed: {exc}")
raise
# ------------------------------------------------------------------
# EPG / Events / manifest — not supported or delegated
# ------------------------------------------------------------------
def get_events(self, **kwargs) -> List[Event]:
"""LIV Golf has no one-time events; streams are linear channels."""
return []
def get_epg(self, channel_id: str, **kwargs) -> List[Dict]:
"""LIV Golf has no EPG."""
return []
def get_manifest(self, content_id: str, **kwargs) -> Optional[str]:
"""
Retrieve the manifest URL for a specific channel ID.
If the channel is not in the channel manager's cache, it triggers a fresh
fetch. LIV Golf channels embed the manifest URL directly.
"""
champion_id = kwargs.get("champion_id", self._champion_id)
logger.info(f"[LivGolfProvider] get_manifest(content_id={content_id})")
try:
channel = self.channel_manager.get_channel(
content_id, champion_id=champion_id
)
if channel:
logger.info(f"[LivGolfProvider] Found manifest for '{content_id}'")
return channel.manifest
logger.warning(f"[LivGolfProvider] Channel '{content_id}' not found")
return None
except Exception as exc:
logger.error(f"[LivGolfProvider] get_manifest failed: {exc}")
return None
def get_catchup_manifest(
self, channel_id: str, start_time: int, end_time: int, **kwargs
) -> Optional[str]:
"""No catch-up support."""
return None
def get_drm(self, content_id: str, **kwargs) -> list:
"""LIV Golf streams are DRM-free."""
return []
def get_dynamic_manifest_params(
self, channel: StreamingChannel, **kwargs
) -> Optional[str]:
return None
def enrich_channel_data(
self, channel: StreamingChannel, **kwargs
) -> Optional[StreamingChannel]:
return None
@staticmethod
def validate_credentials() -> bool:
"""
Anonymous providers do not validate user credentials.
Return True so the base class does not block provider setup.
"""
return True
# ------------------------------------------------------------------
# Class-level helpers
# ------------------------------------------------------------------
@classmethod
def get_static_logo(cls, country: str = None) -> str:
return cls.PROVIDER_LOGO
@classmethod
def get_static_label(cls, country: str = None) -> str:
return PROVIDER_LABEL
resources/settings.xml
-11
View File
@@ -201,17 +201,6 @@
<setting id="bhtelecom_proxy_port" type="number" label="Proxy Port" visible="eq(-2,true)" />
</category>
<!-- =============================================== -->
<!-- LIV GOLF -->
<!-- =============================================== -->
<category id="livgolf" label="LIV Golf">
<setting id="enable_livgolf" type="bool" label="Enable LIV Golf" default="false" />
<setting id="livgolf_country" type="text" label="Country Code" default="global" visible="false" />
<setting id="livgolf_proxy_enabled" type="bool" label="Enable Proxy" default="false" />
<setting id="livgolf_proxy_host" type="text" label="Proxy Host" visible="eq(-1,true)" />
<setting id="livgolf_proxy_port" type="number" label="Proxy Port" visible="eq(-2,true)" />
</category>
<!-- =============================================== -->
<!-- M3U PROVIDER (Generic - enables/disables ALL M3U groups) -->
<!-- =============================================== -->