Detect and handle duplicate Immich people with the same name

When Immich has multiple person records sharing a name (e.g. unmerged
face clusters), winnow would previously run separate jobs for each,
with the second job wiping the first job's output directory — resulting
in far fewer training images than expected.

New behaviour:
- At startup, duplicate names are detected and a warning is printed
  showing asset counts for each duplicate.
- By default (MERGE_DUPLICATE_PEOPLE=false), only the person with the
  most assets is processed; smaller duplicates are skipped cleanly.
- With MERGE_DUPLICATE_PEOPLE=true, the duplicates are permanently
  merged inside Immich via PUT /api/people/{id}/merge (keeps the
  largest), then the people list is re-fetched before jobs run.

Also adds an explicit comment in executor.py confirming that replacement
targets come exclusively from tracker-mapped files, so manually-added
Frigate training images are never selected for deletion.

compose.yml

@@ -26,6 +26,7 @@ services:
       # - SKIP_PEOPLE=Unknown         # Comma-separated; skip these people
       # - MIN_FACE_COUNT=5            # Skip people with fewer than N assets in Immich
       # - YEARS_FILTER=10             # Only include images from the last N years (default: 10)
+      # - MERGE_DUPLICATE_PEOPLE=true # Auto-merge Immich people with the same name (keeps most assets)
 
       # ── Image Quality ─────────────────────────────────────────────────────
       # - MIN_FACE_WIDTH=50           # Minimum face width in pixels (default: 50)

winnow/cli.py

@@ -9,7 +9,7 @@ from rich.prompt import Confirm
 
 from .config import Config, ConfigManager
 from .executor import execute_jobs, upload_to_frigate
-from .immich_api import get_people
+from .immich_api import get_people, merge_people
 from .jobs import _show_preview, auto_configure, interactive_configure
 from .log_config import console, setup_logging
 from .upload_tracker import find_by_crop_dimension, get_person_summary, reset_person
@@ -51,6 +51,85 @@ def _handle_trace_crop(size_str: str) -> None:
     sys.exit(0)
 
 
+def _handle_duplicate_people(people: list[dict]) -> list[dict]:
+    """Warn about or merge Immich people that share the same name.
+
+    Duplicates arise when Immich creates separate person records for the same
+    individual (e.g. unmerged face clusters). Without handling, winnow would
+    run multiple jobs for the same Frigate folder and overwrite its own output,
+    leaving far fewer training images than expected.
+
+    With MERGE_DUPLICATE_PEOPLE=false (default): prints a warning, skips the
+    smaller duplicates so only the person with the most assets is processed,
+    and returns a deduplicated people list.
+    With MERGE_DUPLICATE_PEOPLE=true: merges each duplicate group inside
+    Immich via its API (permanently combines the face records), then
+    re-fetches the people list so the rest of the run sees the merged state.
+    """
+    from collections import defaultdict
+
+    by_name: dict[str, list[dict]] = defaultdict(list)
+    for p in people:
+        name = (p.get("name") or "").strip()
+        if name:
+            by_name[name].append(p)
+
+    duplicates = {name: ps for name, ps in by_name.items() if len(ps) > 1}
+    if not duplicates:
+        return people
+
+    if not Config.MERGE_DUPLICATE_PEOPLE:
+        rprint("\n[bold yellow]⚠  Duplicate person names detected in Immich:[/bold yellow]")
+        for name, ps in sorted(duplicates.items()):
+            ordered = sorted(ps, key=lambda x: x.get("assetCount", 0), reverse=True)
+            entries = ", ".join(
+                f"[dim]{p['id'][:8]}…[/dim] ({p.get('assetCount', 0)} assets)"
+                for p in ordered
+            )
+            rprint(f"  [yellow]{name}[/yellow] → {len(ps)} people: {entries}")
+            skipped = ordered[1:]
+            rprint(
+                f"  [dim]  Processing largest only "
+                f"({ordered[0].get('assetCount', 0)} assets). "
+                f"Skipping {len(skipped)} smaller duplicate(s) to avoid overwriting output.[/dim]"
+            )
+        rprint(
+            "  [dim]Set MERGE_DUPLICATE_PEOPLE=true to permanently merge duplicates "
+            "inside Immich (keeps the person with the most assets).[/dim]\n"
+        )
+        # Return deduplicated list — keep only the largest per name so that
+        # downstream job creation never runs two jobs for the same Frigate folder.
+        skip_ids = {
+            p["id"]
+            for ps in duplicates.values()
+            for p in sorted(ps, key=lambda x: x.get("assetCount", 0), reverse=True)[1:]
+        }
+        return [p for p in people if p["id"] not in skip_ids]
+
+    # Auto-merge: survivor = largest asset count, rest merge into it inside Immich
+    merged_any = False
+    for name, ps in sorted(duplicates.items()):
+        ordered = sorted(ps, key=lambda x: x.get("assetCount", 0), reverse=True)
+        survivor = ordered[0]
+        merge_ids = [p["id"] for p in ordered[1:]]
+        rprint(
+            f"  [cyan]Merging {name!r} inside Immich:[/cyan] keeping "
+            f"[dim]{survivor['id'][:8]}…[/dim] ({survivor.get('assetCount', 0)} assets), "
+            f"absorbing {len(merge_ids)} smaller duplicate(s)..."
+        )
+        if merge_people(survivor["id"], merge_ids):
+            rprint(f"  [green]✓ Merged {name!r}[/green]")
+            merged_any = True
+        else:
+            rprint(f"  [red]✗ Failed to merge {name!r}[/red]")
+
+    if merged_any:
+        rprint("  [dim]Re-fetching people after merge...[/dim]")
+        return get_people()
+
+    return people
+
+
 def main() -> None:
     """Entry point for winnow CLI."""
     try:
