92 lines
3.1 KiB
Python
92 lines
3.1 KiB
Python
"""Telegram bot command sync service.
|
|
|
|
Synchronises @bot_command-decorated handlers with BotCommand DB table
|
|
and Telegram Bot API (set_my_commands) on boot or manual trigger.
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import logging
|
|
from typing import Any
|
|
|
|
from sqlalchemy import select, update
|
|
from sqlalchemy.ext.asyncio import AsyncSession
|
|
|
|
from app.integrations.telegram.client import TelegramBotClient
|
|
from app.integrations.telegram.decorators import get_registered_commands
|
|
from app.models.bot_command import BotCommand
|
|
|
|
logger = logging.getLogger("voidea.telegram.sync")
|
|
|
|
|
|
class TelegramBotSyncService:
|
|
"""Syncs registered bot commands to DB and Telegram API."""
|
|
|
|
def __init__(self, db: AsyncSession, bot_client: TelegramBotClient | None = None):
|
|
self.db = db
|
|
self.bot_client = bot_client
|
|
|
|
async def sync_to_db(self) -> list[BotCommand]:
|
|
"""Ensure all @bot_command handlers have matching BotCommand records."""
|
|
registered = get_registered_commands()
|
|
result = await self.db.execute(select(BotCommand))
|
|
existing = {cmd.name: cmd for cmd in result.scalars().all()}
|
|
|
|
synced: list[BotCommand] = []
|
|
|
|
for reg in registered:
|
|
if reg.name in existing:
|
|
cmd = existing[reg.name]
|
|
if cmd.description != reg.description or cmd.requires_auth != reg.requires_auth:
|
|
cmd.description = reg.description
|
|
cmd.requires_auth = reg.requires_auth
|
|
synced.append(cmd)
|
|
else:
|
|
cmd = BotCommand(
|
|
name=reg.name,
|
|
description=reg.description,
|
|
enabled=True,
|
|
requires_auth=reg.requires_auth,
|
|
)
|
|
self.db.add(cmd)
|
|
synced.append(cmd)
|
|
|
|
await self.db.commit()
|
|
for cmd in synced:
|
|
await self.db.refresh(cmd)
|
|
|
|
logger.info("Synced %d bot commands to DB", len(synced))
|
|
return synced
|
|
|
|
async def sync_to_telegram(self) -> bool:
|
|
"""Push enabled commands to Telegram Bot API via set_my_commands."""
|
|
if not self.bot_client or not self.bot_client.available:
|
|
logger.info("Telegram client not available — skipping set_my_commands")
|
|
return False
|
|
|
|
result = await self.db.execute(
|
|
select(BotCommand).where(BotCommand.enabled == True)
|
|
)
|
|
enabled = result.scalars().all()
|
|
|
|
commands = [
|
|
{"command": cmd.name.lstrip("/"), "description": cmd.description}
|
|
for cmd in enabled
|
|
]
|
|
|
|
ok = await self.bot_client.set_commands(commands)
|
|
if ok:
|
|
logger.info("Pushed %d commands to Telegram", len(commands))
|
|
else:
|
|
logger.warning("Failed to push commands to Telegram")
|
|
return ok
|
|
|
|
async def full_sync(self) -> dict[str, Any]:
|
|
"""Run full sync: DB → Telegram."""
|
|
synced = await self.sync_to_db()
|
|
telegram_ok = await self.sync_to_telegram()
|
|
return {
|
|
"synced_count": len(synced),
|
|
"telegram_sync": telegram_ok,
|
|
}
|