Files
dl-organizer/src/vlm/commands/report.py
T
windyboy dfa18ed405 refactor: consolidate skill docs, add anti-drift tests, and apply audit fixes
DLO-13: Restructure vlm-library-workflow skill as safety contract layer.
- Rewrite SKILL.md (69 lines): safety contract, execution threshold semantics,
  six-step high-risk loop, decision rules, phase skeleton
- Delete redundant references (cli-reference, workflow, command-recipes, dev-guide)
- Add triage.md (failure mapping + preflight) and dev-map.md (module→test mapping)
- Add tests/test_docs_consistency.py: 78 parametrized tests verifying documented
  vlm commands exist in CLI registry
- Add CSV path mismatch test to test_plan_review.py (4th safety gate path)
- Delete vlm-expert.skill (Gemini package, 7 months stale) and README Gemini section

DLO-2 audit fixes: rate limiter injection, symmetric quarantine categories,
review-plan safety gates, parser improvements, planner validation.

CLI modularization: commands/ directory with one module per command group.
2026-09-25 13:50:09 +08:00

134 lines
5.6 KiB
Python

"""Report CLI commands."""
from __future__ import annotations
from pathlib import Path
from typing import Optional
import click
from vlm.cli_helpers import (
command_error,
default_artifact_path,
emit_report,
optional_plan_summary,
resolve_legacy_default_input_path,
run_command,
)
from vlm.context import CLIContext, pass_context
from vlm.io import load_analysis_json, load_inventory_csv
from vlm.reports import (
completeness_from_analysis,
duplicate_groups_from_analysis,
generate_completeness_report,
generate_duplicate_report,
generate_inventory_report,
generate_summary_report,
)
def _run_inventory(ctx: CLIContext, format: str, input: Path, output: Optional[Path]) -> None:
input = resolve_legacy_default_input_path(input, "input", "inventory.csv", "--input")
click.echo(f"Loading inventory from: {input}")
video_files = load_inventory_csv(input)
click.echo(f"Loaded {len(video_files)} files\n")
report_format = "csv" if format == "text" else format
content = generate_inventory_report(video_files, report_format, ctx.config.library_root)
emit_report(content, output)
ctx.logger.info("inventory report: %s files, format=%s", len(video_files), format)
def _run_completeness(
ctx: CLIContext, format: str, input: Path, output: Optional[Path], plan: Optional[Path]
) -> None:
input = resolve_legacy_default_input_path(input, "input", "analysis.json", "--input")
plan_summary = optional_plan_summary(plan)
click.echo(f"Loading analysis from: {input}")
analysis_data = load_analysis_json(input)
seasons = completeness_from_analysis(analysis_data)
click.echo(f"Loaded {len(seasons)} series with gaps\n")
content = generate_completeness_report(
seasons, format, ctx.config.library_root, plan_summary=plan_summary
)
emit_report(content, output)
ctx.logger.info("completeness report: %s series", len(seasons))
def _run_duplicates(
ctx: CLIContext, format: str, input: Path, output: Optional[Path], plan: Optional[Path]
) -> None:
input = resolve_legacy_default_input_path(input, "input", "analysis.json", "--input")
plan_summary = optional_plan_summary(plan)
click.echo(f"Loading analysis from: {input}")
analysis_data = load_analysis_json(input)
groups = duplicate_groups_from_analysis(analysis_data)
click.echo(f"Loaded {len(groups)} duplicate groups\n")
content = generate_duplicate_report(
groups, format, ctx.config.library_root, plan_summary=plan_summary
)
emit_report(content, output)
ctx.logger.info("duplicate report: %s groups", len(groups))
def _run_summary(ctx: CLIContext, input: Path, output: Optional[Path]) -> None:
input = resolve_legacy_default_input_path(input, "input", "inventory.csv", "--input")
click.echo(f"Loading inventory from: {input}")
video_files = load_inventory_csv(input)
click.echo(f"Loaded {len(video_files)} files\n")
content = generate_summary_report(video_files, ctx.config.library_root)
emit_report(content, output)
ctx.logger.info("summary report: %s files", len(video_files))
@click.group()
@pass_context
def report(ctx: CLIContext):
"""Generate inventory, completeness, duplicate, and summary reports."""
@report.command("inventory")
@click.option("--format", type=click.Choice(["csv", "json", "text"], case_sensitive=False), default="text")
@click.option("--input", type=click.Path(path_type=Path), default=lambda: default_artifact_path("inventory.csv"))
@click.option("--output", type=click.Path(path_type=Path), default=None)
@pass_context
def report_inventory(ctx: CLIContext, format: str, input: Path, output: Optional[Path]):
"""List discovered files with metadata."""
run_command(ctx, lambda: _run_inventory(ctx, format, input, output), stage="inventory report")
@report.command("completeness")
@click.option("--format", type=click.Choice(["text", "json"], case_sensitive=False), default="text")
@click.option("--input", type=click.Path(path_type=Path), default=lambda: default_artifact_path("analysis.json"))
@click.option("--output", type=click.Path(path_type=Path), default=None)
@click.option("--plan", type=click.Path(path_type=Path), default=None)
@pass_context
def report_completeness(ctx: CLIContext, format: str, input: Path, output: Optional[Path], plan: Optional[Path]):
"""Show series with episode gaps."""
run_command(
ctx, lambda: _run_completeness(ctx, format, input, output, plan),
stage="completeness report", json_errors=True,
)
@report.command("duplicates")
@click.option("--format", type=click.Choice(["text", "json"], case_sensitive=False), default="text")
@click.option("--input", type=click.Path(path_type=Path), default=lambda: default_artifact_path("analysis.json"))
@click.option("--output", type=click.Path(path_type=Path), default=None)
@click.option("--plan", type=click.Path(path_type=Path), default=None)
@pass_context
def report_duplicates(ctx: CLIContext, format: str, input: Path, output: Optional[Path], plan: Optional[Path]):
"""Show duplicate groups with quality comparison."""
run_command(
ctx, lambda: _run_duplicates(ctx, format, input, output, plan),
stage="duplicate report", json_errors=True,
)
@report.command("summary")
@click.option("--input", type=click.Path(path_type=Path), default=lambda: default_artifact_path("inventory.csv"))
@click.option("--output", type=click.Path(path_type=Path), default=None)
@pass_context
def report_summary(ctx: CLIContext, input: Path, output: Optional[Path]):
"""Show library statistics."""
run_command(ctx, lambda: _run_summary(ctx, input, output), stage="summary report")