Initial: fingerprint DB + Takeout fetch + ingest flows

.gitignore

@@ -0,0 +1,7 @@
+.venv/
+__pycache__/
+*.pyc
+photo_pipeline.db
+*.db-wal
+*.db-shm
+test_flow.py

.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/

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

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}")

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)

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"])

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: {}

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" "$@"

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" "$@"

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/<export>/
+  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/<archive_stem>/.
+
+    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 <manifest-file> [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)