Update analysis and plan files to enhance duplicate handling and reporting

- Updated `analysis.json` with a new generation timestamp.
- Modified `plan.json` to include a new plan ID and created timestamp, and changed operation types from "no-op" to "quarantine" for specific files needing manual review.
- Enhanced the README.md to document the new `--analysis` option for generating execution plans, which now includes a human-readable summary and duplicate handling strategies.
- Introduced a new `duplicate_resolve.py` module to manage duplicate file resolution strategies.
- Improved the execution engine to support quarantine operations and added rollback functionality for quarantined files.

These changes improve the functionality of the Video Library Manager by providing better duplicate management and clearer reporting capabilities.

README.md

@@ -12,7 +12,8 @@ A Python-based CLI tool for managing personal video collections with a safety-fi
 - **Metadata Enrichment**: Add bilingual titles and reputation signals (TMDB + optional AI fallback)
 - **Incremental Performance**: SQLite-backed cache avoids repeated metadata lookups
 - **State Tracking**: Track file status throughout the workflow
-- **Detailed Reporting**: Generate inventory, completeness, and duplicate reports
+- **Detailed Reporting**: Generate inventory, completeness, and duplicate reports (reports can include plan content summary via `--plan`)
+- **Plan–Analysis Integration**: `vlm plan --analysis` applies duplicate resolution (keep by reputation, quarantine rest) and adds a Chinese human summary to the plan for quick review
 
 ## Installation
 
@@ -120,9 +121,21 @@ Create a reviewable plan of file operations:
 vlm plan
 ```
 
-This creates `plan.json` with proposed operations (move, rename, quarantine).
+To let the plan automatically resolve duplicate groups (keep one file per group by reputation, quarantine the rest), pass the analysis file:
 
-**Review the plan** by opening `plan.json` in your editor. You can edit it if needed.
+```bash
+vlm plan --analysis analysis.json
+```
+
+This creates `plan.json` with:
+- Proposed operations (move, rename, quarantine, no-op)
+- **Summary**: counts by operation type and by reason
+- **Human summary** (中文): short narrative for quick review
+- **Metadata**: when using `--analysis`, duplicate groups considered and completeness gaps
+
+Duplicate keep strategy is configurable in `~/.vlm/config.yaml` under `plan.duplicate_keep` (`by_reputation`, `first_seen`, or `manual`). Default is `by_reputation` (prefer external rating; fallback to first-seen).
+
+**Review the plan** by opening `plan.json` in your editor, or read the human summary when you run `vlm execute`. You can edit the plan JSON if needed.
 
 ### 7. Execute (Dry-Run First)
 
@@ -132,7 +145,7 @@ Preview what will happen without making changes:
 vlm execute
 ```
 
-When ready to actually move/rename files:
+Before running, the CLI prints the plan’s **human summary** (or a short summary from counts) so you can confirm at a glance. When ready to actually move/rename/quarantine files:
 
 ```bash
 vlm execute --confirm
@@ -175,9 +188,9 @@ vlm enrich
 vlm analyze
 # Output: analysis.json with 5 series with gaps, 12 duplicate groups
 
-# 6. Generate execution plan
-vlm plan
-# Output: plan.json with 456 operations proposed
+# 6. Generate execution plan (optionally use analysis for duplicate handling)
+vlm plan --analysis analysis.json
+# Output: plan.json with operations, human summary, and duplicate quarantine decisions
 
 # 7. Review the plan
 cat plan.json | less
@@ -271,8 +284,12 @@ vlm analyze --input my_identities.json --output my_analysis.json
 # Generate plan with default files
 vlm plan
 
-# Generate plan with custom files
+# Use analysis so duplicate groups become "keep one + quarantine rest" (by_reputation by default)
+vlm plan --analysis analysis.json
+
+# Custom input/output
 vlm plan --input my_identities.json --output my_plan.json
+vlm plan --input my_identities.json --analysis my_analysis.json --output my_plan.json
 ```
 
 ### Execution
@@ -327,8 +344,12 @@ vlm report inventory --format json --output inventory_report.json
 # Generate completeness report (series with gaps)
 vlm report completeness
 
+# Include plan content summary in the report (human_summary from plan.json)
+vlm report completeness --plan plan.json
+
 # Generate duplicates report
 vlm report duplicates
+vlm report duplicates --plan plan.json
 
 # Generate summary statistics
 vlm report summary
@@ -383,6 +404,11 @@ quarantine_dir: ".quarantine"
 # Logging level (DEBUG, INFO, WARNING, ERROR)
 log_level: "INFO"
 
