metastar/meta-stremio
4
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

adaptation due to meta-core new API

Browse Source
This commit is contained in:
Pierre
2026-05-24 17:44:12 +02:00
parent 56eced073a
commit 37223c442b
3 changed files with 252 additions and 52 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src/storage/leader_client.py
+9 -3
View File
@@ -139,13 +139,19 @@ class LeaderClient:
# api-mediated-access PR D: redisUrl is no longer published by
# meta-core. If it shows up, an old build snuck back in —
# log a warning so it surfaces in the operator's face.
# fail loudly so the rollback can't go unnoticed. Set
# ALLOW_LEGACY_REDIS_URL=1 to downgrade to a warning during
# a deliberate temporary rollback.
if data.get('redisUrl'):
print(
"[LeaderClient] WARNING: meta-core still publishes redisUrl; "
msg = (
"[LeaderClient] meta-core still publishes redisUrl; "
"direct Redis access was retired by the api-mediated-access "
"lockdown. Verify meta-core version."
)
if os.environ.get("ALLOW_LEGACY_REDIS_URL") == "1":
print("WARNING: " + msg)
else:
raise RuntimeError(msg)
self._cached_urls = URLsResponse(
hostname=data.get('hostname', ''),
base_url=data.get('baseUrl', ''),
src/storage/leader_storage.py
+123 -49
View File
@@ -20,6 +20,7 @@ from typing import List, Optional, Callable
from .provider import StorageProvider, VideoMetadata
from .leader_client import LeaderClient, LeaderLockInfo
from .meta_consumer import MetaConsumer
from .meta_core_api_client import MetaCoreApiClient
# Import webdav_client to configure it after leader discovery
import webdav_client
@@ -70,6 +71,7 @@ class LeaderStorage(StorageProvider):
self._leader_client: Optional[LeaderClient] = None
self._client: Optional[redis.Redis] = None
self._api_client: Optional[MetaCoreApiClient] = None
self._connected = False
self._lock = threading.Lock()
@@ -122,23 +124,14 @@ class LeaderStorage(StorageProvider):
# Try to get leader info (short timeout per attempt)
info = self._leader_client.get_leader_info()
# api-mediated-access PR D: redis_url may be empty when
# meta-core no longer advertises it. Synthesise from the
# hostname in that case so meta-stremio's read paths still
# work until they're migrated to HTTP. The synthesised URL
# assumes the docker-network `meta-core` alias or the
# canonical hostname is reachable on port 6379.
redis_url = info.redis_url if info else None
if info and not redis_url and info.hostname:
redis_url = f"redis://{info.hostname}:6379"
print(
"[LeaderStorage] redis_url absent in leader info; "
f"falling back to {redis_url}"
)
if info and redis_url:
print(f"[LeaderStorage] Found leader at {redis_url} (attempt {attempt})")
self._connect_to_redis(redis_url)
# api-mediated-access PR D: when redis_url is absent, run in
# HTTP-only mode against meta-core's /meta/* surface. Redis
# access has been retired; the API client covers every read
# this storage provider needs. Only fall back to Redis when
# both redis_url and api_url are present (transition state).
if info and info.api_url:
print(f"[LeaderStorage] Found leader at {info.api_url} (attempt {attempt})")
self._connect_via_api(info)
if self._connected:
# Configure WebDAV client with leader's internal WebDAV URL (for container-to-container access)
@@ -154,7 +147,7 @@ class LeaderStorage(StorageProvider):
self._leader_client.start_watching()
return
else:
print(f"[LeaderStorage] Failed to connect to Redis, retrying...")
print(f"[LeaderStorage] Failed to connect to meta-core, retrying...")
else:
print(f"[LeaderStorage] Waiting for meta-core leader... (attempt {attempt})")
@@ -188,9 +181,9 @@ class LeaderStorage(StorageProvider):
try:
info = self._leader_client.get_leader_info()
if info and info.redis_url:
print(f"[LeaderStorage] Background: Found leader at {info.redis_url}")
self._connect_to_redis(info.redis_url)
if info and info.api_url:
print(f"[LeaderStorage] Background: Found leader at {info.api_url}")
self._connect_via_api(info)
if self._connected:
# Configure WebDAV client
@@ -228,9 +221,18 @@ class LeaderStorage(StorageProvider):
self._disconnect_redis()
def is_connected(self) -> bool:
"""Check if connected to the storage backend."""
"""Check if connected to the storage backend (Redis or HTTP API)."""
with self._lock:
if not self._connected or not self._client:
if not self._connected:
return False
# HTTP-only mode: probe meta-core's /health.
if self._api_client is not None and self._client is None:
if self._api_client.health():
return True
self._connected = False
return False
# Legacy Redis mode.
if not self._client:
return False
try:
self._client.ping()
@@ -255,9 +257,9 @@ class LeaderStorage(StorageProvider):
try:
info = self._leader_client.get_leader_info()
if info and info.redis_url:
print(f"[LeaderStorage] Reconnecting to leader at {info.redis_url}...")
self._connect_to_redis(info.redis_url)
if info and info.api_url:
print(f"[LeaderStorage] Reconnecting to leader at {info.api_url}...")
self._connect_via_api(info)
if self._connected:
# Reconfigure WebDAV client with new leader's internal WebDAV URL
@@ -274,8 +276,46 @@ class LeaderStorage(StorageProvider):
print(f"[LeaderStorage] Failed to reconnect after {max_attempts} attempts")
def _connect_via_api(self, info: LeaderLockInfo) -> None:
"""Connect to meta-core via its HTTP API. The api_client covers every
read this storage provider does — Redis access is no longer required
(api-mediated-access PR D). Also starts the SSE meta-events consumer
so cache invalidation still flows in real time.
"""
with self._lock:
try:
# Drop any prior state (Redis or otherwise)
self._disconnect_redis_unlocked()
# Set up the HTTP read client + probe meta-core.
api = MetaCoreApiClient(api_url=info.api_url)
if not api.health():
raise RuntimeError(f"meta-core health probe failed at {info.api_url}")
self._api_client = api
self._connected = True
print(f"[LeaderStorage] Connected via meta-core API at {info.api_url}")
# Start SSE consumer for cache invalidation.
try:
self._meta_consumer = MetaConsumer(api_url=info.api_url)
self._meta_consumer.on_change(self._on_metadata_change)
self._meta_consumer.start()
except Exception as e:
print(f"[LeaderStorage] Warning: failed to start meta consumer: {e}")
self._cache_dirty = True
self._notify_ready()
except Exception as e:
print(f"[LeaderStorage] Failed to connect via API: {e}")
self._api_client = None
self._connected = False
def _connect_to_redis(self, url: str) -> None:
"""Connect to Redis."""
"""Connect to Redis. Legacy path — kept for direct REDIS_URL overrides
and ad-hoc tests. Production discovery goes through _connect_via_api.
"""
with self._lock:
try:
# Disconnect existing client
@@ -321,7 +361,7 @@ class LeaderStorage(StorageProvider):
self._disconnect_redis_unlocked()
def _disconnect_redis_unlocked(self) -> None:
"""Disconnect from Redis (without lock)."""
"""Disconnect from Redis and/or API (without lock)."""
# Stop meta consumer
if self._meta_consumer:
try:
@@ -336,6 +376,10 @@ class LeaderStorage(StorageProvider):
except Exception:
pass
self._client = None
# The API client has no socket; just drop the reference.
self._api_client = None
self._connected = False
# Clear cache
@@ -399,10 +443,14 @@ class LeaderStorage(StorageProvider):
return f"{self._prefix}file:{hash_id}/"
def _get_file_metadata(self, hash_id: str) -> dict:
"""Get all metadata for a file using flat key scan."""
"""Get all metadata for a file. Uses meta-core HTTP API when in
HTTP-only mode; falls back to Redis SCAN+GET otherwise."""
if self._api_client is not None:
flat = self._api_client.get_metadata_flat(hash_id)
return dict(flat) if flat else {}
prefix = self._get_file_key_prefix(hash_id)
data = {}
for key in self._client.scan_iter(f"{prefix}*", count=100):
key_str = key if isinstance(key, str) else key.decode()
field = key_str[len(prefix):]
@@ -599,6 +647,15 @@ class LeaderStorage(StorageProvider):
# StorageProvider Interface
# ========================================================================
def _list_hash_ids(self) -> List[str]:
"""Enumerate every hashId. HTTP API in api-mediated mode, SMEMBERS
on file:__index__ in legacy Redis mode."""
if self._api_client is not None:
return self._api_client.list_hash_ids()
index_key = f"{self._prefix}file:__index__"
ids = self._client.smembers(index_key)
return [h if isinstance(h, str) else h.decode() for h in ids]
def get_all_videos(self) -> List[VideoMetadata]:
"""Get all videos from storage."""
if not self.is_connected():
@@ -608,8 +665,7 @@ class LeaderStorage(StorageProvider):
try:
# Get all hash IDs from index
index_key = f"{self._prefix}file:__index__"
hash_ids = self._client.smembers(index_key)
hash_ids = self._list_hash_ids()
for hash_id in hash_ids:
hash_id_str = hash_id if isinstance(hash_id, str) else hash_id.decode()
@@ -670,31 +726,39 @@ class LeaderStorage(StorageProvider):
try:
# Get from index and check each file's imdbid
index_key = f"{self._prefix}file:__index__"
for hash_id in self._client.smembers(index_key):
hash_id_str = hash_id if isinstance(hash_id, str) else hash_id.decode()
prefix = self._get_file_key_prefix(hash_id_str)
# Check imdbid field directly
stored_imdb = self._client.get(f"{prefix}imdbid") or self._client.get(f"{prefix}imdbId")
if stored_imdb:
stored_imdb_str = stored_imdb if isinstance(stored_imdb, str) else stored_imdb.decode()
if stored_imdb_str.lower() == imdb_id:
data = self._get_file_metadata(hash_id_str)
return self._parse_video(hash_id_str, data)
for hash_id_str in self._list_hash_ids():
stored_imdb = self._get_imdb_field(hash_id_str)
if stored_imdb and stored_imdb.lower() == imdb_id:
data = self._get_file_metadata(hash_id_str)
return self._parse_video(hash_id_str, data)
except Exception as e:
print(f"[LeaderStorage] Error finding video by IMDB ID {imdb_id}: {e}")
return None
def _get_imdb_field(self, hash_id: str) -> Optional[str]:
"""Read imdbid (or imdbId) for a file. HTTP API when in api-mediated
mode, direct Redis GET otherwise."""
if self._api_client is not None:
v = self._api_client.get_property(hash_id, "imdbid")
if v:
return v
return self._api_client.get_property(hash_id, "imdbId")
prefix = self._get_file_key_prefix(hash_id)
stored_imdb = self._client.get(f"{prefix}imdbid") or self._client.get(f"{prefix}imdbId")
if not stored_imdb:
return None
return stored_imdb if isinstance(stored_imdb, str) else stored_imdb.decode()
def get_file_path_by_cid(self, cid: str) -> Optional[str]:
"""
Get the file path for a file by its CID.
Looks up file:{cid}/* in Redis and returns the path.
Tries 'path' field first, then falls back to 'filePath'.
This works for any file including poster images.
In api-mediated mode this hits meta-core's GET /api/file/{cid}/info
which already does the reverse-index lookup + canonical-path
resolution. In Redis mode we fall back to reading file:{cid}/path
and file:{cid}/filePath directly.
Returns the relative path if found, None otherwise.
"""
@@ -702,6 +766,11 @@ class LeaderStorage(StorageProvider):
return None
try:
if self._api_client is not None:
# /api/file/{cid}/info already returns a path relative to
# FILES_PATH, normalised. No further trimming required.
return self._api_client.resolve_file_path(cid)
prefix = self._get_file_key_prefix(cid)
# Try 'path' field first (relative path)
path = self._client.get(f"{prefix}path")
@@ -726,6 +795,8 @@ class LeaderStorage(StorageProvider):
return 0
try:
if self._api_client is not None:
return self._api_client.count_hash_ids()
# Use index set for accurate count
index_key = f"{self._prefix}file:__index__"
return self._client.scard(index_key)
@@ -762,8 +833,11 @@ class LeaderStorage(StorageProvider):
if status['connected']:
try:
status['video_count'] = self.get_video_count()
info = self._client.info('memory')
status['memory_used'] = info.get('used_memory_human', 'N/A')
# Memory stats only available in Redis-direct mode; the HTTP
# API doesn't surface them. Skip cleanly when unavailable.
if self._client is not None:
mem = self._client.info('memory')
status['memory_used'] = mem.get('used_memory_human', 'N/A')
except Exception:
pass
src/storage/meta_core_api_client.py
+120
View File
@@ -0,0 +1,120 @@
"""
MetaCoreApiClient — HTTP read client against meta-core.
Mirrors the subset of meta-stremio's storage-read surface that previously
spoke ioredis directly. Backs the LeaderStorage when meta-core's leader info
omits redis_url (post api-mediated-access PR D), so the addon keeps working
without direct Redis access.
Endpoints used:
- GET /meta → enumerate hashIds
- GET /meta/{hash} → flat metadata map for one file
- GET /meta/{hash}/{key} → single property (text/plain)
- GET /api/file/{cid}/info → resolve CID to file path
- GET /health → connectivity probe
"""
from __future__ import annotations
import json
import logging
from typing import Optional, List, Dict
from urllib.parse import quote
from urllib.request import Request, urlopen
from urllib.error import HTTPError, URLError
logger = logging.getLogger(__name__)
class MetaCoreApiClient:
"""HTTP client for meta-core's /meta/* read surface."""
def __init__(self, api_url: str, timeout: int = 30):
self._api_url = api_url.rstrip('/')
self._timeout = timeout
# ----- enumeration ---------------------------------------------------
def list_hash_ids(self) -> List[str]:
"""GET /meta — returns every hashId in file:__index__."""
body = self._get_json('/meta')
if not body:
return []
ids = body.get('hashIds')
return ids if isinstance(ids, list) else []
def count_hash_ids(self) -> int:
body = self._get_json('/meta')
if not body:
return 0
return int(body.get('count', 0))
# ----- per-file -----------------------------------------------------
def get_metadata_flat(self, hash_id: str) -> Optional[Dict[str, str]]:
"""GET /meta/{hash}. Returns the flat property map or None on 404."""
body = self._get_json(f'/meta/{quote(hash_id, safe="")}', allow_404=True)
if not body:
return None
meta = body.get('metadata')
return meta if isinstance(meta, dict) else None
def get_property(self, hash_id: str, key: str) -> Optional[str]:
"""GET /meta/{hash}/{key}. meta-core returns text/plain (or 404)."""
path = f'/meta/{quote(hash_id, safe="")}/{quote(key, safe="/")}'
url = f"{self._api_url}{path}"
try:
req = Request(url, headers={'Accept': 'text/plain'})
with urlopen(req, timeout=self._timeout) as resp:
text = resp.read().decode('utf-8', errors='replace')
return text if text != '' else None
except HTTPError as e:
if e.code == 404:
return None
logger.warning("[MetaCoreApiClient] GET %s → %s", path, e)
return None
except URLError as e:
logger.warning("[MetaCoreApiClient] GET %s failed: %s", path, e)
return None
# ----- CID resolution -----------------------------------------------
def resolve_file_path(self, cid: str) -> Optional[str]:
"""GET /api/file/{cid}/info → relative path (or None on miss)."""
path = f'/api/file/{quote(cid, safe="")}/info'
body = self._get_json(path)
if not body or not body.get('exists'):
return None
return body.get('filePath') or None
# ----- health -------------------------------------------------------
def health(self) -> bool:
try:
req = Request(f"{self._api_url}/health", headers={'Accept': 'application/json'})
with urlopen(req, timeout=min(self._timeout, 5)) as resp:
return resp.status == 200
except Exception:
return False
# ----- internals ----------------------------------------------------
def _get_json(self, path: str, allow_404: bool = False) -> Optional[dict]:
url = f"{self._api_url}{path}"
try:
req = Request(url, headers={'Accept': 'application/json'})
with urlopen(req, timeout=self._timeout) as resp:
raw = resp.read().decode('utf-8', errors='replace')
if not raw:
return None
return json.loads(raw)
except HTTPError as e:
if allow_404 and e.code == 404:
return None
logger.warning("[MetaCoreApiClient] GET %s → %s", path, e)
return None
except URLError as e:
logger.warning("[MetaCoreApiClient] GET %s failed: %s", path, e)
return None
except json.JSONDecodeError as e:
logger.warning("[MetaCoreApiClient] GET %s — invalid JSON: %s", path, e)
return None