import logging import os from typing import List, Optional, Dict from telethon import TelegramClient, events from telethon.tl.types import User from telethon.tl.types.auth import Authorization from telethon.errors import ( FloodWaitError, UserDeactivatedError, ChatAdminRequiredError, PeerIdInvalidError, ChannelInvalidError, UserNotParticipantError ) from app.settings import Config logger = logging.getLogger(__name__) class TelethonClientManager: """Менеджер для работы с Telethon клиентом""" def __init__(self): self.client: Optional[TelegramClient] = None self.is_initialized = False async def initialize(self) -> bool: """Инициализировать Telethon клиент""" try: if not Config.USE_TELETHON: logger.warning("Telethon отключен в конфигурации") return False if not (Config.TELETHON_API_ID and Config.TELETHON_API_HASH): logger.error("TELETHON_API_ID или TELETHON_API_HASH не установлены") return False # Получить путь для сессии session_dir = os.path.join(os.path.dirname(__file__), '..', 'sessions') os.makedirs(session_dir, exist_ok=True) session_path = os.path.join(session_dir, 'telethon_session') self.client = TelegramClient( session_path, api_id=Config.TELETHON_API_ID, api_hash=Config.TELETHON_API_HASH ) await self.client.connect() # Проверить авторизацию if not await self.client.is_user_authorized(): logger.error("Telethon клиент не авторизован. Требуется повторный вход") return False self.is_initialized = True me = await self.client.get_me() logger.info(f"✅ Telethon клиент инициализирован: {me.first_name}") return True except Exception as e: logger.error(f"Ошибка при инициализации Telethon: {e}") return False async def shutdown(self): """Остановить Telethon клиент""" if self.client and self.is_initialized: try: await self.client.disconnect() self.is_initialized = False logger.info("✅ Telethon клиент остановлен") except Exception as e: logger.error(f"Ошибка при остановке Telethon: {e}") async def send_message(self, chat_id: int, text: str, parse_mode: str = "html", disable_web_page_preview: bool = True) -> Optional[int]: """ Отправить сообщение в чат Returns: Optional[int]: ID отправленного сообщения или None при ошибке """ if not self.is_initialized: logger.error("Telethon клиент не инициализирован") return None try: message = await self.client.send_message( chat_id, text, parse_mode=parse_mode, link_preview=not disable_web_page_preview ) logger.info(f"✅ Сообщение отправлено в чат {chat_id} (Telethon)") return message.id except FloodWaitError as e: logger.warning(f"⏳ FloodWait: нужно ждать {e.seconds} секунд") raise except (ChatAdminRequiredError, UserNotParticipantError): logger.error(f"❌ Клиент не администратор или не участник чата {chat_id}") return None except PeerIdInvalidError: logger.error(f"❌ Неверный ID чата: {chat_id}") return None except Exception as e: logger.error(f"❌ Ошибка при отправке сообщения: {e}") return None async def get_chat_members(self, chat_id: int, limit: int = None) -> List[Dict]: """ Получить список участников чата Returns: List[Dict]: Список участников с информацией """ if not self.is_initialized: logger.error("Telethon клиент не инициализирован") return [] try: members = [] async for member in self.client.iter_participants(chat_id, limit=limit): member_info = { 'user_id': str(member.id), 'username': member.username, 'first_name': member.first_name, 'last_name': member.last_name, 'is_bot': member.bot, 'is_admin': member.is_self, # self-check для упрощения } members.append(member_info) logger.info(f"✅ Получено {len(members)} участников из чата {chat_id}") return members except (ChatAdminRequiredError, UserNotParticipantError): logger.error(f"❌ Нет прав получить участников чата {chat_id}") return [] except Exception as e: logger.error(f"❌ Ошибка при получении участников: {e}") return [] async def get_chat_info(self, chat_id: int) -> Optional[Dict]: """Получить информацию о чате""" if not self.is_initialized: logger.error("Telethon клиент не инициализирован") return None try: chat = await self.client.get_entity(chat_id) members_count = None if hasattr(chat, 'participants_count'): members_count = chat.participants_count return { 'id': chat.id, 'title': chat.title if hasattr(chat, 'title') else str(chat.id), 'description': chat.about if hasattr(chat, 'about') else None, 'members_count': members_count, 'is_supergroup': hasattr(chat, 'megagroup') and chat.megagroup, 'is_channel': hasattr(chat, 'broadcast'), 'is_group': hasattr(chat, 'gigagroup') } except Exception as e: logger.error(f"❌ Ошибка при получении информации о чате {chat_id}: {e}") return None async def join_chat(self, chat_link: str) -> Optional[int]: """ Присоединиться к чату по ссылке Returns: Optional[int]: ID чата или None при ошибке """ if not self.is_initialized: logger.error("Telethon клиент не инициализирован") return None try: # Попытаться присоединиться result = await self.client(ImportChatInviteRequest(hash)) chat_id = result.chats[0].id logger.info(f"✅ Присоединился к чату: {chat_id}") return chat_id except Exception as e: logger.error(f"❌ Ошибка при присоединении к чату: {e}") return None async def leave_chat(self, chat_id: int) -> bool: """Покинуть чат""" if not self.is_initialized: logger.error("Telethon клиент не инициализирован") return False try: await self.client.delete_dialog(chat_id, revoke=True) logger.info(f"✅ Покинул чат: {chat_id}") return True except Exception as e: logger.error(f"❌ Ошибка при выходе из чата: {e}") return False async def edit_message(self, chat_id: int, message_id: int, text: str) -> Optional[int]: """ Отредактировать сообщение Returns: Optional[int]: ID отредактированного сообщения или None при ошибке """ if not self.is_initialized: logger.error("Telethon клиент не инициализирован") return None try: message = await self.client.edit_message( chat_id, message_id, text, parse_mode="html" ) logger.info(f"✅ Сообщение отредактировано: {chat_id}/{message_id}") return message.id except Exception as e: logger.error(f"❌ Ошибка при редактировании сообщения: {e}") return None async def delete_message(self, chat_id: int, message_id: int) -> bool: """Удалить сообщение""" if not self.is_initialized: logger.error("Telethon клиент не инициализирован") return False try: await self.client.delete_messages(chat_id, message_id) logger.info(f"✅ Сообщение удалено: {chat_id}/{message_id}") return True except Exception as e: logger.error(f"❌ Ошибка при удалении сообщения: {e}") return False async def search_messages(self, chat_id: int, query: str, limit: int = 100) -> List[Dict]: """ Искать сообщения в чате Returns: List[Dict]: Список найденных сообщений """ if not self.is_initialized: logger.error("Telethon клиент не инициализирован") return [] try: messages = [] async for message in self.client.iter_messages(chat_id, search=query, limit=limit): messages.append({ 'id': message.id, 'text': message.text, 'date': message.date }) logger.info(f"✅ Найдено {len(messages)} сообщений по запросу '{query}'") return messages except Exception as e: logger.error(f"❌ Ошибка при поиске сообщений: {e}") return [] async def get_user_groups(self) -> List[Dict]: """ Получить все группы и супергруппы пользователя Returns: List[Dict]: Список групп с информацией {id, title, slow_mode_delay, members_count} """ if not self.is_initialized: logger.error("Telethon клиент не инициализирован") return [] try: groups = [] from telethon.tl.types import Chat, Channel # Получить все диалоги (чаты/группы) async for dialog in self.client.iter_dialogs(): entity = dialog.entity # Пропустить личные чаты и каналы (только группы и супергруппы) if not isinstance(entity, (Chat, Channel)): continue # Пропустить каналы (broadcast) if isinstance(entity, Channel) and entity.broadcast: continue group_info = { 'chat_id': entity.id, 'title': entity.title if hasattr(entity, 'title') else str(entity.id), 'slow_mode_delay': entity.slowmode_seconds if hasattr(entity, 'slowmode_seconds') else 0, 'members_count': entity.participants_count if hasattr(entity, 'participants_count') else 0, 'is_supergroup': isinstance(entity, Channel), # Channel = supergroup/megagroup } groups.append(group_info) logger.debug(f"📍 Найдена группа: {group_info['title']} (ID: {group_info['chat_id']})") logger.info(f"✅ Получено {len(groups)} групп от Telethon") return groups except Exception as e: logger.error(f"❌ Ошибка при получении групп: {e}") return [] def is_connected(self) -> bool: """Проверить, подключен ли клиент""" return self.is_initialized and self.client is not None # Глобальный экземпляр менеджера telethon_manager = TelethonClientManager()