219 lines
8.5 KiB
Python
219 lines
8.5 KiB
Python
import logging
|
||
|
||
from aiogram import Bot
|
||
|
||
from aiohttp import BasicAuth
|
||
from aiogram.client.session.aiohttp import AiohttpSession
|
||
|
||
from aiogram.enums import ParseMode
|
||
from aiogram.client.default import DefaultBotProperties
|
||
from aiogram.types import InlineKeyboardButton, InlineKeyboardMarkup
|
||
from aiogram.exceptions import (
|
||
TelegramAPIError,
|
||
TelegramBadRequest,
|
||
TelegramNetworkError,
|
||
)
|
||
|
||
from config import Config
|
||
from typing import Dict, Any, List, Optional
|
||
|
||
logger = logging.getLogger(__name__)
|
||
|
||
|
||
class NotificationBot:
|
||
def __init__(self, config: Config):
|
||
self.config = config
|
||
|
||
try:
|
||
if config.proxy_telegram:
|
||
proxy = (
|
||
f"{config.proxy_protocol}://{config.proxy_host}:{config.proxy_port}"
|
||
)
|
||
|
||
if config.proxy_username and config.proxy_password:
|
||
logger.info(
|
||
"Инициализация Telegram бота с прокси: %s://%s:%s (с авторизацией)",
|
||
config.proxy_protocol,
|
||
config.proxy_host,
|
||
config.proxy_port
|
||
)
|
||
auth = BasicAuth(
|
||
login=config.proxy_username, password=config.proxy_password
|
||
)
|
||
self.session = AiohttpSession(proxy=(proxy, auth))
|
||
else:
|
||
logger.info(
|
||
"Инициализация Telegram бота с прокси: %s://%s:%s (без авторизации)",
|
||
config.proxy_protocol,
|
||
config.proxy_host,
|
||
config.proxy_port
|
||
)
|
||
self.session = AiohttpSession(proxy=proxy)
|
||
|
||
self.bot: Bot = Bot(
|
||
token=config.bot_token,
|
||
default=DefaultBotProperties(
|
||
parse_mode=ParseMode.HTML,
|
||
disable_notification=True,
|
||
link_preview_is_disabled=True,
|
||
),
|
||
session=self.session,
|
||
)
|
||
else:
|
||
logger.info("Инициализация Telegram бота без прокси")
|
||
self.bot: Bot = Bot(
|
||
token=config.bot_token,
|
||
default=DefaultBotProperties(
|
||
parse_mode=ParseMode.HTML,
|
||
disable_notification=True,
|
||
link_preview_is_disabled=True,
|
||
),
|
||
)
|
||
|
||
logger.info(
|
||
"Бот для уведомлений инициализирован (admin_id=%s)", config.admin_id
|
||
)
|
||
|
||
except (TelegramAPIError, ValueError, TypeError) as e:
|
||
logger.error(f"Не удалось инициализировать Бот для уведомлений: {e}")
|
||
raise
|
||
|
||
async def _notify(
|
||
self, message: str, reply_to: int = None, link: str = None
|
||
) -> Optional[int]:
|
||
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 TelegramBadRequest as e:
|
||
logger.error("Bad request sending notification: %s", e)
|
||
return None
|
||
except TelegramNetworkError as e:
|
||
logger.error("Network error sending notification: %s", e)
|
||
return None
|
||
except TelegramAPIError as e:
|
||
logger.error("Telegram API error sending notification: %s", e)
|
||
return None
|
||
except Exception as e:
|
||
logger.error("Unexpected error sending notification: %s", e)
|
||
return None
|
||
|
||
async def new_post(self, post: Dict[str, Any], links: List[str]) -> Optional[int]:
|
||
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 (KeyError, AttributeError, TypeError) as e:
|
||
logger.error(f"Неверная структура поста в new_post: {e}")
|
||
return None
|
||
except Exception as e:
|
||
logger.error(f"Неожиданная ошибка в new_post: {e}")
|
||
return None
|
||
|
||
async def success(self, link: str, post_message_id: int) -> Optional[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 Exception as e:
|
||
logger.error(f"Неожиданная ошибка в success: {e}")
|
||
return None
|
||
|
||
async def failure(
|
||
self, post: Dict[str, Any], reason: str, post_message_id: int
|
||
) -> Optional[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 (KeyError, AttributeError, TypeError) as e:
|
||
logger.error(f"Неверная структура поста в failure: {e}")
|
||
return None
|
||
except Exception as e:
|
||
logger.error(f"Неожиданная ошибка в failure: {e}")
|
||
return None
|