web-page-backend/tarot-lab/app/db.py

"""tarot.db SQLite — 5 CRUD + _tarot_row_to_dict + init_db."""
import json
import os
import sqlite3
from typing import Any, Dict, Optional

from .config import DB_PATH


def _conn() -> sqlite3.Connection:
    os.makedirs(os.path.dirname(DB_PATH), exist_ok=True)
    conn = sqlite3.connect(DB_PATH, timeout=120.0)
    conn.row_factory = sqlite3.Row
    conn.execute("PRAGMA journal_mode=WAL")
    conn.execute("PRAGMA busy_timeout=120000")
    return conn


def init_db() -> None:
    with _conn() as conn:
        conn.execute("""
            CREATE TABLE IF NOT EXISTS tarot_readings (
                id                  INTEGER PRIMARY KEY AUTOINCREMENT,
                created_at          TEXT NOT NULL DEFAULT (strftime('%Y-%m-%dT%H:%M:%fZ','now')),
                spread_type         TEXT NOT NULL,
                category            TEXT,
                question            TEXT,
                cards               TEXT NOT NULL,
                interpretation_json TEXT,
                summary             TEXT,
                model               TEXT,
                tokens_in           INTEGER,
                tokens_out          INTEGER,
                cost_usd            REAL,
                confidence          TEXT,
                favorite            INTEGER NOT NULL DEFAULT 0,
                note                TEXT
            )
        """)
        conn.execute("""
            CREATE INDEX IF NOT EXISTS idx_tarot_created
            ON tarot_readings(created_at DESC)
        """)
        conn.execute("""
            CREATE INDEX IF NOT EXISTS idx_tarot_favorite
            ON tarot_readings(favorite, created_at DESC)
        """)


def save_tarot_reading(data: Dict[str, Any]) -> int:
    interp = data.get("interpretation_json") or {}
    summary = interp.get("summary", "") if isinstance(interp, dict) else ""
    with _conn() as conn:
        cur = conn.execute(
            """INSERT INTO tarot_readings
               (spread_type, category, question, cards, interpretation_json,
                summary, model, tokens_in, tokens_out, cost_usd, confidence)
               VALUES (?,?,?,?,?,?,?,?,?,?,?)""",
            (
                data["spread_type"],
                data.get("category"),
                data.get("question"),
                json.dumps(data.get("cards") or [], ensure_ascii=False),
                json.dumps(interp, ensure_ascii=False) if interp else None,
                summary,
                data.get("model"),
                data.get("tokens_in"),
                data.get("tokens_out"),
                data.get("cost_usd"),
                data.get("confidence"),
            ),
        )
        return int(cur.lastrowid)


def get_tarot_reading(reading_id: int) -> Optional[Dict[str, Any]]:
    with _conn() as conn:
        r = conn.execute("SELECT * FROM tarot_readings WHERE id=?", (reading_id,)).fetchone()
    return _tarot_row_to_dict(r) if r else None


def list_tarot_readings(
    page: int = 1, size: int = 20,
    favorite: Optional[bool] = None,
    spread_type: Optional[str] = None,
    category: Optional[str] = None,
) -> Dict[str, Any]:
    wheres, params = [], []
    if favorite is not None:
        wheres.append("favorite=?")
        params.append(1 if favorite else 0)
    if spread_type:
        wheres.append("spread_type=?")
        params.append(spread_type)
    if category:
        wheres.append("category=?")
        params.append(category)
    where_sql = ("WHERE " + " AND ".join(wheres)) if wheres else ""
    offset = (page - 1) * size
    with _conn() as conn:
        total = conn.execute(
            f"SELECT COUNT(*) c FROM tarot_readings {where_sql}", params
        ).fetchone()["c"]
        rows = conn.execute(
            f"SELECT * FROM tarot_readings {where_sql} ORDER BY created_at DESC LIMIT ? OFFSET ?",
            params + [size, offset],
        ).fetchall()
    return {
        "items": [_tarot_row_to_dict(r) for r in rows],
        "page": page, "size": size, "total": int(total),
    }


def update_tarot_reading(reading_id: int, **kwargs) -> None:
    sets, vals = [], []
    if "favorite" in kwargs and kwargs["favorite"] is not None:
        sets.append("favorite=?")
        vals.append(1 if kwargs["favorite"] else 0)
    if "note" in kwargs and kwargs["note"] is not None:
        sets.append("note=?")
        vals.append(kwargs["note"])
    if not sets:
        return
    vals.append(reading_id)
    with _conn() as conn:
        conn.execute(f"UPDATE tarot_readings SET {','.join(sets)} WHERE id=?", vals)


def delete_tarot_reading(reading_id: int) -> None:
    with _conn() as conn:
        conn.execute("DELETE FROM tarot_readings WHERE id=?", (reading_id,))


def _tarot_row_to_dict(r) -> Dict[str, Any]:
    try:
        interp = json.loads(r["interpretation_json"]) if r["interpretation_json"] else None
    except (ValueError, TypeError):
        interp = None
    try:
        cards = json.loads(r["cards"]) if r["cards"] else []
    except (ValueError, TypeError):
        cards = []
    return {
        "id": r["id"],
        "created_at": r["created_at"],
        "spread_type": r["spread_type"],
        "category": r["category"],
        "question": r["question"],
        "cards": cards,
        "interpretation_json": interp,
        "summary": r["summary"],
        "model": r["model"],
        "tokens_in": r["tokens_in"],
        "tokens_out": r["tokens_out"],
        "cost_usd": r["cost_usd"],
        "confidence": r["confidence"],
        "favorite": int(r["favorite"]),
        "note": r["note"],
    }