"""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( identity_file_pairs: list[tuple[MovieIdentity | SeriesIdentity, 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. Uses (identity, file) pairs so that same filename under different paths are not conflated. Args: identity_file_pairs: List of (identity, video_file) in matching order Returns: List of DuplicateGroup objects for files with duplicates """ groups: dict[tuple, list[tuple[MovieIdentity | SeriesIdentity, VideoFile]]] = {} for identity, video_file in identity_file_pairs: if isinstance(identity, MovieIdentity): if identity.year is None: continue key = ("movie", identity.title, identity.year) if key not in groups: groups[key] = [] groups[key].append((identity, video_file)) else: # SeriesIdentity if identity.season is None or not identity.episodes: continue for episode in identity.episodes: key = ("series", identity.title, identity.season, episode) 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