From 4dd1affbe35dfc0dccdfa994f071d8a6cec96809 Mon Sep 17 00:00:00 2001 From: Nirvana Date: Sun, 1 Nov 2026 15:07:41 +0100 Subject: [PATCH] add analytics emitter --- analytics/__init__.py | 17 ++ analytics/emitter.py | 292 +++++++++++++++++++++++++++++++++++ analytics/events.py | 54 +++++++ analytics/integration.py | 99 ++++++++++++ analytics/privacy.py | 73 +++++++++ routes/streams/channels.py | 17 +- routes/streams/events.py | 10 ++ routes/streams/recordings.py | 10 ++ routes/streams/vod.py | 10 ++ service.py | 27 ++++ 10 files changed, 607 insertions(+), 2 deletions(-) create mode 100644 analytics/__init__.py create mode 100644 analytics/emitter.py create mode 100644 analytics/events.py create mode 100644 analytics/integration.py create mode 100644 analytics/privacy.py diff --git a/analytics/__init__.py b/analytics/__init__.py new file mode 100644 index 0000000..24b53e9 --- /dev/null +++ b/analytics/__init__.py @@ -0,0 +1,17 @@ +"""Analytics event emission. Optional and best-effort: it must never affect playback.""" +from .emitter import ( + AnalyticsEmitter, + configure_from_env, + get_emitter, + shutdown_emitter, +) +from .integration import annotate, tracked + +__all__ = [ + "AnalyticsEmitter", + "configure_from_env", + "get_emitter", + "shutdown_emitter", + "annotate", + "tracked", +] \ No newline at end of file diff --git a/analytics/emitter.py b/analytics/emitter.py new file mode 100644 index 0000000..e8340ad --- /dev/null +++ b/analytics/emitter.py @@ -0,0 +1,292 @@ +""" +Asynchronous, best-effort event emitter. + +Design goal: the request thread pays for ONE deque.append and nothing else. + + - emit() is a length check + deque.append. No lock, no condition-variable + notify, no I/O, no logging, no allocation beyond the dict the caller built. + It cannot block and does not wake the consumer thread. + - The consumer thread polls every `flush_interval` seconds (Event.wait), so + there is no per-event cross-thread signalling. Anonymisation, JSON encoding + and HTTP all happen there. + - Overflow drops the NEWEST event and counts it; the consumer logs the drop + count (rate-limited). Nothing is ever logged from a request thread. + - A failing sink drops the batch (no retry queue) and backs off exponentially + up to `max_backoff`, so a dead endpoint is not hammered. + - Counters are intentionally lock-free. `dropped` is incremented from request + threads and may under-count under heavy contention; it is a health signal, + not an audit log. +""" +from __future__ import annotations + +import atexit +import collections +import os +import threading +import time +from typing import Any, Callable, Deque, Dict, List, Mapping, Optional + +import requests + +from streaming_providers.base.utils import logger +from .events import finalize +from .privacy import make_anonymizer + +Sink = Callable[[List[Dict[str, Any]]], None] +Transform = Callable[[Dict[str, Any]], Dict[str, Any]] + + +class HttpSink: + """Batched JSON POST: {"events": [...]}. Raises on non-2xx.""" + + def __init__(self, url: str, *, timeout: float = 5.0, auth_token: Optional[str] = None): + self._url = url + self._timeout = (min(3.05, timeout), timeout) # (connect, read) + self._session = requests.Session() # only ever used from the consumer thread + self._session.headers["User-Agent"] = "ultimate-backend-analytics/1" + if auth_token: + self._session.headers["Authorization"] = f"Bearer {auth_token}" + + def __call__(self, batch: List[Dict[str, Any]]) -> None: + resp = self._session.post(self._url, json={"events": batch}, timeout=self._timeout) + resp.raise_for_status() + + def close(self) -> None: + self._session.close() + + +class AnalyticsEmitter: + def __init__( + self, + sink: Sink, + *, + transform: Optional[Transform] = None, + max_queue: int = 10_000, + batch_size: int = 100, + flush_interval: float = 1.0, + max_backoff: float = 30.0, + warn_interval: float = 60.0, + report_interval: float = 300.0, + ): + self._sink = sink + self._transform = transform or finalize + self._max_queue = max_queue + self._batch_size = batch_size + self._flush_interval = flush_interval + self._max_backoff = max_backoff + self._warn_interval = warn_interval + self._report_interval = report_interval + + self._q: Deque[Dict[str, Any]] = collections.deque() + self._dropped = 0 # queue overflow (approximate, see module docstring) + self._dropped_logged = 0 + self._shipped = 0 + self._failed_batches = 0 + self._failed_events = 0 # lost to sink failure or a malformed event + + self._last_warn: Dict[str, float] = {} + self._last_report = time.monotonic() + self._report_snapshot = (0, 0, 0) + + self._drain_deadline = 0.0 + self._stop = threading.Event() + self._thread = threading.Thread(target=self._run, name="analytics-emitter", daemon=True) + self._thread.start() + + # -- public API (request threads) ------------------------------------ + + def emit(self, raw: Dict[str, Any]) -> None: + """Non-blocking, lock-free, never raises. Drops the event if the queue is full.""" + if len(self._q) >= self._max_queue: + self._dropped += 1 + return + self._q.append(raw) + + # -- public API (anywhere) -------------------------------------------- + + def stats(self) -> Dict[str, int]: + return { + "queue_depth": len(self._q), + "dropped": self._dropped, + "shipped": self._shipped, + "failed_batches": self._failed_batches, + "failed_events": self._failed_events, + } + + def shutdown(self, timeout: float = 5.0) -> None: + """Flush what is queued (bounded by `timeout`), then stop. Idempotent.""" + if self._stop.is_set(): + return + self._drain_deadline = time.monotonic() + timeout + self._stop.set() + self._thread.join(timeout=timeout + 1.0) + close = getattr(self._sink, "close", None) + if callable(close): + try: + close() + except Exception: # noqa: BLE001 + pass + + # -- consumer thread --------------------------------------------------- + + def _warn(self, key: str, message: str) -> None: + now = time.monotonic() + if now - self._last_warn.get(key, -1e9) >= self._warn_interval: + self._last_warn[key] = now + logger.warning(message) + + def _take(self) -> List[Dict[str, Any]]: + batch: List[Dict[str, Any]] = [] + pop = self._q.popleft + while len(batch) < self._batch_size: + try: + batch.append(pop()) + except IndexError: + break + return batch + + def _ship(self, raw_batch: List[Dict[str, Any]]) -> bool: + wire: List[Dict[str, Any]] = [] + for raw in raw_batch: + try: + wire.append(self._transform(raw)) + except Exception: # noqa: BLE001 — one malformed event must not sink the batch + self._failed_events += 1 + if not wire: + return True + try: + self._sink(wire) + except Exception as err: # noqa: BLE001 — sink is pluggable + self._failed_batches += 1 + self._failed_events += len(wire) + self._warn("ship", f"Analytics ship failed, dropped {len(wire)} events: {err}") + return False + self._shipped += len(wire) + return True + + def _report(self) -> None: + dropped = self._dropped + if dropped > self._dropped_logged: + self._warn( + "drop", + f"Analytics queue full: {dropped - self._dropped_logged} events dropped " + f"(total {dropped}). Analytics is best-effort and does not affect playback.", + ) + self._dropped_logged = dropped + now = time.monotonic() + if now - self._last_report >= self._report_interval: + self._last_report = now + snapshot = (self._shipped, self._dropped, self._failed_events) + if snapshot != self._report_snapshot: # stay quiet when idle + self._report_snapshot = snapshot + logger.info(f"Analytics stats: {self.stats()}") + + def _run(self) -> None: + backoff = 0.0 + while not self._stop.is_set(): + wait = self._flush_interval + batch = self._take() + if batch: + if self._ship(batch): + backoff = 0.0 + if len(self._q) >= self._batch_size: + wait = 0.0 # backlog: keep draining without sleeping + else: + backoff = min(max(backoff * 2, 1.0), self._max_backoff) + wait = backoff + self._report() + self._stop.wait(wait) + self._final_flush() + + def _final_flush(self) -> None: + while self._q and time.monotonic() < self._drain_deadline: + if not self._ship(self._take()): + break + if self._q: + logger.warning(f"Analytics shutdown: {len(self._q)} queued events discarded") + + +# -- process-wide singleton ------------------------------------------------- + +_emitter: Optional[AnalyticsEmitter] = None +_lock = threading.Lock() + + +def _env_num(env: Mapping[str, str], name: str, default, cast, minimum): + raw = (env.get(name) or "").strip() + if not raw: + return default + try: + value = cast(raw) + except ValueError: + logger.warning(f"Ignoring invalid {name}={raw!r}; using {default}") + return default + return value if value >= minimum else default + + +def configure_from_env(environ: Optional[Mapping[str, str]] = None) -> Optional[AnalyticsEmitter]: + """ + Create the process-wide emitter from ANALYTICS_* variables. Idempotent. + Never raises. Returns None (analytics fully off, zero overhead) when + ANALYTICS_ENDPOINT_URL is unset or invalid. + + ANALYTICS_ENDPOINT_URL full URL batches are POSTed to (unset = disabled) + ANALYTICS_AUTH_TOKEN optional bearer token + ANALYTICS_MAX_QUEUE default 10000 + ANALYTICS_BATCH_SIZE default 100 + ANALYTICS_FLUSH_INTERVAL seconds, default 1.0 + ANALYTICS_TIMEOUT HTTP read timeout in seconds, default 5.0 + ANALYTICS_ANONYMIZE off | truncate | hash (default off) + ANALYTICS_HASH_SALT required for hash mode + """ + global _emitter + env = os.environ if environ is None else environ + try: + with _lock: + if _emitter is not None: + return _emitter + url = (env.get("ANALYTICS_ENDPOINT_URL") or "").strip() + if not url: + logger.info("Analytics disabled (ANALYTICS_ENDPOINT_URL not set)") + return None + if not url.lower().startswith(("http://", "https://")): + logger.warning(f"Analytics disabled: ANALYTICS_ENDPOINT_URL must be http(s), got {url!r}") + return None + + token = (env.get("ANALYTICS_AUTH_TOKEN") or "").strip() or None + if token and url.lower().startswith("http://"): + logger.warning("Analytics auth token is sent over plain HTTP; use HTTPS") + timeout = _env_num(env, "ANALYTICS_TIMEOUT", 5.0, float, 0.1) + + anonymize = make_anonymizer( + env.get("ANALYTICS_ANONYMIZE", "off"), (env.get("ANALYTICS_HASH_SALT") or "") + ) + _emitter = AnalyticsEmitter( + HttpSink(url, timeout=timeout, auth_token=token), + transform=lambda raw: finalize(anonymize(raw)), + max_queue=_env_num(env, "ANALYTICS_MAX_QUEUE", 10_000, int, 1), + batch_size=_env_num(env, "ANALYTICS_BATCH_SIZE", 100, int, 1), + flush_interval=_env_num(env, "ANALYTICS_FLUSH_INTERVAL", 1.0, float, 0.05), + ) + atexit.register(shutdown_emitter) + logger.info(f"Analytics enabled -> {url.split('?')[0]}") + return _emitter + except Exception as err: # noqa: BLE001 — analytics must never stop the service + logger.warning(f"Analytics could not be initialised: {err}") + return None + + +def get_emitter() -> Optional[AnalyticsEmitter]: + return _emitter + + +def shutdown_emitter(timeout: float = 5.0) -> None: + """Flush and stop the emitter. Safe to call repeatedly or when disabled.""" + global _emitter + with _lock: + emitter, _emitter = _emitter, None + if emitter is not None: + try: + emitter.shutdown(timeout) + except Exception: # noqa: BLE001 + pass \ No newline at end of file diff --git a/analytics/events.py b/analytics/events.py new file mode 100644 index 0000000..ca51936 --- /dev/null +++ b/analytics/events.py @@ -0,0 +1,54 @@ +""" +Analytics event schema. Pure data, no I/O. + +The request thread builds a plain dict of raw values (cheapest possible). +Everything else (anonymisation, None-stripping, schema_version) happens on the +consumer thread via finalize(), so none of it costs the request path anything. +""" +from __future__ import annotations + +from typing import Any, Dict + +try: # TypedDict is documentation only; keep import safe on old interpreters + from typing import TypedDict +except ImportError: # pragma: no cover + TypedDict = dict # type: ignore + +SCHEMA_VERSION = 1 + +# Event vocabulary: the backend only names what happened. +EVENT_MANIFEST_REQUEST = "manifest_request" + +# How the request was served. +DELIVERY_REDIRECT = "redirect" # 302 to the upstream manifest; the client talks to upstream directly afterwards +DELIVERY_PROXIED = "proxied" # manifest body returned (rewritten/decrypted); the media proxy sees the segments +DELIVERY_ERROR = "error" # status >= 400 or an unhandled exception + + +class AnalyticsEvent(TypedDict, total=False): + """Wire format (after finalize). Keys with None values are omitted.""" + event_type: str + ts: float # Unix epoch seconds, taken on the request thread + client_ip: str # raw, truncated or hashed depending on ANALYTICS_ANONYMIZE + provider: str + channel_id: str # id of the content item (channel/event/vod/recording) + content_type: str # "channel" | "event" | "vod" | "recording" + http_status: int + delivery: str # DELIVERY_* + user_agent: str + country: str + is_catchup: bool + drm_variant: str + client_drm: bool + no_proxy: bool + highest_quality_only: bool + proxy: str # only if a route calls annotate(proxy=...); userinfo is always stripped + manifest_refresh_hint: int # only if a route calls annotate(manifest_refresh_hint=...) + schema_version: int + + +def finalize(raw: Dict[str, Any]) -> Dict[str, Any]: + """Drop None values (so consumers can key-presence-check) and stamp the schema version.""" + out = {k: v for k, v in raw.items() if v is not None} + out["schema_version"] = SCHEMA_VERSION + return out \ No newline at end of file diff --git a/analytics/integration.py b/analytics/integration.py new file mode 100644 index 0000000..6b384c4 --- /dev/null +++ b/analytics/integration.py @@ -0,0 +1,99 @@ +""" +The only module route code touches. + + @tracked("channel") wrap a stream handler: one event per call, + def _handle_channel_stream(provider, channel_id): ... + + annotate(is_catchup=..., proxy=...) optional extra fields for the current request + +Cost on the request thread when analytics is enabled: a handful of attribute +reads, one dict literal and one deque.append. When disabled: a single +`get_emitter() is None` check. Everything is wrapped so that no analytics +failure can change a response. +""" +from __future__ import annotations + +import functools +import time +from typing import Any, Callable + +from bottle import HTTPResponse, request, response + +from .emitter import get_emitter +from .events import ( + DELIVERY_ERROR, + DELIVERY_PROXIED, + DELIVERY_REDIRECT, + EVENT_MANIFEST_REQUEST, +) + +_ENV_KEY = "analytics.extra" +_UA_MAX = 256 + + +def annotate(**fields: Any) -> None: + """Attach extra fields to the event of the current request. Never raises.""" + try: + request.environ.setdefault(_ENV_KEY, {}).update(fields) + except Exception: # noqa: BLE001 + pass + + +def _classify(status: int, redirected: bool) -> str: + if status >= 400: + return DELIVERY_ERROR + return DELIVERY_REDIRECT if redirected else DELIVERY_PROXIED + + +def _emit(emitter, content_type: str, provider: str, content_id: str, + status: int, delivery: str) -> None: + try: + ua = request.headers.get("User-Agent") + raw = { + "event_type": EVENT_MANIFEST_REQUEST, + "ts": time.time(), + "client_ip": request.remote_addr, + "provider": provider, + "channel_id": content_id, + "content_type": content_type, + "http_status": status, + "delivery": delivery, + "user_agent": ua[:_UA_MAX] if ua else None, + "country": request.query.get("country") or None, + } + extra = request.environ.get(_ENV_KEY) + if extra: + raw.update(extra) + emitter.emit(raw) + except Exception: # noqa: BLE001 + pass + + +def tracked(content_type: str) -> Callable: + """ + Decorator for `handler(provider, content_id, ...)`. Emits exactly one event + per call, including for redirects (which Bottle raises as HTTPResponse), + handler-set error statuses and unexpected exceptions. Never alters the + handler's return value, raised exception, or response. + """ + def decorator(fn: Callable) -> Callable: + @functools.wraps(fn) + def wrapper(provider, content_id, *args, **kwargs): + emitter = get_emitter() + if emitter is None: + return fn(provider, content_id, *args, **kwargs) + + status, delivery = 500, DELIVERY_ERROR + try: + result = fn(provider, content_id, *args, **kwargs) + status = response.status_code + delivery = _classify(status, redirected=False) + return result + except HTTPResponse as exc: + status = exc.status_code + delivery = _classify(status, redirected=300 <= status < 400) + raise + finally: + _emit(emitter, content_type, provider, content_id, status, delivery) + return wrapper + return decorator \ No newline at end of file diff --git a/analytics/privacy.py b/analytics/privacy.py new file mode 100644 index 0000000..ba37d64 --- /dev/null +++ b/analytics/privacy.py @@ -0,0 +1,73 @@ +""" +Optional anonymisation. Runs on the consumer thread only, never on the request path. + +ANALYTICS_ANONYMIZE: + off client_ip is sent as-is (default) + truncate IPv4 -> /24, IPv6 -> /48 (203.0.113.7 -> 203.0.113.0) + hash salted SHA-256 (first 16 hex chars). Keeps "same client" clustering + without sending the address. Requires ANALYTICS_HASH_SALT; rotate the + salt on a schedule. Note: a salted hash of an IP is pseudonymous, not + anonymous, under GDPR. + +Proxy URLs always lose their userinfo (user:pass@). Unless mode is "off" the +host is also replaced by a short hash, so events still group by proxy. +""" +from __future__ import annotations + +import hashlib +import ipaddress +from typing import Any, Callable, Dict, Optional +from urllib.parse import urlsplit + +from streaming_providers.base.utils import logger + +Anonymizer = Callable[[Dict[str, Any]], Dict[str, Any]] + +_MODES = ("off", "truncate", "hash") + + +def _hash(value: str, salt: str, length: int = 16) -> str: + return hashlib.sha256(f"{salt}|{value}".encode("utf-8")).hexdigest()[:length] + + +def truncate_ip(ip: str) -> Optional[str]: + """IPv4 -> /24 network address, IPv6 -> /48. Returns None if unparsable.""" + try: + addr = ipaddress.ip_address(ip.strip()) + except ValueError: + return None + prefix = 24 if addr.version == 4 else 48 + return str(ipaddress.ip_network(f"{addr}/{prefix}", strict=False).network_address) + + +def sanitize_proxy(proxy: str, *, hide_host: bool, salt: str = "") -> Optional[str]: + """Strip credentials; optionally replace the host with a short hash.""" + parts = urlsplit(proxy) + host = parts.netloc.rpartition("@")[2] + if not parts.scheme or not host: + return None + if hide_host: + return f"{parts.scheme}://{_hash(host, salt, 8)}" + return f"{parts.scheme}://{host}" + + +def make_anonymizer(mode: str = "off", salt: str = "") -> Anonymizer: + mode = (mode or "off").strip().lower() + if mode not in _MODES: + logger.warning(f"Unknown ANALYTICS_ANONYMIZE={mode!r}; falling back to 'truncate'") + mode = "truncate" + if mode == "hash" and not salt: + logger.warning("ANALYTICS_ANONYMIZE=hash needs ANALYTICS_HASH_SALT; falling back to 'truncate'") + mode = "truncate" + + def apply(raw: Dict[str, Any]) -> Dict[str, Any]: + out = dict(raw) + ip = out.get("client_ip") + if ip and mode != "off": + out["client_ip"] = _hash(ip, salt) if mode == "hash" else truncate_ip(ip) + proxy = out.get("proxy") + if proxy: + out["proxy"] = sanitize_proxy(proxy, hide_host=(mode != "off"), salt=salt) + return out + + return apply \ No newline at end of file diff --git a/routes/streams/channels.py b/routes/streams/channels.py index 27a72b9..be0e5a4 100644 --- a/routes/streams/channels.py +++ b/routes/streams/channels.py @@ -8,7 +8,9 @@ channel-specific catchup handling logic. """ import time +from typing import Optional +from analytics.integration import annotate, tracked from bottle import HTTPResponse, request, response from streaming_providers.base.stream_format import StreamFormat from streaming_providers.base.utils import logger @@ -23,6 +25,7 @@ def setup_channel_routes(app, manager, service, helpers): _resolve_stream_unified = helpers["_resolve_stream_unified"] _get_drm_configs = helpers["_get_drm_configs"] + @tracked("channel") def _handle_channel_stream(provider, channel_id): """Single implementation backing /stream/index.mpd — the only channel stream route. Replaces the former /stream/sw-drm/, /stream/noproxy/, @@ -58,8 +61,8 @@ def setup_channel_routes(app, manager, service, helpers): # Always defined so the _resolve_stream_unified call below is # unconditionally safe, even though the ternary guards already # prevent None from being passed when is_catchup is False. - start_time_int: int | None = None - end_time_int: int | None = None + start_time_int: Optional[int] = None + end_time_int: Optional[int] = None if is_catchup: try: @@ -87,6 +90,16 @@ def setup_channel_routes(app, manager, service, helpers): # Window validation (catchup_hours lookup, age check) lives # inside _resolve_stream_unified via _validate_catchup_window. + # Analytics context for @tracked — a dict update on request.environ, + # nothing else. Emission itself happens in the decorator's finally. + annotate( + is_catchup=is_catchup, + drm_variant=drm_variant, + client_drm=client_drm, + no_proxy=no_proxy, + highest_quality_only=highest_quality_only, + ) + return _resolve_stream_unified( CONTENT_TYPE_CHANNEL, provider, channel_id, country=country, diff --git a/routes/streams/events.py b/routes/streams/events.py index 0e3ef58..fe74957 100644 --- a/routes/streams/events.py +++ b/routes/streams/events.py @@ -13,6 +13,7 @@ no_proxy/highest_quality_only query params on the one route, resolved directly through _resolve_stream_unified. """ +from analytics.integration import annotate, tracked from bottle import HTTPResponse, request, response from streaming_providers.base.stream_format import StreamFormat from streaming_providers.base.utils import logger @@ -28,6 +29,7 @@ def setup_event_routes(app, manager, service, helpers): _resolve_stream_unified = helpers["_resolve_stream_unified"] _get_drm_configs = helpers["_get_drm_configs"] + @tracked("event") def _handle_event_stream(provider, event_id): """Single implementation backing /stream/index.mpd — the only event stream route. Replaces the former /stream/sw-drm/, /stream/proxied/, @@ -58,6 +60,14 @@ def setup_event_routes(app, manager, service, helpers): f"highest_quality_only={highest_quality_only}" ) + # Analytics context for @tracked: a dict update on request.environ. + annotate( + drm_variant=drm_variant, + client_drm=client_drm, + no_proxy=no_proxy, + highest_quality_only=highest_quality_only, + ) + return _resolve_stream_unified( CONTENT_TYPE_EVENT, provider, event_id, country=country, diff --git a/routes/streams/recordings.py b/routes/streams/recordings.py index 3a72c88..304796d 100644 --- a/routes/streams/recordings.py +++ b/routes/streams/recordings.py @@ -13,6 +13,7 @@ client_drm on the one route, resolved directly through _resolve_stream_unified. """ +from analytics.integration import annotate, tracked from bottle import HTTPResponse, request, response from streaming_providers.base.stream_format import StreamFormat from streaming_providers.base.utils import logger @@ -28,6 +29,7 @@ def setup_recording_routes(app, manager, service, helpers): _resolve_stream_unified = helpers["_resolve_stream_unified"] _get_drm_configs = helpers["_get_drm_configs"] + @tracked("recording") def _handle_recording_stream(provider, recording_id): """Single implementation backing /stream/index.mpd — the only recording stream route. Replaces the former /stream/proxied/ route, @@ -56,6 +58,14 @@ def setup_recording_routes(app, manager, service, helpers): f"highest_quality_only={highest_quality_only}" ) + # Analytics context for @tracked: a dict update on request.environ. + annotate( + drm_variant=drm_variant, + client_drm=client_drm, + no_proxy=no_proxy, + highest_quality_only=highest_quality_only, + ) + return _resolve_stream_unified( CONTENT_TYPE_RECORDING, provider, recording_id, country=country, diff --git a/routes/streams/vod.py b/routes/streams/vod.py index 198815d..05e33e9 100644 --- a/routes/streams/vod.py +++ b/routes/streams/vod.py @@ -13,6 +13,7 @@ query params on the one route, resolved directly through _resolve_stream_unified. """ +from analytics.integration import annotate, tracked from bottle import HTTPResponse, request, response from streaming_providers.base.stream_format import StreamFormat from streaming_providers.base.utils import logger @@ -28,6 +29,7 @@ def setup_vod_routes(app, manager, service, helpers): _resolve_stream_unified = helpers["_resolve_stream_unified"] _get_drm_configs = helpers["_get_drm_configs"] + @tracked("vod") def _handle_vod_stream(provider, vod_id): """Single implementation backing /stream/index.mpd — the only VOD stream route. Replaces the former /stream/proxied/ and @@ -60,6 +62,14 @@ def setup_vod_routes(app, manager, service, helpers): f"highest_quality_only={highest_quality_only}" ) + # Analytics context for @tracked: a dict update on request.environ. + annotate( + drm_variant=drm_variant, + client_drm=client_drm, + no_proxy=no_proxy, + highest_quality_only=highest_quality_only, + ) + return _resolve_stream_unified( CONTENT_TYPE_VOD, provider, vod_id, country=country, diff --git a/service.py b/service.py index b2b93c0..7a9be3a 100644 --- a/service.py +++ b/service.py @@ -3,6 +3,7 @@ import hashlib import json import math import os +import signal import sys import threading import time @@ -48,6 +49,7 @@ try: get_vfs_instance, is_kodi_environment, ) + from analytics.emitter import configure_from_env, shutdown_emitter except ImportError as import_err: print( f"Ultimate Backend: Critical import failed - {str(import_err)}", file=sys.stderr @@ -114,6 +116,10 @@ class UltimateService: self.vfs = get_vfs_instance(subdir="m3u_cache") logger.info(f"VFS initialized for M3U caching: {self.vfs.base_path}") + # Analytics: optional, best-effort, disabled unless ANALYTICS_ENDPOINT_URL + # is set. configure_from_env() never raises. + configure_from_env() + self.mpd_cache = MPDCacheManager() logger.info(f"MPD cache initialized: {self.mpd_cache.vfs.base_path}") @@ -2223,6 +2229,24 @@ class ThreadedWSGIRefServer(ServerAdapter): server.server_close() +def _install_sigterm_handler(): + """ + Turn SIGTERM (docker stop, systemd) into a normal exit so start_service's + finally block runs. Python's default SIGTERM action kills the process + without running it. No-op outside the main thread (e.g. Kodi mode). + """ + if threading.current_thread() is not threading.main_thread(): + return + + def _handler(signum, frame): + raise SystemExit(0) + + try: + signal.signal(signal.SIGTERM, _handler) + except (ValueError, OSError): + pass + + def start_service(service_instance): """Start the Bottle server""" port = service_instance.server_port @@ -2238,6 +2262,8 @@ def start_service(service_instance): except Exception as e: logger.warning(f"Channel warm-up could not be started: {e}") + _install_sigterm_handler() + try: run( service_instance.app, @@ -2249,6 +2275,7 @@ def start_service(service_instance): ) finally: service_instance.manager.stop_auto_refresh() + shutdown_emitter() # flush queued analytics events (bounded, never raises) def run_kodi_service():