refactored

basethon/json_converter.py

@@ -6,9 +6,9 @@ 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
+from .proxy_parser import ProxyParser
 
 
 class JsonConverter(BaseSession):

basethon/proxy_parser.py

@@ -0,0 +1,113 @@
+from dataclasses import asdict, dataclass
+from typing import Literal
+
+
+@dataclass
+class ThonProxy:
+    proxy_type: Literal["http", "socks5"] | str
+    addr: str
+    port: int
+    username: str | None
+    password: str | None
+    rdns: bool = True
+
+
+class ProxyParser:
+    """Парсер проксей, если что-то не так, то райзит ValueError"""
+
+    def __init__(self, proxy: str, splitter: str = ":"):
+        proxy = proxy.replace("https:", "http:")
+        self.__proxy, self.__splitter = proxy, splitter
+        self.__prefixes = ("http", "socks5", "ss")
+        self.__url_prefixes = ("http://", "socks5://", "ss://")
+
+    @property
+    def splitted(self) -> list[str]:
+        _proxy = self.__proxy
+        for url_prefix in self.__url_prefixes:
+            if not self.__proxy.startswith(url_prefix):
+                continue
+            _url_prefix = url_prefix.replace("//", "")
+            _proxy = self.__proxy.replace(url_prefix, _url_prefix, 1)
+            if "@" in _proxy:
+                _splitted = _proxy.split("@", maxsplit=1)
+                _creds = _splitted[0].replace(_url_prefix, "", 1)
+                return f"{_url_prefix}{_splitted[-1]}:{_creds}".split(":")
+        return _proxy.split(self.__splitter)
+
+    @property
+    def type(self) -> Literal["http", "socks5", "ss"]:
+        p_type = self.splitted[0]
+        match p_type:
+            case "http":
+                return "http"
+            case "socks5":
+                return "socks5"
+            case "ss":
+                return "ss"
+        return "http"
+
+    @property
+    def ip(self) -> str:
+        if self.__proxy.startswith(self.__prefixes):
+            try:
+                return self.splitted[1]
+            except IndexError as e:
+                _message = f"[ProxyParser] Не найден IP в прокси: {self.__proxy}"
+                raise ValueError(_message) from e
+        return self.splitted[0]
+
+    @property
+    def port(self) -> int:
+        _message = f"[ProxyParser] Некорректно задан порт в прокси: {self.__proxy}"
+        if self.__proxy.startswith(self.__prefixes):
+            try:
+                return int(self.splitted[2])
+            except (ValueError, TypeError, IndexError) as e:
+                raise ValueError(_message) from e
+        try:
+            return int(self.splitted[1])
+        except (ValueError, TypeError, IndexError) as e:
+            raise ValueError(_message) from e
+
+    @property
+    def user(self) -> str | None:
+        if self.splitted[0] in self.__prefixes:
+            # [type, ip, port, user, pswd]
+            if len(self.splitted) == 5:
+                return self.splitted[3]
+        # [ip, port, user, pswd]
+        if len(self.splitted) == 4:
+            return self.splitted[2]
+
+    @property
+    def pswd(self) -> str | None:
+        if self.splitted[0] in self.__prefixes:
+            # [type, ip, port, user, pswd]
+            if len(self.splitted) == 5:
+                return self.splitted[4]
+        # [ip, port, user, pswd]
+        if len(self.splitted) == 4:
+            return self.splitted[3]
+
+    @property
+    def url(self) -> str:
+        if self.type == "ss":
+            return ""
+        if not self.user or not self.pswd:
+            return f"{self.type}://{self.ip}:{self.port}"
+        return f"{self.type}://{self.user}:{self.pswd}@{self.ip}:{self.port}"
+
+    @property
+    def thon(self) -> ThonProxy:
+        return ThonProxy(self.type, self.ip, self.port, self.user, self.pswd)
+
+    @property
+    def asdict_thon(self) -> dict:
+        return asdict(self.thon)
+
+    @property
+    def check(self) -> str:
+        if not self.user or not self.pswd:
+            return f"{self.type}:{self.ip}:{self.port}"
+        return f"{self.type}:{self.ip}:{self.port}:{self.user}:{self.pswd}"

poetry.lock

@@ -16,19 +16,18 @@ name = "jsoner"
 version = "0.1.0"
 description = ""
 optional = false
-python-versions = "^3.11"
+python-versions = "^3.10"
 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"
+resolved_reference = "b3a43e5e14609a7939f87e521654b6ae266e2723"
 
 [[package]]
 name = "pyaes"
@@ -53,13 +52,13 @@ files = [
 
 [[package]]
 name = "rsa"
-version = "4.9"
+version = "4.9.1"
 description = "Pure-Python RSA implementation"
 optional = false
-python-versions = ">=3.6,<4"
+python-versions = "<4,>=3.6"
 files = [
-    {file = "rsa-4.9-py3-none-any.whl", hash = "sha256:90260d9058e514786967344d0ef75fa8727eed8a7d2e43ce9f4bcf1b536174f7"},
+    {file = "rsa-4.9.1-py3-none-any.whl", hash = "sha256:68635866661c6836b8d39430f97a996acbd61bfa49406748ea243539fe239762"},
-    {file = "rsa-4.9.tar.gz", hash = "sha256:e38464a49c6c85d7f1351b0126661487a7e0a14a50f1675ec50eb34d4f20ef21"},
+    {file = "rsa-4.9.1.tar.gz", hash = "sha256:e7bdbfdb5497da4c07dfd35530e1a902659db6ff241e39d9953cad06ebd0ae75"},
 ]
 
 [package.dependencies]
@@ -83,21 +82,6 @@ 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"