diff options
Diffstat (limited to 'packages/meshbay-node')
27 files changed, 3580 insertions, 90 deletions
diff --git a/packages/meshbay-node/pyproject.toml b/packages/meshbay-node/pyproject.toml index 4999e3d..58f6f7f 100644 --- a/packages/meshbay-node/pyproject.toml +++ b/packages/meshbay-node/pyproject.toml @@ -18,6 +18,7 @@ dependencies = [ "websockets>=12.0", # hub→node revocation push "aiortc>=1.9", # WebRTC DataChannel for browser P2P (Phase 9) "aiosqlite>=0.20", # async SQLite for chat, audit, bundle stores + "guessit>=4.4", # filename parsing for the Videos app ] [project.optional-dependencies] diff --git a/packages/meshbay-node/src/meshbay_node/daemon.py b/packages/meshbay-node/src/meshbay_node/daemon.py index af85340..bf6bd64 100644 --- a/packages/meshbay-node/src/meshbay_node/daemon.py +++ b/packages/meshbay-node/src/meshbay_node/daemon.py @@ -26,7 +26,7 @@ Usage: """ import asyncio -from dataclasses import asdict +from dataclasses import asdict, replace import base64 import json import logging @@ -38,14 +38,17 @@ from pathlib import Path import uvicorn from meshbay_common import MNP_VERSION -from meshbay_common.protocol import MNP +from meshbay_common.protocol import MNP, index_entry_wire from meshbay_node.audit import AuditStore from meshbay_node.bundle_store import BundleStore from meshbay_node.chat.store import ChatStore from meshbay_node.config import Config, DEFAULT_CONFIG_PATH, load_config, write_example_config -from meshbay_node.roots import RootSet, RootError +from meshbay_node.roots import RootSet, RootError, entry_abs_path from meshbay_node.hub_client import HubClient, HubConfig from meshbay_node.indexer import DirectoryIndexer, IndexCache, GroupIndex +from meshbay_node.indexer.enrich import Enricher +from meshbay_node.media_cache import MediaCache +from meshbay_node.tmdb import TmdbClient from meshbay_node.keystore import create_keystore, load_keystore, load_or_create_keystore from meshbay_node.roster import Roster from meshbay_node.transport import ( @@ -62,6 +65,12 @@ if WEBRTC_AVAILABLE: log = logging.getLogger(__name__) +def _under_video_root(path: str, video_root: str) -> bool: + """Mirrors video-app.js's underVideoRoot: same folder, or a descendant.""" + path = path or "" + return path == video_root or path.startswith(video_root + "/") + + # ── Argon2id calibration ────────────────────────────────────────────────────── def calibrate_argon2(target_ms: int = 500) -> None: @@ -132,6 +141,15 @@ class NodeDaemon: self._last_broadcast_snapshot: dict[str, tuple] = {} self._audit_store: AuditStore | None = None self._bundle_store: BundleStore | None = None + self._media_cache: MediaCache | None = None + self._enricher: Enricher | None = None + self._tmdb_client: TmdbClient | None = None + # A file id attempted at most once per daemon run, success or + # failure — a persistently unprobeable file (corrupt, still being + # written) does not get re-queued on every coalesced broadcast. A + # restart retries everything, matching the "disposable, rebuildable" + # stance the rest of this cache takes (docs/mediacenter.md §1/§2). + self._enriched_attempted: set[str] = set() self._roster: Roster | None = None self._indexers: list[DirectoryIndexer] = [] self._tasks: list[asyncio.Task] = [] @@ -312,6 +330,10 @@ class NodeDaemon: # by the signed operation that changes it. "enabled_apps": await self._roster.enabled_apps( group_cfg.id) if self._roster else list(Roster.DEFAULT_APPS), + # Which folder is the Videos app's entry point for this + # group — "" means the whole group index. + "video_root": await self._roster.video_root( + group_cfg.id) if self._roster else "", } if not groups_ctx: @@ -333,6 +355,26 @@ class NodeDaemon: await self._audit_store.open() log.info("Audit store opened: %s", audit_db) + # 6b. Media cache (Videos app — TMDB metadata + thumbnails). + # Node-wide like audit.db, not per-group: a thumbnail is the same + # bytes regardless of which group happens to share the file + # (docs/mediacenter.md §2/§5.5). + media_cache_db = data_dir / "media_cache.db" + self._media_cache = MediaCache(db_path=media_cache_db) + await self._media_cache.open() + self._enricher = Enricher(self._media_cache) + self._tmdb_client = TmdbClient(roster=self._roster) + # Read once at load, like member_upload/enabled_apps — kept + # current in place by ops.set_tmdb_config (the signed op), + # exposed to every group's handshake ack via `daemon_state` + # (already wired to self._webrtc._ctx below) since this is + # node-wide, not per-group. + tmdb_enabled, tmdb_token, tmdb_language = await self._roster.tmdb_config() + self._state["tmdb_enabled"] = tmdb_enabled + self._state["tmdb_token_customized"] = bool(tmdb_token) + self._state["tmdb_language"] = tmdb_language or "" + log.info("Media cache opened: %s", media_cache_db) + # 5. Denylist denylist = self._denylist @@ -359,6 +401,8 @@ class NodeDaemon: self._webrtc._ctx["node_user_id"] = session.user_id self._webrtc._ctx["audit_store"] = self._audit_store self._webrtc._ctx["bundle_store"] = self._bundle_store + self._webrtc._ctx["media_cache"] = self._media_cache + self._webrtc._ctx["tmdb_client"] = self._tmdb_client self._webrtc._ctx["sk_x25519_raw"] = sk_x_raw self._webrtc._ctx["pk_x25519_raw"] = pk_x_raw self._webrtc._ctx["pk_x25519_b64"] = keys.pk_x25519_b64 @@ -468,6 +512,7 @@ class NodeDaemon: self._state["quic_server"] = self._quic_server self._state["hub"] = hub self._state["reload_fn"] = self._reload_config + self._state["enrich_video_root_fn"] = self._enrich_video_root_now # Rotating a key has to reach every transport holding a copy of it, # and clearing the denylist has to reach the one the handshake # consults — so both are published rather than reachable only @@ -494,6 +539,16 @@ class NodeDaemon: hashes = [e.id for e in gctx["index"].entries] if hashes: await self._register_swarm(hashes, endpoint) + # initial_scan() itself never calls on_change (it predates + # the concept — every existing caller only cared about the + # scan finishing, not about notifying anyone) — but Videos + # app enrichment (duration/thumb_hash/display_title/...) + # hangs entirely off that callback (_broadcast_index_change). + # Without this, every file already on disk at startup — the + # common case, an existing library — would never get + # enriched at all; only a file added later, while the node + # is already running, would trigger it via the watchdog. + await self._on_index_change(indexer) for idx, group_cfg in zip(self._indexers, self._config.groups): gctx = groups_ctx.get(group_cfg.id) @@ -663,6 +718,9 @@ class NodeDaemon: "enabled_apps": ( await self._roster.enabled_apps(group_cfg.id) if self._roster else list(Roster.DEFAULT_APPS)), + "video_root": ( + await self._roster.video_root(group_cfg.id) + if self._roster else ""), "chat_store": store, } groups_ctx[group_cfg.id] = new_ctx @@ -880,6 +938,7 @@ class NodeDaemon: prev = self._last_broadcast_snapshot.get(group_id) delta = None + previous = None if prev is not None: prev_version, prev_entries = prev previous = GroupIndex._snapshot( @@ -887,6 +946,34 @@ class NodeDaemon: delta = idx.diff(previous) self._last_broadcast_snapshot[group_id] = (idx.version, idx.entries_by_id()) + # Videos app (docs/mediacenter.md §5.2): schedule async technical + # probe + title parse + thumbnail generation for every newly-seen + # video entry under the group's configured video_root. Never blocks + # this broadcast — enrichment fields arrive later as their own + # INDEX_DELTA update (_on_enriched below). + new_entries = delta.additions if delta is not None else list(idx.entries) + asyncio.ensure_future(self._enrich_new_video_entries(indexer, new_entries)) + + # A rename/move changes the very filename (or season folder) that + # §3.3/§3.4's title-parse read display_title/season/episode from, + # but leaves the file's content — and so its id and everything + # ffprobe/thumbnailing already found — untouched. Only entries + # whose name or path actually differ from the last broadcast get a + # fresh pass; an update that is enrichment's own field-fill + # (duration/thumb_hash/... landing via _on_enriched below) leaves + # name/path alone and must not re-trigger itself forever. + if delta is not None and delta.updates and previous is not None: + asyncio.ensure_future( + self._reenrich_renamed_video_entries(indexer, delta.updates, previous)) + + # Videos app: a file that leaves the index also loses its thumbnail + # and file->tmdb mapping — the "real deletion obligation" docs/ + # mediacenter.md §2/§8 calls out explicitly rather than leaving + # implicit. tmdb_meta rows are left alone (§2: shared across files). + if delta is not None and delta.deletions and self._media_cache: + for file_id in delta.deletions: + asyncio.ensure_future(self._media_cache.prune_file(file_id)) + # 11.5 — Push to connected WebRTC peers in this group if self._webrtc: if delta is not None: @@ -896,12 +983,9 @@ class NodeDaemon: "group_id": idx.group_id, "base_version": delta.base_version, "version": delta.version, - "additions": [ - {"id": e.id, "name": e.name, "path": e.path, - "size": e.size, "type": e.type, "added_at": e.added_at} - for e in delta.additions - ], + "additions": [index_entry_wire(e) for e in delta.additions], "deletions": delta.deletions, + "updates": [index_entry_wire(e) for e in delta.updates], } else: msg = { @@ -909,11 +993,7 @@ class NodeDaemon: "v": MNP_VERSION, "group_id": idx.group_id, "version": idx.version, - "entries": [ - {"id": e.id, "name": e.name, "path": e.path, - "size": e.size, "type": e.type, "added_at": e.added_at} - for e in idx.entries - ], + "entries": [index_entry_wire(e) for e in idx.entries], } pushed = 0 for session in list(self._webrtc._sessions.values()): @@ -942,6 +1022,103 @@ class NodeDaemon: endpoint = f"webrtc:{self._config.node.quic_port}" asyncio.ensure_future(self._register_swarm(hashes, endpoint)) + async def _enrich_new_video_entries(self, indexer: DirectoryIndexer, entries: list) -> None: + """ + Videos app: fire (never await further) enrichment for unattempted + video entries under the group's configured video_root. + + A group with no video_root set yet does not enrich anything — TMDB + lookups and ffmpeg thumbnailing are real, ongoing per-file cost + (mediacenter.md §5.2/§10), and running them over an operator's whole + shared index before they have chosen which folder is actually their + media library would burn both TMDB's rate limit and the node's CPU + on files that were never meant to be in the Videos app at all. Once a + root is set, `_enrich_video_root_now` (called from ops.set_video_root) + separately sweeps whatever it already contains — this path alone only + ever sees entries new since the last broadcast. + """ + if not self._enricher or not self._roster: + return + video_root = await self._roster.video_root(indexer.group_id) + if not video_root: + return + for entry in entries: + if entry.type != "video" or entry.id in self._enriched_attempted: + continue + if not _under_video_root(entry.path, video_root): + continue + file_path = entry_abs_path(indexer.roots, entry) + if not file_path or not file_path.exists(): + continue + self._enriched_attempted.add(entry.id) + + async def on_done(file_id: str, fields: dict, _indexer=indexer) -> None: + await self._on_enriched(_indexer, file_id, fields) + + self._enricher.spawn(entry, file_path, on_done) + + async def _enrich_video_root_now(self, group_id: str) -> None: + """ + Videos app: sweep a group's existing index for enrichment right + after its video_root is set or changed (ops.set_video_root). + + The ordinary path above only ever looks at entries new since the + last broadcast, so a folder that already had files sitting in it + before it became the video_root would otherwise never get enriched + at all — nothing else re-visits already-indexed entries once they + have been broadcast once. + """ + indexer = self._state.get("indexers", {}).get(group_id) + if not indexer: + return + await self._enrich_new_video_entries(indexer, list(indexer.index.entries)) + + async def _reenrich_renamed_video_entries( + self, indexer: DirectoryIndexer, updates: list, previous: GroupIndex, + ) -> None: + """ + Videos app: found live — a French-named episode file, renamed by + the operator to match its English-named siblings, kept showing as + its own separate poster-grid card (and its own row in Flat list) + indefinitely, because `_enriched_attempted` — there specifically to + stop enrichment's own field-fill from re-triggering itself forever + (see the caller) — also silently blocked the *new* filename from + ever being title-parsed at all. `entry.id in self._enriched_attempted` + is the same content, so simply discarding it here and re-running + the ordinary enrichment path is enough: a fresh ffprobe/thumbnail + for an unchanged file is redundant work, not a correctness issue, + and renames are rare enough that the redundancy is not worth a + separate "title-parse only" code path. + """ + for entry in updates: + if entry.type != "video": + continue + old = previous.get_entry(entry.id) + if old is None or (old.name == entry.name and old.path == entry.path): + continue + self._enriched_attempted.discard(entry.id) + await self._enrich_new_video_entries(indexer, updates) + + async def _on_enriched(self, indexer: DirectoryIndexer, file_id: str, fields: dict) -> None: + """ + Merge enrichment fields into the live index and re-trigger a + broadcast so they reach connected clients as an INDEX_DELTA update + (GroupIndex.diff's `updates`, not `additions` — same id, new fields). + + Builds a *new* IndexEntry via dataclasses.replace rather than + mutating the existing one in place: the diff mechanism compares + against a shallow snapshot of entry *references*, so an in-place + mutation would silently also change what "previous" looks like, + and the change would never show up as a diff (see group_index.py's + diff() docstring). + """ + idx = indexer.index + entry = idx.get_entry(file_id) + if entry is None: + return # removed from the index while enrichment was in flight + idx.add_entry(replace(entry, **fields)) + await self._on_index_change(indexer) + def _drop_group_sessions(self, group_id: str) -> None: """Close live sessions for a revoked group (H4).""" if not self._webrtc or not group_id: @@ -983,6 +1160,12 @@ class NodeDaemon: if self._bundle_store: await self._bundle_store.close() + if self._tmdb_client: + await self._tmdb_client.close() + + if self._media_cache: + await self._media_cache.close() + if self._roster: await self._roster.close() diff --git a/packages/meshbay-node/src/meshbay_node/indexer/enrich.py b/packages/meshbay-node/src/meshbay_node/indexer/enrich.py new file mode 100644 index 0000000..784b2a3 --- /dev/null +++ b/packages/meshbay-node/src/meshbay_node/indexer/enrich.py @@ -0,0 +1,166 @@ +""" +Index-time enrichment for the Videos group app: technical probe (ffprobe), +filename parsing (title_parse), and thumbnail generation (ffmpeg) for a +newly-added video IndexEntry. + +Runs through its own small bounded worker pool — separate from the streaming +transcode pool (docs/mediacenter.md §5.2, mirroring webrtc_server.py's +`_transcode_semaphore`) — so indexing a large library never blocks on this, +and enrichment never competes with an active viewer for CPU. The scan itself +already put the entry in the index with hash/size/type only; this fills in +the rest asynchronously and hands the result back via a callback. +""" + +import asyncio +import logging +from pathlib import Path +from typing import Awaitable, Callable + +import blake3 + +from meshbay_common.protocol import IndexEntry +from meshbay_node.indexer import title_parse +from meshbay_node.indexer.indexer import MEDIA_EXTENSIONS +from meshbay_node.media_cache import MediaCache +from meshbay_node.media_probe import probe_video + +log = logging.getLogger(__name__) + +DEFAULT_MAX_CONCURRENT = 2 +PROBE_TIMEOUT_SECS = 30 +THUMB_TIMEOUT_SECS = 30 +THUMB_WIDTH = 320 +# "Show/SeasonFolder/episode.mkv" is the expected shape, with a little slack +# for an extra wrapper folder — not an attempt to find the exact group root. +MAX_ANCESTOR_DEPTH = 4 +# Bounds the "borrow a title from a sibling episode filename" scan (§3.4) so +# a folder with thousands of files costs a fixed, small amount of work. +MAX_SIBLINGS_CHECKED = 20 + + +def _season_from_ancestors(file_path: Path) -> int | None: + folder = file_path.parent + for _ in range(MAX_ANCESTOR_DEPTH): + if folder is None or folder == folder.parent: + break + season = title_parse.season_from_folder_name(folder.name) + if season is not None: + return season + folder = folder.parent + return None + + +def _title_from_siblings(file_path: Path) -> str | None: + """ + §3.4: an episode filename with no show name in it borrows the title from + a representative sibling in the same folder, never from the folder name + alone (an acronym-named show folder is a real, observed case). + """ + try: + names = sorted(p.name for p in file_path.parent.iterdir() if p.is_file()) + except OSError: + return None + checked = 0 + for name in names: + if name == file_path.name: + continue + if Path(name).suffix.lower() not in MEDIA_EXTENSIONS["video"]: + continue + checked += 1 + if checked > MAX_SIBLINGS_CHECKED: + break + parsed = title_parse.parse_episode_filename(name) + if parsed.display_title: + return parsed.display_title + return None + + +async def _make_thumbnail(file_path: Path, duration: float | None) -> bytes | None: + """One ffmpeg frame grab at ~10% of duration (or 5s if unknown), scaled down.""" + seek = max(0.0, (duration or 50.0) * 0.1) + proc = await asyncio.create_subprocess_exec( + "ffmpeg", "-v", "error", "-ss", str(seek), "-i", str(file_path), + "-frames:v", "1", "-vf", f"scale={THUMB_WIDTH}:-1", + "-f", "image2", "-c:v", "mjpeg", "pipe:1", + stdout=asyncio.subprocess.PIPE, stderr=asyncio.subprocess.PIPE, + ) + try: + stdout, _ = await asyncio.wait_for(proc.communicate(), THUMB_TIMEOUT_SECS) + except asyncio.TimeoutError: + proc.kill() + await proc.wait() + return None + return stdout or None + + +class Enricher: + """Owns the node's bounded index-time enrichment pool.""" + + def __init__(self, media_cache: MediaCache, max_concurrent: int = DEFAULT_MAX_CONCURRENT): + self._media_cache = media_cache + self._sem = asyncio.Semaphore(max_concurrent) + self._tasks: set[asyncio.Task] = set() + + def spawn( + self, entry: IndexEntry, file_path: Path, + on_done: Callable[[str, dict], Awaitable[None]], + ) -> asyncio.Task: + """ + Fire-and-forget one file's enrichment. `on_done(file_id, fields)` is + awaited with the index fields to merge in once ready — never blocks + the caller (a scan or watchdog event). The reference this method + returns is what keeps the task alive; callers should hold it the + same way `WebRTCPeerSession._spawn` holds streaming tasks. + """ + task = asyncio.ensure_future(self._run(entry, file_path, on_done)) + self._tasks.add(task) + + def _cleanup(t: asyncio.Task) -> None: + self._tasks.discard(t) + if not t.cancelled() and t.exception(): + log.error("Enrichment failed for %s: %s", entry.id[:12], t.exception(), + exc_info=t.exception()) + task.add_done_callback(_cleanup) + return task + + async def _run( + self, entry: IndexEntry, file_path: Path, + on_done: Callable[[str, dict], Awaitable[None]], + ) -> None: + async with self._sem: + fields: dict = {} + duration: float | None = None + try: + _codec, duration, _has_audio, width, height = await asyncio.wait_for( + probe_video(str(file_path)), timeout=PROBE_TIMEOUT_SECS) + fields["duration"] = int(duration) if duration else None + fields["width"] = width + fields["height"] = height + except Exception as e: + log.warning("Probe failed for %s: %s", file_path, e) + + ep = title_parse.parse_episode_filename(entry.name) + if ep.episode is not None: + title = ep.display_title or await asyncio.to_thread( + _title_from_siblings, file_path) + season = ep.season + if season is None: + season = await asyncio.to_thread(_season_from_ancestors, file_path) + fields["display_title"] = title or title_parse.naive_title(entry.name) + fields["season"] = season + fields["episode"] = ep.episode + else: + mv = title_parse.parse_movie_filename(entry.name) + fields["display_title"] = mv.display_title or mv.naive_title + + try: + thumb = await _make_thumbnail(file_path, duration) + except Exception as e: + log.warning("Thumbnail generation failed for %s: %s", file_path, e) + thumb = None + if thumb: + thumb_hash = blake3.blake3(thumb).hexdigest() + await self._media_cache.put_thumb(thumb_hash, entry.id, thumb) + fields["thumb_hash"] = thumb_hash + + await on_done(entry.id, fields) diff --git a/packages/meshbay-node/src/meshbay_node/indexer/group_index.py b/packages/meshbay-node/src/meshbay_node/indexer/group_index.py index ec98667..1ce4e0a 100644 --- a/packages/meshbay-node/src/meshbay_node/indexer/group_index.py +++ b/packages/meshbay-node/src/meshbay_node/indexer/group_index.py @@ -78,6 +78,19 @@ class GroupIndex: def get_entry(self, file_id: str) -> IndexEntry | None: return self._entries.get(file_id) + def get_entry_by_path(self, path: str) -> IndexEntry | None: + """ + Linear scan — entries are keyed by content id, not path, and nothing + before the Videos app needed to go the other way (a client always + already has the id from index_sync/index_delta). Fine for an + on-demand, per-tile lookup against a few thousand entries; revisit + if a future caller makes this hot. + """ + for entry in self._entries.values(): + if entry.path == path: + return entry + return None + @property def entries(self) -> list[IndexEntry]: return list(self._entries.values()) @@ -205,16 +218,34 @@ class GroupIndex: # ── Delta ───────────────────────────────────────────────────────────────── def diff(self, previous: "GroupIndex") -> IndexDelta: - """Compute what changed since a previous version of this index.""" + """ + Compute what changed since a previous version of this index. + + A shared id whose entry object now compares unequal (field-by-field, + via IndexEntry's dataclass-generated __eq__) is an update, not an + addition — the Videos app's async enrichment (duration, thumb_hash, + title, ...) replaces an existing entry's fields after the fact via + `add_entry`, which never introduces a new id. This only works + because that replacement always constructs a *new* IndexEntry object + (`dataclasses.replace`, never in-place attribute mutation) — mutating + the same object in place would also mutate `previous`'s copy, since + entries_by_id() is a shallow dict copy, and the two would always + compare equal. + """ prev_ids = set(previous._entries) curr_ids = set(self._entries) additions = [self._entries[i] for i in curr_ids - prev_ids] deletions = list(prev_ids - curr_ids) + updates = [ + self._entries[i] for i in curr_ids & prev_ids + if self._entries[i] != previous._entries[i] + ] return IndexDelta( base_version=previous.version, version=self.version, additions=additions, deletions=deletions, + updates=updates, ) diff --git a/packages/meshbay-node/src/meshbay_node/indexer/title_parse.py b/packages/meshbay-node/src/meshbay_node/indexer/title_parse.py new file mode 100644 index 0000000..ef522ad --- /dev/null +++ b/packages/meshbay-node/src/meshbay_node/indexer/title_parse.py @@ -0,0 +1,174 @@ +""" +Filename -> title/year/season/episode parsing for the Videos group app. + +Wraps `guessit` and layers the fixes from docs/mediacenter.md §3.3/§3.4 on +top of it: none of them are per-title hacks, each is a generic rule found +by validating guessit's raw output against real TMDB search results over a +~1950-file library (movies, TV shows, and a small franchise set). + +Scope is deliberately narrow (§3.5): title, year, season, episode. Technical +facts (resolution, codec, duration) come from ffprobe, never the filename — +a mislabeled `1080p` tag is a real, observed failure mode. + +This module never touches the filesystem or the network. The orchestration +that decides *which* file supplies a show's title (a representative episode +filename, not the folder name — §3.4) lives in the indexer, which has the +directory listing; this module only parses strings it's handed. +""" + +from __future__ import annotations + +import re +from dataclasses import dataclass, field + +from guessit import guessit + +# French edition/release vocabulary guessit's (English-centric) edition list +# doesn't recognize — left stuck to the title instead of stripped as a tag. +_EDITION_PHRASES = ( + r"version\s+longue", + r"version\s+int[ée]grale", + r"remasteris[ée]e?", + r"non[\s._-]*censur[ée]e?", + r"original\s+version", +) +_EDITION_RE = re.compile("|".join(_EDITION_PHRASES), re.IGNORECASE) + +# A season-like ancestor folder: the English/French words plus a number or +# Roman numeral. Vocabulary is a plain tuple so a deployment can extend it +# per locale without touching the regex-building logic. +SEASON_WORDS = ("season", "saison") +_SEASON_RE = re.compile( + r"(?:" + "|".join(SEASON_WORDS) + r")\s*([0-9]+|[ivxlc]+)\b", + re.IGNORECASE, +) +_SPECIALS_RE = re.compile(r"\b(?:bonus|extras?|specials?)\b", re.IGNORECASE) + +_ROMAN_NUMERALS = { + 2: "II", 3: "III", 4: "IV", 5: "V", 6: "VI", + 7: "VII", 8: "VIII", 9: "IX", 10: "X", +} + + +def _roman_to_int(s: str) -> int | None: + values = {"i": 1, "v": 5, "x": 10, "l": 50, "c": 100} + s = s.lower() + if not s or any(c not in values for c in s): + return None + total = 0 + prev = 0 + for c in reversed(s): + v = values[c] + total += v if v >= prev else -v + prev = v + return total or None + + +def _strip_editions(title: str) -> str: + return re.sub(r"\s+", " ", _EDITION_RE.sub(" ", title)).strip() + + +def naive_title(filename: str) -> str: + """ + The mandated fallback (§3.6, §4.1): strip the extension, replace every + `.`/`_`/`-` with a space, drop a trailing parenthesized year, collapse + whitespace. Always computable, never fails, used both as the flat-mode + display name of last resort and as a second TMDB query candidate. + """ + stem = filename.rsplit(".", 1)[0] if "." in filename else filename + stem = re.sub(r"[._-]+", " ", stem) + stem = re.sub(r"\(\s*(19|20)\d{2}\s*\)", " ", stem) + stem = _strip_editions(stem) + return re.sub(r"\s+", " ", stem).strip() + + +def sequel_variants(title: str) -> list[str]: + """ + A trailing sequel digit sometimes has no equivalent in the real TMDB + title, or the real title uses a Roman numeral instead (§3.3 row 4). + Returns extra candidates to try — empty if `title` has no trailing digit. + """ + m = re.match(r"^(.*\S)\s+([2-9])$", title) + if not m: + return [] + base, digit = m.group(1), int(m.group(2)) + variants = [base] + roman = _ROMAN_NUMERALS.get(digit) + if roman: + variants.append(f"{base} {roman}") + return variants + + +def season_from_folder_name(name: str) -> int | None: + """ + §3.4: a season-like ancestor folder, vocabulary-driven rather than + assuming a numeric convention everywhere. A specials/bonus/extras + folder maps to season 0 (matching TMDB's own `season_number: 0`). + Returns None if `name` doesn't look like a season folder at all. + """ + if _SPECIALS_RE.search(name): + return 0 + m = _SEASON_RE.search(name) + if not m: + return None + token = m.group(1) + if token.isdigit(): + return int(token) + return _roman_to_int(token) + + +@dataclass +class ParsedName: + display_title: str | None # None => caller must supply from elsewhere (e.g. a sibling file) + alt_title: str | None = None # guessit's alternative_title, a second query candidate (§3.3 row 1) + naive_title: str = "" # always available, fully punctuation-normalized fallback + year: int | None = None + season: int | None = None + episode: int | None = None + confidence: bool = False # True only when display_title is set and structurally corroborated + + +def parse_movie_filename(filename: str) -> ParsedName: + """Parse a standalone movie filename.""" + g = guessit(filename) + title = g.get("title") + title = str(title).strip() if title else None + if title: + title = _strip_editions(title) + alt = g.get("alternative_title") + alt = _strip_editions(str(alt).strip()) if alt else None + year = g.get("year") + nt = naive_title(filename) + confidence = bool(title) and len(title) >= 2 and year is not None + return ParsedName( + display_title=title or None, alt_title=alt, naive_title=nt, + year=year, confidence=confidence, + ) + + +def parse_episode_filename(filename: str) -> ParsedName: + """ + Parse an episode filename. `display_title` may come back None (e.g. + `S08E02.SUBFRENCH.720p.mkv` carries no show name at all, §3.2) — the + indexer then supplies the show title from a representative sibling + filename in the same folder rather than the folder name itself (§3.4). + """ + g = guessit(filename) + title = g.get("title") + title = str(title).strip() if title else None + if title: + title = _strip_editions(title) + season = g.get("season") + episode = g.get("episode") + # guessit returns a list when it finds more than one candidate (e.g. a + # multi-episode file); take the first as the representative one. + if isinstance(season, list): + season = season[0] if season else None + if isinstance(episode, list): + episode = episode[0] if episode else None + nt = naive_title(filename) + confidence = bool(title) and len(title) >= 2 and season is not None and episode is not None + return ParsedName( + display_title=title or None, naive_title=nt, + season=season, episode=episode, confidence=confidence, + ) diff --git a/packages/meshbay-node/src/meshbay_node/media_cache.py b/packages/meshbay-node/src/meshbay_node/media_cache.py new file mode 100644 index 0000000..6daae3f --- /dev/null +++ b/packages/meshbay-node/src/meshbay_node/media_cache.py @@ -0,0 +1,188 @@ +""" +MeshBay Node — TMDB metadata and thumbnail cache for the Videos group app. + +Node-wide (not per-group, `data_dir/media_cache.db`), same rationale as +`tmdb_enabled`/`tmdb_api_token` living in `group_settings` under the +`group_id=""` sentinel (docs/mediacenter.md §5.5): TMDB is one operator's +budget and credential, and a thumbnail is the same bytes regardless of which +group happens to share the file. + +Disposable and rebuildable, like the rest of the file index (§1, §2) — never +a second identity for a file. Every row here is keyed off a value the node +can already derive (a file's own blake3 id, or a TMDB id), so losing this +database costs re-probing/re-fetching, not data. +""" + +import json +import logging +import time +from pathlib import Path + +import aiosqlite + +log = logging.getLogger(__name__) + +_SCHEMA = """ +CREATE TABLE IF NOT EXISTS file_tmdb ( + file_id TEXT PRIMARY KEY, + tmdb_id TEXT NOT NULL, + media_type TEXT NOT NULL +); +CREATE TABLE IF NOT EXISTS tmdb_meta ( + tmdb_id TEXT NOT NULL, + media_type TEXT NOT NULL, + json TEXT NOT NULL, + fetched_at REAL NOT NULL, + PRIMARY KEY (tmdb_id, media_type) +); +CREATE TABLE IF NOT EXISTS thumbs ( + thumb_hash TEXT PRIMARY KEY, + file_id TEXT NOT NULL, + jpeg BLOB NOT NULL +); +CREATE INDEX IF NOT EXISTS idx_thumbs_file ON thumbs(file_id); +CREATE TABLE IF NOT EXISTS season_meta ( + tmdb_id TEXT NOT NULL, + season INTEGER NOT NULL, + json TEXT NOT NULL, + fetched_at REAL NOT NULL, + PRIMARY KEY (tmdb_id, season) +); +""" + +# TMDB overviews/ratings do drift; a file's own resolved tmdb_id does not +# need re-checking on this schedule, only the metadata blob (§5.4, V3). +TMDB_META_TTL_SECS = 30 * 86400 + + +class MediaCache: + """Async SQLite cache for TMDB lookups and generated thumbnails.""" + + def __init__(self, db_path: Path): + self._db_path = db_path + self._db: aiosqlite.Connection | None = None + + async def open(self) -> None: + self._db_path.parent.mkdir(parents=True, exist_ok=True) + self._db = await aiosqlite.connect(str(self._db_path)) + await self._db.executescript(_SCHEMA) + await self._db.commit() + + async def close(self) -> None: + if self._db: + await self._db.close() + self._db = None + + # ── file -> tmdb id mapping ────────────────────────────────────────────── + + async def get_file_tmdb(self, file_id: str) -> tuple[str, str] | None: + """Returns (tmdb_id, media_type), or None if this file was never resolved.""" + async with self._db.execute( + "SELECT tmdb_id, media_type FROM file_tmdb WHERE file_id = ?", + (file_id,), + ) as cur: + row = await cur.fetchone() + return (row[0], row[1]) if row else None + + async def set_file_tmdb(self, file_id: str, tmdb_id: str, media_type: str) -> None: + await self._db.execute( + "INSERT OR REPLACE INTO file_tmdb (file_id, tmdb_id, media_type) " + "VALUES (?, ?, ?)", + (file_id, tmdb_id, media_type), + ) + await self._db.commit() + + # ── tmdb id -> metadata json ───────────────────────────────────────────── + + async def get_tmdb_meta(self, tmdb_id: str, media_type: str) -> dict | None: + """Returns None on a miss or on an entry older than TMDB_META_TTL_SECS.""" + async with self._db.execute( + "SELECT json, fetched_at FROM tmdb_meta WHERE tmdb_id = ? AND media_type = ?", + (tmdb_id, media_type), + ) as cur: + row = await cur.fetchone() + if not row: + return None + raw_json, fetched_at = row + if time.time() - fetched_at > TMDB_META_TTL_SECS: + return None + return json.loads(raw_json) + + async def set_tmdb_meta(self, tmdb_id: str, media_type: str, meta: dict) -> None: + await self._db.execute( + "INSERT OR REPLACE INTO tmdb_meta (tmdb_id, media_type, json, fetched_at) " + "VALUES (?, ?, ?, ?)", + (tmdb_id, media_type, json.dumps(meta), time.time()), + ) + await self._db.commit() + + # ── tmdb id + season number -> season-level metadata json ──────────────── + # + # A show's own overview (tmdb_meta above) is one static field an operator + # found does not necessarily describe every season alike (mediacenter.md + # §5.4) — this is TMDB's per-season `overview`/`air_date`/`poster_path`, + # fetched and cached independently, on the same staleness schedule. + + async def get_season_meta(self, tmdb_id: str, season: int) -> dict | None: + async with self._db.execute( + "SELECT json, fetched_at FROM season_meta WHERE tmdb_id = ? AND season = ?", + (tmdb_id, season), + ) as cur: + row = await cur.fetchone() + if not row: + return None + raw_json, fetched_at = row + if time.time() - fetched_at > TMDB_META_TTL_SECS: + return None + return json.loads(raw_json) + + async def set_season_meta(self, tmdb_id: str, season: int, meta: dict) -> None: + await self._db.execute( + "INSERT OR REPLACE INTO season_meta (tmdb_id, season, json, fetched_at) " + "VALUES (?, ?, ?, ?)", + (tmdb_id, season, json.dumps(meta), time.time()), + ) + await self._db.commit() + + # ── thumbnails ──────────────────────────────────────────────────────────── + + async def get_thumb(self, thumb_hash: str) -> bytes | None: + async with self._db.execute( + "SELECT jpeg FROM thumbs WHERE thumb_hash = ?", (thumb_hash,), + ) as cur: + row = await cur.fetchone() + return bytes(row[0]) if row else None + + async def get_thumb_hash_by_file_id(self, file_id: str) -> str | None: + """ + A TMDB poster/backdrop is stored under a synthetic file_id + (`tmdb:{poster_path}`, stable across requests for the same image) — + this is how `_fetch_and_cache_poster` recognizes "already fetched" + without knowing the content hash up front (that's only known once + the bytes are downloaded). + """ + async with self._db.execute( + "SELECT thumb_hash FROM thumbs WHERE file_id = ?", (file_id,), + ) as cur: + row = await cur.fetchone() + return row[0] if row else None + + async def put_thumb(self, thumb_hash: str, file_id: str, jpeg: bytes) -> None: + await self._db.execute( + "INSERT OR REPLACE INTO thumbs (thumb_hash, file_id, jpeg) VALUES (?, ?, ?)", + (thumb_hash, file_id, jpeg), + ) + await self._db.commit() + + # ── pruning ─────────────────────────────────────────────────────────────── + + async def prune_file(self, file_id: str) -> None: + """ + Called when a file leaves the index (deletion, unshared root). Removes + its thumbnail and its file->tmdb mapping. `tmdb_meta` rows are left + alone — they're keyed by tmdb_id, not file_id, and other files (other + episodes of the same show) may still reference the same entry. + """ + await self._db.execute("DELETE FROM thumbs WHERE file_id = ?", (file_id,)) + await self._db.execute("DELETE FROM file_tmdb WHERE file_id = ?", (file_id,)) + await self._db.commit() diff --git a/packages/meshbay-node/src/meshbay_node/media_probe.py b/packages/meshbay-node/src/meshbay_node/media_probe.py new file mode 100644 index 0000000..8ad883d --- /dev/null +++ b/packages/meshbay-node/src/meshbay_node/media_probe.py @@ -0,0 +1,70 @@ +""" +ffprobe wrapper shared by stream-time codec detection (transport/webrtc_server.py) +and index-time technical-field enrichment (indexer/enrich.py). + +Split out of webrtc_server.py so the indexer package (which webrtc_server.py +already imports from) can call it too without a circular import. +""" + +import asyncio +import json + +_H264_PROFILES = {"Baseline": "42", "Main": "4d", "High": "64", "High 10": "6e"} + + +async def probe_video(path: str) -> tuple[str | None, float, bool, int | None, int | None]: + """ + Probe video file with ffprobe, return (MSE codec string, duration, + has_audio, width, height). + + The audio half of the codec string is always "mp4a.40.2" (AAC-LC) or + absent — never the source's real audio codec — because the streaming + path always transcodes audio to AAC and never copies it: MSE in every + mainstream browser only decodes AAC/Opus, and a source codec outside + that (AC-3, E-AC-3, DTS, ...) is at best silently unplayable and at + worst, for E-AC-3 at least, makes ffmpeg itself refuse to write the + fragmented MP4 header ("Cannot write moov atom before EAC3 packets + parsed" — reproduced against a real 5.1 E-AC-3 WEB-DL). Video stays + whatever it actually is: it is always copied, never transcoded. + + width/height come from the same ffprobe call (one extra `-show_entries` + field, no second process spawn) — resolution is deliberately never + guessed from the filename (docs/mediacenter.md §3.5). + """ + proc = await asyncio.create_subprocess_exec( + "ffprobe", "-v", "error", + "-show_entries", "stream=codec_name,profile,level,codec_type,width,height", + "-show_entries", "format=duration", + "-of", "json", path, + stdout=asyncio.subprocess.PIPE, stderr=asyncio.subprocess.PIPE, + ) + stdout, _ = await proc.communicate() + info = json.loads(stdout) + duration = float(info.get("format", {}).get("duration", 0)) + + v_codec = "" + has_audio = False + width: int | None = None + height: int | None = None + for s in info.get("streams", []): + if s.get("codec_type") == "video" and not v_codec: + cn = s.get("codec_name", "") + if cn == "h264": + p = _H264_PROFILES.get(s.get("profile", "High"), "64") + lvl = int(s.get("level", 40)) + v_codec = f"avc1.{p}00{lvl:02x}" + elif cn == "hevc": + v_codec = "hev1.1.6.L93.B0" + elif cn == "vp9": + v_codec = "vp09.00.10.08" + elif cn == "av1": + v_codec = "av01.0.01M.08" + width = s.get("width") + height = s.get("height") + elif s.get("codec_type") == "audio": + has_audio = True + + if not v_codec: + return None, duration, has_audio, width, height + codec = f"{v_codec},mp4a.40.2" if has_audio else v_codec + return codec, duration, has_audio, width, height diff --git a/packages/meshbay-node/src/meshbay_node/ops.py b/packages/meshbay-node/src/meshbay_node/ops.py index 154813d..6848e3a 100644 --- a/packages/meshbay-node/src/meshbay_node/ops.py +++ b/packages/meshbay-node/src/meshbay_node/ops.py @@ -731,6 +731,68 @@ async def set_enabled_apps(state: dict, group_id: str, apps: list[str]) -> dict: return {"apps": apps, "group_id": group_id} +# ── TMDB config (Videos app) ───────────────────────────────────────────────── + +async def set_tmdb_config(state: dict, enabled: bool, token: str | None = None, + language: str | None = None) -> dict: + """ + Whether the node calls TMDB at all, whether it uses a custom API token + instead of the shipped default, and in what language it queries TMDB + (docs/mediacenter.md §5.5). + + Node-wide (roster.py group_settings, group_id="") rather than per-group + like set_member_upload/set_enabled_apps: TMDB is one operator's budget, + one credential and one shared cache, not a per-group or per-viewer + concern. `token=""` explicitly clears a previously-set custom token + (reverts to the shipped default); `token=None` leaves whatever was + there unchanged. Same discipline for `language`. + """ + roster = _roster(state) + await roster.set_tmdb_config(enabled, token, language, set_by=state.get("node_user_id", "")) + state["tmdb_enabled"] = enabled + # `token=None` means "leave whatever was there" (§ set_tmdb_config's own + # docstring) — so the customized flag only changes when a value (a real + # token, or "" to clear one) was actually given. + if token is not None: + state["tmdb_token_customized"] = bool(token) + if language is not None: + state["tmdb_language"] = language + log.info("TMDB config: enabled=%s custom_token=%s language=%s", + enabled, bool(token), language or state.get("tmdb_language", "")) + return { + "enabled": enabled, + "token_customized": state.get("tmdb_token_customized", False), + "language": state.get("tmdb_language", ""), + } + + +async def set_video_root(state: dict, group_id: str, path: str) -> dict: + """ + Which folder (possibly a subfolder of a shared root) is the Videos app's + entry point for this group. Same shape as set_enabled_apps: lives on the + node (roster.db), takes effect without a restart, signed by the operator. + `path=""` clears it — the Videos tab then asks for one to be chosen + before anything (including TMDB enrichment, docs/mediacenter.md §5.2) + runs, rather than defaulting to the whole shared index. + + A non-empty path fires (never awaits) a sweep of whatever that folder + already contains: the ordinary per-change enrichment path only ever + looks at files new since the last broadcast, so anything already sitting + in a folder before it became the video_root would otherwise never be + picked up. + """ + roster = _roster(state) + ctx = _group_ctx(state, group_id) + await roster.set_video_root(group_id, path, set_by=state.get("node_user_id", "")) + ctx["video_root"] = path + log.info("Videos root for group %s: %r", group_id[:8], path) + if path: + enrich_fn = state.get("enrich_video_root_fn") + if enrich_fn: + asyncio.ensure_future(enrich_fn(group_id)) + return {"path": path, "group_id": group_id} + + # ── Scan settings ──────────────────────────────────────────────────────────── async def set_scan_settings(state: dict, group_id: str, reconcile_interval_secs: float, diff --git a/packages/meshbay-node/src/meshbay_node/roster.py b/packages/meshbay-node/src/meshbay_node/roster.py index 6eb7651..701462f 100644 --- a/packages/meshbay-node/src/meshbay_node/roster.py +++ b/packages/meshbay-node/src/meshbay_node/roster.py @@ -599,6 +599,52 @@ class Roster: json.dumps(sorted(apps)), set_by) return apps + # TMDB is one operator's budget and credential, not a per-group concern + # (docs/mediacenter.md §5.5) — stored under the group_id="" sentinel, + # the same precedent as `roster.get_member("", user_id)` authorizing the + # operator node-wide (desktop-client-v1.md §6.3). Unset means "on, using + # the shipped default token" — the same "absent means the old behaviour" + # discipline member_upload/enabled_apps already follow. + SETTING_TMDB_ENABLED = "tmdb_enabled" + SETTING_TMDB_TOKEN = "tmdb_api_token" + # A TMDB language tag (e.g. "fr-FR") — one for the whole node, same + # reasoning as the token: one shared cache, not a per-viewer request. + # Unset means TMDB's own default (English) rather than this node + # guessing one. + SETTING_TMDB_LANGUAGE = "tmdb_language" + NODE_WIDE_GROUP_ID = "" + + async def tmdb_config(self) -> tuple[bool, str | None, str | None]: + """Returns (enabled, custom_token_or_None, language_or_None).""" + enabled = (await self.get_setting( + self.NODE_WIDE_GROUP_ID, self.SETTING_TMDB_ENABLED, "1")) != "0" + token = await self.get_setting(self.NODE_WIDE_GROUP_ID, self.SETTING_TMDB_TOKEN) + language = await self.get_setting(self.NODE_WIDE_GROUP_ID, self.SETTING_TMDB_LANGUAGE) + return enabled, (token or None), (language or None) + + async def set_tmdb_config(self, enabled: bool, token: str | None = None, + language: str | None = None, set_by: str = "") -> None: + await self.set_setting(self.NODE_WIDE_GROUP_ID, self.SETTING_TMDB_ENABLED, + "1" if enabled else "0", set_by) + if token is not None: + await self.set_setting(self.NODE_WIDE_GROUP_ID, self.SETTING_TMDB_TOKEN, + token, set_by) + if language is not None: + await self.set_setting(self.NODE_WIDE_GROUP_ID, self.SETTING_TMDB_LANGUAGE, + language, set_by) + + # Which folder is the Videos app's entry point for this group — per-group + # (unlike tmdb_config above), since different groups share different + # trees. Empty/unset means the whole group index, exactly as today. + SETTING_VIDEO_ROOT = "video_root" + + async def video_root(self, group_id: str) -> str: + return await self.get_setting(group_id, self.SETTING_VIDEO_ROOT, "") or "" + + async def set_video_root(self, group_id: str, path: str, set_by: str = "") -> str: + await self.set_setting(group_id, self.SETTING_VIDEO_ROOT, path or "", set_by) + return path or "" + # How often the indexer's reconciliation backstop runs, and how long it # waits after the last change on a file before hashing it. Unset means # the indexer's own defaults — an existing group's behaviour must not diff --git a/packages/meshbay-node/src/meshbay_node/tmdb.py b/packages/meshbay-node/src/meshbay_node/tmdb.py new file mode 100644 index 0000000..448a2b9 --- /dev/null +++ b/packages/meshbay-node/src/meshbay_node/tmdb.py @@ -0,0 +1,177 @@ +""" +TMDB (themoviedb.org) client for the Videos group app. + +Called only by the node, never by a client (docs/mediacenter.md §2): the +node holds the one credential and makes the one request per unique title, +shared by every member. Token resolution order (§5.5): + + 1. an operator-supplied token (roster.py group_settings, group_id="") + 2. the MESHBAY_TMDB_DEFAULT_TOKEN environment variable + 3. none — TMDB lookups are inert (callers get an empty result, never an + exception, so a node with no token configured just serves thumbnails) + +The real secret (whichever token resolves) never appears in source control: +there is no literal fallback value in this file. See mediacenter.md's +implementation notes on why a shipped default is a deployment concern, not +a code concern. + +Results also come back in whatever language the operator configured +(roster.py's `tmdb_language`, e.g. "fr-FR") — one language for the whole +node, same reasoning as the token: one shared cache, not a per-viewer +request. Omitted entirely when unset, which lets TMDB fall back to its own +default (English) rather than this client guessing one. +""" + +import difflib +import logging +import os +import re + +import httpx + +from meshbay_node.roster import Roster + +log = logging.getLogger(__name__) + +_BASE_URL = "https://api.themoviedb.org/3/" +_IMAGE_BASE = "https://image.tmdb.org/t/p/w500" +_TIMEOUT = 10.0 +_DEFAULT_TOKEN_ENV = "MESHBAY_TMDB_DEFAULT_TOKEN" + + +def _normalize(s: str) -> str: + s = s.lower() + s = re.sub(r"[^a-z0-9àâäéèêëïîôöùûüçñ ]+", " ", s) + return re.sub(r"\s+", " ", s).strip() + + +def _best_match(query_title: str, results: list[dict], keys: tuple[str, ...]) -> tuple[dict | None, float]: + """ + Trusts TMDB's own ranking (§3.3's last row — a locally-recomputed + re-rank picked a coincidentally-closer-looking wrong show once): only + the top result is considered. The similarity ratio is returned purely + as a confidence signal for the caller's fallback decision, never used + to pick a different candidate. + """ + if not results: + return None, 0.0 + top = results[0] + qn = _normalize(query_title) + best_ratio = 0.0 + for k in keys: + val = top.get(k) + if val: + best_ratio = max(best_ratio, difflib.SequenceMatcher(None, qn, _normalize(str(val))).ratio()) + return top, best_ratio + + +class TmdbClient: + """One instance per node, holding the resolved token and an httpx client.""" + + def __init__(self, roster: Roster | None = None, + transport: httpx.AsyncBaseTransport | None = None): + self._roster = roster + # `transport` is a test-only seam (httpx.MockTransport) — production + # callers never pass it, and httpx.AsyncClient defaults to real + # network I/O when it's None. + self._client = httpx.AsyncClient(timeout=_TIMEOUT, transport=transport) + + async def close(self) -> None: + await self._client.aclose() + + async def _resolve(self) -> tuple[bool, str | None, str | None]: + """Returns (enabled, token, language). token/language are None when unset.""" + if self._roster is not None: + enabled, custom_token, language = await self._roster.tmdb_config() + else: + enabled, custom_token, language = True, None, None + token = custom_token or os.environ.get(_DEFAULT_TOKEN_ENV) or None + return enabled and bool(token), token, language + + async def _get(self, path: str, params: dict) -> dict | None: + enabled, token, language = await self._resolve() + if not enabled: + return None + if language and "language" not in params: + params = {**params, "language": language} + try: + resp = await self._client.get( + _BASE_URL + path, params=params, + headers={"Authorization": f"Bearer {token}", "accept": "application/json"}, + ) + resp.raise_for_status() + return resp.json() + except httpx.HTTPError as e: + log.warning("TMDB request failed (%s): %s", path, e) + return None + + @staticmethod + def poster_url(path: str | None) -> str | None: + return f"{_IMAGE_BASE}{path}" if path else None + + async def fetch_image(self, url: str) -> bytes | None: + """Fetches a poster/backdrop image. Unauthenticated — image.tmdb.org needs no token.""" + try: + resp = await self._client.get(url, timeout=_TIMEOUT) + resp.raise_for_status() + return resp.content + except httpx.HTTPError as e: + log.warning("TMDB image fetch failed (%s): %s", url, e) + return None + + async def search_movie(self, title: str, year: int | None = None) -> tuple[dict | None, float]: + params = {"query": title, "include_adult": "false"} + if year: + params["year"] = year + data = await self._get("search/movie", params) + results = (data or {}).get("results", []) + return _best_match(title, results, ("title", "original_title")) + + async def search_tv(self, title: str) -> tuple[dict | None, float]: + data = await self._get("search/tv", {"query": title}) + results = (data or {}).get("results", []) + return _best_match(title, results, ("name", "original_name")) + + async def search_movie_results(self, title: str) -> list[dict]: + """ + The raw candidate list (capped), for an operator correcting a wrong + automatic match (§ webrtc_server.py's tmdb_search_req) — unlike + `search_movie`, this doesn't collapse to TMDB's own top result: a + human picks from several, so several is the point. + """ + data = await self._get("search/movie", {"query": title, "include_adult": "false"}) + return (data or {}).get("results", [])[:8] + + async def search_tv_results(self, title: str) -> list[dict]: + data = await self._get("search/tv", {"query": title}) + return (data or {}).get("results", [])[:8] + + async def tv_season(self, tmdb_id: str | int, season: int, + language: str | None = None) -> dict | None: + """`language`, when given, overrides the configured one — same + English-fallback use as `movie_details`/`tv_details`.""" + params = {"language": language} if language else {} + return await self._get(f"tv/{tmdb_id}/season/{season}", params) + + async def movie_details(self, tmdb_id: str | int, language: str | None = None) -> dict | None: + """ + Full details, not the search result: search/movie doesn't return + `runtime` or genre names (only `genre_ids`) at all. + + `language`, when given, overrides the configured one — used for the + English fallback fetch (§ below): TMDB itself doesn't fall back + server-side for an untranslated field, it just returns "" for it, + the same gap the TMDB website itself papers over client-side. + """ + params = {"language": language} if language else {} + return await self._get(f"movie/{tmdb_id}", params) + + async def tv_details(self, tmdb_id: str | int, language: str | None = None) -> dict | None: + params = {"language": language} if language else {} + return await self._get(f"tv/{tmdb_id}", params) + + async def movie_credits(self, tmdb_id: str | int) -> dict | None: + return await self._get(f"movie/{tmdb_id}/credits", {}) + + async def tv_credits(self, tmdb_id: str | int) -> dict | None: + return await self._get(f"tv/{tmdb_id}/credits", {}) diff --git a/packages/meshbay-node/src/meshbay_node/transport/webrtc_server.py b/packages/meshbay-node/src/meshbay_node/transport/webrtc_server.py index aaf1ecf..f4d1dee 100644 --- a/packages/meshbay-node/src/meshbay_node/transport/webrtc_server.py +++ b/packages/meshbay-node/src/meshbay_node/transport/webrtc_server.py @@ -33,6 +33,7 @@ import time from pathlib import Path from typing import Any +import blake3 import jwt import msgpack from aiortc import RTCPeerConnection, RTCSessionDescription, RTCDataChannel @@ -64,6 +65,9 @@ from meshbay_common.adminop import ( OP_MEMBER_UPLOAD, OP_APPS_ENABLED, OP_SET_SCAN_SETTINGS, + OP_TMDB_CONFIG, + OP_VIDEO_ROOT, + OP_TMDB_OVERRIDE, OP_ROOT_ADD, OP_ROOT_REMOVE, OP_GROUP_ATTACH, @@ -84,10 +88,15 @@ from meshbay_common.join import ( join_transcript, ) from meshbay_common.webcrypto import chunk_key_aes, encrypt_chunk_aes -from meshbay_common.protocol import MNP +from meshbay_common.protocol import MNP, index_entry_wire from meshbay_node.indexer import GroupIndex from meshbay_node.indexer.indexer import DirectoryIndexer from meshbay_node import ops +# Re-imported under its original name: every call site and existing test in +# this module still refers to it as `_probe_video`. The implementation lives +# in media_probe.py so the indexer package (imported just above) can call it +# too, for index-time enrichment, without a circular import. +from meshbay_node.media_probe import probe_video as _probe_video from meshbay_node.roots import ( RootSet, entry_abs_path, SAFE_UPLOAD_NAME, safe_subdir, _free_name, ) @@ -163,59 +172,6 @@ STREAM_CREDIT_TIMEOUT = 120 # charged for a slot within this, rather than within the timeout. STREAM_CREDIT_POLL = 3 -_H264_PROFILES = {"Baseline": "42", "Main": "4d", "High": "64", "High 10": "6e"} - - -async def _probe_video(path: str) -> tuple[str | None, float, bool]: - """ - Probe video file with ffprobe, return (MSE codec string, duration, - has_audio). - - The audio half of the codec string is always "mp4a.40.2" (AAC-LC) or - absent — never the source's real audio codec — because _stream_video_ - inner always transcodes audio to AAC and never copies it: MSE in every - mainstream browser only decodes AAC/Opus, and a source codec outside - that (AC-3, E-AC-3, DTS, ...) is at best silently unplayable and at - worst, for E-AC-3 at least, makes ffmpeg itself refuse to write the - fragmented MP4 header ("Cannot write moov atom before EAC3 packets - parsed" — reproduced against a real 5.1 E-AC-3 WEB-DL). Video stays - whatever it actually is: it is always copied, never transcoded. - """ - import json as _json - proc = await asyncio.create_subprocess_exec( - "ffprobe", "-v", "error", - "-show_entries", "stream=codec_name,profile,level,codec_type", - "-show_entries", "format=duration", - "-of", "json", path, - stdout=asyncio.subprocess.PIPE, stderr=asyncio.subprocess.PIPE, - ) - stdout, _ = await proc.communicate() - info = _json.loads(stdout) - duration = float(info.get("format", {}).get("duration", 0)) - - v_codec = "" - has_audio = False - for s in info.get("streams", []): - if s.get("codec_type") == "video" and not v_codec: - cn = s.get("codec_name", "") - if cn == "h264": - p = _H264_PROFILES.get(s.get("profile", "High"), "64") - lvl = int(s.get("level", 40)) - v_codec = f"avc1.{p}00{lvl:02x}" - elif cn == "hevc": - v_codec = "hev1.1.6.L93.B0" - elif cn == "vp9": - v_codec = "vp09.00.10.08" - elif cn == "av1": - v_codec = "av01.0.01M.08" - elif s.get("codec_type") == "audio": - has_audio = True - - if not v_codec: - return None, duration, has_audio - codec = f"{v_codec},mp4a.40.2" if has_audio else v_codec - return codec, duration, has_audio - def _pack(obj: dict) -> bytes: data = msgpack.packb(obj, use_bin_type=True) @@ -417,6 +373,18 @@ class WebRTCPeerSession: self._do_apps_enabled(msg) elif mtype == MNP.SET_SCAN_SETTINGS: self._do_set_scan_settings(msg) + elif mtype == MNP.TMDB_CONFIG: + self._do_tmdb_config(msg) + elif mtype == MNP.VIDEO_ROOT: + self._do_video_root(msg) + elif mtype == MNP.MEDIA_META_REQ: + self._spawn(self._do_media_meta_request(msg)) + elif mtype == MNP.SEASON_META_REQ: + self._spawn(self._do_season_meta_request(msg)) + elif mtype == MNP.TMDB_SEARCH_REQ: + self._spawn(self._do_tmdb_search_request(msg)) + elif mtype == MNP.TMDB_OVERRIDE: + self._do_tmdb_override(msg) elif mtype == MNP.MEMBER_UNPIN: self._do_member_unpin(msg) elif mtype == MNP.GEK_ROTATE: @@ -667,6 +635,20 @@ class WebRTCPeerSession: # setting (or one whose context has not loaded it yet) hides # nothing. "enabled_apps": list(self._group_ctx().get("enabled_apps") or []), + # Which folder the Videos app treats as its entry point for + # this group — "" means the whole group index. + "video_root": self._group_ctx().get("video_root") or "", + # Node-wide (not per-group), same "read once, kept current by + # the signed op" shape — surfaced here rather than only via + # tmdb_config_ack so a client that connects after the operator + # already configured it does not have to wait for a live change + # to find out (docs/mediacenter.md §5.5). + "tmdb_enabled": bool( + self._ctx.get("daemon_state", {}).get("tmdb_enabled", True)), + "tmdb_token_customized": bool( + self._ctx.get("daemon_state", {}).get("tmdb_token_customized", False)), + "tmdb_language": str( + self._ctx.get("daemon_state", {}).get("tmdb_language") or ""), # So a client that connects mid-scan shows the indexing state # immediately, instead of waiting for the next periodic # INDEX_PROGRESS push. Never a path or filename — see @@ -1623,10 +1605,13 @@ class WebRTCPeerSession: except Exception: pass - # Every "application" a group can show — Chat and Files today. Videos, - # Music, Photos join this set (and apps.js's registry, client-side) when - # they land; nothing else about this handler changes. - ALLOWED_APPS = frozenset({"chat", "files"}) + # Every "application" a group can show. Music, Photos join this set (and + # apps.js's registry, client-side) when they land; nothing else about + # this handler changes. DEFAULT_APPS (roster.py) deliberately does not + # include "video" — it is the first app with outbound third-party + # network calls (once TMDB is on), so an operator opts a group in + # explicitly rather than getting it for free (docs/mediacenter.md §5.6). + ALLOWED_APPS = frozenset({"chat", "files", "video"}) def _do_apps_enabled(self, msg: dict) -> None: """ @@ -1677,6 +1662,125 @@ class WebRTCPeerSession: except Exception: pass + def _do_tmdb_config(self, msg: dict) -> None: + """ + Turn TMDB lookups on/off node-wide, optionally set (or clear) a + custom API token, and optionally set the language TMDB is queried + in (e.g. "fr-FR") — one for the whole node, same reasoning as the + token: one shared cache, not a per-viewer request. Signed like the + rest: this turns on outbound third-party network traffic the node + did not have before the Videos app (docs/mediacenter.md §5.5, §8) + — an unsigned toggle would let any member turn on egress the + operator never agreed to. + """ + enabled = msg.get("enabled") + if not isinstance(enabled, bool): + self._send({"type": "error", "detail": "Missing or invalid 'enabled'"}) + return + token = msg.get("token") + if token is not None and not isinstance(token, str): + self._send({"type": "error", "detail": "Invalid 'token'"}) + return + language = msg.get("language") + if language is not None and not isinstance(language, str): + self._send({"type": "error", "detail": "Invalid 'language'"}) + return + if not self._has_admin_authority(): + self._send({"type": "error", "detail": "No authorized key for this"}) + return + # The subject is the signed, audited, human-shown string — it must + # never contain the token itself (it would end up in the audit log + # in plaintext). The actual token travels only in `payload`, which + # is node-side context, never re-sent or re-verified from the wire. + # The language is not a secret, so it travels in the subject itself. + subject = (f"enabled={enabled},custom_token={'yes' if token else 'no'}," + f"language={language or 'default'}") + self._issue_admin_challenge( + OP_TMDB_CONFIG, subject, + payload={"enabled": enabled, "token": token, "language": language}, + group_id="") + + async def _admin_exec_tmdb_config( + self, pending: dict, transcript: bytes, sig: bytes, + ) -> None: + if not await self._verify_admin_sig(transcript, sig): + self._send({"type": "error", "detail": "Signature verification failed"}) + self._audit("admin_auth_failed", f"tmdb_config:{pending['subject']}") + return + p = pending.get("payload") or {} + try: + result = await self._run_op( + ops.set_tmdb_config, p.get("enabled", True), p.get("token"), p.get("language")) + except ops.OpError as e: + self._send({"type": "error", "detail": e.message}) + return + self._audit("tmdb_config", pending["subject"]) + + # Node-wide setting: every connected peer in every group is told, not + # just this group's peers (unlike apps_enabled/member_upload). + notice = { + "type": MNP.TMDB_CONFIG_ACK, "v": MNP_VERSION, + "enabled": result["enabled"], "token_customized": result["token_customized"], + "language": result["language"], + } + for gctx in self._ctx.get("groups", {}).values(): + for session in list(gctx.get("_peers", {}).values()): + try: + session._send(notice) + except Exception: + pass + + def _do_video_root(self, msg: dict) -> None: + """ + Which folder (possibly a subfolder of a shared root) the Videos app + treats as its entry point for this group. Signed like apps_enabled: + it decides what every member's Videos tab shows. + + An empty path is always accepted (it means "the whole group index", + today's behaviour). A non-empty path must resolve to a real, + currently-readable directory — validated against the group's own + roots the same way directory creation/deletion already is, so a + stale or mistyped path is refused before a signature is even asked + for. + """ + path = msg.get("path") + if not isinstance(path, str): + self._send({"type": "error", "detail": "Missing or invalid 'path'"}) + return + path = path.strip("/") + if path: + ctx = self._group_ctx() + resolved = ctx["roots"].resolve(path) if ctx.get("roots") else None + if not resolved or not resolved.is_dir(): + self._send({"type": "error", "detail": "Not a directory in this group"}) + return + if not self._has_admin_authority(): + self._send({"type": "error", "detail": "No authorized key for this"}) + return + self._issue_admin_challenge(OP_VIDEO_ROOT, path) + + async def _admin_exec_video_root( + self, pending: dict, transcript: bytes, sig: bytes, + ) -> None: + path = pending["subject"] + if not await self._verify_admin_sig(transcript, sig): + self._send({"type": "error", "detail": "Signature verification failed"}) + self._audit("admin_auth_failed", f"video_root:{path}") + return + try: + await self._run_op(ops.set_video_root, self._group_id or "", path) + except ops.OpError as e: + self._send({"type": "error", "detail": e.message}) + return + self._audit("video_root", path) + + notice = {"type": MNP.VIDEO_ROOT_ACK, "v": MNP_VERSION, "path": path} + for uid, session in list(self._peer_registry().items()): + try: + session._send(notice) + except Exception: + pass + # Reconcile's backstop and the watchdog debounce (indexer.py # DirectoryIndexer) — how hard the node works on the operator's own # disk, not a member-facing permission. Signed for the same reason as @@ -2138,14 +2242,7 @@ class WebRTCPeerSession: def _do_index_sync(self) -> None: ctx = self._group_ctx() idx = ctx["index"] - entries = [ - { - "id": e.id, "name": e.name, "path": e.path, - "size": e.size, "type": e.type, "added_at": e.added_at, - "uploader_id": e.uploader_id, - } - for e in idx.entries - ] + entries = [index_entry_wire(e) for e in idx.entries] self._send({ "type": MNP.INDEX_SYNC, "v": MNP_VERSION, @@ -2189,12 +2286,38 @@ class WebRTCPeerSession: continue return sorted(out)[:2000] + async def _try_serve_thumbnail( + self, thumb_hash: str, chunk_index: int, gek: bytes | None, + ) -> dict | None: + """ + docs/mediacenter.md §5.3: a thumbnail is served through the same + chunked file_req path as a real file, resolved against the media + cache instead of the index when the id doesn't match a file. Always + a single chunk in practice (a thumbnail-sized JPEG never approaches + CHUNK_SIZE) — a request for any chunk beyond 0 is just a miss. + """ + media_cache = self._ctx.get("media_cache") + if media_cache is None or chunk_index != 0: + return None + jpeg = await media_cache.get_thumb(thumb_hash) + if jpeg is None: + return None + return _encrypt_chunk_bytes( + self._ctx["sk_node"], gek, jpeg, 0, + bytes.fromhex(thumb_hash), thumb_hash, + ) + async def _do_file_request(self, msg: dict) -> None: ctx = self._group_ctx() file_id = msg["file_id"] chunk_index = msg["chunk_index"] entry = ctx["index"].get_entry(file_id) if not entry: + thumb = await self._try_serve_thumbnail(file_id, chunk_index, ctx.get("gek")) + if thumb is not None: + log.debug("file_req file_id=%s chunk=%s: served as thumbnail", file_id[:16], chunk_index) + self._send(thumb) + return log.warning("File not found: %s", file_id[:16]) self._send({"type": "error", "detail": "File not found"}) return @@ -2235,6 +2358,344 @@ class WebRTCPeerSession: if chunk_index == 0: self._audit("file_download", entry.name) + @staticmethod + async def _fetch_and_cache_poster(media_cache, tmdb_client, poster_path: str | None) -> str | None: + """ + Downloads a TMDB poster/backdrop once, caches it under its own + blake3 like a video thumbnail (docs/mediacenter.md §5.4), and + returns the hash a client then fetches via the normal file_req/ + chunk path (§5.3) — no client ever contacts image.tmdb.org directly. + + Checked by the synthetic `tmdb:{poster_path}` id *before* touching + the network: without this, every `media_meta_req` for an + already-cached file re-downloaded the same poster from TMDB (found + live — a poster grid re-fetched both a show's poster and backdrop + from TMDB on every single visit, real added latency and needless + outbound traffic for an image that never changes). + """ + if not poster_path: + return None + synthetic_id = f"tmdb:{poster_path}" + cached_hash = await media_cache.get_thumb_hash_by_file_id(synthetic_id) + if cached_hash is not None: + return cached_hash + content = await tmdb_client.fetch_image(tmdb_client.poster_url(poster_path)) + if content is None: + return None + thumb_hash = blake3.blake3(content).hexdigest() + await media_cache.put_thumb(thumb_hash, synthetic_id, content) + return thumb_hash + + async def _do_media_meta_request(self, msg: dict) -> None: + """ + docs/mediacenter.md §5.4: TMDB metadata for one path, resolved from + the group's index (root+relpath the client already knows from + index_sync/index_delta — never a raw filesystem path off the wire). + """ + path = msg.get("path") + log.debug("media_meta_req path=%r", path) + if not isinstance(path, str) or not path: + self._send({"type": "error", "detail": "Missing path"}) + return + ctx = self._group_ctx() + entry = ctx["index"].get_entry_by_path(path) + if not entry: + self._send({"type": "error", "detail": "File not found"}) + return + + media_cache = self._ctx.get("media_cache") + tmdb_client = self._ctx.get("tmdb_client") + if media_cache is None or tmdb_client is None: + self._send({"type": MNP.MEDIA_META_RESP, "v": MNP_VERSION, + "path": path, "confidence": 0}) + return + + is_show = entry.season is not None and entry.episode is not None + media_type = "tv" if is_show else "movie" + + cached = await media_cache.get_file_tmdb(entry.id) + meta = None + tmdb_id = None + if cached is not None: + tmdb_id, media_type = cached + meta = await media_cache.get_tmdb_meta(tmdb_id, media_type) + + if meta is None: + result, ratio = await self._tmdb_search(tmdb_client, entry, is_show) + if result is None or ratio < 0.6: + self._send({"type": MNP.MEDIA_META_RESP, "v": MNP_VERSION, + "path": path, "confidence": 0}) + return + tmdb_id = str(result["id"]) + meta = await self._tmdb_build_meta(tmdb_client, tmdb_id, media_type, result) + await media_cache.set_file_tmdb(entry.id, tmdb_id, media_type) + await media_cache.set_tmdb_meta(tmdb_id, media_type, meta) + + poster_thumb_hash = await self._fetch_and_cache_poster( + media_cache, tmdb_client, meta.get("poster_path")) + backdrop_thumb_hash = await self._fetch_and_cache_poster( + media_cache, tmdb_client, meta.get("backdrop_path")) + log.debug("media_meta_req path=%r: replying tmdb_id=%s poster=%s backdrop=%s", + path, tmdb_id, poster_thumb_hash, backdrop_thumb_hash) + + resp = { + "type": MNP.MEDIA_META_RESP, "v": MNP_VERSION, "path": path, + "tmdb_id": tmdb_id, "title": meta.get("title"), + "original_title": meta.get("original_title"), + "overview": meta.get("overview"), + "poster_thumb_hash": poster_thumb_hash, + "backdrop_thumb_hash": backdrop_thumb_hash, + "release_date": meta.get("release_date"), + "first_air_date": meta.get("first_air_date"), + "genres": meta.get("genres", []), + "vote_average": meta.get("vote_average"), + "runtime": meta.get("runtime"), + "cast": meta.get("cast", []), + "director": meta.get("director"), + "confidence": meta.get("confidence", 1.0), + } + if is_show: + resp["season"] = entry.season + resp["episode"] = entry.episode + self._send(resp) + + async def _do_season_meta_request(self, msg: dict) -> None: + """ + Per-season TMDB overview/poster/air_date for a multi-season show — + found live: `media_meta_resp`'s one static show-level overview does + not necessarily describe every season alike (a season-3-specific + promotional summary applied to all three seasons of a show). + `tmdb_id` is whatever the client's own prior `media_meta_resp` + already resolved — never re-derived from a path here, so this + never re-runs a TMDB search of its own. + """ + tmdb_id = msg.get("tmdb_id") + season = msg.get("season") + if not isinstance(tmdb_id, str) or not tmdb_id or not isinstance(season, int): + self._send({"type": "error", "detail": "Missing tmdb_id or season"}) + return + media_cache = self._ctx.get("media_cache") + tmdb_client = self._ctx.get("tmdb_client") + if media_cache is None or tmdb_client is None: + self._send({"type": MNP.SEASON_META_RESP, "v": MNP_VERSION, + "tmdb_id": tmdb_id, "season": season, "confidence": 0}) + return + + details = await media_cache.get_season_meta(tmdb_id, season) + if details is None: + fetched = await tmdb_client.tv_season(tmdb_id, season) + if fetched is None: + self._send({"type": MNP.SEASON_META_RESP, "v": MNP_VERSION, + "tmdb_id": tmdb_id, "season": season, "confidence": 0}) + return + # Same per-field English fallback as _tmdb_build_meta: TMDB + # returns "" for an untranslated field rather than falling back + # itself. + if not fetched.get("overview"): + fallback = await tmdb_client.tv_season(tmdb_id, season, language="en-US") or {} + fetched = {**fallback, **{k: v for k, v in fetched.items() if v not in (None, "", [])}} + await media_cache.set_season_meta(tmdb_id, season, fetched) + details = fetched + + poster_thumb_hash = await self._fetch_and_cache_poster( + media_cache, tmdb_client, details.get("poster_path")) + self._send({ + "type": MNP.SEASON_META_RESP, "v": MNP_VERSION, + "tmdb_id": tmdb_id, "season": season, "confidence": 1.0, + "name": details.get("name"), + "overview": details.get("overview"), + "air_date": details.get("air_date"), + "poster_thumb_hash": poster_thumb_hash, + }) + + async def _do_tmdb_search_request(self, msg: dict) -> None: + """ + Candidate TMDB matches for an operator correcting a wrong automatic + match (docs/mediacenter.md, §V-whatever this becomes) — a plain + lookup, not a mutation, so unlike `tmdb_override` this needs no + admin authority: any member can see what TMDB itself would offer, + the same as the automatic search already silently does on their + behalf. Only `tmdb_override` actually changes what everyone sees. + """ + query = msg.get("query") + media_type = msg.get("media_type") + if not isinstance(query, str) or not query.strip() or media_type not in ("movie", "tv"): + self._send({"type": "error", "detail": "Missing query or media_type"}) + return + media_cache = self._ctx.get("media_cache") + tmdb_client = self._ctx.get("tmdb_client") + if media_cache is None or tmdb_client is None: + self._send({"type": MNP.TMDB_SEARCH_RESP, "v": MNP_VERSION, + "query": query, "media_type": media_type, "results": []}) + return + + raw = (await tmdb_client.search_movie_results(query) if media_type == "movie" + else await tmdb_client.search_tv_results(query)) + results = [] + for r in raw: + poster_thumb_hash = await self._fetch_and_cache_poster( + media_cache, tmdb_client, r.get("poster_path")) + results.append({ + "tmdb_id": str(r.get("id")), + "title": r.get("title") or r.get("name"), + "year": (r.get("release_date") or r.get("first_air_date") or "")[:4], + "poster_thumb_hash": poster_thumb_hash, + }) + # media_type echoed back, not just query: a client can fire a "war" + # tv search and a "war" movie search close together, and without it + # the two responses are indistinguishable for keyed matching + # (transport.js's tmdb_search_resp handler). + self._send({"type": MNP.TMDB_SEARCH_RESP, "v": MNP_VERSION, + "query": query, "media_type": media_type, "results": results}) + + def _do_tmdb_override(self, msg: dict) -> None: + """ + An operator correcting a wrong automatic TMDB match. Signed like + video_root/tmdb_config: it replaces what every member sees for a + show/movie, node-wide (media_cache is shared, not per-viewer). + + Applied to every entry sharing the representative file's + display_title — the same grouping the poster grid itself uses + (§3.4/§V6) — not just the one file the operator happened to be + looking at, so the correction actually sticks regardless of which + episode a future render picks as representative. + """ + path = msg.get("path") + tmdb_id = msg.get("tmdb_id") + media_type = msg.get("media_type") + if not isinstance(path, str) or not path: + self._send({"type": "error", "detail": "Missing path"}) + return + if not isinstance(tmdb_id, str) or not tmdb_id or media_type not in ("movie", "tv"): + self._send({"type": "error", "detail": "Missing tmdb_id or media_type"}) + return + ctx = self._group_ctx() + entry = ctx["index"].get_entry_by_path(path) + if not entry: + self._send({"type": "error", "detail": "File not found"}) + return + if not self._has_admin_authority(): + self._send({"type": "error", "detail": "No authorized key for this"}) + return + subject = f"path={path},tmdb_id={tmdb_id},media_type={media_type}" + self._issue_admin_challenge(OP_TMDB_OVERRIDE, subject) + + async def _admin_exec_tmdb_override( + self, pending: dict, transcript: bytes, sig: bytes, + ) -> None: + subject = pending["subject"] + if not await self._verify_admin_sig(transcript, sig): + self._send({"type": "error", "detail": "Signature verification failed"}) + self._audit("admin_auth_failed", f"tmdb_override:{subject}") + return + fields = dict(part.split("=", 1) for part in subject.split(",")) + path, tmdb_id, media_type = fields["path"], fields["tmdb_id"], fields["media_type"] + + ctx = self._group_ctx() + entry = ctx["index"].get_entry_by_path(path) + media_cache = self._ctx.get("media_cache") + if entry is None or media_cache is None: + self._send({"type": "error", "detail": "File or media cache not available"}) + return + target_title = entry.display_title or entry.name + matched = [e for e in ctx["index"].entries + if e.type == "video" and (e.display_title or e.name) == target_title] + for e in matched: + await media_cache.set_file_tmdb(e.id, tmdb_id, media_type) + self._audit("tmdb_override", subject) + + notice = {"type": MNP.TMDB_OVERRIDE_ACK, "v": MNP_VERSION, "path": path, + "tmdb_id": tmdb_id, "media_type": media_type} + for uid, session in list(self._peer_registry().items()): + try: + session._send(notice) + except Exception: + pass + + async def _tmdb_search(self, tmdb_client, entry, is_show: bool): + """ + §3.3's retry ladder: the parsed title first, then a couple of + generic, non-per-title fallbacks — never re-ranking TMDB's own + top result locally (§3.3's last row). + """ + from meshbay_node.indexer import title_parse + + if is_show: + title = entry.display_title or title_parse.naive_title(entry.name) + result, ratio = await tmdb_client.search_tv(title) + if result is None or ratio < 0.6: + naive = title_parse.naive_title(entry.name) + if naive != title: + result, ratio = await tmdb_client.search_tv(naive) + return result, ratio + + parsed = title_parse.parse_movie_filename(entry.name) + title = entry.display_title or parsed.display_title or parsed.naive_title + result, ratio = await tmdb_client.search_movie(title, parsed.year) + if result is not None and ratio >= 0.6: + return result, ratio + for candidate in filter(None, [parsed.alt_title, parsed.naive_title, + *title_parse.sequel_variants(title)]): + if candidate == title: + continue + result2, ratio2 = await tmdb_client.search_movie(candidate, parsed.year) + if result2 is not None and ratio2 > ratio: + result, ratio = result2, ratio2 + if ratio >= 0.6: + break + return result, ratio + + @staticmethod + async def _tmdb_build_meta(tmdb_client, tmdb_id: str, media_type: str, result: dict) -> dict: + """ + `result` (the search hit) only carries `genre_ids` and no `runtime` + at all — the full details endpoint is the actual source for those, + falling back to the search result for anything details somehow + lacks (never expected in practice, just avoids a KeyError-shaped + surprise if TMDB's response ever varies). + """ + details = (await tmdb_client.tv_details(tmdb_id) if media_type == "tv" + else await tmdb_client.movie_details(tmdb_id)) or result + # TMDB doesn't fall back server-side for a field with no translation + # in the configured language — it returns "" (or an empty list) for + # it, not the English text (confirmed live: a French query left + # `overview` empty for a title TMDB has no French translation for). + # The TMDB website covers exactly this gap client-side, by falling + # back to English per field rather than discarding an otherwise-good + # localized response over one empty one — mirrored here the same + # way, at field granularity, not by abandoning the whole response. + if not details.get("overview") or not details.get("poster_path") or not details.get("genres"): + fallback = (await tmdb_client.tv_details(tmdb_id, language="en-US") if media_type == "tv" + else await tmdb_client.movie_details(tmdb_id, language="en-US")) or {} + details = {**fallback, **{k: v for k, v in details.items() if v not in (None, "", [])}} + credits = (await tmdb_client.tv_credits(tmdb_id) if media_type == "tv" + else await tmdb_client.movie_credits(tmdb_id)) + cast = [{"name": c.get("name"), "character": c.get("character")} + for c in (credits or {}).get("cast", [])[:10]] + director = None + if media_type == "movie": + director = next( + (c.get("name") for c in (credits or {}).get("crew", []) + if c.get("job") == "Director"), None) + runtime = details.get("runtime") + if runtime is None and media_type == "tv": + episode_run_times = details.get("episode_run_time") or [] + runtime = episode_run_times[0] if episode_run_times else None + return { + "title": details.get("title") or details.get("name"), + "original_title": details.get("original_title") or details.get("original_name"), + "overview": details.get("overview"), + "poster_path": details.get("poster_path"), + "backdrop_path": details.get("backdrop_path"), + "release_date": details.get("release_date"), + "first_air_date": details.get("first_air_date"), + "genres": [g.get("name") for g in details.get("genres", []) if g.get("name")], + "vote_average": details.get("vote_average"), + "runtime": runtime, + "cast": cast, + "director": director, + } + def _do_stream_segment(self, msg: dict) -> None: self._spawn(self._do_stream_segment_async(msg)) @@ -2738,6 +3199,15 @@ class WebRTCPeerSession: elif pending["op"] == OP_SET_SCAN_SETTINGS: self._spawn( self._admin_exec_set_scan_settings(pending, transcript, sig_bytes)) + elif pending["op"] == OP_TMDB_CONFIG: + self._spawn( + self._admin_exec_tmdb_config(pending, transcript, sig_bytes)) + elif pending["op"] == OP_VIDEO_ROOT: + self._spawn( + self._admin_exec_video_root(pending, transcript, sig_bytes)) + elif pending["op"] == OP_TMDB_OVERRIDE: + self._spawn( + self._admin_exec_tmdb_override(pending, transcript, sig_bytes)) elif pending["op"] == OP_ROOT_ADD: self._spawn( self._admin_exec_root_add(pending, transcript, sig_bytes)) @@ -3009,7 +3479,7 @@ class WebRTCPeerSession: file_hash = bytes.fromhex(entry.id) try: - codec_str, duration, has_audio = await _probe_video(str(file_path)) + codec_str, duration, has_audio, _width, _height = await _probe_video(str(file_path)) except Exception as e: self._send({"type": "error", "detail": f"Probe failed: {e}"}) return @@ -3220,18 +3690,14 @@ class WebRTCPeerSession: await self._pc.close() -def _read_and_encrypt( +def _encrypt_chunk_bytes( sk_node: Ed25519PrivateKey, gek: bytes, - file_path: Path, + plaintext: bytes, chunk_index: int, file_hash: bytes, file_id: str = "", ) -> dict: - with open(file_path, "rb") as f: - f.seek(chunk_index * CHUNK_SIZE) - plaintext = f.read(CHUNK_SIZE) - ckey = chunk_key_aes(gek, file_hash, chunk_index) nonce, ct = encrypt_chunk_aes(ckey, plaintext) @@ -3249,6 +3715,20 @@ def _read_and_encrypt( } +def _read_and_encrypt( + sk_node: Ed25519PrivateKey, + gek: bytes, + file_path: Path, + chunk_index: int, + file_hash: bytes, + file_id: str = "", +) -> dict: + with open(file_path, "rb") as f: + f.seek(chunk_index * CHUNK_SIZE) + plaintext = f.read(CHUNK_SIZE) + return _encrypt_chunk_bytes(sk_node, gek, plaintext, chunk_index, file_hash, file_id) + + class WebRTCTransport: """ Manages WebRTC peer connections for browser clients. diff --git a/packages/meshbay-node/src/meshbay_node/ui/app.py b/packages/meshbay-node/src/meshbay_node/ui/app.py index d5b3f94..18764e8 100644 --- a/packages/meshbay-node/src/meshbay_node/ui/app.py +++ b/packages/meshbay-node/src/meshbay_node/ui/app.py @@ -20,7 +20,7 @@ import time from html import escape from pathlib import Path -from fastapi import FastAPI, WebSocket, WebSocketDisconnect, Query +from fastapi import FastAPI, HTTPException, WebSocket, WebSocketDisconnect, Query from fastapi.responses import HTMLResponse, JSONResponse from meshbay_node import __version__ @@ -361,6 +361,20 @@ def create_ui_app(state: dict) -> FastAPI: state, group_id, bool(payload.get("allowed", False)), )) + # ── Enabled apps (operator only, localhost) ──────────────────────────── + # + # Same loopback shape as member-upload: the Create Group wizard sets this + # once, right after creating the group and before the (potentially long) + # initial scan, so an operator narrowing this down to just Files+Videos + # never briefly has Chat live for other members to notice. + + @app.put("/api/groups/{group_id}/apps") + async def set_enabled_apps(group_id: str, payload: dict): + apps = payload.get("apps") + if not isinstance(apps, list) or not apps: + raise HTTPException(400, "apps must be a non-empty list") + return await _op(lambda: ops.set_enabled_apps(state, group_id, apps)) + # ── Scan settings (operator only, localhost) ────────────────────────── @app.put("/api/groups/{group_id}/scan-settings") diff --git a/packages/meshbay-node/tests/test_enrich.py b/packages/meshbay-node/tests/test_enrich.py new file mode 100644 index 0000000..cff4d50 --- /dev/null +++ b/packages/meshbay-node/tests/test_enrich.py @@ -0,0 +1,130 @@ +"""Tests for indexer/enrich.py — season/title corroboration and the end-to-end pool.""" + +import asyncio +import shutil +import subprocess +from pathlib import Path + +import pytest + +from meshbay_common.protocol import IndexEntry +from meshbay_node.indexer.enrich import Enricher, _season_from_ancestors, _title_from_siblings +from meshbay_node.media_cache import MediaCache + +_HAVE_FFMPEG = shutil.which("ffmpeg") and shutil.which("ffprobe") + + +# ── pure helpers, no ffmpeg needed ─────────────────────────────────────────── + +def test_season_from_ancestors_finds_season_folder(tmp_path): + folder = tmp_path / "Some Show" / "Season 2" + folder.mkdir(parents=True) + ep = folder / "01 - Episode Title.mkv" + ep.touch() + + assert _season_from_ancestors(ep) == 2 + + +def test_season_from_ancestors_none_when_no_season_folder(tmp_path): + folder = tmp_path / "Movies" + folder.mkdir() + f = folder / "Some Movie 2015.mkv" + f.touch() + + assert _season_from_ancestors(f) is None + + +def test_title_from_siblings_borrows_from_a_titled_sibling(tmp_path): + folder = tmp_path / "Acronym Show" + folder.mkdir() + titled = folder / "Some.Show.Name.S01E01.720p.mkv" + untitled = folder / "S01E02.SUBFRENCH.720p.mkv" + titled.touch() + untitled.touch() + + assert _title_from_siblings(untitled) == "Some Show Name" + + +def test_title_from_siblings_none_when_no_titled_sibling(tmp_path): + folder = tmp_path / "Acronym Show" + folder.mkdir() + (folder / "S01E02.mkv").touch() + + assert _title_from_siblings(folder / "S01E02.mkv") is None + + +# ── end-to-end against a real (tiny, synthetic) video file ────────────────── + +pytestmark_ffmpeg = pytest.mark.skipif(not _HAVE_FFMPEG, reason="ffmpeg/ffprobe not installed") + + +def _make_clip(path: Path) -> None: + subprocess.run( + ["ffmpeg", "-hide_banner", "-loglevel", "error", "-y", + "-f", "lavfi", "-i", "testsrc=size=320x240:rate=25:duration=1", + "-c:v", "libx264", "-preset", "ultrafast", "-an", str(path)], + check=True, capture_output=True, + ) + + +@pytest.fixture +async def media_cache(tmp_path): + c = MediaCache(db_path=tmp_path / "media_cache.db") + await c.open() + yield c + await c.close() + + +@pytestmark_ffmpeg +@pytest.mark.asyncio +async def test_enricher_populates_fields_and_stores_thumbnail(tmp_path, media_cache): + clip = tmp_path / "Some.Movie.2015.1080p.mkv" + _make_clip(clip) + entry = IndexEntry(id="fileid1", name=clip.name, path=clip.name, + size=clip.stat().st_size, type="video", added_at=0) + + enricher = Enricher(media_cache) + done = asyncio.get_event_loop().create_future() + + async def on_done(file_id, fields): + done.set_result((file_id, fields)) + + enricher.spawn(entry, clip, on_done) + file_id, fields = await asyncio.wait_for(done, timeout=30) + + assert file_id == "fileid1" + assert fields["width"] == 320 + assert fields["height"] == 240 + assert fields["display_title"] == "Some Movie" + assert fields.get("thumb_hash") + stored = await media_cache.get_thumb(fields["thumb_hash"]) + assert stored is not None and len(stored) > 0 + + +@pytestmark_ffmpeg +@pytest.mark.asyncio +async def test_enricher_handles_episode_with_season_from_folder(tmp_path, media_cache): + # Filename carries only a bare episode number, no SxxExx token — guessit + # confirmed (separately) not to find a season here at all — so the + # season must come from the ancestor folder (§3.4's non-standard case). + folder = tmp_path / "Some Show" / "Saison 3" + folder.mkdir(parents=True) + titled_sibling = folder / "Some.Show.Episode.06.mkv" + titled_sibling.touch() + clip = folder / "Episode.07.720p.mkv" + _make_clip(clip) + entry = IndexEntry(id="fileid2", name=clip.name, path=str(clip.relative_to(tmp_path)), + size=clip.stat().st_size, type="video", added_at=0) + + enricher = Enricher(media_cache) + done = asyncio.get_event_loop().create_future() + + async def on_done(file_id, fields): + done.set_result((file_id, fields)) + + enricher.spawn(entry, clip, on_done) + file_id, fields = await asyncio.wait_for(done, timeout=30) + + assert fields["display_title"] == "Some Show" + assert fields["season"] == 3 + assert fields["episode"] == 7 diff --git a/packages/meshbay-node/tests/test_media_cache.py b/packages/meshbay-node/tests/test_media_cache.py new file mode 100644 index 0000000..b66c448 --- /dev/null +++ b/packages/meshbay-node/tests/test_media_cache.py @@ -0,0 +1,71 @@ +"""Tests for media_cache.py — TMDB/thumbnail cache and its pruning obligation.""" + +import time + +import pytest + +from meshbay_node.media_cache import MediaCache, TMDB_META_TTL_SECS + + +@pytest.fixture +async def cache(tmp_path): + c = MediaCache(db_path=tmp_path / "media_cache.db") + await c.open() + yield c + await c.close() + + +@pytest.mark.asyncio +async def test_file_tmdb_round_trip(cache): + assert await cache.get_file_tmdb("file1") is None + + await cache.set_file_tmdb("file1", "12345", "movie") + + assert await cache.get_file_tmdb("file1") == ("12345", "movie") + + +@pytest.mark.asyncio +async def test_tmdb_meta_round_trip(cache): + assert await cache.get_tmdb_meta("12345", "movie") is None + + await cache.set_tmdb_meta("12345", "movie", {"title": "A Movie", "vote_average": 7.5}) + + meta = await cache.get_tmdb_meta("12345", "movie") + assert meta == {"title": "A Movie", "vote_average": 7.5} + + +@pytest.mark.asyncio +async def test_tmdb_meta_expires_after_ttl(cache): + await cache._db.execute( + "INSERT INTO tmdb_meta (tmdb_id, media_type, json, fetched_at) VALUES (?, ?, ?, ?)", + ("999", "movie", '{"title": "Old"}', time.time() - TMDB_META_TTL_SECS - 1), + ) + await cache._db.commit() + + assert await cache.get_tmdb_meta("999", "movie") is None + + +@pytest.mark.asyncio +async def test_thumb_round_trip(cache): + assert await cache.get_thumb("thumbhash1") is None + + await cache.put_thumb("thumbhash1", "file1", b"\xff\xd8fakejpeg") + + assert await cache.get_thumb("thumbhash1") == b"\xff\xd8fakejpeg" + + +@pytest.mark.asyncio +async def test_prune_file_removes_thumb_and_mapping_but_not_shared_meta(cache): + # Two episodes of the same show share one tmdb_meta row (§2's stated case). + await cache.set_file_tmdb("ep1", "555", "tv") + await cache.set_file_tmdb("ep2", "555", "tv") + await cache.set_tmdb_meta("555", "tv", {"name": "A Show"}) + await cache.put_thumb("thumb-ep1", "ep1", b"jpeg-bytes-1") + + await cache.prune_file("ep1") + + assert await cache.get_file_tmdb("ep1") is None + assert await cache.get_thumb("thumb-ep1") is None + # ep2's own mapping and the shared show metadata both survive + assert await cache.get_file_tmdb("ep2") == ("555", "tv") + assert await cache.get_tmdb_meta("555", "tv") == {"name": "A Show"} diff --git a/packages/meshbay-node/tests/test_poster_cache.py b/packages/meshbay-node/tests/test_poster_cache.py new file mode 100644 index 0000000..bbd824d --- /dev/null +++ b/packages/meshbay-node/tests/test_poster_cache.py @@ -0,0 +1,88 @@ +""" +Bug found live, 2026-08-24: `_fetch_and_cache_poster` downloaded a TMDB +poster/backdrop from `image.tmdb.org` on *every* `media_meta_req`, even for +a file whose TMDB match was already cached — the content-addressed +`thumb_hash` isn't known until the bytes are downloaded, so nothing had +ever checked "have I already fetched this poster_path" first. On a group +with a show split across release folders (§V6), one Videos-tab visit +triggered four to six redundant image downloads; compounded with TMDB +latency (or a stall), this is what an operator saw as posters that "never +finish loading" on a second visit. + +Fixed by keying the `thumbs` cache by a synthetic `tmdb:{poster_path}` id +*before* the network call, mirroring the `file_id` convention `_do_file_request` +already uses to resolve a thumbnail by id. +""" + +import pytest + +from meshbay_node.media_cache import MediaCache +from meshbay_node.transport.webrtc_server import WebRTCPeerSession + +pytestmark = pytest.mark.asyncio + + +@pytest.fixture +async def media_cache(tmp_path): + c = MediaCache(db_path=tmp_path / "media_cache.db") + await c.open() + yield c + await c.close() + + +class FakeTmdbClient: + def __init__(self): + self.fetch_calls = 0 + + @staticmethod + def poster_url(path): + return f"https://image.tmdb.org/t/p/w500{path}" + + async def fetch_image(self, url): + self.fetch_calls += 1 + return b"jpeg-bytes-for-" + url.encode() + + +async def test_second_fetch_for_the_same_poster_path_skips_the_network(media_cache): + client = FakeTmdbClient() + + first = await WebRTCPeerSession._fetch_and_cache_poster( + media_cache, client, "/poster.jpg") + second = await WebRTCPeerSession._fetch_and_cache_poster( + media_cache, client, "/poster.jpg") + + assert first == second, "the same poster_path must yield the same thumb_hash" + assert client.fetch_calls == 1, ( + "a poster already cached must never be re-downloaded from TMDB") + + +async def test_different_poster_paths_are_each_fetched_once(media_cache): + client = FakeTmdbClient() + + poster_hash = await WebRTCPeerSession._fetch_and_cache_poster( + media_cache, client, "/poster.jpg") + backdrop_hash = await WebRTCPeerSession._fetch_and_cache_poster( + media_cache, client, "/backdrop.jpg") + poster_hash_again = await WebRTCPeerSession._fetch_and_cache_poster( + media_cache, client, "/poster.jpg") + + assert poster_hash != backdrop_hash + assert poster_hash == poster_hash_again + assert client.fetch_calls == 2, "one network fetch per distinct poster_path" + + +async def test_none_path_is_a_no_op(media_cache): + client = FakeTmdbClient() + result = await WebRTCPeerSession._fetch_and_cache_poster(media_cache, client, None) + assert result is None + assert client.fetch_calls == 0 + + +async def test_cached_hash_actually_serves_the_downloaded_bytes(media_cache): + client = FakeTmdbClient() + thumb_hash = await WebRTCPeerSession._fetch_and_cache_poster( + media_cache, client, "/poster.jpg") + await WebRTCPeerSession._fetch_and_cache_poster(media_cache, client, "/poster.jpg") + + stored = await media_cache.get_thumb(thumb_hash) + assert stored == b"jpeg-bytes-for-https://image.tmdb.org/t/p/w500/poster.jpg" diff --git a/packages/meshbay-node/tests/test_rename_reenrichment.py b/packages/meshbay-node/tests/test_rename_reenrichment.py new file mode 100644 index 0000000..87e0d1a --- /dev/null +++ b/packages/meshbay-node/tests/test_rename_reenrichment.py @@ -0,0 +1,163 @@ +""" +Bug found live, 2026-08-24: an episode file first named in French (its +release folder mixed languages across seasons) was renamed by the operator +to match its English-named siblings — but kept showing as its own +separate poster-grid card, and its own row in Flat list, indefinitely. + +`_enriched_attempted` (daemon.py) exists so that enrichment's own +field-fill (duration/thumb_hash/display_title/... landing back via +`_on_enriched`) does not re-trigger itself forever — but it also silently +blocked the *new* filename from ever being title-parsed at all, since the +file's content (and so its id) is unchanged by a rename. A rename/move is +exactly the case `_reenrich_renamed_video_entries` exists to detect: same +id, but `name` or `path` differs from the version last broadcast. +""" + +import asyncio + +import pytest +from cryptography.hazmat.primitives.asymmetric.ed25519 import Ed25519PrivateKey + +from meshbay_common.crypto import generate_gek +from meshbay_node.config import Config, HubConfig, NodeConfig, GroupConfig, KeystoreConfig +from meshbay_node.daemon import NodeDaemon +from meshbay_node.indexer import DirectoryIndexer + +from conftest import one_root + +pytestmark = pytest.mark.asyncio + + +def _free_port() -> int: + import socket + with socket.socket() as s: + s.bind(("127.0.0.1", 0)) + return s.getsockname()[1] + + +class _StubRoster: + async def video_root(self, group_id): + return "shared" + + +class _SpyEnricher: + """Records which file ids were actually (re-)scheduled, without + needing a real ffmpeg/ffprobe pipeline for this test.""" + + def __init__(self): + self.spawned = [] + + def spawn(self, entry, file_path, on_done): + self.spawned.append(entry.id) + + async def _noop(): + return None + + return asyncio.ensure_future(_noop()) + + +async def test_a_renamed_file_gets_re_enriched(tmp_path): + group_id = "a" * 32 + shared = tmp_path / "shared" + shared.mkdir() + old_path = shared / "la-guerre-des-mondes-s03e02.mkv" + old_path.write_bytes(b"not a real video, just needs to be indexed as one") + + config = Config( + hub=HubConfig(url="http://localhost:9999", username="testuser"), + node=NodeConfig(quic_port=_free_port(), ui_port=_free_port()), + groups=[GroupConfig( + id=group_id, name="test-group", shared_dir=str(shared), + visibility="private", quic_port=29016, + )], + keystore=KeystoreConfig(path=tmp_path / "keystore.enc"), + data_dir=tmp_path / "data", + ) + daemon = NodeDaemon(config) + daemon._broadcast_coalesce_secs = 0.01 + daemon._enricher = _SpyEnricher() + daemon._roster = _StubRoster() + + indexer = DirectoryIndexer( + roots=one_root(shared), group_id=group_id, + sk_node=Ed25519PrivateKey.generate(), gek=generate_gek(), + # reconcile() below calls this itself — the initial scan doesn't + # (see test_startup_scan_enrichment.py), so that first broadcast is + # still triggered manually, matching _bg_scan's real sequence. + on_change=daemon._on_index_change) + await indexer.initial_scan() + await daemon._on_index_change(indexer) + await asyncio.sleep(0.05) + + entry = next(iter(indexer.index.entries)) + file_id = entry.id + assert daemon._enricher.spawned == [file_id], ( + "the file must be scheduled for enrichment once, under its original name") + + new_path = shared / "war-of-the-worlds-s03e02.mkv" + old_path.rename(new_path) + changed = await indexer.reconcile() + assert changed, "the rename must actually be picked up by reconcile()" + # The re-broadcast is a fire-and-forget task chained behind the + # coalescing timer, itself scheduling another fire-and-forget task — + # poll rather than guess a single sleep long enough for both hops. + for _ in range(30): + if len(daemon._enricher.spawned) >= 2: + break + await asyncio.sleep(0.02) + + renamed_entry = indexer.index.get_entry(file_id) + assert renamed_entry is not None + assert renamed_entry.name == "war-of-the-worlds-s03e02.mkv" + assert daemon._enricher.spawned == [file_id, file_id], ( + "a rename must re-schedule enrichment for the same file id — " + "_enriched_attempted must not permanently block the new filename " + "from ever being title-parsed") + + +async def test_an_unrelated_update_does_not_re_trigger_enrichment(tmp_path): + """ + The other half of the same fix: an update whose name/path did *not* + change (the ordinary case — enrichment's own field-fill, or the + reconcile sweep confirming a file unmodified) must not re-schedule + enrichment. Without this, `_on_enriched` merging a file's own results + back into the index would count as its own trigger and loop forever. + """ + group_id = "a" * 32 + shared = tmp_path / "shared" + shared.mkdir() + path = shared / "movie.mkv" + path.write_bytes(b"not a real video, just needs to be indexed as one") + + config = Config( + hub=HubConfig(url="http://localhost:9999", username="testuser"), + node=NodeConfig(quic_port=_free_port(), ui_port=_free_port()), + groups=[GroupConfig( + id=group_id, name="test-group", shared_dir=str(shared), + visibility="private", quic_port=29017, + )], + keystore=KeystoreConfig(path=tmp_path / "keystore.enc"), + data_dir=tmp_path / "data", + ) + daemon = NodeDaemon(config) + daemon._broadcast_coalesce_secs = 0.01 + daemon._enricher = _SpyEnricher() + daemon._roster = _StubRoster() + + indexer = DirectoryIndexer( + roots=one_root(shared), group_id=group_id, + sk_node=Ed25519PrivateKey.generate(), gek=generate_gek(), + on_change=daemon._on_index_change) + await indexer.initial_scan() + await daemon._on_index_change(indexer) + await asyncio.sleep(0.05) + + entry = next(iter(indexer.index.entries)) + assert daemon._enricher.spawned == [entry.id] + + # A reconcile pass that finds nothing changed at all — not even a + # rename — must not re-schedule anything. + changed = await indexer.reconcile() + await asyncio.sleep(0.05) + assert not changed + assert daemon._enricher.spawned == [entry.id] diff --git a/packages/meshbay-node/tests/test_season_and_search_requests.py b/packages/meshbay-node/tests/test_season_and_search_requests.py new file mode 100644 index 0000000..8ba55fb --- /dev/null +++ b/packages/meshbay-node/tests/test_season_and_search_requests.py @@ -0,0 +1,187 @@ +""" +`_do_season_meta_request` (per-season TMDB overview/poster/air_date, for the +season-tab view — docs/mediacenter.md §5.4's fix for a 3-season show whose +overview read as season-3-specific for every season) and +`_do_tmdb_search_request` (raw TMDB candidates for an operator correcting a +wrong automatic match). Neither is a signed admin op — see each handler's own +docstring for why — so these tests only exercise the read path, unlike +test_tmdb_override_policy.py. +""" + +import pytest + +from meshbay_common.protocol import MNP +from meshbay_node.media_cache import MediaCache +from meshbay_node.transport.webrtc_server import WebRTCPeerSession + +pytestmark = pytest.mark.asyncio + + +def _session(media_cache=None, tmdb_client=None) -> WebRTCPeerSession: + session = WebRTCPeerSession.__new__(WebRTCPeerSession) + session._ctx = {"media_cache": media_cache, "tmdb_client": tmdb_client} + session.sent = [] + session._send = session.sent.append + return session + + +@pytest.fixture +async def media_cache(tmp_path): + c = MediaCache(db_path=tmp_path / "media_cache.db") + await c.open() + yield c + await c.close() + + +class FakeTmdbClient: + def __init__(self, season_json=None): + self.season_json = season_json + self.tv_season_calls = [] + self.movie_search_calls = [] + self.tv_search_calls = [] + + @staticmethod + def poster_url(path): + return f"https://image.tmdb.org/t/p/w500{path}" + + async def fetch_image(self, url): + return b"jpeg-bytes-for-" + url.encode() + + async def tv_season(self, tmdb_id, season, language=None): + self.tv_season_calls.append((tmdb_id, season, language)) + return self.season_json + + async def search_movie_results(self, title): + self.movie_search_calls.append(title) + return [{"id": 111, "title": title, "release_date": "2019-05-01", "poster_path": "/m.jpg"}] + + async def search_tv_results(self, title): + self.tv_search_calls.append(title) + return [{"id": 222, "name": title, "first_air_date": "2021-03-01", "poster_path": "/t.jpg"}] + + +# ── season_meta_req ────────────────────────────────────────────────────────── + +async def test_season_meta_missing_tmdb_id_is_refused(): + session = _session() + await session._do_season_meta_request({"season": 1}) + assert session.sent == [{"type": "error", "detail": "Missing tmdb_id or season"}] + + +async def test_season_meta_non_int_season_is_refused(): + session = _session() + await session._do_season_meta_request({"tmdb_id": "42", "season": "1"}) + assert session.sent == [{"type": "error", "detail": "Missing tmdb_id or season"}] + + +async def test_season_meta_with_no_cache_or_client_reports_zero_confidence(): + session = _session(media_cache=None, tmdb_client=None) + await session._do_season_meta_request({"tmdb_id": "42", "season": 1}) + assert session.sent == [{ + "type": MNP.SEASON_META_RESP, "v": session.sent[0]["v"], + "tmdb_id": "42", "season": 1, "confidence": 0, + }] + + +async def test_season_meta_cache_hit_skips_the_tmdb_call(media_cache): + await media_cache.set_season_meta("42", 3, { + "name": "Season 3", "overview": "cached overview", "air_date": "2023-01-01", + "poster_path": "/cached.jpg", + }) + client = FakeTmdbClient() + session = _session(media_cache=media_cache, tmdb_client=client) + + await session._do_season_meta_request({"tmdb_id": "42", "season": 3}) + + assert client.tv_season_calls == [], "a cached season must not be re-fetched" + resp = session.sent[0] + assert resp["type"] == MNP.SEASON_META_RESP + assert resp["confidence"] == 1.0 + assert resp["overview"] == "cached overview" + + +async def test_season_meta_cache_miss_fetches_and_caches(media_cache): + client = FakeTmdbClient(season_json={ + "name": "Season 1", "overview": "fresh overview", "air_date": "2020-01-01", + "poster_path": "/fresh.jpg", + }) + session = _session(media_cache=media_cache, tmdb_client=client) + + await session._do_season_meta_request({"tmdb_id": "7", "season": 1}) + + assert client.tv_season_calls == [("7", 1, None)] + resp = session.sent[0] + assert resp["overview"] == "fresh overview" + assert resp["poster_thumb_hash"] is not None + cached = await media_cache.get_season_meta("7", 1) + assert cached["overview"] == "fresh overview", "a fetched season must be cached for next time" + + +async def test_season_meta_empty_overview_falls_back_to_english(media_cache): + async def tv_season(tmdb_id, season, language=None): + if language == "en-US": + return {"name": "S1", "overview": "English overview", "air_date": "2020-01-01", + "poster_path": "/p.jpg"} + return {"name": "S1", "overview": "", "air_date": "2020-01-01", "poster_path": "/p.jpg"} + + client = FakeTmdbClient() + client.tv_season = tv_season + session = _session(media_cache=media_cache, tmdb_client=client) + + await session._do_season_meta_request({"tmdb_id": "9", "season": 1}) + + assert session.sent[0]["overview"] == "English overview" + + +# ── tmdb_search_req ────────────────────────────────────────────────────────── + +async def test_search_missing_query_is_refused(): + session = _session() + await session._do_tmdb_search_request({"media_type": "movie"}) + assert session.sent == [{"type": "error", "detail": "Missing query or media_type"}] + + +async def test_search_bad_media_type_is_refused(): + session = _session() + await session._do_tmdb_search_request({"query": "war", "media_type": "album"}) + assert session.sent == [{"type": "error", "detail": "Missing query or media_type"}] + + +async def test_search_with_no_cache_or_client_returns_empty_results(): + session = _session(media_cache=None, tmdb_client=None) + await session._do_tmdb_search_request({"query": "war", "media_type": "tv"}) + assert session.sent == [{ + "type": MNP.TMDB_SEARCH_RESP, "v": session.sent[0]["v"], + "query": "war", "media_type": "tv", "results": [], + }] + + +async def test_search_movie_calls_movie_search_and_echoes_media_type(media_cache): + client = FakeTmdbClient() + session = _session(media_cache=media_cache, tmdb_client=client) + + await session._do_tmdb_search_request({"query": "War of the Worlds", "media_type": "movie"}) + + assert client.movie_search_calls == ["War of the Worlds"] + assert client.tv_search_calls == [] + resp = session.sent[0] + assert resp["type"] == MNP.TMDB_SEARCH_RESP + assert resp["media_type"] == "movie", ( + "media_type must be echoed back — otherwise a movie search and a tv " + "search for the same query are indistinguishable to the client's " + "keyed response matching (transport.js tmdb_search_resp handler)") + assert resp["results"] == [{ + "tmdb_id": "111", "title": "War of the Worlds", "year": "2019", + "poster_thumb_hash": resp["results"][0]["poster_thumb_hash"], + }] + + +async def test_search_tv_calls_tv_search(media_cache): + client = FakeTmdbClient() + session = _session(media_cache=media_cache, tmdb_client=client) + + await session._do_tmdb_search_request({"query": "War of the Worlds", "media_type": "tv"}) + + assert client.tv_search_calls == ["War of the Worlds"] + assert client.movie_search_calls == [] + assert session.sent[0]["media_type"] == "tv" diff --git a/packages/meshbay-node/tests/test_startup_scan_enrichment.py b/packages/meshbay-node/tests/test_startup_scan_enrichment.py new file mode 100644 index 0000000..cdadad9 --- /dev/null +++ b/packages/meshbay-node/tests/test_startup_scan_enrichment.py @@ -0,0 +1,92 @@ +""" +Regression: indexer.initial_scan() (run once at startup, daemon.py's +_bg_scan) never itself calls on_change — that predates the Videos app, and +every existing caller only cared about the scan finishing, not about +notifying anyone. Enrichment (duration/thumb_hash/display_title/...) hangs +entirely off on_change (daemon._broadcast_index_change). + +Without an explicit call to _on_index_change right after the startup scan, +a file already on disk at boot — the common case, an existing library — +would never get enriched at all: only a file added later, while the node +is already running (seen by the watchdog), would trigger it. Found live +against a real library after the first restart with this feature enabled. + +Enrichment only runs once a group has a video_root configured (a group +with none set gets no TMDB/thumbnail work at all, docs/mediacenter.md +§5.2/§10) — this test's fake roster reports the shared root itself as the +configured video_root, so the enrichment-scheduling behaviour under test +is exercised the same way a real operator's group would be. +""" + +import asyncio + +import pytest +from cryptography.hazmat.primitives import serialization +from cryptography.hazmat.primitives.asymmetric.ed25519 import Ed25519PrivateKey +from unittest.mock import MagicMock + +from meshbay_common.crypto import generate_gek +from meshbay_node.config import Config, HubConfig, NodeConfig, GroupConfig, KeystoreConfig +from meshbay_node.daemon import NodeDaemon +from meshbay_node.indexer import DirectoryIndexer +from meshbay_node.indexer.enrich import Enricher +from meshbay_node.media_cache import MediaCache + +from conftest import one_root + +pytestmark = pytest.mark.asyncio + + +def _free_port() -> int: + import socket + with socket.socket() as s: + s.bind(("127.0.0.1", 0)) + return s.getsockname()[1] + + +async def test_a_file_already_on_disk_at_startup_gets_enrichment_scheduled(tmp_path): + shared = tmp_path / "shared" + shared.mkdir() + (shared / "movie.mkv").write_bytes(b"not a real video, just needs to be indexed as one") + + config = Config( + hub=HubConfig(url="http://localhost:9999", username="testuser"), + node=NodeConfig(quic_port=_free_port(), ui_port=_free_port()), + groups=[GroupConfig( + id="a" * 32, name="test-group", shared_dir=str(shared), + visibility="private", quic_port=29012, + )], + keystore=KeystoreConfig(path=tmp_path / "keystore.enc"), + data_dir=tmp_path / "data", + ) + class _StubRoster: + async def video_root(self, group_id): + return "shared" # the root itself, i.e. "enrich the whole thing" + + daemon = NodeDaemon(config) + daemon._broadcast_coalesce_secs = 0.01 # real value would make this test wait 0.5s + daemon._media_cache = MediaCache(db_path=tmp_path / "media_cache.db") + await daemon._media_cache.open() + daemon._enricher = Enricher(daemon._media_cache) + daemon._roster = _StubRoster() + + sk_node = Ed25519PrivateKey.generate() + indexer = DirectoryIndexer( + roots=one_root(shared), group_id="a" * 32, + sk_node=sk_node, gek=generate_gek()) + + # Mirrors _bg_scan's actual sequence in daemon.py. + await indexer.initial_scan() + assert not daemon._enriched_attempted, ( + "nothing should be scheduled before _on_index_change is ever called") + + await daemon._on_index_change(indexer) + await asyncio.sleep(0.05) # let the coalescing timer fire _broadcast_index_change + + entry = next(iter(indexer.index.entries)) + assert entry.id in daemon._enriched_attempted, ( + "a file already on disk at startup must get enrichment scheduled the " + "first time its group's index is broadcast, not only on a later " + "watchdog-detected change to it") + + await daemon._media_cache.close() diff --git a/packages/meshbay-node/tests/test_stream_audio_transcode.py b/packages/meshbay-node/tests/test_stream_audio_transcode.py index 5b1fc46..dde0df4 100644 --- a/packages/meshbay-node/tests/test_stream_audio_transcode.py +++ b/packages/meshbay-node/tests/test_stream_audio_transcode.py @@ -149,13 +149,14 @@ async def test_probe_video_reports_aac_regardless_of_source_audio_codec(tmp_path clip = tmp_path / "clip.mkv" _make_clip(clip, acodec="eac3", channels=6) - codec, duration, has_audio = await _probe_video(str(clip)) + codec, duration, has_audio, width, height = await _probe_video(str(clip)) assert has_audio is True assert duration > 0 assert codec is not None assert "eac3" not in codec and "ec-3" not in codec assert "mp4a.40.2" in codec + assert (width, height) == (320, 240) async def test_probe_video_handles_no_audio_track(tmp_path): @@ -167,8 +168,9 @@ async def test_probe_video_handles_no_audio_track(tmp_path): check=True, capture_output=True, ) - codec, duration, has_audio = await _probe_video(str(clip)) + codec, duration, has_audio, width, height = await _probe_video(str(clip)) assert has_audio is False assert codec is not None and "," not in codec, \ "no audio track must not produce a dangling ',' or a fake audio codec" + assert (width, height) == (320, 240) diff --git a/packages/meshbay-node/tests/test_title_parse.py b/packages/meshbay-node/tests/test_title_parse.py new file mode 100644 index 0000000..1a277f2 --- /dev/null +++ b/packages/meshbay-node/tests/test_title_parse.py @@ -0,0 +1,129 @@ +""" +Tests for indexer/title_parse.py — synthetic filenames only, one per +docs/mediacenter.md §3.3/§3.4 rule. The real ~1950-file library validation +is a manual acceptance step (§11), not something this repo's corpus holds. +""" + +from meshbay_node.indexer.title_parse import ( + ParsedName, + naive_title, + parse_episode_filename, + parse_movie_filename, + season_from_folder_name, + sequel_variants, +) + + +# ── §3.3 row: plain, well-formed movie filename ────────────────────────────── + +def test_plain_movie_filename_parses_confidently(): + r = parse_movie_filename("The.Great.Adventure.2015.1080p.BluRay.x264.mkv") + assert r.display_title == "The Great Adventure" + assert r.year == 2015 + assert r.confidence is True + + +# ── §3.3 row 1: real title in alternative_title ────────────────────────────── + +def test_franchise_numeric_code_exposes_alternative_title(): + r = parse_movie_filename("Franchise.007.-.The.Real.Subtitle.1999.720p.mkv") + assert r.year == 1999 + # guessit lands the franchise fragment in title and the real subtitle in + # alternative_title — both must be surfaced so a caller (tmdb.py) can + # try either, per the design doc's "search both fields" fix. + assert r.alt_title is not None + + +# ── §3.3 row 2: hyphenated title split before a parenthesized year ─────────── + +def test_naive_title_normalizes_hyphens_dots_and_underscores(): + nt = naive_title("Hero-Name.(2002).DVDRip.XviD-GROUP.avi") + assert "-" not in nt + assert "." not in nt + assert "_" not in nt + assert "2002" not in nt # the year itself is stripped, not just its parens + assert "Hero" in nt and "Name" in nt + + +# ── §3.3 row 3: French edition vocabulary stuck to the title ───────────────── + +def test_french_edition_phrases_are_stripped(): + r = parse_movie_filename("Some.Movie.Version.Longue.2010.mkv") + assert "version" not in (r.display_title or "").lower() + assert "longue" not in (r.display_title or "").lower() + + r2 = parse_movie_filename("Autre.Film.Remasterise.1998.mkv") + assert "remasteris" not in (r2.display_title or "").lower() + + +# ── §3.3 row 4: trailing sequel digit / Roman numeral ──────────────────────── + +def test_sequel_variants_strips_digit_and_offers_roman_numeral(): + variants = sequel_variants("Some Sequel 2") + assert "Some Sequel" in variants + assert "Some Sequel II" in variants + + +def test_sequel_variants_empty_when_no_trailing_digit(): + assert sequel_variants("Some Movie") == [] + + +# ── §3.3 row 6: no usable title at all ─────────────────────────────────────── + +def test_low_confidence_when_no_title_or_year(): + r = parse_movie_filename("abc123.mkv") + assert r.confidence is False + # the mandated fallback is always available regardless + assert r.naive_title == "abc123" + + +# ── §3.2/§3.4: episode filename with no show name at all ───────────────────── + +def test_episode_only_filename_has_no_title_but_has_season_episode(): + r = parse_episode_filename("S08E02.SUBFRENCH.720p.mkv") + assert r.display_title is None + assert r.season == 8 + assert r.episode == 2 + # confidence is False (no title yet) — the indexer must supply one from + # a representative sibling filename in the same folder, per §3.4. + assert r.confidence is False + + +def test_episode_filename_with_show_name_parses_confidently(): + r = parse_episode_filename("Some.Show.Name.S02E05.720p.WEB.mkv") + assert r.display_title == "Some Show Name" + assert r.season == 2 + assert r.episode == 5 + assert r.confidence is True + + +# ── §3.4: season-like ancestor folders, including non-English vocabulary ──── + +def test_season_folder_english(): + assert season_from_folder_name("Season 2") == 2 + + +def test_season_folder_french_word(): + assert season_from_folder_name("Saison 3") == 3 + + +def test_season_folder_roman_numeral(): + assert season_from_folder_name("Saison IV") == 4 + + +def test_specials_folder_maps_to_season_zero(): + assert season_from_folder_name("Specials") == 0 + assert season_from_folder_name("Bonus") == 0 + assert season_from_folder_name("Extras") == 0 + + +def test_non_season_folder_name_returns_none(): + assert season_from_folder_name("Some Show Name") is None + + +def test_parsed_name_is_a_plain_dataclass(): + # sanity: constructible with just the one required field, per the + # "None => caller must supply from elsewhere" contract. + p = ParsedName(display_title=None) + assert p.confidence is False + assert p.naive_title == "" diff --git a/packages/meshbay-node/tests/test_tmdb.py b/packages/meshbay-node/tests/test_tmdb.py new file mode 100644 index 0000000..4500feb --- /dev/null +++ b/packages/meshbay-node/tests/test_tmdb.py @@ -0,0 +1,166 @@ +"""Tests for tmdb.py against a mocked httpx transport — no live network in CI.""" + +import httpx +import pytest + +from meshbay_node.tmdb import TmdbClient + + +class FakeRoster: + def __init__(self, enabled: bool = True, token: str | None = "fake-token", + language: str | None = None): + self._enabled = enabled + self._token = token + self._language = language + + async def tmdb_config(self): + return self._enabled, self._token, self._language + + +def _handler(response_map): + def handle(request: httpx.Request) -> httpx.Response: + path = request.url.path + for prefix, body in response_map.items(): + if path.endswith(prefix): + return httpx.Response(200, json=body) + return httpx.Response(404, json={"results": []}) + return handle + + +@pytest.mark.asyncio +async def test_search_movie_returns_top_result_and_confidence(): + body = {"results": [{"id": 42, "title": "The Great Adventure", "release_date": "2015-01-01"}]} + client = TmdbClient( + roster=FakeRoster(), + transport=httpx.MockTransport(_handler({"search/movie": body})), + ) + result, ratio = await client.search_movie("The Great Adventure", 2015) + + assert result is not None + assert result["id"] == 42 + assert ratio > 0.9 + await client.close() + + +@pytest.mark.asyncio +async def test_search_tv_returns_top_result(): + body = {"results": [{"id": 7, "name": "Some Show"}]} + client = TmdbClient( + roster=FakeRoster(), + transport=httpx.MockTransport(_handler({"search/tv": body})), + ) + result, ratio = await client.search_tv("Some Show") + + assert result is not None and result["id"] == 7 + assert ratio > 0.9 + await client.close() + + +@pytest.mark.asyncio +async def test_configured_language_is_sent_to_tmdb(): + captured = {} + + def handle(request: httpx.Request) -> httpx.Response: + captured["language"] = request.url.params.get("language") + return httpx.Response(200, json={"results": []}) + + client = TmdbClient( + roster=FakeRoster(language="fr-FR"), + transport=httpx.MockTransport(handle), + ) + await client.search_movie("Anything") + + assert captured["language"] == "fr-FR" + await client.close() + + +@pytest.mark.asyncio +async def test_no_language_configured_omits_the_param(): + captured = {} + + def handle(request: httpx.Request) -> httpx.Response: + captured["has_language"] = "language" in request.url.params + return httpx.Response(200, json={"results": []}) + + client = TmdbClient( + roster=FakeRoster(language=None), + transport=httpx.MockTransport(handle), + ) + await client.search_movie("Anything") + + assert captured["has_language"] is False + await client.close() + + +@pytest.mark.asyncio +async def test_no_results_returns_none_and_zero_confidence(): + client = TmdbClient( + roster=FakeRoster(), + transport=httpx.MockTransport(_handler({"search/movie": {"results": []}})), + ) + result, ratio = await client.search_movie("Nonexistent Obscure Title") + + assert result is None + assert ratio == 0.0 + await client.close() + + +@pytest.mark.asyncio +async def test_disabled_via_roster_setting_makes_no_request(): + calls = [] + + def handle(request: httpx.Request) -> httpx.Response: + calls.append(request) + return httpx.Response(200, json={"results": []}) + + client = TmdbClient( + roster=FakeRoster(enabled=False), + transport=httpx.MockTransport(handle), + ) + result, ratio = await client.search_movie("Anything") + + assert result is None + assert calls == [] # confirms the disabled check short-circuits before any request + await client.close() + + +@pytest.mark.asyncio +async def test_no_token_resolvable_makes_no_request(monkeypatch): + monkeypatch.delenv("MESHBAY_TMDB_DEFAULT_TOKEN", raising=False) + calls = [] + + def handle(request: httpx.Request) -> httpx.Response: + calls.append(request) + return httpx.Response(200, json={"results": []}) + + client = TmdbClient( + roster=FakeRoster(enabled=True, token=None), + transport=httpx.MockTransport(handle), + ) + result, ratio = await client.search_movie("Anything") + + assert result is None + assert calls == [] + await client.close() + + +@pytest.mark.asyncio +async def test_http_error_returns_none_gracefully(): + def handle(request: httpx.Request) -> httpx.Response: + return httpx.Response(500, json={"status_message": "server error"}) + + client = TmdbClient( + roster=FakeRoster(), + transport=httpx.MockTransport(handle), + ) + result, ratio = await client.search_movie("Anything") + + assert result is None + assert ratio == 0.0 + await client.close() + + +@pytest.mark.asyncio +async def test_poster_url_builds_full_url(): + assert TmdbClient.poster_url("/abc123.jpg") == "https://image.tmdb.org/t/p/w500/abc123.jpg" + assert TmdbClient.poster_url(None) is None diff --git a/packages/meshbay-node/tests/test_tmdb_config_policy.py b/packages/meshbay-node/tests/test_tmdb_config_policy.py new file mode 100644 index 0000000..29ef54c --- /dev/null +++ b/packages/meshbay-node/tests/test_tmdb_config_policy.py @@ -0,0 +1,230 @@ +""" +The operator decides whether the node calls TMDB at all, and whether it uses +a custom API token — docs/mediacenter.md §5.5. Same shape as +test_apps_enabled_policy.py/test_scan_settings_policy.py: a signed operator +instruction, node-wide (group_id="") rather than per-group, stored via +roster.py's group_settings table. + +Specific to this one: the subject signed/audited must never contain the +token itself (it would end up in the audit log in plaintext) — only whether +one was supplied travels there. The token itself only ever travels in +`payload`, which is node-side context never re-sent or re-verified from the +wire (see _issue_admin_challenge's docstring). +""" + +from pathlib import Path + +import pytest + +from meshbay_common.adminop import OP_TMDB_CONFIG +from meshbay_node.indexer.group_index import GroupIndex +from meshbay_node.roster import Roster +from meshbay_node.transport.webrtc_server import WebRTCPeerSession +from cryptography.hazmat.primitives.asymmetric.ed25519 import Ed25519PrivateKey + +from conftest import one_root + +pytestmark = pytest.mark.asyncio + + +def _session(tmp_path: Path, user_id: str, *, operator: str | None = None) -> WebRTCPeerSession: + shared_root = tmp_path / "shared" + shared_root.mkdir(exist_ok=True) + index = GroupIndex(group_id="g" * 32, sk_node=Ed25519PrivateKey.generate()) + ctx = { + "roots": one_root(shared_root), + "index": index, + "sk_node": index.sk_node, + "node_user_id": operator, + } + session = WebRTCPeerSession.__new__(WebRTCPeerSession) + session._ctx = ctx + session._group_id = None + session._user_id = user_id + session._pk_user = "" + session.sent = [] + session._send = session.sent.append + session._audit = lambda *a, **k: None + return session + + +def _fake_challenge(issued: list): + return lambda op, subject, payload=None, group_id=None: issued.append( + (op, subject, payload, group_id)) + + +# ── Refused before a challenge is even issued ─────────────────────────────── + +async def test_missing_enabled_is_refused(tmp_path): + session = _session(tmp_path, "op", operator="op") + session._has_admin_authority = lambda: True + issued = [] + session._issue_admin_challenge = _fake_challenge(issued) + + session._do_tmdb_config({}) + + assert not issued + assert [m for m in session.sent if m.get("type") == "error"] + + +async def test_non_bool_enabled_is_refused(tmp_path): + session = _session(tmp_path, "op", operator="op") + session._has_admin_authority = lambda: True + issued = [] + session._issue_admin_challenge = _fake_challenge(issued) + + session._do_tmdb_config({"enabled": "yes"}) + + assert not issued + assert [m for m in session.sent if m.get("type") == "error"] + + +async def test_non_string_token_is_refused(tmp_path): + session = _session(tmp_path, "op", operator="op") + session._has_admin_authority = lambda: True + issued = [] + session._issue_admin_challenge = _fake_challenge(issued) + + session._do_tmdb_config({"enabled": True, "token": 12345}) + + assert not issued + assert [m for m in session.sent if m.get("type") == "error"] + + +async def test_a_request_with_nobody_to_authorize_it_is_refused(tmp_path): + session = _session(tmp_path, "member-1", operator="the-operator") + session._has_admin_authority = lambda: False + + session._do_tmdb_config({"enabled": False}) + + assert [m for m in session.sent if m.get("type") == "error"] + + +# ── Who may change it, and what gets signed ───────────────────────────────── + +async def test_changing_it_needs_a_signature(tmp_path): + session = _session(tmp_path, "op", operator="op") + session._has_admin_authority = lambda: True + issued = [] + session._issue_admin_challenge = _fake_challenge(issued) + + session._do_tmdb_config({"enabled": True}) + + assert len(issued) == 1 + op, subject, payload, group_id = issued[0] + assert op == OP_TMDB_CONFIG + assert group_id == "", "node-wide, like group_attach/group_detach — not tied to self._group_id" + + +async def test_the_token_itself_never_appears_in_the_signed_subject(tmp_path): + """The subject is what gets audited (self._audit(pending['subject'])) — + a secret must never end up there in plaintext.""" + session = _session(tmp_path, "op", operator="op") + session._has_admin_authority = lambda: True + issued = [] + session._issue_admin_challenge = _fake_challenge(issued) + + secret = "sk-super-secret-tmdb-token" + session._do_tmdb_config({"enabled": True, "token": secret}) + + _, subject, payload, _ = issued[0] + assert secret not in subject + assert payload["token"] == secret, "the real value still has to reach the exec step somehow" + + +async def test_subject_reflects_enabled_and_whether_a_token_was_supplied(tmp_path): + session = _session(tmp_path, "op", operator="op") + session._has_admin_authority = lambda: True + issued = [] + session._issue_admin_challenge = _fake_challenge(issued) + + session._do_tmdb_config({"enabled": False, "token": "x"}) + + _, subject, _, _ = issued[0] + assert subject == "enabled=False,custom_token=yes,language=default" + + +async def test_subject_says_no_custom_token_when_none_given(tmp_path): + session = _session(tmp_path, "op", operator="op") + session._has_admin_authority = lambda: True + issued = [] + session._issue_admin_challenge = _fake_challenge(issued) + + session._do_tmdb_config({"enabled": True}) + + _, subject, _, _ = issued[0] + assert subject == "enabled=True,custom_token=no,language=default" + + +async def test_subject_reflects_a_configured_language(tmp_path): + session = _session(tmp_path, "op", operator="op") + session._has_admin_authority = lambda: True + issued = [] + session._issue_admin_challenge = _fake_challenge(issued) + + session._do_tmdb_config({"enabled": True, "language": "fr-FR"}) + + _, subject, payload, _ = issued[0] + assert subject == "enabled=True,custom_token=no,language=fr-FR" + assert payload["language"] == "fr-FR" + + +async def test_non_string_language_is_refused(tmp_path): + session = _session(tmp_path, "op", operator="op") + session._has_admin_authority = lambda: True + issued = [] + session._issue_admin_challenge = _fake_challenge(issued) + + session._do_tmdb_config({"enabled": True, "language": 42}) + + assert not issued + assert [m for m in session.sent if m.get("type") == "error"] + + +# ── Where it is stored ────────────────────────────────────────────────────── + +async def test_the_setting_lives_on_the_node_and_survives_a_restart(tmp_path): + roster = Roster(db_path=tmp_path / "roster.db") + await roster.open() + try: + enabled, token, language = await roster.tmdb_config() + assert (enabled, token, language) == (True, None, None), ( + "absent must mean on, with the shipped default token, TMDB's own default language") + await roster.set_tmdb_config(True, "my-custom-token", "fr-FR", set_by="op") + enabled, token, language = await roster.tmdb_config() + assert (enabled, token, language) == (True, "my-custom-token", "fr-FR") + finally: + await roster.close() + + reopened = Roster(db_path=tmp_path / "roster.db") + await reopened.open() + try: + assert await reopened.tmdb_config() == (True, "my-custom-token", "fr-FR") + finally: + await reopened.close() + + +async def test_clearing_the_token_reverts_to_the_default(tmp_path): + roster = Roster(db_path=tmp_path / "roster.db") + await roster.open() + try: + await roster.set_tmdb_config(True, "a-token", set_by="op") + assert (await roster.tmdb_config())[1] == "a-token" + + await roster.set_tmdb_config(True, "", set_by="op") + enabled, token, language = await roster.tmdb_config() + assert token is None, "an explicit empty string clears the custom token" + finally: + await roster.close() + + +async def test_omitting_the_token_leaves_it_unchanged(tmp_path): + roster = Roster(db_path=tmp_path / "roster.db") + await roster.open() + try: + await roster.set_tmdb_config(True, "a-token", set_by="op") + await roster.set_tmdb_config(False, None, set_by="op") + enabled, token, language = await roster.tmdb_config() + assert (enabled, token) == (False, "a-token") + finally: + await roster.close() diff --git a/packages/meshbay-node/tests/test_tmdb_language_fallback.py b/packages/meshbay-node/tests/test_tmdb_language_fallback.py new file mode 100644 index 0000000..1a85531 --- /dev/null +++ b/packages/meshbay-node/tests/test_tmdb_language_fallback.py @@ -0,0 +1,98 @@ +""" +TMDB doesn't fall back server-side for a field with no translation in the +requested language — it returns "" (or an empty list) for that one field, +not the English text, confirmed live against a real French query. The TMDB +website itself covers exactly this gap client-side by falling back to +English per field; `_tmdb_build_meta` (webrtc_server.py) mirrors that, +rather than discarding an otherwise-good localized response over one empty +field, or silently showing a blank overview/poster. +""" + +import pytest + +from meshbay_node.transport.webrtc_server import WebRTCPeerSession + +pytestmark = pytest.mark.asyncio + + +class FakeTmdbClient: + def __init__(self, localized: dict, english: dict, credits: dict | None = None): + self._localized = localized + self._english = english + self._credits = credits or {"cast": [], "crew": []} + self.calls: list[tuple[str, str | None]] = [] + + async def movie_details(self, tmdb_id, language=None): + self.calls.append(("movie_details", language)) + return self._english if language == "en-US" else self._localized + + async def tv_details(self, tmdb_id, language=None): + self.calls.append(("tv_details", language)) + return self._english if language == "en-US" else self._localized + + async def movie_credits(self, tmdb_id): + return self._credits + + async def tv_credits(self, tmdb_id): + return self._credits + + +async def test_empty_overview_falls_back_to_english_but_keeps_localized_poster(): + localized = { + "title": "OVNI(s)", "original_title": "OVNI(s)", + "overview": "", # no French translation on TMDB for this field + "poster_path": "/fr-poster.jpg", "backdrop_path": "/fr-backdrop.jpg", + "genres": [{"name": "Comédie"}], "vote_average": 7.2, + "first_air_date": "2016-01-01", + } + english = { + "title": "UFOs", "original_title": "OVNI(s)", + "overview": "A real English overview.", + "poster_path": "/en-poster.jpg", "backdrop_path": "/en-backdrop.jpg", + "genres": [{"name": "Comedy"}], "vote_average": 7.2, + "first_air_date": "2016-01-01", + } + client = FakeTmdbClient(localized, english) + + meta = await WebRTCPeerSession._tmdb_build_meta(client, "108353", "tv", {"id": 108353}) + + assert meta["overview"] == "A real English overview.", ( + "the empty localized field must fall back to the English value") + assert meta["poster_path"] == "/fr-poster.jpg", ( + "a non-empty localized field must NOT be overwritten by the English fallback" + ) + assert meta["genres"] == ["Comédie"], "localized genres were present — kept as-is" + assert ("tv_details", "en-US") in client.calls, "the fallback fetch must actually happen" + + +async def test_fully_populated_localized_response_never_triggers_a_fallback_call(): + localized = { + "title": "OVNI(s)", "overview": "Un résumé complet en français.", + "poster_path": "/fr-poster.jpg", "genres": [{"name": "Comédie"}], + "vote_average": 7.2, + } + client = FakeTmdbClient(localized, english={"overview": "should never be used"}) + + meta = await WebRTCPeerSession._tmdb_build_meta(client, "108353", "tv", {"id": 108353}) + + assert meta["overview"] == "Un résumé complet en français." + assert client.calls == [("tv_details", None)], ( + "a fully translated response must not cost a second TMDB request") + + +async def test_completely_untranslated_response_falls_back_entirely(): + localized = {"overview": "", "poster_path": None, "genres": []} + english = { + "title": "UFOs", "original_title": "OVNI(s)", + "overview": "A real English overview.", "poster_path": "/en-poster.jpg", + "genres": [{"name": "Comedy"}], "vote_average": 7.2, + "release_date": "2016-01-01", + } + client = FakeTmdbClient(localized, english) + + meta = await WebRTCPeerSession._tmdb_build_meta(client, "418517", "movie", {"id": 418517}) + + assert meta["overview"] == "A real English overview." + assert meta["poster_path"] == "/en-poster.jpg" + assert meta["genres"] == ["Comedy"] + assert meta["title"] == "UFOs" diff --git a/packages/meshbay-node/tests/test_tmdb_override_policy.py b/packages/meshbay-node/tests/test_tmdb_override_policy.py new file mode 100644 index 0000000..b8fa6f9 --- /dev/null +++ b/packages/meshbay-node/tests/test_tmdb_override_policy.py @@ -0,0 +1,172 @@ +""" +An operator correcting a wrong automatic TMDB match (found live: a real +show's search consistently matched a season-3-specific promotional TMDB +entry instead of the show itself). Signed like video_root/tmdb_config — +it changes what every member sees, node-wide (media_cache is shared, not +per-viewer) — and, once authorized, applies to every index entry sharing +the representative file's display_title, the same grouping the poster +grid itself uses (§3.4/§V6), not just the one file the operator happened +to be looking at. +""" + +import hashlib +from pathlib import Path + +import pytest +from cryptography.hazmat.primitives.asymmetric.ed25519 import Ed25519PrivateKey + +from meshbay_common.adminop import OP_TMDB_OVERRIDE +from meshbay_common.protocol import IndexEntry, MNP +from meshbay_node.indexer.group_index import GroupIndex +from meshbay_node.media_cache import MediaCache +from meshbay_node.transport.webrtc_server import WebRTCPeerSession + +from conftest import one_root + +pytestmark = pytest.mark.asyncio + + +def _session(tmp_path: Path, user_id: str, *, operator: str | None = None) -> WebRTCPeerSession: + shared_root = tmp_path / "shared" + shared_root.mkdir(exist_ok=True) + index = GroupIndex(group_id="g" * 32, sk_node=Ed25519PrivateKey.generate()) + ctx = { + "roots": one_root(shared_root), + "index": index, + "sk_node": index.sk_node, + "node_user_id": operator, + } + session = WebRTCPeerSession.__new__(WebRTCPeerSession) + session._ctx = ctx + session._group_id = None + session._user_id = user_id + session._pk_user = "" + session.sent = [] + session._send = session.sent.append + session._audit = lambda *a, **k: None + return session + + +def _entry(path: str, name: str, display_title: str) -> IndexEntry: + # A real id is a blake3 content hash; sha256 here is just a stand-in with + # the same property that matters for these tests — deterministic and + # effectively collision-free across the handful of entries a test builds. + # (`hash((path, name)) % 10` was tried here before and is NOT that: it's + # randomized per-process by PYTHONHASHSEED and collides constantly across + # only 10 possible values, silently dropping entries in GroupIndex's + # id-keyed dict.) + digest = hashlib.sha256(f"{path}/{name}".encode()).hexdigest() + return IndexEntry( + id=digest, name=name, path=path, + size=1, type="video", added_at=0, display_title=display_title, + season=1, episode=1, + ) + + +# ── Refused before a challenge is even issued ─────────────────────────────── + +async def test_missing_path_is_refused(tmp_path): + session = _session(tmp_path, "op", operator="op") + session._has_admin_authority = lambda: True + issued = [] + session._issue_admin_challenge = lambda op, subject: issued.append((op, subject)) + + session._do_tmdb_override({"tmdb_id": "123", "media_type": "tv"}) + + assert not issued + assert [m for m in session.sent if m.get("type") == "error"] + + +async def test_missing_tmdb_id_is_refused(tmp_path): + session = _session(tmp_path, "op", operator="op") + session._ctx["index"].add_entry(_entry("shared", "ep.mkv", "Show")) + session._has_admin_authority = lambda: True + issued = [] + session._issue_admin_challenge = lambda op, subject: issued.append((op, subject)) + + session._do_tmdb_override({"path": "shared", "media_type": "tv"}) + + assert not issued + assert [m for m in session.sent if m.get("type") == "error"] + + +async def test_unknown_path_is_refused(tmp_path): + session = _session(tmp_path, "op", operator="op") + session._has_admin_authority = lambda: True + issued = [] + session._issue_admin_challenge = lambda op, subject: issued.append((op, subject)) + + session._do_tmdb_override({"path": "nope", "tmdb_id": "123", "media_type": "tv"}) + + assert not issued + assert [m for m in session.sent if m.get("type") == "error"] + + +async def test_a_request_with_nobody_to_authorize_it_is_refused(tmp_path): + session = _session(tmp_path, "member-1", operator="the-operator") + session._ctx["index"].add_entry(_entry("shared", "ep.mkv", "Show")) + session._has_admin_authority = lambda: False + + session._do_tmdb_override({"path": "shared", "tmdb_id": "123", "media_type": "tv"}) + + assert [m for m in session.sent if m.get("type") == "error"] + + +async def test_a_valid_request_is_signed(tmp_path): + session = _session(tmp_path, "op", operator="op") + session._ctx["index"].add_entry(_entry("shared", "ep.mkv", "War of the Worlds")) + session._has_admin_authority = lambda: True + issued = [] + session._issue_admin_challenge = lambda op, subject: issued.append((op, subject)) + + session._do_tmdb_override({"path": "shared", "tmdb_id": "2255", "media_type": "tv"}) + + assert issued == [(OP_TMDB_OVERRIDE, + "path=shared,tmdb_id=2255,media_type=tv")] + + +# ── Applying the override ─────────────────────────────────────────────────── + +async def test_override_updates_every_entry_sharing_the_display_title(tmp_path): + session = _session(tmp_path, "op", operator="op") + index = session._ctx["index"] + s1 = _entry("shared/S1", "s01e01.mkv", "War of the Worlds") + s2 = _entry("shared/S2", "s02e01.mkv", "War of the Worlds") + s3 = _entry("shared/S3", "s03e02.mkv", "War of the Worlds") + other_show = _entry("shared/Other", "ep.mkv", "A Different Show") + for e in (s1, s2, s3, other_show): + index.add_entry(e) + + media_cache = MediaCache(db_path=tmp_path / "media_cache.db") + await media_cache.open() + try: + session._ctx["media_cache"] = media_cache + # Signature verification itself is exercised generically elsewhere + # (test_roster_pairing.py) — this test is about the policy once a + # signature is known good: which entries actually get updated, and + # who is told about it. + session._verify_admin_sig = lambda transcript, sig: _true() + peer = type("Peer", (), {"sent": []})() + peer._send = peer.sent.append + session._peer_registry = lambda: {"peer-1": peer} + + await session._admin_exec_tmdb_override( + {"subject": "path=shared/S1,tmdb_id=999,media_type=tv"}, + b"transcript", b"sig") + + for e in (s1, s2, s3): + assert await media_cache.get_file_tmdb(e.id) == ("999", "tv"), ( + "every entry sharing the representative file's display_title " + "must be corrected, not just the one the operator clicked on") + assert await media_cache.get_file_tmdb(other_show.id) is None, ( + "a different show's own match must be left alone") + # The ack is broadcast to other connected peers, never echoed onto + # the requester's own `sent` — see the loop in + # _admin_exec_tmdb_override, which sends via each peer's own _send. + assert [m for m in peer.sent if m.get("type") == MNP.TMDB_OVERRIDE_ACK] + finally: + await media_cache.close() + + +async def _true(): + return True diff --git a/packages/meshbay-node/tests/test_video_root_gates_enrichment.py b/packages/meshbay-node/tests/test_video_root_gates_enrichment.py new file mode 100644 index 0000000..df86a79 --- /dev/null +++ b/packages/meshbay-node/tests/test_video_root_gates_enrichment.py @@ -0,0 +1,153 @@ +""" +Videos-app enrichment (ffprobe/thumbnailing/TMDB, mediacenter.md §5.2/§10) +only ever runs for a group that has a video_root configured, and only for +files under it — see daemon.py's _enrich_new_video_entries. Burning TMDB's +rate limit and the node's CPU on an operator's whole shared index before +they have chosen which folder is actually their media library would be +real, ongoing cost for files never meant to be in the Videos app at all. + +Setting or changing the root (ops.set_video_root) fires a one-off sweep +(_enrich_video_root_now) of whatever it already contains: the ordinary +per-broadcast path only ever looks at files new since the last broadcast, +so anything already sitting in a folder before it became the video_root +would otherwise never be picked up. +""" + +import asyncio + +import pytest +from cryptography.hazmat.primitives.asymmetric.ed25519 import Ed25519PrivateKey + +from meshbay_common.crypto import generate_gek +from meshbay_node import ops +from meshbay_node.config import Config, HubConfig, NodeConfig, GroupConfig, KeystoreConfig +from meshbay_node.daemon import NodeDaemon +from meshbay_node.indexer import DirectoryIndexer +from meshbay_node.indexer.enrich import Enricher +from meshbay_node.media_cache import MediaCache +from meshbay_node.roster import Roster + +from conftest import one_root + +pytestmark = pytest.mark.asyncio + + +def _free_port() -> int: + import socket + with socket.socket() as s: + s.bind(("127.0.0.1", 0)) + return s.getsockname()[1] + + +async def _make_daemon(tmp_path, shared, group_id): + config = Config( + hub=HubConfig(url="http://localhost:9999", username="testuser"), + node=NodeConfig(quic_port=_free_port(), ui_port=_free_port()), + groups=[GroupConfig( + id=group_id, name="test-group", shared_dir=str(shared), + visibility="private", quic_port=29014, + )], + keystore=KeystoreConfig(path=tmp_path / "keystore.enc"), + data_dir=tmp_path / "data", + ) + daemon = NodeDaemon(config) + daemon._broadcast_coalesce_secs = 0.01 # real value would make these tests wait 0.5s + daemon._media_cache = MediaCache(db_path=tmp_path / "media_cache.db") + await daemon._media_cache.open() + daemon._enricher = Enricher(daemon._media_cache) + daemon._roster = Roster(db_path=tmp_path / "roster.db") + await daemon._roster.open() + return daemon + + +async def _teardown(daemon): + await daemon._media_cache.close() + await daemon._roster.close() + + +async def test_no_video_root_means_no_enrichment_at_all(tmp_path): + group_id = "a" * 32 + shared = tmp_path / "shared" + shared.mkdir() + (shared / "movie.mkv").write_bytes(b"not a real video, just needs to be indexed as one") + + daemon = await _make_daemon(tmp_path, shared, group_id) + try: + indexer = DirectoryIndexer( + roots=one_root(shared), group_id=group_id, + sk_node=Ed25519PrivateKey.generate(), gek=generate_gek()) + await indexer.initial_scan() + await daemon._on_index_change(indexer) + await asyncio.sleep(0.05) + + assert not daemon._enriched_attempted, ( + "a group with no video_root configured must not enrich anything, " + "not even fall back to the whole index") + finally: + await _teardown(daemon) + + +async def test_only_entries_under_the_configured_root_are_enriched(tmp_path): + group_id = "a" * 32 + shared = tmp_path / "shared" + shared.mkdir() + (shared / "Movies").mkdir() + # Distinct content: the index dedupes by content hash, and two files with + # the same bytes would otherwise collapse into a single entry. + (shared / "Movies" / "in-root.mkv").write_bytes(b"in-root content") + (shared / "outside.mkv").write_bytes(b"outside content") + + daemon = await _make_daemon(tmp_path, shared, group_id) + try: + await daemon._roster.set_video_root(group_id, "shared/Movies", set_by="op") + indexer = DirectoryIndexer( + roots=one_root(shared), group_id=group_id, + sk_node=Ed25519PrivateKey.generate(), gek=generate_gek()) + await indexer.initial_scan() + await daemon._on_index_change(indexer) + await asyncio.sleep(0.05) + + by_name = {e.name: e for e in indexer.index.entries} + assert by_name["in-root.mkv"].id in daemon._enriched_attempted + assert by_name["outside.mkv"].id not in daemon._enriched_attempted, ( + "a file outside the configured video_root must never be enriched") + finally: + await _teardown(daemon) + + +async def test_setting_the_video_root_sweeps_what_it_already_contains(tmp_path): + group_id = "a" * 32 + shared = tmp_path / "shared" + shared.mkdir() + (shared / "Movies").mkdir() + (shared / "Movies" / "already-there.mkv").write_bytes(b"x") + + daemon = await _make_daemon(tmp_path, shared, group_id) + try: + indexer = DirectoryIndexer( + roots=one_root(shared), group_id=group_id, + sk_node=Ed25519PrivateKey.generate(), gek=generate_gek()) + await indexer.initial_scan() + daemon._state["indexers"][group_id] = indexer + + # Broadcast once with nothing configured — nothing should be scheduled. + await daemon._on_index_change(indexer) + await asyncio.sleep(0.05) + assert not daemon._enriched_attempted + + # Now the operator points video_root at the folder that already held + # this file all along. + state = { + "roster": daemon._roster, + "groups_ctx": {group_id: {}}, + "enrich_video_root_fn": daemon._enrich_video_root_now, + } + await ops.set_video_root(state, group_id, "shared/Movies") + await asyncio.sleep(0.05) # let the fire-and-forget sweep actually run + + entry = next(iter(indexer.index.entries)) + assert entry.id in daemon._enriched_attempted, ( + "a file already sitting in the newly-chosen root must be picked " + "up by the sweep, not wait for some unrelated future change") + finally: + await _teardown(daemon) diff --git a/packages/meshbay-node/tests/test_video_root_policy.py b/packages/meshbay-node/tests/test_video_root_policy.py new file mode 100644 index 0000000..8cc1540 --- /dev/null +++ b/packages/meshbay-node/tests/test_video_root_policy.py @@ -0,0 +1,141 @@ +""" +Which folder (possibly a subfolder of a shared root) is the Videos app's +entry point for a group. Same shape as test_apps_enabled_policy.py: a +signed operator instruction, per-group (unlike tmdb_config, which is +node-wide), stored via roster.py's group_settings table. + +Specific to this one: a non-empty path must resolve to a real, readable +directory inside one of the group's own roots before a challenge is ever +issued — refusing a typo up front, the same way an empty apps set is +refused up front rather than round-tripped to the operator's browser. +""" + +from pathlib import Path + +import pytest + +from meshbay_common.adminop import OP_VIDEO_ROOT +from meshbay_node.indexer.group_index import GroupIndex +from meshbay_node.roster import Roster +from meshbay_node.transport.webrtc_server import WebRTCPeerSession +from cryptography.hazmat.primitives.asymmetric.ed25519 import Ed25519PrivateKey + +from conftest import one_root + +pytestmark = pytest.mark.asyncio + + +def _session(tmp_path: Path, user_id: str, *, operator: str | None = None) -> WebRTCPeerSession: + shared_root = tmp_path / "shared" + shared_root.mkdir(exist_ok=True) + (shared_root / "Movies").mkdir() + (shared_root / "Shows").mkdir() + index = GroupIndex(group_id="g" * 32, sk_node=Ed25519PrivateKey.generate()) + ctx = { + "roots": one_root(shared_root), + "index": index, + "sk_node": index.sk_node, + "node_user_id": operator, + } + session = WebRTCPeerSession.__new__(WebRTCPeerSession) + session._ctx = ctx + session._group_id = None + session._user_id = user_id + session._pk_user = "" + session.sent = [] + session._send = session.sent.append + session._audit = lambda *a, **k: None + return session + + +# ── Refused before a challenge is even issued ─────────────────────────────── + +async def test_missing_path_is_refused(tmp_path): + session = _session(tmp_path, "op", operator="op") + session._has_admin_authority = lambda: True + issued = [] + session._issue_admin_challenge = lambda op, subject: issued.append((op, subject)) + + session._do_video_root({}) + + assert not issued + assert [m for m in session.sent if m.get("type") == "error"] + + +async def test_a_nonexistent_folder_is_refused(tmp_path): + session = _session(tmp_path, "op", operator="op") + session._has_admin_authority = lambda: True + issued = [] + session._issue_admin_challenge = lambda op, subject: issued.append((op, subject)) + + session._do_video_root({"path": "shared/Nonexistent"}) + + assert not issued, "a mistyped path must be refused before a signature round trip" + assert [m for m in session.sent if m.get("type") == "error"] + + +async def test_path_traversal_is_refused(tmp_path): + session = _session(tmp_path, "op", operator="op") + session._has_admin_authority = lambda: True + issued = [] + session._issue_admin_challenge = lambda op, subject: issued.append((op, subject)) + + session._do_video_root({"path": "../../etc"}) + + assert not issued + assert [m for m in session.sent if m.get("type") == "error"] + + +async def test_a_request_with_nobody_to_authorize_it_is_refused(tmp_path): + session = _session(tmp_path, "member-1", operator="the-operator") + session._has_admin_authority = lambda: False + + session._do_video_root({"path": "shared/Movies"}) + + assert [m for m in session.sent if m.get("type") == "error"] + + +# ── Accepted cases ─────────────────────────────────────────────────────────── + +async def test_an_empty_path_is_always_accepted(tmp_path): + """Empty means 'the whole group index' — always valid, nothing to resolve.""" + session = _session(tmp_path, "op", operator="op") + session._has_admin_authority = lambda: True + issued = [] + session._issue_admin_challenge = lambda op, subject: issued.append((op, subject)) + + session._do_video_root({"path": ""}) + + assert issued == [(OP_VIDEO_ROOT, "")] + + +async def test_a_real_subfolder_is_accepted_and_signed(tmp_path): + session = _session(tmp_path, "op", operator="op") + session._has_admin_authority = lambda: True + issued = [] + session._issue_admin_challenge = lambda op, subject: issued.append((op, subject)) + + session._do_video_root({"path": "shared/Movies"}) + + assert issued == [(OP_VIDEO_ROOT, "shared/Movies")] + + +# ── Where it is stored ────────────────────────────────────────────────────── + +async def test_the_setting_lives_on_the_node_and_survives_a_restart(tmp_path): + roster = Roster(db_path=tmp_path / "roster.db") + await roster.open() + try: + assert await roster.video_root("g1") == "", "absent must mean the whole group index" + await roster.set_video_root("g1", "shared/Movies", set_by="op") + assert await roster.video_root("g1") == "shared/Movies" + finally: + await roster.close() + + reopened = Roster(db_path=tmp_path / "roster.db") + await reopened.open() + try: + assert await reopened.video_root("g1") == "shared/Movies" + assert await reopened.video_root("g2") == "", "one group's setting must not answer for another" + finally: + await reopened.close() diff --git a/packages/meshbay-node/tests/test_wizard_apps_endpoint.py b/packages/meshbay-node/tests/test_wizard_apps_endpoint.py new file mode 100644 index 0000000..10ab489 --- /dev/null +++ b/packages/meshbay-node/tests/test_wizard_apps_endpoint.py @@ -0,0 +1,76 @@ +""" +Create Group wizard: choosing which apps a brand-new group offers, before +the (potentially long) initial scan — see app.js's CreateGroupWizard. This +is a loopback-only, operator-authenticated endpoint (11.5.3), same shape as +the existing member-upload one: a thin adapter over `ops.set_enabled_apps`, +with only the validation `_do_apps_enabled` (the signed MNP front door) +already does client-side in the wizard, but worth enforcing at this front +door too since nothing else would. +""" + +from pathlib import Path + +import pytest +from cryptography.hazmat.primitives.asymmetric.ed25519 import Ed25519PrivateKey +from fastapi.testclient import TestClient + +from conftest import one_root +from meshbay_node.indexer.group_index import GroupIndex +from meshbay_node.roster import Roster +from meshbay_node.ui.app import create_ui_app + +pytestmark = pytest.mark.asyncio + + +async def _client(tmp_path: Path): + shared = tmp_path / "shared" + shared.mkdir(exist_ok=True) + index = GroupIndex(group_id="g" * 32, sk_node=Ed25519PrivateKey.generate()) + roster = Roster(db_path=tmp_path / "roster.db") + await roster.open() + state = { + "status": "running", + "groups_ctx": {"g" * 32: {"index": index, "roots": one_root(shared)}}, + "indexes": {"g" * 32: index}, + "roster": roster, + } + app = create_ui_app(state) + return TestClient(app), roster + + +async def test_narrowing_the_apps_persists_to_the_roster(tmp_path): + client, roster = await _client(tmp_path) + try: + resp = client.put(f"/api/groups/{'g' * 32}/apps", json={"apps": ["files", "video"]}) + assert resp.status_code == 200, resp.text + assert sorted(resp.json()["apps"]) == ["files", "video"] + assert sorted(await roster.enabled_apps("g" * 32)) == ["files", "video"] + finally: + await roster.close() + + +async def test_empty_apps_list_is_refused(tmp_path): + client, roster = await _client(tmp_path) + try: + resp = client.put(f"/api/groups/{'g' * 32}/apps", json={"apps": []}) + assert resp.status_code == 400 + finally: + await roster.close() + + +async def test_missing_apps_field_is_refused(tmp_path): + client, roster = await _client(tmp_path) + try: + resp = client.put(f"/api/groups/{'g' * 32}/apps", json={}) + assert resp.status_code == 400 + finally: + await roster.close() + + +async def test_unhosted_group_is_refused(tmp_path): + client, roster = await _client(tmp_path) + try: + resp = client.put("/api/groups/" + "z" * 32 + "/apps", json={"apps": ["files"]}) + assert resp.status_code == 404 + finally: + await roster.close() |