215 lines
7.4 KiB
Python
215 lines
7.4 KiB
Python
"""Client helpers for talking to Immich."""
|
|
|
|
from __future__ import annotations
|
|
|
|
from dataclasses import dataclass
|
|
from datetime import datetime, timedelta, timezone
|
|
from typing import Any, Dict, List, Optional, Tuple
|
|
|
|
import requests
|
|
|
|
|
|
class ImmichError(RuntimeError):
|
|
"""Raised when the Immich API is unavailable or returns an error."""
|
|
|
|
|
|
@dataclass
|
|
class ImmichAsset:
|
|
"""Subset of metadata needed by the UI."""
|
|
|
|
id: str
|
|
file_name: str
|
|
captured_at: Optional[str]
|
|
thumbnail_url: str
|
|
preview_url: str
|
|
original_url: str
|
|
web_url: str
|
|
latitude: Optional[float]
|
|
longitude: Optional[float]
|
|
location_label: Optional[str]
|
|
|
|
|
|
class ImmichClient:
|
|
"""Lightweight wrapper around the Immich REST API."""
|
|
|
|
def __init__(self, base_url: str, api_key: str) -> None:
|
|
if not base_url:
|
|
raise ValueError("base_url is required")
|
|
if not api_key:
|
|
raise ValueError("api_key is required")
|
|
|
|
self.base_url = base_url.rstrip("/")
|
|
self.session = requests.Session()
|
|
self.session.headers.update(
|
|
{
|
|
"x-api-key": api_key,
|
|
"accept": "application/json",
|
|
}
|
|
)
|
|
|
|
def _request(self, method: str, path: str, **kwargs: Any) -> Any:
|
|
url = f"{self.base_url}{path}"
|
|
timeout = kwargs.pop("timeout", 15)
|
|
try:
|
|
response = self.session.request(method, url, timeout=timeout, **kwargs)
|
|
response.raise_for_status()
|
|
if response.content:
|
|
return response.json()
|
|
return {}
|
|
except (
|
|
requests.RequestException
|
|
) as exc: # pragma: no cover - network failure path
|
|
raise ImmichError(str(exc)) from exc
|
|
|
|
def _parse_coordinate(self, value: Any) -> Optional[float]:
|
|
if value is None:
|
|
return None
|
|
if isinstance(value, (int, float)):
|
|
return float(value)
|
|
try:
|
|
return float(value)
|
|
except (TypeError, ValueError):
|
|
return None
|
|
|
|
def _build_location_label(self, *sources: Dict[str, Any]) -> Optional[str]:
|
|
parts: List[str] = []
|
|
for source in sources:
|
|
if not source:
|
|
continue
|
|
for key in ("city", "state", "country"):
|
|
value = source.get(key)
|
|
if value and value not in parts:
|
|
parts.append(value)
|
|
return ", ".join(parts) if parts else None
|
|
|
|
def _build_asset(self, asset_data: Dict[str, Any]) -> ImmichAsset:
|
|
asset_id = asset_data.get("id")
|
|
if not asset_id:
|
|
raise ImmichError("Asset payload missing id")
|
|
file_name = (
|
|
asset_data.get("originalFileName") or asset_data.get("fileName") or "Photo"
|
|
)
|
|
captured_at = (
|
|
asset_data.get("fileCreatedAt")
|
|
or asset_data.get("createdAt")
|
|
or asset_data.get("exifInfo", {}).get("dateTimeOriginal")
|
|
)
|
|
exif_info = asset_data.get("exifInfo") or {}
|
|
position = asset_data.get("position") or asset_data.get("geolocation") or {}
|
|
latitude = self._parse_coordinate(
|
|
exif_info.get("latitude")
|
|
or exif_info.get("gpsLatitude")
|
|
or position.get("latitude")
|
|
)
|
|
longitude = self._parse_coordinate(
|
|
exif_info.get("longitude")
|
|
or exif_info.get("gpsLongitude")
|
|
or position.get("longitude")
|
|
)
|
|
location_label = self._build_location_label(exif_info, position)
|
|
|
|
thumbnail_url = (
|
|
f"{self.base_url}/api/assets/{asset_id}/thumbnail?size=thumbnail"
|
|
)
|
|
preview_url = f"{self.base_url}/api/assets/{asset_id}/thumbnail?size=preview"
|
|
original_url = f"{self.base_url}/api/assets/{asset_id}/original"
|
|
web_url = f"{self.base_url}/photos/{asset_id}"
|
|
return ImmichAsset(
|
|
id=asset_id,
|
|
file_name=file_name,
|
|
captured_at=captured_at,
|
|
thumbnail_url=thumbnail_url,
|
|
preview_url=preview_url,
|
|
original_url=original_url,
|
|
web_url=web_url,
|
|
latitude=latitude,
|
|
longitude=longitude,
|
|
location_label=location_label,
|
|
)
|
|
|
|
def get_recent_assets(self, days: int = 3, limit: int = 200) -> List[ImmichAsset]:
|
|
"""Fetch assets created in the last ``days`` days."""
|
|
|
|
since = datetime.now(timezone.utc) - timedelta(days=days)
|
|
since_iso = since.isoformat().replace("+00:00", "Z")
|
|
payload = {
|
|
"size": limit,
|
|
"page": 1,
|
|
"orderBy": "takenAt",
|
|
"orderDirection": "DESC",
|
|
"metadata": {
|
|
"types": ["IMAGE"],
|
|
"takenAfter": since_iso,
|
|
},
|
|
}
|
|
data = self._request("POST", "/api/search/metadata", json=payload)
|
|
|
|
def _normalize_items(value: Any) -> List[Any]:
|
|
if isinstance(value, list):
|
|
return value
|
|
if isinstance(value, dict):
|
|
for key in ("items", "results", "data", "assets"):
|
|
nested = value.get(key)
|
|
if isinstance(nested, list):
|
|
return nested
|
|
return []
|
|
|
|
items: List[Any] = []
|
|
for key in ("items", "assets", "results", "data"):
|
|
candidate = _normalize_items(data.get(key))
|
|
if candidate:
|
|
items = candidate
|
|
break
|
|
|
|
assets: List[ImmichAsset] = []
|
|
for item in items:
|
|
asset_payload: Optional[Any] = None
|
|
|
|
if isinstance(item, dict):
|
|
asset_payload = item.get("asset", item)
|
|
if isinstance(asset_payload, str):
|
|
asset_payload = {"id": asset_payload}
|
|
elif isinstance(item, str):
|
|
asset_payload = {"id": item}
|
|
|
|
if isinstance(asset_payload, dict):
|
|
# Some responses only send the id; fetch missing metadata.
|
|
if set(asset_payload.keys()) == {"id"}:
|
|
assets.append(self.get_asset(asset_payload["id"]))
|
|
else:
|
|
assets.append(self._build_asset(asset_payload))
|
|
|
|
return assets
|
|
|
|
def get_asset(self, asset_id: str) -> ImmichAsset:
|
|
"""Fetch a single asset."""
|
|
|
|
data = self._request("GET", f"/api/assets/{asset_id}")
|
|
return self._build_asset(data)
|
|
|
|
def fetch_asset_content(self, asset_id: str, variant: str) -> Tuple[bytes, str]:
|
|
"""Download binary image content for the requested variant."""
|
|
|
|
variant = variant.lower()
|
|
if variant == "thumbnail":
|
|
path = f"/api/assets/{asset_id}/thumbnail"
|
|
params = {"size": "thumbnail"}
|
|
elif variant == "preview":
|
|
path = f"/api/assets/{asset_id}/thumbnail"
|
|
params = {"size": "preview"}
|
|
elif variant == "original":
|
|
path = f"/api/assets/{asset_id}/original"
|
|
params = None
|
|
else:
|
|
raise ImmichError(f"Unsupported asset variant: {variant}")
|
|
|
|
url = f"{self.base_url}{path}"
|
|
try:
|
|
response = self.session.get(url, params=params, timeout=30)
|
|
response.raise_for_status()
|
|
except requests.RequestException as exc: # pragma: no cover
|
|
raise ImmichError(str(exc)) from exc
|
|
|
|
mime_type = response.headers.get("Content-Type", "application/octet-stream")
|
|
return response.content, mime_type
|