timeout by asyncio
This commit is contained in:
@ -1,6 +1,7 @@
|
||||
from .async_base_session import AsyncBaseSession
|
||||
from .base_data import BaseData
|
||||
from .base_session import BaseSession
|
||||
from .base_thon import BaseData, BaseThon
|
||||
from .base_thon import BaseThon
|
||||
from .json_async_converter import JsonAsyncConverter
|
||||
from .json_converter import JsonConverter
|
||||
from .models import ThonOptions, ThonSearchCodeOptions
|
||||
|
||||
110
basethon/base_data.py
Normal file
110
basethon/base_data.py
Normal file
@ -0,0 +1,110 @@
|
||||
from telethon.sessions import StringSession
|
||||
|
||||
|
||||
class BaseData:
|
||||
def __init__(self, json_data: dict, raise_error: bool):
|
||||
self.__json_data, self.__raise_error = json_data, raise_error
|
||||
|
||||
@property
|
||||
def json_data(self) -> dict:
|
||||
return self.__json_data
|
||||
|
||||
def json_data_edit(self, key: str, value: str | int | bool | None):
|
||||
self.__json_data[key] = value
|
||||
|
||||
@property
|
||||
def session_file(self) -> str:
|
||||
if not (session_file := self.json_data.get("session_file")):
|
||||
if self.__raise_error:
|
||||
raise ValueError("ERROR_SESSION_FILE:BASE_THON")
|
||||
return ""
|
||||
return session_file
|
||||
|
||||
@property
|
||||
def string_session(self) -> StringSession:
|
||||
if not (string_session := self.json_data.get("string_session")):
|
||||
if self.__raise_error:
|
||||
raise ValueError("ERROR_STRING_SESSION:BASE_THON")
|
||||
return StringSession()
|
||||
return StringSession(string_session)
|
||||
|
||||
@property
|
||||
def app_id(self) -> int:
|
||||
"""Api Id"""
|
||||
if api_id := self.json_data.get("api_id"):
|
||||
return api_id
|
||||
if not (app_id := self.json_data.get("app_id")):
|
||||
raise ValueError("ERROR_APP_ID:BASE_THON")
|
||||
return app_id
|
||||
|
||||
@property
|
||||
def app_hash(self) -> str:
|
||||
"""Api Hash"""
|
||||
if api_hash := self.json_data.get("api_hash"):
|
||||
return api_hash
|
||||
if not (app_hash := self.json_data.get("app_hash")):
|
||||
raise ValueError("ERROR_APP_HASH:BASE_THON")
|
||||
return app_hash
|
||||
|
||||
@property
|
||||
def device(self) -> str:
|
||||
"""Device Model"""
|
||||
if device_model := self.json_data.get("device_model"):
|
||||
return device_model
|
||||
if not (device := self.json_data.get("device")):
|
||||
raise ValueError("ERROR_DEVICE:BASE_THON")
|
||||
return device
|
||||
|
||||
@property
|
||||
def sdk(self) -> str:
|
||||
"""System Version"""
|
||||
if system_version := self.json_data.get("system_version"):
|
||||
return system_version
|
||||
if not (sdk := self.json_data.get("sdk")):
|
||||
raise ValueError("ERROR_SDK:BASE_THON")
|
||||
return sdk
|
||||
|
||||
@property
|
||||
def app_version(self) -> str:
|
||||
"""App Version"""
|
||||
if not (app_version := self.json_data.get("app_version")):
|
||||
raise ValueError("ERROR_APP_VERSION:BASE_THON")
|
||||
return app_version
|
||||
|
||||
@property
|
||||
def lang_pack(self) -> str:
|
||||
"""Lang Pack"""
|
||||
if lang_code := self.json_data.get("lang_code"):
|
||||
return lang_code
|
||||
return self.json_data.get("lang_pack", "en")
|
||||
|
||||
@property
|
||||
def system_lang_code(self) -> str:
|
||||
"""System Lang Code"""
|
||||
if system_lang_code := self.json_data.get("system_lang_code"):
|
||||
return system_lang_code
|
||||
return self.json_data.get("system_lang_pack", "en-us")
|
||||
|
||||
@property
|
||||
def twostep(self) -> str | None:
|
||||
"""2FA"""
|
||||
if password := self.json_data.get("password"):
|
||||
return password
|
||||
if twofa := self.json_data.get("twoFA"):
|
||||
return twofa
|
||||
if twostep := self.json_data.get("twostep"):
|
||||
return twostep
|
||||
|
||||
@property
|
||||
def proxy(self) -> dict | tuple:
|
||||
if not (proxy := self.json_data.get("proxy")):
|
||||
if self.__raise_error:
|
||||
raise ValueError("ERROR_PROXY:BASE_THON")
|
||||
return {}
|
||||
return proxy
|
||||
|
||||
@property
|
||||
def tz_offset(self) -> int:
|
||||
if tz_offset := self.json_data.get("tz_offset", 0):
|
||||
return tz_offset
|
||||
return 0
|
||||
@ -7,133 +7,25 @@ from random import randint
|
||||
from typing import Self
|
||||
|
||||
from telethon import TelegramClient
|
||||
from telethon.sessions import StringSession
|
||||
from telethon.tl.functions.account import UpdateStatusRequest
|
||||
from telethon.tl.functions.help import GetCountriesListRequest, GetNearestDcRequest
|
||||
from telethon.tl.functions.langpack import GetLangPackRequest
|
||||
from telethon.types import JsonNumber, JsonObject, JsonObjectValue, JsonString
|
||||
|
||||
from basethon.base_data import BaseData
|
||||
from basethon.models import ThonOptions, ThonSearchCodeOptions
|
||||
|
||||
from .constants import API_PACKS
|
||||
from .exceptions import ThonBannedError
|
||||
|
||||
|
||||
class BaseData:
|
||||
def __init__(self, json_data: dict, raise_error: bool):
|
||||
self.__json_data, self.__raise_error = json_data, raise_error
|
||||
|
||||
@property
|
||||
def json_data(self) -> dict:
|
||||
return self.__json_data
|
||||
|
||||
def json_data_edit(self, key: str, value: str | int | bool | None):
|
||||
self.__json_data[key] = value
|
||||
|
||||
@property
|
||||
def session_file(self) -> str:
|
||||
if not (session_file := self.json_data.get("session_file")):
|
||||
if self.__raise_error:
|
||||
raise ValueError("ERROR_SESSION_FILE:BASE_THON")
|
||||
return ""
|
||||
return session_file
|
||||
|
||||
@property
|
||||
def string_session(self) -> StringSession:
|
||||
if not (string_session := self.json_data.get("string_session")):
|
||||
if self.__raise_error:
|
||||
raise ValueError("ERROR_STRING_SESSION:BASE_THON")
|
||||
return StringSession()
|
||||
return StringSession(string_session)
|
||||
|
||||
@property
|
||||
def app_id(self) -> int:
|
||||
"""Api Id"""
|
||||
if api_id := self.json_data.get("api_id"):
|
||||
return api_id
|
||||
if not (app_id := self.json_data.get("app_id")):
|
||||
raise ValueError("ERROR_APP_ID:BASE_THON")
|
||||
return app_id
|
||||
|
||||
@property
|
||||
def app_hash(self) -> str:
|
||||
"""Api Hash"""
|
||||
if api_hash := self.json_data.get("api_hash"):
|
||||
return api_hash
|
||||
if not (app_hash := self.json_data.get("app_hash")):
|
||||
raise ValueError("ERROR_APP_HASH:BASE_THON")
|
||||
return app_hash
|
||||
|
||||
@property
|
||||
def device(self) -> str:
|
||||
"""Device Model"""
|
||||
if device_model := self.json_data.get("device_model"):
|
||||
return device_model
|
||||
if not (device := self.json_data.get("device")):
|
||||
raise ValueError("ERROR_DEVICE:BASE_THON")
|
||||
return device
|
||||
|
||||
@property
|
||||
def sdk(self) -> str:
|
||||
"""System Version"""
|
||||
if system_version := self.json_data.get("system_version"):
|
||||
return system_version
|
||||
if not (sdk := self.json_data.get("sdk")):
|
||||
raise ValueError("ERROR_SDK:BASE_THON")
|
||||
return sdk
|
||||
|
||||
@property
|
||||
def app_version(self) -> str:
|
||||
"""App Version"""
|
||||
if not (app_version := self.json_data.get("app_version")):
|
||||
raise ValueError("ERROR_APP_VERSION:BASE_THON")
|
||||
return app_version
|
||||
|
||||
@property
|
||||
def lang_pack(self) -> str:
|
||||
"""Lang Pack"""
|
||||
if lang_code := self.json_data.get("lang_code"):
|
||||
return lang_code
|
||||
return self.json_data.get("lang_pack", "en")
|
||||
|
||||
@property
|
||||
def system_lang_code(self) -> str:
|
||||
"""System Lang Code"""
|
||||
if system_lang_code := self.json_data.get("system_lang_code"):
|
||||
return system_lang_code
|
||||
return self.json_data.get("system_lang_pack", "en-us")
|
||||
|
||||
@property
|
||||
def twostep(self) -> str | None:
|
||||
"""2FA"""
|
||||
if password := self.json_data.get("password"):
|
||||
return password
|
||||
if twofa := self.json_data.get("twoFA"):
|
||||
return twofa
|
||||
if twostep := self.json_data.get("twostep"):
|
||||
return twostep
|
||||
|
||||
@property
|
||||
def proxy(self) -> dict | tuple:
|
||||
if not (proxy := self.json_data.get("proxy")):
|
||||
if self.__raise_error:
|
||||
raise ValueError("ERROR_PROXY:BASE_THON")
|
||||
return {}
|
||||
return proxy
|
||||
|
||||
@property
|
||||
def tz_offset(self) -> int:
|
||||
if tz_offset := self.json_data.get("tz_offset", 0):
|
||||
return tz_offset
|
||||
return 0
|
||||
|
||||
|
||||
class BaseThon(BaseData):
|
||||
def __init__(self, options: ThonOptions):
|
||||
self.__item = options.item
|
||||
self.__retries = options.retries
|
||||
self.__timeout = options.timeout
|
||||
self._logger = logging.getLogger("basethon")
|
||||
self._async_check_timeout = options.async_check_timeout
|
||||
super().__init__(options.json_data, options.raise_error)
|
||||
self.__client = self.__get_client()
|
||||
|
||||
@ -187,7 +79,7 @@ class BaseThon(BaseData):
|
||||
with contextlib.suppress(Exception):
|
||||
await self.client(GetCountriesListRequest(self.lang_pack, 0))
|
||||
|
||||
async def check(self) -> str:
|
||||
async def _check(self) -> str:
|
||||
try:
|
||||
await self.client.connect()
|
||||
if not await self.client.is_user_authorized():
|
||||
@ -207,6 +99,14 @@ class BaseThon(BaseData):
|
||||
self._logger.exception(e)
|
||||
return f"ERROR_AUTH:{e}"
|
||||
|
||||
async def check(self) -> str:
|
||||
if not self._async_check_timeout:
|
||||
return await self._check()
|
||||
try:
|
||||
return await asyncio.wait_for(self._check(), self._async_check_timeout)
|
||||
except asyncio.TimeoutError:
|
||||
return "ERROR_AUTH:CONNECTION_ERROR:TIMEOUT"
|
||||
|
||||
async def search_code(self, options: ThonSearchCodeOptions | None = None) -> str:
|
||||
options = options or ThonSearchCodeOptions()
|
||||
future = datetime.now() + timedelta(seconds=options.wait_time)
|
||||
|
||||
@ -9,6 +9,11 @@ class ThonOptions:
|
||||
retries: int = 10
|
||||
timeout: int = 10
|
||||
raise_error: bool = True
|
||||
async_check_timeout: int | None = 0
|
||||
|
||||
def __post_init__(self):
|
||||
if self.async_check_timeout == 0:
|
||||
self.async_check_timeout = self.retries * self.timeout
|
||||
|
||||
|
||||
@dataclass
|
||||
|
||||
Reference in New Issue
Block a user