RaptDramaDump/swagger/serve.py

"""独立的 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()