From 23ca57900398cf6ab12132148e3b1fe25ce0cc41 Mon Sep 17 00:00:00 2001 From: root Date: Sat, 1 Aug 2026 22:12:33 +0500 Subject: [PATCH] feat(modules): add TDATA wrapper and checker --- .gitignore | 2 - modules/checker.py | 171 +++++++++++++++++++++++++++++++++++++++++++++ modules/tdata.py | 117 +++++++++++++++++++++++++++++++ 3 files changed, 288 insertions(+), 2 deletions(-) create mode 100644 modules/checker.py create mode 100644 modules/tdata.py diff --git a/.gitignore b/.gitignore index dfaf780..3aad907 100644 --- a/.gitignore +++ b/.gitignore @@ -189,5 +189,3 @@ pyvenv.cfg pip-selfcheck.json temp -data -tdata diff --git a/modules/checker.py b/modules/checker.py new file mode 100644 index 0000000..7cfd345 --- /dev/null +++ b/modules/checker.py @@ -0,0 +1,171 @@ +import logging + +from dataclasses import dataclass +from phonenumbers import parse, geocoder + +from typing import List, Optional + +from telethon.types import User, Chat, Channel +from telethon.tl.functions.payments import GetStarsStatusRequest +from telethon.tl.functions.contacts import GetContactsRequest + +from modules.tdata import TDATA + +logger = logging.getLogger(__name__) + + +# Unified Result class +@dataclass +class Result: + valid: bool + error: Optional[str] = None + + me: Optional[User] = None + stars: Optional[int] = None + country: Optional[str] = None + contacts: Optional[List[User]] = None + chats: Optional[List[Chat]] = None + channels: Optional[List[Channel]] = None + admin_channels: Optional[List[Channel]] = None + + +class Checker: + def __init__(self, path: str): + self.path = path + self.client = None + logger.info(f"Инициализация проверки для пути: {path}") + try: + self.desktop = TDATA(path=self.path) + except Exception as e: + logger.error(f"Ошибка инициализации TDATA: {e}") + self.desktop = None + + async def __aenter__(self): + return self + + async def __aexit__(self, exc_type, exc_val, exc_tb): + if self.client: + logger.debug("Отключение клиента Telethon") + try: + await self.client.disconnect() + except: + pass + + async def _ensure_client(self): + """Обеспечение инициализации клиента перед операцией""" + if self.client is None: + if self.desktop is None: + raise Exception("TDATA не инициализирована") + logger.debug("Инициализация клиента Telethon из TDATA") + self.client = await self.desktop.convert() + + async def validate(self) -> Result: + """Валидация TDATA""" + try: + if self.desktop is None: + return Result(valid=False, error="TDATA не инициализирована") + + if not await self.desktop.validate(): + return Result(valid=False, error="TDATA невалидна или пуста") + + await self._ensure_client() + me = await self.client.get_me() + + if not me: + return Result( + valid=False, error="Не удалось получить информацию о пользователе" + ) + + return Result(valid=True, me=me) + + except Exception as e: + logger.error(f"Ошибка валидации: {e}") + return Result(valid=False, error=str(e)) + + async def check_all(self) -> Result: + """Полная проверка аккаунта""" + try: + # Сначала валидация + validation = await self.validate() + if not validation.valid: + return validation + + me = validation.me + logger.debug(f"Пользователь: {me.first_name} (+{me.phone})") + + # Собираем данные с обработкой ошибок + stars = None + country = None + contacts = None + chats = None + channels = None + admin_channels = None + + try: + stars_status = await self.client(GetStarsStatusRequest(peer=me)) + stars = stars_status.balance.amount + except Exception as e: + logger.warning(f"Не удалось получить баланс звезд: {e}") + + try: + phone_parse = parse(str(f"+{me.phone}")) + country = geocoder.description_for_number(phone_parse, "ru") + except Exception as e: + logger.warning(f"Не удалось определить страну: {e}") + + try: + logger.debug("Получение контактов") + result = await self.client(GetContactsRequest(hash=0)) + contacts = result.users + except Exception as e: + logger.warning(f"Не удалось получить контакты: {e}") + + try: + logger.debug("Получение всех диалогов") + chats = [] + async for dialog in self.client.iter_dialogs(): + chats.append(dialog.entity) + except Exception as e: + logger.warning(f"Не удалось получить чаты: {e}") + + if chats: + try: + channels = [chat for chat in chats if isinstance(chat, Channel)] + except Exception as e: + logger.warning(f"Не удалось отфильтровать каналы: {e}") + + try: + logger.debug("Проверка прав администратора") + admin_channels = [] + for chat in chats: + if isinstance(chat, Channel): + try: + permissions = await self.client.get_permissions( + chat, "me" + ) + if permissions.is_admin or permissions.is_creator: + admin_channels.append(chat) + except: + pass + except Exception as e: + logger.warning(f"Не удалось проверить админские права: {e}") + + return Result( + valid=True, + me=me, + stars=stars, + country=country, + contacts=contacts, + chats=chats, + channels=channels, + admin_channels=admin_channels, + ) + + except Exception as e: + logger.error(f"Критическая ошибка при проверке: {e}") + return Result(valid=False, error=str(e)) + + +class MassChecker: + def __init__(self): + pass diff --git a/modules/tdata.py b/modules/tdata.py new file mode 100644 index 0000000..3706c66 --- /dev/null +++ b/modules/tdata.py @@ -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