with json converter

basethon/__init__.py

@@ -1,3 +1,5 @@
+from .base_session import BaseSession
 from .base_thon import BaseThon
+from .json_converter import JsonConverter
 
-__all__ = ["BaseThon"]
+__all__ = ["BaseThon", "BaseSession", "JsonConverter"]

basethon/base_client.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,
+        )

basethon/base_session.py

@@ -0,0 +1,26 @@
+from pathlib import Path
+from typing import Generator
+
+from jsoner import json_read_sync
+
+
+class BaseSession:
+    def __init__(self, base_dir: Path, errors_dir: Path, banned_dir: Path):
+        self.base_dir = base_dir
+        self.base_dir.mkdir(exist_ok=True)
+        self.errors_dir = errors_dir
+        self.errors_dir.mkdir(exist_ok=True)
+        self.banned_dir = banned_dir
+        self.banned_dir.mkdir(exist_ok=True)
+        self.json_errors: set[Path] = set()
+
+    def iter_sessions(self) -> Generator:
+        for item in self.base_dir.glob("*.session"):
+            json_file = item.with_suffix(".json")
+            if not json_file.is_file():
+                self.json_errors.add(json_file)
+                continue
+            if not (json_data := json_read_sync(json_file)):
+                self.json_errors.add(json_file)
+                continue
+            yield item, json_file, json_data

basethon/json_converter.py

@@ -0,0 +1,56 @@
+import asyncio
+from pathlib import Path
+from typing import Generator
+
+from jsoner import json_write_sync
+from telethon import TelegramClient
+from telethon.sessions import StringSession
+from tooler import ProxyParser
+
+from .base_session import BaseSession
+
+
+class JsonConverter(BaseSession):
+    def __init__(
+        self,
+        base_dir: Path,
+        errors_dir: Path,
+        banned_dir: Path,
+        proxy: str,
+        json_write: bool = True,
+    ):
+        super().__init__(base_dir, errors_dir, banned_dir)
+        self.__api_id, self.__api_hash = 2040, "b18441a1ff607e10a989891a5462e627"
+        self.__proxy = ProxyParser(proxy).asdict_thon
+        self.__json_write = json_write
+
+    def _main(self, item: Path, json_file: Path, json_data: dict) -> dict:
+        loop = asyncio.new_event_loop()
+        asyncio.set_event_loop(loop)
+        client = TelegramClient(str(item), self.__api_id, self.__api_hash)
+        ss = StringSession()
+        ss._server_address = client.session.server_address  # type: ignore
+        ss._takeout_id = client.session.takeout_id  # type: ignore
+        ss._auth_key = client.session.auth_key  # type: ignore
+        ss._dc_id = client.session.dc_id  # type: ignore
+        ss._port = client.session.port  # type: ignore
+        string_session = ss.save()
+        del ss, client
+        json_data["proxy"] = self.__proxy
+        json_data["string_session"] = string_session
+        if self.__json_write:
+            json_write_sync(json_file, json_data)
+        return json_data
+
+    def iter(self) -> Generator:
+        for item, json_file, json_data in self.iter_sessions():
+            _json_data = self._main(item, json_file, json_data)
+            yield item, json_file, _json_data
+
+    def main(self) -> int:
+        count = 0
+        self.__json_write = True
+        for item, json_file, json_data in self.iter_sessions():
+            self._main(item, json_file, json_data)
+            count += 1
+        return count

poetry.lock

@@ -1,5 +1,35 @@
 # This file is automatically @generated by Poetry 1.8.3 and should not be changed by hand.
 
+[[package]]
+name = "aiofiles"
+version = "23.2.1"
+description = "File support for asyncio."
+optional = false
+python-versions = ">=3.7"
+files = [
+    {file = "aiofiles-23.2.1-py3-none-any.whl", hash = "sha256:19297512c647d4b27a2cf7c34caa7e405c0d60b5560618a29a9fe027b18b0107"},
+    {file = "aiofiles-23.2.1.tar.gz", hash = "sha256:84ec2218d8419404abcb9f0c02df3f34c6e0a68ed41072acfb1cef5cbc29051a"},
+]
+
+[[package]]
+name = "jsoner"
+version = "0.1.0"
+description = ""
+optional = false
+python-versions = "^3.11"
+files = []
+develop = false
+
+[package.dependencies]
+aiofiles = "^23.2.1"
+tooler = {git = "https://github.com/trojvn/tooler.git"}
+
+[package.source]
+type = "git"
+url = "https://github.com/trojvn/jsoner.git"
+reference = "HEAD"
+resolved_reference = "1d714ef57aaac46caf261ba493b38ab00ce7e365"
+
 [[package]]
 name = "pyaes"
 version = "1.6.1"
@@ -52,7 +82,22 @@ rsa = "*"
 [package.extras]
 cryptg = ["cryptg"]
 
+[[package]]
+name = "tooler"
+version = "0.5.0"
+description = ""
+optional = false
+python-versions = "^3.11"
+files = []
+develop = false
+
+[package.source]
+type = "git"
+url = "https://github.com/trojvn/tooler.git"
+reference = "HEAD"
+resolved_reference = "ff04688e6131ec612ea871fe091f76cedf5098ca"
+
 [metadata]
 lock-version = "2.0"
 python-versions = "^3.12"
-content-hash = "cb1c7ee8dae3ec0d7c35c3ee27428c61eb7ff111270b12bf3ac0a66817e4b9ee"
+content-hash = "d48b78dd5bc745693c770b4a2f96a8be461e21385bd8f84ae71f9906672cf433"

pyproject.toml

@@ -8,6 +8,7 @@ readme = "README.md"
 [tool.poetry.dependencies]
 python = "^3.12"
 telethon = "^1.37.0"
+jsoner = {git = "https://github.com/trojvn/jsoner.git"}
 
 
 [build-system]