Initial commit: Video Library Manager
- Add core VLM modules (scanner, parser, planner, executor, analysis) - Add CLI with quarantine, reports, rollback, and state management - Add comprehensive test suite - Add project configuration and documentation - Add .gitignore for Python project
This commit is contained in:
windyboy
@@ -0,0 +1,532 @@
|
||||
"""Report generation for Video Library Manager.
|
||||
|
||||
This module provides functionality to generate various reports about the video library:
|
||||
- Inventory reports (all discovered files with metadata)
|
||||
- Completeness reports (series with episode gaps)
|
||||
- Duplicate reports (duplicate files with quality comparisons)
|
||||
- Summary reports (library statistics)
|
||||
"""
|
||||
|
||||
import csv
|
||||
import json
|
||||
import logging
|
||||
from datetime import datetime, timezone
|
||||
from io import StringIO
|
||||
from pathlib import Path
|
||||
|
||||
from vlm.models import SeasonCompleteness, DuplicateGroup, VideoFile, MovieIdentity, SeriesIdentity
|
||||
|
||||
logger = logging.getLogger(__name__)
|
||||
|
||||
|
||||
def generate_inventory_report(
|
||||
files: list[VideoFile],
|
||||
format: str,
|
||||
library_root: Path
|
||||
) -> str:
|
||||
"""Generate inventory report listing all discovered video files with metadata.
|
||||
|
||||
Args:
|
||||
files: List of VideoFile objects to include in the report
|
||||
format: Output format ("csv" or "json")
|
||||
library_root: Root of the library (included in report metadata)
|
||||
|
||||
Returns:
|
||||
Formatted report as string
|
||||
|
||||
Raises:
|
||||
ValueError: If format is not "csv" or "json"
|
||||
"""
|
||||
if format not in ["csv", "json"]:
|
||||
raise ValueError(f"Invalid format: {format}. Must be 'csv' or 'json'")
|
||||
|
||||
generation_timestamp = datetime.now(timezone.utc).strftime("%Y-%m-%dT%H:%M:%S")
|
||||
|
||||
if format == "json":
|
||||
return _generate_inventory_json(files, generation_timestamp, library_root)
|
||||
else: # csv
|
||||
return _generate_inventory_csv(files, generation_timestamp, library_root)
|
||||
|
||||
|
||||
def _generate_inventory_csv(
|
||||
files: list[VideoFile],
|
||||
timestamp: str,
|
||||
library_root: Path
|
||||
) -> str:
|
||||
"""Generate inventory report in CSV format.
|
||||
|
||||
CSV Schema:
|
||||
- path, filename, size_bytes, modified_timestamp, category, resolution,
|
||||
codec, duration_seconds, bitrate_kbps
|
||||
- Timestamps in ISO 8601 format (YYYY-MM-DDTHH:MM:SS) in UTC
|
||||
- Missing optional values represented as empty strings
|
||||
- Header row always present
|
||||
"""
|
||||
output = StringIO()
|
||||
|
||||
# Write metadata as comments
|
||||
output.write(f"# Generated: {timestamp}\n")
|
||||
output.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(output, fieldnames=fieldnames)
|
||||
writer.writeheader()
|
||||
|
||||
# Write each file
|
||||
for video_file in files:
|
||||
# Format timestamp as ISO 8601 in UTC
|
||||
if video_file.modified_timestamp.tzinfo is None:
|
||||
# Assume local time, convert to UTC
|
||||
modified_utc = video_file.modified_timestamp.replace(tzinfo=timezone.utc)
|
||||
else:
|
||||
modified_utc = video_file.modified_timestamp.astimezone(timezone.utc)
|
||||
|
||||
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)
|
||||
|
||||
return output.getvalue()
|
||||
|
||||
|
||||
def _generate_inventory_json(
|
||||
files: list[VideoFile],
|
||||
timestamp: str,
|
||||
library_root: Path
|
||||
) -> str:
|
||||
"""Generate inventory report in JSON format."""
|
||||
inventory_data = {
|
||||
'metadata': {
|
||||
'generated': 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
|
||||
if video_file.modified_timestamp.tzinfo is None:
|
||||
# Assume local time, convert to UTC
|
||||
modified_utc = video_file.modified_timestamp.replace(tzinfo=timezone.utc)
|
||||
else:
|
||||
modified_utc = video_file.modified_timestamp.astimezone(timezone.utc)
|
||||
|
||||
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)
|
||||
|
||||
return json.dumps(inventory_data, indent=2, ensure_ascii=False)
|
||||
|
||||
|
||||
def generate_completeness_report(
|
||||
analysis: list[SeasonCompleteness],
|
||||
format: str,
|
||||
library_root: Path
|
||||
) -> str:
|
||||
"""Generate completeness report showing series with episode gaps.
|
||||
|
||||
Args:
|
||||
analysis: List of SeasonCompleteness objects with detected gaps
|
||||
format: Output format ("text" or "json")
|
||||
library_root: Root of the library (included in report metadata)
|
||||
|
||||
Returns:
|
||||
Formatted report as string
|
||||
|
||||
Raises:
|
||||
ValueError: If format is not "text" or "json"
|
||||
"""
|
||||
if format not in ["text", "json"]:
|
||||
raise ValueError(f"Invalid format: {format}. Must be 'text' or 'json'")
|
||||
|
||||
generation_timestamp = datetime.now(timezone.utc).strftime("%Y-%m-%dT%H:%M:%S")
|
||||
|
||||
if format == "json":
|
||||
return _generate_completeness_json(analysis, generation_timestamp, library_root)
|
||||
else: # text
|
||||
return _generate_completeness_text(analysis, generation_timestamp, library_root)
|
||||
|
||||
|
||||
def _generate_completeness_text(
|
||||
analysis: list[SeasonCompleteness],
|
||||
timestamp: str,
|
||||
library_root: Path
|
||||
) -> str:
|
||||
"""Generate completeness report in text format."""
|
||||
lines = []
|
||||
lines.append("=" * 80)
|
||||
lines.append("SERIES COMPLETENESS REPORT")
|
||||
lines.append("=" * 80)
|
||||
lines.append(f"Generated: {timestamp}")
|
||||
lines.append(f"Library Root: {library_root}")
|
||||
lines.append(f"Series with gaps: {len(analysis)}")
|
||||
lines.append("")
|
||||
|
||||
if not analysis:
|
||||
lines.append("No series with episode gaps detected.")
|
||||
return "\n".join(lines)
|
||||
|
||||
# Group by series title
|
||||
series_groups = {}
|
||||
for season_data in analysis:
|
||||
if season_data.series_title not in series_groups:
|
||||
series_groups[season_data.series_title] = []
|
||||
series_groups[season_data.series_title].append(season_data)
|
||||
|
||||
# Sort series alphabetically
|
||||
for series_title in sorted(series_groups.keys()):
|
||||
lines.append("-" * 80)
|
||||
lines.append(f"Series: {series_title}")
|
||||
lines.append("-" * 80)
|
||||
|
||||
# Sort seasons by season number
|
||||
seasons = sorted(series_groups[series_title], key=lambda x: x.season)
|
||||
|
||||
for season_data in seasons:
|
||||
lines.append(f" Season {season_data.season:02d}:")
|
||||
lines.append(f" Episodes found: {_format_episode_list(season_data.episodes_found)}")
|
||||
lines.append(f" Episodes missing: {_format_episode_list(season_data.episodes_missing)}")
|
||||
lines.append("")
|
||||
|
||||
return "\n".join(lines)
|
||||
|
||||
|
||||
def _generate_completeness_json(
|
||||
analysis: list[SeasonCompleteness],
|
||||
timestamp: str,
|
||||
library_root: Path
|
||||
) -> str:
|
||||
"""Generate completeness report in JSON format."""
|
||||
report_data = {
|
||||
"metadata": {
|
||||
"generated": timestamp,
|
||||
"library_root": str(library_root),
|
||||
"series_count": len(set(s.series_title for s in analysis))
|
||||
},
|
||||
"series": []
|
||||
}
|
||||
|
||||
# Group by series title
|
||||
series_groups = {}
|
||||
for season_data in analysis:
|
||||
if season_data.series_title not in series_groups:
|
||||
series_groups[season_data.series_title] = []
|
||||
series_groups[season_data.series_title].append(season_data)
|
||||
|
||||
# Build series data
|
||||
for series_title in sorted(series_groups.keys()):
|
||||
seasons_data = []
|
||||
for season_data in sorted(series_groups[series_title], key=lambda x: x.season):
|
||||
seasons_data.append({
|
||||
"season": season_data.season,
|
||||
"episodes_found": season_data.episodes_found,
|
||||
"episodes_missing": season_data.episodes_missing
|
||||
})
|
||||
|
||||
report_data["series"].append({
|
||||
"title": series_title,
|
||||
"seasons": seasons_data
|
||||
})
|
||||
|
||||
return json.dumps(report_data, indent=2, ensure_ascii=False)
|
||||
|
||||
|
||||
def generate_duplicate_report(
|
||||
duplicates: list[DuplicateGroup],
|
||||
format: str,
|
||||
library_root: Path
|
||||
) -> str:
|
||||
"""Generate duplicate report showing duplicate files with quality comparisons.
|
||||
|
||||
Args:
|
||||
duplicates: List of DuplicateGroup objects with duplicate files
|
||||
format: Output format ("text" or "json")
|
||||
library_root: Root of the library (included in report metadata)
|
||||
|
||||
Returns:
|
||||
Formatted report as string
|
||||
|
||||
Raises:
|
||||
ValueError: If format is not "text" or "json"
|
||||
"""
|
||||
if format not in ["text", "json"]:
|
||||
raise ValueError(f"Invalid format: {format}. Must be 'text' or 'json'")
|
||||
|
||||
generation_timestamp = datetime.now(timezone.utc).strftime("%Y-%m-%dT%H:%M:%S")
|
||||
|
||||
if format == "json":
|
||||
return _generate_duplicate_json(duplicates, generation_timestamp, library_root)
|
||||
else: # text
|
||||
return _generate_duplicate_text(duplicates, generation_timestamp, library_root)
|
||||
|
||||
|
||||
def _generate_duplicate_text(
|
||||
duplicates: list[DuplicateGroup],
|
||||
timestamp: str,
|
||||
library_root: Path
|
||||
) -> str:
|
||||
"""Generate duplicate report in text format."""
|
||||
lines = []
|
||||
lines.append("=" * 80)
|
||||
lines.append("DUPLICATE FILES REPORT")
|
||||
lines.append("=" * 80)
|
||||
lines.append(f"Generated: {timestamp}")
|
||||
lines.append(f"Library Root: {library_root}")
|
||||
lines.append(f"Duplicate groups: {len(duplicates)}")
|
||||
lines.append("")
|
||||
|
||||
if not duplicates:
|
||||
lines.append("No duplicate files detected.")
|
||||
return "\n".join(lines)
|
||||
|
||||
# Sort by largest file size first
|
||||
sorted_duplicates = sorted(
|
||||
duplicates,
|
||||
key=lambda g: max(f.size_bytes for f in g.files),
|
||||
reverse=True
|
||||
)
|
||||
|
||||
for idx, group in enumerate(sorted_duplicates, 1):
|
||||
lines.append("-" * 80)
|
||||
|
||||
# Format identity
|
||||
identity = group.identity
|
||||
if isinstance(identity, MovieIdentity):
|
||||
lines.append(f"Group {idx}: {identity.title} ({identity.year})")
|
||||
else: # SeriesIdentity
|
||||
episodes_str = ", ".join(str(e) for e in identity.episodes)
|
||||
lines.append(f"Group {idx}: {identity.title} - S{identity.season:02d}E{episodes_str}")
|
||||
|
||||
lines.append("-" * 80)
|
||||
lines.append(f" Files: {len(group.files)}")
|
||||
lines.append("")
|
||||
|
||||
# Show quality comparison for each file
|
||||
for file_idx, quality_data in enumerate(group.quality_comparison, 1):
|
||||
lines.append(f" File {file_idx}:")
|
||||
lines.append(f" Filename: {quality_data['filename']}")
|
||||
lines.append(f" Path: {quality_data['path']}")
|
||||
lines.append(f" Size: {_format_size(quality_data['size_bytes'])}")
|
||||
|
||||
if 'resolution' in quality_data:
|
||||
lines.append(f" Resolution: {quality_data['resolution']}")
|
||||
|
||||
if 'codec' in quality_data:
|
||||
lines.append(f" Codec: {quality_data['codec']}")
|
||||
|
||||
if 'duration_seconds' in quality_data:
|
||||
lines.append(f" Duration: {_format_duration(quality_data['duration_seconds'])}")
|
||||
|
||||
if 'bitrate_kbps' in quality_data:
|
||||
lines.append(f" Bitrate: {quality_data['bitrate_kbps']} kbps")
|
||||
|
||||
lines.append("")
|
||||
|
||||
return "\n".join(lines)
|
||||
|
||||
|
||||
def _generate_duplicate_json(
|
||||
duplicates: list[DuplicateGroup],
|
||||
timestamp: str,
|
||||
library_root: Path
|
||||
) -> str:
|
||||
"""Generate duplicate report in JSON format."""
|
||||
report_data = {
|
||||
"metadata": {
|
||||
"generated": timestamp,
|
||||
"library_root": str(library_root),
|
||||
"duplicate_groups": len(duplicates)
|
||||
},
|
||||
"duplicates": []
|
||||
}
|
||||
|
||||
for group in duplicates:
|
||||
identity = group.identity
|
||||
|
||||
# Format identity
|
||||
if isinstance(identity, MovieIdentity):
|
||||
identity_data = {
|
||||
"type": "movie",
|
||||
"title": identity.title,
|
||||
"year": identity.year
|
||||
}
|
||||
else: # SeriesIdentity
|
||||
identity_data = {
|
||||
"type": "series",
|
||||
"title": identity.title,
|
||||
"season": identity.season,
|
||||
"episodes": identity.episodes
|
||||
}
|
||||
|
||||
group_data = {
|
||||
"identity": identity_data,
|
||||
"file_count": len(group.files),
|
||||
"files": group.quality_comparison
|
||||
}
|
||||
|
||||
report_data["duplicates"].append(group_data)
|
||||
|
||||
return json.dumps(report_data, indent=2, ensure_ascii=False)
|
||||
|
||||
|
||||
def generate_summary_report(
|
||||
files: list[VideoFile],
|
||||
library_root: Path
|
||||
) -> str:
|
||||
"""Generate summary report with library statistics.
|
||||
|
||||
Args:
|
||||
files: List of all VideoFile objects in the library
|
||||
library_root: Root of the library (included in report metadata)
|
||||
|
||||
Returns:
|
||||
Formatted summary report as text string
|
||||
"""
|
||||
generation_timestamp = datetime.now(timezone.utc).strftime("%Y-%m-%dT%H:%M:%S")
|
||||
|
||||
lines = []
|
||||
lines.append("=" * 80)
|
||||
lines.append("LIBRARY SUMMARY REPORT")
|
||||
lines.append("=" * 80)
|
||||
lines.append(f"Generated: {generation_timestamp}")
|
||||
lines.append(f"Library Root: {library_root}")
|
||||
lines.append("")
|
||||
|
||||
# Calculate total statistics
|
||||
total_files = len(files)
|
||||
total_size = sum(f.size_bytes for f in files)
|
||||
|
||||
lines.append(f"Total Files: {total_files}")
|
||||
lines.append(f"Total Size: {_format_size(total_size)}")
|
||||
lines.append("")
|
||||
|
||||
# Category breakdown
|
||||
lines.append("Category Breakdown:")
|
||||
lines.append("-" * 40)
|
||||
|
||||
category_stats = {}
|
||||
for file in files:
|
||||
category = file.category
|
||||
if category not in category_stats:
|
||||
category_stats[category] = {"count": 0, "size": 0}
|
||||
category_stats[category]["count"] += 1
|
||||
category_stats[category]["size"] += file.size_bytes
|
||||
|
||||
# Sort categories alphabetically
|
||||
for category in sorted(category_stats.keys()):
|
||||
stats = category_stats[category]
|
||||
lines.append(f" {category.capitalize()}:")
|
||||
lines.append(f" Files: {stats['count']}")
|
||||
lines.append(f" Size: {_format_size(stats['size'])}")
|
||||
|
||||
lines.append("")
|
||||
|
||||
return "\n".join(lines)
|
||||
|
||||
|
||||
def _format_episode_list(episodes: list[int]) -> str:
|
||||
"""Format episode list as compact string with ranges.
|
||||
|
||||
Examples:
|
||||
[1, 2, 3, 5, 6, 8] -> "1-3, 5-6, 8"
|
||||
[1, 3, 5] -> "1, 3, 5"
|
||||
"""
|
||||
if not episodes:
|
||||
return "none"
|
||||
|
||||
# Sort episodes
|
||||
sorted_episodes = sorted(episodes)
|
||||
|
||||
# Build ranges
|
||||
ranges = []
|
||||
start = sorted_episodes[0]
|
||||
end = sorted_episodes[0]
|
||||
|
||||
for episode in sorted_episodes[1:]:
|
||||
if episode == end + 1:
|
||||
# Continue current range
|
||||
end = episode
|
||||
else:
|
||||
# End current range and start new one
|
||||
if start == end:
|
||||
ranges.append(str(start))
|
||||
else:
|
||||
ranges.append(f"{start}-{end}")
|
||||
start = episode
|
||||
end = episode
|
||||
|
||||
# Add final range
|
||||
if start == end:
|
||||
ranges.append(str(start))
|
||||
else:
|
||||
ranges.append(f"{start}-{end}")
|
||||
|
||||
return ", ".join(ranges)
|
||||
|
||||
|
||||
def _format_size(size_bytes: int) -> str:
|
||||
"""Format file size in human-readable format.
|
||||
|
||||
Examples:
|
||||
1024 -> "1.00 KB"
|
||||
1048576 -> "1.00 MB"
|
||||
1073741824 -> "1.00 GB"
|
||||
"""
|
||||
for unit in ['B', 'KB', 'MB', 'GB', 'TB']:
|
||||
if size_bytes < 1024.0:
|
||||
return f"{size_bytes:.2f} {unit}"
|
||||
size_bytes /= 1024.0
|
||||
return f"{size_bytes:.2f} PB"
|
||||
|
||||
|
||||
def _format_duration(duration_seconds: float) -> str:
|
||||
"""Format duration in human-readable format.
|
||||
|
||||
Examples:
|
||||
90 -> "1m 30s"
|
||||
3665 -> "1h 1m 5s"
|
||||
"""
|
||||
hours = int(duration_seconds // 3600)
|
||||
minutes = int((duration_seconds % 3600) // 60)
|
||||
seconds = int(duration_seconds % 60)
|
||||
|
||||
parts = []
|
||||
if hours > 0:
|
||||
parts.append(f"{hours}h")
|
||||
if minutes > 0:
|
||||
parts.append(f"{minutes}m")
|
||||
if seconds > 0 or not parts:
|
||||
parts.append(f"{seconds}s")
|
||||
|
||||
return " ".join(parts)
|
||||
Reference in New Issue
Block a user