Fix TheTVDB matching for 04x01 filenames and add all-seasons mode.

Parse NxNN and SxxExx layouts, match across every season when season is 0, and rewrite season plus episode numbers.

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

engine/episode_match.py

@@ -4,63 +4,149 @@ Match episode titles from filenames against a reference episode list (e.g. TheTV
 from __future__ import annotations
 
 import re
+from dataclasses import dataclass
 from difflib import SequenceMatcher
 from typing import Optional
 
 from .tvdb_client import TvdbEpisode
 
-DEFAULT_EPISODE_PATTERN = r"(.*?[Ss]\d+[Ee])(\d+)(-[Ee](\d+))?(.*)"
+# S01E05 - Title or S01E05-E06 - Title
+PATTERN_SXXEXX = re.compile(
+    r"^(.*?[Ss])(\d+)([Ee])(\d+)(-[Ee](\d+))?(.*)$",
+)
+# Show Name 04x01 Title, 4x01 - Title, etc.
+PATTERN_NXNN = re.compile(
+    r"^(.*?)(\d{1,2})[xX](\d{1,4})(?:([\s._-]+)(.+))?$",
+)
+
+DEFAULT_EPISODE_PATTERN = PATTERN_SXXEXX.pattern
+
+
+@dataclass(frozen=True)
+class EpisodeTarget:
+    season: int
+    episode: int
 
 
 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()
+    t = re.sub(r"\s+", " ", t).strip()
+    if t.startswith("the "):
+        t = t[4:]
+    return t
 
 
-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
+def _clean_title(rest: str) -> str:
     title = rest.strip()
+    for prefix in ("- ", "– ", "_ ", ". "):
+        if title.startswith(prefix):
+            title = title[len(prefix) :].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,
-    }
+    return title
+
+
+def parse_episode_stem(stem: str, pattern: str = DEFAULT_EPISODE_PATTERN) -> Optional[dict]:
+    """
+    Parse common TV filename stems (S01E05 or 04x01 styles).
+    Returns dict with format, season, episode, title, padding hints — or None.
+    """
+    del pattern  # legacy param; auto-detect formats instead
+
+    m = PATTERN_SXXEXX.match(stem)
+    if m:
+        prefix_before_s, season_s, _e, ep_s, _dash, ep2_s, rest = m.groups()
+        try:
+            season = int(season_s)
+            old_first = int(ep_s)
+        except ValueError:
+            return None
+        span = 1
+        if ep2_s is not None:
+            try:
+                old_second = int(ep2_s)
+            except ValueError:
+                return None
+            span = old_second - old_first + 1
+            if span < 1:
+                span = 1
+        title = _clean_title(rest)
+        return {
+            "format": "sxxexx",
+            "prefix_before_s": prefix_before_s,
+            "season": season,
+            "season_pad": len(season_s),
+            "episode_pad": len(ep_s),
+            "old_first": old_first,
+            "span": span,
+            "title": title,
+            "suffix": rest,
+        }
+
+    m = PATTERN_NXNN.match(stem)
+    if m:
+        prefix, season_s, ep_s, sep, title_part = m.groups()
+        try:
+            season = int(season_s)
+            episode = int(ep_s)
+        except ValueError:
+            return None
+        title = _clean_title(title_part or "")
+        sep = sep or " "
+        if title and not sep.strip():
+            sep = " "
+        return {
+            "format": "nxnn",
+            "prefix": prefix,
+            "season": season,
+            "season_pad": len(season_s),
+            "episode_pad": len(ep_s),
+            "old_first": episode,
+            "span": 1,
+            "title": title,
+            "title_sep": sep,
+        }
+
+    return None
+
+
+def rewrite_episode_stem(
+    stem: str,
+    target: EpisodeTarget,
+    padding: int = 2,
+    pattern: str = DEFAULT_EPISODE_PATTERN,
+) -> str:
+    """Replace season/episode block in stem, preserving layout and title."""
+    parsed = parse_episode_stem(stem, pattern)
+    if not parsed:
+        return stem
+
+    pad = max(1, padding)
+    new_season = target.season
+    new_ep = target.episode
+    title = parsed["title"]
+    span = parsed["span"]
+
+    if parsed["format"] == "nxnn":
+        s_pad = max(parsed["season_pad"], len(str(new_season)))
+        e_pad = max(parsed["episode_pad"], pad, len(str(new_ep)))
+        block = f"{new_season:0{s_pad}d}x{new_ep:0{e_pad}d}"
+        if title:
+            return f"{parsed['prefix']}{block}{parsed['title_sep']}{title}"
+        return f"{parsed['prefix']}{block}"
+
+    s_pad = max(parsed["season_pad"], len(str(new_season)))
+    e_pad = max(parsed["episode_pad"], pad, len(str(new_ep)))
+    e1 = str(new_ep).zfill(e_pad)
+    prefix = f"{parsed['prefix_before_s']}S{new_season:0{s_pad}d}E"
+    if span <= 1:
+        return f"{prefix}{e1}{parsed['suffix']}"
+    e2 = str(new_ep + span - 1).zfill(e_pad)
+    return f"{prefix}{e1}-E{e2}{parsed['suffix']}"
 
 
 def rewrite_episode_number(
@@ -69,25 +155,16 @@ def rewrite_episode_number(
     padding: int = 2,
     pattern: str = DEFAULT_EPISODE_PATTERN,
 ) -> str:
-    """Replace episode number block in stem, preserving prefix, span, and title."""
+    """Legacy helper: episode only, keep season from filename."""
     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}"
+    return rewrite_episode_stem(
+        stem,
+        EpisodeTarget(season=parsed["season"], episode=new_first_ep),
+        padding=padding,
+        pattern=pattern,
+    )
 
 
 def _similarity(a: str, b: str) -> float:
