-
Notifications
You must be signed in to change notification settings - Fork 0
Expand file tree
/
Copy pathserver.py
More file actions
135 lines (118 loc) · 4.63 KB
/
Copy pathserver.py
File metadata and controls
135 lines (118 loc) · 4.63 KB
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
#!/usr/bin/env python3
# -*- coding: utf-8 -*-
"""PUBG Tactical Map — local server.
Serves the static app AND persists developer-annotated points into the
project file ``points.js``, so the data travels with the project folder.
Run:
python server.py # -> http://127.0.0.1:8123
python server.py 9000 # custom port
Saving points requires this server. Opening index.html directly (file://)
still lets you VIEW points, but not save them.
"""
import json
import os
import sys
from http.server import HTTPServer, SimpleHTTPRequestHandler
from urllib.parse import unquote
ROOT = os.path.dirname(os.path.abspath(__file__))
WEB = os.path.join(ROOT, "src") # web app root (index.html + assets)
TILES = os.path.join(ROOT, "tiles") # generated tile pyramids (kept out of repo)
POINTS_JS = os.path.join(WEB, "points.js")
DEFAULT_PORT = 8123
class Handler(SimpleHTTPRequestHandler):
def translate_path(self, path):
path = unquote(path.split("?", 1)[0])
parts = path.lstrip("/").split("/")
if parts and parts[0] == "tiles":
return os.path.join(TILES, *parts[1:])
return os.path.join(WEB, *parts)
def do_POST(self):
if self.path.split("?")[0] == "/api/points":
self._save_points()
else:
self.send_error(404, "Not Found")
def do_GET(self):
if self.path.split("?")[0] == "/api/points":
self._send_points()
else:
SimpleHTTPRequestHandler.do_GET(self)
def _save_points(self):
try:
length = int(self.headers.get("Content-Length", 0))
raw = self.rfile.read(length)
data = json.loads(raw.decode("utf-8"))
if not isinstance(data, dict):
raise ValueError("points payload must be a JSON object")
content = (
"// AUTO-SAVED by developer mode - user-annotated points.\n"
"// This file is the project's point database. It is rewritten by\n"
"// server.py whenever the developer saves; do not edit by hand.\n"
"window.MAP_POINTS = "
+ json.dumps(data, ensure_ascii=False, indent=1)
+ ";\n"
)
tmp = POINTS_JS + ".tmp"
with open(tmp, "w", encoding="utf-8") as f:
f.write(content)
os.replace(tmp, POINTS_JS) # atomic replace
self._json(200, {"ok": True})
except Exception as e:
self._json(500, {"ok": False, "error": str(e)})
def _send_points(self):
try:
data = {}
if os.path.exists(POINTS_JS):
with open(POINTS_JS, "r", encoding="utf-8") as f:
text = f.read()
start, end = text.find("{"), text.rfind("}")
if start != -1 and end > start:
data = json.loads(text[start:end + 1])
self._json(200, data)
except Exception as e:
self._json(500, {"ok": False, "error": str(e)})
def _json(self, code, obj):
body = json.dumps(obj, ensure_ascii=False).encode("utf-8")
self.send_response(code)
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)
# never cache points.js so a refresh always picks up the latest save
def end_headers(self):
p = (self.path or "").split("?")[0]
if p.endswith("points.js"):
self.send_header("Cache-Control", "no-store, max-age=0")
SimpleHTTPRequestHandler.end_headers(self)
def log_message(self, fmt, *args):
sys.stderr.write("[%s] %s\n" % (self.log_date_time_string(), fmt % args))
def main():
port = DEFAULT_PORT
auto_open = False
for arg in sys.argv[1:]:
if arg == "--open":
auto_open = True
else:
try:
port = int(arg)
except ValueError:
pass
try:
httpd = HTTPServer(("127.0.0.1", port), Handler)
except OSError as e:
print("Cannot start: port %d busy (%s). Try: python server.py 9000" % (port, e))
sys.exit(1)
url = "http://127.0.0.1:%d" % port
print("PUBG Tactical Map server -> %s (Ctrl+C to stop)" % url)
if auto_open:
import threading
import webbrowser
t = threading.Timer(1.2, lambda: webbrowser.open(url))
t.daemon = True
t.start()
try:
httpd.serve_forever()
except KeyboardInterrupt:
print("\nstopped")
if __name__ == "__main__":
main()