qinghe-web/Youthome-Web/serve.py

"""Local web preview with caching disabled for editable frontend files."""

from functools import partial
from http.server import SimpleHTTPRequestHandler, ThreadingHTTPServer
from http.client import HTTPConnection, HTTPSConnection
from pathlib import Path
from urllib.parse import urlsplit
from urllib.request import Request, urlopen
from urllib.error import HTTPError
import argparse


class PreviewHandler(SimpleHTTPRequestHandler):
    # 后端地址，可通过命令行参数覆盖；默认本机 8009
    backend_host = "127.0.0.1"
    backend_port = 8009
    backend_scheme = "http"

    def proxy_api(self):
        length = int(self.headers.get("Content-Length", "0"))
        if length > 1024 * 1024 * 1024:
            self.send_error(413, "Upload too large")
            return
        body = self.rfile.read(length) if length else None
        headers = {}
        for name in ("Content-Type", "access_token"):
            if self.headers.get(name):
                headers[name] = self.headers[name]
        connection = None
        response = None
        try:
            if self.backend_scheme == "https":
                request = Request(
                    f"https://{self.backend_host}:{self.backend_port}{self.path}",
                    data=body, headers=headers, method=self.command,
                )
                try:
                    response = urlopen(request, timeout=120)
                except HTTPError as error:
                    response = error
            else:
                connection = HTTPConnection(self.backend_host, self.backend_port, timeout=120)
                connection.request(self.command, self.path, body=body, headers=headers)
                response = connection.getresponse()
            content = response.read()
            self.send_response(response.status if connection else response.code)
            for name in ("Content-Type", "Cache-Control"):
                value = response.getheader(name) if connection else response.headers.get(name)
                if value:
                    self.send_header(name, value)
            self.send_header("Content-Length", str(len(content)))
            self.end_headers()
            self.wfile.write(content)
        except OSError:
            self.send_error(502, "Backend unavailable")
        finally:
            if response:
                response.close()
            if connection:
                connection.close()

    def proxy_media(self):
        # /media/<path> -> https://video.yoyoeco.com/<path>
        target = self.path[len("/media"):] or "/"
        headers = {}
        if self.headers.get("Range"):
            headers["Range"] = self.headers["Range"]
        connection = HTTPSConnection("video.yoyoeco.com", 443, timeout=60)
        try:
            connection.request("GET", target, headers=headers)
            response = connection.getresponse()
            self.send_response(response.status)
            for name in ("Content-Type", "Content-Length", "Content-Range", "Accept-Ranges", "Cache-Control"):
                value = response.getheader(name)
                if value:
                    self.send_header(name, value)
            self.send_header("Access-Control-Allow-Origin", "*")
            self.end_headers()
            while True:
                chunk = response.read(64 * 1024)
                if not chunk:
                    break
                self.wfile.write(chunk)
        except OSError:
            self.send_error(502, "Media unavailable")
        finally:
            connection.close()

    def do_GET(self):
        if self.path.startswith("/api/"):
            return self.proxy_api()
        if self.path.startswith("/media/"):
            return self.proxy_media()
        return super().do_GET()

    def do_POST(self):
        if self.path.startswith("/api/"):
            return self.proxy_api()
        self.send_error(404)

    def do_DELETE(self):
        if self.path.startswith("/api/"):
            return self.proxy_api()
        self.send_error(404)

    def do_PUT(self):
        if self.path.startswith("/api/"):
            return self.proxy_api()
        self.send_error(404)

    def do_PATCH(self):
        if self.path.startswith("/api/"):
            return self.proxy_api()
        self.send_error(404)

    def do_OPTIONS(self):
        if self.path.startswith("/api/"):
            return self.proxy_api()
        self.send_error(404)

    def end_headers(self):
        if self.path.split("?", 1)[0].endswith((".html", ".js", ".css")):
            self.send_header("Cache-Control", "no-store, max-age=0")
        super().end_headers()


def main():
    parser = argparse.ArgumentParser(description="Serve the Youthome web preview")
    parser.add_argument("--port", type=int, default=4173)
    parser.add_argument("--backend-host", default="127.0.0.1", help="Backend host that /api/* is proxied to")
    parser.add_argument("--backend-port", type=int, default=8009, help="Backend port that /api/* is proxied to")
    parser.add_argument("--backend-url", help="Full HTTP(S) backend origin; overrides host and port")
    args = parser.parse_args()
    if args.backend_url:
        backend = urlsplit(args.backend_url)
        if backend.scheme not in ("http", "https") or not backend.hostname or backend.path not in ("", "/") or backend.query or backend.fragment:
            parser.error("--backend-url must be an HTTP(S) origin without a path")
        PreviewHandler.backend_scheme = backend.scheme
        PreviewHandler.backend_host = backend.hostname
        PreviewHandler.backend_port = backend.port or (443 if backend.scheme == "https" else 80)
    else:
        PreviewHandler.backend_host = args.backend_host
        PreviewHandler.backend_port = args.backend_port
    handler = partial(PreviewHandler, directory=str(Path(__file__).resolve().parent))
    with ThreadingHTTPServer(("127.0.0.1", args.port), handler) as server:
        print(f"Youthome web preview: http://127.0.0.1:{args.port}/pages/feed.html", flush=True)
        print(f"  /api/* -> {PreviewHandler.backend_scheme}://{PreviewHandler.backend_host}:{PreviewHandler.backend_port}", flush=True)
        server.serve_forever()


if __name__ == "__main__":
    main()