windy/dl-organizer
1
0
Fork 0
Code Issues Pull Requests Packages Projects Releases Wiki Activity

Fix identity reconstruction metadata loss with v2 schema

Browse Source 
Implements identities.json v2 schema with embedded video metadata to fix
duplicate resolution by quality, which previously failed due to VideoFile
objects being reconstructed with hardcoded defaults (size_bytes=0,
resolution=None, codec=None).

Changes:
- Add --inventory flag to vlm parse command to embed video metadata
- Update _video_file_from_record() to extract embedded metadata if present
- Add vlm_schema_version field to identities.json (v1.0 or v2.0)
- Maintain backward compatibility with v1 files (no metadata)

Schema v2 format:
- Embeds video_metadata object in each record (movies/series)
- Contains: size_bytes, modified_timestamp, resolution, codec,
  duration_seconds, bitrate_kbps
- Enables accurate quality comparison during duplicate analysis

Testing:
- Added comprehensive unit tests for io.py functions
- Added CLI integration tests for parse command
- Added end-to-end tests for duplicate quality comparison
- All 437 existing tests still pass (1 pre-existing failure in executor)

Documentation:
- Updated README.md with --inventory usage examples
- Updated CLAUDE.md with schema versioning details
- Added workflow examples showing metadata embedding

This fix resolves the critical P0 issue where duplicate resolution by_quality
strategy failed completely due to missing video metadata in reconstructed
VideoFile objects.

