base thon with aenter and aexit

basethon/__init__.py

@@ -0,0 +1,3 @@
+from .base_thon import BaseThon
+
+__all__ = ["BaseThon"]

basethon/base_thon.py

@@ -0,0 +1,176 @@
+import contextlib
+from pathlib import Path
+from typing import Self
+
+from telethon.errors import UserDeactivatedBanError, UserDeactivatedError
+from telethon.sessions import StringSession
+
+from .telegrambaseclient import TelegramClient
+
+TelethonBannedError = (UserDeactivatedError, UserDeactivatedBanError)
+
+
+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
+
+
+class BaseThon(BaseData):
+    def __init__(
+        self,
+        json_data: dict,
+        item: Path | None = None,
+        retries: int = 50,
+        timeout: int = 10,
+        raise_error: bool = True,
+    ):
+        self.__item, self.__retries, self.__timeout = item, retries, timeout
+        super().__init__(json_data, raise_error)
+        self.__client = self.__get_client()
+
+    @property
+    def client(self) -> TelegramClient:
+        return self.__client
+
+    def __get_client(self) -> TelegramClient:
+        __session = str(self.__item) if not self.__item else self.string_session
+        return TelegramClient(
+            session=__session,
+            api_id=self.app_id,
+            api_hash=self.app_hash,
+            device_model=self.device,
+            app_version=self.app_version,
+            system_lang_code=self.system_lang_code,
+            lang_code=self.lang_pack,
+            connection_retries=self.__retries,
+            request_retries=self.__retries,
+            proxy=self.proxy,
+            timeout=self.__timeout,
+        )
+
+    async def check(self) -> str:
+        try:
+            await self.client.connect()
+            if not await self.client.is_user_authorized():
+                return "ERROR_AUTH:BAN_ERROR"
+            return "OK"
+        except ConnectionError:
+            await self.disconnect()
+            return "ERROR_AUTH:CONNECTION_ERROR"
+        except TelethonBannedError:
+            await self.disconnect()
+            return "ERROR_AUTH:BAN_ERROR"
+        except Exception as e:
+            await self.disconnect()
+            return f"ERROR_AUTH:{e}"
+
+    async def disconnect(self):
+        with contextlib.suppress(Exception):
+            await self.client.disconnect()  # type: ignore
+
+    async def __aenter__(self) -> str | Self:
+        r = await self.check()
+        if r != "OK":
+            return r
+        return self
+
+    async def __aexit__(self, exc_type, exc_val, exc_tb):
+        await self.disconnect()

basethon/telegrambaseclient.py

