aboutsummaryrefslogtreecommitdiffstats
path: root/packages/meshbay-node/src
diff options
context:
space:
mode:
authorChristophe Besson <cbesson@gmail.com>2026-08-29 15:43:38 +0200
committerChristophe Besson <cbesson@gmail.com>2026-08-29 15:43:38 +0200
commit5d28d0c96cc8489645178b483835489779cb3887 (patch)
tree90248f1c3211ca37495a774d0bc140109c299abc /packages/meshbay-node/src
parent236e5e811355212945b89c4f7a99df5c837e97c7 (diff)
downloadmeshbay-5d28d0c96cc8489645178b483835489779cb3887.tar.gz
feat(node): V8–V11 — show-branch ladder, year-aware _best_match, wider sequel_variants
V8: the TV/show branch of _tmdb_search used the old "first candidate over 0.6 wins" shape. It now shares one _tmdb_ladder helper with the movie branch — score every candidate query, keep the best, fast-path a confident primary hit. A year lifted off the show's folder name (title_parse.year_in, e.g. "Some.Show.2022.S01") rescues a sub-0.6 hit that lands on the exact year. title_parse.clean_query de-dots a folder-derived title without naive_title's extension-stripping trap. V9: _best_match gains an optional `year`. When the top result is not a confident textual hit (ratio < 0.6) and a year was requested, a different result of that exact release year is preferred — TMDB already year-filtered the search, so this is a hard corroboration, not the fuzzy re-rank §3.3 warns against. A confident top hit is never overridden. search_movie/search_tv forward the year. V10: sequel_variants widened — trailing Roman→digit as well as digit→Roman, spelled-out indices (one..twelve / un..douze / ordinals), and a "Part N" / "Chapitre N" wrapper. Still empty for a trailing word that is not an index or a 4-digit year. V11: when the primary hit is already decent (>= 0.6) and there is nothing more specific to try (no alternative_title, no sequel variant — only a punctuation restatement left), the ladder returns without the extra requests. The clean-title common case is back to one call. Co-Authored-By: Claude Sonnet 5 <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_018BMLQjqFGCize2KtNBT79v
Diffstat (limited to 'packages/meshbay-node/src')
-rw-r--r--packages/meshbay-node/src/meshbay_node/indexer/title_parse.py100
-rw-r--r--packages/meshbay-node/src/meshbay_node/tmdb.py58
-rw-r--r--packages/meshbay-node/src/meshbay_node/transport/webrtc_server.py116
3 files changed, 187 insertions, 87 deletions
diff --git a/packages/meshbay-node/src/meshbay_node/indexer/title_parse.py b/packages/meshbay-node/src/meshbay_node/indexer/title_parse.py
index 24c23bc..018746b 100644
--- a/packages/meshbay-node/src/meshbay_node/indexer/title_parse.py
+++ b/packages/meshbay-node/src/meshbay_node/indexer/title_parse.py
@@ -52,14 +52,8 @@ _SPECIALS_RE = re.compile(r"\b(?:bonus|extras?|specials?)\b", re.IGNORECASE)
# some other word that merely starts with "s" followed by digits.
_SEASON_ABBREV_RE = re.compile(r"^s(\d{1,2})$", re.IGNORECASE)
-_ROMAN_NUMERALS = {
- 2: "II", 3: "III", 4: "IV", 5: "V", 6: "VI",
- 7: "VII", 8: "VIII", 9: "IX", 10: "X",
-}
-
-
def _roman_to_int(s: str) -> int | None:
- values = {"i": 1, "v": 5, "x": 10, "l": 50, "c": 100}
+ values = {"i": 1, "v": 5, "x": 10, "l": 50, "c": 100, "d": 500, "m": 1000}
s = s.lower()
if not s or any(c not in values for c in s):
return None
@@ -72,6 +66,49 @@ def _roman_to_int(s: str) -> int | None:
return total or None
+def _int_to_roman(n: int) -> str | None:
+ if not 1 <= n <= 39:
+ return None
+ out = []
+ for val, sym in ((10, "X"), (9, "IX"), (5, "V"), (4, "IV"), (1, "I")):
+ while n >= val:
+ out.append(sym)
+ n -= val
+ return "".join(out)
+
+
+# Spelled-out sequel indices, English + French, plus a few ordinals.
+_NUMBER_WORDS: dict[str, int] = {
+ "one": 1, "two": 2, "three": 3, "four": 4, "five": 5, "six": 6, "seven": 7,
+ "eight": 8, "nine": 9, "ten": 10, "eleven": 11, "twelve": 12,
+ "first": 1, "second": 2, "third": 3,
+ "un": 1, "deux": 2, "trois": 3, "quatre": 4, "cinq": 5, "sept": 7,
+ "huit": 8, "neuf": 9, "dix": 10, "onze": 11, "douze": 12,
+ "premier": 1, "première": 1, "deuxième": 2, "seconde": 2, "troisième": 3,
+}
+_YEAR_RE = re.compile(r"(?<!\d)(?:19|20)\d{2}(?!\d)")
+
+
+def year_in(text: str) -> int | None:
+ """First 19xx/20xx in `text`, or None — used to lift a year off a show
+ folder name ("Some.Show.2022.S01") for the search fallback (§10.1/V8)."""
+ m = _YEAR_RE.search(text or "")
+ return int(m.group(0)) if m else None
+
+
+def clean_query(s: str) -> str:
+ """
+ Punctuation → spaces for a TMDB query, *without* the extension /
+ parenthesized-year stripping `naive_title` does. `naive_title` assumes
+ a real filename; a show's `display_title` is a folder basename
+ ("Some.Show.Name" — `rsplit('.', 1)` would eat ".Name"), so it needs a
+ gentler normaliser (§10.1/V8).
+ """
+ s = re.sub(r"[._-]+", " ", s or "")
+ s = _strip_editions(s)
+ return re.sub(r"\s+", " ", s).strip()
+
+
def _strip_editions(title: str) -> str:
return re.sub(r"\s+", " ", _EDITION_RE.sub(" ", title)).strip()
@@ -90,21 +127,50 @@ def naive_title(filename: str) -> str:
return re.sub(r"\s+", " ", stem).strip()
+_PART_KEYWORDS = r"part|chapter|volume|vol|partie|chapitre|volet|livre|book|episode"
+_TRAILING_INDEX_RE = re.compile(
+ r"^(?P<base>.+?)(?:\s+(?:" + _PART_KEYWORDS + r"))?"
+ r"\s+(?P<num>\d{1,2}|[ivxlcdm]{1,6}|" + "|".join(_NUMBER_WORDS) + r")$",
+ re.IGNORECASE,
+)
+
+
+def _index_value(tok: str) -> int | None:
+ tok = tok.strip().lower()
+ if tok.isdigit():
+ v = int(tok)
+ return v if 1 <= v <= 39 else None
+ if tok in _NUMBER_WORDS:
+ return _NUMBER_WORDS[tok]
+ return _roman_to_int(tok)
+
+
def sequel_variants(title: str) -> list[str]:
"""
- A trailing sequel digit sometimes has no equivalent in the real TMDB
- title, or the real title uses a Roman numeral instead (§3.3 row 4).
- Returns extra candidates to try — empty if `title` has no trailing digit.
+ A trailing sequel index often has no exact match in the real TMDB
+ title: the file has a digit where TMDB uses a Roman numeral (or the
+ reverse), spells the number out, or wraps it as "Part N" / "Chapitre N"
+ (§3.3 row 4, §10.1/V10). Returns extra candidate titles to try — the
+ base alone, and the index re-rendered as digit and as Roman numeral.
+ Empty when `title` carries no recognisable trailing index.
"""
- m = re.match(r"^(.*\S)\s+([2-9])$", title)
+ m = _TRAILING_INDEX_RE.match(title.strip())
if not m:
return []
- base, digit = m.group(1), int(m.group(2))
- variants = [base]
- roman = _ROMAN_NUMERALS.get(digit)
- if roman:
- variants.append(f"{base} {roman}")
- return variants
+ base = m.group("base").strip()
+ if len(base) < 2:
+ return []
+ n = _index_value(m.group("num"))
+ if n is None:
+ return []
+ tok = m.group("num").lower()
+ out = [base]
+ roman = _int_to_roman(n)
+ if roman and roman.lower() != tok:
+ out.append(f"{base} {roman}")
+ if str(n) != tok:
+ out.append(f"{base} {n}")
+ return [v for v in dict.fromkeys(out) if v != title]
def season_from_folder_name(name: str) -> int | None:
diff --git a/packages/meshbay-node/src/meshbay_node/tmdb.py b/packages/meshbay-node/src/meshbay_node/tmdb.py
index a530660..c9623df 100644
--- a/packages/meshbay-node/src/meshbay_node/tmdb.py
+++ b/packages/meshbay-node/src/meshbay_node/tmdb.py
@@ -51,24 +51,47 @@ def _normalize(s: str) -> str:
return re.sub(r"\s+", " ", s).strip()
-def _best_match(query_title: str, results: list[dict], keys: tuple[str, ...]) -> tuple[dict | None, float]:
+def _release_year_of(item: dict) -> int | None:
+ d = str(item.get("release_date") or item.get("first_air_date") or "")
+ return int(d[:4]) if d[:4].isdigit() else None
+
+
+def _ratio_against(qn: str, item: dict, keys: tuple[str, ...]) -> float:
+ best = 0.0
+ for k in keys:
+ val = item.get(k)
+ if val:
+ best = max(best, difflib.SequenceMatcher(None, qn, _normalize(str(val))).ratio())
+ return best
+
+
+def _best_match(
+ query_title: str, results: list[dict], keys: tuple[str, ...],
+ year: int | None = None,
+) -> tuple[dict | None, float]:
"""
Trusts TMDB's own ranking (§3.3's last row — a locally-recomputed
- re-rank picked a coincidentally-closer-looking wrong show once): only
- the top result is considered. The similarity ratio is returned purely
- as a confidence signal for the caller's fallback decision, never used
- to pick a different candidate.
+ re-rank picked a coincidentally-closer-looking wrong show once): the
+ top result is what's returned. The similarity ratio rides along purely
+ as a confidence signal for the caller's fallback decision.
+
+ One narrow exception (§10.1/V9): when the top result is *not* a
+ confident textual hit (ratio < 0.6) and a `year` was requested, a
+ different result of that **exact** release year is preferred. TMDB
+ already year-filtered the search, so an entry landing on the requested
+ year is a hard corroboration, not the fuzzy string re-rank §3.3 warns
+ against — and this never overrides a confident top hit.
"""
if not results:
return None, 0.0
- top = results[0]
qn = _normalize(query_title)
- best_ratio = 0.0
- for k in keys:
- val = top.get(k)
- if val:
- best_ratio = max(best_ratio, difflib.SequenceMatcher(None, qn, _normalize(str(val))).ratio())
- return top, best_ratio
+ top = results[0]
+ top_ratio = _ratio_against(qn, top, keys)
+ if year is not None and top_ratio < 0.6:
+ for item in results:
+ if _release_year_of(item) == year:
+ return item, _ratio_against(qn, item, keys)
+ return top, top_ratio
class TmdbClient:
@@ -138,12 +161,15 @@ class TmdbClient:
params["year"] = year
data = await self._get("search/movie", params)
results = (data or {}).get("results", [])
- return _best_match(title, results, ("title", "original_title"))
+ return _best_match(title, results, ("title", "original_title"), year=year)
- async def search_tv(self, title: str) -> tuple[dict | None, float]:
- data = await self._get("search/tv", {"query": title})
+ async def search_tv(self, title: str, year: int | None = None) -> tuple[dict | None, float]:
+ params = {"query": title}
+ if year:
+ params["first_air_date_year"] = year
+ data = await self._get("search/tv", params)
results = (data or {}).get("results", [])
- return _best_match(title, results, ("name", "original_name"))
+ return _best_match(title, results, ("name", "original_name"), year=year)
async def search_movie_results(self, title: str) -> list[dict]:
"""
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 788a8f1..c75819e 100644
--- a/packages/meshbay-node/src/meshbay_node/transport/webrtc_server.py
+++ b/packages/meshbay-node/src/meshbay_node/transport/webrtc_server.py
@@ -3173,80 +3173,88 @@ class WebRTCPeerSession:
async def _tmdb_search(self, tmdb_client, entry, is_show: bool):
"""
- §3.3's retry ladder. TMDB's own top result is still trusted per
- query (§3.3's last row — no local re-ranking of *its* list); what
- changed is that the ladder now *scores every candidate query* and
- keeps the best, instead of returning the first that merely clears
- 0.6.
+ §3.3's retry ladder — same shape for movies and shows (§10.1/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`, and
- renders a sequel number where TMDB uses a Roman numeral. A wrong
- film that happened to score ~0.7 against that weak query — a
- same-year making-of documentary, or a franchise entry whose
- localized TMDB title *is* the franchise name — used to win outright
- before `alternative_title` or the Roman-numeral variant was ever
- tried. Found live (2026-08-29): a numbered sequel matched a
- same-year documentary; a two-volume film's second part matched the
- first; several franchise entries matched one early entry.
+ "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)
- result, ratio = await tmdb_client.search_tv(title)
- if result is None or ratio < 0.6:
- naive = title_parse.naive_title(entry.name)
- if naive != title:
- result, ratio = await tmdb_client.search_tv(naive)
- return result, ratio
-
- def _release_year(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
+ 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))
- # Fast path, unchanged in effect: a strong direct hit still returns
- # on the first call, so the common case costs exactly one request
- # and the new ladder below only engages in the ambiguous 0<ratio<0.85
- # zone where every one of the live bugs lived.
- result, ratio = await tmdb_client.search_movie(title, parsed.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)
+ @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
+ (§10.1/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 _apply_year_rescue(res: dict, r: float) -> float:
- # Only for a query whose own top hit is weak on its face
- # (ratio < 0.6): TMDB already year-filtered the search, so its
- # top result landing exactly on the filename's year is a hard
- # corroborating signal that the low string ratio is a
- # localized/rearranged title, not a wrong film. Never lets year
- # equality outrank a genuinely strong textual match elsewhere.
- if r < 0.6 and parsed.year and _release_year(res) == parsed.year:
+ 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 = _apply_year_rescue(best_result, best_score)
+ best_score = _rescue(best_result, best_score)
+ if best_score >= 0.6 and not strong_extra:
+ return best_result, best_score
- for candidate in filter(None, [parsed.alt_title,
- *title_parse.sequel_variants(title),
- parsed.naive_title]):
- if candidate == title:
+ for candidate in extra:
+ if candidate == primary:
continue
- r2, ratio2 = await tmdb_client.search_movie(candidate, parsed.year)
- if r2 is None and parsed.year:
- # A year-filtered search that finds nothing: the filename's
- # year tag may be an edition/regional year TMDB doesn't
- # carry. Retry the same candidate unconstrained before
- # dropping it.
- r2, ratio2 = await tmdb_client.search_movie(candidate)
+ 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 = _apply_year_rescue(r2, ratio2)
+ score2 = _rescue(r2, ratio2)
if score2 > best_score:
best_result, best_score = r2, score2
if best_score >= 0.85: