hyapp-h5/scripts/h5_dev_server.py

#!/usr/bin/env python3
from __future__ import annotations

import argparse
import http.server
import os
import socketserver
import urllib.error
import urllib.parse
import urllib.request


PROXIES = {
    "/__api_test__/": "https://api-test.global-interaction.com/",
    "/__api_local__/": "http://localhost:13000/",
}

HOP_BY_HOP_HEADERS = {
    "connection",
    "content-length",
    "host",
    "keep-alive",
    "proxy-authenticate",
    "proxy-authorization",
    "te",
    "trailer",
    "transfer-encoding",
    "upgrade",
}


class H5DevHandler(http.server.SimpleHTTPRequestHandler):
    def do_OPTIONS(self) -> None:
        if self.proxy_base() is None:
            self.send_response(204)
            self.end_headers()
            return
        self.send_response(204)
        self.send_cors_headers()
        self.end_headers()

    def do_GET(self) -> None:
        if self.path == "/__h5_dev_server__/health":
            self.send_response(200)
            self.send_header("Content-Type", "text/plain; charset=utf-8")
            self.send_header("Content-Length", "2")
            self.end_headers()
            self.wfile.write(b"ok")
            return
        self.handle_request()

    def do_HEAD(self) -> None:
        if self.path == "/__h5_dev_server__/health":
            self.send_response(200)
            self.send_header("Content-Type", "text/plain; charset=utf-8")
            self.send_header("Content-Length", "2")
            self.end_headers()
            return
        if self.proxy_base() is None:
            super().do_HEAD()
            return
        self.proxy_request()

    def do_POST(self) -> None:
        self.handle_request()

    def do_PUT(self) -> None:
        self.handle_request()

    def do_PATCH(self) -> None:
        self.handle_request()

    def do_DELETE(self) -> None:
        self.handle_request()

    def handle_request(self) -> None:
        if self.proxy_base() is None:
            super().do_GET()
            return
        self.proxy_request()

    def proxy_base(self) -> tuple[str, str] | None:
        parsed = urllib.parse.urlsplit(self.path)
        for prefix, target in PROXIES.items():
            if parsed.path.startswith(prefix):
                return prefix, target
        return None

    def proxy_request(self) -> None:
        match = self.proxy_base()
        if match is None:
            self.send_error(404)
            return
        prefix, target_base = match
        parsed = urllib.parse.urlsplit(self.path)
        target_path = parsed.path[len(prefix):]
        target_url = urllib.parse.urljoin(target_base, target_path)
        if parsed.query:
            target_url = target_url + "?" + parsed.query

        body = None
        if self.command in {"POST", "PUT", "PATCH"}:
            length = int(self.headers.get("Content-Length", "0") or "0")
            body = self.rfile.read(length) if length > 0 else b""

        headers = {
            key: value
            for key, value in self.headers.items()
            if key.lower() not in HOP_BY_HOP_HEADERS
        }
        headers["Accept-Encoding"] = "identity"

        request = urllib.request.Request(
            target_url,
            data=body,
            headers=headers,
            method=self.command,
        )
        try:
            with urllib.request.urlopen(request, timeout=30) as response:
                self.send_response(response.status)
                self.copy_response_headers(response.headers)
                self.send_cors_headers()
                self.end_headers()
                self.write_body(response.read())
        except urllib.error.HTTPError as error:
            self.send_response(error.code)
            self.copy_response_headers(error.headers)
            self.send_cors_headers()
            self.end_headers()
            self.write_body(error.read())
        except Exception as error:
            self.send_response(502)
            self.send_header("Content-Type", "application/json; charset=utf-8")
            self.send_cors_headers()
            self.end_headers()
            message = str(error).replace('"', '\\"')
            self.write_body((f'{{"code":"PROXY_ERROR","message":"{message}"}}').encode("utf-8"))

    def write_body(self, body: bytes) -> None:
        if self.command == "HEAD":
            return
        try:
            self.wfile.write(body)
        except BrokenPipeError:
            return

    def copy_response_headers(self, headers: http.client.HTTPMessage) -> None:
        for key, value in headers.items():
            if key.lower() in HOP_BY_HOP_HEADERS:
                continue
            if key.lower() == "content-encoding":
                continue
            self.send_header(key, value)

    def send_cors_headers(self) -> None:
        self.send_header("Access-Control-Allow-Origin", "*")
        self.send_header("Access-Control-Allow-Methods", "GET,POST,PUT,PATCH,DELETE,OPTIONS")
        self.send_header("Access-Control-Allow-Headers", "Authorization,Content-Type,X-App-Code,X-Request-ID")

    def end_headers(self) -> None:
        self.send_header("Cache-Control", "no-store, no-cache, must-revalidate, max-age=0")
        self.send_header("Pragma", "no-cache")
        self.send_header("Expires", "0")
        super().end_headers()


class ReusableTCPServer(socketserver.ThreadingTCPServer):
    allow_reuse_address = True


def main() -> None:
    parser = argparse.ArgumentParser()
    parser.add_argument("--host", default="127.0.0.1")
    parser.add_argument("--port", type=int, default=30000)
    parser.add_argument("--directory", default=os.getcwd())
    args = parser.parse_args()

    handler = lambda *handler_args, **handler_kwargs: H5DevHandler(
        *handler_args,
        directory=args.directory,
        **handler_kwargs,
    )
    with ReusableTCPServer((args.host, args.port), handler) as server:
        print(f"Serving H5 at http://{args.host}:{args.port}/")
        print("API proxy: /__api_test__/ -> https://api-test.global-interaction.com/")
        print("API proxy: /__api_local__/ -> http://localhost:13000/")
        server.serve_forever()


if __name__ == "__main__":
    main()