@@ -0,0 +1,245 @@
+import asyncio
+import collections
+import logging
+import platform
+import re
+import time
+import typing
+
+from telethon import TelegramClient as TC
+from telethon import __name__ as __base_name__
+from telethon._updates import EntityCache as MbEntityCache
+from telethon._updates import MessageBox
+from telethon.extensions import markdown
+from telethon.network import Connection, ConnectionTcpFull, MTProtoSender, TcpMTProxy
+from telethon.sessions import MemorySession, Session, SQLiteSession
+from telethon.tl import functions, types
+
+DEFAULT_DC_ID = 2
+DEFAULT_IPV4_IP = "149.154.167.51"
+DEFAULT_IPV6_IP = "2001:67c:4e8:f002::a"
+DEFAULT_PORT = 443
+
+_base_log = logging.getLogger(__base_name__)
+
+API_PACKS = {
+    4: "android",
+    5: "android",
+    6: "android",
+    8: "ios",
+    2834: "macos",
+    2040: "tdesktop",
+    17349: "tdesktop",
+    21724: "android",
+    16623: "android",
+    2496: "",
+}
+
+
+class TelegramClient(TC):
+    def __init__(
+        self: "TelegramClient",
+        session: "typing.Union[str, Session]",
+        api_id: int,
+        api_hash: str,
+        *,
+        connection: "typing.Type[Connection]" = ConnectionTcpFull,
+        use_ipv6: bool = False,
+        proxy: typing.Union[tuple, dict] = None,  # type: ignore
+        local_addr: typing.Union[str, tuple] = None,  # type: ignore
+        timeout: int = 10,
+        request_retries: int = 5,
+        connection_retries: int = 5,
+        retry_delay: int = 1,
+        auto_reconnect: bool = True,
+        sequential_updates: bool = False,
+        flood_sleep_threshold: int = 60,
+        raise_last_call_error: bool = False,
+        device_model: str = None,  # type: ignore
+        system_version: str = None,  # type: ignore
+        app_version: str = None,  # type: ignore
+        lang_code: str = "en",
+        system_lang_code: str = "en",
+        base_logger: typing.Union[str, logging.Logger] = None,  # type: ignore
+        receive_updates: bool = True,
+        catch_up: bool = False,
+        entity_cache_limit: int = 5000,
+    ):
+        if not api_id or not api_hash:
+            raise ValueError(
+                "Your API ID or Hash cannot be empty or None. "
+                "Refer to telethon.rtfd.io for more information."
+            )
+
+        self._use_ipv6 = use_ipv6
+
+        if isinstance(base_logger, str):
+            base_logger = logging.getLogger(base_logger)
+        elif not isinstance(base_logger, logging.Logger):
+            base_logger = _base_log
+
+        class _Loggers(dict):
+            def __missing__(self, key):
+                if key.startswith("telethon."):
+                    key = key.split(".", maxsplit=1)[1]
+
+                return base_logger.getChild(key)
+
+        self._log = _Loggers()
+
+        # Determine what session object we have
+        if isinstance(session, str) or session is None:
+            try:
+                session = SQLiteSession(session)
+            except ImportError:
+                import warnings
+
+                warnings.warn(
+                    "The sqlite3 module is not available under this "
+                    "Python installation and no custom session "
+                    "instance was given; using MemorySession.\n"
+                    "You will need to re-login every time unless "
+                    "you use another session storage"
+                )
+                session = MemorySession()
+        elif not isinstance(session, Session):
+            raise TypeError("The given session must be a str or a Session instance.")
+
+        # ':' in session.server_address is True if it's an IPv6 address
+        if not session.server_address or (":" in session.server_address) != use_ipv6:
+            session.set_dc(
+                DEFAULT_DC_ID,
+                DEFAULT_IPV6_IP if self._use_ipv6 else DEFAULT_IPV4_IP,
+                DEFAULT_PORT,
+            )
+
+        self.flood_sleep_threshold = flood_sleep_threshold
+
+        self.session = session
+        self.api_id = int(api_id)
+        self.api_hash = api_hash
+
+        if not callable(getattr(self.loop, "sock_connect", None)):
+            raise TypeError(
+                "Event loop of type {} lacks `sock_connect`, which is needed to use proxies.\n\n"
+                "Change the event loop in use to use proxies:\n"
+                "# https://github.com/LonamiWebs/Telethon/issues/1337\n"
+                "import asyncio\n"
+                "asyncio.set_event_loop(asyncio.SelectorEventLoop())".format(
+                    self.loop.__class__.__name__
+                )
+            )
+
+        if local_addr is not None:
+            if use_ipv6 is False and ":" in local_addr:
+                raise TypeError(
+                    "A local IPv6 address must only be used with `use_ipv6=True`."
+                )
+            elif use_ipv6 is True and ":" not in local_addr:
+                raise TypeError(
+                    "`use_ipv6=True` must only be used with a local IPv6 address."
+                )
+
+        self._raise_last_call_error = raise_last_call_error
+
+        self._request_retries = request_retries
+        self._connection_retries = connection_retries
+        self._retry_delay = retry_delay or 0
+        self._proxy = proxy
+        self._local_addr = local_addr
+        self._timeout = timeout
+        self._auto_reconnect = auto_reconnect
+
+        assert isinstance(connection, type)
+        self._connection = connection
+        # noinspection PyUnresolvedReferences
+        init_proxy = (
+            None
+            if not issubclass(connection, TcpMTProxy)
+            else types.InputClientProxy(*connection.address_info(proxy))
+        )
+
+        # Used on connection. Capture the variables in a lambda since
+        # exporting clients need to create this InvokeWithLayerRequest.
+        system = platform.uname()
+
+        if system.machine in ("x86_64", "AMD64"):
+            default_device_model = "PC 64bit"
+        elif system.machine in ("i386", "i686", "x86"):
+            default_device_model = "PC 32bit"
+        else:
+            default_device_model = system.machine
+        default_system_version = re.sub(r"-.+", "", system.release)
+
+        self._init_request = functions.InitConnectionRequest(
+            api_id=self.api_id,
+            device_model=device_model or default_device_model or "Unknown",
+            system_version=system_version or default_system_version or "1.0",
+            app_version=app_version or self.__version__,
+            lang_code=lang_code,
+            system_lang_code=system_lang_code,
+            lang_pack=API_PACKS.get(self.api_id, "android"),
+            query=None,
+            proxy=init_proxy,
+        )
+
+        # Remember flood-waited requests to avoid making them again
+        self._flood_waited_requests = {}  # type: ignore
+
+        # Cache ``{dc_id: (_ExportState, MTProtoSender)}`` for all borrowed senders
+        self._borrowed_senders = {}  # type: ignore
+        self._borrow_sender_lock = asyncio.Lock()
+
+        self._loop = None  # only used as a sanity check
+        self._updates_error = None
+        self._updates_handle = None
+        self._keepalive_handle = None
+        self._last_request = time.time()
+        self._no_updates = not receive_updates
+
+        # Used for non-sequential updates, in order to terminate all pending tasks on disconnect.
+        self._sequential_updates = sequential_updates
+        self._event_handler_tasks = set()  # type: ignore
+
+        self._authorized = None
+
+        # Some further state for subclasses
+        self._event_builders = []  # type: ignore
+
+        # {chat_id: {Conversation}}
+        self._conversations = collections.defaultdict(set)  # type: ignore
+
+        self._albums = {}  # type: ignore
+
+        # Default parse mode
+        self._parse_mode = markdown
+
+        # Some fields to easy signing in. Let {phone: hash} be
+        # a dictionary because the user may change their mind.
+        self._phone_code_hash = {}  # type: ignore
+        self._phone = None
+        self._tos = None
+
+        # A place to store if channels are a megagroup or not (see `edit_admin`)
+        self._megagroup_cache = {}  # type: ignore
+
+        # This is backported from v2 in a very ad-hoc way just to get proper update handling
+        self._catch_up = catch_up
+        self._updates_queue = asyncio.Queue()  # type: ignore
+        self._message_box = MessageBox(self._log["messagebox"])
+        self._mb_entity_cache = (
+            MbEntityCache()
+        )  # required for proper update handling (to know when to getDifference)
+        self._entity_cache_limit = entity_cache_limit
+
+        self._sender = MTProtoSender(
+            self.session.auth_key,
+            loggers=self._log,
+            retries=self._connection_retries,
+            delay=self._retry_delay,
+            auto_reconnect=self._auto_reconnect,
+            connect_timeout=self._timeout,
+            auth_key_callback=self._auth_key_callback,
+            updates_queue=self._updates_queue,
+            auto_reconnect_callback=self._handle_auto_reconnect,
+        )

