#!/usr/bin/env python3 """ Plex Media Library Statistics Generator Scans a Plex-style media library and generates CSV statistics about movies, TV shows, and anime. Uses SQLite for persistent caching to avoid rescanning unchanged files. Author: Media Scraper Tool License: MIT """ import argparse import csv import os import sqlite3 import subprocess import sys import time from dataclasses import dataclass from datetime import datetime from pathlib import Path from typing import Dict, List, Optional, Set, Tuple # Configuration constants VIDEO_EXTENSIONS = {'.mkv', '.mp4', '.m4v', '.avi', '.mov', '.ts', '.m2ts', '.webm', '.mpg', '.mpeg', '.wmv'} BYTES_PER_GB = 1_000_000_000 SECONDS_PER_HOUR = 3600 CSV_DECIMAL_PLACES = 2 @dataclass class MediaFile: """Represents a single video file with its metadata.""" path: str library_root: str library_type: str # 'tv' or 'movies' show_movie: str season: Optional[str] # None for movies size_bytes: int mtime_ns: int duration_seconds: Optional[float] scan_timestamp: float error_message: Optional[str] = None class PlexStatsDB: """Manages SQLite database for caching media file metadata.""" def __init__(self, db_path: Path): """Initialize database connection and create schema if needed.""" self.db_path = db_path self.conn: Optional[sqlite3.Connection] = None self._init_db() def _init_db(self) -> None: """Initialize database connection and create tables if needed.""" self.conn = sqlite3.connect(self.db_path) self.conn.row_factory = sqlite3.Row cursor = self.conn.cursor() # Create media_files table cursor.execute(''' CREATE TABLE IF NOT EXISTS media_files ( id INTEGER PRIMARY KEY AUTOINCREMENT, path TEXT UNIQUE NOT NULL, library_root TEXT NOT NULL, library_type TEXT NOT NULL, show_movie TEXT NOT NULL, season TEXT, size_bytes INTEGER NOT NULL, mtime_ns INTEGER NOT NULL, duration_seconds REAL, scan_timestamp REAL NOT NULL, error_message TEXT ) ''') # Create indexes for common queries cursor.execute('CREATE INDEX IF NOT EXISTS idx_library_root ON media_files(library_root)') cursor.execute('CREATE INDEX IF NOT EXISTS idx_path ON media_files(path)') cursor.execute('CREATE INDEX IF NOT EXISTS idx_show_movie ON media_files(show_movie)') cursor.execute('CREATE INDEX IF NOT EXISTS idx_season ON media_files(season)') self.conn.commit() def close(self) -> None: """Close database connection.""" if self.conn: self.conn.close() def __enter__(self): """Context manager entry.""" return self def __exit__(self, exc_type, exc_val, exc_tb): """Context manager exit.""" self.close() def get_cached_file(self, path: str) -> Optional[MediaFile]: """Retrieve a cached media file by path.""" cursor = self.conn.cursor() cursor.execute('SELECT * FROM media_files WHERE path = ?', (path,)) row = cursor.fetchone() if not row: return None return MediaFile( path=row['path'], library_root=row['library_root'], library_type=row['library_type'], show_movie=row['show_movie'], season=row['season'], size_bytes=row['size_bytes'], mtime_ns=row['mtime_ns'], duration_seconds=row['duration_seconds'], scan_timestamp=row['scan_timestamp'], error_message=row['error_message'] ) def insert_or_update_file(self, media_file: MediaFile) -> None: """Insert or update a media file record.""" cursor = self.conn.cursor() cursor.execute(''' INSERT OR REPLACE INTO media_files (path, library_root, library_type, show_movie, season, size_bytes, mtime_ns, duration_seconds, scan_timestamp, error_message) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?) ''', ( media_file.path, media_file.library_root, media_file.library_type, media_file.show_movie, media_file.season, media_file.size_bytes, media_file.mtime_ns, media_file.duration_seconds, media_file.scan_timestamp, media_file.error_message )) self.conn.commit() def get_files_by_library_root(self, library_root: str) -> List[MediaFile]: """Retrieve all files cached for a specific library root.""" cursor = self.conn.cursor() cursor.execute('SELECT * FROM media_files WHERE library_root = ?', (library_root,)) rows = cursor.fetchall() return [ MediaFile( path=row['path'], library_root=row['library_root'], library_type=row['library_type'], show_movie=row['show_movie'], season=row['season'], size_bytes=row['size_bytes'], mtime_ns=row['mtime_ns'], duration_seconds=row['duration_seconds'], scan_timestamp=row['scan_timestamp'], error_message=row['error_message'] ) for row in rows ] def delete_file(self, path: str) -> None: """Delete a file record from the database.""" cursor = self.conn.cursor() cursor.execute('DELETE FROM media_files WHERE path = ?', (path,)) self.conn.commit() def check_ffprobe_available() -> bool: """Check if ffprobe is available in PATH.""" try: subprocess.run( ['ffprobe', '-version'], capture_output=True, timeout=5, check=True ) return True except (subprocess.CalledProcessError, FileNotFoundError, subprocess.TimeoutExpired): return False def get_duration_from_ffprobe(file_path: str) -> Optional[float]: """ Extract duration in seconds from a video file using ffprobe. Returns: Duration in seconds as float, or None if extraction fails. """ try: result = subprocess.run( [ 'ffprobe', '-v', 'error', '-show_entries', 'format=duration', '-of', 'default=noprint_wrappers=1:nokey=1', file_path ], capture_output=True, timeout=30, text=True, check=False ) if result.returncode != 0: return None duration_str = result.stdout.strip() if not duration_str: return None return float(duration_str) except (subprocess.TimeoutExpired, ValueError, OSError) as e: return None def infer_library_type(library_root: str) -> str: """ Infer library type from directory name. Returns 'tv' or 'movies' based on directory name. Defaults to 'tv' if uncertain. """ root_name = Path(library_root).name.lower() if 'movie' in root_name or 'film' in root_name: return 'movies' return 'tv' def extract_show_movie_name(file_path: str, library_root: str) -> Tuple[str, Optional[str]]: """ Extract show/movie name and season (if applicable) from file path. Args: file_path: Full path to the video file library_root: Root of the library Returns: Tuple of (show/movie_name, season_name) season_name is None for movies. """ relative_path = Path(file_path).relative_to(library_root) parts = relative_path.parts if len(parts) < 2: # Unusual structure, use filename return file_path.split(os.sep)[-2], None show_movie = parts[0] season = parts[1] if len(parts) > 2 else None return show_movie, season def get_video_files_in_library(library_root: Path, library_type: str) -> List[str]: """ Recursively find all video files in a library root. Args: library_root: Path to library root directory library_type: 'tv' or 'movies' Returns: List of absolute paths to video files. """ video_files: List[str] = [] try: for path in library_root.rglob('*'): if path.is_file() and path.suffix.lower() in VIDEO_EXTENSIONS: video_files.append(str(path.resolve())) except (PermissionError, OSError) as e: print(f"Warning: Could not access {library_root}: {e}") return video_files def scan_library( library_root: Path, library_type: str, db: PlexStatsDB, force_rescan: bool = False, verbose: bool = False ) -> Tuple[List[MediaFile], Dict[str, int]]: """ Scan a media library and update cache. Args: library_root: Path to library root library_type: 'tv' or 'movies' db: Database instance force_rescan: If True, ffprobe all files regardless of cache verbose: Print verbose output Returns: Tuple of (list of MediaFile objects, stats dict) """ library_root_str = str(library_root.resolve()) stats = { 'discovered': 0, 'cached_unchanged': 0, 'new': 0, 'changed': 0, 'removed': 0, 'errors': 0 } print(f"\nScanning {library_root_str}...") # Get all video files currently in library current_files = get_video_files_in_library(library_root, library_type) current_files_set: Set[str] = set(current_files) stats['discovered'] = len(current_files) print(f"Files discovered: {stats['discovered']:,}") # Get previously cached files for this library cached_files = db.get_files_by_library_root(library_root_str) cached_by_path: Dict[str, MediaFile] = {f.path: f for f in cached_files} # Find removed files for cached_path in cached_by_path: if cached_path not in current_files_set: db.delete_file(cached_path) stats['removed'] += 1 # Process current files media_files: List[MediaFile] = [] files_to_probe: List[str] = [] for file_path in current_files: try: stat_info = os.stat(file_path) size_bytes = stat_info.st_size mtime_ns = stat_info.st_mtime_ns show_movie, season = extract_show_movie_name(file_path, library_root_str) # Check cache cached = cached_by_path.get(file_path) if ( not force_rescan and cached is not None and cached.size_bytes == size_bytes and cached.mtime_ns == mtime_ns ): # File unchanged, reuse cache media_files.append(cached) stats['cached_unchanged'] += 1 else: # File is new or changed media_file = MediaFile( path=file_path, library_root=library_root_str, library_type=library_type, show_movie=show_movie, season=season, size_bytes=size_bytes, mtime_ns=mtime_ns, duration_seconds=None, scan_timestamp=time.time() ) if cached is not None: stats['changed'] += 1 else: stats['new'] += 1 files_to_probe.append(file_path) media_files.append(media_file) except (OSError, PermissionError) as e: print(f"Warning: Could not access {file_path}: {e}") stats['errors'] += 1 print(f"Cached/unchanged: {stats['cached_unchanged']:,}") print(f"New: {stats['new']}") print(f"Changed: {stats['changed']}") print(f"Removed: {stats['removed']}") # Probe new/changed files if files_to_probe: print(f"\nProbing {len(files_to_probe)} files...") for idx, file_path in enumerate(files_to_probe, 1): # Find the media file object for this path media_file = next((mf for mf in media_files if mf.path == file_path), None) if media_file is None: continue # Print progress show_name = media_file.show_movie[:50] print(f" [{idx}/{len(files_to_probe)}] {show_name}...", flush=True) # Get duration duration = get_duration_from_ffprobe(file_path) if duration is not None: media_file.duration_seconds = duration else: media_file.error_message = "ffprobe failed or returned no duration" stats['errors'] += 1 # Update database db.insert_or_update_file(media_file) # Insert cached files that weren't probed for media_file in media_files: if media_file.path not in files_to_probe: db.insert_or_update_file(media_file) return media_files, stats def aggregate_tv_seasons(media_files: List[MediaFile]) -> Dict[Tuple[str, str], Dict]: """ Aggregate media files by series and season for TV output. Args: media_files: List of MediaFile objects Returns: Dict mapping (series, season) tuples to aggregated stats """ seasons: Dict[Tuple[str, str], Dict] = {} for media_file in media_files: if media_file.season is None: continue key = (media_file.show_movie, media_file.season) if key not in seasons: seasons[key] = { 'series': media_file.show_movie, 'season': media_file.season, 'file_count': 0, 'size_bytes': 0, 'duration_seconds': 0.0, 'error_count': 0 } seasons[key]['file_count'] += 1 seasons[key]['size_bytes'] += media_file.size_bytes if media_file.duration_seconds is not None: seasons[key]['duration_seconds'] += media_file.duration_seconds if media_file.error_message: seasons[key]['error_count'] += 1 return seasons def aggregate_tv_series(season_aggregates: Dict[Tuple[str, str], Dict]) -> Dict[str, Dict]: """ Aggregate season data by series for series-level output. Args: season_aggregates: Dict from aggregate_tv_seasons Returns: Dict mapping series names to aggregated stats """ series: Dict[str, Dict] = {} for (series_name, season_name), season_data in season_aggregates.items(): if series_name not in series: series[series_name] = { 'series': series_name, 'season_count': 0, 'file_count': 0, 'size_bytes': 0, 'duration_seconds': 0.0 } series[series_name]['season_count'] += 1 series[series_name]['file_count'] += season_data['file_count'] series[series_name]['size_bytes'] += season_data['size_bytes'] series[series_name]['duration_seconds'] += season_data['duration_seconds'] return series def aggregate_movies(media_files: List[MediaFile]) -> Dict[str, Dict]: """ Aggregate media files by movie name. Args: media_files: List of MediaFile objects (should all be movies) Returns: Dict mapping movie names to aggregated stats """ movies: Dict[str, Dict] = {} for media_file in media_files: movie_name = media_file.show_movie if movie_name not in movies: movies[movie_name] = { 'movie': movie_name, 'file_count': 0, 'size_bytes': 0, 'duration_seconds': 0.0, 'error_count': 0 } movies[movie_name]['file_count'] += 1 movies[movie_name]['size_bytes'] += media_file.size_bytes if media_file.duration_seconds is not None: movies[movie_name]['duration_seconds'] += media_file.duration_seconds if media_file.error_message: movies[movie_name]['error_count'] += 1 return movies def format_number(value: float, decimal_places: int = CSV_DECIMAL_PLACES) -> float: """Format a number to a specific number of decimal places.""" return round(value, decimal_places) def write_tv_season_csv( season_aggregates: Dict[Tuple[str, str], Dict], output_path: Path, csv_prefix: str ) -> None: """Write season-level TV statistics to CSV.""" csv_file = output_path / f"{csv_prefix}_seasons.csv" # Sort by series, then season sorted_seasons = sorted( season_aggregates.items(), key=lambda x: (x[0][0], x[0][1]) # (series, season) ) with open(csv_file, 'w', newline='') as f: writer = csv.DictWriter( f, fieldnames=[ 'Library', 'Series', 'Season', 'File_Count', 'Size_Bytes', 'Size_GB', 'Duration_Seconds', 'Duration_Hours', 'GB_Per_Hour' ] ) writer.writeheader() for (series, season), data in sorted_seasons: size_gb = format_number(data['size_bytes'] / BYTES_PER_GB) duration_hours = format_number(data['duration_seconds'] / SECONDS_PER_HOUR) gb_per_hour = ( format_number(size_gb / duration_hours) if duration_hours > 0 else 0.0 ) writer.writerow({ 'Library': 'tv', 'Series': series, 'Season': season, 'File_Count': data['file_count'], 'Size_Bytes': data['size_bytes'], 'Size_GB': size_gb, 'Duration_Seconds': data['duration_seconds'], 'Duration_Hours': duration_hours, 'GB_Per_Hour': gb_per_hour }) print(f" {csv_file}") def write_tv_series_csv( series_aggregates: Dict[str, Dict], output_path: Path, csv_prefix: str ) -> None: """Write series-level TV statistics to CSV.""" csv_file = output_path / f"{csv_prefix}_series.csv" # Sort by series name sorted_series = sorted(series_aggregates.items(), key=lambda x: x[0]) with open(csv_file, 'w', newline='') as f: writer = csv.DictWriter( f, fieldnames=[ 'Library', 'Series', 'Season_Count', 'File_Count', 'Size_Bytes', 'Size_GB', 'Duration_Seconds', 'Duration_Hours', 'GB_Per_Hour' ] ) writer.writeheader() for series_name, data in sorted_series: size_gb = format_number(data['size_bytes'] / BYTES_PER_GB) duration_hours = format_number(data['duration_seconds'] / SECONDS_PER_HOUR) gb_per_hour = ( format_number(size_gb / duration_hours) if duration_hours > 0 else 0.0 ) writer.writerow({ 'Library': 'tv', 'Series': series_name, 'Season_Count': data['season_count'], 'File_Count': data['file_count'], 'Size_Bytes': data['size_bytes'], 'Size_GB': size_gb, 'Duration_Seconds': data['duration_seconds'], 'Duration_Hours': duration_hours, 'GB_Per_Hour': gb_per_hour }) print(f" {csv_file}") def write_movies_csv( movies_aggregates: Dict[str, Dict], output_path: Path, csv_prefix: str ) -> None: """Write movie statistics to CSV.""" csv_file = output_path / f"{csv_prefix}_movies.csv" # Sort by movie name sorted_movies = sorted(movies_aggregates.items(), key=lambda x: x[0]) with open(csv_file, 'w', newline='') as f: writer = csv.DictWriter( f, fieldnames=[ 'Library', 'Movie', 'File_Count', 'Size_Bytes', 'Size_GB', 'Duration_Seconds', 'Duration_Hours', 'GB_Per_Hour' ] ) writer.writeheader() for movie_name, data in sorted_movies: size_gb = format_number(data['size_bytes'] / BYTES_PER_GB) duration_hours = format_number(data['duration_seconds'] / SECONDS_PER_HOUR) gb_per_hour = ( format_number(size_gb / duration_hours) if duration_hours > 0 else 0.0 ) writer.writerow({ 'Library': 'movies', 'Movie': movie_name, 'File_Count': data['file_count'], 'Size_Bytes': data['size_bytes'], 'Size_GB': size_gb, 'Duration_Seconds': data['duration_seconds'], 'Duration_Hours': duration_hours, 'GB_Per_Hour': gb_per_hour }) print(f" {csv_file}") def print_scan_summary( library_type: str, media_files: List[MediaFile], stats: Dict[str, int], season_agg: Optional[Dict[Tuple[str, str], Dict]] = None, series_agg: Optional[Dict[str, Dict]] = None, movie_agg: Optional[Dict[str, Dict]] = None, db_path: Path = None, csv_files: List[str] = None ) -> None: """Print a human-readable summary of the scan.""" if csv_files is None: csv_files = [] # Calculate totals total_size_gb = sum(mf.size_bytes for mf in media_files) / BYTES_PER_GB total_duration_seconds = sum( (mf.duration_seconds or 0) for mf in media_files if mf.duration_seconds ) total_duration_hours = total_duration_seconds / SECONDS_PER_HOUR print("\n" + "=" * 60) print("Scan Complete") print("=" * 60) if library_type == 'tv': unique_series = len(series_agg) if series_agg else 0 unique_seasons = len(season_agg) if season_agg else 0 print(f"TV series: {unique_series}") print(f"Seasons: {unique_seasons}") else: unique_movies = len(movie_agg) if movie_agg else 0 print(f"Movies: {unique_movies}") print(f"Video files: {len(media_files):,}") print(f"Total size: {format_number(total_size_gb, 2)} GB") print(f"Total runtime: {format_number(total_duration_hours, 1)} hours") print() print(f"Unchanged: {stats['cached_unchanged']:,}") print(f"New: {stats['new']}") print(f"Changed: {stats['changed']}") print(f"Removed: {stats['removed']}") print(f"Errors: {stats['errors']}") if csv_files: print() print("CSV files written:") for csv_file in csv_files: print(f" {csv_file}") if db_path: print() print("Database:") print(f" {db_path}") print("=" * 60) def main(): """Main entry point.""" parser = argparse.ArgumentParser( description='Generate statistics for Plex-style media libraries.' ) parser.add_argument( 'library_root', help='Path to the media library root (e.g., /plex/tv or /plex/movies)' ) parser.add_argument( '--type', choices=['tv', 'movies'], help='Library type (tv or movies). If omitted, will be inferred from directory name.' ) parser.add_argument( '--db', type=Path, default=Path('plex_stats.db'), help='Path to SQLite database (default: plex_stats.db)' ) parser.add_argument( '--output-dir', type=Path, default=Path('.'), help='Directory for output CSV files (default: current directory)' ) parser.add_argument( '--csv-prefix', help='Prefix for CSV filenames (e.g., "tv" -> tv_seasons.csv). ' 'If omitted, derived from library directory name.' ) parser.add_argument( '--force', action='store_true', help='Force ffprobe scan of all files, ignoring cache' ) parser.add_argument( '--verbose', action='store_true', help='Print verbose output' ) args = parser.parse_args() # Validate inputs library_root = Path(args.library_root).resolve() if not library_root.exists(): print(f"Error: Library root does not exist: {library_root}") sys.exit(1) if not library_root.is_dir(): print(f"Error: Library root is not a directory: {library_root}") sys.exit(1) # Check ffprobe availability if not check_ffprobe_available(): print("Error: ffprobe was not found.") print() print("Install FFmpeg, for example on Ubuntu/Debian:") print() print(" sudo apt install ffmpeg") print() sys.exit(1) # Determine library type library_type = args.type or infer_library_type(str(library_root)) # Determine CSV prefix csv_prefix = args.csv_prefix or library_root.name # Create output directory if needed output_dir = args.output_dir output_dir.mkdir(parents=True, exist_ok=True) # Open database with PlexStatsDB(args.db) as db: # Scan library media_files, stats = scan_library( library_root, library_type, db, force_rescan=args.force, verbose=args.verbose ) # Filter out files with errors for aggregation valid_files = [mf for mf in media_files if mf.duration_seconds is not None] csv_files_written: List[str] = [] # Generate CSV output print("\nGenerating CSV files...") if library_type == 'tv': season_agg = aggregate_tv_seasons(valid_files) series_agg = aggregate_tv_series(season_agg) if season_agg: write_tv_season_csv(season_agg, output_dir, csv_prefix) csv_files_written.append(str(output_dir / f"{csv_prefix}_seasons.csv")) if series_agg: write_tv_series_csv(series_agg, output_dir, csv_prefix) csv_files_written.append(str(output_dir / f"{csv_prefix}_series.csv")) print_scan_summary( library_type, valid_files, stats, season_agg=season_agg, series_agg=series_agg, db_path=args.db, csv_files=csv_files_written ) else: # movies movie_agg = aggregate_movies(valid_files) if movie_agg: write_movies_csv(movie_agg, output_dir, csv_prefix) csv_files_written.append(str(output_dir / f"{csv_prefix}_movies.csv")) print_scan_summary( library_type, valid_files, stats, movie_agg=movie_agg, db_path=args.db, csv_files=csv_files_written ) if __name__ == '__main__': main()