haos-hacs-immich-album-watcher/custom_components/immich_album_watcher/sync.py

"""Optional sync with the standalone Immich Watcher server."""

from __future__ import annotations

import logging
from typing import Any

import aiohttp

from homeassistant.core import HomeAssistant
from homeassistant.helpers.aiohttp_client import async_get_clientsession

_LOGGER = logging.getLogger(__name__)


class ServerSyncClient:
    """Client for communicating with the standalone Immich Watcher server.

    All methods are safe to call even if the server is unreachable --
    they log warnings and return empty/default values. The HA integration
    must never break due to server connectivity issues.
    """

    def __init__(self, hass: HomeAssistant, server_url: str, api_key: str) -> None:
        self._hass = hass
        self._base_url = server_url.rstrip("/")
        self._api_key = api_key

    @property
    def _headers(self) -> dict[str, str]:
        return {"X-API-Key": self._api_key, "Content-Type": "application/json"}

    async def async_get_trackers(self) -> list[dict[str, Any]]:
        """Fetch tracker configurations from the server.

        Returns empty list on any error.
        """
        try:
            session = async_get_clientsession(self._hass)
            async with session.get(
                f"{self._base_url}/api/sync/trackers",
                headers=self._headers,
            ) as response:
                if response.status == 200:
                    return await response.json()
                _LOGGER.warning(
                    "Server sync: failed to fetch trackers (HTTP %d)", response.status
                )
        except aiohttp.ClientError as err:
            _LOGGER.warning("Server sync: connection failed: %s", err)
        return []

    async def async_render_template(
        self, template_id: int, context: dict[str, Any]
    ) -> str | None:
        """Render a server-managed template with context.

        Returns None on any error.
        """
        try:
            session = async_get_clientsession(self._hass)
            async with session.post(
                f"{self._base_url}/api/sync/templates/{template_id}/render",
                headers=self._headers,
                json={"context": context},
            ) as response:
                if response.status == 200:
                    data = await response.json()
                    return data.get("rendered")
                _LOGGER.warning(
                    "Server sync: template render failed (HTTP %d)", response.status
                )
        except aiohttp.ClientError as err:
            _LOGGER.warning("Server sync: template render connection failed: %s", err)
        return None

    async def async_report_event(
        self,
        tracker_name: str,
        event_type: str,
        album_id: str,
        album_name: str,
        details: dict[str, Any] | None = None,
    ) -> bool:
        """Report a detected event to the server for logging.

        Returns True if successfully reported, False on any error.
        Fire-and-forget -- failures are logged but don't affect HA operation.
        """
        try:
            session = async_get_clientsession(self._hass)
            payload = {
                "tracker_name": tracker_name,
                "event_type": event_type,
                "album_id": album_id,
                "album_name": album_name,
                "details": details or {},
            }
            async with session.post(
                f"{self._base_url}/api/sync/events",
                headers=self._headers,
                json=payload,
            ) as response:
                if response.status == 200:
                    _LOGGER.debug("Server sync: event reported for album '%s'", album_name)
                    return True
                _LOGGER.debug(
                    "Server sync: event report failed (HTTP %d)", response.status
                )
        except aiohttp.ClientError as err:
            _LOGGER.debug("Server sync: event report connection failed: %s", err)
        return False

    async def async_check_connection(self) -> bool:
        """Check if the server is reachable."""
        try:
            session = async_get_clientsession(self._hass)
            async with session.get(
                f"{self._base_url}/api/health",
            ) as response:
                return response.status == 200
        except aiohttp.ClientError:
            return False