"""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 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. 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 _log_find_nonzero_exit(returncode: int, stderr_text: str, discovered_count: int) -> None: """Log the explicit contract for non-zero `find` exits. Contract: if `find` emits partial stdout before failing, keep those paths and continue with a warning. If no paths were emitted, return an empty result and log that scan discovery was incomplete. """ stderr_suffix = f": {stderr_text}" if stderr_text else "" if discovered_count > 0: logger.warning( "find exited with code %s; using %s partial scan result(s)%s", returncode, discovered_count, stderr_suffix, ) else: logger.warning( "find exited with code %s and produced no scan results%s", returncode, stderr_suffix, ) 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() 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) if process.returncode != 0: stderr_text = stderr.decode(errors="replace").strip() _log_find_nonzero_exit(process.returncode, stderr_text, len(discovered_paths)) 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]], 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")