107 lines
4.0 KiB
Python
107 lines
4.0 KiB
Python
#!/usr/bin/env python3
|
|
"""Tiny local server for the Pokémon Party OBS overlay.
|
|
|
|
It serves overlay.html/editor.html like `python -m http.server`, plus adds:
|
|
POST /api/party -> validates JSON and writes party.json in this folder.
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import argparse
|
|
import json
|
|
from http.server import SimpleHTTPRequestHandler, ThreadingHTTPServer
|
|
from pathlib import Path
|
|
from typing import Any
|
|
|
|
EMPTY_SLOT = {"species": "", "nickname": "", "sprite": "", "status": "empty"}
|
|
VALID_STATUSES = {"alive", "dead", "boxed", "empty"}
|
|
|
|
|
|
def normalize_slot(slot: Any) -> dict[str, str]:
|
|
if not isinstance(slot, dict):
|
|
return dict(EMPTY_SLOT)
|
|
|
|
species = str(slot.get("species", "") or "").strip().lower().replace(" ", "-")
|
|
nickname = str(slot.get("nickname", "") or "").strip()
|
|
sprite = str(slot.get("sprite", "") or "").strip()
|
|
status = str(slot.get("status", "") or "").strip().lower()
|
|
|
|
if not status or status not in VALID_STATUSES:
|
|
status = "alive" if (species or nickname or sprite) else "empty"
|
|
|
|
if not (species or nickname or sprite):
|
|
return dict(EMPTY_SLOT)
|
|
|
|
return {
|
|
"species": species,
|
|
"nickname": nickname,
|
|
"sprite": sprite,
|
|
"status": status,
|
|
}
|
|
|
|
|
|
def normalize_party(data: Any) -> dict[str, list[dict[str, str]]]:
|
|
party = data.get("party") if isinstance(data, dict) else data
|
|
if not isinstance(party, list):
|
|
raise ValueError("Expected JSON object with a party array, or a party array.")
|
|
|
|
slots = [normalize_slot(slot) for slot in party[:6]]
|
|
while len(slots) < 6:
|
|
slots.append(dict(EMPTY_SLOT))
|
|
return {"party": slots}
|
|
|
|
|
|
class PartyOverlayHandler(SimpleHTTPRequestHandler):
|
|
server_version = "PokemonPartyOverlay/1.0"
|
|
|
|
def do_POST(self) -> None: # noqa: N802 - http.server method name
|
|
if self.path.split("?", 1)[0] != "/api/party":
|
|
self.send_json({"ok": False, "error": "Not found"}, status=404)
|
|
return
|
|
|
|
try:
|
|
length = int(self.headers.get("Content-Length", "0"))
|
|
raw = self.rfile.read(length).decode("utf-8")
|
|
data = json.loads(raw)
|
|
normalized = normalize_party(data)
|
|
party_path = Path(self.directory).resolve() / "party.json"
|
|
party_path.write_text(json.dumps(normalized, indent=2) + "\n", encoding="utf-8")
|
|
except Exception as exc: # Keep this friendly for local streamer use.
|
|
self.send_json({"ok": False, "error": str(exc)}, status=400)
|
|
return
|
|
|
|
self.send_json({"ok": True, "message": "party.json saved"})
|
|
|
|
def send_json(self, payload: dict[str, Any], status: int = 200) -> None:
|
|
body = json.dumps(payload).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)
|
|
|
|
|
|
def main() -> None:
|
|
parser = argparse.ArgumentParser(description="Run the Pokémon Party OBS overlay server.")
|
|
parser.add_argument("--host", default="127.0.0.1", help="Host to bind. Default: 127.0.0.1")
|
|
parser.add_argument("--port", type=int, default=8765, help="Port to bind. Default: 8765")
|
|
parser.add_argument("--directory", default=str(Path(__file__).resolve().parent), help="Overlay folder to serve.")
|
|
args = parser.parse_args()
|
|
|
|
handler = lambda *handler_args, **kwargs: PartyOverlayHandler( # noqa: E731
|
|
*handler_args,
|
|
directory=args.directory,
|
|
**kwargs,
|
|
)
|
|
with ThreadingHTTPServer((args.host, args.port), handler) as httpd:
|
|
print(f"Pokémon Party Overlay server running at http://{args.host}:{args.port}/")
|
|
print(f"Editor: http://{args.host}:{args.port}/editor.html")
|
|
print(f"Overlay: http://{args.host}:{args.port}/overlay.html")
|
|
print("Keep this window open while streaming. Press Ctrl+C to stop.")
|
|
httpd.serve_forever()
|
|
|
|
|
|
if __name__ == "__main__":
|
|
main()
|