diff options
Diffstat (limited to 'packages/meshbay-node/src/meshbay_node/indexer')
3 files changed, 372 insertions, 1 deletions
diff --git a/packages/meshbay-node/src/meshbay_node/indexer/enrich.py b/packages/meshbay-node/src/meshbay_node/indexer/enrich.py new file mode 100644 index 0000000..784b2a3 --- /dev/null +++ b/packages/meshbay-node/src/meshbay_node/indexer/enrich.py @@ -0,0 +1,166 @@ +""" +Index-time enrichment for the Videos group app: technical probe (ffprobe), +filename parsing (title_parse), and thumbnail generation (ffmpeg) for a +newly-added video IndexEntry. + +Runs through its own small bounded worker pool — separate from the streaming +transcode pool (docs/mediacenter.md §5.2, mirroring webrtc_server.py's +`_transcode_semaphore`) — so indexing a large library never blocks on this, +and enrichment never competes with an active viewer for CPU. The scan itself +already put the entry in the index with hash/size/type only; this fills in +the rest asynchronously and hands the result back via a callback. +""" + +import asyncio +import logging +from pathlib import Path +from typing import Awaitable, Callable + +import blake3 + +from meshbay_common.protocol import IndexEntry +from meshbay_node.indexer import title_parse +from meshbay_node.indexer.indexer import MEDIA_EXTENSIONS +from meshbay_node.media_cache import MediaCache +from meshbay_node.media_probe import probe_video + +log = logging.getLogger(__name__) + +DEFAULT_MAX_CONCURRENT = 2 +PROBE_TIMEOUT_SECS = 30 +THUMB_TIMEOUT_SECS = 30 +THUMB_WIDTH = 320 +# "Show/SeasonFolder/episode.mkv" is the expected shape, with a little slack +# for an extra wrapper folder — not an attempt to find the exact group root. +MAX_ANCESTOR_DEPTH = 4 +# Bounds the "borrow a title from a sibling episode filename" scan (§3.4) so +# a folder with thousands of files costs a fixed, small amount of work. +MAX_SIBLINGS_CHECKED = 20 + + +def _season_from_ancestors(file_path: Path) -> int | None: + folder = file_path.parent + for _ in range(MAX_ANCESTOR_DEPTH): + if folder is None or folder == folder.parent: + break + season = title_parse.season_from_folder_name(folder.name) + if season is not None: + return season + folder = folder.parent + return None + + +def _title_from_siblings(file_path: Path) -> str | None: + """ + §3.4: an episode filename with no show name in it borrows the title from + a representative sibling in the same folder, never from the folder name + alone (an acronym-named show folder is a real, observed case). + """ + try: + names = sorted(p.name for p in file_path.parent.iterdir() if p.is_file()) + except OSError: + return None + checked = 0 + for name in names: + if name == file_path.name: + continue + if Path(name).suffix.lower() not in MEDIA_EXTENSIONS["video"]: + continue + checked += 1 + if checked > MAX_SIBLINGS_CHECKED: + break + parsed = title_parse.parse_episode_filename(name) + if parsed.display_title: + return parsed.display_title + return None + + +async def _make_thumbnail(file_path: Path, duration: float | None) -> bytes | None: + """One ffmpeg frame grab at ~10% of duration (or 5s if unknown), scaled down.""" + seek = max(0.0, (duration or 50.0) * 0.1) + proc = await asyncio.create_subprocess_exec( + "ffmpeg", "-v", "error", "-ss", str(seek), "-i", str(file_path), + "-frames:v", "1", "-vf", f"scale={THUMB_WIDTH}:-1", + "-f", "image2", "-c:v", "mjpeg", "pipe:1", + stdout=asyncio.subprocess.PIPE, stderr=asyncio.subprocess.PIPE, + ) + try: + stdout, _ = await asyncio.wait_for(proc.communicate(), THUMB_TIMEOUT_SECS) + except asyncio.TimeoutError: + proc.kill() + await proc.wait() + return None + return stdout or None + + +class Enricher: + """Owns the node's bounded index-time enrichment pool.""" + + def __init__(self, media_cache: MediaCache, max_concurrent: int = DEFAULT_MAX_CONCURRENT): + self._media_cache = media_cache + self._sem = asyncio.Semaphore(max_concurrent) + self._tasks: set[asyncio.Task] = set() + + def spawn( + self, entry: IndexEntry, file_path: Path, + on_done: Callable[[str, dict], Awaitable[None]], + ) -> asyncio.Task: + """ + Fire-and-forget one file's enrichment. `on_done(file_id, fields)` is + awaited with the index fields to merge in once ready — never blocks + the caller (a scan or watchdog event). The reference this method + returns is what keeps the task alive; callers should hold it the + same way `WebRTCPeerSession._spawn` holds streaming tasks. + """ + task = asyncio.ensure_future(self._run(entry, file_path, on_done)) + self._tasks.add(task) + + def _cleanup(t: asyncio.Task) -> None: + self._tasks.discard(t) + if not t.cancelled() and t.exception(): + log.error("Enrichment failed for %s: %s", entry.id[:12], t.exception(), + exc_info=t.exception()) + task.add_done_callback(_cleanup) + return task + + async def _run( + self, entry: IndexEntry, file_path: Path, + on_done: Callable[[str, dict], Awaitable[None]], + ) -> None: + async with self._sem: + fields: dict = {} + duration: float | None = None + try: + _codec, duration, _has_audio, width, height = await asyncio.wait_for( + probe_video(str(file_path)), timeout=PROBE_TIMEOUT_SECS) + fields["duration"] = int(duration) if duration else None + fields["width"] = width + fields["height"] = height + except Exception as e: + log.warning("Probe failed for %s: %s", file_path, e) + + ep = title_parse.parse_episode_filename(entry.name) + if ep.episode is not None: + title = ep.display_title or await asyncio.to_thread( + _title_from_siblings, file_path) + season = ep.season + if season is None: + season = await asyncio.to_thread(_season_from_ancestors, file_path) + fields["display_title"] = title or title_parse.naive_title(entry.name) + fields["season"] = season + fields["episode"] = ep.episode + else: + mv = title_parse.parse_movie_filename(entry.name) + fields["display_title"] = mv.display_title or mv.naive_title + + try: + thumb = await _make_thumbnail(file_path, duration) + except Exception as e: + log.warning("Thumbnail generation failed for %s: %s", file_path, e) + thumb = None + if thumb: + thumb_hash = blake3.blake3(thumb).hexdigest() + await self._media_cache.put_thumb(thumb_hash, entry.id, thumb) + fields["thumb_hash"] = thumb_hash + + await on_done(entry.id, fields) diff --git a/packages/meshbay-node/src/meshbay_node/indexer/group_index.py b/packages/meshbay-node/src/meshbay_node/indexer/group_index.py index ec98667..1ce4e0a 100644 --- a/packages/meshbay-node/src/meshbay_node/indexer/group_index.py +++ b/packages/meshbay-node/src/meshbay_node/indexer/group_index.py @@ -78,6 +78,19 @@ class GroupIndex: def get_entry(self, file_id: str) -> IndexEntry | None: return self._entries.get(file_id) + def get_entry_by_path(self, path: str) -> IndexEntry | None: + """ + Linear scan — entries are keyed by content id, not path, and nothing + before the Videos app needed to go the other way (a client always + already has the id from index_sync/index_delta). Fine for an + on-demand, per-tile lookup against a few thousand entries; revisit + if a future caller makes this hot. + """ + for entry in self._entries.values(): + if entry.path == path: + return entry + return None + @property def entries(self) -> list[IndexEntry]: return list(self._entries.values()) @@ -205,16 +218,34 @@ class GroupIndex: # ── Delta ───────────────────────────────────────────────────────────────── def diff(self, previous: "GroupIndex") -> IndexDelta: - """Compute what changed since a previous version of this index.""" + """ + Compute what changed since a previous version of this index. + + A shared id whose entry object now compares unequal (field-by-field, + via IndexEntry's dataclass-generated __eq__) is an update, not an + addition — the Videos app's async enrichment (duration, thumb_hash, + title, ...) replaces an existing entry's fields after the fact via + `add_entry`, which never introduces a new id. This only works + because that replacement always constructs a *new* IndexEntry object + (`dataclasses.replace`, never in-place attribute mutation) — mutating + the same object in place would also mutate `previous`'s copy, since + entries_by_id() is a shallow dict copy, and the two would always + compare equal. + """ prev_ids = set(previous._entries) curr_ids = set(self._entries) additions = [self._entries[i] for i in curr_ids - prev_ids] deletions = list(prev_ids - curr_ids) + updates = [ + self._entries[i] for i in curr_ids & prev_ids + if self._entries[i] != previous._entries[i] + ] return IndexDelta( base_version=previous.version, version=self.version, additions=additions, deletions=deletions, + updates=updates, ) diff --git a/packages/meshbay-node/src/meshbay_node/indexer/title_parse.py b/packages/meshbay-node/src/meshbay_node/indexer/title_parse.py new file mode 100644 index 0000000..ef522ad --- /dev/null +++ b/packages/meshbay-node/src/meshbay_node/indexer/title_parse.py @@ -0,0 +1,174 @@ +""" +Filename -> title/year/season/episode parsing for the Videos group app. + +Wraps `guessit` and layers the fixes from docs/mediacenter.md §3.3/§3.4 on +top of it: none of them are per-title hacks, each is a generic rule found +by validating guessit's raw output against real TMDB search results over a +~1950-file library (movies, TV shows, and a small franchise set). + +Scope is deliberately narrow (§3.5): title, year, season, episode. Technical +facts (resolution, codec, duration) come from ffprobe, never the filename — +a mislabeled `1080p` tag is a real, observed failure mode. + +This module never touches the filesystem or the network. The orchestration +that decides *which* file supplies a show's title (a representative episode +filename, not the folder name — §3.4) lives in the indexer, which has the +directory listing; this module only parses strings it's handed. +""" + +from __future__ import annotations + +import re +from dataclasses import dataclass, field + +from guessit import guessit + +# French edition/release vocabulary guessit's (English-centric) edition list +# doesn't recognize — left stuck to the title instead of stripped as a tag. +_EDITION_PHRASES = ( + r"version\s+longue", + r"version\s+int[ée]grale", + r"remasteris[ée]e?", + r"non[\s._-]*censur[ée]e?", + r"original\s+version", +) +_EDITION_RE = re.compile("|".join(_EDITION_PHRASES), re.IGNORECASE) + +# A season-like ancestor folder: the English/French words plus a number or +# Roman numeral. Vocabulary is a plain tuple so a deployment can extend it +# per locale without touching the regex-building logic. +SEASON_WORDS = ("season", "saison") +_SEASON_RE = re.compile( + r"(?:" + "|".join(SEASON_WORDS) + r")\s*([0-9]+|[ivxlc]+)\b", + re.IGNORECASE, +) +_SPECIALS_RE = re.compile(r"\b(?:bonus|extras?|specials?)\b", re.IGNORECASE) + +_ROMAN_NUMERALS = { + 2: "II", 3: "III", 4: "IV", 5: "V", 6: "VI", + 7: "VII", 8: "VIII", 9: "IX", 10: "X", +} + + +def _roman_to_int(s: str) -> int | None: + values = {"i": 1, "v": 5, "x": 10, "l": 50, "c": 100} + s = s.lower() + if not s or any(c not in values for c in s): + return None + total = 0 + prev = 0 + for c in reversed(s): + v = values[c] + total += v if v >= prev else -v + prev = v + return total or None + + +def _strip_editions(title: str) -> str: + return re.sub(r"\s+", " ", _EDITION_RE.sub(" ", title)).strip() + + +def naive_title(filename: str) -> str: + """ + The mandated fallback (§3.6, §4.1): strip the extension, replace every + `.`/`_`/`-` with a space, drop a trailing parenthesized year, collapse + whitespace. Always computable, never fails, used both as the flat-mode + display name of last resort and as a second TMDB query candidate. + """ + stem = filename.rsplit(".", 1)[0] if "." in filename else filename + stem = re.sub(r"[._-]+", " ", stem) + stem = re.sub(r"\(\s*(19|20)\d{2}\s*\)", " ", stem) + stem = _strip_editions(stem) + return re.sub(r"\s+", " ", stem).strip() + + +def sequel_variants(title: str) -> list[str]: + """ + A trailing sequel digit sometimes has no equivalent in the real TMDB + title, or the real title uses a Roman numeral instead (§3.3 row 4). + Returns extra candidates to try — empty if `title` has no trailing digit. + """ + m = re.match(r"^(.*\S)\s+([2-9])$", title) + if not m: + return [] + base, digit = m.group(1), int(m.group(2)) + variants = [base] + roman = _ROMAN_NUMERALS.get(digit) + if roman: + variants.append(f"{base} {roman}") + return variants + + +def season_from_folder_name(name: str) -> int | None: + """ + §3.4: a season-like ancestor folder, vocabulary-driven rather than + assuming a numeric convention everywhere. A specials/bonus/extras + folder maps to season 0 (matching TMDB's own `season_number: 0`). + Returns None if `name` doesn't look like a season folder at all. + """ + if _SPECIALS_RE.search(name): + return 0 + m = _SEASON_RE.search(name) + if not m: + return None + token = m.group(1) + if token.isdigit(): + return int(token) + return _roman_to_int(token) + + +@dataclass +class ParsedName: + display_title: str | None # None => caller must supply from elsewhere (e.g. a sibling file) + alt_title: str | None = None # guessit's alternative_title, a second query candidate (§3.3 row 1) + naive_title: str = "" # always available, fully punctuation-normalized fallback + year: int | None = None + season: int | None = None + episode: int | None = None + confidence: bool = False # True only when display_title is set and structurally corroborated + + +def parse_movie_filename(filename: str) -> ParsedName: + """Parse a standalone movie filename.""" + g = guessit(filename) + title = g.get("title") + title = str(title).strip() if title else None + if title: + title = _strip_editions(title) + alt = g.get("alternative_title") + alt = _strip_editions(str(alt).strip()) if alt else None + year = g.get("year") + nt = naive_title(filename) + confidence = bool(title) and len(title) >= 2 and year is not None + return ParsedName( + display_title=title or None, alt_title=alt, naive_title=nt, + year=year, confidence=confidence, + ) + + +def parse_episode_filename(filename: str) -> ParsedName: + """ + Parse an episode filename. `display_title` may come back None (e.g. + `S08E02.SUBFRENCH.720p.mkv` carries no show name at all, §3.2) — the + indexer then supplies the show title from a representative sibling + filename in the same folder rather than the folder name itself (§3.4). + """ + g = guessit(filename) + title = g.get("title") + title = str(title).strip() if title else None + if title: + title = _strip_editions(title) + season = g.get("season") + episode = g.get("episode") + # guessit returns a list when it finds more than one candidate (e.g. a + # multi-episode file); take the first as the representative one. + if isinstance(season, list): + season = season[0] if season else None + if isinstance(episode, list): + episode = episode[0] if episode else None + nt = naive_title(filename) + confidence = bool(title) and len(title) >= 2 and season is not None and episode is not None + return ParsedName( + display_title=title or None, naive_title=nt, + season=season, episode=episode, confidence=confidence, + ) |