feat(immich): ImmichClient read — search_assets, list_tags, resolve_tag_id, download_thumbnail

shared/photoflow/immich/__init__.py

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

shared/photoflow/immich/client.py

@@ -0,0 +1,74 @@
+import requests
+
+
+def _normalize(item: dict) -> dict:
+    exif = item.get("exifInfo") or {}
+    lat = exif.get("latitude")
+    lon = exif.get("longitude")
+    return {
+        "id": item["id"],
+        "original_filename": item.get("originalFileName", ""),
+        "taken_at": item.get("localDateTime", ""),
+        "gps_lat": float(lat) if lat is not None else None,
+        "gps_lon": float(lon) if lon is not None else None,
+        "place_city": exif.get("city"),
+        "place_country": exif.get("country"),
+        "type": item.get("type", ""),
+        "tags": [t.get("name", "") for t in (item.get("tags") or [])],
+        "rating": int(exif.get("rating") or 0),
+        "updated_at": item.get("updatedAt", ""),
+    }
+
+
+class ImmichClient:
+    def __init__(self, base_url: str, api_key: str, timeout: int = 30):
+        self.base_url = base_url.rstrip("/")
+        self.timeout = timeout
+        self.session = requests.Session()
+        self.session.headers.update({"x-api-key": api_key, "Accept": "application/json"})
+
+    def _url(self, path: str) -> str:
+        return f"{self.base_url}{path}"
+
+    def list_tags(self) -> list[dict]:
+        r = self.session.get(self._url("/api/tags"), timeout=self.timeout)
+        r.raise_for_status()
+        return r.json()
+
+    def resolve_tag_id(self, name: str):
+        for tag in self.list_tags():
+            if tag.get("value") == name or tag.get("name") == name:
+                return tag["id"]
+        return None
+
+    def search_assets(self, *, taken_after=None, taken_before=None,
+                      tag_ids=None, updated_after=None) -> list[dict]:
+        body = {"withExif": True}
+        if taken_after:
+            body["takenAfter"] = taken_after
+        if taken_before:
+            body["takenBefore"] = taken_before
+        if tag_ids:
+            body["tagIds"] = tag_ids
+        if updated_after:
+            body["updatedAfter"] = updated_after
+        out = []
+        page = 1
+        while True:
+            payload = dict(body, size=1000, page=page)
+            r = self.session.post(self._url("/api/search/metadata"),
+                                  json=payload, timeout=self.timeout)
+            r.raise_for_status()
+            block = r.json().get("assets", {})
+            out.extend(_normalize(it) for it in block.get("items", []))
+            nxt = block.get("nextPage")
+            if not nxt:
+                break
+            page = int(nxt)
+        return out
+
+    def download_thumbnail(self, asset_id: str) -> bytes:
+        r = self.session.get(self._url(f"/api/assets/{asset_id}/thumbnail?size=preview"),
+                             timeout=self.timeout)
+        r.raise_for_status()
+        return r.content