+# Plan behavior (e.g. when using vlm plan --analysis)
+plan:
+  # Duplicate keep strategy: "by_reputation" (default), "first_seen", or "manual"
+  duplicate_keep: "by_reputation"
+
 # Category mappings (directory name to category)
 categories:
   movie: [movie, movies, films]
@@ -532,8 +558,9 @@ vlm enrich
 # 2. Analyze completeness
 vlm analyze
 
-# 3. View report
+# 3. View report (optionally include plan summary if you have a plan)
 vlm report completeness
+vlm report completeness --plan plan.json
 ```
 
 ### Scenario 3: Finding and Removing Duplicates
@@ -550,12 +577,18 @@ vlm analyze
 # 3. View duplicates with quality comparison
 vlm report duplicates
 
-# 4. Manually quarantine lower quality files
+# 4. Generate plan with analysis: VLM keeps one file per duplicate group (by reputation) and quarantines the rest
+vlm plan --analysis analysis.json
+
+# 5. Review plan (human summary in plan.json and when you run execute)
+vlm execute
+vlm execute --confirm
+
+# Alternatively: manual quarantine without plan
 vlm quarantine add /path/to/lower/quality/file.mkv --reason "duplicate - lower quality"
 
-# 5. Or generate plan and let VLM suggest operations
-vlm plan
-vlm execute
+# Report with plan context
+vlm report duplicates --plan plan.json
 ```
 
 ### Scenario 4: Reorganizing Your Library
@@ -726,10 +759,11 @@ src/vlm/
 ├── providers/          # External metadata providers (TMDB, etc.)
 │   ├── base.py         # Provider interface
 │   └── tmdb.py         # TMDB API client
-├── io.py               # JSON/CSV load/save and plan/analysis input helpers
+├── io.py               # JSON/CSV load/save, load_analysis_json, plan/analysis input helpers
 ├── utils.py            # UTC time, format_size, etc.
 ├── analysis.py         # Completeness and duplicate detection
-├── planner.py          # Execution plan generation
+├── duplicate_resolve.py # Duplicate group keep-index (by_reputation, first_seen, manual)
+├── planner.py          # Execution plan generation (optionally consumes analysis)
 ├── executor.py         # File operations and rollback
 ├── quarantine.py       # Quarantine management
 ├── state.py            # File state tracking

analysis.json

