goyban
39842c72d7
- Tasks get an optional plan date separate from the deadline. Unfinished plans roll over in Today; only missed deadlines count as overdue. - Postpone sheet on every task (tomorrow, weekend, next week, ...) that can move either the plan date or the deadline, with undo. - Backlog view for undated tasks, with suggestions in Today. - Projects with deadline, notes, progress, next step and "falling behind" warnings, shown in Today and in their own view. - Times and the calendar now follow the plan date (else the deadline). - Schema v3 migration (backup taken first); export format v3 includes projects and still imports older files. Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
707 lines
25 KiB
Python
707 lines
25 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 = 3
|
|
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(day, start: Optional[str], end: Optional[str]) -> Optional[str]:
|
|
"""Return an error message if the time fields don't make sense together.
|
|
|
|
`day` is the date the time applies to: the planned date if set, else the deadline.
|
|
"""
|
|
if (start or end) and not day:
|
|
return "A time needs a date (planned or deadline)"
|
|
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
|
|
project_id: Optional[int] = None
|
|
priority: int = Field(0, ge=0, le=3)
|
|
due_date: Optional[date] = None
|
|
planned_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.planned_date or 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
|
|
project_id: Optional[int] = None
|
|
priority: Optional[int] = Field(None, ge=0, le=3)
|
|
due_date: Optional[date] = None
|
|
planned_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 ProjectIn(BaseModel):
|
|
name: str = Field(max_length=100)
|
|
notes: str = Field("", max_length=20000)
|
|
category_id: Optional[int] = None
|
|
due_date: Optional[date] = None
|
|
done: bool = False
|
|
|
|
_name = field_validator("name")(clean_title)
|
|
|
|
|
|
class ProjectPatch(BaseModel):
|
|
name: Optional[str] = Field(None, max_length=100)
|
|
notes: Optional[str] = Field(None, max_length=20000)
|
|
category_id: Optional[int] = None
|
|
due_date: Optional[date] = None
|
|
done: Optional[bool] = None
|
|
|
|
_name = field_validator("name")(clean_title)
|
|
|
|
|
|
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 ExportProject(ProjectIn):
|
|
id: int
|
|
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] = []
|
|
projects: list[ExportProject] = []
|
|
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"],
|
|
"project_id": r["project_id"],
|
|
"priority": r["priority"],
|
|
"due_date": r["due_date"],
|
|
"planned_date": r["planned_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 ensure_project(conn: sqlite3.Connection, project_id: Optional[int]) -> None:
|
|
if project_id is not None and not conn.execute(
|
|
"SELECT 1 FROM projects WHERE id = ?", (project_id,)
|
|
).fetchone():
|
|
raise HTTPException(400, "Unknown project")
|
|
|
|
|
|
def iso(d) -> Optional[str]:
|
|
return d.isoformat() if isinstance(d, date) else d
|
|
|
|
|
|
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()
|
|
cur = conn.execute(
|
|
"""INSERT INTO tasks (title, notes, category_id, project_id, priority, due_date, planned_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("project_id"), t.get("priority", 0),
|
|
iso(t.get("due_date")), iso(t.get("planned_date")),
|
|
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)
|
|
ensure_project(conn, body.project_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 "project_id" in fields:
|
|
ensure_project(conn, fields["project_id"])
|
|
if "tags" in fields:
|
|
fields["tags"] = json.dumps(norm_tags(fields["tags"]))
|
|
for key in ("due_date", "planned_date"):
|
|
if key in fields:
|
|
fields[key] = iso(fields[key])
|
|
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", "planned_date", "start_time", "end_time")}
|
|
error = check_times(merged["planned_date"] or 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
|
|
planned = date.fromisoformat(task["planned_date"]) if task["planned_date"] else None
|
|
nxt = dict(task)
|
|
if due:
|
|
# The deadline drives the series; a plan date keeps its distance to the deadline.
|
|
nxt["due_date"] = next_due(due, task["recurrence"], task["recur_interval"], today)
|
|
nxt["planned_date"] = planned + (nxt["due_date"] - due) if planned else None
|
|
else:
|
|
# No deadline: repeat on plan dates, so the task never shows up as overdue.
|
|
nxt["planned_date"] = next_due(planned, 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)
|
|
|
|
|
|
# --- Project endpoints -------------------------------------------------------
|
|
|
|
def get_project(conn: sqlite3.Connection, project_id: int) -> dict:
|
|
row = conn.execute("SELECT * FROM projects WHERE id = ?", (project_id,)).fetchone()
|
|
if row is None:
|
|
raise HTTPException(404, "Project not found")
|
|
return {**dict(row), "done": bool(row["done"])}
|
|
|
|
|
|
def list_projects(conn: sqlite3.Connection) -> list[dict]:
|
|
return [{**dict(r), "done": bool(r["done"])} for r in conn.execute("SELECT * FROM projects ORDER BY id")]
|
|
|
|
|
|
def insert_project(conn: sqlite3.Connection, p: dict) -> int:
|
|
now = now_iso()
|
|
done = bool(p.get("done"))
|
|
cur = conn.execute(
|
|
"""INSERT INTO projects (name, notes, category_id, due_date, done, done_at, created_at, updated_at)
|
|
VALUES (?, ?, ?, ?, ?, ?, ?, ?)""",
|
|
(
|
|
p["name"], p.get("notes", ""), p.get("category_id"), iso(p.get("due_date")), int(done),
|
|
p.get("done_at") or (now if done else None), p.get("created_at") or now, p.get("updated_at") or now,
|
|
),
|
|
)
|
|
return cur.lastrowid
|
|
|
|
|
|
@app.get("/api/projects")
|
|
def get_projects(conn=Depends(get_conn)):
|
|
return list_projects(conn)
|
|
|
|
|
|
@app.post("/api/projects", status_code=201)
|
|
def create_project(body: ProjectIn, conn=Depends(get_conn)):
|
|
ensure_category(conn, body.category_id)
|
|
with conn:
|
|
project_id = insert_project(conn, body.model_dump())
|
|
return get_project(conn, project_id)
|
|
|
|
|
|
@app.patch("/api/projects/{project_id}")
|
|
def update_project(project_id: int, body: ProjectPatch, conn=Depends(get_conn)):
|
|
current = get_project(conn, project_id)
|
|
fields = body.model_dump(exclude_unset=True)
|
|
for key in ("name", "notes", "done"):
|
|
if key in fields and fields[key] is None:
|
|
del fields[key]
|
|
if "category_id" in fields:
|
|
ensure_category(conn, fields["category_id"])
|
|
if "due_date" in fields:
|
|
fields["due_date"] = iso(fields["due_date"])
|
|
if "done" in fields:
|
|
if fields["done"] and not current["done"]:
|
|
fields["done_at"] = now_iso()
|
|
elif not fields["done"]:
|
|
fields["done_at"] = None
|
|
fields["done"] = int(fields["done"])
|
|
fields["updated_at"] = now_iso()
|
|
with conn:
|
|
conn.execute(
|
|
f"UPDATE projects SET {', '.join(f'{k} = ?' for k in fields)} WHERE id = ?",
|
|
(*fields.values(), project_id),
|
|
)
|
|
return get_project(conn, project_id)
|
|
|
|
|
|
@app.delete("/api/projects/{project_id}", status_code=204)
|
|
def delete_project(project_id: int, conn=Depends(get_conn)):
|
|
"""Deletes the project only; its tasks are kept and simply no longer belong to a project."""
|
|
with conn:
|
|
conn.execute("DELETE FROM projects WHERE id = ?", (project_id,))
|
|
return Response(status_code=204)
|
|
|
|
|
|
# --- 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),
|
|
"projects": list_projects(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 projects")
|
|
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]
|
|
project_map: dict[int, int] = {}
|
|
for p in data.projects:
|
|
fields = p.model_dump()
|
|
fields["category_id"] = cat_map.get(p.category_id) if p.category_id is not None else None
|
|
project_map[p.id] = insert_project(conn, fields)
|
|
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
|
|
fields["project_id"] = project_map.get(t.project_id) if t.project_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), "projects": len(data.projects),
|
|
"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")
|