diff --git a/VERSION b/VERSION index 6d7de6e..21e8796 100644 --- a/VERSION +++ b/VERSION @@ -1 +1 @@ -1.0.2 +1.0.3 diff --git a/build_appimage.sh b/build_appimage.sh index 2729013..fc24e83 100755 --- a/build_appimage.sh +++ b/build_appimage.sh @@ -19,6 +19,8 @@ pyinstaller --noconfirm --onedir --windowed \ --hidden-import=engine \ --hidden-import=engine.rules \ --hidden-import=engine.pipeline \ + --hidden-import=engine.tvdb_client \ + --hidden-import=engine.episode_match \ --hidden-import=gui \ --hidden-import=gui.main_window \ --hidden-import=gui.rule_widgets \ diff --git a/engine/__init__.py b/engine/__init__.py index f76f33c..7019dad 100644 --- a/engine/__init__.py +++ b/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", diff --git a/engine/episode_match.py b/engine/episode_match.py new file mode 100644 index 0000000..a77c26c --- /dev/null +++ b/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 diff --git a/engine/rules.py b/engine/rules.py index 5056cd4..eba7b0a 100644 --- a/engine/rules.py +++ b/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 = "" diff --git a/engine/tvdb_client.py b/engine/tvdb_client.py new file mode 100644 index 0000000..cdac6b1 --- /dev/null +++ b/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 diff --git a/gui/main_window.py b/gui/main_window.py index 01e8b11..4bed955 100644 --- a/gui/main_window.py +++ b/gui/main_window.py @@ -36,6 +36,7 @@ from .rule_widgets import ( CaseRuleWidget, NumberingRuleWidget, EpisodeRenumberRuleWidget, + TvdbEpisodeRenumberRuleWidget, PrefixSuffixRuleWidget, CsvMappingRuleWidget, ) @@ -85,6 +86,7 @@ class MainWindow(QMainWindow): CaseRuleWidget(), NumberingRuleWidget(), EpisodeRenumberRuleWidget(), + TvdbEpisodeRenumberRuleWidget(), PrefixSuffixRuleWidget(), CsvMappingRuleWidget(), ] @@ -96,12 +98,17 @@ class MainWindow(QMainWindow): "5. Case", "6. Numbering", "7. Episode renumber", - "8. Prefix / Suffix", - "9. CSV mapping", + "8. TheTVDB episode match", + "9. Prefix / Suffix", + "10. CSV mapping", ] for title, w in zip(rule_titles, self._rule_widgets): w.enabled_cb.toggled.connect(self._refresh_preview) # always refresh when checkbox toggled w.ruleChanged.connect(self._on_rule_changed) # refresh only when enabled rule’s options change + if hasattr(w, "set_file_names"): + w.set_file_names(self._file_names) + if hasattr(w, "matchCompleted"): + w.matchCompleted.connect(self._refresh_preview) g = QGroupBox(title) g_layout = QVBoxLayout(g) g_layout.setContentsMargins(8, 12, 8, 8) @@ -206,7 +213,13 @@ class MainWindow(QMainWindow): self.preview_table.setSortingEnabled(False) self.preview_table.setRowCount(0) self.preview_status.setText("Add a folder to see files.") + for w in self._rule_widgets: + if hasattr(w, "set_file_names"): + w.set_file_names([]) return + for w in self._rule_widgets: + if hasattr(w, "set_file_names"): + w.set_file_names(self._file_names) # Preserve sort and selection so checking a rule doesn’t reorder or clear selection header = self.preview_table.horizontalHeader() sort_section = header.sortIndicatorSection() diff --git a/gui/rule_widgets.py b/gui/rule_widgets.py index cbfd317..272e361 100644 --- a/gui/rule_widgets.py +++ b/gui/rule_widgets.py @@ -15,8 +15,9 @@ from PyQt6.QtWidgets import ( QStackedWidget, QPushButton, QFileDialog, + QMessageBox, ) -from PyQt6.QtCore import pyqtSignal +from PyQt6.QtCore import pyqtSignal, QThread from engine.rules import ( ReplaceRule, @@ -25,10 +26,13 @@ from engine.rules import ( CaseRule, NumberingRule, EpisodeRenumberRule, + TvdbEpisodeRenumberRule, RegexRule, PrefixSuffixRule, CsvMappingRule, ) +from engine.tvdb_client import TvdbClient, TvdbError, TVDB_API_KEY, SEASON_TYPE_CHOICES +from engine.episode_match import match_filenames_to_episodes class ReplaceRuleWidget(QWidget): @@ -305,6 +309,228 @@ class EpisodeRenumberRuleWidget(QWidget): return r +class _TvdbSearchWorker(QThread): + finished = pyqtSignal(list) + failed = pyqtSignal(str) + + def __init__(self, query: str): + super().__init__() + self.query = query + + def run(self): + try: + client = TvdbClient(TVDB_API_KEY) + results = client.search_series(self.query) + self.finished.emit(results) + except TvdbError as e: + self.failed.emit(str(e)) + except Exception as e: + self.failed.emit(str(e)) + + +class _TvdbMatchWorker(QThread): + finished = pyqtSignal(dict, list, list) + failed = pyqtSignal(str) + + def __init__( + self, + series_id: int, + season: int, + season_type: str, + filenames: list[str], + ): + super().__init__() + self.series_id = series_id + self.season = season + self.season_type = season_type + self.filenames = filenames + + def run(self): + try: + client = TvdbClient(TVDB_API_KEY) + episodes = client.get_season_episodes( + self.series_id, + self.season, + season_type=self.season_type, + ) + if not episodes: + raise TvdbError(f"No episodes found for season {self.season}") + mapping, unmatched, notes = match_filenames_to_episodes(self.filenames, episodes) + self.finished.emit(mapping, unmatched, notes) + except TvdbError as e: + self.failed.emit(str(e)) + except Exception as e: + self.failed.emit(str(e)) + + +class TvdbEpisodeRenumberRuleWidget(QWidget): + ruleChanged = pyqtSignal() + matchCompleted = pyqtSignal() + + def __init__(self, parent=None): + super().__init__(parent) + self._file_names: list[str] = [] + self._episode_mapping: dict[str, int] = {} + self._search_results: list = [] + self._search_worker: _TvdbSearchWorker | None = None + self._match_worker: _TvdbMatchWorker | None = None + + layout = QFormLayout(self) + self.enabled_cb = QCheckBox("Use this rule") + self.enabled_cb.setChecked(False) + self.enabled_cb.toggled.connect(self._emit) + layout.addRow(self.enabled_cb) + + layout.addRow("Source:", QLabel("TheTVDB.com")) + + search_row = QHBoxLayout() + self.search_query = QLineEdit() + self.search_query.setPlaceholderText("Show name to search") + self.search_query.returnPressed.connect(self._search_series) + search_btn = QPushButton("Search") + search_btn.clicked.connect(self._search_series) + search_row.addWidget(self.search_query, 1) + search_row.addWidget(search_btn) + layout.addRow("Show:", search_row) + + self.series_combo = QComboBox() + self.series_combo.setEnabled(False) + self.series_combo.currentIndexChanged.connect(self._emit) + layout.addRow("Pick show:", self.series_combo) + + self.order_type = QComboBox() + for label, value in SEASON_TYPE_CHOICES: + self.order_type.addItem(label, value) + self.order_type.currentIndexChanged.connect(self._emit) + layout.addRow("Episode order:", self.order_type) + + self.season = QSpinBox() + self.season.setMinimum(0) + self.season.setMaximum(99) + self.season.setValue(1) + self.season.valueChanged.connect(self._emit) + layout.addRow("Season:", self.season) + + self.padding = QSpinBox() + self.padding.setMinimum(1) + self.padding.setMaximum(3) + self.padding.setValue(2) + self.padding.valueChanged.connect(self._emit) + layout.addRow("Zero-pad width:", self.padding) + + self.match_btn = QPushButton("Match titles from file list") + self.match_btn.clicked.connect(self._match_titles) + layout.addRow(self.match_btn) + + self.status = QLabel("Search for a show, choose episode order and season, then match titles.") + self.status.setWordWrap(True) + layout.addRow(self.status) + + info = QLabel( + "Reads episode titles from filenames like S01E05 - Episode Title. " + "Fetches the chosen TheTVDB episode order for that season and rewrites SxxExx numbers to match." + ) + info.setWordWrap(True) + layout.addRow(info) + + def set_file_names(self, names: list[str]) -> None: + self._file_names = list(names) + + def _order_label(self) -> str: + return self.order_type.currentText() + + def _order_value(self) -> str: + value = self.order_type.currentData() + return value if value else "default" + + def _emit(self): + self.ruleChanged.emit() + + def _search_series(self): + query = self.search_query.text().strip() + if not query: + QMessageBox.warning(self, "Search", "Enter a show name to search.") + return + self.match_btn.setEnabled(False) + self.series_combo.setEnabled(False) + self.status.setText("Searching TheTVDB…") + self._search_worker = _TvdbSearchWorker(query) + self._search_worker.finished.connect(self._on_search_finished) + self._search_worker.failed.connect(self._on_search_failed) + self._search_worker.start() + + def _on_search_finished(self, results): + self._search_results = results + self.series_combo.clear() + if not results: + self.status.setText("No series found. Try a different search.") + self.match_btn.setEnabled(True) + return + for s in results: + label = s.name + if s.year: + label = f"{label} ({s.year})" + self.series_combo.addItem(label, s.id) + self.series_combo.setEnabled(True) + self.match_btn.setEnabled(True) + self.status.setText(f"Found {len(results)} series. Pick one and match titles.") + self._emit() + + def _on_search_failed(self, message: str): + self.match_btn.setEnabled(True) + self.series_combo.setEnabled(self.series_combo.count() > 0) + self.status.setText(f"Search failed: {message}") + + def _match_titles(self): + if self.series_combo.count() == 0: + QMessageBox.warning(self, "Match", "Search and select a show first.") + return + if not self._file_names: + QMessageBox.warning(self, "Match", "Load a folder with files first.") + return + series_id = self.series_combo.currentData() + if series_id is None: + return + self.match_btn.setEnabled(False) + self.status.setText(f"Fetching episodes ({self._order_label()}) and matching titles…") + self._match_worker = _TvdbMatchWorker( + int(series_id), + self.season.value(), + self._order_value(), + self._file_names, + ) + self._match_worker.finished.connect(self._on_match_finished) + self._match_worker.failed.connect(self._on_match_failed) + self._match_worker.start() + + def _on_match_finished(self, mapping: dict, unmatched: list, notes: list): + self._episode_mapping = mapping + self.match_btn.setEnabled(True) + matched = len(mapping) + total = len(self._file_names) + msg = ( + f"Matched {matched} of {total} file(s) for season {self.season.value()} " + f"({self._order_label()})." + ) + if unmatched: + msg += f" {len(unmatched)} file(s) unmatched." + self.status.setText(msg) + self.matchCompleted.emit() + self._emit() + + def _on_match_failed(self, message: str): + self.match_btn.setEnabled(True) + self.status.setText(f"Match failed: {message}") + + def getRule(self) -> TvdbEpisodeRenumberRule: + r = TvdbEpisodeRenumberRule( + episode_mapping=dict(self._episode_mapping), + padding=self.padding.value(), + ) + r.enabled = self.enabled_cb.isChecked() + return r + + class PrefixSuffixRuleWidget(QWidget): ruleChanged = pyqtSignal()