@@ -1,6 +1,6 @@
 {
   "metadata": {
-    "generated": "2026-02-10T08:49:31",
+    "generated": "2026-02-10T09:54:17",
     "source_identities": "identities.json",
     "total_movies": 637,
     "total_series": 1492

src/vlm/cli.py

@@ -600,8 +600,14 @@ def analyze(ctx: CLIContext, input: Path, output: Path, inventory: Optional[Path
     default=Path('plan.json'),
     help='Path to save execution plan (default: 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):
+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.
@@ -611,11 +617,12 @@ def plan(ctx: CLIContext, input: Path, output: Path):
     
         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:
         from vlm.commands.plan import plan_cmd
-        plan_cmd(ctx, input, output)
+        plan_cmd(ctx, input, output, analysis)
     except FileNotFoundError:
         click.echo(f"Error: Input file not found: {input}", err=True)
         ctx.logger.error(f"Input file not found: {input}")
@@ -678,6 +685,17 @@ def execute(ctx: CLIContext, plan: Path, confirm: bool):
         click.echo(f"Execution plan loaded: {execution_plan.plan_id}")
         click.echo(f"Created at: {execution_plan.created_at}")
         click.echo(f"Total operations: {len(execution_plan.operations)}")
+        if execution_plan.human_summary:
+            click.echo()
+            click.echo(execution_plan.human_summary)
+        elif execution_plan.summary or execution_plan.summary_by_reason:
+            s = execution_plan.summary or {}
+            by_r = execution_plan.summary_by_reason or {}
+            parts = [f"操作统计：共 {s.get('total', len(execution_plan.operations))} 条（move {s.get('move', 0)}，rename {s.get('rename', 0)}，quarantine {s.get('quarantine', 0)}，no-op {s.get('no-op', 0)}）"]
+            if by_r:
+                parts.append("原因分布：" + "；".join(f"{r}: {c}" for r, c in list(by_r.items())[:5]))
+            click.echo()
+            click.echo("\n".join(parts))
         click.echo()
         
         # Display mode warning
@@ -697,7 +715,7 @@ def execute(ctx: CLIContext, plan: Path, confirm: bool):
         click.echo()
         
         # Create execution engine and execute plan
-        engine = ExecutionEngine(logger=logger)
+        engine = ExecutionEngine(logger=logger, config=config)
         results, summary, rollback_log = engine.execute_plan(
             execution_plan,
             mode=mode,
@@ -1036,7 +1054,7 @@ def rollback(ctx: CLIContext, log: Optional[Path]):
         click.echo()
         
         # Create execution engine
-        engine = ExecutionEngine(logger=logger)
+        engine = ExecutionEngine(logger=logger, config=config)
         
         # Load rollback log
         rollback_log = engine.load_rollback_log(log)
@@ -1115,6 +1133,20 @@ def rollback(ctx: CLIContext, log: Optional[Path]):
         sys.exit(1)
 
 
+def _fallback_plan_summary(execution_plan) -> str:
+    """Build a short plan summary from summary and summary_by_reason when human_summary is empty."""
+    s = execution_plan.summary or {}
+    by_r = execution_plan.summary_by_reason or {}
+    total = s.get("total", len(execution_plan.operations))
+    parts = [
+        f"计划操作统计：共 {total} 条（move {s.get('move', 0)}，rename {s.get('rename', 0)}，"
+        f"quarantine {s.get('quarantine', 0)}，no-op {s.get('no-op', 0)}）"
+    ]
+    if by_r:
+        parts.append("原因分布：" + "；".join(f"{r}: {c}" for r, c in list(by_r.items())[:8]))
+    return "\n".join(parts)
+
+
 @main.group()
 @pass_context
 def report(ctx: CLIContext):
@@ -1263,8 +1295,14 @@ def report_inventory(ctx: CLIContext, format: str, input: Path, output: Optional
     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]):
+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.
@@ -1273,15 +1311,25 @@ def report_completeness(ctx: CLIContext, format: str, input: Path, output: Optio
     
         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
     """
     import json
     from vlm.reports import generate_completeness_report
     from vlm.models import SeasonCompleteness
+    from vlm.planner import load_plan
 
     config = ctx.config
     logger = ctx.logger
 
+    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}")
@@ -1307,7 +1355,9 @@ def report_completeness(ctx: CLIContext, format: str, input: Path, output: Optio
         
         # Generate report
         click.echo(f"Generating completeness report in {format} format...")
-        report_content = generate_completeness_report(season_completeness, format, config.library_root)
+        report_content = generate_completeness_report(
+            season_completeness, format, config.library_root, plan_summary=plan_summary
+        )
         
         # Output report
         if output:
@@ -1358,8 +1408,14 @@ def report_completeness(ctx: CLIContext, format: str, input: Path, output: Optio
     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]):
+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
@@ -1369,16 +1425,26 @@ def report_duplicates(ctx: CLIContext, format: str, input: Path, output: Optiona
     
         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
     """
     import json
     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
     
     config = ctx.config
     logger = ctx.logger
 
+    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}")
@@ -1439,7 +1505,9 @@ def report_duplicates(ctx: CLIContext, format: str, input: Path, output: Optiona
         
         # Generate report
         click.echo(f"Generating duplicate report in {format} format...")
-        report_content = generate_duplicate_report(duplicate_groups, format, config.library_root)
+        report_content = generate_duplicate_report(
+            duplicate_groups, format, config.library_root, plan_summary=plan_summary
+        )
         
         # Output report
         if output:

src/vlm/commands/plan.py

@@ -1,20 +1,24 @@
 """Plan command implementation."""
 
+import json
 from pathlib import Path
+from typing import Optional
 
 import click
 
 from vlm.context import CLIContext
-from vlm.io import identities_to_plan_input, load_identities_json
+from vlm.io import identities_to_plan_input, load_identities_json, load_analysis_json
 from vlm.planner import generate_plan, save_plan
 
 
-def plan_cmd(ctx: CLIContext, input: Path, output: Path) -> None:
-    """Generate execution plan from identities."""
+def plan_cmd(ctx: CLIContext, input: Path, output: Path, analysis: Optional[Path] = None) -> None:
+    """Generate execution plan from identities; optionally use analysis for duplicate handling."""
     config = ctx.config
     logger = ctx.logger
 
     click.echo(f"Generating execution plan from: {input}")
+    if analysis:
+        click.echo(f"Using analysis: {analysis}")
     click.echo()
 
     identities_data = load_identities_json(input)
@@ -31,8 +35,21 @@ def plan_cmd(ctx: CLIContext, input: Path, output: Path) -> None:
 
     identities_list = identities_to_plan_input(identities_data)
 
+    analysis_data = None
+    if analysis:
+        if not analysis.exists():
+            click.echo(f"Error: Analysis file not found: {analysis}", err=True)
+            click.echo("Run 'vlm analyze' first to generate analysis.json.", err=True)
+            raise FileNotFoundError(analysis)
+        try:
+            analysis_data = load_analysis_json(analysis)
+        except (json.JSONDecodeError, KeyError) as e:
+            click.echo(f"Error: Invalid or incomplete analysis file: {e}", err=True)
+            click.echo("Run 'vlm analyze' to regenerate analysis.json.", err=True)
+            raise
+
     click.echo("Generating execution plan...")
-    execution_plan = generate_plan(identities_list, config)
+    execution_plan = generate_plan(identities_list, config, analysis_data=analysis_data)
 
     click.echo()
     click.echo("Plan generation complete!")

src/vlm/config.py

@@ -47,6 +47,9 @@ class Config:
     reputation_policy: str = "flag_for_review"
     naming_title_format: str = "{title_zh} {title_en}"
 
+    # Plan settings (e.g. duplicate handling when consuming analysis)
+    duplicate_keep: str = "by_reputation"
+
 
 def load_config(path: Path) -> Config:
     """Load configuration from YAML file."""
@@ -86,6 +89,9 @@ def load_config(path: Path) -> Config:
         "anime": ["anime"]
     })
 
+    plan = data.get("plan", {})
+    duplicate_keep = plan.get("duplicate_keep", "by_reputation")
+
     enrichment = data.get("enrichment")
     if enrichment is None:
         enrichment = data.get("enrich", {})
@@ -126,6 +132,7 @@ def load_config(path: Path) -> Config:
         reputation_low_score_threshold=reputation.get("low_score_threshold", 6.0),
         reputation_policy=reputation.get("policy", "flag_for_review"),
         naming_title_format=naming.get("title_format", "{title_zh} {title_en}"),
+        duplicate_keep=duplicate_keep,
     )
 
 
@@ -168,9 +175,12 @@ def create_default_config(path: Path) -> Config:
         },
     }
 
