add scanner metadata cache reuse and control flags

This commit is contained in:
windyboy
2026-02-09 23:55:35 +08:00
parent 59a3b52fee
commit 53aaeeaedf
3 changed files with 225 additions and 6 deletions
+76 -6
View File
@@ -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,
+89
View File
@@ -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
+60
View File
@@ -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: