Add TheTVDB episode title matching and release v1.0.3.

Match filenames to TheTVDB episode order (default, aired, DVD, etc.) to fix chaotic SxxExx numbering while preserving titles.

Co-authored-by: Cursor <[email protected]>

engine/__init__.py

@@ -1,4 +1,4 @@
-from .rules import Rule, ReplaceRule, InsertRule, RemoveRule, CaseRule, NumberingRule, EpisodeRenumberRule, RegexRule, PrefixSuffixRule, CsvMappingRule
+from .rules import Rule, ReplaceRule, InsertRule, RemoveRule, CaseRule, NumberingRule, EpisodeRenumberRule, TvdbEpisodeRenumberRule, RegexRule, PrefixSuffixRule, CsvMappingRule
 from .pipeline import apply_pipeline, compute_preview
 
 __all__ = [
@@ -9,6 +9,7 @@ __all__ = [
     "CaseRule",
     "NumberingRule",
     "EpisodeRenumberRule",
+    "TvdbEpisodeRenumberRule",
     "RegexRule",
     "PrefixSuffixRule",
     "CsvMappingRule",

engine/episode_match.py

@@ -0,0 +1,159 @@
+"""
+Match episode titles from filenames against a reference episode list (e.g. TheTVDB).
+"""
+from __future__ import annotations
+
+import re
+from difflib import SequenceMatcher
+from typing import Optional
+
+from .tvdb_client import TvdbEpisode
+
+DEFAULT_EPISODE_PATTERN = r"(.*?[Ss]\d+[Ee])(\d+)(-[Ee](\d+))?(.*)"
+
+
+def normalize_title(title: str) -> str:
+    """Lowercase, strip punctuation, collapse whitespace for fuzzy comparison."""
+    t = title.lower()
+    t = re.sub(r"[^\w\s]", " ", t, flags=re.UNICODE)
+    return re.sub(r"\s+", " ", t).strip()
+
+
+def parse_episode_stem(stem: str, pattern: str = DEFAULT_EPISODE_PATTERN) -> Optional[dict]:
+    """
+    Parse SxxExx-style filename stem.
+    Returns dict with prefix, episode numbers, title rest, and span — or None if no match.
+    """
+    try:
+        m = re.match(pattern, stem)
+    except re.error:
+        return None
+    if not m:
+        return None
+    prefix, old_first_s, _range_dash_e, old_second_s, rest = (
+        m.group(1),
+        m.group(2),
+        m.group(3),
+        m.group(4),
+        m.group(5),
+    )
+    try:
+        old_first = int(old_first_s)
+    except ValueError:
+        return None
+    span = 1
+    if old_second_s is not None:
+        try:
+            old_second = int(old_second_s)
+        except ValueError:
+            return None
+        span = old_second - old_first + 1
+        if span < 1:
+            span = 1
+    title = rest.strip()
+    if title.startswith("-") or title.startswith("–"):
+        title = title[1:].strip()
+    if title.startswith("_"):
+        title = title[1:].strip()
+    return {
+        "prefix": prefix,
+        "old_first": old_first,
+        "span": span,
+        "title": title,
+    }
+
+
+def rewrite_episode_number(
+    stem: str,
+    new_first_ep: int,
+    padding: int = 2,
+    pattern: str = DEFAULT_EPISODE_PATTERN,
+) -> str:
+    """Replace episode number block in stem, preserving prefix, span, and title."""
+    parsed = parse_episode_stem(stem, pattern)
+    if not parsed:
+        return stem
+    try:
+        m = re.match(pattern, stem)
+    except re.error:
+        return stem
+    if not m:
+        return stem
+    prefix = m.group(1)
+    rest = m.group(5)
+    span = parsed["span"]
+    pad = max(1, padding)
+    e1 = str(new_first_ep).zfill(pad)
+    if span <= 1:
+        return f"{prefix}{e1}{rest}"
+    e2 = str(new_first_ep + span - 1).zfill(pad)
+    return f"{prefix}{e1}-E{e2}{rest}"
+
+
+def _similarity(a: str, b: str) -> float:
+    if not a or not b:
+        return 0.0
+    if a == b:
+        return 1.0
+    return SequenceMatcher(None, a, b).ratio()
+
+
+def match_filenames_to_episodes(
+    filenames: list[str],
+    episodes: list[TvdbEpisode],
+    pattern: str = DEFAULT_EPISODE_PATTERN,
+    min_score: float = 0.72,
+) -> tuple[dict[str, int], list[str], list[str]]:
+    """
+    Match filenames to TheTVDB episode numbers by title.
+
+    Returns:
+        mapping: original filename -> correct episode number
+        unmatched_files: filenames that could not be matched
+        notes: human-readable match details for UI
+    """
+    file_entries: list[tuple[str, str, str]] = []
+    for name in filenames:
+        stem = name.rsplit(".", 1)[0] if "." in name and not name.startswith(".") else name
+        parsed = parse_episode_stem(stem, pattern)
+        if not parsed or not parsed["title"]:
+            continue
+        norm = normalize_title(parsed["title"])
+        if norm:
+            file_entries.append((name, norm, parsed["title"]))
+
+    ep_entries = [(ep.number, normalize_title(ep.name), ep.name) for ep in episodes]
+
+    pairs: list[tuple[float, str, int, str, str]] = []
+    for fname, fnorm, raw_title in file_entries:
+        for ep_num, enorm, ep_name in ep_entries:
+            score = _similarity(fnorm, enorm)
+            pairs.append((score, fname, ep_num, raw_title, ep_name))
+
+    pairs.sort(key=lambda x: (-x[0], x[1], x[2]))
+
+    mapping: dict[str, int] = {}
+    used_files: set[str] = set()
+    used_eps: set[int] = set()
+    notes: list[str] = []
+
+    for score, fname, ep_num, raw_title, ep_name in pairs:
+        if score < min_score:
+            break
+        if fname in used_files or ep_num in used_eps:
+            continue
+        mapping[fname] = ep_num
+        used_files.add(fname)
+        used_eps.add(ep_num)
+        pct = int(round(score * 100))
+        notes.append(f"{fname}: E{ep_num:02d} ← “{ep_name}” ({pct}% match, file title “{raw_title}”)")
+
+    unmatched_files = [
+        name for name in filenames
+        if name not in mapping
+        and parse_episode_stem(
+            name.rsplit(".", 1)[0] if "." in name and not name.startswith(".") else name,
+            pattern,
+        )
+    ]
+    return mapping, unmatched_files, notes

engine/rules.py

@@ -9,6 +9,8 @@ from dataclasses import dataclass, field
 from pathlib import Path
 from typing import List, Optional
 
+from .episode_match import rewrite_episode_number
+
 
 @dataclass
 class Rule(ABC):
@@ -257,6 +259,38 @@ class EpisodeRenumberRule(Rule):
         return new_stem, ext
 
 
+@dataclass
+class TvdbEpisodeRenumberRule(Rule):
+    """
+    Renumber SxxExx blocks using a pre-built filename → episode number mapping
+    (typically from TheTVDB title matching).
+    """
+    episode_mapping: dict = field(default_factory=dict, repr=False)
+    padding: int = 2
+    pattern: str = r"(.*?[Ss]\d+[Ee])(\d+)(-[Ee](\d+))?(.*)"
+
+    def apply(
+        self,
+        stem: str,
+        ext: str,
+        index: int,
+        total: int,
+        original_name: Optional[str] = None,
+    ) -> tuple[str, str]:
+        if not self.episode_mapping or original_name is None:
+            return stem, ext
+        new_ep = self.episode_mapping.get(original_name)
+        if new_ep is None:
+            return stem, ext
+        new_stem = rewrite_episode_number(
+            stem,
+            int(new_ep),
+            padding=self.padding,
+            pattern=self.pattern,
+        )
+        return new_stem, ext
+
+
 @dataclass
 class PrefixSuffixRule(Rule):
     prefix: str = ""

engine/tvdb_client.py

@@ -0,0 +1,169 @@
+"""
+TheTVDB API v4 client (read-only). Uses stdlib urllib — no extra dependencies.
+"""
+from __future__ import annotations
+
+import json
+import urllib.error
+import urllib.parse
+import urllib.request
+from dataclasses import dataclass
+from typing import Any, Optional
+
+BASE_URL = "https://api4.thetvdb.com/v4"
+TVDB_API_KEY = "78384b24-0f0b-461e-9415-69cce3da35e4"
+
+# Labels and API path segments for /series/{id}/episodes/{season-type}
+SEASON_TYPE_CHOICES: list[tuple[str, str]] = [
+    ("Default (TheTVDB default for show)", "default"),
+    ("Official / aired order", "official"),
+    ("DVD order", "dvd"),
+    ("Absolute order", "absolute"),
+    ("Alternate order", "alternate"),
+    ("Regional order", "regional"),
+]
+
+
+@dataclass(frozen=True)
+class TvdbSeries:
+    id: int
+    name: str
+    year: Optional[str] = None
+
+
+@dataclass(frozen=True)
+class TvdbEpisode:
+    number: int
+    season_number: int
+    name: str
+
+
+class TvdbError(Exception):
+    pass
+
+
+class TvdbClient:
+    def __init__(self, api_key: str, pin: str = ""):
+        self.api_key = api_key.strip()
+        self.pin = pin.strip()
+        self._token: Optional[str] = None
+
+    def _request(
+        self,
+        method: str,
+        path: str,
+        params: Optional[dict[str, Any]] = None,
+        body: Optional[dict[str, Any]] = None,
+    ) -> dict[str, Any]:
+        if not self._token and path != "/login":
+            self.login()
+
+        url = BASE_URL + path
+        if params:
+            query = urllib.parse.urlencode({k: v for k, v in params.items() if v is not None})
+            if query:
+                url = f"{url}?{query}"
+
+        data_bytes: Optional[bytes] = None
+        headers = {"Accept": "application/json"}
+        if body is not None:
+            data_bytes = json.dumps(body).encode("utf-8")
+            headers["Content-Type"] = "application/json"
+        if self._token:
+            headers["Authorization"] = f"Bearer {self._token}"
+
+        req = urllib.request.Request(url, data=data_bytes, headers=headers, method=method)
+        try:
+            with urllib.request.urlopen(req, timeout=30) as resp:
+                payload = json.loads(resp.read().decode("utf-8"))
+        except urllib.error.HTTPError as e:
+            detail = e.read().decode("utf-8", errors="replace")
+            raise TvdbError(f"TheTVDB HTTP {e.code}: {detail}") from e
+        except urllib.error.URLError as e:
+            raise TvdbError(f"TheTVDB network error: {e.reason}") from e
+        except json.JSONDecodeError as e:
+            raise TvdbError("TheTVDB returned invalid JSON") from e
+
+        if payload.get("status") == "failure":
+            msg = payload.get("message") or "Unknown TheTVDB error"
+            raise TvdbError(msg)
+        return payload
+
+    def login(self) -> None:
+        if not self.api_key:
+            raise TvdbError("TheTVDB API key is required")
+        body: dict[str, str] = {"apikey": self.api_key}
+        if self.pin:
+            body["pin"] = self.pin
+        payload = self._request("POST", "/login", body=body)
+        token = payload.get("data", {}).get("token")
+        if not token:
+            raise TvdbError("TheTVDB login did not return a token")
+        self._token = token
+
+    def search_series(self, query: str, limit: int = 25) -> list[TvdbSeries]:
+        query = query.strip()
+        if not query:
+            return []
+        payload = self._request(
+            "GET",
+            "/search",
+            params={"query": query, "type": "series", "limit": limit},
+        )
+        results: list[TvdbSeries] = []
+        for item in payload.get("data") or []:
+            if item.get("type") != "series":
+                continue
+            try:
+                series_id = int(item["id"])
+            except (KeyError, TypeError, ValueError):
+                continue
+            year = item.get("year")
+            if year is not None:
+                year = str(year)
+            results.append(
+                TvdbSeries(
+                    id=series_id,
+                    name=item.get("name") or item.get("title") or f"Series {series_id}",
+                    year=year,
+                )
+            )
+        return results
+
+    def get_season_episodes(
+        self,
+        series_id: int,
+        season: int,
+        season_type: str = "default",
+    ) -> list[TvdbEpisode]:
+        episodes: list[TvdbEpisode] = []
+        page = 0
+        while True:
+            payload = self._request(
+                "GET",
+                f"/series/{series_id}/episodes/{season_type}",
+                params={"page": page, "season": season},
+            )
+            batch = (payload.get("data") or {}).get("episodes") or []
+            if not batch:
+                break
+            for ep in batch:
+                if ep.get("seasonNumber") != season:
+                    continue
+                number = ep.get("number")
+                name = ep.get("name")
+                if number is None or not name:
+                    continue
+                episodes.append(
+                    TvdbEpisode(
+                        number=int(number),
+                        season_number=int(season),
+                        name=str(name),
+                    )
+                )
+            links = payload.get("links") or {}
+            if not links.get("next"):
+                break
+            page += 1
+        episodes.sort(key=lambda e: e.number)
+        return episodes