v0.3.1 (#13)

* update: best-practice of `Clash.Meta` configuration

* feat(add): `Clash.Meta` scaffold

* feat(add): random port develop

.gitignore

@@ -158,4 +158,5 @@ cython_debug/
 # and can be added to the global gitignore or merged into this file. For a more nuclear
 # option (not recommended) you can uncomment the following to ignore the entire idea folder.
 .idea/
-tests
+scaffold/logs
+scaffold/database

requirements_dev.txt

@@ -0,0 +1,2 @@
+requests>=2.31.0
+loguru>=0.7.0

scaffold/__init__.py

@@ -0,0 +1,5 @@
+# -*- coding: utf-8 -*-
+# Time : 2023/7/7 3:10
+# Author : QIN2DIM
+# Github : https://github.com/QIN2DIM
+# Description:

scaffold/clash_meta_apis.py

@@ -0,0 +1,74 @@
+# -*- coding: utf-8 -*-
+# Time : 2023/7/7 3:10
+# Author : QIN2DIM
+# Github : https://github.com/QIN2DIM
+# Description:
+from __future__ import annotations
+
+from dataclasses import dataclass
+from pathlib import Path
+from typing import Dict, Literal, Any
+
+import requests
+from loguru import logger
+from requests.sessions import Session
+
+
+@dataclass
+class ClashMetaAPI:
+ secret: str
+ controller_url: str
+ db: Path
+
+ _session: Session = None
+
+ @classmethod
+ def from_secret(
+ cls, secret: str, controller_url: str | None = None, path_db: Path | None = None
+ ) -> ClashMetaAPI | None:
+ _url = controller_url or "http://127.0.0.1:9090"
+ db = path_db or Path("database")
+ if not secret:
+ logger.warning("请设置你的外部控制密钥", documentation="https://wiki.metacubex.one/api/#_1")
+ instance = cls(secret=secret, controller_url=_url, db=db)
+ instance._session = requests.session()
+ return instance
+
+ @property
+ def headers(self):
+ return {"Authorization": f"Bearer {self.secret}"}
+
+ def get_version(self) -> Dict[str, Any]:
+ api = f"{self.controller_url}/version"
+ return requests.get(api, headers=self.headers).json()
+
+ def get_dns_query(
+ self, name: str, dns_type: Literal["A", "CNAME", "MX"] | None = ""
+ ) -> Dict[str, Any]:
+ api = f"{self.controller_url}/dns/query"
+ params = {"name": name}
+ if dns_type:
+ params["type"] = dns_type
+ return self._session.get(api, headers=self.headers, params=params).json()
+
+ def get_proxies(self):
+ api = f"{self.controller_url}/proxies"
+ return self._session.get(api, headers=self.headers).json()
+
+ def get_traffic(self):
+ api = f"{self.controller_url}/traffic"
+ return self._session.get(api, headers=self.headers).json()
+
+ def flush_fakeip_cache(self):
+ api = f"{self.controller_url}/cache/fakeip/flush"
+ self._session.post(api, headers=self.headers)
+
+ def get_connections(self):
+ api = f"{self.controller_url}/connections"
+ return self._session.get(api, headers=self.headers).json()
+
+ def delete_connection(self, conn_id: str | None = ""):
+ api = f"{self.controller_url}/connections"
+ if conn_id:
+ api = f"{self.controller_url}/connections/:{conn_id}"
+ self._session.delete(api, headers=self.headers)

scaffold/clash_meta_config.yaml

@@ -0,0 +1,91 @@
+dns:
+ enable: true
+ prefer-h3: true
+ listen: 0.0.0.0:53
+ enhanced-mode: fake-ip
+ fake-ip-range: 198.18.0.1/16
+ default-nameserver:
+ - https://223.5.5.5/dns-query
+ nameserver:
+ - "quic://dns.adguard.com"
+ - "https://dns.cloudflare.com/dns-query#h3=true"
+rule-providers:
+ direct:
+ type: http
+ behavior: domain
+ url: "https://cdn.jsdelivr.net/gh/Loyalsoldier/clash-rules@release/direct.txt"
+ path: ./ruleset/direct.yaml
+ interval: 86400
+ proxy:
+ type: http
+ behavior: domain
+ url: "https://cdn.jsdelivr.net/gh/Loyalsoldier/clash-rules@release/proxy.txt"
+ path: ./ruleset/proxy.yaml
+ interval: 86400
+ reject:
+ type: http
+ behavior: domain
+ url: "https://cdn.jsdelivr.net/gh/Loyalsoldier/clash-rules@release/reject.txt"
+ path: ./ruleset/reject.yaml
+ interval: 86400
+ private:
+ type: http
+ behavior: domain
+ url: "https://cdn.jsdelivr.net/gh/Loyalsoldier/clash-rules@release/private.txt"
+ path: ./ruleset/private.yaml
+ interval: 86400
+ apple:
+ type: http
+ behavior: domain
+ url: "https://cdn.jsdelivr.net/gh/Loyalsoldier/clash-rules@release/apple.txt"
+ path: ./ruleset/apple.yaml
+ interval: 86400
+ icloud:
+ type: http
+ behavior: domain
+ url: "https://cdn.jsdelivr.net/gh/Loyalsoldier/clash-rules@release/icloud.txt"
+ path: ./ruleset/icloud.yaml
+ interval: 86400
+ telegramcidr:
+ type: http
+ behavior: ipcidr
+ url: "https://cdn.jsdelivr.net/gh/Loyalsoldier/clash-rules@release/telegramcidr.txt"
+ path: ./ruleset/telegramcidr.yaml
+ interval: 86400
+ lancidr:
+ type: http
+ behavior: ipcidr
+ url: "https://cdn.jsdelivr.net/gh/Loyalsoldier/clash-rules@release/lancidr.txt"
+ path: ./ruleset/lancidr.yaml
+ interval: 86400
+ cncidr:
+ type: http
+ behavior: ipcidr
+ url: "https://cdn.jsdelivr.net/gh/Loyalsoldier/clash-rules@release/cncidr.txt"
+ path: ./ruleset/cncidr.yaml
+ interval: 86400
+ applications:
+ type: http
+ behavior: classical
+ url: "https://cdn.jsdelivr.net/gh/Loyalsoldier/clash-rules@release/applications.txt"
+ path: ./ruleset/applications.yaml
+ interval: 86400
+rules:
+ - RULE-SET,applications,DIRECT
+ - GEOSITE,cn,DIRECT
+ - GEOSITE,private,DIRECT
+ - DOMAIN,clash.razord.top,DIRECT
+ - DOMAIN,yacd.haishan.me,DIRECT
+ - DOMAIN,services.googleapis.cn,PROXY
+ - RULE-SET,reject,REJECT
+ - RULE-SET,direct,DIRECT
+ - RULE-SET,private,DIRECT
+ - RULE-SET,proxy,PROXY
+ - RULE-SET,icloud,DIRECT
+ - RULE-SET,apple,DIRECT
+ - RULE-SET,lancidr,DIRECT
+ - RULE-SET,cncidr,DIRECT
+ - RULE-SET,telegramcidr,PROXY,no-resolve
+ - GEOIP,PRIVATE,DIRECT
+ - GEOIP,CN,DIRECT
+ - MATCH,PROXY

scaffold/toolbox.py

@@ -0,0 +1,75 @@
+# -*- coding: utf-8 -*-
+# Time : 2023/7/7 3:24
+# Author : QIN2DIM
+# Github : https://github.com/QIN2DIM
+# Description:
+from __future__ import annotations
+
+import os
+import sys
+from dataclasses import dataclass
+from os.path import dirname
+from pathlib import Path
+from typing import Literal
+
+from loguru import logger
+
+__all__ = ["project", "init_log"]
+
+
+@dataclass
+class Project:
+ src_point = Path(dirname(__file__))
+ root_point = src_point
+ database = root_point.joinpath("database")
+
+ logs = root_point.joinpath("logs")
+
+ def __post_init__(self):
+ os.makedirs(self.database, exist_ok=True)
+
+
+def init_log(*, stdout_level: Literal["INFO", "DEBUG"] = "DEBUG", **sink_channel):
+ event_logger_format = "<g>{time:YYYY-MM-DD HH:mm:ss}</g> | <lvl>{level}</lvl> - {message}"
+ serialize_format = event_logger_format + "- {extra}"
+ logger.remove()
+ logger.add(
+ sink=sys.stdout, colorize=True, level=stdout_level, format=serialize_format, diagnose=False
+ )
+ if sink_channel.get("error"):
+ logger.add(
+ sink=sink_channel.get("error"),
+ level="ERROR",
+ rotation="1 week",
+ encoding="utf8",
+ diagnose=False,
+ format=serialize_format,
+ )
+ if sink_channel.get("runtime"):
+ logger.add(
+ sink=sink_channel.get("runtime"),
+ level="DEBUG",
+ rotation="20 MB",
+ retention="20 days",
+ encoding="utf8",
+ diagnose=False,
+ format=serialize_format,
+ )
+ if sink_channel.get("serialize"):
+ logger.add(
+ sink=sink_channel.get("serialize"),
+ level="DEBUG",
+ format=serialize_format,
+ encoding="utf8",
+ diagnose=False,
+ serialize=True,
+ )
+ return logger
+
+
+project = Project()
+init_log(
+ error=project.logs.joinpath("error.log"),
+ runtime=project.logs.joinpath("runtime.log"),
+ serialize=project.logs.joinpath("serialize.log"),
+)

tests/test_clash_meta_apis.py

@@ -0,0 +1,61 @@
+# -*- coding: utf-8 -*-
+# Time : 2023/7/7 3:42
+# Author : QIN2DIM
+# Github : https://github.com/QIN2DIM
+# Description:
+from __future__ import annotations
+
+import json
+import socket
+from urllib.parse import urlparse
+
+from loguru import logger
+
+from scaffold.clash_meta_apis import ClashMetaAPI
+from scaffold.toolbox import project, init_log
+
+# 快速生成密钥，添加到 Clash-Verge 外部控制组件
+# python -c "import secrets;print(secrets.token_hex()[:16])"
+CONTROLLER_SECRET = ""
+CONTROLLER_URL = "http://127.0.0.1:9090"
+
+init_log(
+ stdout_level="INFO",
+ error=project.logs.joinpath("error.log"),
+ runtime=project.logs.joinpath("runtime.log"),
+ serialize=project.logs.joinpath("serialize.log"),
+)
+
+clash = ClashMetaAPI.from_secret(CONTROLLER_SECRET, CONTROLLER_URL, project.database)
+
+
+def test_is_live():
+ u = urlparse(CONTROLLER_URL)
+ try:
+ with socket.socket(socket.AF_INET, socket.SOCK_STREAM) as s:
+ host, port = u.netloc.split(":")
+ s.bind((host, int(port)))
+ raise OSError
+ except socket.gaierror:
+ logger.debug("外部控制接口已关闭", netloc=u.netloc, port=u.port)
+ return False
+ except OSError:
+ logger.debug("外部控制接口可达", netloc=u.netloc)
+ return True
+
+
+def test_get_dns_query():
+ clash.flush_fakeip_cache()
+ names = ["www.bilibili.com", "www.baidu.com", "www.google.com", "www.youtube.com"]
+ for name in names:
+ r = clash.get_dns_query(name=name, dns_type="A")
+ fp = clash.db.joinpath(f"dns_query_{name}.json")
+ fp.write_text(json.dumps(r, indent=4))
+ logger.debug("TEST - DNS Query", name=name, result=r)
+
+
+def test_get_connections():
+ conns = clash.get_connections()
+ fp = clash.db.joinpath("connections.json")
+ fp.write_text(json.dumps(conns, indent=4))
+ logger.debug("TEST - Get Connections", conns=conns)