"""Subtitles that belong to the library copy, or the live stream, being played.
vsembed serves a library copy at `/mp4/<kind>/<ident>.mp4` and measures its
sidecar subtitles against that copy's own audio, correcting or withdrawing
each one. Only Jellyfin reads the folder, so on 2026-10-18 Kodi was still
showing wyzie's English for No Other Choice -- the track the server had found
out of sync and drifting -- out of a per-TMDB cache on the Pi (md5-identical
to the unsynced original). The server lists a copy's sidecars at
`/subs/<kind>/<ident>.json` on the same host, and this fetches them.
A LIVE stream (vidcore, HdHub, or /mp4/ before a copy exists) has no copy to
measure. Asked with `?src=`, the server samples the stream's own audio and
places each English track against it (~30-60 s); until then it answers
"pending". The player starts on wyzie's raw tracks and swaps the synced one in
when watch_live() returns it.
Tracks are cached per list URL, and re-downloaded whenever the server's
identity (size, mtime) for one changes, so a re-sync on the server reaches the
Pi on the next play. Only tracks the server lists now are offered; a withdrawn
one may linger on disk but is never handed to the player.
"""
import hashlib
import json
import os
import re
from concurrent.futures import ThreadPoolExecutor
from typing import Any, Callable, Dict, List, Optional, Tuple
import requests
from lib.utils.kodi.utils import kodilog
# The ident is `tt<n>` or `tmdb:<n>`, plus `:<season>:<episode>` for a series
# -- exactly what vsembed's routes accept, so nothing else matches.
_IDENT = r"((?:tt\d+|tmdb(?::|%3A)\d+)(?:(?::|%3A)\d+(?::|%3A)\d+)?)"
_LIBRARY_MP4 = re.compile(
r"^(https?://[^/?#]+)/mp4/(movie|series)/" + _IDENT + r"\.mp4(?:[?#].*)?$",
re.IGNORECASE)
_VSEMBED_STREAM = re.compile(
r"^(https?://[^/?#]+)/(mp4|vidcore|hdhub)/(movie|series)/" + _IDENT
+ r"\.(?:mp4|m3u8|mkv)(?:\?([^#]*))?(?:#.*)?$", re.IGNORECASE)
_HEIGHT = re.compile(r"(?:^|&)h=(\d+)")
# The server is on the LAN; a short wait keeps a down server from delaying
# playback, and the caller falls back to wyzie.
LIST_TIMEOUT = 5
DOWNLOAD_TIMEOUT = 15
DOWNLOAD_WORKERS = 8
IDENTITY_FILE = ".identity.json"
# Live: the server measured ~30 s (vidcore) to ~60 s (HdHub), plus a resolve.
LIVE_INTERVAL = 10
LIVE_DEADLINE = 6 * 60
def library_list_url(stream_url: Optional[str]) -> Optional[str]:
"""`/subs/<kind>/<ident>.json` for a vsembed library copy, else None."""
m = _LIBRARY_MP4.match(stream_url or "")
if not m:
return None
return "{0}/subs/{1}/{2}.json".format(m.group(1), m.group(2), m.group(3))
def live_list_url(stream_url: Optional[str]) -> Optional[str]:
"""The live-sync list URL for any vsembed stream, else None."""
m = _VSEMBED_STREAM.match(stream_url or "")
if not m:
return None
h = _HEIGHT.search(m.group(5) or "")
return "{0}/subs/{1}/{2}.json?src={3}&h={4}".format(
m.group(1), m.group(3), m.group(4), m.group(2).lower(),
h.group(1) if h else "1080")
def watch_live(client: "LibrarySubtitleClient", stream_url: str,
still_playing: Callable[[], bool], wait: Callable[[float], bool],
deadline: float = LIVE_DEADLINE,
interval: float = LIVE_INTERVAL) -> Optional[List[str]]:
"""Ask until the server has placed the stream's subtitles. Paths, or None
if it has none, playback stopped, Kodi is closing or time ran out.
`wait(seconds)` returns False when Kodi is closing."""
for _ in range(int(deadline // interval) + 1):
status, paths = client.poll_live(stream_url)
if status == "ready":
return paths
if status != "pending" or not still_playing():
return None
if not wait(interval):
return None
return None
class LibrarySubtitleClient:
def __init__(self, root: str):
self.root = root
def _folder(self, list_url: str) -> str:
digest = hashlib.sha1(list_url.encode("utf-8")).hexdigest()[:16]
return os.path.join(self.root, digest)
def _listing(self, list_url: str) -> Tuple[str, Optional[List[Dict[str, Any]]]]:
"""(status, tracks). A held copy's answer carries no status: ready."""
try:
response = requests.get(list_url, timeout=LIST_TIMEOUT)
if response.status_code != 200:
kodilog("Library subtitles: {0} -> HTTP {1}".format(
list_url, response.status_code))
return "none", None
data = response.json()
except Exception as e:
kodilog("Library subtitles: {0} failed ({1})".format(list_url, e))
return "none", None
if not isinstance(data, dict):
return "none", None
subs = data.get("subtitles")
if not isinstance(subs, list):
return "none", None
tracks = [s for s in subs if isinstance(s, dict) and s.get("url") and s.get("id")]
return str(data.get("status") or "ready"), tracks
def _materialise(self, list_url: str, listing: List[Dict[str, Any]],
suffix: str = "") -> Optional[List[str]]:
"""Download what is not already current; paths of the listed tracks."""
folder = self._folder(list_url)
os.makedirs(folder, exist_ok=True)
identity_path = os.path.join(folder, IDENTITY_FILE)
try:
with open(identity_path) as f:
held = json.load(f)
except Exception:
held = {}
planned = []
for track in listing:
# The id names a file, and it arrived over the network.
name = os.path.basename(str(track["id"])).strip(". ")
if not name:
continue
planned.append((os.path.join(folder, name + suffix + ".srt"), name, track))
def _current(item):
path, name, track = item
return (os.path.exists(path) and os.path.getsize(path) > 0
and held.get(name) == track.get("identity"))
def _one(item):
path, name, track = item
try:
response = requests.get(track["url"], timeout=DOWNLOAD_TIMEOUT)
if response.status_code != 200 or not response.content:
kodilog("Library subtitles: {0} -> HTTP {1}".format(
name, response.status_code))
return None
tmp = path + ".part"
with open(tmp, "wb") as f:
f.write(response.content)
os.replace(tmp, path)
except Exception as e:
kodilog("Library subtitles: {0} failed ({1})".format(name, e))
return None
return item
todo = [item for item in planned if not _current(item)]
fresh = []
if todo:
with ThreadPoolExecutor(max_workers=DOWNLOAD_WORKERS) as pool:
fresh = [item for item in pool.map(_one, todo) if item]
for _path, name, track in fresh:
held[name] = track.get("identity")
try:
with open(identity_path, "w") as f:
json.dump(held, f)
except OSError as e:
kodilog("Library subtitles: could not record identities ({0})".format(e))
failed = {id(item) for item in todo} - {id(item) for item in fresh}
paths = sorted(item[0] for item in planned
if id(item) not in failed and os.path.exists(item[0]))
kodilog("Library subtitles: {0} track(s) for {1} ({2} downloaded)".format(
len(paths), list_url, len(fresh)))
return paths or None
def get_subtitles(self, stream_url: Optional[str]) -> Optional[List[str]]:
"""Paths of the copy's own subtitles, or None to fall through.
None for anything that is not a library copy, a server that cannot be
reached or predates the route, and a copy with no subtitles at all.
"""
list_url = library_list_url(stream_url)
if not list_url:
return None
_status, listing = self._listing(list_url)
if not listing:
return None
return self._materialise(list_url, listing)
def poll_live(self, stream_url: Optional[str]) -> Tuple[str, Optional[List[str]]]:
"""One ask: ("pending", None), ("ready", paths) or ("none", None)."""
list_url = live_list_url(stream_url)
if not list_url:
return "none", None
status, listing = self._listing(list_url)
if status == "pending":
return "pending", None
if status != "ready" or not listing:
return "none", None
# `.synced` in the name: Kodi shows it beside wyzie's raw English.
paths = self._materialise(list_url, listing, suffix=".synced")
return ("ready", paths) if paths else ("none", None)