513 lines
20 KiB
Python
513 lines
20 KiB
Python
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=config)
|
||
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 update_captcha(self):
|
||
"""Update or add captcha text in the thread spoiler block"""
|
||
captcha_text = self.config._load_captcha()
|
||
captcha_answer = self.config._load_captcha_answer()
|
||
|
||
if not captcha_text or not self.config.enable_captcha or not captcha_answer:
|
||
if not self.config.enable_captcha:
|
||
logger.error("Капча выключена. Редактирование темы невозможно.")
|
||
elif not captcha_answer:
|
||
logger.error(
|
||
"Ответ для капчи не установлен. Редактирование темы невозможно."
|
||
)
|
||
elif not captcha_text:
|
||
logger.error(
|
||
"Вопрос (текст) капчи не установлен. Редактирование темы невозможно."
|
||
)
|
||
|
||
return False
|
||
|
||
try:
|
||
thread = await self.lolz_api.get_thread(
|
||
thread_id=self.config.lolz_thread_id
|
||
)
|
||
thread_text = thread["first_post"].get("post_body", "")
|
||
|
||
# Pattern to match the captcha spoiler block
|
||
captcha_pattern = (
|
||
r"\[spoiler=Капча[^\]]*\]\[CENTER\].*?\[/CENTER\]\[/spoiler\]"
|
||
)
|
||
|
||
if re.search(captcha_pattern, thread_text, re.DOTALL):
|
||
# Replace existing captcha text
|
||
captcha_wrapper = f"[spoiler=Капча (обязательно);align=center][CENTER]{captcha_text}[/CENTER][/spoiler]"
|
||
updated_text = re.sub(
|
||
captcha_pattern, captcha_wrapper, thread_text, flags=re.DOTALL
|
||
)
|
||
|
||
# Check if text actually changed
|
||
if updated_text == thread_text:
|
||
logger.info(
|
||
f"Капча в теме {self.config.lolz_thread_id} уже актуальна, изменений не требуется."
|
||
)
|
||
return True
|
||
|
||
thread_text = updated_text
|
||
logger.info(f"Обновление капчи в теме {self.config.lolz_thread_id}...")
|
||
|
||
else:
|
||
# Add new captcha block
|
||
captcha_wrapper = f"\n\n[spoiler=Капча (обязательно);align=center][CENTER]{captcha_text}[/CENTER][/spoiler]"
|
||
thread_text += captcha_wrapper
|
||
logger.info(f"Добавление капчи в тему {self.config.lolz_thread_id}...")
|
||
|
||
await self.lolz_api.edit_thread(
|
||
thread_id=self.config.lolz_thread_id, post_body=thread_text
|
||
)
|
||
|
||
logger.info(f"Тема {self.config.lolz_thread_id} успешно отредактирована.")
|
||
return True
|
||
|
||
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})"
|
||
)
|
||
|
||
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
|
||
|
||
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
|
||
|
||
if self.config.enable_captcha:
|
||
captcha_answer = self.config._load_captcha_answer()
|
||
if captcha_answer and str(captcha_answer) not in post_content:
|
||
logger.info(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._load_replies()
|
||
|
||
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 _bump_loop(self):
|
||
while True:
|
||
try:
|
||
logger.info(
|
||
f"Проверка статуса поднятия для темы {self.config.lolz_thread_id}..."
|
||
)
|
||
|
||
thread = await self.lolz_api.get_thread(
|
||
thread_id=self.config.lolz_thread_id
|
||
)
|
||
can_bump = await self.lolz_api.can_bump(thread=thread)
|
||
|
||
if can_bump:
|
||
try:
|
||
await self.lolz_api.bump_thread(
|
||
thread_id=self.config.lolz_thread_id
|
||
)
|
||
|
||
except Exception as e:
|
||
logger.error(
|
||
f"Ошибка поднятия темы {self.config.lolz_thread_id}, пропускаю: {e}"
|
||
)
|
||
else:
|
||
logger.info("Тему нельзя поднять.")
|
||
|
||
logger.info(f"Ожидание {self.config.bump_check_interval} секунд...")
|
||
await asyncio.sleep(self.config.bump_check_interval)
|
||
|
||
except KeyboardInterrupt:
|
||
logger.info("Получен сигнал прерывания (Ctrl+C).")
|
||
break
|
||
|
||
except Exception as e:
|
||
logger.exception(f"Критическая ошибка в цикле поднятия темы: {e}")
|
||
await asyncio.sleep(self.config.bump_check_interval)
|
||
|
||
async def _captcha_loop(self):
|
||
"""Loop to periodically check and update captcha text"""
|
||
while True:
|
||
try:
|
||
logger.info(
|
||
f"Проверка обновления капчи для темы {self.config.lolz_thread_id}..."
|
||
)
|
||
|
||
await self.update_captcha()
|
||
|
||
logger.info(
|
||
f"Ожидание {self.config.captcha_check_interval} секунд до следующей проверки капчи..."
|
||
)
|
||
await asyncio.sleep(self.config.captcha_check_interval)
|
||
|
||
except KeyboardInterrupt:
|
||
logger.info("Получен сигнал прерывания (Ctrl+C).")
|
||
break
|
||
|
||
except Exception as e:
|
||
logger.exception(f"Критическая ошибка в цикле обновления капчи: {e}")
|
||
await asyncio.sleep(self.config.captcha_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:
|
||
self.start_page = self.config.start_page
|
||
|
||
if self.config.proxy_telegram:
|
||
proxy = {
|
||
"scheme": self.config.proxy_protocol,
|
||
"hostname": self.config.proxy_host,
|
||
"port": self.config.proxy_port,
|
||
}
|
||
|
||
if self.config.proxy_username and self.config.proxy_password:
|
||
logger.info(
|
||
"Инициализация Pyrogram с прокси: %s://%s:%s (с авторизацией)",
|
||
self.config.proxy_protocol,
|
||
self.config.proxy_host,
|
||
self.config.proxy_port,
|
||
)
|
||
|
||
proxy["username"] = self.config.proxy_username
|
||
proxy["password"] = self.config.proxy_password
|
||
else:
|
||
logger.info(
|
||
"Инициализация Pyrogram с прокси: %s://%s:%s (без авторизации)",
|
||
self.config.proxy_protocol,
|
||
self.config.proxy_host,
|
||
self.config.proxy_port,
|
||
)
|
||
|
||
self.client = Client(
|
||
name=self.SESSION_NAME,
|
||
api_id=self.config.api_id,
|
||
api_hash=self.config.api_hash,
|
||
proxy=proxy,
|
||
)
|
||
|
||
else:
|
||
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.")
|
||
|
||
# _main_loop - Основной цикл для остлеживания темы
|
||
# _bump_loop - Цикл для авто-поднятия темы
|
||
# _captcha_loop - Цикл для обновления капчи
|
||
|
||
if self.config.enable_captcha:
|
||
await self.update_captcha()
|
||
|
||
# Gather all enabled loops
|
||
tasks = [self._main_loop()]
|
||
|
||
if self.config.enable_auto_bump:
|
||
tasks.append(self._bump_loop())
|
||
|
||
if self.config.enable_captcha:
|
||
tasks.append(self._captcha_loop())
|
||
|
||
await asyncio.gather(*tasks)
|
||
|
||
except Exception as e:
|
||
logger.error(f"Ошибка в главном процессе бота: {e}")
|
||
raise
|
||
|
||
finally:
|
||
if self.client and self.client.is_connected:
|
||
try:
|
||
await self.client.stop()
|
||
except Exception as e:
|
||
logger.error(f"Ошибка при остановке клиента: {e}")
|
||
|
||
logger.info("Бот остановлен.")
|