Upload 7 files

server/decoder.py

@@ -0,0 +1,105 @@
+#!/usr/bin/env python
+# -*- coding: utf-8 -*-
+import re
+import json
+from base64 import b64decode
+from typing import Dict, Iterator, List, Union
+from .resource import Resource
+from dataclasses import dataclass
+from urllib import parse
+
+
+class BaseDecoder:
+    def __init__(self, encode_str: str, nameinfo: str = ""):
+        self.encode_str = encode_str
+        self.decode_str: str
+        self.nameinfo = nameinfo
+
+    @staticmethod
+    def _bytesto_str(iobytes: bytes) -> str:
+        return iobytes.decode('utf-8')
+
+
+    def decode(self) -> None:
+        self.decode_str = self._bytesto_str(b64decode(self.encode_str))
+
+
+@dataclass
+class EncodedCfg:
+    encoded_config_str: str
+    nameinfo: str = ""
+
+
+class ListDecoder(BaseDecoder):
+    def iter_encode_config(self) -> Iterator[EncodedCfg]:
+        for config_str in self.decode_str.splitlines():
+            _config_str = re.sub(r"(vmess|ss)://", "", config_str)
+            nameinfo = ""
+            if "#" in _config_str:
+                _config_str, nameinfo = _config_str.split("#", 1)
+                nameinfo = parse.unquote(nameinfo)
+
+            _encoded_config_str = _config_str + (
+                (4 - len(_config_str) % 4) * "="
+            )
+
+            if "trojan" in config_str and "0.0.0.0" not in config_str and '美国' not in nameinfo and '443' not in config_str:
+                if "allowInsecure" in config_str:
+                     # yield _encoded_config_str.replace('allowInsecure=0', 'allowInsecure=1').replace('type=tcp==', '')+"#"+"随机节点"
+                    yield _encoded_config_str.replace('allowInsecure=0', 'allowInsecure=1').replace('type=tcp==', '')+"#"+nameinfo
+                else:
+                    # yield _encoded_config_str.replace('type=tcp==', '')+"&allowInsecure=1"+"#"+"随机节点"
+                    yield _encoded_config_str.replace('type=tcp==', '')+"&allowInsecure=1"+"#"+nameinfo
+
+            if ("vmess" in config_str) :
+                vmess_decoded_data = b64decode(_config_str).decode('utf-8')
+                vmess_info_json = json.loads(vmess_decoded_data)
+                vmess_info_str = json.dumps(vmess_info_json,ensure_ascii=False)#不转译中文
+                if ('倍率提示'not in vmess_info_str)and('导航' not in vmess_info_str) and('443' not in vmess_info_str):
+                    yield config_str.replace('allowInsecure=0', 'allowInsecure=1')
+
+
+
+class ConfigDecoder(BaseDecoder):
+    def get_json(self) -> Dict:
+        if re.match(r"^{.*}$", self.decode_str):
+            return json.loads(self.decode_str)
+        return self._parse_ss()
+
+    def _parse_ss(self):
+        crypt, pw_at_addr, port = self.decode_str.split(":")
+        pw, addr = pw_at_addr.split("@")
+        return {
+            "ps": self.nameinfo,
+            "add": addr,
+            "port": int(port),
+            "method": crypt,
+            "password": pw,
+            "ss": True,
+            "level": 0,
+        }
+
+
+def _get_resource_from_url(url: str) -> str:
+    resource = Resource(url)
+    return resource.get_encoded_data()
+
+
+def _get_listencoded_cfg_from_encoded_str(encoded_str: str) -> List[EncodedCfg]:
+    decoder = ListDecoder(encoded_str)
+    decoder.decode()
+    return [item for item in decoder.iter_encode_config()]
+
+
+
+
+
+def decode_url_to_configs(url: str)->list:
+    encoded_str = _get_resource_from_url(url)
+    ##https://github.com/CareyWang/sub-web 成功用这个订阅解析解决了订阅被cf格挡问题，且可用于sspanel
+    lst_encoded_cfg = _get_listencoded_cfg_from_encoded_str(encoded_str)#lst_encoded_cfg解析为节点后的列表,包含所有节点链接
+    return lst_encoded_cfg
+
+
+
+

server/resource.py

@@ -0,0 +1,18 @@
+#!/usr/bin/env python
+# -*- coding: utf-8 -*-
+import requests
+from .utils import headers
+
+
+class Resource:
+    def __init__(self, url: str):
+        self.url = url
+        self.cutted_str: str
+        self._get_resource()
+
+    def _get_resource(self) -> None:
+        request = requests.get(self.url,headers=headers)
+        self.cutted_str = request.text
+
+    def get_encoded_data(self) -> str:
+        return self.cutted_str + ((4 - len(self.cutted_str) % 4) * "=")

server/utils.py

@@ -0,0 +1,25 @@
+import json
+import os
+import hashlib
+import secrets
+# 获取当前文件所在的目录
+current_dir = os.path.dirname(os.path.realpath(__file__))
+# 读取配置文件
+with open(os.path.join(current_dir, 'uu.json'), 'r') as f:
+    config = json.load(f)
+
+headers = {
+    'accept': 'text/html,application/xhtml+xml,application/xml;q=0.9,image/avif,image/webp,image/apng,*/*;q=0.8,application/signed-exchange;v=b3;q=0.7',
+    'accept-language': 'zh-CN,zh;q=0.9',
+    'cache-control': 'max-age=0',
+    'priority': 'u=0, i',
+    'sec-ch-ua': '"Chromium";v="128", "Not;A=Brand";v="24", "Google Chrome";v="128"',
+    'sec-ch-ua-mobile': '?0',
+    'sec-ch-ua-platform': '"macOS"',
+    'sec-fetch-dest': 'document',
+    'sec-fetch-mode': 'navigate',
+    'sec-fetch-site': 'none',
+    'sec-fetch-user': '?1',
+    'upgrade-insecure-requests': '1',
+    'user-agent': 'Mozilla/5.0 (Macintosh; Intel Mac OS X 10_15_7) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/128.0.0.0 Safari/537.36'
+}

server/uu.json

@@ -0,0 +1,6 @@
+{
+    "baseUrl":"https://rogerxavier-v2b-proxy.hf.space",
+    "purchaseLinkBase":"https://afdian.com/order/create?product_type=1",
+    "planID": "ef3c43c6daa411ee965e5254001e7c00",
+    "skuID": "ef440afcdaa411ee9e165254001e7c00"
+}