239 lines
9.0 KiB
Python
239 lines
9.0 KiB
Python
#!/usr/bin/env python3
|
|
"""Small, dependency-free web service for the feedc0de co-op loading screen."""
|
|
|
|
from __future__ import annotations
|
|
|
|
import argparse
|
|
import json
|
|
import mimetypes
|
|
import re
|
|
import sqlite3
|
|
import sys
|
|
from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer
|
|
from pathlib import Path
|
|
from urllib.parse import parse_qs, unquote, urlsplit
|
|
|
|
|
|
HERE = Path(__file__).resolve().parent
|
|
DEFAULT_DATABASE = HERE.parents[2] / "sv.db"
|
|
DEFAULT_SCREENSHOTS = HERE.parent / "backgrounds"
|
|
STEAM_ID = re.compile(r"^7656119\d{10}$")
|
|
SCREENSHOT = re.compile(r"^[a-z0-9_-]+\.jpg$", re.IGNORECASE)
|
|
RANKS = (
|
|
(0, "Newbie"),
|
|
(10, "Survivor"),
|
|
(50, "Hunter"),
|
|
(150, "Slayer"),
|
|
(500, "Veteran"),
|
|
(1500, "Exterminator"),
|
|
(5000, "Legend"),
|
|
)
|
|
|
|
|
|
def rank_for(score: int) -> str:
|
|
rank = RANKS[0][1]
|
|
for threshold, name in RANKS:
|
|
if score < threshold:
|
|
break
|
|
rank = name
|
|
return rank
|
|
|
|
|
|
class LoadingScreenServer(ThreadingHTTPServer):
|
|
daemon_threads = True
|
|
allow_reuse_address = True
|
|
|
|
def __init__(self, address: tuple[str, int], database: Path, screenshots: Path):
|
|
self.database = database.resolve()
|
|
self.screenshots = screenshots.resolve()
|
|
super().__init__(address, LoadingScreenHandler)
|
|
|
|
|
|
class LoadingScreenHandler(BaseHTTPRequestHandler):
|
|
server_version = "Feedc0deCoopLoading/1.1"
|
|
|
|
def log_message(self, format_string: str, *args: object) -> None:
|
|
# Do not put joining players' Steam IDs into an access log.
|
|
if args and str(args[1]).startswith(("4", "5")):
|
|
super().log_message(format_string, *args)
|
|
|
|
def do_GET(self) -> None: # noqa: N802 - BaseHTTPRequestHandler API
|
|
request = urlsplit(self.path)
|
|
path = unquote(request.path)
|
|
|
|
if path in ("/", "/index.html"):
|
|
self.send_file(HERE / "index.html", "text/html; charset=utf-8", no_cache=True)
|
|
return
|
|
if path == "/assets/style.css":
|
|
self.send_file(HERE / "style.css", "text/css; charset=utf-8")
|
|
return
|
|
if path == "/assets/app.js":
|
|
self.send_file(HERE / "app.js", "application/javascript; charset=utf-8")
|
|
return
|
|
if path.startswith("/assets/screenshots/"):
|
|
self.serve_screenshot(path.rsplit("/", 1)[-1])
|
|
return
|
|
if path == "/api/player":
|
|
steam_id = parse_qs(request.query).get("steamid", [""])[0]
|
|
self.serve_player(steam_id)
|
|
return
|
|
if path == "/api/leaderboard":
|
|
self.serve_leaderboard()
|
|
return
|
|
if path == "/api/health":
|
|
self.send_json({"ok": True, "database": self.server.database.is_file()})
|
|
return
|
|
self.send_error_json(404, "Not found")
|
|
|
|
def do_HEAD(self) -> None: # noqa: N802 - BaseHTTPRequestHandler API
|
|
self.do_GET()
|
|
|
|
def serve_screenshot(self, name: str) -> None:
|
|
if not SCREENSHOT.fullmatch(name):
|
|
self.send_error_json(404, "Not found")
|
|
return
|
|
path = self.server.screenshots / name
|
|
if path.parent != self.server.screenshots or not path.is_file():
|
|
self.send_error_json(404, "Not found")
|
|
return
|
|
self.send_file(path, "image/jpeg")
|
|
|
|
def serve_player(self, steam_id: str) -> None:
|
|
if not STEAM_ID.fullmatch(steam_id):
|
|
self.send_error_json(400, "Invalid SteamID64")
|
|
return
|
|
if not self.server.database.is_file():
|
|
self.send_error_json(503, "Progression database unavailable")
|
|
return
|
|
|
|
try:
|
|
database_uri = f"file:{self.server.database.as_posix()}?mode=ro"
|
|
with sqlite3.connect(database_uri, uri=True, timeout=0.5) as connection:
|
|
connection.execute("PRAGMA query_only = ON")
|
|
connection.execute("PRAGMA busy_timeout = 500")
|
|
row = connection.execute(
|
|
"SELECT player_name, lifetime_score, credits, playtime_seconds "
|
|
"FROM custom_coop_players WHERE steamid64 = ? LIMIT 1",
|
|
(steam_id,),
|
|
).fetchone()
|
|
except sqlite3.Error:
|
|
self.send_error_json(503, "Progression record temporarily unavailable")
|
|
return
|
|
|
|
if row is None:
|
|
payload = {
|
|
"found": False,
|
|
"playerName": "New operative",
|
|
"score": 0,
|
|
"credits": 0,
|
|
"playtimeSeconds": 0,
|
|
"rank": "Newbie",
|
|
}
|
|
else:
|
|
score = max(int(row[1] or 0), 0)
|
|
payload = {
|
|
"found": True,
|
|
"playerName": str(row[0] or "Returning operative")[:64],
|
|
"score": score,
|
|
"credits": max(int(row[2] or 0), 0),
|
|
"playtimeSeconds": max(int(row[3] or 0), 0),
|
|
"rank": rank_for(score),
|
|
}
|
|
self.send_json(payload, no_cache=True)
|
|
|
|
def serve_leaderboard(self) -> None:
|
|
if not self.server.database.is_file():
|
|
self.send_error_json(503, "Progression database unavailable")
|
|
return
|
|
|
|
try:
|
|
database_uri = f"file:{self.server.database.as_posix()}?mode=ro"
|
|
with sqlite3.connect(database_uri, uri=True, timeout=0.5) as connection:
|
|
connection.execute("PRAGMA query_only = ON")
|
|
connection.execute("PRAGMA busy_timeout = 500")
|
|
rows = connection.execute(
|
|
"SELECT player_name, lifetime_score, playtime_seconds "
|
|
"FROM custom_coop_players "
|
|
"ORDER BY lifetime_score DESC, playtime_seconds DESC, player_name COLLATE NOCASE ASC "
|
|
"LIMIT 5"
|
|
).fetchall()
|
|
except sqlite3.Error:
|
|
self.send_error_json(503, "Leaderboard temporarily unavailable")
|
|
return
|
|
|
|
players = []
|
|
for position, row in enumerate(rows, start=1):
|
|
score = max(int(row[1] or 0), 0)
|
|
players.append(
|
|
{
|
|
"position": position,
|
|
"playerName": str(row[0] or "Unknown operative")[:64],
|
|
"score": score,
|
|
"playtimeSeconds": max(int(row[2] or 0), 0),
|
|
"rank": rank_for(score),
|
|
}
|
|
)
|
|
self.send_json({"players": players}, no_cache=True)
|
|
|
|
def send_file(self, path: Path, content_type: str, no_cache: bool = False) -> None:
|
|
try:
|
|
content = path.read_bytes()
|
|
except OSError:
|
|
self.send_error_json(404, "Not found")
|
|
return
|
|
self.send_response(200)
|
|
self.send_common_headers(no_cache=no_cache)
|
|
self.send_header("Content-Type", content_type)
|
|
self.send_header("Content-Length", str(len(content)))
|
|
self.end_headers()
|
|
if self.command != "HEAD":
|
|
self.wfile.write(content)
|
|
|
|
def send_json(self, payload: dict[str, object], status: int = 200, no_cache: bool = False) -> None:
|
|
content = json.dumps(payload, ensure_ascii=False, separators=(",", ":")).encode("utf-8")
|
|
self.send_response(status)
|
|
self.send_common_headers(no_cache=no_cache)
|
|
self.send_header("Content-Type", "application/json; charset=utf-8")
|
|
self.send_header("Content-Length", str(len(content)))
|
|
self.end_headers()
|
|
if self.command != "HEAD":
|
|
self.wfile.write(content)
|
|
|
|
def send_error_json(self, status: int, message: str) -> None:
|
|
self.send_json({"error": message}, status=status, no_cache=True)
|
|
|
|
def send_common_headers(self, no_cache: bool = False) -> None:
|
|
self.send_header("Cache-Control", "no-store" if no_cache else "public, max-age=3600")
|
|
self.send_header("Content-Security-Policy", "default-src 'self'; img-src 'self' data:; style-src 'self'; script-src 'self'; connect-src 'self'")
|
|
self.send_header("Referrer-Policy", "no-referrer")
|
|
self.send_header("X-Content-Type-Options", "nosniff")
|
|
|
|
|
|
def parse_args() -> argparse.Namespace:
|
|
parser = argparse.ArgumentParser(description=__doc__)
|
|
parser.add_argument("--bind", default="0.0.0.0", help="address to listen on (default: all IPv4 interfaces)")
|
|
parser.add_argument("--port", type=int, default=8765, help="HTTP port (default: 8765)")
|
|
parser.add_argument("--database", type=Path, default=DEFAULT_DATABASE, help="path to garrysmod/sv.db")
|
|
parser.add_argument("--screenshots", type=Path, default=DEFAULT_SCREENSHOTS, help="directory of slideshow JPEGs")
|
|
return parser.parse_args()
|
|
|
|
|
|
def main() -> int:
|
|
args = parse_args()
|
|
if not 1 <= args.port <= 65535:
|
|
print("loading screen: port must be between 1 and 65535", file=sys.stderr)
|
|
return 2
|
|
server = LoadingScreenServer((args.bind, args.port), args.database, args.screenshots)
|
|
print(f"loading screen: serving http://{args.bind}:{args.port}", flush=True)
|
|
try:
|
|
server.serve_forever(poll_interval=0.5)
|
|
except KeyboardInterrupt:
|
|
pass
|
|
finally:
|
|
server.server_close()
|
|
return 0
|
|
|
|
|
|
if __name__ == "__main__":
|
|
raise SystemExit(main())
|