goyban
509b1a1320
Tasks can now have a start/end time on their deadline date. The new Calendar view shows a day as a timeline (with a week strip and projected repeats of recurring tasks), and the task form lists what is already planned that day and flags overlaps. Existing databases are migrated to schema v2 automatically, with a backup copy taken first. Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
565 lines
20 KiB
Python
565 lines
20 KiB
Python
import calendar
|
|
import hashlib
|
|
import hmac
|
|
import json
|
|
import os
|
|
import sqlite3
|
|
import time
|
|
from collections import defaultdict, deque
|
|
from contextlib import asynccontextmanager
|
|
from datetime import date, datetime, timedelta, timezone
|
|
from datetime import time as Time
|
|
from pathlib import Path
|
|
from typing import Literal, Optional
|
|
|
|
from fastapi import Depends, FastAPI, HTTPException, Request, Response
|
|
from fastapi.responses import JSONResponse
|
|
from fastapi.staticfiles import StaticFiles
|
|
from pydantic import BaseModel, Field, field_validator, model_validator
|
|
|
|
from . import db
|
|
|
|
# --- Configuration -----------------------------------------------------------
|
|
|
|
APP_PASSWORD = os.environ.get("APP_PASSWORD", "")
|
|
SECRET_KEY = os.environ.get("SECRET_KEY", "")
|
|
COOKIE_SECURE = os.environ.get("COOKIE_SECURE", "false").lower() in ("1", "true", "yes")
|
|
SESSION_DAYS = int(os.environ.get("SESSION_DAYS", "30"))
|
|
|
|
if not APP_PASSWORD or not SECRET_KEY or "change-me" in (APP_PASSWORD, SECRET_KEY):
|
|
raise SystemExit("APP_PASSWORD and SECRET_KEY must be set to real values (see .env.example)")
|
|
|
|
COOKIE_NAME = "whattodo_session"
|
|
CSRF_HEADER = "x-whattodo"
|
|
EXPORT_VERSION = 2
|
|
STATIC_DIR = Path(__file__).parent / "static"
|
|
|
|
# Changing the password (or secret) invalidates all existing sessions.
|
|
_SESSION_KEY = hashlib.sha256(f"{SECRET_KEY}\0{APP_PASSWORD}".encode()).digest()
|
|
|
|
|
|
@asynccontextmanager
|
|
async def lifespan(_app: FastAPI):
|
|
db.init()
|
|
yield
|
|
|
|
|
|
app = FastAPI(title="WhatToDo", lifespan=lifespan, docs_url=None, redoc_url=None, openapi_url=None)
|
|
|
|
|
|
# --- Sessions & security ------------------------------------------------------
|
|
|
|
def make_token() -> str:
|
|
ts = str(int(time.time()))
|
|
sig = hmac.new(_SESSION_KEY, ts.encode(), hashlib.sha256).hexdigest()
|
|
return f"{ts}.{sig}"
|
|
|
|
|
|
def valid_token(token: Optional[str]) -> bool:
|
|
if not token or "." not in token:
|
|
return False
|
|
ts, sig = token.split(".", 1)
|
|
expected = hmac.new(_SESSION_KEY, ts.encode(), hashlib.sha256).hexdigest()
|
|
if not hmac.compare_digest(sig, expected):
|
|
return False
|
|
try:
|
|
return time.time() - int(ts) < SESSION_DAYS * 86400
|
|
except ValueError:
|
|
return False
|
|
|
|
|
|
PUBLIC_API = {"/api/login", "/api/health"}
|
|
CSP = (
|
|
"default-src 'self'; script-src 'self'; style-src 'self'; img-src 'self' data:; "
|
|
"connect-src 'self'; manifest-src 'self'; frame-ancestors 'none'; base-uri 'none'; form-action 'self'"
|
|
)
|
|
|
|
|
|
@app.middleware("http")
|
|
async def guard(request: Request, call_next):
|
|
path = request.url.path
|
|
if path.startswith("/api/"):
|
|
if path not in PUBLIC_API and not valid_token(request.cookies.get(COOKIE_NAME)):
|
|
return JSONResponse({"detail": "Not logged in"}, status_code=401)
|
|
# Custom header on every write request: browsers can't send it cross-site without CORS.
|
|
if request.method not in ("GET", "HEAD", "OPTIONS") and request.headers.get(CSRF_HEADER) != "1":
|
|
return JSONResponse({"detail": "Missing X-WhatToDo header"}, status_code=403)
|
|
response = await call_next(request)
|
|
response.headers["X-Content-Type-Options"] = "nosniff"
|
|
response.headers["Referrer-Policy"] = "same-origin"
|
|
response.headers["X-Frame-Options"] = "DENY"
|
|
response.headers["Content-Security-Policy"] = CSP
|
|
if path.startswith("/api/"):
|
|
response.headers["Cache-Control"] = "no-store"
|
|
else:
|
|
response.headers.setdefault("Cache-Control", "no-cache")
|
|
return response
|
|
|
|
|
|
_failed_logins: dict[str, deque] = defaultdict(deque)
|
|
MAX_FAILS, FAIL_WINDOW = 10, 15 * 60
|
|
|
|
|
|
class LoginIn(BaseModel):
|
|
password: str
|
|
|
|
|
|
@app.post("/api/login")
|
|
def login(body: LoginIn, request: Request, response: Response):
|
|
ip = request.client.host if request.client else "unknown"
|
|
fails, now = _failed_logins[ip], time.time()
|
|
while fails and now - fails[0] > FAIL_WINDOW:
|
|
fails.popleft()
|
|
if len(fails) >= MAX_FAILS:
|
|
raise HTTPException(429, "Too many failed attempts, try again in a few minutes")
|
|
if not hmac.compare_digest(body.password.encode(), APP_PASSWORD.encode()):
|
|
fails.append(now)
|
|
time.sleep(1)
|
|
raise HTTPException(401, "Wrong password")
|
|
fails.clear()
|
|
response.set_cookie(
|
|
COOKIE_NAME, make_token(), max_age=SESSION_DAYS * 86400,
|
|
httponly=True, samesite="lax", secure=COOKIE_SECURE, path="/",
|
|
)
|
|
return {"ok": True}
|
|
|
|
|
|
@app.post("/api/logout")
|
|
def logout(response: Response):
|
|
response.delete_cookie(COOKIE_NAME, path="/")
|
|
return {"ok": True}
|
|
|
|
|
|
@app.get("/api/health")
|
|
def health():
|
|
return {"ok": True}
|
|
|
|
|
|
@app.get("/api/me")
|
|
def me():
|
|
return {"ok": True}
|
|
|
|
|
|
# --- Helpers -----------------------------------------------------------------
|
|
|
|
def get_conn():
|
|
conn = db.connect()
|
|
try:
|
|
yield conn
|
|
finally:
|
|
conn.close()
|
|
|
|
|
|
def now_iso() -> str:
|
|
return datetime.now(timezone.utc).isoformat(timespec="seconds")
|
|
|
|
|
|
def add_months(d: date, n: int) -> date:
|
|
m = d.month - 1 + n
|
|
y, m = d.year + m // 12, m % 12 + 1
|
|
return date(y, m, min(d.day, calendar.monthrange(y, m)[1]))
|
|
|
|
|
|
def step(d: date, recurrence: str, n: int) -> date:
|
|
if recurrence == "daily":
|
|
return d + timedelta(days=n)
|
|
if recurrence == "weekly":
|
|
return d + timedelta(weeks=n)
|
|
if recurrence == "monthly":
|
|
return add_months(d, n)
|
|
return add_months(d, 12 * n) # yearly
|
|
|
|
|
|
def next_due(due: Optional[date], recurrence: str, n: int, today: date) -> date:
|
|
"""Next occurrence after `today`, so a late completion doesn't create an already-overdue task."""
|
|
nxt = step(due or today, recurrence, n)
|
|
while nxt <= today:
|
|
nxt = step(nxt, recurrence, n)
|
|
return nxt
|
|
|
|
|
|
def norm_tags(tags: list[str]) -> list[str]:
|
|
out: list[str] = []
|
|
for t in tags:
|
|
t = t.strip().lstrip("#").strip().lower()
|
|
if t and len(t) <= 40 and t not in out:
|
|
out.append(t)
|
|
return out[:20]
|
|
|
|
|
|
def fmt_time(t) -> Optional[str]:
|
|
return t.strftime("%H:%M") if isinstance(t, Time) else t
|
|
|
|
|
|
def check_times(due, start: Optional[str], end: Optional[str]) -> Optional[str]:
|
|
"""Return an error message if the time fields don't make sense together."""
|
|
if (start or end) and not due:
|
|
return "A time needs a deadline date"
|
|
if end and not start:
|
|
return "An end time needs a start time"
|
|
if start and end and end <= start:
|
|
return "End time must be after start time"
|
|
return None
|
|
|
|
|
|
def clean_title(v: Optional[str]) -> Optional[str]:
|
|
if v is None:
|
|
return v
|
|
v = v.strip()
|
|
if not v:
|
|
raise ValueError("must not be empty")
|
|
return v
|
|
|
|
|
|
# --- Models ------------------------------------------------------------------
|
|
|
|
Recurrence = Literal["none", "daily", "weekly", "monthly", "yearly"]
|
|
HexColor = Field(pattern=r"^#[0-9a-fA-F]{6}$")
|
|
|
|
|
|
class SubtaskIn(BaseModel):
|
|
title: str = Field(min_length=1, max_length=500)
|
|
done: bool = False
|
|
|
|
|
|
class TaskIn(BaseModel):
|
|
title: str = Field(max_length=500)
|
|
notes: str = Field("", max_length=20000)
|
|
category_id: Optional[int] = None
|
|
priority: int = Field(0, ge=0, le=3)
|
|
due_date: Optional[date] = None
|
|
start_time: Optional[Time] = None
|
|
end_time: Optional[Time] = None
|
|
tags: list[str] = []
|
|
recurrence: Recurrence = "none"
|
|
recur_interval: int = Field(1, ge=1, le=365)
|
|
subtasks: list[SubtaskIn] = Field([], max_length=200)
|
|
|
|
_title = field_validator("title")(clean_title)
|
|
|
|
@model_validator(mode="after")
|
|
def _times(self):
|
|
error = check_times(self.due_date, fmt_time(self.start_time), fmt_time(self.end_time))
|
|
if error:
|
|
raise ValueError(error)
|
|
return self
|
|
|
|
|
|
class TaskPatch(BaseModel):
|
|
title: Optional[str] = Field(None, max_length=500)
|
|
notes: Optional[str] = Field(None, max_length=20000)
|
|
category_id: Optional[int] = None
|
|
priority: Optional[int] = Field(None, ge=0, le=3)
|
|
due_date: Optional[date] = None
|
|
start_time: Optional[Time] = None
|
|
end_time: Optional[Time] = None
|
|
tags: Optional[list[str]] = None
|
|
recurrence: Optional[Recurrence] = None
|
|
recur_interval: Optional[int] = Field(None, ge=1, le=365)
|
|
subtasks: Optional[list[SubtaskIn]] = Field(None, max_length=200)
|
|
|
|
_title = field_validator("title")(clean_title)
|
|
|
|
|
|
class CompleteIn(BaseModel):
|
|
today: Optional[date] = None # the browser's local date
|
|
|
|
|
|
class CategoryIn(BaseModel):
|
|
name: str = Field(min_length=1, max_length=50)
|
|
color: str = HexColor
|
|
position: Optional[int] = None
|
|
|
|
|
|
class CategoryPatch(BaseModel):
|
|
name: Optional[str] = Field(None, min_length=1, max_length=50)
|
|
color: Optional[str] = Field(None, pattern=r"^#[0-9a-fA-F]{6}$")
|
|
position: Optional[int] = None
|
|
|
|
|
|
class ExportCategory(BaseModel):
|
|
id: int
|
|
name: str = Field(min_length=1, max_length=50)
|
|
color: str = HexColor
|
|
position: int = 0
|
|
|
|
|
|
class ExportTask(TaskIn):
|
|
done: bool = False
|
|
done_at: Optional[str] = None
|
|
created_at: Optional[str] = None
|
|
updated_at: Optional[str] = None
|
|
|
|
|
|
class ExportData(BaseModel):
|
|
app: str
|
|
version: int
|
|
categories: list[ExportCategory] = []
|
|
tasks: list[ExportTask] = []
|
|
|
|
|
|
# --- Data access -------------------------------------------------------------
|
|
|
|
def row_to_task(r: sqlite3.Row, subtasks: list[dict]) -> dict:
|
|
return {
|
|
"id": r["id"],
|
|
"title": r["title"],
|
|
"notes": r["notes"],
|
|
"category_id": r["category_id"],
|
|
"priority": r["priority"],
|
|
"due_date": r["due_date"],
|
|
"start_time": r["start_time"],
|
|
"end_time": r["end_time"],
|
|
"tags": json.loads(r["tags"]),
|
|
"recurrence": r["recurrence"],
|
|
"recur_interval": r["recur_interval"],
|
|
"done": bool(r["done"]),
|
|
"done_at": r["done_at"],
|
|
"created_at": r["created_at"],
|
|
"updated_at": r["updated_at"],
|
|
"subtasks": subtasks,
|
|
}
|
|
|
|
|
|
def load_tasks(conn: sqlite3.Connection, task_id: Optional[int] = None) -> list[dict]:
|
|
where, params = ("WHERE id = ?", (task_id,)) if task_id is not None else ("", ())
|
|
sub_where = "WHERE task_id = ?" if task_id is not None else ""
|
|
subs: dict[int, list[dict]] = defaultdict(list)
|
|
for s in conn.execute(f"SELECT * FROM subtasks {sub_where} ORDER BY position, id", params):
|
|
subs[s["task_id"]].append({"id": s["id"], "title": s["title"], "done": bool(s["done"])})
|
|
rows = conn.execute(f"SELECT * FROM tasks {where} ORDER BY id", params).fetchall()
|
|
return [row_to_task(r, subs[r["id"]]) for r in rows]
|
|
|
|
|
|
def get_task(conn: sqlite3.Connection, task_id: int) -> dict:
|
|
tasks = load_tasks(conn, task_id)
|
|
if not tasks:
|
|
raise HTTPException(404, "Task not found")
|
|
return tasks[0]
|
|
|
|
|
|
def ensure_category(conn: sqlite3.Connection, category_id: Optional[int]) -> None:
|
|
if category_id is not None and not conn.execute(
|
|
"SELECT 1 FROM categories WHERE id = ?", (category_id,)
|
|
).fetchone():
|
|
raise HTTPException(400, "Unknown category")
|
|
|
|
|
|
def write_subtasks(conn: sqlite3.Connection, task_id: int, subtasks: list[dict]) -> None:
|
|
conn.execute("DELETE FROM subtasks WHERE task_id = ?", (task_id,))
|
|
for i, s in enumerate(subtasks):
|
|
conn.execute(
|
|
"INSERT INTO subtasks (task_id, title, done, position) VALUES (?, ?, ?, ?)",
|
|
(task_id, s["title"].strip(), int(s["done"]), i),
|
|
)
|
|
|
|
|
|
def insert_task(conn: sqlite3.Connection, t: dict, done: bool = False, done_at: Optional[str] = None,
|
|
created_at: Optional[str] = None, updated_at: Optional[str] = None) -> int:
|
|
now = now_iso()
|
|
due = t.get("due_date")
|
|
cur = conn.execute(
|
|
"""INSERT INTO tasks (title, notes, category_id, priority, due_date, start_time, end_time,
|
|
tags, recurrence, recur_interval, done, done_at, created_at, updated_at)
|
|
VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)""",
|
|
(
|
|
t["title"], t.get("notes", ""), t.get("category_id"), t.get("priority", 0),
|
|
due.isoformat() if isinstance(due, date) else due,
|
|
fmt_time(t.get("start_time")), fmt_time(t.get("end_time")),
|
|
json.dumps(norm_tags(t.get("tags", []))), t.get("recurrence", "none"),
|
|
t.get("recur_interval", 1), int(done), done_at, created_at or now, updated_at or now,
|
|
),
|
|
)
|
|
write_subtasks(conn, cur.lastrowid, t.get("subtasks", []))
|
|
return cur.lastrowid
|
|
|
|
|
|
# --- Category endpoints ------------------------------------------------------
|
|
|
|
def list_categories(conn: sqlite3.Connection) -> list[dict]:
|
|
return [dict(r) for r in conn.execute("SELECT * FROM categories ORDER BY position, id")]
|
|
|
|
|
|
@app.get("/api/categories")
|
|
def get_categories(conn=Depends(get_conn)):
|
|
return list_categories(conn)
|
|
|
|
|
|
@app.post("/api/categories", status_code=201)
|
|
def create_category(body: CategoryIn, conn=Depends(get_conn)):
|
|
pos = body.position
|
|
if pos is None:
|
|
pos = conn.execute("SELECT COALESCE(MAX(position), -1) + 1 FROM categories").fetchone()[0]
|
|
try:
|
|
with conn:
|
|
cur = conn.execute(
|
|
"INSERT INTO categories (name, color, position) VALUES (?, ?, ?)",
|
|
(body.name.strip(), body.color, pos),
|
|
)
|
|
except sqlite3.IntegrityError:
|
|
raise HTTPException(409, "A category with that name already exists")
|
|
return dict(conn.execute("SELECT * FROM categories WHERE id = ?", (cur.lastrowid,)).fetchone())
|
|
|
|
|
|
@app.patch("/api/categories/{category_id}")
|
|
def update_category(category_id: int, body: CategoryPatch, conn=Depends(get_conn)):
|
|
ensure_category(conn, category_id)
|
|
fields = body.model_dump(exclude_unset=True, exclude_none=True)
|
|
if "name" in fields:
|
|
fields["name"] = fields["name"].strip()
|
|
if fields:
|
|
try:
|
|
with conn:
|
|
conn.execute(
|
|
f"UPDATE categories SET {', '.join(f'{k} = ?' for k in fields)} WHERE id = ?",
|
|
(*fields.values(), category_id),
|
|
)
|
|
except sqlite3.IntegrityError:
|
|
raise HTTPException(409, "A category with that name already exists")
|
|
return dict(conn.execute("SELECT * FROM categories WHERE id = ?", (category_id,)).fetchone())
|
|
|
|
|
|
@app.delete("/api/categories/{category_id}", status_code=204)
|
|
def delete_category(category_id: int, conn=Depends(get_conn)):
|
|
with conn:
|
|
conn.execute("DELETE FROM categories WHERE id = ?", (category_id,))
|
|
return Response(status_code=204)
|
|
|
|
|
|
# --- Task endpoints ----------------------------------------------------------
|
|
|
|
@app.get("/api/tasks")
|
|
def get_tasks(conn=Depends(get_conn)):
|
|
return load_tasks(conn)
|
|
|
|
|
|
@app.post("/api/tasks", status_code=201)
|
|
def create_task(body: TaskIn, conn=Depends(get_conn)):
|
|
ensure_category(conn, body.category_id)
|
|
with conn:
|
|
task_id = insert_task(conn, body.model_dump())
|
|
return get_task(conn, task_id)
|
|
|
|
|
|
@app.patch("/api/tasks/{task_id}")
|
|
def update_task(task_id: int, body: TaskPatch, conn=Depends(get_conn)):
|
|
current = get_task(conn, task_id)
|
|
fields = body.model_dump(exclude_unset=True)
|
|
subtasks = fields.pop("subtasks", None)
|
|
for key in ("title", "notes", "priority", "tags", "recurrence", "recur_interval"):
|
|
if key in fields and fields[key] is None:
|
|
del fields[key] # these columns can't be cleared, only changed
|
|
if "category_id" in fields:
|
|
ensure_category(conn, fields["category_id"])
|
|
if "tags" in fields:
|
|
fields["tags"] = json.dumps(norm_tags(fields["tags"]))
|
|
if fields.get("due_date") is not None:
|
|
fields["due_date"] = fields["due_date"].isoformat()
|
|
for key in ("start_time", "end_time"):
|
|
if key in fields:
|
|
fields[key] = fmt_time(fields[key])
|
|
merged = {k: fields.get(k, current[k]) for k in ("due_date", "start_time", "end_time")}
|
|
error = check_times(merged["due_date"], merged["start_time"], merged["end_time"])
|
|
if error:
|
|
raise HTTPException(422, error)
|
|
fields["updated_at"] = now_iso()
|
|
with conn:
|
|
conn.execute(
|
|
f"UPDATE tasks SET {', '.join(f'{k} = ?' for k in fields)} WHERE id = ?",
|
|
(*fields.values(), task_id),
|
|
)
|
|
if subtasks is not None:
|
|
write_subtasks(conn, task_id, subtasks)
|
|
return get_task(conn, task_id)
|
|
|
|
|
|
@app.delete("/api/tasks/{task_id}", status_code=204)
|
|
def delete_task(task_id: int, conn=Depends(get_conn)):
|
|
with conn:
|
|
conn.execute("DELETE FROM tasks WHERE id = ?", (task_id,))
|
|
return Response(status_code=204)
|
|
|
|
|
|
@app.post("/api/tasks/{task_id}/complete")
|
|
def complete_task(task_id: int, body: CompleteIn, conn=Depends(get_conn)):
|
|
"""Mark done. A recurring task also spawns its next occurrence (with subtasks unchecked)."""
|
|
task = get_task(conn, task_id)
|
|
if task["done"]:
|
|
return {"task": task, "next": None}
|
|
today = body.today or date.today()
|
|
next_id = None
|
|
with conn:
|
|
now = now_iso()
|
|
conn.execute("UPDATE tasks SET done = 1, done_at = ?, updated_at = ? WHERE id = ?", (now, now, task_id))
|
|
if task["recurrence"] != "none":
|
|
due = date.fromisoformat(task["due_date"]) if task["due_date"] else None
|
|
nxt = dict(task)
|
|
nxt["due_date"] = next_due(due, task["recurrence"], task["recur_interval"], today)
|
|
nxt["subtasks"] = [{"title": s["title"], "done": False} for s in task["subtasks"]]
|
|
next_id = insert_task(conn, nxt)
|
|
return {"task": get_task(conn, task_id), "next": get_task(conn, next_id) if next_id else None}
|
|
|
|
|
|
@app.post("/api/tasks/{task_id}/reopen")
|
|
def reopen_task(task_id: int, conn=Depends(get_conn)):
|
|
get_task(conn, task_id)
|
|
with conn:
|
|
conn.execute(
|
|
"UPDATE tasks SET done = 0, done_at = NULL, updated_at = ? WHERE id = ?", (now_iso(), task_id)
|
|
)
|
|
return get_task(conn, task_id)
|
|
|
|
|
|
# --- Export / import ---------------------------------------------------------
|
|
|
|
@app.get("/api/export")
|
|
def export_data(conn=Depends(get_conn)):
|
|
stamp = datetime.now().strftime("%Y%m%d-%H%M")
|
|
data = {
|
|
"app": "whattodo",
|
|
"version": EXPORT_VERSION,
|
|
"exported_at": now_iso(),
|
|
"categories": list_categories(conn),
|
|
"tasks": load_tasks(conn),
|
|
}
|
|
return Response(
|
|
json.dumps(data, indent=2, ensure_ascii=False),
|
|
media_type="application/json",
|
|
headers={"Content-Disposition": f'attachment; filename="whattodo-export-{stamp}.json"'},
|
|
)
|
|
|
|
|
|
@app.post("/api/import")
|
|
def import_data(data: ExportData, mode: Literal["replace", "merge"] = "merge", conn=Depends(get_conn)):
|
|
"""replace: wipe everything and load the file. merge: add the file's tasks, matching categories by name."""
|
|
if data.app != "whattodo" or data.version > EXPORT_VERSION:
|
|
raise HTTPException(400, "This file is not a WhatToDo export (or is from a newer version)")
|
|
with conn:
|
|
if mode == "replace":
|
|
conn.execute("DELETE FROM subtasks")
|
|
conn.execute("DELETE FROM tasks")
|
|
conn.execute("DELETE FROM categories")
|
|
existing = {r["name"].lower(): r["id"] for r in conn.execute("SELECT id, name FROM categories")}
|
|
next_pos = conn.execute("SELECT COALESCE(MAX(position), -1) + 1 FROM categories").fetchone()[0]
|
|
cat_map: dict[int, int] = {}
|
|
for c in sorted(data.categories, key=lambda c: c.position):
|
|
key = c.name.strip().lower()
|
|
if key not in existing:
|
|
cur = conn.execute(
|
|
"INSERT INTO categories (name, color, position) VALUES (?, ?, ?)",
|
|
(c.name.strip(), c.color, next_pos),
|
|
)
|
|
existing[key] = cur.lastrowid
|
|
next_pos += 1
|
|
cat_map[c.id] = existing[key]
|
|
for t in data.tasks:
|
|
fields = t.model_dump()
|
|
fields["category_id"] = cat_map.get(t.category_id) if t.category_id is not None else None
|
|
insert_task(conn, fields, done=t.done, done_at=t.done_at,
|
|
created_at=t.created_at, updated_at=t.updated_at)
|
|
return {"ok": True, "mode": mode, "categories": len(data.categories), "tasks": len(data.tasks)}
|
|
|
|
|
|
# Static frontend (must be mounted last so it doesn't shadow /api routes).
|
|
app.mount("/", StaticFiles(directory=STATIC_DIR, html=True), name="static")
|