psp-preservation-tool/app/info.py

"""Per-item metadata: cached fetches + user overrides.

Fetched metadata (game details from IGDB, movie details from TheMovieDB /
Wikipedia) is cached in the library DB keyed by ``(release_id, system)`` so it
is only ever requested once.  Users can also hand-tag an item; the override is
stored with ``manual=1`` and always wins over both the cache and fresh fetches.

All database calls are defensive: a transient I/O failure returns ``None`` /
``False`` instead of raising, so a flaky share never crashes the UI.
"""
import json
import sqlite3
import time

from . import config, covers
from .db import get_connection

_failed = {}  # (release_id, system) -> time.time() of last failed fetch
_FAIL_TTL = 1800.0  # don't re-fetch failed metadata for 30 minutes

# Persistent "no metadata" cache so a failed lookup isn't re-tried on restart.
_miss_db = {}
_miss_loaded = False
_MISS_TTL = 86400.0  # retry failed metadata after 1 day


def _load_miss_db():
    global _miss_loaded
    if _miss_loaded:
        return
    _miss_loaded = True
    try:
        conn = get_connection(config.DB_PATH)
        for r in conn.execute(
            "SELECT key, system, ts FROM meta_attempts WHERE status = 'missing'"
        ):
            _miss_db[(r["key"], r["system"])] = r["ts"]
        conn.close()
    except sqlite3.Error:
        pass


def _mark_missing(key, system):
    ts = time.time()
    _miss_db[(key, system)] = ts
    try:
        conn = get_connection(config.DB_PATH)
        conn.execute(
            "INSERT INTO meta_attempts (key, system, status, ts) VALUES (?,?,?,?) "
            "ON CONFLICT(key, system) DO UPDATE SET status=excluded.status, ts=excluded.ts",
            (key, system, "missing", ts),
        )
        conn.commit()
        conn.close()
    except sqlite3.Error:
        pass


def _clear_missing(key, system):
    _miss_db.pop((key, system), None)
    try:
        conn = get_connection(config.DB_PATH)
        conn.execute(
            "DELETE FROM meta_attempts WHERE key = ? AND system = ?", (key, system)
        )
        conn.commit()
        conn.close()
    except sqlite3.Error:
        pass


def _serialize(meta):
    """Strip non-JSON-able or helper keys before persisting."""
    return {k: v for k, v in (meta or {}).items() if k != "manual"}


def load(release_id, system):
    """Return the stored info dict (with ``manual`` set) or None."""
    try:
        conn = get_connection(config.DB_PATH)
        row = conn.execute(
            "SELECT title, meta_json, manual FROM item_info WHERE release_id = ? AND system = ?",
            (str(release_id), system),
        ).fetchone()
        conn.close()
    except sqlite3.Error:
        return None
    if row is None:
        return None
    meta = {}
    try:
        meta = json.loads(row["meta_json"] or "{}")
    except (TypeError, ValueError):
        meta = {}
    if not isinstance(meta, dict):
        meta = {}
    if row["title"]:
        meta["title"] = row["title"]
    meta["manual"] = bool(row["manual"])
    return meta


def save(release_id, system, meta, manual=False, title=None):
    """Persist info for an item.  ``manual`` marks a user override."""
    try:
        conn = get_connection(config.DB_PATH)
        conn.execute(
            """
            INSERT INTO item_info (release_id, system, title, meta_json, manual, updated_at)
            VALUES (?,?,?,?,?,datetime('now'))
            ON CONFLICT(release_id, system) DO UPDATE SET
                title=excluded.title, meta_json=excluded.meta_json,
                manual=excluded.manual, updated_at=excluded.updated_at
            """,
            (
                str(release_id),
                system,
                title if title else (meta.get("title") if manual else None),
                json.dumps(_serialize(meta)),
                1 if manual else 0,
            ),
        )
        conn.commit()
        conn.close()
        return True
    except sqlite3.Error:
        return False


def reset(release_id, system):
    """Delete cached/overridden info so it is re-fetched next time."""
    key = (str(release_id), system)
    _failed.pop(key, None)
    _clear_missing(*key)
    try:
        conn = get_connection(config.DB_PATH)
        conn.execute(
            "DELETE FROM item_info WHERE release_id = ? AND system = ?",
            (str(release_id), system),
        )
        conn.commit()
        conn.close()
    except sqlite3.Error:
        pass


def override_titles():
    """Return {(system, str(release_id)): title} for all manual overrides."""
    try:
        conn = get_connection(config.DB_PATH)
        rows = conn.execute(
            "SELECT release_id, system, title FROM item_info WHERE manual = 1 AND title IS NOT NULL"
        ).fetchall()
        conn.close()
    except sqlite3.Error:
        return {}
    return {(r["system"], str(r["release_id"])): r["title"] for r in rows}


def cached_release_dates():
    """Return {(release_id, system): release_date} from cached metadata only.

    Pure DB read — no network — so list views can show a release date without
    triggering API fetches for every row.
    """
    out = {}
    try:
        conn = get_connection(config.DB_PATH)
        rows = conn.execute("SELECT release_id, system, meta_json FROM item_info").fetchall()
        conn.close()
    except sqlite3.Error:
        return {}
    for r in rows:
        try:
            meta = json.loads(r["meta_json"] or "{}")
        except (TypeError, ValueError):
            continue
        if isinstance(meta, dict) and meta.get("release_date"):
            out[(r["release_id"], r["system"])] = meta["release_date"]
    return out


def get_or_fetch(release_id, system, fetch_title):
    """Return cached info (override or fetched) or fetch-and-cache it."""
    info = load(release_id, system)
    if info is not None:
        return info
    key = (str(release_id), system)
    last = _failed.get(key)
    if last is not None and time.time() - last < _FAIL_TTL:
        return None  # recently failed; don't hammer the API again
    _load_miss_db()
    miss_ts = _miss_db.get(key)
    if miss_ts is not None and time.time() - miss_ts < _MISS_TTL:
        return None
    try:
        if system == "videos":
            meta = covers.get_movie_metadata(fetch_title)
        else:
            meta = covers.get_game_metadata(fetch_title, system=system)
    except Exception:  # noqa: BLE001 - network/API failures must not crash the UI
        meta = None
    if meta:
        save(release_id, system, meta, manual=False)
        meta["manual"] = False
        return meta
    _failed[key] = time.time()
    _mark_missing(*key)
    return None