v0.1.0

.gitignore

@@ -160,4 +160,5 @@ cython_debug/
 .idea/
 tests/
 clashbyte/database
-clashbyte/logs
+clashbyte/logs
+examples/dns_answers

clashbyte/__init__.py

@@ -4,7 +4,7 @@
 # Github     : https://github.com/QIN2DIM
 # Description:
 from .apis import ClashMetaAPI
-from .toolbox import init_log, project
+from .utils import init_log, project
 
 __all__ = ["ClashMetaAPI", "init_log", "project"]
-__version__ = "0.0.5"
+__version__ = "0.1.0"

clashbyte/apis.py

@@ -21,17 +21,13 @@ class Clash:
 
 
 @dataclass
-class ClashMeta:
+class ClashMetaAPI:
     """https://wiki.metacubex.one/api/"""
 
-
-@dataclass
-class ClashMetaAPI:
     secret: str = ""
     controller_url: str = ""
 
     _client = None
-    _core = None
 
     def __post_init__(self):
         if not self.controller_url:
@@ -48,7 +44,6 @@ def __post_init__(self):
             )
             del headers["Authorization"]
         self._client: httpx.Client = httpx.Client(base_url=self.controller_url, headers=headers)
-        self._core: ClashMeta = ClashMeta()
 
     @property
     def is_alive(self):

clashbyte/component/__init__.py

@@ -0,0 +1,9 @@
+# -*- coding: utf-8 -*-
+# Time       : 2023/7/18 16:48
+# Author     : QIN2DIM
+# Github     : https://github.com/QIN2DIM
+# Description:
+from .convertor import Toolkit
+
+kit = Toolkit
+__all__ = ["kit"]

clashbyte/component/convertor.py

@@ -0,0 +1,64 @@
+# -*- coding: utf-8 -*-
+# Time       : 2022/5/22 10:43
+# Author     : QIN2DIM
+# Github     : https://github.com/QIN2DIM
+# Description:
+from __future__ import annotations
+
+import os
+import shutil
+from dataclasses import dataclass
+from pathlib import Path
+from typing import List
+from urllib.parse import urlparse
+
+import yaml
+
+from clashbyte.scheme import Hysteria
+from clashbyte.scheme import Scheme
+from clashbyte.scheme import Tuic
+from scheme import path_clash_config
+
+
+@dataclass
+class Toolkit:
+    @staticmethod
+    def from_link_to_scheme(link: str) -> Scheme | None:
+        parser = urlparse(link)
+        if not parser:
+            return
+        if parser.scheme == "hysteria":
+            return Hysteria.from_urlparser(parser)
+        if parser.scheme == "tuic":
+            return Tuic.from_urlparser(parser)
+
+    @staticmethod
+    def from_clash_to_links():
+        pass
+
+    @staticmethod
+    def gen_clash_config_from_links(links: str | List[str], sp: Path | str | None = None):
+        if isinstance(links, str):
+            links = links.split("\n")
+        proxies = [Toolkit.from_link_to_scheme(link) for link in links]
+        proxies = [p.to_clash_node() for p in proxies if p]
+        proxy_groups = [
+            {"name": "PROXY", "type": "select", "proxies": [p["name"] for p in proxies]}
+        ]
+
+        if sp is None:
+            sp = Path("clash_configs")
+        elif isinstance(sp, str):
+            sp = Path(sp)
+        os.makedirs(sp, exist_ok=True)
+
+        t = yaml.safe_load(path_clash_config.read_text(encoding="utf8"))
+        t.update({"proxies": proxies, "proxy-groups": proxy_groups})
+        fn = "runtime.yaml"
+        fp = sp.joinpath(fn)
+        if fp.exists():
+            shutil.move(fp, sp.joinpath(f"{int(fp.stat().st_mtime)}.yaml"))
+        Path(fn).write_text(yaml.safe_dump(t, allow_unicode=True, sort_keys=False))
+        shutil.copy(fn, fp)
+
+        return t

clashbyte/panel.py

