import re import os import random import logging import asyncio from pyrogram.client import Client from pyrogram.errors import FloodWait from typing import List, Optional, Dict, Any from config import Config from modules.lolz import Lolz from modules.notification_bot import NotificationBot from modules.misc import ProcessedPostsManager logger = logging.getLogger(__name__) class TelegramLinkExtractor: @staticmethod def extract(text: str) -> List[str]: patterns = [ r"https?://(?:www\.)?(?:t\.me|telegram\.me)/([a-zA-Z0-9_]+(?:/\d+)?)", r"\[MEDIA=telegram\]([a-zA-Z0-9_]+(?:/\d+)?)\[/MEDIA\]", r'data-telegram-post="([a-zA-Z0-9_]+/\d+)"', ] all_matches = { f"https://t.me/{match}" for p in patterns for match in re.findall(p, text, re.I) } return list(all_matches) @staticmethod def parse(link: str) -> Optional[tuple[str, Optional[int]]]: match = re.search(r"t\.me/([^/]+)(?:/(\d+))?", link) if match: channel = match.group(1) message_id = int(match.group(2)) if match.group(2) else None return channel, message_id return None class TelegramStarsBot: SESSION_NAME = "storage/stars_bot" def __init__(self, config: Config): self.config = config self.start_page: int = 1 self.lolz_api = Lolz(config.lolz_token) self.processed_manager = ProcessedPostsManager() 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( thread_id=self.config.lolz_thread_id ) if all_posts: post_ids = 0 for post in all_posts: post_id = post.get("post_id") has_comments = await self.lolz_api.has_comments(post=post) if has_comments: self.processed_manager.mark_processed(post_id) post_ids += 1 logger.info(f"Добавлено {post_ids} постов в список обработанных.") else: logger.info("Не найдено постов для добавления в обработанные.") async def send_star(self, channel: str, message_id: Optional[int] = None) -> bool: if not hasattr(self.client, "send_paid_reaction"): logger.error("Платные реакции недоступны. Отправка 'звезд' невозможна.") return False try: for attempt in range(self.config.max_retries): try: if message_id is None: fetched_id = None async for msg in self.client.get_chat_history( chat_id=channel, limit=1 ): fetched_id = msg.id break if fetched_id is None: logger.error( f"Не удалось найти сообщения в канале https://t.me/{channel}" ) return False message_id = fetched_id await self.client.send_paid_reaction( chat_id=channel, message_id=message_id, amount=self.config.stars_count, ) logger.info( f"Отправлено {self.config.stars_count} звезд в https://t.me/{channel}/{message_id}" ) return True except FloodWait as e: wait_time = getattr(e, "value", getattr(e, "x", 10)) + 2 logger.warning( f"FloodWait: необходимо подождать {wait_time} секунд." ) await asyncio.sleep(wait_time) except Exception as e: logger.error( f"Попытка {attempt + 1} отправки звезд не удалась: {e}" ) if attempt < self.config.max_retries - 1: await asyncio.sleep(3 * (attempt + 1)) logger.error( f"Все {self.config.max_retries} попытки отправки звезд в https://t.me/{channel} исчерпаны." ) return False except Exception as e: logger.error(f"Критическая ошибка при отправке звезд: {e}") return False async def _process_single_post(self, post: Dict[str, Any]): post_message_id = None post_id = post.get("post_id") if not post_id: logger.warning("Пост без post_id, пропускаю.") return if self.processed_manager.is_processed(post_id): logger.debug(f"Пост {post_id} уже обработан, пропускаю.") return # Only fetch full post data for unprocessed posts post = await self.lolz_api.get_post(post_id=post_id) poster_user_id = post.get("poster_user_id") poster_username = post.get("poster_username", "Неизвестный пользователь") logger.info( f"Найден новый пост для обработки: ID {post_id} (от {poster_username})" ) if self.config.skip_commented: if await self.lolz_api.has_comments(post=post): logger.info( f"Пост {post_id} уже имеет комментарии. Пропускаю обработку." ) self.processed_manager.mark_processed(post_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) 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: await 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 = await self.notification_bot.new_post( post=post, links=links ) successful_reactions = 0 for link in links: parsed_link = TelegramLinkExtractor.parse(link) if parsed_link: channel, message_id = parsed_link if await self.send_star(channel, message_id): successful_reactions += 1 break # stop after first successful reaction await asyncio.sleep(1) if successful_reactions > 0 and self.config.enable_reply: await asyncio.sleep(self.config.api_delay) if poster_user_id: replies = self.config.reply_templates for reply_options in replies: await asyncio.sleep(self.config.api_delay) 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: await self.notification_bot.success( link=f"https://t.me/{channel}/{message_id}", post_message_id=post_message_id, ) else: logger.warning( f"Пост {post_id}: poster_user_id отсутствует, комментарий не оставлен." ) if self.notification_bot: await 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: await 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) async def _main_loop(self): while True: try: logger.info( f"Проверка новых постов в теме {self.config.lolz_thread_id} начиная со страницы {self.start_page}..." ) posts, last_page = await self.lolz_api.get_thread_posts( self.config.lolz_thread_id, self.start_page ) if posts: for post in posts: try: await self._process_single_post(post) except Exception as e: post_id = post.get("post_id", "?") logger.error(f"Ошибка обработки поста {post_id}, пропускаю: {e}") if last_page > self.start_page: logger.info(f"Мониторинг продвинулся до страницы {last_page}.") self.start_page = last_page else: logger.info("Новых постов для обработки не найдено.") logger.info(f"Ожидание {self.config.check_interval} секунд...") await asyncio.sleep(self.config.check_interval) except KeyboardInterrupt: logger.info("Получен сигнал прерывания (Ctrl+C).") break except Exception as e: logger.exception(f"Критическая ошибка в главном цикле: {e}") await asyncio.sleep(self.config.check_interval) async def start(self): is_first_login = not os.path.exists(f"{self.SESSION_NAME}.session") if is_first_login: logger.info("Сессия Telegram не найдена. Запускаю процесс входа...") else: if self.config.default_choice == 2: logger.info("Выбран режим парсинга существующих постов.") self.client = Client( name=self.SESSION_NAME, api_id=self.config.api_id, api_hash=self.config.api_hash, ) await self.client.start() await self.parse_existing_posts() await self.client.stop() logger.info( "Парсинг завершен. Все существующие посты добавлены в обработанные." ) return self.start_page = self.config.start_page self.client = Client( name=self.SESSION_NAME, api_id=self.config.api_id, api_hash=self.config.api_hash, ) try: await self.client.start() logger.info("Клиент Telegram успешно запущен.") if is_first_login: logger.info("Аккаунт Telegram успешно подключен.") logger.info("Пожалуйста, перезапустите скрипт для начала работы.") return logger.info(f"Проверка начинается со страницы: {self.start_page}") logger.info("Бот в работе. Для остановки нажмите Ctrl+C.") await self._main_loop() finally: if self.client and self.client.is_connected: await self.client.stop() logger.info("Бот остановлен.")