296 lines
10 KiB
Python
296 lines
10 KiB
Python
"""
|
|
Map Editor HTTP Server
|
|
|
|
Serves editor/client/public/ as static files with smart routing:
|
|
/ -> index.html
|
|
/page -> page.html or page/index.html
|
|
/index -> index.html
|
|
|
|
API endpoints (all paths relative to project root):
|
|
GET /api/ls?path=<dir> directory listing
|
|
GET /api/read?path=<file> file contents
|
|
GET /api/find?path=<dir>&ext=<.json> recursive file listing by extension
|
|
POST /api/write?path=<file> write file (JSON body: {"content":"..."})
|
|
POST /api/mkdir?path=<dir> create directory
|
|
POST /api/delete?path=<file> delete a file
|
|
POST /api/compile-chunk?x=&y=&z= recompile a chunk JSON to .dcf
|
|
"""
|
|
|
|
import os
|
|
import re
|
|
import sys
|
|
import json
|
|
import mimetypes
|
|
import subprocess
|
|
from http.server import HTTPServer, BaseHTTPRequestHandler
|
|
from urllib.parse import urlparse, parse_qs, unquote
|
|
from pathlib import Path
|
|
|
|
CHUNK_COORD_RE = re.compile(r"^-?\d+$")
|
|
|
|
PORT = 3000
|
|
SCRIPT_DIR = Path(__file__).parent.resolve()
|
|
PUBLIC_DIR = SCRIPT_DIR.parent / "client" / "public"
|
|
PROJECT_ROOT = SCRIPT_DIR.parent.parent.resolve()
|
|
|
|
|
|
class Handler(BaseHTTPRequestHandler):
|
|
def do_GET(self):
|
|
parsed = urlparse(self.path)
|
|
path = unquote(parsed.path)
|
|
if path.startswith("/api/"):
|
|
params = parse_qs(parsed.query)
|
|
if path == "/api/ls":
|
|
self.api_ls(params)
|
|
elif path == "/api/read":
|
|
self.api_read(params)
|
|
elif path == "/api/find":
|
|
self.api_find(params)
|
|
else:
|
|
self.send_error(404)
|
|
else:
|
|
self.serve_static(path)
|
|
|
|
def do_POST(self):
|
|
parsed = urlparse(self.path)
|
|
params = parse_qs(parsed.query)
|
|
if parsed.path == "/api/write":
|
|
self.api_write(params)
|
|
elif parsed.path == "/api/mkdir":
|
|
self.api_mkdir(params)
|
|
elif parsed.path == "/api/delete":
|
|
self.api_delete(params)
|
|
elif parsed.path == "/api/compile-chunk":
|
|
self.api_compile_chunk(params)
|
|
else:
|
|
self.send_error(404)
|
|
|
|
# ------------------------------------------------------------------
|
|
# Helpers
|
|
# ------------------------------------------------------------------
|
|
|
|
def resolve_project_path(self, rel):
|
|
rel = rel.lstrip("/")
|
|
target = (PROJECT_ROOT / rel).resolve()
|
|
if not str(target).startswith(str(PROJECT_ROOT)):
|
|
return None
|
|
return target
|
|
|
|
def send_json(self, data, code=200):
|
|
body = json.dumps(data, indent=2).encode("utf-8")
|
|
self.send_response(code)
|
|
self.send_header("Content-Type", "application/json")
|
|
self.send_header("Content-Length", len(body))
|
|
self.end_headers()
|
|
self.wfile.write(body)
|
|
|
|
def read_body(self):
|
|
length = int(self.headers.get("Content-Length", 0))
|
|
return self.rfile.read(length)
|
|
|
|
# ------------------------------------------------------------------
|
|
# API handlers
|
|
# ------------------------------------------------------------------
|
|
|
|
def api_ls(self, params):
|
|
rel = params.get("path", [""])[0]
|
|
target = self.resolve_project_path(rel)
|
|
if target is None:
|
|
self.send_json({"error": "invalid path"}, 400)
|
|
return
|
|
if not target.exists():
|
|
self.send_json({"error": "path not found"}, 404)
|
|
return
|
|
if not target.is_dir():
|
|
self.send_json({"error": "not a directory"}, 400)
|
|
return
|
|
|
|
entries = []
|
|
for entry in sorted(target.iterdir(), key=lambda e: (e.is_file(), e.name)):
|
|
entries.append({
|
|
"name": entry.name,
|
|
"type": "dir" if entry.is_dir() else "file",
|
|
"size": entry.stat().st_size if entry.is_file() else None,
|
|
})
|
|
self.send_json({
|
|
"path": str(target.relative_to(PROJECT_ROOT)),
|
|
"entries": entries,
|
|
})
|
|
|
|
def api_read(self, params):
|
|
rel = params.get("path", [""])[0]
|
|
target = self.resolve_project_path(rel)
|
|
if target is None:
|
|
self.send_json({"error": "invalid path"}, 400)
|
|
return
|
|
if not target.exists():
|
|
self.send_json({"error": "file not found"}, 404)
|
|
return
|
|
if not target.is_file():
|
|
self.send_json({"error": "not a file"}, 400)
|
|
return
|
|
|
|
try:
|
|
content = target.read_text(encoding="utf-8")
|
|
self.send_json({
|
|
"path": str(target.relative_to(PROJECT_ROOT)),
|
|
"content": content,
|
|
})
|
|
except UnicodeDecodeError:
|
|
import base64
|
|
content = base64.b64encode(target.read_bytes()).decode("ascii")
|
|
self.send_json({
|
|
"path": str(target.relative_to(PROJECT_ROOT)),
|
|
"content": content,
|
|
"encoding": "base64",
|
|
})
|
|
|
|
def api_write(self, params):
|
|
rel = params.get("path", [""])[0]
|
|
target = self.resolve_project_path(rel)
|
|
if target is None:
|
|
self.send_json({"error": "invalid path"}, 400)
|
|
return
|
|
|
|
body = self.read_body()
|
|
try:
|
|
data = json.loads(body)
|
|
content = data.get("content", "")
|
|
except (json.JSONDecodeError, UnicodeDecodeError):
|
|
content = body.decode("utf-8")
|
|
|
|
target.parent.mkdir(parents=True, exist_ok=True)
|
|
target.write_text(content, encoding="utf-8")
|
|
self.send_json({"ok": True, "path": str(target.relative_to(PROJECT_ROOT))})
|
|
|
|
def api_mkdir(self, params):
|
|
rel = params.get("path", [""])[0]
|
|
target = self.resolve_project_path(rel)
|
|
if target is None:
|
|
self.send_json({"error": "invalid path"}, 400)
|
|
return
|
|
target.mkdir(parents=True, exist_ok=True)
|
|
self.send_json({"ok": True, "path": str(target.relative_to(PROJECT_ROOT))})
|
|
|
|
def api_delete(self, params):
|
|
rel = params.get("path", [""])[0]
|
|
target = self.resolve_project_path(rel)
|
|
if target is None:
|
|
self.send_json({"error": "invalid path"}, 400)
|
|
return
|
|
if target.is_file():
|
|
target.unlink()
|
|
self.send_json({"ok": True, "path": str(target.relative_to(PROJECT_ROOT))})
|
|
|
|
def api_find(self, params):
|
|
rel = params.get("path", [""])[0]
|
|
ext = params.get("ext", [""])[0]
|
|
target = self.resolve_project_path(rel)
|
|
if target is None:
|
|
self.send_json({"error": "invalid path"}, 400)
|
|
return
|
|
if not target.is_dir():
|
|
self.send_json({"error": "not a directory"}, 400)
|
|
return
|
|
|
|
pattern = f"*{ext}" if ext else "*"
|
|
paths = sorted(
|
|
str(p.relative_to(target)).replace(os.sep, "/")
|
|
for p in target.rglob(pattern)
|
|
if p.is_file()
|
|
)
|
|
self.send_json({"path": str(target.relative_to(PROJECT_ROOT)), "files": paths})
|
|
|
|
def api_compile_chunk(self, params):
|
|
x = params.get("x", [""])[0]
|
|
y = params.get("y", [""])[0]
|
|
z = params.get("z", [""])[0]
|
|
if not (CHUNK_COORD_RE.match(x) and CHUNK_COORD_RE.match(y) and CHUNK_COORD_RE.match(z)):
|
|
self.send_json({"error": "invalid chunk coordinates"}, 400)
|
|
return
|
|
|
|
json_path = PROJECT_ROOT / "assetsraw" / "chunks" / f"chunk_{x}_{y}_{z}.json"
|
|
if not json_path.is_file():
|
|
self.send_json({"error": "chunk json not found"}, 404)
|
|
return
|
|
|
|
result = subprocess.run(
|
|
[sys.executable, "-m", "tools.asset.chunk", str(json_path)],
|
|
cwd=str(PROJECT_ROOT),
|
|
capture_output=True,
|
|
text=True,
|
|
)
|
|
self.send_json({
|
|
"ok": result.returncode == 0,
|
|
"returncode": result.returncode,
|
|
"stdout": result.stdout,
|
|
"stderr": result.stderr,
|
|
})
|
|
|
|
# ------------------------------------------------------------------
|
|
# Static file serving
|
|
# ------------------------------------------------------------------
|
|
|
|
def serve_static(self, path):
|
|
clean = path.lstrip("/")
|
|
candidates = []
|
|
|
|
if clean == "" or clean == "index" or clean == "index.html":
|
|
candidates = ["index.html"]
|
|
else:
|
|
candidates.append(clean)
|
|
if "." not in Path(clean).name:
|
|
candidates.append(clean + ".html")
|
|
candidates.append(clean + "/index.html")
|
|
|
|
for candidate in candidates:
|
|
file_path = PUBLIC_DIR / candidate
|
|
if file_path.is_file():
|
|
self.serve_file(file_path)
|
|
return
|
|
|
|
self.send_error(404)
|
|
|
|
def serve_file(self, file_path):
|
|
mime, _ = mimetypes.guess_type(str(file_path))
|
|
mime = mime or "application/octet-stream"
|
|
|
|
template = PUBLIC_DIR / "template.html"
|
|
if (
|
|
mime == "text/html"
|
|
and file_path.name != "template.html"
|
|
and template.is_file()
|
|
):
|
|
fragment = file_path.read_text(encoding="utf-8")
|
|
shell = template.read_text(encoding="utf-8")
|
|
marker = shell.find('id="editor"')
|
|
if marker == -1:
|
|
content = shell.encode("utf-8")
|
|
else:
|
|
tag_end = shell.index(">", marker) + 1
|
|
close_idx = shell.index("</div>", tag_end)
|
|
content = (shell[:tag_end] + fragment + shell[close_idx:]).encode("utf-8")
|
|
else:
|
|
content = file_path.read_bytes()
|
|
|
|
self.send_response(200)
|
|
self.send_header("Content-Type", mime)
|
|
self.send_header("Content-Length", len(content))
|
|
self.end_headers()
|
|
self.wfile.write(content)
|
|
|
|
def log_message(self, fmt, *args):
|
|
print(f" {self.address_string()} {fmt % args}")
|
|
|
|
|
|
if __name__ == "__main__":
|
|
print(f"Map Editor")
|
|
print(f" http://localhost:{PORT}")
|
|
print(f" public : {PUBLIC_DIR}")
|
|
print(f" project: {PROJECT_ROOT}")
|
|
server = HTTPServer(("", PORT), Handler)
|
|
try:
|
|
server.serve_forever()
|
|
except KeyboardInterrupt:
|
|
print("\nStopped.")
|