diff --git a/apps/trip-cluster/app/cli.py b/apps/trip-cluster/app/cli.py index f81c172..956a55d 100644 --- a/apps/trip-cluster/app/cli.py +++ b/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 diff --git a/apps/trip-cluster/app/ingest.py b/apps/trip-cluster/app/ingest.py new file mode 100644 index 0000000..d207ad0 --- /dev/null +++ b/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 .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} diff --git a/apps/trip-cluster/tests/test_ingest.py b/apps/trip-cluster/tests/test_ingest.py new file mode 100644 index 0000000..25896b8 --- /dev/null +++ b/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()