import os import logging import asyncio from pathlib import Path from telethon import TelegramClient from telethon.tl.functions.messages import ImportChatInviteRequest from telethon.tl.functions.channels import JoinChannelRequest from telethon.errors import ( UserAlreadyParticipantError, InviteHashExpiredError, InviteHashInvalidError, FloodWaitError, ChannelInvalidError ) # Инициализация логирования внутри контура юзербота logger = logging.getLogger("CyberAgentUserClient") # Сбор конфигурационных данных с безопасными фолбеками TELEGRAM_API_ID = os.environ.get("TELEGRAM_API_ID") or os.environ.get("API_ID") TELEGRAM_API_HASH = os.environ.get("TELEGRAM_API_HASH") or os.environ.get("API_HASH") # Максимальный размер медиафайла для загрузки (25 МБ) — защита от переполнения хранилища MAX_FILE_SIZE_BYTES = 25 * 1024 * 1024 # Переменная для хранения синглтона клиента client = None async def get_client(): """Ленивая инициализация синглтона Telethon с проверкой конфигурации""" global client if client is None: if not TELEGRAM_API_ID or not TELEGRAM_API_HASH: raise ValueError("Критическая ошибка среды: Переменные TELEGRAM_API_ID и TELEGRAM_API_HASH не заданы.") try: api_id_int = int(TELEGRAM_API_ID) except ValueError: raise ValueError(f"Переменная TELEGRAM_API_ID должна быть числом. Получено: '{TELEGRAM_API_ID}'") # Инициализируем клиент. Файл сессии будет сохранен как user_session.session # Привязываем к текущему запущенному event loop внутри FastAPI/Uvicorn client = TelegramClient('user_session', api_id_int, TELEGRAM_API_HASH) # Безопасное подключение к серверам Telegram без блокировки рантайма if not client.is_connected(): await client.connect() return client async def check_authorization(tg_client) -> bool: """Проверка, что сессия активна и не требует ручного ввода кода в консоль""" try: return await tg_client.is_user_authorized() except Exception as e: logger.error(f"Ошибка проверки авторизации юзербота: {e}") return False async def search_in_group(group_username: str, query: str) -> str: """Поиск сообщений по ключевому слову в конкретной группе/канале""" try: cl = await get_client() if not await check_authorization(cl): return "❌ Ошибка: Юзербот не авторизован. Перевыпустите файл `user_session.session` локально." # Очистка юзернейма группы от префиксов ссылок target_group = group_username.strip().replace("https://t.me/", "").replace("t.me/", "") results = [] # Итерируемся по сообщениям с ограничением в 10 результатов async for message in cl.iter_messages(target_group, search=query, limit=10): sender = message.sender_id or "Аноним" date_str = message.date.strftime("%Y-%m-%d %H:%M:%S") if message.date else "Неизвестно" text_snippet = message.text or "[Вложение/Медиа без текста]" results.append(f"[{date_str}] ID {sender}: {text_snippet}") return "\n\n".join(results) if results else "ℹ️ По данному поисковому запросу ничего не найдено." except ChannelInvalidError: return f"❌ Ошибка: Не удалось получить доступ к целевому чату '{group_username}'. Убедитесь, что юзербот состоит в нем." except ValueError as ve: return f"❌ Ошибка валидации сущности чата: {str(ve)}" except Exception as e: logger.error(f"Сбой при поиске по ключевому слову: {str(e)}") return f"❌ Ошибка Telethon при поиске по контуру: {str(e)}" async def send_private_message(username: str, text: str) -> str: """Прямая отправка сообщения пользователю или в чат от лица юзербота""" try: cl = await get_client() if not await check_authorization(cl): return "❌ Ошибка: Сессия юзербота не авторизована." await cl.send_message(username.strip(), text) return f"✅ Сообщение успешно отправлено адресату {username}" except Exception as e: return f"❌ Ошибка при отправке приватного сообщения: {str(e)}" async def join_telegram_chat(link_or_username: str) -> str: """ Автоматическое вступление юзербота в закрытые группы (по инвайт-ссылкам) и открытые чаты/каналы (по публичным ссылкам или юзернеймам). """ try: cl = await get_client() if not await check_authorization(cl): return "❌ Ошибка: Юзербот не имеет активной авторизации." target = link_or_username.strip() # Определение приватной инвайт-ссылки (содержит '+' или 'joinchat/') if "+" in target or "joinchat/" in target: if "+" in target: invite_hash = target.split("+")[-1] else: invite_hash = target.split("joinchat/")[-1] # Отрезаем GET-параметры реферальных или буст-ссылок invite_hash = invite_hash.split("?")[0].split("/")[0] await cl(ImportChatInviteRequest(invite_hash)) return "✅ Юзербот успешно вступил в приватный чат по инвайт-ссылке!" else: # Обработка публичных каналов, групп или @username clean_target = target.replace("https://", "").replace("http://", "").replace("t.me/", "").replace("telegram.me/", "").replace("@", "") clean_target = clean_target.split("/")[0].split("?")[0] await cl(JoinChannelRequest(clean_target)) return f"✅ Юзербот успешно вступил в чат/канал: @{clean_target}" except UserAlreadyParticipantError: return "ℹ️ Юзербот уже состоит в данном чате." except (InviteHashExpiredError, InviteHashInvalidError): return "❌ Ошибка: Ссылка-приглашение недействительна, просрочена или была отозвана администраторами." except FloodWaitError as e: logger.warning(f"Сработал лимитатор Telegram API: FloodWait на {e.seconds} секунд.") return f"❌ Защитный контур Telegram: Сработал флуд-контроль. Требуется пауза {e.seconds} сек." except Exception as e: logger.error(f"Непредвиденный сбой входа в чат: {str(e)}") return f"❌ Не удалось выполнить вход в чат. Ошибка API Telethon: {str(e)}" # ===================================================================== # МОДУЛЬ СКАЧИВАНИЯ МЕДИАФАЙЛОВ И ДОКУМЕНТОВ # ===================================================================== async def download_media_from_message(chat_id: int, message_id: int) -> dict: """ Находит сообщение по ID, проверяет ограничения безопасности (размер), скачивает в изолированную директорию downloads/ и возвращает метаданные. """ try: cl = await get_client() if not await check_authorization(cl): return {"status": "error", "message": "Юзербот не авторизован."} # Получаем объект сообщения напрямую из чата msg = await cl.get_messages(int(chat_id), ids=int(message_id)) if not msg or not msg.media: return {"status": "empty", "message": "Сообщение не содержит медиафайлов."} # Высокоуровневая и безопасная проверка размера через Telethon File API file_size = msg.file.size if msg.file else 0 if file_size > MAX_FILE_SIZE_BYTES: size_mb = round(file_size / (1024 * 1024), 2) logger.warning(f"Превышен лимит размера: {size_mb}MB в сообщении {message_id}") return {"status": "error", "message": f"Файл слишком велик ({size_mb} МБ). Лимит обработки ядра — 25 МБ."} # Создаем изолированную директорию для загрузок через Path-api download_dir = Path("downloads") download_dir.mkdir(exist_ok=True) # Определяем MIME-категорию file_mime = "document" if msg.voice: file_mime = "voice" elif msg.audio: file_mime = "audio" elif msg.photo: file_mime = "photo" elif msg.file and msg.file.ext: ext_lower = msg.file.ext.lower() if ext_lower == ".pdf": file_mime = "pdf" elif ext_lower in [".txt", ".log", ".json", ".py", ".md"]: file_mime = "text" # Безопасное извлечение расширения файла file_extension = msg.file.ext.lower() if (msg.file and msg.file.ext) else ".bin" # Формируем уникальное имя файла для исключения Race Conditions target_filename = f"{file_mime}_{chat_id}_{message_id}{file_extension}" full_path = download_dir / target_filename logger.info(f"Запуск асинхронного скачивания медиа {target_filename} ({round(file_size/1024)} КБ)...") downloaded_path = await cl.download_media(msg, file=str(full_path)) if downloaded_path and os.path.exists(downloaded_path): return { "status": "success", "file_path": downloaded_path, "mime_type": file_mime, "file_size": os.path.getsize(downloaded_path) } return {"status": "error", "message": "Не удалось сохранить файл на диск."} except Exception as e: logger.error(f"Ошибка в модуле download_media: {str(e)}") return {"status": "error", "message": str(e)}