bilibili-api-backup/server.py at main · Utility-Software-Collection/bilibili-api-backup · 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
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
import enum
import inspect
import socket
import sys

import httpx
from sanic import Sanic, json
from bilibili_api import *

app = Sanic("SimpleNetworkAccess")


def enum_to_value(obj):
    if isinstance(obj, enum.Enum):
        return obj.value
    elif isinstance(obj, dict):
        return {key: enum_to_value(value) for key, value in obj.items()}
    elif isinstance(obj, list):
        return [enum_to_value(item) for item in obj]
    elif hasattr(obj, '__dict__'):
        return {key: enum_to_value(value) for key, value in vars(obj).items()}
    else:
        return obj


def parse_param(param_map):
    for key, value in param_map.items():
        if key == "credential":
            continue
        if ":parse" in value:
            process = value.replace(":parse", "")
            param_map[key] = eval(process)


@app.get("/<package:str>/<clazz:str>/<func:str>")
async def req_by_clazz(request, package, clazz, func):
    try:
        package_attr = getattr(sys.modules[__name__], package)
        clazz_attr = getattr(package_attr, clazz)
        clazz_params = inspect.signature(clazz_attr).parameters
        cred_obj = None
        cred_dict = dict()
        clazz_dict = dict()
        for i in request.args:
            if i == "sessdata" or i == "bili_jct" or i == "buvid3" or i == "ac_time_value":
                cred_dict[i] = request.args[i][0]
                # continue
            if i in clazz_params.keys():
                clazz_dict[i] = request.args[i][0]

        parse_param(clazz_dict)

        if cred_dict:
            cred_obj = Credential(**cred_dict)

        if "credential" in clazz_params.keys():
            clazz_dict["credential"] = cred_obj

        func_dict = dict()
        clazz_obj = clazz_attr(**clazz_dict)
        func_attr = getattr(clazz_obj, func)
        func_params = inspect.signature(func_attr).parameters

        if "credential" in func_params.keys():
            func_dict["credential"] = cred_obj

        for i in request.args:
            if i in func_params.keys():
                func_dict[i] = request.args[i][0]

        parse_param(func_dict)

        if inspect.isasyncgenfunction(func_attr) or inspect.iscoroutinefunction(func_attr):
            result = await func_attr(**func_dict)
        else:
            result = func_attr(**func_dict)
        result_no_enum = enum_to_value(result)

    except Exception as e:
        wrap_result = dict()
        wrap_result["code"] = -1
        wrap_result["message"] = str(e)
        return json(wrap_result)

    wrap_result = dict()
    wrap_result["code"] = 0
    wrap_result["data"] = result_no_enum
    return json(wrap_result)


@app.get("/<package:str>/<func:str>")
async def req_by_static(request, package, func):
    try:
        package_attr = getattr(sys.modules[__name__], package)
        func_attr = getattr(package_attr, func)
        func_params = inspect.signature(func_attr).parameters
        cred_obj = None
        cred_dict = dict()
        func_dict = dict()

        for i in request.args:
            if i == "sessdata" or i == "bili_jct" or i == "buvid3" or i == "ac_time_value":
                cred_dict[i] = request.args[i][0]
                # continue
            if i in func_params:
                func_dict[i] = request.args[i][0]

        if cred_dict:
            cred_obj = Credential(**cred_dict)

        if package == "Credential":
            func_attr = getattr(cred_obj, func)

        if "credential" in func_params.keys():
            func_dict["credential"] = cred_obj

        parse_param(func_dict)

        if inspect.isasyncgenfunction(func_attr) or inspect.iscoroutinefunction(func_attr):
            result = await func_attr(**func_dict)
        else:
            result = func_attr(**func_dict)
        result_no_enum = enum_to_value(result)

    except Exception as e:
        wrap_result = dict()
        wrap_result["code"] = -1
        wrap_result["message"] = str(e)
        return json(wrap_result)

    wrap_result = dict()
    wrap_result["code"] = 0
    wrap_result["data"] = result_no_enum
    return json(wrap_result)


if __name__ == '__main__':
    port = int(sys.argv[1]) if len(sys.argv) > 1 else 9000
    # 手动检测端口占用
    sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
    try:
        sock.bind(("0.0.0.0", port))
        sock.close()
    except OSError:
        print(f"端口 {port} 已被占用，请更换端口或关闭占用该端口的程序。")
        sys.exit(1)

    app.run(host="0.0.0.0", port=port, dev=False)