136 lines
5.1 KiB
Python
136 lines
5.1 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) -> 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}'")
|
|
client = await self.desktop.ToTelethon(
|
|
session=tempPath, flag=UseCurrentSession, api=self.api
|
|
)
|
|
|
|
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
|