656 lines
30 KiB
Python
656 lines
30 KiB
Python
"""HTTP API under /api, and the built front end (static/) for everything else."""
|
|
import hashlib
|
|
import json
|
|
import re
|
|
import secrets
|
|
import sqlite3
|
|
import uuid
|
|
from contextlib import asynccontextmanager
|
|
from datetime import date, datetime, time, timedelta, timezone
|
|
from typing import Annotated, Literal
|
|
from zoneinfo import available_timezones
|
|
|
|
from fastapi import Depends, FastAPI, File, Form, HTTPException, Request, UploadFile
|
|
from fastapi.responses import FileResponse, JSONResponse, Response
|
|
from fastapi.staticfiles import StaticFiles
|
|
from pydantic import BaseModel, Field, field_validator
|
|
|
|
from parentcalendar import config, db, ics, scans
|
|
|
|
START_WORKER = True # tests turn the background extraction worker off
|
|
|
|
|
|
@asynccontextmanager
|
|
async def lifespan(app):
|
|
db.setup()
|
|
stop = scans.start_worker()[0] if START_WORKER else None
|
|
yield
|
|
if stop:
|
|
stop.set()
|
|
|
|
|
|
app = FastAPI(title="ParentCalendar", lifespan=lifespan)
|
|
|
|
|
|
@app.exception_handler(sqlite3.IntegrityError)
|
|
def integrity_error(request, exc): # unknown child, source or calendar id, and the like
|
|
return JSONResponse({"detail": f"invalid reference: {exc}"}, status_code=400)
|
|
|
|
|
|
def get_db():
|
|
conn = db.connect()
|
|
try:
|
|
yield conn
|
|
finally:
|
|
conn.close()
|
|
|
|
|
|
DB = Annotated[sqlite3.Connection, Depends(get_db)]
|
|
Name = Annotated[str, Field(min_length=1, max_length=80)]
|
|
Color = Annotated[str, Field(pattern=r"^#[0-9a-fA-F]{6}$")]
|
|
Ids = Annotated[list[int], Field(max_length=200)]
|
|
Kind = Literal["no_school", "early_release", "event", "game", "practice", "deadline", "pack_lunch"]
|
|
Month = Annotated[str, Field(pattern=r"^\d{4}-(0[1-9]|1[0-2])$")]
|
|
Day, Clock = date, time # models have fields named "date", which would hide the type
|
|
|
|
|
|
def found(row, what):
|
|
if row is None:
|
|
raise HTTPException(404, f"{what} not found")
|
|
return row
|
|
|
|
|
|
def set_children(conn, table, key, row_id, child_ids):
|
|
conn.execute(f"DELETE FROM {table} WHERE {key} = ?", (row_id,))
|
|
conn.executemany(f"INSERT INTO {table} ({key}, child_id) VALUES (?, ?)", [(row_id, c) for c in set(child_ids)])
|
|
|
|
|
|
CHILD_IDS = "(SELECT json_group_array(child_id) FROM {table} WHERE {key} = {alias}.id) AS child_ids"
|
|
|
|
|
|
# --- Household setup: user, children, calendars, sources ------------------------------------------------------------
|
|
|
|
@app.get("/api/state")
|
|
def state(conn: DB):
|
|
return {
|
|
"timezone": conn.execute("SELECT timezone FROM user WHERE id = 1").fetchone()[0],
|
|
"show_scan_stats": config.SHOW_SCAN_STATS,
|
|
"public_url": config.PUBLIC_URL,
|
|
"about": {"version": config.APP_VERSION, "commit": config.APP_COMMIT, "repo_url": config.REPO_URL},
|
|
"children": db.all_dicts(conn.execute("SELECT * FROM child ORDER BY id")),
|
|
"calendars": db.all_dicts(conn.execute(
|
|
f"SELECT c.*, {CHILD_IDS.format(table='calendar_child', key='calendar_id', alias='c')} FROM calendar c ORDER BY id")),
|
|
"sources": db.all_dicts(conn.execute(
|
|
f"SELECT s.*, {CHILD_IDS.format(table='source_child', key='source_id', alias='s')} FROM source s ORDER BY id")),
|
|
}
|
|
|
|
|
|
class UserIn(BaseModel):
|
|
timezone: str
|
|
|
|
@field_validator("timezone")
|
|
@classmethod
|
|
def known(cls, v):
|
|
if v not in available_timezones():
|
|
raise ValueError("unknown time zone")
|
|
return v
|
|
|
|
|
|
@app.put("/api/user")
|
|
def update_user(body: UserIn, conn: DB):
|
|
with conn:
|
|
conn.execute("UPDATE user SET timezone = ? WHERE id = 1", (body.timezone,))
|
|
return state(conn)
|
|
|
|
|
|
class ChildIn(BaseModel):
|
|
display_name: Name
|
|
color: Color = "#3b82f6"
|
|
|
|
|
|
@app.post("/api/children")
|
|
def add_child(body: ChildIn, conn: DB):
|
|
with conn:
|
|
conn.execute("INSERT INTO child (display_name, color) VALUES (?, ?)", (body.display_name.strip(), body.color))
|
|
return state(conn)
|
|
|
|
|
|
@app.put("/api/children/{child_id}")
|
|
def update_child(child_id: int, body: ChildIn, conn: DB):
|
|
with conn:
|
|
found(conn.execute("UPDATE child SET display_name = ?, color = ? WHERE id = ? RETURNING id",
|
|
(body.display_name.strip(), body.color, child_id)).fetchone(), "child")
|
|
return state(conn)
|
|
|
|
|
|
@app.delete("/api/children/{child_id}")
|
|
def delete_child(child_id: int, conn: DB):
|
|
with conn:
|
|
conn.execute("DELETE FROM child WHERE id = ?", (child_id,))
|
|
return state(conn)
|
|
|
|
|
|
class CalendarIn(BaseModel):
|
|
name: Name
|
|
color: Color = "#3b82f6"
|
|
child_ids: Ids = []
|
|
source_ids: Ids = [] # empty: every source
|
|
alarm_minutes: int | None = Field(None, ge=0, le=10080)
|
|
|
|
|
|
@app.post("/api/calendars")
|
|
def add_calendar(body: CalendarIn, conn: DB):
|
|
with conn:
|
|
cid = conn.execute("INSERT INTO calendar (name, color, source_ids, feed_token, alarm_minutes) VALUES (?, ?, ?, ?, ?)",
|
|
(body.name.strip(), body.color, json.dumps(body.source_ids), secrets.token_urlsafe(24),
|
|
body.alarm_minutes)).lastrowid
|
|
set_children(conn, "calendar_child", "calendar_id", cid, body.child_ids)
|
|
return state(conn)
|
|
|
|
|
|
@app.put("/api/calendars/{calendar_id}")
|
|
def update_calendar(calendar_id: int, body: CalendarIn, conn: DB):
|
|
with conn:
|
|
found(conn.execute("UPDATE calendar SET name = ?, color = ?, source_ids = ?, alarm_minutes = ? WHERE id = ? RETURNING id",
|
|
(body.name.strip(), body.color, json.dumps(body.source_ids), body.alarm_minutes, calendar_id)
|
|
).fetchone(), "calendar")
|
|
set_children(conn, "calendar_child", "calendar_id", calendar_id, body.child_ids)
|
|
return state(conn)
|
|
|
|
|
|
@app.delete("/api/calendars/{calendar_id}")
|
|
def delete_calendar(calendar_id: int, conn: DB):
|
|
with conn:
|
|
conn.execute("DELETE FROM calendar WHERE id = ?", (calendar_id,))
|
|
return state(conn)
|
|
|
|
|
|
class SourceIn(BaseModel):
|
|
name: Name
|
|
layout_type: Literal["month_grid", "menu", "list"] = "month_grid"
|
|
double_read: bool = False
|
|
ignore_labels: Annotated[list[Annotated[str, Field(min_length=1, max_length=40)]], Field(max_length=50)] = []
|
|
applies_filter: Annotated[str, Field(max_length=200)] | None = None
|
|
prompt_notes: Annotated[str, Field(max_length=2000)] | None = None
|
|
default_calendar_id: int | None = None
|
|
child_ids: Ids = []
|
|
|
|
|
|
def source_values(body):
|
|
return (body.name.strip(), body.layout_type, body.double_read, json.dumps(body.ignore_labels),
|
|
body.applies_filter or None, body.prompt_notes or None, body.default_calendar_id)
|
|
|
|
|
|
@app.post("/api/sources")
|
|
def add_source(body: SourceIn, conn: DB):
|
|
with conn:
|
|
sid = conn.execute("INSERT INTO source (name, layout_type, double_read, ignore_labels, applies_filter,"
|
|
" prompt_notes, default_calendar_id) VALUES (?, ?, ?, ?, ?, ?, ?)", source_values(body)).lastrowid
|
|
set_children(conn, "source_child", "source_id", sid, body.child_ids)
|
|
return state(conn)
|
|
|
|
|
|
@app.put("/api/sources/{source_id}")
|
|
def update_source(source_id: int, body: SourceIn, conn: DB):
|
|
with conn:
|
|
found(conn.execute("UPDATE source SET name = ?, layout_type = ?, double_read = ?, ignore_labels = ?,"
|
|
" applies_filter = ?, prompt_notes = ?, default_calendar_id = ? WHERE id = ? RETURNING id",
|
|
source_values(body) + (source_id,)).fetchone(), "source")
|
|
set_children(conn, "source_child", "source_id", source_id, body.child_ids)
|
|
return state(conn)
|
|
|
|
|
|
@app.delete("/api/sources/{source_id}")
|
|
def delete_source(source_id: int, conn: DB):
|
|
with conn:
|
|
conn.execute("DELETE FROM source WHERE id = ?", (source_id,))
|
|
return state(conn)
|
|
|
|
|
|
# --- Scans: upload, review, publish ----------------------------------------------------------------------------------
|
|
|
|
EVENT_SELECT = f"SELECT e.*, {CHILD_IDS.format(table='event_child', key='event_id', alias='e')} FROM event e"
|
|
|
|
|
|
def scan_detail(conn, scan_id):
|
|
scan = found(db.one(conn, "SELECT s.*, src.name AS source_name, src.layout_type FROM scan s"
|
|
" JOIN source src ON src.id = s.source_id WHERE s.id = ?", (scan_id,)), "scan")
|
|
scan["events"] = db.all_dicts(conn.execute(EVENT_SELECT + " WHERE e.scan_id = ? ORDER BY e.date, e.id", (scan_id,)))
|
|
old_ids = sorted({e["replaces_id"] for e in scan["events"] if e["replaces_id"]})
|
|
old = {r["id"]: r for r in db.all_dicts(conn.execute(
|
|
f"SELECT * FROM event WHERE id IN ({','.join('?' * len(old_ids))})", old_ids))} if old_ids else {}
|
|
for e in scan["events"]: # what an "updated" draft would change, for review
|
|
prev = old.get(e["replaces_id"])
|
|
e["replaces"] = {f: prev[f] for f in scans.MATCH_FIELDS} if prev else None
|
|
scan["menu_items"] = db.all_dicts(conn.execute("SELECT * FROM menu_item WHERE scan_id = ? ORDER BY date", (scan_id,)))
|
|
scan["can_undo"] = conn.execute(
|
|
"SELECT 1 FROM correction c JOIN event e ON e.id = c.event_id WHERE e.scan_id = ? AND c.undone = 0 LIMIT 1",
|
|
(scan_id,)).fetchone() is not None and scan["status"] != "published"
|
|
return scan
|
|
|
|
|
|
@app.post("/api/scans")
|
|
def upload_scan(conn: DB, source_id: Annotated[int, Form()], month: Annotated[Month, Form()],
|
|
files: Annotated[list[UploadFile], File()]):
|
|
found(db.one(conn, "SELECT id FROM source WHERE id = ?", (source_id,)), "source")
|
|
if not 1 <= len(files) <= 10:
|
|
raise HTTPException(400, "upload 1 to 10 files")
|
|
data = [f.file.read(config.MAX_UPLOAD_BYTES + 1) for f in files]
|
|
try:
|
|
scan_id = scans.save_upload(conn, source_id, month, data)
|
|
except scans.UploadError as e:
|
|
raise HTTPException(400, str(e)) from e
|
|
return scan_detail(conn, scan_id)
|
|
|
|
|
|
class PastedText(BaseModel):
|
|
source_id: int
|
|
month: Month
|
|
text: Annotated[str, Field(min_length=1, max_length=20000)]
|
|
|
|
|
|
@app.post("/api/scans/text")
|
|
def paste_scan(body: PastedText, conn: DB):
|
|
found(db.one(conn, "SELECT id FROM source WHERE id = ?", (body.source_id,)), "source")
|
|
text = body.text.strip()
|
|
if not text:
|
|
raise HTTPException(400, "paste some text")
|
|
return scan_detail(conn, scans.save_text(conn, body.source_id, body.month, text))
|
|
|
|
|
|
@app.get("/api/scans")
|
|
def list_scans(conn: DB):
|
|
return db.all_dicts(conn.execute(
|
|
"SELECT s.id, s.source_id, src.name AS source_name, s.hint_month, s.status, s.error, s.created_at,"
|
|
" s.published_at, s.seconds, (SELECT count(*) FROM event e WHERE e.scan_id = s.id) AS event_count"
|
|
" FROM scan s JOIN source src ON src.id = s.source_id ORDER BY s.id DESC LIMIT 50"))
|
|
|
|
|
|
@app.get("/api/scans/{scan_id}")
|
|
def get_scan(scan_id: int, conn: DB):
|
|
return scan_detail(conn, scan_id)
|
|
|
|
|
|
@app.get("/api/scans/{scan_id}/pages/{page}")
|
|
def scan_page(scan_id: int, page: int, conn: DB):
|
|
scan = found(db.one(conn, "SELECT image_keys FROM scan WHERE id = ?", (scan_id,)), "scan")
|
|
if not 1 <= page <= len(scan["image_keys"]):
|
|
raise HTTPException(404, "page not found (photos are deleted after the retention period)")
|
|
key = scan["image_keys"][page - 1]
|
|
path = config.IMAGES_DIR / scans.processed_key(key)
|
|
return FileResponse(path if path.exists() else config.IMAGES_DIR / key, media_type="image/jpeg")
|
|
|
|
|
|
def require_status(scan, *allowed):
|
|
if scan["status"] not in allowed:
|
|
raise HTTPException(409, f"scan is {scan['status']}")
|
|
|
|
|
|
@app.post("/api/scans/{scan_id}/retry")
|
|
def retry_scan(scan_id: int, conn: DB):
|
|
require_status(scan_detail(conn, scan_id), "failed")
|
|
with conn:
|
|
conn.execute("UPDATE scan SET status = 'queued', error = NULL WHERE id = ?", (scan_id,))
|
|
conn.execute("INSERT INTO job (scan_id) VALUES (?)", (scan_id,))
|
|
return scan_detail(conn, scan_id)
|
|
|
|
|
|
@app.post("/api/scans/{scan_id}/publish")
|
|
def publish_scan(scan_id: int, conn: DB):
|
|
require_status(scan_detail(conn, scan_id), "parsed", "reviewed")
|
|
scans.publish(conn, scan_id)
|
|
return scan_detail(conn, scan_id)
|
|
|
|
|
|
@app.delete("/api/scans/{scan_id}")
|
|
def abandon_scan(scan_id: int, conn: DB):
|
|
require_status(scan_detail(conn, scan_id), "queued", "parsed", "failed", "reviewed")
|
|
with conn:
|
|
conn.execute("DELETE FROM event WHERE scan_id = ?", (scan_id,)) # all drafts, since it was never published
|
|
conn.execute("DELETE FROM job WHERE scan_id = ? AND status = 'queued'", (scan_id,))
|
|
conn.execute("UPDATE scan SET status = 'abandoned' WHERE id = ?", (scan_id,))
|
|
return {"ok": True}
|
|
|
|
|
|
@app.post("/api/scans/{scan_id}/undo")
|
|
def undo(scan_id: int, conn: DB):
|
|
"""Revert the most recent edit or bulk action in this scan's review."""
|
|
require_status(scan_detail(conn, scan_id), "parsed", "reviewed")
|
|
last = conn.execute("SELECT c.batch FROM correction c JOIN event e ON e.id = c.event_id"
|
|
" WHERE e.scan_id = ? AND c.undone = 0 ORDER BY c.id DESC LIMIT 1", (scan_id,)).fetchone()
|
|
if last:
|
|
rows = db.all_dicts(conn.execute("SELECT * FROM correction WHERE batch = ? ORDER BY id DESC", (last[0],)))
|
|
with conn:
|
|
for c in rows:
|
|
write_field(conn, c["event_id"], c["field"], c["before"])
|
|
conn.execute("UPDATE correction SET undone = 1 WHERE batch = ?", (last[0],))
|
|
return scan_detail(conn, scan_id)
|
|
|
|
|
|
# --- Events: edit in review or on the calendar screen ----------------------------------------------------------------
|
|
|
|
class EventChanges(BaseModel):
|
|
"""Fields left out stay as they are; null clears end_date, times, notes and location."""
|
|
date: Day | None = None
|
|
end_date: Day | None = None
|
|
all_day: bool | None = None
|
|
start_time: Clock | None = None
|
|
end_time: Clock | None = None
|
|
title: Name | None = None
|
|
notes: Annotated[str, Field(max_length=2000)] | None = None
|
|
location: Annotated[str, Field(max_length=200)] | None = None
|
|
kind: Kind | None = None
|
|
include: bool | None = None
|
|
child_ids: Ids | None = None
|
|
|
|
|
|
class EventIn(EventChanges):
|
|
date: Day
|
|
title: Name
|
|
scan_id: int | None = None # adding during review: a draft in that scan
|
|
source_id: int | None = None
|
|
child_ids: Ids = []
|
|
|
|
|
|
EDITABLE = set(EventChanges.model_fields)
|
|
|
|
|
|
def stored(field, value):
|
|
if isinstance(value, date):
|
|
return value.isoformat()
|
|
if isinstance(value, time):
|
|
return value.strftime("%H:%M")
|
|
if field == "title" and value:
|
|
return value.strip()
|
|
return value
|
|
|
|
|
|
def write_field(conn, event_id, field, value):
|
|
if field == "child_ids":
|
|
set_children(conn, "event_child", "event_id", event_id, value or [])
|
|
elif field in EDITABLE or field == "status":
|
|
conn.execute(f"UPDATE event SET {field} = ? WHERE id = ?", (value, event_id))
|
|
|
|
|
|
def check_event(e):
|
|
if e["end_date"] and e["end_date"] < e["date"]:
|
|
raise HTTPException(422, "end date is before the start date")
|
|
if not e["all_day"] and not e["start_time"]:
|
|
raise HTTPException(422, "a timed event needs a start time")
|
|
if e["all_day"]:
|
|
e["start_time"] = e["end_time"] = None
|
|
elif e["end_time"] and e["start_time"] and (e["end_date"] or e["date"]) == e["date"] and e["end_time"] <= e["start_time"]:
|
|
raise HTTPException(422, "end time is before the start time")
|
|
|
|
|
|
def change_event(conn, event_id, changes, batch):
|
|
"""Apply changes, recording each as a correction (undo, and later few-shot examples).
|
|
A change to a confirmed event bumps its SEQUENCE so calendar apps take the update."""
|
|
event = found(db.one(conn, EVENT_SELECT + " WHERE e.id = ?", (event_id,)), "event")
|
|
if event["status"] == "deleted" and changes.get("status") != "draft":
|
|
raise HTTPException(409, "event is deleted")
|
|
after = dict(event) | {k: stored(k, v) for k, v in changes.items()}
|
|
check_event(after)
|
|
changed = [k for k in changes if k != "child_ids" and after[k] != event[k]]
|
|
changed += ["start_time", "end_time"] if after["all_day"] and (event["start_time"] or event["end_time"]) else []
|
|
if "child_ids" in changes and sorted(set(after["child_ids"])) != sorted(event["child_ids"]):
|
|
changed.append("child_ids")
|
|
for field in dict.fromkeys(changed):
|
|
conn.execute("INSERT INTO correction (event_id, batch, field, before, after) VALUES (?, ?, ?, ?, ?)",
|
|
(event_id, batch, field, json.dumps(event[field]), json.dumps(after[field])))
|
|
write_field(conn, event_id, field, after[field])
|
|
if changed and event["status"] == "confirmed":
|
|
conn.execute("UPDATE event SET sequence = sequence + 1, updated_at = strftime('%Y-%m-%d %H:%M:%f', 'now') WHERE id = ?", (event_id,))
|
|
if changed and event["scan_id"]:
|
|
conn.execute("UPDATE scan SET status = 'reviewed' WHERE id = ? AND status = 'parsed'", (event["scan_id"],))
|
|
|
|
|
|
def event_detail(conn, event_id):
|
|
return db.one(conn, EVENT_SELECT + " WHERE e.id = ?", (event_id,))
|
|
|
|
|
|
@app.patch("/api/events/{event_id}")
|
|
def edit_event(event_id: int, body: EventChanges, conn: DB):
|
|
with conn:
|
|
change_event(conn, event_id, body.model_dump(exclude_unset=True), uuid.uuid4().hex)
|
|
return event_detail(conn, event_id)
|
|
|
|
|
|
@app.delete("/api/events/{event_id}")
|
|
def delete_event(event_id: int, conn: DB):
|
|
with conn:
|
|
change_event(conn, event_id, {"status": "deleted"}, uuid.uuid4().hex)
|
|
return event_detail(conn, event_id)
|
|
|
|
|
|
@app.post("/api/events")
|
|
def add_event(body: EventIn, conn: DB):
|
|
values = {k: stored(k, v) for k, v in body.model_dump().items()}
|
|
values["all_day"] = values["all_day"] if values["all_day"] is not None else not values["start_time"]
|
|
values["kind"] = values["kind"] or "event"
|
|
check_event(values)
|
|
status = "confirmed"
|
|
if body.scan_id:
|
|
scan = found(db.one(conn, "SELECT status, source_id FROM scan WHERE id = ?", (body.scan_id,)), "scan")
|
|
require_status(scan, "parsed", "reviewed")
|
|
status, values["source_id"] = "draft", scan["source_id"]
|
|
with conn:
|
|
event_id = conn.execute(
|
|
"INSERT INTO event (uid, scan_id, source_id, date, end_date, all_day, start_time, end_time, title, notes,"
|
|
" location, kind, status, match) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)",
|
|
(scans.new_uid(), body.scan_id, values["source_id"], values["date"], values["end_date"], values["all_day"],
|
|
values["start_time"], values["end_time"], values["title"], values["notes"], values["location"],
|
|
values["kind"], status, "new" if body.scan_id else None)).lastrowid
|
|
set_children(conn, "event_child", "event_id", event_id, body.child_ids)
|
|
if body.scan_id: # undoing the add hides it again
|
|
conn.execute("INSERT INTO correction (event_id, batch, field, before, after) VALUES (?, ?, 'status', ?, ?)",
|
|
(event_id, uuid.uuid4().hex, json.dumps("deleted"), json.dumps("draft")))
|
|
return event_detail(conn, event_id)
|
|
|
|
|
|
class BulkIn(BaseModel):
|
|
ids: Annotated[list[int], Field(min_length=1, max_length=500)]
|
|
changes: EventChanges | None = None
|
|
delete: bool = False
|
|
|
|
|
|
@app.post("/api/events/bulk")
|
|
def bulk_edit(body: BulkIn, conn: DB):
|
|
"""One change (or delete) for many events, undone together."""
|
|
batch = uuid.uuid4().hex
|
|
changes = {"status": "deleted"} if body.delete else (body.changes.model_dump(exclude_unset=True) if body.changes else {})
|
|
with conn:
|
|
for event_id in body.ids:
|
|
change_event(conn, event_id, changes, batch)
|
|
return db.all_dicts(conn.execute(EVENT_SELECT + f" WHERE e.id IN ({','.join('?' * len(body.ids))})", body.ids))
|
|
|
|
|
|
def calendar_events(conn, calendar, statuses, start=None, end=None):
|
|
"""Events in a calendar's view: tagged with one of its children, from one of its sources (if it lists any)."""
|
|
sql = (EVENT_SELECT + f" WHERE e.status IN ({','.join('?' * len(statuses))}) AND EXISTS (SELECT 1 FROM event_child ec"
|
|
" JOIN calendar_child cc ON cc.child_id = ec.child_id WHERE ec.event_id = e.id AND cc.calendar_id = ?)")
|
|
args = [*statuses, calendar["id"]]
|
|
if calendar["source_ids"]:
|
|
sql += f" AND e.source_id IN ({','.join('?' * len(calendar['source_ids']))})"
|
|
args += calendar["source_ids"]
|
|
if start:
|
|
sql += " AND COALESCE(e.end_date, e.date) >= ?"
|
|
args.append(start)
|
|
if end:
|
|
sql += " AND e.date <= ?"
|
|
args.append(end)
|
|
return db.all_dicts(conn.execute(sql + " ORDER BY e.date, e.start_time, e.id", args))
|
|
|
|
|
|
@app.get("/api/events")
|
|
def list_events(calendar_id: int, conn: DB, start: date | None = None, end: date | None = None):
|
|
calendar = found(db.one(conn, "SELECT * FROM calendar WHERE id = ?", (calendar_id,)), "calendar")
|
|
return calendar_events(conn, calendar, ["confirmed"], start and start.isoformat(), end and end.isoformat())
|
|
|
|
|
|
class MenuItemIn(BaseModel):
|
|
items: Annotated[list[Annotated[str, Field(min_length=1, max_length=80)]], Field(max_length=20)]
|
|
no_lunch: bool
|
|
|
|
|
|
@app.patch("/api/menu_items/{item_id}")
|
|
def edit_menu_item(item_id: int, body: MenuItemIn, conn: DB):
|
|
"""Fix a misread menu day, in review or after publishing: menus never reach a calendar file, so the
|
|
lunch screen shows the change right away. The food is kept when no_lunch is set, so unticking it brings it back."""
|
|
items = [i.strip() for i in body.items if i.strip()]
|
|
with conn:
|
|
found(conn.execute("UPDATE menu_item SET items = ?, no_lunch = ?, flags = '[]' WHERE id = ? RETURNING id",
|
|
(json.dumps(items), body.no_lunch, item_id)).fetchone(), "menu item")
|
|
return db.one(conn, "SELECT * FROM menu_item WHERE id = ?", (item_id,))
|
|
|
|
|
|
# --- Lunch: each child picks the days to pack ------------------------------------------------------------------------
|
|
|
|
def month_bounds(month):
|
|
y, m = map(int, month.split("-"))
|
|
first = date(y, m, 1)
|
|
return first, (date(y + m // 12, m % 12 + 1, 1) - timedelta(days=1))
|
|
|
|
|
|
def lunch_days(conn, child_id, first, last):
|
|
"""Weekdays with a published menu for this child, with closures and picks."""
|
|
menus = {}
|
|
for m in db.all_dicts(conn.execute(
|
|
"SELECT m.*, src.name AS source_name FROM menu_item m JOIN scan s ON s.id = m.scan_id"
|
|
" JOIN source src ON src.id = m.source_id JOIN source_child sc ON sc.source_id = m.source_id"
|
|
" WHERE sc.child_id = ? AND src.layout_type = 'menu' AND s.status = 'published' AND m.date BETWEEN ? AND ?"
|
|
" ORDER BY s.published_at, s.id", (child_id, first.isoformat(), last.isoformat()))):
|
|
menus.setdefault(m["date"], {})[m["source_id"]] = m # a later scan of the same menu replaces the earlier one
|
|
closures = db.all_dicts(conn.execute(
|
|
"SELECT e.date, e.end_date, e.title FROM event e JOIN event_child ec ON ec.event_id = e.id"
|
|
" WHERE ec.child_id = ? AND e.kind = 'no_school' AND e.status = 'confirmed'"
|
|
" AND e.date <= ? AND COALESCE(e.end_date, e.date) >= ?", (child_id, last.isoformat(), first.isoformat())))
|
|
picks = {r[0] for r in conn.execute(
|
|
"SELECT e.date FROM event e JOIN event_child ec ON ec.event_id = e.id WHERE ec.child_id = ?"
|
|
" AND e.kind = 'pack_lunch' AND e.status = 'confirmed' AND e.date BETWEEN ? AND ?",
|
|
(child_id, first.isoformat(), last.isoformat()))}
|
|
days = []
|
|
for day, by_source in sorted(menus.items()):
|
|
closed = next((c["title"] for c in closures if c["date"] <= day <= (c["end_date"] or c["date"])), None)
|
|
no_lunch = all(m["no_lunch"] for m in by_source.values())
|
|
greyed = bool(closed) or no_lunch
|
|
days.append({"date": day, "menus": list(by_source.values()), "closed": closed, "no_lunch": no_lunch,
|
|
"greyed": greyed, "picked": day in picks,
|
|
"warning": "No school this day now" if greyed and day in picks else None})
|
|
return days
|
|
|
|
|
|
@app.get("/api/lunch")
|
|
def get_lunch(child_id: int, month: Month, conn: DB):
|
|
found(db.one(conn, "SELECT id FROM child WHERE id = ?", (child_id,)), "child")
|
|
return lunch_days(conn, child_id, *month_bounds(month))
|
|
|
|
|
|
class PickIn(BaseModel):
|
|
child_id: int
|
|
date: Day
|
|
picked: bool
|
|
|
|
|
|
class PicksIn(BaseModel):
|
|
child_id: int
|
|
dates: Annotated[list[Day], Field(min_length=1, max_length=31)]
|
|
picked: bool
|
|
|
|
|
|
def set_pick(conn, child, day, picked):
|
|
"""A pick is a 'Pack lunch' event for that child, 8:00-8:30: picking creates or restores it (same UID), unpicking deletes it.
|
|
The event's notes are the menu's first line (the main dish), so the calendar shows what's being skipped.
|
|
Returns False when picking a day with no school lunch (no menu, no school, or no lunch)."""
|
|
entry = next(iter(lunch_days(conn, child["id"], day, day)), None)
|
|
if picked and (entry is None or entry["greyed"]):
|
|
return False
|
|
existing = db.one(conn, "SELECT e.* FROM event e JOIN event_child ec ON ec.event_id = e.id WHERE ec.child_id = ?"
|
|
" AND e.kind = 'pack_lunch' AND e.date = ? ORDER BY e.id LIMIT 1", (child["id"], day.isoformat()))
|
|
status = "confirmed" if picked else "deleted"
|
|
notes = next((m["items"][0] for m in entry["menus"] if m["items"]), None) if entry else None
|
|
if existing and existing["status"] != status:
|
|
# Re-picking refreshes the notes from the menu, in case it was fixed since.
|
|
conn.execute("UPDATE event SET status = ?, notes = COALESCE(?, notes), sequence = sequence + 1,"
|
|
" updated_at = strftime('%Y-%m-%d %H:%M:%f', 'now') WHERE id = ?", (status, notes if picked else None, existing["id"]))
|
|
elif not existing and picked:
|
|
event_id = conn.execute(
|
|
"INSERT INTO event (uid, source_id, date, all_day, start_time, end_time, title, notes, kind, status)"
|
|
" VALUES (?, ?, ?, 0, '08:00', '08:30', ?, ?, 'pack_lunch', 'confirmed')", # ponytail: fixed time
|
|
(scans.new_uid(), entry["menus"][0]["source_id"], day.isoformat(), f"Pack lunch — {child['display_name']}", notes)).lastrowid
|
|
conn.execute("INSERT INTO event_child VALUES (?, ?)", (event_id, child["id"]))
|
|
return True
|
|
|
|
|
|
@app.post("/api/lunch")
|
|
def pick_lunch(body: PickIn, conn: DB):
|
|
child = found(db.one(conn, "SELECT * FROM child WHERE id = ?", (body.child_id,)), "child")
|
|
with conn:
|
|
if not set_pick(conn, child, body.date, body.picked):
|
|
raise HTTPException(409, "no school lunch to replace that day")
|
|
return lunch_days(conn, child["id"], *month_bounds(body.date.isoformat()[:7]))
|
|
|
|
|
|
@app.post("/api/lunch/bulk")
|
|
def pick_lunch_days(body: PicksIn, conn: DB):
|
|
"""Select all / clear all. Days with no school lunch are skipped when picking, not refused."""
|
|
child = found(db.one(conn, "SELECT * FROM child WHERE id = ?", (body.child_id,)), "child")
|
|
with conn: # all or nothing
|
|
for day in sorted(set(body.dates)):
|
|
set_pick(conn, child, day, body.picked)
|
|
return lunch_days(conn, child["id"], *month_bounds(min(body.dates).isoformat()[:7]))
|
|
|
|
|
|
# --- Export --------------------------------------------------------------------------------------------------------
|
|
|
|
@app.get("/api/calendars/{calendar_id}/export.ics")
|
|
def export_ics(calendar_id: int, conn: DB, everything: bool = False):
|
|
"""New or changed events since this calendar's last download (deletions included as CANCELLED),
|
|
or every event with everything=true."""
|
|
calendar = found(db.one(conn, "SELECT * FROM calendar WHERE id = ?", (calendar_id,)), "calendar")
|
|
tz = conn.execute("SELECT timezone FROM user WHERE id = 1").fetchone()[0]
|
|
cutoff = (datetime.now(timezone.utc) - timedelta(days=config.CANCELLED_DAYS)).strftime("%Y-%m-%d %H:%M:%S")
|
|
events = [e for e in calendar_events(conn, calendar, ["confirmed", "deleted"])
|
|
if e["status"] == "confirmed" or e["updated_at"] >= cutoff]
|
|
if not everything and calendar["last_exported_at"]:
|
|
events = [e for e in events if e["updated_at"] > calendar["last_exported_at"]]
|
|
else:
|
|
events = [e for e in events if e["status"] == "confirmed"] # nothing to cancel in a first import
|
|
body = ics.build(calendar["name"], events, tz, calendar["alarm_minutes"])
|
|
with conn:
|
|
conn.execute("UPDATE calendar SET last_exported_at = strftime('%Y-%m-%d %H:%M:%f', 'now') WHERE id = ?", (calendar_id,))
|
|
filename = re.sub(r"[^A-Za-z0-9-]+", "-", calendar["name"]).strip("-") or "calendar"
|
|
return Response(body, media_type="text/calendar; charset=utf-8",
|
|
headers={"Content-Disposition": f'attachment; filename="{filename}.ics"'})
|
|
|
|
|
|
@app.post("/api/calendars/{calendar_id}/feed_token")
|
|
def reset_feed_token(calendar_id: int, conn: DB):
|
|
"""A new feed link. The old one stops working, so everyone subscribed has to subscribe again."""
|
|
found(db.one(conn, "SELECT id FROM calendar WHERE id = ?", (calendar_id,)), "calendar")
|
|
with conn:
|
|
conn.execute("UPDATE calendar SET feed_token = ? WHERE id = ?", (secrets.token_urlsafe(24), calendar_id))
|
|
return state(conn)
|
|
|
|
|
|
@app.get("/feed/{token}.ics")
|
|
def feed(token: str, request: Request, conn: DB):
|
|
"""A calendar's subscription feed, the only URL that's public. The token in the path is the secret.
|
|
The past 12 months and everything ahead, with deletions as CANCELLED for CANCELLED_DAYS."""
|
|
calendar = found(db.one(conn, "SELECT * FROM calendar WHERE feed_token = ?", (token,)), "feed")
|
|
tz = conn.execute("SELECT timezone FROM user WHERE id = 1").fetchone()[0]
|
|
cutoff = (datetime.now(timezone.utc) - timedelta(days=config.CANCELLED_DAYS)).strftime("%Y-%m-%d %H:%M:%S")
|
|
start = (date.today() - timedelta(days=config.FEED_PAST_DAYS)).isoformat()
|
|
events = [e for e in calendar_events(conn, calendar, ["confirmed", "deleted"], start)
|
|
if e["status"] == "confirmed" or e["updated_at"] >= cutoff]
|
|
body = ics.build(calendar["name"], events, tz, calendar["alarm_minutes"], color=calendar["color"], refresh="PT1H")
|
|
etag = '"' + hashlib.sha256(body).hexdigest()[:32] + '"'
|
|
headers = {"ETag": etag, "Cache-Control": "private, max-age=900"}
|
|
if request.headers.get("if-none-match") == etag:
|
|
return Response(status_code=304, headers=headers)
|
|
return Response(body, media_type="text/calendar; charset=utf-8", headers=headers)
|
|
|
|
|
|
if config.STATIC_DIR.exists():
|
|
app.mount("/", StaticFiles(directory=config.STATIC_DIR, html=True), name="static")
|