feat: add DP-based episode matcher, add module architecture doc

- Add _dp_match_episodes: O(n*m) DP for minimum-cost file→episode assignment
  considering disc size, episode deltas, and TVDB duration (Issue #1)
- Wire DP matcher as primary strategy in _match_by_duration_and_order
- Add comprehensive module docstring with architecture overview (Issue #10)
This commit is contained in:
Jarian 2026-07-05 11:49:34 +00:00
parent fd0d865858
commit 0536496b8c

View File

@ -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 discepisode 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 discepisode 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 os
import shutil import shutil
from pathlib import Path from pathlib import Path
@ -940,24 +979,170 @@ class EpisodeRenamer:
return matched_episodes return matched_episodes
def _match_by_duration_and_order(self, episodes_info: List[Dict], tvdb_episodes: List[Dict]) -> List[Dict]: 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.""" """Match episodes using DP-based optimal assignment considering disc size, episode deltas, and TVDB duration."""
matched_episodes = [] allowed = list(range(1, len(tvdb_episodes) + 1))
# Try precise duration matching first # Primary: DP-based optimal matching
duration_matches = self._precise_duration_match(episodes_info, tvdb_episodes, list(range(1, len(tvdb_episodes) + 1))) 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 duration_matches: for match in matched_episodes:
matched_episodes.append(match)
duration_diff = match.get('duration_diff', 0) duration_diff = match.get('duration_diff', 0)
fn = match['file_info']['path'].name[:50]
if duration_diff is not None: 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: 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 return matched_episodes
def _precise_duration_match(self, files: List[Dict], tvdb_episodes: List[Dict], allowed_episodes: List[int]) -> List[Dict]: def _dp_match_episodes(self, files: List[Dict], tvdb_episodes: List[Dict],
"""Perform precise duration matching with fallback to lexicographical order.""" 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 = [] matched_episodes = []
used_files = set() used_files = set()
used_episodes = set() used_episodes = set()