#!/usr/bin/env python3 from __future__ import annotations import argparse import http.server import json import os import socketserver import urllib.error import urllib.parse import urllib.request PROXIES = { "/__api_prod__/": "https://api.global-interaction.com/", "/__api_test__/": "https://api-test.global-interaction.com/", "/__api_local__/": "http://127.0.0.1:13000/", } HOP_BY_HOP_HEADERS = { "connection", "content-length", "host", "keep-alive", "proxy-connection", "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"" # 代理只转发业务相关 header；Host、Connection 等 hop-by-hop header 必须由 urllib 按目标域重建， # 否则生产网关会看到本地 Host 或错误的连接语义，排查时会像业务接口异常。 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_proxy_response(response.status, response.headers, response.read()) except urllib.error.HTTPError as error: self.send_proxy_response(error.code, error.headers, error.read()) except Exception as error: body = json.dumps( {"code": "PROXY_ERROR", "message": str(error)} ).encode("utf-8") self.send_response(502) self.send_header("Content-Type", "application/json; charset=utf-8") self.send_header("Content-Length", str(len(body))) self.send_cors_headers() self.end_headers() self.write_body(body) def send_proxy_response( self, status: int, headers: http.client.HTTPMessage, body: bytes ) -> None: self.send_response(status) self.copy_response_headers(headers) self.send_header("Content-Length", str(len(body))) self.send_cors_headers() self.end_headers() self.write_body(body) 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=8788) 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(f"Manager center: http://{args.host}:{args.port}/gonghui/manager-center/") print("API proxy: /__api_prod__/ -> https://api.global-interaction.com/") print("API proxy: /__api_test__/ -> https://api-test.global-interaction.com/") print("API proxy: /__api_local__/ -> http://127.0.0.1:13000/") server.serve_forever() if __name__ == "__main__": main()