"""Inventory scanner for discovering and cataloging video files. This module implements the core scanning functionality for the Video Library Manager, including file discovery via `find`, metadata extraction, and categorization based on directory structure. """ import csv import json import logging import os import subprocess 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 ) -> 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 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 [] 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) for index, file_path in enumerate(discovered_paths, start=1): video_file = _create_video_file(file_path, root, config.categories) 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. Uses the system `find` command for traversal speed and falls back to Python recursion if `find` is unavailable. """ try: return _discover_video_paths_with_find(root, video_extensions) except FileNotFoundError: logger.warning("`find` command not available - falling back to Python recursion") return _discover_video_paths_recursive(root, video_extensions) def _discover_video_paths_with_find(root: Path, video_extensions: list[str]) -> list[Path]: """Discover matching video files using the system `find` command.""" normalized_extensions = [ext.lower() for ext in video_extensions if ext] if not normalized_extensions: return [] command: list[str] = ["find", str(root), "-type", "f", "("] for index, extension in enumerate(normalized_extensions): if index > 0: command.append("-o") command.extend(["-iname", f"*{extension}"]) command.extend([")", "-print0"]) process = subprocess.Popen( command, stdout=subprocess.PIPE, stderr=subprocess.PIPE ) stdout, stderr = process.communicate() if process.returncode != 0: stderr_text = stderr.decode(errors="replace").strip() if stderr_text: logger.warning(f"find reported issues while scanning: {stderr_text}") discovered_paths: list[Path] = [] for path_bytes in stdout.split(b"\0"): if not path_bytes: continue file_path = Path(os.fsdecode(path_bytes)) if _is_hidden_path(file_path, root): continue discovered_paths.append(file_path) return discovered_paths def _discover_video_paths_recursive(root: Path, video_extensions: list[str]) -> list[Path]: """Fallback discovery using Python directory traversal.""" 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] def _create_video_file( file_path: Path, library_root: Path, categories_config: dict[str, list[str]] ) -> 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 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) # 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 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(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")