+    plan_content = {"duplicate_keep": default_config.duplicate_keep}
+
     yaml_content = {
         "library_root": str(default_config.library_root),
         "video_extensions": default_config.video_extensions,
+        "plan": plan_content,
         "templates": {
             "movie_dir": default_config.movie_template,
             "series_dir": default_config.series_template,
@@ -314,5 +324,9 @@ def validate_config(config: Config) -> list[str]:
         errors.append("tmdb_region must be a string when set")
     if not isinstance(config.tmdb_include_adult, bool):
         errors.append("tmdb_include_adult must be a boolean")
+    if config.duplicate_keep not in ("by_reputation", "first_seen", "manual"):
+        errors.append(
+            f"duplicate_keep must be one of 'by_reputation', 'first_seen', 'manual', got: {config.duplicate_keep!r}"
+        )
 
     return errors

src/vlm/duplicate_resolve.py

@@ -0,0 +1,51 @@
+"""Duplicate group resolution: choose which file to keep when consuming analysis."""
+
+from pathlib import Path
+from typing import Union
+
+from vlm.models import MovieIdentity, SeriesIdentity
+
+
+def choose_keep_index(
+    items: list[tuple[Path, Union[MovieIdentity, SeriesIdentity]]],
+    strategy: str,
+) -> int | None:
+    """Choose the index of the item to keep in a duplicate group.
+
+    Strategies:
+    - by_reputation: Prefer items with reputation_score; then sort by score desc,
+      then reputation_votes desc, then first_seen (input order). Keep index 0 after sort.
+    - first_seen: Keep the first item (index 0).
+    - manual: Return None; caller should not generate quarantine ops, only record in metadata.
+
+    Args:
+        items: List of (path, identity) for the duplicate group.
+        strategy: One of "by_reputation", "first_seen", "manual".
+
+    Returns:
+        Index in [0, len(items)) to keep, or None for manual.
+    """
+    if strategy == "manual":
+        return None
+    if strategy == "first_seen":
+        return 0
+    if strategy == "by_reputation":
+        return _by_reputation_index(items)
+    return 0
+
+
+def _by_reputation_index(
+    items: list[tuple[Path, Union[MovieIdentity, SeriesIdentity]]],
+) -> int:
+    """Sort by: has reputation > no reputation; then score desc; then votes desc; then order. Return 0."""
+    def key(idx_reason: tuple[int, tuple[Path, Union[MovieIdentity, SeriesIdentity]]]) -> tuple:
+        idx, (path, identity) = idx_reason
+        has_rep = identity.reputation_score is not None
+        score = identity.reputation_score if identity.reputation_score is not None else -1.0
+        votes = identity.reputation_votes if identity.reputation_votes is not None else -1
+        # Prefer has reputation (True > False), then higher score, then higher votes, then lower index
+        return (not has_rep, -score, -votes, idx)
+
+    indexed = list(enumerate(items))
+    indexed.sort(key=key)
+    return indexed[0][0]

src/vlm/executor.py

@@ -14,21 +14,32 @@ from pathlib import Path
 from typing import Optional
 from uuid import uuid4
 
+from .config import Config
 from .logging_config import get_logger, log_operation
 from .models import ExecutionPlan, FileOperation, OperationResult, RollbackLog
+from .quarantine import QuarantineManager
 from .utils import ensure_utc, utc_now
 
 
 class ExecutionEngine:
     """Engine for executing file operations safely with dry-run support."""
 
-    def __init__(self, logger: Optional[logging.Logger] = None):
+    def __init__(
+        self,
+        logger: Optional[logging.Logger] = None,
+        config: Optional[Config] = None,
+    ):
         """Initialize the execution engine.
 
         Args:
             logger: Optional logger instance (uses default if not provided)
+            config: Optional config (required for quarantine operations)
         """
         self.logger = logger or get_logger()
+        self.config = config
+        self._quarantine_manager: Optional[QuarantineManager] = (
+            QuarantineManager(config, self.logger) if config else None
+        )
     
     def execute_plan(
         self,
@@ -135,6 +146,39 @@ class ExecutionEngine:
                 executed_at=executed_at
             )
 
+        # Handle quarantine operations (no destination_path; use QuarantineManager)
+        if operation.operation_type == "quarantine":
+            if not self._quarantine_manager:
+                return OperationResult(
+                    operation=operation,
+                    success=False,
+                    error_message="Config required for quarantine operations",
+                    executed_at=executed_at
+                )
+            if mode == "dry-run":
+                log_operation(
+                    self.logger,
+                    logging.INFO,
+                    f"[DRY-RUN] Would quarantine: {operation.source_path} ({operation.reason})",
+                    operation_type="execute",
+                    file_path=operation.source_path
+                )
+                return OperationResult(
+                    operation=operation,
+                    success=True,
+                    error_message=None,
+                    executed_at=executed_at
+                )
+            result = self._quarantine_manager.quarantine_file(
+                operation.source_path, operation.reason
+            )
+            return OperationResult(
+                operation=operation,
+                success=result.success,
+                error_message=result.error_message,
+                executed_at=result.executed_at
+            )
+
         # Handle conflicted operations
         if operation.has_conflict:
             log_operation(
@@ -171,11 +215,16 @@ class ExecutionEngine:
         Returns:
             OperationResult indicating what would happen
         """
+        dest = operation.destination_path
+        msg = (
+            f"[DRY-RUN] Would quarantine: {operation.source_path} ({operation.reason})"
+            if operation.operation_type == "quarantine"
+            else f"[DRY-RUN] Would {operation.operation_type}: {operation.source_path} -> {dest}"
+        )
         log_operation(
             self.logger,
             logging.INFO,
-            f"[DRY-RUN] Would {operation.operation_type}: "
-            f"{operation.source_path} -> {operation.destination_path}",
+            msg,
             operation_type="execute",
             file_path=operation.source_path
         )
@@ -589,4 +638,3 @@ class ExecutionEngine:
             summary_msg,
             operation_type="rollback"
         )
-

src/vlm/io.py

@@ -17,11 +17,21 @@ __all__ = [
     "save_inventory_csv",
     "load_identities_json",
     "save_identities_json",
+    "load_analysis_json",
     "identities_to_plan_input",
     "identities_to_analysis_input",
 ]
 
 
+def load_analysis_json(path: Path) -> dict:
+    """Load analysis result from JSON file (metadata, completeness, duplicates).
+
+    Caller should check file existence and handle missing/invalid keys.
+    """
+    with open(path, "r", encoding="utf-8") as f:
+        return json.load(f)
+
+
 def load_identities_json(path: Path) -> dict:
     """Load identities from JSON file."""
     with open(path, "r", encoding="utf-8") as f:

src/vlm/models.py

@@ -134,11 +134,18 @@ class ExecutionPlan:
         created_at: Timestamp when the plan was created
         operations: List of file operations to execute
         summary: Dictionary with operation counts by type
+        summary_by_reason: Optional count per reason string (for human review)
+        human_summary: Optional short Chinese narrative summary of the plan
+        metadata: Optional dict (e.g. analysis_source, duplicate_groups_considered,
+            completeness_seasons_with_gaps) when plan was built from analysis
     """
     plan_id: str
     created_at: datetime
     operations: list[FileOperation]
     summary: dict
+    summary_by_reason: dict = field(default_factory=dict)
+    human_summary: str = ""
+    metadata: dict = field(default_factory=dict)
 
 
 @dataclass

src/vlm/planner.py

@@ -8,9 +8,10 @@ import json
 import uuid
 from datetime import datetime
 from pathlib import Path
-from typing import Union
+from typing import Optional, Union
 
 from vlm.config import Config
+from vlm.duplicate_resolve import choose_keep_index
 from vlm.utils import ensure_utc, utc_now
 from vlm.models import (
     ExecutionPlan,
@@ -20,38 +21,71 @@ from vlm.models import (
     VideoFile,
 )
 
+QUARANTINE_REASON_DUPLICATE = "重复项（已按外部评分保留一条）"
+
 
 def generate_plan(
     identities: list[tuple[VideoFile, Union[MovieIdentity, SeriesIdentity, None]]],
-    config: Config
+    config: Config,
+    analysis_data: Optional[dict] = None,
 ) -> ExecutionPlan:
-    """Generate an execution plan from parsed identities.
+    """Generate an execution plan from parsed identities; optionally apply analysis duplicates.
 
-    Creates file operations for organizing video files based on their parsed
-    identities and configuration templates. Handles movies, series, anime,
-    and other categories according to v1 constraints.
-    
-    Args:
-        identities: List of tuples containing (VideoFile, parsed_identity)
-        config: Configuration with templates and settings
-        
-    Returns:
-        ExecutionPlan with all file operations and summary
+    When analysis_data is provided and duplicate_keep is not "manual", duplicate groups
+    are resolved (one kept, rest quarantined) according to config.duplicate_keep.
     """
     operations = []
-    
     for video_file, identity in identities:
         operation = _create_operation(video_file, identity, config)
         operations.append(operation)
 
-    # Generate summary counts
+    metadata: dict = {}
+    if analysis_data is not None:
+        metadata["analysis_source"] = analysis_data.get("metadata", {}).get("source_identities", "")
+        metadata["duplicate_groups_considered"] = len(analysis_data.get("duplicates", []))
+        metadata["completeness_seasons_with_gaps"] = len(analysis_data.get("completeness", []))
+
+        if config.duplicate_keep != "manual":
+            path_to_index = {str(vf.path): i for i, (vf, _) in enumerate(identities)}
+            for dup in analysis_data.get("duplicates", []):
+                paths = dup.get("files", [])
+                indices = [path_to_index[p] for p in paths if p in path_to_index]
+                items = [
+                    (identities[i][0].path, identities[i][1])
+                    for i in indices
+                    if identities[i][1] is not None
+                    and isinstance(identities[i][1], (MovieIdentity, SeriesIdentity))
+                ]
+                if not items:
+                    continue
+                keep_idx = choose_keep_index(items, config.duplicate_keep)
+                if keep_idx is None:
+                    continue
+                keep_identity_index = indices[keep_idx]
+                quarantine_indices = set(indices) - {keep_identity_index}
+                for i in quarantine_indices:
+                    vf = identities[i][0]
+                    operations[i] = FileOperation(
+                        operation_type="quarantine",
+                        source_path=vf.path,
+                        destination_path=None,
+                        reason=QUARANTINE_REASON_DUPLICATE,
+                        has_conflict=False,
+                        conflict_reason=None,
+                    )
+
     summary = _generate_summary(operations)
+    summary_by_reason = _generate_summary_by_reason(operations)
+    human_summary = _generate_human_summary(operations, summary, summary_by_reason, metadata)
 
     return ExecutionPlan(
         plan_id=str(uuid.uuid4()),
         created_at=utc_now(),
         operations=operations,
-        summary=summary
+        summary=summary,
+        summary_by_reason=summary_by_reason,
+        human_summary=human_summary,
+        metadata=metadata,
     )
 
 
@@ -326,6 +360,46 @@ def _generate_summary(operations: list[FileOperation]) -> dict:
     return summary
 
 
+def _generate_summary_by_reason(operations: list[FileOperation]) -> dict[str, int]:
+    """Aggregate operation counts by reason string."""
+    out: dict[str, int] = {}
+    for op in operations:
+        r = op.reason
+        out[r] = out.get(r, 0) + 1
+    return out
+
+
+def _generate_human_summary(
+    operations: list[FileOperation],
+    summary: dict,
+    summary_by_reason: dict[str, int],
+    metadata: dict,
+) -> str:
+    """Build a short Chinese narrative summary of the plan."""
+    total = summary.get("total", len(operations))
+    move = summary.get("move", 0)
+    rename = summary.get("rename", 0)
+    quarantine = summary.get("quarantine", 0)
+    noop = summary.get("no-op", 0)
+    conflicts = sum(1 for op in operations if op.has_conflict)
+
+    parts = [
+        f"本计划共 {total} 条操作：move {move}，rename {rename}，quarantine {quarantine}，no-op {noop}。"
+    ]
+    if summary_by_reason:
+        reason_lines = [f"  - {r}: {c} 条" for r, c in sorted(summary_by_reason.items(), key=lambda x: -x[1])[:10]]
+        parts.append("原因分布：")
+        parts.extend(reason_lines)
+    if conflicts > 0:
+        parts.append(f"冲突 {conflicts} 条。")
+    if metadata:
+        dup = metadata.get("duplicate_groups_considered", 0)
+        gaps = metadata.get("completeness_seasons_with_gaps", 0)
+        if dup or gaps:
+            parts.append(f"依据 analysis：重复组 {dup} 个；剧集缺口 {gaps} 季。")
+    return "\n".join(parts)
+
+
 def save_plan(plan: ExecutionPlan, output_path: Path) -> None:
     """Save execution plan to JSON file.
     
@@ -351,7 +425,10 @@ def save_plan(plan: ExecutionPlan, output_path: Path) -> None:
             }
             for op in plan.operations
         ],
-        "summary": plan.summary
+        "summary": plan.summary,
+        "summary_by_reason": plan.summary_by_reason,
+        "human_summary": plan.human_summary,
+        "metadata": plan.metadata,
     }
     
     # Write to JSON file with indentation for human readability
@@ -398,5 +475,8 @@ def load_plan(input_path: Path) -> ExecutionPlan:
         plan_id=plan_dict["plan_id"],
         created_at=created_at,
         operations=operations,
-        summary=plan_dict["summary"]
+        summary=plan_dict["summary"],
+        summary_by_reason=plan_dict.get("summary_by_reason", {}),
+        human_summary=plan_dict.get("human_summary", ""),
+        metadata=plan_dict.get("metadata", {}),
     )

