126 lines
5 KiB
Python
126 lines
5 KiB
Python
#!/usr/bin/env python3
|
|
"""Serve one prototype and save validated playtest JSONL into the repository."""
|
|
|
|
from __future__ import annotations
|
|
|
|
import argparse
|
|
import json
|
|
import os
|
|
from pathlib import Path
|
|
import re
|
|
from http.server import SimpleHTTPRequestHandler, ThreadingHTTPServer
|
|
import tempfile
|
|
from urllib.parse import urlparse
|
|
|
|
|
|
MAX_LOG_BYTES = 10 * 1024 * 1024
|
|
SAFE_FILENAME = re.compile(r"^[A-Za-z0-9][A-Za-z0-9._-]{0,159}\.jsonl$")
|
|
|
|
|
|
class PlaytestHandler(SimpleHTTPRequestHandler):
|
|
server_version = "FindMyFunPlaytestServer/1"
|
|
|
|
def do_GET(self) -> None:
|
|
if urlparse(self.path).path == "/api/playtest-log":
|
|
self._json_response(200, {"ready": True, "destination": "JSONL/"})
|
|
return
|
|
super().do_GET()
|
|
|
|
def do_POST(self) -> None:
|
|
if urlparse(self.path).path != "/api/playtest-log":
|
|
self._json_response(404, {"error": "unknown endpoint"})
|
|
return
|
|
|
|
raw_length = self.headers.get("Content-Length")
|
|
try:
|
|
length = int(raw_length or "")
|
|
except ValueError:
|
|
self._json_response(411, {"error": "valid Content-Length required"})
|
|
return
|
|
if length <= 0 or length > MAX_LOG_BYTES:
|
|
self._json_response(413, {"error": f"log must be between 1 and {MAX_LOG_BYTES} bytes"})
|
|
return
|
|
|
|
filename = self.headers.get("X-Playtest-Filename", "")
|
|
if not SAFE_FILENAME.fullmatch(filename) or ".." in filename:
|
|
self._json_response(400, {"error": "unsafe or invalid JSONL filename"})
|
|
return
|
|
|
|
body = self.rfile.read(length)
|
|
try:
|
|
text = body.decode("utf-8")
|
|
events = [json.loads(line) for line in text.splitlines() if line.strip()]
|
|
except (UnicodeDecodeError, json.JSONDecodeError) as error:
|
|
self._json_response(400, {"error": f"invalid JSONL: {error}"})
|
|
return
|
|
if not events or not all(isinstance(event, dict) for event in events):
|
|
self._json_response(400, {"error": "JSONL must contain event objects"})
|
|
return
|
|
|
|
experiment = events[0].get("experiment")
|
|
session_id = events[0].get("session_id")
|
|
if not experiment or not session_id:
|
|
self._json_response(400, {"error": "first event must contain experiment and session_id"})
|
|
return
|
|
if any(event.get("experiment") != experiment or event.get("session_id") != session_id for event in events):
|
|
self._json_response(400, {"error": "all events must belong to one experiment and session"})
|
|
return
|
|
|
|
destination = self.server.log_directory / filename
|
|
normalized = text if text.endswith("\n") else text + "\n"
|
|
temporary_path: Path | None = None
|
|
try:
|
|
with tempfile.NamedTemporaryFile("w", encoding="utf-8", dir=self.server.log_directory, prefix=".playtest-", suffix=".tmp", delete=False) as temporary:
|
|
temporary.write(normalized)
|
|
temporary.flush()
|
|
os.fsync(temporary.fileno())
|
|
temporary_path = Path(temporary.name)
|
|
os.replace(temporary_path, destination)
|
|
except OSError as error:
|
|
if temporary_path is not None:
|
|
temporary_path.unlink(missing_ok=True)
|
|
self._json_response(500, {"error": f"could not save log: {error}"})
|
|
return
|
|
|
|
self._json_response(201, {"saved": True, "path": f"JSONL/{filename}", "events": len(events), "bytes": len(normalized.encode("utf-8"))})
|
|
|
|
def _json_response(self, status: int, payload: dict[str, object]) -> None:
|
|
body = (json.dumps(payload) + "\n").encode("utf-8")
|
|
self.send_response(status)
|
|
self.send_header("Content-Type", "application/json; charset=utf-8")
|
|
self.send_header("Content-Length", str(len(body)))
|
|
self.send_header("Cache-Control", "no-store")
|
|
self.end_headers()
|
|
self.wfile.write(body)
|
|
|
|
|
|
class PlaytestServer(ThreadingHTTPServer):
|
|
def __init__(self, address: tuple[str, int], handler: type[PlaytestHandler], directory: Path, log_directory: Path):
|
|
self.log_directory = log_directory
|
|
self.log_directory.mkdir(parents=True, exist_ok=True)
|
|
super().__init__(address, lambda *args, **kwargs: handler(*args, directory=str(directory), **kwargs))
|
|
|
|
|
|
def main() -> None:
|
|
parser = argparse.ArgumentParser()
|
|
parser.add_argument("--directory", type=Path, required=True)
|
|
parser.add_argument("--log-directory", type=Path, required=True)
|
|
parser.add_argument("--port", type=int, default=8000)
|
|
args = parser.parse_args()
|
|
|
|
directory = args.directory.resolve(strict=True)
|
|
log_directory = args.log_directory.resolve()
|
|
server = PlaytestServer(("127.0.0.1", args.port), PlaytestHandler, directory, log_directory)
|
|
print(f"Serving {directory} at http://127.0.0.1:{args.port}")
|
|
print(f"Playtest logs save to {log_directory}")
|
|
try:
|
|
server.serve_forever()
|
|
except KeyboardInterrupt:
|
|
pass
|
|
finally:
|
|
server.server_close()
|
|
|
|
|
|
if __name__ == "__main__":
|
|
main()
|
|
|