241 lines
10 KiB
Python
241 lines
10 KiB
Python
from __future__ import annotations
|
|
|
|
from datetime import datetime, timezone
|
|
from pathlib import Path
|
|
from uuid import uuid4
|
|
import mimetypes
|
|
|
|
import httpx
|
|
from fastapi import HTTPException
|
|
|
|
from app.config import settings
|
|
|
|
|
|
def _api_base_url() -> str:
|
|
base = settings.immich_base_url.rstrip("/")
|
|
if not base.endswith("/api"):
|
|
base = f"{base}/api"
|
|
return base
|
|
|
|
|
|
class ImmichClient:
|
|
def __init__(self) -> None:
|
|
if not settings.immich_api_key:
|
|
raise HTTPException(status_code=500, detail="IMMICH_API_KEY manquante dans .env")
|
|
self.base_url = _api_base_url()
|
|
self.headers = {
|
|
"x-api-key": settings.immich_api_key,
|
|
"Accept": "application/json",
|
|
}
|
|
self.timeout = settings.request_timeout_seconds
|
|
|
|
async def list_albums(self) -> list[dict]:
|
|
async with httpx.AsyncClient(timeout=self.timeout) as client:
|
|
response = await client.get(f"{self.base_url}/albums", headers=self.headers)
|
|
self._raise_for_immich(response)
|
|
return response.json()
|
|
|
|
async def create_album(self, album_name: str, description: str | None = None) -> dict:
|
|
payload = {"albumName": album_name}
|
|
if description:
|
|
payload["description"] = description
|
|
async with httpx.AsyncClient(timeout=self.timeout) as client:
|
|
response = await client.post(
|
|
f"{self.base_url}/albums",
|
|
headers={**self.headers, "Content-Type": "application/json"},
|
|
json=payload,
|
|
)
|
|
self._raise_for_immich(response)
|
|
return response.json()
|
|
|
|
async def get_or_create_album(self, album_name: str, description: str | None = None) -> dict:
|
|
albums = await self.list_albums()
|
|
for album in albums:
|
|
if album.get("albumName") == album_name:
|
|
return album
|
|
return await self.create_album(album_name, description=description)
|
|
|
|
async def upload_asset(self, file_path: Path, original_filename: str) -> dict:
|
|
now = datetime.now(timezone.utc).isoformat()
|
|
mime_type = mimetypes.guess_type(original_filename)[0] or "application/octet-stream"
|
|
device_asset_id = f"{settings.upload_device_id}-{uuid4()}-{original_filename}"
|
|
|
|
data = {
|
|
"deviceAssetId": device_asset_id,
|
|
"deviceId": settings.upload_device_id,
|
|
"fileCreatedAt": now,
|
|
"fileModifiedAt": now,
|
|
"isFavorite": "false",
|
|
}
|
|
|
|
with file_path.open("rb") as file_handle:
|
|
files = {"assetData": (original_filename, file_handle, mime_type)}
|
|
async with httpx.AsyncClient(timeout=self.timeout) as client:
|
|
response = await client.post(
|
|
f"{self.base_url}/assets",
|
|
headers=self.headers,
|
|
data=data,
|
|
files=files,
|
|
)
|
|
self._raise_for_immich(response)
|
|
return response.json()
|
|
|
|
async def add_assets_to_album(self, album_id: str, asset_ids: list[str]) -> list[dict]:
|
|
payload = {"ids": asset_ids}
|
|
async with httpx.AsyncClient(timeout=self.timeout) as client:
|
|
response = await client.put(
|
|
f"{self.base_url}/albums/{album_id}/assets",
|
|
headers={**self.headers, "Content-Type": "application/json"},
|
|
json=payload,
|
|
)
|
|
self._raise_for_immich(response)
|
|
return response.json()
|
|
|
|
async def get_asset_thumbnail(self, asset_id: str, size: str = "preview") -> tuple[bytes, str]:
|
|
async with httpx.AsyncClient(timeout=self.timeout, follow_redirects=True) as client:
|
|
response = await client.get(
|
|
f"{self.base_url}/assets/{asset_id}/thumbnail",
|
|
headers={"x-api-key": settings.immich_api_key, "Accept": "image/*"},
|
|
params={"size": size},
|
|
)
|
|
self._raise_for_immich(response)
|
|
media_type = response.headers.get("content-type", "image/jpeg")
|
|
return response.content, media_type
|
|
|
|
async def get_asset_original(self, asset_id: str) -> tuple[bytes, str]:
|
|
async with httpx.AsyncClient(timeout=self.timeout, follow_redirects=True) as client:
|
|
response = await client.get(
|
|
f"{self.base_url}/assets/{asset_id}/original",
|
|
headers={"x-api-key": settings.immich_api_key, "Accept": "*/*"},
|
|
)
|
|
self._raise_for_immich(response)
|
|
media_type = response.headers.get("content-type", "application/octet-stream")
|
|
return response.content, media_type
|
|
|
|
|
|
async def list_people(self, q: str | None = None, with_hidden: bool = False, limit: int = 100) -> list[dict]:
|
|
"""Retourne les personnes reconnues par Immich.
|
|
|
|
Si q est fourni, on utilise /search/person. Sinon, on parcourt /people
|
|
avec pagination. Le parsing est volontairement tolérant car Immich a déjà
|
|
retourné soit une liste directe, soit un objet contenant une clé people.
|
|
"""
|
|
if q and q.strip():
|
|
return await self.search_people(q.strip(), with_hidden=with_hidden)
|
|
|
|
collected: list[dict] = []
|
|
page = 1
|
|
page_size = min(max(limit, 1), 100)
|
|
while len(collected) < limit:
|
|
async with httpx.AsyncClient(timeout=self.timeout) as client:
|
|
response = await client.get(
|
|
f"{self.base_url}/people",
|
|
headers=self.headers,
|
|
params={"page": page, "size": page_size, "withHidden": str(with_hidden).lower()},
|
|
)
|
|
self._raise_for_immich(response)
|
|
payload = response.json()
|
|
people = self._extract_people(payload)
|
|
if not people:
|
|
break
|
|
collected.extend(people)
|
|
has_next = bool(payload.get("hasNextPage")) if isinstance(payload, dict) else len(people) >= page_size
|
|
if not has_next or len(people) < page_size:
|
|
break
|
|
page += 1
|
|
return collected[:limit]
|
|
|
|
async def search_people(self, name: str, with_hidden: bool = False) -> list[dict]:
|
|
async with httpx.AsyncClient(timeout=self.timeout) as client:
|
|
response = await client.get(
|
|
f"{self.base_url}/search/person",
|
|
headers=self.headers,
|
|
params={"name": name, "withHidden": str(with_hidden).lower()},
|
|
)
|
|
self._raise_for_immich(response)
|
|
payload = response.json()
|
|
return self._extract_people(payload)
|
|
|
|
async def get_person(self, person_id: str) -> dict:
|
|
async with httpx.AsyncClient(timeout=self.timeout) as client:
|
|
response = await client.get(f"{self.base_url}/people/{person_id}", headers=self.headers)
|
|
self._raise_for_immich(response)
|
|
return response.json()
|
|
|
|
async def get_person_thumbnail(self, person_id: str) -> tuple[bytes, str]:
|
|
async with httpx.AsyncClient(timeout=self.timeout, follow_redirects=True) as client:
|
|
response = await client.get(
|
|
f"{self.base_url}/people/{person_id}/thumbnail",
|
|
headers={"x-api-key": settings.immich_api_key, "Accept": "image/*"},
|
|
)
|
|
self._raise_for_immich(response)
|
|
media_type = response.headers.get("content-type", "image/jpeg")
|
|
return response.content, media_type
|
|
|
|
|
|
async def list_person_assets(self, person_id: str, page: int = 1, size: int = 100) -> list[dict]:
|
|
"""Retourne les assets complets où Immich a reconnu cette personne.
|
|
|
|
Immich recommande maintenant /search/metadata pour filtrer les assets,
|
|
notamment par personIds, avec pagination.
|
|
"""
|
|
page = max(page, 1)
|
|
size = min(max(size, 1), 250)
|
|
payload = {
|
|
"personIds": [person_id],
|
|
"page": page,
|
|
"size": size,
|
|
}
|
|
async with httpx.AsyncClient(timeout=self.timeout) as client:
|
|
response = await client.post(
|
|
f"{self.base_url}/search/metadata",
|
|
headers={**self.headers, "Content-Type": "application/json"},
|
|
json=payload,
|
|
)
|
|
self._raise_for_immich(response)
|
|
return self._extract_assets(response.json())
|
|
|
|
async def get_asset_info(self, asset_id: str) -> dict:
|
|
async with httpx.AsyncClient(timeout=self.timeout) as client:
|
|
response = await client.get(f"{self.base_url}/assets/{asset_id}", headers=self.headers)
|
|
self._raise_for_immich(response)
|
|
return response.json()
|
|
|
|
@staticmethod
|
|
def _extract_assets(payload: object) -> list[dict]:
|
|
if isinstance(payload, list):
|
|
return [item for item in payload if isinstance(item, dict)]
|
|
if isinstance(payload, dict):
|
|
assets = payload.get("assets")
|
|
if isinstance(assets, list):
|
|
return [item for item in assets if isinstance(item, dict)]
|
|
if isinstance(assets, dict):
|
|
items = assets.get("items")
|
|
if isinstance(items, list):
|
|
return [item for item in items if isinstance(item, dict)]
|
|
for key in ("items", "results", "data"):
|
|
value = payload.get(key)
|
|
if isinstance(value, list):
|
|
return [item for item in value if isinstance(item, dict)]
|
|
return []
|
|
|
|
@staticmethod
|
|
def _extract_people(payload: object) -> list[dict]:
|
|
if isinstance(payload, list):
|
|
return [item for item in payload if isinstance(item, dict)]
|
|
if isinstance(payload, dict):
|
|
for key in ("people", "items", "results", "data"):
|
|
value = payload.get(key)
|
|
if isinstance(value, list):
|
|
return [item for item in value if isinstance(item, dict)]
|
|
return []
|
|
|
|
@staticmethod
|
|
def _raise_for_immich(response: httpx.Response) -> None:
|
|
if response.status_code >= 400:
|
|
try:
|
|
detail = response.json()
|
|
except Exception:
|
|
detail = response.text
|
|
raise HTTPException(status_code=response.status_code, detail={"immich_error": detail})
|