src/vlm/quarantine.py

@@ -510,6 +510,25 @@ class QuarantineManager:
         
         return entries
     
+    def find_quarantine_path_by_original(self, original_path: Path) -> Optional[Path]:
+        """Find the quarantine path for a file that was quarantined from original_path.
+        
+        Used when rolling back a quarantine operation when the rollback log only
+        has the original (source) path and not the actual quarantine destination.
+        
+        Args:
+            original_path: The original path of the file before quarantine
+        
+        Returns:
+            The path where the file was moved in quarantine, or None if not found
+        """
+        for category in ("movie", "series"):
+            manifest = self._load_manifest(category)
+            for entry in manifest.entries:
+                if entry.original_path == original_path:
+                    return entry.quarantine_path
+        return None
+    
     def restore_from_quarantine(
         self,
         quarantine_path: Path

src/vlm/reports.py

@@ -151,7 +151,8 @@ def _generate_inventory_json(
 def generate_completeness_report(
     analysis: list[SeasonCompleteness],
     format: str,
-    library_root: Path
+    library_root: Path,
+    plan_summary: str | None = None,
 ) -> str:
     """Generate completeness report showing series with episode gaps.
 
@@ -159,6 +160,7 @@ def generate_completeness_report(
         analysis: List of SeasonCompleteness objects with detected gaps
         format: Output format ("text" or "json")
         library_root: Root of the library (included in report metadata)
+        plan_summary: Optional plan content summary to prepend (when --plan was used)
 
     Returns:
         Formatted report as string
@@ -172,9 +174,19 @@ def generate_completeness_report(
     generation_timestamp = datetime.now(timezone.utc).strftime("%Y-%m-%dT%H:%M:%S")
 
     if format == "json":
-        return _generate_completeness_json(analysis, generation_timestamp, library_root)
-    else:  # text
-        return _generate_completeness_text(analysis, generation_timestamp, library_root)
+        body = _generate_completeness_json(analysis, generation_timestamp, library_root)
+    else:
+        body = _generate_completeness_text(analysis, generation_timestamp, library_root)
+
+    if plan_summary:
+        if format == "text":
+            section = "计划内容总结\n" + "-" * 40 + "\n" + plan_summary + "\n\n"
+            return section + body
+        else:
+            data = json.loads(body)
+            data["plan_summary"] = plan_summary
+            return json.dumps(data, indent=2, ensure_ascii=False)
+    return body
 
 
 def _generate_completeness_text(
@@ -264,7 +276,8 @@ def _generate_completeness_json(
 def generate_duplicate_report(
     duplicates: list[DuplicateGroup],
     format: str,
-    library_root: Path
+    library_root: Path,
+    plan_summary: str | None = None,
 ) -> str:
     """Generate duplicate report showing duplicate files with quality comparisons.
 
@@ -272,6 +285,7 @@ def generate_duplicate_report(
         duplicates: List of DuplicateGroup objects with duplicate files
         format: Output format ("text" or "json")
         library_root: Root of the library (included in report metadata)
+        plan_summary: Optional plan content summary to prepend (when --plan was used)
 
     Returns:
         Formatted report as string
@@ -285,9 +299,19 @@ def generate_duplicate_report(
     generation_timestamp = datetime.now(timezone.utc).strftime("%Y-%m-%dT%H:%M:%S")
 
     if format == "json":
-        return _generate_duplicate_json(duplicates, generation_timestamp, library_root)
-    else:  # text
-        return _generate_duplicate_text(duplicates, generation_timestamp, library_root)
+        body = _generate_duplicate_json(duplicates, generation_timestamp, library_root)
+    else:
+        body = _generate_duplicate_text(duplicates, generation_timestamp, library_root)
+
+    if plan_summary:
+        if format == "text":
+            section = "计划内容总结\n" + "-" * 40 + "\n" + plan_summary + "\n\n"
+            return section + body
+        else:
+            data = json.loads(body)
+            data["plan_summary"] = plan_summary
+            return json.dumps(data, indent=2, ensure_ascii=False)
+    return body
 
 
 def _generate_duplicate_text(

tests/test_executor.py

@@ -10,6 +10,7 @@ from uuid import uuid4
 
 import pytest
 
+from vlm.config import Config
 from vlm.executor import ExecutionEngine
 from vlm.models import ExecutionPlan, FileOperation
 
@@ -904,3 +905,43 @@ class TestRollbackExecution:
         actual_failed = sum(1 for r in rollback_results if not r.success)
         assert rollback_summary["successful"] == actual_success
         assert rollback_summary["failed"] == actual_failed
+
+    def test_rollback_quarantine_operation(self, tmp_path):
+        """Test that rollback restores quarantined files via QuarantineManager."""
+        library_root = tmp_path / "library"
+        library_root.mkdir()
+        (library_root / "movie").mkdir()
+        config = Config(library_root=library_root, quarantine_dir=".quarantine")
+        engine = ExecutionEngine(logger=logging.getLogger("test_executor"), config=config)
+
+        original_path = library_root / "movie" / "Duplicate (2020).mkv"
+        original_path.write_text("movie content")
+
+        plan = ExecutionPlan(
+            plan_id=str(uuid4()),
+            created_at=datetime.now(),
+            operations=[
+                FileOperation(
+                    operation_type="quarantine",
+                    source_path=original_path,
+                    destination_path=None,
+                    reason="duplicate",
+                    has_conflict=False,
+                    conflict_reason=None,
+                ),
+            ],
+            summary={"quarantine": 1},
+        )
+
+        results, _, rollback_log = engine.execute_plan(plan, mode="execute", confirmed=True)
+        assert results[0].success
+        assert not original_path.exists()
+        quarantine_path = config.library_root / "movie" / ".quarantine" / "Duplicate (2020).mkv"
+        assert quarantine_path.exists()
+
+        rollback_results, rollback_summary = engine.rollback(rollback_log)
+        assert rollback_summary["successful"] == 1
+        assert rollback_summary["failed"] == 0
+        assert original_path.exists()
+        assert original_path.read_text() == "movie content"
+        assert not quarantine_path.exists()