864 lines
27 KiB
Python
864 lines
27 KiB
Python
#!/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()
|