alexei.dolgolyov
d0bc767e98
- Command templates now match notification template style: type icons,
linked filenames via album shared links, location, favorite status
- Media mode sends text message first, then media as reply (was media-only)
- Search/find/person/place resolve asset public URLs from tracked albums'
shared links (share/{key}/photos/{id})
- Albums/summary commands include album public_url in context
- Enriched command template preview sample context with public_url, city,
country, is_favorite
- Extract sanitizePreview to shared lib/sanitize.ts
- Command template preview now renders HTML links (was raw text)
- Global provider filter moved above search in sidebar
- CLAUDE.md: template consistency + context variable sync rules
132 lines
5.0 KiB
Python
132 lines
5.0 KiB
Python
"""Album-related Immich bot commands: albums, favorites, summary."""
|
|
|
|
from __future__ import annotations
|
|
|
|
import asyncio
|
|
import logging
|
|
from typing import Any
|
|
|
|
import aiohttp
|
|
|
|
from notify_bridge_core.providers.immich.asset_utils import get_public_url
|
|
|
|
from ...database.models import ServiceProvider, TelegramBot
|
|
from ...services import make_immich_provider
|
|
from ..handler import _get_notification_trackers_for_providers, _render_cmd_template
|
|
from .common import _format_assets, build_asset_dict
|
|
|
|
_LOGGER = logging.getLogger(__name__)
|
|
|
|
|
|
async def _cmd_albums(
|
|
bot: TelegramBot, providers_map: dict[int, ServiceProvider], locale: str,
|
|
) -> dict[str, Any]:
|
|
provider_ids = set(providers_map.keys())
|
|
trackers = await _get_notification_trackers_for_providers(provider_ids)
|
|
if not trackers:
|
|
return {"albums": []}
|
|
|
|
albums_data: list[dict] = []
|
|
async with aiohttp.ClientSession() as http:
|
|
for tracker in trackers:
|
|
provider = providers_map.get(tracker.provider_id)
|
|
if not provider or provider.type != "immich":
|
|
continue
|
|
immich = make_immich_provider(http, provider)
|
|
album_ids = tracker.collection_ids or []
|
|
if not album_ids:
|
|
continue
|
|
|
|
ext_domain = (provider.config.get("external_domain") or provider.config.get("url", "")).rstrip("/")
|
|
album_results = await asyncio.gather(
|
|
*[immich.client.get_album(aid) for aid in album_ids],
|
|
return_exceptions=True,
|
|
)
|
|
link_results = await asyncio.gather(
|
|
*[immich.client.get_shared_links(aid) for aid in album_ids],
|
|
return_exceptions=True,
|
|
)
|
|
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)
|
|
albums_data.append({
|
|
"name": f"{album_id[:8]}...", "asset_count": "?", "id": album_id,
|
|
})
|
|
elif 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,
|
|
})
|
|
|
|
return {"albums": albums_data}
|
|
|
|
|
|
async def cmd_favorites(
|
|
bot: TelegramBot, providers_map: dict[int, ServiceProvider],
|
|
all_album_ids: list[str], count: int, locale: str,
|
|
response_mode: str, client: Any,
|
|
cmd_templates: dict[str, dict[str, str]],
|
|
) -> str | list[dict[str, Any]]:
|
|
"""Handle /favorites command with concurrent album fetching."""
|
|
album_ids = all_album_ids[:10]
|
|
if not album_ids:
|
|
return _format_assets([], "favorites", "", locale, response_mode, client, cmd_templates)
|
|
|
|
results = await asyncio.gather(
|
|
*[client.get_album(aid) for aid in album_ids],
|
|
return_exceptions=True,
|
|
)
|
|
|
|
fav_assets: list[dict[str, Any]] = []
|
|
for album_id, result in zip(album_ids, results):
|
|
if isinstance(result, Exception):
|
|
_LOGGER.warning("Failed to fetch album %s: %s", album_id, result)
|
|
continue
|
|
if result:
|
|
for aid, asset in list(result.assets.items())[:50]:
|
|
if asset.is_favorite and len(fav_assets) < count:
|
|
fav_assets.append(build_asset_dict(asset))
|
|
if len(fav_assets) >= count:
|
|
break
|
|
|
|
return _format_assets(fav_assets, "favorites", "", locale, response_mode, client, cmd_templates)
|
|
|
|
|
|
async def cmd_summary(
|
|
client: Any, all_album_ids: list[str], locale: str,
|
|
cmd_templates: dict[str, dict[str, str]],
|
|
external_domain: str = "",
|
|
) -> str:
|
|
"""Handle /summary command with concurrent album fetching."""
|
|
if not all_album_ids:
|
|
return _render_cmd_template(cmd_templates, "summary", locale, {"albums": []})
|
|
|
|
album_results = await asyncio.gather(
|
|
*[client.get_album(aid) for aid in all_album_ids],
|
|
return_exceptions=True,
|
|
)
|
|
link_results = await asyncio.gather(
|
|
*[client.get_shared_links(aid) for aid in all_album_ids],
|
|
return_exceptions=True,
|
|
)
|
|
ext = external_domain.rstrip("/")
|
|
|
|
albums_data: list[dict] = []
|
|
for album_id, result, links in zip(all_album_ids, album_results, link_results):
|
|
if isinstance(result, Exception):
|
|
_LOGGER.warning("Failed to fetch album %s: %s", album_id, result)
|
|
continue
|
|
if result:
|
|
pub_url = ""
|
|
if not isinstance(links, Exception) and ext:
|
|
pub_url = get_public_url(ext, links) or ""
|
|
albums_data.append({
|
|
"name": result.name, "asset_count": result.asset_count,
|
|
"id": album_id, "public_url": pub_url,
|
|
})
|
|
|
|
return _render_cmd_template(cmd_templates, "summary", locale, {"albums": albums_data})
|