card-grader/store.py

"""SQLite: settings and a log of past grade estimates.

One database, two tables. There is no inventory or pricing here — this app
does exactly one thing (estimate a PSA grade from photos) and remembers what
it told you, so you can look back at a card without re-running the estimate.
"""

import json
import os
import sqlite3
import time

BASE_DIR = os.path.dirname(os.path.abspath(__file__))
# Overridable so a container can point this at a mounted volume (e.g.
# /data/grades.db) instead of the app's own directory, which is what makes
# the data survive a container recreate/image update.
DB_PATH = os.environ.get("CARD_GRADER_DB_PATH") or os.path.join(BASE_DIR, "grades.db")

SCHEMA = """
CREATE TABLE IF NOT EXISTS settings (
    key   TEXT PRIMARY KEY,
    value TEXT
);

CREATE TABLE IF NOT EXISTS grades (
    id                   INTEGER PRIMARY KEY AUTOINCREMENT,
    created_at           TEXT NOT NULL,
    label                TEXT,              -- your own name for the card, optional
    card_type            TEXT,              -- pokemon | sports | other_tcg | other
    card_note            TEXT,              -- what the model read off the card
    image_count          INTEGER,
    thumbnail            TEXT,              -- small JPEG, base64 — first photo only
    model                TEXT,
    estimated_grade      INTEGER,
    grade_low            INTEGER,
    grade_high           INTEGER,
    confidence           TEXT,
    categories_json      TEXT,
    edge_measurements_json     TEXT,
    centering_measurement_json TEXT,
    aspect_measurement_json    TEXT,
    limitations_json     TEXT,
    note                 TEXT,
    estimated_cost       REAL,
    usage_json           TEXT
);

CREATE INDEX IF NOT EXISTS idx_grades_created ON grades(created_at DESC);
"""

DEFAULT_SETTINGS = {
    "anthropic_api_key": "",
    "vision_model": "claude-sonnet-5",
    "vision_effort": "low",
}


def connect():
    conn = sqlite3.connect(DB_PATH)
    conn.row_factory = sqlite3.Row
    conn.execute("PRAGMA foreign_keys = ON")
    return conn


def init():
    conn = connect()
    try:
        conn.executescript(SCHEMA)
        # CREATE TABLE IF NOT EXISTS never touches an already-existing table,
        # so a column added after cards were already graded needs its own
        # migration — guarded because re-running this against a database
        # that already has the column would otherwise error every startup.
        try:
            conn.execute("ALTER TABLE grades ADD COLUMN aspect_measurement_json TEXT")
        except sqlite3.OperationalError:
            pass
        for key, value in DEFAULT_SETTINGS.items():
            conn.execute(
                "INSERT OR IGNORE INTO settings (key, value) VALUES (?, ?)",
                (key, json.dumps(value)),
            )
        conn.commit()
    finally:
        conn.close()


def now():
    return time.strftime("%Y-%m-%dT%H:%M:%S")


# --------------------------------------------------------------- settings


def get_settings():
    conn = connect()
    try:
        rows = conn.execute("SELECT key, value FROM settings").fetchall()
    finally:
        conn.close()
    out = dict(DEFAULT_SETTINGS)
    for row in rows:
        try:
            out[row["key"]] = json.loads(row["value"])
        except (ValueError, TypeError):
            out[row["key"]] = row["value"]
    return out


def save_settings(updates):
    conn = connect()
    try:
        for key, value in updates.items():
            if key not in DEFAULT_SETTINGS:
                continue
            conn.execute(
                "INSERT INTO settings (key, value) VALUES (?, ?) "
                "ON CONFLICT(key) DO UPDATE SET value=excluded.value",
                (key, json.dumps(value)),
            )
        conn.commit()
    finally:
        conn.close()
    return get_settings()


# ------------------------------------------------------------------ grades


def save_grade(grade, thumbnail=None, label=None):
    """Persist one grading result. Returns the new row's id."""
    conn = connect()
    try:
        cur = conn.execute(
            "INSERT INTO grades "
            "(created_at, label, card_type, card_note, image_count, thumbnail, "
            " model, estimated_grade, "
            " grade_low, grade_high, confidence, categories_json, "
            " edge_measurements_json, centering_measurement_json, "
            " aspect_measurement_json, "
            " limitations_json, note, estimated_cost, usage_json) "
            "VALUES (?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?)",
            (
                now(), label, grade.get("card_type"), grade.get("card_note"),
                grade.get("image_count"), thumbnail,
                (grade.get("usage") or {}).get("model"),
                grade.get("estimated_grade"), grade.get("grade_low"),
                grade.get("grade_high"), grade.get("confidence"),
                json.dumps(grade.get("categories")),
                json.dumps(grade.get("edge_measurements")),
                json.dumps(grade.get("centering_measurement")),
                json.dumps(grade.get("aspect_measurement")),
                json.dumps(grade.get("limitations")),
                grade.get("note"), grade.get("estimated_cost"),
                json.dumps(grade.get("usage")),
            ),
        )
        conn.commit()
        return cur.lastrowid
    finally:
        conn.close()


def _row_to_grade(row):
    d = dict(row)
    for key in ("categories_json", "edge_measurements_json",
                "centering_measurement_json", "aspect_measurement_json",
                "limitations_json", "usage_json"):
        out_key = key[:-len("_json")]
        raw = d.pop(key, None)
        try:
            d[out_key] = json.loads(raw) if raw else None
        except (ValueError, TypeError):
            d[out_key] = None
    return d


def list_grades(limit=200):
    conn = connect()
    try:
        rows = conn.execute(
            "SELECT * FROM grades ORDER BY created_at DESC, id DESC LIMIT ?",
            (limit,),
        ).fetchall()
    finally:
        conn.close()
    return [_row_to_grade(r) for r in rows]


def get_grade(grade_id):
    conn = connect()
    try:
        row = conn.execute("SELECT * FROM grades WHERE id = ?", (grade_id,)).fetchone()
    finally:
        conn.close()
    return _row_to_grade(row) if row else None


def update_grade_label(grade_id, label):
    conn = connect()
    try:
        conn.execute("UPDATE grades SET label = ? WHERE id = ?", (label, grade_id))
        conn.commit()
    finally:
        conn.close()
    return get_grade(grade_id)


def delete_grade(grade_id):
    conn = connect()
    try:
        conn.execute("DELETE FROM grades WHERE id = ?", (grade_id,))
        conn.commit()
    finally:
        conn.close()