feat(modules): add TDATA wrapper and checker

2026-08-01 22:12:33 +05:00
parent c48cd2f500
commit 23ca579003
3 changed files with 288 additions and 2 deletions
@@ -0,0 +1,117 @@
+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: str = "tdata"):
+        try:
+            self.path = path
+            if not Path(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, 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