"""Command-line interface for Video Library Manager. This module provides the main CLI entry point using Click framework. It implements global options (--config, --log-level) and error handling. """ import json import sys import traceback from pathlib import Path from typing import Optional import click import yaml from click.core import ParameterSource from vlm.config import Config, load_config, create_default_config, validate_config from vlm.context import CLIContext, pass_context from vlm.logging_config import setup_logging, get_logger from vlm.plan_render import fallback_plan_summary, preferred_plan_summary, render_review_preview from vlm.utils import format_size def default_config_path() -> Path: """Return the default config path resolved at runtime.""" return Path.home() / ".vlm" / "config.yaml" def _workspace_dir_from_context() -> Path: """Resolve workspace directory from CLI context at runtime.""" click_ctx = click.get_current_context(silent=True) if click_ctx and isinstance(click_ctx.obj, CLIContext): return click_ctx.obj.config.workspace_dir return Path("artifacts") def default_artifact_path(filename: str) -> Path: """Build default artifact path for a filename at runtime.""" return _workspace_dir_from_context() / filename def _is_default_parameter(parameter_name: str) -> bool: """Check whether a parameter value came from Click default.""" click_ctx = click.get_current_context(silent=True) if click_ctx is None: return False return click_ctx.get_parameter_source(parameter_name) == ParameterSource.DEFAULT def resolve_legacy_default_input_path( current_path: Path, parameter_name: str, legacy_filename: str, option_name: str, ) -> Path: """Fallback to legacy root path when default workspace file is missing. This keeps stage-A compatibility for users who still have legacy artifacts in repository root while printing a migration warning. """ if not _is_default_parameter(parameter_name): return current_path workspace_default = default_artifact_path(legacy_filename) legacy_default = Path(legacy_filename) if current_path != workspace_default: return current_path if current_path.exists() or not legacy_default.exists(): return current_path click.echo( "Warning: detected legacy default input at " f"{legacy_default.resolve()}. Please migrate to " f"{workspace_default.resolve()} (example: {option_name} {workspace_default.resolve()}).", err=True, ) return legacy_default def _command_error(ctx: CLIContext, user_message: str, logger_message: str, *, exc_info: bool = False) -> None: """Print a command error message, log it, and exit consistently.""" click.echo(user_message, err=True) ctx.logger.error(logger_message, exc_info=exc_info) raise SystemExit(1) def _load_or_create_config(config: Path) -> Config: """Load configuration from disk or create a default config file.""" if config.exists(): try: cfg = load_config(config) except yaml.YAMLError as e: click.echo(f"Error: Invalid YAML syntax in configuration file: {e}", err=True) click.echo("Using default configuration values.", err=True) cfg = create_default_config(config) except ValueError as e: click.echo(f"Error: {e}", err=True) click.echo("Using default configuration values.", err=True) cfg = create_default_config(config) else: click.echo(f"Configuration file not found at {config}", err=True) click.echo("Creating default configuration...", err=True) cfg = create_default_config(config) click.echo(f"Default configuration created at {config}", err=True) validation_errors = validate_config(cfg) if validation_errors: click.echo("Configuration validation errors:", err=True) for error in validation_errors: click.echo(f" - {error}", err=True) click.echo("Please fix the configuration file and try again.", err=True) raise ValueError("invalid configuration") return cfg def _initialize_cli_context(config: Path, log_level: Optional[str]) -> CLIContext: """Load config, apply CLI overrides, and build the CLI context.""" cfg = _load_or_create_config(config) if log_level: cfg.log_level = log_level.upper() logger = setup_logging(log_level=cfg.log_level) return CLIContext(config=cfg, logger=logger) @click.group() @click.option( '--config', type=click.Path(path_type=Path), default=default_config_path, help='Path to configuration file (default: ~/.vlm/config.yaml)' ) @click.option( '--log-level', type=click.Choice(['DEBUG', 'INFO', 'WARNING', 'ERROR'], case_sensitive=False), default=None, help='Set logging level (overrides config file)' ) @click.pass_context def main(ctx, config: Path, log_level: Optional[str]): """Video Library Manager - A tool for managing personal video collections. VLM helps you organize, analyze, and maintain your video library with a safety-first approach. All operations are reversible and require explicit confirmation before making changes. Common workflow: 1. vlm scan - Discover all video files 2. vlm parse - Extract titles, years, seasons, episodes 3. vlm enrich - Enrich parsed identities with external metadata 4. vlm analyze - Detect gaps and duplicates 5. vlm plan - Generate execution plan 6. vlm execute - Execute plan (dry-run by default) 7. vlm execute --confirm - Actually execute operations Use 'vlm COMMAND --help' for more information on a specific command. """ # Ensure context object exists ctx.ensure_object(dict) try: ctx.obj = _initialize_cli_context(config, log_level) except Exception as e: traceback.print_exc(file=sys.stderr) click.echo(f"Error initializing VLM: {e}", err=True) sys.exit(1) @main.command() @click.option( '--output', type=click.Path(path_type=Path), default=lambda: default_artifact_path('inventory.csv'), help='Output file for inventory (default: artifacts/inventory.csv)' ) @click.option( '--metadata/--no-metadata', default=True, help='Extract video metadata via ffprobe (default: enabled)' ) @click.option( '--reuse-from', type=click.Path(exists=True, path_type=Path), default=None, help='Reuse metadata cache from an existing inventory CSV (default: output file if it exists)' ) @click.option( '--force-refresh-metadata', is_flag=True, default=False, help='Ignore metadata cache and re-run ffprobe for all files' ) @pass_context def scan( ctx: CLIContext, output: Path, metadata: bool, reuse_from: Optional[Path], force_refresh_metadata: bool ): """Scan library and generate inventory. Discovers all video files in the library and records their metadata. This is a read-only operation that does not modify any files. Example: vlm scan # Save to inventory.csv vlm scan --output my_library.csv # Save to custom file vlm scan --no-metadata # Faster scan without ffprobe vlm scan --reuse-from old.csv # Reuse prior metadata cache vlm scan --force-refresh-metadata # Re-run ffprobe for all files """ try: from vlm.commands.scan import scan_cmd scan_cmd(ctx, output, metadata, reuse_from, force_refresh_metadata) except Exception as e: click.echo(f"Error during scan: {e}", err=True) ctx.logger.error(f"Scan failed: {e}", exc_info=True) sys.exit(1) @main.command() @click.option( '--input', type=click.Path(path_type=Path), default=lambda: default_artifact_path('inventory.csv'), help='Input inventory CSV file (default: artifacts/inventory.csv)' ) @click.option( '--output', type=click.Path(path_type=Path), default=lambda: default_artifact_path('identities.json'), help='Output file for parsed identities (default: artifacts/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, 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 (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 """ try: input = resolve_legacy_default_input_path(input, "input", "inventory.csv", "--input") from vlm.commands.parse import parse_cmd parse_cmd(ctx, input, output, inventory) except FileNotFoundError: _command_error( ctx, f"Error: Input file not found: {input}", f"Input file not found: {input}", ) except ValueError as e: _command_error(ctx, f"Error: {e}", f"Parse failed: {e}") except OSError as e: _command_error( ctx, f"Error reading/writing files: {e}", f"Parse file I/O failed: {e}", exc_info=True, ) @main.command() @click.option( '--input', type=click.Path(path_type=Path), default=lambda: default_artifact_path('identities.json'), help='Path to identities JSON file (default: artifacts/identities.json)' ) @click.option( '--output', type=click.Path(path_type=Path), default=None, help='Output path (default: overwrite input file)' ) @click.option( '--refresh-changed-only', is_flag=True, default=False, help='Refresh only records whose identity fingerprint changed (incremental behavior)' ) @click.option( '--refresh-all', is_flag=True, default=False, help='Bypass cache and re-enrich all records' ) @click.option( '--timeout', type=int, default=6, show_default=True, help='Per-request timeout in seconds' ) @click.option( '--retries', type=int, default=2, show_default=True, help='Retry attempts for provider/API requests' ) @pass_context def enrich( ctx: CLIContext, input: Path, output: Optional[Path], refresh_changed_only: bool, refresh_all: bool, timeout: int, retries: int, ): """Enrich identities with translation and reputation metadata. Applies incremental cache-backed enrichment to parsed identities and writes results back into identities JSON. """ try: input = resolve_legacy_default_input_path(input, "input", "identities.json", "--input") from vlm.commands.enrich import enrich_cmd enrich_cmd( ctx, input, output, refresh_changed_only, refresh_all, timeout, retries, ) except FileNotFoundError: _command_error( ctx, f"Error: Input file not found: {input}", f"Input file not found: {input}", ) except json.JSONDecodeError as e: _command_error( ctx, f"Error: Failed to parse JSON file: {e}", f"JSON parsing failed during enrich: {e}", exc_info=True, ) except ValueError as e: _command_error(ctx, f"Error: {e}", f"Enrich validation failed: {e}") except OSError as e: _command_error( ctx, f"Error reading/writing files: {e}", f"Enrich file I/O failed: {e}", exc_info=True, ) @main.command() @click.option( '--input', type=click.Path(path_type=Path), default=lambda: default_artifact_path('identities.json'), help='Path to parsed identities JSON file (default: artifacts/identities.json)' ) @click.option( '--output', type=click.Path(path_type=Path), default=lambda: default_artifact_path('analysis.json'), help='Path to save analysis results (default: artifacts/analysis.json)' ) @click.option( '--inventory', type=click.Path(exists=True, path_type=Path), default=None, help='Optional inventory CSV to merge size/resolution/codec for duplicate quality comparison' ) @pass_context def analyze(ctx: CLIContext, input: Path, output: Path, inventory: Optional[Path]): """Analyze completeness and duplicates. Detects episode gaps in series and identifies potential duplicate files. Provides quality comparison data for duplicates. Example: vlm analyze # Use default files vlm analyze --input my_identities.json # Custom input vlm analyze --inventory inventory.csv # Merge metadata for quality comparison vlm analyze --output my_analysis.json # Custom output """ try: input = resolve_legacy_default_input_path(input, "input", "identities.json", "--input") from vlm.commands.analyze import analyze_cmd analyze_cmd(ctx, input, output, inventory) except FileNotFoundError: _command_error( ctx, f"Error: Input file not found: {input}", f"Input file not found: {input}", ) except json.JSONDecodeError as e: _command_error( ctx, f"Error: Failed to parse JSON file: {e}", f"JSON parsing failed: {e}", exc_info=True, ) except Exception as e: _command_error( ctx, f"Error during analysis: {e}", f"Analysis failed: {e}", exc_info=True, ) @main.command() @click.option( '--input', type=click.Path(path_type=Path), default=lambda: default_artifact_path('identities.json'), help='Path to parsed identities JSON file (default: artifacts/identities.json)' ) @click.option( '--output', type=click.Path(path_type=Path), default=lambda: default_artifact_path('plan.json'), help='Path to save execution plan (default: artifacts/plan.json)' ) @click.option( '--analysis', type=click.Path(path_type=Path), default=None, help='Path to analysis JSON (optional); when provided, duplicate groups are applied to the plan' ) @pass_context def plan(ctx: CLIContext, input: Path, output: Path, analysis: Path | None): """Generate execution plan. Creates a structured, reviewable plan of all file operations to be performed. The plan can be edited before execution. Example: vlm plan # Use default files vlm plan --input my_identities.json # Custom input vlm plan --analysis analysis.json # Use analysis for duplicate handling vlm plan --output my_plan.json # Custom output """ try: input = resolve_legacy_default_input_path(input, "input", "identities.json", "--input") from vlm.commands.plan import plan_cmd plan_cmd(ctx, input, output, analysis) except FileNotFoundError: _command_error( ctx, f"Error: Input file not found: {input}", f"Input file not found: {input}", ) except json.JSONDecodeError as e: _command_error( ctx, f"Error: Failed to parse JSON file: {e}", f"JSON parsing failed: {e}", exc_info=True, ) except Exception as e: _command_error( ctx, f"Error during plan generation: {e}", f"Plan generation failed: {e}", exc_info=True, ) @main.command(name="review-plan") @click.option( '--input', type=click.Path(exists=True, path_type=Path), default=lambda: default_artifact_path('plan.json'), help='Path to execution plan JSON file (default: artifacts/plan.json)' ) @click.option( '--output', type=click.Path(path_type=Path), default=lambda: default_artifact_path('plan_manual_review.csv'), help='Path to save manual review CSV (default: artifacts/plan_manual_review.csv)' ) @click.option( '--season-threshold', type=int, default=20, show_default=True, help='Flag operations with season >= this value as high risk' ) @click.option( '--episode-threshold', type=int, default=40, show_default=True, help='Flag operations with episode >= this value as high risk' ) @click.option( '--preview-limit', type=int, default=10, show_default=True, help='Number of high-risk operations to preview in console output' ) @click.option( '--show-all', is_flag=True, default=False, help='Show all high-risk operations in console preview' ) @pass_context def review_plan_cmd( ctx: CLIContext, input: Path, output: Path, season_threshold: int, episode_threshold: int, preview_limit: int, show_all: bool, ): """Review a plan and export high-risk operations for manual confirmation.""" from vlm.planner import load_plan from vlm.plan_review import review_plan, save_review_csv logger = ctx.logger try: input = resolve_legacy_default_input_path(input, "input", "plan.json", "--input") if season_threshold < 1 or episode_threshold < 1: click.echo("Error: thresholds must be >= 1", err=True) sys.exit(1) if preview_limit < 1: click.echo("Error: --preview-limit must be >= 1", err=True) sys.exit(1) click.echo(f"Loading plan: {input}") execution_plan = load_plan(input) rows, counters = review_plan( execution_plan, season_threshold=season_threshold, episode_threshold=episode_threshold, ) save_review_csv(rows, output) click.echo() click.echo("Plan overview:") click.echo(preferred_plan_summary(execution_plan)) click.echo() click.echo("Plan review summary:") click.echo(f" Total operations: {counters['total_operations']}") click.echo(f" High-risk operations: {counters['high_risk_operations']}") click.echo(f" manual_review: {counters['manual_review']}") click.echo(f" sample_source: {counters['sample_source']}") click.echo(f" high_season: {counters['high_season']}") click.echo(f" high_episode: {counters['high_episode']}") click.echo(f" conflicts: {counters['conflicts']}") click.echo() click.echo("High-risk operations preview:") if rows: preview_lines, hidden_count = render_review_preview( rows, preview_limit=preview_limit, show_all=show_all, ) for line in preview_lines: click.echo(line) if hidden_count > 0: click.echo( f" ... and {hidden_count} more high-risk operations " "(use --show-all to display all)" ) else: click.echo(" (none)") click.echo() click.echo(f"Saved manual review CSV to: {output}") logger.info( "Plan review completed: total=%s high_risk=%s output=%s", counters["total_operations"], counters["high_risk_operations"], output, ) except FileNotFoundError: _command_error(ctx, f"Error: Plan file not found: {input}", f"Plan file not found: {input}") except json.JSONDecodeError as e: _command_error( ctx, f"Error: Failed to parse plan JSON: {e}", f"Plan review JSON parsing failed: {e}", exc_info=True, ) except Exception as e: _command_error( ctx, f"Error during plan review: {e}", f"Plan review failed: {e}", exc_info=True, ) @main.command(name="apply-review") @click.option( '--plan', type=click.Path(exists=True, path_type=Path), default=Path('plan.json'), help='Path to execution plan JSON file (default: plan.json)' ) @click.option( '--csv', type=click.Path(exists=True, path_type=Path), default=Path('plan_manual_review.csv'), help='Path to the modified manual review CSV (default: plan_manual_review.csv)' ) @click.option( '--output', type=click.Path(path_type=Path), default=None, help='Path to save updated plan (default: overwrite input plan)' ) @pass_context def apply_review_cmd( ctx: CLIContext, plan: Path, csv: Path, output: Optional[Path], ): """Apply modifications from a manual review CSV back to the plan JSON. This command reads the 'operation_type' column from the CSV and updates the corresponding operations in the plan. This is the primary way to manually approve or reject high-risk operations. """ from vlm.planner import load_plan, save_plan, apply_review_to_plan logger = ctx.logger output_path = output or plan try: click.echo(f"Loading plan: {plan}") execution_plan = load_plan(plan) click.echo(f"Applying review from: {csv}") updated_plan = apply_review_to_plan(execution_plan, csv) save_plan(updated_plan, output_path) click.echo(f"Successfully updated plan saved to: {output_path}") # Calculate changes for user feedback modified = 0 for i, op in enumerate(updated_plan.operations): if op.operation_type != execution_plan.operations[i].operation_type: modified += 1 click.echo(f"Total operations modified: {modified}") logger.info("Applied review from %s to %s, modified %s ops", csv, output_path, modified) except Exception as e: _command_error( ctx, f"Error applying review: {e}", f"Apply review failed: {e}", exc_info=True, ) @main.command() @click.option( '--plan', type=click.Path(path_type=Path), default=lambda: default_artifact_path('plan.json'), help='Path to execution plan JSON file (default: artifacts/plan.json)' ) @click.option( '--confirm', is_flag=True, default=False, help='Actually execute operations (default is dry-run)' ) @click.option( '--yes', is_flag=True, default=False, help='Skip confirmation prompt (auto-approve)' ) @click.option( '--verbose-ops', is_flag=True, default=False, help='Print per-operation dry-run logs at INFO level' ) @click.option( '--preserve-directories', is_flag=True, default=False, help='Preserve empty source directories instead of allowing them to be destroyed' ) @click.option( '--safe-mode', is_flag=True, default=False, help='Enable safe mode: prevent any operations that would destroy directories' ) @pass_context def execute(ctx: CLIContext, plan: Path, confirm: bool, yes: bool, verbose_ops: bool, preserve_directories: bool, safe_mode: bool): """Execute plan (defaults to dry-run, requires --confirm). Executes file operations from a plan. Defaults to dry-run mode which simulates operations without making changes. Use --confirm to actually execute operations. Example: vlm execute # Dry-run with plan.json vlm execute --plan my_plan.json # Dry-run with custom plan vlm execute --confirm # Actually execute operations (with prompt) vlm execute --confirm --yes # Execute without confirmation prompt """ try: plan = resolve_legacy_default_input_path(plan, "plan", "plan.json", "--plan") from vlm.commands.execute import execute_cmd execute_cmd(ctx, plan, confirm, yes, verbose_ops, preserve_directories, safe_mode) except FileNotFoundError: _command_error( ctx, f"Error: File not found: {plan}", f"Execution file not found: {plan}", ) except ValueError as e: _command_error(ctx, f"Error: {e}", f"Execution validation failed: {e}") except OSError as e: _command_error( ctx, f"Error during execution: {e}", f"Execution I/O failed: {e}", exc_info=True, ) @main.group() @pass_context def quarantine(ctx: CLIContext): """Manage quarantined files. Quarantine operations allow you to safely isolate unwanted files for review before deletion. Only movie and series files can be quarantined in v1. """ pass @quarantine.command('list') @click.option( '--category', type=click.Choice(['movie', 'series'], case_sensitive=False), default=None, help='Filter by category (movie or series)' ) @pass_context def quarantine_list(ctx: CLIContext, category: Optional[str]): """List quarantined files. Shows all files currently in quarantine with their original locations, quarantine timestamps, and reasons. Example: vlm quarantine list # List all quarantined files vlm quarantine list --category movie # List only movie files """ from vlm.quarantine import QuarantineManager config = ctx.config logger = ctx.logger try: # Create quarantine manager manager = QuarantineManager(config, logger) # List quarantined files entries = manager.list_quarantined(category=category) if not entries: if category: click.echo(f"No quarantined files found in category '{category}'.") else: click.echo("No quarantined files found.") return # Display quarantined files click.echo() if category: click.echo(f"Quarantined files in category '{category}':") else: click.echo("Quarantined files:") click.echo("=" * 80) for i, entry in enumerate(entries, 1): click.echo(f"\n[{i}] {entry.quarantine_path.name}") click.echo(f" Category: {entry.category}") click.echo(f" Original: {entry.original_path}") click.echo(f" Quarantine: {entry.quarantine_path}") click.echo(f" Size: {format_size(entry.size_bytes)}") click.echo(f" Quarantined: {entry.quarantined_at.strftime('%Y-%m-%d %H:%M:%S')}") if entry.reason: click.echo(f" Reason: {entry.reason}") click.echo() click.echo("=" * 80) click.echo(f"Total: {len(entries)} quarantined file(s)") click.echo() logger.info(f"Listed {len(entries)} quarantined files" + (f" from category '{category}'" if category else "")) except Exception as e: _command_error( ctx, f"Error listing quarantined files: {e}", f"Failed to list quarantined files: {e}", exc_info=True, ) @quarantine.command('add') @click.argument('file', type=click.Path(exists=True, path_type=Path)) @click.option( '--reason', type=str, default=None, help='Reason for quarantining the file' ) @pass_context def quarantine_add(ctx: CLIContext, file: Path, reason: Optional[str]): """Add file to quarantine. Moves a file to the category-specific quarantine directory. Only movie and series files can be quarantined in v1. Anime and other files will be rejected. Example: vlm quarantine add /path/to/movie.mkv vlm quarantine add /path/to/series.mkv --reason "duplicate" """ from vlm.quarantine import QuarantineManager config = ctx.config logger = ctx.logger try: # Create quarantine manager manager = QuarantineManager(config, logger) # Display confirmation click.echo(f"Quarantining file: {file}") if reason: click.echo(f"Reason: {reason}") click.echo() # Quarantine the file result = manager.quarantine_file(file, reason=reason) if result.success: click.echo(f"✓ File successfully quarantined!") click.echo(f" Original: {result.operation.source_path}") click.echo(f" Quarantine: {result.operation.destination_path}") click.echo() click.echo("To restore this file, run:") click.echo(f" vlm quarantine restore {result.operation.destination_path}") else: _command_error( ctx, f"✗ Failed to quarantine file: {result.error_message}", f"Quarantine failed for {file}: {result.error_message}", ) logger.info(f"Quarantined file: {file}") except ValueError as e: # Category restriction error _command_error(ctx, f"Error: {e}", f"Quarantine rejected: {e}") except Exception as e: _command_error( ctx, f"Error quarantining file: {e}", f"Failed to quarantine file: {e}", exc_info=True, ) @quarantine.command('restore') @click.argument('file', type=click.Path(exists=True, path_type=Path)) @pass_context def quarantine_restore(ctx: CLIContext, file: Path): """Restore file from quarantine. Moves a file from quarantine back to its original location. This is a best-effort operation that may fail if the original location is occupied. Example: vlm quarantine restore /path/to/.quarantine/movie.mkv """ from vlm.quarantine import QuarantineManager config = ctx.config logger = ctx.logger try: # Create quarantine manager manager = QuarantineManager(config, logger) # Display confirmation click.echo(f"Restoring file from quarantine: {file}") click.echo() # Restore the file result = manager.restore_from_quarantine(file) if result.success: click.echo(f"✓ File successfully restored!") click.echo(f" Quarantine: {result.operation.source_path}") click.echo(f" Restored to: {result.operation.destination_path}") else: if result.operation.has_conflict: click.echo(f"✗ Cannot restore: {result.operation.conflict_reason}", err=True) click.echo(f" Original location: {result.operation.destination_path}", err=True) else: click.echo(f"✗ Failed to restore file: {result.error_message}", err=True) _command_error( ctx, f"Error restoring file: {result.error_message or result.operation.conflict_reason or 'unknown error'}", f"Failed to restore file: {file}", ) logger.info(f"Restored file from quarantine: {file}") except Exception as e: _command_error( ctx, f"Error restoring file: {e}", f"Failed to restore file: {e}", exc_info=True, ) @main.command() @click.option( '--log', type=click.Path(exists=True, path_type=Path), default=None, help='Path to rollback log JSON file' ) @pass_context def rollback(ctx: CLIContext, log: Optional[Path]): """Rollback previous execution (best-effort). Attempts to reverse file operations from a previous execution by moving files from their destination back to their source. This is a best-effort operation that may not succeed if files have been modified or moved. Operations are processed in LIFO (Last In, First Out) order for best-effort restoration. All rollback attempts are logged with detailed results. Example: vlm rollback # Find latest rollback log vlm rollback --log rollback_.json # Use specific log vlm rollback --log ~/.vlm/rollback/rollback_*.json """ try: from vlm.commands.execute import rollback_cmd rollback_cmd(ctx, log) except FileNotFoundError as e: _command_error(ctx, f"Error: {e}", f"Rollback log not found: {e}") except ValueError as e: _command_error(ctx, f"Error: {e}", f"Rollback failed: {e}") except OSError as e: _command_error( ctx, f"Error during rollback: {e}", f"Rollback failed: {e}", exc_info=True, ) def _fallback_plan_summary(execution_plan) -> str: """Build a short plan summary from summary and summary_by_reason when human_summary is empty.""" return fallback_plan_summary(execution_plan) @main.group() @pass_context def report(ctx: CLIContext): """Generate and export reports. Generate various reports about your video library including inventory, completeness analysis, duplicate detection, and summary statistics. """ pass @report.command('inventory') @click.option( '--format', type=click.Choice(['csv', 'json', 'text'], case_sensitive=False), default='text', help='Output format (default: text)' ) @click.option( '--input', type=click.Path(path_type=Path), default=lambda: default_artifact_path('inventory.csv'), help='Input inventory CSV file (default: artifacts/inventory.csv)' ) @click.option( '--output', type=click.Path(path_type=Path), default=None, help='Output file (default: print to console)' ) @pass_context def report_inventory(ctx: CLIContext, format: str, input: Path, output: Optional[Path]): """Generate inventory report. Lists all discovered video files with metadata in the specified format. Example: vlm report inventory # Text format to console vlm report inventory --format csv # CSV format to console vlm report inventory --format json --output inventory_report.json """ from vlm.reports import generate_inventory_report from vlm.io import load_inventory_csv config = ctx.config logger = ctx.logger try: input = resolve_legacy_default_input_path(input, "input", "inventory.csv", "--input") # Load inventory from CSV click.echo(f"Loading inventory from: {input}") video_files = load_inventory_csv(input) click.echo(f"Loaded {len(video_files)} files") click.echo() # Generate report click.echo(f"Generating inventory report in {format} format...") # For text format, use CSV format as the text representation report_format = 'csv' if format == 'text' else format report_content = generate_inventory_report(video_files, report_format, config.library_root) # Output report if output: # Save to file output.parent.mkdir(parents=True, exist_ok=True) with open(output, 'w', encoding='utf-8') as f: f.write(report_content) click.echo(f"Report saved to: {output}") else: # Print to console click.echo() click.echo(report_content) logger.info(f"Generated inventory report in {format} format with {len(video_files)} files") except FileNotFoundError: _command_error(ctx, f"Error: Input file not found: {input}", f"Input file not found: {input}") except Exception as e: _command_error( ctx, f"Error generating inventory report: {e}", f"Inventory report generation failed: {e}", exc_info=True, ) @report.command('completeness') @click.option( '--format', type=click.Choice(['text', 'json'], case_sensitive=False), default='text', help='Output format (default: text)' ) @click.option( '--input', type=click.Path(path_type=Path), default=lambda: default_artifact_path('analysis.json'), help='Input analysis JSON file (default: artifacts/analysis.json)' ) @click.option( '--output', type=click.Path(path_type=Path), default=None, help='Output file (default: print to console)' ) @click.option( '--plan', type=click.Path(path_type=Path), default=None, help='Optional plan JSON; when provided, report includes plan content summary' ) @pass_context def report_completeness(ctx: CLIContext, format: str, input: Path, output: Optional[Path], plan: Optional[Path]): """Generate completeness report. Shows series with episode gaps detected through heuristic analysis. Example: vlm report completeness # Text format to console vlm report completeness --format json # JSON format to console vlm report completeness --plan plan.json # Include plan content summary vlm report completeness --format text --output completeness.txt """ from vlm.reports import generate_completeness_report from vlm.models import SeasonCompleteness from vlm.planner import load_plan from vlm.io import load_analysis_json config = ctx.config logger = ctx.logger input = resolve_legacy_default_input_path(input, "input", "analysis.json", "--input") plan_summary = None if plan: if not plan.exists(): click.echo(f"Error: Plan file not found: {plan}", err=True) sys.exit(1) execution_plan = load_plan(plan) plan_summary = execution_plan.human_summary or _fallback_plan_summary(execution_plan) try: # Load analysis from JSON click.echo(f"Loading analysis from: {input}") analysis_data = load_analysis_json(input) # Extract completeness data completeness_list = analysis_data.get('completeness', []) # Convert to SeasonCompleteness objects season_completeness = [] for c in completeness_list: season_completeness.append(SeasonCompleteness( series_title=c['series_title'], season=c['season'], episodes_found=c['episodes_found'], episodes_missing=c['episodes_missing'] )) click.echo(f"Loaded {len(season_completeness)} series with gaps") click.echo() # Generate report click.echo(f"Generating completeness report in {format} format...") report_content = generate_completeness_report( season_completeness, format, config.library_root, plan_summary=plan_summary ) # Output report if output: # Save to file output.parent.mkdir(parents=True, exist_ok=True) with open(output, 'w', encoding='utf-8') as f: f.write(report_content) click.echo(f"Report saved to: {output}") else: # Print to console click.echo() click.echo(report_content) logger.info(f"Generated completeness report in {format} format with {len(season_completeness)} series") except FileNotFoundError: _command_error(ctx, f"Error: Input file not found: {input}", f"Input file not found: {input}") except json.JSONDecodeError as e: _command_error( ctx, f"Error: Failed to parse JSON file: {e}", f"JSON parsing failed: {e}", exc_info=True, ) except Exception as e: _command_error( ctx, f"Error generating completeness report: {e}", f"Completeness report generation failed: {e}", exc_info=True, ) @report.command('duplicates') @click.option( '--format', type=click.Choice(['text', 'json'], case_sensitive=False), default='text', help='Output format (default: text)' ) @click.option( '--input', type=click.Path(path_type=Path), default=lambda: default_artifact_path('analysis.json'), help='Input analysis JSON file (default: artifacts/analysis.json)' ) @click.option( '--output', type=click.Path(path_type=Path), default=None, help='Output file (default: print to console)' ) @click.option( '--plan', type=click.Path(path_type=Path), default=None, help='Optional plan JSON; when provided, report includes plan content summary' ) @pass_context def report_duplicates(ctx: CLIContext, format: str, input: Path, output: Optional[Path], plan: Optional[Path]): """Generate duplicate report. Shows duplicate files with quality comparison data to help decide which files to keep. Example: vlm report duplicates # Text format to console vlm report duplicates --format json # JSON format to console vlm report duplicates --plan plan.json # Include plan content summary vlm report duplicates --format text --output duplicates.txt """ from vlm.reports import generate_duplicate_report from vlm.models import DuplicateGroup, MovieIdentity, SeriesIdentity, VideoFile from vlm.planner import load_plan from datetime import datetime, timezone from vlm.io import load_analysis_json config = ctx.config logger = ctx.logger input = resolve_legacy_default_input_path(input, "input", "analysis.json", "--input") plan_summary = None if plan: if not plan.exists(): click.echo(f"Error: Plan file not found: {plan}", err=True) sys.exit(1) execution_plan = load_plan(plan) plan_summary = execution_plan.human_summary or _fallback_plan_summary(execution_plan) try: # Load analysis from JSON click.echo(f"Loading analysis from: {input}") analysis_data = load_analysis_json(input) # Extract duplicates data duplicates_list = analysis_data.get('duplicates', []) # Convert to DuplicateGroup objects duplicate_groups = [] for d in duplicates_list: identity_data = d['identity'] # Reconstruct identity if identity_data['type'] == 'movie': identity = MovieIdentity( title=identity_data['title'], year=identity_data.get('year'), confidence=1.0, needs_review=False, original_filename="" ) else: # series identity = SeriesIdentity( title=identity_data['title'], season=identity_data.get('season'), episodes=identity_data.get('episodes', []), confidence=1.0, needs_review=False, original_filename="" ) quality_by_path = { str(item.get("path", "")): item for item in d.get("quality_comparison", []) } # Reconstruct VideoFile objects from file paths and preserve size metadata files = [] for file_path in d['files']: quality = quality_by_path.get(str(file_path), {}) files.append(VideoFile( path=Path(file_path), filename=Path(file_path).name, size_bytes=int(quality.get("size_bytes", 0) or 0), modified_timestamp=datetime.now(timezone.utc), category="", resolution=quality.get("resolution"), codec=quality.get("codec"), duration_seconds=quality.get("duration_seconds"), bitrate_kbps=quality.get("bitrate_kbps"), )) duplicate_groups.append(DuplicateGroup( identity=identity, files=files, quality_comparison=d['quality_comparison'] )) click.echo(f"Loaded {len(duplicate_groups)} duplicate groups") click.echo() # Generate report click.echo(f"Generating duplicate report in {format} format...") report_content = generate_duplicate_report( duplicate_groups, format, config.library_root, plan_summary=plan_summary ) # Output report if output: # Save to file output.parent.mkdir(parents=True, exist_ok=True) with open(output, 'w', encoding='utf-8') as f: f.write(report_content) click.echo(f"Report saved to: {output}") else: # Print to console click.echo() click.echo(report_content) logger.info(f"Generated duplicate report in {format} format with {len(duplicate_groups)} groups") except FileNotFoundError: _command_error(ctx, f"Error: Input file not found: {input}", f"Input file not found: {input}") except json.JSONDecodeError as e: _command_error( ctx, f"Error: Failed to parse JSON file: {e}", f"JSON parsing failed: {e}", exc_info=True, ) except Exception as e: _command_error( ctx, f"Error generating duplicate report: {e}", f"Duplicate report generation failed: {e}", exc_info=True, ) @report.command('summary') @click.option( '--input', type=click.Path(path_type=Path), default=lambda: default_artifact_path('inventory.csv'), help='Input inventory CSV file (default: artifacts/inventory.csv)' ) @click.option( '--output', type=click.Path(path_type=Path), default=None, help='Output file (default: print to console)' ) @pass_context def report_summary(ctx: CLIContext, input: Path, output: Optional[Path]): """Generate summary report. Shows library statistics including total file count, size, and category breakdown. Example: vlm report summary # Print to console vlm report summary --output summary.txt # Save to file """ from vlm.reports import generate_summary_report from vlm.io import load_inventory_csv config = ctx.config logger = ctx.logger try: input = resolve_legacy_default_input_path(input, "input", "inventory.csv", "--input") # Load inventory from CSV click.echo(f"Loading inventory from: {input}") video_files = load_inventory_csv(input) click.echo(f"Loaded {len(video_files)} files") click.echo() # Generate report click.echo("Generating summary report...") report_content = generate_summary_report(video_files, config.library_root) # Output report if output: # Save to file output.parent.mkdir(parents=True, exist_ok=True) with open(output, 'w', encoding='utf-8') as f: f.write(report_content) click.echo(f"Report saved to: {output}") else: # Print to console click.echo() click.echo(report_content) logger.info(f"Generated summary report with {len(video_files)} files") except FileNotFoundError: _command_error(ctx, f"Error: Input file not found: {input}", f"Input file not found: {input}") except Exception as e: _command_error( ctx, f"Error generating summary report: {e}", f"Summary report generation failed: {e}", exc_info=True, ) @main.group() @pass_context def state(ctx: CLIContext): """Manage file states. Track file statuses and user decisions throughout the workflow. """ pass @state.command('show') @click.argument('file', type=click.Path(path_type=Path)) @pass_context def state_show(ctx: CLIContext, file: Path): """Show state for a file. Displays the current status, reason, and last update timestamp for a file. Example: vlm state show /path/to/movie.mkv """ from vlm.state import StateManager config = ctx.config logger = ctx.logger try: # Get state store path state_path = Path.home() / ".vlm" / "state.json" # Create state manager manager = StateManager(state_path) # Get file state file_state = manager.get_file_state(file) if file_state is None: click.echo(f"No state found for file: {file}") click.echo("This file has not been tracked yet.") else: click.echo(f"State for file: {file}") click.echo() click.echo(f" Status: {file_state.status}") if file_state.reason: click.echo(f" Reason: {file_state.reason}") click.echo(f" Updated: {file_state.updated_at.strftime('%Y-%m-%d %H:%M:%S')}") logger.info(f"Showed state for file: {file}") except Exception as e: _command_error( ctx, f"Error showing file state: {e}", f"Failed to show file state: {e}", exc_info=True, ) @state.command('set') @click.argument('file', type=click.Path(path_type=Path)) @click.option( '--status', type=click.Choice(['reviewed', 'ignored', 'planned', 'executed', 'quarantined'], case_sensitive=False), required=True, help='Status to set for the file' ) @click.option( '--reason', type=str, default=None, help='Optional reason for the status' ) @pass_context def state_set(ctx: CLIContext, file: Path, status: str, reason: Optional[str]): """Set state for a file. Updates the status and optional reason for a file. This operation is idempotent - setting the same status multiple times will update the timestamp and reason. Valid statuses: reviewed, ignored, planned, executed, quarantined Example: vlm state set /path/to/movie.mkv --status reviewed vlm state set /path/to/movie.mkv --status ignored --reason "duplicate" """ from vlm.state import StateManager config = ctx.config logger = ctx.logger try: # Get state store path state_path = Path.home() / ".vlm" / "state.json" # Create state manager manager = StateManager(state_path) # Set file state manager.set_file_state(file, status, reason) # Save state manager.save() # Display confirmation click.echo(f"✓ State updated for file: {file}") click.echo(f" Status: {status}") if reason: click.echo(f" Reason: {reason}") logger.info(f"Set state for file {file}: status={status}, reason={reason}") except ValueError as e: _command_error(ctx, f"Error: {e}", f"Invalid status: {e}") except Exception as e: _command_error( ctx, f"Error setting file state: {e}", f"Failed to set file state: {e}", exc_info=True, ) @state.command('query') @click.option( '--status', type=click.Choice(['reviewed', 'ignored', 'planned', 'executed', 'quarantined'], case_sensitive=False), required=True, help='Status to query for' ) @pass_context def state_query(ctx: CLIContext, status: str): """Query files by status. Lists all files with the specified status. Example: vlm state query --status ignored vlm state query --status reviewed """ from vlm.state import StateManager config = ctx.config logger = ctx.logger try: # Get state store path state_path = Path.home() / ".vlm" / "state.json" # Create state manager manager = StateManager(state_path) # Query files by status file_states = manager.query_by_status(status) if not file_states: click.echo(f"No files found with status '{status}'.") return # Display results click.echo(f"Files with status '{status}':") click.echo("=" * 80) click.echo() for i, file_state in enumerate(file_states, 1): click.echo(f"[{i}] {file_state.file_path}") if file_state.reason: click.echo(f" Reason: {file_state.reason}") click.echo(f" Updated: {file_state.updated_at.strftime('%Y-%m-%d %H:%M:%S')}") click.echo() click.echo("=" * 80) click.echo(f"Total: {len(file_states)} file(s)") logger.info(f"Queried files with status '{status}': {len(file_states)} found") except Exception as e: _command_error( ctx, f"Error querying file states: {e}", f"Failed to query file states: {e}", exc_info=True, ) @state.command('clear') @click.argument('file', type=click.Path(path_type=Path)) @pass_context def state_clear(ctx: CLIContext, file: Path): """Clear state for a file. Removes the state tracking for a file. Example: vlm state clear /path/to/movie.mkv """ from vlm.state import StateManager config = ctx.config logger = ctx.logger try: # Get state store path state_path = Path.home() / ".vlm" / "state.json" # Create state manager manager = StateManager(state_path) # Check if state exists file_state = manager.get_file_state(file) if file_state is None: click.echo(f"No state found for file: {file}") click.echo("Nothing to clear.") return # Clear file state manager.clear_state(file) # Save state manager.save() # Display confirmation click.echo(f"✓ State cleared for file: {file}") logger.info(f"Cleared state for file: {file}") except Exception as e: _command_error( ctx, f"Error clearing file state: {e}", f"Failed to clear file state: {e}", exc_info=True, ) @main.group(name='config') @pass_context def config_cmd(ctx: CLIContext): """Manage configuration. Initialize, view, and validate configuration settings. """ pass @config_cmd.command('init') @click.option( '--path', type=click.Path(path_type=Path), default=default_config_path, help='Path where configuration file should be created' ) @pass_context def config_init(ctx: CLIContext, path: Path): """Initialize configuration file with defaults.""" try: if path.exists(): click.echo(f"Configuration file already exists at {path}", err=True) if not click.confirm("Overwrite existing configuration?"): click.echo("Configuration initialization cancelled.") return create_default_config(path) click.echo(f"Configuration file created at {path}") click.echo("Edit this file to customize your settings.") except Exception as e: _command_error( ctx, f"Error creating configuration: {e}", f"Configuration creation failed: {e}", exc_info=True, ) @config_cmd.command('show') @pass_context def config_show(ctx: CLIContext): """Show current configuration.""" cfg = ctx.config click.echo("Current configuration:") click.echo(f" Library root: {cfg.library_root}") click.echo(f" Video extensions: {', '.join(cfg.video_extensions)}") click.echo(f" Movie template: {cfg.movie_template}") click.echo(f" Series template: {cfg.series_template}") click.echo(f" Movie filename template: {cfg.movie_filename_template}") click.echo(f" Series filename template: {cfg.series_filename_template}") click.echo(f" Quarantine directory: {cfg.quarantine_dir}") click.echo(f" Workspace directory: {cfg.workspace_dir}") click.echo(f" Log level: {cfg.log_level}") @config_cmd.command('validate') @pass_context def config_validate(ctx: CLIContext): """Validate configuration.""" cfg = ctx.config errors = validate_config(cfg) if not errors: click.echo("Configuration is valid.") else: click.echo("Configuration validation errors:", err=True) for error in errors: click.echo(f" - {error}", err=True) _command_error(ctx, "Configuration validation failed.", "Configuration validation failed") if __name__ == '__main__': main()