94 lines
3.4 KiB
Python
94 lines
3.4 KiB
Python
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
|
|
|
|
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()
|