136 lines
5.2 KiB
Python
136 lines
5.2 KiB
Python
import logging
|
||
|
||
from uuid import uuid4
|
||
from pathlib import Path
|
||
|
||
from opentele2.td import TDesktop
|
||
from opentele2.tl import TelegramClient
|
||
from opentele2.api import UseCurrentSession, API
|
||
from opentele2.exception import (
|
||
TFileNotFound,
|
||
TDesktopNotLoaded,
|
||
TDesktopUnauthorized,
|
||
)
|
||
|
||
logger = logging.getLogger(__name__)
|
||
|
||
|
||
class TDATA:
|
||
def __init__(self, path: Path, temp_path: Path):
|
||
logger.debug(f"Инициализация TDATA с путем: {path}")
|
||
try:
|
||
self.path = path
|
||
self.temp_path = temp_path
|
||
|
||
if not self.path.exists():
|
||
logger.error(f"Путь '{self.path}' не существует")
|
||
raise FileNotFoundError(f"Путь '{self.path}' не существует")
|
||
|
||
self.desktop = None
|
||
self.telethon = None
|
||
self.api = API.TelegramDesktop
|
||
|
||
logger.debug("Загрузка TDesktop")
|
||
self.desktop = self.__get_telegram_desktop()
|
||
if self.desktop:
|
||
logger.debug("TDesktop успешно загружен")
|
||
|
||
except Exception as e:
|
||
logger.error(f"Не удалось инициализировать TDATA: {e}")
|
||
raise
|
||
|
||
def __get_telegram_desktop(self) -> TDesktop | None:
|
||
logger.debug("Попытка загрузки TDesktop")
|
||
try:
|
||
if self.desktop:
|
||
return self.desktop
|
||
|
||
desktop = TDesktop(self.path.absolute(), api=self.api)
|
||
logger.debug("TDesktop создан успешно")
|
||
return desktop
|
||
|
||
except TFileNotFound as e:
|
||
logger.warning(f"Файлы TDATA не найдены или неполные: {e}")
|
||
return None
|
||
|
||
except Exception as e:
|
||
logger.error(f"Не удалось инициализировать TDesktop: {e}")
|
||
raise
|
||
|
||
def __gen_session_path(self) -> str:
|
||
logger.debug("Генерация уникального пути для сессии")
|
||
tempPath = f"temp/{uuid4()}.session"
|
||
|
||
while Path(tempPath).exists():
|
||
logger.debug(f"Путь сессии '{tempPath}' уже существует, генерация нового")
|
||
tempPath = f"temp/{uuid4()}.session"
|
||
|
||
logger.debug(f"Сгенерирован путь сессии: {tempPath}")
|
||
return tempPath
|
||
|
||
async def validate(self) -> bool:
|
||
logger.debug("Валидация TDATA")
|
||
if self.desktop is None:
|
||
logger.warning("TDesktop не инициализирован")
|
||
return False
|
||
|
||
is_loaded = self.desktop.isLoaded()
|
||
if is_loaded:
|
||
logger.debug("TDATA содержит загруженные аккаунты")
|
||
else:
|
||
logger.warning("TDATA не содержит загруженных аккаунтов")
|
||
|
||
return is_loaded
|
||
|
||
async def is_valid(self, client: TelegramClient, disconnect: bool = True) -> bool:
|
||
logger.debug("Проверка авторизации Telethon клиента")
|
||
try:
|
||
await client.connect()
|
||
logger.debug("Подключение к Telegram установлено")
|
||
|
||
authorized = await client.is_user_authorized()
|
||
|
||
if not authorized:
|
||
logger.error("Пользователь не авторизован")
|
||
raise TDesktopUnauthorized()
|
||
|
||
logger.debug("Клиент успешно авторизован")
|
||
self.telethon = client
|
||
return True
|
||
|
||
except Exception as e:
|
||
logger.error(f"Не удалось валидировать клиент: {e}")
|
||
return False
|
||
|
||
finally:
|
||
if disconnect:
|
||
logger.debug("Отключение клиента после валидации")
|
||
await client.disconnect()
|
||
|
||
async def convert(self, lang_code: str = "en") -> TelegramClient:
|
||
logger.debug("Начало конвертации TDATA в Telethon сессию")
|
||
if not await self.validate():
|
||
logger.error("TDATA невалидна, конвертация невозможна")
|
||
raise TDesktopNotLoaded("TDATA невалидна или не авторизована")
|
||
|
||
tempPath = self.__gen_session_path()
|
||
|
||
try:
|
||
logger.debug(f"Создание Telethon сессии по пути '{tempPath}' с языком '{lang_code}'")
|
||
client = await self.desktop.ToTelethon(
|
||
session=tempPath, flag=UseCurrentSession, api=self.api, lang_code=lang_code
|
||
)
|
||
|
||
logger.debug("Валидация созданной сессии")
|
||
valid_status = await self.is_valid(client, disconnect=False)
|
||
if not valid_status:
|
||
logger.error("Созданная сессия невалидна")
|
||
raise TDesktopUnauthorized("TDATA невалидна или не авторизована")
|
||
|
||
logger.debug(f"Конвертация успешна: {tempPath}")
|
||
return self.telethon
|
||
|
||
except Exception as e:
|
||
logger.error(f"Не удалось конвертировать TDATA: {e}")
|
||
raise
|