track uploaded files to prevent dupes. uses immich ID of original image

if_curator/cli.py

@@ -18,6 +18,7 @@ from .embeddings import is_embedding_available, load_embedding_model
 from .image_processing import process_face_mode, process_full_mode, process_object_mode
 from .immich_api import fetch_all_assets, fetch_face_data, fetch_full_image, filter_recent_assets, get_people
 from .logging import console, setup_logging
+from .upload_tracker import filter_already_uploaded, mark_uploaded
 
 logger = logging.getLogger(__name__)
 
@@ -92,8 +93,16 @@ def _configure_person(person: dict, people: list[dict]) -> dict | None:
 
     rprint(f"  Found [bold]{len(all_assets)}[/bold] total, [bold]{len(recent_assets)}[/bold] in range ({years} years).")
 
+    # Filter out assets already uploaded to Frigate
+    before_dedup = len(recent_assets)
+    new_asset_ids = set(filter_already_uploaded([a["id"] for a in recent_assets]))
+    recent_assets = [a for a in recent_assets if a["id"] in new_asset_ids]
+    skipped = before_dedup - len(recent_assets)
+    if skipped:
+        rprint(f"  [dim]Skipped {skipped} assets already uploaded to Frigate.[/dim]")
+
     if not recent_assets:
-        rprint("  [dim]Skipping (0 recent images).[/dim]")
+        rprint("  [dim]Skipping (0 new images after dedup).[/dim]")
         return None
 
     # Strategy selection
@@ -186,8 +195,16 @@ def auto_configure(people: list[dict]) -> list[dict]:
 
         rprint(f"  {name}: {len(all_assets)} total, {len(recent_assets)} recent")
 
+        # Filter out assets already uploaded to Frigate
+        before_dedup = len(recent_assets)
+        new_asset_ids = set(filter_already_uploaded([a["id"] for a in recent_assets]))
+        recent_assets = [a for a in recent_assets if a["id"] in new_asset_ids]
+        skipped = before_dedup - len(recent_assets)
+        if skipped:
+            rprint(f"  [dim]Skipped {skipped} assets already uploaded to Frigate.[/dim]")
+
         if not recent_assets:
-            rprint(f"  [dim]Skipping {name} (0 recent images).[/dim]")
+            rprint(f"  [dim]Skipping {name} (0 new images after dedup).[/dim]")
             continue
 
         has_embedding = is_embedding_available(entity_type)
