421 lines
17 KiB
Python
421 lines
17 KiB
Python
#!/usr/bin/env python3
|
|
"""Small, dependency-free server for a private period tracker."""
|
|
|
|
from __future__ import annotations
|
|
|
|
import argparse
|
|
import csv
|
|
import io
|
|
import json
|
|
import logging
|
|
import mimetypes
|
|
import re
|
|
import sqlite3
|
|
from contextlib import closing
|
|
from datetime import date, datetime, timedelta, 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"
|
|
DAY_PATH = re.compile(r"^/api/days/(\d{4}-\d{2}-\d{2})$")
|
|
MAX_BODY = 64 * 1024
|
|
MAX_BATCH = 400
|
|
FLOWS = ("spotting", "period", "light", "medium", "heavy")
|
|
PERIOD_FLOWS = frozenset(("period", "light", "medium", "heavy"))
|
|
SYMPTOMS = (
|
|
("cramps", "Cramps"),
|
|
("headache", "Headache"),
|
|
("bloating", "Bloating"),
|
|
("tender-breasts", "Tender breasts"),
|
|
("acne", "Acne"),
|
|
("fatigue", "Fatigue"),
|
|
("back-pain", "Back pain"),
|
|
("nausea", "Nausea"),
|
|
("cravings", "Cravings"),
|
|
("mood-swings", "Mood swings"),
|
|
)
|
|
SYMPTOM_IDS = frozenset(symptom_id for symptom_id, _ in SYMPTOMS)
|
|
|
|
mimetypes.add_type("application/manifest+json", ".webmanifest")
|
|
|
|
|
|
def utc_now() -> str:
|
|
return datetime.now(timezone.utc).isoformat(timespec="seconds")
|
|
|
|
|
|
def parse_day(value: object) -> date | None:
|
|
"""A calendar date that is not in the future.
|
|
|
|
One day of slack covers a phone that is ahead of the server's timezone.
|
|
"""
|
|
if not isinstance(value, str) or not re.fullmatch(r"\d{4}-\d{2}-\d{2}", value):
|
|
return None
|
|
try:
|
|
parsed = date.fromisoformat(value)
|
|
except ValueError:
|
|
return None
|
|
if parsed.year < 1990 or parsed > date.today() + timedelta(days=1):
|
|
return None
|
|
return parsed
|
|
|
|
|
|
def clean_entry(value: object) -> dict | None:
|
|
"""Validate one day's log; returns None when it is malformed."""
|
|
if not isinstance(value, dict):
|
|
return None
|
|
flow = value.get("flow")
|
|
symptoms = value.get("symptoms", [])
|
|
if flow is not None and flow not in FLOWS:
|
|
return None
|
|
if not isinstance(symptoms, list) or not all(item in SYMPTOM_IDS for item in symptoms):
|
|
return None
|
|
return {"flow": flow, "symptoms": sorted(set(symptoms))}
|
|
|
|
|
|
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 days (
|
|
day TEXT PRIMARY KEY,
|
|
flow TEXT,
|
|
symptoms TEXT NOT NULL DEFAULT '[]',
|
|
updated_at TEXT NOT NULL
|
|
);
|
|
CREATE TABLE IF NOT EXISTS settings (
|
|
key TEXT PRIMARY KEY,
|
|
value TEXT NOT NULL
|
|
);
|
|
INSERT OR IGNORE INTO settings (key, value) VALUES ('share_details', 'false');
|
|
"""
|
|
)
|
|
|
|
def days(self) -> dict[str, dict]:
|
|
with closing(self.connect()) as db, db:
|
|
rows = db.execute("SELECT day, flow, symptoms FROM days ORDER BY day").fetchall()
|
|
return {row["day"]: {"flow": row["flow"], "symptoms": json.loads(row["symptoms"])} for row in rows}
|
|
|
|
def share_details(self) -> bool:
|
|
with closing(self.connect()) as db, db:
|
|
row = db.execute("SELECT value FROM settings WHERE key = 'share_details'").fetchone()
|
|
return row is not None and row["value"] == "true"
|
|
|
|
def set_share_details(self, enabled: bool) -> None:
|
|
with closing(self.connect()) as db, db:
|
|
db.execute(
|
|
"UPDATE settings SET value = ? WHERE key = 'share_details'",
|
|
("true" if enabled else "false",),
|
|
)
|
|
|
|
@staticmethod
|
|
def _write(db: sqlite3.Connection, day: str, entry: dict, changed_at: str) -> None:
|
|
# An empty log is the same as no log, so the row goes away.
|
|
if entry["flow"] is None and not entry["symptoms"]:
|
|
db.execute("DELETE FROM days WHERE day = ?", (day,))
|
|
return
|
|
db.execute(
|
|
"""
|
|
INSERT INTO days (day, flow, symptoms, updated_at) VALUES (?, ?, ?, ?)
|
|
ON CONFLICT(day) DO UPDATE SET
|
|
flow = excluded.flow,
|
|
symptoms = excluded.symptoms,
|
|
updated_at = excluded.updated_at
|
|
""",
|
|
(day, entry["flow"], json.dumps(entry["symptoms"]), changed_at),
|
|
)
|
|
|
|
def set_day(self, day: str, entry: dict) -> None:
|
|
with closing(self.connect()) as db, db:
|
|
self._write(db, day, entry, utc_now())
|
|
|
|
def fill_days(self, entries: dict[str, dict]) -> int:
|
|
"""Add logs for days that have none yet; existing logs always win."""
|
|
changed_at = utc_now()
|
|
added = 0
|
|
with closing(self.connect()) as db, db:
|
|
db.execute("BEGIN IMMEDIATE")
|
|
for day, entry in sorted(entries.items()):
|
|
if db.execute("SELECT 1 FROM days WHERE day = ?", (day,)).fetchone():
|
|
continue
|
|
self._write(db, day, entry, changed_at)
|
|
added += 1
|
|
return added
|
|
|
|
def backup(self, destination: Path, keep: int) -> Path:
|
|
"""Write today's consistent snapshot and prune all but the newest `keep`."""
|
|
destination.mkdir(parents=True, exist_ok=True)
|
|
target = destination / f"flowtrack-{date.today().isoformat()}.sqlite3"
|
|
partial = target.with_suffix(".partial")
|
|
with closing(self.connect()) as source, closing(sqlite3.connect(partial)) as copy:
|
|
source.backup(copy)
|
|
partial.replace(target)
|
|
snapshots = sorted(destination.glob("flowtrack-*.sqlite3"))
|
|
for old in snapshots[:-keep] if keep > 0 else []:
|
|
old.unlink()
|
|
return target
|
|
|
|
|
|
class TrackerServer(ThreadingHTTPServer):
|
|
daemon_threads = True
|
|
|
|
def __init__(self, address: tuple[str, int], store: Store, editors: set[str], viewers: set[str], dev_user: str | None = None):
|
|
super().__init__(address, Handler)
|
|
self.store = store
|
|
self.editors = editors
|
|
self.viewers = viewers
|
|
self.dev_user = dev_user
|
|
|
|
|
|
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", "DENY")
|
|
self.send_header("Referrer-Policy", "no-referrer")
|
|
self.send_header("Permissions-Policy", "camera=(), microphone=(), geolocation=()")
|
|
self.send_header(
|
|
"Content-Security-Policy",
|
|
"default-src 'self'; img-src 'self' data:; object-src 'none'; base-uri 'none'; frame-ancestors 'none'",
|
|
)
|
|
super().end_headers()
|
|
|
|
def send_body(self, body: bytes, content_type: str, status: HTTPStatus = HTTPStatus.OK, extra: dict | None = None) -> None:
|
|
self.send_response(status)
|
|
self.send_header("Content-Type", content_type)
|
|
self.send_header("Content-Length", str(len(body)))
|
|
self.send_header("Cache-Control", "no-store")
|
|
for name, value in (extra or {}).items():
|
|
self.send_header(name, value)
|
|
self.end_headers()
|
|
self.wfile.write(body)
|
|
|
|
def json_response(self, payload: dict, status: HTTPStatus = HTTPStatus.OK) -> None:
|
|
self.send_body(json.dumps(payload, separators=(",", ":")).encode(), "application/json; charset=utf-8", status)
|
|
|
|
def error_json(self, message: str, status: HTTPStatus) -> None:
|
|
self.json_response({"error": message}, status)
|
|
|
|
def role(self) -> tuple[str, str | None]:
|
|
# nginx overwrites Remote-User with the Authelia identity, so it can
|
|
# be trusted here; the app itself only listens on loopback.
|
|
user = self.headers.get("Remote-User", "").strip() or self.server.dev_user or ""
|
|
if user in self.server.editors:
|
|
return user, "editor"
|
|
if user in self.server.viewers:
|
|
return user, "viewer"
|
|
return user, None
|
|
|
|
def visible_days(self, role: str) -> dict[str, dict]:
|
|
days = self.server.store.days()
|
|
if role == "editor" or self.server.store.share_details():
|
|
return days
|
|
# Without sharing a viewer sees which days were period days, nothing else.
|
|
return {
|
|
day: {"flow": "period", "symptoms": []}
|
|
for day, entry in days.items()
|
|
if entry["flow"] in PERIOD_FLOWS
|
|
}
|
|
|
|
def read_json(self) -> object:
|
|
if self.headers.get_content_type() != "application/json":
|
|
self.error_json("Expected application/json", HTTPStatus.UNSUPPORTED_MEDIA_TYPE)
|
|
return None
|
|
try:
|
|
length = int(self.headers.get("Content-Length", "0"))
|
|
except ValueError:
|
|
length = -1
|
|
if length <= 0 or length > MAX_BODY:
|
|
self.error_json("Invalid request size", HTTPStatus.BAD_REQUEST)
|
|
return None
|
|
try:
|
|
return json.loads(self.rfile.read(length))
|
|
except (json.JSONDecodeError, UnicodeDecodeError):
|
|
self.error_json("Invalid JSON", HTTPStatus.BAD_REQUEST)
|
|
return None
|
|
|
|
def do_GET(self) -> None:
|
|
path = urlparse(self.path).path
|
|
if path == "/health":
|
|
self.json_response({"status": "ok"})
|
|
return
|
|
if not path.startswith("/api/"):
|
|
self.serve_static(path)
|
|
return
|
|
user, role = self.role()
|
|
if role is None:
|
|
self.error_json("Not allowed", HTTPStatus.FORBIDDEN)
|
|
return
|
|
if path == "/api/state":
|
|
self.json_response(
|
|
{
|
|
"user": user,
|
|
"role": role,
|
|
"shareDetails": self.server.store.share_details(),
|
|
"symptoms": [{"id": symptom_id, "label": label} for symptom_id, label in SYMPTOMS],
|
|
"days": self.visible_days(role),
|
|
}
|
|
)
|
|
return
|
|
if path in ("/api/export.json", "/api/export.csv"):
|
|
if role != "editor":
|
|
self.error_json("Only the owner can export", HTTPStatus.FORBIDDEN)
|
|
return
|
|
self.export(path.endswith(".csv"))
|
|
return
|
|
self.error_json("Not found", HTTPStatus.NOT_FOUND)
|
|
|
|
def do_PUT(self) -> None:
|
|
path = urlparse(self.path).path
|
|
_, role = self.role()
|
|
if role != "editor":
|
|
self.error_json("Read-only access", HTTPStatus.FORBIDDEN)
|
|
return
|
|
match = DAY_PATH.fullmatch(path)
|
|
if match:
|
|
if parse_day(match.group(1)) is None:
|
|
self.error_json("Invalid or future date", HTTPStatus.BAD_REQUEST)
|
|
return
|
|
payload = self.read_json()
|
|
if payload is None:
|
|
return
|
|
entry = clean_entry(payload)
|
|
if entry is None:
|
|
self.error_json("Invalid day log", HTTPStatus.BAD_REQUEST)
|
|
return
|
|
self.server.store.set_day(match.group(1), entry)
|
|
self.json_response({"day": match.group(1), **entry})
|
|
return
|
|
if path == "/api/settings":
|
|
payload = self.read_json()
|
|
if payload is None:
|
|
return
|
|
share = payload.get("shareDetails") if isinstance(payload, dict) else None
|
|
if not isinstance(share, bool):
|
|
self.error_json("shareDetails must be boolean", HTTPStatus.BAD_REQUEST)
|
|
return
|
|
self.server.store.set_share_details(share)
|
|
self.json_response({"shareDetails": share})
|
|
return
|
|
self.error_json("Not found", HTTPStatus.NOT_FOUND)
|
|
|
|
def do_POST(self) -> None:
|
|
path = urlparse(self.path).path
|
|
_, role = self.role()
|
|
if role != "editor":
|
|
self.error_json("Read-only access", HTTPStatus.FORBIDDEN)
|
|
return
|
|
if path != "/api/days":
|
|
self.error_json("Not found", HTTPStatus.NOT_FOUND)
|
|
return
|
|
payload = self.read_json()
|
|
if payload is None:
|
|
return
|
|
days = payload.get("days") if isinstance(payload, dict) else None
|
|
if not isinstance(days, dict) or not days or len(days) > MAX_BATCH:
|
|
self.error_json("days must be a non-empty object", HTTPStatus.BAD_REQUEST)
|
|
return
|
|
entries = {}
|
|
for day, value in days.items():
|
|
entry = clean_entry(value)
|
|
if parse_day(day) is None or entry is None:
|
|
self.error_json(f"Invalid log for {day}", HTTPStatus.BAD_REQUEST)
|
|
return
|
|
entries[day] = entry
|
|
self.json_response({"added": self.server.store.fill_days(entries)})
|
|
|
|
def export(self, as_csv: bool) -> None:
|
|
days = self.server.store.days()
|
|
stamp = date.today().isoformat()
|
|
if as_csv:
|
|
labels = dict(SYMPTOMS)
|
|
buffer = io.StringIO()
|
|
writer = csv.writer(buffer)
|
|
writer.writerow(["date", "flow", "symptoms"])
|
|
for day, entry in days.items():
|
|
writer.writerow([day, entry["flow"] or "", "; ".join(labels[item] for item in entry["symptoms"])])
|
|
body, content_type = buffer.getvalue().encode(), "text/csv; charset=utf-8"
|
|
else:
|
|
body = json.dumps({"exportedAt": utc_now(), "days": days}, indent=2).encode()
|
|
content_type = "application/json; charset=utf-8"
|
|
extension = "csv" if as_csv else "json"
|
|
self.send_body(body, content_type, extra={
|
|
"Content-Disposition": f'attachment; filename="flowtrack-{stamp}.{extension}"'
|
|
})
|
|
|
|
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",
|
|
"application/manifest+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)))
|
|
# Revalidate code on every load so a deploy reaches the home-screen app
|
|
# the next time it opens.
|
|
self.send_header("Cache-Control", "no-cache" if textual else "public, max-age=86400")
|
|
self.end_headers()
|
|
self.wfile.write(body)
|
|
|
|
|
|
def main() -> None:
|
|
parser = argparse.ArgumentParser(description="Private period tracker")
|
|
parser.add_argument("--host", default="127.0.0.1")
|
|
parser.add_argument("--port", type=int, default=8093)
|
|
parser.add_argument("--database", default="./flowtrack.sqlite3")
|
|
parser.add_argument("--editor", action="append", default=[], help="user who owns and edits the data (repeatable)")
|
|
parser.add_argument("--viewer", action="append", default=[], help="user with read-only access (repeatable)")
|
|
parser.add_argument("--dev-user", help="identity to assume when no Remote-User header is sent (local testing only)")
|
|
parser.add_argument("--backup-to", type=Path, help="write a snapshot of the database to this directory and exit")
|
|
parser.add_argument("--keep", type=int, default=30, help="snapshots to keep with --backup-to")
|
|
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()
|
|
if args.backup_to:
|
|
logging.info("Wrote %s", store.backup(args.backup_to, args.keep))
|
|
return
|
|
server = TrackerServer((args.host, args.port), store, set(args.editor), set(args.viewer), args.dev_user)
|
|
logging.info("Listening on http://%s:%s", args.host, args.port)
|
|
server.serve_forever()
|
|
|
|
|
|
if __name__ == "__main__":
|
|
main()
|