"""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 from typing import List, Dict, Optional import re class EpisodeRenamer: """Renames episode files to Jellyfin format and moves extras to subfolder.""" def __init__(self, folder_path: str, show_name: str, season_number: int, disc_mapping: Optional[Dict[int, List[int]]] = None, auto_delete_duplicates: bool = False): """Initialize with folder path, show name, season number, optional disc mapping, and auto-delete option.""" self.folder_path = Path(folder_path) self.show_name = show_name self.season_number = season_number self.disc_mapping = disc_mapping # {disc_num: [episode_numbers]} self.auto_delete_duplicates = auto_delete_duplicates self.extras_folder = self.folder_path / "extras" self.delete_folder = self.folder_path / "delete me" def _create_extras_folder(self) -> bool: """Create extras folder if it doesn't exist.""" try: self.extras_folder.mkdir(exist_ok=True) return True except Exception as e: print(f"Error creating extras folder: {e}") return False def _create_delete_folder(self) -> bool: """Create delete me folder if it doesn't exist.""" try: self.delete_folder.mkdir(exist_ok=True) return True except Exception as e: print(f"Error creating delete me folder: {e}") return False def _sanitize_filename(self, filename: str) -> str: """Sanitize filename to remove invalid characters.""" # Remove invalid characters for file systems invalid_chars = ['<', '>', ':', '"', '/', '\\', '|', '?', '*'] for char in invalid_chars: filename = filename.replace(char, '') # Replace multiple spaces with single space filename = re.sub(r'\s+', ' ', filename) return filename.strip() def _generate_episode_filename(self, episode_number: int, original_extension: str = ".mkv") -> str: """Generate Jellyfin-compatible episode filename.""" # Format: "Show Name s01e01.mkv" season_str = f"{self.season_number:02d}" episode_str = f"{episode_number:02d}" filename = f"{self.show_name} s{season_str}e{episode_str}{original_extension}" return self._sanitize_filename(filename) def move_extras_to_folder(self, extras: List[Dict]) -> List[str]: """Move extra files to the extras folder.""" if not self._create_extras_folder(): return [] moved_files = [] for extra in extras: source_path = extra['path'] dest_path = self.extras_folder / source_path.name try: # Check if destination already exists if dest_path.exists(): print(f"Warning: {dest_path.name} already exists in extras folder, skipping.") continue shutil.move(str(source_path), str(dest_path)) moved_files.append(str(dest_path)) print(f"Moved to extras: {source_path.name}") except Exception as e: print(f"Error moving {source_path.name} to extras: {e}") return moved_files def _compute_file_hash(self, file_path: Path, num_bytes: int = 65536) -> str: """Compute a hash of the first N bytes of a file for duplicate detection.""" import hashlib h = hashlib.sha256() try: with open(file_path, 'rb') as f: h.update(f.read(num_bytes)) return h.hexdigest() except Exception: return "" def detect_and_move_duplicates(self, episodes: List[Dict]) -> List[str]: """Detect duplicate episodes and either delete them or move them to delete me folder. Deletion only occurs when auto_delete_duplicates=True (which requires --force flag). Without --force, duplicates are moved to 'delete me' folder for review. """ processed_files = [] duplicates_found = [] # Group episodes by duration (within 1 minute tolerance) duration_groups = {} for episode in episodes: duration = episode['duration_minutes'] # Find existing group within 1 minute tolerance matching_group = None for group_duration in duration_groups.keys(): if abs(duration - group_duration) <= 0.5: # Tighter tolerance for exact duplicates matching_group = group_duration break if matching_group: duration_groups[matching_group].append(episode) else: duration_groups[duration] = [episode] # Identify duplicates (groups with more than one episode) for duration, group_episodes in duration_groups.items(): if len(group_episodes) > 1: print(f"\nFound {len(group_episodes)} potential duplicates with ~{duration:.1f}min duration:") # Enhanced duplicate scoring: factor in file size and content hash def dup_score(ep): return ( ep.get('disc_number', 999) if 'disc_number' in ep else 999, -ep['size_gb'], ep['path'].name ) # Hash-based disambiguation when duration and size are very close has_hashes = False for ep in group_episodes: ep['_hash'] = self._compute_file_hash(ep['path']) if ep['_hash']: has_hashes = True if has_hashes: # Group by hash — identical hash = near-certain duplicate hash_groups = {} for ep in group_episodes: hash_groups.setdefault(ep['_hash'], []).append(ep) for hash_val, hash_eps in hash_groups.items(): if len(hash_eps) > 1: hash_eps.sort(key=dup_score) print(f" Keeping: {hash_eps[0]['path'].name} ({hash_eps[0]['size_gb']:.2f}GB) [hash match]") for dup in hash_eps[1:]: print(f" Duplicate (hash): {dup['path'].name} ({dup['size_gb']:.2f}GB)") duplicates_found.append(dup) elif len(hash_eps) == 1: # Unique hash — still check against duration group group_episodes_copy = [e for e in group_episodes if e['_hash'] == hash_val] if len(group_episodes_copy) == 1: continue # Sort by disc number (keep lower disc numbers) then file size (largest first) for consistent ordering group_episodes.sort(key=dup_score) # Keep the first file (from earliest disc, largest size), mark others as duplicates keeper = group_episodes[0] duplicates = group_episodes[1:] disc_info = "" if hasattr(keeper['path'], 'name'): # Try to extract disc info from filename for display disc_match = re.search(r'Disc\s*(\d+)', keeper['path'].name, re.IGNORECASE) if disc_match: disc_info = f" (Disc {disc_match.group(1)})" print(f" Keeping: {keeper['path'].name}{disc_info} ({keeper['size_gb']:.2f}GB)") for duplicate in duplicates: dup_disc_info = "" if hasattr(duplicate['path'], 'name'): disc_match = re.search(r'Disc\s*(\d+)', duplicate['path'].name, re.IGNORECASE) if disc_match: dup_disc_info = f" (Disc {disc_match.group(1)})" print(f" Duplicate: {duplicate['path'].name}{dup_disc_info} ({duplicate['size_gb']:.2f}GB)") duplicates_found.append(duplicate) # Handle duplicates based on auto_delete_duplicates flag if self.auto_delete_duplicates: import datetime manifest_path = self.folder_path / "deletion_manifest.json" manifest_entries = [] for duplicate in duplicates_found: source_path = duplicate['path'] try: manifest_entries.append({ 'filename': source_path.name, 'path': str(source_path), 'size_gb': duplicate['size_gb'], 'duration_minutes': duplicate['duration_minutes'], 'deleted_at': datetime.datetime.now().isoformat() }) source_path.unlink() processed_files.append(str(source_path)) print(f"Deleted duplicate: {source_path.name}") except Exception as e: print(f"Error deleting {source_path.name}: {e}") try: import json as _json with open(manifest_path, 'w') as f: _json.dump(manifest_entries, f, indent=2) print(f"\nDeletion manifest written to: {manifest_path}") except Exception as e: print(f"Warning: Could not write deletion manifest: {e}") else: # Move duplicates to delete me folder (original behavior) if not self._create_delete_folder(): return [] for duplicate in duplicates_found: source_path = duplicate['path'] dest_path = self.delete_folder / source_path.name try: # Check if destination already exists if dest_path.exists(): print(f"Warning: {dest_path.name} already exists in delete me folder, skipping.") continue shutil.move(str(source_path), str(dest_path)) processed_files.append(str(dest_path)) print(f"Moved to delete me: {source_path.name}") except Exception as e: print(f"Error moving {source_path.name} to delete me: {e}") return processed_files def _extract_disc_info(self, filename: str) -> Dict: """Extract disc number and track number from filename.""" # Look for patterns like "Disc 1", "Disk 1", "Disc1", "disc-1", "DISC 1_t03" disc_match = re.search(r'(?i)\bdis[ck][\s._-]*(\d{1,3})(?!\d)', filename) disc_number = int(disc_match.group(1)) if disc_match else None # Look for track patterns like "_t01", "_t22", "track01", allow small separators track_match = re.search(r'(?i)(?:_t|track)[\s._-]?(\d+)', filename) track_number = int(track_match.group(1)) if track_match else None return { 'disc_number': disc_number, 'track_number': track_number, 'filename': filename } def _estimate_episodes_per_disc(self, episodes_info: List[Dict], total_episodes: int) -> Dict[int, List[int]]: """Estimate which episodes are on which disc based on file distribution and storage constraints.""" # Group files by disc and sort within each disc discs = {} for info in episodes_info: disc = info['disc_number'] if disc: if disc not in discs: discs[disc] = [] discs[disc].append(info) if not discs: return {} # Sort files within each disc by filename (lexicographical order) for disc_num in discs: discs[disc_num].sort(key=lambda x: x['filename']) # Sort discs by number sorted_disc_nums = sorted(discs.keys()) episodes_per_disc = {} # Calculate storage-aware episode distribution episode_number = 1 for disc_num in sorted_disc_nums: files_on_disc = discs[disc_num] episodes_on_this_disc = [] # Calculate total storage used by episodes on this disc total_disc_size_gb = sum(info['file_info']['size_gb'] for info in files_on_disc) # Estimate Blu-ray capacity constraints # Standard Blu-ray: ~45GB usable (48GB total - overhead) # Account for extras that might also be on the disc estimated_episode_capacity_gb = 45.0 # Conservative estimate # If this disc is near capacity, reduce episode count to be conservative if total_disc_size_gb > 40.0: # Stricter threshold: 40GB # This disc is likely at capacity limits # Reduce episode count by 1 if we're close to limits available_episodes = max(1, len(files_on_disc) - 1) available_episodes = min(available_episodes, total_episodes - episode_number + 1) print(f" Warning: Disc {disc_num} near capacity ({total_disc_size_gb:.1f}GB), reducing episode allocation") elif total_disc_size_gb > 35.0: # Warning threshold: 35GB # Show warning but don't reduce count yet available_episodes = min(len(files_on_disc), total_episodes - episode_number + 1) print(f" Warning: Disc {disc_num} approaching capacity ({total_disc_size_gb:.1f}GB), may have storage constraints") else: available_episodes = min(len(files_on_disc), total_episodes - episode_number + 1) for _ in range(available_episodes): if episode_number <= total_episodes: episodes_on_this_disc.append(episode_number) episode_number += 1 episodes_per_disc[disc_num] = episodes_on_this_disc # Diagnostics: print estimated mapping per disc with storage info print("\nEstimated episodes-per-disc mapping (considering storage constraints):") total_assigned = 0 for d in sorted(episodes_per_disc.keys()): files_on_disc = [f for f in episodes_info if f['disc_number'] == d] total_size = sum(f['file_info']['size_gb'] for f in files_on_disc) print(f" Disc {d}: Episodes {episodes_per_disc[d]} ({len(files_on_disc)} files, {total_size:.1f}GB)") total_assigned += len(episodes_per_disc[d]) # Flag potential storage issues if len(files_on_disc) > len(episodes_per_disc[d]): extra_files = len(files_on_disc) - len(episodes_per_disc[d]) print(f" ⚠ Warning: {extra_files} extra files detected - may indicate storage constraints") # Ensure we haven't lost any episodes due to capacity constraints if total_assigned < total_episodes: missing_episodes = total_episodes - total_assigned print(f" ⚠ Warning: {missing_episodes} episodes unassigned due to capacity constraints") # Try to assign remaining episodes to discs with available space remaining_episodes = [] for ep in range(1, total_episodes + 1): assigned = False for disc_episodes in episodes_per_disc.values(): if ep in disc_episodes: assigned = True break if not assigned: remaining_episodes.append(ep) # Find discs with the most available space to assign remaining episodes if remaining_episodes: disc_space = {} for d in sorted(episodes_per_disc.keys()): files_on_disc = [f for f in episodes_info if f['disc_number'] == d] total_size = sum(f['file_info']['size_gb'] for f in files_on_disc) available_space = 45.0 - total_size # Remaining Blu-ray capacity disc_space[d] = available_space # Assign remaining episodes to discs with most space for ep in remaining_episodes: best_disc = max(disc_space.keys(), key=lambda d: disc_space[d]) episodes_per_disc[best_disc].append(ep) disc_space[best_disc] -= 10.0 # Rough estimate: reduce available space print(f" ↻ Reassigning Episode {ep} to Disc {best_disc} (most available space)") return episodes_per_disc def rename_episodes(self, episodes: List[Dict], tvdb_episodes: Optional[List[Dict]] = None) -> List[Dict]: """ Rename episode files to Jellyfin format. Args: episodes: List of episode file info dictionaries tvdb_episodes: Optional TVDB episode data for validation Returns: List of rename operations performed """ if not episodes: print("No episodes to rename.") return [] # If we have TVDB data, try to match by duration if tvdb_episodes: matched_episodes = self._match_episodes_by_duration(episodes, tvdb_episodes) else: # Fallback: assume files are in episode order when sorted by size sorted_episodes = sorted(episodes, key=lambda x: x['size_bytes'], reverse=True) matched_episodes = [] for i, episode in enumerate(sorted_episodes, 1): matched_episodes.append({ 'file_info': episode, 'episode_number': i, 'tvdb_info': None }) # Perform renames renamed_files = [] for match in matched_episodes: file_info = match['file_info'] episode_number = match['episode_number'] source_path = file_info['path'] # Get file extension original_extension = source_path.suffix # Generate new filename new_filename = self._generate_episode_filename(episode_number, original_extension) new_path = source_path.parent / new_filename # Check if destination already exists if new_path.exists() and new_path != source_path: print(f"Warning: {new_filename} already exists, skipping rename of {source_path.name}") continue try: # Rename the file source_path.rename(new_path) rename_info = { 'original_name': source_path.name, 'new_name': new_filename, 'episode_number': episode_number, 'file_size_gb': file_info['size_gb'], 'duration_minutes': file_info['duration_minutes'] } if match['tvdb_info']: rename_info['tvdb_duration'] = match['tvdb_info']['runtime'] rename_info['tvdb_name'] = match['tvdb_info']['name'] renamed_files.append(rename_info) print(f"Renamed: {source_path.name} → {new_filename}") except Exception as e: print(f"Error renaming {source_path.name}: {e}") return renamed_files def _match_episodes_by_duration(self, episodes: List[Dict], tvdb_episodes: List[Dict]) -> List[Dict]: """Match episode files with TVDB episode data. Use disc mapping if provided, otherwise use duration matching.""" matched_episodes = [] # Extract disc and track information episodes_info = [] for episode in episodes: # Handle both path objects and filename strings if isinstance(episode, dict) and 'path' in episode: filename = episode['path'].name elif isinstance(episode, dict) and 'name' in episode: filename = episode['name'] else: print(f"Warning: Unexpected episode format: {episode}") continue disc_info = self._extract_disc_info(filename) episodes_info.append({ 'file_info': episode, 'disc_number': disc_info['disc_number'], 'track_number': disc_info['track_number'], 'filename': filename, 'duration': episode.get('duration_minutes', 0) }) # Sort by lexicographical order (this should match episode order) episodes_info.sort(key=lambda x: x['filename']) # Detect and handle duplicates before processing print("\n=== Duplicate Detection ===") episode_files = [info['file_info'] for info in episodes_info] duplicate_files = self.detect_and_move_duplicates(episode_files) # Remove duplicates from episodes_info if duplicate_files: duplicate_names = {Path(f).name for f in duplicate_files} episodes_info = [info for info in episodes_info if info['filename'] not in duplicate_names] print(f"Removed {len(duplicate_files)} duplicates from episode processing") # Use disc mapping if provided if self.disc_mapping: print("\n=== Using Disc Mapping ===") matched_episodes = self._match_using_disc_mapping(episodes_info, tvdb_episodes) else: # Original complex matching logic for when no disc mapping is provided print("\n=== Using Duration/Disc Analysis ===") matched_episodes = self._match_using_duration_analysis(episodes_info, tvdb_episodes) # Sort by episode number for final output matched_episodes.sort(key=lambda x: x['episode_number']) return matched_episodes def _match_using_disc_mapping(self, episodes_info: List[Dict], tvdb_episodes: List[Dict]) -> List[Dict]: """Match episodes using the provided disc mapping.""" matched_episodes = [] # Group files by disc disc_groups = {} for info in episodes_info: disc_num = info['disc_number'] if disc_num is not None: disc_groups.setdefault(disc_num, []).append(info) print(f"Files found on {len(disc_groups)} discs") for disc_num in sorted(disc_groups.keys()): files = disc_groups[disc_num] expected_episodes = self.disc_mapping.get(disc_num, []) print(f"Disc {disc_num}: {len(files)} files, expecting episodes {expected_episodes}") # Sort files on this disc by filename files.sort(key=lambda x: x['filename']) # Match files to episodes in order for i, file_info in enumerate(files): if i < len(expected_episodes): episode_number = expected_episodes[i] # Find the TVDB episode tvdb_episode = None for ep in tvdb_episodes: if ep['episode_number'] == episode_number: tvdb_episode = ep break matched_episodes.append({ 'file_info': file_info['file_info'], 'episode_number': episode_number, 'tvdb_info': tvdb_episode }) print(f" Episode {episode_number}: {file_info['filename']}") else: print(f" Warning: Extra file on disc {disc_num}: {file_info['filename']}") return matched_episodes def _match_using_duration_analysis(self, episodes_info: List[Dict], tvdb_episodes: List[Dict]) -> List[Dict]: """Use the original complex duration/disc analysis matching.""" # Diagnostics: overview of discs/files and durations try: disc_groups = {} for info in episodes_info: disc_groups.setdefault(info['disc_number'], []).append(info) print("=== Disc/File Overview ===") for disc_key in sorted([d for d in disc_groups.keys() if d is not None]): files = sorted(disc_groups[disc_key], key=lambda x: x['filename']) print(f"Disc {disc_key}: {len(files)} candidate file(s)") for f in files: try: print(f" - {f['filename']} ({f['duration']:.1f} min)") except Exception: pass except Exception: pass # If we have disc information, enforce disc→episode ordering so early discs map to earlier episodes if any(info['disc_number'] for info in episodes_info): total_eps = len(tvdb_episodes) episodes_per_disc = self._estimate_episodes_per_disc(episodes_info, total_eps) if episodes_per_disc: print("Found disc information - enforcing disc-to-episode order constraints...") matched_episodes = self._match_by_duration_and_disc(episodes_info, tvdb_episodes, episodes_per_disc) else: print("Found disc information - using flexible duration-first matching...") matched_episodes = self._flexible_duration_match(episodes_info, tvdb_episodes) else: # No disc info, use lexicographical order with precise duration matching print("No disc information found, using lexicographical order with precise duration matching...") matched_episodes = self._match_by_duration_and_order(episodes_info, tvdb_episodes) return matched_episodes def _flexible_duration_match(self, episodes_info: List[Dict], tvdb_episodes: List[Dict]) -> List[Dict]: """Match episodes using sequential assignment with constraint validation.""" # Sort episodes info by lexicographical order (this determines episode sequence) sorted_episodes_info = sorted(episodes_info, key=lambda x: x['filename']) # Analyze disc capacity constraints disc_capacity = self._analyze_disc_capacity(sorted_episodes_info) print(f" Sequential assignment ({len(sorted_episodes_info)} files → {len(tvdb_episodes)} episodes)...") if disc_capacity: print(f" Disc capacity constraints: {disc_capacity}") # Use strict sequential assignment with constraint validation assignments = self._sequential_assignment_with_constraints(sorted_episodes_info, tvdb_episodes, disc_capacity) return assignments def _sequential_assignment_with_constraints(self, files: List[Dict], tvdb_episodes: List[Dict], disc_capacity: Dict[int, int]) -> List[Dict]: """Assign files to episodes sequentially while respecting all constraints.""" assignments = [] # Track disc usage disc_assignments = {} for disc_num in disc_capacity: disc_assignments[disc_num] = 0 # Track which episodes have been assigned used_episodes = set() # Go through files in lexicographical order for file_idx, file_info in enumerate(files): best_assignment = None best_score = float('inf') # For each file, try to assign it to the next available episode in sequential order for episode_number in range(1, len(tvdb_episodes) + 1): if episode_number in used_episodes: continue # Find the TVDB episode for this episode number tvdb_episode = None for ep in tvdb_episodes: if ep['episode_number'] == episode_number: tvdb_episode = ep break if not tvdb_episode: continue # Check all constraints if self._validate_sequential_assignment(file_info, tvdb_episode, disc_assignments, disc_capacity): file_duration = file_info['duration'] tvdb_duration = tvdb_episode.get('runtime', 0) if tvdb_duration > 0: duration_diff = abs(file_duration - tvdb_duration) # Prefer episodes closer to the file's natural position, but also consider duration position_penalty = abs(episode_number - (file_idx + 1)) * 0.5 # Small penalty for out-of-order total_score = duration_diff + position_penalty if total_score < best_score: best_score = total_score best_assignment = (episode_number, tvdb_episode, duration_diff) # If we find a very good match (within 1 minute and correct position), take it immediately if duration_diff <= 1.0 and episode_number == (file_idx + 1): break else: # No TVDB duration, just assign by position if episode_number == (file_idx + 1): best_assignment = (episode_number, tvdb_episode, None) break # Make the best assignment we found if best_assignment: episode_number, tvdb_episode, duration_diff = best_assignment # Update tracking used_episodes.add(episode_number) file_disc = file_info['disc_number'] if file_disc and file_disc in disc_assignments: disc_assignments[file_disc] += 1 assignments.append({ 'file_info': file_info['file_info'], 'episode_number': episode_number, 'tvdb_info': tvdb_episode, 'duration_diff': duration_diff, 'assignment_cost': duration_diff or 0 }) disc_num = file_info['disc_number'] or "?" if duration_diff is not None: if duration_diff <= 1.0: print(f" Disc {disc_num}: {file_info['filename'][:50]}... → Episode {episode_number} (Δ{duration_diff:.1f}min) ✓") elif duration_diff <= 2.0: print(f" Disc {disc_num}: {file_info['filename'][:50]}... → Episode {episode_number} (Δ{duration_diff:.1f}min) ~ acceptable") else: print(f" Disc {disc_num}: {file_info['filename'][:50]}... → Episode {episode_number} (Δ{duration_diff:.1f}min) ⚠ large difference") else: print(f" Disc {disc_num}: {file_info['filename'][:50]}... → Episode {episode_number} (no TVDB duration)") else: # No valid assignment found disc_num = file_info['disc_number'] or "?" print(f" Disc {disc_num}: {file_info['filename'][:50]}... → No valid assignment (all constraints violated)") # Sort by episode number for output assignments.sort(key=lambda x: x['episode_number']) return assignments def _validate_sequential_assignment(self, file_info: Dict, tvdb_episode: Dict, disc_assignments: Dict[int, int], disc_capacity: Dict[int, int]) -> bool: """Validate a sequential assignment against all constraints.""" file_duration = file_info['duration'] tvdb_duration = tvdb_episode.get('runtime', 0) file_disc = file_info['disc_number'] # Constraint 1: Duration difference must be ≤ 1 minute (but allow if no better options) if tvdb_duration > 0: duration_diff = abs(file_duration - tvdb_duration) if duration_diff > 1.0: # Allow violations if duration is close (within 2 minutes) for edge cases if duration_diff > 2.0: return False # Constraint 2: Video length should be >= TVDB duration (with reasonable tolerance) if tvdb_duration > 0: # Allow files to be up to 2 minutes shorter (encoding differences) if file_duration < (tvdb_duration - 2.0): return False # Constraint 3: Disc capacity constraint (strict) if file_disc and disc_capacity: max_episodes_on_disc = disc_capacity.get(file_disc, 0) current_episodes_on_disc = disc_assignments.get(file_disc, 0) if current_episodes_on_disc >= max_episodes_on_disc: return False return True def _find_optimal_assignment(self, files: List[Dict], tvdb_episodes: List[Dict], disc_capacity: Dict[int, int]) -> List[Dict]: """Find optimal assignment using constraint-aware matching.""" assignments = [] used_files = set() used_episodes = set() # First pass: Find perfect duration matches that satisfy all constraints for file_info in files: if id(file_info) in used_files: continue best_match = None best_score = float('inf') for tvdb_ep in tvdb_episodes: if tvdb_ep['episode_number'] in used_episodes: continue # Check if this assignment would be valid file_duration = file_info['duration'] tvdb_duration = tvdb_ep.get('runtime', 0) # Apply constraints if not self._assignment_satisfies_constraints(file_info, tvdb_ep, assignments, disc_capacity): continue # Calculate match quality if tvdb_duration > 0: duration_diff = abs(file_duration - tvdb_duration) # Prefer matches within 1 minute if duration_diff <= 1.0: score = duration_diff if score < best_score: best_score = score best_match = tvdb_ep # Accept good matches if best_match and best_score <= 1.0: assignments.append({ 'file_info': file_info['file_info'], 'episode_number': best_match['episode_number'], 'tvdb_info': best_match, 'duration_diff': best_score, 'assignment_cost': best_score }) used_files.add(id(file_info)) used_episodes.add(best_match['episode_number']) disc_num = file_info['disc_number'] or "?" print(f" Disc {disc_num} optimal match: {file_info['filename'][:50]}... → Episode {best_match['episode_number']} (Δ{best_score:.1f}min)") # Second pass: Sequential assignment for remaining files (respecting order) remaining_files = [f for f in files if id(f) not in used_files] remaining_episodes = [ep for ep in tvdb_episodes if ep['episode_number'] not in used_episodes] remaining_episodes.sort(key=lambda x: x['episode_number']) for file_info in remaining_files: if not remaining_episodes: break # Try to assign to the next available episode in order for tvdb_ep in remaining_episodes[:]: if self._assignment_satisfies_constraints(file_info, tvdb_ep, assignments, disc_capacity): file_duration = file_info['duration'] tvdb_duration = tvdb_ep.get('runtime', 0) duration_diff = abs(file_duration - tvdb_duration) if tvdb_duration > 0 else None assignments.append({ 'file_info': file_info['file_info'], 'episode_number': tvdb_ep['episode_number'], 'tvdb_info': tvdb_ep, 'duration_diff': duration_diff, 'assignment_cost': duration_diff or 0 }) remaining_episodes.remove(tvdb_ep) disc_num = file_info['disc_number'] or "?" if duration_diff is not None: print(f" Disc {disc_num} sequential: {file_info['filename'][:50]}... → Episode {tvdb_ep['episode_number']} (Δ{duration_diff:.1f}min)") else: print(f" Disc {disc_num} sequential: {file_info['filename'][:50]}... → Episode {tvdb_ep['episode_number']} (no duration)") break # Sort by episode number for output assignments.sort(key=lambda x: x['episode_number']) return assignments def _assignment_satisfies_constraints(self, file_info: Dict, tvdb_episode: Dict, current_assignments: List[Dict], disc_capacity: Dict[int, int]) -> bool: """Check if assigning a file to an episode satisfies all constraints.""" file_duration = file_info['duration'] tvdb_duration = tvdb_episode.get('runtime', 0) file_disc = file_info['disc_number'] # Constraint 1: Duration difference must be ≤ 1 minute if tvdb_duration > 0: duration_diff = abs(file_duration - tvdb_duration) if duration_diff > 1.0: return False # Constraint 2: Video length should be >= TVDB duration (with small tolerance) if tvdb_duration > 0: if file_duration < (tvdb_duration - 0.5): return False # Constraint 3: Disc capacity constraint if file_disc and disc_capacity: max_episodes_on_disc = disc_capacity.get(file_disc, 0) # Count current assignments to this disc current_disc_assignments = sum(1 for a in current_assignments if a['file_info']['path'].name.find(f'Disc {file_disc}') != -1) if current_disc_assignments >= max_episodes_on_disc: return False return True def _analyze_disc_capacity(self, episodes_info: List[Dict]) -> Dict[int, int]: """Analyze the maximum number of episodes each disc can hold based on storage constraints.""" disc_capacity = {} # Group files by disc disc_files = {} for info in episodes_info: disc_num = info['disc_number'] if disc_num: if disc_num not in disc_files: disc_files[disc_num] = [] disc_files[disc_num].append(info) # Analyze each disc's capacity considering storage limits for disc_num, files in disc_files.items(): file_count = len(files) total_size_gb = sum(f['file_info']['size_gb'] for f in files) # Blu-ray capacity constraints max_bluray_capacity_gb = 48.0 usable_capacity_gb = 45.0 # Account for overhead and extras if total_size_gb > usable_capacity_gb * 0.8: # 80% threshold # This disc is likely near capacity # Estimate how many episodes can actually fit based on average episode size avg_episode_size = total_size_gb / file_count if file_count > 0 else 10.0 estimated_episodes = min(file_count, int(usable_capacity_gb / avg_episode_size)) # Be conservative - if we're unsure, use the smaller number disc_capacity[disc_num] = min(estimated_episodes, file_count) print(f" Storage constraint analysis for Disc {disc_num}:") print(f" Total files: {file_count}, Total size: {total_size_gb:.1f}GB") print(f" Estimated episode capacity: {estimated_episodes} (limited by {usable_capacity_gb}GB Blu-ray)") else: # Disc has plenty of space, use file count disc_capacity[disc_num] = file_count return disc_capacity def _is_valid_assignment(self, file_info: Dict, tvdb_episode: Dict, disc_capacity: Dict[int, int], episode_index: int, all_files: List[Dict]) -> bool: """Validate if assigning a file to an episode satisfies all constraints.""" file_duration = file_info['duration'] tvdb_duration = tvdb_episode.get('runtime', 0) episode_num = tvdb_episode['episode_number'] file_disc = file_info['disc_number'] # Constraint 1: Duration difference must be ≤ 1 minute if tvdb_duration > 0: duration_diff = abs(file_duration - tvdb_duration) if duration_diff > 1.0: return False # Constraint 2: Video length should be >= TVDB duration # (Allow small tolerance for encoding differences) if tvdb_duration > 0: if file_duration < (tvdb_duration - 0.5): # 30 second tolerance return False # Constraint 3: Disc capacity constraint # In diagonal traversal, we're assigning file_index to episode_index # So we need to check if this disc has already been "filled up" by previous assignments if file_disc and disc_capacity: max_episodes_on_disc = disc_capacity.get(file_disc, 0) # Count how many previous files from this disc have been assigned episodes_already_assigned_to_disc = 0 for prev_file_idx in range(episode_index): # Previous assignments in diagonal if prev_file_idx < len(all_files): prev_file = all_files[prev_file_idx] if prev_file['disc_number'] == file_disc: episodes_already_assigned_to_disc += 1 # Adding this assignment would exceed disc capacity if episodes_already_assigned_to_disc >= max_episodes_on_disc: return False return True def _match_by_duration_and_disc(self, episodes_info: List[Dict], tvdb_episodes: List[Dict], episodes_per_disc: Dict[int, List[int]]) -> List[Dict]: """Match episodes using disc constraints and precise duration matching.""" matched_episodes = [] used_episode_numbers = set() # Process each disc in order for disc_num in sorted(episodes_per_disc.keys()): files_on_disc = [f for f in episodes_info if f['disc_number'] == disc_num] files_on_disc.sort(key=lambda x: x['filename']) allowed_episodes = episodes_per_disc[disc_num] print(f"\nDisc {disc_num}: allowed episode numbers {allowed_episodes}") # Get TVDB episodes for this disc disc_tvdb_episodes = [ep for ep in tvdb_episodes if ep['episode_number'] in allowed_episodes] # Match files to episodes by duration within this disc disc_matches = self._precise_duration_match(files_on_disc, disc_tvdb_episodes, allowed_episodes) for match in disc_matches: if match['episode_number'] not in used_episode_numbers: matched_episodes.append(match) used_episode_numbers.add(match['episode_number']) duration_diff = match.get('duration_diff', 0) if duration_diff is not None: print(f" Disc {disc_num} match: {match['file_info']['path'].name[:50]}... → Episode {match['episode_number']} (Δ{duration_diff:.1f}min)") else: print(f" Disc {disc_num} fallback: {match['file_info']['path'].name[:50]}... → Episode {match['episode_number']} (no duration match)") return matched_episodes def _match_by_duration_and_order(self, episodes_info: List[Dict], tvdb_episodes: List[Dict]) -> List[Dict]: """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" DP match: {fn}... → Episode {match['episode_number']} (Δ{duration_diff:.1f}min)") else: print(f" DP fallback: {fn}... → Episode {match['episode_number']} (no duration match)") return matched_episodes 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) allowed_set = set(allowed_episodes) 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_num = tvdb_episodes[ei]['episode_number'] if ep_num not in allowed_set: continue tdur = tvdb_episodes[ei].get('runtime', 0) cost[fi][ei] = 0.0 if tdur == 0 else abs(fdur - tdur) # DP: dp[i][j] = min cost to assign first i files from first j episodes # Transition: # dp[i][j] = min(dp[i][j-1], # skip episode j-1 # dp[i-1][j-1] + cost) # assign file i-1 to episode j-1 dp = [[INF] * (n_episodes + 1) for _ in range(n_files + 1)] for j in range(n_episodes + 1): dp[0][j] = 0.0 for i in range(1, n_files + 1): dp[i][0] = INF for j in range(1, n_episodes + 1): dp[i][j] = dp[i][j - 1] c = cost[i - 1][j - 1] if c < INF and dp[i - 1][j - 1] < INF: dp[i][j] = min(dp[i][j], dp[i - 1][j - 1] + c) # Backtrack best_j = min(range(1, n_episodes + 1), key=lambda j: dp[n_files][j]) if dp[n_files][best_j] >= INF: return self._fallback_sequential(files, tvdb_episodes, allowed_episodes) assignment = {} i, j = n_files, best_j while i > 0 and j > 0: if dp[i][j] == dp[i][j - 1]: j -= 1 elif dp[i - 1][j - 1] < INF and cost[i - 1][j - 1] < INF: assignment[i - 1] = j - 1 i -= 1 j -= 1 else: break 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, }) assigned_files = set(assignment.keys()) assigned_eps = set(assignment.values()) remaining_files = [f for idx, f in enumerate(files) if idx not in assigned_files] remaining_eps = [tvdb_episodes[idx] for idx in range(n_episodes) if idx 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]: """DP-based optimal duration matching within allowed episodes. Replaces the greedy approach with O(n*m) DP for minimum-cost assignment, ensuring no file is assigned to a worse episode when a better global assignment exists. """ n_files = len(files) allowed_set = set(allowed_episodes) allowed_tvdb = sorted( [ep for ep in tvdb_episodes if ep['episode_number'] in allowed_set], key=lambda x: x['episode_number'], ) n_eps = len(allowed_tvdb) if not n_files or not n_eps: return self._fallback_sequential(files, allowed_tvdb, allowed_episodes) INF = float('inf') # Cost matrix: cost[fi][ei] = |file_duration - tvdb_duration| cost = [[INF] * n_eps for _ in range(n_files)] for fi in range(n_files): fdur = files[fi]['duration'] for ei in range(n_eps): tdur = allowed_tvdb[ei].get('runtime', 0) cost[fi][ei] = 0.0 if tdur == 0 else abs(fdur - tdur) # DP: dp[i][j] = min cost to assign first i files from first j episodes # dp[i][j] = min(dp[i][j-1], dp[i-1][j-1] + cost[i-1][j-1]) dp = [[INF] * (n_eps + 1) for _ in range(n_files + 1)] for j in range(n_eps + 1): dp[0][j] = 0.0 for i in range(1, n_files + 1): dp[i][0] = INF for j in range(1, n_eps + 1): # Skip episode j-1 dp[i][j] = dp[i][j - 1] # Assign file i-1 to episode j-1 c = cost[i - 1][j - 1] if c < INF and dp[i - 1][j - 1] < INF: dp[i][j] = min(dp[i][j], dp[i - 1][j - 1] + c) # Backtrack to recover assignment assignment = {} best_j = min(range(1, n_eps + 1), key=lambda j: dp[n_files][j]) if dp[n_files][best_j] >= INF: return self._fallback_sequential(files, allowed_tvdb, allowed_episodes) i, j = n_files, best_j while i > 0 and j > 0: if j > 0 and dp[i][j] == dp[i][j - 1]: j -= 1 elif dp[i - 1][j - 1] < INF and cost[i - 1][j - 1] < INF: assignment[i - 1] = j - 1 i -= 1 j -= 1 else: break matched_episodes = [] for fi, ei in assignment.items(): ep = allowed_tvdb[ei] fdur = files[fi]['duration'] tdur = ep.get('runtime', 0) diff = abs(fdur - tdur) if tdur > 0 else None matched_episodes.append({ 'file_info': files[fi]['file_info'], 'episode_number': ep['episode_number'], 'tvdb_info': ep, 'duration_diff': diff, 'assignment_cost': diff or 0, }) # Fallback for unassigned files assigned_fi = set(assignment.keys()) assigned_ei = set(assignment.values()) remaining_files = [f for idx, f in enumerate(files) if idx not in assigned_fi] remaining_eps = [e for idx, e in enumerate(allowed_tvdb) if idx not in assigned_ei] 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, 'assignment_cost': 0, }) return matched_episodes def _fallback_sequential(self, files: List[Dict], tvdb_eps: List[Dict], allowed_episodes: List[int]) -> List[Dict]: """Sequential fallback when DP cannot produce an assignment.""" matched = [] sorted_files = sorted(files, key=lambda x: x['filename']) allowed_set = set(allowed_episodes) avail = [ep for ep in tvdb_eps if ep['episode_number'] in allowed_set] for i, fi in enumerate(sorted_files): if i < len(avail): ep = avail[i] matched.append({ 'file_info': fi['file_info'], 'episode_number': ep['episode_number'], 'tvdb_info': ep, 'duration_diff': None, }) return matched