feat: 增加独立的 Swagger 接口调试台
This commit is contained in:
Nixevol
@@ -0,0 +1,167 @@
|
||||
"""独立的 Swagger 调试台。
|
||||
|
||||
浏览器打开本机页面，请求由这里转发到 https://apis.raptdrama.com。
|
||||
不导入、不修改抓取脚本。
|
||||
"""
|
||||
import gzip
|
||||
import json
|
||||
from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer
|
||||
from pathlib import Path
|
||||
from urllib.error import HTTPError, URLError
|
||||
from urllib.request import Request, urlopen
|
||||
|
||||
ROOT = Path(__file__).resolve().parent
|
||||
UPSTREAM = "https://apis.raptdrama.com"
|
||||
PORT = 8787
|
||||
|
||||
DEFAULT_HEADERS = {
|
||||
"Accept": "application/json",
|
||||
"User-Agent": "RaptDrama/1.1.81 (Android 12; 23113RKC6C; Redmi)",
|
||||
"x-app-build": "139",
|
||||
"x-app-version": "1.1.81",
|
||||
"x-device-type": "phone",
|
||||
"x-device-brand": "Redmi",
|
||||
"x-device-model": "23113RKC6C",
|
||||
"x-device-id": "rd_c9f0fa17a39f96aa",
|
||||
"x-device-fingerprint": "0a4e4b212cbb0a2434cf804c8e30ab58",
|
||||
"x-device-physical": "true",
|
||||
"x-os-name": "Android",
|
||||
"x-os-version": "12",
|
||||
"accept-language": "en-US",
|
||||
"accept-language-custom": "en-US",
|
||||
}
|
||||
|
||||
HOP_HEADERS = {
|
||||
"connection",
|
||||
"keep-alive",
|
||||
"proxy-authenticate",
|
||||
"proxy-authorization",
|
||||
"te",
|
||||
"trailers",
|
||||
"transfer-encoding",
|
||||
"upgrade",
|
||||
"host",
|
||||
"content-length",
|
||||
"accept-encoding",
|
||||
}
|
||||
|
||||
CONTENT_TYPES = {
|
||||
".html": "text/html; charset=utf-8",
|
||||
".yaml": "application/yaml; charset=utf-8",
|
||||
".yml": "application/yaml; charset=utf-8",
|
||||
".css": "text/css; charset=utf-8",
|
||||
".js": "text/javascript; charset=utf-8",
|
||||
}
|
||||
|
||||
|
||||
class Handler(BaseHTTPRequestHandler):
|
||||
protocol_version = "HTTP/1.1"
|
||||
|
||||
def do_GET(self):
|
||||
self._route()
|
||||
|
||||
def do_POST(self):
|
||||
self._route()
|
||||
|
||||
def do_PUT(self):
|
||||
self._route()
|
||||
|
||||
def do_DELETE(self):
|
||||
self._route()
|
||||
|
||||
def do_OPTIONS(self):
|
||||
self.send_response(204)
|
||||
self._cors()
|
||||
self.send_header("Content-Length", "0")
|
||||
self.end_headers()
|
||||
|
||||
def _route(self):
|
||||
path = self.path.split("?", 1)[0]
|
||||
if path in ("/", "/index.html"):
|
||||
self._send_file(ROOT / "index.html")
|
||||
return
|
||||
if path == "/openapi.yaml":
|
||||
self._send_file(ROOT / "openapi.yaml")
|
||||
return
|
||||
if path.startswith("/ui/"):
|
||||
target = (ROOT / "ui" / path[len("/ui/"):]).resolve()
|
||||
if ROOT / "ui" not in target.parents and target != ROOT / "ui":
|
||||
self.send_error(404)
|
||||
return
|
||||
self._send_file(target)
|
||||
return
|
||||
if path == "/proxy" or path.startswith("/proxy/"):
|
||||
self._proxy()
|
||||
return
|
||||
self.send_error(404)
|
||||
|
||||
def _send_file(self, path: Path):
|
||||
if not path.is_file():
|
||||
self.send_error(404)
|
||||
return
|
||||
payload = path.read_bytes()
|
||||
self.send_response(200)
|
||||
self.send_header("Content-Type", CONTENT_TYPES.get(path.suffix.lower(), "application/octet-stream"))
|
||||
self.send_header("Content-Length", str(len(payload)))
|
||||
self._cors()
|
||||
self.end_headers()
|
||||
self.wfile.write(payload)
|
||||
|
||||
def _proxy(self):
|
||||
rest = self.path[len("/proxy"):] or "/"
|
||||
if not rest.startswith("/"):
|
||||
rest = "/" + rest
|
||||
length = int(self.headers.get("Content-Length") or 0)
|
||||
body = self.rfile.read(length) if length and self.command not in ("GET", "HEAD") else None
|
||||
headers = {}
|
||||
for key, value in self.headers.items():
|
||||
if key.lower() in HOP_HEADERS:
|
||||
continue
|
||||
headers[key] = value
|
||||
present = {key.lower() for key in headers}
|
||||
for key, value in DEFAULT_HEADERS.items():
|
||||
if key.lower() not in present:
|
||||
headers[key] = value
|
||||
if body is not None and "content-type" not in present:
|
||||
headers["Content-Type"] = "application/json"
|
||||
request = Request(UPSTREAM + rest, data=body, headers=headers, method=self.command)
|
||||
try:
|
||||
with urlopen(request, timeout=30) as response:
|
||||
payload = response.read()
|
||||
status = response.status
|
||||
content_type = response.headers.get("Content-Type", "application/json; charset=utf-8")
|
||||
except HTTPError as exc:
|
||||
payload = exc.read()
|
||||
status = exc.code
|
||||
content_type = exc.headers.get("Content-Type", "application/json; charset=utf-8")
|
||||
except URLError as exc:
|
||||
payload = json.dumps({"error": str(exc.reason)}, ensure_ascii=False).encode("utf-8")
|
||||
status = 502
|
||||
content_type = "application/json; charset=utf-8"
|
||||
if payload[:2] == b"\x1f\x8b":
|
||||
payload = gzip.decompress(payload)
|
||||
self.send_response(status)
|
||||
self.send_header("Content-Type", content_type)
|
||||
self.send_header("Content-Length", str(len(payload)))
|
||||
self._cors()
|
||||
self.end_headers()
|
||||
self.wfile.write(payload)
|
||||
|
||||
def _cors(self):
|
||||
self.send_header("Access-Control-Allow-Origin", "*")
|
||||
self.send_header("Access-Control-Allow-Headers", "*")
|
||||
self.send_header("Access-Control-Allow-Methods", "GET, POST, PUT, DELETE, OPTIONS")
|
||||
|
||||
def log_message(self, fmt, *args):
|
||||
print(f"{self.command} {self.path} {args[1] if len(args) > 1 else ''}", flush=True)
|
||||
|
||||
|
||||
def main():
|
||||
server = ThreadingHTTPServer(("127.0.0.1", PORT), Handler)
|
||||
print(f"Swagger 调试台: http://127.0.0.1:{PORT}", flush=True)
|
||||
print("请求会转发到 https://apis.raptdrama.com", flush=True)
|
||||
server.serve_forever()
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
main()
|
||||
Reference in New Issue
Block a user