Phase 10: Telegram bot commands + Phase 11: Snackbar notifications
All checks were successful
Validate / Hassfest (push) Successful in 3s

Phase 10 — Telegram Bot Commands:
- Add commands_config JSON field to TelegramBot model (enabled cmds,
  default count, response mode, rate limits, locale)
- Create command handler with 14 commands: /status, /albums, /events,
  /summary, /latest, /memory, /random, /search, /find, /person,
  /place, /favorites, /people, /help
- Add search_smart, search_metadata, search_by_person, get_random,
  download_asset, get_asset_thumbnail to ImmichClient
- Auto-register commands with Telegram setMyCommands API (EN+RU)
- Rate limiting per chat per command category
- Media mode: download thumbnails and send as photos to Telegram
- Webhook handler routes /commands before falling through to AI chat
- Frontend: expandable Commands section per bot with checkboxes,
  count/mode/locale settings, rate limit inputs, sync button

Phase 11 — Snackbar Notifications:
- Create snackbar store (snackbar.svelte.ts) with $state rune
- Create Snackbar component with fly/fade transitions, typed colors
- Mount globally in +layout.svelte
- Replace all alert() calls with typed snackbar notifications
- Add success snacks to all CRUD operations across all pages
- 4 types: success (3s), error (5s), info (3s), warning (4s)
- Max 3 visible, auto-dismiss, manual dismiss via X button

Both: Add ~30 i18n keys (EN+RU) for commands UI and snack messages.

Co-Authored-By: Claude Opus 4.6 (1M context) <noreply@anthropic.com>
This commit is contained in:
2026-03-19 21:39:05 +03:00
parent ffce3ee337
commit e6ff0a423a
20 changed files with 1384 additions and 70 deletions

View File

@@ -358,5 +358,189 @@ class ImmichClient:
return False
async def search_smart(
self,
query: str,
album_ids: list[str] | None = None,
limit: int = 10,
) -> list[dict[str, Any]]:
"""Semantic search via Immich CLIP (smart search).
Args:
query: Natural language search query
album_ids: Optional list of album IDs to scope results to
limit: Max results to return
Returns:
List of asset dicts from search results
"""
payload: dict[str, Any] = {"query": query, "page": 1, "size": limit}
try:
async with self._session.post(
f"{self._url}/api/search/smart",
headers=self._json_headers,
json=payload,
) as response:
if response.status == 200:
data = await response.json()
items = data.get("assets", {}).get("items", [])
if album_ids:
# Post-filter: only keep assets from tracked albums
tracked = set(album_ids)
items = [
a for a in items
if any(
alb.get("id") in tracked
for alb in a.get("albums", [])
)
]
return items[:limit]
_LOGGER.warning("Smart search failed: HTTP %s", response.status)
except aiohttp.ClientError as err:
_LOGGER.warning("Smart search error: %s", err)
return []
async def search_metadata(
self,
query: str,
album_ids: list[str] | None = None,
limit: int = 10,
) -> list[dict[str, Any]]:
"""Search assets by metadata (filename, description).
Args:
query: Text to search for
album_ids: Optional list of album IDs to scope results to
limit: Max results to return
Returns:
List of asset dicts from search results
"""
payload: dict[str, Any] = {
"originalFileName": query,
"page": 1,
"size": limit,
}
try:
async with self._session.post(
f"{self._url}/api/search/metadata",
headers=self._json_headers,
json=payload,
) as response:
if response.status == 200:
data = await response.json()
items = data.get("assets", {}).get("items", [])
if album_ids:
tracked = set(album_ids)
items = [
a for a in items
if any(
alb.get("id") in tracked
for alb in a.get("albums", [])
)
]
return items[:limit]
_LOGGER.warning("Metadata search failed: HTTP %s", response.status)
except aiohttp.ClientError as err:
_LOGGER.warning("Metadata search error: %s", err)
return []
async def search_by_person(
self,
person_id: str,
limit: int = 10,
) -> list[dict[str, Any]]:
"""Find assets containing a specific person.
Args:
person_id: Immich person ID
limit: Max results to return
Returns:
List of asset dicts
"""
try:
async with self._session.get(
f"{self._url}/api/people/{person_id}/assets",
headers=self._headers,
) as response:
if response.status == 200:
data = await response.json()
return data[:limit]
_LOGGER.warning("Person assets failed: HTTP %s", response.status)
except aiohttp.ClientError as err:
_LOGGER.warning("Person assets error: %s", err)
return []
async def get_random_assets(
self,
count: int = 5,
) -> list[dict[str, Any]]:
"""Get random assets from Immich.
Args:
count: Number of random assets to return
Returns:
List of asset dicts
"""
try:
async with self._session.get(
f"{self._url}/api/assets/random",
headers=self._headers,
params={"count": count},
) as response:
if response.status == 200:
return await response.json()
_LOGGER.warning("Random assets failed: HTTP %s", response.status)
except aiohttp.ClientError as err:
_LOGGER.warning("Random assets error: %s", err)
return []
async def download_asset(self, asset_id: str) -> bytes | None:
"""Download an asset's original file.
Args:
asset_id: The asset ID to download
Returns:
Raw bytes of the asset, or None on failure
"""
try:
async with self._session.get(
f"{self._url}/api/assets/{asset_id}/original",
headers=self._headers,
) as response:
if response.status == 200:
return await response.read()
_LOGGER.warning("Asset download failed: HTTP %s", response.status)
except aiohttp.ClientError as err:
_LOGGER.warning("Asset download error: %s", err)
return None
async def get_asset_thumbnail(self, asset_id: str, size: str = "preview") -> bytes | None:
"""Download an asset's thumbnail/preview.
Args:
asset_id: The asset ID
size: "thumbnail" (small) or "preview" (larger)
Returns:
Raw bytes of the thumbnail, or None on failure
"""
try:
async with self._session.get(
f"{self._url}/api/assets/{asset_id}/thumbnail",
headers=self._headers,
params={"size": size},
) as response:
if response.status == 200:
return await response.read()
_LOGGER.warning("Thumbnail download failed: HTTP %s", response.status)
except aiohttp.ClientError as err:
_LOGGER.warning("Thumbnail download error: %s", err)
return None
class ImmichApiError(Exception):
"""Raised when an Immich API call fails."""