stardew-tracker/app.py

#!/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()