alexei.dolgolyov/notify-bridge
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases 31 Wiki Activity

feat: remove hardcoded command templates, enforce template system exclusively

Browse Source 
- Remove ALL hardcoded EN/RU fallback strings from handler.py — every
  command response now renders through CommandTemplateSlot templates
- _render_cmd_template now returns error placeholder instead of None
  when template is missing, ensuring no silent failures
- Fix register_commands_with_telegram tuple unpacking bug (was ignoring
  cmd_template_slots from _resolve_command_context)
- Auto-assign system default template (matching locale) on command
  config creation when none specified
- Add command_template_config_id to CommandConfigCreate model
- Remove "no template" option from frontend dropdown — template is
  now required for command configs
- Auto-select first matching template when creating new command config
- Fix || vs ?? for command_template_config_id, default_count, and
  rate_limits in frontend edit function (0 was treated as falsy)
This commit is contained in:
Admin
2026-03-21 22:53:07 +03:00
parent 3e3a6f0777
commit ddcbfdaa0b
3 changed files with 102 additions and 156 deletions
Download Patch File Download Diff File Expand all files Collapse all files
frontend/src/routes/command-configs/+page.svelte
+11 -5
View File
@@ -62,7 +62,14 @@
finally { loaded = true; }
}
function openNew() { form = defaultForm(); editing = null; showForm = true; }
function openNew() {
form = defaultForm();
// Auto-select first matching template for the default provider_type
const match = cmdTemplateConfigs.find((c: any) => c.provider_type === form.provider_type);
if (match) form.command_template_config_id = match.id;
editing = null;
showForm = true;
}
function editConfig(cfg: any) {
form = {
name: cfg.name,
@@ -71,9 +78,9 @@
enabled_commands: [...(cfg.enabled_commands || [])],
locale: cfg.locale || 'en',
response_mode: cfg.response_mode || 'media',
default_count: cfg.default_count || 5,
rate_limits: { search: cfg.rate_limits?.search || 30, default: cfg.rate_limits?.default || 10 },
command_template_config_id: cfg.command_template_config_id || null,
default_count: cfg.default_count ?? 5,
rate_limits: { search: cfg.rate_limits?.search ?? 30, default: cfg.rate_limits?.default ?? 10 },
command_template_config_id: cfg.command_template_config_id ?? null,
};
editing = cfg.id;
showForm = true;
@@ -167,7 +174,6 @@
<label for="cc-template" class="block text-sm font-medium mb-1">{t('commandConfig.responseTemplate')}</label>
<select id="cc-template" bind:value={form.command_template_config_id}
class="w-full px-3 py-2 border border-[var(--color-border)] rounded-md text-sm bg-[var(--color-background)]">
<option value={null}>— {t('commandConfig.noTemplate')} —</option>
{#each cmdTemplateConfigs.filter((c: any) => c.provider_type === form.provider_type) as tpl}
<option value={tpl.id}>{tpl.name}{tpl.user_id === 0 ? ' (System)' : ''}</option>
{/each}
packages/server/src/notify_bridge_server/api/command_configs.py
+41 -2
View File
@@ -10,7 +10,7 @@ from sqlmodel.ext.asyncio.session import AsyncSession
from ..auth.dependencies import get_current_user
from ..database.engine import get_session
from ..database.models import CommandConfig, CommandTracker, User
from ..database.models import CommandConfig, CommandTemplateConfig, CommandTracker, User
_LOGGER = logging.getLogger(__name__)
@@ -26,6 +26,7 @@ class CommandConfigCreate(BaseModel):
response_mode: str = "media"
default_count: int = 5
rate_limits: dict[str, Any] = {}
command_template_config_id: int | None = None
class CommandConfigUpdate(BaseModel):
@@ -36,6 +37,7 @@ class CommandConfigUpdate(BaseModel):
response_mode: str | None = None
default_count: int | None = None
rate_limits: dict[str, Any] | None = None
command_template_config_id: int | None = None
@router.get("")
@@ -65,7 +67,18 @@ async def create_command_config(
detail=f"Invalid provider_type. Must be one of: {', '.join(valid_types)}",
)
config = CommandConfig(user_id=user.id, **body.model_dump())
data = body.model_dump()
# Auto-assign system default template if none specified
if not data.get("command_template_config_id"):
locale = data.get("locale", "en")
provider_type = data.get("provider_type", "immich")
default_tpl = await _find_system_default_template(
session, provider_type, locale
)
if default_tpl:
data["command_template_config_id"] = default_tpl.id
config = CommandConfig(user_id=user.id, **data)
session.add(config)
await session.commit()
await session.refresh(config)
@@ -91,9 +104,13 @@ async def update_command_config(
session: AsyncSession = Depends(get_session),
):
"""Update a command config."""
from sqlalchemy.orm.attributes import flag_modified
config = await _get_user_config(session, config_id, user.id)
for field, value in body.model_dump(exclude_unset=True).items():
setattr(config, field, value)
# JSON columns need explicit dirty flag for SQLAlchemy change detection
if field in ("rate_limits", "enabled_commands"):
flag_modified(config, field)
session.add(config)
await session.commit()
await session.refresh(config)
@@ -129,6 +146,7 @@ def _config_response(c: CommandConfig) -> dict:
"response_mode": c.response_mode,
"default_count": c.default_count,
"rate_limits": c.rate_limits or {},
"command_template_config_id": c.command_template_config_id,
"created_at": c.created_at.isoformat(),
}
@@ -140,3 +158,24 @@ async def _get_user_config(
if not config or config.user_id != user_id:
raise HTTPException(status_code=404, detail="Command config not found")
return config
async def _find_system_default_template(
session: AsyncSession, provider_type: str, locale: str
) -> CommandTemplateConfig | None:
"""Find a system default (user_id=0) command template matching provider + locale."""
# Try exact locale match first (e.g. "Default Commands (EN)" for locale "en")
locale_upper = locale.upper()
result = await session.exec(
select(CommandTemplateConfig).where(
CommandTemplateConfig.user_id == 0,
CommandTemplateConfig.provider_type == provider_type,
)
)
templates = result.all()
# Match by locale suffix in name, e.g. "(EN)" or "(RU)"
for tpl in templates:
if f"({locale_upper})" in tpl.name:
return tpl
# Fallback: return first system template for this provider
return templates[0] if templates else None
packages/server/src/notify_bridge_server/commands/handler.py
+50 -149
View File
@@ -55,11 +55,12 @@ def _check_rate_limit(bot_id: int, chat_id: str, cmd: str, limits: dict[str, int
def _render_cmd_template(
templates: dict[str, str], slot_name: str, context: dict[str, Any]
) -> str | None:
"""Try to render a command template. Returns None if no template or error."""
) -> str:
"""Render a command template. Returns rendered string or error placeholder."""
template_str = templates.get(slot_name)
if not template_str:
return None
_LOGGER.warning("No command template found for slot '%s'", slot_name)
return f"[No template: {slot_name}]"
try:
from jinja2.sandbox import SandboxedEnvironment
env = SandboxedEnvironment(autoescape=False)
@@ -67,17 +68,15 @@ def _render_cmd_template(
return tmpl.render(**context)
except Exception as e:
_LOGGER.warning("Failed to render command template '%s': %s", slot_name, e)
return None
return f"[Template error: {slot_name}]"
async def _resolve_command_context(
bot: TelegramBot,
) -> list[tuple[CommandTracker, CommandConfig, ServiceProvider]]:
) -> tuple[list[tuple[CommandTracker, CommandConfig, ServiceProvider]], dict[str, str]]:
"""Resolve all enabled command trackers, configs, and providers for a bot.
Finds CommandTrackerListener rows where listener_type="telegram_bot"
and listener_id=bot.id, then loads the full chain:
CommandTrackerListener -> CommandTracker (enabled) -> CommandConfig + ServiceProvider.
Returns (context_tuples, cmd_template_slots).
"""
engine = get_engine()
async with AsyncSession(engine) as session:
@@ -91,7 +90,7 @@ async def _resolve_command_context(
listeners = result.all()
if not listeners:
return []
return [], {}
tuples: list[tuple[CommandTracker, CommandConfig, ServiceProvider]] = []
for listener in listeners:
@@ -161,14 +160,7 @@ async def handle_command(
enabled, locale, response_mode, default_count, rate_limits = _merge_command_context(ctx_tuples)
if cmd == "start":
result = _render_cmd_template(cmd_templates, "start", {"locale": locale, "bot_name": bot.name})
if result:
return result
msgs = {
"en": "Hi! I'm your Notify Bridge bot. Use /help to see available commands.",
"ru": "Привет! Я бот Notify Bridge. Используйте /help для списка команд.",
}
return msgs.get(locale, msgs["en"])
return _render_cmd_template(cmd_templates, "start", {"locale": locale, "bot_name": bot.name})
if cmd not in enabled and cmd != "start":
return None # Silently ignore disabled commands
@@ -176,14 +168,7 @@ async def handle_command(
# Rate limit check
wait = _check_rate_limit(bot.id, chat_id, cmd, rate_limits)
if wait is not None:
result = _render_cmd_template(cmd_templates, "rate_limited", {"wait": wait, "locale": locale})
if result:
return result
msgs = {
"en": f"Please wait {wait}s before using this command again.",
"ru": f"Подождите {wait} сек. перед повторным использованием.",
}
return msgs.get(locale, msgs["en"])
return _render_cmd_template(cmd_templates, "rate_limited", {"wait": wait, "locale": locale})
count = min(count_override or default_count, 20)
@@ -192,50 +177,39 @@ async def handle_command(
for _, _, provider in ctx_tuples:
providers_map[provider.id] = provider
# Dispatch — each handler returns (fallback_text, template_context)
# Template is tried first; if no template, fallback is returned.
# Dispatch — each handler returns template context dict
if cmd == "help":
fallback, ctx = _cmd_help(enabled, locale)
ctx = _cmd_help(enabled, locale)
elif cmd == "status":
fallback, ctx = await _cmd_status(bot, providers_map, locale)
ctx = await _cmd_status(bot, providers_map, locale)
elif cmd == "albums":
fallback, ctx = await _cmd_albums(bot, providers_map, locale)
ctx = await _cmd_albums(bot, providers_map, locale)
elif cmd == "events":
fallback, ctx = await _cmd_events(bot, providers_map, count, locale)
ctx = await _cmd_events(bot, providers_map, count, locale)
elif cmd == "people":
fallback, ctx = await _cmd_people(providers_map, locale)
ctx = await _cmd_people(providers_map, locale)
elif 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)
else:
return None
# Try template, fall back to hardcoded
rendered = _render_cmd_template(cmd_templates, cmd, {**ctx, "locale": locale})
return rendered if rendered else fallback
return _render_cmd_template(cmd_templates, cmd, {**ctx, "locale": locale})
def _cmd_help(enabled: list[str], locale: str) -> tuple[str, dict]:
def _cmd_help(enabled: list[str], locale: str) -> dict[str, Any]:
commands = []
lines = []
for cmd in enabled:
desc = COMMAND_DESCRIPTIONS.get(cmd, {})
desc_text = desc.get(locale, desc.get("en", ""))
commands.append({"name": cmd, "description": desc_text})
lines.append(f"/{cmd} — {desc_text}")
header = {"en": "Available commands:", "ru": "Доступные команды:"}
fallback = header.get(locale, header["en"]) + "\n" + "\n".join(lines)
return fallback, {"commands": commands}
return {"commands": commands}
async def _get_notification_trackers_for_providers(
provider_ids: set[int],
) -> list[NotificationTracker]:
"""Get notification trackers for the given provider IDs.
Used by commands like albums, events, status that need notification
tracker data (collection_ids, event logs).
"""
"""Get notification trackers for the given provider IDs."""
if not provider_ids:
return []
engine = get_engine()
@@ -273,7 +247,7 @@ async def _check_native_memory(bot: TelegramBot) -> bool:
return False
async def _cmd_status(bot: TelegramBot, providers_map: dict[int, ServiceProvider], locale: str) -> tuple[str, dict]:
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)
@@ -288,34 +262,16 @@ async def _cmd_status(bot: TelegramBot, providers_map: dict[int, ServiceProvider
last_event = result.first()
last_str = last_event.created_at.strftime("%Y-%m-%d %H:%M") if last_event else "-"
ctx = {"trackers_active": active, "trackers_total": total, "total_albums": total_albums, "last_event": last_str}
if locale == "ru":
fallback = (
f"📊 Статус\n"
f"Трекеры: {active}/{total} активных\n"
f"Альбомы: {total_albums}\n"
f"Последнее событие: {last_str}"
)
else:
fallback = (
f"📊 Status\n"
f"Trackers: {active}/{total} active\n"
f"Albums: {total_albums}\n"
f"Last event: {last_str}"
)
return fallback, ctx
return {"trackers_active": active, "trackers_total": total, "total_albums": total_albums, "last_event": last_str}
async def _cmd_albums(bot: TelegramBot, providers_map: dict[int, ServiceProvider], locale: str) -> tuple[str, dict]:
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:
fallback = "No tracked albums." if locale == "en" else "Нет отслеживаемых альбомов."
return fallback, {"albums": []}
return {"albums": []}
albums_data: list[dict] = []
lines = []
async with aiohttp.ClientSession() as http:
for tracker in trackers:
provider = providers_map.get(tracker.provider_id)
@@ -327,22 +283,18 @@ async def _cmd_albums(bot: TelegramBot, providers_map: dict[int, ServiceProvider
album = await immich.client.get_album(album_id)
if album:
albums_data.append({"name": album.name, "asset_count": album.asset_count, "id": album_id})
lines.append(f" • {album.name} ({album.asset_count} assets)")
except Exception:
lines.append(f" • {album_id[:8]}... (error)")
albums_data.append({"name": f"{album_id[:8]}...", "asset_count": "?", "id": album_id})
header = "📚 Tracked albums:" if locale == "en" else "📚 Отслеживаемые альбомы:"
fallback = header + "\n" + "\n".join(lines) if lines else header + "\n (none)"
return fallback, {"albums": albums_data}
return {"albums": albums_data}
async def _cmd_events(bot: TelegramBot, providers_map: dict[int, ServiceProvider], count: int, locale: str) -> tuple[str, dict]:
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:
fallback = "No events." if locale == "en" else "Нет событий."
return fallback, {"events": []}
return {"events": []}
engine = get_engine()
async with AsyncSession(engine) as session:
@@ -354,23 +306,13 @@ async def _cmd_events(bot: TelegramBot, providers_map: dict[int, ServiceProvider
)
events = result.all()
if not events:
fallback = "No events yet." if locale == "en" else "Пока нет событий."
return fallback, {"events": []}
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]
header = f"📋 Last {len(events)} events:" if locale == "en" else f"📋 Последние {len(events)} событий:"
lines = []
for e in events:
ts = e.created_at.strftime("%m/%d %H:%M")
lines.append(f" {ts} — {e.event_type}: {e.collection_name}")
fallback = header + "\n" + "\n".join(lines)
return fallback, {"events": events_data}
return {"events": events_data}
async def _cmd_people(providers_map: dict[int, ServiceProvider], locale: str) -> tuple[str, dict]:
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:
@@ -381,24 +323,18 @@ async def _cmd_people(providers_map: dict[int, ServiceProvider], locale: str) ->
people = await immich.client.get_people()
all_people.update(people)
if not all_people:
fallback = "No people detected." if locale == "en" else "Люди не обнаружены."
return fallback, {"people": []}
names = sorted(all_people.values())
header = f"👥 {len(names)} people:" if locale == "en" else f"👥 {len(names)} людей:"
fallback = header + "\n" + ", ".join(names)
return fallback, {"people": names}
return {"people": names}
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, str] | None = None,
cmd_templates: dict[str, str],
) -> str | list[dict[str, Any]]:
"""Handle commands that need Immich API access and may return media."""
if not providers_map:
return "No trackers configured." if locale == "en" else "Трекеры не настроены."
return _render_cmd_template(cmd_templates, "no_results", {"command": cmd, "query": args, "locale": locale})
# Get notification trackers for album data
provider_ids = set(providers_map.keys())
@@ -415,26 +351,27 @@ async def _cmd_immich(
provider = p
break
if not provider:
return "Server not found." if locale == "en" else "Сервер не найден."
return _render_cmd_template(cmd_templates, "no_results", {"command": cmd, "query": args, "locale": locale})
async with aiohttp.ClientSession() as http:
immich = make_immich_provider(http, provider)
client = immich.client
if cmd == "search":
if not args:
return "Usage: /search <query>" if locale == "en" else "Использование: /search <запрос>"
return _render_cmd_template(cmd_templates, "no_results", {"command": cmd, "query": "", "locale": locale})
assets = await client.search_smart(args, album_ids=all_album_ids, limit=count)
return _format_assets(assets, cmd, args, locale, response_mode, client, cmd_templates)
if cmd == "find":
if not args:
return "Usage: /find <text>" if locale == "en" else "Использование: /find <текст>"
return _render_cmd_template(cmd_templates, "no_results", {"command": cmd, "query": "", "locale": locale})
assets = await client.search_metadata(args, album_ids=all_album_ids, limit=count)
return _format_assets(assets, cmd, args, locale, response_mode, client, cmd_templates)
if cmd == "person":
if not args:
return "Usage: /person <name>" if locale == "en" else "Использование: /person <имя>"
return _render_cmd_template(cmd_templates, "no_results", {"command": "person", "query": "", "locale": locale})
people = await client.get_people()
person_id = None
for pid, pname in people.items():
@@ -442,13 +379,13 @@ async def _cmd_immich(
person_id = pid
break
if not person_id:
return f"Person '{args}' not found." if locale == "en" else f"Человек '{args}' не найден."
return _render_cmd_template(cmd_templates, "no_results", {"command": "person", "query": args, "locale": locale})
assets = await client.search_by_person(person_id, limit=count)
return _format_assets(assets, cmd, args, locale, response_mode, client, cmd_templates)
if cmd == "place":
if not args:
return "Usage: /place <location>" if locale == "en" else "Использование: /place <место>"
return _render_cmd_template(cmd_templates, "no_results", {"command": "place", "query": "", "locale": locale})
assets = await client.search_smart(
f"photos taken in {args}", album_ids=all_album_ids, limit=count
)
@@ -508,20 +445,14 @@ async def _cmd_immich(
if cmd == "summary":
albums_data: list[dict] = []
lines = []
for album_id in all_album_ids:
try:
album = await client.get_album(album_id)
if album:
albums_data.append({"name": album.name, "asset_count": album.asset_count, "id": album_id})
lines.append(f" • {album.name}: {album.asset_count} assets")
except Exception:
pass
rendered = _render_cmd_template(cmd_templates or {}, "summary", {"albums": albums_data, "locale": locale})
if rendered:
return rendered
header = f"📋 Album summary ({len(lines)}):" if locale == "en" else f"📋 Сводка альбомов ({len(lines)}):"
return header + "\n" + "\n".join(lines) if lines else header
return _render_cmd_template(cmd_templates, "summary", {"albums": albums_data, "locale": locale})
if cmd == "memory":
# Check if any linked tracking config uses native memories
@@ -537,7 +468,6 @@ async def _cmd_immich(
for mem in memories:
year = mem.get("data", {}).get("year")
for raw_asset in mem.get("assets", []):
# Optional album filtering
if tracked_ids:
asset_albums = raw_asset.get("albums", [])
if not any(a.get("id") in tracked_ids for a in asset_albums):
@@ -572,23 +502,20 @@ async def _cmd_immich(
memory_assets = memory_assets[:count]
if not memory_assets:
return "No memories for today." if locale == "en" else "Нет воспоминаний за сегодня."
return _render_cmd_template(cmd_templates, "no_results", {"command": "memory", "query": "", "locale": locale})
return _format_assets(memory_assets, cmd, "", locale, response_mode, client, cmd_templates)
return "Unknown command." if locale == "en" else "Неизвестная команда."
return None
def _format_assets(
assets: list[dict[str, Any]], cmd: str, query: str,
locale: str, response_mode: str, client: Any,
cmd_templates: dict[str, str] | None = None,
cmd_templates: dict[str, str],
) -> str | list[dict[str, Any]]:
"""Format asset results as text or media payload."""
if not assets:
rendered = _render_cmd_template(cmd_templates or {}, "no_results", {"command": cmd, "query": query, "locale": locale})
if rendered:
return rendered
return {"en": "No results found.", "ru": "Ничего не найдено."}.get(locale, "No results found.")
return _render_cmd_template(cmd_templates, "no_results", {"command": cmd, "query": query, "locale": locale})
if response_mode == "media":
media_items = []
@@ -606,34 +533,12 @@ def _format_assets(
})
return media_items
# Text mode — try template first
# Map command names to template slot names (search/find/person/place share "search" slot)
# Text mode — render via template
slot_map = {"find": "search", "person": "search", "place": "search"}
slot_name = slot_map.get(cmd, cmd)
rendered = _render_cmd_template(cmd_templates or {}, slot_name, {
return _render_cmd_template(cmd_templates, slot_name, {
"assets": assets, "query": query, "command": cmd, "count": len(assets), "locale": locale,
})
if rendered:
return rendered
# Hardcoded fallback
header_map = {
"search": {"en": f'🔍 Results for "{query}":', "ru": f'🔍 Результаты для "{query}":'},
"find": {"en": f'📄 Files matching "{query}":', "ru": f'📄 Файлы по запросу "{query}":'},
"person": {"en": f"👤 Photos of {query}:", "ru": f"👤 Фото {query}:"},
"place": {"en": f"📍 Photos from {query}:", "ru": f"📍 Фото из {query}:"},
"favorites": {"en": "⭐ Favorites:", "ru": "⭐ Избранное:"},
"latest": {"en": "📸 Latest:", "ru": "📸 Последние:"},
"random": {"en": "🎲 Random:", "ru": "🎲 Случайные:"},
"memory": {"en": "📅 On this day:", "ru": "📅 В этот день:"},
}
header = header_map.get(cmd, {}).get(locale, f"Results ({len(assets)}):")
lines = []
for a in assets:
name = a.get("originalFileName", a.get("id", "?")[:8])
year = a.get("year", "")
lines.append(f" • {name} ({year})" if year else f" • {name}")
return header + "\n" + "\n".join(lines)
async def send_media_group(
@@ -711,13 +616,9 @@ async def send_media_group(
async def register_commands_with_telegram(bot: TelegramBot) -> bool:
"""Register enabled commands with Telegram BotFather API.
Resolves all command trackers and configs for this bot, merges
enabled commands (union), and calls setMyCommands.
"""
ctx = await _resolve_command_context(bot)
enabled, locale, _, _, _ = _merge_command_context(ctx)
"""Register enabled commands with Telegram BotFather API."""
ctx_tuples, _ = await _resolve_command_context(bot)
enabled, locale, _, _, _ = _merge_command_context(ctx_tuples)
commands = []
for cmd in enabled: