""" Telegram Client API (MTProto) service для отправки сообщений без /start Использует Pyrogram для подключения через пользовательский аккаунт Поддерживает несколько аккаунтов из БД """ import asyncio import os from typing import Optional, Dict from sqlalchemy.orm import Session from app.core.config import get_settings import logging # Опциональный импорт Pyrogram try: from pyrogram import Client from pyrogram.errors import ( FloodWait, UserNotParticipant, ChatWriteForbidden, PeerIdInvalid, UsernameNotOccupied, UsernameInvalid ) PYROGRAM_AVAILABLE = True except ImportError: PYROGRAM_AVAILABLE = False Client = None FloodWait = Exception UserNotParticipant = Exception ChatWriteForbidden = Exception PeerIdInvalid = Exception UsernameNotOccupied = Exception UsernameInvalid = Exception logger = logging.getLogger(__name__) _settings = get_settings() # Кэш клиентов по account_id (вместо глобального singleton) _telegram_clients: Dict[int, Client] = {} _client_locks: Dict[int, asyncio.Lock] = {} _clients_lock = asyncio.Lock() # Блокировка для управления словарем клиентов async def get_telegram_client(account_id: Optional[int] = None, db: Optional[Session] = None) -> Optional[Client]: """ Получить или создать Telegram Client для аккаунта Args: account_id: ID аккаунта из БД. Если None, используется аккаунт из env (legacy) db: Database session для загрузки аккаунта Returns: Client: Pyrogram клиент или None если не настроен """ if not PYROGRAM_AVAILABLE: logger.warning("Pyrogram не установлен. Установите: pip install pyrogram tgcrypto") return None global _telegram_clients, _client_locks # Если указан account_id, загружаем из БД if account_id and db: from app.db.models import TelegramMtprotoAccount account = db.query(TelegramMtprotoAccount).filter( TelegramMtprotoAccount.id == account_id, TelegramMtprotoAccount.status == "active" ).first() if not account: logger.warning(f"MTProto аккаунт {account_id} не найден или не активен") return None # Получаем или создаем lock для этого аккаунта async with _clients_lock: if account_id not in _client_locks: _client_locks[account_id] = asyncio.Lock() account_lock = _client_locks[account_id] # Если клиент уже создан, проверяем его if account_id in _telegram_clients: client = _telegram_clients[account_id] if client.is_connected: try: me = await client.get_me() if me: return client except Exception as e: logger.warning(f"Ошибка проверки клиента {account_id}: {e}, пересоздаем") async with account_lock: if account_id in _telegram_clients: try: await _telegram_clients[account_id].stop() except: pass del _telegram_clients[account_id] # Создаем новый клиент async with account_lock: # Двойная проверка if account_id in _telegram_clients and _telegram_clients[account_id].is_connected: return _telegram_clients[account_id] try: # Расшифровываем API hash from app.utils.encryption import decrypt_token api_hash = decrypt_token(account.api_hash_enc) if not api_hash: logger.error(f"Не удалось расшифровать API hash для аккаунта {account_id}") return None # Создаем путь для сессии # ВАЖНО: Используем фиксированное имя для сессии, не связанное с session_path # session_path используется для временного хранения phone_code_hash session_base_path = os.path.join(os.getcwd(), _settings.TELEGRAM_SESSION_PATH) os.makedirs(session_base_path, exist_ok=True) session_name = f"mtproto_{account_id}" # Создаем клиент # name - имя сессии без расширения, Pyrogram добавит .session автоматически # workdir - директория где будет сохранен файл сессии client = Client( name=session_name, api_id=account.api_id, api_hash=api_hash, phone_number=account.phone, workdir=session_base_path ) # Подключаемся - Pyrogram автоматически подключится если сессия валидна # Если сессия невалидна или отсутствует, нужно авторизоваться через API try: await client.start() # Проверяем, что клиент действительно авторизован me = await client.get_me() if not me: logger.warning(f"Клиент {account_id} подключен, но не авторизован") try: await client.stop() except: pass return None except Exception as e: # Если требуется код или сессия невалидна - это нормально # Нужно авторизоваться через API эндпоинт logger.warning(f"Требуется авторизация для аккаунта {account_id}: {e}") try: await client.stop() except: pass return None # Обновляем username в БД если изменился me = await client.get_me() if me and me.username != account.username: account.username = me.username db.commit() _telegram_clients[account_id] = client logger.info(f"MTProto аккаунт {account_id} подключен: @{me.username if me else 'N/A'}") return client except Exception as e: logger.error(f"Ошибка создания MTProto клиента для аккаунта {account_id}: {e}", exc_info=True) return None # Legacy: используем настройки из env (для обратной совместимости) if not _settings.TELEGRAM_API_ID or not _settings.TELEGRAM_API_HASH or not _settings.TELEGRAM_PHONE: logger.warning("Telegram Client API не настроен: отсутствуют настройки в env или аккаунт не указан") return None # Используем account_id=0 для legacy аккаунта из env legacy_account_id = 0 async with _clients_lock: if legacy_account_id not in _client_locks: _client_locks[legacy_account_id] = asyncio.Lock() legacy_lock = _client_locks[legacy_account_id] if legacy_account_id in _telegram_clients: client = _telegram_clients[legacy_account_id] if client.is_connected: try: me = await client.get_me() if me: return client except Exception as e: logger.warning(f"Ошибка проверки legacy клиента: {e}") async with legacy_lock: if legacy_account_id in _telegram_clients: try: await _telegram_clients[legacy_account_id].stop() except: pass del _telegram_clients[legacy_account_id] async with legacy_lock: if legacy_account_id in _telegram_clients and _telegram_clients[legacy_account_id].is_connected: return _telegram_clients[legacy_account_id] try: session_path = os.path.join(os.getcwd(), _settings.TELEGRAM_SESSION_PATH) os.makedirs(session_path, exist_ok=True) session_name = "legacy_session" # Фиксированное имя для legacy сессии client = Client( name=session_name, api_id=_settings.TELEGRAM_API_ID, api_hash=_settings.TELEGRAM_API_HASH, phone_number=_settings.TELEGRAM_PHONE, workdir=session_path ) await client.start() _telegram_clients[legacy_account_id] = client logger.info(f"Legacy MTProto клиент подключен: {client.me.username if client.me else 'N/A'}") return client except Exception as e: logger.error(f"Ошибка создания legacy MTProto клиента: {e}") return None async def send_message_via_mtproto( recipient: str, message: str, parse_mode: Optional[str] = None, account_id: Optional[int] = None, db: Optional[Session] = None ) -> tuple[bool, Optional[str]]: """ Отправить сообщение через Telegram Client API (MTProto) Позволяет отправлять сообщения без /start, так как использует пользовательский аккаунт Args: recipient: username (@username) или chat_id (число) message: текст сообщения parse_mode: режим парсинга (Markdown, HTML, None) account_id: ID MTProto аккаунта из БД (опционально) db: Database session (требуется если указан account_id) Returns: tuple[bool, Optional[str]]: (success, error_message) """ if not PYROGRAM_AVAILABLE: return False, "Pyrogram не установлен. Установите: pip install pyrogram tgcrypto" client = await get_telegram_client(account_id=account_id, db=db) if not client: if account_id: return False, f"MTProto аккаунт {account_id} не настроен или недоступен" return False, "Telegram Client API не настроен. Проверьте настройки или добавьте MTProto аккаунт" try: # Парсим recipient # recipient может быть строкой или числом if isinstance(recipient, int): # Уже число - используем как chat_id target = recipient else: recipient_str = str(recipient) # Если это username (начинается с @), убираем @ if recipient_str.startswith('@'): username = recipient_str[1:] target = username elif recipient_str.lstrip('-').isdigit(): # Это числовой chat_id (строка) target = int(recipient_str) else: # Попробуем как username target = recipient_str # Конвертируем parse_mode для Pyrogram pyrogram_parse_mode = None if parse_mode == "Markdown": pyrogram_parse_mode = "markdown" elif parse_mode == "HTML": pyrogram_parse_mode = "html" # Отправляем сообщение sent_message = await client.send_message( chat_id=target, text=message, parse_mode=pyrogram_parse_mode ) logger.info(f"Сообщение отправлено через MTProto: {sent_message.id}") return True, None except FloodWait as e: error_msg = f"Превышен лимит запросов. Подождите {e.value} секунд" logger.warning(error_msg) return False, error_msg except (UserNotParticipant, ChatWriteForbidden) as e: error_msg = f"Нет доступа к чату: {str(e)}" logger.warning(error_msg) return False, error_msg except (PeerIdInvalid, UsernameNotOccupied, UsernameInvalid) as e: error_msg = f"Неверный получатель ({recipient}): {str(e)}" logger.warning(error_msg) return False, error_msg except Exception as e: error_msg = f"Ошибка отправки через MTProto: {str(e)}" logger.error(error_msg, exc_info=True) return False, error_msg async def is_mtproto_available(account_id: Optional[int] = None, db: Optional[Session] = None) -> bool: """Проверить доступен ли Telegram Client API""" if not PYROGRAM_AVAILABLE: return False client = await get_telegram_client(account_id=account_id, db=db) return client is not None and client.is_connected async def get_default_mtproto_account(tenant_id: int, db: Session) -> Optional[int]: """ Получить ID аккаунта по умолчанию для тенанта Returns: int: ID аккаунта или None если нет аккаунтов """ if not PYROGRAM_AVAILABLE: return None from app.db.models import TelegramMtprotoAccount # Сначала ищем аккаунт помеченный как default account = db.query(TelegramMtprotoAccount).filter( TelegramMtprotoAccount.tenant_id == tenant_id, TelegramMtprotoAccount.status == "active", TelegramMtprotoAccount.is_default == True ).first() if account: return account.id # Если нет default, берем первый активный account = db.query(TelegramMtprotoAccount).filter( TelegramMtprotoAccount.tenant_id == tenant_id, TelegramMtprotoAccount.status == "active" ).first() return account.id if account else None async def get_mtproto_account_by_phone(tenant_id: int, customer_phone: str, db: Session) -> Optional[int]: """ Получить ID MTProto аккаунта по номеру телефона клиента Логика: ищем MTProto аккаунт с номером телефона, совпадающим с номером клиента Args: tenant_id: ID тенанта customer_phone: Номер телефона клиента (например, +79991234567) db: Database session Returns: int: ID MTProto аккаунта или None если не найден """ if not PYROGRAM_AVAILABLE or not customer_phone: return None from app.db.models import TelegramMtprotoAccount from sqlalchemy import func # Нормализуем номер телефона (убираем пробелы, дефисы и т.д.) normalized_phone = customer_phone.strip().replace(' ', '').replace('-', '').replace('(', '').replace(')', '') # Ищем аккаунт с таким же номером телефона (нормализуем для сравнения) # Получаем все аккаунты и сравниваем нормализованные номера в Python accounts = db.query(TelegramMtprotoAccount).filter( TelegramMtprotoAccount.tenant_id == tenant_id, TelegramMtprotoAccount.status == "active" ).all() for account in accounts: # Нормализуем номер аккаунта для сравнения account_phone_normalized = account.phone.strip().replace(' ', '').replace('-', '').replace('(', '').replace(')', '') if account_phone_normalized == normalized_phone: return account.id if account: return account.id # Если не нашли по точному совпадению, пробуем найти по умолчанию return await get_default_mtproto_account(tenant_id, db) async def close_telegram_client(account_id: Optional[int] = None): """Закрыть соединение с Telegram Client""" if not PYROGRAM_AVAILABLE: return global _telegram_clients if account_id is not None: # Закрываем конкретный клиент if account_id in _telegram_clients: try: await _telegram_clients[account_id].stop() except Exception as e: logger.error(f"Ошибка закрытия клиента {account_id}: {e}") finally: del _telegram_clients[account_id] else: # Закрываем все клиенты for acc_id, client in list(_telegram_clients.items()): try: await client.stop() except Exception as e: logger.error(f"Ошибка закрытия клиента {acc_id}: {e}") _telegram_clients.clear()