"""Analysis engine for detecting completeness issues and duplicates. This module provides functionality to analyze video collections for: - Series completeness (detecting episode gaps) - Duplicate detection (finding duplicate content) - Quality comparison (comparing video quality metrics) """ from vlm.models import SeriesIdentity, SeasonCompleteness, DuplicateGroup, VideoFile, MovieIdentity def analyze_series_completeness(episodes: list[SeriesIdentity]) -> list[SeasonCompleteness]: """Analyze series completeness and detect episode gaps using heuristic detection. This function groups episodes by series title and season, then detects gaps in the episode sequence using heuristic detection. For each season, it finds the minimum and maximum episode numbers and identifies missing episodes in that range [min, max]. Note: This is heuristic gap detection only. It does NOT calculate percentages or determine if seasons are "complete" (v1 constraint: no external metadata). Args: episodes: List of parsed series identities Returns: List of SeasonCompleteness objects for seasons with detected gaps """ from vlm.parser import group_episodes # Group episodes by (title, season) grouped = group_episodes(episodes) completeness_results = [] # Analyze each season for (series_title, season), episode_list in grouped.items(): # Collect all episode numbers from this season all_episode_numbers = set() for episode in episode_list: all_episode_numbers.update(episode.episodes) # Convert to sorted list episodes_found = sorted(all_episode_numbers) # Find min and max episode numbers if not episodes_found: continue min_episode = min(episodes_found) max_episode = max(episodes_found) # Detect gaps in the range [min, max] expected_episodes = set(range(min_episode, max_episode + 1)) found_episodes_set = set(episodes_found) missing_episodes = sorted(expected_episodes - found_episodes_set) # Only include seasons with gaps if missing_episodes: completeness_results.append(SeasonCompleteness( series_title=series_title, season=season, episodes_found=episodes_found, episodes_missing=missing_episodes )) return completeness_results def detect_duplicates( identities: list[MovieIdentity | SeriesIdentity], files: list[VideoFile] ) -> list[DuplicateGroup]: """Detect duplicate video files and provide quality comparison data. Groups files by normalized identity (title+year for movies, title+season+episode for series) and identifies groups with multiple files as potential duplicates. Args: identities: List of parsed identities (movies or series) files: List of video files corresponding to the identities Returns: List of DuplicateGroup objects for files with duplicates """ # Create a mapping from original filename to VideoFile for quick lookup file_map = {file.filename: file for file in files} # Group identities by normalized identity groups: dict[tuple, list[tuple[MovieIdentity | SeriesIdentity, VideoFile]]] = {} for identity in identities: # Create grouping key based on identity type if isinstance(identity, MovieIdentity): # For movies: group by (title, year) # Skip if year is None (needs review) if identity.year is None: continue key = ('movie', identity.title, identity.year) else: # SeriesIdentity # For series: group by (title, season, episode) # Skip if season is None or episodes is empty (needs review) if identity.season is None or not identity.episodes: continue # For multi-episode files, use the first episode for grouping # Each episode in the list should be treated separately for episode in identity.episodes: key = ('series', identity.title, identity.season, episode) # Get the corresponding VideoFile video_file = file_map.get(identity.original_filename) if video_file is None: continue # Add to group if key not in groups: groups[key] = [] groups[key].append((identity, video_file)) continue # Get the corresponding VideoFile for movies video_file = file_map.get(identity.original_filename) if video_file is None: continue # Add to group if key not in groups: groups[key] = [] groups[key].append((identity, video_file)) # Filter groups to only those with multiple files (duplicates) duplicate_groups = [] for key, items in groups.items(): if len(items) > 1: # Extract identities and files # Use the first identity as the representative representative_identity = items[0][0] duplicate_files = [item[1] for item in items] # Generate quality comparison data quality_comparison = compare_quality(duplicate_files) duplicate_groups.append(DuplicateGroup( identity=representative_identity, files=duplicate_files, quality_comparison=quality_comparison )) return duplicate_groups def compare_quality(files: list[VideoFile]) -> list[dict]: """Compare video quality metrics for a set of files. Extracts and compares resolution, codec, file size, and other quality indicators to help users decide which files to keep. Args: files: List of video files to compare Returns: List of dictionaries with quality comparison data for each file """ comparison_data = [] for file in files: quality_info = { 'filename': file.filename, 'path': str(file.path), 'size_bytes': file.size_bytes, } # Add optional metadata if available if file.resolution is not None: quality_info['resolution'] = file.resolution if file.codec is not None: quality_info['codec'] = file.codec if file.duration_seconds is not None: quality_info['duration_seconds'] = file.duration_seconds if file.bitrate_kbps is not None: quality_info['bitrate_kbps'] = file.bitrate_kbps comparison_data.append(quality_info) return comparison_data