51 lines
1.4 KiB
Python
51 lines
1.4 KiB
Python
import asyncio
|
|
import logging
|
|
|
|
import pydantic
|
|
from aiogram import Bot, Dispatcher, Router
|
|
|
|
|
|
class TelegramConfig(pydantic.BaseModel):
|
|
TOKEN: str
|
|
DROP_PENDING_UPDATES: bool = True
|
|
|
|
|
|
class TelegramHelper:
|
|
def __init__(self, config: TelegramConfig, *routers: Router) -> None:
|
|
self.bot: Bot = Bot(token=config.TOKEN)
|
|
self.dispatcher: Dispatcher = Dispatcher()
|
|
self._polling_task: asyncio.Task | None = None
|
|
self._drop_pending_updates: bool = config.DROP_PENDING_UPDATES
|
|
|
|
for router in routers:
|
|
self.dispatcher.include_router(router)
|
|
|
|
logging.info('Telegram bot initialized')
|
|
|
|
async def start_polling(self) -> None:
|
|
"""Start bot polling in background"""
|
|
if self._polling_task is not None:
|
|
logging.warning('Bot polling already started')
|
|
return
|
|
|
|
self._polling_task = asyncio.create_task(
|
|
self.dispatcher.start_polling(self.bot, drop_pending_updates=self._drop_pending_updates)
|
|
)
|
|
|
|
logging.info('Telegram bot polling started')
|
|
|
|
async def stop_polling(self) -> None:
|
|
"""Stop bot polling and cleanup"""
|
|
if self._polling_task is None:
|
|
logging.warning('Bot polling not started')
|
|
return
|
|
|
|
self._polling_task.cancel()
|
|
|
|
if self.bot.session is not None:
|
|
await self.bot.session.close()
|
|
|
|
self._polling_task = None
|
|
|
|
logging.info('Telegram bot stopped')
|