568 lines
18 KiB
Python
568 lines
18 KiB
Python
"""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 _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 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
|
|
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)
|
|
|
|
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
|
|
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)
|
|
|
|
return json.dumps(inventory_data, indent=2, ensure_ascii=False)
|
|
|
|
|
|
def generate_completeness_report(
|
|
analysis: list[SeasonCompleteness],
|
|
format: str,
|
|
library_root: Path,
|
|
plan_summary: str | None = None,
|
|
) -> 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)
|
|
plan_summary: Optional plan content summary to prepend (when --plan was used)
|
|
|
|
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":
|
|
body = _generate_completeness_json(analysis, generation_timestamp, library_root)
|
|
else:
|
|
body = _generate_completeness_text(analysis, generation_timestamp, library_root)
|
|
|
|
if plan_summary:
|
|
if format == "text":
|
|
section = "计划内容总结\n" + "-" * 40 + "\n" + plan_summary + "\n\n"
|
|
return section + body
|
|
else:
|
|
data = json.loads(body)
|
|
data["plan_summary"] = plan_summary
|
|
return json.dumps(data, indent=2, ensure_ascii=False)
|
|
return body
|
|
|
|
|
|
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,
|
|
plan_summary: str | None = None,
|
|
) -> 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)
|
|
plan_summary: Optional plan content summary to prepend (when --plan was used)
|
|
|
|
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":
|
|
body = _generate_duplicate_json(duplicates, generation_timestamp, library_root)
|
|
else:
|
|
body = _generate_duplicate_text(duplicates, generation_timestamp, library_root)
|
|
|
|
if plan_summary:
|
|
if format == "text":
|
|
section = "计划内容总结\n" + "-" * 40 + "\n" + plan_summary + "\n\n"
|
|
return section + body
|
|
else:
|
|
data = json.loads(body)
|
|
data["plan_summary"] = plan_summary
|
|
return json.dumps(data, indent=2, ensure_ascii=False)
|
|
return body
|
|
|
|
|
|
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)
|
|
|
|
def _group_max_size(group: DuplicateGroup) -> int:
|
|
quality_sizes = [
|
|
int(item.get("size_bytes", 0) or 0)
|
|
for item in group.quality_comparison
|
|
if isinstance(item, dict)
|
|
]
|
|
if quality_sizes:
|
|
return max(quality_sizes)
|
|
if group.files:
|
|
return max(f.size_bytes for f in group.files)
|
|
return 0
|
|
|
|
# Sort by largest file size first
|
|
sorted_duplicates = sorted(
|
|
duplicates,
|
|
key=_group_max_size,
|
|
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)
|