vless2subcon/main.py at main · deadlyedge/vless2subcon · GitHub

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
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
import yaml
import requests
from fastapi import FastAPI
from urllib.parse import urlparse, parse_qs
from fastapi.responses import FileResponse


app = FastAPI()


def read_config(path="config.yaml") -> dict:
    with open(path, "r", encoding="UTF-8") as f:
        return yaml.load(f, Loader=yaml.Loader)


def write_yaml(data: dict, path="output.yaml"):
    with open(path, "w", encoding="UTF-8") as f:
        yaml.dump(data, f, allow_unicode=True)


def trans_url_to_info(url) -> dict:
    server_info = urlparse(url)
    # print(server_info)

    # 展开query parameters并合并到一个字典中
    params = {key: value[0] for key, value in parse_qs(server_info.query).items()}
    params["path"] = "/" if "path" not in params.keys() else params["path"]

    if server_info.scheme == "hysteria2":
        return {
            "name": server_info.fragment,
            "type": "hysteria2",
            "server": server_info.hostname,
            "port": server_info.port,
            "obfs": "salamander",
            "obfs-password": params["obfs-password"],
            "sni": params["sni"],
            "skip-cert-verify": bool(params["insecure"]),
            "password": params["auth"],
            "alpn": ["h3"],
        }
    elif server_info.scheme == "vless":
        return {
            "name": server_info.fragment,
            "server": server_info.hostname,
            "port": server_info.port,
            "client-fingerprint": params["fp"],
            "type": server_info.scheme,
            "uuid": server_info.username,
            "tls": (params["security"] == "tls"),
            "tfo": False,
            "servername": params["host"],
            "skip-cert-verify": False,
            "network": params["type"],
            "ws-opts": {
                "path": params["path"],
                "headers": {"Host": params["host"]},
            },
        }
    else:
        raise KeyError("不支持的订阅转换协议: ", server_info.scheme)


@app.get("/{urls:path}")
async def parse_url_data(urls: str = ""):
    # 初始化
    config = read_config()
    try:
        sub_url = config["subscription_url"]
        vless_servers = config["vless_servers"] if config["vless_servers"] else []
    except (AttributeError, TypeError, KeyError):
        """
        检查配置文件 config.yaml
        subscription_url:
        vless_servers:
        """
        sub_url = "https://my.sub.domain"
        vless_servers = []
    servers = []

    # 如果传入了添加的vless server
    if urls:
        for url in urls.split("|"):
            vless_servers.append(url)

    # print(vless_servers)

    # 转换所有的自建vless
    for url in vless_servers:
        # print(url)
        try:
            servers.append(trans_url_to_info(url))
        except KeyError:
            return "订阅解滤器中不包含必要的参数"

    # 读取订阅
    try:
        output = yaml.load(requests.get(sub_url).text, Loader=yaml.Loader)
    except requests.exceptions.ConnectionError:
        config = {
            "subscription_url": "https://my.sub.url.address",
            # "subscription_url": sub_url,
            "vless_servers": vless_servers,
        }
        write_yaml(config, path="config.yaml")
        return "首先在设置文件【config.yaml】中填写订阅转换地址"

    # 插入自建vless
    for index, server in enumerate(servers):
        output["proxies"].insert(index, server)
    for group in output["proxy-groups"]:
        if not group["name"].endswith("节点"):
            for server in servers:
                group["proxies"].append(server["name"])

    # 保存设置
    config = {
        "subscription_url": sub_url,
        "vless_servers": vless_servers,
    }
    write_yaml(config, path="config.yaml")

    # 写入输出yaml文件
    write_yaml(output, path="output.yaml")

    return FileResponse("output.yaml")


if __name__ == "__main__":
    import uvicorn

    uvicorn.run(app, host="127.0.0.1", port=8000)