feat: comprehensive code review fixes — security, performance, quality
Backend security: - Reject Gitea webhooks when webhook_secret is empty (was silently skipping) - Add slowapi rate limiting on login (5/min) and setup (3/min) endpoints - Add CORS middleware with configurable origins - Mask telegram_webhook_secret in settings API response - Protect system-owned command template configs from regular user modification - Increase minimum password length to 8 characters Backend performance: - Batch queries in _resolve_command_context (3 queries instead of 3N) - Concurrent album fetching with asyncio.gather in immich commands - Singleton Jinja2 SandboxedEnvironment (reuse instead of per-render creation) - TTLCache for rate limits (bounded memory, auto-eviction) - Optional aiohttp session reuse in send_reply/send_media_group Backend code quality: - Extract dispatch_helpers.py (shared link_data loading + event filtering) - Extract database/seeds.py from main.py (490 lines → dedicated module) - Split immich_handler.py (415 lines) into commands/immich/ subpackage - Replace bare except blocks with logged warnings - Add per-provider config validation (Pydantic models) - Truncate command input to 512 chars - Expose usage_* and desc_* slots in capabilities and variables API Frontend security: - CSS.escape() for user-controlled querySelector in highlight.ts - Client-side password min 8 chars validation on setup and password change Frontend code quality: - Replace any types with proper interfaces across top files - Decompose targets/+page.svelte into TargetForm + ReceiverSection - Fix $derived.by usage, $state mutation patterns - Add console.warn to empty catch blocks Frontend UX: - Auth redirect via goto() with "Redirecting..." state - Platform-aware Ctrl/Cmd K keyboard hint - Remove stat-card hover transform Frontend accessibility: - Modal: role=dialog, aria-modal, focus trap, restore focus - EntitySelect/IconGridSelect: listbox/option roles, aria-selected/disabled
This commit is contained in:
@@ -0,0 +1,5 @@
|
||||
"""Immich command handler subpackage."""
|
||||
|
||||
from .handler import ImmichCommandHandler
|
||||
|
||||
__all__ = ["ImmichCommandHandler"]
|
||||
@@ -0,0 +1,113 @@
|
||||
"""Album-related Immich bot commands: albums, favorites, summary."""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import asyncio
|
||||
import logging
|
||||
from typing import Any
|
||||
|
||||
import aiohttp
|
||||
|
||||
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
|
||||
|
||||
_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
|
||||
|
||||
results = await asyncio.gather(
|
||||
*[immich.client.get_album(aid) for aid in album_ids],
|
||||
return_exceptions=True,
|
||||
)
|
||||
for album_id, result in zip(album_ids, 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:
|
||||
albums_data.append({
|
||||
"name": result.name, "asset_count": result.asset_count, "id": album_id,
|
||||
})
|
||||
|
||||
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({
|
||||
"id": asset.id, "originalFileName": asset.filename,
|
||||
"type": asset.type,
|
||||
})
|
||||
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]],
|
||||
) -> str:
|
||||
"""Handle /summary command with concurrent album fetching."""
|
||||
if not all_album_ids:
|
||||
return _render_cmd_template(cmd_templates, "summary", locale, {"albums": []})
|
||||
|
||||
results = await asyncio.gather(
|
||||
*[client.get_album(aid) for aid in all_album_ids],
|
||||
return_exceptions=True,
|
||||
)
|
||||
|
||||
albums_data: list[dict] = []
|
||||
for album_id, result in zip(all_album_ids, results):
|
||||
if isinstance(result, Exception):
|
||||
_LOGGER.warning("Failed to fetch album %s: %s", album_id, result)
|
||||
continue
|
||||
if result:
|
||||
albums_data.append({
|
||||
"name": result.name, "asset_count": result.asset_count, "id": album_id,
|
||||
})
|
||||
|
||||
return _render_cmd_template(cmd_templates, "summary", locale, {"albums": albums_data})
|
||||
@@ -0,0 +1,49 @@
|
||||
"""Shared helpers, imports, and constants for Immich command handlers."""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import logging
|
||||
from typing import Any
|
||||
|
||||
from ...services import make_immich_provider
|
||||
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",
|
||||
}
|
||||
|
||||
|
||||
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 | list[dict[str, Any]]:
|
||||
"""Format asset results as text or media payload."""
|
||||
if not assets:
|
||||
return _render_cmd_template(cmd_templates, "no_results", locale, {"command": cmd, "query": query})
|
||||
|
||||
if response_mode == "media":
|
||||
media_items = []
|
||||
for asset in assets:
|
||||
asset_id = asset.get("id", "")
|
||||
filename = asset.get("originalFileName", "")
|
||||
year = asset.get("year", "")
|
||||
caption = f"{filename} ({year})" if year else filename
|
||||
media_items.append({
|
||||
"type": "photo",
|
||||
"asset_id": asset_id,
|
||||
"caption": caption,
|
||||
"thumbnail_url": f"{client.url}/api/assets/{asset_id}/thumbnail?size=preview",
|
||||
"api_key": client.api_key,
|
||||
})
|
||||
return media_items
|
||||
|
||||
slot_map = {"find": "search", "person": "search", "place": "search"}
|
||||
slot_name = slot_map.get(cmd, cmd)
|
||||
return _render_cmd_template(cmd_templates, slot_name, locale, {
|
||||
"assets": assets, "query": query, "command": cmd, "count": len(assets),
|
||||
})
|
||||
@@ -0,0 +1,199 @@
|
||||
"""Event-related Immich bot commands: events, latest, memory, random."""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import asyncio
|
||||
import logging
|
||||
import random as rng
|
||||
from datetime import datetime, timezone
|
||||
from typing import Any
|
||||
|
||||
from sqlmodel import select
|
||||
from sqlmodel.ext.asyncio.session import AsyncSession
|
||||
|
||||
from ...database.engine import get_engine
|
||||
from ...database.models import (
|
||||
EventLog, NotificationTarget, NotificationTrackerTarget,
|
||||
ServiceProvider, TelegramBot, TrackingConfig,
|
||||
)
|
||||
from ..handler import _get_notification_trackers_for_providers, _render_cmd_template
|
||||
from .common import _format_assets
|
||||
|
||||
_LOGGER = logging.getLogger(__name__)
|
||||
|
||||
|
||||
async def _cmd_events(
|
||||
bot: TelegramBot, providers_map: dict[int, ServiceProvider],
|
||||
count: int, locale: str,
|
||||
) -> dict[str, Any]:
|
||||
provider_ids = set(providers_map.keys())
|
||||
trackers = await _get_notification_trackers_for_providers(provider_ids)
|
||||
tracker_ids = [t.id for t in trackers]
|
||||
if not tracker_ids:
|
||||
return {"events": []}
|
||||
|
||||
engine = get_engine()
|
||||
async with AsyncSession(engine) as session:
|
||||
result = await session.exec(
|
||||
select(EventLog)
|
||||
.where(EventLog.tracker_id.in_(tracker_ids))
|
||||
.order_by(EventLog.created_at.desc())
|
||||
.limit(count)
|
||||
)
|
||||
events = result.all()
|
||||
|
||||
events_data = [
|
||||
{"type": e.event_type, "album": e.collection_name,
|
||||
"count": e.assets_count, "date": e.created_at.strftime("%m/%d %H:%M")}
|
||||
for e in events
|
||||
]
|
||||
return {"events": events_data}
|
||||
|
||||
|
||||
async def cmd_latest(
|
||||
client: Any, all_album_ids: list[str], count: int,
|
||||
locale: str, response_mode: str,
|
||||
cmd_templates: dict[str, dict[str, str]],
|
||||
) -> str | list[dict[str, Any]]:
|
||||
"""Handle /latest command with concurrent album fetching."""
|
||||
album_ids = all_album_ids[:10]
|
||||
if not album_ids:
|
||||
return _format_assets([], "latest", "", locale, response_mode, client, cmd_templates)
|
||||
|
||||
results = await asyncio.gather(
|
||||
*[client.get_album(aid) for aid in album_ids],
|
||||
return_exceptions=True,
|
||||
)
|
||||
|
||||
latest_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())[:count]:
|
||||
latest_assets.append({
|
||||
"id": asset.id, "originalFileName": asset.filename,
|
||||
"type": asset.type, "createdAt": asset.created_at,
|
||||
})
|
||||
|
||||
latest_assets.sort(key=lambda a: a.get("createdAt", ""), reverse=True)
|
||||
return _format_assets(latest_assets[:count], "latest", "", locale, response_mode, client, cmd_templates)
|
||||
|
||||
|
||||
async def cmd_random(
|
||||
client: Any, all_album_ids: list[str], count: int,
|
||||
locale: str, response_mode: str,
|
||||
cmd_templates: dict[str, dict[str, str]],
|
||||
) -> str | list[dict[str, Any]]:
|
||||
"""Handle /random command with concurrent album fetching."""
|
||||
album_ids = all_album_ids[:10]
|
||||
if not album_ids:
|
||||
return _format_assets([], "random", "", locale, response_mode, client, cmd_templates)
|
||||
|
||||
results = await asyncio.gather(
|
||||
*[client.get_album(aid) for aid in album_ids],
|
||||
return_exceptions=True,
|
||||
)
|
||||
|
||||
random_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:
|
||||
asset_list = list(result.assets.values())
|
||||
sampled = rng.sample(asset_list, min(count, len(asset_list)))
|
||||
for asset in sampled:
|
||||
random_assets.append({
|
||||
"id": asset.id, "originalFileName": asset.filename,
|
||||
"type": asset.type,
|
||||
})
|
||||
|
||||
rng.shuffle(random_assets)
|
||||
return _format_assets(random_assets[:count], "random", "", locale, response_mode, client, cmd_templates)
|
||||
|
||||
|
||||
async def _check_native_memory(bot: TelegramBot) -> bool:
|
||||
"""Check if any tracker-target linked to this bot uses native memory source."""
|
||||
engine = get_engine()
|
||||
async with AsyncSession(engine) as session:
|
||||
result = await session.exec(
|
||||
select(NotificationTarget).where(
|
||||
NotificationTarget.type == "telegram",
|
||||
NotificationTarget.user_id == bot.user_id,
|
||||
)
|
||||
)
|
||||
targets = result.all()
|
||||
bot_target_ids = {t.id for t in targets if t.config.get("bot_token") == bot.token}
|
||||
if not bot_target_ids:
|
||||
return False
|
||||
tt_result = await session.exec(
|
||||
select(NotificationTrackerTarget).where(
|
||||
NotificationTrackerTarget.target_id.in_(bot_target_ids)
|
||||
)
|
||||
)
|
||||
for tt in tt_result.all():
|
||||
if tt.tracking_config_id:
|
||||
tc = await session.get(TrackingConfig, tt.tracking_config_id)
|
||||
if tc and tc.memory_source == "native":
|
||||
return True
|
||||
return False
|
||||
|
||||
|
||||
async def cmd_memory(
|
||||
bot: TelegramBot, client: Any, all_album_ids: list[str], count: int,
|
||||
locale: str, response_mode: str,
|
||||
cmd_templates: dict[str, dict[str, str]],
|
||||
) -> str | list[dict[str, Any]]:
|
||||
"""Handle /memory command with concurrent album fetching."""
|
||||
use_native = await _check_native_memory(bot)
|
||||
today = datetime.now(timezone.utc)
|
||||
memory_assets: list[dict[str, Any]] = []
|
||||
|
||||
if use_native:
|
||||
memories = await client.get_memories()
|
||||
tracked_ids = set(all_album_ids) if all_album_ids else None
|
||||
for mem in memories:
|
||||
year = mem.get("data", {}).get("year")
|
||||
for raw_asset in mem.get("assets", []):
|
||||
if tracked_ids:
|
||||
asset_albums = raw_asset.get("albums", [])
|
||||
if not any(a.get("id") in tracked_ids for a in asset_albums):
|
||||
continue
|
||||
memory_assets.append({
|
||||
"id": raw_asset.get("id", ""),
|
||||
"originalFileName": raw_asset.get("originalFileName", ""),
|
||||
"type": raw_asset.get("type", "IMAGE"),
|
||||
"createdAt": raw_asset.get("fileCreatedAt", raw_asset.get("createdAt", "")),
|
||||
"year": year,
|
||||
})
|
||||
else:
|
||||
album_ids = all_album_ids[:10]
|
||||
if album_ids:
|
||||
results = await asyncio.gather(
|
||||
*[client.get_album(aid) for aid in album_ids],
|
||||
return_exceptions=True,
|
||||
)
|
||||
month_day = (today.month, today.day)
|
||||
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 result.assets.items():
|
||||
try:
|
||||
dt = datetime.fromisoformat(asset.created_at.replace("Z", "+00:00"))
|
||||
if (dt.month, dt.day) == month_day and dt.year != today.year:
|
||||
memory_assets.append({
|
||||
"id": asset.id, "originalFileName": asset.filename,
|
||||
"type": asset.type, "createdAt": asset.created_at,
|
||||
"year": dt.year,
|
||||
})
|
||||
except (ValueError, AttributeError):
|
||||
pass
|
||||
|
||||
memory_assets = memory_assets[:count]
|
||||
if not memory_assets:
|
||||
return _render_cmd_template(cmd_templates, "no_results", locale, {"command": "memory", "query": ""})
|
||||
return _format_assets(memory_assets, "memory", "", locale, response_mode, client, cmd_templates)
|
||||
@@ -0,0 +1,168 @@
|
||||
"""Immich-specific bot command handler — main dispatch class."""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import logging
|
||||
from typing import Any
|
||||
|
||||
import aiohttp
|
||||
from sqlmodel import select
|
||||
from sqlmodel.ext.asyncio.session import AsyncSession
|
||||
|
||||
from ...database.engine import get_engine
|
||||
from ...database.models import (
|
||||
CommandConfig, CommandTracker, EventLog,
|
||||
ServiceProvider, TelegramBot,
|
||||
)
|
||||
from ...services import make_immich_provider
|
||||
from ..base import ProviderCommandHandler
|
||||
from ..handler import _get_notification_trackers_for_providers, _render_cmd_template
|
||||
from .albums import _cmd_albums, cmd_favorites, cmd_summary
|
||||
from .common import _IMMICH_COMMANDS
|
||||
from .events import _cmd_events, cmd_latest, cmd_memory, cmd_random
|
||||
from .search import cmd_find, cmd_person, cmd_place, cmd_search
|
||||
|
||||
_LOGGER = logging.getLogger(__name__)
|
||||
|
||||
|
||||
async def _cmd_status(
|
||||
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)
|
||||
active = sum(1 for t in trackers if t.enabled)
|
||||
total = len(trackers)
|
||||
total_albums = sum(len(t.collection_ids or []) for t in trackers)
|
||||
|
||||
engine = get_engine()
|
||||
async with AsyncSession(engine) as session:
|
||||
result = await session.exec(
|
||||
select(EventLog).order_by(EventLog.created_at.desc()).limit(1)
|
||||
)
|
||||
last_event = result.first()
|
||||
last_str = last_event.created_at.strftime("%Y-%m-%d %H:%M") if last_event else "-"
|
||||
|
||||
return {
|
||||
"trackers_active": active, "trackers_total": total,
|
||||
"total_albums": total_albums, "last_event": last_str,
|
||||
}
|
||||
|
||||
|
||||
async def _cmd_people(
|
||||
providers_map: dict[int, ServiceProvider], locale: str,
|
||||
) -> dict[str, Any]:
|
||||
all_people: dict[str, str] = {}
|
||||
async with aiohttp.ClientSession() as http:
|
||||
for provider in providers_map.values():
|
||||
if provider.type != "immich":
|
||||
continue
|
||||
immich = make_immich_provider(http, provider)
|
||||
people = await immich.client.get_people()
|
||||
all_people.update(people)
|
||||
names = sorted(all_people.values())
|
||||
return {"people": names}
|
||||
|
||||
|
||||
class ImmichCommandHandler(ProviderCommandHandler):
|
||||
"""Handles all Immich-specific bot commands."""
|
||||
|
||||
provider_type = "immich"
|
||||
|
||||
def get_provider_commands(self) -> set[str]:
|
||||
return _IMMICH_COMMANDS
|
||||
|
||||
def get_rate_categories(self) -> dict[str, str]:
|
||||
return {
|
||||
"search": "search", "find": "search", "person": "search",
|
||||
"place": "search", "favorites": "search", "people": "search",
|
||||
}
|
||||
|
||||
async def handle(
|
||||
self,
|
||||
cmd: str,
|
||||
args: str,
|
||||
count: int,
|
||||
locale: str,
|
||||
response_mode: str,
|
||||
providers_map: dict[int, ServiceProvider],
|
||||
cmd_templates: dict[str, dict[str, str]],
|
||||
bot: TelegramBot,
|
||||
ctx_tuples: list[tuple[CommandTracker, CommandConfig, ServiceProvider]],
|
||||
) -> str | list[dict[str, Any]] | None:
|
||||
if cmd == "status":
|
||||
ctx = await _cmd_status(bot, providers_map, locale)
|
||||
return _render_cmd_template(cmd_templates, "status", locale, ctx)
|
||||
if cmd == "albums":
|
||||
ctx = await _cmd_albums(bot, providers_map, locale)
|
||||
return _render_cmd_template(cmd_templates, "albums", locale, ctx)
|
||||
if cmd == "events":
|
||||
ctx = await _cmd_events(bot, providers_map, count, locale)
|
||||
return _render_cmd_template(cmd_templates, "events", locale, ctx)
|
||||
if cmd == "people":
|
||||
ctx = await _cmd_people(providers_map, locale)
|
||||
return _render_cmd_template(cmd_templates, "people", locale, ctx)
|
||||
if cmd in ("search", "find", "person", "place", "latest",
|
||||
"random", "favorites", "summary", "memory"):
|
||||
return await _cmd_immich(
|
||||
bot, cmd, args, count, locale, response_mode,
|
||||
providers_map, cmd_templates,
|
||||
)
|
||||
return None
|
||||
|
||||
|
||||
async def _cmd_immich(
|
||||
bot: TelegramBot, cmd: str, args: str, count: int, locale: str,
|
||||
response_mode: str, providers_map: dict[int, ServiceProvider],
|
||||
cmd_templates: dict[str, dict[str, str]],
|
||||
) -> str | list[dict[str, Any]]:
|
||||
"""Handle commands that need Immich API access and may return media."""
|
||||
if not providers_map:
|
||||
return _render_cmd_template(cmd_templates, "no_results", locale, {"command": cmd, "query": args})
|
||||
|
||||
provider_ids = set(providers_map.keys())
|
||||
notification_trackers = await _get_notification_trackers_for_providers(provider_ids)
|
||||
|
||||
all_album_ids: list[str] = []
|
||||
for t in notification_trackers:
|
||||
all_album_ids.extend(t.collection_ids or [])
|
||||
|
||||
provider: ServiceProvider | None = None
|
||||
for p in providers_map.values():
|
||||
if p.type == "immich":
|
||||
provider = p
|
||||
break
|
||||
if not provider:
|
||||
return _render_cmd_template(cmd_templates, "no_results", locale, {"command": cmd, "query": args})
|
||||
|
||||
async with aiohttp.ClientSession() as http:
|
||||
immich = make_immich_provider(http, provider)
|
||||
client = immich.client
|
||||
|
||||
if cmd == "search":
|
||||
return await cmd_search(client, args, all_album_ids, count, locale, response_mode, cmd_templates)
|
||||
|
||||
if cmd == "find":
|
||||
return await cmd_find(client, args, all_album_ids, count, locale, response_mode, cmd_templates)
|
||||
|
||||
if cmd == "person":
|
||||
return await cmd_person(client, args, count, locale, response_mode, cmd_templates)
|
||||
|
||||
if cmd == "place":
|
||||
return await cmd_place(client, args, all_album_ids, count, locale, response_mode, cmd_templates)
|
||||
|
||||
if cmd == "favorites":
|
||||
return await cmd_favorites(bot, providers_map, all_album_ids, count, locale, response_mode, client, cmd_templates)
|
||||
|
||||
if cmd == "latest":
|
||||
return await cmd_latest(client, all_album_ids, count, locale, response_mode, cmd_templates)
|
||||
|
||||
if cmd == "random":
|
||||
return await cmd_random(client, all_album_ids, count, locale, response_mode, cmd_templates)
|
||||
|
||||
if cmd == "summary":
|
||||
return await cmd_summary(client, all_album_ids, locale, cmd_templates)
|
||||
|
||||
if cmd == "memory":
|
||||
return await cmd_memory(bot, client, all_album_ids, count, locale, response_mode, cmd_templates)
|
||||
|
||||
return None
|
||||
@@ -0,0 +1,66 @@
|
||||
"""Search-related Immich bot commands: search, find, person, place."""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
from typing import Any
|
||||
|
||||
from ..handler import _render_cmd_template
|
||||
from .common import _format_assets
|
||||
|
||||
|
||||
async def cmd_search(
|
||||
client: Any, args: str, all_album_ids: list[str], count: int,
|
||||
locale: str, response_mode: str,
|
||||
cmd_templates: dict[str, dict[str, str]],
|
||||
) -> str | list[dict[str, Any]]:
|
||||
"""Handle /search command."""
|
||||
if not args:
|
||||
return _render_cmd_template(cmd_templates, "no_results", locale, {"command": "search", "query": ""})
|
||||
assets = await client.search_smart(args, album_ids=all_album_ids, limit=count)
|
||||
return _format_assets(assets, "search", args, locale, response_mode, client, cmd_templates)
|
||||
|
||||
|
||||
async def cmd_find(
|
||||
client: Any, args: str, all_album_ids: list[str], count: int,
|
||||
locale: str, response_mode: str,
|
||||
cmd_templates: dict[str, dict[str, str]],
|
||||
) -> str | list[dict[str, Any]]:
|
||||
"""Handle /find command."""
|
||||
if not args:
|
||||
return _render_cmd_template(cmd_templates, "no_results", locale, {"command": "find", "query": ""})
|
||||
assets = await client.search_metadata(args, album_ids=all_album_ids, limit=count)
|
||||
return _format_assets(assets, "find", args, locale, response_mode, client, cmd_templates)
|
||||
|
||||
|
||||
async def cmd_person(
|
||||
client: Any, args: str, count: int,
|
||||
locale: str, response_mode: str,
|
||||
cmd_templates: dict[str, dict[str, str]],
|
||||
) -> str | list[dict[str, Any]]:
|
||||
"""Handle /person command."""
|
||||
if not args:
|
||||
return _render_cmd_template(cmd_templates, "no_results", locale, {"command": "person", "query": ""})
|
||||
people = await client.get_people()
|
||||
person_id = None
|
||||
for pid, pname in people.items():
|
||||
if args.lower() in pname.lower():
|
||||
person_id = pid
|
||||
break
|
||||
if not person_id:
|
||||
return _render_cmd_template(cmd_templates, "no_results", locale, {"command": "person", "query": args})
|
||||
assets = await client.search_by_person(person_id, limit=count)
|
||||
return _format_assets(assets, "person", args, locale, response_mode, client, cmd_templates)
|
||||
|
||||
|
||||
async def cmd_place(
|
||||
client: Any, args: str, all_album_ids: list[str], count: int,
|
||||
locale: str, response_mode: str,
|
||||
cmd_templates: dict[str, dict[str, str]],
|
||||
) -> str | list[dict[str, Any]]:
|
||||
"""Handle /place command."""
|
||||
if not args:
|
||||
return _render_cmd_template(cmd_templates, "no_results", locale, {"command": "place", "query": ""})
|
||||
assets = await client.search_smart(
|
||||
f"photos taken in {args}", album_ids=all_album_ids, limit=count
|
||||
)
|
||||
return _format_assets(assets, "place", args, locale, response_mode, client, cmd_templates)
|
||||
Reference in New Issue
Block a user