#!/usr/bin/env python3 """Small, dependency-free server for the shared Stardew tracker.""" from __future__ import annotations import argparse import json import logging import mimetypes import re import sqlite3 from contextlib import closing from datetime import datetime, timezone from http import HTTPStatus from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer from pathlib import Path from urllib.parse import unquote, urlparse APP_DIR = Path(__file__).resolve().parent STATIC_DIR = APP_DIR / "static" ITEM_ID = re.compile(r"^[a-z0-9][a-z0-9-]{0,79}$") MAX_BODY = 2048 def utc_now() -> str: return datetime.now(timezone.utc).isoformat(timespec="seconds") class Store: def __init__(self, path: str | Path): self.path = str(path) def connect(self) -> sqlite3.Connection: connection = sqlite3.connect(self.path, timeout=10) connection.row_factory = sqlite3.Row connection.execute("PRAGMA journal_mode=WAL") connection.execute("PRAGMA busy_timeout=10000") return connection def initialize(self) -> None: with closing(self.connect()) as db, db: db.executescript( """ CREATE TABLE IF NOT EXISTS checks ( item_id TEXT PRIMARY KEY, checked INTEGER NOT NULL CHECK (checked IN (0, 1)), updated_at TEXT NOT NULL, updated_by TEXT NOT NULL ); CREATE TABLE IF NOT EXISTS metadata ( key TEXT PRIMARY KEY, value TEXT NOT NULL ); INSERT OR IGNORE INTO metadata (key, value) VALUES ('revision', '0'); """ ) def snapshot(self) -> dict: with closing(self.connect()) as db, db: revision = int( db.execute("SELECT value FROM metadata WHERE key = 'revision'").fetchone()[0] ) rows = db.execute( "SELECT item_id, checked, updated_at, updated_by FROM checks ORDER BY item_id" ).fetchall() latest_row = db.execute( "SELECT updated_at, updated_by FROM checks ORDER BY updated_at DESC, rowid DESC LIMIT 1" ).fetchone() return { "revision": revision, "updatedAt": latest_row["updated_at"] if latest_row else None, "updatedBy": latest_row["updated_by"] if latest_row else None, "checks": {row["item_id"]: bool(row["checked"]) for row in rows}, } def set_check(self, item_id: str, checked: bool, user: str) -> dict: changed_at = utc_now() with closing(self.connect()) as db, db: db.execute("BEGIN IMMEDIATE") db.execute( """ INSERT INTO checks (item_id, checked, updated_at, updated_by) VALUES (?, ?, ?, ?) ON CONFLICT(item_id) DO UPDATE SET checked = excluded.checked, updated_at = excluded.updated_at, updated_by = excluded.updated_by """, (item_id, int(checked), changed_at, user), ) db.execute( "UPDATE metadata SET value = CAST(value AS INTEGER) + 1 WHERE key = 'revision'" ) revision = int( db.execute("SELECT value FROM metadata WHERE key = 'revision'").fetchone()[0] ) return { "itemId": item_id, "checked": checked, "revision": revision, "updatedAt": changed_at, "updatedBy": user, } class TrackerServer(ThreadingHTTPServer): daemon_threads = True def __init__(self, address: tuple[str, int], store: Store): super().__init__(address, Handler) self.store = store class Handler(BaseHTTPRequestHandler): server: TrackerServer def log_message(self, format: str, *args: object) -> None: logging.info("%s - %s", self.address_string(), format % args) def end_headers(self) -> None: self.send_header("X-Content-Type-Options", "nosniff") self.send_header("X-Frame-Options", "SAMEORIGIN") self.send_header("Referrer-Policy", "no-referrer") self.send_header("Permissions-Policy", "camera=(), microphone=(), geolocation=()") super().end_headers() def json_response(self, payload: dict, status: HTTPStatus = HTTPStatus.OK) -> None: body = json.dumps(payload, separators=(",", ":")).encode() 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 error_json(self, message: str, status: HTTPStatus) -> None: self.json_response({"error": message}, status) def authenticated_user(self) -> str: value = self.headers.get("Remote-User", "").strip() return value[:120] or "authelia-user" def do_GET(self) -> None: path = urlparse(self.path).path if path == "/health": self.json_response({"status": "ok"}) return if path == "/api/state": self.json_response(self.server.store.snapshot()) return self.serve_static(path) def do_POST(self) -> None: path = urlparse(self.path).path if path != "/api/checks": self.error_json("Not found", HTTPStatus.NOT_FOUND) return if self.headers.get_content_type() != "application/json": self.error_json("Expected application/json", HTTPStatus.UNSUPPORTED_MEDIA_TYPE) return try: length = int(self.headers.get("Content-Length", "0")) except ValueError: self.error_json("Invalid content length", HTTPStatus.BAD_REQUEST) return if length <= 0 or length > MAX_BODY: self.error_json("Invalid request size", HTTPStatus.BAD_REQUEST) return try: payload = json.loads(self.rfile.read(length)) except (json.JSONDecodeError, UnicodeDecodeError): self.error_json("Invalid JSON", HTTPStatus.BAD_REQUEST) return item_id = payload.get("itemId") if isinstance(payload, dict) else None checked = payload.get("checked") if isinstance(payload, dict) else None if not isinstance(item_id, str) or not ITEM_ID.fullmatch(item_id): self.error_json("Invalid itemId", HTTPStatus.BAD_REQUEST) return if not isinstance(checked, bool): self.error_json("checked must be boolean", HTTPStatus.BAD_REQUEST) return result = self.server.store.set_check(item_id, checked, self.authenticated_user()) self.json_response(result) def serve_static(self, request_path: str) -> None: relative = "index.html" if request_path == "/" else unquote(request_path.lstrip("/")) if not relative or ".." in Path(relative).parts: self.send_error(HTTPStatus.NOT_FOUND) return target = (STATIC_DIR / relative).resolve() try: target.relative_to(STATIC_DIR.resolve()) except ValueError: self.send_error(HTTPStatus.NOT_FOUND) return if not target.is_file(): self.send_error(HTTPStatus.NOT_FOUND) return body = target.read_bytes() content_type = mimetypes.guess_type(target.name)[0] or "application/octet-stream" self.send_response(HTTPStatus.OK) textual = content_type.startswith("text/") or content_type in { "application/javascript", "application/json", "image/svg+xml", } self.send_header("Content-Type", f"{content_type}; charset=utf-8" if textual else content_type) self.send_header("Content-Length", str(len(body))) # Source filenames are intentionally stable. Always revalidate them so # two household clients cannot run different interaction code for an # hour after a deployment. self.send_header("Cache-Control", "no-cache" if textual else "public, max-age=3600") self.end_headers() self.wfile.write(body) def main() -> None: parser = argparse.ArgumentParser(description="Shared two-year Stardew Valley tracker") parser.add_argument("--host", default="127.0.0.1") parser.add_argument("--port", type=int, default=8089) parser.add_argument("--database", default="./progress.sqlite3") args = parser.parse_args() logging.basicConfig(level=logging.INFO, format="%(asctime)s %(levelname)s %(message)s") Path(args.database).parent.mkdir(parents=True, exist_ok=True) store = Store(args.database) store.initialize() server = TrackerServer((args.host, args.port), store) logging.info("Listening on http://%s:%s", args.host, args.port) server.serve_forever() if __name__ == "__main__": main()