Co-Authored-By: Claude Sonnet 4.5 <noreply@anthropic.com>
This commit is contained in:
windyboy
2026-02-13 09:44:51 +08:00
co-authored by Claude Sonnet 4.5
parent 1f55eab304
commit 065195b83b
7 changed files with 845 additions and 39 deletions
Download Patch File Download Diff File Expand all files Collapse all files
CLAUDE.md
+43 -1
View File
@@ -39,12 +39,31 @@ vlm config init
# Common workflow
vlm scan # Discover files
vlm parse # Extract identities
vlm parse # Extract identities (v1 schema)
vlm parse --inventory inventory.csv # Extract identities with embedded metadata (v2 schema, recommended)
vlm enrich # (Optional) Enrich titles/reputation via TMDB
vlm enrich --refresh-all # Force full refresh (ignore cache)
vlm analyze # Detect gaps/duplicates
vlm plan # Generate execution plan
vlm plan --analysis analysis.json # Generate plan with duplicate resolution
vlm execute # Dry-run (default)
vlm execute --confirm # Actually execute
vlm rollback # Undo executed operations
# Reporting
vlm report summary # Overview statistics
vlm report inventory # File inventory
vlm report completeness # Series with missing episodes
vlm report duplicates # Duplicate files with quality comparison
# Quarantine management
vlm quarantine list # List quarantined files
vlm quarantine add <file> --reason "duplicate"
vlm quarantine restore <file>
# State management
vlm state show <file>
vlm state set <file> --status reviewed
```
## Architecture
@@ -53,6 +72,8 @@ vlm execute --confirm # Actually execute
VLM follows a read-first, multi-stage pipeline:
1. **Scan** → discovers video files, extracts metadata via ffprobe (optional), saves to inventory.csv
2. **Parse** → extracts titles/years/seasons/episodes from filenames, saves to identities.json
- Use `--inventory inventory.csv` to embed video metadata (v2 schema) for accurate duplicate resolution by quality
- Without `--inventory`, produces v1 schema (lightweight, no embedded metadata)
3. **Enrich** (optional) → adds bilingual titles and reputation (TMDB); updates identities.json in place; uses SQLite cache for incremental runs
4. **Analyze** → detects episode gaps and duplicates, saves to analysis.json
5. **Plan** → generates reviewable execution plan (plan.json) with file operations
@@ -143,6 +164,27 @@ This allows files in `/library/movies/` or `/library/films/` to be recognized as
**Migration Note:** If you have existing directories with non-standard names (like "movies" or "tv"), update your config.yaml and re-run `vlm scan` to fix categorization. No files will be moved.
### Schema Versioning
**identities.json Schema Versions:**
- **v1** (default without `--inventory`): Lightweight schema without embedded metadata
- Records contain: path, filename, category, title, year/season/episodes, confidence, needs_review
- VideoFile objects reconstructed with defaults: size_bytes=0, resolution=None, codec=None
- Suitable for basic organization workflows
- **v2** (with `--inventory`): Enhanced schema with embedded video metadata
- All v1 fields plus `video_metadata` object containing:
- size_bytes, modified_timestamp, resolution, codec, duration_seconds, bitrate_kbps
- Enables accurate duplicate resolution by quality (compare resolution, codec, file size)
- Required for `by_quality` duplicate resolution strategy
- Backward compatible: v1 files load without errors
**Implementation Details:**
- `_video_file_from_record()` in io.py extracts embedded metadata if present
- Parse command with `--inventory` flag loads inventory.csv and embeds metadata in output
- Schema version stored in `vlm_schema_version` field at root level of identities.json
### File Discovery
Uses system `find` command for speed, falls back to Python recursion if unavailable. Hidden paths (starting with `.`) are skipped automatically.
README.md
+28 -11
View File
@@ -68,12 +68,19 @@ Extract titles, years, seasons, and episodes from filenames:
vlm parse
```
This creates `identities.json` with parsed information.
**For accurate duplicate resolution by quality, embed video metadata:**
```bash
vlm parse --inventory inventory.csv
```
This creates `identities.json` with parsed information. When using `--inventory`, video metadata (size, resolution, codec) is embedded, enabling accurate quality comparison during duplicate analysis.
**What it extracts:**
- **Movies**: Title and year (e.g., "Inception (2010)")
- **Series**: Title, season, and episode numbers (e.g., "Breaking Bad S01E01")
- **Confidence scores**: Indicates parsing reliability
- **Video metadata** (with `--inventory`): Size, resolution, codec, duration, bitrate
### 4. Enrich Titles and Reputation (Optional but Recommended)
@@ -180,9 +187,9 @@ vlm config init
vlm scan
# Output: inventory.csv with 1234 files discovered
# 3. Parse filenames
vlm parse
# Output: identities.json with parsed titles and episodes
# 3. Parse filenames with metadata embedding (recommended for duplicate resolution)
vlm parse --inventory inventory.csv
# Output: identities.json with parsed titles, episodes, and embedded video metadata (v2 schema)
# 4. Enrich identities (translation + reputation)
vlm enrich
@@ -190,7 +197,7 @@ vlm enrich
# 5. Analyze for gaps and duplicates
vlm analyze
# Output: analysis.json with 5 series with gaps, 12 duplicate groups
# Output: analysis.json with 5 series with gaps, 12 duplicate groups (accurate quality comparison)
# 6. Generate execution plan (optionally use analysis for duplicate handling)
vlm plan --analysis analysis.json
@@ -246,13 +253,23 @@ vlm scan --output my_library.csv
### Parsing
```bash
# Parse with default files
# Parse with default files (v1 schema - no metadata embedding)
vlm parse
# Parse with metadata embedding (v2 schema - enables quality comparison)
vlm parse --inventory inventory.csv
# Parse with custom input/output
vlm parse --input my_inventory.csv --output my_identities.json
# Parse with metadata from custom inventory
vlm parse --input my_inventory.csv --output my_identities.json --inventory my_inventory.csv
```
**Schema Versions:**
- **v1** (without `--inventory`): Lightweight identities, suitable for basic organization
- **v2** (with `--inventory`): Embeds video metadata, required for accurate duplicate resolution by quality
### Enrichment
```bash
@@ -557,7 +574,7 @@ vlm report summary
```bash
# 1. Scan and parse
vlm scan
vlm parse
vlm parse --inventory inventory.csv
vlm enrich
# 2. Analyze completeness
@@ -571,15 +588,15 @@ vlm report completeness --plan plan.json
### Scenario 3: Finding and Removing Duplicates
```bash
# 1. Scan and parse
# 1. Scan and parse (with --inventory for accurate quality comparison)
vlm scan
vlm parse
vlm parse --inventory inventory.csv
vlm enrich
# 2. Analyze for duplicates
vlm analyze
# 3. View duplicates with quality comparison
# 3. View duplicates with quality comparison (now shows actual resolution/codec/size)
vlm report duplicates
# 4. Generate plan with analysis: VLM keeps one file per duplicate group (by reputation) and quarantines the rest
@@ -604,7 +621,7 @@ vlm report duplicates --plan plan.json
# 2. Scan and parse
vlm scan
vlm parse
vlm parse --inventory inventory.csv
vlm enrich
# 3. Generate plan
src/vlm/cli.py
+68 -20
View File
@@ -169,16 +169,26 @@ def scan(
default=Path('identities.json'),
help='Output file for parsed identities (default: identities.json)'
)
@click.option(
'--inventory',
type=click.Path(exists=True, path_type=Path),
default=None,
help='Inventory CSV to embed video metadata (enables v2 schema with quality data)'
)
@pass_context
def parse(ctx: CLIContext, input: Path, output: Path):
def parse(ctx: CLIContext, input: Path, output: Path, inventory: Optional[Path]):
"""Parse identities from filenames.
Extracts movie titles, years, series titles, seasons, and episodes
from video filenames in the inventory.
Use --inventory to embed video metadata (size, resolution, codec) in the output,
which enables accurate duplicate resolution by quality in the analyze stage.
Example:
vlm parse # Use default files
vlm parse # Use default files (v1 schema)
vlm parse --inventory inventory.csv # Embed metadata (v2 schema)
vlm parse --input my_inventory.csv # Custom input
vlm parse --output parsed_identities.json # Custom output
"""
@@ -186,15 +196,25 @@ def parse(ctx: CLIContext, input: Path, output: Path):
import json
from datetime import datetime, timezone
from vlm.parser import parse_movie, parse_series
from vlm.io import load_inventory_csv
config = ctx.config
logger = ctx.logger
try:
# Display parse start message
click.echo(f"Parsing identities from: {input}")
# Load video metadata from inventory if provided
path_to_metadata = {}
if inventory:
click.echo(f"Loading video metadata from: {inventory}")
inventory_files = load_inventory_csv(inventory)
path_to_metadata = {str(vf.path): vf for vf in inventory_files}
click.echo(f"Loaded metadata for {len(path_to_metadata)} files")
click.echo()
# Load inventory from CSV
video_files = []
with open(input, 'r', encoding='utf-8') as csvfile:
@@ -203,7 +223,7 @@ def parse(ctx: CLIContext, input: Path, output: Path):
for line in csvfile:
if not line.startswith('#'):
lines.append(line)
# Parse CSV
reader = csv.DictReader(lines)
for row in reader:
@@ -212,7 +232,7 @@ def parse(ctx: CLIContext, input: Path, output: Path):
'filename': row['filename'],
'category': row['category']
})
click.echo(f"Loaded {len(video_files)} files from inventory")
click.echo()
@@ -221,27 +241,48 @@ def parse(ctx: CLIContext, input: Path, output: Path):
series_identities = []
anime_files = []
other_files = []
def get_video_metadata(file_path: str) -> dict:
"""Extract video metadata from inventory if available."""
if not path_to_metadata:
return {}
vf = path_to_metadata.get(file_path)
if not vf:
return {}
return {
'size_bytes': vf.size_bytes,
'modified_timestamp': vf.modified_timestamp.isoformat(),
'resolution': vf.resolution,
'codec': vf.codec,
'duration_seconds': vf.duration_seconds,
'bitrate_kbps': vf.bitrate_kbps,
}
for vf in video_files:
filename = vf['filename']
category = vf['category']
file_path = vf['path']
video_metadata = get_video_metadata(file_path)
if category == 'movie':
identity = parse_movie(filename, extensions=config.video_extensions)
movie_identities.append({
'path': vf['path'],
record = {
'path': file_path,
'filename': filename,
'category': category,
'title': identity.title,
'year': identity.year,
'confidence': identity.confidence,
'needs_review': identity.needs_review
})
}
if video_metadata:
record['video_metadata'] = video_metadata
movie_identities.append(record)
elif category == 'series':
identity = parse_series(filename, extensions=config.video_extensions)
series_identities.append({
'path': vf['path'],
record = {
'path': file_path,
'filename': filename,
'category': category,
'title': identity.title,
@@ -249,7 +290,10 @@ def parse(ctx: CLIContext, input: Path, output: Path):
'episodes': identity.episodes,
'confidence': identity.confidence,
'needs_review': identity.needs_review
})
}
if video_metadata:
record['video_metadata'] = video_metadata
series_identities.append(record)
elif category == 'anime':
# Anime files are not parsed in v1
@@ -299,8 +343,12 @@ def parse(ctx: CLIContext, input: Path, output: Path):
# Build JSON structure
generation_timestamp = datetime.now(timezone.utc).strftime("%Y-%m-%dT%H:%M:%S")
# Use v2 schema if video metadata was embedded
schema_version = "2.0" if path_to_metadata else "1.0"
identities_data = {
'vlm_schema_version': schema_version,
'metadata': {
'generated': generation_timestamp,
'source_inventory': str(input),
src/vlm/io.py
+21 -7
View File
@@ -46,17 +46,31 @@ def save_identities_json(data: dict, path: Path) -> None:
def _video_file_from_record(record: dict) -> VideoFile:
"""Build a minimal VideoFile from an identities record (no inventory metadata)."""
"""Build a VideoFile from an identities record.
If the record contains embedded video_metadata (v2 schema), use it.
Otherwise, use defaults (v1 schema backward compatibility).
"""
vm = record.get("video_metadata", {})
# Parse modified_timestamp if provided (v2), otherwise use current time (v1)
modified_timestamp_str = vm.get("modified_timestamp")
if modified_timestamp_str:
from datetime import datetime
modified_timestamp = datetime.fromisoformat(modified_timestamp_str)
else:
modified_timestamp = utc_now()
return VideoFile(
path=Path(record["path"]),
filename=record["filename"],
size_bytes=0,
modified_timestamp=utc_now(),
size_bytes=vm.get("size_bytes", 0),
modified_timestamp=modified_timestamp,
category=record["category"],
resolution=None,
codec=None,
duration_seconds=None,
bitrate_kbps=None,
resolution=vm.get("resolution"),
codec=vm.get("codec"),
duration_seconds=vm.get("duration_seconds"),
bitrate_kbps=vm.get("bitrate_kbps"),
)
tests/test_cli_parse.py
+171
View File
@@ -0,0 +1,171 @@
"""Tests for the parse CLI command."""
import json
from pathlib import Path
from click.testing import CliRunner
from vlm.cli import main
class TestParseCommand:
"""Tests for vlm parse command."""
def test_parse_without_inventory_v1_schema(self, tmp_path):
"""Test parse without --inventory flag produces v1 schema."""
# Create a simple inventory CSV (without metadata)
inventory_csv = tmp_path / "inventory.csv"
inventory_csv.write_text(
"# vlm inventory\n"
"path,filename,size_bytes,modified_timestamp,category,resolution,codec,duration_seconds,bitrate_kbps\n"
"/library/movie/Matrix (1999).mkv,Matrix (1999).mkv,1000000,2024-01-01T00:00:00,movie,,,\n"
"/library/series/Breaking Bad S01E01.mkv,Breaking Bad S01E01.mkv,500000,2024-01-02T00:00:00,series,,,\n"
)
output_json = tmp_path / "identities.json"
runner = CliRunner()
result = runner.invoke(
main,
[
"--config",
str(tmp_path / "config.yaml"),
"parse",
"--input",
str(inventory_csv),
"--output",
str(output_json),
],
)
# Command should succeed
assert result.exit_code == 0
# Load output and verify v1 schema (no version field or version is 1.0)
with open(output_json) as f:
data = json.load(f)
# v1 schema - no vlm_schema_version or it's "1.0"
assert data.get("vlm_schema_version", "1.0") == "1.0"
# Verify no video_metadata in records
if data.get("movies"):
for movie in data["movies"]:
assert "video_metadata" not in movie
if data.get("series"):
for series in data["series"]:
assert "video_metadata" not in series
def test_parse_with_inventory_v2_schema(self, tmp_path):
"""Test parse with --inventory flag produces v2 schema with embedded metadata."""
# Create inventory CSV with full metadata
inventory_csv = tmp_path / "inventory.csv"
inventory_csv.write_text(
"# vlm inventory\n"
"path,filename,size_bytes,modified_timestamp,category,resolution,codec,duration_seconds,bitrate_kbps\n"
"/library/movie/Matrix (1999).mkv,Matrix (1999).mkv,2000000000,2024-01-01T10:00:00+00:00,movie,1920x1080,h264,7200.5,5000\n"
"/library/series/Breaking Bad S01E01.mkv,Breaking Bad S01E01.mkv,500000000,2024-01-02T15:00:00+00:00,series,1280x720,h264,2700.0,3000\n"
)
output_json = tmp_path / "identities.json"
runner = CliRunner()
result = runner.invoke(
main,
[
"--config",
str(tmp_path / "config.yaml"),
"parse",
"--input",
str(inventory_csv),
"--output",
str(output_json),
"--inventory",
str(inventory_csv),
],
)
# Command should succeed
assert result.exit_code == 0
assert "Loading video metadata from:" in result.output
assert "Loaded metadata for 2 files" in result.output
# Load output and verify v2 schema
with open(output_json) as f:
data = json.load(f)
# v2 schema version
assert data["vlm_schema_version"] == "2.0"
# Verify video_metadata embedded in movie record
assert len(data["movies"]) == 1
movie = data["movies"][0]
assert "video_metadata" in movie
assert movie["video_metadata"]["size_bytes"] == 2000000000
assert movie["video_metadata"]["resolution"] == "1920x1080"
assert movie["video_metadata"]["codec"] == "h264"
assert movie["video_metadata"]["duration_seconds"] == 7200.5
assert movie["video_metadata"]["bitrate_kbps"] == 5000
# Verify video_metadata embedded in series record
assert len(data["series"]) == 1
series = data["series"][0]
assert "video_metadata" in series
assert series["video_metadata"]["size_bytes"] == 500000000
assert series["video_metadata"]["resolution"] == "1280x720"
assert series["video_metadata"]["codec"] == "h264"
assert series["video_metadata"]["duration_seconds"] == 2700.0
assert series["video_metadata"]["bitrate_kbps"] == 3000
def test_parse_with_inventory_partial_metadata(self, tmp_path):
"""Test parse with inventory that has partial metadata."""
# Create inventory with some files having metadata, some not
inventory_csv = tmp_path / "inventory.csv"
inventory_csv.write_text(
"# vlm inventory\n"
"path,filename,size_bytes,modified_timestamp,category,resolution,codec,duration_seconds,bitrate_kbps\n"
"/library/movie/Matrix (1999).mkv,Matrix (1999).mkv,2000000000,2024-01-01T10:00:00+00:00,movie,1920x1080,h264,,\n"
"/library/movie/Inception (2010).mkv,Inception (2010).mkv,1500000000,2024-01-02T10:00:00+00:00,movie,,,\n"
)
output_json = tmp_path / "identities.json"
runner = CliRunner()
result = runner.invoke(
main,
[
"--config",
str(tmp_path / "config.yaml"),
"parse",
"--input",
str(inventory_csv),
"--output",
str(output_json),
"--inventory",
str(inventory_csv),
],
)
# Command should succeed
assert result.exit_code == 0
# Load output and verify
with open(output_json) as f:
data = json.load(f)
assert data["vlm_schema_version"] == "2.0"
assert len(data["movies"]) == 2
# First movie has partial metadata
movie1 = data["movies"][0]
assert movie1["video_metadata"]["size_bytes"] == 2000000000
assert movie1["video_metadata"]["resolution"] == "1920x1080"
assert movie1["video_metadata"]["codec"] == "h264"
assert movie1["video_metadata"]["duration_seconds"] is None
assert movie1["video_metadata"]["bitrate_kbps"] is None
# Second movie has minimal metadata
movie2 = data["movies"][1]
assert movie2["video_metadata"]["size_bytes"] == 1500000000
assert movie2["video_metadata"]["resolution"] is None
assert movie2["video_metadata"]["codec"] is None
tests/test_duplicate_quality_metadata.py
+252
View File
@@ -0,0 +1,252 @@
"""Integration tests for duplicate resolution with embedded video metadata."""
import json
from pathlib import Path
import pytest
from vlm.io import load_identities_json, identities_to_analysis_input
from vlm.analysis import detect_duplicates
class TestDuplicateQualityWithMetadata:
"""Test that embedded metadata enables accurate quality comparison."""
def test_duplicate_quality_comparison_with_v2_metadata(self, tmp_path):
"""Test that v2 identities with embedded metadata produce accurate quality comparison."""
# Create v2 identities.json with two duplicates having different quality
identities_file = tmp_path / "identities.json"
data = {
"vlm_schema_version": "2.0",
"metadata": {
"generated": "2024-02-13T12:00:00",
"source_inventory": "inventory.csv",
"total_files": 2,
},
"movies": [
{
"path": "/library/movie/Matrix (1999) 1080p.mkv",
"filename": "Matrix (1999) 1080p.mkv",
"category": "movie",
"title": "The Matrix",
"year": 1999,
"confidence": 0.9,
"needs_review": False,
"video_metadata": {
"size_bytes": 2000000000,
"modified_timestamp": "2024-01-01T10:00:00+00:00",
"resolution": "1920x1080",
"codec": "h264",
"duration_seconds": 7200.0,
"bitrate_kbps": 5000,
},
},
{
"path": "/library/movie/Matrix (1999) 720p.mkv",
"filename": "Matrix (1999) 720p.mkv",
"category": "movie",
"title": "The Matrix",
"year": 1999,
"confidence": 0.9,
"needs_review": False,
"video_metadata": {
"size_bytes": 1000000000,
"modified_timestamp": "2024-01-02T10:00:00+00:00",
"resolution": "1280x720",
"codec": "h264",
"duration_seconds": 7200.0,
"bitrate_kbps": 2500,
},
},
],
"series": [],
"anime": [],
"other": [],
}
# Save identities
with open(identities_file, "w") as f:
json.dump(data, f, indent=2)
# Load identities
loaded = load_identities_json(identities_file)
# Convert to analysis input
movie_identities, series_identities, video_files = identities_to_analysis_input(loaded)
# Create identity-file pairs
identity_file_pairs = list(zip(movie_identities, video_files))
# Detect duplicates
duplicates = detect_duplicates(identity_file_pairs)
# Verify we found 1 duplicate group
assert len(duplicates) == 1
duplicate_group = duplicates[0]
# Verify duplicate group has 2 files
assert len(duplicate_group.files) == 2
# Verify quality comparison has actual metadata (not zeros/nulls)
quality_comparison = duplicate_group.quality_comparison
assert len(quality_comparison) == 2
# First file (1080p) should have correct metadata
file1 = quality_comparison[0]
assert file1["size_bytes"] == 2000000000
assert file1["resolution"] == "1920x1080"
assert file1["codec"] == "h264"
assert file1["bitrate_kbps"] == 5000
# Second file (720p) should have correct metadata
file2 = quality_comparison[1]
assert file2["size_bytes"] == 1000000000
assert file2["resolution"] == "1280x720"
assert file2["codec"] == "h264"
assert file2["bitrate_kbps"] == 2500
def test_duplicate_quality_comparison_with_v1_no_metadata(self, tmp_path):
"""Test that v1 identities without metadata produce limited quality comparison."""
# Create v1 identities.json without embedded metadata (backward compatibility)
identities_file = tmp_path / "identities.json"
data = {
# No vlm_schema_version (v1)
"metadata": {
"generated": "2024-02-13T12:00:00",
"source_inventory": "inventory.csv",
"total_files": 2,
},
"movies": [
{
"path": "/library/movie/Inception (2010) 1080p.mkv",
"filename": "Inception (2010) 1080p.mkv",
"category": "movie",
"title": "Inception",
"year": 2010,
"confidence": 0.9,
"needs_review": False,
# No video_metadata
},
{
"path": "/library/movie/Inception (2010) 720p.mkv",
"filename": "Inception (2010) 720p.mkv",
"category": "movie",
"title": "Inception",
"year": 2010,
"confidence": 0.9,
"needs_review": False,
# No video_metadata
},
],
"series": [],
"anime": [],
"other": [],
}
# Save identities
with open(identities_file, "w") as f:
json.dump(data, f, indent=2)
# Load identities
loaded = load_identities_json(identities_file)
# Convert to analysis input
movie_identities, series_identities, video_files = identities_to_analysis_input(loaded)
# Create identity-file pairs
identity_file_pairs = list(zip(movie_identities, video_files))
# Detect duplicates
duplicates = detect_duplicates(identity_file_pairs)
# Verify we found 1 duplicate group
assert len(duplicates) == 1
duplicate_group = duplicates[0]
# Verify quality comparison exists but has limited data (v1 defaults)
quality_comparison = duplicate_group.quality_comparison
assert len(quality_comparison) == 2
# v1 files have size_bytes=0 and no resolution/codec
for quality_info in quality_comparison:
assert quality_info["size_bytes"] == 0
assert "resolution" not in quality_info # None values are not included
assert "codec" not in quality_info
def test_series_duplicate_quality_comparison_with_v2_metadata(self, tmp_path):
"""Test series duplicates with embedded metadata."""
identities_file = tmp_path / "identities.json"
data = {
"vlm_schema_version": "2.0",
"metadata": {
"generated": "2024-02-13T12:00:00",
"source_inventory": "inventory.csv",
"total_files": 2,
},
"movies": [],
"series": [
{
"path": "/library/series/Breaking Bad S01E01 1080p.mkv",
"filename": "Breaking Bad S01E01 1080p.mkv",
"category": "series",
"title": "Breaking Bad",
"season": 1,
"episodes": [1],
"confidence": 0.9,
"needs_review": False,
"video_metadata": {
"size_bytes": 800000000,
"modified_timestamp": "2024-01-01T10:00:00+00:00",
"resolution": "1920x1080",
"codec": "hevc",
"duration_seconds": 2700.0,
"bitrate_kbps": 4000,
},
},
{
"path": "/library/series/Breaking Bad S01E01 720p.mkv",
"filename": "Breaking Bad S01E01 720p.mkv",
"category": "series",
"title": "Breaking Bad",
"season": 1,
"episodes": [1],
"confidence": 0.9,
"needs_review": False,
"video_metadata": {
"size_bytes": 400000000,
"modified_timestamp": "2024-01-02T10:00:00+00:00",
"resolution": "1280x720",
"codec": "h264",
"duration_seconds": 2700.0,
"bitrate_kbps": 2000,
},
},
],
"anime": [],
"other": [],
}
# Save and load
with open(identities_file, "w") as f:
json.dump(data, f, indent=2)
loaded = load_identities_json(identities_file)
movie_identities, series_identities, video_files = identities_to_analysis_input(loaded)
# Create identity-file pairs for series
identity_file_pairs = list(zip(series_identities, video_files))
# Detect duplicates
duplicates = detect_duplicates(identity_file_pairs)
# Verify duplicate detected
assert len(duplicates) == 1
duplicate_group = duplicates[0]
assert len(duplicate_group.files) == 2
# Verify quality comparison has metadata
quality_comparison = duplicate_group.quality_comparison
assert quality_comparison[0]["resolution"] == "1920x1080"
assert quality_comparison[0]["codec"] == "hevc"
assert quality_comparison[1]["resolution"] == "1280x720"
assert quality_comparison[1]["codec"] == "h264"
tests/test_io.py
+262
View File
@@ -0,0 +1,262 @@
"""Tests for the I/O layer module."""
import json
from datetime import datetime, timezone
from pathlib import Path
import pytest
from vlm.io import (
_video_file_from_record,
load_identities_json,
save_identities_json,
)
from vlm.models import VideoFile
class TestVideoFileFromRecord:
"""Tests for _video_file_from_record function."""
def test_video_file_from_record_with_metadata_v2(self):
"""Test loading VideoFile from v2 record with embedded metadata."""
record = {
"path": "/library/movie/The Matrix (1999).mkv",
"filename": "The Matrix (1999).mkv",
"category": "movie",
"title": "The Matrix",
"year": 1999,
"confidence": 0.9,
"needs_review": False,
"video_metadata": {
"size_bytes": 1234567890,
"modified_timestamp": "2024-01-01T10:00:00+00:00",
"resolution": "1920x1080",
"codec": "h264",
"duration_seconds": 7200.5,
"bitrate_kbps": 5000,
},
}
vf = _video_file_from_record(record)
assert vf.path == Path("/library/movie/The Matrix (1999).mkv")
assert vf.filename == "The Matrix (1999).mkv"
assert vf.category == "movie"
assert vf.size_bytes == 1234567890
assert vf.resolution == "1920x1080"
assert vf.codec == "h264"
assert vf.duration_seconds == 7200.5
assert vf.bitrate_kbps == 5000
assert isinstance(vf.modified_timestamp, datetime)
def test_video_file_from_record_without_metadata_v1(self):
"""Test loading VideoFile from v1 record without metadata (backward compatibility)."""
record = {
"path": "/library/movie/Inception (2010).mkv",
"filename": "Inception (2010).mkv",
"category": "movie",
"title": "Inception",
"year": 2010,
"confidence": 0.9,
"needs_review": False,
}
vf = _video_file_from_record(record)
assert vf.path == Path("/library/movie/Inception (2010).mkv")
assert vf.filename == "Inception (2010).mkv"
assert vf.category == "movie"
# v1 defaults
assert vf.size_bytes == 0
assert vf.resolution is None
assert vf.codec is None
assert vf.duration_seconds is None
assert vf.bitrate_kbps is None
assert isinstance(vf.modified_timestamp, datetime)
def test_video_file_from_record_partial_metadata(self):
"""Test loading VideoFile with partial metadata (some fields missing)."""
record = {
"path": "/library/series/Breaking Bad S01E01.mkv",
"filename": "Breaking Bad S01E01.mkv",
"category": "series",
"title": "Breaking Bad",
"season": 1,
"episodes": [1],
"confidence": 0.9,
"needs_review": False,
"video_metadata": {
"size_bytes": 987654321,
"modified_timestamp": "2024-02-01T15:30:00+00:00",
"resolution": "1280x720",
# codec missing
# duration_seconds missing
# bitrate_kbps missing
},
}
vf = _video_file_from_record(record)
assert vf.size_bytes == 987654321
assert vf.resolution == "1280x720"
assert vf.codec is None
assert vf.duration_seconds is None
assert vf.bitrate_kbps is None
class TestIdentitiesJsonVersioning:
"""Tests for identities.json schema versioning."""
def test_load_identities_v2_with_metadata(self, tmp_path):
"""Test loading v2 identities.json with video metadata."""
identities_file = tmp_path / "identities.json"
data = {
"vlm_schema_version": "2.0",
"metadata": {
"generated": "2024-01-01T10:00:00",
"source_inventory": "inventory.csv",
"total_files": 1,
},
"movies": [
{
"path": "/library/movie/Test (2024).mkv",
"filename": "Test (2024).mkv",
"category": "movie",
"title": "Test",
"year": 2024,
"confidence": 0.9,
"needs_review": False,
"video_metadata": {
"size_bytes": 1000000,
"modified_timestamp": "2024-01-01T00:00:00+00:00",
"resolution": "1920x1080",
"codec": "h264",
"duration_seconds": 3600.0,
"bitrate_kbps": 2500,
},
}
],
"series": [],
"anime": [],
"other": [],
}
save_identities_json(data, identities_file)
loaded = load_identities_json(identities_file)
assert loaded["vlm_schema_version"] == "2.0"
assert len(loaded["movies"]) == 1
assert "video_metadata" in loaded["movies"][0]
assert loaded["movies"][0]["video_metadata"]["size_bytes"] == 1000000
def test_load_identities_v1_backward_compatibility(self, tmp_path):
"""Test loading v1 identities.json without version field (backward compatibility)."""
identities_file = tmp_path / "identities.json"
data = {
# No vlm_schema_version field (v1)
"metadata": {
"generated": "2024-01-01T10:00:00",
"source_inventory": "inventory.csv",
"total_files": 1,
},
"movies": [
{
"path": "/library/movie/Old (2023).mkv",
"filename": "Old (2023).mkv",
"category": "movie",
"title": "Old",
"year": 2023,
"confidence": 0.9,
"needs_review": False,
# No video_metadata field
}
],
"series": [],
"anime": [],
"other": [],
}
save_identities_json(data, identities_file)
loaded = load_identities_json(identities_file)
# v1 files don't have version field
assert "vlm_schema_version" not in loaded or loaded.get("vlm_schema_version") is None
assert len(loaded["movies"]) == 1
assert "video_metadata" not in loaded["movies"][0]
class TestIdentitiesJsonRoundTrip:
"""Tests for identities.json save/load round-trip."""
def test_save_and_load_v2_identities(self, tmp_path):
"""Test saving and loading v2 identities with metadata."""
identities_file = tmp_path / "identities_v2.json"
original_data = {
"vlm_schema_version": "2.0",
"metadata": {
"generated": "2024-02-13T12:00:00",
"source_inventory": "test_inventory.csv",
"total_files": 2,
},
"movies": [
{
"path": "/library/movie/Movie1 (2024).mkv",
"filename": "Movie1 (2024).mkv",
"category": "movie",
"title": "Movie1",
"year": 2024,
"confidence": 0.9,
"needs_review": False,
"video_metadata": {
"size_bytes": 2000000000,
"modified_timestamp": "2024-02-10T08:00:00+00:00",
"resolution": "3840x2160",
"codec": "hevc",
"duration_seconds": 7200.0,
"bitrate_kbps": 8000,
},
}
],
"series": [
{
"path": "/library/series/Show S01E01.mkv",
"filename": "Show S01E01.mkv",
"category": "series",
"title": "Show",
"season": 1,
"episodes": [1],
"confidence": 0.9,
"needs_review": False,
"video_metadata": {
"size_bytes": 500000000,
"modified_timestamp": "2024-02-11T10:00:00+00:00",
"resolution": "1920x1080",
"codec": "h264",
"duration_seconds": 2700.0,
"bitrate_kbps": 3000,
},
}
],
"anime": [],
"other": [],
}
save_identities_json(original_data, identities_file)
loaded_data = load_identities_json(identities_file)
# Verify structure
assert loaded_data["vlm_schema_version"] == "2.0"
assert len(loaded_data["movies"]) == 1
assert len(loaded_data["series"]) == 1
# Verify movie metadata preserved
movie = loaded_data["movies"][0]
assert movie["video_metadata"]["size_bytes"] == 2000000000
assert movie["video_metadata"]["resolution"] == "3840x2160"
assert movie["video_metadata"]["codec"] == "hevc"
# Verify series metadata preserved
series = loaded_data["series"][0]
assert series["video_metadata"]["size_bytes"] == 500000000
assert series["video_metadata"]["resolution"] == "1920x1080"
assert series["video_metadata"]["codec"] == "h264"