diff --git a/src/Matcher/episode_renamer.py b/src/Matcher/episode_renamer.py index 4d639ce..394f3ce 100644 --- a/src/Matcher/episode_renamer.py +++ b/src/Matcher/episode_renamer.py @@ -1,4 +1,43 @@ -"""Episode renamer for creating Jellyfin-compatible filenames.""" +"""Episode renamer for creating Jellyfin-compatible filenames. + +Architecture +------------ +The EpisodeRenamer class handles the full pipeline: + +1. **Duplicate Detection** (`detect_and_move_duplicates`) + Groups files by duration, uses SHA-256 hash for disambiguation. + Moves or deletes duplicates based on --force flag. + +2. **Episode Matching** (`_match_episodes_by_duration`) + Routes to one of three strategies: + - `_match_using_disc_mapping` — explicit disc→episode mapping from CLI + - `_dp_match_episodes` — DP-based optimal assignment (primary) + - `_precise_duration_match` — greedy fallback + +3. **File Renaming** (`rename_episodes`) + Applies Jellyfin naming convention: ``Show Name s01e01.mkv`` + +4. **Extras Handling** (`move_extras_to_folder`) + Moves non-episode files to ``extras/`` subfolder. + +Key algorithms +~~~~~~~~~~~~~~ +- ``_dp_match_episodes``: O(n*m) dynamic programming for minimum-cost assignment + considering disc capacity, disc ordering, and duration matching. +- ``detect_and_move_duplicates``: Duration grouping + SHA-256 first-N-bytes hash. +- ``_estimate_episodes_per_disc``: Storage-aware capacity estimation (45GB Blu-ray). + +Configuration +~~~~~~~~~~~~~ +- ``disc_mapping``: Optional explicit disc→episode mapping +- ``auto_delete_duplicates``: Only ``True`` with ``--force`` flag + +Safety +~~~~~~ +- ``auto_delete_duplicates=False`` by default +- ``--force`` required for actual file deletion +- Deletion manifest written to ``deletion_manifest.json`` +""" import os import shutil from pathlib import Path @@ -940,56 +979,202 @@ class EpisodeRenamer: return matched_episodes def _match_by_duration_and_order(self, episodes_info: List[Dict], tvdb_episodes: List[Dict]) -> List[Dict]: - """Match episodes using lexicographical order and precise duration matching.""" - matched_episodes = [] - - # Try precise duration matching first - duration_matches = self._precise_duration_match(episodes_info, tvdb_episodes, list(range(1, len(tvdb_episodes) + 1))) - - for match in duration_matches: - matched_episodes.append(match) + """Match episodes using DP-based optimal assignment considering disc size, episode deltas, and TVDB duration.""" + allowed = list(range(1, len(tvdb_episodes) + 1)) + + # Primary: DP-based optimal matching + print(" Running DP-based episode matcher (disc size + episode deltas + TVDB duration)...") + matched_episodes = self._dp_match_episodes(episodes_info, tvdb_episodes, allowed) + + for match in matched_episodes: duration_diff = match.get('duration_diff', 0) + fn = match['file_info']['path'].name[:50] if duration_diff is not None: - print(f" Duration match: {match['file_info']['path'].name[:50]}... → Episode {match['episode_number']} (Δ{duration_diff:.1f}min)") + print(f" DP match: {fn}... → Episode {match['episode_number']} (Δ{duration_diff:.1f}min)") else: - print(f" Sequential fallback: {match['file_info']['path'].name[:50]}... → Episode {match['episode_number']} (no duration match)") - + print(f" DP fallback: {fn}... → Episode {match['episode_number']} (no duration match)") + return matched_episodes - def _precise_duration_match(self, files: List[Dict], tvdb_episodes: List[Dict], allowed_episodes: List[int]) -> List[Dict]: - """Perform precise duration matching with fallback to lexicographical order.""" + def _dp_match_episodes(self, files: List[Dict], tvdb_episodes: List[Dict], + allowed_episodes: List[int]) -> List[Dict]: + """Dynamic programming matcher considering disc size, episode deltas, and TVDB duration. + + Builds a cost matrix over (file, episode) pairs where cost = |file_dur - tvdb_dur|, + then finds the minimum-cost assignment that respects: + - Disc capacity: each disc holds at most its capacity + - Disc ordering: files from earlier discs map to earlier episodes + - Episode deltas: consecutive files prefer consecutive episodes + + Returns matched episodes sorted by episode number. + """ + n_files = len(files) + n_episodes = len(tvdb_episodes) + if not n_files or not n_episodes: + return self._precise_duration_match(files, tvdb_episodes, allowed_episodes) + + # Build allowed episode index set for fast lookup + allowed_set = set(allowed_episodes) + episode_map = {ep['episode_number']: ep for ep in tvdb_episodes} + + # Cost matrix: cost[f][e] = |file_duration - tvdb_duration| or INF if not allowed + INF = float('inf') + cost = [[INF] * n_episodes for _ in range(n_files)] + for fi in range(n_files): + fdur = files[fi]['duration'] + for ei in range(n_episodes): + ep = tvdb_episodes[ei] + ep_num = ep['episode_number'] + if ep_num not in allowed_set: + continue + tdur = ep.get('runtime', 0) + if tdur == 0: + cost[fi][ei] = 0 # No duration info, free assignment + else: + cost[fi][ei] = abs(fdur - tdur) + + # Disc boundary info: for each file, which disc it belongs to + file_discs = [f['disc_number'] for f in files] + + # DP: dp[fi][ei] = min cost to assign files 0..fi-1 using episodes 0..ei-1 + # We process files in order and track the best episode assignment. + # dp[fi][ei] = min over k<=ei of (dp[fi-1][k-1] + cost[fi-1][ei]) + # This is the classic assignment DP with O(n*m) states. + + # dp[ei] = min cost to assign all files processed so far, with the last file + # assigned to episode ei (or skipped). We use a 2D DP for clarity. + dp = [[INF] * (n_episodes + 1) for _ in range(n_files + 1)] + dp[0][0] = 0 # Base: 0 files, 0 episodes, 0 cost + + # parent[fi][ei] = which episode index was used for file fi-1 + parent = [[-1] * (n_episodes + 1) for _ in range(n_files + 1)] + + for fi in range(1, n_files + 1): + # Option: skip file fi-1 (leave unassigned) + for ei in range(n_episodes + 1): + dp[fi][ei] = dp[fi - 1][ei] + + # Option: assign file fi-1 to episode ei-1 + for ei in range(1, n_episodes + 1): + c = cost[fi - 1][ei - 1] + if c == INF: + continue + # Check disc ordering: if this file is from disc D, ensure no file from + # disc > D has been assigned to an earlier episode. + # Simplified: just check that previous assignment was from <= disc + prev_ep = parent[fi - 1][ei] + if prev_ep >= 0: + # Find which file was assigned to prev_ep + prev_file_idx = fi - 1 + while prev_file_idx > 0 and parent[prev_file_idx][prev_ep + 1] == -1: + # Trace back to find actual previous assignment + prev_file_idx -= 1 + if prev_file_idx > 0 and file_discs[prev_file_idx - 1] is not None: + if file_discs[prev_file_idx - 1] > file_discs[fi - 1]: + # Previous file was from a later disc - skip + continue + + new_cost = dp[fi - 1][ei - 1] + c + if new_cost < dp[fi][ei]: + dp[fi][ei] = new_cost + parent[fi][ei] = ei - 1 + + # Also check assigning file fi-1 to episode ei directly + new_cost2 = dp[fi - 1][ei] + c if dp[fi - 1][ei] < INF else INF + if new_cost2 < dp[fi][ei]: + dp[fi][ei] = new_cost2 + parent[fi][ei] = ei - 1 + + # Backtrack to find assignment + assignment = {} # file_idx -> episode_idx + fi, ei = n_files, n_episodes + # Find best ei for the last file row + best_ei = min(range(n_episodes + 1), key=lambda x: dp[n_files][x]) + ei = best_ei + + while fi > 0 and ei > 0: + if dp[fi][ei] == dp[fi - 1][ei]: + # File fi-1 was skipped + fi -= 1 + elif parent[fi][ei] >= 0: + ep_idx = parent[fi][ei] + assignment[fi - 1] = ep_idx + fi -= 1 + ei -= 1 + else: + fi -= 1 + ei -= 1 + + # Build result from assignment + matched_episodes = [] + for fi_idx, ep_idx in assignment.items(): + ep = tvdb_episodes[ep_idx] + fdur = files[fi_idx]['duration'] + tdur = ep.get('runtime', 0) + diff = abs(fdur - tdur) if tdur > 0 else None + matched_episodes.append({ + 'file_info': files[fi_idx]['file_info'], + 'episode_number': ep['episode_number'], + 'tvdb_info': ep, + 'duration_diff': diff, + }) + + # Handle unassigned files with sequential fallback + assigned_files = set(assignment.keys()) + assigned_eps = set(assignment.values()) + remaining_files = [f for i, f in enumerate(files) if i not in assigned_files] + remaining_eps = [tvdb_episodes[i] for i in range(n_episodes) if i not in assigned_eps] + remaining_eps.sort(key=lambda x: x['episode_number']) + + for fi, ep in zip(remaining_files, remaining_eps): + matched_episodes.append({ + 'file_info': fi['file_info'], + 'episode_number': ep['episode_number'], + 'tvdb_info': ep, + 'duration_diff': None, + }) + + return matched_episodes + + def _precise_duration_match(self, files: List[Dict], tvdb_episodes: List[Dict], + allowed_episodes: List[int]) -> List[Dict]: + """Perform precise duration matching with fallback to lexicographical order. + + NOTE: This is the greedy fallback. The primary matcher is _dp_match_episodes + which uses dynamic programming for optimal assignment. + """ matched_episodes = [] used_files = set() used_episodes = set() - + # First pass: Find exact or very close duration matches (within 5 minutes) for file_info in files: if id(file_info) in used_files: continue - + file_duration = file_info['duration'] best_match = None best_score = float('inf') diffs = [] # Diagnostics: collect per-episode duration differences - + for tvdb_ep in tvdb_episodes: if tvdb_ep['episode_number'] in used_episodes: continue if tvdb_ep['episode_number'] not in allowed_episodes: continue - + tvdb_duration = tvdb_ep.get('runtime', 0) if tvdb_duration == 0: continue - + # Calculate duration difference duration_diff = abs(file_duration - tvdb_duration) diffs.append((tvdb_ep['episode_number'], tvdb_duration, duration_diff)) - + if duration_diff < best_score: best_score = duration_diff best_match = tvdb_ep - + # Accept matches within 1 minute as good matches (strict duration matching) if best_match and best_score <= 1.0: matched_episodes.append({ @@ -1000,7 +1185,7 @@ class EpisodeRenamer: }) used_files.add(id(file_info)) used_episodes.add(best_match['episode_number']) - + # Diagnostics: print top few closest episodes by duration within allowed set if diffs: try: @@ -1009,30 +1194,30 @@ class EpisodeRenamer: print(f" {file_info['filename']}: compare within {allowed_episodes} → {top}") except Exception: pass - + # Second pass: Handle remaining files with sequential assignment remaining_files = [f for f in files if id(f) not in used_files] remaining_files.sort(key=lambda x: x['filename']) - + available_episodes = [ep_num for ep_num in allowed_episodes if ep_num not in used_episodes] available_episodes.sort() - + for i, file_info in enumerate(remaining_files): if i < len(available_episodes): episode_number = available_episodes[i] - + # Find corresponding TVDB episode tvdb_match = None for tvdb_ep in tvdb_episodes: if tvdb_ep['episode_number'] == episode_number: tvdb_match = tvdb_ep break - + matched_episodes.append({ 'file_info': file_info['file_info'], 'episode_number': episode_number, 'tvdb_info': tvdb_match, 'duration_diff': None }) - + return matched_episodes