diff --git a/README.md b/README.md index bc8fd84..8185267 100644 --- a/README.md +++ b/README.md @@ -240,14 +240,14 @@ vlm analyze vlm plan --analysis artifacts/analysis.json # Output: artifacts/plan.json with operations, human summary, and duplicate quarantine decisions -# 7. Review the plan in terminal (summary + high-risk preview) +# 7. Review the plan in terminal (summary + high-risk preview + verdict) vlm review-plan # Optional: interactive full-screen review (install `.[tui]` first) vlm review-plan --tui -# Optional: control preview size -vlm review-plan --preview-limit 20 -# Optional: show every high-risk operation in terminal -vlm review-plan --show-all +# Enriched CSV (title, relative paths, quality) + duplicate grouping +vlm review-plan --identities artifacts/identities.json --analysis artifacts/analysis.json --group-by duplicate +# Spot-check random safe moves; write target tree preview +vlm review-plan --sample-safe 5 --structure-preview artifacts/plan_structure.txt # Edit artifacts/plan_manual_review.csv in Excel/Numbers # Sync your manual decisions back to artifacts/plan.json vlm apply-review @@ -257,8 +257,8 @@ vlm apply-review vlm execute # Shows what would happen without making changes (respecting your manual edits) -# 9. Execute with confirmation -vlm execute --confirm +# 9. Execute with confirmation (optional review gate) +vlm execute --confirm --require-review # Actually performs the file operations # 10. If needed, rollback diff --git a/src/vlm/cli.py b/src/vlm/cli.py index 5b9dbd8..d5c6599 100644 --- a/src/vlm/cli.py +++ b/src/vlm/cli.py @@ -17,7 +17,14 @@ 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.plan_render import ( + fallback_plan_summary, + preferred_plan_summary, + render_review_footer, + render_review_preview, + render_review_verdict, + duplicate_groups_from_plan, +) from vlm.utils import format_size @@ -538,6 +545,38 @@ def plan(ctx: CLIContext, input: Path, output: Path, analysis: Path | None): default=False, help='Interactive Textual UI (requires: uv pip install -e ".[tui]")' ) +@click.option( + '--identities', + type=click.Path(path_type=Path), + default=None, + help='Identities JSON for enriched review rows (default: artifacts/identities.json if present)' +) +@click.option( + '--analysis', + type=click.Path(path_type=Path), + default=None, + help='Analysis JSON for duplicate grouping and TUI quality pane' +) +@click.option( + '--group-by', + type=click.Choice(['none', 'reason', 'title', 'duplicate'], case_sensitive=False), + default='none', + show_default=True, + help='Reorder review rows for display/export' +) +@click.option( + '--sample-safe', + type=int, + default=0, + show_default=True, + help='Include N random non-high-risk move operations for spot-checking' +) +@click.option( + '--structure-preview', + type=click.Path(path_type=Path), + default=None, + help='Write target library tree preview to this file (e.g. artifacts/plan_structure.txt)' +) @pass_context def review_plan_cmd( ctx: CLIContext, @@ -548,10 +587,17 @@ def review_plan_cmd( preview_limit: int, show_all: bool, tui: bool, + identities: Optional[Path], + analysis: Optional[Path], + group_by: str, + sample_safe: int, + structure_preview: Optional[Path], ): """Review a plan and export high-risk operations for manual confirmation.""" + from vlm.io import load_analysis_json, load_identities_json from vlm.planner import load_plan - from vlm.plan_review import review_plan, save_review_csv + from vlm.plan_review import GROUP_BY_CHOICES, prepare_review_rows, save_review_csv + from vlm.plan_structure_preview import write_structure_preview logger = ctx.logger @@ -563,6 +609,13 @@ def review_plan_cmd( if preview_limit < 1: click.echo("Error: --preview-limit must be >= 1", err=True) sys.exit(1) + if sample_safe < 0: + click.echo("Error: --sample-safe must be >= 0", err=True) + sys.exit(1) + group_by = group_by.lower() + if group_by not in GROUP_BY_CHOICES: + click.echo(f"Error: invalid --group-by {group_by}", err=True) + sys.exit(1) if tui: if not _review_plan_tui_streams_ok(): @@ -571,15 +624,46 @@ def review_plan_cmd( click.echo(f"Loading plan: {input}") execution_plan = load_plan(input) - rows, counters = review_plan( + + identities_data = None + identities_path = identities + if identities_path is None: + default_id = default_artifact_path("identities.json") + if default_id.is_file(): + identities_path = default_id + if identities_path is not None and identities_path.is_file(): + identities_data = load_identities_json(identities_path) + click.echo(f"Loaded identities: {identities_path}") + + analysis_data = None + if analysis is not None and analysis.is_file(): + analysis_data = load_analysis_json(analysis) + click.echo(f"Loaded analysis: {analysis}") + + rows, counters = prepare_review_rows( execution_plan, season_threshold=season_threshold, episode_threshold=episode_threshold, + library_root=ctx.config.library_root, + identities_data=identities_data, + analysis_data=analysis_data, + sample_safe=sample_safe, + group_by=group_by, ) + if structure_preview is not None: + write_structure_preview( + execution_plan, + ctx.config.library_root, + structure_preview, + ) + click.echo(f"Wrote structure preview: {structure_preview}") + if tui: + from vlm.plan_review import build_duplicate_path_maps from vlm.review_tui import ReviewTUIContext, run_plan_review_tui + _, path_to_quality = build_duplicate_path_maps(analysis_data) tui_ctx = ReviewTUIContext( rows=rows, counters=counters, @@ -587,6 +671,7 @@ def review_plan_cmd( output_csv=output, plan_input=input, summary_text=preferred_plan_summary(execution_plan), + path_to_quality=path_to_quality, ) rc = run_plan_review_tui(tui_ctx) if rc != 0: @@ -617,6 +702,9 @@ def review_plan_cmd( click.echo(f" high_episode: {counters['high_episode']}") click.echo(f" conflicts: {counters['conflicts']}") + click.echo() + click.echo(render_review_verdict(counters)) + click.echo() click.echo("High-risk operations preview:") if rows: @@ -624,6 +712,7 @@ def review_plan_cmd( rows, preview_limit=preview_limit, show_all=show_all, + library_root=ctx.config.library_root, ) for line in preview_lines: click.echo(line) @@ -638,6 +727,16 @@ def review_plan_cmd( click.echo() click.echo(f"Saved manual review CSV to: {output}") + dup_groups = duplicate_groups_from_plan(execution_plan) + click.echo() + for line in render_review_footer( + counters=counters, + output_csv=output, + plan_input=input, + duplicate_groups=dup_groups, + ): + click.echo(line) + logger.info( "Plan review completed: total=%s high_risk=%s output=%s", counters["total_operations"], @@ -764,8 +863,30 @@ def apply_review_cmd( default=False, help='Enable safe mode: prevent any operations that would destroy directories' ) +@click.option( + '--require-review', + is_flag=True, + default=False, + help='With --confirm, require a current plan_manual_review.csv when high-risk ops exist' +) +@click.option( + '--review-csv', + type=click.Path(path_type=Path), + default=None, + help='Review CSV path for --require-review (default: beside plan file)' +) @pass_context -def execute(ctx: CLIContext, plan: Path, confirm: bool, yes: bool, verbose_ops: bool, preserve_directories: bool, safe_mode: bool): +def execute( + ctx: CLIContext, + plan: Path, + confirm: bool, + yes: bool, + verbose_ops: bool, + preserve_directories: bool, + safe_mode: bool, + require_review: bool, + review_csv: Optional[Path], +): """Execute plan (defaults to dry-run, requires --confirm). Executes file operations from a plan. Defaults to dry-run mode which @@ -782,7 +903,17 @@ def execute(ctx: CLIContext, plan: Path, confirm: bool, yes: bool, verbose_ops: 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) + execute_cmd( + ctx, + plan, + confirm, + yes, + verbose_ops, + preserve_directories, + safe_mode, + require_review=require_review, + review_csv=review_csv, + ) except FileNotFoundError: _command_error( ctx, diff --git a/src/vlm/commands/execute.py b/src/vlm/commands/execute.py index fe1ac96..2ad9cd6 100644 --- a/src/vlm/commands/execute.py +++ b/src/vlm/commands/execute.py @@ -42,6 +42,8 @@ def execute_cmd( verbose_ops: bool, preserve_directories: bool, safe_mode: bool, + require_review: bool = False, + review_csv: Path | None = None, ) -> None: """Execute an execution plan in dry-run or execute mode.""" config = ctx.config @@ -53,6 +55,15 @@ def execute_cmd( execution_plan = load_plan(plan) + if require_review and confirm: + from vlm.plan_review import check_review_requirements + + csv_path = review_csv or plan.parent / "plan_manual_review.csv" + review_errors = check_review_requirements(execution_plan, plan, csv_path) + if review_errors: + formatted = "\n".join(f" - {e}" for e in review_errors) + raise ValueError(f"REVIEW REQUIRED BEFORE EXECUTE:\n{formatted}") + if preserve_directories: click.echo("Directory preservation is enabled (plan metadata/operations will be honored).") click.echo() diff --git a/src/vlm/io.py b/src/vlm/io.py index 179e5cb..064332a 100644 --- a/src/vlm/io.py +++ b/src/vlm/io.py @@ -282,6 +282,11 @@ def execution_plan_to_record(plan: ExecutionPlan) -> PlanJSON: "reason": op.reason, "has_conflict": op.has_conflict, "conflict_reason": op.conflict_reason, + **( + {"review_context": op.review_context} + if op.review_context + else {} + ), } for op in plan.operations ], @@ -305,6 +310,7 @@ def execution_plan_from_record(data: object) -> ExecutionPlan: reason=op["reason"], has_conflict=op["has_conflict"], conflict_reason=op.get("conflict_reason"), + review_context=op.get("review_context") or {}, ) for op in plan_dict["operations"] ] diff --git a/src/vlm/models.py b/src/vlm/models.py index 4ee6223..662c275 100644 --- a/src/vlm/models.py +++ b/src/vlm/models.py @@ -128,6 +128,7 @@ class FileOperation: reason: str has_conflict: bool conflict_reason: Optional[str] = None + review_context: dict = field(default_factory=dict) @dataclass diff --git a/src/vlm/plan_render.py b/src/vlm/plan_render.py index 11bc70b..a45cf8b 100644 --- a/src/vlm/plan_render.py +++ b/src/vlm/plan_render.py @@ -2,8 +2,11 @@ from __future__ import annotations +from pathlib import Path from typing import Any +from vlm.review_display import display_path, risk_flags_to_labels + def fallback_plan_summary(execution_plan: Any) -> str: """Build a short plan summary from summary and summary_by_reason. @@ -34,10 +37,68 @@ def preferred_plan_summary(execution_plan: Any) -> str: return fallback_plan_summary(execution_plan) +def duplicate_groups_from_plan(execution_plan: Any) -> int: + metadata = getattr(execution_plan, "metadata", {}) or {} + if not isinstance(metadata, dict): + return 0 + val = metadata.get("duplicate_groups_considered", 0) + snapshot = metadata.get("validation_snapshot", {}) + if isinstance(snapshot, dict) and not val: + val = snapshot.get("duplicate_groups_considered", 0) + try: + return int(val) + except (TypeError, ValueError): + return 0 + + +def render_review_verdict(counters: dict[str, int]) -> str: + """Return a one-line review verdict for execute readiness.""" + high_risk = counters.get("high_risk_operations", 0) + if high_risk > 0: + return f"Verdict: BLOCKED ({high_risk} high-risk operation(s) need review)" + return "Verdict: OK TO DRY-RUN (no high-risk operations)" + + +def render_review_footer( + *, + counters: dict[str, int], + output_csv: Path, + plan_input: Path, + duplicate_groups: int = 0, +) -> list[str]: + """Build actionable next-step lines after review-plan output.""" + lines: list[str] = [] + high_risk = counters.get("high_risk_operations", 0) + + if high_risk > 0: + lines.append("Next steps:") + lines.append( + f" 1. Review high-risk items: vlm review-plan --tui --input {plan_input} " + f"--output {output_csv}" + ) + lines.append( + f" Or edit {output_csv} (set operation_type to no-op to reject), then:" + ) + lines.append(f" vlm apply-review --plan {plan_input} --csv {output_csv}") + lines.append(f" 2. Dry-run after apply-review: vlm execute --plan {plan_input}") + else: + lines.append("Next steps:") + lines.append(f" 1. Dry-run: vlm execute --plan {plan_input}") + lines.append(f" 2. Execute with confirmation: vlm execute --plan {plan_input} --confirm") + + if duplicate_groups > 0: + lines.append( + f" Duplicate quality comparison: vlm report duplicates --plan {plan_input}" + ) + + return lines + + def render_review_preview( rows: list[dict[str, str]], preview_limit: int = 10, show_all: bool = False, + library_root: Path | None = None, ) -> tuple[list[str], int]: """Render high-risk review rows for console preview. @@ -53,14 +114,39 @@ def render_review_preview( for row in selected: idx = row.get("index", "?") operation_type = row.get("operation_type", "") - flags = row.get("risk_flags", "") or "none" + flags_raw = row.get("risk_flags", "") + flags_label = risk_flags_to_labels(flags_raw, max_len=80) if flags_raw else "(none)" source_path = row.get("source_path", "") destination_path = row.get("destination_path", "") reason = row.get("reason", "") - lines.append(f" - [{idx}] {operation_type} | flags={flags}") - lines.append(f" source: {source_path}") - lines.append(f" destination: {destination_path or '(none)'}") + source_name = row.get("source_name") or ( + Path(source_path).name if source_path else "" + ) + dest_name = row.get("dest_name") or ( + Path(destination_path).name if destination_path else "" + ) + rel_source = row.get("rel_source", "") + rel_dest = row.get("rel_dest", "") + if library_root is not None and source_path and not rel_source: + rel_source = display_path(Path(source_path), library_root) + if library_root is not None and destination_path and not rel_dest: + rel_dest = display_path(Path(destination_path), library_root) + + change = f"{source_name} -> {dest_name}" if dest_name else source_name + lines.append(f" - [{idx}] {operation_type} | {flags_label}") + if change: + lines.append(f" change: {change}") + if rel_source: + lines.append(f" source: {rel_source}") + elif source_path: + lines.append(f" source: {source_path}") + if rel_dest: + lines.append(f" destination: {rel_dest}") + elif destination_path: + lines.append(f" destination: {destination_path}") + elif destination_path == "": + lines.append(" destination: (none)") lines.append(f" reason: {reason}") return lines, remaining diff --git a/src/vlm/plan_review.py b/src/vlm/plan_review.py index 607baf5..30411e2 100644 --- a/src/vlm/plan_review.py +++ b/src/vlm/plan_review.py @@ -3,11 +3,40 @@ from __future__ import annotations import csv +import random import re from pathlib import Path +from typing import Any -from vlm.models import ExecutionPlan, FileOperation -from vlm.utils import is_sample_path +from vlm.models import ExecutionPlan, FileOperation, MovieIdentity, SeriesIdentity +from vlm.review_display import display_path +from vlm.utils import format_size, is_sample_path + +REVIEW_CSV_BASE_FIELDS = [ + "index", + "operation_type", + "risk_flags", + "source_path", + "destination_path", + "reason", +] + +REVIEW_CSV_ENRICHED_FIELDS = [ + "title", + "category", + "season", + "episode", + "source_name", + "dest_name", + "rel_source", + "rel_dest", + "quality_hint", + "duplicate_group_id", +] + +REVIEW_CSV_ALL_FIELDS = REVIEW_CSV_BASE_FIELDS + REVIEW_CSV_ENRICHED_FIELDS + +GROUP_BY_CHOICES = ("none", "reason", "title", "duplicate") def _extract_season_episode(operation: FileOperation) -> tuple[int | None, int | None]: @@ -34,6 +63,220 @@ def _extract_season_episode(operation: FileOperation) -> tuple[int | None, int | return season, episode +def _quality_hint_from_record(record: dict[str, Any]) -> str: + vm = record.get("video_metadata") or {} + if not isinstance(vm, dict): + vm = {} + parts: list[str] = [] + if vm.get("resolution"): + parts.append(str(vm["resolution"])) + if vm.get("codec"): + parts.append(str(vm["codec"])) + size = vm.get("size_bytes") + if size: + parts.append(format_size(int(size))) + return " / ".join(parts) + + +def build_identity_lookup(identities_data: dict[str, Any]) -> dict[str, dict[str, str]]: + """Map source file path strings to review enrichment fields.""" + lookup: dict[str, dict[str, str]] = {} + + def _add_record(record: dict[str, Any], category: str) -> None: + path = record.get("path") + if not path: + return + path_key = str(path) + title = str(record.get("display_title") or record.get("title") or "") + ctx: dict[str, str] = { + "title": title, + "category": category, + "quality_hint": _quality_hint_from_record(record), + } + if category == "movie": + year = record.get("year") + if year is not None: + ctx["episode"] = "" + ctx["season"] = "" + elif category == "series": + season = record.get("season") + episodes = record.get("episodes") or [] + if season is not None: + ctx["season"] = str(season) + if episodes: + ctx["episode"] = ",".join(str(e) for e in episodes) + lookup[path_key] = ctx + try: + lookup[str(Path(path_key).resolve())] = ctx + except OSError: + pass + + for rec in identities_data.get("movies", []) or []: + if isinstance(rec, dict): + _add_record(rec, "movie") + for rec in identities_data.get("series", []) or []: + if isinstance(rec, dict): + _add_record(rec, "series") + for section in ("anime", "other"): + for rec in identities_data.get(section, []) or []: + if isinstance(rec, dict): + _add_record(rec, section) + + return lookup + + +def build_duplicate_path_maps( + analysis_data: dict[str, Any] | None, +) -> tuple[dict[str, str], dict[str, dict[str, Any]]]: + """Return path -> duplicate_group_id and path -> quality_comparison entry.""" + path_to_group: dict[str, str] = {} + path_to_quality: dict[str, dict[str, Any]] = {} + if not analysis_data: + return path_to_group, path_to_quality + + for group_idx, dup in enumerate(analysis_data.get("duplicates", []) or []): + if not isinstance(dup, dict): + continue + identity = dup.get("identity") or {} + if not isinstance(identity, dict): + identity = {} + title = identity.get("title", "") + itype = identity.get("type", "") + year = identity.get("year") + season = identity.get("season") + episodes = identity.get("episodes") or [] + if itype == "movie" and year is not None: + group_id = f"movie:{title}:{year}" + elif season is not None and episodes: + group_id = f"series:{title}:{season}:{episodes[0]}" + else: + group_id = f"dup:{group_idx}" + + quality_list = dup.get("quality_comparison") or [] + if not isinstance(quality_list, list): + quality_list = [] + for file_path in dup.get("files", []) or []: + path_key = str(file_path) + path_to_group[path_key] = group_id + for qc in quality_list: + if isinstance(qc, dict) and str(qc.get("path")) == path_key: + path_to_quality[path_key] = qc + break + + return path_to_group, path_to_quality + + +def build_review_context( + video_file_path: Path, + identity: MovieIdentity | SeriesIdentity | None, + *, + category: str | None = None, + duplicate_group_id: str = "", + keep_candidate: bool | None = None, +) -> dict[str, Any]: + """Build plan-time review_context for a single operation.""" + ctx: dict[str, Any] = {} + if category: + ctx["category"] = category + elif identity is not None: + ctx["category"] = "movie" if isinstance(identity, MovieIdentity) else "series" + + if identity is not None: + ctx["title"] = identity.title + if isinstance(identity, MovieIdentity): + if identity.year is not None: + ctx["year"] = identity.year + else: + if identity.season is not None: + ctx["season"] = identity.season + if identity.episodes: + ctx["episode"] = identity.episodes[0] + + if duplicate_group_id: + ctx["duplicate_group_id"] = duplicate_group_id + if keep_candidate is not None: + ctx["keep_candidate"] = keep_candidate + + return ctx + + +def enrich_review_row( + row: dict[str, str], + *, + library_root: Path | None = None, + identity_lookup: dict[str, dict[str, str]] | None = None, + path_to_duplicate_group: dict[str, str] | None = None, + operation: FileOperation | None = None, +) -> dict[str, str]: + """Add optional enrichment columns to a review row dict.""" + enriched = dict(row) + source_path = row.get("source_path", "") + dest_path = row.get("destination_path", "") + + enriched["source_name"] = Path(source_path).name if source_path else "" + enriched["dest_name"] = Path(dest_path).name if dest_path else "" + + if library_root is not None and source_path: + enriched["rel_source"] = display_path(Path(source_path), library_root) + if library_root is not None and dest_path: + enriched["rel_dest"] = display_path(Path(dest_path), library_root) + + ctx = (operation.review_context if operation else None) or {} + if isinstance(ctx, dict) and ctx: + if ctx.get("title"): + enriched["title"] = str(ctx["title"]) + if ctx.get("category"): + enriched["category"] = str(ctx["category"]) + if ctx.get("season") is not None: + enriched["season"] = str(ctx["season"]) + if ctx.get("episode") is not None: + enriched["episode"] = str(ctx["episode"]) + if ctx.get("duplicate_group_id"): + enriched["duplicate_group_id"] = str(ctx["duplicate_group_id"]) + + if identity_lookup and source_path: + id_ctx = identity_lookup.get(source_path) or identity_lookup.get( + str(Path(source_path).resolve()) + ) + if id_ctx: + for key in ("title", "category", "season", "episode", "quality_hint"): + if id_ctx.get(key) and not enriched.get(key): + enriched[key] = id_ctx[key] + + if path_to_duplicate_group and source_path: + gid = path_to_duplicate_group.get(source_path) + if gid and not enriched.get("duplicate_group_id"): + enriched["duplicate_group_id"] = gid + + return enriched + + +def enrich_review_rows( + rows: list[dict[str, str]], + plan: ExecutionPlan, + *, + library_root: Path | None = None, + identity_lookup: dict[str, dict[str, str]] | None = None, + path_to_duplicate_group: dict[str, str] | None = None, +) -> list[dict[str, str]]: + """Enrich all review rows using plan operations and optional lookups.""" + op_by_index = {i + 1: op for i, op in enumerate(plan.operations)} + result: list[dict[str, str]] = [] + for row in rows: + idx = int(row["index"]) + op = op_by_index.get(idx) + result.append( + enrich_review_row( + row, + library_root=library_root, + identity_lookup=identity_lookup, + path_to_duplicate_group=path_to_duplicate_group, + operation=op, + ) + ) + return result + + def review_plan( plan: ExecutionPlan, season_threshold: int = 20, @@ -70,6 +313,12 @@ def review_plan( if op.has_conflict: flags.append("conflict") counters["conflicts"] += 1 + rc = op.review_context if hasattr(op, "review_context") else {} + if isinstance(rc, dict) and rc.get("duplicate_group_id") and op.operation_type == "quarantine": + if "duplicate" not in flags: + flags.append("duplicate") + elif "duplicate" in reason_l and "duplicate" not in flags: + flags.append("duplicate") if flags: counters["high_risk_operations"] += 1 rows.append( @@ -86,12 +335,158 @@ def review_plan( return rows, counters +def append_sample_safe_rows( + plan: ExecutionPlan, + rows: list[dict[str, str]], + sample_count: int, + *, + rng: random.Random | None = None, +) -> list[dict[str, str]]: + """Append random non-high-risk move operations for spot-checking.""" + if sample_count < 1: + return rows + + high_risk_indices = {int(r["index"]) for r in rows} + candidates: list[tuple[int, FileOperation]] = [] + for idx, op in enumerate(plan.operations, start=1): + if idx in high_risk_indices: + continue + if op.operation_type != "move": + continue + candidates.append((idx, op)) + + if not candidates: + return rows + + picker = rng or random.Random() + picked = picker.sample(candidates, min(sample_count, len(candidates))) + extra: list[dict[str, str]] = [] + for idx, op in sorted(picked, key=lambda x: x[0]): + extra.append( + { + "index": str(idx), + "operation_type": op.operation_type, + "risk_flags": "spot_check", + "source_path": str(op.source_path), + "destination_path": str(op.destination_path) if op.destination_path else "", + "reason": op.reason, + } + ) + return rows + extra + + +def group_review_rows( + rows: list[dict[str, str]], + group_by: str, +) -> list[dict[str, str]]: + """Reorder review rows for display/export grouping (does not change plan).""" + if group_by == "none" or not rows: + return rows + + def sort_key(row: dict[str, str]) -> tuple: + if group_by == "reason": + return (row.get("reason", ""), int(row.get("index", 0))) + if group_by == "title": + return (row.get("title", ""), row.get("duplicate_group_id", ""), int(row.get("index", 0))) + if group_by == "duplicate": + gid = row.get("duplicate_group_id", "") + if not gid: + gid = f"_ungrouped_{row.get('index', '')}" + return (gid, int(row.get("index", 0))) + return (int(row.get("index", 0)),) + + return sorted(rows, key=sort_key) + + +def check_review_requirements( + plan: ExecutionPlan, + plan_path: Path, + review_csv: Path, + *, + season_threshold: int = 20, + episode_threshold: int = 40, +) -> list[str]: + """Return human-readable errors when review is required but missing or stale.""" + _, counters = review_plan( + plan, + season_threshold=season_threshold, + episode_threshold=episode_threshold, + ) + high_risk = counters.get("high_risk_operations", 0) + if high_risk == 0: + return [] + + errors: list[str] = [] + if not review_csv.is_file(): + errors.append( + f"Review required: {high_risk} high-risk operation(s) but review CSV not found: {review_csv}. " + f"Run: vlm review-plan --input {plan_path}" + ) + return errors + + try: + if review_csv.stat().st_mtime < plan_path.stat().st_mtime: + errors.append( + f"Review CSV is older than plan ({review_csv}). " + f"Re-run review-plan and apply-review before execute --confirm." + ) + except OSError as exc: + errors.append(f"Could not compare review CSV timestamps: {exc}") + + return errors + + +def prepare_review_rows( + plan: ExecutionPlan, + *, + season_threshold: int = 20, + episode_threshold: int = 40, + library_root: Path | None = None, + identities_data: dict[str, Any] | None = None, + analysis_data: dict[str, Any] | None = None, + sample_safe: int = 0, + group_by: str = "none", +) -> tuple[list[dict[str, str]], dict[str, int]]: + """Build, enrich, optionally sample and group review rows.""" + rows, counters = review_plan( + plan, + season_threshold=season_threshold, + episode_threshold=episode_threshold, + ) + if sample_safe > 0: + rows = append_sample_safe_rows(plan, rows, sample_safe) + + identity_lookup = ( + build_identity_lookup(identities_data) if identities_data else None + ) + path_to_dup_group, _ = build_duplicate_path_maps(analysis_data) + rows = enrich_review_rows( + rows, + plan, + library_root=library_root, + identity_lookup=identity_lookup, + path_to_duplicate_group=path_to_dup_group, + ) + if group_by and group_by != "none": + rows = group_review_rows(rows, group_by) + return rows, counters + + def save_review_csv(rows: list[dict[str, str]], output: Path) -> None: """Write review rows to CSV.""" output.parent.mkdir(parents=True, exist_ok=True) - fields = ["index", "operation_type", "risk_flags", "source_path", "destination_path", "reason"] - with open(output, "w", encoding="utf-8", newline="") as f: - writer = csv.DictWriter(f, fieldnames=fields) + fieldnames = list(REVIEW_CSV_BASE_FIELDS) + for row in rows: + for key in row: + if key in REVIEW_CSV_ENRICHED_FIELDS and key not in fieldnames: + fieldnames.append(key) + for f in REVIEW_CSV_ENRICHED_FIELDS: + if f in fieldnames: + continue + if any(f in row for row in rows): + fieldnames.append(f) + + with open(output, "w", encoding="utf-8", newline="") as fh: + writer = csv.DictWriter(fh, fieldnames=fieldnames, extrasaction="ignore") writer.writeheader() writer.writerows(rows) - diff --git a/src/vlm/plan_structure_preview.py b/src/vlm/plan_structure_preview.py new file mode 100644 index 0000000..8be40cc --- /dev/null +++ b/src/vlm/plan_structure_preview.py @@ -0,0 +1,64 @@ +"""Build a target library tree preview from an execution plan.""" + +from __future__ import annotations + +from collections import defaultdict +from pathlib import Path + +from vlm.models import ExecutionPlan +from vlm.review_display import display_path + + +def build_structure_preview_lines( + plan: ExecutionPlan, + library_root: Path, + *, + max_titles: int = 40, + max_paths_per_title: int = 8, +) -> list[str]: + """Return ASCII lines showing where move/rename operations will place files.""" + by_title: dict[str, list[str]] = defaultdict(list) + + for op in plan.operations: + if op.operation_type not in ("move", "rename") or not op.destination_path: + continue + ctx = op.review_context if isinstance(op.review_context, dict) else {} + title = str(ctx.get("title") or op.destination_path.parent.parent.name) + rel = display_path(op.destination_path, library_root) + by_title[title].append(rel) + + lines = [ + "Target library structure preview (move/rename destinations)", + f"Library root: {library_root}", + "", + ] + if not by_title: + lines.append("(no move/rename destinations)") + return lines + + sorted_titles = sorted(by_title.keys())[:max_titles] + for title in sorted_titles: + paths = sorted(set(by_title[title]))[:max_paths_per_title] + lines.append(f"{title}/") + for p in paths: + lines.append(f" {p}") + remaining = len(by_title[title]) - len(paths) + if remaining > 0: + lines.append(f" ... +{remaining} more path(s)") + lines.append("") + + if len(by_title) > max_titles: + lines.append(f"... +{len(by_title) - max_titles} more title(s)") + + return lines + + +def write_structure_preview( + plan: ExecutionPlan, + library_root: Path, + output_path: Path, +) -> None: + """Write structure preview text to a file.""" + lines = build_structure_preview_lines(plan, library_root) + output_path.parent.mkdir(parents=True, exist_ok=True) + output_path.write_text("\n".join(lines) + "\n", encoding="utf-8") diff --git a/src/vlm/planner.py b/src/vlm/planner.py index 166c5f4..f43405e 100644 --- a/src/vlm/planner.py +++ b/src/vlm/planner.py @@ -9,12 +9,12 @@ import uuid from pathlib import Path from typing import Optional, Union +from dataclasses import replace + from vlm.config import Config from vlm.duplicate_resolve import DuplicateResolutionError, choose_keep_index -from vlm.io import ( - load_execution_plan, - save_execution_plan, -) +from vlm.io import load_execution_plan, save_execution_plan +from vlm.plan_review import build_duplicate_path_maps, build_review_context from vlm.utils import ( canonical_path_str, is_sample_path, @@ -143,6 +143,7 @@ def generate_plan( metadata: dict = {} validation_snapshot: dict[str, object] = {} duplicate_resolution_issues: list[dict[str, object]] = [] + duplicate_keep_paths: set[str] = set() if analysis_data is not None: validation_snapshot["analysis_source"] = analysis_data.get("metadata", {}).get("source_identities", "") validation_snapshot["duplicate_groups_considered"] = len(analysis_data.get("duplicates", [])) @@ -204,6 +205,7 @@ def generate_plan( if keep_idx is None: continue keep_identity_index = valid_indices[keep_idx] + duplicate_keep_paths.add(str(identities[keep_identity_index][0].path)) quarantine_indices = set(valid_indices) - {keep_identity_index} reason = _select_duplicate_quarantine_reason( config.duplicate_keep, @@ -240,6 +242,14 @@ def generate_plan( validation_snapshot["duplicate_resolution_issues"] = duplicate_resolution_issues metadata["duplicate_resolution_issues"] = duplicate_resolution_issues + path_to_dup_group, _ = build_duplicate_path_maps(analysis_data) + operations = _stamp_review_context_on_operations( + operations, + identities, + path_to_dup_group=path_to_dup_group, + duplicate_keep_paths=duplicate_keep_paths, + ) + if validation_snapshot: validation_snapshot["captured_at"] = utc_now().isoformat() metadata["validation_snapshot"] = validation_snapshot @@ -555,6 +565,36 @@ def _create_series_operation( ) +def _stamp_review_context_on_operations( + operations: list[FileOperation], + identities: list[tuple], + *, + path_to_dup_group: dict[str, str], + duplicate_keep_paths: set[str], +) -> list[FileOperation]: + """Attach review_context to file operations aligned with identity indices.""" + stamped: list[FileOperation] = [] + for i, op in enumerate(operations): + if i < len(identities): + vf, identity = identities[i] + if isinstance(identity, (MovieIdentity, SeriesIdentity)): + gid = path_to_dup_group.get(str(vf.path), "") + keep_candidate = None + if gid: + keep_candidate = str(vf.path) in duplicate_keep_paths + ctx = build_review_context( + vf.path, + identity, + category=vf.category, + duplicate_group_id=gid, + keep_candidate=keep_candidate, + ) + stamped.append(replace(op, review_context=ctx)) + continue + stamped.append(op) + return stamped + + def _generate_summary(operations: list[FileOperation]) -> dict: """Generate summary statistics for operations. @@ -691,7 +731,6 @@ def save_plan(plan: ExecutionPlan, output_path: Path) -> None: save_execution_plan(plan, output_path) - def load_plan(input_path: Path) -> ExecutionPlan: """Load execution plan from JSON file. diff --git a/src/vlm/review_display.py b/src/vlm/review_display.py index 7f4ce0b..c6f453c 100644 --- a/src/vlm/review_display.py +++ b/src/vlm/review_display.py @@ -26,7 +26,7 @@ def risk_flags_to_labels(flags: str, *, max_len: int = 24) -> str: return text[: max_len - 1] + "…" -def _display_path(p: Path, library_root: Path) -> str: +def display_path(p: Path, library_root: Path) -> str: try: resolved = p.resolve() root = library_root.resolve() @@ -44,10 +44,10 @@ def format_paths_for_detail( """Build multi-line before/after path text for review detail panes.""" source = Path(source_s) dest = Path(dest_s) if dest_s.strip() else None - src_line = _display_path(source, library_root) + src_line = display_path(source, library_root) lines = [f"来源: {src_line}", f"完整: {source}"] if dest is not None: - dst_line = _display_path(dest, library_root) + dst_line = display_path(dest, library_root) lines.append(f"目标: {dst_line}") lines.append(f"完整: {dest}") else: diff --git a/src/vlm/review_tui.py b/src/vlm/review_tui.py index 8745e1b..4ca68d6 100644 --- a/src/vlm/review_tui.py +++ b/src/vlm/review_tui.py @@ -2,9 +2,10 @@ from __future__ import annotations -from dataclasses import dataclass +from dataclasses import dataclass, field from pathlib import Path + from vlm.plan_review import save_review_csv from vlm.review_display import ( build_csv_rows, @@ -12,22 +13,28 @@ from vlm.review_display import ( review_row_status_symbol, risk_flags_to_labels, ) +from vlm.utils import format_size -try: - from textual import on - from textual.app import App, ComposeResult - from textual.binding import Binding - from textual.containers import Container, Horizontal, ScrollableContainer - from textual.geometry import Size - from textual.screen import ModalScreen, Screen - from textual.widgets import DataTable, Footer, Static -except ImportError as exc: # pragma: no cover - exercised via runtime fallback - TEXTUAL_IMPORT_ERROR: ImportError | None = exc -else: - TEXTUAL_IMPORT_ERROR = None +FILTER_CHOICES = ("all", "manual_review", "sample_source", "conflict", "duplicate") -MISSING_TEXTUAL_MESSAGE = 'Textual is not installed. Install with: uv pip install -e ".[tui]"' +def _row_matches_filter(row: dict[str, str], filter_key: str) -> bool: + if filter_key == "all": + return True + flags = row.get("risk_flags", "") + if filter_key in flags.split("|"): + return True + if filter_key == "duplicate" and row.get("duplicate_group_id"): + return True + return False + + +def _change_label(row: dict[str, str]) -> str: + src = row.get("source_name") or Path(row.get("source_path", "")).name + dst = row.get("dest_name") or "" + if dst: + return f"{src} → {dst}" + return src @dataclass(frozen=True) @@ -40,8 +47,24 @@ class ReviewTUIContext: output_csv: Path plan_input: Path summary_text: str + path_to_quality: dict[str, dict] = field(default_factory=dict) +try: + from textual import on + from textual.app import App, ComposeResult + from textual.binding import Binding + from textual.containers import Container, Horizontal, ScrollableContainer + from textual.geometry import Size + from textual.screen import ModalScreen, Screen + from textual.widgets import DataTable, Footer, Static +except ImportError as exc: # pragma: no cover + TEXTUAL_IMPORT_ERROR: ImportError | None = exc +else: + TEXTUAL_IMPORT_ERROR = None + +MISSING_TEXTUAL_MESSAGE = 'Textual is not installed. Install with: uv pip install -e ".[tui]"' + if TEXTUAL_IMPORT_ERROR is None: def _index_from_row_key(row_key) -> int: # noqa: ANN001 - RowKey | str @@ -52,7 +75,6 @@ if TEXTUAL_IMPORT_ERROR is None: return int(row_key) return int(val) - class SummaryScreen(Screen): """Migration summary; Enter continues, q aborts.""" @@ -80,6 +102,7 @@ if TEXTUAL_IMPORT_ERROR is None: f"将要写入: {self._ctx.output_csv}", "", "默认仅审核标记为高危的操作（与 CSV 行一致）。", + "筛选: 1全部 2需人工 3样片 4冲突 5重复 · g 驳回整组重复", "", "Enter 进入审核 · q 退出", ] @@ -139,8 +162,14 @@ if TEXTUAL_IMPORT_ERROR is None: Binding("a", "keep_row", "保留", show=True), Binding("r", "reject_row", "驳回", show=True), Binding("u", "undo_row", "撤销", show=True), + Binding("g", "reject_group", "驳回组", show=True), Binding("s", "save", "保存", show=True), Binding("q", "request_quit", "退出", show=True), + Binding("1", "filter_all", show=False), + Binding("2", "filter_manual", show=False), + Binding("3", "filter_sample", show=False), + Binding("4", "filter_conflict", show=False), + Binding("5", "filter_duplicate", show=False), ] def __init__(self, ctx: ReviewTUIContext) -> None: @@ -153,6 +182,7 @@ if TEXTUAL_IMPORT_ERROR is None: int(r["index"]): r["operation_type"] for r in ctx.rows } self.op_by_index: dict[int, str] = dict(self.initial_op_by_index) + self._filter = "all" def compose(self) -> ComposeResult: plan_s = str(self.ctx.plan_input) @@ -163,12 +193,13 @@ if TEXTUAL_IMPORT_ERROR is None: f" / {self.ctx.counters['total_operations']}" ) yield Static(hdr, id="header_line") + yield Static("", id="filter_line") with Horizontal(id="body"): yield DataTable(id="review_table", cursor_type="row", zebra_stripes=True) with ScrollableContainer(id="detail_scroll"): yield Static("", id="detail_text") yield Static( - "↑↓ j/k 移动 · a 保留 · r 驳回(no-op) · u 撤销本条 · s 保存退出 · q 退出", + "1-5 筛选 · ↑↓ j/k · a 保留 · r 驳回 · g 驳回重复组 · u 撤销 · s 保存 · q 退出", id="footer_line", ) yield Footer() @@ -180,6 +211,12 @@ if TEXTUAL_IMPORT_ERROR is None: background: $primary-darken-2; color: $text; } + #filter_line { + dock: top; + padding: 0 1; + background: $panel; + color: $text-muted; + } #footer_line { dock: bottom; padding: 0 1; @@ -214,35 +251,58 @@ if TEXTUAL_IMPORT_ERROR is None: } """ - def on_mount(self) -> None: - table = self.query_one("#review_table", DataTable) - table.cursor_type = "row" + def _filtered_rows(self) -> list[dict[str, str]]: + return [ + r + for r in self.ctx.rows + if _row_matches_filter(r, self._filter) + ] + + def _update_filter_line(self) -> None: + labels = { + "all": "全部", + "manual_review": "需人工", + "sample_source": "样片", + "conflict": "冲突", + "duplicate": "重复", + } + visible = len(self._filtered_rows()) + self.query_one("#filter_line", Static).update( + f"筛选: {labels.get(self._filter, self._filter)} · 显示 {visible}/{len(self.ctx.rows)} 条" + ) + + def _rebuild_table(self) -> None: + table = self._table() + table.clear(columns=True) table.add_column(" ", key="sym", width=3) table.add_column("#", key="idx", width=4) - table.add_column("类型", key="op", width=11) - table.add_column("风险", key="risk", width=18) - table.add_column("文件", key="file") + table.add_column("类型", key="op", width=10) + table.add_column("风险", key="risk", width=14) + table.add_column("变更", key="change") - for r in self.ctx.rows: + for r in self._filtered_rows(): idx = int(r["index"]) - sym = self._symbol_for(idx) table.add_row( - sym, + self._symbol_for(idx), str(idx), self.op_by_index[idx], risk_flags_to_labels(r["risk_flags"]), - Path(r["source_path"]).name, + _change_label(r), key=str(idx), ) - self._apply_body_layout(self.app.size) + self._update_filter_line() if table.row_count > 0: table.focus() self._refresh_detail(_index_from_row_key(table.ordered_rows[0].key)) else: self.query_one("#detail_text", Static).update( - "无高危项。按 s 保存仅含表头的 CSV（与无 --tui 行为一致）。" + "当前筛选无条目。按 1 显示全部，或 s 保存 CSV。" ) + def on_mount(self) -> None: + self._rebuild_table() + self._apply_body_layout(self.app.size) + def _table(self) -> DataTable: return self.query_one("#review_table", DataTable) @@ -278,6 +338,33 @@ if TEXTUAL_IMPORT_ERROR is None: idx = _index_from_row_key(event.row_key) self._refresh_detail(idx) + def _format_quality_block(self, index: int) -> str: + r = self._by_index[index] + gid = r.get("duplicate_group_id", "") + if not gid: + return "" + + lines = [f"重复组: {gid}"] + group_paths = [ + other + for other in self.ctx.rows + if other.get("duplicate_group_id") == gid + ] + for other in group_paths: + oidx = int(other["index"]) + path = other.get("source_path", "") + qc = self.ctx.path_to_quality.get(path, {}) + if qc: + res = qc.get("resolution", "?") + size = format_size(int(qc.get("size_bytes", 0) or 0)) + mark = "★" if self.op_by_index.get(oidx) != "no-op" else " " + lines.append(f" {mark} [{oidx}] {Path(path).name}: {res} {size}") + else: + hint = other.get("quality_hint", "") + mark = "★" if self.op_by_index.get(oidx) != "no-op" else " " + lines.append(f" {mark} [{oidx}] {Path(path).name}: {hint or '(no metadata)'}") + return "\n".join(lines) + def _refresh_detail(self, index: int) -> None: r = self._by_index[index] op = self.op_by_index[index] @@ -287,23 +374,29 @@ if TEXTUAL_IMPORT_ERROR is None: self.ctx.library_root, ) risk_cn = risk_flags_to_labels(r["risk_flags"], max_len=120) + title = r.get("title", "") summary = ( - f"#{index} · {op} · {Path(r['source_path']).name}" + f"#{index} · {op} · {_change_label(r)}" + + (f" · {title}" if title else "") + (f" · {risk_cn}" if risk_cn else "") ) + quality_block = self._format_quality_block(index) text = ( f"{summary}\n\n" f"变更\n{paths}\n\n" f"依据\n{r['reason']}\n\n" f"标记\n{r['risk_flags']}" ) + if quality_block: + text += f"\n\n画质对比\n{quality_block}" self.query_one("#detail_text", Static).update(text) def _refresh_row_cells(self, index: int) -> None: table = self._table() key = str(index) - sym = self._symbol_for(index) - table.update_cell(key, "sym", sym) + if key not in [str(_index_from_row_key(r.key)) for r in table.ordered_rows]: + return + table.update_cell(key, "sym", self._symbol_for(index)) table.update_cell(key, "op", self.op_by_index[index]) def _update_dirty_header(self) -> None: @@ -347,9 +440,43 @@ if TEXTUAL_IMPORT_ERROR is None: self._refresh_detail(idx) self._update_dirty_header() + def action_reject_group(self) -> None: + idx = self._current_index() + if idx is None: + return + gid = self._by_index[idx].get("duplicate_group_id", "") + if not gid: + self.action_reject_row() + return + for r in self.ctx.rows: + if r.get("duplicate_group_id") == gid: + self.op_by_index[int(r["index"])] = "no-op" + self._rebuild_table() + self._update_dirty_header() + def action_undo_row(self) -> None: self.action_keep_row() + def action_filter_all(self) -> None: + self._filter = "all" + self._rebuild_table() + + def action_filter_manual(self) -> None: + self._filter = "manual_review" + self._rebuild_table() + + def action_filter_sample(self) -> None: + self._filter = "sample_source" + self._rebuild_table() + + def action_filter_conflict(self) -> None: + self._filter = "conflict" + self._rebuild_table() + + def action_filter_duplicate(self) -> None: + self._filter = "duplicate" + self._rebuild_table() + def action_save(self) -> None: out_rows = build_csv_rows(self.ctx.rows, self.op_by_index) save_review_csv(out_rows, self.ctx.output_csv) @@ -402,3 +529,4 @@ else: def run_plan_review_tui(ctx: ReviewTUIContext) -> int: """Raise a friendly error when the optional Textual dependency is missing.""" raise RuntimeError(MISSING_TEXTUAL_MESSAGE) from TEXTUAL_IMPORT_ERROR + diff --git a/tests/test_cli_review_plan.py b/tests/test_cli_review_plan.py index 2d307ca..aa85759 100644 --- a/tests/test_cli_review_plan.py +++ b/tests/test_cli_review_plan.py @@ -104,6 +104,10 @@ def test_review_plan_generates_csv_summary_and_preview(tmp_path): assert "source:" in result.output assert "destination:" in result.output assert "reason:" in result.output + assert "Verdict: BLOCKED" in result.output + assert "需人工判断" in result.output or "样片路径" in result.output + assert "Next steps:" in result.output + assert "apply-review" in result.output assert output_csv.exists() with open(output_csv, "r", encoding="utf-8", newline="") as f: diff --git a/tests/test_plan_review.py b/tests/test_plan_review.py new file mode 100644 index 0000000..c040716 --- /dev/null +++ b/tests/test_plan_review.py @@ -0,0 +1,124 @@ +"""Tests for plan review helpers.""" + +import json +from pathlib import Path + +from vlm.models import ExecutionPlan, FileOperation +from vlm.plan_review import ( + build_identity_lookup, + check_review_requirements, + enrich_review_rows, + prepare_review_rows, + review_plan, +) +from vlm.plan_render import render_review_verdict +from vlm.plan_structure_preview import build_structure_preview_lines +from vlm.planner import load_plan, save_plan +from vlm.utils import utc_now + + +def _minimal_plan(operations: list[FileOperation]) -> ExecutionPlan: + return ExecutionPlan( + plan_id="t", + created_at=utc_now(), + operations=operations, + summary={"total": len(operations), "move": 1, "rename": 0, "quarantine": 0, "no-op": 0}, + ) + + +def test_render_review_verdict_blocked_and_ok(): + assert "BLOCKED" in render_review_verdict({"high_risk_operations": 2}) + assert "OK TO DRY-RUN" in render_review_verdict({"high_risk_operations": 0}) + + +def test_build_identity_lookup_and_enrich(tmp_path): + identities = { + "vlm_schema_version": "1.0", + "movies": [], + "series": [ + { + "path": str(tmp_path / "dl/Show.S01E01.mkv"), + "filename": "Show.S01E01.mkv", + "category": "series", + "title": "Show", + "season": 1, + "episodes": [1], + "confidence": 1.0, + "needs_review": False, + "video_metadata": {"resolution": "1080p", "size_bytes": 1000}, + } + ], + } + lookup = build_identity_lookup(identities) + op = FileOperation( + operation_type="move", + source_path=tmp_path / "dl/Show.S01E01.mkv", + destination_path=tmp_path / "lib/series/Show/Season 01/S01E01.mkv", + reason="organize", + has_conflict=False, + review_context={"title": "Show", "category": "series", "season": 1, "episode": 1}, + ) + plan = _minimal_plan([op]) + rows = [ + { + "index": "1", + "operation_type": "move", + "risk_flags": "", + "source_path": str(op.source_path), + "destination_path": str(op.destination_path), + "reason": op.reason, + } + ] + enriched = enrich_review_rows(rows, plan, identity_lookup=lookup, library_root=tmp_path / "lib") + assert enriched[0]["title"] == "Show" + assert enriched[0]["quality_hint"] + assert enriched[0]["source_name"] == "Show.S01E01.mkv" + + +def test_check_review_requirements_missing_csv(tmp_path): + plan_path = tmp_path / "plan.json" + op = FileOperation( + operation_type="no-op", + source_path=tmp_path / "Series.S20E01.mkv", + destination_path=None, + reason="Series needs manual review (season exceeds configured threshold)", + has_conflict=False, + ) + plan = _minimal_plan([op]) + save_plan(plan, plan_path) + loaded = load_plan(plan_path) + errors = check_review_requirements(loaded, plan_path, tmp_path / "missing.csv") + assert len(errors) == 1 + assert "not found" in errors[0] + + +def test_structure_preview_lines(tmp_path): + lib = tmp_path / "library" + op = FileOperation( + operation_type="move", + source_path=tmp_path / "a.mkv", + destination_path=lib / "series/Show/Season 01/S01E01.mkv", + reason="x", + has_conflict=False, + review_context={"title": "Show"}, + ) + lines = build_structure_preview_lines(_minimal_plan([op]), lib) + assert any("Show" in line for line in lines) + + +def test_prepare_review_rows_sample_safe(tmp_path): + ops = [ + FileOperation( + operation_type="move", + source_path=tmp_path / f"f{i}.mkv", + destination_path=tmp_path / f"lib/f{i}.mkv", + reason="organize", + has_conflict=False, + ) + for i in range(5) + ] + plan = _minimal_plan(ops) + rows, counters = prepare_review_rows(plan, sample_safe=2) + assert counters["high_risk_operations"] == 0 + assert len(rows) == 2 + assert all(r["risk_flags"] == "spot_check" for r in rows)