Initial shared Stardew Valley tracker
This commit is contained in:
commit
a33f7a297d
14 changed files with 1171 additions and 0 deletions
231
app.py
Normal file
231
app.py
Normal file
|
|
@ -0,0 +1,231 @@
|
|||
#!/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 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 self.connect() as 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 self.connect() as 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 self.connect() as 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)))
|
||||
self.send_header("Cache-Control", "no-cache" if target.name == "index.html" 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()
|
||||
Loading…
Add table
Add a link
Reference in a new issue