Spaces:
Running
Running
File size: 3,960 Bytes
b29bec8 3f92a54 505d4ba 3f92a54 505d4ba 3f92a54 505d4ba 3f92a54 b29bec8 3f92a54 b29bec8 3f92a54 |
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 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:
# 使用正则表达式去除 type=... 参数，包括#前的部分
# 这里我们匹配 &type= 后面跟着任意字符直到下一个 & 或 # 或字符串结束 ,去除type=tcp== 或者type=tcp= 这类影响链接的多余部分
_encoded_config_str_without_type = re.sub(r'([&?]type=[^&#]*)', '',_encoded_config_str)
if "allowInsecure" in config_str:
yield _encoded_config_str_without_type.replace('allowInsecure=0', 'allowInsecure=1')+"#"+nameinfo
else:
yield _encoded_config_str_without_type+"&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')
if ("ss://" in config_str) and ("套餐" not in nameinfo) and('到期' not in nameinfo) and('流量' not in nameinfo) and('剩余' not in nameinfo):
#shadowsocks节点保留
yield config_str
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
|