"""
Stremio addon that extracts HLS stream URLs from fmovies.co (primary, via netoda.tech)
with a vsembed.ru/CloudStream Pro fallback (currently broken — Cloudflare Turnstile).
The fmovies path runs node + Playwright via subprocess to handle the JS-driven extraction.
"""
import bisect
import calendar
import contextlib
import html
import json
import logging
import os
import random
import re
import shutil
import subprocess
import tempfile
from concurrent.futures import ThreadPoolExecutor
import threading
import time
import unicodedata
from dataclasses import dataclass, field
from typing import Optional
from urllib.parse import urlparse, unquote, urljoin, parse_qs, quote
import requests
from flask import Flask, jsonify, request, redirect, abort, Response, send_file
from werkzeug.exceptions import HTTPException
NODE_HELPER = os.path.join(os.path.dirname(os.path.abspath(__file__)),
"node_helper", "fmovies_resolve.mjs")
NODE_BROWSE = os.path.join(os.path.dirname(os.path.abspath(__file__)),
"node_helper", "fmovies_browse.mjs")
RESOLVER_TIMEOUT = 60 # seconds — Playwright launch + extraction
BROWSE_TIMEOUT = 40 # seconds — fmovies search / meta page
FMOVIES_BASE = "https://fmoviess.org"
logging.basicConfig(level=logging.INFO, format="%(asctime)s %(levelname)s %(message)s")
log = logging.getLogger("vsembed")
VSEMBED_BASE = "https://vsembed.ru"
UA = (
"Mozilla/5.0 (X11; Linux x86_64) AppleWebKit/537.36 "
"(KHTML, like Gecko) Chrome/120.0.0.0 Safari/537.36"
)
REQUEST_TIMEOUT = 15
CACHE_TTL = 30 * 60 # 30 min — m3u8 URLs are signed and expire
@dataclass
class Stream:
url: str
quality: str # "1080p" etc
name: str # "fmovies" / "CloudStream Pro"
referer: str = "https://netoda.tech/"
title_extra: str = ""
def _random_fwd_ip() -> str:
return (
f"{random.randint(1, 223)}.{random.randint(0, 255)}."
f"{random.randint(0, 255)}.{random.randint(0, 255)}"
)
def _http_get(url: str, referer: Optional[str] = None) -> str:
headers = {
"User-Agent": UA,
"Accept": "text/html,application/xhtml+xml,application/xml",
"X-Forwarded-For": _random_fwd_ip(),
}
if referer:
headers["Referer"] = referer
r = requests.get(url, headers=headers, timeout=REQUEST_TIMEOUT, allow_redirects=True)
r.raise_for_status()
return r.text
def extract_vsembed(imdb_id: str, tv: bool = False, season: int = 0, episode: int = 0) -> list[Stream]:
"""Run the 3-step WebStreamr-style extraction against vsembed.ru. Maps to
fmovies' "Server 2" (srv-2) embed path → cloudnestra player → m3u8."""
path = (
f"/embed/tv/{imdb_id}/{season}/{episode}"
if tv
else f"/embed/movie/{imdb_id}/"
)
embed_url = VSEMBED_BASE + path
log.info("extract: %s", embed_url)
html = _http_get(embed_url)
html = html.replace("<!--", "").replace("-->", "")
server_match = re.search(
r'<div class="server"\s*data-hash="([^"]+)"[^>]*>\s*CloudStream Pro\s*</div>',
html,
)
if not server_match:
raise RuntimeError("CloudStream Pro server not found in embed page")
data_hash = server_match.group(1)
iframe_match = re.search(r'id="player_iframe"[^>]*src="([^"]+)"', html)
if not iframe_match:
raise RuntimeError("player_iframe not found")
iframe_src = iframe_match.group(1)
if iframe_src.startswith("//"):
iframe_src = "https:" + iframe_src
iframe_origin = f"https://{urlparse(iframe_src).netloc}"
rcp_url = f"{iframe_origin}/rcp/{data_hash}"
rcp_html = _http_get(rcp_url, referer=VSEMBED_BASE + "/")
src_match = re.search(r"src:\s*'([^']+)'", rcp_html)
if not src_match:
raise RuntimeError("no src in rcp response")
player_path = src_match.group(1)
player_url = player_path if player_path.startswith("http") else iframe_origin + player_path
player_html = _http_get(player_url, referer=rcp_url)
file_match = re.search(r"(https://[^']*?\{v\d+\}[^']*?) or", player_html)
if not file_match:
file_match = re.search(r'"(https://[^"]*?\{v\d+\}[^"]*?)"', player_html)
if not file_match:
raise RuntimeError("no m3u8 URL in player page")
host = urlparse(iframe_origin).netloc
m3u8_url = re.sub(r"\{v\d+\}", host, file_match.group(1))
return [Stream(url=m3u8_url, quality="auto (up to 1080p)", name="CloudStream Pro",
referer="https://cloudnestra.com/")]
def _resolve_with(resolver_args: list[str], label: str) -> list[Stream]:
"""Run the node+Playwright resolver and parse its JSON stream list."""
log.info("fmovies resolve: %s", label)
try:
proc = subprocess.run(["node", NODE_HELPER, *resolver_args],
capture_output=True, text=True, timeout=RESOLVER_TIMEOUT)
except subprocess.TimeoutExpired:
raise RuntimeError(f"fmovies resolver timed out after {RESOLVER_TIMEOUT}s")
if proc.returncode != 0 and not proc.stdout.strip():
raise RuntimeError(f"fmovies resolver exit {proc.returncode}: {proc.stderr.strip()[:300]}")
try:
out = json.loads(proc.stdout.strip().splitlines()[-1])
except Exception as exc:
raise RuntimeError(f"fmovies resolver bad JSON: {exc}; stdout={proc.stdout[:300]}")
if "error" in out:
log.warning("fmovies: %s", out["error"])
return []
streams = []
for s in out.get("streams", []):
streams.append(Stream(
url=s["url"],
quality=s.get("quality", "1080p"),
name=s.get("name", "fmovies"),
referer=s.get("referer") or "https://netoda.tech/",
title_extra=s.get("title", ""),
))
return streams
def extract_fmovies(imdb_id: str, tv: bool = False, season: int = 0, episode: int = 0) -> list[Stream]:
"""Resolve via fmovies.co + netoda.tech, found by searching for the cinemeta title."""
args = ["series" if tv else "movie", imdb_id]
if tv:
args += [str(season), str(episode)]
return _resolve_with(args, " ".join(args))
def extract_fmovies_url(film_url: str, tv: bool = False, season: int = 0, episode: int = 0) -> list[Stream]:
"""Resolve a KNOWN fmovies film page directly (fmv: catalog path — no search/cinemeta)."""
args = ["url", film_url, "series" if tv else "movie"]
if tv:
args += [str(season), str(episode)]
return _resolve_with(args, "url " + film_url)
def _run_browse(args: list[str]):
"""Run the fmovies browse helper (search / meta) and return its parsed JSON."""
try:
proc = subprocess.run(["node", NODE_BROWSE, *args],
capture_output=True, text=True, timeout=BROWSE_TIMEOUT)
except subprocess.TimeoutExpired:
raise RuntimeError(f"fmovies browse timed out after {BROWSE_TIMEOUT}s")
out = proc.stdout.strip()
if not out:
raise RuntimeError(f"fmovies browse no output (exit {proc.returncode}): {proc.stderr.strip()[:200]}")
data = json.loads(out.splitlines()[-1])
if "error" in data:
raise RuntimeError(data["error"])
return data
_cache: dict[str, tuple[float, list[Stream]]] = {}
# ---------------------------- source circuit breaker ------------------------
# These extractors are scrapers on the wrong side of an arms race: when a source
# rotates, EVERY call still costs its full timeout (fmovies ~27s + vsembed ~4s)
# and returns nothing. That is pure latency on the /stream hot path — and Stremio
# clients give up long before it: Jacktook's HTTP client timed out
# ("HTTPConnectionPool ... Read timed out") on a 31s response, so the user never
# even saw the lazy vidcore row that WOULD have played.
#
# So trip a breaker per source: after N consecutive failures skip it entirely,
# then let a single probe through once the cooldown expires. Self-healing — when
# a source comes back, the next probe closes the breaker and rows return.
# State is PERSISTED: an in-memory breaker resets on every restart, which would
# hand the user another full-timeout request each time. Cooldown is deliberately
# long — these rotations last days-to-weeks, and every probe costs a real ~31s
# on someone's hot path, so re-probing often is worse than being briefly stale.
SOURCE_FAIL_THRESHOLD = 2
SOURCE_COOLDOWN = 6 * 3600
SOURCE_STATE_FILE = os.path.join(os.path.dirname(os.path.abspath(__file__)), "source_breaker.json")
try:
with open(SOURCE_STATE_FILE) as _f:
_source_state: dict[str, dict] = json.load(_f)
except Exception:
_source_state = {}
def _source_save() -> None:
try:
with open(SOURCE_STATE_FILE, "w") as f:
json.dump(_source_state, f)
except Exception as exc:
log.warning("breaker save failed: %s", exc)
def _source_should_try(name: str) -> bool:
st = _source_state.get(name)
if not st or st["fails"] < SOURCE_FAIL_THRESHOLD:
return True
if time.time() - st["last"] >= SOURCE_COOLDOWN:
log.info("source %s: cooldown expired, probing", name)
return True # single probe
return False
def _source_record(name: str, ok: bool) -> None:
st = _source_state.setdefault(name, {"fails": 0, "last": 0.0})
if ok:
if st["fails"] >= SOURCE_FAIL_THRESHOLD:
log.info("source %s: recovered, breaker closed", name)
st["fails"] = 0
else:
st["fails"] += 1
if st["fails"] == SOURCE_FAIL_THRESHOLD:
log.warning("source %s: %d consecutive failures, skipping for %ds",
name, st["fails"], SOURCE_COOLDOWN)
st["last"] = time.time()
_source_save()
def _try_source(name: str, fn, *args, **kwargs) -> list[Stream]:
if not _source_should_try(name):
log.info("source %s: breaker open, skipped", name)
return []
try:
out = fn(*args, **kwargs)
except Exception as exc:
log.warning("%s extract failed: %s", name, exc)
_source_record(name, False)
return []
_source_record(name, bool(out)) # ran but empty == still a failure
return out
# ------------------------------- source registry ----------------------------
# Sources used to be hardcoded: extract_streams named fmovies and vsembed
# inline, and vidcore was bolted on beside them as a special case. There was no
# shape into which a fourth could go, and nowhere to record what a source can
# actually do.
#
# Three properties matter. The third is the one that cost a day to learn.
@dataclass(frozen=True)
class Source:
id: str
name: str # display label on the row
container: str # "hls-fmp4" | "progressive-mp4"
ingestable: bool # can a PLAIN HTTP CLIENT fetch the bytes?
browser: bool # resolving needs xvfb+patchright (~11s)
heights: tuple = () # qualities offered as rows
resolve: Optional[object] = None
enabled: bool = True
note: str = ""
# How a source knows it has the RIGHT title. "tmdb": it is asked by id, so
# identity is asserted. "filename": it answers a search and we match the
# file's name, so identity is inferred. The library prefers the former.
identity: str = "tmdb"
SOURCES: dict[str, Source] = {}
# A progressive file seeks natively; HLS on this stack does not (that is the
# whole subject of #4), so progressive sorts first.
_CONTAINER_RANK = {"progressive-mp4": 0, "progressive-mkv": 0, "hls-fmp4": 1}
def register_source(src: Source) -> None:
if src.id in SOURCES:
# Two sources sharing an id would make row order and breaker state
# silently ambiguous — exactly the kind of thing that is discovered
# months later as "sometimes the rows come out in a different order".
raise ValueError(f"duplicate source id: {src.id}")
SOURCES[src.id] = src
def get_source(sid: str) -> Optional[Source]:
return SOURCES.get(sid)
def ordered_sources() -> list[Source]:
"""Enabled sources, best container first. `sorted` is stable, so sources
within a container class keep registration order."""
return sorted((s for s in SOURCES.values() if s.enabled),
key=lambda s: _CONTAINER_RANK.get(s.container, 9))
def ingest_source() -> Optional[Source]:
"""The source the library should be fed from, or None.
`ingestable` is a veto that ranking cannot override, and that is the entire
point of this function. vidlink sorts first on every visible merit — a
progressive MP4, resolved in 3s, with exact byte sizes — and ffmpeg cannot
fetch a single byte of it (measured 2026-08-11: 428/403/429 across six
client/URL combinations, including byte-exact browser headers). The
local-first design rests on ffmpeg pulling bytes independently of any
player, so a browser-only source can never feed the library however good its
API looks.
"""
return next(iter(ingest_sources()), None)
def ingest_sources() -> list[Source]:
"""Every source that may feed the library, in the order to try them.
Not the playback order. For playback a progressive file wins because it
seeks; for the LIBRARY the question is whether the file is the right title,
because a wrongly-named copy is permanent. So a source asked by TMDB id goes
before one whose answer we matched by filename, whatever its container.
"""
return sorted((s for s in SOURCES.values() if s.enabled and s.ingestable),
key=lambda s: 0 if s.identity == "tmdb" else 1)
def extract_streams(imdb_id: str, tv: bool = False, season: int = 0, episode: int = 0) -> list[Stream]:
"""Gather streams from both fmovies (srv-1, netoda/voxzer direct) and vsembed
(srv-2, cloudnestra embed). Return both so Jacktook shows two source rows —
when one path's CDN is broken (e.g. voxzer 404), the other usually plays.
Each is behind the circuit breaker above, so a dead source costs ~nothing."""
streams = []
streams.extend(_try_source("fmovies", extract_fmovies, imdb_id, tv=tv, season=season, episode=episode))
streams.extend(_try_source("vsembed", extract_vsembed, imdb_id, tv=tv, season=season, episode=episode))
return streams
def cached_extract(key: str, *args, **kwargs) -> list[Stream]:
now = time.time()
entry = _cache.get(key)
if entry and entry[0] > now:
log.info("cache hit: %s", key)
return entry[1]
streams = extract_streams(*args, **kwargs)
if streams:
_cache[key] = (now + CACHE_TTL, streams)
return streams
def cached_extract_url(key: str, film_url: str, tv: bool = False, season: int = 0, episode: int = 0) -> list[Stream]:
"""fmv: path — resolve a known fmovies film page (fmovies-only, no vsembed fallback)."""
now = time.time()
entry = _cache.get(key)
if entry and entry[0] > now:
log.info("cache hit: %s", key)
return entry[1]
streams = _try_source("fmovies", extract_fmovies_url, film_url, tv=tv, season=season, episode=episode)
if streams:
_cache[key] = (now + CACHE_TTL, streams)
return streams
# Browse (catalog/meta) cache — keep Playwright off the hot path during Kodi redraws.
_browse_cache: dict[str, tuple[float, dict]] = {}
BROWSE_TTL = 10 * 60
def cached_browse(args: list[str]) -> dict:
key = "|".join(args)
now = time.time()
entry = _browse_cache.get(key)
if entry and entry[0] > now:
return entry[1]
data = _run_browse(args)
_browse_cache[key] = (now + BROWSE_TTL, data)
return data
def _fmv_film_url(mid: str, slug: str) -> str:
return f"{FMOVIES_BASE}/film/{slug}-{mid}/"
# ------------------------------- TMDB (flixlab) -----------------------------
# vidcore/flixlab is TMDB-keyed, so the flixlab-native catalog is a TMDB browse:
# search -> fxl:<tmdb> ids -> meta (seasons/episodes from TMDB) -> stream resolved
# by driving vidcore.net directly. No scraping and no browser on the browse path.
TMDB_KEY = "b70756b7083d9ee60f849d82d94a0d80" # same public key the resolvers use
TMDB_BASE = "https://api.themoviedb.org/3"
TMDB_IMG = "https://image.tmdb.org/t/p/w342"
TMDB_TTL = 60 * 60
_tmdb_cache: dict[str, tuple[float, dict]] = {}
def _tmdb_get(path: str, params: Optional[dict] = None) -> dict:
key = path + "|" + json.dumps(params or {}, sort_keys=True)
now = time.time()
entry = _tmdb_cache.get(key)
if entry and entry[0] > now:
return entry[1]
q = {"api_key": TMDB_KEY}
q.update(params or {})
r = requests.get(f"{TMDB_BASE}/{path}", params=q,
headers={"User-Agent": UA}, timeout=REQUEST_TIMEOUT)
r.raise_for_status()
data = r.json()
_tmdb_cache[key] = (now + TMDB_TTL, data)
return data
def _tmdb_poster(p: Optional[str]) -> str:
return TMDB_IMG + p if p else ""
def _tmdb_metas(ctype: str, results: list[dict]) -> list[dict]:
"""TMDB result rows -> Stremio metas with fxl: ids."""
metas = []
for r in results[:40]:
if not r.get("id"):
continue
name = r.get("name") or r.get("title") or ""
date = r.get("first_air_date") or r.get("release_date") or ""
metas.append({
"id": f"fxl:{r['id']}",
"type": ctype,
"name": f"{name} ({date[:4]})" if date[:4] else name,
"poster": _tmdb_poster(r.get("poster_path")),
"description": r.get("overview") or "",
})
return metas
def _tmdb_search(ctype: str, query: str) -> list[dict]:
"""Search TMDB -> Stremio metas with fxl: ids."""
kind = "tv" if ctype == "series" else "movie"
data = _tmdb_get(f"search/{kind}", {"query": query, "include_adult": "false"})
return _tmdb_metas(ctype, data.get("results", []))
# Browse catalogs. vidcore is TMDB-keyed, so discovery is a plain TMDB list --
# no scraping and no browser on the browse path, and every row is playable by
# construction. These exist because the addon previously exposed search-only
# catalogs, and `stremio_only_catalogs` hides Jacktook's native TMDB browse:
# between them there was no way to discover anything without typing a title.
# "latest" uses discover rather than a ready-made list because TMDB has none:
# sorting by release date alone surfaces unreleased placeholders and entries with
# no audience at all, so it is bounded to titles already out (<= today) with a
# small vote floor -- high enough to drop junk, low enough that genuinely new
# releases still appear.
def _today() -> str:
return time.strftime("%Y-%m-%d")
BROWSE_ENDPOINTS = {
"trending": {
"series": ("trending/tv/week", {}),
"movie": ("trending/movie/week", {}),
},
"popular": {
"series": ("tv/popular", {}),
"movie": ("movie/popular", {}),
},
"top": {
"series": ("tv/top_rated", {}),
"movie": ("movie/top_rated", {}),
},
"latest": {
"series": ("discover/tv", {"sort_by": "first_air_date.desc",
"vote_count.gte": "10"}),
"movie": ("discover/movie", {"sort_by": "primary_release_date.desc",
"vote_count.gte": "30"}),
},
# What is on right now: in cinemas / currently airing.
"now": {
"series": ("tv/on_the_air", {}),
"movie": ("movie/now_playing", {}),
},
}
def _tmdb_browse(ctype: str, kind: str, page: int = 1) -> list[dict]:
"""One page of a TMDB browse list -> Stremio metas."""
endpoint, extra = BROWSE_ENDPOINTS[kind][ctype]
params = {"page": str(max(1, page))}
params.update(extra)
if kind == "latest": # exclude anything not actually released yet
params["first_air_date.lte" if ctype == "series"
else "primary_release_date.lte"] = _today()
data = _tmdb_get(endpoint, params)
return _tmdb_metas(ctype, data.get("results", []))
# --- genre browse -----------------------------------------------------------
# Stremio passes a genre by NAME in the `genre` extra, but TMDB discover filters
# by numeric id, so the name->id map is fetched once per type and cached.
_genre_cache: dict[str, tuple[float, dict]] = {}
GENRE_TTL = 24 * 3600
def _tmdb_genres(ctype: str) -> dict:
"""{genre name: tmdb id} for a type, cached for a day."""
kind = "tv" if ctype == "series" else "movie"
hit = _genre_cache.get(kind)
now = time.time()
if hit and hit[0] > now:
return hit[1]
data = _tmdb_get(f"genre/{kind}/list")
mapping = {g["name"]: g["id"] for g in data.get("genres", []) if g.get("name")}
if mapping:
_genre_cache[kind] = (now + GENRE_TTL, mapping)
return mapping
def _tmdb_by_genre(ctype: str, genre: str, page: int = 1) -> list[dict]:
"""Discover a type filtered to one genre, most popular first."""
gid = _tmdb_genres(ctype).get(genre)
if not gid:
log.warning("unknown genre %r for %s", genre, ctype)
return []
endpoint = "discover/tv" if ctype == "series" else "discover/movie"
data = _tmdb_get(endpoint, {
"with_genres": str(gid),
"sort_by": "popularity.desc",
"page": str(max(1, page)),
"vote_count.gte": "50",
})
return _tmdb_metas(ctype, data.get("results", []))
def _tmdb_series_videos(tmdb: str) -> tuple[dict, list[dict]]:
"""Show details + every episode across all seasons, as Stremio `videos`.
Jacktook builds its season/episode menu straight from `videos[]`, so listing
all seasons here is what makes S2/S3/... browsable (the fmovies catalog can
only ever offer one season per entry — fmovies pages each season separately).
"""
show = _tmdb_get(f"tv/{tmdb}")
seasons = [s.get("season_number") for s in show.get("seasons", [])
if (s.get("season_number") or 0) >= 1]
videos: list[dict] = []
if seasons:
wanted = seasons[:20] # append_to_response caps at 20 sub-requests
full = _tmdb_get(f"tv/{tmdb}", {
"append_to_response": ",".join(f"season/{n}" for n in wanted)})
for n in wanted:
for e in (full.get(f"season/{n}") or {}).get("episodes", []):
ep = e.get("episode_number")
if not ep:
continue
videos.append({
"id": f"fxl:{tmdb}:{n}:{ep}",
"title": e.get("name") or f"Episode {ep}",
"season": n,
"episode": ep,
"released": (e.get("air_date") or "") and e["air_date"] + "T00:00:00.000Z",
"overview": e.get("overview") or "",
"thumbnail": _tmdb_poster(e.get("still_path")),
})
return show, videos
def _tmdb_from_title(ctype: str, title: str) -> Optional[str]:
"""Best-effort title -> TMDB id, for fmovies (fmv:) entries which carry no id."""
try:
data = _tmdb_get(f"search/{'tv' if ctype == 'series' else 'movie'}",
{"query": title, "include_adult": "false"})
except Exception as exc:
log.warning("tmdb title search failed for %r: %s", title, exc)
return None
results = data.get("results") or []
return str(results[0]["id"]) if results else None
def _fmv_slug_to_title(slug: str) -> tuple[str, int]:
"""`x-men-the-animated-series-season-3` -> ("x men the animated series", 3)."""
m = re.search(r"-season-(\d+)$", slug or "")
season = int(m.group(1)) if m else 0
base = re.sub(r"-season-\d+$", "", slug or "")
return base.replace("-", " ").strip(), season
# ----------------------------- Stremio addon API ----------------------------
app = Flask(__name__)
# TMDB's own genre names, per type (they differ). Declared statically so the
# manifest never depends on a live TMDB call at startup; _tmdb_genres() maps
# these names back to ids when a genre is actually browsed.
MOVIE_GENRES = [
"Action", "Adventure", "Animation", "Comedy", "Crime", "Documentary",
"Drama", "Family", "Fantasy", "History", "Horror", "Music", "Mystery",
"Romance", "Science Fiction", "Thriller", "War", "Western",
]
TV_GENRES = [
"Action & Adventure", "Animation", "Comedy", "Crime", "Documentary",
"Drama", "Family", "Kids", "Mystery", "Reality", "Sci-Fi & Fantasy",
"Soap", "War & Politics", "Western",
]
MANIFEST = {
"id": "org.tabertron.vsembed",
"version": "0.34.0",
"name": "Tabertron VsEmbed",
"description": "Self-hosted extractor with TMDB browse + search, played via vidcore and HdHub",
"types": ["movie", "series"],
# stream ids: tt (cinemeta-driven), fmv (fmovies-native catalog),
# fxl (flixlab/vidcore-native catalog, TMDB-keyed).
# meta/catalog: native browse so Stremio clients (Jacktook) search each source
# directly and see exactly what IT can play, bypassing TMDB/cinemeta gaps.
"resources": ["stream", "meta", "catalog"],
"idPrefixes": ["tt", "fmv", "fxl"],
# The fmovies catalogs are gone: fmovies and vsembed both died to Cloudflare
# Turnstile on 2026-07-27 and their circuit breakers sit permanently open, so
# those rows could only ever fail. `fmv:` ids stay supported for anything
# already cached client-side; they just aren't advertised for browsing.
#
# Jacktook renders a search catalog as BOTH "Search <name>" and "<name>", so
# the name must read correctly with and without that prefix. "flixlab Search"
# gave "Search flixlab Search"; plain "Search" gave "Search Search". "Titles"
# reads as "Search Titles" / "Titles".
"catalogs": [
{"type": "series", "id": "flixlab-trending-series", "name": "Trending",
"extra": [{"name": "skip"}]},
{"type": "movie", "id": "flixlab-trending-movie", "name": "Trending",
"extra": [{"name": "skip"}]},
{"type": "series", "id": "flixlab-popular-series", "name": "Popular",
"extra": [{"name": "skip"}]},
{"type": "movie", "id": "flixlab-popular-movie", "name": "Popular",
"extra": [{"name": "skip"}]},
{"type": "series", "id": "flixlab-top-series", "name": "Top Rated",
"extra": [{"name": "skip"}]},
{"type": "movie", "id": "flixlab-top-movie", "name": "Top Rated",
"extra": [{"name": "skip"}]},
{"type": "series", "id": "flixlab-latest-series", "name": "Latest",
"extra": [{"name": "skip"}]},
{"type": "movie", "id": "flixlab-latest-movie", "name": "Latest",
"extra": [{"name": "skip"}]},
{"type": "series", "id": "flixlab-now-series", "name": "On Air",
"extra": [{"name": "skip"}]},
{"type": "movie", "id": "flixlab-now-movie", "name": "In Theaters",
"extra": [{"name": "skip"}]},
# Genre options are declared here so a client can offer a picker instead
# of us shipping one catalog row per genre. Names are TMDB's own and
# differ per type (tv has "Sci-Fi & Fantasy", movie has "Science Fiction"),
# so they are passed straight back to TMDB and mapped to ids server-side.
{"type": "series", "id": "flixlab-genre-series", "name": "Genres",
"extra": [{"name": "genre", "options": TV_GENRES}, {"name": "skip"}]},
{"type": "movie", "id": "flixlab-genre-movie", "name": "Genres",
"extra": [{"name": "genre", "options": MOVIE_GENRES}, {"name": "skip"}]},
{"type": "series", "id": "flixlab-series", "name": "Titles",
"extra": [{"name": "search", "isRequired": True}]},
{"type": "movie", "id": "flixlab-movie", "name": "Titles",
"extra": [{"name": "search", "isRequired": True}]},
],
"behaviorHints": {"configurable": False, "adult": False},
}
@app.after_request
def add_cors(resp):
resp.headers["Access-Control-Allow-Origin"] = "*"
resp.headers["Access-Control-Allow-Headers"] = "*"
return resp
@app.route("/manifest.json")
def manifest():
return jsonify(MANIFEST)
# --------------------------- vidcore (additive, lazy) -----------------------
# vidcore.net (flixlab's PRIMARY server) covers titles the fmovies/vsembed path
# lacks (e.g. TMDB-fragmented shows). It needs a real browser (patchright+xvfb),
# so it is LAZY: /stream returns an extra "vidcore" ROW pointing at /vidcore/...,
# which resolves ON PLAY and 302-redirects to the m3u8. The existing sources are
# completely untouched — this is purely one additional option.
VIDCORE_HELPER = os.path.join(os.path.dirname(os.path.abspath(__file__)),
"node_helper", "vidcore_resolve.mjs")
VIDCORE_TIMEOUT = 140
_vidcore_cache: dict[str, tuple[float, str]] = {}
# One lock per title, so concurrent requests for the SAME stream serialise while
# different titles still resolve in parallel.
_vidcore_locks: dict[str, "threading.Lock"] = {}
_vidcore_locks_guard = threading.Lock()
VIDCORE_TTL = 30 * 60
# Reputation: learn which titles vidcore actually has (recorded on play), so the
# row is HIDDEN for known misses — no more empty ~60s row on titles vidcore lacks
# (e.g. Lestat). Misses are re-probed after MISS_TTL since vidcore's catalog grows.
#
# 2026-08-10: this hid titles that play perfectly. The resolver's single "no url"
# exit meant "vidcore lacks this" and "the poll window expired first" arrived
# identically, and one unlucky cold start cost a title a week. Measured that day:
# X-Men '97 S2E7 missed on a 57s cold resolve then hit 9s later; S2E8 and
# tmdb:287238 S1E5 were both stored as misses while resolving in 11.3s and 6s.
# With fmovies+vsembed breaker-tripped since Jul 28, a hidden row is not a
# degraded experience — it is the title having no sources at all.
#
# Two changes close that. The resolver now reports WHY it found nothing, and only
# a positive finding ("absent") counts; and absence must be observed twice in a
# row before the row is hidden, the same 2-strike shape the source breaker above
# already uses. A hit anywhere in between clears the count.
VIDCORE_REP_FILE = os.path.join(os.path.dirname(os.path.abspath(__file__)), "vidcore_reputation.json")
VIDCORE_MISS_TTL = 7 * 24 * 3600 # re-check a "miss" after 7 days
VIDCORE_MISS_STRIKES = 2 # consecutive absences before hiding the row
try:
with open(VIDCORE_REP_FILE) as _f:
_vidcore_rep: dict = json.load(_f)
except Exception:
_vidcore_rep = {}
def _rep_key(kind: str, imdb: str, season: int = 0, episode: int = 0) -> str:
return f"{kind}:{imdb}:{season}:{episode}"
def _rep_outcome(out: dict) -> Optional[str]:
"""Classify a resolver result: "hit" | "miss" | None.
None means inconclusive — nothing is written, so the title stays retryable.
Anything short of the resolver positively reporting `absent` lands here,
including a reason we don't recognise: mistaking slowness for absence is the
expensive error, so unknown states must fail towards re-probing.
"""
if (out.get("streams") or [{}])[0].get("url"):
return "hit"
return "miss" if out.get("reason") == "absent" else None
def _rep_record(key: str, status: str) -> None: # status: "hit" | "miss"
prev = _vidcore_rep.get(key) or {}
strikes = int(prev.get("strikes", 0))
# Strikes must be CONSECUTIVE and recent: two absences a fortnight apart say
# nothing about a catalog that grows, so an entry past its TTL starts over.
if time.time() - prev.get("ts", 0) >= VIDCORE_MISS_TTL:
strikes = 0
strikes = strikes + 1 if status == "miss" else 0
_vidcore_rep[key] = {"status": status, "ts": time.time(), "strikes": strikes}
try:
with open(VIDCORE_REP_FILE, "w") as f:
json.dump(_vidcore_rep, f)
except Exception as exc:
log.warning("vidcore rep save failed: %s", exc)
def _rep_is_known_miss(key: str) -> bool:
r = _vidcore_rep.get(key)
if not r or r.get("status") != "miss":
return False
if time.time() - r.get("ts", 0) >= VIDCORE_MISS_TTL:
return False
# Entries written before the fix carry no strike count and cannot be told
# apart from the false ones — two of the three checked on 2026-08-10 were
# false. Re-probe them (one ~20s resolve) rather than keep hiding a title
# that plays; they re-file correctly under the rules above.
return int(r.get("strikes", 0)) >= VIDCORE_MISS_STRIKES
# Serve a single VARIANT playlist, never vidcore's MASTER. Two reasons, both
# measured on the Pi (2026-07-28, X-Men '97 S2E5):
# 1. PAUSE. The master carries no EXT-X-PLAYLIST-TYPE:VOD and no EXT-X-ENDLIST,
# so Kodi treats it as a LIVE stream and refuses Player.PlayPause
# ("Failed to execute method") while still honouring Stop — exactly the
# "remote pause does nothing" symptom. The variant has both markers and
# pauses correctly.
# 2. DECODE. Jacktook never sets inputstream.adaptive (its only ISA call is the
# mpd trailer path), so HLS lands on Kodi's bare ffmpeg demuxer, which picks
# the HIGHEST-bandwidth variant — vidcore lists 2160p @ ~15.8Mbps first, and
# the Pi's V4L2 mem2mem H.264 decoder cannot decode 4K.
VIDCORE_MAX_HEIGHT = 1080
# The qualities offered as rows, best first. Declared rather than read from the
# playlist because the row list is built on /stream, which must stay fast — the
# master is only known after a resolve (seconds, a browser). Asking for a height
# a title does not carry is harmless: _cap_variant picks the best at or below it
# and never returns nothing. 2160p is deliberately absent — it is listed first by
# vidcore at ~15.8Mbps and the Pi's V4L2 mem2mem H.264 decoder cannot decode it.
VIDCORE_HEIGHTS = (1080, 720, 480)
# The adaptive row. Offered FIRST because it is the one that puts quality
# selection in the player OSD (inputstream.adaptive exposes a video-stream
# picker; the bare ffmpeg demuxer does not), which is where choosing it is
# actually useful. The fixed-height rows stay below it as a fallback: they are
# single variants, so they play even if ISA fails to engage on the client.
VIDCORE_AUTO = "auto"
# Parsed variant lists, keyed by master URL. Three quality rows for one title
# would otherwise re-fetch and re-parse the same playlist three times, on the
# play path, inside the ~30s budget Kodi's demuxer allows.
_variant_cache: dict[str, tuple[float, list[tuple[int, int, str, str]]]] = {}
def _variants(master_url: str) -> list[tuple[int, int, str, str]]:
"""[(height, bandwidth, uri, stream_inf_line)] from a master, best first.
The raw #EXT-X-STREAM-INF line is kept so a master can be rebuilt for
inputstream.adaptive without re-deriving attributes we were handed.
Empty means "do not rewrite this URL" — either it is already a media
playlist, or the fetch failed. Only a successful parse is cached: caching a
failure would keep a title degraded for the full TTL over one bad request.
"""
hit = _variant_cache.get(master_url)
if hit and hit[0] > time.time():
return hit[1]
try:
# Deliberately NOT _http_get: that injects a random X-Forwarded-For, and
# these m3u8 tokens are IP-CIDR locked to the LAN's egress, so a spoofed
# forwarding header risks a 403 on a URL we already know is good.
r = requests.get(master_url, timeout=REQUEST_TIMEOUT,
headers={"User-Agent": UA, "Referer": "https://vidcore.net/"})
r.raise_for_status()
body = r.text
except Exception as exc:
log.warning("vidcore cap: master fetch failed (%s) — serving master", exc)
return []
out: list[tuple[int, int, str, str]] = []
lines = body.splitlines()
for i, line in enumerate(lines):
# NB: #EXT-X-I-FRAME-STREAM-INF carries its URI inline and must not match.
if not line.startswith("#EXT-X-STREAM-INF"):
continue
uri = next((s for s in (ln.strip() for ln in lines[i + 1:])
if s and not s.startswith("#")), None)
if not uri:
continue
rm = re.search(r"RESOLUTION=\d+x(\d+)", line)
bm = re.search(r"BANDWIDTH=(\d+)", line)
out.append((int(rm.group(1)) if rm else 0,
int(bm.group(1)) if bm else 0, uri, line.strip()))
out.sort(reverse=True)
_variant_cache[master_url] = (time.time() + VIDCORE_TTL, out)
return out
def _cap_variant(master_url: str, max_height: int = VIDCORE_MAX_HEIGHT) -> str:
"""Resolve a master playlist to its best variant at or below `max_height`.
Returns `master_url` unchanged if it is already a media playlist or if
anything at all goes wrong — capping is an optimisation and must never
become a new way for a working stream to fail.
"""
variants = _variants(master_url)
if not variants:
return master_url
best = next((v for v in variants if v[0] <= max_height), None)
if best is None:
# Nothing that small exists. Serve the LOWEST variant rather than the
# master: the master carries no VOD/ENDLIST markers, so handing it back
# makes Kodi treat the stream as live and refuse to pause — the bug this
# function exists to prevent, sneaking back in through the fallback.
best = variants[-1]
log.info("vidcore cap: no variant <= %dp — serving lowest (%dp)",
max_height, best[0])
else:
log.info("vidcore cap: serving %dp variant", best[0])
return urljoin(master_url, best[2])
# Resolving takes 7-16s and occasionally much longer, essentially all of it
# vidcore's player rather than our startup (measured 2026-08-10: launch ~0.7s,
# page load ~1.2s, m3u8 5.3s-15s+). On the PLAY path that lands inside Kodi's
# ~30s demuxer budget, and overrunning it kills playback outright — on
# 2026-08-10 the addon returned a good stream 16s after Kodi had given up.
#
# So start resolving when the source list is requested instead. Opening an
# episode's sources is a strong intent signal and buys the whole gap between
# looking and pressing play. If they play before it finishes, the per-title lock
# means the play request joins the resolve already in flight rather than
# starting a second one.
#
# Bounded to one at a time: browsing quickly must not spawn a browser per
# episode looked at. A skipped prefetch costs nothing — the play path still
# resolves exactly as before.
VIDCORE_PREFETCH_MAX = 1
_prefetch_sem = threading.Semaphore(VIDCORE_PREFETCH_MAX)
# How long a PLAY request may queue behind a resolve already in flight.
#
# The prefetch above shares a per-title lock with the play path, so until this
# existed, pressing play meant waiting out whatever the prefetch was doing --
# with no bound short of VIDCORE_TIMEOUT. Measured 2026-08-12 on Supergirl
# (tmdb:1081003): /stream started a prefetch that ran COLD (launch=28866ms after
# 25h idle, against 595-760ms on every warm resolve) and took 57.7s; the play
# request arrived 4s later, queued for all of it, then paid its own 11.4s resolve
# on top and answered at 68s. Kodi allows an HTTP open 20s -- CCurlFile::Stat and
# CurlFile::Open each timed out at exactly that -- so it had given up 2.2s before
# the answer arrived. The prefetch, whose entire purpose is to take resolving off
# the play path, had put it back on and made that play slower than no prefetch at
# all (11.4s, which would have worked).
#
# 8s leaves room to inherit a resolve that is nearly done, while leaving over half
# of Kodi's budget for the redirect and the CDN's own first bytes. Overrunning it
# is not a slower play, it is a failed one.
VIDCORE_PLAY_DEADLINE = 8
def _prefetch_vidcore(kind: str, ident: str, season: int = 0,
episode: int = 0) -> Optional[threading.Thread]:
"""Warm the cache in the background. Returns the thread, or None if skipped."""
key = f"{kind}:{ident}:{season}:{episode}"
hit = _vidcore_cache.get(key)
if hit and hit[0] > time.time():
return None
if not _prefetch_sem.acquire(blocking=False):
log.info("vidcore prefetch: busy, skipping %s", key)
return None
def run() -> None:
try:
_resolve_vidcore(kind, ident, season, episode)
except Exception as exc:
log.warning("vidcore prefetch failed for %s: %s", key, exc)
finally:
_prefetch_sem.release()
t = threading.Thread(target=run, name=f"prefetch-{key}", daemon=True)
t.start()
return t
def _resolve_vidcore(kind: str, ident: str, season: int = 0, episode: int = 0,
wait: Optional[float] = None) -> Optional[str]:
"""`ident` is `tt<n>` (cinemeta path) or `tmdb:<n>` (fxl:/fmv: catalog paths).
`wait` bounds how long to queue behind a resolve someone else already
started. None means wait as long as it takes, which is right for anything
running in the background and wrong for anything a player is waiting on.
"""
key = f"{kind}:{ident}:{season}:{episode}"
now = time.time()
hit = _vidcore_cache.get(key)
if hit and hit[0] > now:
return hit[1]
# Kodi issues GET + HEAD + GET for a single stream url, so without this each
# play launched three browsers for the same title. They then collided on the
# resolver's Chromium profile and all but one died instantly. The profile is
# now per-run, but launching three browsers to answer one question is still
# pure waste: hold a per-title lock so the followers wait and take the cache.
with _vidcore_locks_guard:
lock = _vidcore_locks.setdefault(key, threading.Lock())
if not (lock.acquire() if wait is None else lock.acquire(timeout=wait)):
# Someone else is still working. Take a cached answer if one appeared
# while we waited, otherwise give up: their resolve carries on and fills
# the cache, so the cost of stopping here is one click, not the title.
log.info("vidcore: %s in flight past %ss — not waiting", key, wait)
hit = _vidcore_cache.get(key)
return hit[1] if hit and hit[0] > time.time() else None
try:
hit = _vidcore_cache.get(key) # a waiter may have filled it
if hit and hit[0] > time.time():
return hit[1]
return _resolve_vidcore_locked(key, kind, ident, season, episode)
finally:
lock.release()
def _master_playlist(master_url: str,
max_height: int = VIDCORE_MAX_HEIGHT) -> Optional[str]:
"""Rebuild vidcore's master with everything above `max_height` removed.
This is the manifest for inputstream.adaptive. Kodi only offers a
video-stream picker in the player OSD when ISA is driving the stream, and
ISA needs a ladder rather than the single variant the fixed-height rows
serve — so this is what puts quality selection *during* playback.
2160p is stripped rather than merely not-preferred: an adaptive client on a
fast link climbs to the top of whatever ladder it is given, and vidcore
lists 2160p first at ~15.8Mbps, which the Pi's V4L2 decoder cannot decode.
URIs are made absolute because we serve this body from our own host, so a
relative URI would resolve against us instead of the CDN. Only the manifest
is rewritten — the video bytes are never proxied.
Returns None if there is no ladder to build (already a media playlist, or
the fetch failed), leaving the caller to fall back to a variant redirect.
"""
variants = _variants(master_url)
if not variants:
return None
keep = [v for v in variants if v[0] <= max_height] or [variants[-1]]
lines = ["#EXTM3U"]
for _height, _bw, uri, inf in keep:
lines.append(inf)
lines.append(urljoin(master_url, uri))
return "\n".join(lines) + "\n"
def _resolve_overhead_ms(out: dict, wall_ms: int) -> Optional[int]:
"""Wall time around the subprocess minus what node says it spent.
The gap is everything OUTSIDE node — xvfb-run allocating a display, process
spawn, interpreter start. It exists because a resolve that should have
aborted at ~24s took 57s on 2026-08-10 and killed a play, and launch is
measurably only ~0.7s of that. Returns None rather than raising: a
diagnostic must never be able to fail a resolve that otherwise worked.
"""
timing = out.get("timing")
if not isinstance(timing, dict):
return None
total = timing.get("total_ms")
if not isinstance(total, (int, float)) or isinstance(total, bool):
return None
return int(wall_ms - total)
def _log_resolve_timing(key: str, out: dict, wall_ms: int) -> None:
t = out.get("timing") if isinstance(out.get("timing"), dict) else {}
log.info("vidcore timing %s: wall=%dms node=%sms (launch=%s goto=%s poll=%s) outside-node=%s",
key, wall_ms, t.get("total_ms"), t.get("launch_ms"),
t.get("goto_ms"), t.get("poll_ms"), _resolve_overhead_ms(out, wall_ms))
def _resolve_vidcore_locked(key: str, kind: str, ident: str,
season: int, episode: int) -> Optional[str]:
now = time.time()
args = ["series" if kind == "series" else "movie", ident]
if kind == "series":
args += [str(season), str(episode)]
log.info("vidcore resolve: %s", " ".join(args))
t_start = time.time()
try:
proc = subprocess.run(["xvfb-run", "-a", "node", VIDCORE_HELPER, *args],
capture_output=True, text=True, timeout=VIDCORE_TIMEOUT,
cwd=os.path.dirname(os.path.dirname(VIDCORE_HELPER)))
except subprocess.TimeoutExpired:
log.warning("vidcore resolve timed out for %s after %dms",
key, int((time.time() - t_start) * 1000))
return None
wall_ms = int((time.time() - t_start) * 1000)
lines = (proc.stdout or "").strip().splitlines()
if not lines:
log.warning("vidcore resolve no output for %s: %s", key, (proc.stderr or "")[-200:])
return None
try:
out = json.loads(lines[-1])
except Exception as exc:
log.warning("vidcore resolve bad JSON for %s: %s", key, exc)
return None
_log_resolve_timing(key, out, wall_ms)
url = (out.get("streams") or [{}])[0].get("url")
outcome = _rep_outcome(out)
if url:
# Cache the MASTER, not a capped variant: one resolve now serves every
# quality row, so switching bitrate is a cache hit rather than another
# browser launch.
_vidcore_cache[key] = (now + VIDCORE_TTL, url)
else:
log.info("vidcore: %s (reason=%s)", out.get("error"), out.get("reason"))
if outcome:
_rep_record(key, outcome)
else:
# Inconclusive — the player came up but we gave up first, or the resolver
# failed on our side. Record nothing: the title stays offered and retryable.
log.info("vidcore: %s inconclusive, not recorded", key)
return url
VIDCORE_IDENT = r"tt\d+|tmdb:\d+"
def _vidcore_row(self_host: str, kind: str, ident: str, season: int = 0,
episode: int = 0, height=VIDCORE_MAX_HEIGHT) -> Stream:
if kind == "series":
path = f"/vidcore/series/{ident}:{season}:{episode}.m3u8"
tag = f"S{season:02d}E{episode:02d} " # Jacktook drops TV results without it
else:
path = f"/vidcore/movie/{ident}.m3u8"
tag = ""
label = "Auto" if height == VIDCORE_AUTO else f"{height}p"
# `adaptive=1` is how the row asks the client for inputstream.adaptive. It
# rides in the URL rather than in behaviorHints so no flag has to be plumbed
# through every playback-data path in the Jacktook fork, and so a row that
# is NOT adaptive (a single variant) can never accidentally opt in.
query = f"h={height}" + ("&adaptive=1" if height == VIDCORE_AUTO else "")
# The height rides in the query string so the route patterns (and the ident
# regexes they enforce) are untouched, and so it survives Kodi appending its
# own `|Referer=..&User-Agent=..` suffix to the URL.
return Stream(url=f"http://{self_host}{path}?{query}", quality=label,
name="vidcore", referer="https://vidcore.net/",
title_extra=f"{tag}{label}\nvidcore (flixlab)")
def _requested_height() -> int:
"""Quality asked for by the row the user picked; defaults to the cap."""
try:
h = int(request.args.get("h", VIDCORE_MAX_HEIGHT))
except (TypeError, ValueError):
return VIDCORE_MAX_HEIGHT
return h if 144 <= h <= 4320 else VIDCORE_MAX_HEIGHT
def _serve_vidcore(url: str):
"""Answer a play request: the ISA ladder for `h=auto`, else one variant.
`h=auto` returns the manifest as a BODY rather than redirecting, because the
ladder has to be filtered (no 2160p) before the client sees it. A redirect
would hand over vidcore's own master, 4K and all.
"""
if request.args.get("h") == "auto":
body = _master_playlist(url)
if body:
return Response(body, mimetype="application/vnd.apple.mpegurl")
log.info("vidcore: no ladder to build — falling back to a variant")
return redirect(_cap_variant(url, _requested_height()), code=302)
def _local_redirect(kind: str, ident: str, season: int, episode: int,
height: int):
"""302 to the local copy when we hold one, else None.
Checked BEFORE resolving, which is the substance of this rather than the
redirect itself: a resolve costs ~11s and launches a browser, and paying
that for a title already sitting on disk is pure waste.
Redirecting rather than serving inline hands the player a URL that actually
ends in .mp4, and routes it through /mp4/ — the single place local files are
served, so Range support and rolling forward are not duplicated here.
"""
if not _playable_path(kind, ident, season, episode, height):
return None
path = (f"/mp4/series/{ident}:{season}:{episode}.mp4" if kind == "series"
else f"/mp4/movie/{ident}.mp4")
log.info("vidcore: serving %s from the local copy", ident)
return redirect(f"{path}?h={height}", code=302)
def _log_client_headers(route: str) -> None:
"""Record which request headers the client actually forwarded.
Kodi only sends these if the player passed them through — Jacktook has to
append them to the URL as `|Referer=..&User-Agent=..` for that to happen.
Logging them is the only way to confirm a client honours our proxyHeaders.
"""
log.info("%s: client Referer=%r UA=%r", route,
request.headers.get("Referer"), request.headers.get("User-Agent"))
@app.route("/vidcore/movie/<path:ident>.m3u8")
def vidcore_movie(ident: str):
_log_client_headers("vidcore/movie")
if not re.fullmatch(VIDCORE_IDENT, ident):
abort(404)
height = _requested_height()
local = _local_redirect("movie", ident, 0, 0, height)
if local:
return local
url = _resolve_vidcore("movie", ident, wait=VIDCORE_PLAY_DEADLINE)
if not url:
abort(404)
# Playing is the honest "I chose this" signal, so keep a copy of what is
# being watched. Idempotent, and skipped when the library already has it.
_ensure_ingest("movie", ident, height=height)
return _serve_vidcore(url)
@app.route("/vidcore/series/<path:ident>.m3u8")
def vidcore_series(ident: str):
_log_client_headers("vidcore/series")
m = re.fullmatch(rf"({VIDCORE_IDENT}):(\d+):(\d+)", ident)
if not m:
abort(404)
season, episode = int(m.group(2)), int(m.group(3))
height = _requested_height()
local = _local_redirect("series", m.group(1), season, episode, height)
if local:
return local
url = _resolve_vidcore("series", m.group(1), season, episode,
wait=VIDCORE_PLAY_DEADLINE)
if not url:
abort(404)
_ensure_ingest("series", m.group(1), season, episode, height=height)
_queue_next_episode(m.group(1), season, episode, height)
return _serve_vidcore(url)
# --------------------------- offline MP4 copies -----------------------------
# The only way to get SEEK on this source. vidcore's HLS is muxed fMP4: Kodi's
# ffmpeg demuxer plays it (audio works) but ignores seeks, and
# inputstream.adaptive seeks but plays no audio because it feeds only the first
# track of a muxed fMP4. Neither is fixable from here -- ISA is a compiled
# binary and Kodi's seek never even reaches the demuxer.
#
# So hand the player something it can seek: a plain progressive MP4. The remux
# is a STREAM COPY, so there is no re-encode, no quality loss and almost no CPU
# -- measured on stortron at ~5x realtime (120s of video in 23.9s), ~1.33GB per
# 28-minute episode at 1080p and ~0.67GB at 720p.
MP4_DIR = os.path.join(os.path.dirname(os.path.abspath(__file__)), "mp4-cache")
MP4_REMUX_TIMEOUT = 45 * 60
_mp4_jobs: dict[str, str] = {} # key -> "running" | "done" | "failed: ..."
_mp4_jobs_guard = threading.Lock()
_mp4_sem = threading.Semaphore(2) # concurrent remuxes
# One lock per library FILE, because the job key above is not one-to-one with
# files: Kodi's cinemeta rows ask by `tt…`, the flixlab catalog and the web UI by
# `tmdb:…`, and both name the same destination. Keyed on the ident alone, two
# jobs wrote one .part at once (labs#12). Locks are never removed -- one small
# object per title ever fetched, which is bounded by the library itself.
_dest_locks: dict[str, threading.Lock] = {}
_dest_locks_guard = threading.Lock()
def _dest_lock(dest: str) -> threading.Lock:
with _dest_locks_guard:
return _dest_locks.setdefault(os.path.abspath(dest), threading.Lock())
# What a live job is doing, for the download manager (labs#13). In memory only,
# deliberately: the persisted queue keeps its running/done/failed semantics,
# which the resume path depends on, and a phase is only meaningful while the
# process that owns the job is alive. Entries exist for exactly the life of
# _run_job.
# phase: waiting -> resolving -> downloading
# source: vidcore | hdhub | None (None = nothing fetched, e.g. already held)
_job_info: dict[str, dict] = {}
_cancel_events: dict[str, threading.Event] = {}
def _job_update(key: str, **fields) -> None:
"""Annotate a live job. A no-op for a key with no live job, so
_run_ingest stays callable on its own (as its tests call it)."""
info = _job_info.get(key)
if info is not None:
info.update(fields)
def _acquire_or_cancel(lock, cancel: threading.Event,
timeout: Optional[float] = None) -> bool:
"""Acquire `lock` (a Lock or Semaphore) unless `cancel` fires first.
Polls in short slices because neither primitive can wait on an Event too.
Returns False if cancelled or timed out, and then holds nothing.
"""
deadline = None if timeout is None else time.time() + timeout
while not cancel.is_set():
if lock.acquire(timeout=0.25):
if cancel.is_set():
lock.release()
return False
return True
if deadline is not None and time.time() > deadline:
return False
return False
def _cancel_job(key: str) -> bool:
"""Ask a live job to stop. True if there was one to ask.
It stops at the next point it can: immediately while waiting, when the
resolve returns, or within one MP4_STALL_POLL while ffmpeg is copying
(_run_remux kills it). The .part is left where it is -- ffmpeg's -y reuses
it on the next attempt, and nothing in this module deletes to recover.
"""
with _mp4_jobs_guard:
event = _cancel_events.get(key)
if event is None:
return False
event.set()
log.info("ingest: cancel requested for %s", key)
return True
# Every finished attempt, one JSON object per line, appended and never
# rewritten. The queue file cannot serve as history: it holds one LATEST state
# per key, so a failure followed by a success leaves no trace of the failure.
#
# Appends are serialised under a lock and each is a single write of one line.
# The reader skips a line it cannot parse rather than failing: this file is
# display-only and nothing ever rebuilds it from what was read, so the shape of
# the 2026-08-31 queue loss (an unreadable read persisted as empty) cannot occur.
INGEST_HISTORY_FILE = os.path.join(os.path.dirname(os.path.abspath(__file__)),
"ingest_history.jsonl")
_history_guard = threading.Lock()
def _append_history(entry: dict) -> None:
"""Never raises: losing a history line must not fail the job it records."""
try:
line = json.dumps(entry, ensure_ascii=False) + "\n"
with _history_guard:
with open(INGEST_HISTORY_FILE, "a", encoding="utf-8") as f:
f.write(line)
except Exception as exc:
log.warning("history append failed: %s", exc)
def _read_history(limit: int = 500) -> list[dict]:
"""Newest first. A missing file is an empty history."""
try:
with open(INGEST_HISTORY_FILE, encoding="utf-8") as f:
lines = f.readlines()
except FileNotFoundError:
return []
except OSError as exc:
log.warning("history unreadable: %s", exc)
return []
out = []
for line in reversed(lines):
try:
entry = json.loads(line)
except ValueError:
continue
if isinstance(entry, dict):
out.append(entry)
if len(out) >= limit:
break
return out
def _record_attempt(key: str, info: dict, state: str, started: float) -> None:
dest = info.get("dest")
size = None
if dest and state == "done":
try:
size = os.path.getsize(dest)
except OSError:
pass
_append_history({
"ts": time.time(), "started": started, "key": key,
"kind": info.get("kind"), "ident": info.get("ident"),
"season": info.get("season"), "episode": info.get("episode"),
"height": info.get("height"), "title": info.get("title"),
"label": info.get("label"), "outcome": state,
"source": info.get("source"), "bytes": size,
"took": round(time.time() - started, 1), "dest": dest,
})
# A human verdict that one source's copy of one FILE is wrong, in a way no
# measurement here can see. 2026-10-14, VisionQuest S01E01: vidcore's only audio
# track is tagged `eng` and is French; re-fetching took the same copy again,
# because vidcore is always tried first. Keyed by library path, like the dest
# lock, because a Kodi `tt…` job and a web `tmdb:…` job name the same file.
# Read by _run_ingest, so every way into the library honours it.
SOURCE_REJECTS_FILE = os.path.join(os.path.dirname(os.path.abspath(__file__)),
"source_rejects.json")
_rejects_guard = threading.Lock()
def _read_rejects() -> dict:
"""RAISES on an unreadable file -- the writer's read (see _read_job_queue)."""
try:
with open(SOURCE_REJECTS_FILE) as f:
data = json.load(f)
except FileNotFoundError:
return {}
if not isinstance(data, dict):
raise ValueError("source rejects is not a JSON object")
return data
def _rejected_sources(dest: str) -> set:
"""Never raises. Unreadable reads as nothing rejected: a record we cannot
read must not become a way for a working source to stop."""
try:
return set(_read_rejects().get(os.path.abspath(dest)) or [])
except Exception as exc:
log.warning("source rejects unreadable (%s) — rejecting nothing", exc)
return set()
def _change_rejects(dest: str, sid: str, rejected: bool) -> None:
"""Never raises. An unreadable file is left exactly as it is."""
try:
with _rejects_guard:
data = _read_rejects()
key = os.path.abspath(dest)
ids = set(data.get(key) or [])
ids = ids | {sid} if rejected else ids - {sid}
if ids:
data[key] = sorted(ids)
else:
data.pop(key, None)
tmp = SOURCE_REJECTS_FILE + ".tmp"
with open(tmp, "w") as f:
json.dump(data, f, indent=1)
os.replace(tmp, SOURCE_REJECTS_FILE)
log.info("source %s %s for %s", sid,
"rejected" if rejected else "allowed again", os.path.basename(dest))
except Exception as exc:
log.warning("source rejects not saved: %s", exc)
def _reject_source(dest: str, sid: str) -> None:
_change_rejects(dest, sid, True)
def _allow_source(dest: str, sid: str) -> None:
_change_rejects(dest, sid, False)
def _ingest_order(dest: str) -> list[str]:
"""Source ids that may fill this file, in the order they will be tried."""
rejected = _rejected_sources(dest)
return [s.id for s in ingest_sources() if s.id not in rejected]
# Job state was in-memory only, so a restart lost every intent: a copy that was
# mid-flight never resumed, and since a partial writes to `<dest>.part` while the
# row keys off the finished file, nothing surfaced the loss. Rolling forward
# makes that worse — it queues work nobody is waiting on, so nobody notices.
#
# Nothing is deleted to recover: an orphaned .part is simply reused, because
# ffmpeg runs with -y and writes the same path.
INGEST_QUEUE_FILE = os.path.join(os.path.dirname(os.path.abspath(__file__)),
"ingest_queue.json")
# Guards the read-modify-write below. Without it two jobs finishing together
# both load the queue, both add their own key, and the second write drops the
# first -- and a season finishing episode after episode is exactly that, over
# and over.
_queue_guard = threading.Lock()
def _read_job_queue() -> dict:
"""The queue as stored. RAISES if the file exists but cannot be read.
The raising version exists so a writer can tell "there is no queue yet" from
"there is a queue and I could not read it". Those must not be treated alike:
the first is an empty dict, the second is every job on record.
"""
try:
with open(INGEST_QUEUE_FILE) as f:
data = json.load(f)
except FileNotFoundError:
return {}
if not isinstance(data, dict):
raise ValueError("ingest queue is not a JSON object")
return data
def _load_job_queue() -> dict:
"""Persisted job states. Never raises — this is a convenience, and a bad
file must not be a reason to refuse to boot."""
try:
return _read_job_queue()
except Exception as exc:
log.warning("ingest queue unreadable (%s) — reading as empty", exc)
return {}
def _set_job_state(key: str, state: str, args: Optional[tuple] = None) -> None:
"""Record a job's state in memory and on disk.
`args` is what makes a resume possible: a key alone cannot be re-run.
"""
_mp4_jobs[key] = state
try:
with _queue_guard:
# The RAISING read, deliberately. _load_job_queue answers {} for a
# file it cannot parse, which is right for a reader and catastrophic
# for a writer: it would rebuild the queue from nothing and persist
# that, so one unreadable read destroys every job on record. Letting
# it raise here costs a single unrecorded state change instead.
queue = _read_job_queue()
entry = queue.get(key) or {}
entry["state"] = state
if args is not None:
entry["args"] = list(args)
entry["ts"] = time.time()
queue[key] = entry
# Written to a temp file and renamed, for the same reason the
# ingest itself writes a .part. open(..., "w") TRUNCATES first, so a
# reader arriving in that window got "Expecting value: line 1
# column 1" -- and _load_job_queue catches everything and returns {}
# by design, so a torn read did not fail loudly. It reported an
# empty queue, and the next write persisted that. Every job on
# record, gone.
tmp = INGEST_QUEUE_FILE + ".tmp"
with open(tmp, "w") as f:
json.dump(queue, f)
os.replace(tmp, INGEST_QUEUE_FILE)
except Exception as exc: # persistence is never load-bearing
log.warning("ingest queue save failed: %s", exc)
def _resume_ingests() -> None:
"""Rehydrate job state at startup and restart anything left mid-flight.
Finished jobs are rehydrated too, not just running ones: without that, the
next request for an already-copied title starts a duplicate ingest.
A FAILED job is not retried. A failure that recurs every boot would launch a
browser on each one; the user can still ask for it by playing the row.
"""
for key, entry in _load_job_queue().items():
state = entry.get("state") or ""
if state != "running":
_mp4_jobs[key] = state
continue
# Deliberately NOT rehydrated as "running": _start_mp4_job refuses a key
# already marked running, so pre-setting it here would silently prevent
# the very restart this function exists to perform. The job sets it.
args = entry.get("args")
if not args or len(args) != 5:
log.info("ingest resume: %s has no arguments — skipped", key)
continue
log.info("ingest resume: %s", key)
_start_mp4_job(*args)
def _mp4_key(kind: str, ident: str, season: int, episode: int, height: int) -> str:
return f"{kind}:{ident}:{season}:{episode}:{height}"
def _mp4_path(kind: str, ident: str, season: int, episode: int, height: int) -> str:
"""Cache path for one remuxed copy.
The ident is slugified rather than interpolated: it arrives from the URL and
contains colons, and must never be able to walk out of the cache directory.
"""
slug = re.sub(r"[^A-Za-z0-9]+", "_", f"{kind}-{ident}-{season}-{episode}-{height}")
return os.path.join(MP4_DIR, f"{slug}.mp4")
def _mp4_ready(kind: str, ident: str, season: int, episode: int, height: int) -> bool:
"""True only for a COMPLETE copy.
An in-progress remux writes to `<name>.part`; the moov atom that makes the
file seekable is written last, so a partial file is both truncated and
unseekable and must never be served.
"""
path = _mp4_path(kind, ident, season, episode, height)
try:
return os.path.getsize(path) > 0
except OSError:
return False
# ---------------------------- library ingest --------------------------------
# A finished copy is not cache -- it is library content, and it is permanent. It
# lands in the Jellyfin library on /mnt/usb, where Jellyfin indexes it and Kodi
# plays it through the same plugin:// path as everything else. Seek, resume and
# pause all work there natively, which is what makes the muxed-fMP4 problem above
# stop mattering for anything we have actually fetched.
#
# The layout is the one Jellyfin identifies on, with the TMDB id stated
# explicitly:
# Shows/<Title> (<Year>) [tmdbid-<id>]/Season 02/<Title> (<Year>) - S02E07 - <Ep>.mp4
# Movies/<Title> (<Year>) [tmdbid-<id>]/<Title> (<Year>).mp4
#
# Stating the id matters. The 2026-07-28 library cleanup found Jellyfin's own
# identification is ~92% right -- fine for browsing, NOT safe to rename blindly
# (Bombshell matched the Hedy Lamarr documentary; The_Fall matched The Fall Guy
# at only 8% runtime error). That applied to retrofitting unknown files. Here the
# title was resolved FROM its TMDB id, so identity is asserted, not inferred, and
# no runtime cross-check is needed.
LIBRARY_ROOT = "/mnt/usb"
# The volume is exFAT (verified 2026-08-11: findmnt -> exfat /dev/sdb1), which
# rejects these outright. They are REPLACED rather than dropped so a title stays
# readable and distinguishable -- dropping would turn "Ep: One" into "Ep One".
# The colon maps to U+FF1A because the library already holds
# `Ghost in the Shell：STAND ALONE COMPLEX` -- an existing convention, not a new
# invention. Slashes become "-" because a path separator reads naturally as one.
EXFAT_SUBSTITUTIONS = {
":": "：", "*": "＊", "?": "？", '"': "＂",
"<": "＜", ">": "＞", "|": "｜", "/": "-", "\\": "-",
}
def _safe_component(name: str) -> str:
"""One path component, safe on exFAT and never empty.
Never returns "" or a value containing a separator, so a composed path
cannot escape LIBRARY_ROOT no matter what TMDB (or a caller) supplies.
"""
out = "".join(EXFAT_SUBSTITUTIONS.get(c, c) for c in name or "")
out = "".join(" " if ord(c) < 32 else c for c in out)
out = re.sub(r"\s+", " ", out).strip()
# Leading dots hide the folder from Jellyfin entirely; trailing dots and
# spaces are silently altered by exFAT/Windows, which would break the
# os.replace that finalises an ingest.
out = out.strip(". ")
return out or "_"
def _library_path(kind: str, title: str, year: str, tmdb_id: str,
season: int = 0, episode: int = 0,
episode_title: str = "") -> str:
"""Destination for one finished copy, in Jellyfin's layout."""
stem = f"{title} ({year})" if year else title
folder = _safe_component(f"{stem} [tmdbid-{tmdb_id}]")
if kind == "series":
name = f"{_safe_component(stem)} - S{season:02d}E{episode:02d}"
if episode_title:
name += f" - {_safe_component(episode_title)}"
return os.path.join(LIBRARY_ROOT, "Shows", folder,
f"Season {season:02d}", f"{name}.mp4")
return os.path.join(LIBRARY_ROOT, "Movies", folder,
f"{_safe_component(stem)}.mp4")
def _part_path(dest: str) -> str:
"""Where an in-progress copy is written.
A SIBLING of its destination, not a temp dir: os.replace() cannot cross a
filesystem boundary (EXDEV), and the library is a separate exFAT volume from
the repo. Writing to <repo>/mp4-cache and renaming onto /mnt/usb would fail
on every single ingest.
"""
return dest + ".part"
# Naming is only as safe as the identity behind it, so a title we cannot
# identify is NOT ingested. The 2026-07-28 cleanup exists because wrongly-named
# media accumulated; a bad file in the library costs more than a missing one.
_library_meta_cache: dict[str, Optional[dict]] = {}
def _tmdb_id_for(kind: str, ident: str) -> Optional[str]:
"""`tmdb:<n>` verbatim, or a `tt<n>` translated through TMDB's find."""
if ident.startswith("tmdb:"):
return ident.split(":", 1)[1]
data = _tmdb_get(f"find/{ident}", {"external_source": "imdb_id"})
key = "tv_results" if kind == "series" else "movie_results"
results = data.get(key) or []
return str(results[0]["id"]) if results else None
def _library_meta(kind: str, ident: str, season: int = 0,
episode: int = 0) -> Optional[dict]:
"""Everything needed to name one file, or None if it cannot be identified.
A "not found" answer is cached (it is stable), but a TMDB *failure* is not:
marking a title permanently unidentifiable because of one bad request is the
same mistake that hid playable titles for a week -- inconclusive must not
harden into absent.
"""
key = f"{kind}:{ident}:{season}:{episode}"
if key in _library_meta_cache:
return _library_meta_cache[key]
try:
tmdb_id = _tmdb_id_for(kind, ident)
if not tmdb_id:
log.info("library: %s has no TMDB id — not ingestable", ident)
_library_meta_cache[key] = None
return None
if kind == "series":
show = _tmdb_get(f"tv/{tmdb_id}")
title = show.get("name") or ""
year = (show.get("first_air_date") or "")[:4]
ep = _tmdb_get(f"tv/{tmdb_id}/season/{season}/episode/{episode}")
episode_title = ep.get("name") or ""
# Minutes, and free: this is the response we already had to fetch
# for the episode title, and it is cached with the rest of the meta.
runtime = ep.get("runtime")
else:
film = _tmdb_get(f"movie/{tmdb_id}")
title = film.get("title") or ""
year = (film.get("release_date") or "")[:4]
episode_title = ""
runtime = film.get("runtime")
except Exception as exc:
# Deliberately NOT cached — see docstring.
log.warning("library meta failed for %s: %s", key, exc)
return None
if not title:
log.info("library: %s resolved to no title — not ingestable", ident)
_library_meta_cache[key] = None
return None
meta = {"tmdb_id": tmdb_id, "title": title,
"year": year, "episode_title": episode_title, "runtime": runtime}
_library_meta_cache[key] = meta
return meta
# Whether a copy already on disk can be trusted, measured against TMDB's
# `runtime`. A LOOSER floor than REMUX_MIN_COMPLETENESS, deliberately, because
# it compares a different kind of number: the remux check compares an output
# against the EXTINF sum of the playlist it came from -- a measurement, which
# agreed to 0.003% on real data -- whereas `runtime` is editorial metadata. The
# fourteen known-good copies in the library measured 92-101% of it. A floor
# tight enough for the former would reject nearly all of the latter.
LIBRARY_MIN_COMPLETENESS = 0.85
# Copies whose length was measured against their own source at the moment they
# were written. That is strictly better evidence than TMDB's runtime, so it
# WINS over it -- and it has to, or a title whose upstream genuinely only holds
# a short version would loop for ever: the runtime gate rejects the copy, the
# re-ingest fetches the same short source, the playlist check passes it, and the
# runtime gate rejects it again on the next play.
LIBRARY_VERIFIED_FILE = os.path.join(os.path.dirname(os.path.abspath(__file__)),
"library_verified.json")
_duration_cache: dict[tuple, Optional[float]] = {}
def _load_verified() -> dict:
"""Never raises: like the job queue, this is evidence, not load-bearing."""
try:
with open(LIBRARY_VERIFIED_FILE) as f:
data = json.load(f)
return data if isinstance(data, dict) else {}
except Exception:
return {}
def _record_verified(path: str, seconds: float, expected: float) -> None:
try:
store = _load_verified()
store[os.path.abspath(path)] = {"seconds": seconds, "expected": expected,
"ts": time.time()}
with open(LIBRARY_VERIFIED_FILE, "w") as f:
json.dump(store, f)
except Exception as exc:
log.warning("verified store save failed: %s", exc)
def _forget_verified(path: str) -> None:
"""Drop a copy's verification record.
Only ever called for a FORCED re-ingest. The record attests that the file
matched the source it came from, which is not the question a human asking
for a re-download is raising, so it must not be allowed to veto them.
"""
try:
store = _load_verified()
if store.pop(os.path.abspath(path), None) is None:
return
with open(LIBRARY_VERIFIED_FILE, "w") as f:
json.dump(store, f)
except Exception as exc:
log.warning("verified store update failed: %s", exc)
def _cached_media_duration(path: str) -> Optional[float]:
"""_media_duration, keyed on the file's identity rather than its name.
Every play of a local copy asks this, so it must not spawn an ffprobe each
time. Size and mtime are in the key so a re-ingest invalidates it for free.
"""
try:
st = os.stat(path)
except OSError:
return None
key = (os.path.abspath(path), st.st_size, int(st.st_mtime))
if key not in _duration_cache:
_duration_cache[key] = _media_duration(path)
return _duration_cache[key]
def _copy_is_complete(path: str, meta: Optional[dict]) -> bool:
"""Is the file on disk a whole copy, or the 21% of an episode we published?
Both _library_has and _playable_path go through here, and they MUST agree:
if only the serving side rejected a partial, the copy would stop being
served but the ingest would still report "already held" and never replace
it. Rejecting it in both is what makes the repair automatic.
Everything unknown resolves to True. This is the opposite asymmetry from
_remux_to_mp4, and for a reason: there we had just written the file and knew
exactly how long it should be, so an unreadable result was a broken file.
Here the file has been serving fine and the only doubt comes from a third
party's metadata, so a check that cannot run must not condemn it.
"""
if os.path.abspath(path) in _load_verified():
return True
runtime = (meta or {}).get("runtime")
try:
expected = float(runtime) * 60
except (TypeError, ValueError):
return True # TMDB has no runtime for this title
if expected <= 0:
return True
actual = _cached_media_duration(path)
if actual is None:
return True # unprobeable, but it was here first
if actual >= expected * LIBRARY_MIN_COMPLETENESS:
return True
log.warning("library: %s holds %.0fs of an expected %.0fs (%.0f%%) — "
"not treating it as a copy",
os.path.basename(path), actual, expected, actual / expected * 100)
return False
def _library_has(kind: str, ident: str, season: int = 0,
episode: int = 0) -> bool:
"""True if the library already holds this episode, at ANY height.
Height is deliberately absent from the library path: the library holds one
copy per episode. A later request at a different quality is a no-op rather
than a second file, and nothing is ever deleted to make room for an upgrade.
A partial copy does not count, so an ingest is free to replace one.
"""
meta = _library_meta(kind, ident, season, episode)
if not meta:
return False
dest = _library_path(kind, meta["title"], meta["year"], meta["tmdb_id"],
season, episode, meta["episode_title"])
try:
if os.path.getsize(dest) <= 0:
return False
except OSError:
return False
return _copy_is_complete(dest, meta)
# How much of the source a remux must actually contain before it is published.
#
# An exit code cannot answer this. ffmpeg treats an HLS input that stops early
# -- an expired token, a 403 on a segment part-way down the ladder -- as end of
# stream, not as an error: it finalises the moov atom over what it got and exits
# 0. On 2026-08-26 that published Strange New Worlds S04E04 at 10.6 of 51
# minutes (21%), S04E05 at 9.8 of 55 (18%) and Furious S01E07 at 27.7 of 52
# (53%), each logged as `ingest done`.
#
# The tolerance absorbs the drift between an EXTINF sum and a container
# duration, which do not agree to the frame. It does not need to be generous:
# the smallest real loss observed was a little over half the episode, and
# failing here is cheap -- the copy is simply retried -- whereas failing to
# catch one is permanent, because _library_has asks only whether a non-empty
# file exists and will answer "done" for ever after.
REMUX_MIN_COMPLETENESS = 0.99
def _parse_playlist_duration(body: str) -> Optional[float]:
"""Total seconds of a media playlist, summed from its #EXTINF tags.
None when there are none to sum -- a master playlist, or a body that is not
a playlist at all. That has to be distinct from 0.0: zero compares as
complete against anything and would wave through the exact failure this
exists to catch.
"""
total = 0.0
found = False
for line in body.splitlines():
if not line.startswith("#EXTINF:"):
continue
try:
total += float(line[len("#EXTINF:"):].split(",")[0])
found = True
except ValueError:
continue
return total if found else None
def _playlist_duration(variant_url: str) -> Optional[float]:
"""How long the source says this variant is. None if it cannot be asked."""
try:
# Deliberately NOT _http_get, for the same reason as _variants: that
# injects a random X-Forwarded-For and these tokens are IP-CIDR locked.
r = requests.get(variant_url, timeout=REQUEST_TIMEOUT,
headers={"User-Agent": UA,
"Referer": "https://vidcore.net/"})
r.raise_for_status()
return _parse_playlist_duration(r.text)
except Exception as exc:
log.warning("remux: could not read source duration (%s)", exc)
return None
def _media_duration(path: str) -> Optional[float]:
"""Seconds of playable content in a local file, or None if unreadable."""
try:
out = subprocess.run(
["ffprobe", "-v", "error", "-show_entries", "format=duration",
"-of", "default=nw=1:nk=1", path],
capture_output=True, text=True, timeout=60).stdout.strip()
return float(out) if out else None
except Exception as exc:
log.warning("remux: could not probe %s (%s)", os.path.basename(path), exc)
return None
# A remux that stops producing bytes is dead, and the wall clock is the worst
# possible way to find out. 2026-09-03, No Other Choice (tmdb 639988): ffmpeg
# wrote 2.85 GB in ~9 minutes (~5.3 MB/s), then produced nothing for 36 more
# until MP4_REMUX_TIMEOUT killed it -- reported as "timed out after 2700
# seconds", which reads as too-slow and was not. Five minutes without a byte,
# against a copy that runs at megabytes per second, is unambiguous.
MP4_STALL_TIMEOUT = 5 * 60
# How often the watchdog looks. Cheap (one stat), and the resolution only needs
# to be fine relative to the stall window.
MP4_STALL_POLL = 2.0
class RemuxStalled(RuntimeError):
"""The output stopped growing. Distinct from RemuxTimedOut on purpose: one
says the source went quiet, the other says the copy is genuinely too long,
and they want different responses."""
class RemuxTimedOut(RuntimeError):
"""The copy exceeded its wall-clock budget while still making progress."""
class RemuxCancelled(RuntimeError):
"""Someone pressed Cancel. Not a failure of the source, so it must never be
treated as one -- in particular it must not send _run_ingest on to the
HdHub fallback, which would fetch the very title the user just stopped."""
# The job a thread is running, so _run_remux can see its cancel event without
# the event being threaded through _remux_to_mp4 and _copy_progressive (whose
# two-argument shape the test fakes also rely on). Set by _run_job for the life
# of one ingest; absent on any thread that is not running a job.
_job_ctx = threading.local()
def _cancel_requested() -> bool:
event = getattr(_job_ctx, "cancel", None)
return bool(event and event.is_set())
def _read_back(f) -> str:
"""Everything written to an open temp file, from the start. Never raises:
this only ever supplies a diagnostic, and a diagnostic that can fail the
operation it describes is worse than no diagnostic."""
try:
f.seek(0)
return f.read()
except Exception:
return ""
class _RemuxResult:
"""What subprocess.run would have returned, minus the parts nothing reads.
Named rather than a tuple because the caller checks `.returncode` and
`.stderr`, and those two being positional is how they get swapped.
"""
__slots__ = ("returncode", "stderr")
def __init__(self, returncode: int, stderr: str):
self.returncode = returncode
self.stderr = stderr
def _run_remux(cmd: list[str], out_path: str, stall_timeout: float,
hard_timeout: float):
"""Run `cmd`, killing it if `out_path` stops growing for `stall_timeout`.
The watchdog is OUTSIDE ffmpeg deliberately. ffmpeg's own -rw_timeout and
-reconnect flags only cover the stalls ffmpeg itself notices; this covers
every mode, including the one where it reconnects happily and still writes
nothing.
Progress is measured on the output file rather than on ffmpeg's stderr,
because stderr is silent at -v error precisely when things are going well.
A missing file counts as no progress -- ffmpeg may not have created it yet,
and "not created for five minutes" is also a stall.
"""
# stderr goes to a FILE, not a pipe. A pipe deadlocks: ffmpeg fills the 64K
# buffer, blocks writing, therefore stops writing the OUTPUT too -- and the
# watchdog below reads that as a stall and kills a process that was only
# waiting for us to read. The failing case is exactly the one whose stderr
# is worth having.
def _size() -> int:
try:
return os.path.getsize(out_path)
except OSError:
return 0
with tempfile.TemporaryFile(mode="w+", encoding="utf-8",
errors="replace") as errf:
proc = subprocess.Popen(cmd, stdout=subprocess.DEVNULL, stderr=errf)
started = time.time()
last_size = -1
last_growth = started
try:
while True:
try:
proc.wait(timeout=MP4_STALL_POLL)
break
except subprocess.TimeoutExpired:
pass
if _cancel_requested():
raise RemuxCancelled("cancelled")
now = time.time()
size = _size()
if size > last_size:
last_size, last_growth = size, now
if now - last_growth > stall_timeout:
raise RemuxStalled(
f"no output for {now - last_growth:.0f}s "
f"after {last_size} bytes in {now - started:.0f}s")
if now - started > hard_timeout:
raise RemuxTimedOut(
f"still running after {now - started:.0f}s "
f"({last_size} bytes)")
# Read stderr back even on the way out: a stall or a timeout is exactly
# when ffmpeg's last words are worth having, and the caller only ever
# sees the exception message.
except BaseException as exc:
# Kill rather than terminate: ffmpeg reading a stalled socket does
# not always act on SIGTERM, and leaving it alive holds the .part
# open and the semaphore slot with it.
proc.kill()
proc.wait()
tail = _read_back(errf)[-300:]
if tail and isinstance(exc, (RemuxStalled, RemuxTimedOut)):
raise type(exc)(f"{exc} — ffmpeg said: {tail}") from None
raise
return _RemuxResult(proc.returncode, _read_back(errf))
# ---------------------------------------------------------------- subtitles
#
# The HLS stream carries none. Measured 2026-09-04 on No Other Choice (tmdb
# 639988): the master lists four muxed A/V variants and no #EXT-X-MEDIA at all,
# so `-c copy` is not dropping anything -- there is nothing there.
#
# The vidcore PLAYER gets them from a separate endpoint, found in the 2026-07-21
# deobfuscation of vidcore's bundle and re-verified today: `/wyzie` is a proxy
# in front of Wyzie, returning `[{display, language, url, encoding}]`. It needs
# no browser and, unlike sub.wyzie.ru direct (now `401 API key required`), no
# key of ours.
#
# All 32 tracks for that movie fetch in 2.9s / 3.6 MB with 8 workers, every one
# plain SRT, so there is no case for choosing a language here. Everything is
# taken and the player picks.
WYZIE_URL = "https://vidcore.net/wyzie"
SUBTITLE_TIMEOUT = 25
# Measured: 32 tracks in 2.9s at 8, slowest single request 0.95s. Higher buys
# nothing against a set this size and is rude to a host doing us a favour.
SUBTITLE_WORKERS = 8
# wyzie speaks OpenSubtitles' codes, and several are not ISO at all.
#
# The targets are NOT guessed: they are the codes Jellyfin's own
# /Localization/Cultures list answers to, checked 2026-09-04 after a first
# attempt shipped `yue` and `zh-Hant` and Jellyfin filed the Cantonese track as
# "Undefined". Both are perfectly good language tags; they are simply not in the
# table doing the lookup, which is the only thing that matters here.
#
# pb -> pt-br "Portuguese (Brazil)"
# ea -> es-419 "Spanish; Latin"
# zt -> zh-tw "Chinese (Traditional)" (zh-Hant is NOT in the table)
# zc -> zh-hk "Chinese (Hong Kong)" (yue is NOT in the table)
#
# `ze` is deliberately absent: Jellyfin knows it natively as "Chinese
# (Bilingual)", so mapping it to `zh` threw away a distinction the server was
# ready to make.
#
# Anything not listed is passed through: an unrecognised code still names a
# distinct file, and dropping a subtitle because we did not recognise its label
# is the worse failure.
WYZIE_LANG_MAP = {
"pb": "pt-BR", # Portuguese (BR)
"ea": "es-419", # Spanish (LA)
"zt": "zh-TW", # Chinese (traditional)
"zc": "zh-HK", # Chinese (Cantonese)
}
def _http_json(url: str, timeout: int = REQUEST_TIMEOUT):
"""GET and parse JSON. Raises on anything but a 2xx with a JSON body.
Deliberately NOT _http_get: that injects a random X-Forwarded-For, which is
for evading a source's rate limit and has no business on an endpoint that is
answering us honestly.
"""
r = requests.get(url, timeout=timeout,
headers={"User-Agent": UA, "Referer": "https://vidcore.net/"})
r.raise_for_status()
return r.json()
def _fetch_subtitle(url: str, timeout: int = SUBTITLE_TIMEOUT) -> bytes:
"""One subtitle file's bytes. Raises on anything but a 2xx."""
r = requests.get(url, timeout=timeout,
headers={"User-Agent": UA, "Referer": "https://vidcore.net/"})
r.raise_for_status()
return r.content
def _wyzie_tracks(tmdb_id: str, season: int = 0, episode: int = 0) -> list[dict]:
"""Available subtitle tracks, or [] -- never an exception.
A title Wyzie lacks answers 404 with a zero-length body (verified on X-Men
'97 S2E8, which genuinely has none while S1E1 has 34). That is an ordinary
outcome, not a failure, and an ingest must not fail over it. The same goes
for the endpoint being down: a copy without subtitles is worth having.
"""
url = f"{WYZIE_URL}?id={tmdb_id}"
if season and episode:
url += f"&season={season}&episode={episode}"
try:
data = _http_json(url)
except Exception as exc:
log.info("subtitles: none for %s (%s)", tmdb_id, exc)
return []
if not isinstance(data, list):
log.warning("subtitles: unexpected answer shape for %s", tmdb_id)
return []
tracks = [t for t in data if isinstance(t, dict) and t.get("url")]
log.info("subtitles: %d track(s) for %s", len(tracks), tmdb_id)
return tracks
def _sidecar_paths(dest_mp4: str, tracks: list[dict]) -> list[tuple[str, dict]]:
"""(path, track) for each track, with names that cannot collide or escape.
The language code arrives inside a remote JSON body and ends up in a
filename, so it goes through _safe_component like everything else that
composes a library path.
"""
stem = os.path.splitext(dest_mp4)[0]
seen: dict[str, int] = {}
out = []
for track in tracks:
code = WYZIE_LANG_MAP.get(track.get("language"), track.get("language"))
code = _safe_component(code or "und")
n = seen[code] = seen.get(code, 0) + 1
# The first of a language gets the bare suffix, which is the one both
# Jellyfin and Kodi read. Later ones are still offered, just ranked
# below it, rather than silently overwriting it.
suffix = code if n == 1 else f"{code}.{n}"
out.append((f"{stem}.{suffix}.srt", track))
return out
def _write_sidecars(dest_mp4: str, tracks: list[dict]) -> list[str]:
"""Download `tracks` next to `dest_mp4`. Returns the paths that exist after.
Sidecars rather than a muxed track because Jellyfin indexes the directory,
and because a subtitle added later must not mean re-copying 3 GB of video.
Nothing here may fail the ingest: subtitles are an addition to a copy that
is already complete and published. One track failing costs that track.
"""
if not tracks:
return []
planned = _sidecar_paths(dest_mp4, tracks)
todo = [(p, t) for p, t in planned
if not (os.path.exists(p) and os.path.getsize(p) > 0)]
done = [p for p, _ in planned if p not in {q for q, _ in todo}]
def _one(item):
path, track = item
try:
body = _fetch_subtitle(track["url"])
except Exception as exc:
log.info("subtitles: %s failed (%s)", os.path.basename(path), exc)
return None
if not body:
# An empty file is worse than none: the player offers a track that
# shows nothing, and the "already on disk" check would keep it.
log.info("subtitles: %s came back empty", os.path.basename(path))
return None
tmp = _part_path(path)
try:
with open(tmp, "wb") as f:
f.write(body)
os.replace(tmp, path)
except Exception as exc:
log.warning("subtitles: could not write %s (%s)",
os.path.basename(path), exc)
return None
return path
if todo:
with ThreadPoolExecutor(max_workers=SUBTITLE_WORKERS) as pool:
done += [p for p in pool.map(_one, todo) if p]
if done:
log.info("subtitles: %d file(s) beside %s",
len(done), os.path.basename(dest_mp4))
return sorted(done)
def _remux_to_mp4(variant_url: str, out_path: str) -> None:
"""Stream-copy an HLS variant into a progressive MP4.
`-movflags +faststart` puts the moov atom at the front so the file can be
seeked over HTTP. Written to `.part` and renamed, so a crashed or killed
remux can never leave a half file looking complete -- and, since ffmpeg's
exit code says nothing about how much of the source it got, only renamed
once the output has been measured against the playlist it came from.
"""
tmp = _part_path(out_path)
# The destination's OWN directory, not MP4_DIR: an ingest writes into the
# library, where the show folder and season folder may not exist yet.
os.makedirs(os.path.dirname(out_path), exist_ok=True)
# Asked BEFORE the remux: these playlists are signed and expire in ~4h, so
# after a 45-minute copy the URL may no longer answer.
expected = _playlist_duration(variant_url)
proc = _run_remux(
["ffmpeg", "-v", "error", "-y",
"-headers", "Referer: https://vidcore.net/\r\n",
# An I/O operation that goes quiet errors instead of blocking, and the
# reconnect flags then retry it. Without these a CDN that stops
# answering mid-file is indistinguishable from one that is merely slow,
# and ffmpeg waits on it for ever (2026-09-03: 36 minutes).
# -seg_max_retry defaults to 0 on the hls demuxer, i.e. one bad segment
# is fatal, which is the wrong default for a 45-minute copy.
"-rw_timeout", "30000000",
"-reconnect", "1",
"-reconnect_streamed", "1",
"-reconnect_on_network_error", "1",
"-reconnect_delay_max", "30",
"-seg_max_retry", "3",
# -f mp4 is REQUIRED: ffmpeg picks the muxer from the file extension
# and the output is written to `<name>.part`, which it does not
# recognise ("Error initializing the muxer ...: Invalid argument").
"-i", variant_url, "-c", "copy", "-movflags", "+faststart",
"-f", "mp4", tmp],
tmp, stall_timeout=MP4_STALL_TIMEOUT, hard_timeout=MP4_REMUX_TIMEOUT,
)
if proc.returncode != 0 or not os.path.exists(tmp) or os.path.getsize(tmp) == 0:
raise RuntimeError(f"ffmpeg exit {proc.returncode}: {(proc.stderr or '')[-300:]}")
if expected:
actual = _media_duration(tmp)
# An unmeasurable output is refused rather than waved through: we know
# what it should be, so failing to read it back is a broken file. The
# reverse -- an unmeasurable SOURCE -- falls back to publishing, because
# a check that cannot run must not become a new way for a working
# stream to fail (the same rule _cap_variant follows).
if actual is None or actual < expected * REMUX_MIN_COMPLETENESS:
# The .part is deliberately left behind. Nothing is deleted to
# recover; ffmpeg runs with -y and writes the same path, so the
# retry overwrites it and the evidence survives until it does.
raise RuntimeError(
f"incomplete remux: got {actual if actual else 0:.0f}s of "
f"{expected:.0f}s ({(actual or 0) / expected * 100:.0f}%) — "
f"not published")
log.info("remux verified: %.0fs of %.0fs", actual, expected)
# Recorded against the FINAL path, and only for a copy measured against
# its own source. _copy_is_complete trusts this over TMDB's runtime,
# which is what stops a genuinely-short upstream from being re-fetched
# on every play for ever.
_record_verified(out_path, actual, expected)
os.replace(tmp, out_path)
# ------------------------------- subtitle sync ------------------------------
# wyzie tracks are not cut for OUR copy. Measured 2026-10-04 with ffsubsync on
# No Other Choice, against speech extracted once from the film: none of 32
# tracks matched. 23 sat at -8.9 s (another cut/intro), 7 including English at
# -2.2 s with scale 1.001 (24 vs 23.976 fps: drift reaching ~8 s by the end --
# the "keeps going out of sync" report), two at -18/-20 s. Corrected tracks
# re-measured at 0.000-0.010 s / 1.000 with every cue kept.
#
# Per track: in sync -> untouched. Fixable (the correction re-measures clean)
# -> original moved to SUBS_ORIGINAL_DIR, outside the library, never deleted;
# corrected copy takes its place. Unfixable (the correction does not hold: a
# different edit, which offset+scale cannot repair) -> original moved to the
# store and the library stops offering it. A measurement that cannot run
# leaves the track alone: a check that fails must not cost a subtitle.
SUBSYNC_BIN = os.path.join(os.path.dirname(os.path.abspath(__file__)),
".venv", "bin", "ffsubsync")
SUBSYNC_TOLERANCE = 0.3 # seconds; the families measured differ by >= 2
SUBSYNC_TIMEOUT = 15 * 60 # speech extraction of a 2h19 film took 70 s
SUBS_ORIGINAL_DIR = os.path.join(os.path.dirname(os.path.abspath(__file__)),
"subs-original")
SUBTITLE_SYNC_FILE = os.path.join(os.path.dirname(os.path.abspath(__file__)),
"subtitle_sync.json")
_subsync_guard = threading.Lock()
# Only these subtitle languages are measured and repaired ("focus on english",
# 2026-10-04): it is what gets watched, and syncing all ~32 wyzie tracks took
# ~5 min per film where English alone takes ~1.5. Others stay as fetched.
# None = every language (the mechanics tests use that).
SUBSYNC_LANGUAGES: Optional[tuple] = ("en",)
# Which reference a result was measured against. 2 = audio padded to its
# timestamps (2026-10-18, see _subsync_reference). A result without it came
# from the gap-dropping reference and is measured again from its original.
SUBSYNC_REF_VERSION = 2
def _sidecar_lang(dest: str, path: str) -> str:
"""`<stem>.<code>[.<n>].srt` -> code."""
stem = os.path.splitext(os.path.basename(dest))[0] + "."
middle = os.path.basename(path)[len(stem):-len(".srt")]
return re.sub(r"\.\d+$", "", middle)
def _subsync_available() -> bool:
return os.access(SUBSYNC_BIN, os.X_OK)
def _parse_subsync(out: str) -> Optional[dict]:
"""offset/scale/score from ffsubsync's own log lines, or None."""
nums = {}
for name, pattern in (("score", r"score: ([-\d.]+)"),
("offset", r"offset seconds: ([-\d.]+)"),
("scale", r"framerate scale factor: ([-\d.]+)")):
m = re.search(pattern, out or "")
if not m:
return None
nums[name] = float(m.group(1))
return nums
def _subsync_reference(dest: str, workdir: str) -> Optional[str]:
"""Extract the copy's speech ONCE into workdir; the npz is the reference
every track is then measured against (~6 s each instead of ~70 s).
The audio goes through a WAV padded to its own timestamps first. ffsubsync
pipes raw PCM out of ffmpeg, which has no timestamps, so a gap in the audio
is dropped and everything after it lands early. No Other Choice's audio
starts at 9.254 s: the 10-04 reference ran ~9.3 s early, "corrected" the
English by -2.25 s when it needed +7.01 s, and moved 23 tracks that were
in sync by -8.9 s. Padded, the full file and three timestamp-seeked
windows agree (+7.01 / +7.03 s). The WAV sits in workdir, which is also
where --serialize-speech writes the npz.
"""
sidecars = _sidecars_of(dest)
if not sidecars:
return None
wav = os.path.join(workdir, "ref.wav")
try:
subprocess.run(["ffmpeg", "-v", "error", "-y", "-i", dest, "-vn",
"-ac", "1", "-ar", "16000",
"-af", "aresample=async=1:first_pts=0", wav],
check=True, capture_output=True, timeout=SUBSYNC_TIMEOUT)
subprocess.run([SUBSYNC_BIN, wav, "--serialize-speech"], cwd=workdir,
capture_output=True, text=True, timeout=SUBSYNC_TIMEOUT)
except Exception as exc:
log.warning("subsync: speech extraction failed for %s (%s)",
os.path.basename(dest), exc)
return None
npz = os.path.join(workdir, "ref.npz")
return npz if os.path.exists(npz) else None
def _subsync_measure(ref: str, srt_in: str, srt_out: Optional[str]) -> Optional[dict]:
"""Measure one track against the reference; write the correction to
srt_out when given. None if ffsubsync could not produce a measurement."""
try:
out = subprocess.run([SUBSYNC_BIN, ref, "-i", srt_in,
"-o", srt_out or os.devnull],
capture_output=True, text=True,
timeout=SUBSYNC_TIMEOUT)
except Exception as exc:
log.warning("subsync: %s not measured (%s)", os.path.basename(srt_in), exc)
return None
return _parse_subsync(out.stdout + out.stderr)
def _in_tolerance(m: Optional[dict]) -> bool:
return bool(m) and abs(m["offset"]) <= SUBSYNC_TOLERANCE \
and round(m["scale"], 3) == 1.0
def _load_sync() -> dict:
try:
with open(SUBTITLE_SYNC_FILE) as f:
data = json.load(f)
return data if isinstance(data, dict) else {}
except FileNotFoundError:
return {}
except Exception as exc:
log.warning("subsync: record unreadable (%s)", exc)
raise
def _file_identity(path: str) -> Optional[list]:
try:
st = os.stat(path)
return [st.st_size, int(st.st_mtime)]
except OSError:
return None
def _keep_original(path: str) -> str:
"""Copy a sidecar into the store, mirroring its library path. Never
overwrites an earlier original: a later one gets a numeric suffix."""
rel = os.path.relpath(path, LIBRARY_ROOT)
target = os.path.join(SUBS_ORIGINAL_DIR, rel)
os.makedirs(os.path.dirname(target), exist_ok=True)
n, candidate = 1, target
while os.path.exists(candidate):
n += 1
candidate = f"{target}.{n}"
# copyfile, not copy2: copystat fails on the exFAT library (2026-08-31).
shutil.copyfile(path, candidate)
return candidate
def _restore_stale_originals(dest: str, records: dict) -> bool:
"""Put back the original of every track an older reference corrected or
withdrew, so it is measured afresh. The superseded correction goes to the
store first: nothing is deleted. A track the user kept is theirs.
Returns whether the library changed. Never raises."""
folder = os.path.dirname(os.path.abspath(dest))
stem = os.path.splitext(os.path.basename(dest))[0]
pattern = re.compile(re.escape(stem) + r"\.[^./\\]+(\.\d+)?\.srt$")
changed = False
for key, rec in records.items():
if (os.path.dirname(key) != folder
or not pattern.fullmatch(os.path.basename(key))
or rec.get("ref") == SUBSYNC_REF_VERSION
or rec.get("status") not in ("corrected", "unfixable")
or not os.path.exists(rec.get("original") or "")):
continue
try:
if os.path.exists(key):
_keep_original(key)
part = _part_path(key)
shutil.copyfile(rec["original"], part)
os.replace(part, key)
changed = True
log.info("subsync: %s restored from its original for re-measure",
os.path.basename(key))
except Exception as exc:
log.warning("subsync: %s not restored (%s)", os.path.basename(key), exc)
return changed
def _sync_subtitles(dest: str) -> dict:
"""Measure and repair every not-yet-checked sidecar beside one copy.
Idempotent: a track whose file identity matches its record is skipped, so
the ingest path and the backfill share this and re-runs cost nothing.
Never raises.
"""
result = {"tracks": 0, "in_sync": 0, "corrected": 0, "unfixable": 0,
"unmeasured": 0}
if not _subsync_available():
return result
try:
records = _load_sync()
except Exception:
return result # unreadable record: touch nothing
changed = _restore_stale_originals(dest, records)
def current(p: str) -> bool:
rec = records.get(os.path.abspath(p)) or {}
return rec.get("identity") == _file_identity(p) and (
rec.get("status") == "kept" or rec.get("ref") == SUBSYNC_REF_VERSION)
# In scope: the focus languages, plus any track an earlier pass measured
# -- the all-language pass of 10-04 left its results on every language.
todo = [p for p in _sidecars_of(dest)
if (SUBSYNC_LANGUAGES is None
or _sidecar_lang(dest, p) in SUBSYNC_LANGUAGES
or os.path.abspath(p) in records)
and not current(p)]
result["tracks"] = len(todo)
if not todo:
if changed:
_notify_jellyfin(dest, "Modified")
return result
fresh: dict[str, dict] = {}
with tempfile.TemporaryDirectory(prefix="subsync-") as work:
ref = _subsync_reference(dest, work)
if not ref:
result["unmeasured"] = len(todo)
return result
for path in todo:
if _cancel_requested():
break
# Output names KEEP the .srt extension: ffsubsync picks the output
# format from it, and for anything else (".synced", /dev/null) it
# logs "unsupported output format", writes nothing and exits 0 --
# after printing the offsets. That withdrew 32 good tracks once.
name = os.path.basename(path)
corrected = os.path.join(work, "synced-" + name)
m = _subsync_measure(ref, path, corrected)
if not m:
result["unmeasured"] += 1
continue
entry = {"offset": m["offset"], "scale": m["scale"], "ts": time.time(),
"ref": SUBSYNC_REF_VERSION}
try:
if _in_tolerance(m):
entry["status"] = "in_sync"
else:
check = (_subsync_measure(ref, corrected,
os.path.join(work, "recheck-" + name))
if os.path.exists(corrected) else None)
if check is None:
# No correction to judge, or no way to judge it: we
# could not repair this track, which is NOT evidence it
# is unfixable. A check that cannot run must not cost
# the user a subtitle.
log.warning("subsync: %s measured (%.2f s) but not "
"repairable here -- left as it was",
name, m["offset"])
result["unmeasured"] += 1
continue
entry["original"] = _keep_original(path)
if _in_tolerance(check):
# Across filesystems (/tmp -> the USB library), so a
# copy to .part and a rename, not os.replace.
part = _part_path(path)
shutil.copyfile(corrected, part)
os.replace(part, path)
entry["status"] = "corrected"
else:
os.remove(path) # the original is in the store
entry["status"] = "unfixable"
changed = True
except Exception as exc:
log.warning("subsync: %s left as it was (%s)",
os.path.basename(path), exc)
result["unmeasured"] += 1
continue
entry["identity"] = _file_identity(path)
fresh[os.path.abspath(path)] = entry
result[entry["status"]] += 1
log.info("subsync: %s %s (offset %.2f s, scale %.3f)",
os.path.basename(path), entry["status"],
m["offset"], m["scale"])
try:
with _subsync_guard:
records = _load_sync()
records.update(fresh)
tmp = SUBTITLE_SYNC_FILE + ".tmp"
with open(tmp, "w") as f:
json.dump(records, f)
os.replace(tmp, SUBTITLE_SYNC_FILE)
except Exception as exc:
log.warning("subsync: record save failed (%s)", exc)
if changed:
_notify_jellyfin(dest, "Modified")
return result
def _subtitle_summary(dest: str) -> str:
"""One line for the watch page: what the sync found for this copy."""
present = _sidecars_of(dest)
if not present:
return "none"
try:
records = _load_sync()
except Exception:
records = {}
stem = os.path.splitext(os.path.abspath(dest))[0] + "."
if SUBSYNC_LANGUAGES is not None:
return _subtitle_summary_focused(dest, present, records, stem)
counts = {"corrected": 0, "in_sync": 0, "unchecked": 0}
for p in present:
rec = records.get(os.path.abspath(p)) or {}
status = rec.get("status") if rec.get("identity") == _file_identity(p) else None
counts[status if status in counts else "unchecked"] += 1
removed = sum(1 for k, v in records.items()
if k.startswith(stem) and v.get("status") == "unfixable")
parts = [f"{len(present)} tracks"]
for label, n in (("corrected", counts["corrected"]),
("in sync", counts["in_sync"]),
("not yet checked", counts["unchecked"]),
("removed (out of sync)", removed)):
if n:
parts.append(f"{n} {label}")
return " · ".join(parts)
def _subtitle_summary_focused(dest: str, present: list, records: dict,
stem: str) -> str:
"""English first: what the sync did to it, then how many other languages
sit beside the copy unchecked."""
focus = [p for p in present if _sidecar_lang(dest, p) in SUBSYNC_LANGUAGES]
others = len(present) - len(focus)
removed = [k for k, v in records.items()
if k.startswith(stem) and v.get("status") == "unfixable"
and _sidecar_lang(dest, k) in SUBSYNC_LANGUAGES]
def describe(p: str) -> str:
rec = records.get(os.path.abspath(p)) or {}
if rec.get("identity") != _file_identity(p):
return "not yet checked"
if rec.get("status") == "corrected":
return (f"corrected (was {rec['offset']:+.2f} s, "
f"speed {rec['scale']:.3f})").replace("+", "")
if rec.get("status") == "kept":
# Withdrawn as out of sync, then restored by the user: the sync
# skips it (identity matches) and says what it measured.
return f"kept as fetched (measured {rec['offset']:.2f} s off)"
return {"in_sync": "in sync"}.get(rec.get("status"), "not yet checked")
if len(focus) == 1:
head = f"English: {describe(focus[0])}"
elif focus:
states = [describe(p).split(" (")[0] for p in focus]
head = f"English ({len(focus)}): " + " · ".join(
f"{states.count(x)} {x}" for x in dict.fromkeys(states))
else:
head = ("English: removed (out of sync)" if removed else "no English")
if others:
head += (f" · {others} other language{'s' if others != 1 else ''} "
"not checked")
return head
# OpenSubtitles through its stremio addon -- the host Jacktook already uses, no
# key. Asked only when a copy has NO English after wyzie and the sync: on
# 2026-10-18 wyzie listed 9 tracks for Hope and 503'd every file, while this
# had 4 English tracks, each named for its release, all measuring 0.13 s /
# 1.000 against our copy. Capped, because each one becomes a menu entry.
OPENSUBS_URL = "https://opensubtitles-v3.strem.io"
OPENSUBS_MAX_TRACKS = 3
def _opensubs_tracks(kind: str, imdb_id: str, season: int = 0,
episode: int = 0) -> list[dict]:
"""English tracks in _write_sidecars' shape, or [] -- never an exception."""
path = (f"series/{imdb_id}:{season}:{episode}" if kind == "series"
else f"movie/{imdb_id}")
try:
r = requests.get(f"{OPENSUBS_URL}/subtitles/{path}.json",
timeout=REQUEST_TIMEOUT, headers={"User-Agent": UA})
if r.status_code != 200:
log.info("opensubs: none for %s (HTTP %s)", imdb_id, r.status_code)
return []
subs = r.json().get("subtitles") or []
except Exception as exc:
log.info("opensubs: none for %s (%s)", imdb_id, exc)
return []
english = [{"language": "en", "url": s["url"]} for s in subs
if isinstance(s, dict) and s.get("lang") == "eng" and s.get("url")]
log.info("opensubs: %d English track(s) for %s", len(english), imdb_id)
return english[:OPENSUBS_MAX_TRACKS]
def _opensubs_for(kind: str, tmdb_id: str, season: int = 0,
episode: int = 0) -> list[dict]:
"""OpenSubtitles is keyed by imdb id only. A seam: conftest refuses it, so
no test reaches TMDB or OpenSubtitles without a stub."""
imdb_id = _imdb_id_for(kind, f"tmdb:{tmdb_id}")
return _opensubs_tracks(kind, imdb_id, season, episode) if imdb_id else []
def _has_english(dest: str) -> bool:
return any(_sidecar_lang(dest, p) == "en" for p in _sidecars_of(dest))
def _sync_subtitles_safely(dest: str) -> None:
# Measured against THIS copy's audio, because subtitle tracks are cut for
# other releases (labs#13). The copy is already published; this only ever
# replaces or withdraws sidecars, and never raises.
_job_update(getattr(_job_ctx, "key", None) or "", phase="syncing subtitles")
try:
_sync_subtitles(dest)
except Exception as exc:
log.warning("subsync: skipped for %s (%s)", os.path.basename(dest), exc)
def _fetch_subtitles_for(dest: str, meta: dict, season: int, episode: int,
kind: str = "movie") -> None:
"""Put whatever subtitles exist beside a published copy. Never raises.
Called AFTER the copy is complete and published, so there is nothing a
subtitle failure could usefully abort -- and an ingest that reported failure
because a subtitle host was down would send the whole 3 GB round again.
"""
try:
tracks = _wyzie_tracks(meta["tmdb_id"], season=season, episode=episode)
_write_sidecars(dest, tracks)
except Exception as exc:
log.warning("subtitles: skipped for %s (%s)",
os.path.basename(dest), exc)
_sync_subtitles_safely(dest)
# Wyzie gave nothing usable in English (none, a 503, or withdrawn by the
# sync as a different cut): ask the second source, and measure that too.
try:
if _has_english(dest):
return
if not _write_sidecars(dest, _opensubs_for(kind, meta["tmdb_id"],
season, episode)):
return
except Exception as exc:
log.warning("opensubs: skipped for %s (%s)", os.path.basename(dest), exc)
return
_sync_subtitles_safely(dest)
# Once a day, the ingest's subtitle step again for every fetched copy that has
# never had English (2026-10-18: wyzie 503'd all evening and 43 copies had
# none, with nothing to retry them). "Never had": English measured and
# withdrawn as another cut already had both sources tried.
SUBS_RETRY_FIRST_DELAY = 10 * 60
SUBS_RETRY_INTERVAL = 24 * 3600
def _never_had_english(dest: str, records: dict) -> bool:
if _has_english(dest):
return False
folder = os.path.dirname(os.path.abspath(dest))
stem = os.path.splitext(os.path.basename(dest))[0]
pattern = re.compile(re.escape(stem) + r"\.en(\.\d+)?\.srt$")
return not any(os.path.dirname(k) == folder and pattern.fullmatch(os.path.basename(k))
for k in records)
def _subtitle_retry_pass() -> dict:
"""One pass over the fetched copies. Never raises."""
result = {"checked": 0, "retried": 0, "gained": 0}
try:
records = _load_sync()
except Exception:
return result # unreadable record: touch nothing
for dest, keys in sorted(_fetched_dests().items()):
if not os.path.exists(dest):
continue
result["checked"] += 1
if not _never_had_english(dest, records):
continue
if any(_mp4_jobs.get(k) == "running" for k in keys):
continue # its own ingest asks for subtitles
parts = keys[0].split(":") # kind:ident:season:episode:height
kind, ident = parts[0], ":".join(parts[1:-3])
season, episode = int(parts[-3]), int(parts[-2])
try:
meta = _library_meta(kind, ident, season, episode)
if not meta:
continue
result["retried"] += 1
_fetch_subtitles_for(dest, meta, season, episode, kind=kind)
if _has_english(dest):
result["gained"] += 1
except Exception as exc:
log.warning("subs-retry: %s skipped (%s)", os.path.basename(dest), exc)
log.info("subs-retry: %d fetched copies, %d without English retried, %d gained it",
result["checked"], result["retried"], result["gained"])
return result
def _subtitle_retry_loop() -> None:
time.sleep(SUBS_RETRY_FIRST_DELAY)
while True:
try:
_subtitle_retry_pass()
except Exception as exc:
log.warning("subs-retry: pass failed (%s)", exc)
time.sleep(SUBS_RETRY_INTERVAL)
def _start_subtitle_retry() -> None:
threading.Thread(target=_subtitle_retry_loop, name="subs-retry",
daemon=True).start()
def _run_ingest(kind: str, ident: str, season: int, episode: int,
height: int, force: bool = False) -> str:
"""Fetch one title into the library. Synchronous; returns the final state.
Split out from the thread so the orchestration is testable without one, and
so the ordering below is explicit: identify FIRST, and decline outright if we
cannot. Everything after that point writes a file, and a file we cannot name
correctly is worse than no file at all.
Resolution happens HERE rather than when the job was queued: variant URLs are
signed and IP-locked with a ~4h lifetime, so a job that sat in a queue would
otherwise start against a dead link.
"""
key = _mp4_key(kind, ident, season, episode, height)
# The registry decides whether ANYTHING may feed the library; the order
# below (vidcore, then HdHub) is ingest_sources()'s order, written out.
if not ingest_source():
return "failed: no ingestable source"
# Checked before the resolve, so a full volume costs no browser at all.
if not _enough_space():
log.warning("ingest: declining %s — under the free-space floor", key)
return "failed: below the free-space floor"
meta = _library_meta(kind, ident, season, episode)
if not meta:
return f"failed: cannot identify {ident}"
dest = _library_path(kind, meta["title"], meta["year"], meta["tmdb_id"],
season, episode, meta["episode_title"])
if force:
# A human is asserting the copy is wrong in a way no measurement can
# see: the wrong episode, the wrong season, unwatchable audio. Its
# verification record only ever attested that it matched the source it
# came from, so it does not get to veto them.
_forget_verified(dest)
log.info("ingest: forced re-fetch of %s", os.path.basename(dest))
elif _library_has(kind, ident, season, episode):
log.info("ingest: library already holds %s", os.path.basename(dest))
# Still ask for subtitles. This early return exists so an existing copy
# costs no browser, and one HTTP call does not change that -- while
# skipping it would mean the library only ever gains subtitles for
# things fetched after today.
_fetch_subtitles_for(dest, meta, season, episode, kind=kind)
return "done"
# vidcore first: it is asked by TMDB id, so what it returns IS this title.
# HdHub is the fallback because its answer is matched by filename -- see
# ingest_sources(). A failure of either kind (no resolve, or a resolve onto
# a CDN that refuses the bytes, as on 2026-09-28) falls through. A source a
# human rejected for this file is skipped (SOURCE_REJECTS_FILE).
order = _ingest_order(dest)
if not order:
return "failed: every source is rejected for this title"
started = time.time()
failure = None
_job_update(key, phase="resolving", dest=dest)
url = (_resolve_vidcore(kind, ident, season, episode)
if "vidcore" in order else None)
# The resolve is a browser subprocess and is let run to its end; Cancel is
# honoured the moment it returns, before any byte is fetched.
if _cancel_requested():
return "cancelled"
if "vidcore" not in order:
failure = "failed: vidcore rejected for this title"
elif url:
variant = _cap_variant(url, height)
# The preview plays exactly what is being copied (labs#13).
picked = next(((h, bw) for h, bw, uri, _ in _variants(url)
if urljoin(url, uri) == variant), (None, None))
_job_update(key, preview={"type": "hls", "url": variant,
"height": picked[0], "bandwidth": picked[1]})
log.info("ingest start: %s -> %s", key, dest)
_job_update(key, phase="downloading", source="vidcore",
downloading_since=time.time())
try:
_remux_to_mp4(variant, dest)
except RemuxCancelled:
log.info("ingest cancelled: %s", key)
return "cancelled"
except Exception as exc:
log.warning("ingest failed for %s: %s", key, exc)
failure = f"failed: {exc}"
else:
failure = "failed: could not resolve title"
if failure:
if _cancel_requested():
return "cancelled"
if "hdhub" not in order:
return f"{failure}; hdhub: rejected for this title"
started = time.time()
_job_update(key, phase="downloading", source="hdhub",
downloading_since=started)
why = _ingest_from_hdhub(kind, ident, season, episode, height, dest, meta)
# _ingest_from_hdhub turns every exception into a reason string, a
# cancel included, so the event is the authority here, not `why`.
if _cancel_requested():
log.info("ingest cancelled: %s", key)
return "cancelled"
if why:
log.info("ingest: hdhub fallback for %s declined: %s", key, why)
return f"{failure}; hdhub: {why}"
log.info("ingest done: %s in %.0fs (%.2f GB)", key,
time.time() - started, os.path.getsize(dest) / 1e9)
_fetch_subtitles_for(dest, meta, season, episode, kind=kind)
# After the sidecars, so one notification covers the copy and them.
_notify_jellyfin(dest, "Created")
return "done"
def _start_mp4_job(kind: str, ident: str, season: int, episode: int,
height: int, force: bool = False) -> str:
"""Kick off an ingest in the background. Returns the job state.
The job holds no reference to any playback session, so pausing or stopping
the player cannot affect it — which is the whole point. Nothing here may ever
become cancellable-on-stop.
"""
key = _mp4_key(kind, ident, season, episode, height)
with _mp4_jobs_guard:
state = _mp4_jobs.get(key)
if state == "running":
return state
# The arguments are persisted alongside the state so a restart mid-remux
# can pick the job up again — a key on its own cannot be re-run.
_set_job_state(key, "running", args=(kind, ident, season, episode, height))
threading.Thread(target=_run_job, name=f"mp4-{key}",
args=(kind, ident, season, episode, height, force),
daemon=True).start()
return "running"
def _run_job(kind: str, ident: str, season: int, episode: int,
height: int, force: bool = False,
cancel: Optional[threading.Event] = None) -> str:
"""One ingest, through the remux semaphore, with its state recorded.
Split out of the thread so a season can drive episodes SEQUENTIALLY through
the same gate. Spawning a job per episode instead would leave every one past
the second blocked on the semaphore for up to MP4_REMUX_TIMEOUT, and a long
season would start reporting "remux slots busy" for episodes nothing was
actually wrong with.
"""
key = _mp4_key(kind, ident, season, episode, height)
started = time.time()
# A season passes in an event it registered BEFORE checking its own cancel
# flag, so a season Cancel cannot fall between choosing this episode and
# this job becoming cancellable.
cancel = cancel or threading.Event()
label = f"S{season:02d}E{episode:02d}" if kind == "series" else "movie"
with _mp4_jobs_guard:
_job_info[key] = {"phase": "waiting", "source": None, "dest": None,
"started": started, "kind": kind, "ident": ident,
"season": season, "episode": episode,
"height": height, "label": label, "title": ident}
_cancel_events[key] = cancel
# A season drives episodes through here directly, never through
# _start_mp4_job, so without this an episode being fetched by a season
# read as not-running -- and the web page offered to fetch it again.
if _mp4_jobs.get(key) != "running":
_set_job_state(key, "running",
args=(kind, ident, season, episode, height))
_job_ctx.cancel = cancel
_job_ctx.key = key
meta = None
acquired = False
try:
# The file is claimed BEFORE a remux slot, so a job waiting on another
# ident's copy of the same file does not sit on one of the two slots
# while it waits. Once it gets the file, _run_ingest's _library_has
# check finds the copy the first job made and answers "done" without
# fetching again. No identity (TMDB down) means no file to claim;
# _run_ingest then declines on its own, before writing anything.
meta = _library_meta(kind, ident, season, episode)
if meta:
_job_update(key, title=meta["title"])
claim = (_dest_lock(_library_path(kind, meta["title"], meta["year"],
meta["tmdb_id"], season, episode,
meta["episode_title"]))
if meta else None)
# Both waits poll, so a job still queued behind another can be
# cancelled without first waiting out the one in front of it.
if claim is not None and not _acquire_or_cancel(claim, cancel):
state = "cancelled"
else:
try:
acquired = _acquire_or_cancel(_mp4_sem, cancel,
timeout=MP4_REMUX_TIMEOUT)
if cancel.is_set():
state = "cancelled"
elif not acquired:
state = "failed: remux slots busy"
else:
state = _run_ingest(kind, ident, season, episode, height,
force)
finally:
if acquired:
_mp4_sem.release()
if claim is not None:
claim.release()
except Exception as exc: # never leave a job "running"
log.warning("ingest crashed for %s: %s", key, exc)
state = f"failed: {exc}"
finally:
_job_ctx.cancel = None
_job_ctx.key = None
info = _job_info.get(key) or {}
_record_attempt(key, info, state, started)
with _mp4_jobs_guard:
_set_job_state(key, state)
_job_info.pop(key, None)
_cancel_events.pop(key, None)
return state
def _released_episodes(tmdb_id: str, season: int) -> list[int]:
"""Episode numbers of this season that have actually aired.
By air date, never by episode count: TMDB lists a running season's unaired
episodes too, and asking vidcore for one costs a full browser resolve to
learn nothing. An episode with no air_date at all counts as unreleased --
that field is how "released" is expressed here, and guessing the other way
queues work that cannot succeed.
"""
try:
data = _tmdb_get(f"tv/{tmdb_id}/season/{season}")
except Exception as exc:
log.warning("season listing failed for %s S%s: %s", tmdb_id, season, exc)
return []
today = time.strftime("%Y-%m-%d")
out = []
for ep in data.get("episodes") or []:
air = (ep.get("air_date") or "").strip()
num = ep.get("episode_number")
if air and air <= today and isinstance(num, int):
out.append(num)
return sorted(out)
_season_jobs: dict[str, str] = {}
_season_guard = threading.Lock()
# Live season jobs only, like _job_info: position in the episode list and the
# episode being fetched, for the download manager.
_season_info: dict[str, dict] = {}
_season_cancel: dict[str, threading.Event] = {}
def _cancel_season(key: str) -> bool:
"""Stop a season job: no further episodes, and the current one cancelled."""
with _season_guard:
event = _season_cancel.get(key)
if event is None:
return False
# Both set under the season lock -- the same lock the loop holds while
# it registers the next episode -- so no episode can start unseen.
event.set()
current = (_season_info.get(key) or {}).get("current_cancel")
if current is not None:
current.set()
log.info("season: cancel requested for %s", key)
return True
def _start_season_job(ident: str, season: int, height: int,
force: bool = False) -> str:
"""Fetch every released episode of a season, one at a time.
This is _queue_next_episode's lookahead of 1 raised to a whole season -- the
same machinery, the same persistent queue, the same semaphore. Nothing new
holds the work: a restart mid-season resumes from ingest_queue.json exactly
as it already did for a single episode.
"""
key = f"season:{ident}:{season}:{height}"
cancel = threading.Event()
with _season_guard:
if _season_jobs.get(key) == "running":
return "running"
_season_jobs[key] = "running"
_season_cancel[key] = cancel
_season_info[key] = {"ident": ident, "season": season,
"height": height, "episodes": [], "index": 0,
"current": None, "started": time.time()}
def run() -> None:
done = 0
try:
meta = _library_meta("series", ident, season, 1)
episodes = _released_episodes(meta["tmdb_id"], season) if meta else []
_season_info[key].update(episodes=episodes,
title=meta["title"] if meta else ident)
log.info("season %s S%02d: %d released episode(s)",
ident, season, len(episodes))
for i, episode in enumerate(episodes, 1):
if not force and _library_has("series", ident, season, episode):
continue
episode_cancel = threading.Event()
with _season_guard:
if cancel.is_set():
break
_season_info[key].update(
index=i, current_cancel=episode_cancel,
current=_mp4_key("series", ident, season, episode,
height))
_run_job("series", ident, season, episode, height, force,
cancel=episode_cancel)
done += 1
except Exception as exc:
log.warning("season job crashed for %s: %s", key, exc)
finally:
with _season_guard:
_season_jobs[key] = (f"cancelled ({done})" if cancel.is_set()
else f"done ({done})")
_season_cancel.pop(key, None)
_season_info.pop(key, None)
threading.Thread(target=run, name=f"season-{key}", daemon=True).start()
return "running"
def _playable_path(kind: str, ident: str, season: int = 0, episode: int = 0,
height: int = VIDCORE_MAX_HEIGHT) -> Optional[str]:
"""The best complete local file for this title, or None.
Library first, then the legacy mp4-cache. The fallback is not vestigial:
copies made before the library existed must stay reachable. Nothing here
deletes, so nothing here may orphan either.
Falls back cleanly when TMDB is unreachable — _library_meta returns None and
the legacy path still answers.
"Complete" is enforced here rather than at the row, because Jacktook caches
the source list for 24h: a row can be a day stale, but this is on the path
of every play. It is the only place that can refuse a partial copy no matter
which cached row asked for it.
"""
meta = _library_meta(kind, ident, season, episode)
if meta:
dest = _library_path(kind, meta["title"], meta["year"], meta["tmdb_id"],
season, episode, meta["episode_title"])
try:
if os.path.getsize(dest) > 0 and _copy_is_complete(dest, meta):
return dest
except OSError:
pass
legacy = _mp4_path(kind, ident, season, episode, height)
try:
if os.path.getsize(legacy) > 0 and _copy_is_complete(legacy, meta):
return legacy
except OSError:
pass
return None
def _ensure_ingest(kind: str, ident: str, season: int = 0, episode: int = 0,
height: int = VIDCORE_MAX_HEIGHT) -> None:
"""Start a copy of this title unless there is a reason not to.
Idempotent: _start_mp4_job already refuses to run the same key twice, so
repeated plays of the same episode enqueue nothing new.
"""
if not ident:
return
if not _library_meta(kind, ident, season, episode):
return # cannot be named, so cannot be ingested
if _library_has(kind, ident, season, episode):
return
_start_mp4_job(kind, ident, season, episode, height)
def _ingest_progress(kind: str, ident: str, season: int = 0,
episode: int = 0) -> Optional[int]:
"""Bytes written so far for an in-flight copy, or None if none is running.
Read off the .part file rather than parsed out of ffmpeg: it is exact, needs
no -progress plumbing, and cannot drift from reality — the bytes on disk ARE
the progress.
A percentage is deliberately not offered. It would need a total size we do
not know up front (the variant's BANDWIDTH is a ceiling, not a measurement),
and a confidently wrong percentage reads worse than an honest byte count.
"""
meta = _library_meta(kind, ident, season, episode)
if not meta:
return None
dest = _library_path(kind, meta["title"], meta["year"], meta["tmdb_id"],
season, episode, meta["episode_title"])
try:
return os.path.getsize(_part_path(dest))
except OSError:
return None
def _next_episode(tmdb_id: str, season: int,
episode: int) -> Optional[tuple[int, int]]:
"""What follows this episode per TMDB, or None if nothing does.
TMDB is the bound on rolling forward: it knows where a season ends and where
the series does. Any failure returns None — stopping is always safe, whereas
guessing an episode number fetches something that does not exist.
"""
try:
cur = _tmdb_get(f"tv/{tmdb_id}/season/{season}")
numbers = {e.get("episode_number") for e in cur.get("episodes", [])}
if episode + 1 in numbers:
return (season, episode + 1)
nxt = _tmdb_get(f"tv/{tmdb_id}/season/{season + 1}")
following = sorted(n for n in (e.get("episode_number")
for e in nxt.get("episodes", [])) if n)
if following:
return (season + 1, following[0])
except Exception as exc:
log.info("rolling: no next episode after S%02dE%02d (%s)",
season, episode, exc)
return None
def _queue_next_episode(ident: str, season: int, episode: int,
height: int) -> None:
"""Fetch the episode after this one, so a series becomes a local library.
This is the only trigger that fetches something the user did not choose
directly, so it stops on the FIRST sign of doubt, at a deliberately lower
threshold than _rep_is_known_miss uses for hiding a row.
The asymmetry is intentional. Hiding a row costs the user the title outright
if we are wrong, so that needs two consecutive strikes. Declining a
speculative prefetch costs nothing — the episode stays fully playable, and
playing it resolves normally and re-files its own reputation. So one strike
is enough to not go looking.
"""
meta = _library_meta("series", ident, season, episode)
if not meta:
return
nxt = _next_episode(meta["tmdb_id"], season, episode)
if not nxt:
return
ns, ne = nxt
rep = _vidcore_rep.get(_rep_key("series", ident, ns, ne)) or {}
if rep.get("status") == "miss" and int(rep.get("strikes", 0)) >= 1:
log.info("rolling: S%02dE%02d previously absent — not prefetching", ns, ne)
return
_ensure_ingest("series", ident, ns, ne, height)
def _serve_mp4(kind: str, ident: str, season: int, episode: int):
height = _requested_height()
if kind == "series":
# Also from here, not just the vidcore route: once a series is in the
# library it is played from disk, and rolling forward would otherwise
# stall the moment it started working.
_queue_next_episode(ident, season, episode, height)
path = _playable_path(kind, ident, season, episode, height)
if path:
# conditional=True is what answers Range requests, and Range is what
# makes the file seekable in the player.
return send_file(path, mimetype="video/mp4", conditional=True)
# No copy yet. Start one, and hand the player the live stream meanwhile:
# this row has to be playable the instant it is clicked, and a 503 would
# surface as a playback failure while the copy silently proceeded. The same
# URL serves the local file once it lands, so the row upgrades itself.
_ensure_ingest(kind, ident, season, episode, height)
# A progressive file first when HdHub has one: it seeks, and it is not on
# vidcore's CDN -- which on 2026-09-28 was banned outright, turning every
# vidcore redirect from here into a dead link.
if _hdhub_enabled():
picked = _hdhub_pick(kind, ident, season, episode, height)
if picked:
log.info("mp4: no copy yet — serving hdhub %s", picked[0].filename)
return redirect(picked[0].url, code=302)
url = _resolve_vidcore(kind, ident, season, episode,
wait=VIDCORE_PLAY_DEADLINE)
if not url:
abort(404)
return _serve_vidcore(url)
@app.route("/mp4/movie/<path:ident>.mp4")
def mp4_movie(ident: str):
if not re.fullmatch(VIDCORE_IDENT, ident):
abort(404)
return _serve_mp4("movie", ident, 0, 0)
@app.route("/mp4/series/<path:ident>.mp4")
def mp4_series(ident: str):
m = re.fullmatch(rf"({VIDCORE_IDENT}):(\d+):(\d+)", ident)
if not m:
abort(404)
return _serve_mp4("series", m.group(1), int(m.group(2)), int(m.group(3)))
# ------------------------- a library copy's own subtitles -------------------
# The sync (labs#13) corrects the sidecars beside a copy, and only Jellyfin
# reads that folder. Kodi plays the copy through /mp4/, so Jacktook fetched
# wyzie itself into a per-TMDB cache: on 2026-10-18 the Pi's No Other Choice
# English.srt was byte-identical to the UNSYNCED original -- the drift the sync
# had already fixed. These routes hand a player the file's own sidecars under
# the same id the file is served by. Like /web/file/, they never fetch.
def _subs_target(kind: str, ident: str) -> Optional[tuple]:
"""(kind, ident, season, episode) from a /subs/ ident, or None."""
if kind == "movie" and re.fullmatch(VIDCORE_IDENT, ident):
return kind, ident, 0, 0
m = re.fullmatch(rf"({VIDCORE_IDENT}):(\d+):(\d+)", ident)
if kind == "series" and m:
return kind, m.group(1), int(m.group(2)), int(m.group(3))
return None
def _subs_by_id(kind: str, ident: str) -> dict:
"""{id: sidecar path} for the held copy; id is `<code>[.<n>]`."""
target = _subs_target(kind, ident)
if not target:
abort(404)
path = _playable_path(*target)
if not path:
return {}
stem = os.path.splitext(os.path.basename(path))[0] + "."
return {os.path.basename(p)[len(stem):-len(".srt")]: p
for p in _sidecars_of(path)}
@app.route("/subs/<kind>/<path:ident>.json")
def subs_list(kind: str, ident: str):
live = _live_request(kind, ident)
if live:
return jsonify(_live_answer(kind, ident, *live))
out = []
for sid, p in sorted(_subs_by_id(kind, ident).items()):
out.append({"id": sid, "lang": re.sub(r"\.\d+$", "", sid),
"url": f"http://{request.host}/subs/{kind}/{ident}/{sid}.srt",
# Lets a client tell a re-synced file from the one it holds.
"identity": _file_identity(p)})
return jsonify({"subtitles": out})
@app.route("/subs/<kind>/<path:ident>/<sid>.srt")
def subs_file(kind: str, ident: str, sid: str):
# Looked up among the copy's own sidecars (or the live sync's own list),
# never joined onto a path.
live = _live_request(kind, ident)
path = (_live_file(kind, ident, sid, *live) if live
else _subs_by_id(kind, ident).get(sid))
if not path:
abort(404)
return send_file(path, mimetype="application/x-subrip")
# ---------------------------- live streams: subtitles synced to the stream ---
# A stream played from vidcore or HdHub has no copy to measure against, and
# taking all its audio costs what the download does (Hope: 767 s). So six
# 150 s audio windows of the stream being played are fetched (~150 MB, ~1 min)
# and window_align places each English track against them -- see
# test_window_align.py for why that, and not ffsubsync's own search. Only a
# track the windows could place is offered; the player keeps what it had for
# the rest. Results live in LIVE_SUBS_DIR, outside the library: a cache, not
# library state.
LIVE_SUBS_DIR = os.path.join(os.path.dirname(os.path.abspath(__file__)), "live-subs")
LIVE_SOURCES = ("vidcore", "hdhub", "mp4")
LIVE_WINDOW_FRACTIONS = (0.1, 0.25, 0.4, 0.55, 0.7, 0.85)
# 6 x 150 s, not 3 x 300 s: on Hope 3 x 300 left one window that matched (a
# quiet stretch, then music the VAD called 97% speech) and no answer; six
# shorter windows gave three that agreed.
LIVE_WINDOW_SECONDS = 150
LIVE_MAX_TRACKS = 3
LIVE_READY_TTL = 3 * 86400
LIVE_NONE_RETRY = 30 * 60 # "no answer" may be a 503 that clears
_live_jobs: dict[str, str] = {}
_live_guard = threading.Lock()
# One at a time: each is ~150 MB of segments and six VAD passes on stortron.
_live_slot = threading.Semaphore(1)
def _live_request(kind: str, ident: str) -> Optional[tuple]:
"""(src, height) when this asks about a stream with no copy behind it,
else None. A held copy is what plays, whatever `src` says: /vidcore/ and
/hdhub/ redirect to it."""
src = request.args.get("src")
if src is None:
return None
if src not in LIVE_SOURCES:
abort(404)
target = _subs_target(kind, ident)
if not target:
abort(404)
if _playable_path(*target):
return None
try:
height = int(request.args.get("h") or VIDCORE_MAX_HEIGHT)
except ValueError:
abort(404)
if height not in VIDCORE_HEIGHTS:
height = VIDCORE_MAX_HEIGHT
return src, height
def _live_key(kind: str, ident: str, src: str, height: int) -> str:
return f"{kind}:{ident}:{src}:{height}"
def _live_dir(key: str) -> str:
return os.path.join(LIVE_SUBS_DIR, re.sub(r"[^\w.-]", "_", key))
def _live_state(key: str) -> Optional[dict]:
"""The finished result for `key`, or None (none yet, or out of date)."""
try:
with open(os.path.join(_live_dir(key), "state.json")) as f:
state = json.load(f)
except Exception:
return None
age = time.time() - float(state.get("ts") or 0)
ttl = LIVE_READY_TTL if state.get("status") == "ready" else LIVE_NONE_RETRY
return state if age < ttl else None
def _start_live_sync(key: str, target) -> None:
threading.Thread(target=target, name=f"livesubs-{key}", daemon=True).start()
def _live_answer(kind: str, ident: str, src: str, height: int) -> dict:
key = _live_key(kind, ident, src, height)
state = _live_state(key)
if state is None:
with _live_guard:
start = key not in _live_jobs
if start:
_live_jobs[key] = "running"
# Started outside the guard: the job takes it again when it finishes.
if start:
_start_live_sync(key, lambda: _run_live_sync(key, kind, ident,
src, height))
state = _live_state(key)
if state is None:
return {"status": "pending", "subtitles": []}
return {"status": state["status"], "subtitles": [
dict(t, url=(f"http://{request.host}/subs/{kind}/{ident}/{t['id']}.srt"
f"?src={src}&h={height}"))
for t in state.get("tracks") or []]}
def _live_file(kind: str, ident: str, sid: str, src: str,
height: int) -> Optional[str]:
key = _live_key(kind, ident, src, height)
state = _live_state(key) or {}
if sid not in {t["id"] for t in state.get("tracks") or []}:
return None
path = os.path.join(_live_dir(key), sid + ".srt")
return path if os.path.exists(path) else None
def _run_live_sync(key: str, kind: str, ident: str, src: str, height: int) -> None:
"""The job: never raises, always leaves a state behind."""
state = {"status": "none", "tracks": []}
try:
_, base, season, episode = _subs_target(kind, ident)
with _live_slot:
state = _live_sync(kind, base, season, episode, src, height,
_live_dir(key))
except Exception as exc:
log.warning("livesubs: %s failed (%s)", key, exc)
finally:
state["ts"] = time.time()
try:
os.makedirs(_live_dir(key), exist_ok=True)
tmp = os.path.join(_live_dir(key), "state.json.part")
with open(tmp, "w") as f:
json.dump(state, f)
os.replace(tmp, os.path.join(_live_dir(key), "state.json"))
except Exception as exc:
log.warning("livesubs: %s state not saved (%s)", key, exc)
with _live_guard:
_live_jobs.pop(key, None)
log.info("livesubs: %s -> %s (%s)", key, state["status"],
", ".join(f"{t['id']} {t['status']} {t['offset']:+.2f}s"
for t in state.get("tracks") or []) or "nothing placed")
def _live_sync(kind: str, ident: str, season: int, episode: int, src: str,
height: int, out_dir: str) -> dict:
import window_align as wa # numpy: only this path needs it
source = _live_source(kind, ident, season, episode, src, height)
if not source:
return {"status": "none", "tracks": []}
tracks = _live_tracks(kind, ident, season, episode)
if not tracks:
return {"status": "none", "tracks": []}
with tempfile.TemporaryDirectory(prefix="livesubs-") as work:
windows = _live_windows(source, work)
os.makedirs(out_dir, exist_ok=True)
placed = []
for sid, text in tracks:
m = wa.align(wa.srt_cues(text), windows)
if not m:
log.info("livesubs: %s %s -- the windows do not agree", ident, sid)
continue
entry = {"id": sid, "lang": re.sub(r"\.\d+$", "", sid),
"offset": m["offset"], "scale": m["scale"]}
if _in_tolerance(m):
entry["status"], body = "in_sync", text
else:
body = wa.apply(text, m["scale"], m["offset"])
if not _in_tolerance(wa.align(wa.srt_cues(body), windows)):
log.info("livesubs: %s %s -- correction does not hold", ident, sid)
continue
entry["status"] = "corrected"
part = os.path.join(out_dir, sid + ".srt.part")
with open(part, "w", encoding="utf-8") as f:
f.write(body)
os.replace(part, os.path.join(out_dir, sid + ".srt"))
placed.append(entry)
return {"status": "ready" if placed else "none", "tracks": placed}
def _live_tracks(kind: str, ident: str, season: int, episode: int) -> list:
"""[(id, text)] English tracks: wyzie's, else OpenSubtitles'."""
meta = _library_meta(kind, ident, season, episode)
if not meta:
return []
def fetch(tracks):
out = []
for t in tracks:
if t.get("language") != "en" or not t.get("url"):
continue
try:
body = _fetch_subtitle(t["url"])
except Exception as exc:
log.info("livesubs: track failed (%s)", exc)
continue
if body:
out.append(body.decode("utf-8", errors="replace"))
if len(out) >= LIVE_MAX_TRACKS:
break
return out
texts = fetch(_wyzie_tracks(meta["tmdb_id"], season=season, episode=episode))
if not texts:
try:
texts = fetch(_opensubs_for(kind, meta["tmdb_id"], season, episode))
except Exception as exc:
log.info("livesubs: opensubs skipped (%s)", exc)
return [("en" if n == 0 else f"en.{n + 1}", t) for n, t in enumerate(texts)]
def _live_source(kind: str, ident: str, season: int, episode: int, src: str,
height: int) -> Optional[dict]:
"""What the player is actually playing, for its audio. /mp4/ without a
copy serves HdHub first, then vidcore -- the same order _serve_mp4 uses."""
if src == "mp4":
src = ("hdhub" if _hdhub_enabled()
and _hdhub_pick(kind, ident, season, episode, height) else "vidcore")
if src == "vidcore":
master = _resolve_vidcore(kind, ident, season, episode)
if not master:
return None
# Every variant carries the same audio on the same clock; the
# smallest is the cheapest to sample (Hope 480p: 144 MB for 6 x 150 s).
return {"kind": "hls", "url": _cap_variant(master, min(VIDCORE_HEIGHTS)),
"headers": {"User-Agent": UA, "Referer": "https://vidcore.net/"}}
picked = _hdhub_pick(kind, ident, season, episode, height)
if not picked:
return None
f, probe = picked
audio = _probe_audio(probe)
english = [a for a in audio if _audio_lang(a) in ("eng", "en")] or audio
try:
duration = float(((probe or {}).get("format") or {}).get("duration"))
except (TypeError, ValueError):
return None
return {"kind": "file", "url": f.url, "headers": {}, "duration": duration,
"audio": english[0]["index"] if english else None}
def _hls_segments(body: str, base: str) -> tuple:
"""(init url or None, [(start seconds, url)]) from a media playlist."""
init, segs, t, dur = None, [], 0.0, None
for line in body.splitlines():
line = line.strip()
if line.startswith("#EXT-X-MAP:"):
m = re.search(r'URI="([^"]+)"', line)
init = urljoin(base, m.group(1)) if m else None
elif line.startswith("#EXTINF:"):
try:
dur = float(line[8:].split(",")[0])
except ValueError:
dur = None
elif line and not line.startswith("#") and dur is not None:
segs.append((round(t, 3), urljoin(base, line)))
t += dur
dur = None
return init, segs
def _window_plan(segs: list, fractions=LIVE_WINDOW_FRACTIONS,
seconds: float = LIVE_WINDOW_SECONDS) -> list:
"""[(first, end)] segment index ranges covering each window."""
if len(segs) < 2:
return []
starts = [s for s, _ in segs]
total = starts[-1] + (starts[-1] - starts[-2])
plan = []
for f in fractions:
i0 = bisect.bisect_left(starts, total * f)
i1 = bisect.bisect_left(starts, total * f + seconds)
if i0 < i1:
plan.append((i0, i1))
return plan
def _probe_start(path: str, stream: Optional[str] = None) -> float:
args = ["ffprobe", "-v", "error"]
args += (["-select_streams", stream, "-show_entries", "stream=start_time"]
if stream else ["-show_entries", "format=start_time"])
out = subprocess.run(args + ["-of", "csv=p=0", path], capture_output=True,
text=True, timeout=60).stdout
return float(out.strip().splitlines()[0].strip(","))
def _hls_window_wavs(source: dict, work: str) -> list:
"""[(position, wav)] for an HLS variant. ffmpeg cannot seek vidcore's
muxed fMP4 (it read for 2.5 min and wrote nothing), so each window is the
init segment plus the segments covering it, concatenated: a valid fMP4
whose own timestamps place it. Position = its audio start minus the
stream's start, i.e. the player's clock."""
headers = source["headers"]
r = requests.get(source["url"], headers=headers, timeout=REQUEST_TIMEOUT)
r.raise_for_status()
init, segs = _hls_segments(r.text, source["url"])
init_bytes = b""
if init:
init_bytes = requests.get(init, headers=headers, timeout=REQUEST_TIMEOUT).content
def get(url):
resp = requests.get(url, headers=headers, timeout=30)
resp.raise_for_status()
return resp.content
def assemble(name, urls):
path = os.path.join(work, name)
with ThreadPoolExecutor(max_workers=4) as pool:
parts = list(pool.map(get, urls))
with open(path, "wb") as f:
f.write(init_bytes)
for p in parts:
f.write(p)
return path
out = []
if len(segs) < 2:
return out
base = _probe_start(assemble("first.mp4", [segs[0][1]]))
for n, (i0, i1) in enumerate(_window_plan(segs)):
try:
path = assemble(f"win{n}.mp4", [u for _, u in segs[i0:i1]])
pos = _probe_start(path, "a:0") - base
wav = os.path.join(work, f"w{n}.wav")
subprocess.run(["ffmpeg", "-v", "error", "-y", "-i", path, "-vn",
"-ac", "1", "-ar", "16000", "-af", "aresample=async=1",
wav], check=True, capture_output=True, timeout=300)
out.append((pos, wav))
except Exception as exc:
log.info("livesubs: window %d skipped (%s)", n, exc)
return out
def _file_window_wavs(source: dict, work: str) -> list:
"""[(position, wav)] for a progressive file: ranged seeks, which a file
(unlike vidcore's HLS) allows. The English audio stream, whatever the
player has selected: timing is the video's either way."""
out = []
for n, f in enumerate(LIVE_WINDOW_FRACTIONS):
pos = int(source["duration"] * f)
wav = os.path.join(work, f"w{n}.wav")
cmd = ["ffmpeg", "-v", "error", "-y", "-rw_timeout", "30000000"]
if source.get("headers"):
cmd += ["-headers", "".join(f"{k}: {v}\r\n"
for k, v in source["headers"].items())]
cmd += ["-ss", str(pos), "-t", str(LIVE_WINDOW_SECONDS), "-i", source["url"]]
cmd += (["-map", f"0:{source['audio']}"] if source.get("audio") is not None
else ["-vn"])
cmd += ["-ac", "1", "-ar", "16000",
"-af", "aresample=async=1:first_pts=0", wav]
try:
subprocess.run(cmd, check=True, capture_output=True, timeout=300)
out.append((float(pos), wav))
except Exception as exc:
log.info("livesubs: window %d skipped (%s)", n, exc)
return out
def _live_windows(source: dict, work: str) -> list:
"""[(position, speech array)] -- VAD by ffsubsync, one window at a time."""
import numpy as np
wavs = (_hls_window_wavs if source["kind"] == "hls" else _file_window_wavs)(
source, work)
out = []
for pos, wav in wavs:
try:
subprocess.run([SUBSYNC_BIN, wav, "--serialize-speech"], cwd=work,
capture_output=True, timeout=SUBSYNC_TIMEOUT)
out.append((pos, np.load(os.path.splitext(wav)[0] + ".npz")["speech"]))
except Exception as exc:
log.info("livesubs: no speech from %s (%s)", os.path.basename(wav), exc)
return out
# ------------------------------ HdHub (direct files) ------------------------
# The second live source, and the first that hands over FILES rather than a
# player. Found 2026-09-28 in cinewave's bundle (watch.cinewave.qzz.io), where it
# is one of 22 servers and the only one that is not an embed: a real Stremio
# addon whose streams are plain mkv links on Cloudflare R2 and pixeldrain.
#
# It exists because one source is one outage. That evening Cloudflare blocked
# vidcore's CDN host `moon.quietridge.top` outright ("restricted ... due to Terms
# of Service violations", even on `/`), every master 403'd from every referer,
# vidcore's own player sat on "LOADING THINGS UP...", and nothing not already in
# the library could play.
#
# Measured from stortron before any of this was written:
# * asked by IMDb id it answers in 0.2-0.4s; asked by `tmdb:` it says "No
# streams found" (after up to 21s). So it is always asked by IMDb id.
# * R2 and pixeldrain rows answer a plain Range GET with 206, ffprobe reads
# them, and they download at ~10 MB/s -- ffmpeg can copy them, which is the
# property vidlink never had. Everything else ("10Gbps Download Only" on
# hubcloud, hubdrive) is HTML, a 400, or unmeasured, and is not used.
# * matching is fuzzy: Tony (2026) came back with "NCIS: Tony & Ziva" rows.
# * many rows are HEVC Main10 or 4K, which the Pi's H.264 decoder cannot play.
# * dual-audio files flag HINDI as the default track.
# * R2 links are pre-signed for 8h from when HdHub CACHED them, not from when
# we ask; one handed over at 22:50 UTC expired at 23:25.
HDHUB_BASE = ("https://hdhub.thevolecitor.qzz.io/"
# base64 of {"torbox":"unset","qualities":"2160p,1080p,720p",
# "sort":"desc"} -- the configuration cinewave itself sends.
"eyJ0b3Jib3giOiJ1bnNldCIsInF1YWxpdGllcyI6IjIxNjBwLDEwODBwLDcyMHAi"
"LCJzb3J0IjoiZGVzYyJ9")
HDHUB_TIMEOUT = 10
HDHUB_TTL = 20 * 60
# Hosts measured to serve bytes to a plain HTTP client. An allowlist rather than
# a blocklist: an unmeasured host is exactly how a dead row gets offered.
HDHUB_HOSTS = (".r2.cloudflarestorage.com", ".r2.dev", "pixeldrain.dev",
"pixeldrain.com")
# What cinewave filters out too, matched against the row's own text.
HDHUB_REJECT = ("10gbps", "hubcloud", "hubcdn", "castle", "file not found")
# A link must outlive what it is used for. Pressing play needs minutes; a copy
# of a 3.7 GB file at ~10 MB/s needs ~6 minutes plus reconnects, and a reconnect
# after expiry is a 403 that leaves a partial behind.
HDHUB_PLAY_MARGIN = 10 * 60
HDHUB_INGEST_MARGIN = 60 * 60
# A filename match is inferred identity. The last check before a file is named
# into the library is its length against TMDB's runtime: a different film or
# episode is almost never within 15% of this one.
HDHUB_RUNTIME_WINDOW = (0.85, 1.15)
# The play path has ~20s before Kodi gives up on an open. Probing is ~1-2s a
# file, so a handful of tries fits; an unbounded walk down a dead list does not.
HDHUB_PICK_BUDGET = 12
HDHUB_PICK_TRIES = 4
HDHUB_PROBE_TIMEOUT = 15
# Audio the MP4 container holds with a stream copy. DTS and TrueHD do not go in
# without re-encoding, so a file whose only audio is one of them is refused.
MP4_AUDIO = {"aac", "ac3", "eac3", "mp3"}
_FOREIGN = ("hindi", "tamil", "telugu", "malayalam", "kannada", "bengali",
"punjabi", "marathi", "dual", "multi")
_UNDECODABLE = re.compile(
r"x265|h\.?265|hevc|10.?bit|\bav1\b|\bdts\b|truehd|2160p|\b4k\b|\buhd\b", re.I)
@dataclass(frozen=True)
class HdHubFile:
url: str
filename: str
height: int
size: int
host: str
english_only: bool
_hdhub_cache: dict[tuple, tuple[float, list]] = {}
_hdhub_probe_cache: dict[str, tuple[float, Optional[dict]]] = {}
_hdhub_pick_cache: dict[tuple, tuple[float, tuple]] = {}
def _hdhub_query(kind: str, imdb: str, season: int = 0, episode: int = 0) -> list:
"""HdHub's raw stream rows for one title. Raises on any failure."""
key = (kind, imdb, season, episode)
hit = _hdhub_cache.get(key)
if hit and hit[0] > time.time():
return hit[1]
path = (f"series/{imdb}:{season}:{episode}" if kind == "series"
else f"movie/{imdb}")
r = requests.get(f"{HDHUB_BASE}/stream/{path}.json",
headers={"User-Agent": UA}, timeout=HDHUB_TIMEOUT)
r.raise_for_status()
rows = r.json().get("streams") or []
_hdhub_cache[key] = (time.time() + HDHUB_TTL, rows)
return rows
def _imdb_id_for(kind: str, ident: str) -> Optional[str]:
"""`tt<n>` verbatim, or a `tmdb:<n>` translated through TMDB."""
if ident.startswith("tt"):
return ident
tmdb_id = ident.split(":", 1)[1]
data = _tmdb_get(f"{'tv' if kind == 'series' else 'movie'}/{tmdb_id}/external_ids")
return data.get("imdb_id") or None
def _norm(text: str) -> str:
"""Lowercase words only: diacritics folded, punctuation to spaces."""
text = unicodedata.normalize("NFKD", text or "")
text = "".join(c for c in text if not unicodedata.combining(c)).lower()
text = text.replace("&", " and ")
return re.sub(r"[^a-z0-9]+", " ", text).strip()
def _hdhub_filename(description: str) -> str:
"""The file name: first line of the description, minus its [tags]."""
first = (description or "").split("\n", 1)[0]
return re.sub(r"^(\s*\[[^\]]*\])+\s*", "", first).strip()
def _hdhub_title_ok(filename: str, title: str, year: str, kind: str,
season: int = 0, episode: int = 0) -> bool:
"""Is this file the title asked for?
The title must LEAD the name -- which is what refuses "Ncis.Tony.&.Ziva"
for Tony -- and what follows it must be the year (movies, +/-1) or the exact
SxxEyy (series, optionally after a year). A season pack or a neighbouring
episode is a different file, not a close match.
"""
f, t = _norm(filename), _norm(title)
if not t or not f.startswith(t + " "):
return False
rest = f[len(t):].split()
if kind == "series":
tag = f"s{season:02d}e{episode:02d}"
i = 1 if re.fullmatch(r"(19|20)\d\d", rest[0]) else 0
return len(rest) > i and rest[i] == tag
if any(re.fullmatch(r"s\d{1,2}e\d{1,3}.*", tok) for tok in rest):
return False
if not (year and re.fullmatch(r"(19|20)\d\d", rest[0])):
return False
return abs(int(rest[0]) - int(year)) <= 1
def _hdhub_height(text: str) -> int:
"""Vertical resolution a row declares, or 0. `4KHDHub 1080p` is 1080: the
site's name starts with 4K, the row does not."""
m = re.search(r"(\d{3,4})p\b", text or "", re.I)
if m:
return int(m.group(1))
return 2160 if re.search(r"\b4k\b", text or "", re.I) else 0
def _hdhub_decodable(name: str, filename: str) -> bool:
"""Could the Pi's H.264 decoder play this, judging by what it says?"""
height = _hdhub_height(name) or _hdhub_height(filename)
if not height or height > VIDCORE_MAX_HEIGHT:
return False
return not _UNDECODABLE.search(filename or "")
def _hdhub_languages(filename: str) -> tuple[bool, bool]:
"""(has English audio, has ONLY English audio), from the name.
A name that mentions no language at all is taken as English: that is how
ordinary release names read. The probe checks the real tags later.
"""
words = _norm(filename).split()
foreign = any(w in words for w in _FOREIGN)
english = "english" in words or "eng" in words
return (english or not foreign, not foreign)
def _url_seconds_left(url: str, now: Optional[float] = None) -> Optional[float]:
"""Seconds until a pre-signed (AWS-style) URL expires; None if unsigned."""
q = parse_qs(urlparse(url).query)
try:
signed = calendar.timegm(time.strptime(q["X-Amz-Date"][0], "%Y%m%dT%H%M%SZ"))
lifetime = int(q["X-Amz-Expires"][0])
except (KeyError, IndexError, ValueError):
return None
return signed + lifetime - (time.time() if now is None else now)
def _hdhub_files(kind: str, ident: str, season: int = 0, episode: int = 0,
max_height: int = VIDCORE_MAX_HEIGHT,
margin: float = HDHUB_PLAY_MARGIN) -> list[HdHubFile]:
"""Files HdHub has for this title that we could use, best first.
Never raises: HdHub is a third party, and its being down must cost a row,
not the source list.
"""
try:
meta = _library_meta(kind, ident, season, episode)
if not meta:
return []
imdb = _imdb_id_for(kind, ident)
if not imdb:
return []
rows = _hdhub_query(kind, imdb, season, episode)
except Exception as exc:
log.warning("hdhub: lookup failed for %s %s: %s", kind, ident, exc)
return []
files = []
for r in rows:
url = r.get("url") or ""
name = r.get("name") or ""
desc = r.get("description") or ""
if not url or any(x in (name + " " + desc).lower() for x in HDHUB_REJECT):
continue
host = (urlparse(url).hostname or "").lower()
if not any(host == h.lstrip(".") or host.endswith(h) for h in HDHUB_HOSTS):
continue
filename = _hdhub_filename(desc)
if not _hdhub_title_ok(filename, meta["title"], meta["year"], kind,
season, episode):
continue
if not _hdhub_decodable(name, filename):
continue
height = _hdhub_height(name) or _hdhub_height(filename)
if height > max_height:
continue
has_english, english_only = _hdhub_languages(filename)
if not has_english:
continue
left = _url_seconds_left(url)
if left is not None and left < margin:
continue
size = int((r.get("behaviorHints") or {}).get("videoSize") or 0)
files.append(HdHubFile(url=url, filename=filename, height=height,
size=size, host=host, english_only=english_only))
# English-only first (no track juggling needed), then R2 (pixeldrain limits
# hotlinked downloads), then resolution, then size as the quality tiebreak.
files.sort(key=lambda x: (not x.english_only, ".r2." not in x.host + ".",
-x.height, -x.size))
return files
def _hdhub_probe(url: str, filename: Optional[str] = None) -> Optional[dict]:
"""ffprobe's view of a remote file, or None. Cached, including failures
(briefly), so a dead link is not re-probed on every click."""
hit = _hdhub_probe_cache.get(url)
if hit and hit[0] > time.time():
return hit[1]
result = None
try:
out = subprocess.run(
["ffprobe", "-v", "error", "-rw_timeout", "10000000",
"-show_entries",
"stream=index,codec_type,codec_name,height:stream_tags=language"
":stream_disposition=attached_pic:format=duration",
"-of", "json", url],
capture_output=True, text=True, timeout=HDHUB_PROBE_TIMEOUT)
if out.returncode == 0:
result = json.loads(out.stdout)
except Exception as exc:
log.info("hdhub: probe failed for %s (%s)", filename or url[:60], exc)
_hdhub_probe_cache[url] = (time.time() + (HDHUB_TTL if result else 300), result)
return result
def _probe_video(p: Optional[dict]) -> Optional[dict]:
"""The real video stream -- not the cover art some releases attach."""
for s in (p or {}).get("streams") or []:
if s.get("codec_type") == "video" and not (s.get("disposition") or {}).get("attached_pic"):
return s
return None
def _probe_audio(p: Optional[dict]) -> list[dict]:
"""Audio streams an MP4 can carry with a stream copy."""
return [s for s in (p or {}).get("streams") or []
if s.get("codec_type") == "audio" and s.get("codec_name") in MP4_AUDIO]
def _audio_lang(s: dict) -> str:
return ((s.get("tags") or {}).get("language") or "").lower()
def _probe_usable(p: Optional[dict]) -> bool:
"""Can the Pi play this file, with English sound? The file decides, not
its name: a name can omit HEVC, the stream cannot."""
v = _probe_video(p)
if not v or v.get("codec_name") != "h264" or (v.get("height") or 0) > 1088:
return False
audio = _probe_audio(p)
if not audio:
return False
tagged = [a for a in audio if _audio_lang(a) not in ("", "und")]
return not tagged or any(_audio_lang(a) in ("eng", "en") for a in tagged)
def _hdhub_pick(kind: str, ident: str, season: int = 0, episode: int = 0,
height: int = VIDCORE_MAX_HEIGHT,
margin: float = HDHUB_PLAY_MARGIN) -> Optional[tuple]:
"""(file, probe) for the best file that actually checks out, or None."""
key = (kind, ident, season, episode, height)
hit = _hdhub_pick_cache.get(key)
if hit and hit[0] > time.time():
left = _url_seconds_left(hit[1][0].url)
if left is None or left >= margin:
return hit[1]
deadline = time.time() + HDHUB_PICK_BUDGET
for f in _hdhub_files(kind, ident, season, episode, max_height=height,
margin=margin)[:HDHUB_PICK_TRIES]:
if time.time() > deadline:
log.info("hdhub: pick budget spent for %s", ident)
break
p = _hdhub_probe(f.url, f.filename)
if _probe_usable(p):
_hdhub_pick_cache[key] = (time.time() + HDHUB_TTL, (f, p))
return f, p
log.info("hdhub: %s not usable, trying the next", f.filename)
return None
_hdhub_prefetch_sem = threading.Semaphore(2)
def _hdhub_prefetch(kind: str, ident: str, season: int = 0,
episode: int = 0) -> None:
"""Probe the best file while the source list is being read, so pressing
play is a cache hit. Skipped, not queued, when already busy."""
if not _hdhub_prefetch_sem.acquire(blocking=False):
return
def run() -> None:
try:
_hdhub_pick(kind, ident, season, episode)
except Exception as exc:
log.warning("hdhub prefetch failed for %s: %s", ident, exc)
finally:
_hdhub_prefetch_sem.release()
threading.Thread(target=run, name=f"hdhub-{ident}", daemon=True).start()
def _hdhub_enabled() -> bool:
src = get_source("hdhub")
return bool(src and src.enabled)
def _append_hdhub_rows(out: list, kind: str, ident: Optional[str],
season: int = 0, episode: int = 0) -> None:
"""One row per resolution HdHub has a usable-looking file at.
Built from HdHub's listing (0.2-0.4s) without probing: probing is seconds
per file and belongs off the source list. The play route probes and falls
through to the next file, so a row whose top file is bad still plays.
"""
if not ident or not _hdhub_enabled():
return
files = _hdhub_files(kind, ident, season, episode)
if not files:
return
try:
_hdhub_prefetch(kind, ident, season, episode)
except Exception as exc: # never let a warm-up break /stream
log.warning("hdhub prefetch could not start: %s", exc)
if kind == "series":
path = f"/hdhub/series/{ident}:{season}:{episode}.mkv"
tag = f"S{season:02d}E{episode:02d} " # Jacktook drops TV rows without it
else:
path = f"/hdhub/movie/{ident}.mkv"
tag = ""
seen = set()
for f in files:
if f.height in seen:
continue
seen.add(f.height)
size = f" · {f.size / 1e9:.1f} GB" if f.size else ""
audio = "English" if f.english_only else "English + other audio"
out.append(_stream_dict(Stream(
url=f"http://{request.host}{path}?h={f.height}",
quality=f"{f.height}p", name="HdHub", referer="",
title_extra=f"{tag}{f.height}p{size}\nHdHub · {audio} · seekable")))
def _serve_hdhub(kind: str, ident: str, season: int, episode: int):
height = _requested_height()
local = _local_redirect(kind, ident, season, episode, height)
if local:
return local
picked = _hdhub_pick(kind, ident, season, episode, height)
if not picked:
abort(404)
_ensure_ingest(kind, ident, season, episode, height=height)
if kind == "series":
_queue_next_episode(ident, season, episode, height)
log.info("hdhub: serving %s", picked[0].filename)
return redirect(picked[0].url, code=302)
@app.route("/hdhub/movie/<path:ident>.mkv")
def hdhub_movie(ident: str):
if not re.fullmatch(VIDCORE_IDENT, ident):
abort(404)
return _serve_hdhub("movie", ident, 0, 0)
@app.route("/hdhub/series/<path:ident>.mkv")
def hdhub_series(ident: str):
m = re.fullmatch(rf"({VIDCORE_IDENT}):(\d+):(\d+)", ident)
if not m:
abort(404)
return _serve_hdhub("series", m.group(1), int(m.group(2)), int(m.group(3)))
def _copy_progressive(url: str, probe: dict, out_path: str) -> None:
"""Stream-copy a remote file into a progressive MP4, English audio first.
English goes first AND is flagged default because Kodi plays the default
track, and these dual-audio files flag Hindi. Subtitle streams are not
copied: the library gets wyzie sidecars beside every copy, and SRT does not
stream-copy into MP4. Measured against the source's own duration before it
is published, for the reason _remux_to_mp4 is: ffmpeg exits 0 on an input
that stops early.
"""
tmp = _part_path(out_path)
os.makedirs(os.path.dirname(out_path), exist_ok=True)
video = _probe_video(probe)
audio = _probe_audio(probe)
if not video or not audio:
raise RuntimeError("no copyable video/audio in source")
audio = ([a for a in audio if _audio_lang(a) in ("eng", "en")]
+ [a for a in audio if _audio_lang(a) not in ("eng", "en")])
cmd = ["ffmpeg", "-v", "error", "-y",
"-rw_timeout", "30000000",
"-reconnect", "1",
"-reconnect_streamed", "1",
"-reconnect_on_network_error", "1",
"-reconnect_delay_max", "30",
"-i", url, "-map", f"0:{video['index']}"]
for a in audio:
cmd += ["-map", f"0:{a['index']}"]
cmd += ["-c", "copy"]
for i in range(len(audio)):
cmd += [f"-disposition:a:{i}", "default" if i == 0 else "0"]
# -f mp4 is REQUIRED, as in _remux_to_mp4: the output is `<name>.part`.
cmd += ["-movflags", "+faststart", "-f", "mp4", tmp]
proc = _run_remux(cmd, tmp, stall_timeout=MP4_STALL_TIMEOUT,
hard_timeout=MP4_REMUX_TIMEOUT)
if proc.returncode != 0 or not os.path.exists(tmp) or os.path.getsize(tmp) == 0:
raise RuntimeError(f"ffmpeg exit {proc.returncode}: {(proc.stderr or '')[-300:]}")
try:
expected = float(((probe or {}).get("format") or {}).get("duration") or 0) or None
except (TypeError, ValueError):
expected = None
if expected:
actual = _media_duration(tmp)
if actual is None or actual < expected * REMUX_MIN_COMPLETENESS:
raise RuntimeError(
f"incomplete copy: got {actual if actual else 0:.0f}s of "
f"{expected:.0f}s — not published")
log.info("copy verified: %.0fs of %.0fs", actual, expected)
_record_verified(out_path, actual, expected)
os.replace(tmp, out_path)
def _ingest_from_hdhub(kind: str, ident: str, season: int, episode: int,
height: int, dest: str, meta: dict) -> Optional[str]:
"""Copy the title from HdHub into `dest`. None on success, else why not."""
if not _hdhub_enabled():
return "disabled"
picked = _hdhub_pick(kind, ident, season, episode, height,
margin=HDHUB_INGEST_MARGIN)
if not picked:
return "no usable file"
f, p = picked
_job_update(_mp4_key(kind, ident, season, episode, height),
preview={"type": "file", "url": f.url, "filename": f.filename,
"height": f.height, "size": f.size, "probe": p})
try:
seconds = float((p.get("format") or {}).get("duration") or 0)
expected = float(meta.get("runtime") or 0) * 60
except (TypeError, ValueError):
seconds = expected = 0
lo, hi = HDHUB_RUNTIME_WINDOW
if seconds and expected and not (lo * expected <= seconds <= hi * expected):
return (f"{f.filename} runs {seconds / 60:.0f} min but TMDB says "
f"{expected / 60:.0f} — not taken as this title")
log.info("ingest start (hdhub): %s -> %s", f.filename, dest)
try:
_copy_progressive(f.url, p, dest)
except Exception as exc:
return str(exc)
return None
def _vidcore_ident_for_fmv(ctype: str, slug: str) -> Optional[str]:
"""fmovies carries no imdb/tmdb id — bridge its slug to TMDB so the (lazy)
vidcore row can be offered on catalog entries too."""
title, _ = _fmv_slug_to_title(slug)
tmdb = _tmdb_from_title(ctype, title)
if not tmdb:
log.info("fmv->tmdb: no match for %r", title)
return None
return f"tmdb:{tmdb}"
def _append_mp4_row(out: list, kind: str, ident: Optional[str],
season: int = 0, episode: int = 0) -> None:
"""Offer a local remuxed copy, FIRST, when one exists.
Only listed when the copy is complete, so the row can never be a dead click:
it is the one row on this source where seek, resume and pause all work, and
it plays from local disk with no CDN token to expire.
"""
if not ident:
return
if kind == "series":
path = f"/mp4/series/{ident}:{season}:{episode}.mp4"
tag = f"S{season:02d}E{episode:02d} "
else:
path = f"/mp4/movie/{ident}.mp4"
tag = ""
def row(label: str, height: int) -> None:
out.append(_stream_dict(Stream(
url=f"http://{request.host}{path}?h={height}",
quality=label, name="local",
referer="https://vidcore.net/",
title_extra=f"{tag}{label}\nlocal copy")))
for height in VIDCORE_HEIGHTS:
local = _playable_path(kind, ident, season, episode, height)
if not local:
continue
# A library copy is height-agnostic — one copy per episode — so the
# height is not part of its label. It is still sent in the query string
# because the legacy cache path IS keyed by it.
in_library = local.startswith(LIBRARY_ROOT + os.sep)
return row("in library — seekable" if in_library
else f"{height}p seekable", height)
# No copy yet, but the row is still offered: it plays the live stream and
# keeps a copy as it goes. Offering it ONLY once a copy existed is what made
# the feature unreachable — the row had to exist for a copy to be made, and
# a copy had to exist for the row to be offered.
if not _library_meta(kind, ident, season, episode):
return # cannot be named, so the row would be a lie
running = _mp4_jobs.get(
_mp4_key(kind, ident, season, episode, VIDCORE_MAX_HEIGHT)) == "running"
if not running:
return row("play & keep", VIDCORE_MAX_HEIGHT)
done = _ingest_progress(kind, ident, season, episode)
row(f"saving to library… {done / 1e6:.0f} MB" if done
else "saving to library…", VIDCORE_MAX_HEIGHT)
def _append_action_rows(out: list, kind: str, ident: Optional[str],
season: int = 0, episode: int = 0) -> None:
"""Rows that DO something, for the judgments no measurement can make.
The completeness gate catches a copy that is SHORT. It cannot catch the
wrong episode, the wrong season, or audio nobody can listen to -- those need
eyes, and the eyes are in front of the TV, not the audit page. So the escape
hatch has to be here, in the list Jacktook already puts up before playing,
which is the one surface Kore can drive.
Listed LAST, after every row anyone actually came for: a Kodi source list is
navigated by arrow key.
"""
if not ident or not _library_meta(kind, ident, season, episode):
return # cannot be named, so the row would be a lie
if kind == "series":
refetch = f"/refetch/series/{ident}:{season}:{episode}.m3u8"
tag = f"S{season:02d}E{episode:02d} "
else:
refetch = f"/refetch/movie/{ident}.m3u8"
tag = ""
def row(path: str, quality: str, title: str) -> None:
out.append(_stream_dict(Stream(
url=f"http://{request.host}{path}?h={VIDCORE_MAX_HEIGHT}",
quality=quality, name="local", referer="https://vidcore.net/",
title_extra=title)))
# Only when a copy actually exists. With none, _append_mp4_row's "play &
# keep" row already fetches one, and this would be offering to redo nothing.
if _playable_path(kind, ident, season, episode, VIDCORE_MAX_HEIGHT):
row(refetch, "re-download",
f"{tag}wrong copy? re-download\nplays live meanwhile")
if kind == "series":
row(f"/season/series/{ident}:{season}:{episode}.m3u8", "season",
f"get season {season} — aired episodes\nplays this one meanwhile")
def _append_vidcore_row(out: list, kind: str, ident: Optional[str],
season: int = 0, episode: int = 0) -> None:
"""Additive lazy vidcore rows — one per quality, hidden for known misses.
One row per height is what gives the user a bitrate choice: a Stremio client
renders each stream as a source to pick, and they all share a single resolve,
so only the first pick pays for the browser. Each still points at ONE variant
playlist, which is what keeps pause working — offering the master instead
would be adaptive, but Kodi reads it as a live stream (and Jacktook never
engages inputstream.adaptive, so nothing would adapt anyway).
"""
if not ident:
return
if _rep_is_known_miss(_rep_key(kind, ident, season, episode)):
return
try:
# Start resolving now, while the user is still reading the source list.
_prefetch_vidcore(kind, ident, season, episode)
except Exception as exc: # never let a warm-up break /stream
log.warning("vidcore prefetch could not start: %s", exc)
# Fixed variants only. The adaptive row is NOT offered: inputstream.adaptive
# cannot play this CDN's audio, and the cause is in the media, not in how we
# ask for it. Measured 2026-08-10 by pulling a segment and probing it:
#
# init-s1080p-v1-a1.mp4 -> index=0 h264 video
# index=1 aac audio 48000Hz 2ch (real audio,
# mean_volume -32.0dB, max -14.7dB)
#
# Video and audio are MUXED into one fMP4 track set. ISA builds its stream
# list from the manifest CODECS ("avc1...,mp4a..."), so it reports two
# streams and opens the aac decoder from that metadata, but its HLS fMP4
# reader only feeds the first track -- the audio stream gets no samples,
# which is the "CVideoPlayerAudio::Process - stream stalled" line, and the
# result is perfect video with no sound. ffmpeg demuxes the container itself
# and gets both tracks, which is why the fixed rows have audio.
#
# The cost is seek: ffmpeg accepts a seek on these segments and ignores it
# (request 50%, stay at 1.2%), while ISA seeks correctly. On this source the
# two are mutually exclusive, and audio wins. Seek needs a source that is not
# muxed fMP4 HLS -- a progressive MP4 (vidlink) seeks natively under ffmpeg.
src = get_source("vidcore")
for height in (src.heights if src else VIDCORE_HEIGHTS):
out.append(_stream_dict(
_vidcore_row(request.host, kind, ident, season, episode, height=height)))
# These MUST be GET. A Stremio row is a URL the player opens, and Kodi opens it
# with GET -- the exact opposite of the POST-only rule on /library/reingest.
#
# It is safe here for a specific reason rather than by luck. Kodi issues
# GET + HEAD + GET for a single stream URL (measured 2026-08-10, when a fixed
# Chromium profile turned that into three resolvers fighting over one browser
# session), so anything these start fires up to three times. Both job starters
# already refuse a duplicate: _start_mp4_job returns early while its key is
# "running", _start_season_job while that season is. That idempotency is
# LOAD-BEARING here, not incidental -- remove it and one click becomes three
# remuxes of the same episode.
#
# Each answers with the live upstream, which is _serve_mp4's established idiom:
# a row has to be playable the instant it is clicked, and a 503 would surface as
# a playback failure while the work went on invisibly.
def _serve_action(kind: str, ident: str, season: int, episode: int,
whole_season: bool = False):
height = _requested_height()
if whole_season:
_start_season_job(ident, season, height)
else:
_start_mp4_job(kind, ident, season, episode, height, force=True)
url = _resolve_vidcore(kind, ident, season, episode,
wait=VIDCORE_PLAY_DEADLINE)
if not url:
abort(404)
return _serve_vidcore(url)
@app.route("/refetch/movie/<path:ident>.m3u8")
def refetch_movie(ident: str):
if not re.fullmatch(VIDCORE_IDENT, ident):
abort(404)
return _serve_action("movie", ident, 0, 0)
@app.route("/refetch/series/<path:ident>.m3u8")
def refetch_series(ident: str):
m = re.fullmatch(rf"({VIDCORE_IDENT}):(\d+):(\d+)", ident)
if not m:
abort(404)
return _serve_action("series", m.group(1), int(m.group(2)), int(m.group(3)))
@app.route("/season/series/<path:ident>.m3u8")
def season_series_row(ident: str):
"""Queue the season, and play the episode it was asked from meanwhile."""
m = re.fullmatch(rf"({VIDCORE_IDENT}):(\d+):(\d+)", ident)
if not m:
abort(404)
return _serve_action("series", m.group(1), int(m.group(2)),
int(m.group(3)), whole_season=True)
@app.route("/stream/movie/<path:ident>.json")
def stream_movie(ident: str):
ident = ident.strip()
fxl = re.fullmatch(r"fxl:(\d+)", ident)
if fxl: # flixlab-native catalog: vidcore only
out: list = []
_append_mp4_row(out, "movie", f"tmdb:{fxl.group(1)}")
_append_hdhub_rows(out, "movie", f"tmdb:{fxl.group(1)}")
_append_vidcore_row(out, "movie", f"tmdb:{fxl.group(1)}")
_append_action_rows(out, "movie", f"tmdb:{fxl.group(1)}")
return jsonify({"streams": out})
fmv = re.fullmatch(r"fmv:(\d+):([a-z0-9\-]+)", ident)
if fmv:
mid, slug = fmv.group(1), fmv.group(2)
try:
streams = cached_extract_url(f"fmvm:{mid}", _fmv_film_url(mid, slug), tv=False)
except Exception as exc:
log.exception("fmv movie resolve failed for %s: %s", ident, exc)
streams = []
out = [_stream_dict(s) for s in streams]
_append_vidcore_row(out, "movie", _vidcore_ident_for_fmv("movie", slug))
return jsonify({"streams": out})
if not re.fullmatch(r"tt\d+", ident):
return jsonify({"streams": []})
try:
streams = cached_extract(f"movie:{ident}", ident, tv=False)
except Exception as exc:
log.exception("extract failed for %s: %s", ident, exc)
streams = []
out = [_stream_dict(s) for s in streams]
_append_hdhub_rows(out, "movie", ident)
_append_vidcore_row(out, "movie", ident) # additive gap-filler
return jsonify({"streams": out})
@app.route("/stream/series/<path:ident>.json")
def stream_series(ident: str):
fxl = re.fullmatch(r"fxl:(\d+):(\d+):(\d+)", ident)
if fxl: # flixlab-native catalog: vidcore only
tmdb, season, episode = fxl.group(1), int(fxl.group(2)), int(fxl.group(3))
out: list = []
_append_mp4_row(out, "series", f"tmdb:{tmdb}", season, episode)
_append_hdhub_rows(out, "series", f"tmdb:{tmdb}", season, episode)
_append_vidcore_row(out, "series", f"tmdb:{tmdb}", season, episode)
_append_action_rows(out, "series", f"tmdb:{tmdb}", season, episode)
return jsonify({"streams": out})
fmv = re.fullmatch(r"fmv:(\d+):([a-z0-9\-]+):(\d+):(\d+)", ident)
if fmv:
mid, slug = fmv.group(1), fmv.group(2)
season, episode = int(fmv.group(3)), int(fmv.group(4))
try:
streams = cached_extract_url(
f"fmvs:{mid}:{season}:{episode}",
_fmv_film_url(mid, slug), tv=True, season=season, episode=episode,
)
except Exception as exc:
log.exception("fmv series resolve failed for %s: %s", ident, exc)
streams = []
se_tag = f"S{season:02d}E{episode:02d}"
out = [_stream_dict(s, se_tag=se_tag) for s in streams]
_append_vidcore_row(out, "series", _vidcore_ident_for_fmv("series", slug), season, episode)
return jsonify({"streams": out})
match = re.fullmatch(r"(tt\d+):(\d+):(\d+)", ident)
if not match:
return jsonify({"streams": []})
imdb_id, season, episode = match.group(1), int(match.group(2)), int(match.group(3))
try:
streams = cached_extract(
f"series:{imdb_id}:{season}:{episode}",
imdb_id,
tv=True,
season=season,
episode=episode,
)
except Exception as exc:
log.exception("extract failed for %s: %s", ident, exc)
streams = []
se_tag = f"S{season:02d}E{episode:02d}"
out = [_stream_dict(s, se_tag=se_tag) for s in streams]
_append_hdhub_rows(out, "series", imdb_id, season, episode)
_append_vidcore_row(out, "series", imdb_id, season, episode) # additive
return jsonify({"streams": out})
def _stream_dict(s: Stream, se_tag: str = "") -> dict:
if s.title_extra:
title = s.title_extra
elif se_tag:
title = f"{se_tag} {s.quality}\n{s.name}"
else:
title = f"{s.quality}\n{s.name}"
hints: dict = {"notWebReady": True}
# No referer, no proxyHeaders: a plain file host needs none, and sending a
# made-up one only gives it something to refuse.
if s.referer:
hints["proxyHeaders"] = {"request": {"User-Agent": UA,
"Referer": s.referer}}
return {
"name": s.name,
"title": title,
"url": s.url,
"behaviorHints": hints,
}
# ----------------------- fmovies-native catalog + meta ----------------------
# Lets Stremio clients (Jacktook) search fmovies directly and browse exactly
# what is playable. Catalog ids are "fmv:<mid>:<slug>"; selecting an episode
# routes back to /stream/series/fmv:<mid>:<slug>:<season>:<episode>.json.
def _meta_flixlab(ctype: str, tmdb: str):
"""flixlab-native meta, straight from TMDB (vidcore is TMDB-keyed).
Unlike the fmovies catalog — where each entry IS one season, because fmovies
pages seasons separately — this returns every season's episodes in `videos`,
so Jacktook renders a full season list.
"""
try:
if ctype == "series":
show, videos = _tmdb_series_videos(tmdb)
else:
show, videos = _tmdb_get(f"movie/{tmdb}"), []
except Exception as exc:
log.exception("flixlab meta failed for %s: %s", tmdb, exc)
return jsonify({"meta": {}})
name = show.get("name") or show.get("title") or tmdb
date = show.get("first_air_date") or show.get("release_date") or ""
meta_obj = {
"id": f"fxl:{tmdb}",
"type": ctype,
"name": name,
"poster": _tmdb_poster(show.get("poster_path")),
"background": _tmdb_poster(show.get("backdrop_path")) or _tmdb_poster(show.get("poster_path")),
"description": show.get("overview") or "",
"releaseInfo": date[:4],
"genres": [g["name"] for g in show.get("genres", [])] or ["flixlab"],
}
if ctype == "series":
meta_obj["videos"] = videos
return jsonify({"meta": meta_obj})
def _search_query_from_extra(extra: str) -> str:
for part in (extra or "").split("/"):
if part.startswith("search="):
return unquote(part[len("search="):])
return ""
def _skip_from_extra(extra: str) -> int:
"""Stremio pages browse catalogs with `skip=<n>`; TMDB pages by 20."""
for part in (extra or "").split("/"):
if part.startswith("skip="):
try:
return max(0, int(unquote(part[len("skip="):])))
except ValueError:
return 0
return 0
def _genre_from_extra(extra: str) -> str:
for part in (extra or "").split("/"):
if part.startswith("genre="):
return unquote(part[len("genre="):])
return ""
def _browse_kind(cid: str) -> Optional[str]:
"""Map a browse catalog id (flixlab-<kind>-<type>) to its TMDB list."""
for kind in BROWSE_ENDPOINTS:
if cid.startswith(f"flixlab-{kind}-"):
return kind
return None
@app.route("/catalog/<ctype>/<cid>.json")
@app.route("/catalog/<ctype>/<cid>/<path:extra>.json")
def catalog(ctype: str, cid: str, extra: str = ""):
if ctype not in ("series", "movie"):
return jsonify({"metas": []})
if cid.startswith("flixlab-genre-"):
genre = _genre_from_extra(extra)
if not genre:
return jsonify({"metas": []}) # picker not chosen yet
page = _skip_from_extra(extra) // 20 + 1
try:
return jsonify({"metas": _tmdb_by_genre(ctype, genre, page)})
except Exception as exc:
log.exception("genre browse %s/%s failed: %s", genre, ctype, exc)
return jsonify({"metas": []})
browse = _browse_kind(cid)
if browse:
# TMDB pages by 20; Stremio asks with skip=, so convert to a page number.
page = _skip_from_extra(extra) // 20 + 1
try:
return jsonify({"metas": _tmdb_browse(ctype, browse, page)})
except Exception as exc:
log.exception("browse %s/%s page %d failed: %s", browse, ctype, page, exc)
return jsonify({"metas": []})
query = _search_query_from_extra(extra)
if not query:
return jsonify({"metas": []}) # search-only catalog (no browse listing)
if cid.startswith("flixlab"): # flixlab/vidcore is TMDB-keyed — search TMDB
try:
return jsonify({"metas": _tmdb_search(ctype, query)})
except Exception as exc:
log.exception("flixlab catalog search failed for %r: %s", query, exc)
return jsonify({"metas": []})
try:
data = cached_browse(["search", query])
except Exception as exc:
log.exception("catalog search failed for %r: %s", query, exc)
return jsonify({"metas": []})
metas = []
for r in data.get("results", []):
if r.get("type") != ctype:
continue
metas.append({
"id": f"fmv:{r['mid']}:{r['slug']}",
"type": r["type"],
"name": r.get("title") or r["slug"],
"poster": r.get("poster") or "",
})
return jsonify({"metas": metas})
@app.route("/meta/<ctype>/<path:meta_id>.json")
def meta(ctype: str, meta_id: str):
parts = meta_id.split(":")
if parts[0] == "fxl" and len(parts) >= 2:
return _meta_flixlab(ctype, parts[1])
if len(parts) < 3 or parts[0] != "fmv":
return jsonify({"meta": {}})
mid, slug = parts[1], parts[2]
try:
data = cached_browse(["meta", mid, slug, ctype])
except Exception as exc:
log.exception("meta failed for %s: %s", meta_id, exc)
return jsonify({"meta": {}})
season = data.get("season") or 1
meta_obj = {
"id": meta_id,
"type": ctype,
"name": data.get("name") or slug,
"poster": data.get("poster") or "",
"background": data.get("poster") or "",
"genres": ["fmovies"],
}
if ctype == "series":
videos = []
for e in data.get("episodes", []):
ep = int(e["episode"])
videos.append({
"id": f"{meta_id}:{season}:{ep}",
"title": e.get("title") or f"Episode {ep}",
"season": season,
"episode": ep,
"released": "",
})
meta_obj["videos"] = videos
return jsonify({"meta": meta_obj})
# -------------------------------- retention ---------------------------------
# The library is NOT a cache. test_library_ingest says so in its first line, and
# the volume agrees: 745 GB of it, the overwhelming majority never fetched by
# this addon at all. So retention has no authority over Shows/ and Movies/. It
# may reclaim only what it can PROVE is derived, and it may never delete an
# episode to make room for another one.
#
# What protects the disk is therefore refusal, not deletion: below the floor an
# ingest declines to start. That is the honest order of preference -- a copy we
# have is worth more than a copy we were about to make, so the new one yields.
# It also fails visibly, as a row on the audit page, instead of quietly eating
# the last of the volume.
#
# The floor is deliberately generous. At roughly 1.3 GB an episode it leaves
# about twenty episodes of headroom, which is enough to notice and act before
# anything is actually at risk.
LIBRARY_MIN_FREE_BYTES = 25 * 1024 ** 3
def _disk() -> Optional[tuple[int, int, int]]:
"""(total, used, free) for the library volume, or None if unreadable."""
try:
usage = shutil.disk_usage(LIBRARY_ROOT)
return usage.total, usage.used, usage.free
except OSError as exc:
log.warning("retention: cannot stat %s (%s)", LIBRARY_ROOT, exc)
return None
def _enough_space() -> bool:
"""False when an ingest should decline to start.
Unreadable disk means True: this is a brake, and a brake that cannot read
the road must not stop the car. The same rule the completeness gate follows
for a missing runtime.
"""
disk = _disk()
if disk is None:
return True
return disk[2] >= LIBRARY_MIN_FREE_BYTES
def _reclaimable() -> list[dict]:
"""Artifacts that are PROVABLY derived, and therefore safe to remove.
Driven from the ingest queue rather than by walking the volume. That keeps
it bounded -- twenty stats instead of a crawl over 745 GB -- and it keeps
the scope honest: these are the copies this addon made, and the only ones it
can reason about. A file it cannot tie back to a job it ran is left alone,
which is why the single legacy copy in mp4-cache survives: X-Men '97 S02E07
was never ingested to the library, so it is the ONLY copy of that episode
and nothing here is entitled to touch it.
"""
out: list[dict] = []
now = time.time()
for key, entry in _load_job_queue().items():
args = entry.get("args") or []
if len(args) < 5:
continue
kind, ident, season, episode, height = args[:5]
season, episode, height = int(season), int(episode), int(height)
meta = _library_meta(kind, ident, season, episode)
if not meta:
continue
dest = _library_path(kind, meta["title"], meta["year"], meta["tmdb_id"],
season, episode, meta["episode_title"])
part = _part_path(dest)
try:
st = os.stat(part)
# Only once it cannot still be in flight. A remux may legitimately
# run for MP4_REMUX_TIMEOUT, and the running job would rather find
# its own file than a surprise.
if (_mp4_jobs.get(key) != "running"
and now - st.st_mtime > MP4_REMUX_TIMEOUT):
out.append({"path": part, "bytes": st.st_size,
"reason": "abandoned partial remux"})
except OSError:
pass
# A legacy mp4-cache copy is redundant ONLY once the library holds the
# same episode and that copy passes the completeness gate. Both halves
# matter: without the second, a truncated library copy would be grounds
# for deleting the good legacy one.
legacy = _mp4_path(kind, ident, season, episode, height)
if legacy != dest and os.path.exists(legacy) and _library_has(
kind, ident, season, episode):
try:
out.append({"path": legacy, "bytes": os.path.getsize(legacy),
"reason": "superseded by the library copy"})
except OSError:
pass
return out
def _reclaim() -> dict:
"""Remove everything _reclaimable() currently vouches for.
The set is RE-DERIVED here rather than taken from the request. A caller does
not get to name a path for deletion: the only paths that can ever be removed
are ones this function independently proved were derived, a moment ago.
"""
freed, removed = 0, []
for item in _reclaimable():
try:
os.remove(item["path"])
except OSError as exc:
log.warning("reclaim: %s (%s)", item["path"], exc)
continue
freed += item["bytes"]
removed.append(item["path"])
log.info("reclaimed %s (%s, %.0f MB)", os.path.basename(item["path"]),
item["reason"], item["bytes"] / 1e6)
return {"removed": removed, "freed": freed}
# ------------------------------ library audit -------------------------------
# The truncation this exists for was invisible for five days and was found by
# watching an episode, not by any check. Every number needed to spot it was
# already here: what is on disk, what TMDB says it should be, and whether the
# copy was measured against its own source when it was written.
#
# Scope is what WE fetched -- the ingest queue -- not the whole library. Those
# are the copies this addon is answerable for and the only ones the button can
# act on. The rest of /mnt/usb arrived by other means, and offering to
# re-download it would be offering something we cannot do.
#
# This is a page rather than a row in Kodi because Jacktook caches its source
# list for 24h, so a row can be a day stale, and because the client is Kore,
# which cannot drive Kodi's own UI. A LAN page is answered fresh, every time,
# from the same phone.
AUDIT_OK, AUDIT_SHORT, AUDIT_MISSING, AUDIT_UNKNOWN = (
"ok", "short", "missing", "unknown")
def _audit_rows() -> list[dict]:
"""One row per title this addon has ever fetched, newest first."""
verified = _load_verified()
rows = []
for key, entry in _load_job_queue().items():
args = entry.get("args") or []
if len(args) < 5:
continue # pre-resume entry; cannot be re-run
kind, ident, season, episode, height = args[:5]
meta = _library_meta(kind, ident, int(season), int(episode))
row = {"key": key, "kind": kind, "ident": ident,
"season": int(season), "episode": int(episode),
"height": int(height), "state": entry.get("state") or "",
"ts": entry.get("ts") or 0,
"title": meta["title"] if meta else ident,
"label": (f"S{int(season):02d}E{int(episode):02d}"
if kind == "series" else "movie")}
if not meta:
row.update(status=AUDIT_UNKNOWN, seconds=None, expected=None,
pct=None, verified=False, path=None)
rows.append(row)
continue
path = _library_path(kind, meta["title"], meta["year"], meta["tmdb_id"],
int(season), int(episode), meta["episode_title"])
row["path"] = path
row["verified"] = os.path.abspath(path) in verified
exists = os.path.exists(path) and os.path.getsize(path) > 0
row["seconds"] = _cached_media_duration(path) if exists else None
try:
row["expected"] = float(meta.get("runtime")) * 60
except (TypeError, ValueError):
row["expected"] = None
if not exists:
row["status"], row["pct"] = AUDIT_MISSING, None
elif not row["expected"] or row["seconds"] is None:
# Present and playable, but nothing to measure it against. Reported
# as its own state rather than folded into "ok" -- an unchecked copy
# and a checked one are not the same claim.
row["status"], row["pct"] = AUDIT_UNKNOWN, None
else:
row["pct"] = row["seconds"] / row["expected"] * 100
row["status"] = (AUDIT_OK
if _copy_is_complete(path, meta) else AUDIT_SHORT)
rows.append(row)
rows.sort(key=lambda r: r["ts"], reverse=True)
return rows
_AUDIT_CSS = """
body{font:14px/1.5 system-ui,sans-serif;margin:1.5rem;color:#222}
table{border-collapse:collapse;width:100%;max-width:60rem}
th,td{text-align:left;padding:.4rem .6rem;border-bottom:1px solid #ddd}
th{font-weight:600;border-bottom:2px solid #999}
tr.short{background:#fdecea}
tr.missing,tr.unknown{background:#fff8e1}
.tag{font-size:.75rem;padding:.1rem .4rem;border-radius:.2rem;background:#eee}
.tag.short{background:#d32f2f;color:#fff}
.tag.ok{background:#2e7d32;color:#fff}
button{padding:.3rem .7rem;cursor:pointer}
td.num{text-align:right;font-variant-numeric:tabular-nums}
p.disk{color:#555}
p.low{color:#b71c1c;font-weight:600}
"""
def _gb(n):
return f"{n / 1024 ** 3:.0f} GB"
def _disk_html() -> str:
"""The budget, and whatever is provably safe to remove.
Shown even when there is nothing to reclaim: the number that matters is the
headroom, and the point of putting it here is that the season button can
spend it twelve gigabytes at a time.
"""
disk = _disk()
out = []
if disk:
total, _used, free = disk
low = free < LIBRARY_MIN_FREE_BYTES
out.append(
f"<p class='{'low' if low else 'disk'}'>"
f"{_gb(free)} free of {_gb(total)}"
f" · floor {_gb(LIBRARY_MIN_FREE_BYTES)}"
+ (" · <b>ingests are declining until this clears</b>"
if low else "") + "</p>")
items = _reclaimable()
if not items:
return "".join(out) + "<p class=disk>Nothing to reclaim.</p>"
freed = sum(i["bytes"] for i in items)
out.append(f"<form method=post action=/library/reclaim><p class=disk>"
f"{len(items)} reclaimable ({freed / 1e6:.0f} MB): "
+ "; ".join(f"{html.escape(os.path.basename(i['path']))} "
f"— {html.escape(i['reason'])}" for i in items)
+ " <button>Reclaim</button></p></form>")
return "".join(out)
def _audit_html(rows: list[dict]) -> str:
def mins(seconds):
return "—" if not seconds else f"{seconds / 60:.0f}m"
body = []
short = sum(1 for r in rows if r["status"] == AUDIT_SHORT)
body.append(f"<h1>Library ({len(rows)} fetched"
+ (f", <b>{short} incomplete</b>" if short else "") + ")</h1>")
body.append(_disk_html())
body.append("<table><tr><th>Title</th><th></th><th>On disk</th>"
"<th>Expected</th><th></th><th>Checked</th><th></th></tr>")
for r in rows:
pct = "—" if r["pct"] is None else f"{r['pct']:.0f}%"
body.append(
f"<tr class={r['status']}>"
f"<td>{html.escape(str(r['title']))}</td>"
f"<td>{html.escape(r['label'])}</td>"
f"<td class=num>{mins(r['seconds'])}</td>"
f"<td class=num>{mins(r['expected'])}</td>"
f"<td class=num><span class='tag {r['status']}'>{pct if r['status'] != AUDIT_MISSING else 'missing'}</span></td>"
f"<td>{'source' if r['verified'] else '—'}</td>"
"<td><form method=post action=/library/reingest>"
f"<input type=hidden name=kind value='{html.escape(r['kind'])}'>"
f"<input type=hidden name=ident value='{html.escape(r['ident'])}'>"
f"<input type=hidden name=season value='{r['season']}'>"
f"<input type=hidden name=episode value='{r['episode']}'>"
f"<input type=hidden name=height value='{r['height']}'>"
"<button name=force value=1>Re-download</button>"
"</form>"
+ ("<form method=post action=/library/season>"
f"<input type=hidden name=ident value='{html.escape(r['ident'])}'>"
f"<input type=hidden name=season value='{r['season']}'>"
f"<input type=hidden name=height value='{r['height']}'>"
f"<button>Get season {r['season']}</button></form>"
if r["kind"] == "series" else "")
+ "</td></tr>")
body.append("</table>")
return (f"<!doctype html><meta charset=utf-8>"
f"<meta name=viewport content='width=device-width,initial-scale=1'>"
f"<title>vsembed library</title><style>{_AUDIT_CSS}</style>"
+ "".join(body))
@app.route("/library")
def library_audit():
return Response(_audit_html(_audit_rows()), mimetype="text/html")
@app.route("/library/audit.json")
def library_audit_json():
return jsonify(rows=_audit_rows())
@app.route("/library/reingest", methods=["POST"])
def library_reingest():
"""POST only, deliberately. Kodi issues speculative GETs all over this
service -- the parent-directory scan on every play is already a known 404
-- and a re-download must never be something a stray fetch can start.
"""
kind = request.form.get("kind") or ""
ident = request.form.get("ident") or ""
if kind not in ("movie", "series") or not ident:
abort(400)
try:
season = int(request.form.get("season") or 0)
episode = int(request.form.get("episode") or 0)
height = int(request.form.get("height") or VIDCORE_MAX_HEIGHT)
except ValueError:
abort(400)
state = _start_mp4_job(kind, ident, season, episode, height,
force=bool(request.form.get("force")))
log.info("library: re-ingest requested for %s:%s:%s:%s -> %s",
kind, ident, season, episode, state)
return redirect("/library", code=303)
@app.route("/library/reclaim", methods=["POST"])
def library_reclaim():
"""POST only, like every other action here.
Takes no arguments on purpose. There is no path parameter to get wrong,
because the caller does not choose what goes -- _reclaim re-derives the set
itself and will only ever remove what it can prove is derived.
"""
result = _reclaim()
log.info("library: reclaimed %d file(s), %.0f MB",
len(result["removed"]), result["freed"] / 1e6)
return redirect("/library", code=303)
@app.route("/library/season", methods=["POST"])
def library_season():
"""POST only, for the same reason as the re-ingest route."""
ident = request.form.get("ident") or ""
if not ident:
abort(400)
try:
season = int(request.form.get("season") or 0)
height = int(request.form.get("height") or VIDCORE_MAX_HEIGHT)
except ValueError:
abort(400)
state = _start_season_job(ident, season, height)
log.info("library: season fetch requested for %s S%02d -> %s",
ident, season, state)
return redirect("/library", code=303)
# --------------------------------- web UI -----------------------------------
# Add a title to the library from a browser, separately from Kodi (labs#12).
# Until this existed a download could only start from Kodi -- a play, an action
# row, or a title already listed on /library. You could not start from a title.
#
# It feeds the same pipeline as everything else -- _ensure_ingest and
# _start_season_job -- with ident `tmdb:<id>`, which is what Kodi's fxl rows
# send, so a download started here and a play started there share a job key.
#
# Nothing here stores state. What a page shows is read from the same places the
# rest of the addon already trusts: _library_has for "in library", _mp4_jobs for
# running / failed, the .part size for progress.
#
# Every action is POST, like /library: Kodi fires speculative GETs at this
# service, and a GET must never be able to start a download. LAN only, no auth
# -- the same exposure /library has always had.
WEB_REFRESH_SECONDS = 10
_WEB_CSS = """
:root{color-scheme:light dark;--bg:#fafafa;--fg:#1d1d1f;--mute:#666;--line:#ddd;
--card:#fff;--ok:#2e7d32;--run:#1565c0;--bad:#c62828}
@media(prefers-color-scheme:dark){:root{--bg:#121212;--fg:#e8e8e8;--mute:#9a9a9a;
--line:#333;--card:#1c1c1c;--ok:#66bb6a;--run:#64b5f6;--bad:#ef5350}}
*{box-sizing:border-box}
body{font:15px/1.5 system-ui,sans-serif;margin:0;background:var(--bg);color:var(--fg)}
header,main{max-width:64rem;margin:0 auto;padding:.75rem 1rem}
header{display:flex;flex-wrap:wrap;gap:.5rem 1rem;align-items:center;
border-bottom:1px solid var(--line)}
header a{color:inherit;text-decoration:none;font-weight:600}
header form{display:flex;gap:.4rem;flex:1;min-width:16rem}
input[type=search]{flex:1;min-width:0;padding:.45rem .6rem;font:inherit;
border:1px solid var(--line);border-radius:.3rem;background:var(--card);color:inherit}
select,button{font:inherit;padding:.4rem .7rem;border:1px solid var(--line);
border-radius:.3rem;background:var(--card);color:inherit;cursor:pointer}
button.go{background:var(--run);border-color:var(--run);color:#fff}
button.del{color:var(--bad);border-color:var(--bad)}
a.watch{font-size:.85rem;margin-left:.3rem}
h1{font-size:1.4rem;margin:.5rem 0}
h2{font-size:1.05rem;margin:1.2rem 0 .5rem}
.grid{display:grid;grid-template-columns:repeat(auto-fill,minmax(8.5rem,1fr));gap:.8rem}
.card{display:block;color:inherit;text-decoration:none}
.card img,.card .noimg{width:100%;aspect-ratio:2/3;object-fit:cover;
border-radius:.3rem;background:var(--line);display:block}
.card span{display:block;font-size:.85rem;margin-top:.3rem}
.title{display:flex;gap:1rem;flex-wrap:wrap}
.title img{width:10rem;border-radius:.3rem}
.title .about{flex:1;min-width:16rem}
.mute{color:var(--mute)}
.tabs{display:flex;flex-wrap:wrap;gap:.3rem;margin:.8rem 0}
.tabs a{padding:.2rem .6rem;border:1px solid var(--line);border-radius:.3rem;
color:inherit;text-decoration:none}
.tabs a.on{background:var(--run);border-color:var(--run);color:#fff}
table{border-collapse:collapse;width:100%}
td{padding:.45rem .4rem;border-bottom:1px solid var(--line);vertical-align:middle}
td.n{width:3rem;color:var(--mute);font-variant-numeric:tabular-nums}
td.act{text-align:right;white-space:nowrap}
form.dl{display:inline-flex;gap:.3rem;margin:0}
.st{font-size:.85rem}.st.held{color:var(--ok)}.st.running{color:var(--run)}
.st.failed{color:var(--bad)}
p.warn{color:var(--bad);font-weight:600}
"""
def _web_page(title: str, body: str, refresh: bool = False,
status: int = 200) -> Response:
head = (f"<!doctype html><meta charset=utf-8>"
f"<meta name=viewport content='width=device-width,initial-scale=1'>"
+ (f"<meta http-equiv=refresh content={WEB_REFRESH_SECONDS}>"
if refresh else "")
+ f"<title>{html.escape(title)}</title><style>{_WEB_CSS}</style>")
nav = ("<header><a href='/'>vsembed</a>"
"<a href='/web/downloads'>Downloads</a><a href='/library'>Library</a>"
"<form method=get action='/web/search'>"
"<input type=search name=q placeholder='Find a film or show' required>"
"<select name=type><option value=''>All</option>"
"<option value=series>Series</option>"
"<option value=movie>Movies</option></select>"
"<button>Search</button></form></header>")
return Response(head + nav + f"<main>{body}</main>", status=status,
mimetype="text/html")
def _web_cards(ctype: str, metas: list[dict]) -> str:
out = []
for m in metas:
tmdb = str(m.get("id") or "").removeprefix("fxl:")
if not tmdb.isdigit():
continue
img = (f"<img loading=lazy alt='' src='{html.escape(m['poster'])}'>"
if m.get("poster") else "<div class=noimg></div>")
out.append(f"<a class=card href='/web/title/{ctype}/{tmdb}'>{img}"
f"<span>{html.escape(m.get('name') or tmdb)}</span></a>")
return "<div class=grid>" + "".join(out) + "</div>"
def _web_state(kind: str, tmdb: str, season: int = 0,
episode: int = 0) -> dict:
"""What the library knows about one file: held, running, failed or none.
A "done" job whose file is gone (or short -- _library_has applies the
completeness gate) reads as "none", so it can simply be asked for again.
Jobs are looked up at every height: the library holds one copy per episode
whatever height fetched it.
"""
ident = f"tmdb:{tmdb}"
if _library_has(kind, ident, season, episode):
dest = _web_dest(kind, tmdb, season, episode)
return {"status": "held", "source": _source_of(dest) if dest else None}
states = [_mp4_jobs.get(_mp4_key(kind, ident, season, episode, h))
for h in VIDCORE_HEIGHTS]
if "running" in states:
key = next(_mp4_key(kind, ident, season, episode, h)
for h, st in zip(VIDCORE_HEIGHTS, states) if st == "running")
return {"status": "running", "key": key,
"source": (_job_info.get(key) or {}).get("source"),
"bytes": _ingest_progress(kind, ident, season, episode)}
failed = next((s for s in states if s and s.startswith("failed")), None)
if failed:
return {"status": "failed", "why": failed}
if "cancelled" in states:
return {"status": "failed", "why": "cancelled"}
return {"status": "none"}
def _source_name(sid: Optional[str]) -> str:
src = get_source(sid or "")
return src.name if src else (sid or "")
def _web_dest(kind: str, tmdb: str, season: int, episode: int) -> Optional[str]:
"""The library path for one file, or None when TMDB cannot name it."""
meta = _library_meta(kind, f"tmdb:{tmdb}", season, episode)
if not meta:
return None
return os.path.abspath(_library_path(kind, meta["title"], meta["year"],
meta["tmdb_id"], season, episode,
meta["episode_title"]))
def _web_state_html(state: dict, watch: Optional[str] = None) -> str:
s = state["status"]
if s == "held":
src = _source_name(state.get("source"))
return (f"<span class='st held'>in library{' · ' + html.escape(src) if src else ''}</span>"
+ (f" <a class=watch href='{watch}'>Watch</a>" if watch else ""))
if s == "running":
n = state.get("bytes")
got = f" · {n / 1e6:.0f} MB" if n else " · starting"
src = _source_name(state.get("source"))
frm = f" from {html.escape(src)}" if src else ""
return (f"<span class='st running'>downloading{frm}{got}</span>"
+ _preview_link(state.get("key")))
if s == "failed":
return f"<span class='st failed'>{html.escape(state['why'])}</span>"
return ""
def _web_download_form(kind: str, tmdb: str, season: int = 0, episode: int = 0,
scope: str = "one", label: str = "Download") -> str:
heights = "".join(f"<option value={h}>{h}p</option>" for h in VIDCORE_HEIGHTS)
return ("<form class=dl method=post action='/web/download'>"
f"<input type=hidden name='kind' value='{kind}'>"
f"<input type=hidden name='tmdb' value='{tmdb}'>"
f"<input type=hidden name='season' value='{season}'>"
f"<input type=hidden name='episode' value='{episode}'>"
f"<input type=hidden name='scope' value='{scope}'>"
f"<select name=height aria-label=Quality>{heights}</select>"
f"<button class=go>{label}</button></form>"
+ _web_source_line(kind, tmdb, season, episode))
def _web_source_line(kind: str, tmdb: str, season: int, episode: int) -> str:
"""Where a Download will fetch from, and what a human has ruled out.
A season button has no single file, so it shows the default order; each
episode it fetches still honours its own rejections in _run_ingest.
"""
dest = None if kind == "series" and not episode else _web_dest(
kind, tmdb, season, episode)
order = _ingest_order(dest) if dest else [s.id for s in ingest_sources()]
names = [html.escape(_source_name(sid)) for sid in order]
where = ("from " + ", else ".join(names)) if names else "no source left"
out = f" <span class=mute>{where}</span>"
for sid in sorted(_rejected_sources(dest)) if dest else []:
out += (f" <span class=mute>· {html.escape(_source_name(sid))} rejected</span>"
+ _web_source_form("allow", kind, tmdb, season, episode, sid,
"Allow again"))
return out
def _web_source_form(action: str, kind: str, tmdb: str, season: int,
episode: int, sid: str, label: str) -> str:
return (f"<form class=dl method=post action='/web/source/{action}'>"
f"<input type=hidden name='kind' value='{kind}'>"
f"<input type=hidden name='tmdb' value='{tmdb}'>"
f"<input type=hidden name='season' value='{season}'>"
f"<input type=hidden name='episode' value='{episode}'>"
f"<input type=hidden name='source' value='{html.escape(sid)}'>"
f"<button>{label}</button></form>")
def _web_low_space() -> str:
return ("" if _enough_space() else
"<p class=warn>The library volume is under its free-space floor — "
"downloads will decline until space is cleared.</p>")
@app.route("/")
def index():
rows = []
for label, ctype, kind in (("Trending series", "series", "trending"),
("Trending films", "movie", "trending"),
("Popular series", "series", "popular"),
("Popular films", "movie", "popular")):
try:
metas = _tmdb_browse(ctype, kind)
except Exception as exc: # one dead row, not a dead page
log.warning("web: %s browse failed: %s", label, exc)
continue
if metas:
rows.append(f"<h2>{label}</h2>" + _web_cards(ctype, metas[:12]))
if not rows:
rows.append("<p class=mute>TMDB did not answer — search still works "
"once it does.</p>")
body = ("<p class=mute>Find a title and download it into the library on "
"stortron; Jellyfin and Kodi pick it up from there. "
"Fetched so far: <a href='/library'>/library</a> · "
"Stremio install URL: <code>/manifest.json</code></p>"
+ "".join(rows))
return _web_page("vsembed", body)
@app.route("/web/search")
def web_search():
query = (request.args.get("q") or "").strip()
ctype = request.args.get("type") or ""
types = [ctype] if ctype in ("series", "movie") else ["series", "movie"]
if not query:
return redirect("/", code=303)
parts = [f"<h1>Results for “{html.escape(query)}”</h1>"]
for t in types:
try:
metas = _tmdb_search(t, query)
except Exception as exc:
log.warning("web: search %r (%s) failed: %s", query, t, exc)
parts.append(f"<p class=warn>TMDB search failed for {t}.</p>")
continue
heading = "Series" if t == "series" else "Films"
parts.append(f"<h2>{heading}</h2>" + (_web_cards(t, metas) if metas else
"<p class=mute>Nothing found.</p>"))
return _web_page(f"{query} — vsembed", "".join(parts))
def _web_header(show: dict) -> str:
name = show.get("name") or show.get("title") or ""
year = (show.get("first_air_date") or show.get("release_date") or "")[:4]
poster = _tmdb_poster(show.get("poster_path"))
img = f"<img alt='' src='{html.escape(poster)}'>" if poster else ""
return (f"<div class=title>{img}<div class=about>"
f"<h1>{html.escape(name)}"
+ (f" <span class=mute>({year})</span>" if year else "") + "</h1>"
f"<p>{html.escape(show.get('overview') or '')}</p>")
def _web_movie(tmdb: str) -> Response:
film = _tmdb_get(f"movie/{tmdb}")
state = _web_state("movie", tmdb)
action = ("" if state["status"] == "running"
else _web_delete_form("movie", tmdb, 0, 0)
if state["status"] == "held"
else _web_download_form("movie", tmdb))
body = (_web_header(film) + _web_low_space()
+ f"<p>{_web_state_html(state, f'/web/watch/movie/{tmdb}')} "
f"{action}</p></div></div>")
return _web_page(film.get("title") or tmdb, body,
refresh=state["status"] == "running")
def _web_series(tmdb: str) -> Response:
show = _tmdb_get(f"tv/{tmdb}")
seasons = [s["season_number"] for s in show.get("seasons") or []
if isinstance(s.get("season_number"), int)
and s["season_number"] >= 1]
try:
season = int(request.args.get("s") or (seasons[0] if seasons else 1))
except ValueError:
abort(400)
tabs = "".join(
f"<a href='/web/title/series/{tmdb}?s={n}'"
+ (" class=on" if n == season else "") + f">Season {n}</a>"
for n in seasons)
episodes = _tmdb_get(f"tv/{tmdb}/season/{season}").get("episodes") or []
today = time.strftime("%Y-%m-%d")
ident = f"tmdb:{tmdb}"
running = any(_season_jobs.get(f"season:{ident}:{season}:{h}") == "running"
for h in VIDCORE_HEIGHTS)
rows, aired_missing = [], 0
owned = set(_fetched_dests())
for ep in episodes:
num = ep.get("episode_number")
if not isinstance(num, int):
continue
air = (ep.get("air_date") or "").strip()
name = html.escape(ep.get("name") or f"Episode {num}")
# By air date, never by count -- the rule _released_episodes uses, so
# this page and the season job agree on what exists.
if not air or air > today:
rows.append(f"<tr><td class=n>{num}</td><td>{name}</td>"
"<td class=act><span class='st mute'>not aired</span>"
"</td></tr>")
continue
state = _web_state("series", tmdb, season, num)
running = running or state["status"] == "running"
if state["status"] not in ("held", "running"):
aired_missing += 1
action = ("" if state["status"] == "running"
else _web_delete_form("series", tmdb, season, num, owned)
if state["status"] == "held"
else _web_download_form("series", tmdb, season, num))
rows.append(f"<tr><td class=n>{num}</td><td>{name}<br>"
f"<span class=mute>{air}</span></td>"
f"<td class=act>"
f"{_web_state_html(state, f'/web/watch/series/{tmdb}/{season}/{num}')}"
f" {action}</td></tr>")
season_action = (
f"<p>{_web_download_form('series', tmdb, season, 0, 'season', f'Get all {aired_missing} missing')}</p>"
if aired_missing > 1 else "")
body = (_web_header(show) + _web_low_space() + "</div></div>"
+ f"<div class=tabs>{tabs}</div>" + season_action
+ "<table>" + "".join(rows) + "</table>")
return _web_page(show.get("name") or tmdb, body, refresh=running)
@app.route("/web/title/<kind>/<tmdb>")
def web_title(kind: str, tmdb: str):
if kind not in ("movie", "series") or not tmdb.isdigit():
abort(404)
try:
return _web_movie(tmdb) if kind == "movie" else _web_series(tmdb)
except HTTPException:
raise # our own abort(), not TMDB's
except Exception:
# Broad like the catalog routes: every failure here is a TMDB call.
# The traceback goes to the journal so a real bug is not filed as one.
log.exception("web: title page failed for %s/%s", kind, tmdb)
return _web_page("TMDB unavailable",
"<p class=warn>TMDB did not answer for this title. "
"Reload in a moment.</p>", status=502)
@app.route("/web/download", methods=["POST"])
def web_download():
"""POST only -- see the section comment. Validated strictly: these values
become a job key, a TMDB path and, via _library_path, a filename."""
f = request.form
kind, tmdb = f.get("kind") or "", f.get("tmdb") or ""
if kind not in ("movie", "series") or not tmdb.isdigit():
abort(400)
try:
season = int(f.get("season") or 0)
episode = int(f.get("episode") or 0)
height = int(f.get("height") or VIDCORE_MAX_HEIGHT)
except ValueError:
abort(400)
if height not in VIDCORE_HEIGHTS:
abort(400)
ident = f"tmdb:{tmdb}"
scope = f.get("scope") or "one"
back = f"/web/title/{kind}/{tmdb}"
if kind == "movie":
season = episode = 0
_ensure_ingest("movie", ident, 0, 0, height)
elif scope == "season":
if season < 1:
abort(400)
state = _start_season_job(ident, season, height)
log.info("web: season fetch %s S%02d -> %s", ident, season, state)
else:
if season < 1 or episode < 1:
abort(400)
_ensure_ingest("series", ident, season, episode, height)
log.info("web: download requested %s:%s:%s:%s @%sp",
kind, ident, season, episode, height)
if kind == "series":
back += f"?s={season}"
return redirect(back, code=303)
def _human_bytes(n: Optional[float]) -> str:
if not n:
return "—"
return f"{n / 1e9:.1f} GB" if n >= 1e9 else f"{n / 1e6:.0f} MB"
def _human_secs(s: Optional[float]) -> str:
if s is None:
return "—"
s = int(s)
return f"{s // 60}m {s % 60:02d}s" if s >= 60 else f"{s}s"
def _title_href(kind: Optional[str], ident: Optional[str],
season: Optional[int]) -> Optional[str]:
"""Title page for a job, when its ident is a TMDB one (web and fxl jobs).
A tt... ident from a cinemeta row has no page to link to."""
if kind not in ("movie", "series") or not (ident or "").startswith("tmdb:"):
return None
href = f"/web/title/{kind}/{ident.split(':', 1)[1]}"
return href + f"?s={season}" if kind == "series" else href
def _downloads_active_html() -> tuple[str, bool]:
now = time.time()
rows = []
with _season_guard:
seasons = {k: dict(v) for k, v in _season_info.items()}
for key, s in seasons.items():
total = len(s.get("episodes") or [])
where = (f"episode {s.get('index') or 0} of {total}" if total
else "listing episodes")
rows.append(
f"<tr><td>{html.escape(str(s.get('title') or s.get('ident')))}"
f"<br><span class=mute>Season {s.get('season')} · {where}</span></td>"
f"<td>season</td><td>{s.get('height')}p</td><td></td>"
f"<td>{_human_secs(now - (s.get('started') or now))}</td>"
f"<td class=act>{_web_cancel_form(key, 'Cancel season')}</td></tr>")
with _mp4_jobs_guard:
jobs = {k: dict(v) for k, v in _job_info.items()}
for key, j in sorted(jobs.items(), key=lambda kv: kv[1].get("started") or 0):
phase = j.get("phase") or ""
progress = ""
if phase == "downloading" and j.get("dest"):
try:
got = os.path.getsize(_part_path(j["dest"]))
except OSError:
got = 0
since = j.get("downloading_since") or j.get("started") or now
rate = got / max(now - since, 1e-6)
progress = f" · {_human_bytes(got)} · {rate / 1e6:.1f} MB/s"
title = html.escape(str(j.get("title") or j.get("ident")))
href = _title_href(j.get("kind"), j.get("ident"), j.get("season"))
if href:
title = f"<a href='{href}'>{title}</a>"
rows.append(
f"<tr><td>{title}<br><span class=mute>"
f"{html.escape(str(j.get('label') or ''))}</span></td>"
f"<td><span class='st running'>{html.escape(phase)}{progress}</span></td>"
f"<td>{j.get('height')}p</td>"
f"<td>{html.escape(j.get('source') or '')}</td>"
f"<td>{_human_secs(now - (j.get('started') or now))}</td>"
f"<td class=act>{_preview_link(key)} "
f"{_web_cancel_form(key, 'Cancel')}</td></tr>")
if not rows:
return "<p class=mute>Nothing downloading.</p>", False
return ("<table><tr><th>Title</th><th>Status</th><th>Quality</th>"
"<th>Source</th><th>Elapsed</th><th></th></tr>"
+ "".join(rows) + "</table>"), True
def _web_cancel_form(key: str, label: str) -> str:
return ("<form class=dl method=post action='/web/cancel'>"
f"<input type=hidden name='key' value='{html.escape(key)}'>"
f"<button>{label}</button></form>")
def _downloads_history_html() -> str:
entries = _read_history(200)
if not entries:
return "<p class=mute>No downloads yet.</p>"
rows = []
for h in entries:
outcome = str(h.get("outcome") or "")
cls = ("held" if outcome == "done" else
"mute" if outcome == "cancelled" else "failed")
title = html.escape(str(h.get("title") or h.get("ident") or ""))
href = _title_href(h.get("kind"), h.get("ident"), h.get("season"))
if href:
title = f"<a href='{href}'>{title}</a>"
when = time.strftime("%b %d %H:%M", time.localtime(h.get("ts") or 0))
source = h.get("source") or ("already held" if outcome == "done" else "")
rows.append(
f"<tr><td class=mute>{when}</td>"
f"<td>{title}<br><span class=mute>"
f"{html.escape(str(h.get('label') or ''))}</span></td>"
f"<td><span class='st {cls}'>{html.escape(outcome)}</span></td>"
f"<td>{html.escape(source)}</td>"
f"<td class=n>{_human_bytes(h.get('bytes'))}</td>"
f"<td class=n>{_human_secs(h.get('took'))}</td></tr>")
return ("<table><tr><th>When</th><th>Title</th><th>Outcome</th>"
"<th>Source</th><th>Size</th><th>Took</th></tr>"
+ "".join(rows) + "</table>")
@app.route("/web/downloads")
def web_downloads():
active, live = _downloads_active_html()
body = ("<h1>Downloads</h1>" + _web_low_space()
+ "<h2>Now</h2>" + active
+ "<h2>History</h2>" + _downloads_history_html())
return _web_page("Downloads — vsembed", body, refresh=live)
@app.route("/web/cancel", methods=["POST"])
def web_cancel():
"""POST only. A key that no longer names a live job is not an error: the
job finished between the page rendering and the click."""
key = request.form.get("key") or ""
if key.startswith("season:"):
_cancel_season(key)
elif key:
_cancel_job(key)
return redirect("/web/downloads", code=303)
# ---------------------------------- watch -----------------------------------
# Play a library copy in the browser to judge what it actually is.
#
# Audio a browser cannot decode plays as SILENCE, and it is the DEFAULT-flagged
# track that decides it (see _browser_audio). Measured 2026-10-04 in Chromium:
# canPlayType for ec-3 and ac-3 is "", an eac3-only file decoded 0 audio bytes,
# and the browser never switches to another track it could have played.
# Jellyfin gets sound out of these by re-encoding the audio into an HLS stream
# per viewer (labs#13 has the trace); that machinery is deliberately not
# rebuilt here. Picture-only is enough to judge picture quality, and the page
# says so rather than leaving the silence to be mistaken for a broken copy.
BROWSER_SILENT_AUDIO = ("ac3", "eac3", "dts", "truehd", "mlp")
_quality_cache: dict[tuple, Optional[dict]] = {}
def _ffprobe_file(path: str) -> Optional[dict]:
try:
out = subprocess.run(
["ffprobe", "-v", "error", "-show_entries",
"stream=index,codec_type,codec_name,profile,width,height,"
"avg_frame_rate,bit_rate,channels,pix_fmt:stream_tags=language"
":stream_disposition=default:format=duration,size,bit_rate",
"-of", "json", path],
capture_output=True, text=True, timeout=60)
return json.loads(out.stdout) if out.returncode == 0 else None
except Exception as exc:
log.warning("watch: could not probe %s (%s)", os.path.basename(path), exc)
return None
def _quality(path: str) -> Optional[dict]:
"""ffprobe's view of a local file, keyed on its identity like
_cached_media_duration, so a page view does not spawn ffprobe each time
and a re-ingest invalidates it for free."""
try:
st = os.stat(path)
except OSError:
return None
key = (os.path.abspath(path), st.st_size, int(st.st_mtime))
if key not in _quality_cache:
_quality_cache[key] = _ffprobe_file(path)
return _quality_cache[key]
def _browser_audio(audio: list[dict]) -> Optional[dict]:
"""The track a browser plays: the DEFAULT-flagged one, not the first.
Measured 2026-10-04 in Chromium: aac first + not default decoded 0 audio
bytes; eac3 first + aac default played the aac. With no flag recorded
(HdHub's probe does not ask for it), the first track is the best guess."""
for a in audio:
if (a.get("disposition") or {}).get("default"):
return a
return audio[0] if audio else None
def _fps(rate: Optional[str]) -> Optional[float]:
try:
num, den = (rate or "").split("/")
return float(num) / float(den) if float(den) else None
except ValueError:
return None
def _mbps(bits: Optional[str]) -> Optional[str]:
try:
return f"{int(bits) / 1e6:.1f} Mbps"
except (TypeError, ValueError):
return None
def _source_of(path: str) -> Optional[str]:
"""Where a copy came from, if this addon's history recorded it."""
target = os.path.abspath(path)
for h in _read_history(2000):
if (h.get("outcome") == "done" and h.get("dest")
and os.path.abspath(h["dest"]) == target and h.get("source")):
return h["source"]
return None
def _web_target(kind: str, tmdb: str, season: int, episode: int):
"""(path, meta) of a held copy, or abort(404). Never fetches anything."""
if kind not in ("movie", "series") or not tmdb.isdigit():
abort(404)
if kind == "movie":
season = episode = 0
ident = f"tmdb:{tmdb}"
path = _playable_path(kind, ident, season, episode)
if not path:
abort(404)
return path, _library_meta(kind, ident, season, episode)
def _serve_web_file(kind: str, tmdb: str, season: int = 0, episode: int = 0):
path, _meta = _web_target(kind, tmdb, season, episode)
return send_file(path, mimetype="video/mp4", conditional=True)
@app.route("/web/file/movie/<tmdb>.mp4")
def web_file_movie(tmdb: str):
return _serve_web_file("movie", tmdb)
@app.route("/web/file/series/<tmdb>/<int:season>/<int:episode>.mp4")
def web_file_series(tmdb: str, season: int, episode: int):
return _serve_web_file("series", tmdb, season, episode)
@app.route("/web/file/<kind>/<path:rest>")
def web_file_other(kind: str, rest: str):
abort(404)
def _quality_html(path: str, meta: Optional[dict]) -> str:
q = _quality(path) or {}
streams = q.get("streams") or []
video = next((s for s in streams if s.get("codec_type") == "video"), None)
audio = [s for s in streams if s.get("codec_type") == "audio"]
fmt = q.get("format") or {}
rows = []
if video:
fps = _fps(video.get("avg_frame_rate"))
rows.append(("Picture",
f"{video.get('width')}×{video.get('height')} · "
f"{video.get('codec_name')} {video.get('profile') or ''}"
+ (f" · {fps:.2f} fps" if fps else "")))
rate = _mbps(video.get("bit_rate"))
rows.append(("Video bitrate", rate or
f"{_mbps(fmt.get('bit_rate')) or '—'} (whole file)"))
for i, a in enumerate(audio, 1):
lang = (a.get("tags") or {}).get("language") or "und"
rows.append((f"Audio {i}",
f"{a.get('codec_name')} · {a.get('channels') or '?'} ch · {lang}"
+ (f" · {_mbps(a.get('bit_rate'))}" if a.get("bit_rate") else "")))
try:
seconds = float(fmt.get("duration"))
except (TypeError, ValueError):
seconds = None
try:
expected = float((meta or {}).get("runtime")) * 60
except (TypeError, ValueError):
expected = None
length = _human_secs(seconds)
if seconds and expected:
length += f" · {seconds / expected * 100:.0f}% of TMDB's {expected / 60:.0f} min"
rows.append(("Length", length))
rows.append(("Size", _human_bytes(os.path.getsize(path))))
rows.append(("Source", _source_of(path) or "not recorded"))
rows.append(("Subtitles", _subtitle_summary(path)))
warn = ""
played = _browser_audio(audio)
if played and played.get("codec_name") in BROWSER_SILENT_AUDIO:
warn = ("<p class=warn>This copy plays without sound here: its default "
f"audio track is {html.escape(played.get('codec_name'))}, "
"which browsers cannot decode (and they do not switch to "
"another track). Kodi and Jellyfin play it with sound.</p>")
return warn + "<table>" + "".join(
f"<tr><td class=mute>{k}</td><td>{html.escape(str(v))}</td></tr>"
for k, v in rows) + "</table>"
def _web_watch(kind: str, tmdb: str, season: int = 0, episode: int = 0):
path, meta = _web_target(kind, tmdb, season, episode)
title = (meta or {}).get("title") or tmdb
if kind == "series":
file_url = f"/web/file/series/{tmdb}/{season}/{episode}.mp4"
back = f"/web/title/series/{tmdb}?s={season}"
sub = f"S{season:02d}E{episode:02d}"
if (meta or {}).get("episode_title"):
sub += f" · {meta['episode_title']}"
else:
file_url = f"/web/file/movie/{tmdb}.mp4"
back = f"/web/title/movie/{tmdb}"
sub = (meta or {}).get("year") or ""
body = (f"<p><a href='{back}'>← {html.escape(title)}</a></p>"
f"<h1>{html.escape(title)} <span class=mute>{html.escape(sub)}</span></h1>"
f"<video controls preload=metadata playsinline src='{file_url}'"
" style='width:100%;max-height:70vh;background:#000'></video>"
"<h2>What this copy is</h2>" + _quality_html(path, meta)
+ _web_replace_form(kind, tmdb, season, episode, path)
+ _web_delete_form(kind, tmdb, season, episode))
return _web_page(f"{title} {sub} — watch", body)
# ---------------------------------- delete ----------------------------------
# Remove a copy found to be bad. Immediate -- chosen over a trash folder -- so
# the guard rails are about SCOPE. Only copies this addon fetched can go: the
# library is ~745 GB, nearly all of it arrived by other means, and nothing here
# may touch that. Ownership is by FILE, like the dest lock, because a Kodi
# cinemeta job (tt…) and a web job (tmdb:…) name the same destination.
#
# No re-fetch block afterwards (the user's choice): a later play or download
# simply fetches it again.
def _fetched_dests() -> dict[str, list[str]]:
"""{absolute library path: [queue keys that fetched it]}.
A title TMDB cannot identify right now is left out, which makes it
undeletable for the moment -- the safe direction for a destructive action.
"""
out: dict[str, list[str]] = {}
for key, entry in _load_job_queue().items():
args = entry.get("args") or []
if len(args) < 5:
continue
kind, ident = args[0], args[1]
season, episode = int(args[2]), int(args[3])
meta = _library_meta(kind, ident, season, episode)
if not meta:
continue
dest = _library_path(kind, meta["title"], meta["year"], meta["tmdb_id"],
season, episode, meta["episode_title"])
out.setdefault(os.path.abspath(dest), []).append(key)
return out
def _sidecars_of(path: str) -> list[str]:
"""Exactly the names _sidecar_paths writes for this copy:
`<stem>.<code>.srt` and `<stem>.<code>.<n>.srt`. Matched by pattern, not
a bare glob, so nothing that merely starts with the same name is caught."""
folder = os.path.dirname(path)
stem = os.path.splitext(os.path.basename(path))[0]
pattern = re.compile(re.escape(stem) + r"\.[^./\\]+(\.\d+)?\.srt$")
try:
names = os.listdir(folder)
except OSError:
return []
return [os.path.join(folder, n) for n in names if pattern.fullmatch(n)]
def _prune_empty_dirs(path: str) -> None:
"""Remove the season and show folders a delete leaves empty -- and stop at
Shows/ or Movies/, which are never removed. rmdir refuses a non-empty
folder, so this cannot take anything with it."""
root = os.path.abspath(LIBRARY_ROOT)
stops = {root, os.path.join(root, "Shows"), os.path.join(root, "Movies")}
folder = os.path.dirname(os.path.abspath(path))
while folder.startswith(root + os.sep) and folder not in stops:
try:
os.rmdir(folder)
except OSError:
return
folder = os.path.dirname(folder)
def _delete_copy(kind: str, tmdb: str, season: int, episode: int) -> tuple[int, str]:
"""(http status, what happened). 200 on success."""
ident = f"tmdb:{tmdb}"
meta = _library_meta(kind, ident, season, episode)
if not meta:
return 404, "title not identifiable"
dest = os.path.abspath(_library_path(kind, meta["title"], meta["year"],
meta["tmdb_id"], season, episode,
meta["episode_title"]))
owners = _fetched_dests().get(dest)
if not owners:
return 404, "not a copy this addon fetched"
lock = _dest_lock(dest)
if not lock.acquire(blocking=False):
return 409, "a download is writing this file"
try:
legacy = [p for h in VIDCORE_HEIGHTS
for p in {_mp4_path(kind, ident, season, episode, h)}
if os.path.exists(p)]
if not os.path.exists(dest) and not legacy:
return 404, "no copy on disk"
freed = 0
for p in ([dest] if os.path.exists(dest) else []) + _sidecars_of(dest) + legacy:
try:
freed += os.path.getsize(p)
os.remove(p)
log.info("delete: removed %s", p)
except OSError as exc:
log.warning("delete: could not remove %s (%s)", p, exc)
_forget_verified(dest)
_prune_empty_dirs(dest)
with _mp4_jobs_guard:
for key in owners:
_set_job_state(key, "deleted")
_append_history({
"ts": time.time(), "key": owners[0], "kind": kind, "ident": ident,
"season": season, "episode": episode, "title": meta["title"],
"label": (f"S{season:02d}E{episode:02d}" if kind == "series"
else "movie"),
"outcome": "deleted", "source": None, "bytes": freed,
"took": 0, "dest": dest})
finally:
lock.release()
# In the background: measured live, Jellyfin took ~5 s to answer, and the
# click waited on it. The delete is complete either way.
threading.Thread(target=_notify_jellyfin, args=(dest, "Deleted"),
name="jellyfin-notify", daemon=True).start()
return 200, f"deleted {os.path.basename(dest)}"
# Jellyfin's real-time monitor picked up an ingested file on its own on
# 2026-10-04 (3 s before the ingest logged "done"), so for adds this is a
# backstop. It is here for what the monitor is not proven on -- deletes and
# subtitle sidecars -- and for an event the monitor misses. The key lives in a
# gitignored file beside this one (api key "stremio-vsembed").
JELLYFIN_CONFIG = os.path.join(os.path.dirname(os.path.abspath(__file__)),
"jellyfin.json")
_jellyfin_cfg: Optional[dict] = None
def _notify_jellyfin(path: str, update_type: str) -> None:
"""POST /Library/Media/Updated for one path. Never raises, never blocks
for long: it reports an action and must not be able to fail it."""
global _jellyfin_cfg
try:
if _jellyfin_cfg is None:
with open(JELLYFIN_CONFIG) as f:
_jellyfin_cfg = json.load(f)
r = requests.post(
_jellyfin_cfg["url"].rstrip("/") + "/Library/Media/Updated",
json={"Updates": [{"Path": path, "UpdateType": update_type}]},
headers={"X-Emby-Token": _jellyfin_cfg["api_key"]}, timeout=5)
r.raise_for_status()
log.info("jellyfin: %s %s", update_type, os.path.basename(path))
except FileNotFoundError:
return
except Exception as exc:
log.warning("jellyfin: notify %s failed for %s (%s)",
update_type, os.path.basename(path), exc)
def _web_delete_form(kind: str, tmdb: str, season: int, episode: int,
owned: Optional[set] = None) -> str:
"""The Delete button, only for a copy this addon fetched. `owned` lets a
page that renders many rows compute _fetched_dests once."""
meta = _library_meta(kind, f"tmdb:{tmdb}", season, episode)
if not meta:
return ""
dest = os.path.abspath(_library_path(kind, meta["title"], meta["year"],
meta["tmdb_id"], season, episode,
meta["episode_title"]))
if dest not in (owned if owned is not None else _fetched_dests()):
return ""
name = meta["title"] + (f" S{season:02d}E{episode:02d}" if kind == "series" else "")
prompt = json.dumps(f"Delete {name}? The file and its subtitles are removed "
"now; this cannot be undone.")
return ("<form class=dl method=post action='/web/delete' "
f"onsubmit='return confirm({html.escape(prompt, quote=True)})'>"
f"<input type=hidden name='kind' value='{kind}'>"
f"<input type=hidden name='tmdb' value='{tmdb}'>"
f"<input type=hidden name='season' value='{season}'>"
f"<input type=hidden name='episode' value='{episode}'>"
"<button class=del>Delete</button></form>")
@app.route("/web/delete", methods=["POST"])
def web_delete():
"""POST only, like every action here: a stray GET must never delete."""
f = request.form
kind, tmdb = f.get("kind") or "", f.get("tmdb") or ""
if kind not in ("movie", "series") or not tmdb.isdigit():
abort(400)
try:
season = int(f.get("season") or 0)
episode = int(f.get("episode") or 0)
except ValueError:
abort(400)
if kind == "movie":
season = episode = 0
status, what = _delete_copy(kind, tmdb, season, episode)
log.info("web: delete %s:%s:%s:%s -> %s %s",
kind, tmdb, season, episode, status, what)
if status != 200:
return _web_page("Not deleted", f"<p class=warn>{html.escape(what)}</p>",
status=status)
back = f"/web/title/{kind}/{tmdb}" + (f"?s={season}" if kind == "series" else "")
return redirect(back, code=303)
# ---------------------------------- replace ---------------------------------
# A copy that is wrong in a way only a person notices (2026-10-14: VisionQuest
# S01E01's audio tagged eng, spoken French). Delete + Download fetched the same
# copy again, because vidcore is always first. Replace rejects the source the
# copy came from, for this file, and forces a re-fetch from the next one. The
# forced job writes a .part and renames, so the old copy stays until the new
# one is whole -- and stays for good if no other source has it.
def _web_form_target(f) -> tuple[str, str, int, int]:
kind, tmdb = f.get("kind") or "", f.get("tmdb") or ""
if kind not in ("movie", "series") or not tmdb.isdigit():
abort(400)
try:
season = int(f.get("season") or 0)
episode = int(f.get("episode") or 0)
except ValueError:
abort(400)
if kind == "movie":
season = episode = 0
return kind, tmdb, season, episode
def _web_back(kind: str, tmdb: str, season: int) -> str:
return f"/web/title/{kind}/{tmdb}" + (f"?s={season}" if kind == "series" else "")
def _replace_copy(kind: str, tmdb: str, season: int, episode: int,
height: int) -> tuple[int, str]:
"""(http status, what happened). 200 when the re-fetch was started."""
dest = _web_dest(kind, tmdb, season, episode)
if not dest or not os.path.exists(dest):
return 404, "no copy on disk"
# The forced job overwrites the file: the same scope rule as Delete.
if dest not in _fetched_dests():
return 404, "not a copy this addon fetched"
source = _source_of(dest)
if not source:
return 409, "where this copy came from was not recorded"
if not [sid for sid in _ingest_order(dest) if sid != source]:
return 409, f"no source other than {_source_name(source)} is left"
_reject_source(dest, source)
state = _start_mp4_job(kind, f"tmdb:{tmdb}", season, episode, height,
force=True)
return 200, f"{_source_name(source)} rejected, re-fetch {state}"
@app.route("/web/replace", methods=["POST"])
def web_replace():
"""POST only, like every action here."""
kind, tmdb, season, episode = _web_form_target(request.form)
try:
height = int(request.form.get("height") or VIDCORE_MAX_HEIGHT)
except ValueError:
abort(400)
if height not in VIDCORE_HEIGHTS:
abort(400)
status, what = _replace_copy(kind, tmdb, season, episode, height)
log.info("web: replace %s:%s:%s:%s -> %s %s",
kind, tmdb, season, episode, status, what)
if status != 200:
return _web_page("Not replaced", f"<p class=warn>{html.escape(what)}</p>",
status=status)
return redirect(_web_back(kind, tmdb, season), code=303)
@app.route("/web/source/<action>", methods=["POST"])
def web_source(action: str):
"""Reject or allow again one source for one file."""
if action not in ("reject", "allow"):
abort(404)
kind, tmdb, season, episode = _web_form_target(request.form)
sid = request.form.get("source") or ""
if sid not in SOURCES:
abort(400)
dest = _web_dest(kind, tmdb, season, episode)
if not dest:
abort(404)
(_reject_source if action == "reject" else _allow_source)(dest, sid)
return redirect(_web_back(kind, tmdb, season), code=303)
def _web_replace_form(kind: str, tmdb: str, season: int, episode: int,
path: str) -> str:
"""'Bad copy' button, when there is somewhere else to get it from."""
if os.path.abspath(path) not in _fetched_dests():
return ""
source = _source_of(path)
others = [sid for sid in _ingest_order(path) if sid != source]
if not source or not others:
return ""
prompt = json.dumps(f"Stop using {_source_name(source)} for this and fetch it "
f"from {_source_name(others[0])}? The current copy stays "
"until the new one is complete.")
return ("<form class=dl method=post action='/web/replace' "
f"onsubmit='return confirm({html.escape(prompt, quote=True)})'>"
f"<input type=hidden name='kind' value='{kind}'>"
f"<input type=hidden name='tmdb' value='{tmdb}'>"
f"<input type=hidden name='season' value='{season}'>"
f"<input type=hidden name='episode' value='{episode}'>"
f"<button>Bad copy — replace from {html.escape(_source_name(others[0]))}"
"</button></form>")
@app.route("/web/watch/movie/<tmdb>")
def web_watch_movie(tmdb: str):
return _web_watch("movie", tmdb)
@app.route("/web/watch/series/<tmdb>/<int:season>/<int:episode>")
def web_watch_series(tmdb: str, season: int, episode: int):
return _web_watch("series", tmdb, season, episode)
# --------------------------------- preview ----------------------------------
# Watch the stream a RUNNING download is copying, judge it, then let it finish
# or Cancel. It plays the very source the job copies -- recorded in
# _job_info[key]["preview"] when the copy starts -- not a second resolve. The
# .part cannot be played instead: +faststart writes the moov atom last.
#
# Measured 2026-10-04 (labs#13):
# vidcore variant playlists are REFERER-GATED -- 403 bare, 200 with
# `Referer: https://vidcore.net/` (6/6 on one egress), CORS `*`.
# A browser cannot set Referer, so they play through the relay
# below, which adds it.
# HdHub direct files, no referer. Chromium plays h264 in .mkv as well as
# .mp4, so the browser plays them straight from the source.
HLS_JS = "https://cdnjs.cloudflare.com/ajax/libs/hls.js/1.6.16/hls.min.js"
VIDCORE_REFERER = "https://vidcore.net/"
def _preview_link(key: Optional[str]) -> str:
if not key or not (_job_info.get(key) or {}).get("preview"):
return ""
return (f" <a class=watch href='/web/preview?key={quote(key, safe='')}'>"
"Preview</a>")
def _live_preview(key: str) -> Optional[dict]:
return (_job_info.get(key) or {}).get("preview")
def _preview_quality_html(pv: dict) -> str:
if pv["type"] == "hls":
rate = f"{pv['bandwidth'] / 1e6:.1f} Mbps" if pv.get("bandwidth") else "?"
rows = [("Source", "vidcore (HLS)"),
("Variant", f"{pv.get('height') or '?'}p · {rate}")]
return "<table>" + "".join(
f"<tr><td class=mute>{k}</td><td>{html.escape(v)}</td></tr>"
for k, v in rows) + "</table>"
probe = pv.get("probe") or {}
streams = probe.get("streams") or []
video = next((s for s in streams if s.get("codec_type") == "video"), {})
audio = [s for s in streams if s.get("codec_type") == "audio"]
rows = [("Source", "HdHub (direct file)"),
("File", pv.get("filename") or ""),
("Picture", f"{video.get('codec_name') or '?'} · "
f"{video.get('height') or pv.get('height') or '?'}p"),
("Size", _human_bytes(pv.get("size")))]
for i, a in enumerate(audio, 1):
lang = (a.get("tags") or {}).get("language") or "und"
rows.append((f"Audio {i}", f"{a.get('codec_name')} · {lang}"))
warn = ""
played = _browser_audio(audio)
if played and played.get("codec_name") in BROWSER_SILENT_AUDIO:
warn = ("<p class=warn>This preview plays without sound: the default "
f"audio track is {html.escape(played.get('codec_name'))}, "
"which browsers cannot decode. The copy keeps it.</p>")
return warn + "<table>" + "".join(
f"<tr><td class=mute>{k}</td><td>{html.escape(str(v))}</td></tr>"
for k, v in rows) + "</table>"
@app.route("/web/preview")
def web_preview():
key = request.args.get("key") or ""
info = _job_info.get(key)
if info is None:
state = _mp4_jobs.get(key) or ""
parts = key.split(":")
if state and len(parts) >= 5 and parts[0] in ("movie", "series"):
# Finished (or stopped) since the link was rendered: the title page
# has the outcome, and Watch if it landed.
href = _title_href(parts[0], ":".join(parts[1:-3]), int(parts[-3]))
if href:
return redirect(href, code=303)
abort(404)
title = html.escape(str(info.get("title") or info.get("ident")))
head = (f"<h1>Preview: {title} <span class=mute>"
f"{html.escape(str(info.get('label') or ''))}</span></h1>")
pv = info.get("preview")
if not pv:
return _web_page("Preview", head
+ f"<p class=mute>{html.escape(info.get('phase') or '')}"
" — the stream appears here once the download starts "
"copying.</p>" + _web_cancel_form(key, "Cancel download"),
refresh=True)
k = quote(key, safe="")
if pv["type"] == "hls":
player = (
"<video id=pv controls playsinline style='width:100%;max-height:70vh;"
"background:#000'></video>"
f"<script src='{HLS_JS}'></script><script>"
"(function(){var v=document.getElementById('pv'),"
f"src='/web/relay/playlist?key={k}';"
"if(window.Hls&&Hls.isSupported()){var h=new Hls();h.loadSource(src);"
"h.attachMedia(v);}else{v.src=src;}})();</script>")
else:
player = (f"<video controls playsinline preload=metadata src='{html.escape(pv['url'])}' "
"style='width:100%;max-height:70vh;background:#000'></video>")
body = (head + player
+ "<p>Downloading in the background. Keep it, or "
+ _web_cancel_form(key, "Cancel download") + "</p>"
+ "<h2>What is being copied</h2>" + _preview_quality_html(pv))
return _web_page(f"Preview — {info.get('title') or ''}", body)
def _relay_target(key: str) -> dict:
pv = _live_preview(key)
if not pv or pv.get("type") != "hls":
abort(404)
return pv
@app.route("/web/relay/playlist")
def web_relay_playlist():
key = request.args.get("key") or ""
pv = _relay_target(key)
try:
r = requests.get(pv["url"], headers={"User-Agent": UA,
"Referer": VIDCORE_REFERER},
timeout=REQUEST_TIMEOUT)
r.raise_for_status()
except Exception as exc:
log.warning("preview: playlist fetch failed for %s (%s)", key, exc)
abort(502)
k = quote(key, safe="")
# Exactly the URLs this playlist lists may be relayed. Not "same host as
# the playlist": measured live, segments came from a different CDN host
# (paperlake.top) than the playlist itself.
allowed = pv.setdefault("allowed", set())
def relayed(uri: str) -> str:
absolute = urljoin(pv["url"], uri)
allowed.add(absolute)
return f"/web/relay/seg?key={k}&u={quote(absolute, safe='')}"
out = []
for line in r.text.splitlines():
if line and not line.startswith("#"):
out.append(relayed(line.strip()))
elif 'URI="' in line:
out.append(re.sub(r'URI="([^"]+)"',
lambda m: f'URI="{relayed(m.group(1))}"', line))
else:
out.append(line)
return Response("\n".join(out) + "\n",
mimetype="application/vnd.apple.mpegurl")
@app.route("/web/relay/seg")
def web_relay_seg():
"""One segment, fetched with the referer the CDN demands. Only a URL the
job's own relayed playlist listed -- never an open proxy."""
key = request.args.get("key") or ""
pv = _relay_target(key)
u = request.args.get("u") or ""
if u not in pv.get("allowed", ()):
abort(403)
headers = {"User-Agent": UA, "Referer": VIDCORE_REFERER}
if request.headers.get("Range"):
headers["Range"] = request.headers["Range"]
try:
up = requests.get(u, headers=headers, timeout=REQUEST_TIMEOUT, stream=True)
except Exception as exc:
log.warning("preview: segment fetch failed for %s (%s)", key, exc)
abort(502)
if up.status_code >= 400:
up.close()
abort(502)
passed = {h: up.headers[h] for h in ("Content-Length", "Content-Range",
"Accept-Ranges") if h in up.headers}
# Segments arrive disguised as text/html; the player only needs bytes.
return Response(up.iter_content(64 * 1024), status=up.status_code,
headers=passed, mimetype="video/mp4",
direct_passthrough=True)
# The sources this addon knows about. Declared here, at the end, so each entry
# can point at the resolver it uses. Dead and unusable sources are RECORDED
# rather than deleted: the next person to notice that vidlink serves progressive
# MP4 should find the measurement, not repeat the day it took to make it.
SOURCES_DEFAULT: dict[str, Source] = {s.id: s for s in (
Source(
id="vidcore", name="vidcore (flixlab)",
container="hls-fmp4", # muxed fMP4 — why every copy needs a remux
ingestable=True, browser=True, heights=VIDCORE_HEIGHTS,
resolve=_resolve_vidcore,
note="The only live source. Resolve ~11s via xvfb+patchright. Muxed "
"fMP4: ffmpeg plays it with audio but ignores seeks, ISA seeks but "
"starves the audio track — see labs#4.",
),
Source(
id="vidlink", name="vidlink", container="progressive-mp4",
ingestable=False, browser=True, heights=(1080, 480, 360), enabled=False,
note="Measured 2026-08-11. The API is excellent: "
"vidlink.pro/api/b/tv/<id>/<s>/<e> returns every quality with "
"signed URLs, exact byte sizes and codec in ~3s. But the bytes sit "
"behind a proxy (requiresProxy:true) that rejects curl AND ffmpeg "
"with 428 even given byte-exact browser headers — TLS fingerprint "
"or session binding, not headers. Unusable for ingest without "
"curl-impersonate. Media is HEVC; Pi decode unverified.",
),
Source(
id="hdhub", name="HdHub", container="progressive-mkv",
ingestable=True, browser=False, heights=(1080, 720, 480),
resolve=_hdhub_pick, identity="filename",
note="Added 2026-09-28 when Cloudflare banned vidcore's CDN host. A "
"Stremio addon (found in cinewave's server list) returning direct "
"mkv files on R2/pixeldrain: plain HTTP, ~10 MB/s, seekable. Asked "
"by IMDb id (tmdb: finds nothing). Identity is a FILENAME match, "
"so the library tries vidcore first and checks runtime.",
),
Source(
id="fmovies", name="fmovies", container="hls-fmp4",
ingestable=True, browser=True, enabled=False,
note="Breaker-open since 2026-07-28 (Cloudflare Turnstile).",
),
Source(
id="vsembed", name="CloudStream Pro", container="hls-fmp4",
ingestable=True, browser=False, enabled=False,
note="Breaker-open since 2026-07-28 (Cloudflare Turnstile).",
),
)}
SOURCES.update(SOURCES_DEFAULT)
if __name__ == "__main__":
import argparse
parser = argparse.ArgumentParser()
parser.add_argument("--host", default="0.0.0.0")
parser.add_argument("--port", type=int, default=7220)
args = parser.parse_args()
# Pick up anything a previous process left mid-flight before serving.
_resume_ingests()
_start_subtitle_retry()
app.run(host=args.host, port=args.port, debug=False, threaded=True)