app/data/notebook.py

"""SQLite-backed notebook for saved analyses."""
from __future__ import annotations
import sqlite3
import json
from datetime import datetime
from app.config import NOTEBOOK_DB


def _conn():
    NOTEBOOK_DB.parent.mkdir(parents=True, exist_ok=True)
    c = sqlite3.connect(str(NOTEBOOK_DB))
    c.execute("""
        CREATE TABLE IF NOT EXISTS analyses (
            id INTEGER PRIMARY KEY AUTOINCREMENT,
            ts TEXT NOT NULL,
            title TEXT NOT NULL,
            kind TEXT NOT NULL,
            wave INTEGER,
            spec_json TEXT NOT NULL,
            result_json TEXT NOT NULL,
            notes TEXT
        )
    """)
    return c


def save_analysis(title: str, kind: str, wave: int, spec: dict, result: dict, notes: str = "") -> int:
    c = _conn()
    cur = c.execute(
        "INSERT INTO analyses (ts, title, kind, wave, spec_json, result_json, notes) VALUES (?,?,?,?,?,?,?)",
        (datetime.utcnow().isoformat(), title, kind, wave,
         json.dumps(spec, default=str), json.dumps(result, default=str), notes),
    )
    c.commit()
    aid = cur.lastrowid
    c.close()
    return aid


def list_analyses() -> list[dict]:
    c = _conn()
    rows = c.execute(
        "SELECT id, ts, title, kind, wave, notes FROM analyses ORDER BY ts DESC"
    ).fetchall()
    c.close()
    return [
        {"id": r[0], "ts": r[1], "title": r[2], "kind": r[3], "wave": r[4], "notes": r[5]}
        for r in rows
    ]


def get_analysis(aid: int) -> dict | None:
    c = _conn()
    row = c.execute("SELECT * FROM analyses WHERE id = ?", (aid,)).fetchone()
    c.close()
    if not row:
        return None
    return {
        "id": row[0], "ts": row[1], "title": row[2], "kind": row[3], "wave": row[4],
        "spec": json.loads(row[5]), "result": json.loads(row[6]), "notes": row[7],
    }


def delete_analysis(aid: int) -> None:
    c = _conn()
    c.execute("DELETE FROM analyses WHERE id = ?", (aid,))
    c.commit()
    c.close()