"""Inventory scanner for discovering and cataloging video files. This module implements the core scanning functionality for the Video Library Manager, including file discovery, metadata extraction, and categorization based on directory structure. """ import csv import json import logging import os import subprocess from concurrent.futures import ThreadPoolExecutor, as_completed from datetime import datetime, timezone from pathlib import Path from typing import Callable, Iterator, Optional from vlm.config import Config from vlm.models import VideoFile logger = logging.getLogger(__name__) def _normalize_to_utc(timestamp: datetime) -> datetime: """Normalize a datetime to a UTC instant.""" if timestamp.tzinfo is None: timestamp = timestamp.astimezone() return timestamp.astimezone(timezone.utc) def scan_library( root: Path, config: Config, progress_callback: Optional[Callable[[int, int], None]] = None, include_video_metadata: bool = True, metadata_cache: Optional[dict[str, VideoFile]] = None ) -> list[VideoFile]: """Recursively scan library for video files. Discovers all video files matching configured extensions within the library root, records their metadata, and categorizes them based on directory structure. Args: root: Root directory to scan config: Configuration object with video extensions and settings progress_callback: Optional callback for progress updates include_video_metadata: Whether to run ffprobe metadata extraction metadata_cache: Optional cached inventory keyed by file path Returns: List of VideoFile objects representing discovered files Note: - Handles inaccessible files gracefully by logging errors and continuing - Performs read-only operations without modifying any files or directories - Categorizes files based on parent directory structure (movie/series/anime/other) """ logger.info(f"Starting library scan at: {root}") if not root.exists(): logger.error(f"Library root does not exist: {root}") return [] if not root.is_dir(): logger.error(f"Library root is not a directory: {root}") return [] if include_video_metadata: import shutil if not shutil.which("ffprobe"): logger.warning( "ffprobe command not found in PATH. Video metadata extraction will be skipped. " "Only file-level information (size, mtime) will be recorded." ) include_video_metadata = False video_files = [] file_count = 0 discovered_paths = _discover_video_paths(root, config.video_extensions) total_paths = len(discovered_paths) if progress_callback is not None: progress_callback(0, total_paths) max_workers = max(1, min(config.enrichment_max_concurrency, total_paths or 1)) use_parallel = include_video_metadata and total_paths > 1 and max_workers > 1 if use_parallel: indexed_results: list[tuple[int, VideoFile]] = [] with ThreadPoolExecutor(max_workers=max_workers) as pool: future_to_index = { pool.submit( _create_video_file, file_path, root, config.categories, include_video_metadata, metadata_cache, ): index for index, file_path in enumerate(discovered_paths, start=1) } for completed_count, future in enumerate(as_completed(future_to_index), start=1): index = future_to_index[future] try: video_file = future.result() except Exception as exc: logger.error(f"Failed to scan file #{index}: {exc}") video_file = None if video_file is not None: indexed_results.append((index, video_file)) file_count += 1 if progress_callback is not None: progress_callback(completed_count, total_paths) if file_count % 100 == 0 and file_count > 0: logger.debug(f"Scanned {file_count} files so far...") indexed_results.sort(key=lambda item: item[0]) video_files = [item[1] for item in indexed_results] else: for index, file_path in enumerate(discovered_paths, start=1): video_file = _create_video_file( file_path, root, config.categories, include_video_metadata=include_video_metadata, metadata_cache=metadata_cache ) if video_file is None: if progress_callback is not None: progress_callback(index, total_paths) continue video_files.append(video_file) file_count += 1 if progress_callback is not None: progress_callback(index, total_paths) if file_count % 100 == 0: logger.debug(f"Scanned {file_count} files so far...") logger.info(f"Scan complete. Found {file_count} video files") return video_files def _discover_video_paths(root: Path, video_extensions: list[str]) -> list[Path]: """Discover matching video files under root using os.scandir recursion.""" discovered_paths: list[Path] = [] for file_path in _scan_directory_recursive(root, video_extensions): if _is_hidden_path(file_path, root): continue discovered_paths.append(file_path) return discovered_paths def _is_hidden_path(file_path: Path, library_root: Path) -> bool: """Return True when any path component under root starts with a dot.""" try: relative_path = file_path.relative_to(library_root) except ValueError: return file_path.name.startswith(".") return any(part.startswith(".") for part in relative_path.parts) def _scan_directory_recursive( directory: Path, video_extensions: list[str] ) -> Iterator[Path]: """Recursively scan a directory for video files. Args: directory: Directory to scan video_extensions: List of configured video extensions Yields: File paths for each discovered video file """ try: # Use os.scandir for efficient directory traversal with os.scandir(directory) as entries: for entry in entries: try: # Skip hidden files and directories (starting with .) if entry.name.startswith('.'): continue if entry.is_file(follow_symlinks=False): # Check if file has video extension file_path = Path(entry.path) if _is_video_file(file_path, video_extensions): yield file_path elif entry.is_dir(follow_symlinks=False): # Recursively scan subdirectory subdir_path = Path(entry.path) yield from _scan_directory_recursive(subdir_path, video_extensions) except (OSError, PermissionError) as e: # Handle inaccessible files/directories gracefully logger.error(f"Cannot access {entry.path}: {e}") continue except (OSError, PermissionError) as e: # Handle inaccessible directory logger.error(f"Cannot access directory {directory}: {e}") def _is_video_file(file_path: Path, video_extensions: list[str]) -> bool: """Check if file has a video extension. Args: file_path: Path to file video_extensions: List of valid video extensions (e.g., [".mp4", ".mkv"]) Returns: True if file has a video extension, False otherwise """ file_extension = file_path.suffix.lower() return file_extension in [ext.lower() for ext in video_extensions] DEFAULT_SIDECAR_EXTENSIONS = (".srt", ".ass", ".sub", ".idx", ".sup", ".nfo") def find_sidecar_companions( video_path: Path, sidecar_extensions: tuple = DEFAULT_SIDECAR_EXTENSIONS, ) -> list[Path]: """Find sidecar files in the same directory that belong to a video file. Conservative matching: a companion must share the video's exact stem, optionally followed by dot-separated alphabetic suffix tokens (e.g. language or track tags such as ``zh`` or ``en.forced``). Numeric or otherwise non-alphabetic tokens are rejected so unrelated files are never associated. Args: video_path: Path to the video file sidecar_extensions: Sidecar extensions to consider (case-insensitive) Returns: Sorted list of companion paths (empty when none are found). """ parent = video_path.parent try: with os.scandir(parent) as it: entries = [e for e in it if e.is_file(follow_symlinks=False)] except OSError: return [] stem = video_path.stem exts = {ext.lower() for ext in sidecar_extensions} companions: list[Path] = [] for entry in entries: name = entry.name suffix = Path(name).suffix.lower() if suffix not in exts: continue base = name[: -len(suffix)] if base == stem: companions.append(Path(entry.path)) continue if not base.startswith(stem + "."): continue tokens = base[len(stem) + 1:].split(".") if tokens and all(token and token.isalpha() for token in tokens): companions.append(Path(entry.path)) return sorted(companions, key=lambda p: p.name) def _create_video_file( file_path: Path, library_root: Path, categories_config: dict[str, list[str]], include_video_metadata: bool = True, metadata_cache: Optional[dict[str, VideoFile]] = None ) -> Optional[VideoFile]: """Create VideoFile object from file path. Extracts file metadata and categorizes based on directory structure. Optionally extracts video metadata using ffprobe if available. Args: file_path: Path to video file library_root: Root of the library (for categorization) categories_config: Mapping of category names to directory name lists include_video_metadata: Whether to run ffprobe metadata extraction metadata_cache: Optional cached inventory keyed by file path Returns: VideoFile object or None if file cannot be accessed """ try: # Get file stats stat = file_path.stat() size_bytes = stat.st_size modified_timestamp = datetime.fromtimestamp(stat.st_mtime, tz=timezone.utc) # Categorize based on directory structure category = categorize_file(file_path, library_root, categories_config) # Reuse cached metadata when file identity is unchanged. video_metadata: dict = {} if include_video_metadata: cache_key = str(file_path) cached_file = metadata_cache.get(cache_key) if metadata_cache else None file_mtime_seconds = int(modified_timestamp.timestamp()) if ( cached_file is not None and cached_file.size_bytes == size_bytes and int(_normalize_to_utc(cached_file.modified_timestamp).timestamp()) == file_mtime_seconds ): video_metadata = { 'resolution': cached_file.resolution, 'codec': cached_file.codec, 'duration_seconds': cached_file.duration_seconds, 'bitrate_kbps': cached_file.bitrate_kbps } else: # Extract video metadata using ffprobe (optional, non-blocking) video_metadata = extract_metadata(file_path) # Create VideoFile object with optional metadata return VideoFile( path=file_path, filename=file_path.name, size_bytes=size_bytes, modified_timestamp=modified_timestamp, category=category, resolution=video_metadata.get('resolution'), codec=video_metadata.get('codec'), duration_seconds=video_metadata.get('duration_seconds'), bitrate_kbps=video_metadata.get('bitrate_kbps') ) except (OSError, PermissionError) as e: logger.error(f"Cannot read file metadata for {file_path}: {e}") return None def load_inventory_csv(input_path: Path) -> list[VideoFile]: """Load inventory CSV into VideoFile objects.""" from datetime import timezone video_files: list[VideoFile] = [] with open(input_path, 'r', encoding='utf-8') as csvfile: lines = [line for line in csvfile if not line.startswith('#')] reader = csv.DictReader(lines) for row in reader: modified_timestamp = datetime.fromisoformat(row['modified_timestamp']) if modified_timestamp.tzinfo is None: modified_timestamp = modified_timestamp.replace(tzinfo=timezone.utc) resolution = row.get('resolution') or None codec = row.get('codec') or None duration_seconds = float(row['duration_seconds']) if row.get('duration_seconds') else None bitrate_kbps = int(row['bitrate_kbps']) if row.get('bitrate_kbps') else None video_files.append( VideoFile( path=Path(row['path']), filename=row['filename'], size_bytes=int(row['size_bytes']), modified_timestamp=modified_timestamp, category=row['category'], resolution=resolution, codec=codec, duration_seconds=duration_seconds, bitrate_kbps=bitrate_kbps ) ) return video_files def categorize_file( file_path: Path, library_root: Path, categories_config: dict[str, list[str]] ) -> str: """Determine category based on directory structure. Categories are determined by the top-level directory within the library root, matched against configured category mappings. Args: file_path: Path to video file library_root: Root of the library categories_config: Mapping of category names to directory name lists Example: {"movie": ["movie", "movies"], "series": ["series", "tv"]} Returns: Category string: "movie", "series", "anime", or "other" """ try: # Get relative path from library root relative_path = file_path.relative_to(library_root) # Get the first component of the relative path (top-level directory) parts = relative_path.parts if len(parts) > 0: top_level_dir = parts[0].lower() # Check against configured category mappings for category, dir_names in categories_config.items(): # Case-insensitive matching if top_level_dir in [name.lower() for name in dir_names]: return category # No match found return "other" else: # File is directly in library root return "other" except ValueError: # File is not within library root logger.warning(f"File {file_path} is not within library root {library_root}") return "other" def extract_metadata(file_path: Path) -> dict: """Extract video metadata using ffprobe. Attempts to extract resolution, codec, duration, and bitrate from video file using ffprobe. If ffprobe is not available or fails, returns empty dict. This is a non-blocking operation that gracefully handles failures. Args: file_path: Path to video file Returns: Dictionary with optional keys: - resolution: str (e.g., "1920x1080") - codec: str (e.g., "h264") - duration_seconds: float - bitrate_kbps: int Note: - Returns empty dict if ffprobe is not available - Returns empty dict if ffprobe fails to extract metadata - Logs warnings for failures but does not raise exceptions """ try: # Run ffprobe to get video stream information in JSON format result = subprocess.run( [ 'ffprobe', '-v', 'quiet', # Suppress ffprobe output '-print_format', 'json', # Output as JSON '-show_streams', # Show stream information '-show_format', # Show format information str(file_path) ], capture_output=True, text=True, timeout=10 # 10 second timeout to prevent hanging ) if result.returncode != 0: logger.debug(f"ffprobe failed for {file_path.name}: {result.stderr}") return {} # Parse JSON output probe_data = json.loads(result.stdout) # Extract metadata from the first video stream metadata = {} # Find the first video stream video_stream = None for stream in probe_data.get('streams', []): if stream.get('codec_type') == 'video': video_stream = stream break if video_stream: # Extract resolution width = video_stream.get('width') height = video_stream.get('height') if width and height: metadata['resolution'] = f"{width}x{height}" # Extract codec codec_name = video_stream.get('codec_name') if codec_name: metadata['codec'] = codec_name # Extract duration and bitrate from format section format_info = probe_data.get('format', {}) # Extract duration duration = format_info.get('duration') if duration: try: metadata['duration_seconds'] = float(duration) except (ValueError, TypeError): pass # Extract bitrate bitrate = format_info.get('bit_rate') if bitrate: try: # Convert from bits/sec to kbits/sec metadata['bitrate_kbps'] = int(float(bitrate) / 1000) except (ValueError, TypeError): pass if metadata: logger.debug(f"Extracted metadata for {file_path.name}: {metadata}") return metadata except FileNotFoundError: # ffprobe not installed or not in PATH logger.debug("ffprobe not available - skipping metadata extraction") return {} except subprocess.TimeoutExpired: logger.warning(f"ffprobe timeout for {file_path.name} - skipping metadata") return {} except json.JSONDecodeError as e: logger.warning(f"Failed to parse ffprobe output for {file_path.name}: {e}") return {} except Exception as e: # Catch any other unexpected errors logger.warning(f"Unexpected error extracting metadata for {file_path.name}: {e}") return {} def save_inventory_csv(files: list[VideoFile], output: Path, library_root: Path) -> None: """Save inventory to CSV format (primary format). Generates a CSV file with all file metadata following the defined schema: path, filename, size_bytes, modified_timestamp, category, resolution, codec, duration_seconds, bitrate_kbps Args: files: List of VideoFile objects to export output: Path to output CSV file library_root: Root of the library (included in report metadata) Note: - Timestamps are formatted as ISO 8601 (YYYY-MM-DDTHH:MM:SS) in UTC - Missing optional values are represented as empty strings - Header row is always present with column names - Generation timestamp and library root are included as comment lines """ logger.info(f"Saving inventory to CSV: {output}") # Ensure output directory exists output.parent.mkdir(parents=True, exist_ok=True) # Get generation timestamp in UTC generation_timestamp = datetime.now(timezone.utc).strftime("%Y-%m-%dT%H:%M:%S") with open(output, 'w', newline='', encoding='utf-8') as csvfile: # Write metadata as comments csvfile.write("# vlm_schema_version: 1.0\n") csvfile.write(f"# Generated: {generation_timestamp}\n") csvfile.write(f"# Library Root: {library_root}\n") # Define CSV schema fieldnames = [ 'path', 'filename', 'size_bytes', 'modified_timestamp', 'category', 'resolution', 'codec', 'duration_seconds', 'bitrate_kbps' ] writer = csv.DictWriter(csvfile, fieldnames=fieldnames) writer.writeheader() # Write each file for video_file in files: # Format timestamp as ISO 8601 in UTC modified_utc = _normalize_to_utc(video_file.modified_timestamp) row = { 'path': str(video_file.path), 'filename': video_file.filename, 'size_bytes': video_file.size_bytes, 'modified_timestamp': modified_utc.strftime("%Y-%m-%dT%H:%M:%S"), 'category': video_file.category, 'resolution': video_file.resolution or '', 'codec': video_file.codec or '', 'duration_seconds': video_file.duration_seconds if video_file.duration_seconds is not None else '', 'bitrate_kbps': video_file.bitrate_kbps if video_file.bitrate_kbps is not None else '' } writer.writerow(row) logger.info(f"Saved {len(files)} files to CSV inventory") def save_inventory_json(files: list[VideoFile], output: Path, library_root: Path) -> None: """Save inventory to JSON format (optional export format). Generates a JSON file with all file metadata and report metadata. Args: files: List of VideoFile objects to export output: Path to output JSON file library_root: Root of the library (included in report metadata) Note: - Timestamps are formatted as ISO 8601 strings - Missing optional values are represented as null - Generation timestamp and library root are included in metadata section """ logger.info(f"Saving inventory to JSON: {output}") # Ensure output directory exists output.parent.mkdir(parents=True, exist_ok=True) # Get generation timestamp in UTC generation_timestamp = datetime.now(timezone.utc).strftime("%Y-%m-%dT%H:%M:%S") # Build JSON structure inventory_data = { 'metadata': { 'generated': generation_timestamp, 'library_root': str(library_root), 'file_count': len(files) }, 'files': [] } # Add each file for video_file in files: # Format timestamp as ISO 8601 in UTC modified_utc = _normalize_to_utc(video_file.modified_timestamp) file_data = { 'path': str(video_file.path), 'filename': video_file.filename, 'size_bytes': video_file.size_bytes, 'modified_timestamp': modified_utc.strftime("%Y-%m-%dT%H:%M:%S"), 'category': video_file.category, 'resolution': video_file.resolution, 'codec': video_file.codec, 'duration_seconds': video_file.duration_seconds, 'bitrate_kbps': video_file.bitrate_kbps } inventory_data['files'].append(file_data) # Write JSON file with pretty formatting with open(output, 'w', encoding='utf-8') as jsonfile: json.dump(inventory_data, jsonfile, indent=2, ensure_ascii=False) logger.info(f"Saved {len(files)} files to JSON inventory")