feat(trip-cluster): ingest — scopeable, incremental, processed read-back

apps/trip-cluster/app/cli.py

@@ -18,6 +18,19 @@ def _immich(deps, cfg):
     return ImmichClient(cfg.immich_url, cfg.immich_api_key)
 
 
+def cmd_ingest(deps, *, date_from, date_to, tag, subset, full) -> int:
+    cfg = deps["config"]
+    store = _store(cfg)
+    client = _immich(deps, cfg)
+    from app.ingest import run_ingest
+    res = run_ingest(client, store, cfg.thumbs_dir, date_from=date_from,
+                     date_to=date_to, tag=tag, subset=subset, full=full)
+    store.close()
+    print(f"Ingested {res['fetched']} asset(s); "
+          f"{res['processed_marked']} already-processed.")
+    return 0
+
+
 def cmd_serve(deps) -> int:
     from app import create_app
     create_app(deps["config"]).run(host="0.0.0.0", port=8084)
@@ -51,7 +64,10 @@ def main(argv=None) -> int:
     deps = {"config": load_config()}
     if args.command == "serve":
         return cmd_serve(deps)
-    # ingest / cluster / apply are wired in later tasks.
+    if args.command == "ingest":
+        return cmd_ingest(deps, date_from=args.date_from, date_to=args.date_to,
+                          tag=args.tag, subset=args.subset, full=args.full)
+    # cluster / apply are wired in later tasks.
     print(f"Command '{args.command}' is not implemented yet.")
     return 1
 

apps/trip-cluster/app/ingest.py

@@ -0,0 +1,80 @@
+import datetime
+import logging
+import os
+
+from photoflow.core.models import Asset
+from photoflow.immich import pipeline
+
+log = logging.getLogger(__name__)
+
+
+def _now() -> str:
+    return datetime.datetime.now(datetime.timezone.utc).isoformat()
+
+
+def run_ingest(client, store, thumbs_dir, *, date_from=None, date_to=None,
+               tag=None, subset=None, full=False) -> dict:
+    os.makedirs(thumbs_dir, exist_ok=True)
+
+    tag_ids = None
+    if tag:
+        tid = client.resolve_tag_id(tag)
+        if not tid:
+            raise ValueError(f"Tag {tag!r} not found in Immich")
+        tag_ids = [tid]
+
+    updated_after = None if full else store.get_meta("last_ingest_at")
+
+    assets = client.search_assets(taken_after=date_from, taken_before=date_to,
+                                  tag_ids=tag_ids, updated_after=updated_after)
+    if subset is not None:
+        assets = assets[:subset]
+
+    processed_marked = 0
+    max_updated = updated_after or ""
+    tag_counts: dict = {}
+
+    for a in assets:
+        # Revision (2): sanitize the Immich-provided id before using it as a
+        # path component, guarding against path traversal (mirror /thumb proxy).
+        safe_id = os.path.basename(a["id"])
+        thumb_path = os.path.join(thumbs_dir, f"{safe_id}.jpg")
+        # Revision (1): treat an existing 0-byte file as missing.
+        need_thumb = (not os.path.exists(thumb_path)
+                      or os.path.getsize(thumb_path) == 0)
+        if need_thumb:
+            try:
+                data = client.download_thumbnail(a["id"])
+                # Write to a temp path and os.replace into place so a failure
+                # never leaves a 0-byte <id>.jpg behind.
+                tmp_path = f"{thumb_path}.tmp"
+                with open(tmp_path, "wb") as f:
+                    f.write(data)
+                os.replace(tmp_path, thumb_path)
+            except Exception:
+                log.exception("Thumbnail download failed for asset %s", a["id"])
+                continue
+
+        store.upsert_asset(Asset(
+            immich_id=a["id"], taken_at=a["taken_at"], gps_lat=a["gps_lat"],
+            gps_lon=a["gps_lon"], place_city=a["place_city"],
+            place_country=a["place_country"], type=a["type"],
+            thumb_path=thumb_path, ingested_at=_now(), updated_at=a["updated_at"]))
+        store.set_asset_tags(a["id"], a["tags"])
+        for t in a["tags"]:
+            tag_counts[t] = tag_counts.get(t, 0) + 1
+
+        if pipeline.PROCESSED in a["tags"]:
+            store.mark_processed(a["id"])
+            processed_marked += 1
+
+        if a["updated_at"] and a["updated_at"] > max_updated:
+            max_updated = a["updated_at"]
+
+    for name, count in tag_counts.items():
+        store.upsert_tag(name, count=count)
+
+    if max_updated:
+        store.set_meta("last_ingest_at", max_updated)
+
+    return {"fetched": len(assets), "processed_marked": processed_marked}

apps/trip-cluster/tests/test_ingest.py

