5 Commits
Author SHA1 Message Date
Worph ba6e152b4f align with metadata standard
Build and Push Docker Image / build (push) Failing after 13s
2026-09-13 15:56:46 +02:00
Worph c1b8a2e731 gitattributes 2026-07-14 16:08:01 +02:00
Pierre 6cf93b72a7 improve doc + bugfix 2026-05-25 23:48:20 +02:00
Pierre 37223c442b adaptation due to meta-core new API 2026-05-24 17:44:12 +02:00
Pierre 56eced073a adaptation related to meta-core api changes 2026-05-24 17:26:29 +02:00
6 changed files with 608 additions and 299 deletions
+6
View File
@@ -0,0 +1,6 @@
# FIX CRLF always when developer has not set
# Linux/Mac: git config --global core.autocrlf input
# Windows: git config --global core.autocrlf true
# Auto detect text files and perform LF normalization
* text=auto
* eol=lf
+148 -175
View File
@@ -1,16 +1,11 @@
# Meta-Stremio
A standalone Stremio addon service for streaming personal media with on-the-fly HLS transcoding. Part of the MetaMesh decentralized media ecosystem.
Stremio addon for the MetaMesh stack. Reads file metadata from the shared KV store, reads files via WebDAV (or direct mount), and serves HLS with on-the-fly FFmpeg transcoding plus the standard Stremio addon protocol.
External port in the dev stack: **8182** (auth-gated, via Caddy + nginx-hash-lock). Internal container port is `7000`. Debug-direct port (bypasses the perimeter): **18182**. Container name: `metastremio-app` (backend) / `metastremio` (hash-lock proxy).
## Overview
Meta-Stremio is a **process-read** service that:
- Reads file metadata from the shared KV store (Redis)
- Reads media files via WebDAV (from meta-sort) or direct volume access
- Provides HLS streaming with adaptive real-time FFmpeg transcoding
- Serves as a Stremio addon for seamless playback
- Supports SMB/rclone mounts via meta-sort's WebDAV endpoint
```
┌─────────────────────────────────────────────────────────────────────┐
│ MetaMesh Ecosystem │
@@ -24,245 +19,223 @@ Meta-Stremio is a **process-read** service that:
│ ▼ ▼ ▼ │
│ ┌─────────────────────────────────────────────────────┐ │
│ │ Shared KV Store (Redis) │ │
│ │ Leader election via flock on shared filesystem │ │
│ │ DB stored in: DATA/Apps/meta-core/DB │ │
│ │ meta-core is leader, exposes /urls + /meta APIs │ │
│ └─────────────────────────────────────────────────────┘ │
│ │ │
│ ▼ │
│ ┌─────────────────────────────────────────────────────┐ │
│ │ Shared DATA Volume (media files) │ │
│ │ meta-sort writes → meta-stremio/meta-fuse reads │ │
│ │ File access via WebDAV (or direct mount) │ │
│ └─────────────────────────────────────────────────────┘ │
│ │
└─────────────────────────────────────────────────────────────────────┘
```
## Features
### High Performance
- **Adaptive Quality** - Dynamic preset + CRF adjustment targeting 60-80% transcode ratio
- **Intelligent Prefetch** - 4 segments ahead for smooth playback
- **Single-threaded Encode** - 100% CPU per segment for fastest encoding
- **Muxed Segments** - Video+audio in one FFmpeg call
- **Segment Caching** - Persistent cache for repeated access
### Stremio Integration
- **Catalog Endpoint** - Browse media library by type (movies, series)
- **Meta Endpoint** - Detailed video information with codec details
- **Stream Endpoint** - Multiple stream options per video
- **Multi-Audio** - Separate HLS streams per audio track
- **Subtitles** - Extract and serve embedded subtitles as VTT
- **Language Configuration** - Configure display language (uses metadata stored by meta-sort's TMDB plugin)
### Storage Abstraction
- **Leader Storage** - Auto-discovers meta-sort's Redis via leader election (recommended)
- **Redis Storage** - Direct Redis connection via URL
## Project Structure
```
meta-stremio/
├── README.md
├── CLAUDE.md # Development guide
├── Dockerfile
├── docker-compose.yml
├── requirements.txt # redis + requests
├── src/
│ ├── server.py # HTTP server + routing
│ ├── stremio.py # Stremio handlers + storage integration
│ ├── transcoder.py # HLS transcoding engine
│ ├── fileserver.py # File serving (direct or WebDAV)
│ ├── webdav_client.py # WebDAV client for meta-sort file access
│ └── storage/
│ ├── __init__.py
│ ├── provider.py # Abstract StorageProvider
│ ├── redis_storage.py # Direct Redis connection
│ ├── leader_storage.py # Leader-aware Redis connection
│ └── leader_discovery.py # Leader election discovery
└── www/
└── index.html # Dashboard
```
- **HLS transcoding** — adaptive preset+CRF targeting a 60–80% transcode ratio, prefetch of N segments ahead, persistent segment cache.
- **Stremio protocol** — manifest, catalog, meta, stream; one HLS stream per audio track; subtitles extracted to VTT.
- **Direct file serving** — range-request capable, for clients that can play the source directly.
- **Storage abstractions** — `LeaderStorage` (auto-discovers Redis via meta-core), `RedisStorage` (direct URL), `direct` mode (reads from meta-core's HTTP API + SSE meta-stream).
- **Path-token gate for addon URLs** — when `HASH_API_SEED` is set, addon paths must be prefixed with `/s/<token>`. Dashboard/browser paths are protected separately by the upstream hash-lock + OIDC sidecar (Stremio clients can't carry cookies, so they get the path-token instead).
- **Language-configurable manifest** — `displayLanguage` in the per-install config (base64-in-path) selects which `titles` translation meta-sort's TMDB plugin populated.
## Installation
### Docker (Recommended)
### Docker (in the MetaMesh dev stack)
```bash
# Build
docker build -t meta-stremio .
cd dev
./scripts/start.sh
```
# Run with leader discovery (recommended - auto-discovers meta-sort's Redis)
This brings up `metastremio-app` on port 18182 (debug-direct) and the auth-gated entry on `https://metastremio-dev.localhost:8182`.
### Docker (standalone)
```bash
docker build -t meta-stremio packages/meta-stremio
# Leader mode (auto-discover Redis via meta-core)
docker run -d \
-p 7000:7000 \
-p 8182:7000 \
-v /path/to/media:/files:ro \
-v /path/to/meta-core:/meta-core \
meta-stremio
# Run with direct Redis connection
# Direct Redis mode
docker run -d \
-p 7000:7000 \
-p 8182:7000 \
-v /path/to/media:/files:ro \
-e STORAGE_MODE=redis \
-e REDIS_URL=redis://your-redis:6379 \
meta-stremio
```
### Docker Compose
The container always listens on port `7000` internally — map it to whatever you want externally.
### Standalone (development)
```bash
# Run as part of the MetaMesh stack (uses leader discovery)
cd dev
./start.sh
```
### Standalone (Development)
```bash
# Prerequisites: Python 3.9+, FFmpeg
# Install dependencies
pip install -r requirements.txt
# Run
# Requires Python 3.9+ and FFmpeg.
pip install -r requirements.txt # redis, watchdog, Pillow, requests
cd src
python server.py
```
## Configuration
### Environment Variables
### Environment variables
| Variable | Default | Description |
|----------|---------|-------------|
| `PORT` | `7000` | Server listen port |
| `MEDIA_DIR` | `/data/media` | Media files directory |
| `CACHE_DIR` | `/data/cache` | Transcoded segment cache |
| `STORAGE_MODE` | `leader` | `leader` or `redis` |
| `META_CORE_PATH` | `/meta-core` | Path to meta-core shared volume (for leader discovery) |
| `FILES_PATH` | `/files` | Path to files volume |
| `REDIS_URL` | `redis://localhost:6379` | Redis connection URL (for `redis` mode) |
| `REDIS_PREFIX` | `meta-sort:` | Redis key prefix |
| `SEGMENT_DURATION` | `4` | HLS segment length (seconds) |
| `PREFETCH_SEGMENTS` | `4` | Segments to prefetch ahead |
| `SCHEME` | `auto` | URL scheme (`http`, `https`, `auto`) |
| `META_CORE_WEBDAV_URL` | - | WebDAV URL for file access (e.g., `http://meta-core/webdav`) |
|---|---|---|
| `PORT` | `7000` | Internal HTTP listen port (the dev stack maps 18182 / 8182 to this). |
| `BASE_URL` | — | External URL emitted in manifests / poster URLs (e.g. `https://metastremio-dev.localhost:8182`). |
| `MEDIA_DIR` | `/files/watch` | Media directory (consumed by `transcoder.py`). |
| `CACHE_DIR` | `/data/cache` | Transcoded segment cache. |
| `STORAGE_MODE` | `leader` | `leader`, `redis`, or `direct`. |
| `META_CORE_PATH` | `/meta-core` | Shared meta-core volume; used by `LeaderClient` for `kv-leader.info`. |
| `META_CORE_URL` | — | meta-core HTTP API (e.g. `http://metacore-app:9000`); required for `STORAGE_MODE=direct` since there is no leader file to read. |
| `FILES_PATH` | `/files` | Files volume. |
| `REDIS_URL` | `redis://localhost:6379` | Redis URL for `STORAGE_MODE=redis`. |
| `REDIS_PREFIX` | `meta-sort:` | Redis key prefix. |
| `SCHEME` | `auto` | URL scheme for generated URLs: `http`, `https`, or `auto`. |
| `SEGMENT_DURATION` | `4` | HLS segment length (s). |
| `PREFETCH_SEGMENTS` | `4` | Segments to prefetch ahead. |
| `META_CORE_WEBDAV_URL` | — | When set, files are read over HTTP from meta-core's WebDAV instead of the filesystem (lets meta-stremio reach SMB / rclone mounts that only exist inside meta-core). |
| `HASH_API_SEED` | — | When set, derives a 16-char path-token; all addon routes must be prefixed with `/s/<token>`. Dashboard routes bypass. |
**WebDAV File Access**: When `META_CORE_WEBDAV_URL` is set, meta-stremio reads files via HTTP from meta-core's WebDAV server instead of direct filesystem access. This enables access to SMB/rclone mounts that are dynamically mounted inside meta-core.
## API endpoints
## API Endpoints
All routes below are matched in `src/server.py`. When `HASH_API_SEED` is set, **every addon path is prefixed with `/s/<token>`**; the dashboard paths in the first table are *not* prefixed (they're protected by hash-lock/OIDC upstream).
### Dashboard / browser
### Dashboard
| Endpoint | Method | Description |
|----------|--------|-------------|
| `/` | GET | Setup dashboard |
| `/health` | GET | Health check with storage status |
### Dashboard API
| Endpoint | Method | Description |
|----------|--------|-------------|
|---|---|---|
| `/` , `/index.html` | GET | Setup dashboard |
| `/configure` | GET | Per-install language configuration page |
| `/health` | GET | Liveness + storage status |
| `/api/stats` | GET | Library statistics |
| `/api/library` | GET | Full library list |
| `/api/library` | GET | Full library list (videos + count) |
| `/api/services` | GET | Discovered MetaMesh services (proxied from `meta-core /services`) |
| `/api/languages` | GET | Languages selectable from the `/configure` page |
### Stremio addon protocol
`/manifest.json` and `/stremio/manifest.json` are aliases; same for catalog/meta/stream. Stremio per-install configuration is encoded as URL-safe base64 JSON between the `/s/<token>` prefix and the addon path (e.g. `/s/<token>/<configB64>/manifest.json`).
### Stremio Protocol
| Endpoint | Method | Description |
|----------|--------|-------------|
| `/manifest.json` | GET | Addon manifest |
| `/catalog/:type/:id.json` | GET | Video catalog |
|---|---|---|
| `/manifest.json` | GET, HEAD | Addon manifest |
| `/stremio/manifest.json` | GET, HEAD | Alias of `/manifest.json` |
| `/catalog/:type/:id.json` | GET | Catalog (with optional `…/:extra.json`) |
| `/stremio/catalog/:type/:id.json` | GET | Alias |
| `/meta/:type/:id.json` | GET | Video metadata |
| `/stremio/meta/:type/:id.json` | GET | Alias |
| `/stream/:type/:id.json` | GET | Stream URLs |
| `/stremio/stream/:type/:id.json` | GET | Alias |
### File serving
### Transcoding
| Endpoint | Method | Description |
|----------|--------|-------------|
| `/transcode/:path/master.m3u8` | GET | ABR master playlist |
| `/transcode/:path/master_original_a0.m3u8` | GET | Original quality playlist |
| `/transcode/:path/stream_a0_720p.m3u8` | GET | Quality stream playlist |
| `/transcode/:path/seg_a0_720p_0.ts` | GET | Video segment |
| `/transcode/:path/subtitle_0.vtt` | GET | Subtitle track |
| `/direct/:path` | GET | Direct file serving |
| `/transcode/metrics` | GET | Transcoding metrics |
|---|---|---|
| `/file/{cid}` | GET, HEAD | Serve file by CID |
| `/file/{cid}/w{width}` | GET, HEAD | Serve resized image by CID (uses Pillow) |
| `/poster/{cid}` | GET, HEAD | Legacy alias of `/file/{cid}` |
| `/poster/{cid}/w{width}` | GET, HEAD | Legacy alias of `/file/{cid}/w{width}` |
| `/direct/{path}` | GET, HEAD | Direct file serving with HTTP `Range` support |
## KV Store Schema
### Transcoder
Meta-stremio reads from meta-sort's Redis using this key format:
| Endpoint | Method | Description |
|---|---|---|
| `/transcode/{path}/master.m3u8` | GET, HEAD | ABR master playlist |
| `/transcode/{path}/master_{resolution}.m3u8` | GET, HEAD | Master playlist for a single resolution |
| `/transcode/{path}/master_a{audio}.m3u8` | GET, HEAD | Master playlist for a single audio track |
| `/transcode/{path}/master_{resolution}_a{audio}.m3u8` | GET, HEAD | Combined variant |
| `/transcode/{path}/stream_a{audio}_{resolution}.m3u8` | GET, HEAD | Per-variant stream playlist |
| `/transcode/{path}/seg_a{audio}_{resolution}_{n}.ts` | GET, HEAD | Video segment |
| `/transcode/{path}/subtitle_{idx}.m3u8` | GET, HEAD | Subtitle playlist |
| `/transcode/{path}/subtitle_{idx}.vtt` | GET, HEAD | Subtitle VTT |
| `/transcode/metrics` | GET | Transcoder metrics (includes `total_files`) |
| `/transcode/reset-metrics` | POST | Reset transcoder metrics |
## Project structure
```
meta-sort:file:{hashId} → Hash containing all metadata
- filePath → /media/movies/Movie.mkv
- title → Movie Title
- type → movie|series|anime
- year → 2024
- season → 1 (for series)
- episode → 1 (for series)
- duration → 7200 (seconds)
- width → 1920
- height → 1080
- videoCodec → h264
- audioCodec → aac
- audioTracks → JSON string
- subtitles → JSON string
- imdbId → tt1234567
- poster → URL
meta-stremio/
├── README.md
├── Dockerfile
├── docker-compose.yml
├── docker/ # container entry scripts
├── requirements.txt # redis, watchdog, Pillow, requests
├── src/
│ ├── server.py # HTTP server + routing (entry point)
│ ├── stremio.py # Stremio handlers + library/language helpers
│ ├── transcoder.py # HLS transcoding engine + metrics + segment/subtitle managers
│ ├── fileserver.py # serve_file(cid, width) — file/poster endpoints
│ ├── poster.py # poster URL/CID helpers
│ ├── webdav_client.py # WebDAV client (file_exists, stream_file, stream_range, get_file_size)
│ └── storage/
│ ├── __init__.py
│ ├── provider.py # StorageProvider + VideoMetadata abstract base
│ ├── redis_storage.py # Direct Redis backend
│ ├── leader_storage.py # Leader-aware Redis backend
│ ├── leader_client.py # Reads /meta-core/locks/kv-leader.info + calls meta-core /urls
│ ├── meta_consumer.py # SSE consumer from meta-core /meta stream
│ ├── meta_core_api_client.py # HTTP client for meta-core REST API
│ └── service_registration.py # Registers meta-stremio with meta-core /services
└── www/
├── index.html # Dashboard
└── configure.html # Language configuration page
```
## Stream Types
## KV store schema
Each video provides multiple stream options:
meta-stremio reads from the per-file Redis hash `/file/{cid}` populated by meta-sort and its plugins. Field semantics (`videoType`, `originalTitle`, `titles`, `season`, `episode`, `movieYear`, `cid_*`, etc.) are documented in the repo-root [`METADATA_KEYS.md`](../../METADATA_KEYS.md) — that is the single source of truth, and matches the `@metazla/meta-interface` types.
1. **Direct File** - Original file, no transcoding
- Best quality, may not play on all devices
## Stream types
2. **HLS Original** - Transcoded at source resolution
- H.264/AAC in HLS container
- One stream per audio track
For each video, the addon advertises multiple streams:
3. **HLS ABR** - Adaptive bitrate with quality ladder
- Multiple resolutions (1080p, 720p, 480p, 360p)
- Automatic quality switching
1. **Direct file** — original bytes, served via `/direct/...` with `Range` support. Best quality, may not play on every device.
2. **HLS original** — transcoded at source resolution, H.264/AAC, one stream per audio track.
3. **HLS ABR** — adaptive ladder. Quality presets are baked into `transcoder.py`.
## Transcoding Details
### Adaptive Quality System
The transcoder automatically adjusts encoding settings to maintain a 60-80% transcode ratio (segment encode time / segment duration):
- **Too Fast (<60%)**: Increase quality (lower CRF, slower preset)
- **Too Slow (>80%)**: Decrease quality (higher CRF, faster preset)
### Quality Ladder
| Resolution | Video Bitrate | Audio Bitrate |
|------------|---------------|---------------|
| Resolution | Video bitrate | Audio bitrate |
|---|---|---|
| 1080p | 5000 kbps | 192 kbps |
| 720p | 3000 kbps | 128 kbps |
| 480p | 1500 kbps | 128 kbps |
| 360p | 800 kbps | 96 kbps |
| 720p | 3000 kbps | 128 kbps |
| 480p | 1500 kbps | 128 kbps |
| 360p | 800 kbps | 96 kbps |
## Adding to Stremio
1. Open the dashboard at `http://localhost:7000`
2. Click "Install in Stremio" or copy the manifest URL
3. In Stremio: Settings → Addons → Enter URL → Install
1. Open the dashboard at `https://metastremio-dev.localhost:8182/` (or debug-direct `http://localhost:18182/`).
2. Pick your display language on `/configure` if you want non-English titles.
3. Use the **Install in Stremio** button (uses the `stremio://` protocol handler) or copy the manifest URL into Stremio: Settings → Addons → Add Addon.
## Development
## Debugging
```bash
# Run in development mode (requires meta-sort running with Redis)
cd src
python server.py
# Container logs
docker compose -f dev/docker-compose.yml logs -f metastremio-app
# Or with direct Redis URL
STORAGE_MODE=redis REDIS_URL=redis://localhost:6379 python server.py
# Live health
curl -k https://metastremio-dev.localhost:8182/health
curl http://localhost:18182/health # debug-direct (no auth, no TLS)
# View logs
tail -f /data/cache/*.log
# Transcoder metrics
curl http://localhost:18182/transcode/metrics
curl -X POST http://localhost:18182/transcode/reset-metrics
```
## Related Projects
Container lifecycle: do **not** `docker restart metastremio-app` to apply config — use the dev-stack reload script (`./scripts/reload-stremio.sh`) so supervisord restarts the right process inside the container.
- **[meta-sort](../meta-sort)** - File indexer (writes metadata to Redis)
- **[meta-fuse](../meta-fuse)** - Virtual filesystem (reads metadata)
- **[meta-orbit](../meta-orbit)** - P2P metadata sharing
## Related projects
- **[meta-sort](../meta-sort)** — file indexer; writes the metadata meta-stremio reads.
- **[meta-fuse](../meta-fuse)** — virtual filesystem; reads the same KV store.
- **[meta-share](../meta-share)** — decentralised metadata sharing.
## License
+15
View File
@@ -137,6 +137,21 @@ class LeaderClient:
data = json.loads(response.read().decode())
# api-mediated-access PR D: redisUrl is no longer published by
# meta-core. If it shows up, an old build snuck back in —
# 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'):
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', ''),
+149 -42
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,9 +124,14 @@ class LeaderStorage(StorageProvider):
# Try to get leader info (short timeout per attempt)
info = self._leader_client.get_leader_info()
if info and info.redis_url:
print(f"[LeaderStorage] Found leader at {info.redis_url} (attempt {attempt})")
self._connect_to_redis(info.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)
@@ -140,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})")
@@ -174,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
@@ -214,9 +221,30 @@ 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).
HTTP-only mode auto-recovers from transient meta-core outages: if
the health probe fails we drop the connected flag but keep the
api_client so the *next* call probes again. The background
reconnect thread is only needed for Redis-mode where the
connection itself is sticky.
"""
with self._lock:
if not self._connected or not self._client:
# HTTP-only mode: probe meta-core's /health on every call. This
# is cheap (a single HTTP GET) and naturally recovers when
# meta-core comes back without needing a separate reconnect
# loop. The api_client object stays valid through transient
# failures — it's stateless aside from the configured URL.
if self._api_client is not None and self._client is None:
if self._api_client.health():
self._connected = True
return True
self._connected = False
return False
# Legacy Redis mode.
if not self._connected:
return False
if not self._client:
return False
try:
self._client.ping()
@@ -241,9 +269,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
@@ -260,8 +288,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
@@ -274,11 +340,19 @@ class LeaderStorage(StorageProvider):
print(f"[LeaderStorage] Connected to Redis at {url}")
# Start meta events consumer for real-time updates
# Start meta events consumer (HTTP SSE against meta-core).
# Prefer the api URL we discovered via the LeaderClient; fall
# back to META_CORE_URL env so STORAGE_MODE=direct deployments
# (no leader discovery) still get live updates.
try:
self._meta_consumer = MetaConsumer(self._client, self._prefix)
self._meta_consumer.on_change(self._on_metadata_change)
self._meta_consumer.start()
info = self._leader_client.get_cached_leader_info() if self._leader_client else None
api_url = (info.api_url if info else None) or os.environ.get("META_CORE_URL")
if api_url:
self._meta_consumer = MetaConsumer(api_url=api_url)
self._meta_consumer.on_change(self._on_metadata_change)
self._meta_consumer.start()
else:
print("[LeaderStorage] Warning: api_url unavailable; meta consumer not started")
except Exception as e:
print(f"[LeaderStorage] Warning: failed to start meta consumer: {e}")
@@ -299,7 +373,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:
@@ -314,6 +388,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
@@ -377,10 +455,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):]
@@ -494,10 +576,9 @@ class LeaderStorage(StorageProvider):
# Parse release date
release_date = data.get('releasedate', data.get('releaseDate'))
# Parse episode title - try episodeTitle or extract from titles/eng
# Parse episode title. `titles/*` is the work's name set, never an
# episode title (METADATA_KEYS.md §3).
episode_title = data.get('episodeTitle', data.get('episodeName'))
if not episode_title:
episode_title = data.get('titles/eng', data.get('titles/en'))
# Parse tagline
tagline = data.get('tagline')
@@ -577,6 +658,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():
@@ -586,8 +676,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()
@@ -648,31 +737,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.
"""
@@ -680,6 +777,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")
@@ -704,6 +806,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)
@@ -740,8 +844,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
+170 -82
View File
@@ -1,153 +1,241 @@
"""
Meta Events Consumer for meta-stremio.
Consumes the meta:events Redis stream published by meta-core.
Provides real-time notifications when metadata changes.
Consumes the meta:events stream published by meta-core. Originally used a
Redis Streams XREADGROUP loop; migrated to meta-core's HTTP SSE endpoint at
/api/events/meta as part of the api-mediated-access lockdown (see
meta-core docs/api-mediated-access.md, PR C).
The callback signature is unchanged so leader_storage.py needs no changes:
each event is delivered as (key, event_type) where key has the form
"file:<hashId>/<field>" and event_type is "set" / "del" / "expire".
"""
from __future__ import annotations
import threading
import json
import logging
import os
import threading
import time
from pathlib import Path
from typing import Callable, List, Optional
from urllib.parse import urlparse
try:
from redis import Redis
REDIS_AVAILABLE = True
except ImportError:
REDIS_AVAILABLE = False
Redis = None
import urllib.request
import urllib.error
logger = logging.getLogger(__name__)
class MetaConsumer:
"""
Consumes meta:events stream for real-time metadata updates.
Streams meta:events from meta-core over SSE and dispatches a filtered
subset of changes to registered callbacks.
Uses Redis consumer groups for reliable delivery.
Filters events to only notify about interesting field changes.
Cursor durability: the last-seen event id is persisted to a file (default
/meta-core/cursors/meta-stremio-meta.cursor) so the consumer resumes
cleanly after a restart. On a gap event the cursor is reset to the
server-supplied resumeFrom value.
"""
STREAM = "meta:events"
GROUP = "stremio-consumer"
CONSUMER = "stremio-1"
STREAM_PATH = "/api/events/meta"
# Fields that trigger cache invalidation
# Fields whose changes invalidate cached Stremio responses.
INTERESTING_FIELDS = [
"tmdb", "tmdbId", "title", "poster", "backdrop",
"imdbId", "imdbid", "year", "type", "rating",
"description", "plot", "genres", "fileType"
"description", "plot", "genres", "fileType",
]
def __init__(self, redis_client: Redis, prefix: str = ""):
if not REDIS_AVAILABLE:
raise ImportError("redis package not installed")
self._client = redis_client
self._prefix = prefix
def __init__(
self,
api_url: str,
cursor_path: str = "/meta-core/cursors/meta-stremio-meta.cursor",
request_timeout: int = 35,
) -> None:
self._api_url = api_url.rstrip("/")
self._cursor_path = cursor_path
self._request_timeout = request_timeout
self._running = False
self._thread: Optional[threading.Thread] = None
self._callbacks: List[Callable[[str, str], None]] = []
self._cursor: str = self._load_cursor()
self._stop_event = threading.Event()
def on_change(self, callback: Callable[[str, str], None]) -> 'MetaConsumer':
"""
Register callback for metadata changes.
callback(key, event_type): Called when a file's metadata changes.
- key: The Redis key that changed (e.g., "file:abc123/tmdb")
- event_type: The operation type ("set", "del", etc.)
"""
def on_change(self, callback: Callable[[str, str], None]) -> "MetaConsumer":
"""Register a callback. See module docstring for the contract."""
self._callbacks.append(callback)
return self
def start(self) -> None:
"""Start consuming events."""
if self._running:
return
# Create consumer group if not exists
try:
self._client.xgroup_create(
self.STREAM, self.GROUP, id="$", mkstream=True
)
logger.info(f"[MetaConsumer] Created consumer group '{self.GROUP}'")
except Exception as e:
if "BUSYGROUP" not in str(e):
logger.warning(f"[MetaConsumer] Group creation error: {e}")
self._running = True
self._stop_event.clear()
self._thread = threading.Thread(target=self._consume_loop, daemon=True)
self._thread.start()
logger.info("[MetaConsumer] Started consuming meta:events stream")
logger.info(
"[MetaConsumer] Started SSE consumer for %s (cursor=%s)",
f"{self._api_url}{self.STREAM_PATH}",
self._cursor or "$",
)
def stop(self) -> None:
"""Stop consuming events."""
if not self._running:
return
self._running = False
self._stop_event.set()
if self._thread:
self._thread.join(timeout=5)
self._thread = None
# Best-effort flush of any unsaved cursor advance.
self._save_cursor()
logger.info("[MetaConsumer] Stopped")
def is_running(self) -> bool:
"""Check if consumer is running."""
return self._running
# ---------- internals -----------------------------------------------
def _load_cursor(self) -> str:
try:
with open(self._cursor_path, "r", encoding="utf-8") as f:
value = f.read().strip()
if value:
logger.info("[MetaConsumer] Resuming from cursor %s", value)
return value
except FileNotFoundError:
return ""
except Exception as e:
logger.warning(
"[MetaConsumer] Could not read cursor %s: %s", self._cursor_path, e
)
return ""
def _save_cursor(self) -> None:
if not self._cursor:
return
try:
Path(self._cursor_path).parent.mkdir(parents=True, exist_ok=True)
with open(self._cursor_path, "w", encoding="utf-8") as f:
f.write(self._cursor)
except Exception as e:
logger.warning(
"[MetaConsumer] Could not write cursor %s: %s", self._cursor_path, e
)
def _consume_loop(self) -> None:
"""Main consumer loop."""
backoff = 0.5
max_backoff = 30.0
while self._running:
try:
# Read from stream with 2s block timeout
entries = self._client.xreadgroup(
self.GROUP,
self.CONSUMER,
{self.STREAM: ">"},
count=100,
block=2000
)
if not entries:
continue
for stream_name, messages in entries:
for msg_id, fields in messages:
self._process_event(msg_id, fields)
# Acknowledge
self._client.xack(self.STREAM, self.GROUP, msg_id)
self._stream_once()
backoff = 0.5 # successful read → reset backoff
except Exception as e:
if self._running:
logger.error(f"[MetaConsumer] Error consuming: {e}")
if not self._running:
return
logger.warning(
"[MetaConsumer] SSE error, reconnecting in %.1fs: %s",
backoff, e,
)
# interruptible sleep
self._stop_event.wait(timeout=backoff)
backoff = min(backoff * 2, max_backoff)
def _process_event(self, msg_id: str, fields: dict) -> None:
"""Process a single event from the stream."""
# Handle both bytes and string keys
key = fields.get(b"key") or fields.get("key")
event_type = fields.get(b"type") or fields.get("type")
def _stream_once(self) -> None:
url = f"{self._api_url}{self.STREAM_PATH}"
req = urllib.request.Request(url, method="GET")
req.add_header("Accept", "text/event-stream")
req.add_header("Cache-Control", "no-cache")
if self._cursor:
req.add_header("Last-Event-ID", self._cursor)
if key is None or event_type is None:
# urllib.request doesn't natively grok SSE; we open the socket and
# read line-by-line until the connection drops or the thread is
# asked to stop.
with urllib.request.urlopen(req, timeout=self._request_timeout) as resp:
buffer = ""
since_save = time.time()
for chunk in resp:
if not self._running:
return
buffer += chunk.decode("utf-8", errors="replace")
# Process complete events (separated by blank lines).
while "\n\n" in buffer:
event_chunk, buffer = buffer.split("\n\n", 1)
self._handle_chunk(event_chunk)
# Flush cursor periodically.
if time.time() - since_save > 2.0:
self._save_cursor()
since_save = time.time()
def _handle_chunk(self, chunk: str) -> None:
event_id = ""
event_type = ""
data_lines: List[str] = []
for line in chunk.split("\n"):
if not line or line.startswith(":"):
continue # comment / heartbeat
colon = line.find(":")
if colon == -1:
continue
field = line[:colon]
value = line[colon + 1:]
if value.startswith(" "):
value = value[1:]
if field == "id":
event_id = value
elif field == "event":
event_type = value
elif field == "data":
data_lines.append(value)
if not data_lines:
return
# Decode bytes if needed
key = key.decode() if isinstance(key, bytes) else key
event_type = event_type.decode() if isinstance(event_type, bytes) else event_type
try:
data = json.loads("\n".join(data_lines))
except json.JSONDecodeError as e:
logger.warning(
"[MetaConsumer] Skipping unparseable event %s: %s",
event_id or "(no id)", e,
)
return
# Filter: only care about interesting field changes
# `gap` is a synthetic event signalling that the server trimmed the
# cursor out of retention. Reset to the resumeFrom value and
# continue. Stremio doesn't bootstrap from the stream so loss is
# acceptable — it just means the cache may be stale for a moment.
if event_type == "gap":
resume = data.get("resumeFrom")
if isinstance(resume, str) and resume:
self._cursor = resume
logger.warning(
"[MetaConsumer] Stream cursor trimmed; resuming from %s",
resume,
)
return
key = data.get("key")
if not key or not event_type:
return
if not self._is_interesting_field(key):
# Still advance the cursor so we don't replay this event after
# a restart.
if event_id:
self._cursor = event_id
return
# Notify callbacks
for callback in self._callbacks:
try:
callback(key, event_type)
except Exception as e:
logger.error(f"[MetaConsumer] Callback error: {e}")
logger.error("[MetaConsumer] Callback error: %s", e)
if event_id:
self._cursor = event_id
def _is_interesting_field(self, key: str) -> bool:
"""Check if the key represents an interesting field."""
# Key format: file:{hashId}/{field}
for field in self.INTERESTING_FIELDS:
if key.endswith(f"/{field}"):
return True
+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