@@ -0,0 +1,60 @@
+# -*- coding: utf-8 -*-
+# Time       : 2023/7/18 16:43
+# Author     : QIN2DIM
+# Github     : https://github.com/QIN2DIM
+# Description:
+from __future__ import annotations
+
+from dataclasses import dataclass
+
+import pywebio
+from pywebio.input import *
+from pywebio.output import *
+
+pywebio.config(theme="minty")
+
+
+@dataclass
+class Panel:
+    TITLE = "云彩姬@订阅转换"
+
+    def startup(self):
+        """
+        actions("\n", buttons=[
+                {"label": "生成订阅链接", "value": "生成订阅链接", "color": "secondary"},
+                {"label": "生成短链接", "value": "生成短链接", "color": "secondary"},
+            ], name="action_link"),
+        actions("\n", buttons=[
+                {"label": "上传配置", "value": "上传配置", "color": "info"},
+                {"label": "一键导入Clash", "value": "一键导入Clash", "color": "info"},
+            ], name="action_inner")
+        """
+        toast("Just a demo", position="right", duration=3, color="##78c2ad")
+
+        data = input_group(
+            label=self.TITLE,
+            inputs=[
+                textarea(
+                    label="订阅链接",
+                    placeholder="支持 hysteria 链接，多个链接每行一个或用 | 分隔",
+                    rows=4,
+                    name="sharelink",
+                ),
+                select(label="客户端", options=["NekoRay", "Clash Verge"], name="client"),
+                actions(
+                    label="label",
+                    buttons=[
+                        {"label": "上传配置", "value": "上传配置", "color": "info"},
+                        {"label": "一键导入 Clash", "value": "一键导入 Clash", "color": "info"},
+                    ],
+                    name="clash_config",
+                ),
+            ],
+        )
+
+        put_text(data)
+
+
+if __name__ == "__main__":
+    panel = Panel()
+    panel.startup()

clashbyte/scheme/__init__.py

@@ -0,0 +1,10 @@
+# -*- coding: utf-8 -*-
+# Time       : 2023/7/18 16:45
+# Author     : QIN2DIM
+# Github     : https://github.com/QIN2DIM
+# Description:
+from ._scheme import Scheme, path_clash_config
+from .hysteria import Hysteria
+from .tuic import Tuic
+
+__all__ = ["Hysteria", "Tuic", "Scheme", "path_clash_config"]

clashbyte/scheme/_default.yaml

@@ -0,0 +1,29 @@
+dns:
+  enable: true
+  enhanced-mode: fake-ip
+  nameserver:
+    - "https://dns.google/dns-query#PROXY"
+    - "https://security.cloudflare-dns.com/dns-query#PROXY"
+    - "quic://dns.adguard-dns.com"
+  proxy-server-nameserver:
+    - system
+    - https://223.5.5.5/dns-query
+  nameserver-policy:
+    geosite:cn:
+      - system
+      - https://223.5.5.5/dns-query
+rules:
+  - GEOSITE,category-scholar-!cn,PROXY
+  - GEOSITE,category-ads-all,REJECT
+  - GEOSITE,youtube,PROXY
+  - GEOSITE,google,PROXY
+  - GEOSITE,cn,DIRECT
+  - GEOSITE,private,DIRECT
+  - GEOSITE,steam@cn,DIRECT
+  - GEOSITE,category-games@cn,DIRECT
+  - GEOSITE,geolocation-!cn,PROXY
+  - GEOIP,private,DIRECT,no-resolve
+  - GEOIP,telegram,PROXY
+  - GEOIP,CN,DIRECT
+  - DST-PORT,80/8080/443/8443,PROXY
+  - MATCH,DIRECT

clashbyte/scheme/_scheme.py

@@ -0,0 +1,28 @@
+# -*- coding: utf-8 -*-
+# Time       : 2023/7/18 17:26
+# Author     : QIN2DIM
+# Github     : https://github.com/QIN2DIM
+# Description:
+from abc import ABC, abstractmethod
+from dataclasses import dataclass
+from pathlib import Path
+from typing import Dict, Any
+from urllib.parse import ParseResult
+
+path_clash_config = Path(__file__).parent.joinpath("_default.yaml")
+
+
+@dataclass
+class Scheme(ABC):
+    @classmethod
+    @abstractmethod
+    def from_urlparser(cls, parser: ParseResult):
+        ...
+
+    @abstractmethod
+    def to_sharelink(self) -> str:
+        ...
+
+    @abstractmethod
+    def to_clash_node(self, **kwargs) -> Dict[str, Any]:
+        ...

