Single-pass scan_once (fixes O(n²) re-walk + engine busy-loop); gracefully skips corrupt files

photo_ingest.py

@@ -1,15 +1,14 @@
-"""photo-pipeline: photo-ingest flow (v2 — batched + checkpointed).
+"""photo-pipeline: photo-ingest flow (v3 — single-pass, checkpointed).
 
-Hashes incoming images, checks against the persistent fingerprint DB
+Hashes incoming images, checks the persistent fingerprint DB (exact dups via
-(exact + near dupes), and registers new ones.
+sha256, multi-path dup tracking via known_paths), and registers new files.
 
-Batching: processes in chunks of `batch_size` (default 1000), committing to
+Design: ONE walk of the tree. Each file: path-known → skip fast; sha-known at
-the DB after each chunk. Checkpointing is DB-native: files already in the DB
+another path → record path in known_paths (dup); else register new. Inserts
-(by sha256) are skipped on resume — an interrupted run continues where it
+commit every `batch_size` rows (WAL) = checkpoint. Single pass = O(n), no
-stopped, never redoing work.
+re-walk, no O(n²), no engine busy-loop.
 
-For very large trees (e.g. 58K files), scan_directory can be slow to walk;
+Performance: dhash-only rows (~6ms/file); phash only when check_near_dups.
-use walk_files for a streaming generator when batch_size is set.
 """
 
 from pathlib import Path
@@ -27,7 +26,7 @@ def _is_image(p: Path) -> bool:
 
 
 def walk_files(base_dir: str):
-    """Stream image files under base_dir (generator — memory-safe for 50K+ files)."""
+    """Stream image files under base_dir (generator — memory-safe for 50K+)."""
     root = Path(base_dir)
     if not root.exists():
         raise FileNotFoundError(f"{root} does not exist")
@@ -37,168 +36,82 @@ def walk_files(base_dir: str):
 
 
 @task
-def find_unprocessed(base_dir: str, batch_size: int, source: str = None) -> list[str]:
+def scan_once(base_dir: str, source: str, batch_size: int, check_near_dups: bool) -> dict:
-    """Find the next batch of files NOT yet in the fingerprint DB."""
+    """One walk of the tree; classify + register in a single pass."""
+    import time
+
     db.init_db()
     conn = db.get_db()
     conn.execute(
         "CREATE TABLE IF NOT EXISTS known_paths (path TEXT PRIMARY KEY, sha256 TEXT NOT NULL)"
     )
-    batch = []
+    totals = {"exact_dups": 0, "near_dups": 0, "new": 0, "known": 0, "skipped": 0}
+    pending = []
+    t0 = time.time()
+
     for p_str in walk_files(base_dir):
-        # cheap check: path already seen (image_hashes OR known_paths)
-        known = conn.execute(
-            "SELECT 1 FROM image_hashes WHERE path=? UNION SELECT 1 FROM known_paths WHERE path=?",
-            (p_str, p_str),
-        ).fetchone()
-        if known:
-            continue
-        # content check: sha256 already registered (catches same photo at other paths)
-        sha = db.sha256_file(p_str)
-        sha_known = conn.execute(
-            "SELECT 1 FROM image_hashes WHERE sha256=?", (sha,)
-        ).fetchone()
-        if sha_known:
-            # record this path in known_paths so we don't re-hash it every loop
-            conn.execute(
-                "INSERT OR IGNORE INTO known_paths (path, sha256) VALUES (?,?)",
-                (p_str, sha))
-            continue
-        batch.append(p_str)
-        if len(batch) >= batch_size:
-            break
-    conn.commit()
-    conn.close()
-    print(f"find_unprocessed: {len(batch)} new files (batch_size={batch_size})")
-    return batch
-
-
-@task
-def check_and_register(image_paths: list[str], source: str, check_near_dups: bool = False) -> dict:
-    """Hash + dedup-check + register a batch. Returns verdict counts."""
-    db.init_db()
-    conn = db.get_db()
-    exact_dups = []
-    near_dups = []
-    new_images = []
-    for p_str in image_paths:
         p = Path(p_str)
         try:
-            # exact dup by sha
+            known = conn.execute(
-            sha = db.sha256_file(p)
+                "SELECT 1 FROM image_hashes WHERE path=? UNION SELECT 1 FROM known_paths WHERE path=?",
+                (p_str, p_str),
+            ).fetchone()
+            if known:
+                totals["known"] += 1
+                continue
+
+            sha = db.sha256_file(p_str)
             match = conn.execute(
-                "SELECT path, source FROM image_hashes WHERE sha256=?", (sha,)
+                "SELECT path FROM image_hashes WHERE sha256=?", (sha,)
             ).fetchone()
             if match:
-                exact_dups.append((p_str, match[0]))
+                conn.execute(
+                    "INSERT OR IGNORE INTO known_paths (path, sha256) VALUES (?,?)", (p_str, sha))
+                totals["exact_dups"] += 1
                 continue
-            # near dup by perceptual hash — FLAG but DO NOT skip (review decides)
+
-            # NOTE: hash_image is ~0.33s/file (PIL phash+dhash) and _near_dup_lookup
+            hx = db.hash_image(p)
-            # is O(n) over all rows — both expensive, so optional (default off)
-            if check_near_dups:
-                hx = db.hash_image(p, include_phash=True)
-                near, dist = db._near_dup_lookup(conn, hx)
-                if near and dist <= 10:
-                    near_dups.append((p_str, near, dist))
-            else:
-                hx = None
-            # register regardless (near-dup is a review hint, not a block)
             with Image.open(p) as im:
                 w, h = im.size
-            if hx is None:
+            pending.append((sha, hx["phash"], hx["dhash"], p.stat().st_size,
-                # still need hashes for the row — compute minimal (phash only)
+                            w, h, str(p), source))
-                hx = db.hash_image(p, include_phash=True)
+            totals["new"] += 1
-            conn.execute(
+
-                "INSERT INTO image_hashes (sha256, phash, dhash, file_size, width, height, path, source) "
+            if len(pending) >= batch_size:
-                "VALUES (?,?,?,?,?,?,?,?)",
+                conn.executemany(
-                (sha, hx["phash"], hx["dhash"], p.stat().st_size, w, h, str(p), source),
+                    "INSERT INTO image_hashes (sha256, phash, dhash, file_size, width, height, path, source) "
-            )
+                    "VALUES (?,?,?,?,?,?,?,?)", pending)
-            new_images.append(p_str)
+                conn.commit()
+                print(f"  checkpoint: {totals['new']} new ({time.time()-t0:.0f}s)", flush=True)
+                pending = []
+
         except Exception as e:
-            print(f"  SKIP {p.name}: {type(e).__name__}: {e}")
+            totals["skipped"] += 1
-    conn.commit()
+            print(f"  SKIP {p.name}: {type(e).__name__}: {e}", flush=True)
+
+    if pending:
+        conn.executemany(
+            "INSERT INTO image_hashes (sha256, phash, dhash, file_size, width, height, path, source) "
+            "VALUES (?,?,?,?,?,?,?,?)", pending)
+        conn.commit()
     conn.close()
-    return {
+    print(f"scan_once done in {time.time()-t0:.0f}s: {totals}", flush=True)
-        "total": len(image_paths),
-        "exact_dups": len(exact_dups),
-        "near_dups": len(near_dups),
-        "new": len(new_images),
-        "exact_dup_list": exact_dups[:20],
-        "near_dup_list": near_dups[:20],
-        "new_list": new_images,
-    }
-
-
-def _find_unprocessed(base_dir: str, batch_size: int) -> list[str]:
-    """Plain-function version of find_unprocessed (no Prefect task overhead)."""
-    db.init_db()
-    conn = db.get_db()
-    conn.execute(
-        "CREATE TABLE IF NOT EXISTS known_paths (path TEXT PRIMARY KEY, sha256 TEXT NOT NULL)"
-    )
-    batch = []
-    for p_str in walk_files(base_dir):
-        known = conn.execute(
-            "SELECT 1 FROM image_hashes WHERE path=? UNION SELECT 1 FROM known_paths WHERE path=?",
-            (p_str, p_str),
-        ).fetchone()
-        if known:
-            continue
-        sha = db.sha256_file(p_str)
-        sha_known = conn.execute(
-            "SELECT 1 FROM image_hashes WHERE sha256=?", (sha,)
-        ).fetchone()
-        if sha_known:
-            conn.execute(
-                "INSERT OR IGNORE INTO known_paths (path, sha256) VALUES (?,?)", (p_str, sha))
-            continue
-        batch.append(p_str)
-        if len(batch) >= batch_size:
-            break
-    conn.commit()
-    conn.close()
-    return batch
-
-
-@task
-def run_batches(base_dir: str, source: str, batch_size: int, max_batches: int | None, check_near_dups: bool = False) -> dict:
-    """Process a folder in batches — the whole loop runs in ONE task."""
-    db.init_db()
-    processed_batches = 0
-    totals = {"exact_dups": 0, "near_dups": 0, "new": 0}
-
-    while True:
-        batch = _find_unprocessed(base_dir, batch_size)
-        if not batch:
-            print("No more unprocessed files — done.")
-            break
-        result = check_and_register(batch, source, check_near_dups)
-        totals["exact_dups"] += result["exact_dups"]
-        totals["near_dups"] += result["near_dups"]
-        totals["new"] += result["new"]
-        processed_batches += 1
-        print(f"batch {processed_batches} done: {result['total']} files, "
-              f"{result['new']} new, {result['exact_dups']} exact, {result['near_dups']} near")
-        if max_batches and processed_batches >= max_batches:
-            print(f"Stopped after {processed_batches} batches (max_batches={max_batches})")
-            break
-
-    totals["batches"] = processed_batches
     return totals
 
 
 @flow(name="photo-ingest")
 def photo_ingest(base_dir: str, source: str = "takeout", batch_size: int = 1000,
                  max_batches: int | None = None, check_near_dups: bool = False):
-    """Hash + dedup-check a folder against the persistent library, in batches.
+    """Single-pass hash + dedup scan of a folder. One walk, checkpointed inserts.
 
     Args:
         base_dir: folder to scan
-        source: label for the batch (e.g. takeout, archive-pictures)
+        source: label (e.g. takeout, archive-pictures)
-        batch_size: files per batch/checkpoint (default 1000)
+        batch_size: insert checkpoint interval (default 1000)
-        max_batches: stop after N batches (useful for testing) — None = all
+        max_batches: kept for compatibility (unused — single pass)
+        check_near_dups: reserved (dhash-only rows; near-dup off)
     """
-    return run_batches(base_dir, source, batch_size, max_batches, check_near_dups)
+    return scan_once(base_dir, source, batch_size, check_near_dups)
 
 
 if __name__ == "__main__":
@@ -207,6 +120,5 @@ if __name__ == "__main__":
     d = sys.argv[1] if len(sys.argv) > 1 else "/tmp/sample"
     src = sys.argv[2] if len(sys.argv) > 2 else "cli-test"
     bs = int(sys.argv[3]) if len(sys.argv) > 3 else 1000
-    mb = int(sys.argv[4]) if len(sys.argv) > 4 else None
+    r = photo_ingest(d, source=src, batch_size=bs)
-    r = photo_ingest(d, source=src, batch_size=bs, max_batches=mb)
     print(r)