diff --git a/apps/trip-cluster/app/cli.py b/apps/trip-cluster/app/cli.py index 4bb7023..c87109f 100644 --- a/apps/trip-cluster/app/cli.py +++ b/apps/trip-cluster/app/cli.py @@ -41,6 +41,30 @@ def cmd_cluster(deps, *, gap_factor) -> int: return 0 +def cmd_apply(deps, *, yes) -> int: + cfg = deps["config"] + store = _store(cfg) + from app.writeback import apply_all, APPLYABLE + pending = [c for c in store.all_clusters() if c.status in APPLYABLE] + if not pending: + print("Nothing to apply.") + store.close() + return 0 + if not yes: + ans = input(f"Apply {len(pending)} cluster decision(s) to Immich? [y/N] ") + if ans.strip().lower() not in ("y", "yes"): + print("Aborted.") + store.close() + return 1 + client = _immich(deps, cfg) + results = apply_all(client, store) + store.close() + ok = sum(len(r["succeeded"]) for r in results) + bad = sum(len(r["failed"]) for r in results) + print(f"Applied {ok} tag write(s); {bad} failure(s) across {len(results)} cluster(s).") + return 0 if bad == 0 else 1 + + def cmd_serve(deps) -> int: from app import create_app create_app(deps["config"]).run(host="0.0.0.0", port=8084) @@ -79,7 +103,8 @@ def main(argv=None) -> int: tag=args.tag, subset=args.subset, full=args.full) if args.command == "cluster": return cmd_cluster(deps, gap_factor=args.gap_factor) - # apply is wired in a later task. + if args.command == "apply": + return cmd_apply(deps, yes=args.yes) print(f"Command '{args.command}' is not implemented yet.") return 1 diff --git a/apps/trip-cluster/app/writeback.py b/apps/trip-cluster/app/writeback.py new file mode 100644 index 0000000..7416583 --- /dev/null +++ b/apps/trip-cluster/app/writeback.py @@ -0,0 +1,57 @@ +from photoflow.immich import pipeline + +APPLYABLE = ("approved", "non_trip", "skipped") + + +def _apply_tag(client, store, asset_ids, action, tag, tag_id): + todo = [a for a in asset_ids if not store.already_applied(a, action, tag)] + if not todo: + return [], [] + try: + client.tag_assets(tag_id, todo) + except Exception as e: # noqa: BLE001 — recorded, surfaced + for a in todo: + store.log_writeback(a, action, tag, f"error:{e}") + return [], [(a, str(e)) for a in todo] + for a in todo: + store.log_writeback(a, action, tag, "ok") + return todo, [] + + +def apply_cluster(client, store, cluster_id) -> dict: + c = store.get_cluster(cluster_id) + if c is None or c.status not in APPLYABLE: + return {"cluster_id": cluster_id, "status": c.status if c else None, + "succeeded": [], "failed": []} + + included = [a.immich_id for a, m in store.cluster_members(cluster_id) if m.included] + succeeded, failed = [], [] + + if c.status == "approved": + tag = c.decided_name or c.suggested_name + ok, fail = _apply_tag(client, store, included, "trip", tag, client.upsert_tag(tag)) + succeeded += ok + failed += fail + elif c.status == "non_trip": + ok, fail = _apply_tag(client, store, included, "non-trip", pipeline.NON_TRIP, + client.upsert_tag(pipeline.NON_TRIP)) + succeeded += ok + failed += fail + # 'skipped': no content/non-trip tag, only processed below. + + failed_ids = {i for i, _ in failed} + proc_targets = [a for a in included if a not in failed_ids] + if proc_targets: + _apply_tag(client, store, proc_targets, "processed", pipeline.PROCESSED, + client.upsert_tag(pipeline.PROCESSED)) + for a in proc_targets: + if store.already_applied(a, "processed", pipeline.PROCESSED): + store.mark_processed(a) + + return {"cluster_id": cluster_id, "status": c.status, + "succeeded": succeeded, "failed": failed} + + +def apply_all(client, store) -> list: + return [apply_cluster(client, store, c.id) + for c in store.all_clusters() if c.status in APPLYABLE] diff --git a/apps/trip-cluster/tests/test_writeback.py b/apps/trip-cluster/tests/test_writeback.py new file mode 100644 index 0000000..9deb8d6 --- /dev/null +++ b/apps/trip-cluster/tests/test_writeback.py @@ -0,0 +1,113 @@ +from photoflow.core import Store +from photoflow.core.models import Asset, Cluster, ClusterMember +from photoflow.immich import pipeline +from app.writeback import apply_cluster, apply_all + + +class FakeImmich: + def __init__(self): + self.tagged = [] + self.fail_tag_id = None + self._ids = {} + + def upsert_tag(self, name): + self._ids.setdefault(name, f"id:{name}") + return self._ids[name] + + def tag_assets(self, tag_id, ids): + if self.fail_tag_id is not None and tag_id == self.fail_tag_id: + raise RuntimeError("boom") + self.tagged.append((tag_id, list(ids))) + + +def _store(tmp_path): + s = Store(str(tmp_path / "t.db")).connect() + for i in ("a", "b", "x"): + s.upsert_asset(Asset(immich_id=i, taken_at="2019-06-01")) + return s + + +def _approved(s, name="Venice"): + cid = s.insert_cluster( + Cluster(start_at="2019-06-01", end_at="2019-06-02", suggested_name=name, + status="approved", decided_name=name), + [ClusterMember(cluster_id=0, immich_id="a"), + ClusterMember(cluster_id=0, immich_id="b"), + ClusterMember(cluster_id=0, immich_id="x", included=False, flagged_coverage=True)]) + return cid + + +def test_apply_approved_tags_included_then_processed(tmp_path): + s = _store(tmp_path) + cid = _approved(s) + client = FakeImmich() + res = apply_cluster(client, s, cid) + assert sorted(res["succeeded"]) == ["a", "b"] and res["failed"] == [] + # trip tag on a,b ; processed on a,b ; x (excluded) never tagged + assert ("id:Venice", ["a", "b"]) in client.tagged + assert ("id:_pipeline/processed", ["a", "b"]) in client.tagged + assert all("x" not in ids for _, ids in client.tagged) + assert s.get_asset("a").processed is True + s.close() + + +def test_apply_is_idempotent(tmp_path): + s = _store(tmp_path) + cid = _approved(s) + client = FakeImmich() + apply_cluster(client, s, cid) + before = len(client.tagged) + apply_cluster(client, s, cid) # second run writes nothing new + assert len(client.tagged) == before + s.close() + + +def test_partial_failure_leaves_retryable(tmp_path): + s = _store(tmp_path) + cid = _approved(s) + client = FakeImmich() + client.fail_tag_id = "id:Venice" # trip tag write fails + res = apply_cluster(client, s, cid) + assert res["succeeded"] == [] and sorted(i for i, _ in res["failed"]) == ["a", "b"] + assert s.get_asset("a").processed is False # not marked processed on failure + assert s.already_applied("a", "trip", "Venice") is False # retryable + s.close() + + +def test_apply_non_trip(tmp_path): + s = _store(tmp_path) + cid = s.insert_cluster( + Cluster(start_at="2019-06-01", end_at="2019-06-02", status="non_trip"), + [ClusterMember(cluster_id=0, immich_id="a")]) + client = FakeImmich() + apply_cluster(client, s, cid) + assert ("id:_pipeline/non-trip", ["a"]) in client.tagged + assert ("id:_pipeline/processed", ["a"]) in client.tagged + s.close() + + +def test_apply_all_reports_per_cluster(tmp_path): + s = _store(tmp_path) + _approved(s, name="Venice") + client = FakeImmich() + results = apply_all(client, s) + assert len(results) == 1 and results[0]["status"] == "approved" + s.close() + + +def test_apply_skipped(tmp_path): + s = _store(tmp_path) + cid = s.insert_cluster( + Cluster(start_at="2019-06-01", end_at="2019-06-02", status="skipped"), + [ClusterMember(cluster_id=0, immich_id="a"), + ClusterMember(cluster_id=0, immich_id="b")]) + client = FakeImmich() + res = apply_cluster(client, s, cid) + # skipped: only _pipeline/processed is written, no content/non-trip tag + assert res["status"] == "skipped" + assert res["succeeded"] == [] and res["failed"] == [] + assert ("id:_pipeline/processed", ["a", "b"]) in client.tagged + assert len(client.tagged) == 1 # nothing but the processed tag + assert s.get_asset("a").processed is True + assert s.get_asset("b").processed is True + s.close()