diff options
4 files changed, 744 insertions, 721 deletions
diff --git a/packages/meshbay-hub/src/meshbay_hub/static/transport.js b/packages/meshbay-hub/src/meshbay_hub/static/transport.js index fb698e8..817f182 100644 --- a/packages/meshbay-hub/src/meshbay_hub/static/transport.js +++ b/packages/meshbay-hub/src/meshbay_hub/static/transport.js @@ -1457,7 +1457,7 @@ class MeshBayTransport { * path names the *folder* a file is in (indexer.py's `_virtual_dir`), so * two files sharing a folder (any multi-episode season) would resolve to * whichever entry the node's index happened to return first (found live - * via the Music app's identical bug, 2026-08-25 — see webrtc_server.py's + * via the Music app's identical bug, 2026-08-25 — see apps/video_meta.py's * `_do_media_meta_request`). * `confidence: 0` (no tmdb_id, no fields) means no confident match — * the caller falls back to a thumbnail-only card (§4.1), not an error. @@ -1522,8 +1522,8 @@ class MeshBayTransport { * node-wide (media_cache is shared, not per-viewer) — an unsigned * override would let any member vandalize another show's metadata. * Applies to every file sharing the representative one's display_title, - * not just the file the operator happened to be looking at (webrtc_ - * server.py's _admin_exec_tmdb_override). Keyed by `fileId`, not a path + * not just the file the operator happened to be looking at (apps/ + * video_meta.py's _admin_exec_tmdb_override). Keyed by `fileId`, not a path * — same reasoning as fetchMediaMeta above. */ async overrideTmdbMatch(fileId, tmdbId, mediaType, signFn) { @@ -1573,7 +1573,7 @@ class MeshBayTransport { }); if (msg.type === 'error') throw new Error(msg.detail); if (msg.type === 'admin_challenge') { - // Must match the node's subject byte-for-byte (webrtc_server.py + // Must match the node's subject byte-for-byte (apps/video_meta.py // _do_tmdb_config) — the token itself is never part of the subject // (it would end up in the audit log in plaintext), only whether one // was supplied. The language is not a secret, so it appears as-is. @@ -1596,7 +1596,7 @@ class MeshBayTransport { }); if (msg.type === 'error') throw new Error(msg.detail); if (msg.type === 'admin_challenge') { - // Must match the node's subject byte-for-byte (webrtc_server.py + // Must match the node's subject byte-for-byte (apps/video_meta.py // _do_tmdb_enabled): Python's f"{bool}" is "True"/"False", not JS's // lowercase. const subject = enabled ? 'True' : 'False'; diff --git a/packages/meshbay-node/src/meshbay_node/transport/webrtc/apps/video_meta.py b/packages/meshbay-node/src/meshbay_node/transport/webrtc/apps/video_meta.py new file mode 100644 index 0000000..9222ad8 --- /dev/null +++ b/packages/meshbay-node/src/meshbay_node/transport/webrtc/apps/video_meta.py @@ -0,0 +1,728 @@ +"""What the node does for the Videos app's catalogue: TMDB matching, posters, +season pages, and the operator's TMDB switches.""" + +import logging +import time + +import blake3 +from meshbay_common import MNP_VERSION +from meshbay_common.adminop import ( + OP_TMDB_CONFIG, + OP_TMDB_ENABLED, + OP_TMDB_OVERRIDE, + OP_TMDB_REMATCH, +) +from meshbay_common.protocol import MNP + +from meshbay_node import ops + +log = logging.getLogger("meshbay_node.transport.webrtc_server") + + +# A free-text TMDB search spends the *operator's* credential, which is rated by +# TMDB and shared by everyone in the group: one member typing in the search box +# can exhaust what every other member's automatic matching depends on, and the +# operator is the one who has to notice. §6.5's rule is a bound and a named +# adversary in the same commit; this one arrived without either. +# +# Per member rather than per connection, unlike link previews: three tabs +# is one person, and a ceiling a tab can multiply is not a ceiling. Kept in the +# group context so it survives a reconnect, which is the other thing a per-session +# count cannot do. +# +# Generous next to what a person types — ten searches a minute is a search every +# six seconds, sustained — and small next to a loop. +_TMDB_SEARCH_WINDOW = 60.0 +_TMDB_SEARCH_PER_MEMBER = 10 +_TMDB_SEARCH_NODE = 30 + + +class VideoMetaMixin: + def _do_tmdb_config(self, msg: dict) -> None: + """ + Optionally set (or clear) a custom TMDB API token, and optionally set + the language TMDB is queried in (e.g. "fr-FR") — one for the whole + node, since both are one operator's shared credential/cache, not a + per-group concern (see _do_tmdb_enabled for the per-group on/off + switch). Signed like the rest: this changes outbound third-party + network traffic the node did not have before the Videos app + (docs/MESHBAY_DESIGN.md §9.7, §6.5) — an unsigned change would let any + member alter egress the operator never agreed to. + """ + 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"custom_token={'yes' if token else 'no'},language={language or 'default'}" + self._issue_admin_challenge( + OP_TMDB_CONFIG, subject, + payload={"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("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/the root ops/the + # per-group tmdb_enabled below). + notice = { + "type": MNP.TMDB_CONFIG_ACK, "v": MNP_VERSION, + "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_tmdb_enabled(self, msg: dict) -> None: + """ + Whether TMDB lookups run for this group at all. Per-group, unlike + tmdb_config's token/language — see ops.set_tmdb_enabled. Signed like + app_directories: it decides whether this group's members' Videos tab ever + makes outbound TMDB traffic. + """ + enabled = msg.get("enabled") + if not isinstance(enabled, bool): + self._send({"type": "error", "detail": "Missing or invalid 'enabled'"}) + return + if not self._has_admin_authority(): + self._send({"type": "error", "detail": "No authorized key for this"}) + return + self._issue_admin_challenge(OP_TMDB_ENABLED, str(enabled)) + + async def _admin_exec_tmdb_enabled( + self, pending: dict, transcript: bytes, sig: bytes, + ) -> None: + enabled = pending["subject"] == "True" + if not await self._verify_admin_sig(transcript, sig): + self._send({"type": "error", "detail": "Signature verification failed"}) + self._audit("admin_auth_failed", f"tmdb_enabled:{pending['subject']}") + return + try: + await self._run_op(ops.set_tmdb_enabled, self._group_id or "", enabled) + except ops.OpError as e: + self._send({"type": "error", "detail": e.message}) + return + self._audit("tmdb_enabled", pending["subject"]) + + notice = {"type": MNP.TMDB_ENABLED_ACK, "v": MNP_VERSION, "enabled": enabled} + for uid, session in list(self._peer_registry().items()): + try: + session._send(notice) + except Exception: + pass + + @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/MESHBAY_DESIGN.md §9.7), 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/MESHBAY_DESIGN.md §9.7: TMDB metadata for one file, resolved from + the group's index by its content id (root+relpath the client already + knows from index_sync/index_delta identify the entry; its own `id` + is what actually names one file — never a raw filesystem path off + the wire). + + Keyed by `file_id`, not `path`: `IndexEntry.path` is the *folder* a + file is in (indexer.py's `_virtual_dir`), so two files in the same + folder — any multi-episode season, routinely — shared the same + `.path`, and a lookup by it could silently resolve to the wrong + entry (found live via the Music app's identical bug, 2026-08-25). + """ + file_id = msg.get("file_id") + log.debug("media_meta_req file_id=%r", file_id) + if not isinstance(file_id, str) or not file_id: + self._send({"type": "error", "detail": "Missing file_id"}) + return + ctx = self._group_ctx() + entry = ctx["index"].get_entry(file_id) + 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") + # Per-group, not node-wide (docs/MESHBAY_DESIGN.md §9.7, 2026-08-24): + # treated exactly like "no client configured" — same silent, no-error + # degradation, since a member's Videos tab already has to handle "no + # TMDB match" as the ordinary case. + if media_cache is None or tmdb_client is None or not ctx.get("tmdb_enabled", True): + self._send({"type": MNP.MEDIA_META_RESP, "v": MNP_VERSION, + "file_id": file_id, "confidence": 0}) + return + + # A video the indexer has seen but not yet *enriched* has no + # display_title (enrich.py always sets one) and season/episode still + # None — so the movie/show split reads "movie" and would hand its raw + # filename to TMDB's movie search. During a slow initial scan with a + # browser on the Videos tab that is a storm of + # `search/movie?query=<raw filename>` (found live 2026-08-29, an + # 8-minute scan). While un-enriched we never *search*: we serve a + # cached match if there is one (§ below), else confidence 0 and the + # client refetches once the index delta carries the enriched fields. + enriched = bool(entry.display_title) + + 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: + cached_tmdb_id, cached_media_type = cached + # Serve the cached match when its kind still agrees with the + # entry's current classification — OR when the entry is not + # enriched yet: its season/episode aren't populated, so the + # movie/show split above is not meaningful, and the cached kind + # (set when this file WAS enriched) is the reliable one. This is + # what keeps a restart from re-querying TMDB for everything + # already resolved: the storm was an un-enriched show episode + # looking like a "movie" and treating its own valid "tv" match + # as stale. + # + # Once enriched, the strict `cached_media_type == media_type` + # check still stands: an enrichment fix that reclassifies a + # folder movie->tv must drop the stale movie-era match and + # re-resolve (found live — a Specials-folder fix left hundreds + # of files answering with their wrong-kind match forever). + if cached_media_type == media_type or not enriched: + media_type = cached_media_type + is_show = media_type == "tv" + tmdb_id = cached_tmdb_id + meta = await media_cache.get_tmdb_meta(tmdb_id, media_type) + + if meta is None: + if not enriched: + self._send({"type": MNP.MEDIA_META_RESP, "v": MNP_VERSION, + "file_id": file_id, "confidence": 0}) + return + 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, + "file_id": file_id, "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 file_id=%r: replying tmdb_id=%s poster=%s backdrop=%s", + file_id, tmdb_id, poster_thumb_hash, backdrop_thumb_hash) + + resp = { + "type": MNP.MEDIA_META_RESP, "v": MNP_VERSION, "file_id": file_id, + "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") + # Per-group, not node-wide (docs/MESHBAY_DESIGN.md §9.7, 2026-08-24) — + # same silent zero-confidence degradation as "no client configured". + if (media_cache is None or tmdb_client is None + or not self._group_ctx().get("tmdb_enabled", True)): + 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/MESHBAY_DESIGN.md §9.7, §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") + # Per-group, not node-wide (docs/MESHBAY_DESIGN.md §9.7, 2026-08-24) — + # same silent empty-results degradation as "no client configured": + # a member with TMDB off for this group sees the same "type it in + # yourself" affordance either way, never an error. + if (media_cache is None or tmdb_client is None + or not self._group_ctx().get("tmdb_enabled", True)): + self._send({"type": MNP.TMDB_SEARCH_RESP, "v": MNP_VERSION, + "query": query, "media_type": media_type, "results": []}) + return + + # Refused out loud, not as an empty result: "no matches" is what the + # client draws for an empty list, and telling somebody their film is + # unknown when the node simply declined to ask is a worse answer than + # the truth. `video-app.js`'s `runSearch` puts `detail` on screen. + if not self._tmdb_search_rate_ok(): + log.info("tmdb_search_req: rate-limited (user=%s)", (self._user_id or "")[:8]) + self._send({ + "type": "error", + "detail": "Too many searches in the last minute. This spends the " + "operator's search quota, which everyone in the group " + "shares — try again shortly.", + "code": "tmdb_search_rate_limited", + }) + 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 + app_directories/tmdb_config: it replaces what every member sees for a + show/movie, node-wide (media_cache is shared, not per-viewer). + + For a **show**, applied to every entry sharing the representative + file's display_title — the same grouping the poster grid uses + (§3.4/§V6) — so the correction sticks regardless of which episode a + future render picks as representative. For a **movie** it is applied + to that one file only: guessit gives a whole franchise the same + display_title, and a fan-out there corrected the wrong films (found + live, 2026-08-29). See `_admin_exec_tmdb_override`. + + Keyed by `file_id`, not `path` — see `_do_media_meta_request`'s + docstring for why a folder-level path cannot name one file. + """ + file_id = msg.get("file_id") + tmdb_id = msg.get("tmdb_id") + media_type = msg.get("media_type") + if not isinstance(file_id, str) or not file_id: + self._send({"type": "error", "detail": "Missing file_id"}) + 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(file_id) + 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"file_id={file_id},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(",")) + file_id, tmdb_id, media_type = fields["file_id"], fields["tmdb_id"], fields["media_type"] + + ctx = self._group_ctx() + entry = ctx["index"].get_entry(file_id) + 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 + # This only ever recorded the file->tmdb_id mapping, never the + # metadata tmdb_id names — _do_media_meta_request's cache check + # (entry, cache) both agree on media_type, so it trusted the + # mapping — but found nothing under this *new* id in tmdb_meta + # (nothing had ever fetched it), and silently fell through to a + # fresh search using the file's own title, exactly the one that + # produced the wrong match in the first place. Confirmed live: an + # override "stuck" for shows only because their own title happened + # to be enough for that fallback search to land on the right + # answer anyway, coincidentally — never because the override itself + # was actually being honored — and was invisible until a movie + # whose own title search kept landing on the same wrong result + # exposed it. Fetching and storing the real metadata up front is + # what makes the *override* the thing a later lookup finds. + tmdb_client = self._ctx.get("tmdb_client") + if tmdb_client is not None: + meta = await self._tmdb_build_meta(tmdb_client, tmdb_id, media_type, {}) + await media_cache.set_tmdb_meta(tmdb_id, media_type, meta) + # A show's episodes are many files that legitimately share one match, + # and which episode a render picks as representative rotates — so a + # show override fans out across every entry with the same + # display_title. A *movie* is one file: fanning out by display_title + # there is a bug — guessit gives every + # "<franchise> - <year> - <subtitle>.mkv" the same display_title, so + # "Fix match" on one entry rewrote the whole franchise (found live, + # 2026-08-29). Each corrected file is also marked as a manual + # override so ops.rematch_video / a rename never wipe it. + is_show = entry.season is not None and entry.episode is not None + if is_show: + 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] + else: + matched = [entry] + for e in matched: + await media_cache.set_file_tmdb(e.id, tmdb_id, media_type) + await media_cache.mark_tmdb_override(e.id) + self._audit("tmdb_override", subject) + + notice = {"type": MNP.TMDB_OVERRIDE_ACK, "v": MNP_VERSION, "file_id": file_id, + "tmdb_id": tmdb_id, "media_type": media_type} + for uid, session in list(self._peer_registry().items()): + try: + session._send(notice) + except Exception: + pass + + def _do_tmdb_rematch(self, msg: dict) -> None: + """ + An operator dropping one file's cached TMDB match so it re-resolves + with the current matcher (V13) — the one-click alternative to + the full search-and-pick "Fix match" flow, and reachable without + SSH (`meshbay-node video rematch` clears a whole group). Signed like + `tmdb_override`: `media_cache` is shared node-wide. + """ + file_id = msg.get("file_id") + if not isinstance(file_id, str) or not file_id: + self._send({"type": "error", "detail": "Missing file_id"}) + return + if not self._group_ctx()["index"].get_entry(file_id): + 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 + self._issue_admin_challenge(OP_TMDB_REMATCH, f"file_id={file_id}") + + async def _admin_exec_tmdb_rematch( + 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_rematch:{subject}") + return + file_id = dict(part.split("=", 1) for part in subject.split(","))["file_id"] + media_cache = self._ctx.get("media_cache") + if media_cache is None: + self._send({"type": "error", "detail": "Media cache not available"}) + return + await media_cache.drop_tmdb_match(file_id) + self._audit("tmdb_rematch", subject) + + notice = {"type": MNP.TMDB_REMATCH_ACK, "v": MNP_VERSION, "file_id": file_id} + 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): + """ + docs/MESHBAY_DESIGN.md §9.7's scored ladder — same shape for movies + and shows (V8). + TMDB's own top result is still trusted per query (§3.3's last row — + no local re-ranking of *its* list); what the ladder adds is that it + *scores every candidate query* and keeps the best, instead of + returning the first that merely clears 0.6. + + The bare parsed title is the weakest query: guessit drops a + "Volume 2", strips a real subtitle into `alternative_title`, renders + a sequel index where TMDB spells it differently, and a show's folder + name can carry a year or release-group noise. A wrong entry that + scored ~0.7 against that weak query — a same-year making-of + documentary, a franchise entry whose localized TMDB title *is* the + franchise name, a season-specific promo entry standing in for a + whole show — used to win outright before a stronger candidate was + ever tried. Found live (2026-08-29). + """ + from meshbay_node.indexer import title_parse + + if is_show: + title = entry.display_title or title_parse.naive_title(entry.name) + name_naive = title_parse.naive_title(entry.name) + year = title_parse.year_in(title) or title_parse.year_in(entry.name) + extra = [c for c in (name_naive, title_parse.clean_query(title)) + if c and c != title] + return await self._tmdb_ladder( + tmdb_client.search_tv, title, extra, year, strong_extra=False) + + parsed = title_parse.parse_movie_filename(entry.name) + title = entry.display_title or parsed.display_title or parsed.naive_title + strong = [c for c in (parsed.alt_title, *title_parse.sequel_variants(title)) if c] + extra = [c for c in (*strong, parsed.naive_title) if c and c != title] + return await self._tmdb_ladder( + tmdb_client.search_movie, title, extra, parsed.year, + strong_extra=bool(strong)) + + @staticmethod + async def _tmdb_ladder(search_fn, primary: str, extra: list[str], + year: int | None, strong_extra: bool): + """ + `search_fn(query, year) -> (result|None, ratio)`. Try `primary`, + return at once on a confident hit (ratio >= 0.85 — the common case, + one request). Otherwise score each `extra` candidate and keep the + best. `strong_extra` says whether `extra` contains anything more + specific than a punctuation-normalised restatement of `primary` + (an alternative_title, a sequel variant); when it does not and the + primary hit is already decent, the remaining calls are skipped + (V11 — they almost never win and cost a round trip each). + """ + def _year_of(res: dict) -> int | None: + d = str(res.get("release_date") or res.get("first_air_date") or "") + return int(d[:4]) if d[:4].isdigit() else None + + def _rescue(res: dict, r: float) -> float: + # A sub-0.6 hit whose result lands on the exact requested year: + # TMDB already year-filtered the search, so this is a hard + # corroboration that the low ratio is a localised/rearranged + # title, not a wrong entry. Never overrides a confident hit. + if r < 0.6 and year and _year_of(res) == year: + return max(r, 0.6) + return r + + result, ratio = await search_fn(primary, year) + if result is not None and ratio >= 0.85: + return result, ratio + best_result, best_score = (result, ratio) if result is not None else (None, 0.0) + if best_result is not None: + best_score = _rescue(best_result, best_score) + if best_score >= 0.6 and not strong_extra: + return best_result, best_score + + for candidate in extra: + if candidate == primary: + continue + r2, ratio2 = await search_fn(candidate, year) + if r2 is None and year: + # A year-filtered search that finds nothing: the year tag + # may be an edition/regional year TMDB doesn't carry. Retry + # the candidate unconstrained before dropping it. + r2, ratio2 = await search_fn(candidate, None) + if r2 is None: + continue + score2 = _rescue(r2, ratio2) + if score2 > best_score: + best_result, best_score = r2, score2 + if best_score >= 0.85: + break + return best_result, best_score + + @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) + else: + # A series has no single director, and `tv_credits`' crew is the + # aggregate one — routinely empty, and never a "Director" job. + # TMDB models the equivalent credit as `created_by` on the show + # itself, which is what its own page shows; several creators are + # ordinary, and they read as one line in the detail modal. + director = ", ".join( + name for name in + (c.get("name") for c in details.get("created_by") or []) + if name) or 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 _tmdb_search_rate_ok(self) -> bool: + """ + True when this search is within both the member's window and the node's; + records it when so, and trims both to the window on every call so neither + list can grow without bound. + + Both are checked because they answer different questions: the member's + keeps one person from spending everyone's quota, and the node's keeps a + group of them from doing it together. + """ + now = time.monotonic() + w = _TMDB_SEARCH_WINDOW + ctx = self._group_ctx() + by_member = ctx.setdefault("tmdb_search_hits", {}) + who = self._user_id or "" + mine = [t for t in by_member.get(who, []) if now - t < w] + node = [t for t in self._ctx.get("tmdb_search_hits_node", []) if now - t < w] + if len(mine) >= _TMDB_SEARCH_PER_MEMBER or len(node) >= _TMDB_SEARCH_NODE: + by_member[who] = mine + self._ctx["tmdb_search_hits_node"] = node + return False + mine.append(now) + node.append(now) + by_member[who] = mine + self._ctx["tmdb_search_hits_node"] = node + return True 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 9afb06c..5b7f700 100644 --- a/packages/meshbay-node/src/meshbay_node/transport/webrtc_server.py +++ b/packages/meshbay-node/src/meshbay_node/transport/webrtc_server.py @@ -148,6 +148,7 @@ from meshbay_node.roster import KIND_ACCOUNT, KIND_LINK from meshbay_node.transfers import TransferSlots from meshbay_node.transport.webrtc.apps.music import MusicMixin from meshbay_node.transport.webrtc.apps.subtitles import SubtitlesMixin +from meshbay_node.transport.webrtc.apps.video_meta import VideoMetaMixin from meshbay_node.transport.webrtc.channel import ( _REPLY_TO, _DataChannelBuffer, @@ -232,23 +233,6 @@ _LINK_PREVIEW_RATE_WINDOW = 60.0 _LINK_PREVIEW_RATE_PER_CONN = 15 _LINK_PREVIEW_RATE_NODE = 60 -# A free-text TMDB search spends the *operator's* credential, which is rated by -# TMDB and shared by everyone in the group: one member typing in the search box -# can exhaust what every other member's automatic matching depends on, and the -# operator is the one who has to notice. §6.5's rule is a bound and a named -# adversary in the same commit; this one arrived without either. -# -# Per member rather than per connection, unlike link previews above: three tabs -# is one person, and a ceiling a tab can multiply is not a ceiling. Kept in the -# group context so it survives a reconnect, which is the other thing a per-session -# count cannot do. -# -# Generous next to what a person types — ten searches a minute is a search every -# six seconds, sustained — and small next to a loop. -_TMDB_SEARCH_WINDOW = 60.0 -_TMDB_SEARCH_PER_MEMBER = 10 -_TMDB_SEARCH_NODE = 30 - # Chat limits. A message is a member-supplied write onto the operator's disk # (`chat.db`, where retention is a manual CLI command — §6.6), relayed from there # to every other connected member and turned into a notification for every member @@ -370,7 +354,7 @@ _WEBRTC_TRACE = os.environ.get("MESHBAY_WEBRTC_TRACE") == "1" _WEBRTC_TRACE_INTERVAL_S = 30.0 -class WebRTCPeerSession(MusicMixin, SubtitlesMixin): +class WebRTCPeerSession(VideoMetaMixin, MusicMixin, SubtitlesMixin): """One WebRTC peer connection, handling MNP over a DataChannel.""" def __init__(self, pc: RTCPeerConnection, node_ctx: dict, peer_id: str = ""): @@ -2346,108 +2330,6 @@ class WebRTCPeerSession(MusicMixin, SubtitlesMixin): except Exception: pass - def _do_tmdb_config(self, msg: dict) -> None: - """ - Optionally set (or clear) a custom TMDB API token, and optionally set - the language TMDB is queried in (e.g. "fr-FR") — one for the whole - node, since both are one operator's shared credential/cache, not a - per-group concern (see _do_tmdb_enabled for the per-group on/off - switch). Signed like the rest: this changes outbound third-party - network traffic the node did not have before the Videos app - (docs/MESHBAY_DESIGN.md §9.7, §6.5) — an unsigned change would let any - member alter egress the operator never agreed to. - """ - 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"custom_token={'yes' if token else 'no'},language={language or 'default'}" - self._issue_admin_challenge( - OP_TMDB_CONFIG, subject, - payload={"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("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/the root ops/the - # per-group tmdb_enabled below). - notice = { - "type": MNP.TMDB_CONFIG_ACK, "v": MNP_VERSION, - "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_tmdb_enabled(self, msg: dict) -> None: - """ - Whether TMDB lookups run for this group at all. Per-group, unlike - tmdb_config's token/language — see ops.set_tmdb_enabled. Signed like - app_directories: it decides whether this group's members' Videos tab ever - makes outbound TMDB traffic. - """ - enabled = msg.get("enabled") - if not isinstance(enabled, bool): - self._send({"type": "error", "detail": "Missing or invalid 'enabled'"}) - return - if not self._has_admin_authority(): - self._send({"type": "error", "detail": "No authorized key for this"}) - return - self._issue_admin_challenge(OP_TMDB_ENABLED, str(enabled)) - - async def _admin_exec_tmdb_enabled( - self, pending: dict, transcript: bytes, sig: bytes, - ) -> None: - enabled = pending["subject"] == "True" - if not await self._verify_admin_sig(transcript, sig): - self._send({"type": "error", "detail": "Signature verification failed"}) - self._audit("admin_auth_failed", f"tmdb_enabled:{pending['subject']}") - return - try: - await self._run_op(ops.set_tmdb_enabled, self._group_id or "", enabled) - except ops.OpError as e: - self._send({"type": "error", "detail": e.message}) - return - self._audit("tmdb_enabled", pending["subject"]) - - notice = {"type": MNP.TMDB_ENABLED_ACK, "v": MNP_VERSION, "enabled": enabled} - for uid, session in list(self._peer_registry().items()): - try: - session._send(notice) - except Exception: - pass - # ── App directories (generic) ──────────────────────────────────────── def _do_app_directories(self, msg: dict) -> None: @@ -3903,566 +3785,6 @@ class WebRTCPeerSession(MusicMixin, SubtitlesMixin): if not leased and (chunk_index + 1) * CHUNK_SIZE >= entry.size: self._leaseless.finish(str(file_id)) - @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/MESHBAY_DESIGN.md §9.7), 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/MESHBAY_DESIGN.md §9.7: TMDB metadata for one file, resolved from - the group's index by its content id (root+relpath the client already - knows from index_sync/index_delta identify the entry; its own `id` - is what actually names one file — never a raw filesystem path off - the wire). - - Keyed by `file_id`, not `path`: `IndexEntry.path` is the *folder* a - file is in (indexer.py's `_virtual_dir`), so two files in the same - folder — any multi-episode season, routinely — shared the same - `.path`, and a lookup by it could silently resolve to the wrong - entry (found live via the Music app's identical bug, 2026-08-25). - """ - file_id = msg.get("file_id") - log.debug("media_meta_req file_id=%r", file_id) - if not isinstance(file_id, str) or not file_id: - self._send({"type": "error", "detail": "Missing file_id"}) - return - ctx = self._group_ctx() - entry = ctx["index"].get_entry(file_id) - 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") - # Per-group, not node-wide (docs/MESHBAY_DESIGN.md §9.7, 2026-08-24): - # treated exactly like "no client configured" — same silent, no-error - # degradation, since a member's Videos tab already has to handle "no - # TMDB match" as the ordinary case. - if media_cache is None or tmdb_client is None or not ctx.get("tmdb_enabled", True): - self._send({"type": MNP.MEDIA_META_RESP, "v": MNP_VERSION, - "file_id": file_id, "confidence": 0}) - return - - # A video the indexer has seen but not yet *enriched* has no - # display_title (enrich.py always sets one) and season/episode still - # None — so the movie/show split reads "movie" and would hand its raw - # filename to TMDB's movie search. During a slow initial scan with a - # browser on the Videos tab that is a storm of - # `search/movie?query=<raw filename>` (found live 2026-08-29, an - # 8-minute scan). While un-enriched we never *search*: we serve a - # cached match if there is one (§ below), else confidence 0 and the - # client refetches once the index delta carries the enriched fields. - enriched = bool(entry.display_title) - - 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: - cached_tmdb_id, cached_media_type = cached - # Serve the cached match when its kind still agrees with the - # entry's current classification — OR when the entry is not - # enriched yet: its season/episode aren't populated, so the - # movie/show split above is not meaningful, and the cached kind - # (set when this file WAS enriched) is the reliable one. This is - # what keeps a restart from re-querying TMDB for everything - # already resolved: the storm was an un-enriched show episode - # looking like a "movie" and treating its own valid "tv" match - # as stale. - # - # Once enriched, the strict `cached_media_type == media_type` - # check still stands: an enrichment fix that reclassifies a - # folder movie->tv must drop the stale movie-era match and - # re-resolve (found live — a Specials-folder fix left hundreds - # of files answering with their wrong-kind match forever). - if cached_media_type == media_type or not enriched: - media_type = cached_media_type - is_show = media_type == "tv" - tmdb_id = cached_tmdb_id - meta = await media_cache.get_tmdb_meta(tmdb_id, media_type) - - if meta is None: - if not enriched: - self._send({"type": MNP.MEDIA_META_RESP, "v": MNP_VERSION, - "file_id": file_id, "confidence": 0}) - return - 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, - "file_id": file_id, "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 file_id=%r: replying tmdb_id=%s poster=%s backdrop=%s", - file_id, tmdb_id, poster_thumb_hash, backdrop_thumb_hash) - - resp = { - "type": MNP.MEDIA_META_RESP, "v": MNP_VERSION, "file_id": file_id, - "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") - # Per-group, not node-wide (docs/MESHBAY_DESIGN.md §9.7, 2026-08-24) — - # same silent zero-confidence degradation as "no client configured". - if (media_cache is None or tmdb_client is None - or not self._group_ctx().get("tmdb_enabled", True)): - 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/MESHBAY_DESIGN.md §9.7, §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") - # Per-group, not node-wide (docs/MESHBAY_DESIGN.md §9.7, 2026-08-24) — - # same silent empty-results degradation as "no client configured": - # a member with TMDB off for this group sees the same "type it in - # yourself" affordance either way, never an error. - if (media_cache is None or tmdb_client is None - or not self._group_ctx().get("tmdb_enabled", True)): - self._send({"type": MNP.TMDB_SEARCH_RESP, "v": MNP_VERSION, - "query": query, "media_type": media_type, "results": []}) - return - - # Refused out loud, not as an empty result: "no matches" is what the - # client draws for an empty list, and telling somebody their film is - # unknown when the node simply declined to ask is a worse answer than - # the truth. `video-app.js`'s `runSearch` puts `detail` on screen. - if not self._tmdb_search_rate_ok(): - log.info("tmdb_search_req: rate-limited (user=%s)", (self._user_id or "")[:8]) - self._send({ - "type": "error", - "detail": "Too many searches in the last minute. This spends the " - "operator's search quota, which everyone in the group " - "shares — try again shortly.", - "code": "tmdb_search_rate_limited", - }) - 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 - app_directories/tmdb_config: it replaces what every member sees for a - show/movie, node-wide (media_cache is shared, not per-viewer). - - For a **show**, applied to every entry sharing the representative - file's display_title — the same grouping the poster grid uses - (§3.4/§V6) — so the correction sticks regardless of which episode a - future render picks as representative. For a **movie** it is applied - to that one file only: guessit gives a whole franchise the same - display_title, and a fan-out there corrected the wrong films (found - live, 2026-08-29). See `_admin_exec_tmdb_override`. - - Keyed by `file_id`, not `path` — see `_do_media_meta_request`'s - docstring for why a folder-level path cannot name one file. - """ - file_id = msg.get("file_id") - tmdb_id = msg.get("tmdb_id") - media_type = msg.get("media_type") - if not isinstance(file_id, str) or not file_id: - self._send({"type": "error", "detail": "Missing file_id"}) - 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(file_id) - 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"file_id={file_id},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(",")) - file_id, tmdb_id, media_type = fields["file_id"], fields["tmdb_id"], fields["media_type"] - - ctx = self._group_ctx() - entry = ctx["index"].get_entry(file_id) - 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 - # This only ever recorded the file->tmdb_id mapping, never the - # metadata tmdb_id names — _do_media_meta_request's cache check - # (entry, cache) both agree on media_type, so it trusted the - # mapping — but found nothing under this *new* id in tmdb_meta - # (nothing had ever fetched it), and silently fell through to a - # fresh search using the file's own title, exactly the one that - # produced the wrong match in the first place. Confirmed live: an - # override "stuck" for shows only because their own title happened - # to be enough for that fallback search to land on the right - # answer anyway, coincidentally — never because the override itself - # was actually being honored — and was invisible until a movie - # whose own title search kept landing on the same wrong result - # exposed it. Fetching and storing the real metadata up front is - # what makes the *override* the thing a later lookup finds. - tmdb_client = self._ctx.get("tmdb_client") - if tmdb_client is not None: - meta = await self._tmdb_build_meta(tmdb_client, tmdb_id, media_type, {}) - await media_cache.set_tmdb_meta(tmdb_id, media_type, meta) - # A show's episodes are many files that legitimately share one match, - # and which episode a render picks as representative rotates — so a - # show override fans out across every entry with the same - # display_title. A *movie* is one file: fanning out by display_title - # there is a bug — guessit gives every - # "<franchise> - <year> - <subtitle>.mkv" the same display_title, so - # "Fix match" on one entry rewrote the whole franchise (found live, - # 2026-08-29). Each corrected file is also marked as a manual - # override so ops.rematch_video / a rename never wipe it. - is_show = entry.season is not None and entry.episode is not None - if is_show: - 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] - else: - matched = [entry] - for e in matched: - await media_cache.set_file_tmdb(e.id, tmdb_id, media_type) - await media_cache.mark_tmdb_override(e.id) - self._audit("tmdb_override", subject) - - notice = {"type": MNP.TMDB_OVERRIDE_ACK, "v": MNP_VERSION, "file_id": file_id, - "tmdb_id": tmdb_id, "media_type": media_type} - for uid, session in list(self._peer_registry().items()): - try: - session._send(notice) - except Exception: - pass - - def _do_tmdb_rematch(self, msg: dict) -> None: - """ - An operator dropping one file's cached TMDB match so it re-resolves - with the current matcher (V13) — the one-click alternative to - the full search-and-pick "Fix match" flow, and reachable without - SSH (`meshbay-node video rematch` clears a whole group). Signed like - `tmdb_override`: `media_cache` is shared node-wide. - """ - file_id = msg.get("file_id") - if not isinstance(file_id, str) or not file_id: - self._send({"type": "error", "detail": "Missing file_id"}) - return - if not self._group_ctx()["index"].get_entry(file_id): - 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 - self._issue_admin_challenge(OP_TMDB_REMATCH, f"file_id={file_id}") - - async def _admin_exec_tmdb_rematch( - 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_rematch:{subject}") - return - file_id = dict(part.split("=", 1) for part in subject.split(","))["file_id"] - media_cache = self._ctx.get("media_cache") - if media_cache is None: - self._send({"type": "error", "detail": "Media cache not available"}) - return - await media_cache.drop_tmdb_match(file_id) - self._audit("tmdb_rematch", subject) - - notice = {"type": MNP.TMDB_REMATCH_ACK, "v": MNP_VERSION, "file_id": file_id} - 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): - """ - docs/MESHBAY_DESIGN.md §9.7's scored ladder — same shape for movies - and shows (V8). - TMDB's own top result is still trusted per query (§3.3's last row — - no local re-ranking of *its* list); what the ladder adds is that it - *scores every candidate query* and keeps the best, instead of - returning the first that merely clears 0.6. - - The bare parsed title is the weakest query: guessit drops a - "Volume 2", strips a real subtitle into `alternative_title`, renders - a sequel index where TMDB spells it differently, and a show's folder - name can carry a year or release-group noise. A wrong entry that - scored ~0.7 against that weak query — a same-year making-of - documentary, a franchise entry whose localized TMDB title *is* the - franchise name, a season-specific promo entry standing in for a - whole show — used to win outright before a stronger candidate was - ever tried. Found live (2026-08-29). - """ - from meshbay_node.indexer import title_parse - - if is_show: - title = entry.display_title or title_parse.naive_title(entry.name) - name_naive = title_parse.naive_title(entry.name) - year = title_parse.year_in(title) or title_parse.year_in(entry.name) - extra = [c for c in (name_naive, title_parse.clean_query(title)) - if c and c != title] - return await self._tmdb_ladder( - tmdb_client.search_tv, title, extra, year, strong_extra=False) - - parsed = title_parse.parse_movie_filename(entry.name) - title = entry.display_title or parsed.display_title or parsed.naive_title - strong = [c for c in (parsed.alt_title, *title_parse.sequel_variants(title)) if c] - extra = [c for c in (*strong, parsed.naive_title) if c and c != title] - return await self._tmdb_ladder( - tmdb_client.search_movie, title, extra, parsed.year, - strong_extra=bool(strong)) - - @staticmethod - async def _tmdb_ladder(search_fn, primary: str, extra: list[str], - year: int | None, strong_extra: bool): - """ - `search_fn(query, year) -> (result|None, ratio)`. Try `primary`, - return at once on a confident hit (ratio >= 0.85 — the common case, - one request). Otherwise score each `extra` candidate and keep the - best. `strong_extra` says whether `extra` contains anything more - specific than a punctuation-normalised restatement of `primary` - (an alternative_title, a sequel variant); when it does not and the - primary hit is already decent, the remaining calls are skipped - (V11 — they almost never win and cost a round trip each). - """ - def _year_of(res: dict) -> int | None: - d = str(res.get("release_date") or res.get("first_air_date") or "") - return int(d[:4]) if d[:4].isdigit() else None - - def _rescue(res: dict, r: float) -> float: - # A sub-0.6 hit whose result lands on the exact requested year: - # TMDB already year-filtered the search, so this is a hard - # corroboration that the low ratio is a localised/rearranged - # title, not a wrong entry. Never overrides a confident hit. - if r < 0.6 and year and _year_of(res) == year: - return max(r, 0.6) - return r - - result, ratio = await search_fn(primary, year) - if result is not None and ratio >= 0.85: - return result, ratio - best_result, best_score = (result, ratio) if result is not None else (None, 0.0) - if best_result is not None: - best_score = _rescue(best_result, best_score) - if best_score >= 0.6 and not strong_extra: - return best_result, best_score - - for candidate in extra: - if candidate == primary: - continue - r2, ratio2 = await search_fn(candidate, year) - if r2 is None and year: - # A year-filtered search that finds nothing: the year tag - # may be an edition/regional year TMDB doesn't carry. Retry - # the candidate unconstrained before dropping it. - r2, ratio2 = await search_fn(candidate, None) - if r2 is None: - continue - score2 = _rescue(r2, ratio2) - if score2 > best_score: - best_result, best_score = r2, score2 - if best_score >= 0.85: - break - return best_result, best_score - - @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) - else: - # A series has no single director, and `tv_credits`' crew is the - # aggregate one — routinely empty, and never a "Director" job. - # TMDB models the equivalent credit as `created_by` on the show - # itself, which is what its own page shows; several creators are - # ordinary, and they read as one line in the detail modal. - director = ", ".join( - name for name in - (c.get("name") for c in details.get("created_by") or []) - if name) or 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_chat_message(self, msg: dict) -> None: """ Store one message and hand it to everyone else in this group. @@ -4780,33 +4102,6 @@ class WebRTCPeerSession(MusicMixin, SubtitlesMixin): if isinstance(m.payload, bytes) else m.payload) return row - def _tmdb_search_rate_ok(self) -> bool: - """ - True when this search is within both the member's window and the node's; - records it when so, and trims both to the window on every call so neither - list can grow without bound. - - Both are checked because they answer different questions: the member's - keeps one person from spending everyone's quota, and the node's keeps a - group of them from doing it together. - """ - now = time.monotonic() - w = _TMDB_SEARCH_WINDOW - ctx = self._group_ctx() - by_member = ctx.setdefault("tmdb_search_hits", {}) - who = self._user_id or "" - mine = [t for t in by_member.get(who, []) if now - t < w] - node = [t for t in self._ctx.get("tmdb_search_hits_node", []) if now - t < w] - if len(mine) >= _TMDB_SEARCH_PER_MEMBER or len(node) >= _TMDB_SEARCH_NODE: - by_member[who] = mine - self._ctx["tmdb_search_hits_node"] = node - return False - mine.append(now) - node.append(now) - by_member[who] = mine - self._ctx["tmdb_search_hits_node"] = node - return True - def _link_preview_rate_ok(self) -> bool: """ True when this preview fetch is within both the per-connection and the diff --git a/packages/meshbay-node/tests/test_tmdb_search_bound.py b/packages/meshbay-node/tests/test_tmdb_search_bound.py index 486a2c2..3153b15 100644 --- a/packages/meshbay-node/tests/test_tmdb_search_bound.py +++ b/packages/meshbay-node/tests/test_tmdb_search_bound.py @@ -21,7 +21,7 @@ simply declined to ask is a worse answer than the truth. """ import pytest -from meshbay_node.transport import webrtc_server +from meshbay_node.transport.webrtc.apps import video_meta from meshbay_node.transport.webrtc_server import WebRTCPeerSession GROUP = "g" * 32 @@ -97,8 +97,8 @@ async def test_a_member_at_the_ceiling_does_not_stop_another_one(node, monkeypat Alice exhausts her own window; Bob, who has typed nothing, must be served exactly as if she had not been there. """ - monkeypatch.setattr(webrtc_server, "_TMDB_SEARCH_PER_MEMBER", 3) - monkeypatch.setattr(webrtc_server, "_TMDB_SEARCH_NODE", 100) + monkeypatch.setattr(video_meta, "_TMDB_SEARCH_PER_MEMBER", 3) + monkeypatch.setattr(video_meta, "_TMDB_SEARCH_NODE", 100) ctx, tmdb = node alice = _member(ctx, "alice") @@ -119,8 +119,8 @@ async def test_one_member_cannot_spend_the_whole_node_quota(node, monkeypatch): because the operator's credential is one credential however many people hold the search box down. """ - monkeypatch.setattr(webrtc_server, "_TMDB_SEARCH_PER_MEMBER", 100) - monkeypatch.setattr(webrtc_server, "_TMDB_SEARCH_NODE", 2) + monkeypatch.setattr(video_meta, "_TMDB_SEARCH_PER_MEMBER", 100) + monkeypatch.setattr(video_meta, "_TMDB_SEARCH_NODE", 2) ctx, tmdb = node alice, bob = _member(ctx, "alice"), _member(ctx, "bob") @@ -138,8 +138,8 @@ async def test_a_members_count_survives_their_reconnection(node, monkeypatch): reconnect wide, and a client that drops its DataChannel between searches has no ceiling at all. """ - monkeypatch.setattr(webrtc_server, "_TMDB_SEARCH_PER_MEMBER", 2) - monkeypatch.setattr(webrtc_server, "_TMDB_SEARCH_NODE", 100) + monkeypatch.setattr(video_meta, "_TMDB_SEARCH_PER_MEMBER", 2) + monkeypatch.setattr(video_meta, "_TMDB_SEARCH_NODE", 100) ctx, tmdb = node first = _member(ctx, "alice") @@ -154,7 +154,7 @@ async def test_a_members_count_survives_their_reconnection(node, monkeypatch): async def test_a_refusal_is_said_out_loud_and_not_drawn_as_no_matches(node, monkeypatch): - monkeypatch.setattr(webrtc_server, "_TMDB_SEARCH_PER_MEMBER", 0) + monkeypatch.setattr(video_meta, "_TMDB_SEARCH_PER_MEMBER", 0) ctx, _ = node alice = _member(ctx, "alice") @@ -172,7 +172,7 @@ async def test_the_windows_do_not_grow_without_bound(node, monkeypatch): The lists are trimmed on every call, so the thing that bounds a member also bounds what remembering them costs. """ - monkeypatch.setattr(webrtc_server, "_TMDB_SEARCH_WINDOW", 0.0) + monkeypatch.setattr(video_meta, "_TMDB_SEARCH_WINDOW", 0.0) ctx, tmdb = node alice = _member(ctx, "alice") |