@@ -98,55 +175,73 @@ def _similarity(a: str, b: str) -> float:
     return SequenceMatcher(None, a, b).ratio()
 
 
+def _coerce_target(value: EpisodeTarget | tuple[int, int] | int, parsed: dict) -> EpisodeTarget:
+    if isinstance(value, EpisodeTarget):
+        return value
+    if isinstance(value, tuple):
+        return EpisodeTarget(season=int(value[0]), episode=int(value[1]))
+    return EpisodeTarget(season=parsed["season"], episode=int(value))
+
+
 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]]:
+    min_score: float = 0.65,
+    season_filter: int = 0,
+) -> tuple[dict[str, EpisodeTarget], list[str], list[str]]:
     """
-    Match filenames to TheTVDB episode numbers by title.
+    Match filenames to TheTVDB episodes by title.
 
-    Returns:
-        mapping: original filename -> correct episode number
-        unmatched_files: filenames that could not be matched
-        notes: human-readable match details for UI
+    season_filter: 0 = use all episodes; else only episodes from that season.
+    Returns mapping filename -> (season, episode), unmatched list, notes.
     """
-    file_entries: list[tuple[str, str, str]] = []
+    if season_filter > 0:
+        episodes = [ep for ep in episodes if ep.season_number == season_filter]
+
+    file_entries: list[tuple[str, str, str, dict]] = []
     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
+        if season_filter > 0 and parsed["season"] != season_filter:
+            continue
         norm = normalize_title(parsed["title"])
         if norm:
-            file_entries.append((name, norm, parsed["title"]))
+            file_entries.append((name, norm, parsed["title"], parsed))
 
-    ep_entries = [(ep.number, normalize_title(ep.name), ep.name) for ep in episodes]
+    ep_entries = [
+        (ep.season_number, 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:
+    pairs: list[tuple[float, str, int, int, str, str]] = []
+    for fname, fnorm, raw_title, _parsed in file_entries:
+        for season, ep_num, enorm, ep_name in ep_entries:
             score = _similarity(fnorm, enorm)
-            pairs.append((score, fname, ep_num, raw_title, ep_name))
+            pairs.append((score, fname, season, ep_num, raw_title, ep_name))
 
-    pairs.sort(key=lambda x: (-x[0], x[1], x[2]))
+    pairs.sort(key=lambda x: (-x[0], x[1], x[2], x[3]))
 
-    mapping: dict[str, int] = {}
+    mapping: dict[str, EpisodeTarget] = {}
     used_files: set[str] = set()
-    used_eps: set[int] = set()
+    used_eps: set[tuple[int, int]] = set()
     notes: list[str] = []
 
-    for score, fname, ep_num, raw_title, ep_name in pairs:
+    for score, fname, season, ep_num, raw_title, ep_name in pairs:
         if score < min_score:
             break
-        if fname in used_files or ep_num in used_eps:
+        ep_key = (season, ep_num)
+        if fname in used_files or ep_key in used_eps:
             continue
-        mapping[fname] = ep_num
+        mapping[fname] = EpisodeTarget(season=season, episode=ep_num)
         used_files.add(fname)
-        used_eps.add(ep_num)
+        used_eps.add(ep_key)
         pct = int(round(score * 100))
-        notes.append(f"{fname}: E{ep_num:02d} ← “{ep_name}” ({pct}% match, file title “{raw_title}”)")
+        notes.append(
+            f"{fname}: S{season:02d}E{ep_num:02d} ← “{ep_name}” ({pct}% match, file title “{raw_title}”)"
+        )
 
     unmatched_files = [
         name for name in filenames

engine/rules.py

@@ -9,7 +9,7 @@ from dataclasses import dataclass, field
 from pathlib import Path
 from typing import List, Optional
 
-from .episode_match import rewrite_episode_number
+from .episode_match import rewrite_episode_stem, EpisodeTarget, _coerce_target, parse_episode_stem
 
 
 @dataclass
@@ -279,12 +279,16 @@ class TvdbEpisodeRenumberRule(Rule):
     ) -> 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:
+        target_raw = self.episode_mapping.get(original_name)
+        if target_raw is None:
             return stem, ext
-        new_stem = rewrite_episode_number(
+        parsed = parse_episode_stem(stem, self.pattern)
+        if not parsed:
+            return stem, ext
+        target = _coerce_target(target_raw, parsed)
+        new_stem = rewrite_episode_stem(
             stem,
-            int(new_ep),
+            target,
             padding=self.padding,
             pattern=self.pattern,
         )

engine/tvdb_client.py

@@ -189,5 +189,44 @@ class TvdbClient:
             if not links.get("next"):
                 break
             page += 1
-        episodes.sort(key=lambda e: e.number)
+        episodes.sort(key=lambda e: (e.season_number, e.number))
+        return episodes
+
+    def get_all_episodes(
+        self,
+        series_id: int,
+        season_type: str = "default",
+    ) -> list[TvdbEpisode]:
+        """Fetch every episode for a series (all seasons), paginated."""
+        episodes: list[TvdbEpisode] = []
+        page = 0
+        while True:
+            payload = self._request(
+                "GET",
+                f"/series/{series_id}/episodes/{season_type}",
+                params={"page": page},
+            )
+            batch = (payload.get("data") or {}).get("episodes") or []
+            if not batch:
+                break
+            for ep in batch:
+                number = ep.get("number")
+                season = ep.get("seasonNumber")
+                name = ep.get("name")
+                if number is None or season is None or not name:
+                    continue
+                if int(season) == 0:
+                    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.season_number, e.number))
         return episodes