@@ -103,6 +182,8 @@ def main() -> None:
             rprint("[bold red]Could not fetch people from Immich. Check URL/Key.[/bold red]")
             return
 
+        people = _handle_duplicate_people(people)
+
         # Auto mode when no TTY (Docker, cron, pipes) — the primary use case.
         # A TTY means local interactive use; AUTO_MODE=true overrides that for scripting.
         auto_mode = not sys.stdin.isatty() or os.environ.get("AUTO_MODE", "").lower() in ("true", "1", "yes")

winnow/config.py

@@ -36,6 +36,7 @@ class _Config:
 
     # People filtering
     MIN_FACE_COUNT: int = 0
+    MERGE_DUPLICATE_PEOPLE: bool = False
 
     # Output quality
     FACE_MARGIN: float = 0.15
@@ -60,6 +61,7 @@ class _Config:
         self.YEARS_FILTER = int(os.getenv("YEARS_FILTER", "10"))
         self.MIN_FACE_WIDTH = int(os.getenv("MIN_FACE_WIDTH", "90"))
         self.MIN_FACE_COUNT = int(os.getenv("MIN_FACE_COUNT", "0"))
+        self.MERGE_DUPLICATE_PEOPLE = os.getenv("MERGE_DUPLICATE_PEOPLE", "false").lower() in ("true", "1", "yes")
         self.BLUR_THRESHOLD = float(os.getenv("BLUR_THRESHOLD", "120.0"))
         self.MIN_CONFIDENCE = float(os.getenv("MIN_CONFIDENCE", "0.7"))
         self.MAX_AUTO_IMAGES = int(os.getenv("MAX_AUTO_IMAGES", "80"))

winnow/executor.py

@@ -376,6 +376,8 @@ def upload_to_frigate(jobs: list[dict]) -> None:
             # Snapshot live Frigate files for post-upload reconciliation diff only.
             # effective_count is sourced from the tracker (mapped files) so that
             # manually-added Frigate files don't consume winnow's managed quota.
+            # Replacement targets also come exclusively from the tracker, so manually
+            # added files are never selected for deletion — only winnow-uploaded ones.
             _snapshot = (
                 all_frigate_files.get(name, []) if all_frigate_files is not None
                 else get_frigate_person_files(name)

winnow/immich_api.py

@@ -45,6 +45,26 @@ def get_people() -> list[dict]:
         return []
 
 
+def merge_people(survivor_id: str, merge_ids: list[str]) -> bool:
+    """Merge duplicate people into survivor via Immich's merge endpoint.
+
+    The survivor (identified by survivor_id) absorbs all faces and assets
+    from the people in merge_ids, which are then removed from Immich.
+    """
+    try:
+        resp = requests.put(
+            f"{Config.IMMICH_URL}/api/people/{survivor_id}/merge",
+            headers={**get_headers(), "Content-Type": "application/json"},
+            json={"ids": merge_ids},
+            timeout=30,
+        )
+        resp.raise_for_status()
+        return True
+    except requests.RequestException as e:
+        logger.error(f"Failed to merge people into {survivor_id}: {e}")
+        return False
+
+
 def fetch_all_assets(person: dict) -> list[dict]:
     """Fetch all assets for a person with pagination."""
     name = person.get("name", "Unknown")