veenm
e39a3d93f7
A Home Assistant app with its own sidebar tab: free-form grid editor, live entity widgets (lights, climate, car, calendar, Immich photos, Afvalwijzer waste collection and more), themes, tabs with a side menu, kiosk mode and English/Dutch translations. Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
160 lines
6.8 KiB
Python
160 lines
6.8 KiB
Python
"""Read-only Immich access for the Photos widget.
|
||
|
||
The API key stays on the server: the browser only ever sees album names,
|
||
asset ids and image bytes proxied through these routes.
|
||
"""
|
||
|
||
import hashlib
|
||
import logging
|
||
import re
|
||
import time
|
||
|
||
import aiohttp
|
||
from aiohttp import web
|
||
|
||
LOG = logging.getLogger("dashboard_studio.immich")
|
||
ID_RE = re.compile(r"^[0-9a-fA-F-]{8,64}$")
|
||
CACHE_SECONDS = 600
|
||
NOT_CONFIGURED = "Immich isn’t set up. Add its URL and an API key on the app’s Configuration tab."
|
||
|
||
|
||
class Immich:
|
||
def __init__(self, url, api_key, mock=False):
|
||
self.url = (url or "").rstrip("/")
|
||
self.key = api_key or ""
|
||
self.mock = mock and not (self.url and self.key)
|
||
self.session = None
|
||
self._assets = {} # album id -> (fetched_at, [assets])
|
||
|
||
@property
|
||
def configured(self):
|
||
return self.mock or bool(self.url and self.key)
|
||
|
||
async def start(self):
|
||
self.session = aiohttp.ClientSession(
|
||
headers={"x-api-key": self.key, "Accept": "application/json"},
|
||
timeout=aiohttp.ClientTimeout(total=20),
|
||
)
|
||
|
||
async def close(self):
|
||
if self.session:
|
||
await self.session.close()
|
||
|
||
async def _json(self, method, path, **kwargs):
|
||
async with self.session.request(method, f"{self.url}/api{path}", **kwargs) as resp:
|
||
if resp.status in (401, 403):
|
||
raise web.HTTPBadGateway(text="Immich rejected the API key. Check it on the app’s Configuration tab.")
|
||
if resp.status >= 400:
|
||
raise web.HTTPBadGateway(text=f"Immich returned HTTP {resp.status} for {path}.")
|
||
return await resp.json()
|
||
|
||
# ------------------------------------------------------------ data
|
||
|
||
async def albums(self):
|
||
if self.mock:
|
||
return [
|
||
{"id": "11111111-aaaa-4bbb-8ccc-000000000001", "name": "Summer holiday", "count": 6},
|
||
{"id": "11111111-aaaa-4bbb-8ccc-000000000002", "name": "Family", "count": 4},
|
||
{"id": "11111111-aaaa-4bbb-8ccc-000000000003", "name": "Garden", "count": 3},
|
||
]
|
||
rows = await self._json("GET", "/albums")
|
||
shared = await self._json("GET", "/albums", params={"shared": "true"})
|
||
seen, out = set(), []
|
||
for a in rows + shared:
|
||
if a["id"] in seen:
|
||
continue
|
||
seen.add(a["id"])
|
||
out.append({"id": a["id"], "name": a.get("albumName") or "Untitled", "count": a.get("assetCount", 0)})
|
||
return sorted(out, key=lambda a: a["name"].lower())
|
||
|
||
async def album_assets(self, album_id):
|
||
cached = self._assets.get(album_id)
|
||
if cached and time.time() - cached[0] < CACHE_SECONDS:
|
||
return cached[1]
|
||
if self.mock:
|
||
n = int(album_id[-1]) + 2
|
||
assets = [{"id": f"{album_id[:-4]}{i:04d}", "date": f"2026-0{(i % 9) + 1}-1{i % 9}T12:00:00Z"} for i in range(n * 2)]
|
||
else:
|
||
assets = await self._fetch_album_assets(album_id)
|
||
self._assets[album_id] = (time.time(), assets)
|
||
return assets
|
||
|
||
async def _fetch_album_assets(self, album_id):
|
||
album = await self._json("GET", f"/albums/{album_id}")
|
||
items = album.get("assets")
|
||
if not items and album.get("assetCount"):
|
||
# Newer Immich versions leave assets out of the album response; search instead.
|
||
items, page = [], 1
|
||
while page and len(items) < 5000:
|
||
res = await self._json("POST", "/search/metadata", json={"albumIds": [album_id], "type": "IMAGE", "size": 1000, "page": page})
|
||
block = res.get("assets", {})
|
||
items += block.get("items", [])
|
||
page = block.get("nextPage")
|
||
page = int(page) if page else None
|
||
return [
|
||
{"id": a["id"], "date": a.get("localDateTime") or a.get("fileCreatedAt")}
|
||
for a in items or []
|
||
if a.get("type", "IMAGE") == "IMAGE" and not a.get("isTrashed")
|
||
]
|
||
|
||
async def preview(self, asset_id):
|
||
if self.mock:
|
||
return self._mock_image(asset_id), "image/svg+xml"
|
||
async with self.session.get(f"{self.url}/api/assets/{asset_id}/thumbnail", params={"size": "preview"}) as resp:
|
||
if resp.status >= 400:
|
||
raise web.HTTPBadGateway(text=f"Immich returned HTTP {resp.status} for that photo.")
|
||
return await resp.read(), resp.content_type or "image/jpeg"
|
||
|
||
@staticmethod
|
||
def _mock_image(asset_id):
|
||
h = hashlib.md5(asset_id.encode()).hexdigest()
|
||
c1, c2 = f"#{h[:6]}", f"#{h[6:12]}"
|
||
return f"""<svg xmlns="http://www.w3.org/2000/svg" viewBox="0 0 1200 800">
|
||
<defs><linearGradient id="g" x1="0" y1="0" x2="1" y2="1"><stop offset="0" stop-color="{c1}"/><stop offset="1" stop-color="{c2}"/></linearGradient></defs>
|
||
<rect width="1200" height="800" fill="url(#g)"/><circle cx="{300 + int(h[12:14], 16) * 3}" cy="260" r="110" fill="#fff" opacity=".35"/>
|
||
<path d="M0 640 L300 420 L520 600 L760 360 L1200 700 L1200 800 L0 800Z" fill="#000" opacity=".25"/>
|
||
<text x="40" y="760" fill="#fff" font-family="sans-serif" font-size="32" opacity=".8">Mock photo {asset_id[-4:]}</text></svg>""".encode()
|
||
|
||
# ------------------------------------------------------------ routes
|
||
|
||
def routes(self):
|
||
def guard():
|
||
if not self.configured:
|
||
raise web.HTTPServiceUnavailable(text=NOT_CONFIGURED)
|
||
|
||
async def albums(request):
|
||
guard()
|
||
try:
|
||
return web.json_response(await self.albums())
|
||
except aiohttp.ClientError as err:
|
||
LOG.error("Immich albums: %s", err)
|
||
raise web.HTTPBadGateway(text="Couldn’t reach Immich. Check the URL on the app’s Configuration tab.")
|
||
|
||
async def assets(request):
|
||
guard()
|
||
album_id = request.match_info["album_id"]
|
||
if not ID_RE.match(album_id):
|
||
raise web.HTTPBadRequest()
|
||
try:
|
||
return web.json_response(await self.album_assets(album_id))
|
||
except aiohttp.ClientError as err:
|
||
LOG.error("Immich album %s: %s", album_id, err)
|
||
raise web.HTTPBadGateway(text="Couldn’t reach Immich.")
|
||
|
||
async def preview(request):
|
||
guard()
|
||
asset_id = request.match_info["asset_id"]
|
||
if not ID_RE.match(asset_id):
|
||
raise web.HTTPBadRequest()
|
||
try:
|
||
body, ctype = await self.preview(asset_id)
|
||
except aiohttp.ClientError:
|
||
raise web.HTTPBadGateway(text="Couldn’t reach Immich.")
|
||
return web.Response(body=body, content_type=ctype, headers={"Cache-Control": "private, max-age=86400"})
|
||
|
||
return [
|
||
web.get("/api/immich/albums", albums),
|
||
web.get("/api/immich/albums/{album_id}/assets", assets),
|
||
web.get("/api/immich/assets/{asset_id}/preview", preview),
|
||
]
|