249 lines
9.9 KiB
Python
249 lines
9.9 KiB
Python
import logging
|
||
import asyncio
|
||
from typing import Optional, Tuple
|
||
from telegram.error import TelegramError, BadRequest, Forbidden
|
||
from telethon.errors import FloodWaitError, UserDeactivatedError, ChatAdminRequiredError
|
||
|
||
from app.handlers.telethon_client import telethon_manager
|
||
from app.handlers.sender import MessageSender
|
||
from app.database.member_repository import GroupStatisticsRepository
|
||
from app.settings import Config
|
||
|
||
logger = logging.getLogger(__name__)
|
||
|
||
|
||
class HybridMessageSender:
|
||
"""
|
||
Гибридный отправитель сообщений.
|
||
Пытается отправить как бот, при ошибке переключается на Pyrogram клиента.
|
||
"""
|
||
|
||
def __init__(self, bot, db_session):
|
||
self.bot = bot
|
||
self.db_session = db_session
|
||
self.message_sender = MessageSender(bot, db_session)
|
||
self.stats_repo = GroupStatisticsRepository(db_session)
|
||
|
||
async def send_message(self, chat_id: str, message_text: str,
|
||
group_id: int = None,
|
||
parse_mode: str = "HTML",
|
||
disable_web_page_preview: bool = True) -> Tuple[bool, Optional[str]]:
|
||
"""
|
||
Отправить сообщение с гибридной логикой.
|
||
|
||
Сначала пытается отправить как бот, если ошибка - переходит на клиент.
|
||
|
||
Returns:
|
||
Tuple[bool, Optional[str]]: (успешность, метод_отправки)
|
||
Методы: 'bot', 'client', None если оба способа не работают
|
||
"""
|
||
|
||
# Попытка 1: отправить как бот
|
||
try:
|
||
logger.info(f"Попытка отправить сообщение как бот в {chat_id}")
|
||
await self.message_sender.send_message(
|
||
chat_id=chat_id,
|
||
message_text=message_text,
|
||
parse_mode=parse_mode,
|
||
disable_web_page_preview=disable_web_page_preview
|
||
)
|
||
|
||
if group_id:
|
||
await self.stats_repo.update_send_capabilities(group_id, can_bot=True, can_client=False)
|
||
logger.info(f"Сообщение успешно отправлено ботом в {chat_id}")
|
||
return True, "bot"
|
||
|
||
except (BadRequest, Forbidden) as e:
|
||
# Ошибки которые означают что бот не может писать
|
||
logger.warning(f"Бот не может отправить сообщение в {chat_id}: {e}")
|
||
|
||
if group_id:
|
||
await self.stats_repo.update_send_capabilities(group_id, can_bot=False, can_client=False)
|
||
|
||
# Если Telethon отключен или не инициализирован - выходим
|
||
if not Config.USE_TELETHON or not telethon_manager.is_connected():
|
||
logger.error(f"Telethon недоступен, не удалось отправить сообщение в {chat_id}")
|
||
return False, None
|
||
|
||
# Попытка 2: отправить как клиент
|
||
return await self._send_via_telethon(chat_id, message_text, group_id)
|
||
|
||
except TelegramError as e:
|
||
logger.error(f"Ошибка Telegram при отправке в {chat_id}: {e}")
|
||
|
||
if group_id:
|
||
await self.stats_repo.increment_failed_messages(group_id)
|
||
|
||
return False, None
|
||
|
||
except Exception as e:
|
||
logger.error(f"Неожиданная ошибка при отправке в {chat_id}: {e}")
|
||
|
||
if group_id:
|
||
await self.stats_repo.increment_failed_messages(group_id)
|
||
|
||
return False, None
|
||
|
||
async def _send_via_telethon(self, chat_id: str, message_text: str,
|
||
group_id: int = None) -> Tuple[bool, Optional[str]]:
|
||
"""Отправить сообщение через Telethon клиент"""
|
||
|
||
if not telethon_manager.is_connected():
|
||
logger.error("Telethon клиент не инициализирован")
|
||
return False, None
|
||
|
||
try:
|
||
# Конвертировать chat_id в int для Telethon
|
||
try:
|
||
numeric_chat_id = int(chat_id)
|
||
except ValueError:
|
||
# Если это строка типа "-100123456789"
|
||
numeric_chat_id = int(chat_id)
|
||
|
||
logger.info(f"Попытка отправить сообщение через Telethon в {numeric_chat_id}")
|
||
|
||
message_id = await telethon_manager.send_message(
|
||
chat_id=numeric_chat_id,
|
||
text=message_text,
|
||
parse_mode="html",
|
||
disable_web_page_preview=True
|
||
)
|
||
|
||
if message_id:
|
||
if group_id:
|
||
await self.stats_repo.increment_sent_messages(group_id, via_client=True)
|
||
await self.stats_repo.update_send_capabilities(group_id, can_bot=False, can_client=True)
|
||
|
||
logger.info(f"✅ Сообщение успешно отправлено через Telethon в {numeric_chat_id}")
|
||
return True, "client"
|
||
else:
|
||
if group_id:
|
||
await self.stats_repo.increment_failed_messages(group_id)
|
||
|
||
return False, None
|
||
|
||
except FloodWaitError as e:
|
||
logger.warning(f"⏳ FloodWait от Telethon: нужно ждать {e.seconds} сек")
|
||
|
||
if group_id:
|
||
await self.stats_repo.increment_failed_messages(group_id)
|
||
|
||
# Ожидание и повторная попытка
|
||
await asyncio.sleep(min(e.seconds, Config.TELETHON_FLOOD_WAIT_MAX))
|
||
return await self._send_via_telethon(chat_id, message_text, group_id)
|
||
|
||
except (ChatAdminRequiredError, UserDeactivatedError):
|
||
logger.error(f"❌ Telethon клиент не администратор в {chat_id}")
|
||
|
||
if group_id:
|
||
await self.stats_repo.update_send_capabilities(group_id, can_bot=False, can_client=False)
|
||
|
||
return False, None
|
||
|
||
except Exception as e:
|
||
logger.error(f"❌ Ошибка Telethon при отправке в {chat_id}: {e}")
|
||
|
||
if group_id:
|
||
await self.stats_repo.increment_failed_messages(group_id)
|
||
|
||
return False, None
|
||
|
||
async def send_message_with_retry(self, chat_id: str, message_text: str,
|
||
group_id: int = None,
|
||
max_retries: int = None) -> Tuple[bool, Optional[str]]:
|
||
"""
|
||
Отправить сообщение с повторными попытками
|
||
|
||
Args:
|
||
chat_id: ID чата
|
||
message_text: Текст сообщения
|
||
group_id: ID группы в БД (для отслеживания статистики)
|
||
max_retries: Максимум повторов (по умолчанию из Config)
|
||
|
||
Returns:
|
||
Tuple[bool, Optional[str]]: (успешность, метод_отправки)
|
||
"""
|
||
|
||
if max_retries is None:
|
||
max_retries = Config.MAX_RETRIES
|
||
|
||
for attempt in range(max_retries):
|
||
try:
|
||
success, method = await self.send_message(
|
||
chat_id=chat_id,
|
||
message_text=message_text,
|
||
group_id=group_id,
|
||
parse_mode="HTML"
|
||
)
|
||
|
||
if success:
|
||
return True, method
|
||
|
||
# Ждать перед повторной попыткой
|
||
if attempt < max_retries - 1:
|
||
wait_time = Config.RETRY_DELAY * (attempt + 1)
|
||
logger.info(f"Повтор попытки {attempt + 1}/{max_retries} через {wait_time}с")
|
||
await asyncio.sleep(wait_time)
|
||
|
||
except Exception as e:
|
||
logger.error(f"Ошибка при попытке {attempt + 1}: {e}")
|
||
|
||
if attempt < max_retries - 1:
|
||
await asyncio.sleep(Config.RETRY_DELAY)
|
||
|
||
logger.error(f"Не удалось отправить сообщение в {chat_id} после {max_retries} попыток")
|
||
|
||
if group_id:
|
||
await self.stats_repo.increment_failed_messages(group_id)
|
||
|
||
return False, None
|
||
|
||
async def bulk_send(self, chat_ids: list, message_text: str,
|
||
group_ids: list = None,
|
||
use_slow_mode: bool = False) -> dict:
|
||
"""
|
||
Массовая отправка сообщений
|
||
|
||
Returns:
|
||
dict: {
|
||
'total': количество чатов,
|
||
'success': успешно отправлено,
|
||
'failed': ошибок,
|
||
'via_bot': через бот,
|
||
'via_client': через клиент
|
||
}
|
||
"""
|
||
|
||
results = {
|
||
'total': len(chat_ids),
|
||
'success': 0,
|
||
'failed': 0,
|
||
'via_bot': 0,
|
||
'via_client': 0
|
||
}
|
||
|
||
for idx, chat_id in enumerate(chat_ids):
|
||
group_id = group_ids[idx] if group_ids else None
|
||
|
||
success, method = await self.send_message_with_retry(
|
||
chat_id=str(chat_id),
|
||
message_text=message_text,
|
||
group_id=group_id
|
||
)
|
||
|
||
if success:
|
||
results['success'] += 1
|
||
if method == 'bot':
|
||
results['via_bot'] += 1
|
||
elif method == 'client':
|
||
results['via_client'] += 1
|
||
else:
|
||
results['failed'] += 1
|
||
|
||
# Slow mode
|
||
if use_slow_mode and idx < len(chat_ids) - 1:
|
||
await asyncio.sleep(Config.MIN_SEND_INTERVAL)
|
||
|
||
logger.info(f"Массовая отправка завершена: {results}")
|
||
return results
|