312 lines
12 KiB
Python
312 lines
12 KiB
Python
#!/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
|
|
SEASONS = ("spring", "summer", "fall", "winter")
|
|
|
|
|
|
def utc_now() -> str:
|
|
return datetime.now(timezone.utc).isoformat(timespec="seconds")
|
|
|
|
|
|
def valid_game_date(value: object) -> bool:
|
|
def whole(number: object, low: int, high: int) -> bool:
|
|
return isinstance(number, int) and not isinstance(number, bool) and low <= number <= high
|
|
|
|
return (
|
|
isinstance(value, dict)
|
|
and whole(value.get("year"), 1, 2)
|
|
and value.get("season") in SEASONS
|
|
and whole(value.get("day"), 1, 28)
|
|
)
|
|
|
|
|
|
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');
|
|
-- A NULL date means the crop was planted before dates were recorded.
|
|
CREATE TABLE IF NOT EXISTS plantings (
|
|
item_id TEXT PRIMARY KEY,
|
|
year INTEGER,
|
|
season TEXT,
|
|
day INTEGER,
|
|
updated_at TEXT NOT NULL,
|
|
updated_by TEXT NOT NULL
|
|
);
|
|
"""
|
|
)
|
|
# Version 1.4 stored planting as an undated "planted-<crop>" check.
|
|
db.execute(
|
|
"""
|
|
INSERT OR IGNORE INTO plantings (item_id, updated_at, updated_by)
|
|
SELECT substr(item_id, 9), updated_at, updated_by FROM checks
|
|
WHERE item_id LIKE 'planted-%' AND checked = 1
|
|
"""
|
|
)
|
|
if db.execute("DELETE FROM checks WHERE item_id LIKE 'planted-%'").rowcount:
|
|
self._bump_revision(db)
|
|
|
|
@staticmethod
|
|
def _bump_revision(db: sqlite3.Connection) -> int:
|
|
db.execute("UPDATE metadata SET value = CAST(value AS INTEGER) + 1 WHERE key = 'revision'")
|
|
return int(db.execute("SELECT value FROM metadata WHERE key = 'revision'").fetchone()[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()
|
|
plantings = db.execute(
|
|
"SELECT item_id, year, season, day FROM plantings ORDER BY item_id"
|
|
).fetchall()
|
|
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},
|
|
"plantings": {
|
|
row["item_id"]: {"year": row["year"], "season": row["season"], "day": row["day"]}
|
|
for row in plantings
|
|
},
|
|
}
|
|
|
|
def set_planting(self, item_id: str, planted: dict | None, user: str) -> dict:
|
|
changed_at = utc_now()
|
|
with closing(self.connect()) as db, db:
|
|
db.execute("BEGIN IMMEDIATE")
|
|
if planted is None:
|
|
db.execute("DELETE FROM plantings WHERE item_id = ?", (item_id,))
|
|
else:
|
|
db.execute(
|
|
"""
|
|
INSERT INTO plantings (item_id, year, season, day, updated_at, updated_by)
|
|
VALUES (?, ?, ?, ?, ?, ?)
|
|
ON CONFLICT(item_id) DO UPDATE SET
|
|
year = excluded.year,
|
|
season = excluded.season,
|
|
day = excluded.day,
|
|
updated_at = excluded.updated_at,
|
|
updated_by = excluded.updated_by
|
|
""",
|
|
(item_id, planted["year"], planted["season"], planted["day"], changed_at, user),
|
|
)
|
|
revision = self._bump_revision(db)
|
|
return {
|
|
"itemId": item_id,
|
|
"planted": planted,
|
|
"revision": revision,
|
|
"updatedAt": changed_at,
|
|
"updatedBy": user,
|
|
}
|
|
|
|
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),
|
|
)
|
|
revision = self._bump_revision(db)
|
|
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 not in ("/api/checks", "/api/plantings"):
|
|
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
|
|
if not isinstance(item_id, str) or not ITEM_ID.fullmatch(item_id):
|
|
self.error_json("Invalid itemId", HTTPStatus.BAD_REQUEST)
|
|
return
|
|
if path == "/api/plantings":
|
|
planted = payload.get("planted")
|
|
if planted is not None and not valid_game_date(planted):
|
|
self.error_json("planted must be null or a {year, season, day} date", HTTPStatus.BAD_REQUEST)
|
|
return
|
|
if planted is not None:
|
|
planted = {key: planted[key] for key in ("year", "season", "day")}
|
|
self.json_response(self.server.store.set_planting(item_id, planted, self.authenticated_user()))
|
|
return
|
|
checked = payload.get("checked")
|
|
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()
|