feat(trip-cluster): idempotent write-back + CLI apply with confirmation

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
 

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]

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()