1import logging
 2
 3from aiogram import Bot
 4from aiogram.client.default import DefaultBotProperties
 5from aiogram.client.session.aiohttp import AiohttpSession
 6from aiogram.enums import ParseMode
 7from aiogram.exceptions import TelegramAPIError
 8
 9from app.config import TELEGRAM_PROXY_URL
10from app.config.env import (
11    TELEGRAM_API_TOKEN,
12    TELEGRAM_ADMIN_ID,
13    TELEGRAM_LOGGER_CHANNEL_ID,
14)
15from app.models.notification import Notification
16from app.notification.helper import create_text
17
18logger = logging.getLogger(__name__)
19
20
21class BotManager:
22    _instance = None
23
24    @classmethod
25    async def get_instance(cls):
26        if cls._instance is None and TELEGRAM_API_TOKEN:
27            if TELEGRAM_PROXY_URL:
28                session = AiohttpSession(proxy=TELEGRAM_PROXY_URL)
29            else:
30                session = None
31
32            cls._instance = Bot(
33                token=TELEGRAM_API_TOKEN,
34                default=DefaultBotProperties(parse_mode=ParseMode.HTML),
35                session=session,
36            )
37            try:
38                await cls._instance.get_me()
39            except:
40                logger.error("Telegram API token is not valid.")
41        return cls._instance
42
43
44async def send_message(
45    message: str,
46    parse_mode=ParseMode.HTML,
47):
48    if not (bot := await BotManager.get_instance()):
49        return
50
51    for recipient_id in (TELEGRAM_ADMIN_ID or []) + [
52        TELEGRAM_LOGGER_CHANNEL_ID
53    ]:
54        if not recipient_id:
55            continue
56        try:
57            await bot.send_message(
58                recipient_id,
59                message,
60                parse_mode=parse_mode,
61            )
62        except TelegramAPIError as e:
63            logger.error(e)
64
65
66async def send_notification(notif: Notification):
67    text = create_text(notif)
68    await send_message(text)