database.py - dev - vaskaz/flexin_flow

database.py
173 строки5 KB
vaskaz
newborn
10 июн 2026, 00:16
10 июн 2026, 00:16dc9da65
"""SQLite database models for retro game library."""

import sqlite3
import os
import hashlib
from pathlib import Path
from dataclasses import dataclass, asdict
from typing import Optional

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


@dataclass
class RomFile:
    path: str          # Absolute path
    filename: str      # Just the filename
    size_bytes: int
    platform: str      # e.g. "nes", "snes", "gba"
    platform_name: str # e.g. "Nintendo Entertainment System"
    file_hash: str     # SHA256 of first 64KB (fast fingerprint)
    title: str         # Guessed game title from filename
    emulator: str      # Suggested emulator
    cover_url: str = ""     # Local cached cover path
    description: str = ""   # Game description
    id: Optional[int] = None


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


def init_db():
    conn = get_conn()
    conn.executescript("""
        CREATE TABLE IF NOT EXISTS roms (
            id INTEGER PRIMARY KEY AUTOINCREMENT,
            path TEXT UNIQUE NOT NULL,
            filename TEXT NOT NULL,
            size_bytes INTEGER NOT NULL,
            platform TEXT NOT NULL,
            platform_name TEXT NOT NULL,
            file_hash TEXT NOT NULL,
            title TEXT NOT NULL,
            emulator TEXT NOT NULL DEFAULT '',
            cover_url TEXT NOT NULL DEFAULT '',
            description TEXT NOT NULL DEFAULT ''
        );

        CREATE INDEX IF NOT EXISTS idx_roms_platform ON roms(platform);
        CREATE INDEX IF NOT EXISTS idx_roms_title ON roms(title);
        CREATE INDEX IF NOT EXISTS idx_roms_hash ON roms(file_hash);
    """)
    conn.commit()
    conn.close()


def fast_hash(filepath: str) -> str:
    """SHA256 of first 64KB (fast fingerprint for dedup)."""
    h = hashlib.sha256()
    try:
        with open(filepath, "rb") as f:
            chunk = f.read(65536)
            h.update(chunk)
    except (OSError, PermissionError):
        return ""
    return h.hexdigest()


def insert_rom(rom: RomFile) -> bool:
    """Insert or update a ROM record. Returns True if new/updated."""
    conn = get_conn()
    try:
        conn.execute("""
            INSERT INTO roms (path, filename, size_bytes, platform, platform_name,
                              file_hash, title, emulator, cover_url, description)
            VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?)
            ON CONFLICT(path) DO UPDATE SET
                size_bytes=excluded.size_bytes,
                platform=excluded.platform,
                platform_name=excluded.platform_name,
                file_hash=excluded.file_hash,
                title=excluded.title,
                emulator=excluded.emulator,
                cover_url=excluded.cover_url,
                description=excluded.description
        """, (
            rom.path, rom.filename, rom.size_bytes, rom.platform,
            rom.platform_name, rom.file_hash, rom.title,
            rom.emulator, rom.cover_url, rom.description,
        ))
        conn.commit()
        return True
    except sqlite3.Error:
        return False
    finally:
        conn.close()


def remove_stale(paths_in_use: set[str]):
    """Remove DB entries for files that no longer exist on disk."""
    conn = get_conn()
    try:
        cursor = conn.execute("SELECT id, path FROM roms")
        for row in cursor.fetchall():
            if row["path"] not in paths_in_use:
                conn.execute("DELETE FROM roms WHERE id = ?", (row["id"],))
        conn.commit()
    finally:
        conn.close()


def get_all_roms() -> list[dict]:
    conn = get_conn()
    rows = conn.execute("SELECT * FROM roms ORDER BY platform, title").fetchall()
    conn.close()
    return [dict(r) for r in rows]


def get_rom_by_id(rom_id: int) -> Optional[dict]:
    conn = get_conn()
    row = conn.execute("SELECT * FROM roms WHERE id = ?", (rom_id,)).fetchone()
    conn.close()
    return dict(row) if row else None


def get_platforms() -> list[dict]:
    conn = get_conn()
    rows = conn.execute("""
        SELECT platform, platform_name, COUNT(*) as count
        FROM roms GROUP BY platform ORDER BY platform_name
    """).fetchall()
    conn.close()
    return [dict(r) for r in rows]


def search_roms(query: str) -> list[dict]:
    conn = get_conn()
    like = f"%{query}%"
    rows = conn.execute(
        "SELECT * FROM roms WHERE title LIKE ? OR filename LIKE ? ORDER BY title",
        (like, like)
    ).fetchall()
    conn.close()
    return [dict(r) for r in rows]


def search_roms_by_filters(platform: str = "", search: str = "") -> list[dict]:
    conn = get_conn()
    sql = "SELECT * FROM roms WHERE 1=1"
    params = []
    if platform:
        sql += " AND platform = ?"
        params.append(platform)
    if search:
        sql += " AND (title LIKE ? OR filename LIKE ?)"
        params.extend([f"%{search}%", f"%{search}%"])
    sql += " ORDER BY title"
    rows = conn.execute(sql, params).fetchall()
    conn.close()
    return [dict(r) for r in rows]


def get_stats() -> dict:
    conn = get_conn()
    total = conn.execute("SELECT COUNT(*) as c FROM roms").fetchone()["c"]
    platforms = conn.execute("SELECT COUNT(DISTINCT platform) as c FROM roms").fetchone()["c"]
    total_size = conn.execute("SELECT COALESCE(SUM(size_bytes), 0) as s FROM roms").fetchone()["s"]
    conn.close()
    return {"total_roms": total, "platforms": platforms, "total_size_bytes": total_size}