jarianc/EpisodeMatcher
1
0
Fork 0
Code Issues Pull Requests Actions 1 Packages Projects Releases Wiki Activity

Initial Commit

Browse Source
This commit is contained in:
Jarian Cottingham 2025-08-07 21:15:00 -05:00
commit 03b03d676f
23 changed files with 2103 additions and 0 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

39
Prompt.md Normal file
View File

@ -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
<tv show> s<season number>e<episode number>. 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.

144
README.md Normal file
View File

@ -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)

8
config.json Normal file
View File

@ -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
}
}

227
episode_matcher.py Executable file
View File

@ -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 <folder_path> <show_name> <season_number> [--api-key <tvdb_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()

3
requirements.txt Normal file
View File

@ -0,0 +1,3 @@
requests>=2.25.1
pathlib2>=2.3.5
pymediainfo>=5.1.0

54
setup_config.py Executable file
View File

@ -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()

68
src/.tvdb_cache/game_of_thrones_s01.json Normal file
View File

@ -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
}

68
src/.tvdb_cache/game_of_thrones_s02.json Normal file
View File

@ -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
}

5
src/Matcher/__init__.py Normal file
View File

@ -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']

BIN
src/Matcher/__pycache__/__init__.cpython-313.pyc Normal file
View File

Binary file not shown.

BIN
src/Matcher/__pycache__/episode_renamer.cpython-313.pyc Normal file
View File

Binary file not shown.

BIN
src/Matcher/__pycache__/file_classifier.cpython-313.pyc Normal file
View File

Binary file not shown.

666
src/Matcher/episode_renamer.py Normal file
View File

@ -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

310
src/Matcher/file_classifier.py Normal file
View File

@ -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)")

4
src/TVDBProvider/__init__.py Normal file
View File

@ -0,0 +1,4 @@
"""TVDB Provider module for fetching show and episode data."""
from .tvdb_client import TVDBClient
__all__ = ['TVDBClient']

BIN
src/TVDBProvider/__pycache__/__init__.cpython-313.pyc Normal file
View File

Binary file not shown.

BIN
src/TVDBProvider/__pycache__/tvdb_cache.cpython-313.pyc Normal file
View File

Binary file not shown.

BIN
src/TVDBProvider/__pycache__/tvdb_client.cpython-313.pyc Normal file
View File

Binary file not shown.

126
src/TVDBProvider/tvdb_cache.py Normal file
View File

@ -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)

192
src/TVDBProvider/tvdb_client.py Normal file
View File

@ -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

BIN
src/__pycache__/config.cpython-313.pyc Normal file
View File

Binary file not shown.

105
src/config.py Normal file
View File

@ -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()

84
test_components.py Normal file
View File

@ -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()