120 lines
4.2 KiB
Python
120 lines
4.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):
|
|
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
|
|
self.desktop = self.__get_telegram_desktop()
|
|
logger.debug(f"TDATA инициализирована с путем: {self.path}")
|
|
|
|
except Exception as e:
|
|
logger.error(f"Не удалось инициализировать TDATA: {e}")
|
|
raise
|
|
|
|
def __get_telegram_desktop(self) -> TDesktop | None:
|
|
try:
|
|
if self.desktop:
|
|
return self.desktop
|
|
|
|
return TDesktop(self.path.absolute(), api=self.api)
|
|
|
|
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"
|
|
|
|
return tempPath
|
|
|
|
async def validate(self) -> bool:
|
|
logger.debug("Валидация TDATA")
|
|
if self.desktop is None:
|
|
logger.warning("TDesktop не инициализирован, TDATA невалидна")
|
|
return False
|
|
|
|
return self.desktop.isLoaded()
|
|
|
|
async def valid(self, client: TelegramClient, disconnect: bool = True) -> bool:
|
|
try:
|
|
logger.debug("Проверка авторизации клиента Telethon")
|
|
await client.connect()
|
|
|
|
authorized = await client.is_user_authorized()
|
|
|
|
if not authorized:
|
|
logger.error("Пользователь не авторизован")
|
|
raise TDesktopUnauthorized()
|
|
|
|
logger.debug("Клиент успешно авторизован")
|
|
self.telethon = client
|
|
return True
|
|
|
|
except:
|
|
logger.exception("Не удалось валидировать клиент")
|
|
return False
|
|
|
|
finally:
|
|
if disconnect:
|
|
logger.debug("Отключение клиента после валидации")
|
|
await client.disconnect()
|
|
|
|
async def convert(self) -> TelegramClient:
|
|
logger.debug("Конвертация TDATA в сессию Telethon")
|
|
if not await self.validate():
|
|
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
|
|
)
|
|
|
|
valid_status = await self.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.exception(f"Не удалось конвертировать TDATA: {e}")
|
|
raise
|