path: root/packages/meshbay-node/src/meshbay_node/transport/webrtc/apps/video_meta.py

"""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