@@ -219,7 +236,11 @@ def _resolve_strategy(strategy: str, has_embedding: bool) -> tuple[int | str, st
 
 
 def upload_to_frigate(jobs: list[dict]) -> None:
-    """Upload processed face crops to Frigate via API with detailed logging."""
+    """Upload processed face crops to Frigate via API with detailed logging.
+
+    After each successful upload, records the Immich asset ID in the
+    upload tracker so it is skipped on future runs.
+    """
     frigate_url = os.environ.get("FRIGATE_URL", "")
     if not frigate_url:
         rprint("[yellow]⚠️  FRIGATE_URL not set, skipping upload.[/yellow]")
@@ -228,17 +249,24 @@ def upload_to_frigate(jobs: list[dict]) -> None:
     rprint("\n[bold cyan]📤 Uploading to Frigate[/bold cyan]")
     rprint(f"  Target: [dim]{frigate_url}[/dim]")
 
-    # Count total files to upload
+    # Build a mapping of output filenames → Immich asset IDs
+    # from the asset_map stored on each job during execute_jobs()
+    filename_to_asset_id: dict[str, dict[str, str]] = {}
     total_files = 0
     for job in jobs:
         name = job["person"]["name"]
         person_dir = os.path.join(Config.OUTPUT_DIR, name)
         if not os.path.isdir(person_dir):
             continue
-        total_files += sum(
+        person_files = sorted(
-            1 for f in os.listdir(person_dir)
+            f for f in os.listdir(person_dir)
             if f.lower().endswith((".jpg", ".jpeg", ".png", ".webp"))
         )
+        total_files += len(person_files)
+
+        # Recover asset IDs from the job's asset_map
+        asset_map = job.get("asset_map", {})
+        filename_to_asset_id[name] = asset_map
 
     if total_files == 0:
         rprint("  [yellow]No images found to upload.[/yellow]")
@@ -279,6 +307,9 @@ def upload_to_frigate(jobs: list[dict]) -> None:
                 progress.console.print(f"  [dim]⏭️  {name}: no images found[/dim]")
                 continue
 
+            # Get the asset map for this person
+            asset_map = filename_to_asset_id.get(name, {})
+
             progress.console.print(f"  📁 {name}: uploading {len(person_files)} image(s)...")
             person_uploaded = 0
             person_failed = 0
@@ -299,6 +330,12 @@ def upload_to_frigate(jobs: list[dict]) -> None:
                             uploaded += 1
                             person_uploaded += 1
                             success = True
+
+                            # Mark this asset as uploaded so it's skipped on future runs
+                            asset_id = asset_map.get(fname)
+                            if asset_id:
+                                mark_uploaded(asset_id)
+
                             break
                         else:
                             if attempt < max_retries:
@@ -458,7 +495,11 @@ def _enrich_asset_with_face_data(asset: dict, person: dict) -> dict:
 
 
 def execute_jobs(jobs: list[dict]) -> None:
-    """Download and process images for all jobs."""
+    """Download and process images for all jobs.
+
+    Builds an asset_map per job (filename → Immich asset ID) so that
+    upload_to_frigate() can mark assets as uploaded after success.
+    """
     if not jobs:
         return
 
@@ -484,6 +525,9 @@ def execute_jobs(jobs: list[dict]) -> None:
             person_dir = os.path.join(Config.OUTPUT_DIR, name)
             os.makedirs(person_dir, exist_ok=True)
 
+            # Track filename → asset_id mapping for upload dedup
+            asset_map: dict[str, str] = {}
+
             count = 0
             for asset in assets:
                 try:
@@ -514,6 +558,15 @@ def execute_jobs(jobs: list[dict]) -> None:
                             else process_full_mode(img, person_dir, count)
                         )
                         if saved:
+                            # Record which asset produced which output file
+                            filename = f"{count}.jpg"
+                            asset_map[filename] = asset["id"]
+                            # Also record object-mode variant filenames
+                            if mode == "object":
+                                for f in sorted(os.listdir(person_dir)):
+                                    if f.startswith(f"{count}_") and f not in asset_map:
+                                        asset_map[f] = asset["id"]
+
                             count += 1
                         else:
                             progress.console.print(
@@ -525,6 +578,9 @@ def execute_jobs(jobs: list[dict]) -> None:
                 progress.advance(job_task)
                 progress.advance(overall_task)
 
+            # Store asset_map on the job so upload_to_frigate can use it
+            job["asset_map"] = asset_map
+
             progress.remove_task(job_task)
 
             # Log how many images were actually saved vs selected
@@ -582,3 +638,4 @@ def main() -> None:
 
 if __name__ == "__main__":
     main()
+

upload_tracker.py

@@ -0,0 +1,78 @@
+"""Persistent tracker for Immich asset IDs already uploaded to Frigate.
+
+Prevents duplicate uploads across runs by recording each successfully
+uploaded asset ID in a JSON file within the configured CACHE_DIR.
+"""
+
+import json
+import logging
+from pathlib import Path
+
+logger = logging.getLogger(__name__)
+
+UPLOAD_TRACKER_FILE = "frigate_uploaded_ids.json"
+
+
+def _tracker_path() -> Path:
+    """Return path to the tracker file, using Config.CACHE_DIR if available."""
+    try:
+        from .config import Config
+
+        return Path(Config.CACHE_DIR) / UPLOAD_TRACKER_FILE
+    except (ImportError, AttributeError):
+        return Path(UPLOAD_TRACKER_FILE)
+
+
+def load_uploaded_ids() -> set[str]:
+    """Load the set of Immich asset IDs already uploaded to Frigate."""
+    path = _tracker_path()
+    if not path.exists():
+        return set()
+    try:
+        with open(path) as f:
+            data = json.load(f)
+            return set(data.get("uploaded_asset_ids", []))
+    except (json.JSONDecodeError, OSError) as e:
+        logger.warning(f"Could not load upload tracker: {e}")
+        return set()
+
+
+def save_uploaded_ids(uploaded_ids: set[str]) -> None:
+    """Persist the set of uploaded Immich asset IDs to disk."""
+    path = _tracker_path()
+    path.parent.mkdir(parents=True, exist_ok=True)
+    with open(path, "w") as f:
+        json.dump({"uploaded_asset_ids": sorted(uploaded_ids)}, f, indent=2)
+
+
+def mark_uploaded(asset_id: str) -> None:
+    """Mark a single Immich asset ID as uploaded to Frigate."""
+    ids = load_uploaded_ids()
+    ids.add(asset_id)
+    save_uploaded_ids(ids)
+    logger.debug(f"Marked asset {asset_id} as uploaded to Frigate")
+
+
+def is_uploaded(asset_id: str) -> bool:
+    """Check if an Immich asset ID has already been uploaded to Frigate."""
+    return asset_id in load_uploaded_ids()
+
+
+def filter_already_uploaded(asset_ids: list[str]) -> list[str]:
+    """Return only asset IDs that have NOT yet been uploaded to Frigate.
+
+    Logs how many were skipped so the user knows dedup is working.
+    """
+    uploaded = load_uploaded_ids()
+    new_ids = [aid for aid in asset_ids if aid not in uploaded]
+    skipped = len(asset_ids) - len(new_ids)
+    if skipped:
+        logger.info(f"Skipping {skipped} assets already uploaded to Frigate")
+    return new_ids
+
+
+def reset_uploaded_ids() -> None:
+    """Clear the tracker — useful for re-training from scratch."""
+    save_uploaded_ids(set())
+    logger.info("Cleared Frigate upload tracker")
+