clashbyte/scheme/hysteria.py

@@ -0,0 +1,108 @@
+# -*- coding: utf-8 -*-
+# Time       : 2023/7/18 16:45
+# Author     : QIN2DIM
+# Github     : https://github.com/QIN2DIM
+# Description:
+from __future__ import annotations
+
+from dataclasses import dataclass
+from typing import Literal
+from urllib.parse import ParseResult
+
+from clashbyte.scheme._scheme import Scheme
+from clashbyte.utils import from_dict_to_cls
+
+
+@dataclass
+class Hysteria(Scheme):
+    # hostname or IP address of the server to connect to (required)
+    host: str
+
+    # port of the server to connect to (required)
+    port: int
+
+    # upstream bandwidth in Mbps (required)
+    upmbps: int
+
+    # downstream bandwidth in Mbps (required)
+    downmbps: int
+
+    # multiport skip (optional)
+    mport: str = None
+
+    # protocol to use ("udp", "wechat-video", "faketcp") (optional, default: "udp")
+    protocol: Literal["udp", "wechat-video", "faketcp"] = "udp"
+
+    # authentication payload (string) (optional)
+    auth: str = None
+
+    # SNI for TLS (optional)
+    peer: str = None
+
+    # insecure: ignore certificate errors (optional)
+    insecure: bool = False
+
+    # QUIC ALPN (optional)
+    alpn: str = "h3"
+
+    # Obfuscation mode (optional, empty or "xplus")
+    obfs: Literal["", "xplus"] = None
+
+    # Obfuscation password (optional)
+    obfsParam: str = None
+
+    # remarks alias (optional)
+    remarks: str = None
+
+    @classmethod
+    def from_urlparser(cls, parser: ParseResult):
+        """
+        从节点分享链接反序列化节点对象
+
+        https://hysteria.network/zh/docs/uri-scheme/
+
+        :param parser: Hysteria URL Scheme Parser
+        :return:
+        """
+        host, port = parser.netloc.split(":")
+        data = {"host": host, "port": int(port), "remarks": parser.fragment}
+        for e in parser.query.split("&"):
+            k, v = e.split("=")
+            data[k] = v
+        return from_dict_to_cls(cls, data)
+
+    def to_sharelink(self) -> str:
+        t = "hysteria://{netloc}?{query}#{fragment}"
+        netloc = f"{self.host}:{self.port}"
+        queries = []
+        for k in self.__dict__:
+            if not self.__dict__[k]:
+                continue
+            v = self.__dict__[k]
+            queries.append(f"{k}={v}")
+        query = "&".join(queries)
+        fragment = self.remarks
+        sharelink = t.format(netloc=netloc, query=query, fragment=fragment)
+        return sharelink
+
+    def to_clash_node(self, **kwargs):
+        self.remarks = self.remarks or self.host
+        node = {
+            "name": self.remarks,
+            "type": "hysteria",
+            "server": self.host,
+            "port": self.port,
+            "ports": self.mport,
+            "alpn": [self.alpn],
+            "protocol": self.protocol,
+            "up": self.upmbps,
+            "down": self.downmbps,
+            "sni": self.peer,
+            "skip-cert-verify": self.insecure,
+            "auth_str": self.auth,
+            "obfs": self.obfsParam,  # auto fill
+        }
+        if kwargs:
+            node.update(**kwargs)
+        node = {k: v for k, v in node.items() if v is not None}
+        return node

clashbyte/scheme/tuic.py

@@ -0,0 +1,25 @@
+# -*- coding: utf-8 -*-
+# Time       : 2023/7/18 16:47
+# Author     : QIN2DIM
+# Github     : https://github.com/QIN2DIM
+# Description:
+from __future__ import annotations
+
+from dataclasses import dataclass
+from typing import Dict, Any
+from urllib.parse import ParseResult
+
+from clashbyte.scheme._scheme import Scheme
+
+
+@dataclass
+class Tuic(Scheme):
+    @classmethod
+    def from_urlparser(cls, parser: ParseResult):
+        pass
+
+    def to_sharelink(self) -> str:
+        pass
+
+    def to_clash_node(self, **kwargs) -> Dict[str, Any]:
+        pass