diff --git a/photo_ingest.py b/photo_ingest.py index fe6f79d..3ba8e3c 100644 --- a/photo_ingest.py +++ b/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 -(exact + near dupes), and registers new ones. +Hashes incoming images, checks the persistent fingerprint DB (exact dups via +sha256, multi-path dup tracking via known_paths), and registers new files. -Batching: processes in chunks of `batch_size` (default 1000), committing to -the DB after each chunk. Checkpointing is DB-native: files already in the DB -(by sha256) are skipped on resume — an interrupted run continues where it -stopped, never redoing work. +Design: ONE walk of the tree. Each file: path-known → skip fast; sha-known at +another path → record path in known_paths (dup); else register new. Inserts +commit every `batch_size` rows (WAL) = checkpoint. Single pass = O(n), no +re-walk, no O(n²), no engine busy-loop. -For very large trees (e.g. 58K files), scan_directory can be slow to walk; -use walk_files for a streaming generator when batch_size is set. +Performance: dhash-only rows (~6ms/file); phash only when check_near_dups. """ 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]: - """Find the next batch of files NOT yet in the fingerprint DB.""" +def scan_once(base_dir: str, source: str, batch_size: int, check_near_dups: bool) -> dict: + """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 - sha = db.sha256_file(p) + 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: + 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 - # 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) + + hx = db.hash_image(p) with Image.open(p) as im: w, h = im.size - if hx is None: - # still need hashes for the row — compute minimal (phash only) - hx = db.hash_image(p, include_phash=True) - conn.execute( - "INSERT INTO image_hashes (sha256, phash, dhash, file_size, width, height, path, source) " - "VALUES (?,?,?,?,?,?,?,?)", - (sha, hx["phash"], hx["dhash"], p.stat().st_size, w, h, str(p), source), - ) - new_images.append(p_str) + pending.append((sha, hx["phash"], hx["dhash"], p.stat().st_size, + w, h, str(p), source)) + totals["new"] += 1 + + if len(pending) >= batch_size: + conn.executemany( + "INSERT INTO image_hashes (sha256, phash, dhash, file_size, width, height, path, source) " + "VALUES (?,?,?,?,?,?,?,?)", pending) + 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}") - conn.commit() + totals["skipped"] += 1 + 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 { - "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 + print(f"scan_once done in {time.time()-t0:.0f}s: {totals}", flush=True) 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) - batch_size: files per batch/checkpoint (default 1000) - max_batches: stop after N batches (useful for testing) — None = all + source: label (e.g. takeout, archive-pictures) + batch_size: insert checkpoint interval (default 1000) + 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, max_batches=mb) + r = photo_ingest(d, source=src, batch_size=bs) print(r)