Blame for db.py

"""SQLite database layer for Prompt Garage."""

import os
import sqlite3

DB_PATH = os.path.join(os.path.dirname(os.path.abspath(__file__)), "prompts.db")

SCHEMA = """
CREATE TABLE IF NOT EXISTS prompts (
    id INTEGER PRIMARY KEY,
    title TEXT NOT NULL,
    content TEXT NOT NULL,
    source TEXT,
    is_starred INTEGER DEFAULT 0,
    tags TEXT,
    model TEXT,
    rating INTEGER,
    use_count INTEGER DEFAULT 0,
    created_at TEXT DEFAULT (datetime('now')),
    updated_at TEXT DEFAULT (datetime('now'))
);

CREATE VIRTUAL TABLE IF NOT EXISTS prompts_fts USING fts5(
    title, content, tags, content=prompts, content_rowid=id
);

CREATE TRIGGER IF NOT EXISTS prompts_ai AFTER INSERT ON prompts BEGIN
    INSERT INTO prompts_fts(rowid, title, content, tags)
    VALUES (new.id, new.title, new.content, new.tags);
END;

CREATE TRIGGER IF NOT EXISTS prompts_ad AFTER DELETE ON prompts BEGIN
    INSERT INTO prompts_fts(prompts_fts, rowid, title, content, tags)
    VALUES('delete', old.id, old.title, old.content, old.tags);
END;

CREATE TRIGGER IF NOT EXISTS prompts_au AFTER UPDATE ON prompts BEGIN
    INSERT INTO prompts_fts(prompts_fts, rowid, title, content, tags)
    VALUES('delete', old.id, old.title, old.content, old.tags);
    INSERT INTO prompts_fts(rowid, title, content, tags)
    VALUES (new.id, new.title, new.content, new.tags);
END;
"""


def get_db() -> sqlite3.Connection:
    conn = sqlite3.connect(DB_PATH)
    conn.row_factory = sqlite3.Row
    conn.execute("PRAGMA journal_mode=WAL")
    return conn


def init_db():
    conn = get_db()
    conn.executescript(SCHEMA)
    conn.close()


def add_prompt(title, content, source="manual", tags=None, model=None, rating=None, is_starred=0):
    conn = get_db()
    cur = conn.execute(
        "INSERT INTO prompts (title, content, source, tags, model, rating, is_starred) VALUES (?, ?, ?, ?, ?, ?, ?)",
        (title, content, source, tags, model, rating, is_starred),
    )
    conn.commit()
    pid = cur.lastrowid
    conn.close()
    return pid


def list_prompts(starred=None, tag=None, model=None, limit=20):
    conn = get_db()
    query = "SELECT * FROM prompts WHERE 1=1"
    params = []
    if starred:
        query += " AND is_starred = 1"
    if tag:
        query += " AND tags LIKE ?"
        params.append(f"%{tag}%")
    if model:
        query += " AND model = ?"
        params.append(model)
    query += " ORDER BY updated_at DESC LIMIT ?"
    params.append(limit)
    rows = conn.execute(query, params).fetchall()
    conn.close()
    return rows


def get_prompt(pid):
    conn = get_db()
    row = conn.execute("SELECT * FROM prompts WHERE id = ?", (pid,)).fetchone()
    conn.close()
    return row


def delete_prompt(pid):
    conn = get_db()
    conn.execute("DELETE FROM prompts WHERE id = ?", (pid,))
    conn.commit()
    conn.close()


def update_prompt(pid, **kwargs):
    conn = get_db()
    sets = ", ".join(f"{k} = ?" for k in kwargs)
    vals = list(kwargs.values()) + [pid]
    conn.execute(f"UPDATE prompts SET {sets}, updated_at = datetime('now') WHERE id = ?", vals)
    conn.commit()
    conn.close()


def toggle_star(pid):
    conn = get_db()
    conn.execute("UPDATE prompts SET is_starred = 1 - is_starred, updated_at = datetime('now') WHERE id = ?", (pid,))
    conn.commit()
    row = conn.execute("SELECT is_starred FROM prompts WHERE id = ?", (pid,)).fetchone()
    conn.close()
    return row["is_starred"] if row else None


def set_rating(pid, rating):
    update_prompt(pid, rating=rating)


def increment_use(pid):
    conn = get_db()
    conn.execute("UPDATE prompts SET use_count = use_count + 1, updated_at = datetime('now') WHERE id = ?", (pid,))
    conn.commit()
    conn.close()


def search_fts(query):
    conn = get_db()
    rows = conn.execute(
        "SELECT p.* FROM prompts p JOIN prompts_fts f ON p.id = f.rowid WHERE prompts_fts MATCH ? ORDER BY rank",
        (query,),
    ).fetchall()
    conn.close()
    return rows


def get_stats():
    conn = get_db()
    total = conn.execute("SELECT COUNT(*) as c FROM prompts").fetchone()["c"]
    starred = conn.execute("SELECT COUNT(*) as c FROM prompts WHERE is_starred = 1").fetchone()["c"]
    by_source = conn.execute("SELECT source, COUNT(*) as c FROM prompts GROUP BY source").fetchall()
    by_model = conn.execute("SELECT model, COUNT(*) as c FROM prompts WHERE model IS NOT NULL GROUP BY model").fetchall()
    by_tag_raw = conn.execute("SELECT tags FROM prompts WHERE tags IS NOT NULL AND tags != ''").fetchall()
    conn.close()

    tag_counts = {}
    for row in by_tag_raw:
        for t in row["tags"].split(","):
            t = t.strip()
            if t:
                tag_counts[t] = tag_counts.get(t, 0) + 1

    return {
        "total": total,
        "starred": starred,
        "by_source": {r["source"]: r["c"] for r in by_source},
        "by_model": {r["model"]: r["c"] for r in by_model},
        "by_tag": tag_counts,
    }


def list_top_used(limit=20):
    """Return prompts with use_count > 0, sorted by use_count descending."""
    conn = get_db()
    rows = conn.execute(
        "SELECT * FROM prompts WHERE use_count > 0 ORDER BY use_count DESC LIMIT ?",
        (limit,),
    ).fetchall()
    conn.close()
    return rows


def list_untagged(limit=100):
    """Return prompts that have no tags (NULL or empty string)."""
    conn = get_db()
    rows = conn.execute(
        "SELECT * FROM prompts WHERE tags IS NULL OR tags = '' LIMIT ?",
        (limit,),
    ).fetchall()
    conn.close()
    return rows


def prompt_exists_by_content(content_hash):
    """Check if a prompt with this content already exists (dedup by hash stored in source field)."""
    conn = get_db()
    row = conn.execute("SELECT id FROM prompts WHERE source LIKE ?", (f"%{content_hash}%",)).fetchone()
    conn.close()
    return row is not None