117 lines
3.8 KiB
Python
117 lines
3.8 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,
|
|
),
|
|
)
|
|
|
|
except:
|
|
raise
|
|
|
|
async def _notify(self, message: str, reply_to: int, link: str = None):
|
|
try:
|
|
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,
|
|
)
|
|
|
|
return sent_message.message_id
|
|
|
|
except Exception as e:
|
|
logger.error(f"Ошибка отправки уведомления: {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>'
|
|
|
|
links_msg = f"<blockquote>\n"
|
|
for link in links:
|
|
links_msg += f"{link}\n"
|
|
|
|
links_msg += "</blockquote>"
|
|
|
|
message = (
|
|
f"🚩 Новый ответ в теме {thread_link}:\n"
|
|
f"▪️ Пользователь: {user_link}\n"
|
|
f"▪️ Ссылки: {links_msg}"
|
|
)
|
|
|
|
post_message_id = await self._notify(message=message, link=permalink)
|
|
return post_message_id
|
|
|
|
except:
|
|
return None
|
|
|
|
async def success(self, link: str, post_message_id: int):
|
|
try:
|
|
message = (
|
|
f"🌟 Звезда отправлена:\n"
|
|
f"▪️ Количество: {self.config.stars_count} ⭐️\n"
|
|
f"▪️ Ссылка на сообщение: {link}"
|
|
)
|
|
|
|
success_message_id = await self._notify(
|
|
message=message, reply_to=post_message_id, link=link
|
|
)
|
|
return success_message_id
|
|
|
|
except:
|
|
return None
|
|
|
|
async def failure(self, post: Dict[str, Any], reason: str, post_message_id: int):
|
|
try:
|
|
permalink = post.get("links").get("permalink")
|
|
message = (
|
|
f"❌ Не удалось отправить звезду:\n"
|
|
f"▪️ Причина: <code>{reason}</code>"
|
|
)
|
|
|
|
success_message_id = await self._notify(
|
|
message=message, reply_to=post_message_id, link=permalink
|
|
)
|
|
return success_message_id
|
|
|
|
except:
|
|
return None
|