steam-backlog-enforcer/steam_backlog_enforcer/config.py

"""Configuration management for Steam Backlog Enforcer."""

from __future__ import annotations

import contextlib
from dataclasses import dataclass, field, fields
from datetime import UTC, datetime, timedelta
import json
import logging
import os
from pathlib import Path
import tempfile
from typing import Any

from steam_backlog_enforcer._state_migrations import (
    backfill_assignment_times,
    migrate_legacy_manual_pick,
)

CONFIG_DIR = Path.home() / ".config" / "steam_backlog_enforcer"
CONFIG_FILE = CONFIG_DIR / "config.json"
STATE_FILE = CONFIG_DIR / "state.json"
SNAPSHOT_FILE = CONFIG_DIR / "snapshot.json"
LOG_FILE = CONFIG_DIR / "enforcer.log"

# Steam store domains to block.
BLOCKED_DOMAINS = [
    "store.steampowered.com",
    "checkout.steampowered.com",
    "store.akamai.steamstatic.com",
    "storefront.steampowered.com",
    "store.cloudflare.steamstatic.com",
]

HOSTS_FILE = Path("/etc/hosts")

logger = logging.getLogger(__name__)


def _atomic_write(path: Path, data: str, *, mode: int | None = None) -> None:
    """Write data to a file atomically via a temporary file + rename.

    Args:
        path: Destination file.
        data: Text to write.
        mode: Permission bits for the result. ``None`` keeps ``mkstemp``'s
            0600, which is what files holding secrets (``config.json`` carries
            ``steam_api_key``) must stay at. Callers whose file has to be
            readable by an unprivileged reader pass it explicitly; it is set on
            the temporary file so the mode lands with the rename rather than
            leaving a window where the destination exists at 0600.
    """
    path.parent.mkdir(parents=True, exist_ok=True)
    fd, tmp = tempfile.mkstemp(dir=path.parent, suffix=".tmp")
    tmp_path = Path(tmp)
    try:
        os.write(fd, data.encode("utf-8"))
        if mode is not None:
            os.fchmod(fd, mode)
        # Keep the owner, or a root daemon write locks the user's CLI out.
        with contextlib.suppress(OSError):  # absent file / non-root: skip
            original = path.stat()
            os.fchown(fd, original.st_uid, original.st_gid)
        os.close(fd)
        tmp_path.replace(path)
    except BaseException:
        with contextlib.suppress(OSError):
            os.close(fd)
        with contextlib.suppress(OSError):
            tmp_path.unlink()
        raise


# The public name, for callers outside this module. The private one predates
# it and is still what most of the package imports (and what tests patch).
atomic_write = _atomic_write


def _field_names(cls: type) -> frozenset[str]:
    """The dataclass's own field names, for filtering a loaded JSON object."""
    return frozenset(f.name for f in fields(cls))


@dataclass
class Config:
    """User configuration."""

    steam_api_key: str = ""
    steam_id: str = ""
    block_store: bool = True
    kill_unauthorized_games: bool = True
    uninstall_other_games: bool = True
    desktop_notifications: bool = True
    max_manual_picks: int = 2
    """How many games may be manually locked in at once.

    All active picks stay installed and visible; the enforcer treats them as
    one allowed set. Raising this weakens enforcement proportionally.
    """
    book_ledger_path: str = "~/.local/share/book_guard/ledger.json"
    """book-guard's signed ledger: the reading hour's only transport."""
    workout_status_url: str = "http://127.0.0.1:8770/api/status"
    """Where to ask whether today has a counted workout. Loopback only."""
    workout_log_path: str = "~/src/screen-locker/screen_locker/log.json"
    """screen-locker's log.json; a change re-asks that URL at once."""
    leetcode_ledger_path: str = "~/.local/share/leetcode_guard/ledger.json"
    leetcode_status_url: str = "http://127.0.0.1:8771/api/status"
    """LeetCode's transports: ledger first, this loopback URL as fallback."""
    count_launcher_processes: bool = True
    """Whether time spent in non-Steam launchers counts against the budget.

    Set False if a launcher legitimately idles in the tray all day. Leaving it
    True is the safer default: under-counting is the failure mode that quietly
    defeats the budget entirely.
    """
    counted_processes: list[dict[str, Any]] | None = None
    """Non-Steam games that count against the budget, beyond the launcher set.

    Each entry is ``{"id": ..., "label": ..., "names": [...]}``. ``None`` means
    "use the built-in defaults" (osu!lazer); an explicit ``[]`` means "none",
    which is why this is not simply an empty list. Parsed and validated by
    ``leisure_watch.procscan`` (once ``_counted_procs``) — the field exists
    here only so that ``save()`` round-trips a hand-edited list instead of
    dropping it.

    Deliberately separate from ``LAUNCHER_PROCESS_NAMES``: that frozenset is
    also the total-block kill list, so adding a game there has side effects
    this list does not.
    """
    playtime_enforcement: bool = True
    """Master switch for the daily gaming budget.

    Disabling stops the cutoff but never strands a live block — the release
    path runs regardless, so "disabled" cannot come to mean "blocked forever".
    """

    def save(self) -> None:
        """Persist config to disk."""
        _atomic_write(
            CONFIG_FILE,
            json.dumps(self.__dict__, indent=2) + "\n",
        )

    @classmethod
    def load(cls) -> Config:
        """Load config from disk, or return defaults."""
        if CONFIG_FILE.exists():
            data = json.loads(CONFIG_FILE.read_text(encoding="utf-8"))
            return cls(**{k: v for k, v in data.items() if k in _field_names(cls)})
        return cls()


