Initial commit: memo app with FastAPI, SQLite, APScheduler, Discord notifications

2026-04-30 14:03:11 +02:00
commit 748ee8ae75
11 changed files with 727 additions and 0 deletions
@@ -0,0 +1,39 @@
+import sqlite3
+import os
+from contextlib import contextmanager
+
+DB_PATH = os.environ.get("DB_PATH", "/data/memo.db")
+
+SCHEMA = """
+CREATE TABLE IF NOT EXISTS memos (
+    id INTEGER PRIMARY KEY AUTOINCREMENT,
+    title TEXT NOT NULL,
+    description TEXT DEFAULT '',
+    reminder_time TEXT NOT NULL DEFAULT '09:00',
+    created_at TEXT DEFAULT (datetime('now', 'localtime')),
+    done_at TEXT,
+    is_done INTEGER DEFAULT 0
+);
+
+CREATE TABLE IF NOT EXISTS settings (
+    key TEXT PRIMARY KEY,
+    value TEXT NOT NULL DEFAULT ''
+);
+
+INSERT OR IGNORE INTO settings (key, value) VALUES ('discord_webhook', '');
+INSERT OR IGNORE INTO settings (key, value) VALUES ('timezone', 'Europe/Paris');
+"""
+
+@contextmanager
+def get_db():
+    conn = sqlite3.connect(DB_PATH)
+    conn.row_factory = sqlite3.Row
+    try:
+        yield conn
+        conn.commit()
+    finally:
+        conn.close()
+
+def init_db():
+    with get_db() as conn:
+        conn.executescript(SCHEMA)
@@ -0,0 +1,28 @@
+import requests
+from .database import get_db
+
+def get_webhook_url() -> str:
+    with get_db() as conn:
+        row = conn.execute("SELECT value FROM settings WHERE key = 'discord_webhook'").fetchone()
+        return row["value"] if row else ""
+
+def send_reminder(memo: dict):
+    url = get_webhook_url()
+    if not url:
+        return
+
+    description = memo["description"] or ""
+    desc_line = f"\n> {description}" if description else ""
+
+    payload = {
+        "embeds": [{
+            "title": f"🔔 Rappel : {memo['title']}",
+            "description": f"{desc_line}\n📅 Créé le {memo['created_at'][:10]}",
+            "color": 0xf59e0b,
+            "footer": {"text": "memo.nas.percolouco.com"}
+        }]
+    }
+    try:
+        requests.post(url, json=payload, timeout=5)
+    except Exception:
+        pass
@@ -0,0 +1,130 @@
+from fastapi import FastAPI, HTTPException, Request
+from fastapi.responses import HTMLResponse
+from fastapi.staticfiles import StaticFiles
+from fastapi.templating import Jinja2Templates
+from pydantic import BaseModel
+from typing import Optional
+import os
+
+from .database import init_db, get_db
+from . import scheduler as sched
+
+app = FastAPI(title="Memo")
+
+BASE_DIR = os.path.dirname(os.path.dirname(__file__))
+app.mount("/static", StaticFiles(directory=os.path.join(BASE_DIR, "static")), name="static")
+templates = Jinja2Templates(directory=os.path.join(BASE_DIR, "templates"))
+
+@app.on_event("startup")
+def startup():
+    init_db()
+    sched.start()
+
+@app.on_event("shutdown")
+def shutdown():
+    sched.stop()
+
+@app.get("/", response_class=HTMLResponse)
+def index(request: Request):
+    return templates.TemplateResponse("index.html", {"request": request})
+
+
+# --- Memo models ---
+
+class MemoCreate(BaseModel):
+    title: str
+    description: Optional[str] = ""
+    reminder_time: str = "09:00"
+
+class MemoUpdate(BaseModel):
+    title: Optional[str] = None
+    description: Optional[str] = None
+    reminder_time: Optional[str] = None
+
+
+# --- Memo routes ---
+
+@app.get("/api/memos")
+def list_memos(done: Optional[str] = None):
+    with get_db() as conn:
+        if done == "true":
+            rows = conn.execute("SELECT * FROM memos WHERE is_done = 1 ORDER BY done_at DESC").fetchall()
+        elif done == "false":
+            rows = conn.execute("SELECT * FROM memos WHERE is_done = 0 ORDER BY reminder_time, created_at").fetchall()
+        else:
+            rows = conn.execute("SELECT * FROM memos ORDER BY is_done, reminder_time, created_at").fetchall()
+        return [dict(r) for r in rows]
+
+@app.post("/api/memos", status_code=201)
+def create_memo(memo: MemoCreate):
+    with get_db() as conn:
+        cur = conn.execute(
+            "INSERT INTO memos (title, description, reminder_time) VALUES (?, ?, ?)",
+            (memo.title.strip(), memo.description.strip(), memo.reminder_time)
+        )
+        row = conn.execute("SELECT * FROM memos WHERE id = ?", (cur.lastrowid,)).fetchone()
+        return dict(row)
+
+@app.put("/api/memos/{memo_id}")
+def update_memo(memo_id: int, memo: MemoUpdate):
+    with get_db() as conn:
+        row = conn.execute("SELECT * FROM memos WHERE id = ?", (memo_id,)).fetchone()
+        if not row:
+            raise HTTPException(status_code=404, detail="Memo introuvable")
+        fields, values = [], []
+        if memo.title is not None:
+            fields.append("title = ?"); values.append(memo.title.strip())
+        if memo.description is not None:
+            fields.append("description = ?"); values.append(memo.description.strip())
+        if memo.reminder_time is not None:
+            fields.append("reminder_time = ?"); values.append(memo.reminder_time)
+        if not fields:
+            return dict(row)
+        values.append(memo_id)
+        conn.execute(f"UPDATE memos SET {', '.join(fields)} WHERE id = ?", values)
+        return dict(conn.execute("SELECT * FROM memos WHERE id = ?", (memo_id,)).fetchone())
+
+@app.post("/api/memos/{memo_id}/done")
+def mark_done(memo_id: int):
+    with get_db() as conn:
+        row = conn.execute("SELECT * FROM memos WHERE id = ?", (memo_id,)).fetchone()
+        if not row:
+            raise HTTPException(status_code=404, detail="Memo introuvable")
+        conn.execute(
+            "UPDATE memos SET is_done = 1, done_at = datetime('now', 'localtime') WHERE id = ?",
+            (memo_id,)
+        )
+        return dict(conn.execute("SELECT * FROM memos WHERE id = ?", (memo_id,)).fetchone())
+
+@app.post("/api/memos/{memo_id}/undone")
+def mark_undone(memo_id: int):
+    with get_db() as conn:
+        row = conn.execute("SELECT * FROM memos WHERE id = ?", (memo_id,)).fetchone()
+        if not row:
+            raise HTTPException(status_code=404, detail="Memo introuvable")
+        conn.execute("UPDATE memos SET is_done = 0, done_at = NULL WHERE id = ?", (memo_id,))
+        return dict(conn.execute("SELECT * FROM memos WHERE id = ?", (memo_id,)).fetchone())
+
+@app.delete("/api/memos/{memo_id}", status_code=204)
+def delete_memo(memo_id: int):
+    with get_db() as conn:
+        conn.execute("DELETE FROM memos WHERE id = ?", (memo_id,))
+
+
+# --- Settings ---
+
+class SettingsUpdate(BaseModel):
+    discord_webhook: Optional[str] = None
+
+@app.get("/api/settings")
+def get_settings():
+    with get_db() as conn:
+        rows = conn.execute("SELECT key, value FROM settings").fetchall()
+        return {r["key"]: r["value"] for r in rows}
+
+@app.post("/api/settings")
+def update_settings(data: SettingsUpdate):
+    with get_db() as conn:
+        if data.discord_webhook is not None:
+            conn.execute("UPDATE settings SET value = ? WHERE key = 'discord_webhook'", (data.discord_webhook,))
+    return {"ok": True}
@@ -0,0 +1,24 @@
+from apscheduler.schedulers.background import BackgroundScheduler
+from apscheduler.triggers.cron import CronTrigger
+from datetime import datetime
+from .database import get_db
+from .discord_notify import send_reminder
+
+scheduler = BackgroundScheduler(timezone="Europe/Paris")
+
+def check_reminders():
+    now = datetime.now().strftime("%H:%M")
+    with get_db() as conn:
+        memos = conn.execute(
+            "SELECT * FROM memos WHERE is_done = 0 AND reminder_time = ?",
+            (now,)
+        ).fetchall()
+        for memo in memos:
+            send_reminder(dict(memo))
+
+def start():
+    scheduler.add_job(check_reminders, CronTrigger(minute="*"), id="check_reminders", replace_existing=True)
+    scheduler.start()
+
+def stop():
+    scheduler.shutdown()