feat(bot): add admin notification via Telegram bot

modules/bot.py

@@ -0,0 +1,116 @@
+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"Перейти")]]
+                )
+
+            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

modules/telegram.py

@@ -11,6 +11,7 @@ from typing import List, Optional, Dict, Any
 
 from config import Config
 from modules.lolz import Lolz
+from modules.bot import NotificationBot
 from modules.misc import ProcessedPostsManager
 
 logger = logging.getLogger(__name__)
@@ -57,6 +58,10 @@ class TelegramStarsBot:
 
         self.client: Optional[Client] = None
 
+        self.notification_bot = None
+        if config.bot_token and config.admin_id and config.notify_admin:
+            self.notification_bot = NotificationBot(config)
+
     async def parse_existing_posts(self):
         logger.info("Начинаю парсинг всех существующих постов в теме...")
         all_posts = await self.lolz_api.get_all_thread_posts(
@@ -137,8 +142,11 @@ class TelegramStarsBot:
             return False
 
     async def _process_single_post(self, post: Dict[str, Any]):
+        post_message_id = None
+
         post_id = post.get("post_id")
         poster_user_id = post.get("poster_user_id")
+        poster_username = post.get("poster_username", "Неизвестный пользователь")
 
         if not post_id or self.processed_manager.is_processed(post_id):
             if post_id:
@@ -147,7 +155,9 @@ class TelegramStarsBot:
                 logger.warning("Пост без post_id, пропускаю.")
             return
 
-        logger.info(f"Найден новый пост для обработки: ID {post_id}")
+        logger.info(
+            f"Найден новый пост для обработки: ID {post_id} (от {poster_username})"
+        )
 
         if self.config.skip_commented:
             if await self.lolz_api.has_comments(post=post):
@@ -155,21 +165,47 @@ class TelegramStarsBot:
                     f"Пост {post_id} уже имеет комментарии. Пропускаю обработку."
                 )
                 self.processed_manager.mark_processed(post_id)
+
+                if self.notification_bot:
+                    self.notification_bot.failure(
+                        post=post,
+                        reason=f"пост уже имеет комментарии",
+                        post_message_id=post_message_id,
+                    )
+
                 return
 
         post_content = post.get("post_body_html") or post.get("post_body")
         if not post_content:
             logger.warning(f"У поста {post_id} отсутствует содержимое. Пропускаем.")
             self.processed_manager.mark_processed(post_id)
+
+            if self.notification_bot:
+                self.notification_bot.failure(
+                    post=post,
+                    reason=f"у поста отсутствует содержимое",
+                    post_message_id=post_message_id,
+                )
+
             return
 
         links = TelegramLinkExtractor.extract(post_content)
         if not links:
             logger.info(f"В посте {post_id} не найдено ссылок Telegram.")
             self.processed_manager.mark_processed(post_id)
+
+            if self.notification_bot:
+                self.notification_bot.failure(
+                    post=post,
+                    reason=f"в посте не найдено ссылок Telegram",
+                    post_message_id=post_message_id,
+                )
+
             return
 
         logger.info(f"В посте {post_id} найдено ссылок: {len(links)}.")
+        if self.notification_bot:
+            post_message_id = self.notification_bot.new_post(post=post, links=links)
 
         successful_reactions = 0
         for link in links:
@@ -194,15 +230,36 @@ class TelegramStarsBot:
 
                     reply_message = f"[userids={poster_user_id};align=left]{random.choice(reply_options)}[/userids]"
                     await self.lolz_api.create_comment(post_id, reply_message)
+
+                if self.notification_bot:
+                    self.notification_bot.success(
+                        link=parsed_link, post_message_id=post_message_id
+                    )
+
             else:
                 logger.warning(
                     f"Пост {post_id}: poster_user_id отсутствует, комментарий не оставлен."
                 )
+
+                if self.notification_bot:
+                    self.notification_bot.failure(
+                        post=post,
+                        reason=f"poster_user_id отсутствует",
+                        post_message_id=post_message_id,
+                    )
+
         elif successful_reactions == 0:
             logger.warning(
                 f"Пост {post_id}: ни одна звезда не была отправлена успешно ({len(links)} ссылок)."
             )
 
+            if self.notification_bot:
+                self.notification_bot.failure(
+                    post=post,
+                    reason=f"ни одна звезда не была отправлена успешно",
+                    post_message_id=post_message_id,
+                )
+
         logger.info(f"Пост {post_id} полностью обработан.")
         self.processed_manager.mark_processed(post_id)
 
@@ -277,16 +334,8 @@ class TelegramStarsBot:
                 logger.info("Пожалуйста, перезапустите скрипт для начала работы.")
                 return
 
-            logger.info("=" * 40)
-            logger.info(
-                f"Комментарии на форуме: {'ВКЛЮЧЕНЫ' if self.config.enable_reply else 'ВЫКЛЮЧЕНЫ'}"
-            )
-            logger.info(
-                f"Пропуск постов с комментариями: {'ВКЛЮЧЕН' if self.config.skip_commented else 'ВЫКЛЮЧЕН'}"
-            )
             logger.info(f"Проверка начинается со страницы: {self.start_page}")
             logger.info("Бот в работе. Для остановки нажмите Ctrl+C.")
-            logger.info("=" * 40)
             await self._main_loop()
 
         finally: