shared/photoflow/immich/__init__.py

@@ -1,3 +1,4 @@
 from photoflow.immich.client import ImmichClient
+from photoflow.immich import pipeline
 
-__all__ = ["ImmichClient"]
+__all__ = ["ImmichClient", "pipeline"]

shared/photoflow/immich/client.py

@@ -72,3 +72,22 @@ class ImmichClient:
                              timeout=self.timeout)
         r.raise_for_status()
         return r.content
+
+    def upsert_tag(self, name: str) -> str:
+        r = self.session.put(self._url("/api/tags"),
+                             json={"tags": [name]}, timeout=self.timeout)
+        r.raise_for_status()
+        for tag in r.json():
+            if tag.get("value") == name or tag.get("name") == name:
+                return tag["id"]
+        resolved = self.resolve_tag_id(name)
+        if resolved is None:
+            raise RuntimeError(f"upsert_tag: could not resolve id for {name!r}")
+        return resolved
+
+    def tag_assets(self, tag_id: str, asset_ids: list[str]) -> None:
+        if not asset_ids:
+            return
+        r = self.session.put(self._url(f"/api/tags/{tag_id}/assets"),
+                             json={"ids": asset_ids}, timeout=self.timeout)
+        r.raise_for_status()

shared/photoflow/immich/pipeline.py

@@ -0,0 +1,11 @@
+ROOT = "_pipeline"
+PROCESSED = f"{ROOT}/processed"
+NON_TRIP = f"{ROOT}/non-trip"
+
+
+def ai_rating(n: int) -> str:
+    return f"{ROOT}/ai-rating/{n}"
+
+
+def is_pipeline_tag(name: str) -> bool:
+    return name == ROOT or name.startswith(ROOT + "/")

shared/tests/test_immich_client.py

@@ -48,3 +48,16 @@ def test_download_thumbnail(httpserver):
         b"\xff\xd8\xffjpegbytes", content_type="image/jpeg")
     c = ImmichClient(httpserver.url_for(""), "k")
     assert c.download_thumbnail("a").startswith(b"\xff\xd8\xff")
+
+
+def test_upsert_tag_returns_id_by_value(httpserver):
+    httpserver.expect_request("/api/tags", method="PUT").respond_with_json(
+        [{"id": "p1", "name": "non-trip", "value": "_pipeline/non-trip"}])
+    c = ImmichClient(httpserver.url_for(""), "k")
+    assert c.upsert_tag("_pipeline/non-trip") == "p1"
+
+
+def test_tag_assets_posts_ids(httpserver):
+    httpserver.expect_request("/api/tags/p1/assets", method="PUT").respond_with_json({"ok": True})
+    c = ImmichClient(httpserver.url_for(""), "k")
+    c.tag_assets("p1", ["a", "b"])  # should not raise

shared/tests/test_pipeline.py

@@ -0,0 +1,15 @@
+from photoflow.immich import pipeline
+
+
+def test_constants_and_helpers():
+    assert pipeline.ROOT == "_pipeline"
+    assert pipeline.PROCESSED == "_pipeline/processed"
+    assert pipeline.NON_TRIP == "_pipeline/non-trip"
+    assert pipeline.ai_rating(4) == "_pipeline/ai-rating/4"
+
+
+def test_is_pipeline_tag():
+    assert pipeline.is_pipeline_tag("_pipeline/processed") is True
+    assert pipeline.is_pipeline_tag("_pipeline") is True
+    assert pipeline.is_pipeline_tag("Italy 2019") is False
+    assert pipeline.is_pipeline_tag("Kiev") is False