Compare commits
| Author | SHA1 | Date | |
|---|---|---|---|
 Worph
|
ba6e152b4f | ||
|
Worph
|
c1b8a2e731 | ||
|
Pierre
|
6cf93b72a7 | ||
|
Pierre
|
37223c442b | ||
|
Pierre
|
56eced073a | ||
|
Pierre
|
28cb433724 |
@@ -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
|
||||||
@@ -1,16 +1,11 @@
|
|||||||
# Meta-Stremio
|
# 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
|
## 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 │
|
│ MetaMesh Ecosystem │
|
||||||
@@ -24,215 +19,190 @@ Meta-Stremio is a **process-read** service that:
|
|||||||
│ ▼ ▼ ▼ │
|
│ ▼ ▼ ▼ │
|
||||||
│ ┌─────────────────────────────────────────────────────┐ │
|
│ ┌─────────────────────────────────────────────────────┐ │
|
||||||
│ │ Shared KV Store (Redis) │ │
|
│ │ Shared KV Store (Redis) │ │
|
||||||
│ │ Leader election via flock on shared filesystem │ │
|
│ │ meta-core is leader, exposes /urls + /meta APIs │ │
|
||||||
│ │ DB stored in: DATA/Apps/meta-core/DB │ │
|
|
||||||
│ └─────────────────────────────────────────────────────┘ │
|
│ └─────────────────────────────────────────────────────┘ │
|
||||||
│ │ │
|
│ │ │
|
||||||
│ ▼ │
|
│ ▼ │
|
||||||
│ ┌─────────────────────────────────────────────────────┐ │
|
│ ┌─────────────────────────────────────────────────────┐ │
|
||||||
│ │ Shared DATA Volume (media files) │ │
|
│ │ File access via WebDAV (or direct mount) │ │
|
||||||
│ │ meta-sort writes → meta-stremio/meta-fuse reads │ │
|
|
||||||
│ └─────────────────────────────────────────────────────┘ │
|
│ └─────────────────────────────────────────────────────┘ │
|
||||||
│ │
|
|
||||||
└─────────────────────────────────────────────────────────────────────┘
|
└─────────────────────────────────────────────────────────────────────┘
|
||||||
```
|
```
|
||||||
|
|
||||||
## Features
|
## Features
|
||||||
|
|
||||||
### High Performance
|
- **HLS transcoding** — adaptive preset+CRF targeting a 60–80% transcode ratio, prefetch of N segments ahead, persistent segment cache.
|
||||||
- **Adaptive Quality** - Dynamic preset + CRF adjustment targeting 60-80% transcode ratio
|
- **Stremio protocol** — manifest, catalog, meta, stream; one HLS stream per audio track; subtitles extracted to VTT.
|
||||||
- **Intelligent Prefetch** - 4 segments ahead for smooth playback
|
- **Direct file serving** — range-request capable, for clients that can play the source directly.
|
||||||
- **Single-threaded Encode** - 100% CPU per segment for fastest encoding
|
- **Storage abstractions** — `LeaderStorage` (auto-discovers Redis via meta-core), `RedisStorage` (direct URL), `direct` mode (reads from meta-core's HTTP API + SSE meta-stream).
|
||||||
- **Muxed Segments** - Video+audio in one FFmpeg call
|
- **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).
|
||||||
- **Segment Caching** - Persistent cache for repeated access
|
- **Language-configurable manifest** — `displayLanguage` in the per-install config (base64-in-path) selects which `titles` translation meta-sort's TMDB plugin populated.
|
||||||
|
|
||||||
### 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
|
|
||||||
```
|
|
||||||
|
|
||||||
## Installation
|
## Installation
|
||||||
|
|
||||||
### Docker (Recommended)
|
### Docker (in the MetaMesh dev stack)
|
||||||
|
|
||||||
```bash
|
```bash
|
||||||
# Build
|
cd dev
|
||||||
docker build -t meta-stremio .
|
./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 \
|
docker run -d \
|
||||||
-p 7000:7000 \
|
-p 8182:7000 \
|
||||||
-v /path/to/media:/files:ro \
|
-v /path/to/media:/files:ro \
|
||||||
-v /path/to/meta-core:/meta-core \
|
-v /path/to/meta-core:/meta-core \
|
||||||
meta-stremio
|
meta-stremio
|
||||||
|
|
||||||
# Run with direct Redis connection
|
# Direct Redis mode
|
||||||
docker run -d \
|
docker run -d \
|
||||||
-p 7000:7000 \
|
-p 8182:7000 \
|
||||||
-v /path/to/media:/files:ro \
|
-v /path/to/media:/files:ro \
|
||||||
-e STORAGE_MODE=redis \
|
-e STORAGE_MODE=redis \
|
||||||
-e REDIS_URL=redis://your-redis:6379 \
|
-e REDIS_URL=redis://your-redis:6379 \
|
||||||
meta-stremio
|
meta-stremio
|
||||||
```
|
```
|
||||||
|
|
||||||
### Docker Compose
|
The container always listens on port `7000` internally — map it to whatever you want externally.
|
||||||
|
|
||||||
|
### Standalone (development)
|
||||||
|
|
||||||
```bash
|
```bash
|
||||||
# Run as part of the MetaMesh stack (uses leader discovery)
|
# Requires Python 3.9+ and FFmpeg.
|
||||||
cd dev
|
pip install -r requirements.txt # redis, watchdog, Pillow, requests
|
||||||
./start.sh
|
|
||||||
```
|
|
||||||
|
|
||||||
### Standalone (Development)
|
|
||||||
|
|
||||||
```bash
|
|
||||||
# Prerequisites: Python 3.9+, FFmpeg
|
|
||||||
|
|
||||||
# Install dependencies
|
|
||||||
pip install -r requirements.txt
|
|
||||||
|
|
||||||
# Run
|
|
||||||
cd src
|
cd src
|
||||||
python server.py
|
python server.py
|
||||||
```
|
```
|
||||||
|
|
||||||
## Configuration
|
## Configuration
|
||||||
|
|
||||||
### Environment Variables
|
### Environment variables
|
||||||
|
|
||||||
| Variable | Default | Description |
|
| Variable | Default | Description |
|
||||||
|----------|---------|-------------|
|
|---|---|---|
|
||||||
| `PORT` | `7000` | Server listen port |
|
| `PORT` | `7000` | Internal HTTP listen port (the dev stack maps 18182 / 8182 to this). |
|
||||||
| `MEDIA_DIR` | `/data/media` | Media files directory |
|
| `BASE_URL` | — | External URL emitted in manifests / poster URLs (e.g. `https://metastremio-dev.localhost:8182`). |
|
||||||
| `CACHE_DIR` | `/data/cache` | Transcoded segment cache |
|
| `MEDIA_DIR` | `/files/watch` | Media directory (consumed by `transcoder.py`). |
|
||||||
| `STORAGE_MODE` | `leader` | `leader` or `redis` |
|
| `CACHE_DIR` | `/data/cache` | Transcoded segment cache. |
|
||||||
| `META_CORE_PATH` | `/meta-core` | Path to meta-core shared volume (for leader discovery) |
|
| `STORAGE_MODE` | `leader` | `leader`, `redis`, or `direct`. |
|
||||||
| `FILES_PATH` | `/files` | Path to files volume |
|
| `META_CORE_PATH` | `/meta-core` | Shared meta-core volume; used by `LeaderClient` for `kv-leader.info`. |
|
||||||
| `REDIS_URL` | `redis://localhost:6379` | Redis connection URL (for `redis` mode) |
|
| `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. |
|
||||||
| `REDIS_PREFIX` | `meta-sort:` | Redis key prefix |
|
| `FILES_PATH` | `/files` | Files volume. |
|
||||||
| `SEGMENT_DURATION` | `4` | HLS segment length (seconds) |
|
| `REDIS_URL` | `redis://localhost:6379` | Redis URL for `STORAGE_MODE=redis`. |
|
||||||
| `PREFETCH_SEGMENTS` | `4` | Segments to prefetch ahead |
|
| `REDIS_PREFIX` | `meta-sort:` | Redis key prefix. |
|
||||||
| `SCHEME` | `auto` | URL scheme (`http`, `https`, `auto`) |
|
| `SCHEME` | `auto` | URL scheme for generated URLs: `http`, `https`, or `auto`. |
|
||||||
| `META_CORE_WEBDAV_URL` | - | WebDAV URL for file access (e.g., `http://meta-core/webdav`) |
|
| `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 |
|
| Endpoint | Method | Description |
|
||||||
|----------|--------|-------------|
|
|---|---|---|
|
||||||
| `/` | GET | Setup dashboard |
|
| `/` , `/index.html` | GET | Setup dashboard |
|
||||||
| `/health` | GET | Health check with storage status |
|
| `/configure` | GET | Per-install language configuration page |
|
||||||
|
| `/health` | GET | Liveness + storage status |
|
||||||
### Dashboard API
|
|
||||||
| Endpoint | Method | Description |
|
|
||||||
|----------|--------|-------------|
|
|
||||||
| `/api/stats` | GET | Library statistics |
|
| `/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 |
|
| Endpoint | Method | Description |
|
||||||
|----------|--------|-------------|
|
|---|---|---|
|
||||||
| `/manifest.json` | GET | Addon manifest |
|
| `/manifest.json` | GET, HEAD | Addon manifest |
|
||||||
| `/catalog/:type/:id.json` | GET | Video catalog |
|
| `/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 |
|
| `/meta/:type/:id.json` | GET | Video metadata |
|
||||||
|
| `/stremio/meta/:type/:id.json` | GET | Alias |
|
||||||
| `/stream/:type/:id.json` | GET | Stream URLs |
|
| `/stream/:type/:id.json` | GET | Stream URLs |
|
||||||
|
| `/stremio/stream/:type/:id.json` | GET | Alias |
|
||||||
|
|
||||||
|
### File serving
|
||||||
|
|
||||||
### Transcoding
|
|
||||||
| Endpoint | Method | Description |
|
| Endpoint | Method | Description |
|
||||||
|----------|--------|-------------|
|
|---|---|---|
|
||||||
| `/transcode/:path/master.m3u8` | GET | ABR master playlist |
|
| `/file/{cid}` | GET, HEAD | Serve file by CID |
|
||||||
| `/transcode/:path/master_original_a0.m3u8` | GET | Original quality playlist |
|
| `/file/{cid}/w{width}` | GET, HEAD | Serve resized image by CID (uses Pillow) |
|
||||||
| `/transcode/:path/stream_a0_720p.m3u8` | GET | Quality stream playlist |
|
| `/poster/{cid}` | GET, HEAD | Legacy alias of `/file/{cid}` |
|
||||||
| `/transcode/:path/seg_a0_720p_0.ts` | GET | Video segment |
|
| `/poster/{cid}/w{width}` | GET, HEAD | Legacy alias of `/file/{cid}/w{width}` |
|
||||||
| `/transcode/:path/subtitle_0.vtt` | GET | Subtitle track |
|
| `/direct/{path}` | GET, HEAD | Direct file serving with HTTP `Range` support |
|
||||||
| `/direct/:path` | GET | Direct file serving |
|
|
||||||
| `/transcode/metrics` | GET | Transcoding metrics |
|
|
||||||
|
|
||||||
## 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
|
meta-stremio/
|
||||||
- filePath → /media/movies/Movie.mkv
|
├── README.md
|
||||||
- title → Movie Title
|
├── Dockerfile
|
||||||
- type → movie|series|anime
|
├── docker-compose.yml
|
||||||
- year → 2024
|
├── docker/ # container entry scripts
|
||||||
- season → 1 (for series)
|
├── requirements.txt # redis, watchdog, Pillow, requests
|
||||||
- episode → 1 (for series)
|
├── src/
|
||||||
- duration → 7200 (seconds)
|
│ ├── server.py # HTTP server + routing (entry point)
|
||||||
- width → 1920
|
│ ├── stremio.py # Stremio handlers + library/language helpers
|
||||||
- height → 1080
|
│ ├── transcoder.py # HLS transcoding engine + metrics + segment/subtitle managers
|
||||||
- videoCodec → h264
|
│ ├── fileserver.py # serve_file(cid, width) — file/poster endpoints
|
||||||
- audioCodec → aac
|
│ ├── poster.py # poster URL/CID helpers
|
||||||
- audioTracks → JSON string
|
│ ├── webdav_client.py # WebDAV client (file_exists, stream_file, stream_range, get_file_size)
|
||||||
- subtitles → JSON string
|
│ └── storage/
|
||||||
- imdbId → tt1234567
|
│ ├── __init__.py
|
||||||
- poster → URL
|
│ ├── 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
|
## Stream types
|
||||||
- Best quality, may not play on all devices
|
|
||||||
|
|
||||||
2. **HLS Original** - Transcoded at source resolution
|
For each video, the addon advertises multiple streams:
|
||||||
- H.264/AAC in HLS container
|
|
||||||
- One stream per audio track
|
|
||||||
|
|
||||||
3. **HLS ABR** - Adaptive bitrate with quality ladder
|
1. **Direct file** — original bytes, served via `/direct/...` with `Range` support. Best quality, may not play on every device.
|
||||||
- Multiple resolutions (1080p, 720p, 480p, 360p)
|
2. **HLS original** — transcoded at source resolution, H.264/AAC, one stream per audio track.
|
||||||
- Automatic quality switching
|
3. **HLS ABR** — adaptive ladder. Quality presets are baked into `transcoder.py`.
|
||||||
|
|
||||||
## Transcoding Details
|
| Resolution | Video bitrate | Audio bitrate |
|
||||||
|
|---|---|---|
|
||||||
### 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 |
|
|
||||||
|------------|---------------|---------------|
|
|
||||||
| 1080p | 5000 kbps | 192 kbps |
|
| 1080p | 5000 kbps | 192 kbps |
|
||||||
| 720p | 3000 kbps | 128 kbps |
|
| 720p | 3000 kbps | 128 kbps |
|
||||||
| 480p | 1500 kbps | 128 kbps |
|
| 480p | 1500 kbps | 128 kbps |
|
||||||
@@ -240,29 +210,32 @@ The transcoder automatically adjusts encoding settings to maintain a 60-80% tran
|
|||||||
|
|
||||||
## Adding to Stremio
|
## Adding to Stremio
|
||||||
|
|
||||||
1. Open the dashboard at `http://localhost:7000`
|
1. Open the dashboard at `https://metastremio-dev.localhost:8182/` (or debug-direct `http://localhost:18182/`).
|
||||||
2. Click "Install in Stremio" or copy the manifest URL
|
2. Pick your display language on `/configure` if you want non-English titles.
|
||||||
3. In Stremio: Settings → Addons → Enter URL → Install
|
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
|
```bash
|
||||||
# Run in development mode (requires meta-sort running with Redis)
|
# Container logs
|
||||||
cd src
|
docker compose -f dev/docker-compose.yml logs -f metastremio-app
|
||||||
python server.py
|
|
||||||
|
|
||||||
# Or with direct Redis URL
|
# Live health
|
||||||
STORAGE_MODE=redis REDIS_URL=redis://localhost:6379 python server.py
|
curl -k https://metastremio-dev.localhost:8182/health
|
||||||
|
curl http://localhost:18182/health # debug-direct (no auth, no TLS)
|
||||||
|
|
||||||
# View logs
|
# Transcoder metrics
|
||||||
tail -f /data/cache/*.log
|
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)
|
## Related projects
|
||||||
- **[meta-fuse](../meta-fuse)** - Virtual filesystem (reads metadata)
|
|
||||||
- **[meta-orbit](../meta-orbit)** - P2P metadata sharing
|
- **[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
|
## License
|
||||||
|
|
||||||
|
|||||||
+70
-8
@@ -28,6 +28,7 @@ import json
|
|||||||
import signal
|
import signal
|
||||||
import sys
|
import sys
|
||||||
import base64
|
import base64
|
||||||
|
import hashlib
|
||||||
from http.server import HTTPServer, BaseHTTPRequestHandler
|
from http.server import HTTPServer, BaseHTTPRequestHandler
|
||||||
from urllib.parse import urlparse, unquote
|
from urllib.parse import urlparse, unquote
|
||||||
import threading
|
import threading
|
||||||
@@ -44,6 +45,23 @@ SCHEME = os.environ.get('SCHEME', 'auto').lower().strip()
|
|||||||
META_CORE_PATH = os.environ.get('META_CORE_PATH', '/meta-core')
|
META_CORE_PATH = os.environ.get('META_CORE_PATH', '/meta-core')
|
||||||
BASE_URL = os.environ.get('BASE_URL', '')
|
BASE_URL = os.environ.get('BASE_URL', '')
|
||||||
|
|
||||||
|
# Stremio addon path-token protection. When HASH_API_SEED is set, every addon
|
||||||
|
# path (manifest, catalog, meta, stream, transcode, direct, file, poster) must
|
||||||
|
# be prefixed with /s/<token>. Dashboard paths bypass this — Caddy + the
|
||||||
|
# nginx-hash-lock sidecar gate those with OIDC, since browsers can carry
|
||||||
|
# cookies but Stremio clients cannot.
|
||||||
|
HASH_API_SEED = os.environ.get('HASH_API_SEED', '').strip()
|
||||||
|
API_KEY = hashlib.sha256(HASH_API_SEED.encode()).hexdigest()[:16] if HASH_API_SEED else ''
|
||||||
|
API_PREFIX = f'/s/{API_KEY}' if API_KEY else ''
|
||||||
|
|
||||||
|
# Paths the addon serves for browsers, not Stremio clients. These are gated
|
||||||
|
# upstream by hash-lock/OIDC instead of the path-token.
|
||||||
|
DASHBOARD_PATHS = frozenset({
|
||||||
|
'/', '/index.html', '/configure', '/health',
|
||||||
|
'/api/stats', '/api/library', '/api/services', '/api/languages',
|
||||||
|
'/transcode/metrics',
|
||||||
|
})
|
||||||
|
|
||||||
# Initialize storage
|
# Initialize storage
|
||||||
storage = stremio.init_storage()
|
storage = stremio.init_storage()
|
||||||
|
|
||||||
@@ -133,10 +151,32 @@ class Handler(BaseHTTPRequestHandler):
|
|||||||
self.send_header('Access-Control-Allow-Headers', 'Content-Type, Range')
|
self.send_header('Access-Control-Allow-Headers', 'Content-Type, Range')
|
||||||
self.end_headers()
|
self.end_headers()
|
||||||
|
|
||||||
|
def strip_api_prefix(self, path: str) -> str | None:
|
||||||
|
"""Remove the /s/<token> prefix from addon paths.
|
||||||
|
|
||||||
|
Returns the de-prefixed path, or None when the token is missing/wrong
|
||||||
|
(caller should send 401). Dashboard paths are returned untouched —
|
||||||
|
they are protected by hash-lock/OIDC upstream, not by the token.
|
||||||
|
"""
|
||||||
|
if not API_KEY:
|
||||||
|
return path
|
||||||
|
if path in DASHBOARD_PATHS:
|
||||||
|
return path
|
||||||
|
if path.startswith(API_PREFIX + '/'):
|
||||||
|
return path[len(API_PREFIX):]
|
||||||
|
if path == API_PREFIX:
|
||||||
|
return '/'
|
||||||
|
return None
|
||||||
|
|
||||||
def do_HEAD(self):
|
def do_HEAD(self):
|
||||||
parsed = urlparse(self.path)
|
parsed = urlparse(self.path)
|
||||||
path = unquote(parsed.path)
|
path = unquote(parsed.path)
|
||||||
|
|
||||||
|
path = self.strip_api_prefix(path)
|
||||||
|
if path is None:
|
||||||
|
self.send_error(401, "Unauthorized")
|
||||||
|
return
|
||||||
|
|
||||||
if path.startswith('/stremio/') or path == '/' or path.startswith('/manifest.json'):
|
if path.startswith('/stremio/') or path == '/' or path.startswith('/manifest.json'):
|
||||||
self.send_response(200)
|
self.send_response(200)
|
||||||
self.send_header('Access-Control-Allow-Origin', '*')
|
self.send_header('Access-Control-Allow-Origin', '*')
|
||||||
@@ -181,6 +221,11 @@ class Handler(BaseHTTPRequestHandler):
|
|||||||
parsed = urlparse(self.path)
|
parsed = urlparse(self.path)
|
||||||
path = unquote(parsed.path)
|
path = unquote(parsed.path)
|
||||||
|
|
||||||
|
path = self.strip_api_prefix(path)
|
||||||
|
if path is None:
|
||||||
|
self.send_error(401, "Unauthorized")
|
||||||
|
return
|
||||||
|
|
||||||
# Reset metrics
|
# Reset metrics
|
||||||
if path == '/transcode/reset-metrics':
|
if path == '/transcode/reset-metrics':
|
||||||
transcoder.reset_metrics()
|
transcoder.reset_metrics()
|
||||||
@@ -192,8 +237,16 @@ class Handler(BaseHTTPRequestHandler):
|
|||||||
parsed = urlparse(self.path)
|
parsed = urlparse(self.path)
|
||||||
raw_path = unquote(parsed.path)
|
raw_path = unquote(parsed.path)
|
||||||
|
|
||||||
|
# Strip path-token (when HASH_API_SEED is set) before any further
|
||||||
|
# parsing; the existing base64-config-in-path scheme then runs on the
|
||||||
|
# tail, so URLs look like /s/<token>/<configB64>/manifest.json.
|
||||||
|
stripped = self.strip_api_prefix(raw_path)
|
||||||
|
if stripped is None:
|
||||||
|
self.send_error(401, "Unauthorized")
|
||||||
|
return
|
||||||
|
|
||||||
# Parse config from path (e.g., /eyJsYW5n.../manifest.json)
|
# Parse config from path (e.g., /eyJsYW5n.../manifest.json)
|
||||||
path, config = parse_config_from_path(raw_path)
|
path, config = parse_config_from_path(stripped)
|
||||||
|
|
||||||
# Root - serve setup page
|
# Root - serve setup page
|
||||||
if path == '/' or path == '/index.html':
|
if path == '/' or path == '/index.html':
|
||||||
@@ -325,11 +378,14 @@ class Handler(BaseHTTPRequestHandler):
|
|||||||
if os.path.exists(html_path):
|
if os.path.exists(html_path):
|
||||||
with open(html_path, 'r', encoding='utf-8') as f:
|
with open(html_path, 'r', encoding='utf-8') as f:
|
||||||
content = f.read()
|
content = f.read()
|
||||||
|
prefix_script = f'<script>window.API_PREFIX = "{API_PREFIX}";</script>'
|
||||||
|
content = content.replace('</head>', f'{prefix_script}\n</head>', 1)
|
||||||
|
encoded = content.encode('utf-8')
|
||||||
self.send_response(200)
|
self.send_response(200)
|
||||||
self.send_header('Content-Type', 'text/html')
|
self.send_header('Content-Type', 'text/html')
|
||||||
self.send_header('Content-Length', len(content.encode('utf-8')))
|
self.send_header('Content-Length', len(encoded))
|
||||||
self.end_headers()
|
self.end_headers()
|
||||||
self.wfile.write(content.encode('utf-8'))
|
self.wfile.write(encoded)
|
||||||
else:
|
else:
|
||||||
# Fallback minimal page
|
# Fallback minimal page
|
||||||
content = f"""<!DOCTYPE html>
|
content = f"""<!DOCTYPE html>
|
||||||
@@ -338,7 +394,7 @@ class Handler(BaseHTTPRequestHandler):
|
|||||||
<body style="font-family: sans-serif; background: #1a1a2e; color: #fff; padding: 2rem;">
|
<body style="font-family: sans-serif; background: #1a1a2e; color: #fff; padding: 2rem;">
|
||||||
<h1>Meta-Stremio</h1>
|
<h1>Meta-Stremio</h1>
|
||||||
<p>Install URL: <code>{self.get_base_url()}/manifest.json</code></p>
|
<p>Install URL: <code>{self.get_base_url()}/manifest.json</code></p>
|
||||||
<p><a href="stremio://{self.headers.get('Host', 'localhost')}/manifest.json" style="color: #4dabf7;">Install in Stremio</a></p>
|
<p><a href="stremio://{self.headers.get('Host', 'localhost')}{API_PREFIX}/manifest.json" style="color: #4dabf7;">Install in Stremio</a></p>
|
||||||
</body>
|
</body>
|
||||||
</html>"""
|
</html>"""
|
||||||
self.send_response(200)
|
self.send_response(200)
|
||||||
@@ -364,6 +420,7 @@ class Handler(BaseHTTPRequestHandler):
|
|||||||
# Inline configure page (fallback)
|
# Inline configure page (fallback)
|
||||||
base_url = self.get_base_url()
|
base_url = self.get_base_url()
|
||||||
host = self.headers.get('Host', 'localhost')
|
host = self.headers.get('Host', 'localhost')
|
||||||
|
api_prefix = API_PREFIX
|
||||||
languages = stremio.get_supported_languages()
|
languages = stremio.get_supported_languages()
|
||||||
|
|
||||||
lang_options = '\n'.join(
|
lang_options = '\n'.join(
|
||||||
@@ -508,6 +565,7 @@ class Handler(BaseHTTPRequestHandler):
|
|||||||
<script>
|
<script>
|
||||||
const baseUrl = '{base_url}';
|
const baseUrl = '{base_url}';
|
||||||
const host = '{host}';
|
const host = '{host}';
|
||||||
|
const apiPrefix = '{api_prefix}';
|
||||||
|
|
||||||
function encodeConfig(config) {{
|
function encodeConfig(config) {{
|
||||||
const json = JSON.stringify(config);
|
const json = JSON.stringify(config);
|
||||||
@@ -523,12 +581,12 @@ class Handler(BaseHTTPRequestHandler):
|
|||||||
if (displayLang === 'en') {{
|
if (displayLang === 'en') {{
|
||||||
// No config needed for default
|
// No config needed for default
|
||||||
url = baseUrl + '/manifest.json';
|
url = baseUrl + '/manifest.json';
|
||||||
stremioUrl = 'stremio://' + host + '/manifest.json';
|
stremioUrl = 'stremio://' + host + apiPrefix + '/manifest.json';
|
||||||
}} else {{
|
}} else {{
|
||||||
const config = {{ displayLanguage: displayLang }};
|
const config = {{ displayLanguage: displayLang }};
|
||||||
const encoded = encodeConfig(config);
|
const encoded = encodeConfig(config);
|
||||||
url = baseUrl + '/' + encoded + '/manifest.json';
|
url = baseUrl + '/' + encoded + '/manifest.json';
|
||||||
stremioUrl = 'stremio://' + host + '/' + encoded + '/manifest.json';
|
stremioUrl = 'stremio://' + host + apiPrefix + '/' + encoded + '/manifest.json';
|
||||||
}}
|
}}
|
||||||
|
|
||||||
document.getElementById('installUrl').textContent = url;
|
document.getElementById('installUrl').textContent = url;
|
||||||
@@ -827,7 +885,7 @@ class Handler(BaseHTTPRequestHandler):
|
|||||||
proto = 'http' if is_localhost else 'https'
|
proto = 'http' if is_localhost else 'https'
|
||||||
|
|
||||||
clean_host = host.replace(':80', '').replace(':443', '')
|
clean_host = host.replace(':80', '').replace(':443', '')
|
||||||
return f"{proto}://{clean_host}"
|
return f"{proto}://{clean_host}{API_PREFIX}"
|
||||||
|
|
||||||
|
|
||||||
class ThreadedServer(HTTPServer):
|
class ThreadedServer(HTTPServer):
|
||||||
@@ -875,8 +933,12 @@ def main():
|
|||||||
print(f"Storage: {storage_type} | Videos: {video_count}")
|
print(f"Storage: {storage_type} | Videos: {video_count}")
|
||||||
print(f"Media: {transcoder.MEDIA_DIR} | Cache: {transcoder.CACHE_DIR}")
|
print(f"Media: {transcoder.MEDIA_DIR} | Cache: {transcoder.CACHE_DIR}")
|
||||||
print(f"Segment: {transcoder.SEGMENT_DURATION}s | Prefetch: {transcoder.PREFETCH_SEGMENTS} segments")
|
print(f"Segment: {transcoder.SEGMENT_DURATION}s | Prefetch: {transcoder.PREFETCH_SEGMENTS} segments")
|
||||||
print(f"Manifest URL: http://localhost:{PORT}/manifest.json")
|
print(f"Manifest URL: http://localhost:{PORT}{API_PREFIX}/manifest.json")
|
||||||
print(f"Dashboard: http://localhost:{PORT}/")
|
print(f"Dashboard: http://localhost:{PORT}/")
|
||||||
|
if API_KEY:
|
||||||
|
print(f"Path-token protection: enabled (prefix: {API_PREFIX})")
|
||||||
|
else:
|
||||||
|
print("Path-token protection: disabled (set HASH_API_SEED to enable)")
|
||||||
print("Adaptive quality: target 60-80% transcode ratio")
|
print("Adaptive quality: target 60-80% transcode ratio")
|
||||||
|
|
||||||
if service_discovery:
|
if service_discovery:
|
||||||
|
|||||||
@@ -137,6 +137,21 @@ class LeaderClient:
|
|||||||
|
|
||||||
data = json.loads(response.read().decode())
|
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(
|
self._cached_urls = URLsResponse(
|
||||||
hostname=data.get('hostname', ''),
|
hostname=data.get('hostname', ''),
|
||||||
base_url=data.get('baseUrl', ''),
|
base_url=data.get('baseUrl', ''),
|
||||||
|
|||||||
+145
-38
@@ -20,6 +20,7 @@ from typing import List, Optional, Callable
|
|||||||
from .provider import StorageProvider, VideoMetadata
|
from .provider import StorageProvider, VideoMetadata
|
||||||
from .leader_client import LeaderClient, LeaderLockInfo
|
from .leader_client import LeaderClient, LeaderLockInfo
|
||||||
from .meta_consumer import MetaConsumer
|
from .meta_consumer import MetaConsumer
|
||||||
|
from .meta_core_api_client import MetaCoreApiClient
|
||||||
|
|
||||||
# Import webdav_client to configure it after leader discovery
|
# Import webdav_client to configure it after leader discovery
|
||||||
import webdav_client
|
import webdav_client
|
||||||
@@ -70,6 +71,7 @@ class LeaderStorage(StorageProvider):
|
|||||||
|
|
||||||
self._leader_client: Optional[LeaderClient] = None
|
self._leader_client: Optional[LeaderClient] = None
|
||||||
self._client: Optional[redis.Redis] = None
|
self._client: Optional[redis.Redis] = None
|
||||||
|
self._api_client: Optional[MetaCoreApiClient] = None
|
||||||
self._connected = False
|
self._connected = False
|
||||||
self._lock = threading.Lock()
|
self._lock = threading.Lock()
|
||||||
|
|
||||||
@@ -122,9 +124,14 @@ class LeaderStorage(StorageProvider):
|
|||||||
# Try to get leader info (short timeout per attempt)
|
# Try to get leader info (short timeout per attempt)
|
||||||
info = self._leader_client.get_leader_info()
|
info = self._leader_client.get_leader_info()
|
||||||
|
|
||||||
if info and info.redis_url:
|
# api-mediated-access PR D: when redis_url is absent, run in
|
||||||
print(f"[LeaderStorage] Found leader at {info.redis_url} (attempt {attempt})")
|
# HTTP-only mode against meta-core's /meta/* surface. Redis
|
||||||
self._connect_to_redis(info.redis_url)
|
# 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:
|
if self._connected:
|
||||||
# Configure WebDAV client with leader's internal WebDAV URL (for container-to-container access)
|
# 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()
|
self._leader_client.start_watching()
|
||||||
return
|
return
|
||||||
else:
|
else:
|
||||||
print(f"[LeaderStorage] Failed to connect to Redis, retrying...")
|
print(f"[LeaderStorage] Failed to connect to meta-core, retrying...")
|
||||||
else:
|
else:
|
||||||
print(f"[LeaderStorage] Waiting for meta-core leader... (attempt {attempt})")
|
print(f"[LeaderStorage] Waiting for meta-core leader... (attempt {attempt})")
|
||||||
|
|
||||||
@@ -174,9 +181,9 @@ class LeaderStorage(StorageProvider):
|
|||||||
try:
|
try:
|
||||||
info = self._leader_client.get_leader_info()
|
info = self._leader_client.get_leader_info()
|
||||||
|
|
||||||
if info and info.redis_url:
|
if info and info.api_url:
|
||||||
print(f"[LeaderStorage] Background: Found leader at {info.redis_url}")
|
print(f"[LeaderStorage] Background: Found leader at {info.api_url}")
|
||||||
self._connect_to_redis(info.redis_url)
|
self._connect_via_api(info)
|
||||||
|
|
||||||
if self._connected:
|
if self._connected:
|
||||||
# Configure WebDAV client
|
# Configure WebDAV client
|
||||||
@@ -214,9 +221,30 @@ class LeaderStorage(StorageProvider):
|
|||||||
self._disconnect_redis()
|
self._disconnect_redis()
|
||||||
|
|
||||||
def is_connected(self) -> bool:
|
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:
|
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
|
return False
|
||||||
try:
|
try:
|
||||||
self._client.ping()
|
self._client.ping()
|
||||||
@@ -241,9 +269,9 @@ class LeaderStorage(StorageProvider):
|
|||||||
try:
|
try:
|
||||||
info = self._leader_client.get_leader_info()
|
info = self._leader_client.get_leader_info()
|
||||||
|
|
||||||
if info and info.redis_url:
|
if info and info.api_url:
|
||||||
print(f"[LeaderStorage] Reconnecting to leader at {info.redis_url}...")
|
print(f"[LeaderStorage] Reconnecting to leader at {info.api_url}...")
|
||||||
self._connect_to_redis(info.redis_url)
|
self._connect_via_api(info)
|
||||||
|
|
||||||
if self._connected:
|
if self._connected:
|
||||||
# Reconfigure WebDAV client with new leader's internal WebDAV URL
|
# 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")
|
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:
|
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:
|
with self._lock:
|
||||||
try:
|
try:
|
||||||
# Disconnect existing client
|
# Disconnect existing client
|
||||||
@@ -274,11 +340,19 @@ class LeaderStorage(StorageProvider):
|
|||||||
|
|
||||||
print(f"[LeaderStorage] Connected to Redis at {url}")
|
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:
|
try:
|
||||||
self._meta_consumer = MetaConsumer(self._client, self._prefix)
|
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.on_change(self._on_metadata_change)
|
||||||
self._meta_consumer.start()
|
self._meta_consumer.start()
|
||||||
|
else:
|
||||||
|
print("[LeaderStorage] Warning: api_url unavailable; meta consumer not started")
|
||||||
except Exception as e:
|
except Exception as e:
|
||||||
print(f"[LeaderStorage] Warning: failed to start meta consumer: {e}")
|
print(f"[LeaderStorage] Warning: failed to start meta consumer: {e}")
|
||||||
|
|
||||||
@@ -299,7 +373,7 @@ class LeaderStorage(StorageProvider):
|
|||||||
self._disconnect_redis_unlocked()
|
self._disconnect_redis_unlocked()
|
||||||
|
|
||||||
def _disconnect_redis_unlocked(self) -> None:
|
def _disconnect_redis_unlocked(self) -> None:
|
||||||
"""Disconnect from Redis (without lock)."""
|
"""Disconnect from Redis and/or API (without lock)."""
|
||||||
# Stop meta consumer
|
# Stop meta consumer
|
||||||
if self._meta_consumer:
|
if self._meta_consumer:
|
||||||
try:
|
try:
|
||||||
@@ -314,6 +388,10 @@ class LeaderStorage(StorageProvider):
|
|||||||
except Exception:
|
except Exception:
|
||||||
pass
|
pass
|
||||||
self._client = None
|
self._client = None
|
||||||
|
|
||||||
|
# The API client has no socket; just drop the reference.
|
||||||
|
self._api_client = None
|
||||||
|
|
||||||
self._connected = False
|
self._connected = False
|
||||||
|
|
||||||
# Clear cache
|
# Clear cache
|
||||||
@@ -377,10 +455,14 @@ class LeaderStorage(StorageProvider):
|
|||||||
return f"{self._prefix}file:{hash_id}/"
|
return f"{self._prefix}file:{hash_id}/"
|
||||||
|
|
||||||
def _get_file_metadata(self, hash_id: str) -> dict:
|
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)
|
prefix = self._get_file_key_prefix(hash_id)
|
||||||
data = {}
|
data = {}
|
||||||
|
|
||||||
for key in self._client.scan_iter(f"{prefix}*", count=100):
|
for key in self._client.scan_iter(f"{prefix}*", count=100):
|
||||||
key_str = key if isinstance(key, str) else key.decode()
|
key_str = key if isinstance(key, str) else key.decode()
|
||||||
field = key_str[len(prefix):]
|
field = key_str[len(prefix):]
|
||||||
@@ -494,10 +576,9 @@ class LeaderStorage(StorageProvider):
|
|||||||
# Parse release date
|
# Parse release date
|
||||||
release_date = data.get('releasedate', data.get('releaseDate'))
|
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'))
|
episode_title = data.get('episodeTitle', data.get('episodeName'))
|
||||||
if not episode_title:
|
|
||||||
episode_title = data.get('titles/eng', data.get('titles/en'))
|
|
||||||
|
|
||||||
# Parse tagline
|
# Parse tagline
|
||||||
tagline = data.get('tagline')
|
tagline = data.get('tagline')
|
||||||
@@ -577,6 +658,15 @@ class LeaderStorage(StorageProvider):
|
|||||||
# StorageProvider Interface
|
# 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]:
|
def get_all_videos(self) -> List[VideoMetadata]:
|
||||||
"""Get all videos from storage."""
|
"""Get all videos from storage."""
|
||||||
if not self.is_connected():
|
if not self.is_connected():
|
||||||
@@ -586,8 +676,7 @@ class LeaderStorage(StorageProvider):
|
|||||||
|
|
||||||
try:
|
try:
|
||||||
# Get all hash IDs from index
|
# Get all hash IDs from index
|
||||||
index_key = f"{self._prefix}file:__index__"
|
hash_ids = self._list_hash_ids()
|
||||||
hash_ids = self._client.smembers(index_key)
|
|
||||||
|
|
||||||
for hash_id in hash_ids:
|
for hash_id in hash_ids:
|
||||||
hash_id_str = hash_id if isinstance(hash_id, str) else hash_id.decode()
|
hash_id_str = hash_id if isinstance(hash_id, str) else hash_id.decode()
|
||||||
@@ -648,16 +737,9 @@ class LeaderStorage(StorageProvider):
|
|||||||
|
|
||||||
try:
|
try:
|
||||||
# Get from index and check each file's imdbid
|
# Get from index and check each file's imdbid
|
||||||
index_key = f"{self._prefix}file:__index__"
|
for hash_id_str in self._list_hash_ids():
|
||||||
for hash_id in self._client.smembers(index_key):
|
stored_imdb = self._get_imdb_field(hash_id_str)
|
||||||
hash_id_str = hash_id if isinstance(hash_id, str) else hash_id.decode()
|
if stored_imdb and stored_imdb.lower() == imdb_id:
|
||||||
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)
|
data = self._get_file_metadata(hash_id_str)
|
||||||
return self._parse_video(hash_id_str, data)
|
return self._parse_video(hash_id_str, data)
|
||||||
|
|
||||||
@@ -666,13 +748,28 @@ class LeaderStorage(StorageProvider):
|
|||||||
|
|
||||||
return None
|
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]:
|
def get_file_path_by_cid(self, cid: str) -> Optional[str]:
|
||||||
"""
|
"""
|
||||||
Get the file path for a file by its CID.
|
Get the file path for a file by its CID.
|
||||||
|
|
||||||
Looks up file:{cid}/* in Redis and returns the path.
|
In api-mediated mode this hits meta-core's GET /api/file/{cid}/info
|
||||||
Tries 'path' field first, then falls back to 'filePath'.
|
which already does the reverse-index lookup + canonical-path
|
||||||
This works for any file including poster images.
|
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.
|
Returns the relative path if found, None otherwise.
|
||||||
"""
|
"""
|
||||||
@@ -680,6 +777,11 @@ class LeaderStorage(StorageProvider):
|
|||||||
return None
|
return None
|
||||||
|
|
||||||
try:
|
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)
|
prefix = self._get_file_key_prefix(cid)
|
||||||
# Try 'path' field first (relative path)
|
# Try 'path' field first (relative path)
|
||||||
path = self._client.get(f"{prefix}path")
|
path = self._client.get(f"{prefix}path")
|
||||||
@@ -704,6 +806,8 @@ class LeaderStorage(StorageProvider):
|
|||||||
return 0
|
return 0
|
||||||
|
|
||||||
try:
|
try:
|
||||||
|
if self._api_client is not None:
|
||||||
|
return self._api_client.count_hash_ids()
|
||||||
# Use index set for accurate count
|
# Use index set for accurate count
|
||||||
index_key = f"{self._prefix}file:__index__"
|
index_key = f"{self._prefix}file:__index__"
|
||||||
return self._client.scard(index_key)
|
return self._client.scard(index_key)
|
||||||
@@ -740,8 +844,11 @@ class LeaderStorage(StorageProvider):
|
|||||||
if status['connected']:
|
if status['connected']:
|
||||||
try:
|
try:
|
||||||
status['video_count'] = self.get_video_count()
|
status['video_count'] = self.get_video_count()
|
||||||
info = self._client.info('memory')
|
# Memory stats only available in Redis-direct mode; the HTTP
|
||||||
status['memory_used'] = info.get('used_memory_human', 'N/A')
|
# 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:
|
except Exception:
|
||||||
pass
|
pass
|
||||||
|
|
||||||
|
|||||||
+170
-82
@@ -1,153 +1,241 @@
|
|||||||
"""
|
"""
|
||||||
Meta Events Consumer for meta-stremio.
|
Meta Events Consumer for meta-stremio.
|
||||||
|
|
||||||
Consumes the meta:events Redis stream published by meta-core.
|
Consumes the meta:events stream published by meta-core. Originally used a
|
||||||
Provides real-time notifications when metadata changes.
|
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
|
from __future__ import annotations
|
||||||
|
|
||||||
import threading
|
import json
|
||||||
import logging
|
import logging
|
||||||
|
import os
|
||||||
|
import threading
|
||||||
|
import time
|
||||||
|
from pathlib import Path
|
||||||
from typing import Callable, List, Optional
|
from typing import Callable, List, Optional
|
||||||
|
from urllib.parse import urlparse
|
||||||
|
|
||||||
try:
|
import urllib.request
|
||||||
from redis import Redis
|
import urllib.error
|
||||||
REDIS_AVAILABLE = True
|
|
||||||
except ImportError:
|
|
||||||
REDIS_AVAILABLE = False
|
|
||||||
Redis = None
|
|
||||||
|
|
||||||
logger = logging.getLogger(__name__)
|
logger = logging.getLogger(__name__)
|
||||||
|
|
||||||
|
|
||||||
class MetaConsumer:
|
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.
|
Cursor durability: the last-seen event id is persisted to a file (default
|
||||||
Filters events to only notify about interesting field changes.
|
/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"
|
STREAM_PATH = "/api/events/meta"
|
||||||
GROUP = "stremio-consumer"
|
|
||||||
CONSUMER = "stremio-1"
|
|
||||||
|
|
||||||
# Fields that trigger cache invalidation
|
# Fields whose changes invalidate cached Stremio responses.
|
||||||
INTERESTING_FIELDS = [
|
INTERESTING_FIELDS = [
|
||||||
"tmdb", "tmdbId", "title", "poster", "backdrop",
|
"tmdb", "tmdbId", "title", "poster", "backdrop",
|
||||||
"imdbId", "imdbid", "year", "type", "rating",
|
"imdbId", "imdbid", "year", "type", "rating",
|
||||||
"description", "plot", "genres", "fileType"
|
"description", "plot", "genres", "fileType",
|
||||||
]
|
]
|
||||||
|
|
||||||
def __init__(self, redis_client: Redis, prefix: str = ""):
|
def __init__(
|
||||||
if not REDIS_AVAILABLE:
|
self,
|
||||||
raise ImportError("redis package not installed")
|
api_url: str,
|
||||||
|
cursor_path: str = "/meta-core/cursors/meta-stremio-meta.cursor",
|
||||||
self._client = redis_client
|
request_timeout: int = 35,
|
||||||
self._prefix = prefix
|
) -> None:
|
||||||
|
self._api_url = api_url.rstrip("/")
|
||||||
|
self._cursor_path = cursor_path
|
||||||
|
self._request_timeout = request_timeout
|
||||||
self._running = False
|
self._running = False
|
||||||
self._thread: Optional[threading.Thread] = None
|
self._thread: Optional[threading.Thread] = None
|
||||||
self._callbacks: List[Callable[[str, str], 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':
|
def on_change(self, callback: Callable[[str, str], None]) -> "MetaConsumer":
|
||||||
"""
|
"""Register a callback. See module docstring for the contract."""
|
||||||
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.)
|
|
||||||
"""
|
|
||||||
self._callbacks.append(callback)
|
self._callbacks.append(callback)
|
||||||
return self
|
return self
|
||||||
|
|
||||||
def start(self) -> None:
|
def start(self) -> None:
|
||||||
"""Start consuming events."""
|
|
||||||
if self._running:
|
if self._running:
|
||||||
return
|
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._running = True
|
||||||
|
self._stop_event.clear()
|
||||||
self._thread = threading.Thread(target=self._consume_loop, daemon=True)
|
self._thread = threading.Thread(target=self._consume_loop, daemon=True)
|
||||||
self._thread.start()
|
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:
|
def stop(self) -> None:
|
||||||
"""Stop consuming events."""
|
|
||||||
if not self._running:
|
if not self._running:
|
||||||
return
|
return
|
||||||
|
|
||||||
self._running = False
|
self._running = False
|
||||||
|
self._stop_event.set()
|
||||||
if self._thread:
|
if self._thread:
|
||||||
self._thread.join(timeout=5)
|
self._thread.join(timeout=5)
|
||||||
self._thread = None
|
self._thread = None
|
||||||
|
# Best-effort flush of any unsaved cursor advance.
|
||||||
|
self._save_cursor()
|
||||||
logger.info("[MetaConsumer] Stopped")
|
logger.info("[MetaConsumer] Stopped")
|
||||||
|
|
||||||
def is_running(self) -> bool:
|
def is_running(self) -> bool:
|
||||||
"""Check if consumer is running."""
|
|
||||||
return self._running
|
return self._running
|
||||||
|
|
||||||
def _consume_loop(self) -> None:
|
# ---------- internals -----------------------------------------------
|
||||||
"""Main consumer loop."""
|
|
||||||
while self._running:
|
def _load_cursor(self) -> str:
|
||||||
try:
|
try:
|
||||||
# Read from stream with 2s block timeout
|
with open(self._cursor_path, "r", encoding="utf-8") as f:
|
||||||
entries = self._client.xreadgroup(
|
value = f.read().strip()
|
||||||
self.GROUP,
|
if value:
|
||||||
self.CONSUMER,
|
logger.info("[MetaConsumer] Resuming from cursor %s", value)
|
||||||
{self.STREAM: ">"},
|
return value
|
||||||
count=100,
|
except FileNotFoundError:
|
||||||
block=2000
|
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
|
||||||
)
|
)
|
||||||
|
|
||||||
if not entries:
|
def _consume_loop(self) -> None:
|
||||||
continue
|
backoff = 0.5
|
||||||
|
max_backoff = 30.0
|
||||||
for stream_name, messages in entries:
|
while self._running:
|
||||||
for msg_id, fields in messages:
|
try:
|
||||||
self._process_event(msg_id, fields)
|
self._stream_once()
|
||||||
# Acknowledge
|
backoff = 0.5 # successful read → reset backoff
|
||||||
self._client.xack(self.STREAM, self.GROUP, msg_id)
|
|
||||||
|
|
||||||
except Exception as e:
|
except Exception as e:
|
||||||
if self._running:
|
if not self._running:
|
||||||
logger.error(f"[MetaConsumer] Error consuming: {e}")
|
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:
|
def _stream_once(self) -> None:
|
||||||
"""Process a single event from the stream."""
|
url = f"{self._api_url}{self.STREAM_PATH}"
|
||||||
# Handle both bytes and string keys
|
req = urllib.request.Request(url, method="GET")
|
||||||
key = fields.get(b"key") or fields.get("key")
|
req.add_header("Accept", "text/event-stream")
|
||||||
event_type = fields.get(b"type") or fields.get("type")
|
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
|
return
|
||||||
|
|
||||||
# Decode bytes if needed
|
try:
|
||||||
key = key.decode() if isinstance(key, bytes) else key
|
data = json.loads("\n".join(data_lines))
|
||||||
event_type = event_type.decode() if isinstance(event_type, bytes) else event_type
|
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):
|
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
|
return
|
||||||
|
|
||||||
# Notify callbacks
|
|
||||||
for callback in self._callbacks:
|
for callback in self._callbacks:
|
||||||
try:
|
try:
|
||||||
callback(key, event_type)
|
callback(key, event_type)
|
||||||
except Exception as e:
|
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:
|
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:
|
for field in self.INTERESTING_FIELDS:
|
||||||
if key.endswith(f"/{field}"):
|
if key.endswith(f"/{field}"):
|
||||||
return True
|
return True
|
||||||
|
|||||||
@@ -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
|
||||||
+3
-2
@@ -477,11 +477,12 @@
|
|||||||
|
|
||||||
<script>
|
<script>
|
||||||
const baseUrl = window.location.origin;
|
const baseUrl = window.location.origin;
|
||||||
const manifestUrl = `${baseUrl}/manifest.json`;
|
const apiPrefix = window.API_PREFIX || '';
|
||||||
|
const manifestUrl = `${baseUrl}${apiPrefix}/manifest.json`;
|
||||||
|
|
||||||
// Set URLs
|
// Set URLs
|
||||||
document.getElementById('manifestUrl').textContent = manifestUrl;
|
document.getElementById('manifestUrl').textContent = manifestUrl;
|
||||||
document.getElementById('installLink').href = `stremio://${window.location.host}/manifest.json`;
|
document.getElementById('installLink').href = `stremio://${window.location.host}${apiPrefix}/manifest.json`;
|
||||||
|
|
||||||
function copyUrl() {
|
function copyUrl() {
|
||||||
navigator.clipboard.writeText(manifestUrl).then(() => {
|
navigator.clipboard.writeText(manifestUrl).then(() => {
|
||||||
|
|||||||
Reference in New Issue
Block a user