@dataclass
class State:
    """Persistent state across runs."""

    current_app_id: int | None = None
    current_game_name: str = ""
    finished_app_ids: list[int] = field(default_factory=list)
    skipped_until: dict[str, str] = field(default_factory=dict)
    """Map of ``str(app_id)`` → ISO-8601 UTC timestamp when the skip expires.

    Games in this map are excluded from auto-assignment until the timestamp
    elapses. Populated when the user declines a freshly-picked game via the
    interactive prompt in ``cmd_done``.
    """
    enforcement_started_at: str = ""
    """ISO-8601 UTC timestamp set on the first game assignment."""
    manual_pick_app_id: int | None = None
    manual_pick_game_name: str = ""
    manual_pick_started_at: str = ""
    """Legacy single-slot manual pick, migrated into ``manual_picks`` on load.

    Kept as fields so an older ``state.json`` still deserialises; they are
    blanked once migrated and are never written again.
    """
    manual_picks: list[dict[str, Any]] = field(default_factory=list)
    """Manual picks, newest last; inactive ones stay until a new choice.

    Each entry is ``{"app_id": int, "game_name": str, "started_at": iso}``,
    plus ``"released_at": iso`` once it earned a new achievement. Plain dicts
    because ``save`` serialises ``self.__dict__`` straight to JSON.
    """
    current_assigned_at: str = ""
    """ISO time the current game was assigned; unlocks after it release it."""
    last_assigned_at: dict[str, str] = field(default_factory=dict)
    """``str(app_id)`` → ISO time of its latest assignment (pick ordering)."""
    released_at: dict[str, str] = field(default_factory=dict)
    """``str(app_id)`` → ISO time it was released below 100% (tampering)."""
    store_unblocked_until: str = ""
    """ISO deadline of the open store window (``unblock``); CLI-written only."""

    def skip_for_days(self, app_id: int, days: int) -> None:
        """Mark ``app_id`` as skipped for ``days`` days from now (UTC)."""
        expires = datetime.now(UTC) + timedelta(days=days)
        self.skipped_until[str(app_id)] = expires.isoformat()

    def active_skipped_ids(self) -> set[int]:
        """Return currently-skipped app IDs and prune expired entries.

        Mutates ``self.skipped_until`` to drop expired or malformed entries.
        Callers should ``save()`` if they want the prune persisted.
        """
        now = datetime.now(UTC)
        active: set[int] = set()
        to_remove: list[str] = []
        for aid_str, ts in self.skipped_until.items():
            try:
                expiry = datetime.fromisoformat(ts)
            except ValueError:
                to_remove.append(aid_str)
                continue
            if expiry > now:
                active.add(int(aid_str))
            else:
                to_remove.append(aid_str)
        for aid_str in to_remove:
            del self.skipped_until[aid_str]
        return active

    def save(self) -> None:
        """Persist state to disk."""
        _atomic_write(
            STATE_FILE,
            json.dumps(self.__dict__, indent=2) + "\n",
        )

    @classmethod
    def load(cls) -> State:
        """Load state from disk, or return defaults."""
        if STATE_FILE.exists():
            try:
                data = json.loads(STATE_FILE.read_text(encoding="utf-8"))
            except json.JSONDecodeError, OSError, ValueError:
                logger.warning("Corrupt state file, using defaults.")
                return cls()
            state = cls(**{k: v for k, v in data.items() if k in _field_names(cls)})
            migrated = migrate_legacy_manual_pick(state)
            if backfill_assignment_times(state) or migrated:
                state.save()
            return state
        return cls()