diff --git a/src/vlm/scanner.py b/src/vlm/scanner.py index 05f0cde..985576b 100644 --- a/src/vlm/scanner.py +++ b/src/vlm/scanner.py @@ -30,7 +30,9 @@ def _normalize_to_utc(timestamp: datetime) -> datetime: def scan_library( root: Path, config: Config, - progress_callback: Optional[Callable[[int, int], None]] = None + 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. @@ -40,6 +42,9 @@ def scan_library( 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 @@ -68,7 +73,13 @@ def scan_library( 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) + 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) @@ -215,7 +226,9 @@ def _is_video_file(file_path: Path, video_extensions: list[str]) -> bool: def _create_video_file( file_path: Path, library_root: Path, - categories_config: dict[str, list[str]] + 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. @@ -226,6 +239,8 @@ def _create_video_file( 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 @@ -238,9 +253,28 @@ def _create_video_file( # 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) + + # 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( @@ -260,6 +294,42 @@ def _create_video_file( 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, diff --git a/tests/test_cli_scan.py b/tests/test_cli_scan.py new file mode 100644 index 0000000..47fe30c --- /dev/null +++ b/tests/test_cli_scan.py @@ -0,0 +1,89 @@ +"""Tests for CLI scan command options.""" + +from pathlib import Path +from unittest.mock import patch + +from click.testing import CliRunner + +from vlm.cli import main + + +def test_scan_no_metadata_passes_flag_to_scanner(tmp_path): + """Test `scan --no-metadata` disables ffprobe metadata extraction.""" + library_root = tmp_path / "library" + library_root.mkdir(parents=True) + + config_file = tmp_path / "config.yaml" + config_file.write_text( + f""" +library_root: {library_root} +video_extensions: + - .mp4 +categories: + movie: [movie, movies] + series: [series, tv, shows] + anime: [anime] +""".strip() + ) + + runner = CliRunner() + with patch("vlm.scanner.scan_library", return_value=[]) as mock_scan, patch( + "vlm.scanner.save_inventory_csv" + ) as mock_save: + result = runner.invoke( + main, + ["--config", str(config_file), "scan", "--no-metadata", "--output", str(tmp_path / "inv.csv")] + ) + + assert result.exit_code == 0 + assert mock_scan.call_count == 1 + _, kwargs = mock_scan.call_args + assert kwargs["include_video_metadata"] is False + assert "progress_callback" in kwargs + assert mock_save.call_count == 1 + + +def test_scan_force_refresh_metadata_disables_cache(tmp_path): + """Test `scan --force-refresh-metadata` skips cache loading.""" + library_root = tmp_path / "library" + library_root.mkdir(parents=True) + + config_file = tmp_path / "config.yaml" + config_file.write_text( + f""" +library_root: {library_root} +video_extensions: + - .mp4 +categories: + movie: [movie, movies] + series: [series, tv, shows] + anime: [anime] +""".strip() + ) + + output_file = tmp_path / "inventory.csv" + output_file.write_text( + "# Generated: 2026-02-09T00:00:00\n" + f"# Library Root: {library_root}\n" + "path,filename,size_bytes,modified_timestamp,category,resolution,codec,duration_seconds,bitrate_kbps\n" + ) + + runner = CliRunner() + with patch("vlm.scanner.load_inventory_csv") as mock_load_cache, patch( + "vlm.scanner.scan_library", return_value=[] + ) as mock_scan, patch("vlm.scanner.save_inventory_csv") as mock_save: + result = runner.invoke( + main, + [ + "--config", str(config_file), "scan", + "--output", str(output_file), + "--force-refresh-metadata" + ] + ) + + assert result.exit_code == 0 + mock_load_cache.assert_not_called() + _, kwargs = mock_scan.call_args + assert kwargs["include_video_metadata"] is True + assert kwargs["metadata_cache"] is None + assert mock_save.call_count == 1 diff --git a/tests/test_scanner.py b/tests/test_scanner.py index 7b8bf5a..b34a1ac 100644 --- a/tests/test_scanner.py +++ b/tests/test_scanner.py @@ -3,6 +3,7 @@ import os import tempfile import time +import csv from datetime import datetime, timezone from pathlib import Path from unittest.mock import patch, MagicMock @@ -17,6 +18,8 @@ from vlm.scanner import ( categorize_file, scan_library, extract_metadata, + save_inventory_csv, + load_inventory_csv, ) @@ -698,6 +701,63 @@ class TestExtractMetadata: assert vf.duration_seconds is None assert vf.bitrate_kbps is None + def test_scan_library_with_metadata_disabled(self, tmp_path): + """Test that metadata extraction can be disabled for faster scans.""" + movie_dir = tmp_path / "movie" + movie_dir.mkdir() + + video_file = movie_dir / "test.mp4" + video_file.touch() + + config = Config(library_root=tmp_path) + + with patch('subprocess.run') as mock_run: + result = scan_library(tmp_path, config, include_video_metadata=False) + + assert len(result) == 1 + assert result[0].filename == "test.mp4" + assert result[0].resolution is None + assert result[0].codec is None + assert result[0].duration_seconds is None + assert result[0].bitrate_kbps is None + mock_run.assert_not_called() + + def test_scan_library_reuses_cached_metadata_when_unchanged(self, tmp_path): + """Test unchanged files reuse metadata from previous inventory cache.""" + movie_dir = tmp_path / "movie" + movie_dir.mkdir() + video_file = movie_dir / "test.mp4" + video_file.write_text("test") + + original = VideoFile( + path=video_file, + filename=video_file.name, + size_bytes=video_file.stat().st_size, + modified_timestamp=datetime.fromtimestamp(video_file.stat().st_mtime, tz=timezone.utc), + category="movie", + resolution="1920x1080", + codec="h264", + duration_seconds=120.5, + bitrate_kbps=5000, + ) + + cache_file = tmp_path / "cached_inventory.csv" + save_inventory_csv([original], cache_file, tmp_path) + cached_entries = load_inventory_csv(cache_file) + metadata_cache = {str(vf.path): vf for vf in cached_entries} + + config = Config(library_root=tmp_path) + with patch("subprocess.run") as mock_run: + result = scan_library(tmp_path, config, metadata_cache=metadata_cache) + + assert len(result) == 1 + scanned = result[0] + assert scanned.resolution == "1920x1080" + assert scanned.codec == "h264" + assert scanned.duration_seconds == 120.5 + assert scanned.bitrate_kbps == 5000 + mock_run.assert_not_called() + class TestInventoryReports: