notify-bridge/packages/server/src/notify_bridge_server/commands/immich/common.py

"""Shared helpers, imports, and constants for Immich command handlers."""

from __future__ import annotations

import asyncio
import logging
from typing import Any

from notify_bridge_core.providers.immich.asset_utils import get_public_url

from ..handler import _render_cmd_template

_LOGGER = logging.getLogger(__name__)

_IMMICH_COMMANDS = {
    "status", "albums", "events", "people",
    "search", "find", "person", "place",
    "latest", "random", "favorites", "summary", "memory",
}


async def fetch_albums_with_links(
    client: Any,
    album_ids: list[str],
    ext_domain: str,
    *,
    include_failed: bool = True,
) -> list[dict[str, Any]]:
    """Fetch albums and their shared links concurrently.

    Returns a list of album data dicts with keys: name, asset_count, id,
    public_url, and ``_album`` (the raw album object for callers that need
    asset-level access).

    When *include_failed* is True, albums that fail to fetch are included
    with placeholder data (``"?"`` for counts).  When False, they are
    silently skipped.
    """
    album_results = await asyncio.gather(
        *[client.get_album(aid) for aid in album_ids],
        return_exceptions=True,
    )
    link_results = await asyncio.gather(
        *[client.get_shared_links(aid) for aid in album_ids],
        return_exceptions=True,
    )

    albums_data: list[dict[str, Any]] = []
    for album_id, result, links in zip(album_ids, album_results, link_results):
        if isinstance(result, Exception):
            _LOGGER.warning("Failed to fetch album %s: %s", album_id, result)
            if include_failed:
                albums_data.append({
                    "name": f"{album_id[:8]}...", "asset_count": "?",
                    "id": album_id, "public_url": "", "_album": None,
                })
            continue
        if result:
            pub_url = ""
            if not isinstance(links, Exception) and ext_domain:
                pub_url = get_public_url(ext_domain, links) or ""
            albums_data.append({
                "name": result.name, "asset_count": result.asset_count,
                "id": album_id, "public_url": pub_url, "_album": result,
            })
    return albums_data


def build_asset_dict(
    asset: Any,
    *,
    public_url: str = "",
    year: int | None = None,
) -> dict[str, Any]:
    """Build a rich asset dict for command templates from an ImmichAssetInfo or raw dict."""
    if isinstance(asset, dict):
        d = {
            "id": asset.get("id", ""),
            "originalFileName": asset.get("originalFileName", asset.get("filename", "")),
            "type": asset.get("type", "IMAGE"),
            "createdAt": asset.get("createdAt", asset.get("created_at", asset.get("fileCreatedAt", ""))),
            "city": asset.get("city", ""),
            "country": asset.get("country", ""),
            "is_favorite": asset.get("is_favorite", asset.get("isFavorite", False)),
            "public_url": asset.get("public_url", public_url),
        }
        if year or asset.get("year"):
            d["year"] = year or asset.get("year")
        return d
    # ImmichAssetInfo dataclass
    return {
        "id": asset.id,
        "originalFileName": asset.filename,
        "type": asset.type,
        "createdAt": asset.created_at,
        "city": getattr(asset, "city", "") or "",
        "country": getattr(asset, "country", "") or "",
        "is_favorite": getattr(asset, "is_favorite", False),
        "public_url": public_url,
        **({"year": year} if year else {}),
    }


def _format_assets(
    assets: list[dict[str, Any]], cmd: str, query: str,
    locale: str, response_mode: str, client: Any,
    cmd_templates: dict[str, dict[str, str]],
) -> str | dict[str, Any]:
    """Format asset results as text or a text-plus-media payload.

    Returns:
        str: rendered text when *response_mode* is ``"text"`` (or no assets).
        dict: ``{"text": ..., "media": [...]}`` when *response_mode* is
              ``"media"`` and assets are present.
    """
    if not assets:
        return _render_cmd_template(cmd_templates, "no_results", locale, {"command": cmd, "query": query})

    slot_map = {"find": "search", "person": "search", "place": "search"}
    slot_name = slot_map.get(cmd, cmd)
    text = _render_cmd_template(cmd_templates, slot_name, locale, {
        "assets": assets, "query": query, "command": cmd, "count": len(assets),
    })

    if response_mode == "media":
        media_items: list[dict[str, Any]] = []
        for asset in assets:
            asset_id = asset.get("id", "")
            media_items.append({
                "type": "photo",
                "asset_id": asset_id,
                "caption": "",
                "thumbnail_url": f"{client.url}/api/assets/{asset_id}/thumbnail?size=preview",
                "api_key": client.api_key,
            })
        # Return text message + media items — text is sent first, media as reply
        return {"text": text, "media": media_items}

    return text