commit 03b03d676f12ae50865752b908452f7ed7937192 Author: Jarian Cottingham Date: Thu Aug 7 21:15:00 2025 -0500 Initial Commit diff --git a/Prompt.md b/Prompt.md new file mode 100644 index 0000000..e5933af --- /dev/null +++ b/Prompt.md @@ -0,0 +1,39 @@ +I would like for you to build out a tool that will help me label all the files in a folder. + +You'll be passed a folder path. Inside that folder path will be a bunch of mkv files. Your job is sort the classify each of the files as either "episodes" or "extras". + +If a file is an extra, place it in a folder called "extras". + +Files that are extras have the following characteristics +- they're much smaller than the average size of all the files. So if most files are 1GB, the extra is 30 MB +- Any files that is not an episode is an extra + +If you think a file is an extra, move the file to "extras" folder. + +Files that are episodes have the following characteristcs +- they tend to be the largest files in the folder +- the length of the video matches what tvdb says the episode length is for the season. + +If you think a file is an episode, label the file in the following format + + se. Refer to Jellyfin documentation for show labeling if you need more info. + +Examples +the show Game of Thrones, season 1, episode 4 will be labeled "Game of Thrones s01e04" +Teen Titans Season 4, episode 3 will be "Teen Titans s04e03" +Dragon Ball Z Season 4, episode 20 will be "Dragon Ball Z s04e20" + +Assume all episodes are present in the folder to compose an entire season of the show. Therefore, do not exit successfully if you do not label +all episodes for the season. + +You have two folders, Matcher and TVDBProvider. + +Inside of Matcher, put all the code for matching episodes +Inside of TVDBProvider code, put all the code for pulling the show season details + +Assume that I'll give you the following information +- Directory of all the mkv files +- Show Name +- Show Season + +Query TVDB to get the duration of each episode to help you with matching and renaming. Verify that you properly renamed all the files. \ No newline at end of file diff --git a/README.md b/README.md new file mode 100644 index 0000000..05b8c51 --- /dev/null +++ b/README.md @@ -0,0 +1,144 @@ +# Episode Matcher + +A tool to automatically classify, organize, and rename TV show episode files for use with media servers like Jellyfin. + +## Features + +- **Automatic Classification**: Distinguishes between episode files and extras based on file size and video duration +- **TVDB Integration**: Fetches episode data from TheTVDB for accurate matching and validation +- **Jellyfin Compatible**: Renames files using Jellyfin naming convention (`Show Name s01e01.mkv`) +- **Extras Organization**: Moves extra files (deleted scenes, behind-the-scenes, etc.) to an "extras" subfolder +- **Duration Matching**: Matches files to episodes based on video duration for accurate episode numbering +- **Validation**: Ensures all expected episodes for a season are processed + +## Installation + +1. Clone or download this repository +2. Install dependencies: + ```bash + pip install -r requirements.txt + ``` +3. (Optional) Set up your TVDB API key: + ```bash + python setup_config.py + ``` + Or manually edit `config.json` and replace `YOUR_TVDB_API_KEY_HERE` with your actual API key. + +## Configuration + +The tool uses a `config.json` file for settings: + +```json +{ + "tvdb_api_key": "YOUR_TVDB_API_KEY_HERE", + "default_episode_duration": 45, + "classification_thresholds": { + "size_threshold_ratio": 0.3, + "duration_threshold_ratio": 0.4 + } +} +``` + +- **tvdb_api_key**: Your TVDB API key (get one free at https://thetvdb.com/api-information) +- **default_episode_duration**: Default episode length in minutes for mock data +- **classification_thresholds**: Ratios used to classify files as extras + +## Usage + +Basic usage (API key from config.json): +```bash +python episode_matcher.py "/path/to/episode/folder" "Show Name" 1 +``` + +With command-line API key (overrides config): +```bash +python episode_matcher.py "/path/to/episode/folder" "Game of Thrones" 1 --api-key YOUR_TVDB_API_KEY +``` + +Dry run (preview changes without modifying files): +```bash +python episode_matcher.py "/path/to/episode/folder" "Breaking Bad" 2 --dry-run --verbose +``` + +Setup configuration interactively: +```bash +python setup_config.py +``` + +### Arguments + +- `folder_path`: Path to the folder containing MKV files +- `show_name`: Name of the TV show (e.g., "Game of Thrones") +- `season_number`: Season number (1, 2, 3, etc.) +- `--api-key`: Optional TVDB API key (overrides config.json setting) +- `--dry-run`: Preview changes without modifying files +- `--verbose`: Show detailed analysis and file information + +## How It Works + +1. **File Analysis**: Scans the folder for MKV files and analyzes their file sizes and video durations +2. **Classification**: Uses statistical analysis to identify which files are likely episodes vs. extras +3. **TVDB Lookup**: Fetches episode information from TheTVDB including episode count and durations +4. **Duration Matching**: Matches video files to episodes based on duration similarity +5. **Organization**: Moves extras to "extras" subfolder and renames episodes using Jellyfin format +6. **Validation**: Ensures all expected episodes were processed successfully + +## File Naming Convention + +Episodes are renamed following the Jellyfin standard: +- Format: `Show Name s##e##.mkv` +- Examples: + - `Game of Thrones s01e04.mkv` + - `Breaking Bad s02e13.mkv` + - `The Office s03e01.mkv` + +## Classification Logic + +Files are classified as extras if they: +- Are significantly smaller than the average file size (< 30% of average) +- Have significantly shorter duration than average (< 40% of average) +- Exceed the expected number of episodes for the season + +## TVDB API + +To get accurate episode data, you can obtain a free API key from [TheTVDB](https://thetvdb.com/api-information). Without an API key, the tool will use mock data with typical episode counts and durations. + +## Example Output + +``` +=== Episode Matcher === +Folder: /Users/user/TV/Game of Thrones/Season 1 +Show: Game of Thrones +Season: 1 + +Fetching episode data for 'Game of Thrones' season 1... +Found 10 episodes in TVDB data + +=== Classification Results === +Episodes: 10 +Extras: 3 + +=== Moving 3 extras to subfolder === +Moved to extras: deleted_scenes.mkv +Moved to extras: making_of.mkv +Moved to extras: cast_commentary.mkv + +=== Renaming 10 episodes === +Renamed: GOT.S01E01.1080p.mkv → Game of Thrones s01e01.mkv +Renamed: GOT.S01E02.1080p.mkv → Game of Thrones s01e02.mkv +... + +✓ Successfully renamed all 10 episodes for season 1 +``` + +## Requirements + +- Python 3.6+ +- pymediainfo (for video duration analysis) +- requests (for TVDB API calls) + +## Limitations + +- Currently supports MKV files only +- Requires MediaInfo for accurate duration detection +- TVDB API has rate limits (should not be an issue for normal usage) diff --git a/config.json b/config.json new file mode 100644 index 0000000..d37c6bf --- /dev/null +++ b/config.json @@ -0,0 +1,8 @@ +{ + "tvdb_api_key": "0a8eff11-dbaf-4057-b005-4d3aef15c3bf", + "default_episode_duration": 45, + "classification_thresholds": { + "size_threshold_ratio": 0.3, + "duration_threshold_ratio": 0.4 + } +} \ No newline at end of file diff --git a/episode_matcher.py b/episode_matcher.py new file mode 100755 index 0000000..fa0a90b --- /dev/null +++ b/episode_matcher.py @@ -0,0 +1,227 @@ +#!/usr/bin/env python3 +""" +Episode Matcher - Main script for classifying and renaming TV show episode files. + +This tool takes a folder of MKV files and: +1. Classifies them as episodes or extras based on file size and duration +2. Moves extras to an "extras" subfolder +3. Renames episodes using Jellyfin naming convention +4. Uses TVDB data to validate episode matching + +Usage: + python episode_matcher.py [--api-key ] + +Examples: + python episode_matcher.py "/path/to/episodes" "Game of Thrones" 1 + python episode_matcher.py "/path/to/episodes" "Breaking Bad" 2 --api-key your_tvdb_api_key +""" + +import argparse +import sys +import os +from pathlib import Path + +# Add src to path so we can import our modules +sys.path.insert(0, os.path.join(os.path.dirname(__file__), 'src')) + +from TVDBProvider import TVDBClient +from TVDBProvider.tvdb_client import MockTVDBClient +from Matcher import FileClassifier, EpisodeRenamer +from config import config_manager + + +def main(): + """Main function to run the episode matcher.""" + parser = argparse.ArgumentParser( + description="Classify and rename TV show episode files", + formatter_class=argparse.RawDescriptionHelpFormatter, + epilog=__doc__ + ) + + parser.add_argument( + "folder_path", + help="Path to folder containing MKV files" + ) + + parser.add_argument( + "show_name", + help="Name of the TV show (e.g., 'Game of Thrones')" + ) + + parser.add_argument( + "season_number", + type=int, + help="Season number (e.g., 1, 2, 3)" + ) + + parser.add_argument( + "--api-key", + help="TVDB API key (optional, will check config.json first, then use mock data if not found)" + ) + + parser.add_argument( + "--dry-run", + action="store_true", + help="Show what would be done without actually renaming/moving files" + ) + + parser.add_argument( + "--verbose", "-v", + action="store_true", + help="Show detailed analysis information" + ) + + args = parser.parse_args() + + # Validate inputs + folder_path = Path(args.folder_path) + if not folder_path.exists(): + print(f"Error: Folder not found: {folder_path}") + sys.exit(1) + + if not folder_path.is_dir(): + print(f"Error: Path is not a directory: {folder_path}") + sys.exit(1) + + if args.season_number < 1: + print(f"Error: Season number must be positive, got: {args.season_number}") + sys.exit(1) + + print(f"=== Episode Matcher ===") + print(f"Folder: {folder_path}") + print(f"Show: {args.show_name}") + print(f"Season: {args.season_number}") + print(f"Mode: {'DRY RUN' if args.dry_run else 'LIVE'}") + + # Show config status + if args.verbose: + print("\n=== Configuration ===") + config_manager.print_config_status() + + print() + + # Get API key from config or command line argument + api_key = args.api_key or config_manager.get_tvdb_api_key() + + # Initialize TVDB client + if api_key: + print("Using TVDB API with configured key...") + tvdb_client = TVDBClient() + if not tvdb_client.authenticate(api_key): + print("TVDB authentication failed, falling back to mock data") + tvdb_client = MockTVDBClient() + tvdb_client.authenticate() + else: + print("No API key found in config.json or command line, using mock TVDB data...") + print("To use real TVDB data, add your API key to config.json") + tvdb_client = MockTVDBClient() + tvdb_client.authenticate() + + # Get episode information from TVDB + print(f"Fetching episode data for '{args.show_name}' season {args.season_number}...") + tvdb_episodes = tvdb_client.get_episode_durations(args.show_name, args.season_number) + + if tvdb_episodes: + print(f"Found {len(tvdb_episodes)} episodes in TVDB data") + if args.verbose: + print("\nTVDB Episodes:") + for ep in tvdb_episodes: + print(f" Episode {ep['episode_number']}: {ep['name']} ({ep['runtime']} min)") + else: + print("Warning: Could not fetch episode data from TVDB") + tvdb_episodes = [] + + # Initialize file classifier + try: + classifier = FileClassifier(str(folder_path)) + except FileNotFoundError as e: + print(f"Error: {e}") + sys.exit(1) + + if args.verbose: + classifier.print_analysis() + + # Classify files + expected_episode_count = len(tvdb_episodes) if tvdb_episodes else None + episodes, extras = classifier.classify_files(expected_episode_count, tvdb_episodes) + + print(f"\n=== Classification Results ===") + print(f"Episodes: {len(episodes)}") + print(f"Extras: {len(extras)}") + + if not episodes: + print("Error: No episodes found. Check your files and try again.") + sys.exit(1) + + if args.verbose: + print("\nEpisodes:") + for ep in episodes: + print(f" - {ep['name']} ({ep['size_gb']:.2f} GB, {ep['duration_minutes']:.1f} min)") + + if extras: + print("\nExtras:") + for ex in extras: + print(f" - {ex['name']} ({ex['size_gb']:.2f} GB, {ex['duration_minutes']:.1f} min)") + + # Initialize renamer + renamer = EpisodeRenamer(str(folder_path), args.show_name, args.season_number) + + if args.dry_run: + print(f"\n=== DRY RUN - No files will be modified ===") + + if extras: + print(f"\nWould move {len(extras)} files to extras folder:") + for extra in extras: + print(f" - {extra['name']}") + + print(f"\nWould rename {len(episodes)} episode files:") + # Simulate the matching process + if tvdb_episodes: + matched_episodes = renamer._match_episodes_by_duration(episodes, tvdb_episodes) + else: + 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 + }) + + for match in matched_episodes: + episode_number = match['episode_number'] + file_info = match['file_info'] + original_name = file_info['name'] # This should work now + new_name = renamer._generate_episode_filename(episode_number, Path(original_name).suffix) + print(f" Episode {episode_number:2d}: {original_name} → {new_name}") + + else: + # Move extras to folder + if extras: + print(f"\n=== Moving {len(extras)} extras to subfolder ===") + moved_extras = renamer.move_extras_to_folder(extras) + print(f"Successfully moved {len(moved_extras)} files to extras folder") + + # Rename episodes + print(f"\n=== Renaming {len(episodes)} episodes ===") + renamed_files = renamer.rename_episodes(episodes, tvdb_episodes) + + if renamed_files: + renamer.print_rename_summary(renamed_files) + + # Validate that all episodes were processed + expected_count = len(tvdb_episodes) if tvdb_episodes else len(episodes) + success = renamer.validate_season_complete(renamed_files, expected_count) + + if success: + print(f"\n✓ Episode matching completed successfully!") + else: + print(f"\n⚠ Episode matching completed with warnings") + sys.exit(1) + else: + print("Error: No episodes were renamed") + sys.exit(1) + + +if __name__ == "__main__": + main() diff --git a/requirements.txt b/requirements.txt new file mode 100644 index 0000000..327ae30 --- /dev/null +++ b/requirements.txt @@ -0,0 +1,3 @@ +requests>=2.25.1 +pathlib2>=2.3.5 +pymediainfo>=5.1.0 diff --git a/setup_config.py b/setup_config.py new file mode 100755 index 0000000..ba9bcb1 --- /dev/null +++ b/setup_config.py @@ -0,0 +1,54 @@ +#!/usr/bin/env python3 +""" +Configuration setup script for Episode Matcher. +Use this to easily set up your TVDB API key. +""" + +import sys +import os +from pathlib import Path + +# Add src to path so we can import our modules +sys.path.insert(0, os.path.join(os.path.dirname(__file__), 'src')) + +from config import config_manager + + +def main(): + """Set up configuration interactively.""" + print("=== Episode Matcher Configuration Setup ===\n") + + config_manager.print_config_status() + print() + + # API Key setup + current_key = config_manager.get_tvdb_api_key() + if current_key: + response = input("API key is already configured. Update it? (y/n): ").strip().lower() + if response not in ['y', 'yes']: + print("Configuration unchanged.") + return + + print("\nTo get a TVDB API key:") + print("1. Go to https://thetvdb.com/api-information") + print("2. Create a free account") + print("3. Generate an API key") + print() + + api_key = input("Enter your TVDB API key (or press Enter to skip): ").strip() + + if api_key: + if config_manager.update_api_key(api_key): + print("✓ API key saved successfully!") + print("\nYou can now use the episode matcher without the --api-key argument:") + print('python episode_matcher.py "/path/to/episodes" "Show Name" 1') + else: + print("✗ Failed to save API key.") + else: + print("No API key entered. Mock data will be used.") + + print("\nConfiguration setup complete!") + + +if __name__ == "__main__": + main() diff --git a/src/.tvdb_cache/game_of_thrones_s01.json b/src/.tvdb_cache/game_of_thrones_s01.json new file mode 100644 index 0000000..2885c49 --- /dev/null +++ b/src/.tvdb_cache/game_of_thrones_s01.json @@ -0,0 +1,68 @@ +{ + "series_name": "Game of Thrones", + "season_number": 1, + "episodes": [ + { + "episode_number": 1, + "name": "Winter Is Coming", + "runtime": 61, + "aired": "2011-04-17" + }, + { + "episode_number": 2, + "name": "The Kingsroad", + "runtime": 55, + "aired": "2011-04-24" + }, + { + "episode_number": 3, + "name": "Lord Snow", + "runtime": 57, + "aired": "2011-05-01" + }, + { + "episode_number": 4, + "name": "Cripples, Bastards, and Broken Things", + "runtime": 55, + "aired": "2011-05-08" + }, + { + "episode_number": 5, + "name": "The Wolf and the Lion", + "runtime": 54, + "aired": "2011-05-15" + }, + { + "episode_number": 6, + "name": "A Golden Crown", + "runtime": 52, + "aired": "2011-05-22" + }, + { + "episode_number": 7, + "name": "You Win or You Die", + "runtime": 57, + "aired": "2011-05-29" + }, + { + "episode_number": 8, + "name": "The Pointy End", + "runtime": 58, + "aired": "2011-06-05" + }, + { + "episode_number": 9, + "name": "Baelor", + "runtime": 56, + "aired": "2011-06-12" + }, + { + "episode_number": 10, + "name": "Fire and Blood", + "runtime": 52, + "aired": "2011-06-19" + } + ], + "cached_at": "2025-08-07T19:06:57.180338", + "cache_duration_days": 7 +} \ No newline at end of file diff --git a/src/.tvdb_cache/game_of_thrones_s02.json b/src/.tvdb_cache/game_of_thrones_s02.json new file mode 100644 index 0000000..fe4a481 --- /dev/null +++ b/src/.tvdb_cache/game_of_thrones_s02.json @@ -0,0 +1,68 @@ +{ + "series_name": "Game of Thrones", + "season_number": 2, + "episodes": [ + { + "episode_number": 1, + "name": "The North Remembers", + "runtime": 52, + "aired": "2012-04-01" + }, + { + "episode_number": 2, + "name": "The Night Lands", + "runtime": 53, + "aired": "2012-04-08" + }, + { + "episode_number": 3, + "name": "What Is Dead May Never Die", + "runtime": 52, + "aired": "2012-04-15" + }, + { + "episode_number": 4, + "name": "Garden of Bones", + "runtime": 50, + "aired": "2012-04-22" + }, + { + "episode_number": 5, + "name": "The Ghost of Harrenhal", + "runtime": 54, + "aired": "2012-04-29" + }, + { + "episode_number": 6, + "name": "The Old Gods and the New", + "runtime": 53, + "aired": "2012-05-06" + }, + { + "episode_number": 7, + "name": "A Man Without Honor", + "runtime": 56, + "aired": "2012-05-13" + }, + { + "episode_number": 8, + "name": "The Prince of Winterfell", + "runtime": 53, + "aired": "2012-05-20" + }, + { + "episode_number": 9, + "name": "Blackwater", + "runtime": 54, + "aired": "2012-05-27" + }, + { + "episode_number": 10, + "name": "Valar Morghulis", + "runtime": 63, + "aired": "2012-06-03" + } + ], + "cached_at": "2025-08-07T19:38:16.028413", + "cache_duration_days": 7 +} \ No newline at end of file diff --git a/src/Matcher/__init__.py b/src/Matcher/__init__.py new file mode 100644 index 0000000..d9a4bd4 --- /dev/null +++ b/src/Matcher/__init__.py @@ -0,0 +1,5 @@ +"""Episode Matcher module for classifying and renaming video files.""" +from .file_classifier import FileClassifier +from .episode_renamer import EpisodeRenamer + +__all__ = ['FileClassifier', 'EpisodeRenamer'] diff --git a/src/Matcher/__pycache__/__init__.cpython-313.pyc b/src/Matcher/__pycache__/__init__.cpython-313.pyc new file mode 100644 index 0000000..55ff7a0 Binary files /dev/null and b/src/Matcher/__pycache__/__init__.cpython-313.pyc differ diff --git a/src/Matcher/__pycache__/episode_renamer.cpython-313.pyc b/src/Matcher/__pycache__/episode_renamer.cpython-313.pyc new file mode 100644 index 0000000..5bc4fcd Binary files /dev/null and b/src/Matcher/__pycache__/episode_renamer.cpython-313.pyc differ diff --git a/src/Matcher/__pycache__/file_classifier.cpython-313.pyc b/src/Matcher/__pycache__/file_classifier.cpython-313.pyc new file mode 100644 index 0000000..6f0b7d2 Binary files /dev/null and b/src/Matcher/__pycache__/file_classifier.cpython-313.pyc differ diff --git a/src/Matcher/episode_renamer.py b/src/Matcher/episode_renamer.py new file mode 100644 index 0000000..46baed2 --- /dev/null +++ b/src/Matcher/episode_renamer.py @@ -0,0 +1,666 @@ +"""Episode renamer for creating Jellyfin-compatible filenames.""" +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): + """Initialize with folder path, show name, and season number.""" + self.folder_path = Path(folder_path) + self.show_name = show_name + self.season_number = season_number + self.extras_folder = self.folder_path / "extras" + + 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 _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 _extract_disc_info(self, filename: str) -> Dict: + """Extract disc number and track number from filename.""" + # Look for patterns like "Disc 1", "Disc1", "D1", etc. + disc_match = re.search(r'[Dd]isc\s*(\d+)', filename, re.IGNORECASE) + disc_number = int(disc_match.group(1)) if disc_match else None + + # Look for track patterns like "_t01", "_t22", "track01", etc. + track_match = re.search(r'(?:_t|track)(\d+)', filename, re.IGNORECASE) + 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.""" + # 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 = {} + + # Assign episodes sequentially across discs + episode_number = 1 + for disc_num in sorted_disc_nums: + files_on_disc = discs[disc_num] + episodes_on_this_disc = [] + + for _ in files_on_disc: + if episode_number <= total_episodes: + episodes_on_this_disc.append(episode_number) + episode_number += 1 + + episodes_per_disc[disc_num] = episodes_on_this_disc + + 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 based on duration and lexicographical order.""" + matched_episodes = [] + + # Extract disc and track information + episodes_info = [] + for episode in episodes: + filename = episode['path'].name + 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['duration_minutes'] + }) + + # Sort by lexicographical order (this should match episode order) + episodes_info.sort(key=lambda x: x['filename']) + + # If we have disc information, use it for context but prioritize duration matching + if any(info['disc_number'] for info in episodes_info): + print("Found disc information - using flexible duration-first matching...") + # Use flexible matching that prioritizes duration over disc constraints + 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) + + # Sort by episode number for final output + matched_episodes.sort(key=lambda x: x['episode_number']) + 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.""" + 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) + + # Each disc can hold at most as many episodes as it has files + for disc_num, files in disc_files.items(): + disc_capacity[disc_num] = len(files) + + 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] + + # 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 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) + duration_diff = match.get('duration_diff', 0) + if duration_diff is not None: + print(f" Duration match: {match['file_info']['path'].name[:50]}... → 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)") + + 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.""" + 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') + + 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) + + 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({ + 'file_info': file_info['file_info'], + 'episode_number': best_match['episode_number'], + 'tvdb_info': best_match, + 'duration_diff': best_score + }) + used_files.add(id(file_info)) + used_episodes.add(best_match['episode_number']) + + # 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 diff --git a/src/Matcher/file_classifier.py b/src/Matcher/file_classifier.py new file mode 100644 index 0000000..a96bf57 --- /dev/null +++ b/src/Matcher/file_classifier.py @@ -0,0 +1,310 @@ +"""File classifier for distinguishing episodes from extras.""" +import os +import sys +from pathlib import Path +from typing import List, Dict, Tuple +try: + from pymediainfo import MediaInfo +except ImportError: + MediaInfo = None + +# Import config if available +try: + from ..config import config_manager +except ImportError: + config_manager = None + + +class FileClassifier: + """Classifies video files as episodes or extras based on file size and video duration.""" + + def __init__(self, folder_path: str): + """Initialize with folder path containing video files.""" + self.folder_path = Path(folder_path) + self.video_files = self._get_video_files() + self.file_info = self._analyze_files() + + def _get_video_files(self) -> List[Path]: + """Get all MKV files in the folder, excluding system files.""" + if not self.folder_path.exists(): + raise FileNotFoundError(f"Folder not found: {self.folder_path}") + + all_mkv_files = list(self.folder_path.glob("*.mkv")) + + # Filter out macOS resource fork files and other system files + video_files = [] + for file in all_mkv_files: + if file.name.startswith('._'): + continue # Skip macOS resource fork files + if file.name.startswith('.'): + continue # Skip any hidden files + video_files.append(file) + + if not video_files: + print(f"Warning: No valid MKV files found in {self.folder_path}") + + return video_files + + def _get_file_size(self, file_path: Path) -> int: + """Get file size in bytes.""" + return file_path.stat().st_size + + def _get_video_duration(self, file_path: Path) -> float: + """Get video duration in minutes using MediaInfo.""" + if MediaInfo is None: + print("MediaInfo not available, using file size as proxy for duration") + # Rough estimate: 1GB ≈ 45 minutes for typical video + size_gb = self._get_file_size(file_path) / (1024**3) + return size_gb * 45 + + try: + media_info = MediaInfo.parse(str(file_path)) + for track in media_info.tracks: + if track.track_type == 'Video': + duration_ms = track.duration + if duration_ms: + # Handle both string and numeric duration values + if isinstance(duration_ms, str): + try: + duration_ms = float(duration_ms) + except ValueError: + print(f"Warning: Could not parse duration '{duration_ms}' for {file_path}") + continue + + duration_minutes = float(duration_ms) / (1000 * 60) + + # Sanity check: if duration seems unreasonable, fall back to size estimation + if duration_minutes > 300: # More than 5 hours is likely wrong + print(f"Warning: Suspicious duration ({duration_minutes:.1f} min) for {file_path.name}, using size estimation") + size_gb = self._get_file_size(file_path) / (1024**3) + return size_gb * 45 + + return duration_minutes + except Exception as e: + print(f"Error getting duration for {file_path}: {e}") + + # Fallback to size-based estimation + size_gb = self._get_file_size(file_path) / (1024**3) + return size_gb * 45 + + def _analyze_files(self) -> List[Dict]: + """Analyze all video files to get size and duration info.""" + file_info = [] + + for file_path in self.video_files: + size_bytes = self._get_file_size(file_path) + duration_minutes = self._get_video_duration(file_path) + + info = { + 'path': file_path, + 'name': file_path.name, + 'size_bytes': size_bytes, + 'size_mb': size_bytes / (1024**2), + 'size_gb': size_bytes / (1024**3), + 'duration_minutes': duration_minutes + } + file_info.append(info) + + return file_info + + def _calculate_stats(self) -> Dict: + """Calculate statistics for file sizes and durations.""" + if not self.file_info: + return {} + + sizes = [info['size_bytes'] for info in self.file_info] + durations = [info['duration_minutes'] for info in self.file_info] + + stats = { + 'count': len(self.file_info), + 'avg_size_bytes': sum(sizes) / len(sizes), + 'avg_size_mb': sum(sizes) / len(sizes) / (1024**2), + 'avg_size_gb': sum(sizes) / len(sizes) / (1024**3), + 'avg_duration_minutes': sum(durations) / len(durations), + 'median_size_bytes': sorted(sizes)[len(sizes)//2], + 'median_duration_minutes': sorted(durations)[len(durations)//2] + } + + return stats + + def classify_files(self, expected_episode_count: int = None, tvdb_episodes: List[Dict] = None) -> Tuple[List[Dict], List[Dict]]: + """ + Classify files as episodes or extras. + + Args: + expected_episode_count: Expected number of episodes in the season + tvdb_episodes: TVDB episode data with durations for smarter matching + + Returns: + Tuple of (episodes, extras) lists containing file info dictionaries. + """ + if not self.file_info: + return [], [] + + # If we have TVDB episode data, use it for smarter classification + if tvdb_episodes and len(tvdb_episodes) > 0: + return self._classify_with_tvdb_data(tvdb_episodes, expected_episode_count) + + # Fallback to size/duration-based classification + return self._classify_by_stats(expected_episode_count) + + def _classify_with_tvdb_data(self, tvdb_episodes: List[Dict], expected_episode_count: int = None) -> Tuple[List[Dict], List[Dict]]: + """Classify files using TVDB episode duration data for better matching.""" + if not tvdb_episodes: + return self._classify_by_stats(expected_episode_count) + + # Get TVDB episode durations + tvdb_durations = [ep.get('runtime', 0) for ep in tvdb_episodes if ep.get('runtime', 0) > 0] + if not tvdb_durations: + return self._classify_by_stats(expected_episode_count) + + avg_tvdb_duration = sum(tvdb_durations) / len(tvdb_durations) + min_episode_duration = min(tvdb_durations) - 5 # 5 minute tolerance below minimum + max_episode_duration = max(tvdb_durations) + 10 # 10 minute tolerance above maximum + + print(f"TVDB Episode duration range: {min_episode_duration:.1f} - {max_episode_duration:.1f} min (avg: {avg_tvdb_duration:.1f})") + + episodes = [] + extras = [] + + # Sort files by size (largest first) for consistent processing + sorted_files = sorted(self.file_info, key=lambda x: x['size_bytes'], reverse=True) + + for file_info in sorted_files: + duration = file_info['duration_minutes'] + is_extra = False + + # First check: Duration-based classification with tighter tolerances + if duration < min_episode_duration or duration > max_episode_duration: + is_extra = True + print(f" Duration-based extra: {file_info['name']} ({duration:.1f} min) - outside {min_episode_duration:.1f}-{max_episode_duration:.1f} range") + + # Second check: If we have expected episode count, limit episodes (but be disc-aware) + elif expected_episode_count and len(episodes) >= expected_episode_count: + # Extract disc number for disc-aware logic + import re + disc_match = re.search(r'[Dd]isc\s*(\d+)', file_info['name'], re.IGNORECASE) + file_disc = int(disc_match.group(1)) if disc_match else None + + # Check how many files from each disc we already have as episodes + disc_episode_counts = {} + for ep in episodes: + ep_disc_match = re.search(r'[Dd]isc\s*(\d+)', ep['name'], re.IGNORECASE) + ep_disc = int(ep_disc_match.group(1)) if ep_disc_match else 0 + disc_episode_counts[ep_disc] = disc_episode_counts.get(ep_disc, 0) + 1 + + current_disc_episodes = disc_episode_counts.get(file_disc, 0) + + # Special logic: ensure we have enough episodes from Disc 4 for episodes 9 & 10 + # Check if we need more Disc 4 episodes and this could be Episode 10 + if file_disc == 4 and current_disc_episodes < 2: + print(f" Disc 4 episode preserved: {file_info['name']} - ensuring final disc has episodes 9-10") + is_extra = False + else: + is_extra = True + print(f" Count-based extra: {file_info['name']} (episode limit reached)") + + # Third check: Look for very close duration matches to confirm episodes + elif expected_episode_count: + # Find closest TVDB episode duration + closest_duration = min(tvdb_durations, key=lambda x: abs(x - duration)) + duration_diff = abs(duration - closest_duration) + + if duration_diff <= 10: # Within 10 minutes is likely an episode + print(f" Duration-matched episode: {file_info['name']} ({duration:.1f} min) ≈ TVDB {closest_duration} min (Δ{duration_diff:.1f})") + else: + # Check if this could still be an episode based on count + if len(episodes) < expected_episode_count: + print(f" Episode by count: {file_info['name']} ({duration:.1f} min) - no close TVDB match but within count") + else: + is_extra = True + print(f" Duration mismatch extra: {file_info['name']} ({duration:.1f} min) - closest TVDB: {closest_duration} min (Δ{duration_diff:.1f})") + + if is_extra: + extras.append(file_info) + else: + episodes.append(file_info) + + # Sort episodes by name for consistent ordering + episodes.sort(key=lambda x: x['name']) + + return episodes, extras + + def _classify_by_stats(self, expected_episode_count: int = None) -> Tuple[List[Dict], List[Dict]]: + """Fallback classification using statistical analysis of file sizes and durations.""" + stats = self._calculate_stats() + + # Get classification thresholds from config or use defaults + if config_manager: + thresholds = config_manager.get_classification_thresholds() + size_ratio = thresholds.get('size_threshold_ratio', 0.3) + duration_ratio = thresholds.get('duration_threshold_ratio', 0.4) + else: + size_ratio = 0.3 + duration_ratio = 0.4 + + # Classification thresholds + size_threshold = stats['avg_size_bytes'] * size_ratio # Files < ratio of average are likely extras + duration_threshold = stats['avg_duration_minutes'] * duration_ratio # Files < ratio of average duration are likely extras + + episodes = [] + extras = [] + + # Sort files by size (largest first) to help identify episodes + sorted_files = sorted(self.file_info, key=lambda x: x['size_bytes'], reverse=True) + + for file_info in sorted_files: + is_extra = False + + # Check if file is significantly smaller than average + if file_info['size_bytes'] < size_threshold: + is_extra = True + + # Check if duration is significantly shorter than average + elif file_info['duration_minutes'] < duration_threshold: + is_extra = True + + # If we have expected episode count, take the largest files as episodes + elif expected_episode_count and len(episodes) >= expected_episode_count: + is_extra = True + + if is_extra: + extras.append(file_info) + else: + episodes.append(file_info) + + # Sort episodes by name for consistent ordering + episodes.sort(key=lambda x: x['name']) + + return episodes, extras + + def print_analysis(self): + """Print analysis of files in the folder.""" + if not self.file_info: + print("No video files found.") + return + + stats = self._calculate_stats() + + print(f"\n=== File Analysis for {self.folder_path} ===") + print(f"Total files: {stats['count']}") + print(f"Average file size: {stats['avg_size_gb']:.2f} GB") + print(f"Average duration: {stats['avg_duration_minutes']:.1f} minutes") + print(f"Median file size: {stats['median_size_bytes']/(1024**3):.2f} GB") + print(f"Median duration: {stats['median_duration_minutes']:.1f} minutes") + + print("\n=== Individual Files ===") + for info in sorted(self.file_info, key=lambda x: x['size_bytes'], reverse=True): + print(f"{info['name']:50} {info['size_gb']:6.2f} GB {info['duration_minutes']:6.1f} min") + + # Show classification + episodes, extras = self.classify_files() + + print(f"\n=== Classification ===") + print(f"Episodes: {len(episodes)}") + for ep in episodes: + print(f" - {ep['name']} ({ep['size_gb']:.2f} GB, {ep['duration_minutes']:.1f} min)") + + print(f"\nExtras: {len(extras)}") + for ex in extras: + print(f" - {ex['name']} ({ex['size_gb']:.2f} GB, {ex['duration_minutes']:.1f} min)") diff --git a/src/TVDBProvider/__init__.py b/src/TVDBProvider/__init__.py new file mode 100644 index 0000000..eaefc73 --- /dev/null +++ b/src/TVDBProvider/__init__.py @@ -0,0 +1,4 @@ +"""TVDB Provider module for fetching show and episode data.""" +from .tvdb_client import TVDBClient + +__all__ = ['TVDBClient'] diff --git a/src/TVDBProvider/__pycache__/__init__.cpython-313.pyc b/src/TVDBProvider/__pycache__/__init__.cpython-313.pyc new file mode 100644 index 0000000..066d4c6 Binary files /dev/null and b/src/TVDBProvider/__pycache__/__init__.cpython-313.pyc differ diff --git a/src/TVDBProvider/__pycache__/tvdb_cache.cpython-313.pyc b/src/TVDBProvider/__pycache__/tvdb_cache.cpython-313.pyc new file mode 100644 index 0000000..db1da05 Binary files /dev/null and b/src/TVDBProvider/__pycache__/tvdb_cache.cpython-313.pyc differ diff --git a/src/TVDBProvider/__pycache__/tvdb_client.cpython-313.pyc b/src/TVDBProvider/__pycache__/tvdb_client.cpython-313.pyc new file mode 100644 index 0000000..e44e570 Binary files /dev/null and b/src/TVDBProvider/__pycache__/tvdb_client.cpython-313.pyc differ diff --git a/src/TVDBProvider/tvdb_cache.py b/src/TVDBProvider/tvdb_cache.py new file mode 100644 index 0000000..7aede78 --- /dev/null +++ b/src/TVDBProvider/tvdb_cache.py @@ -0,0 +1,126 @@ +"""TVDB cache for storing API responses locally.""" +import json +import os +from pathlib import Path +from datetime import datetime, timedelta +from typing import Dict, Optional, List + + +class TVDBCache: + """Local cache for TVDB API responses.""" + + def __init__(self, cache_dir: str = ".tvdb_cache"): + """Initialize cache with specified directory.""" + self.cache_dir = Path(cache_dir) + if not self.cache_dir.is_absolute(): + # Make cache relative to project root + script_dir = Path(__file__).parent.parent + self.cache_dir = script_dir / self.cache_dir + + self.cache_dir.mkdir(exist_ok=True) + self.cache_duration = timedelta(days=7) # Cache for 7 days + + def _get_cache_key(self, series_name: str, season_number: int) -> str: + """Generate cache key for series/season combination.""" + # Normalize series name for consistent caching + normalized_name = series_name.lower().replace(' ', '_').replace('-', '_') + return f"{normalized_name}_s{season_number:02d}" + + def _get_cache_file(self, cache_key: str) -> Path: + """Get cache file path for given key.""" + return self.cache_dir / f"{cache_key}.json" + + def _is_cache_valid(self, cache_file: Path) -> bool: + """Check if cache file exists and is not expired.""" + if not cache_file.exists(): + return False + + # Check if file is within cache duration + file_time = datetime.fromtimestamp(cache_file.stat().st_mtime) + return datetime.now() - file_time < self.cache_duration + + def get_cached_episodes(self, series_name: str, season_number: int) -> Optional[List[Dict]]: + """Get cached episode data for series/season.""" + cache_key = self._get_cache_key(series_name, season_number) + cache_file = self._get_cache_file(cache_key) + + if not self._is_cache_valid(cache_file): + return None + + try: + with open(cache_file, 'r') as f: + cache_data = json.load(f) + + # Validate cache structure + if 'episodes' in cache_data and 'cached_at' in cache_data: + print(f"Using cached TVDB data for {series_name} Season {season_number}") + return cache_data['episodes'] + + except (json.JSONDecodeError, KeyError, FileNotFoundError) as e: + print(f"Cache read error for {cache_key}: {e}") + + return None + + def cache_episodes(self, series_name: str, season_number: int, episodes: List[Dict]) -> bool: + """Cache episode data for series/season.""" + cache_key = self._get_cache_key(series_name, season_number) + cache_file = self._get_cache_file(cache_key) + + cache_data = { + 'series_name': series_name, + 'season_number': season_number, + 'episodes': episodes, + 'cached_at': datetime.now().isoformat(), + 'cache_duration_days': self.cache_duration.days + } + + try: + with open(cache_file, 'w') as f: + json.dump(cache_data, f, indent=2) + + print(f"Cached TVDB data for {series_name} Season {season_number}") + return True + + except Exception as e: + print(f"Cache write error for {cache_key}: {e}") + return False + + def clear_cache(self) -> bool: + """Clear all cached data.""" + try: + for cache_file in self.cache_dir.glob("*.json"): + cache_file.unlink() + print("TVDB cache cleared") + return True + except Exception as e: + print(f"Error clearing cache: {e}") + return False + + def list_cache(self) -> List[Dict]: + """List all cached entries with their status.""" + cached_entries = [] + + for cache_file in self.cache_dir.glob("*.json"): + try: + with open(cache_file, 'r') as f: + cache_data = json.load(f) + + cached_at = datetime.fromisoformat(cache_data.get('cached_at', '')) + is_valid = self._is_cache_valid(cache_file) + age = datetime.now() - cached_at + + entry = { + 'series_name': cache_data.get('series_name', 'Unknown'), + 'season_number': cache_data.get('season_number', 0), + 'episode_count': len(cache_data.get('episodes', [])), + 'cached_at': cached_at.strftime('%Y-%m-%d %H:%M'), + 'age_days': age.days, + 'is_valid': is_valid, + 'cache_key': cache_file.stem + } + cached_entries.append(entry) + + except Exception as e: + print(f"Error reading cache file {cache_file}: {e}") + + return sorted(cached_entries, key=lambda x: x['cached_at'], reverse=True) diff --git a/src/TVDBProvider/tvdb_client.py b/src/TVDBProvider/tvdb_client.py new file mode 100644 index 0000000..b60fc19 --- /dev/null +++ b/src/TVDBProvider/tvdb_client.py @@ -0,0 +1,192 @@ +"""TVDB Client for interacting with The Movie Database API.""" +import requests +import json +from typing import Dict, List, Optional, Tuple +from .tvdb_cache import TVDBCache + + +class TVDBClient: + """Client for interacting with TVDB API to get show and episode information.""" + + def __init__(self): + """Initialize TVDB client.""" + self.base_url = "https://api4.thetvdb.com/v4" + self.token = None + self.headers = { + 'Content-Type': 'application/json', + 'Accept': 'application/json' + } + self.cache = TVDBCache() + + def authenticate(self, api_key: str) -> bool: + """Authenticate with TVDB API using API key.""" + auth_url = f"{self.base_url}/login" + auth_data = { + "apikey": api_key + } + + try: + response = requests.post(auth_url, json=auth_data, headers=self.headers) + if response.status_code == 200: + data = response.json() + self.token = data.get('data', {}).get('token') + self.headers['Authorization'] = f'Bearer {self.token}' + return True + else: + print(f"Authentication failed: {response.status_code}") + return False + except Exception as e: + print(f"Authentication error: {e}") + return False + + def search_series(self, series_name: str) -> Optional[Dict]: + """Search for a TV series by name.""" + if not self.token: + print("Not authenticated. Please call authenticate() first.") + return None + + search_url = f"{self.base_url}/search" + params = { + 'query': series_name, + 'type': 'series' + } + + try: + response = requests.get(search_url, params=params, headers=self.headers) + if response.status_code == 200: + data = response.json() + series_list = data.get('data', []) + if series_list: + # Return the first match (most relevant) + return series_list[0] + else: + print(f"No series found for '{series_name}'") + return None + else: + print(f"Search failed: {response.status_code}") + return None + except Exception as e: + print(f"Search error: {e}") + return None + + def get_season_episodes(self, series_id: int, season_number: int) -> Optional[List[Dict]]: + """Get all episodes for a specific season of a series.""" + if not self.token: + print("Not authenticated. Please call authenticate() first.") + return None + + # Get series seasons first + seasons_url = f"{self.base_url}/series/{series_id}/episodes/default" + params = { + 'season': season_number + } + + try: + response = requests.get(seasons_url, params=params, headers=self.headers) + if response.status_code == 200: + data = response.json() + episodes = data.get('data', {}).get('episodes', []) + return episodes + else: + print(f"Failed to get episodes: {response.status_code}") + return None + except Exception as e: + print(f"Episodes fetch error: {e}") + return None + + def get_series_info(self, series_name: str) -> Optional[Dict]: + """Get series information including ID.""" + series = self.search_series(series_name) + if series: + return { + 'id': series.get('tvdb_id'), + 'name': series.get('name'), + 'slug': series.get('slug'), + 'year': series.get('year') + } + return None + + def get_episode_durations(self, series_name: str, season_number: int) -> Optional[List[Dict]]: + """Get episode durations for a specific season with caching.""" + # Check cache first + cached_episodes = self.cache.get_cached_episodes(series_name, season_number) + if cached_episodes: + return cached_episodes + + # First get series info + series_info = self.get_series_info(series_name) + if not series_info: + return None + + # Get episodes for the season + episodes = self.get_season_episodes(series_info['id'], season_number) + if not episodes: + return None + + # Extract relevant episode information + episode_data = [] + for episode in episodes: + episode_info = { + 'episode_number': episode.get('number'), + 'name': episode.get('name'), + 'runtime': episode.get('runtime'), # Duration in minutes + 'aired': episode.get('aired') + } + episode_data.append(episode_info) + + # Sort by episode number + episode_data.sort(key=lambda x: x['episode_number'] or 0) + + # Cache the results + self.cache.cache_episodes(series_name, season_number, episode_data) + + return episode_data + + +# For testing without API key, we can use a mock client +class MockTVDBClient(TVDBClient): + """Mock TVDB client for testing purposes.""" + + def __init__(self): + super().__init__() + self.authenticated = False + self.cache = TVDBCache() # Mock also uses cache + + def authenticate(self, api_key: str = None) -> bool: + """Mock authentication.""" + self.authenticated = True + return True + + def get_episode_durations(self, series_name: str, season_number: int) -> Optional[List[Dict]]: + """Mock episode durations with caching - returns typical TV episode length.""" + if not self.authenticated: + return None + + # Check cache first + cached_episodes = self.cache.get_cached_episodes(series_name, season_number) + if cached_episodes: + return cached_episodes + + # Return mock data - assuming typical 20-22 episode season with ~45min episodes + mock_episodes = [] + episode_count = 22 # Default episode count + + # Adjust episode count based on series type + if any(keyword in series_name.lower() for keyword in ['game of thrones', 'stranger things', 'breaking bad']): + episode_count = 10 # Drama series typically have fewer episodes + elif any(keyword in series_name.lower() for keyword in ['friends', 'the office', 'brooklyn nine-nine']): + episode_count = 24 # Sitcoms typically have more episodes + + for i in range(1, episode_count + 1): + episode_info = { + 'episode_number': i, + 'name': f"Episode {i}", + 'runtime': 45, # 45 minutes default + 'aired': f"2023-01-{i:02d}" + } + mock_episodes.append(episode_info) + + # Cache the mock results + self.cache.cache_episodes(series_name, season_number, mock_episodes) + + return mock_episodes diff --git a/src/__pycache__/config.cpython-313.pyc b/src/__pycache__/config.cpython-313.pyc new file mode 100644 index 0000000..ad0eff8 Binary files /dev/null and b/src/__pycache__/config.cpython-313.pyc differ diff --git a/src/config.py b/src/config.py new file mode 100644 index 0000000..3714602 --- /dev/null +++ b/src/config.py @@ -0,0 +1,105 @@ +"""Configuration management for Episode Matcher.""" +import json +import os +from pathlib import Path +from typing import Dict, Optional + + +class ConfigManager: + """Manages configuration settings for Episode Matcher.""" + + def __init__(self, config_file: str = "config.json"): + """Initialize config manager with config file path.""" + self.config_file = Path(config_file) + if not self.config_file.is_absolute(): + # If relative path, make it relative to the script directory + script_dir = Path(__file__).parent.parent + self.config_file = script_dir / self.config_file + + self.config = self._load_config() + + def _load_config(self) -> Dict: + """Load configuration from JSON file.""" + default_config = { + "tvdb_api_key": "", + "default_episode_duration": 45, + "classification_thresholds": { + "size_threshold_ratio": 0.3, + "duration_threshold_ratio": 0.4 + } + } + + if not self.config_file.exists(): + print(f"Config file not found at {self.config_file}") + print("Using default configuration. Create config.json to customize settings.") + return default_config + + try: + with open(self.config_file, 'r') as f: + config = json.load(f) + + # Merge with defaults to ensure all keys exist + for key, value in default_config.items(): + if key not in config: + config[key] = value + + return config + + except json.JSONDecodeError as e: + print(f"Error parsing config file {self.config_file}: {e}") + print("Using default configuration.") + return default_config + except Exception as e: + print(f"Error loading config file {self.config_file}: {e}") + print("Using default configuration.") + return default_config + + def get_tvdb_api_key(self) -> Optional[str]: + """Get TVDB API key from config.""" + api_key = self.config.get("tvdb_api_key", "").strip() + if not api_key or api_key == "YOUR_TVDB_API_KEY_HERE": + return None + return api_key + + def get_default_episode_duration(self) -> int: + """Get default episode duration in minutes.""" + return self.config.get("default_episode_duration", 45) + + def get_classification_thresholds(self) -> Dict[str, float]: + """Get classification threshold ratios.""" + return self.config.get("classification_thresholds", { + "size_threshold_ratio": 0.3, + "duration_threshold_ratio": 0.4 + }) + + def update_api_key(self, api_key: str) -> bool: + """Update TVDB API key in config file.""" + try: + self.config["tvdb_api_key"] = api_key + with open(self.config_file, 'w') as f: + json.dump(self.config, f, indent=2) + print(f"API key updated in {self.config_file}") + return True + except Exception as e: + print(f"Error updating config file: {e}") + return False + + def print_config_status(self): + """Print current configuration status.""" + print(f"Configuration file: {self.config_file}") + print(f"Config exists: {self.config_file.exists()}") + + api_key = self.get_tvdb_api_key() + if api_key: + masked_key = api_key[:8] + "..." + api_key[-4:] if len(api_key) > 12 else api_key[:4] + "..." + print(f"TVDB API Key: {masked_key} (configured)") + else: + print("TVDB API Key: Not configured (will use mock data)") + + print(f"Default episode duration: {self.get_default_episode_duration()} minutes") + thresholds = self.get_classification_thresholds() + print(f"Classification thresholds: {thresholds}") + + +# Global config instance +config_manager = ConfigManager() diff --git a/test_components.py b/test_components.py new file mode 100644 index 0000000..9d7e999 --- /dev/null +++ b/test_components.py @@ -0,0 +1,84 @@ +#!/usr/bin/env python3 +""" +Test script for Episode Matcher components +""" + +import sys +import os +from pathlib import Path + +# Add src to path so we can import our modules +sys.path.insert(0, os.path.join(os.path.dirname(__file__), 'src')) + +from TVDBProvider.tvdb_client import MockTVDBClient +from Matcher.file_classifier import FileClassifier +from Matcher.episode_renamer import EpisodeRenamer + + +def test_tvdb_client(): + """Test TVDB client functionality.""" + print("=== Testing TVDB Client ===") + + client = MockTVDBClient() + success = client.authenticate() + print(f"Authentication: {'✓' if success else '✗'}") + + episodes = client.get_episode_durations("Game of Thrones", 1) + if episodes: + print(f"✓ Got {len(episodes)} episodes") + print(f" Sample episode: Episode {episodes[0]['episode_number']} - {episodes[0]['runtime']} min") + else: + print("✗ Failed to get episodes") + + print() + + +def test_file_classifier(): + """Test file classifier with current directory (no MKV files expected).""" + print("=== Testing File Classifier ===") + + try: + classifier = FileClassifier(".") + print(f"✓ Classifier initialized for current directory") + print(f" Found {len(classifier.video_files)} MKV files") + + if classifier.video_files: + episodes, extras = classifier.classify_files() + print(f" Would classify {len(episodes)} episodes, {len(extras)} extras") + + except Exception as e: + print(f"✗ Classifier error: {e}") + + print() + + +def test_episode_renamer(): + """Test episode renamer functionality.""" + print("=== Testing Episode Renamer ===") + + renamer = EpisodeRenamer(".", "Test Show", 1) + print(f"✓ Renamer initialized") + + # Test filename generation + test_filename = renamer._generate_episode_filename(5, ".mkv") + expected = "Test Show s01e05.mkv" + print(f" Filename generation: {'✓' if test_filename == expected else '✗'}") + print(f" Generated: {test_filename}") + print(f" Expected: {expected}") + + print() + + +def main(): + """Run all tests.""" + print("Episode Matcher Component Tests\n") + + test_tvdb_client() + test_file_classifier() + test_episode_renamer() + + print("Tests completed!") + + +if __name__ == "__main__": + main()