psp-preservation-tool/app/db.py

"""SQLite persistence layer (standard-library sqlite3, no ORM)."""
import shutil
import sqlite3
from pathlib import Path

from .config import COVERS_DIR, DB_PATH, LEGACY_COVERS_DIR, LEGACY_DB_PATH

SCHEMA = """
CREATE TABLE IF NOT EXISTS dats (
    id           INTEGER PRIMARY KEY,
    name         TEXT NOT NULL UNIQUE,
    version      TEXT,
    description  TEXT,
    system       TEXT NOT NULL,
    source_path  TEXT,
    ingested_at  TEXT NOT NULL
);

CREATE TABLE IF NOT EXISTS releases (
    id           INTEGER PRIMARY KEY,
    dat_id       INTEGER NOT NULL REFERENCES dats(id) ON DELETE CASCADE,
    name         TEXT NOT NULL,
    description  TEXT,
    region       TEXT,
    serial       TEXT
);

CREATE TABLE IF NOT EXISTS roms (
    id           INTEGER PRIMARY KEY,
    release_id   INTEGER NOT NULL REFERENCES releases(id) ON DELETE CASCADE,
    name         TEXT NOT NULL,
    size         INTEGER,
    crc32        TEXT,
    md5          TEXT,
    sha1         TEXT,
    serial       TEXT
);

CREATE INDEX IF NOT EXISTS idx_releases_serial ON releases(serial);
CREATE INDEX IF NOT EXISTS idx_releases_name   ON releases(name);
CREATE INDEX IF NOT EXISTS idx_releases_region ON releases(region);
CREATE INDEX IF NOT EXISTS idx_roms_crc32      ON roms(crc32);
CREATE INDEX IF NOT EXISTS idx_roms_md5        ON roms(md5);
CREATE INDEX IF NOT EXISTS idx_roms_sha1       ON roms(sha1);

CREATE TABLE IF NOT EXISTS files (
    id          INTEGER PRIMARY KEY,
    path        TEXT NOT NULL UNIQUE,
    name        TEXT NOT NULL,
    size        INTEGER NOT NULL,
    mtime_ns    INTEGER,
    crc32       TEXT,
    md5         TEXT,
    sha1        TEXT,
    serial      TEXT,
    title       TEXT,
    release_id  INTEGER,
    error       TEXT,
    scanned_at  TEXT NOT NULL
);

CREATE INDEX IF NOT EXISTS idx_files_path ON files(path);

CREATE TABLE IF NOT EXISTS choices (
    serial     TEXT NOT NULL,
    size       INTEGER NOT NULL,
    release_id INTEGER,
    PRIMARY KEY (serial, size)
);

CREATE TABLE IF NOT EXISTS item_info (
    release_id  TEXT NOT NULL,
    system      TEXT NOT NULL,
    title       TEXT,
    meta_json   TEXT,
    manual      INTEGER NOT NULL DEFAULT 0,
    updated_at  TEXT NOT NULL,
    PRIMARY KEY (release_id, system)
);

CREATE TABLE IF NOT EXISTS cover_attempts (
    key    TEXT NOT NULL,
    system TEXT NOT NULL,
    status TEXT NOT NULL,
    ts     REAL NOT NULL,
    PRIMARY KEY (key, system)
);

CREATE TABLE IF NOT EXISTS meta_attempts (
    key    TEXT NOT NULL,
    system TEXT NOT NULL,
    status TEXT NOT NULL,
    ts     REAL NOT NULL,
    PRIMARY KEY (key, system)
);

CREATE TABLE IF NOT EXISTS save_sigs (
    save_id  TEXT NOT NULL,
    size     INTEGER NOT NULL,
    mtime_ns INTEGER NOT NULL,
    sig      TEXT NOT NULL,
    PRIMARY KEY (save_id, size, mtime_ns)
);

CREATE TABLE IF NOT EXISTS ps1_entries (
    id           INTEGER PRIMARY KEY,
    source_path  TEXT NOT NULL UNIQUE,
    title        TEXT,
    serial       TEXT,
    region       TEXT,
    disc         INTEGER NOT NULL DEFAULT 1,
    disc_total   INTEGER NOT NULL DEFAULT 1,
    output_eboot TEXT,
    status       TEXT NOT NULL DEFAULT 'pending',
    error        TEXT,
    updated_at   TEXT NOT NULL
);
"""


def get_connection(path=None):
    target = path if path is not None else DB_PATH
    Path(target).parent.mkdir(parents=True, exist_ok=True)
    conn = sqlite3.connect(str(target))
    conn.row_factory = sqlite3.Row
    conn.execute("PRAGMA foreign_keys = ON")
    conn.execute("PRAGMA busy_timeout = 10000")
    conn.execute("PRAGMA journal_mode = WAL")
    conn.execute("PRAGMA synchronous = NORMAL")
    return conn


def _migrate(conn):
    """Bring databases created before ``release_id`` up to date."""
    cols = {r["name"] for r in conn.execute("PRAGMA table_info(files)")}
    if "release_id" not in cols:
        conn.execute("ALTER TABLE files ADD COLUMN release_id INTEGER")


def _integrity_ok(path):
    try:
        conn = sqlite3.connect(str(path))
        ok = conn.execute("PRAGMA integrity_check").fetchone()[0] == "ok"
        conn.close()
        return ok
    except sqlite3.Error:
        return False


def ensure_state():
    """One-time move of legacy on-share DB + cover cache onto local storage.

    SQLite over SMB/FUSE corrupts under concurrent access; local storage is
    fast and reliable.  Runs once, is a no-op afterwards.
    """
    if DB_PATH != LEGACY_DB_PATH and not DB_PATH.exists() and LEGACY_DB_PATH.exists():
        if _integrity_ok(LEGACY_DB_PATH):
            try:
                DB_PATH.parent.mkdir(parents=True, exist_ok=True)
                shutil.copy2(LEGACY_DB_PATH, DB_PATH)
            except OSError:
                pass
    if COVERS_DIR != LEGACY_COVERS_DIR and not COVERS_DIR.exists() and LEGACY_COVERS_DIR.exists():
        try:
            COVERS_DIR.parent.mkdir(parents=True, exist_ok=True)
            shutil.copytree(LEGACY_COVERS_DIR, COVERS_DIR)
        except OSError:
            pass


def init_db(path=None):
    if path is None:
        ensure_state()
    target = path if path is not None else DB_PATH
    Path(target).parent.mkdir(parents=True, exist_ok=True)
    conn = get_connection(target)
    conn.executescript(SCHEMA)
    _migrate(conn)
    conn.commit()
    return conn