@@ -0,0 +1,104 @@
+import os
+from photoflow.core import Store
+from photoflow.immich import pipeline
+from app.ingest import run_ingest
+
+
+class FakeImmich:
+    def __init__(self, assets, tag_map=None, thumb=b"\xff\xd8\xffjpeg"):
+        self._assets = assets
+        self._tag_map = tag_map or {}
+        self._thumb = thumb
+        self.searches = []
+
+    def resolve_tag_id(self, name):
+        return self._tag_map.get(name)
+
+    def search_assets(self, **kwargs):
+        self.searches.append(kwargs)
+        return list(self._assets)
+
+    def download_thumbnail(self, asset_id):
+        return self._thumb
+
+
+def _asset(i, taken, tags=None, updated="2026-01-01T00:00:00Z"):
+    return {"id": i, "original_filename": f"{i}.jpg", "taken_at": taken,
+            "gps_lat": None, "gps_lon": None, "place_city": None,
+            "place_country": None, "type": "IMAGE", "tags": tags or [],
+            "rating": 0, "updated_at": updated}
+
+
+def test_ingest_upserts_assets_tags_and_thumbs(tmp_path):
+    store = Store(str(tmp_path / "t.db")).connect()
+    thumbs = str(tmp_path / "thumbs")
+    os.makedirs(thumbs, exist_ok=True)
+    client = FakeImmich([_asset("a", "2019-06-01", tags=["Italy 2019"]),
+                         _asset("b", "2019-06-02")])
+    res = run_ingest(client, store, thumbs, date_from="2019-01-01", date_to="2020-01-01")
+    assert res["fetched"] == 2
+    assert store.get_asset("a").taken_at == "2019-06-01"
+    assert store.asset_tags("a") == ["Italy 2019"]
+    assert os.path.exists(os.path.join(thumbs, "a.jpg"))
+    assert store.get_meta("last_ingest_at") == "2026-01-01T00:00:00Z"
+    store.close()
+
+
+def test_ingest_marks_processed_from_pipeline_tag(tmp_path):
+    store = Store(str(tmp_path / "t.db")).connect()
+    thumbs = str(tmp_path / "thumbs"); os.makedirs(thumbs, exist_ok=True)
+    client = FakeImmich([_asset("a", "2019-06-01", tags=[pipeline.PROCESSED])])
+    res = run_ingest(client, store, thumbs)
+    assert res["processed_marked"] == 1
+    assert store.get_asset("a").processed is True
+    store.close()
+
+
+def test_ingest_incremental_passes_updated_after(tmp_path):
+    store = Store(str(tmp_path / "t.db")).connect()
+    thumbs = str(tmp_path / "thumbs"); os.makedirs(thumbs, exist_ok=True)
+    store.set_meta("last_ingest_at", "2026-05-01T00:00:00Z")
+    client = FakeImmich([_asset("a", "2019-06-01")])
+    run_ingest(client, store, thumbs)
+    assert client.searches[0].get("updated_after") == "2026-05-01T00:00:00Z"
+    # --full ignores it
+    client2 = FakeImmich([_asset("a", "2019-06-01")])
+    run_ingest(client2, store, thumbs, full=True)
+    assert client2.searches[0].get("updated_after") is None
+    store.close()
+
+
+def test_ingest_subset_and_tag_resolution(tmp_path):
+    store = Store(str(tmp_path / "t.db")).connect()
+    thumbs = str(tmp_path / "thumbs"); os.makedirs(thumbs, exist_ok=True)
+    client = FakeImmich([_asset("a", "2019-06-01"), _asset("b", "2019-06-02")],
+                        tag_map={"Italy 2019": "t1"})
+    res = run_ingest(client, store, thumbs, tag="Italy 2019", subset=1)
+    assert res["fetched"] == 1
+    assert client.searches[0].get("tag_ids") == ["t1"]
+    store.close()
+
+
+def test_ingest_resilient_to_thumb_failure_and_zero_byte(tmp_path):
+    store = Store(str(tmp_path / "t.db")).connect()
+    thumbs = str(tmp_path / "thumbs"); os.makedirs(thumbs, exist_ok=True)
+
+    class FailThenOk(FakeImmich):
+        def download_thumbnail(self, asset_id):
+            if asset_id == "a":
+                raise RuntimeError("boom")
+            return self._thumb
+
+    # Pre-existing 0-byte file for "b" must be treated as missing and refetched.
+    with open(os.path.join(thumbs, "b.jpg"), "wb"):
+        pass
+
+    client = FailThenOk([_asset("a", "2019-06-01"), _asset("b", "2019-06-02")])
+    res = run_ingest(client, store, thumbs)
+    # The run completes despite "a" failing.
+    assert res["fetched"] == 2
+    # No 0-byte file left for the failed asset.
+    assert not os.path.exists(os.path.join(thumbs, "a.jpg"))
+    # The 0-byte file for "b" was replaced with real bytes.
+    assert os.path.getsize(os.path.join(thumbs, "b.jpg")) > 0
+    store.close()