100 lines
3.1 KiB
Python
100 lines
3.1 KiB
Python
import sqlite3
|
|
from pathlib import Path
|
|
|
|
DB_PATH = Path("data/cardpull.db")
|
|
|
|
|
|
def conn():
|
|
c = sqlite3.connect(str(DB_PATH))
|
|
c.row_factory = sqlite3.Row
|
|
c.execute("PRAGMA journal_mode=WAL")
|
|
c.execute("PRAGMA foreign_keys=ON")
|
|
return c
|
|
|
|
|
|
def _add_column(c, table, col, decl):
|
|
cols = [r[1] for r in c.execute(f"PRAGMA table_info({table})").fetchall()]
|
|
if col not in cols:
|
|
c.execute(f"ALTER TABLE {table} ADD COLUMN {col} {decl}")
|
|
|
|
|
|
def init():
|
|
DB_PATH.parent.mkdir(parents=True, exist_ok=True)
|
|
with conn() as c:
|
|
c.executescript("""
|
|
CREATE TABLE IF NOT EXISTS decks (
|
|
slug TEXT PRIMARY KEY,
|
|
deck_name TEXT NOT NULL,
|
|
status TEXT NOT NULL DEFAULT 'queued',
|
|
commander TEXT,
|
|
price_usd REAL DEFAULT 0,
|
|
done INTEGER DEFAULT 0,
|
|
total INTEGER DEFAULT 0
|
|
);
|
|
CREATE TABLE IF NOT EXISTS cards (
|
|
id INTEGER PRIMARY KEY AUTOINCREMENT,
|
|
deck_slug TEXT NOT NULL REFERENCES decks(slug) ON DELETE CASCADE,
|
|
name TEXT NOT NULL,
|
|
position INTEGER NOT NULL,
|
|
filename TEXT,
|
|
back_filename TEXT,
|
|
scry_url TEXT,
|
|
price_usd REAL DEFAULT 0,
|
|
fetch_status TEXT DEFAULT 'pending'
|
|
);
|
|
CREATE TABLE IF NOT EXISTS logs (
|
|
id INTEGER PRIMARY KEY AUTOINCREMENT,
|
|
deck_slug TEXT NOT NULL REFERENCES decks(slug) ON DELETE CASCADE,
|
|
line TEXT NOT NULL
|
|
);
|
|
""")
|
|
_add_column(c, "cards", "back_filename", "TEXT")
|
|
_add_column(c, "cards", "scry_url", "TEXT")
|
|
|
|
|
|
def get_deck(slug: str):
|
|
with conn() as c:
|
|
return c.execute("SELECT * FROM decks WHERE slug=?", (slug,)).fetchone()
|
|
|
|
|
|
def get_decks():
|
|
with conn() as c:
|
|
return c.execute("SELECT * FROM decks ORDER BY rowid").fetchall()
|
|
|
|
|
|
def get_cards(slug: str):
|
|
with conn() as c:
|
|
return c.execute(
|
|
"SELECT * FROM cards WHERE deck_slug=? ORDER BY position, id", (slug,)
|
|
).fetchall()
|
|
|
|
|
|
def get_logs(slug: str):
|
|
with conn() as c:
|
|
rows = c.execute("SELECT line FROM logs WHERE deck_slug=? ORDER BY id", (slug,)).fetchall()
|
|
return [r["line"] for r in rows]
|
|
|
|
|
|
def add_log(slug: str, line: str):
|
|
with conn() as c:
|
|
c.execute("INSERT INTO logs (deck_slug, line) VALUES (?,?)", (slug, line))
|
|
|
|
|
|
def update_deck(slug: str, **kw):
|
|
if not kw:
|
|
return
|
|
sets = ", ".join(f"{k}=?" for k in kw)
|
|
with conn() as c:
|
|
c.execute(f"UPDATE decks SET {sets} WHERE slug=?", [*kw.values(), slug])
|
|
|
|
|
|
def recalc_done(slug: str):
|
|
with conn() as c:
|
|
done = c.execute(
|
|
"SELECT COUNT(*) FROM cards WHERE deck_slug=? AND fetch_status IN ('done','failed','skipped')",
|
|
(slug,)
|
|
).fetchone()[0]
|
|
total = c.execute(
|
|
"SELECT COUNT(*) FROM cards WHERE deck_slug=?", (slug,)
|
|
).fetchone()[0]
|
|
c.execute("UPDATE decks SET done=?, total=? WHERE slug=?", (done, total, slug)) |