Compressor/3d/server.py

#!/usr/bin/env python3
"""3D 部件查看与配色 - 本地演示服务器 (纯标准库实现, 无需安装依赖)

用法:
    python3 server.py              # 默认端口 9000, 浏览器打开 http://127.0.0.1:9000
    python3 server.py --port 9100

功能:
    GET  /                    首页 (index.html)
    GET  /js/... /css/...     前端静态资源
    GET  /models/<file>       GLB/glTF 模型文件
    GET  /api/models          返回 models 目录下可用的模型列表 (JSON)
    GET  /api/state?model=xx  读取某个模型的配色状态 (JSON)
    POST /api/state?model=xx  保存配色状态到 data/color_state.json
"""

import argparse
import json
import mimetypes
import os
import re
import urllib.parse
from http.server import SimpleHTTPRequestHandler, ThreadingHTTPServer

BASE = os.path.dirname(os.path.abspath(__file__))
STATIC_DIR = os.path.join(BASE, "static")
MODELS_DIR = os.path.join(BASE, "models")
DATA_DIR = os.path.join(BASE, "data")
DATA_FILE = os.path.join(DATA_DIR, "color_state.json")

EXTRA_TYPES = {
    ".glb": "model/gltf-binary",
    ".gltf": "model/gltf+json",
    ".js": "text/javascript; charset=utf-8",
    ".json": "application/json; charset=utf-8",
    ".css": "text/css; charset=utf-8",
    ".html": "text/html; charset=utf-8",
}

# 允许的扩展名 -> 相对子目录
ROUTES = {
    "/models": ("models", {".glb", ".gltf"}),
    "/js": ("static/js", {".js", ".mjs"}),
    "/css": ("static/css", {".css"}),
    "/images": ("static/images", {".png", ".jpg", ".jpeg", ".gif", ".svg", ".webp"}),
}

SAFE_NAME = re.compile(r"^[\w\-. ]+$")


class Handler(SimpleHTTPRequestHandler):
    protocol_version = "HTTP/1.1"

    # ---- 工具 ----
    def _guess_type(self, path):
        ext = os.path.splitext(path)[1].lower()
        if ext in EXTRA_TYPES:
            return EXTRA_TYPES[ext]
        return mimetypes.guess_type(path)[0] or "application/octet-stream"

    def _send_bytes(self, code, body: bytes, ctype="application/octet-stream"):
        self.send_response(code)
        self.send_header("Content-Type", ctype)
        self.send_header("Content-Length", str(len(body)))
        self.send_header("Cache-Control", "no-store")
        self.end_headers()
        self.wfile.write(body)

    def _send_json(self, obj, code=200):
        self._send_bytes(code, json.dumps(obj, ensure_ascii=False).encode("utf-8"),
                         "application/json; charset=utf-8")

    def _send_file(self, full_path, ctype=None):
        try:
            with open(full_path, "rb") as f:
                body = f.read()
        except OSError:
            self.send_error(404, "Not Found")
            return
        self._send_bytes(200, body, ctype or self._guess_type(full_path))

    def _safe_join(self, root, rel):
        """防路径穿越"""
        full = os.path.abspath(os.path.join(root, rel))
        if not full.startswith(os.path.abspath(root) + os.sep) and full != os.path.abspath(root):
            return None
        return full

    def _resolve_static(self, path):
        for prefix, (sub, exts) in ROUTES.items():
            if path == prefix or path.startswith(prefix + "/"):
                rel = path[len(prefix):].lstrip("/")
                root = os.path.join(BASE, sub)
                full = self._safe_join(root, rel) if rel else root
                if full and os.path.isfile(full):
                    ext = os.path.splitext(full)[1].lower()
                    return full if ext in exts else None
                if full and os.path.isdir(full):
                    idx = os.path.join(full, "index.html")
                    if os.path.isfile(idx):
                        return idx
                return None
        return None

    # ---- 数据读写 ----
    def _load_state(self):
        try:
            with open(DATA_FILE, "r", encoding="utf-8") as f:
                return json.load(f)
        except (OSError, json.JSONDecodeError):
            return {}

    def _save_state(self, state):
        os.makedirs(DATA_DIR, exist_ok=True)
        tmp = DATA_FILE + ".tmp"
        with open(tmp, "w", encoding="utf-8") as f:
            json.dump(state, f, ensure_ascii=False, indent=2)
        os.replace(tmp, DATA_FILE)

    # ---- handlers ----
    def do_GET(self):
        parsed = urllib.parse.urlparse(self.path)
        path = urllib.parse.unquote(parsed.path)
        query = urllib.parse.parse_qs(parsed.query)

        if path == "/api/models":
            try:
                names = [f for f in os.listdir(MODELS_DIR)
                         if os.path.isfile(os.path.join(MODELS_DIR, f))
                         and SAFE_NAME.match(f)
                         and f.lower().endswith((".glb", ".gltf"))]
            except OSError:
                names = []
            self._send_json(sorted(names))
            return

        if path == "/api/state":
            model = (query.get("model") or [""])[0]
            state = self._load_state()
            self._send_json(state.get(model, {}))
            return

        if path == "/" or path == "":
            self._send_file(os.path.join(STATIC_DIR, "index.html"))
            return

        static_file = self._resolve_static(path)
        if static_file:
            self._send_file(static_file)
            return

        if path.startswith("/models/"):
            rel = path[len("/models/"):]
            if SAFE_NAME.match(rel):
                full = self._safe_join(MODELS_DIR, rel)
                if full and os.path.isfile(full):
                    self._send_file(full)
                    return
        self.send_error(404, "Not Found")

    def do_POST(self):
        parsed = urllib.parse.urlparse(self.path)
        path = urllib.parse.unquote(parsed.path)
        query = urllib.parse.parse_qs(parsed.query)

        if path == "/api/state":
            try:
                length = int(self.headers.get("Content-Length") or 0)
                body = self.rfile.read(length) if length else b"{}"
                data = json.loads(body.decode("utf-8"))
            except (ValueError, json.JSONDecodeError):
                self.send_error(400, "Bad JSON")
                return
            model = (query.get("model") or [""])[0]
            if not model:
                self.send_error(400, "缺少 model 参数")
                return
            if not isinstance(data, dict):
                self.send_error(400, "body 应为 JSON 对象")
                return
            state = self._load_state()
            state[model] = data
            self._save_state(state)
            self._send_json({"ok": True, "model": model, "count": len(data)})
            return

        self.send_error(405, "Method Not Allowed")

    # ---- 日志 ----
    def log_message(self, fmt, *args):
        print(f"[server] {self.address_string()} - {fmt % args}")


def main():
    ap = argparse.ArgumentParser(description="3D 部件查看与配色本地服务器")
    ap.add_argument("--host", default="127.0.0.1")
    ap.add_argument("--port", type=int, default=9000)
    args = ap.parse_args()

    httpd = ThreadingHTTPServer((args.host, args.port), Handler)
    url = f"http://{args.host}:{args.port}"
    print("=" * 56)
    print("  3D 部件查看与配色 演示服务器已启动")
    print(f"  请用浏览器打开: {url}")
    print(f"  模型目录: {MODELS_DIR}")
    print(f"  配色数据: {DATA_FILE}")
    print("  按 Ctrl+C 停止")
    print("=" * 56)
    try:
        httpd.serve_forever()
    except KeyboardInterrupt:
        print("\n已停止")


if __name__ == "__main__":
    main()