poetry.lock

@@ -0,0 +1,58 @@
+# This file is automatically @generated by Poetry 1.8.3 and should not be changed by hand.
+
+[[package]]
+name = "pyaes"
+version = "1.6.1"
+description = "Pure-Python Implementation of the AES block-cipher and common modes of operation"
+optional = false
+python-versions = "*"
+files = [
+    {file = "pyaes-1.6.1.tar.gz", hash = "sha256:02c1b1405c38d3c370b085fb952dd8bea3fadcee6411ad99f312cc129c536d8f"},
+]
+
+[[package]]
+name = "pyasn1"
+version = "0.6.1"
+description = "Pure-Python implementation of ASN.1 types and DER/BER/CER codecs (X.208)"
+optional = false
+python-versions = ">=3.8"
+files = [
+    {file = "pyasn1-0.6.1-py3-none-any.whl", hash = "sha256:0d632f46f2ba09143da3a8afe9e33fb6f92fa2320ab7e886e2d0f7672af84629"},
+    {file = "pyasn1-0.6.1.tar.gz", hash = "sha256:6f580d2bdd84365380830acf45550f2511469f673cb4a5ae3857a3170128b034"},
+]
+
+[[package]]
+name = "rsa"
+version = "4.9"
+description = "Pure-Python RSA implementation"
+optional = false
+python-versions = ">=3.6,<4"
+files = [
+    {file = "rsa-4.9-py3-none-any.whl", hash = "sha256:90260d9058e514786967344d0ef75fa8727eed8a7d2e43ce9f4bcf1b536174f7"},
+    {file = "rsa-4.9.tar.gz", hash = "sha256:e38464a49c6c85d7f1351b0126661487a7e0a14a50f1675ec50eb34d4f20ef21"},
+]
+
+[package.dependencies]
+pyasn1 = ">=0.1.3"
+
+[[package]]
+name = "telethon"
+version = "1.37.0"
+description = "Full-featured Telegram client library for Python 3"
+optional = false
+python-versions = ">=3.5"
+files = [
+    {file = "Telethon-1.37.0.tar.gz", hash = "sha256:e5e43cff1c1b34e2f9c2b395215beb6e9bda706b69def7efff4f55b23c9c4374"},
+]
+
+[package.dependencies]
+pyaes = "*"
+rsa = "*"
+
+[package.extras]
+cryptg = ["cryptg"]
+
+[metadata]
+lock-version = "2.0"
+python-versions = "^3.12"
+content-hash = "cb1c7ee8dae3ec0d7c35c3ee27428c61eb7ff111270b12bf3ac0a66817e4b9ee"

pyproject.toml

@@ -7,6 +7,7 @@ readme = "README.md"
 
 [tool.poetry.dependencies]
 python = "^3.12"
+telethon = "^1.37.0"
 
 
 [build-system]