- default state of new providers is off

- introduce channel cache
This commit is contained in:
Nirvana
2026-10-27 11:47:04 +01:00
parent c03322ac7d
commit 79714e328f
7 changed files with 581 additions and 22 deletions
@@ -0,0 +1,276 @@
# streaming_providers/base/channel_cache.py
"""
In-memory channel-list cache for the provider layer.
What it gives the PVR client:
* a warm cache: GET /api/providers/<p>/channels is answered from memory
* single-flight: concurrent callers (warm-up, PVR, M3U generation) share ONE
upstream load per (provider, params); nobody logs in twice in parallel
* stale-while-revalidate: past the TTL the old list is served at once and
refreshed in a background thread; only a cold miss blocks
* a status per provider (idle / warming / ready / error) for /api/providers
Memory only, by design: channel objects are not serialised anywhere yet.
invalidate() must be called whenever the provider instance or its account
changes (credentials, reinitialize, enable/disable, rediscover). A load that
was already running when invalidate() was called is delivered to its waiters
but never stored (generation counter).
"""
import re
import threading
import time
from dataclasses import dataclass
from typing import Any, Callable, Dict, Hashable, List, Optional, Tuple
from .utils.logger import logger
DEFAULT_TTL = 24 * 3600.0 # channel lists are stable: fresh for 24 h
DEFAULT_MAX_STALE = 7 * 24 * 3600.0 # serve stale (and refresh) up to 7 days; older = cold miss
EMPTY_RESULT_TTL = 30.0 # an empty list is probably a transient failure: retry soon
REFRESH_BACKOFF = 60.0 # after a failed background refresh, wait before the next one
ERROR_TTL = 60.0 # a failed COLD load is remembered: no upstream login per client retry
Key = Tuple[str, Hashable]
_URL_RE = re.compile(r"https?://\S+")
def _safe_error(exc: BaseException) -> str:
"""Error text for the API: no URLs (they may carry tokens), max 200 chars."""
return _URL_RE.sub("<url>", f"{type(exc).__name__}: {exc}")[:200]
@dataclass
class _Entry:
channels: List[Any]
fetched_at: float
ttl: float
retry_after: float = 0.0
owner: Any = None # the provider instance this list came from
class _Flight:
__slots__ = ("done", "result", "error", "gen", "owner")
def __init__(self, gen: int, owner: Any = None):
self.owner = owner
self.done = threading.Event()
self.result: Optional[List[Any]] = None
self.error: Optional[BaseException] = None
self.gen = gen
class ChannelCache:
def __init__(
self,
ttl: float = DEFAULT_TTL,
max_stale: float = DEFAULT_MAX_STALE,
empty_ttl: float = EMPTY_RESULT_TTL,
refresh_backoff: float = REFRESH_BACKOFF,
error_ttl: float = ERROR_TTL,
clock: Callable[[], float] = time.monotonic,
):
self._ttl = ttl
self._max_stale = max_stale
self._empty_ttl = empty_ttl
self._backoff = refresh_backoff
self._error_ttl = error_ttl
self._clock = clock
self._lock = threading.Lock()
self._entries: Dict[Key, _Entry] = {}
self._flights: Dict[Key, _Flight] = {}
self._generation: Dict[str, int] = {}
self._last_error: Dict[str, str] = {}
self._failures: Dict[Key, Tuple[BaseException, float, Any]] = {}
# ------------------------------------------------------------------
def get(
self,
provider: str,
loader: Callable[[], List[Any]],
*,
params: Hashable = (),
ttl: Optional[float] = None,
owner: Any = None,
) -> List[Any]:
"""
Return the channel list, loading it through `loader` when needed.
`owner` is the provider instance. A list (or failure, or running load)
that belongs to a DIFFERENT instance is never served: replacing an
instance invalidates its cache by identity, whatever code path did it.
"""
key: Key = (provider, params)
ttl = self._ttl if ttl is None else ttl
now = self._clock()
with self._lock:
entry = self._entries.get(key)
if entry is not None and owner is not None and entry.owner is not owner:
del self._entries[key]
entry = None
if entry is not None:
age = now - entry.fetched_at
if age < entry.ttl:
return list(entry.channels)
if age < self._max_stale:
stale = list(entry.channels)
if now >= entry.retry_after:
flight, created = self._flight_locked(key, owner)
if created:
self._spawn_refresh(key, provider, loader, flight, ttl)
return stale
if key not in self._flights:
failure = self._failures.get(key)
if failure is not None:
if owner is not None and failure[2] is not owner:
del self._failures[key]
elif now - failure[1] < self._error_ttl:
# Same answer as the failed load, without asking the upstream again.
raise failure[0].with_traceback(None)
else:
del self._failures[key]
flight, created = self._flight_locked(key, owner)
if created:
self._run(key, provider, loader, flight, ttl)
else:
flight.done.wait()
if flight.error is not None:
raise flight.error
return list(flight.result or [])
def refresh(
self,
provider: str,
loader: Callable[[], List[Any]],
*,
params: Hashable = (),
ttl: Optional[float] = None,
owner: Any = None,
) -> bool:
"""
Reload now, in the calling thread, even if the entry is still fresh.
Single-flight like get(): returns False (and does nothing) when a load
for this key is already running. On failure the old list is kept and
the exception propagates (used by the background refresher).
"""
key: Key = (provider, params)
with self._lock:
flight, created = self._flight_locked(key, owner)
if not created:
return False
self._run(key, provider, loader, flight, self._ttl if ttl is None else ttl)
return True
def due(self, fraction: float) -> List[Key]:
"""Keys whose age reached `fraction` of their TTL and are not in backoff."""
now = self._clock()
with self._lock:
return [
key
for key, e in self._entries.items()
if now - e.fetched_at >= e.ttl * fraction and now >= e.retry_after
]
def invalidate(self, provider: Optional[str] = None) -> None:
"""Drop cached lists (one provider, or all) and discard loads in flight."""
with self._lock:
if provider is None:
names = (
set(self._generation)
| {k[0] for k in self._entries}
| {k[0] for k in self._flights}
)
self._entries.clear()
self._last_error.clear()
self._failures.clear()
else:
names = {provider}
for key in [k for k in self._entries if k[0] == provider]:
del self._entries[key]
self._last_error.pop(provider, None)
for key in [k for k in self._failures if k[0] == provider]:
del self._failures[key]
for name in names:
self._generation[name] = self._generation.get(name, 0) + 1
def status(self, provider: str) -> Dict[str, Any]:
"""Snapshot for the API: state is idle | warming | ready | error."""
now = self._clock()
with self._lock:
entries = [e for k, e in self._entries.items() if k[0] == provider]
loading = any(k[0] == provider for k in self._flights)
error = self._last_error.get(provider)
if entries:
newest = max(entries, key=lambda e: e.fetched_at)
age = now - newest.fetched_at
return {
"state": "ready",
"channel_count": len(newest.channels),
"age_seconds": int(age),
"stale": age >= newest.ttl,
"refreshing": loading,
"error": error,
}
if loading:
return {"state": "warming", "error": error}
if error:
return {"state": "error", "error": error}
return {"state": "idle"}
# ------------------------------------------------------------------
def _flight_locked(self, key: Key, owner: Any = None) -> Tuple[_Flight, bool]:
flight = self._flights.get(key)
if flight is not None:
if owner is None or flight.owner is owner:
return flight, False
# Running load belongs to a replaced instance: discard its result, start anew.
self._generation[key[0]] = self._generation.get(key[0], 0) + 1
flight = _Flight(self._generation.get(key[0], 0), owner)
self._flights[key] = flight
return flight, True
def _spawn_refresh(self, key, provider, loader, flight, ttl) -> None:
def work():
try:
self._run(key, provider, loader, flight, ttl)
except Exception as exc: # already recorded in _last_error
logger.warning(f"ChannelCache: background refresh of '{provider}' failed: {exc}")
threading.Thread(target=work, name=f"channel-refresh-{provider}", daemon=True).start()
def _run(self, key: Key, provider: str, loader, flight: _Flight, ttl: float) -> None:
try:
channels = list(loader())
except Exception as exc:
flight.error = exc
with self._lock:
if self._generation.get(provider, 0) == flight.gen:
self._last_error[provider] = _safe_error(exc)
self._failures[key] = (exc, self._clock(), flight.owner)
entry = self._entries.get(key)
if entry is not None: # keep serving stale, but do not hammer upstream
entry.retry_after = self._clock() + self._backoff
raise
else:
flight.result = channels
with self._lock:
if self._generation.get(provider, 0) == flight.gen:
self._entries[key] = _Entry(
channels,
self._clock(),
ttl if channels else min(ttl, self._empty_ttl),
owner=flight.owner,
)
self._last_error.pop(provider, None)
self._failures.pop(key, None)
finally:
with self._lock:
if self._flights.get(key) is flight:
del self._flights[key]
if flight.result is None and flight.error is None:
flight.error = RuntimeError("channel load aborted")
flight.done.set()
@@ -4,8 +4,11 @@
Channel-related operations separated from core registry.
"""
from typing import Dict, List, Optional
import threading
import time
from typing import Any, Dict, Iterable, List, Optional
from .channel_cache import ChannelCache
from .models import StreamingChannel
from .utils.logger import logger
@@ -13,20 +16,46 @@ from .utils.logger import logger
class ChannelOperations:
"""Handles all channel-related operations."""
def __init__(self, registry):
def __init__(self, registry, cache: Optional[ChannelCache] = None):
self.registry = registry
self.cache = cache or ChannelCache()
self._refresher: Optional[threading.Thread] = None
self._refresher_stop = threading.Event()
logger.debug("ChannelOperations: Initialized")
def get_channels(
self, provider_name: str, fetch_manifests: bool = False, **kwargs
) -> List[StreamingChannel]:
"""Get channels from a specific provider."""
"""
Get channels from a specific provider (cached, see ChannelCache).
`fetch_manifests` is accepted for compatibility but has no effect (the
enrichment it controlled is disabled); it is deliberately NOT part of
the cache key -- callers disagree on its default and would otherwise
split one list into two cache entries.
"""
provider = self.registry.get_provider(provider_name)
if not provider:
raise ValueError(f"Provider '{provider_name}' not found or disabled")
channels = provider.get_channels(**kwargs)
logger.info(f"Retrieved {len(channels)} channels from '{provider_name}'")
# None-valued kwargs (e.g. country=None from the route) do not change the
# result, so they must not split the cache key from the warm-up call.
params = tuple(sorted((k, v) for k, v in kwargs.items() if v is not None))
try:
hash(params)
except TypeError:
return self._load(provider_name, provider, kwargs) # uncacheable
# A provider whose Channel objects embed expiring URLs/tokens can
# shorten its cache with a class attribute CHANNEL_CACHE_TTL (seconds).
ttl = getattr(provider, "CHANNEL_CACHE_TTL", None)
channels = self.cache.get(
provider_name,
lambda: self._load(provider_name, provider, kwargs),
params=params,
ttl=ttl,
owner=provider,
)
# if fetch_manifests and not provider.uses_dynamic_manifests:
# enriched = []
@@ -39,6 +68,146 @@ class ChannelOperations:
return channels
@staticmethod
def _load(provider_name: str, provider, kwargs: Dict[str, Any]) -> List[StreamingChannel]:
channels = provider.get_channels(**kwargs)
logger.info(f"Retrieved {len(channels)} channels from '{provider_name}'")
return channels
# ------------------------------------------------------------------
# Warm-up / cache state
# ------------------------------------------------------------------
def warm_up(self, provider_names: Optional[Iterable[str]] = None, max_parallel: int = 4) -> None:
"""
Load channel lists into the cache in the background (returns at once).
Goes through get_channels(), so auth + fetch happen exactly as for a
client request and concurrent requests join the running load.
"""
names = list(provider_names) if provider_names is not None else self.registry.list_providers()
if not names:
return
def run_all():
gate = threading.Semaphore(max(1, max_parallel))
def one(name: str):
try:
self.get_channels(name)
except Exception as exc:
logger.warning(f"ChannelOperations: warm-up of '{name}' failed: {exc}")
finally:
gate.release()
for name in names:
gate.acquire()
threading.Thread(
target=one, args=(name,), name=f"channel-warmup-{name}", daemon=True
).start()
threading.Thread(target=run_all, name="channel-warmup", daemon=True).start()
logger.info(f"ChannelOperations: warm-up started for {len(names)} providers")
# ------------------------------------------------------------------
# Automatic refresh: keeps the cache warm for every client, forever
# ------------------------------------------------------------------
def start_auto_refresh(
self,
check_interval: float = 900.0,
refresh_fraction: float = 0.8,
error_retry: float = 3600.0,
max_parallel: int = 4,
) -> None:
"""
Background thread that, every `check_interval` seconds:
* reloads lists that reached `refresh_fraction` of their TTL, so a
client never meets a stale or cold list while the backend runs
* (re)loads providers that have NO list yet: never loaded (idle) at
once, failed ones (error) at most every `error_retry` seconds, so
wrong credentials are not retried against the upstream login
every few minutes.
A failed reload keeps the old list. Idempotent.
"""
if self._refresher is not None and self._refresher.is_alive():
return
self._refresher_stop.clear()
attempts: Dict[str, float] = {}
def loop():
while not self._refresher_stop.wait(check_interval):
try:
self.refresh_tick(refresh_fraction, error_retry, attempts, max_parallel)
except Exception as exc:
logger.warning(f"ChannelOperations: refresh tick failed: {exc}")
self._refresher = threading.Thread(target=loop, name="channel-refresher", daemon=True)
self._refresher.start()
logger.info(f"ChannelOperations: auto refresh every {int(check_interval)}s")
def stop_auto_refresh(self) -> None:
self._refresher_stop.set()
def refresh_tick(
self,
refresh_fraction: float,
error_retry: float,
attempts: Dict[str, float],
max_parallel: int = 4,
join_timeout: float = 300.0,
) -> None:
"""
One refresher pass (public so it can be tested without waiting).
A hung provider cannot block the refresher beyond `join_timeout`; its
load keeps running and is skipped by later ticks (single-flight).
"""
now = time.monotonic()
targets: Dict[Any, Dict[str, Any]] = {}
for name, params in self.cache.due(refresh_fraction):
targets[(name, params)] = dict(params)
for name in self.registry.list_providers():
state = self.cache.status(name)["state"]
if state == "idle" or (state == "error" and now - attempts.get(name, 0.0) >= error_retry):
attempts[name] = now
targets[(name, ())] = {}
if not targets:
return
gate = threading.Semaphore(max(1, max_parallel))
threads = []
def one(name: str, params, kwargs: Dict[str, Any]):
try:
provider = self.registry.get_provider(name)
if provider is None:
return
self.cache.refresh(
name,
lambda: self._load(name, provider, kwargs),
params=params,
ttl=getattr(provider, "CHANNEL_CACHE_TTL", None),
owner=provider,
)
except Exception as exc:
logger.warning(f"ChannelOperations: refresh of '{name}' failed: {exc}")
finally:
gate.release()
for (name, params), kwargs in targets.items():
gate.acquire()
t = threading.Thread(target=one, args=(name, params, kwargs), name=f"channel-refresh-{name}", daemon=True)
t.start()
threads.append(t)
deadline = time.monotonic() + join_timeout
for t in threads:
t.join(max(0.0, deadline - time.monotonic()))
def get_channel_state(self, provider_name: str) -> Dict[str, Any]:
"""idle | warming | ready | error (+ count/age/stale for ready)."""
return self.cache.status(provider_name)
def invalidate_channels(self, provider_name: Optional[str] = None) -> None:
self.cache.invalidate(provider_name)
def get_channel_manifest(self, provider_name: str, channel_id: str, **kwargs) -> Optional[str]:
"""Get manifest URL for a specific channel."""
provider = self.registry.get_provider(provider_name)
@@ -70,4 +239,4 @@ class ChannelOperations:
result[name] = []
logger.info(f"Retrieved {total} total channels")
return result
return result
+29 -3
View File
@@ -82,6 +82,9 @@ class ProviderManager:
"""
logger.info(f"ProviderManager: Re-discovering providers for country '{country}'...")
discovered = self.registry.discover_all_providers(country)
# Rediscovery builds fresh instances (auth state is gone): drop cached lists.
self.channel_ops.invalidate_channels()
self.channel_ops.warm_up()
logger.info(f"ProviderManager: Re-discovery complete. Found {len(discovered)} providers")
return discovered
@@ -89,16 +92,38 @@ class ProviderManager:
return self.registry.get_provider(provider_name)
def set_provider_enabled(self, provider_name: str, enabled: bool) -> bool:
return self.registry.set_provider_enabled(provider_name, enabled)
success = self.registry.set_provider_enabled(provider_name, enabled)
self.channel_ops.invalidate_channels(provider_name)
if success and enabled:
self.channel_ops.warm_up([provider_name])
return success
def get_all_providers_metadata(self) -> List[Dict[str, Any]]:
return self.registry.get_all_providers_metadata()
def reinitialize_provider(self, provider_name: str) -> bool:
return self.registry.reinitialize_provider(provider_name)
success = self.registry.reinitialize_provider(provider_name)
# New instance / possibly new account: the cached list is not trustworthy.
self.channel_ops.invalidate_channels(provider_name)
if success:
self.channel_ops.warm_up([provider_name])
return success
def reinitialize_providers(self, provider_names: List[str]) -> Dict[str, bool]:
return {name: self.registry.reinitialize_provider(name) for name in provider_names}
return {name: self.reinitialize_provider(name) for name in provider_names}
def warm_up(self, provider_names: Optional[List[str]] = None) -> None:
"""Preload channel lists of enabled providers in the background."""
self.channel_ops.warm_up(provider_names)
def start_auto_refresh(self, **kwargs) -> None:
self.channel_ops.start_auto_refresh(**kwargs)
def stop_auto_refresh(self) -> None:
self.channel_ops.stop_auto_refresh()
def get_channel_state(self, provider_name: str) -> Dict[str, Any]:
return self.channel_ops.get_channel_state(provider_name)
def reinitialize_all_providers(self) -> Dict[str, bool]:
enabled = self.registry.get_enabled_providers()
@@ -112,6 +137,7 @@ class ProviderManager:
def clear_providers(self):
self.registry.clear_providers()
self.channel_ops.invalidate_channels()
@staticmethod
def get_provider_class(provider_name: str):
@@ -341,9 +341,31 @@ class ProviderRegistry:
if instance:
self.providers[instance_name] = instance
self._migrate_enable_default()
logger.info(f"ProviderRegistry: Discovered {len(discovered)} provider instances")
return discovered
def _migrate_enable_default(self) -> None:
"""
One-time: freeze the just-discovered enable state in providers_enabled.json
and make "disabled" the default for providers without an entry.
M3U groups share the single enable key "m3u".
"""
try:
from .settings.provider_enable_manager import ProviderEnableManager
manager = ProviderEnableManager()
if not manager.needs_default_migration():
return
states = {
("m3u" if m.plugin_name == "m3u" else name): m.enabled
for name, m in self.provider_metadata.items()
}
manager.migrate_default_to_disabled(states)
except Exception as e:
logger.warning(f"ProviderRegistry: enable-default migration skipped: {e}")
def _discover_m3u_groups(self, plugin_class, default_country: str) -> List[str]:
"""
Discover M3U groups and register each as a separate provider instance.
@@ -3,7 +3,8 @@
Provider Enable Manager - Handles provider enable/disable status with precedence:
1. Kodi settings (if in Kodi environment and setting exists)
2. providers_enabled.json file
3. Default to enabled (True)
3. Default from the file's "default_enabled" (True for legacy files without the key,
False once migrate_default_to_disabled() has run)
Works in both Kodi and standalone environments.
"""
@@ -117,6 +118,10 @@ class ProviderEnableManager:
)
del data["providers"][provider]
# Persisted default for providers without an explicit entry
if "default_enabled" in data and not isinstance(data["default_enabled"], bool):
del data["default_enabled"]
# Add last_updated if missing
if "last_updated" not in data:
data["last_updated"] = current_time
@@ -230,7 +235,7 @@ class ProviderEnableManager:
Precedence:
1. Kodi settings (if in Kodi and setting exists)
2. providers_enabled.json file
3. Default to True
3. File's "default_enabled" (legacy files: True; migrated files: False)
Args:
provider_name: Provider name (e.g., "joyn_de", "rtlplus")
@@ -250,8 +255,8 @@ class ProviderEnableManager:
if file_enabled is not None:
return file_enabled
# 3. Default to enabled
return True
# 3. Persisted default (legacy files have no key -> unchanged behaviour)
return bool(data.get("default_enabled", True))
def get_enabled_source(self, provider_name: str) -> EnableSource:
"""
@@ -422,6 +427,7 @@ class ProviderEnableManager:
default_data = {
"version": self.DEFAULT_VERSION,
"last_updated": time.time(),
"default_enabled": False,
"providers": {},
}
@@ -464,6 +470,51 @@ class ProviderEnableManager:
),
}
def needs_default_migration(self) -> bool:
"""
True if providers_enabled.json is missing or readable but has no
"default_enabled" key. An unreadable / malformed file returns False:
never rewrite what we cannot parse, the legacy default (True) stays.
"""
try:
if not self.vfs.exists(self.DEFAULT_FILENAME):
return True
raw = self.vfs.read_json(self.DEFAULT_FILENAME)
except Exception as e:
logger.warning(f"ProviderEnableManager: cannot read file for migration check: {e}")
return False
if not isinstance(raw, dict):
return False
return "default_enabled" not in raw
def migrate_default_to_disabled(self, current_states: Dict[str, bool]) -> Tuple[bool, str, int]:
"""
One-time migration: freeze today's effective state, then flip the default.
Every provider in `current_states` (name -> enabled as it was just
discovered under the OLD default) gets an explicit entry unless the file
already has one; afterwards providers WITHOUT an entry default to
disabled. Existing installs therefore keep exactly what they have.
Returns (success, message, entries_added). Idempotent.
"""
if not self.needs_default_migration():
return True, "default already migrated (or file unreadable)", 0
data = self._load_file(force_reload=True)
added = 0
for name, enabled in current_states.items():
if name not in data["providers"]:
data["providers"][name] = bool(enabled)
added += 1
data["default_enabled"] = False
if not self._save_file(data):
return False, "Failed to save migrated provider defaults", 0
message = f"Default flipped to disabled; froze {added} provider entries"
logger.info(f"ProviderEnableManager: {message}")
return True, message, added
def migrate_from_kodi(self) -> Tuple[bool, str, int]:
"""
Migrate all Kodi enable settings to the file.
@@ -516,4 +567,4 @@ class ProviderEnableManager:
except Exception as e:
message = f"Error migrating from Kodi: {e}"
logger.error(f"ProviderEnableManager: {message}")
return False, message, 0
return False, message, 0
+5
View File
@@ -67,6 +67,9 @@ def setup_provider_routes(app, manager, service):
try:
# Get metadata for ALL providers (enabled + disabled)
all_metadata = manager.get_all_providers_metadata()
for metadata in all_metadata:
if metadata["enabled"]:
metadata["channel_state"] = manager.get_channel_state(metadata["name"])
# For backward compatibility, also get details for enabled providers
enabled_providers = []
@@ -131,6 +134,8 @@ def setup_provider_routes(app, manager, service):
"enabled": metadata["enabled"],
"instance_ready": metadata["instance_ready"],
"requires_credentials": metadata["requires_credentials"],
# idle | warming | ready | error (+ channel_count, age_seconds, stale)
"channel_state": metadata.get("channel_state"),
}
enabled_providers.append(provider_details)
+18 -8
View File
@@ -2231,14 +2231,24 @@ def start_service(service_instance):
# Determine if we should run in debug mode
debug_mode = service_instance.env_manager.get_config("debug_mode", False)
run(
service_instance.app,
server=ThreadedWSGIRefServer,
host="0.0.0.0",
port=port,
quiet=not debug_mode,
debug=debug_mode,
)
# Preload channel lists in the background; the server below starts at once.
try:
service_instance.manager.warm_up()
service_instance.manager.start_auto_refresh()
except Exception as e:
logger.warning(f"Channel warm-up could not be started: {e}")
try:
run(
service_instance.app,
server=ThreadedWSGIRefServer,
host="0.0.0.0",
port=port,
quiet=not debug_mode,
debug=debug_mode,
)
finally:
service_instance.manager.stop_auto_refresh()
def run_kodi_service():