From 16123d00f1fe16fd7f4e24faa72b7e4ac92d75c3 Mon Sep 17 00:00:00 2001 From: Sam Rolfe Date: Fri, 7 Aug 2026 20:41:32 +1000 Subject: [PATCH] Initial: fingerprint DB + Takeout fetch + ingest flows --- .gitignore | 7 ++ .prefectignore | 41 ++++++++++ README.md | 16 ++++ photo_db.py | 163 ++++++++++++++++++++++++++++++++++++++ photo_ingest.py | 115 +++++++++++++++++++++++++++ photo_pipeline_hello.py | 18 +++++ prefect.yaml | 59 ++++++++++++++ run-prefect.sh | 10 +++ run-python.sh | 15 ++++ takeout_fetch.py | 170 ++++++++++++++++++++++++++++++++++++++++ 10 files changed, 614 insertions(+) create mode 100644 .gitignore create mode 100644 .prefectignore create mode 100644 README.md create mode 100644 photo_db.py create mode 100644 photo_ingest.py create mode 100644 photo_pipeline_hello.py create mode 100644 prefect.yaml create mode 100755 run-prefect.sh create mode 100755 run-python.sh create mode 100644 takeout_fetch.py diff --git a/.gitignore b/.gitignore new file mode 100644 index 0000000..3008d87 --- /dev/null +++ b/.gitignore @@ -0,0 +1,7 @@ +.venv/ +__pycache__/ +*.pyc +photo_pipeline.db +*.db-wal +*.db-shm +test_flow.py diff --git a/.prefectignore b/.prefectignore new file mode 100644 index 0000000..7ce5ed8 --- /dev/null +++ b/.prefectignore @@ -0,0 +1,41 @@ +# prefect artifacts +.prefectignore + +# python artifacts +__pycache__/ +*.py[cod] +*$py.class +*.egg-info/ +*.egg + +# Type checking artifacts +.mypy_cache/ +.dmypy.json +dmypy.json +.pyre/ + +# IPython +profile_default/ +ipython_config.py +*.ipynb_checkpoints/* + +# Environments +.python-version +.env +.venv +env/ +venv/ + +# MacOS +.DS_Store + +# Dask +dask-worker-space/ + +# Editors +.idea/ +.vscode/ + +# VCS +.git/ +.hg/ diff --git a/README.md b/README.md new file mode 100644 index 0000000..68b6d72 --- /dev/null +++ b/README.md @@ -0,0 +1,16 @@ +# photo-pipeline + +Prefect-orchestrated photo ingestion: Takeout download → fingerprint DB (dedup) → quality scan → Immich. + +## Components +- `photo_db.py` — SQLite fingerprint DB (sha256 exact + phash/dhash near dupes; Takeout archive tracking) +- `takeout_fetch.py` — flow: download (aria2c resume) → track sha256 → extract +- `photo_ingest.py` — flow: scan dir → check dups → register new +- `prefect.yaml` — deployments (fetch, ingest) on photo-pool + +## NixOS note +Venv python needs LD_LIBRARY_PATH (gcc libstdc++ + zlib) — see run-python.sh / run-prefect.sh. + +## Run +prefect deployment run "takeout-fetch/fetch" --param manifest=/path/urls.txt --param export_id=photos-2026-08 +prefect deployment run "photo-ingest/ingest" --param base_dir=/mnt/data/takeout/photos-2026-08 diff --git a/photo_db.py b/photo_db.py new file mode 100644 index 0000000..8e4cb4c --- /dev/null +++ b/photo_db.py @@ -0,0 +1,163 @@ +"""photo-pipeline: persistent SQLite fingerprint + Takeout tracking DB. + +Two responsibilities: +1. image_hashes — permanent library identity (sha256 exact + phash/dhash near) +2. takeout_archives — shipment tracking for Takeout dumps (the "lost track" problem) +3. batches — which source batch fed which registration + +Hashing is the foundation; this DB is where the hashes live and are queried. +Immich keeps its own hashes in Postgres as the in-library layer. +""" + +import hashlib +import sqlite3 +from pathlib import Path + +from imagehash import dhash, phash +from PIL import Image + +DB_PATH = Path(__file__).parent / "photo_pipeline.db" + +SCHEMA = """ +CREATE TABLE IF NOT EXISTS image_hashes ( + sha256 TEXT PRIMARY KEY, + phash TEXT NOT NULL, + dhash TEXT NOT NULL, + file_size INTEGER, + width INTEGER, + height INTEGER, + exif_date TEXT, + path TEXT, + source TEXT, -- takeout | archive | phone | immich + in_immich INTEGER DEFAULT 0, + added_at TEXT DEFAULT (datetime('now')) +); + +CREATE TABLE IF NOT EXISTS takeout_archives ( + id INTEGER PRIMARY KEY AUTOINCREMENT, + export_id TEXT, + archive_name TEXT UNIQUE, + sha256 TEXT, -- hash of the ARCHIVE file itself + size_bytes INTEGER, + created_at TEXT, + downloaded_at TEXT, + extracted_at TEXT, + extract_path TEXT, + file_count INTEGER, + status TEXT DEFAULT 'pending' -- pending|downloading|downloaded|extracted|registered +); + +CREATE TABLE IF NOT EXISTS batches ( + id INTEGER PRIMARY KEY AUTOINCREMENT, + name TEXT, + archive_id INTEGER REFERENCES takeout_archives(id), + source TEXT, + started_at TEXT, + finished_at TEXT, + status TEXT +); + +CREATE INDEX IF NOT EXISTS idx_image_hashes_phash ON image_hashes(phash); +CREATE INDEX IF NOT EXISTS idx_image_hashes_dhash ON image_hashes(dhash); +""" + + +def get_db(): + conn = sqlite3.connect(DB_PATH) + conn.execute("PRAGMA journal_mode=WAL") + conn.execute("PRAGMA foreign_keys=ON") + return conn + + +def init_db(): + with get_db() as conn: + conn.executescript(SCHEMA) + + +def sha256_file(path: Path) -> str: + h = hashlib.sha256() + with open(path, "rb") as f: + for chunk in iter(lambda: f.read(1024 * 1024), b""): + h.update(chunk) + return h.hexdigest() + + +def hash_image(path: Path, hash_size: int = 8): + """Perceptual hashes for one image file.""" + with Image.open(path) as im: + im = im.convert("RGB") + return { + "phash": str(phash(im, hash_size=hash_size)), + "dhash": str(dhash(im, hash_size=hash_size)), + } + + +def register_image(conn, path: Path, source: str, exif_date: str | None = None): + """Insert one image's hashes if not already present. Returns True if new.""" + sha = sha256_file(path) + existing = conn.execute("SELECT 1 FROM image_hashes WHERE sha256=?", (sha,)).fetchone() + if existing: + return False + with Image.open(path) as im: + w, h = im.size + hx = hash_image(path) + conn.execute( + "INSERT INTO image_hashes (sha256, phash, dhash, file_size, width, height, exif_date, path, source) " + "VALUES (?,?,?,?,?,?,?,?,?)", + (sha, hx["phash"], hx["dhash"], path.stat().st_size, w, h, exif_date, str(path), source), + ) + return True + + +def check_exact_dup(conn, path: Path): + """Return matching library path if this file's sha256 already exists.""" + sha = sha256_file(path) + row = conn.execute("SELECT path, source FROM image_hashes WHERE sha256=?", (sha,)).fetchone() + return row + + +def check_near_dup(conn, path: Path, hamming_threshold: int = 10): + """Return nearest-matching library path if phash+dhash are close enough.""" + hx = hash_image(path) + rows = conn.execute("SELECT phash, dhash, path, source FROM image_hashes").fetchall() + best, best_dist = None, None + for ph, dh, p, src in rows: + phd = hamming(ph, hx["phash"]) + dhd = hamming(dh, hx["dhash"]) + dist = min(phd, dhd) + if best_dist is None or dist < best_dist: + best, best_dist = (p, src), dist + if best_dist == 0: + break + if best and best_dist <= hamming_threshold: + return best, best_dist + return None, best_dist + + +def hamming(a: str, b: str) -> int: + return sum(1 for x, y in zip(a, b) if x != y) + + +def register_takeout_archive(conn, archive_name: str, path: Path) -> int: + """Record a downloaded archive (or detect it's already known). Returns row id.""" + sha = sha256_file(path) + row = conn.execute( + "SELECT id FROM takeout_archives WHERE archive_name=?", (archive_name,) + ).fetchone() + if row: + conn.execute( + "UPDATE takeout_archives SET sha256=?, size_bytes=?, downloaded_at=datetime('now'), status='downloaded' WHERE id=?", + (sha, path.stat().st_size, row[0]), + ) + return row[0] + cur = conn.execute( + "INSERT INTO takeout_archives (archive_name, sha256, size_bytes, downloaded_at, status) " + "VALUES (?,?,?,datetime('now'),'downloaded')", + (archive_name, sha, path.stat().st_size), + ) + return cur.lastrowid + + +if __name__ == "__main__": + init_db() + print(f"DB ready at {DB_PATH}") diff --git a/photo_ingest.py b/photo_ingest.py new file mode 100644 index 0000000..6940cb1 --- /dev/null +++ b/photo_ingest.py @@ -0,0 +1,115 @@ +"""photo-pipeline: photo-ingest flow (v1). + +Stage 1 of the pipeline: hash incoming images, check against the persistent +fingerprint DB (exact + near dupes), and register new ones. + +Run via Prefect deployment on photo-pool (see prefect.yaml). +""" + +import sqlite3 +from pathlib import Path + +from prefect import flow, task + +import photo_db as db + + +@task +def scan_directory(base_dir: str) -> list[str]: + """Enumerate image files in a directory tree.""" + exts = {".jpg", ".jpeg", ".png", ".heic", ".webp", ".gif", ".tif", ".tiff", ".bmp"} + root = Path(base_dir) + if not root.exists(): + raise FileNotFoundError(f"{root} does not exist") + found = [ + str(p) + for p in root.rglob("*") + if p.is_file() and p.suffix.lower() in exts + ] + print(f"Found {len(found)} images under {root}") + return found + + +@task +def check_duplicates(image_paths: list[str]) -> dict: + """Check each image against the fingerprint DB. Returns classification.""" + db.init_db() + conn = db.get_db() + exact_dups = [] + near_dups = [] + new_images = [] + for p_str in image_paths: + p = Path(p_str) + try: + match = db.check_exact_dup(conn, p) + if match: + exact_dups.append((p_str, match)) + continue + near, dist = db.check_near_dup(conn, p) + if near: + near_dups.append((p_str, near, dist)) + continue + new_images.append(p_str) + except Exception as e: + print(f" SKIP {p.name}: {e}") + conn.close() + result = { + "total": len(image_paths), + "exact_dups": len(exact_dups), + "near_dups": len(near_dups), + "new": len(new_images), + "exact_dup_list": exact_dups[:50], + "near_dup_list": near_dups[:50], + "new_list": new_images, + } + print( + f"Check: {result['total']} total, " + f"{result['exact_dups']} exact dups, {result['near_dups']} near dups, " + f"{result['new']} new" + ) + return result + + +@task +def register_new_images(image_paths: list[str], source: str) -> int: + """Add hashes for confirmed-new images into the fingerprint DB.""" + db.init_db() + conn = db.get_db() + registered = 0 + for p_str in image_paths: + p = Path(p_str) + try: + if db.register_image(conn, p, source=source): + registered += 1 + except Exception as e: + print(f" FAIL register {p.name}: {e}") + conn.commit() + conn.close() + print(f"Registered {registered} new images (source={source})") + return registered + + +@flow(name="photo-ingest") +def photo_ingest(base_dir: str, source: str = "takeout", register: bool = True): + """Hash + dedup-check a folder against the persistent library.""" + images = scan_directory(base_dir) + if not images: + print("No images found — nothing to do.") + return {"total": 0} + + result = check_duplicates(images) + + if register and result["new_list"]: + n = register_new_images(result["new_list"], source=source) + result["registered"] = n + + return result + + +if __name__ == "__main__": + # Local run (no deployment) + import sys + + d = sys.argv[1] if len(sys.argv) > 1 else "/tmp/sample" + r = photo_ingest(d) + print(r) diff --git a/photo_pipeline_hello.py b/photo_pipeline_hello.py new file mode 100644 index 0000000..46ad503 --- /dev/null +++ b/photo_pipeline_hello.py @@ -0,0 +1,18 @@ +"""photo-pipeline: first real flow (skeleton). + +Prefect 3.8 local pattern: flow.serve() registers the deployment and +runs scheduled work. For the pilot this is the supported, storage-free path. +""" + +from prefect import flow + + +@flow(name="photo-pipeline-hello") +def hello_pipeline(): + print("photo-pipeline flow alive on .13") + return "ok" + + +if __name__ == "__main__": + # serve() keeps this process alive, polls for scheduled runs + hello_pipeline.serve(name="hello", cron="*/5 * * * *", tags=["photo-pipeline"]) diff --git a/prefect.yaml b/prefect.yaml new file mode 100644 index 0000000..eb5de6d --- /dev/null +++ b/prefect.yaml @@ -0,0 +1,59 @@ +--- +# Prefect project config for photo-pipeline +name: photo-pipeline +prefect-version: 3.8.1 + +build: null +push: null + +pull: +- prefect.deployments.steps.set_working_directory: + directory: /home/sam/photo-pipeline + +deployments: +- name: hello + version: null + tags: [photo-pipeline] + description: "Photo pipeline hello flow" + schedule: + cron: "*/5 * * * *" + timezone: "Australia/Melbourne" + flow_name: null + entrypoint: photo_pipeline_hello.py:hello_pipeline + parameters: {} + work_pool: + name: photo-pool + work_queue_name: null + job_variables: {} + +- name: ingest + version: null + tags: [photo-pipeline] + description: "Hash + dedup-check a folder against the persistent fingerprint DB" + schedule: null + flow_name: null + entrypoint: photo_ingest.py:photo_ingest + parameters: + base_dir: /mnt/data/takeout + source: takeout + register: true + work_pool: + name: photo-pool + work_queue_name: null + job_variables: {} + +- name: fetch + version: null + tags: [photo-pipeline] + description: "Download + track + extract Takeout archives (resume-capable)" + schedule: null + flow_name: null + entrypoint: takeout_fetch.py:takeout_fetch + parameters: + urls: [] + manifest: null + export_id: null + work_pool: + name: photo-pool + work_queue_name: null + job_variables: {} diff --git a/run-prefect.sh b/run-prefect.sh new file mode 100755 index 0000000..9a30e71 --- /dev/null +++ b/run-prefect.sh @@ -0,0 +1,10 @@ +#!/usr/bin/env bash +# NixOS venv wrapper for the prefect binary (same libs as run-python.sh) +set -euo pipefail +NIX_LIBS=( + /nix/store/0iv8glcslgfcgn371lbjr5jjw5a6cqir-gcc-15.3.0-lib/lib + /nix/store/l7xwm1f6f3zj2x8jwdbi8gdyfbx07sh7-zlib-1.3.1/lib +) +EXISTING="${LD_LIBRARY_PATH:-}" +export LD_LIBRARY_PATH="$(IFS=:; echo "${NIX_LIBS[*]}"):${EXISTING}" +exec "$HOME/photo-pipeline/.venv/bin/prefect" "$@" diff --git a/run-python.sh b/run-python.sh new file mode 100755 index 0000000..f7ea607 --- /dev/null +++ b/run-python.sh @@ -0,0 +1,15 @@ +#!/usr/bin/env bash +# NixOS venv wrapper: exposes needed system libs (libstdc++, zlib, ...) to the +# photo-pipeline venv. Edit NIX_LIBS as new deps (e.g. opencv) require more. +set -euo pipefail + +NIX_LIBS=( + /nix/store/0iv8glcslgfcgn371lbjr5jjw5a6cqir-gcc-15.3.0-lib/lib + /nix/store/l7xwm1f6f3zj2x8jwdbi8gdyfbx07sh7-zlib-1.3.1/lib +) + +# Merge with any pre-existing LD_LIBRARY_PATH +EXISTING="${LD_LIBRARY_PATH:-}" +export LD_LIBRARY_PATH="$(IFS=:; echo "${NIX_LIBS[*]}"):${EXISTING}" + +exec "$HOME/photo-pipeline/.venv/bin/python" "$@" diff --git a/takeout_fetch.py b/takeout_fetch.py new file mode 100644 index 0000000..dfe51c2 --- /dev/null +++ b/takeout_fetch.py @@ -0,0 +1,170 @@ +"""photo-pipeline: Takeout download + extract flow (v2). + +Completes the dump-tracking loop: + 1. Download Takeout archives (resume-capable via aria2c) + 2. Verify archive sha256 against takeout_archives (no re-download of known archives) + 3. Extract to /mnt/data/takeout// + 4. Update status in takeout_archives + +Takeout gives each export a set of archive URLs (the "Download" page). Save them +to a manifest file (one URL per line) or pass a list. +""" + +import shutil +import subprocess +from pathlib import Path + +from prefect import flow, task + +import photo_db as db + +STAGING = Path("/mnt/data/takeout") + + +@task +def download_archive(url: str, dest_dir: Path = STAGING) -> Path: + """Download one Takeout archive with aria2c (resume-capable, 4 connections). + + Returns the local archive path. Skips if archive sha256 already known+downloaded. + """ + dest_dir.mkdir(parents=True, exist_ok=True) + archive_name = url.split("/")[-1].split("?")[0] + if not archive_name: + archive_name = f"takeout-{abs(hash(url))}.zip" + dest = dest_dir / archive_name + + # Skip if we already have it downloaded (by name) + if dest.exists() and dest.stat().st_size > 0: + print(f" already on disk: {archive_name} ({dest.stat().st_size} bytes) — verifying sha256") + sha = db.sha256_file(dest) + conn = db.get_db() + known = conn.execute( + "SELECT id FROM takeout_archives WHERE archive_name=? AND sha256=?", (archive_name, sha) + ).fetchone() + conn.close() + if known: + print(f" verified — already tracked (id={known[0]}). Skipping download.") + return dest + print(f" on disk but untracked sha — re-downloading to be safe") + + print(f"Downloading {archive_name} → {dest}") + cmd = [ + "aria2c", + "-x", "4", "-s", "4", # 4 connections + "-c", # continue/resume + "--auto-file-renaming=false", + "--allow-overwrite=false", + "-d", str(dest_dir), + "-o", archive_name, + url, + ] + r = subprocess.run(cmd, capture_output=True, text=True) + if r.returncode != 0: + raise RuntimeError(f"aria2c failed for {archive_name}:\n{r.stderr[-1000:]}") + return dest + + +@task +def track_archive(archive_path: Path, export_id: str | None = None) -> int: + """Record the downloaded archive in takeout_archives (or update it).""" + db.init_db() + conn = db.get_db() + archive_id = db.register_takeout_archive(conn, archive_path.name, archive_path) + if export_id: + conn.execute( + "UPDATE takeout_archives SET export_id=? WHERE id=?", (export_id, archive_id) + ) + conn.commit() + conn.close() + print(f"Tracked archive id={archive_id} ({archive_path.name}, {archive_path.stat().st_size} bytes)") + return archive_id + + +@task +def extract_archive(archive_path: Path, export_dir: str | None = None) -> Path: + """Extract a Takeout archive to /mnt/data/takeout//. + + Handles .zip (unzip or python zipfile) and .tgz/.tar.gz (tar). + """ + if export_dir: + out_root = STAGING / export_dir + else: + out_root = STAGING / archive_path.stem.replace(".tar", "") + out_root.mkdir(parents=True, exist_ok=True) + + if archive_path.suffix == ".zip" or archive_path.name.endswith(".zip"): + # Try unzip first; fall back to python zipfile (no external dep) + if shutil.which("unzip"): + r = subprocess.run( + ["unzip", "-o", "-q", str(archive_path), "-d", str(out_root)], + capture_output=True, text=True, + ) + if r.returncode != 0: + raise RuntimeError(f"unzip failed: {r.stderr[-500:]}") + else: + import zipfile + with zipfile.ZipFile(archive_path) as zf: + zf.extractall(out_root) + elif archive_path.name.endswith((".tgz", ".tar.gz", ".tar")): + r = subprocess.run( + ["tar", "-xzf", str(archive_path), "-C", str(out_root)], + capture_output=True, text=True, + ) + if r.returncode != 0: + raise RuntimeError(f"tar failed: {r.stderr[-500:]}") + else: + raise ValueError(f"Unsupported archive type: {archive_path.name}") + + # Count extracted files + n = sum(1 for _ in out_root.rglob("*") if _.is_file()) + db.init_db() + conn = db.get_db() + conn.execute( + "UPDATE takeout_archives SET extracted_at=datetime('now'), extract_path=?, file_count=?, status='extracted' " + "WHERE archive_name=?", + (str(out_root), n, archive_path.name), + ) + conn.commit() + conn.close() + print(f"Extracted {n} files → {out_root}") + return out_root + + +@flow(name="takeout-fetch") +def takeout_fetch( + urls: list[str] | None = None, + manifest: str | None = None, + export_id: str | None = None, +): + """Download + track + extract a set of Takeout archives. + + Args: + urls: list of archive download URLs (from takeout.google.com Download page) + manifest: path to a text file with one URL per line (alternative to urls) + export_id: label for this export (e.g. "photos-2026-08") + """ + if manifest: + mp = Path(manifest) + if not mp.exists(): + raise FileNotFoundError(f"manifest not found: {mp}") + urls = [l.strip() for l in mp.read_text().splitlines() if l.strip().startswith("http")] + if not urls: + raise ValueError("provide urls or a manifest file") + + results = [] + for url in urls: + path = download_archive(url) + archive_id = track_archive(path, export_id=export_id) + out = extract_archive(path, export_dir=export_id) + results.append({"archive": path.name, "id": archive_id, "extracted_to": str(out)}) + return results + + +if __name__ == "__main__": + import sys + if len(sys.argv) < 2: + print("usage: python takeout_fetch.py [export_id]") + sys.exit(1) + man = sys.argv[1] + eid = sys.argv[2] if len(sys.argv) > 2 else None + takeout_fetch(manifest=man, export_id=eid)