484 lines
18 KiB
Python
484 lines
18 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 re
|
|
import sqlite3
|
|
import struct
|
|
import sys
|
|
import threading
|
|
import time
|
|
from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer
|
|
from pathlib import Path
|
|
from urllib.parse import parse_qs, unquote, urlencode, urlsplit
|
|
from urllib.request import Request, urlopen
|
|
|
|
|
|
HERE = Path(__file__).resolve().parent
|
|
DEFAULT_DATABASE = HERE.parents[2] / "sv.db"
|
|
DEFAULT_SCREENSHOTS = HERE.parent / "backgrounds"
|
|
DEFAULT_MAP_PREVIEWS = HERE / "map-previews"
|
|
DEFAULT_WORKSHOP_CACHE = HERE.parents[5] / "steam_cache/content/4000"
|
|
STEAM_ID = re.compile(r"^7656119\d{10}$")
|
|
SCREENSHOT = re.compile(r"^[a-z0-9_-]+\.jpg$", re.IGNORECASE)
|
|
MAP_PREVIEW = re.compile(r"^[a-z0-9_!+\[\]-]+\.jpg$", re.IGNORECASE)
|
|
RANKS = (
|
|
(0, "Newbie"),
|
|
(10, "Survivor"),
|
|
(50, "Hunter"),
|
|
(150, "Slayer"),
|
|
(500, "Veteran"),
|
|
(1500, "Exterminator"),
|
|
(5000, "Legend"),
|
|
)
|
|
MAX_GMA_STRING = 8 * 1024 * 1024
|
|
MAX_PREVIEW_BYTES = 5 * 1024 * 1024
|
|
WORKSHOP_DETAILS_URL = (
|
|
"https://api.steampowered.com/ISteamRemoteStorage/"
|
|
"GetPublishedFileDetails/v1/"
|
|
)
|
|
|
|
|
|
class WorkshopPreviewIndex:
|
|
"""Discover embedded thumbnails and Workshop IDs from cached GMAs."""
|
|
|
|
def __init__(self, root: Path, refresh_seconds: int = 300):
|
|
self.root = root.resolve()
|
|
self.refresh_seconds = refresh_seconds
|
|
self.sources: dict[str, tuple[Path, int | None, int, str, str]] = {}
|
|
self.preview_cache: dict[str, tuple[bytes, str]] = {}
|
|
self.lock = threading.Lock()
|
|
self.ready = threading.Event()
|
|
self.stop_event = threading.Event()
|
|
self.last_error = ""
|
|
self.last_scan = 0.0
|
|
self.thread = threading.Thread(
|
|
target=self._scan_loop,
|
|
name="workshop-preview-index",
|
|
daemon=True,
|
|
)
|
|
|
|
def start(self) -> None:
|
|
self.thread.start()
|
|
|
|
def stop(self) -> None:
|
|
self.stop_event.set()
|
|
|
|
@staticmethod
|
|
def _read_exact(source, size: int) -> bytes:
|
|
content = source.read(size)
|
|
if len(content) != size:
|
|
raise ValueError("truncated GMA")
|
|
return content
|
|
|
|
@staticmethod
|
|
def _read_cstring(source) -> str:
|
|
content = bytearray()
|
|
while len(content) <= MAX_GMA_STRING:
|
|
character = source.read(1)
|
|
if not character:
|
|
raise ValueError("unterminated GMA string")
|
|
if character == b"\0":
|
|
return content.decode("utf-8", errors="replace")
|
|
content.extend(character)
|
|
raise ValueError("oversized GMA string")
|
|
|
|
@classmethod
|
|
def inspect_gma(cls, path: Path) -> dict[str, tuple[Path, int | None, int, str, str]]:
|
|
with path.open("rb") as source:
|
|
if cls._read_exact(source, 4) != b"GMAD":
|
|
raise ValueError("invalid GMA magic")
|
|
cls._read_exact(source, 1 + 8 + 8)
|
|
for _ in range(4):
|
|
cls._read_cstring(source)
|
|
cls._read_exact(source, 4)
|
|
|
|
entries: list[tuple[str, int]] = []
|
|
while True:
|
|
number = struct.unpack("<I", cls._read_exact(source, 4))[0]
|
|
if number == 0:
|
|
break
|
|
name = cls._read_cstring(source).replace("\\", "/").lower()
|
|
size = struct.unpack("<Q", cls._read_exact(source, 8))[0]
|
|
cls._read_exact(source, 4)
|
|
entries.append((name, size))
|
|
|
|
data_offset = source.tell()
|
|
|
|
workshop_id = path.parent.name if path.parent.name.isdigit() else path.stem
|
|
if not workshop_id.isdigit():
|
|
workshop_id = ""
|
|
offsets: dict[str, tuple[int, int]] = {}
|
|
offset = data_offset
|
|
maps = []
|
|
for name, size in entries:
|
|
offsets[name] = (offset, size)
|
|
match = re.fullmatch(r"maps/([^/]+)\.bsp", name, re.IGNORECASE)
|
|
if match:
|
|
maps.append(match.group(1).lower())
|
|
offset += size
|
|
|
|
result = {}
|
|
for map_name in maps:
|
|
preview = None
|
|
content_type = ""
|
|
for extension, candidate_type in (("png", "image/png"), ("jpg", "image/jpeg"), ("jpeg", "image/jpeg")):
|
|
candidate = offsets.get(f"maps/thumb/{map_name}.{extension}")
|
|
if candidate:
|
|
preview = candidate
|
|
content_type = candidate_type
|
|
break
|
|
if preview:
|
|
result[map_name] = (path, preview[0], preview[1], content_type, workshop_id)
|
|
elif workshop_id:
|
|
result[map_name] = (path, None, 0, "", workshop_id)
|
|
return result
|
|
|
|
def scan(self) -> None:
|
|
discovered = {}
|
|
errors = 0
|
|
if self.root.is_dir():
|
|
for path in sorted(self.root.glob("*/*.gma")):
|
|
try:
|
|
for map_name, source in self.inspect_gma(path).items():
|
|
current = discovered.get(map_name)
|
|
if current is None or (current[1] is None and source[1] is not None):
|
|
discovered[map_name] = source
|
|
except (OSError, ValueError, struct.error):
|
|
errors += 1
|
|
with self.lock:
|
|
self.sources = discovered
|
|
self.last_error = f"{errors} unreadable GMA(s)" if errors else ""
|
|
self.last_scan = time.time()
|
|
self.ready.set()
|
|
|
|
def _scan_loop(self) -> None:
|
|
while not self.stop_event.is_set():
|
|
self.scan()
|
|
self.stop_event.wait(self.refresh_seconds)
|
|
|
|
@staticmethod
|
|
def _download_workshop_preview(workshop_id: str) -> tuple[bytes, str] | None:
|
|
request = Request(
|
|
WORKSHOP_DETAILS_URL,
|
|
data=urlencode({"itemcount": "1", "publishedfileids[0]": workshop_id}).encode("ascii"),
|
|
headers={"User-Agent": "feedc0de-coop-map-preview/1"},
|
|
)
|
|
with urlopen(request, timeout=10) as response:
|
|
details = json.load(response)["response"]["publishedfiledetails"][0]
|
|
preview_url = str(details.get("preview_url") or "")
|
|
if details.get("result") != 1 or not preview_url.startswith(("https://", "http://")):
|
|
return None
|
|
with urlopen(Request(preview_url, headers={"User-Agent": "feedc0de-coop-map-preview/1"}), timeout=15) as response:
|
|
content = response.read(MAX_PREVIEW_BYTES + 1)
|
|
content_type = response.headers.get_content_type()
|
|
if not content or len(content) > MAX_PREVIEW_BYTES or not content_type.startswith("image/"):
|
|
return None
|
|
return content, content_type
|
|
|
|
def get(self, map_name: str) -> tuple[bytes, str] | None:
|
|
self.ready.wait(timeout=15)
|
|
with self.lock:
|
|
source = self.sources.get(map_name.lower())
|
|
if source is None:
|
|
return None
|
|
path, offset, size, content_type, workshop_id = source
|
|
if offset is not None:
|
|
if size <= 0 or size > MAX_PREVIEW_BYTES:
|
|
return None
|
|
try:
|
|
with path.open("rb") as gma:
|
|
gma.seek(offset)
|
|
content = gma.read(size)
|
|
except OSError:
|
|
return None
|
|
return (content, content_type) if len(content) == size else None
|
|
if not workshop_id:
|
|
return None
|
|
with self.lock:
|
|
cached = self.preview_cache.get(workshop_id)
|
|
if cached:
|
|
return cached
|
|
try:
|
|
downloaded = self._download_workshop_preview(workshop_id)
|
|
except (OSError, ValueError, KeyError, json.JSONDecodeError):
|
|
return None
|
|
if downloaded:
|
|
with self.lock:
|
|
self.preview_cache[workshop_id] = downloaded
|
|
return downloaded
|
|
|
|
def health(self) -> dict[str, object]:
|
|
with self.lock:
|
|
return {
|
|
"root": self.root.is_dir(),
|
|
"maps": len(self.sources),
|
|
"scanComplete": self.ready.is_set(),
|
|
"lastError": self.last_error,
|
|
}
|
|
|
|
|
|
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,
|
|
map_previews: Path,
|
|
workshop_cache: Path,
|
|
):
|
|
self.database = database.resolve()
|
|
self.screenshots = screenshots.resolve()
|
|
self.map_previews = map_previews.resolve()
|
|
self.workshop_previews = WorkshopPreviewIndex(workshop_cache)
|
|
super().__init__(address, LoadingScreenHandler)
|
|
self.workshop_previews.start()
|
|
|
|
def server_close(self) -> None:
|
|
self.workshop_previews.stop()
|
|
super().server_close()
|
|
|
|
|
|
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.startswith("/assets/map-previews/"):
|
|
self.serve_map_preview(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(),
|
|
"mapPreviews": self.server.map_previews.is_dir(),
|
|
"workshopPreviews": self.server.workshop_previews.health(),
|
|
}
|
|
)
|
|
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_map_preview(self, name: str) -> None:
|
|
if not MAP_PREVIEW.fullmatch(name):
|
|
self.send_error_json(404, "Not found")
|
|
return
|
|
path = self.server.map_previews / name
|
|
if path.parent == self.server.map_previews and path.is_file():
|
|
self.send_file(path, "image/jpeg")
|
|
return
|
|
preview = self.server.workshop_previews.get(Path(name).stem)
|
|
if not preview:
|
|
self.send_error_json(404, "Not found")
|
|
return
|
|
self.send_bytes(*preview)
|
|
|
|
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_bytes(content, content_type, no_cache=no_cache)
|
|
|
|
def send_bytes(self, content: bytes, content_type: str, no_cache: bool = False) -> None:
|
|
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")
|
|
parser.add_argument("--map-previews", type=Path, default=DEFAULT_MAP_PREVIEWS, help="directory of map-vote JPEGs")
|
|
parser.add_argument(
|
|
"--workshop-cache",
|
|
type=Path,
|
|
default=DEFAULT_WORKSHOP_CACHE,
|
|
help="Workshop content/4000 directory scanned for map previews",
|
|
)
|
|
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,
|
|
args.map_previews,
|
|
args.workshop_cache,
|
|
)
|
|
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())
|