steam-backlog-enforcer/steam_backlog_enforcer/_web_library.py

"""``GET /api/library``: every owned game, for the web library browser.

Joins the owned-games records (names, playtime, last played: see
:mod:`._owned_apps_cache`), the achievement snapshot, the HLTB cache, the
installed manifests and the state. Eligibility for "pick my own game" comes
from :func:`steam_backlog_enforcer._own_pick.ineligible_reason`, the same
rule the job enforces on the answer.

Reads local files only, except once: when no refresh has stored owned-game
records yet, it asks Steam for them (the snapshot alone misses every game
without achievements).
"""

from __future__ import annotations

from dataclasses import dataclass
from typing import TYPE_CHECKING, Any

from steam_backlog_enforcer import config as _config
from steam_backlog_enforcer._actions import allowed_app_ids
from steam_backlog_enforcer._hltb_types import _read_raw_cache
from steam_backlog_enforcer._own_pick import ineligible_reason
from steam_backlog_enforcer._owned_apps_cache import (
    load_owned_records,
    owned_cache_stamp,
    refresh_owned_records,
)
from steam_backlog_enforcer._snapshot import load_snapshot
from steam_backlog_enforcer._web_io import ok
from steam_backlog_enforcer.config import Config, State
from steam_backlog_enforcer.game_install import get_installed_games
from steam_backlog_enforcer.steam_api import GameInfo

if TYPE_CHECKING:
    from steam_backlog_enforcer._web_io import Reply, Request


@dataclass(frozen=True)
class _SnapshotMemo:
    """The snapshot as achievement-free ``GameInfo``s, keyed by file stamp."""

    stamp: tuple[int, int]
    games: dict[int, GameInfo]


@dataclass
class _Memos:
    """The process-wide memos, each keyed by the stamp of its source file."""

    snapshot: _SnapshotMemo | None = None
    owned_ids: tuple[tuple[int, int], set[int]] | None = None


_MEMOS = _Memos()


def _snapshot_games() -> dict[int, GameInfo]:
    """Snapshot rows by app id; re-parsed only when the file changes.

    The snapshot is ~11 MB of per-achievement detail the library never
    shows, so one parse per file version, not one per request.
    """
    try:
        st = _config.SNAPSHOT_FILE.stat()
    except OSError:
        return {}
    stamp = (st.st_mtime_ns, st.st_size)
    memo = _MEMOS.snapshot
    if memo is None or memo.stamp != stamp:
        games = {
            int(row["app_id"]): GameInfo(
                app_id=int(row["app_id"]),
                name=str(row.get("name") or ""),
                total_achievements=int(row.get("total_achievements") or 0),
                unlocked_achievements=int(row.get("unlocked_achievements") or 0),
                playtime_minutes=int(row.get("playtime_minutes") or 0),
                completionist_hours=float(row.get("completionist_hours") or -1),
            )
            for row in load_snapshot() or []
            if "app_id" in row
        }
        memo = _MEMOS.snapshot = _SnapshotMemo(stamp, games)
    return memo.games


def _owned_records(config: Config) -> dict[int, dict[str, Any]]:
    """Owned-game records by app id; fetched once when none are cached."""
    records = load_owned_records(config.steam_id)
    if records is None:
        records = refresh_owned_records(config) or []
    return {int(r["app_id"]): r for r in records if "app_id" in r}


def _hltb_hours(
    app_id: int, game: GameInfo | None, hltb: dict[int, Any]
) -> float | None:
    """Completionist hours: the HLTB cache, else the snapshot, else unknown."""
    hours = float(hltb.get(app_id, {}).get("hours") or -1)
    if hours <= 0 and game is not None:
        hours = game.completionist_hours
    return round(hours, 1) if hours > 0 else None


def owned_app_ids() -> set[int]:
    """Every app id the library can list (the art route serves only these).

    Never asks Steam: an image request must not turn into an API call. Kept
    per owned-cache file version, since a grid load asks once per cover.
    """
    stamp = owned_cache_stamp()
    snapshot = _snapshot_games()
    memo = _MEMOS.owned_ids
    if memo is None or memo[0] != stamp:
        records = load_owned_records(Config.load().steam_id) or []
        ids = {int(r["app_id"]) for r in records if "app_id" in r}
        memo = _MEMOS.owned_ids = (stamp, ids)
    return set(snapshot) | memo[1]


def _reason(
    app_id: int,
    game: GameInfo | None,
    record: dict[str, Any],
    state: State,
    skipped: set[int],
) -> str | None:
    """:func:`ineligible_reason`, saying *why* a game has no scan data."""
    if game is None and record.get("has_stats") is False:
        return "No achievements"
    if game is None and record.get("has_stats"):
        return "Not scanned yet"
    return ineligible_reason(app_id, game, state, skipped)


def library_rows(state: State, config: Config) -> list[dict[str, object]]:
    """One contract ``LibraryGame`` per owned game, in no particular order."""
    snapshot = _snapshot_games()
    records = _owned_records(config)
    hltb = _read_raw_cache()
    installed = {app_id for app_id, _ in get_installed_games()}
    allowed = allowed_app_ids(state)
    skipped = state.active_skipped_ids()
    rows: list[dict[str, object]] = []
    for app_id in records.keys() | snapshot.keys():
        game = snapshot.get(app_id)
        record = records.get(app_id, {})
        rows.append(
            {
                "app_id": app_id,
                "name": record.get("name") or (game.name if game else str(app_id)),
                "achievements_total": game.total_achievements if game else 0,
                "achievements_unlocked": game.unlocked_achievements if game else 0,
                "hltb_hours": _hltb_hours(app_id, game, hltb),
                "playtime_minutes": int(
                    record.get("playtime_minutes")
                    or (game.playtime_minutes if game else 0)
                ),
                "last_played": int(record.get("last_played") or 0) or None,
                "installed": app_id in installed,
                "assigned": app_id in allowed,
                "ineligible_reason": _reason(app_id, game, record, state, skipped),
            }
        )
    return rows


def library_view(_request: Request) -> Reply:
    """``GET /api/library`` — every owned game as ``LibraryPayload``."""
    return ok({"games": library_rows(State.load(), Config.load())})