diff --git a/.env.example b/.env.example new file mode 100644 index 0000000..9759193 --- /dev/null +++ b/.env.example @@ -0,0 +1,14 @@ +API_ID= +API_HASH= + +LOLZ_TOKEN= +LOLZ_THREAD_ID= + +STARS_COUNT=1 + +CHECK_INTERVAL=30 +API_DELAY=5 +MAX_RETRIES=3 + +ENABLE_REPLY=true +SKIP_COMMENTED=true diff --git a/.gitignore b/.gitignore index d7f14ff..6de4d95 100644 --- a/.gitignore +++ b/.gitignore @@ -188,3 +188,5 @@ pyvenv.cfg .venv pip-selfcheck.json +processed.txt +*.log diff --git a/bot.py b/bot.py deleted file mode 100644 index f8d892a..0000000 --- a/bot.py +++ /dev/null @@ -1,435 +0,0 @@ -import asyncio -import json -import re -import os -import logging -import random -import argparse -import sys -from typing import Set, Optional, Dict, Any, List, Union - -import aiohttp -from pyrogram import Client -from pyrogram.errors import FloodWait - -class ColoredFormatter(logging.Formatter): - COLORS = { - "GREY": "\x1b[38;20m", - "BLUE": "\x1b[34;20m", - "GREEN": "\x1b[32;20m", - "YELLOW": "\x1b[33;20m", - "RED": "\x1b[31;20m", - "BOLD_RED": "\x1b[31;1m", - "RESET": "\x1b[0m", - } - - def __init__(self, fmt: str): - super().__init__() - self.fmt = fmt - self.FORMATS = { - logging.DEBUG: self.COLORS["GREY"] + self.fmt + self.COLORS["RESET"], - logging.INFO: self.COLORS["BLUE"] + "ℹ️ " + self.fmt + self.COLORS["RESET"], - logging.WARNING: self.COLORS["YELLOW"] + "⚠️ " + self.fmt + self.COLORS["RESET"], - logging.ERROR: self.COLORS["RED"] + "❌ " + self.fmt + self.COLORS["RESET"], - logging.CRITICAL: self.COLORS["BOLD_RED"] + "🔥 " + self.fmt + self.COLORS["RESET"], - "SUCCESS": self.COLORS["GREEN"] + "✅ " + self.fmt + self.COLORS["RESET"], - } - - def format(self, record): - log_fmt = self.FORMATS.get(getattr(record, "levelformat", record.levelno)) - formatter = logging.Formatter(log_fmt, "%Y-%m-%d %H:%M:%S") - return formatter.format(record) - -def setup_logger(): - logger = logging.getLogger(__name__) - logger.setLevel(logging.INFO) - - if not logger.handlers: - handler = logging.StreamHandler(sys.stdout) - handler.setFormatter(ColoredFormatter("%(message)s")) - logger.addHandler(handler) - - def success(message, *args, **kws): - logger._log(logging.INFO, message, args, extra={'levelformat': 'SUCCESS'}, **kws) - - logger.success = success - return logger - -logger = setup_logger() - -class Config: - def __init__(self, config_file: str = "config.json", cli_thread_id: Optional[str] = None, reset: bool = False): - self.config_file = config_file - self.cli_thread_id = cli_thread_id - - if reset or not self._config_exists_and_is_valid(): - self._run_interactive_setup() - - with open(self.config_file, 'r', encoding='utf-8') as f: - self._config_data = json.load(f) - logger.success("Конфигурация успешно загружена.") - - def _config_exists_and_is_valid(self) -> bool: - if not os.path.exists(self.config_file): - return False - try: - with open(self.config_file, 'r', encoding='utf-8') as f: - config = json.load(f) - return config.get("api_id") != "YOUR_API_ID" - except (json.JSONDecodeError, KeyError): - return False - - def _run_interactive_setup(self): - logger.info("Запускаю мастер первоначальной настройки...") - config = { - "api_id": input("Введите ваш API ID: "), - "api_hash": input("Введите ваш API Hash: "), - "phone_number": input("Введите ваш номер телефона (в международном формате, +...): "), - "lolz_token": input("Введите ваш Lolzteam API токен: "), - "forum_thread_id": input("Введите ID темы на форуме для отслеживания: "), - "stars_count": 3, - "check_interval": 30, - "api_delay": 5, - "max_retries": 3, - "processed_posts_file": "processed_posts.json", - "enable_reply": True, - "reply_templates": ["Готово! Отправил звезды. ⭐", "Выполнено.", "Сделал.", "+rep"], - "skip_posts_with_comments": True, - } - with open(self.config_file, 'w', encoding='utf-8') as f: - json.dump(config, f, indent=4, ensure_ascii=False) - - logger.success(f"Конфигурация сохранена в файл '{self.config_file}'.") - logger.info("Теперь перезапустите скрипт для входа в аккаунт Telegram.") - sys.exit() - - def __getattr__(self, name: str) -> Any: - if name == "forum_thread_id" and self.cli_thread_id: - return self.cli_thread_id - return self._config_data.get(name, True if name == "skip_posts_with_comments" else None) - -class ProcessedPostsManager: - def __init__(self, file_path: str): - self.file_path = file_path - self.processed_posts: Set[int] = self._load() - - def _load(self) -> Set[int]: - try: - with open(self.file_path, 'r', encoding='utf-8') as f: - return set(json.load(f)) - except (FileNotFoundError, json.JSONDecodeError): - return set() - - def _save(self): - with open(self.file_path, 'w', encoding='utf-8') as f: - json.dump(list(self.processed_posts), f, indent=4) - - def is_processed(self, post_id: int) -> bool: - return post_id in self.processed_posts - - def mark_processed(self, post_id: int): - self.processed_posts.add(post_id) - self._save() - - def add_existing_posts(self, post_ids: List[int]): - for post_id in post_ids: - self.processed_posts.add(post_id) - self._save() - -class LolzAPI: - def __init__(self, token: str): - self.base_url = "https://prod-api.lolz.live" - self.headers = {"Authorization": f"Bearer {token}", "Content-Type": "application/json"} - - async def _request(self, method: str, endpoint: str, is_retry: bool = False, **kwargs) -> Optional[Dict[str, Any]]: - url = f"{self.base_url}{endpoint}" - try: - async with aiohttp.ClientSession() as session: - async with session.request(method, url, headers=self.headers, **kwargs) as response: - if response.status == 200: - return await response.json() - - error_text = await response.text() - - if response.status == 403 and "Необходимо подождать" in error_text and not is_retry: - logger.warning("Обнаружен флуд-контроль API. Ожидание 3 секунды перед повторной попыткой...") - await asyncio.sleep(3) - return await self._request(method, endpoint, is_retry=True, **kwargs) - - if response.status == 429: - logger.warning("Превышен лимит запросов к API, ожидание 10 секунд...") - await asyncio.sleep(10) - else: - logger.error(f"Ошибка API {response.status} для {url}: {error_text}") - return None - except aiohttp.ClientError as e: - logger.error(f"Ошибка соединения с API: {e}") - return None - - async def get_thread_posts(self, thread_id: Union[str, int], start_page: int = 1) -> List[Dict[str, Any]]: - all_posts = [] - page = start_page - while True: - params = {"thread_id": thread_id, "page": page, "order": "post_date_reverse"} - data = await self._request("GET", "/posts", params=params) - posts = data.get("posts", []) if data else [] - if not posts: - break - all_posts.extend(posts) - logger.info(f"Получено {len(posts)} постов из темы {thread_id} на странице {page}.") - page += 1 - await asyncio.sleep(1) - if all_posts: - logger.info(f"Всего получено {len(all_posts)} постов из темы {thread_id}.") - else: - logger.info(f"Постов в теме {thread_id} не найдено.") - return all_posts - - async def get_all_thread_posts(self, thread_id: Union[str, int]) -> List[Dict[str, Any]]: - all_posts = [] - page = 1 - while True: - params = {"thread_id": thread_id, "page": page} - data = await self._request("GET", "/posts", params=params) - posts = data.get("posts", []) if data else [] - if not posts: - break - all_posts.extend(posts) - logger.info(f"Получено {len(posts)} постов из темы {thread_id} на странице {page}.") - page += 1 - await asyncio.sleep(1) - return all_posts - - async def get_post_comments(self, post_id: int) -> List[Dict[str, Any]]: - params = {"post_id": post_id} - data = await self._request("GET", "/posts/comments", params=params) - if data and "comments" in data: - return data["comments"] - return [] - - async def has_comments(self, post_id: int) -> bool: - comments = await self.get_post_comments(post_id) - return len(comments) > 0 - - async def create_comment(self, post_id: int, comment_body: str) -> bool: - payload = {"comment_body": comment_body} - logger.info(f"Публикую комментарий к посту {post_id}...") - response = await self._request("POST", f"/posts/{post_id}/comments", json=payload) - - if response and response.get("comment"): - logger.success(f"Комментарий к посту {post_id} успешно опубликован.") - return True - else: - logger.error(f"Не удалось опубликовать комментарий к посту {post_id}.") - return False - -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 = "stars_bot_session" - - def __init__(self, config: Config): - self.config = config - self.lolz_api = LolzAPI(config.lolz_token) - self.processed_manager = ProcessedPostsManager(config.processed_posts_file) - self.client: Optional[Client] = None - self.start_page: int = 1 - - async def parse_existing_posts(self): - logger.info("Начинаю парсинг всех существующих постов в теме...") - all_posts = await self.lolz_api.get_all_thread_posts(self.config.forum_thread_id) - - if all_posts: - post_ids = [post.get("post_id") for post in all_posts if post.get("post_id")] - self.processed_manager.add_existing_posts(post_ids) - logger.success(f"Добавлено {len(post_ids)} постов в список обработанных.") - else: - logger.info("Не найдено постов для добавления в обработанные.") - - async def send_stars_reaction(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: - async for message in self.client.get_chat_history(f"@{channel}", limit=1): - message_id = message.id - break - - if message_id is None: - logger.error(f"Не удалось найти сообщения в канале @{channel}") - return False - - await self.client.send_paid_reaction(f"@{channel}", message_id, self.config.stars_count) - logger.success(f"Отправлено {self.config.stars_count} звезд в @{channel}/{message_id}") - return True - except FloodWait as e: - logger.warning(f"FloodWait: необходимо подождать {e.x + 2} секунд.") - await asyncio.sleep(e.x + 2) - except Exception as e: - logger.error(f"Попытка {attempt + 1} отправки звезд не удалась: {e}") - if attempt < self.config.max_retries - 1: - await asyncio.sleep(3 * (attempt + 1)) - return False - except Exception as e: - logger.error(f"Критическая ошибка при отправке звезд: {e}") - return False - - async def _process_single_post(self, post: Dict[str, Any]): - post_id = post.get("post_id") - post_user_id = post.get("poster_user_id") - - if not post_id or self.processed_manager.is_processed(post_id): - return - - logger.info(f"Найден новый пост для обработки: ID {post_id}") - - if self.config.skip_posts_with_comments: - if await self.lolz_api.has_comments(post_id): - 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) - return - - successful_reactions = 0 - for link in links: - parsed_link = TelegramLinkExtractor.parse(link) - if parsed_link: - channel, message_id = parsed_link - if await self.send_stars_reaction(channel, message_id): - successful_reactions += 1 - await asyncio.sleep(1) - - if successful_reactions > 0 and self.config.enable_reply: - await asyncio.sleep(self.config.api_delay) - reply_message = random.choice(self.config.reply_templates) - - if post_user_id: - reply_message = f"[userids={post_user_id};align=left]{reply_message}[/userids]" - - await self.lolz_api.create_comment(post_id, reply_message) - - 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.forum_thread_id} начиная со страницы {self.start_page}...") - posts = await self.lolz_api.get_thread_posts(self.config.forum_thread_id, self.start_page) - - if posts: - for post in reversed(posts): - await self._process_single_post(post) - 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: - choice = input("1 - Начать мониторинг\n2 - Спарсить все существующие посты\nВыберите действие (1 или 2): ") - if choice == "2": - logger.info("Выбран режим парсинга существующих постов.") - self.client = Client(self.SESSION_NAME, self.config.api_id, self.config.api_hash) - await self.client.start() - await self.parse_existing_posts() - await self.client.stop() - logger.success("Парсинг завершен. Все существующие посты добавлены в обработанные.") - return - - start_page_input = input("Введите номер страницы для начала проверки (или нажмите Enter для проверки с первой страницы): ") - try: - self.start_page = int(start_page_input) if start_page_input.strip() else 1 - if self.start_page < 1: - logger.error("Номер страницы должен быть положительным. Использую первую страницу.") - self.start_page = 1 - except ValueError: - logger.error("Введено некорректное значение. Использую первую страницу.") - self.start_page = 1 - - self.client = Client(self.SESSION_NAME, self.config.api_id, self.config.api_hash) - - try: - await self.client.start() - logger.success("Клиент Telegram успешно запущен.") - - if is_first_login: - logger.success("Аккаунт Telegram успешно подключен.") - logger.info("Пожалуйста, перезапустите скрипт для начала работы.") - return - - logger.info("=" * 40) - logger.info(f"Комментарии на форуме: {'ВКЛЮЧЕНЫ' if self.config.enable_reply else 'ВЫКЛЮЧЕНЫ'}") - logger.info(f"Пропуск постов с комментариями: {'ВКЛЮЧЕН' if self.config.skip_posts_with_comments else 'ВЫКЛЮЧЕН'}") - logger.info(f"Проверка начинается со страницы: {self.start_page}") - logger.info("Бот в работе. Для остановки нажмите Ctrl+C.") - logger.info("=" * 40) - await self._main_loop() - - finally: - if self.client and self.client.is_connected: - await self.client.stop() - logger.info("Бот остановлен.") - -def main(): - parser = argparse.ArgumentParser(description='Telegram Stars Bot для Lolzteam') - parser.add_argument('--thread-id', type=str, help='ID темы форума (переопределяет config.json)') - parser.add_argument('--reset-config', action='store_true', help='Запустить мастер настройки заново') - args = parser.parse_args() - - try: - config = Config(cli_thread_id=args.thread_id, reset=args.reset_config) - bot = TelegramStarsBot(config) - asyncio.run(bot.start()) - except (ValueError, TypeError) as e: - logger.critical(f"Ошибка в конфигурации или при запуске: {e}") - except Exception as e: - logger.critical(f"Фатальная ошибка при запуске бота: {e}") - -if __name__ == "__main__": - main() diff --git a/replies.txt b/replies.txt new file mode 100644 index 0000000..e69de29 diff --git a/requirements.txt b/requirements.txt new file mode 100644 index 0000000..620ea16 Binary files /dev/null and b/requirements.txt differ