152 lines
5.6 KiB
Python
152 lines
5.6 KiB
Python
import logging
|
|
|
|
from aiogram import Bot
|
|
from aiogram.types import InlineKeyboardButton, InlineKeyboardMarkup
|
|
from aiogram.client.default import DefaultBotProperties
|
|
|
|
from config import Config
|
|
from typing import Dict, Any, List
|
|
|
|
logger = logging.getLogger(__name__)
|
|
|
|
|
|
class NotificationBot:
|
|
def __init__(self, config: Config):
|
|
self.config = config
|
|
|
|
try:
|
|
self.bot: Bot = Bot(
|
|
token=config.bot_token,
|
|
default=DefaultBotProperties(
|
|
parse_mode="HTML",
|
|
disable_notification=True,
|
|
link_preview_is_disabled=True,
|
|
),
|
|
)
|
|
logger.info(
|
|
"Бот для уведомлений инициализирован (admin_id=%s)", config.admin_id
|
|
)
|
|
|
|
except:
|
|
logger.exception("Не удалось инициализировать Бот для уведомлений")
|
|
raise
|
|
|
|
async def _notify(self, message: str, reply_to: int = None, link: str = None):
|
|
try:
|
|
logger.debug("Отправка уведомления (reply_to=%s, link=%s)", reply_to, link)
|
|
keyboard = None
|
|
if link:
|
|
keyboard = InlineKeyboardMarkup(
|
|
inline_keyboard=[[InlineKeyboardButton(text=f"Перейти", url=link)]]
|
|
)
|
|
|
|
if not reply_to:
|
|
sent_message = await self.bot.send_message(
|
|
chat_id=self.config.admin_id, text=message, reply_markup=keyboard
|
|
)
|
|
else:
|
|
sent_message = await self.bot.send_message(
|
|
chat_id=self.config.admin_id,
|
|
text=message,
|
|
reply_to_message_id=reply_to,
|
|
reply_markup=keyboard,
|
|
)
|
|
|
|
logger.debug(
|
|
"Уведомление отправлено (message_id=%s)", sent_message.message_id
|
|
)
|
|
return sent_message.message_id
|
|
|
|
except Exception as e:
|
|
logger.error("Ошибка отправки уведомления: %s", e)
|
|
|
|
async def new_post(self, post: Dict[str, Any], links: List[str]):
|
|
try:
|
|
thread = post.get("thread")
|
|
|
|
thread_title = thread.get("thread_title")
|
|
permalink = post.get("links").get("permalink")
|
|
thread_link = f'<a href="{permalink}">{thread_title}</a>'
|
|
|
|
poster_username = post.get("poster_username")
|
|
poster_user_id = post.get("poster_user_id")
|
|
poster_link = f"https://lolz.team/members/{poster_user_id}"
|
|
user_link = f'<a href="{poster_link}">{poster_username}</a>'
|
|
|
|
logger.info(
|
|
"Новый пост: пользователь=%s (id=%s), тема=%r, ссылок=%d",
|
|
poster_username,
|
|
poster_user_id,
|
|
thread_title,
|
|
len(links),
|
|
)
|
|
|
|
links_msg = f"<blockquote>"
|
|
for idx, link in enumerate(links):
|
|
links_msg += f'<a href="{link}">Ссылка №{idx + 1}</a>\n'
|
|
links_msg += "</blockquote>"
|
|
|
|
message = (
|
|
f"🚩 Новый ответ в теме {thread_link}:\n\n"
|
|
f"▪️ <b>Пользователь</b>: {user_link}\n"
|
|
f"▪️ <b>Ссылки</b>: {links_msg}"
|
|
)
|
|
|
|
post_message_id = await self._notify(message=message, link=permalink)
|
|
logger.debug(
|
|
"Уведомление о новом посте отправлено (message_id=%s)", post_message_id
|
|
)
|
|
return post_message_id
|
|
|
|
except:
|
|
logger.exception("Ошибка в new_post")
|
|
raise
|
|
|
|
async def success(self, link: str, post_message_id: int):
|
|
try:
|
|
logger.info(
|
|
"Звезда успешно отправлена: link=%s, stars=%s",
|
|
link,
|
|
self.config.stars_count,
|
|
)
|
|
message = (
|
|
f"🌟 Звезда отправлена:\n\n"
|
|
f"▪️ <b>Количество</b>: {self.config.stars_count}\n"
|
|
f'▪️ <b>Ссылка на сообщение</b>: <a href="{link}">тык</a>'
|
|
)
|
|
|
|
success_message_id = await self._notify(
|
|
message=message, reply_to=post_message_id, link=link
|
|
)
|
|
logger.debug(
|
|
"Уведомление об успехе отправлено (message_id=%s)", success_message_id
|
|
)
|
|
return success_message_id
|
|
|
|
except:
|
|
logger.exception("Ошибка в success")
|
|
raise
|
|
|
|
async def failure(self, post: Dict[str, Any], reason: str, post_message_id: int):
|
|
try:
|
|
permalink = post.get("links").get("permalink")
|
|
logger.warning(
|
|
"Не удалось отправить звезду: reason=%r, link=%s", reason, permalink
|
|
)
|
|
message = (
|
|
f"❌ Не удалось отправить звезду:\n\n"
|
|
f"▪️ <b>Причина</b>: <code>{reason}</code>"
|
|
)
|
|
|
|
failure_message_id = await self._notify(
|
|
message=message, reply_to=post_message_id, link=permalink
|
|
)
|
|
logger.debug(
|
|
"Уведомление об ошибке отправлено (message_id=%s)", failure_message_id
|
|
)
|
|
return failure_message_id
|
|
|
|
except:
|
|
logger.exception("Ошибка в failure")
|
|
raise
|