kuhyx/steam-backlog-enforcer
1
0
Fork 0
mirror of https://github.com/kuhyx/steam-backlog-enforcer.git synced 2026-11-11 13:27:33 +01:00
Code Issues Actions 5 Packages Projects Releases Wiki Activity

Split _actions, _whitelist, _cmd_done and hltb under the cap

Browse Source 
Each keeps its name and serves the moved names through a module __getattr__
(PEP 562) rather than a top-level re-export: every child imports back from
its parent, so a plain import would be circular. The lookup goes through
importlib.import_module, which is a call rather than an import statement,
so it needs no linter suppression. Return type is a _Reexport alias -- mypy
rejects `object` as not callable at the call sites, ruff rejects a bare Any.

Also fixes two leaks where tests wrote to the user's real config, both
caused by a redirected constant moving to a new module while conftest kept
patching the old one:

- APPROVED_EXCEPTIONS_FILE / EXCEPTION_AUDIT_LOG moved to _whitelist_locking;
  the suite had appended 27 fixture entries to the live exception_audit.log.
- _OWNED_IDS_CACHE_FILE was never redirected at all: it is derived from
  CONFIG_DIR at import time, so it captured the real path before the fixture
  could patch it, and the suite overwrote the user's owned-games cache.

Both are now verified by md5summing ~/.config/steam_backlog_enforcer before
and after a full run, which is the only check that actually catches this --
the tests pass either way.

1347 tests, 100.00% coverage.

Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_016A1EYiv7e7HGhET84oa9M3
This commit is contained in:
Krzysztof Rudnicki 2026-08-24 14:55:57 +02:00
parent 07e6e15b6e
commit d8ebbf978b
21 changed files with 1046 additions and 722 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

158
steam_backlog_enforcer/_actions.py
View File

@ -9,26 +9,20 @@ there is a single tested implementation of the underlying behaviour.
from __future__ import annotations
from datetime import datetime, timedelta, timezone
from typing import TYPE_CHECKING, Any
from datetime import datetime, timezone
import importlib
from typing import TYPE_CHECKING, Any, TypeAlias
from steam_backlog_enforcer._allowed_games import (
active_manual_picks,
allowed_games,
)
from steam_backlog_enforcer._total_block import get_total_block_status
from steam_backlog_enforcer.game_install import get_installed_games, is_protected_app
# Marks these two as an intentional re-export (they're imported from
# _allowed_games above, not defined here) -- without this, mypy's
# --no-implicit-reexport flags every downstream `from _actions import
# active_manual_picks` as an error. Everything else in this file is
# accessible normally; this only needs to list names imported-then-reexported.
__all__ = [
"active_manual_picks",
"allowed_games",
]
from steam_backlog_enforcer.store_blocker import is_store_blocked
if TYPE_CHECKING:
from steam_backlog_enforcer.config import State
@ -160,123 +154,31 @@ def apply_manual_pick(
return None
def manual_pick_grace_remaining(state: State, app_id: int) -> float | None:
"""Return days left in *app_id*'s grace window, or ``None``.
``None`` means "no grace window applies": *app_id* is not an active manual
pick, or its timestamp is missing/malformed. A value <= 0 means the window
has closed.
Args:
state: The loaded enforcer state.
app_id: The manually-picked app id to measure.
Returns:
Fractional days remaining before the pick can no longer be abandoned,
or ``None`` when the question does not apply.
"""
pick = find_manual_pick(state, app_id)
if pick is None or not pick.get("started_at"):
return None
try:
started = datetime.fromisoformat(pick["started_at"])
except ValueError:
return None
deadline = started + timedelta(days=MANUAL_GRACE_DAYS)
return (deadline - datetime.now(timezone.utc)).total_seconds() / 86400
def can_abandon_manual_pick(state: State, app_id: int) -> bool:
"""Return ``True`` if *app_id* is still inside its grace window.
Args:
state: The loaded enforcer state.
app_id: The manually-picked app id to check.
Returns:
Whether ``abandon_manual_pick`` would be accepted right now.
"""
remaining = manual_pick_grace_remaining(state, app_id)
return remaining is not None and remaining > 0
def abandon_manual_pick(state: State, app_id: int) -> bool:
"""Drop one manual pick and persist ``state``, if still inside its grace window.
Only the named pick is dropped: any other active pick keeps its own lock
and deadline. The abandoned app id goes onto the existing skip cooldown so
``scan`` will not hand the same game straight back. State-only: like
``apply_manual_pick`` this performs no uninstall/hide cascade, so the MCP
server can call it without touching the filesystem.
Args:
state: The enforcer state to mutate and save.
app_id: The manually-picked app id to back out of.
Returns:
``True`` if the pick was abandoned, ``False`` if its grace window has
closed (or it is not an active pick), in which case ``state`` is
untouched.
"""
if not can_abandon_manual_pick(state, app_id):
return False
state.skip_for_days(app_id, ABANDON_COOLDOWN_DAYS)
state.manual_picks = [
p for p in active_manual_picks(state) if p.get("app_id") != app_id
]
# The abandoned pick may also have been the current assignment; hand the
# assignment to a surviving pick so the enforcer keeps guarding it, or
# clear it so 'scan' can reassign.
if state.current_app_id == app_id:
survivor = state.manual_picks[-1] if state.manual_picks else None
state.current_app_id = survivor["app_id"] if survivor else None
state.current_game_name = survivor["game_name"] if survivor else ""
state.save()
return True
def status_payload(state: State) -> dict[str, Any]:
"""Build the structured status snapshot that ``cmd_status`` renders as text.
Pure data, no stdout — safe for the MCP server. Reads only ``State`` and the
stdout-free leaf helpers; never constructs ``Config`` and never exposes the
Steam API key.
Args:
state: The loaded enforcer state.
Returns:
A JSON-ready dict describing the current enforcement status.
"""
total_block = get_total_block_status()
installed = get_installed_games()
real_games = [(aid, name) for aid, name in installed if not is_protected_app(aid)]
assigned_installed = (
any(aid == state.current_app_id for aid, _ in installed)
if state.current_app_id
else None
)
return {
"current_app_id": state.current_app_id,
"current_game_name": state.current_game_name or None,
"finished_count": len(state.finished_app_ids),
"store_blocked": is_store_blocked(),
"installed_count": len(real_games),
"assigned_game_installed": assigned_installed,
"total_block": {
"active": total_block.active,
"days_remaining": round(total_block.days_remaining, 1),
"until": total_block.until.isoformat() if total_block.until else None,
},
"manual_pick_locked": is_manual_pick_locked(state),
"manual_picks": [
{
**pick,
"grace_days_left": manual_pick_grace_remaining(state, pick["app_id"]),
}
for pick in active_manual_picks(state)
],
_MOVED_TO_MANUAL_PICK_LIFECYCLE = frozenset(
{
"abandon_manual_pick",
"can_abandon_manual_pick",
"manual_pick_grace_remaining",
"status_payload",
}
)
# Whatever the re-exported name turns out to be -- a function, a class or
# a constant. Aliased so the annotation is a name rather than a bare Any.
_Reexport: TypeAlias = Any
def __getattr__(name: str) -> _Reexport:
"""Re-export the names that moved to :mod:`_manual_pick_lifecycle`.
Deferred rather than imported at the top because _manual_pick_lifecycle imports
back from this module, so a module-level import would be circular.
"""
if name in _MOVED_TO_MANUAL_PICK_LIFECYCLE:
module = importlib.import_module(
"steam_backlog_enforcer._manual_pick_lifecycle",
)
return getattr(module, name)
msg = f"module {__name__!r} has no attribute {name!r}"
raise AttributeError(msg)

186
steam_backlog_enforcer/_cmd_done.py
View File

@ -2,21 +2,14 @@
from __future__ import annotations
import importlib
import logging
import sys
from typing import TYPE_CHECKING, Any, TypeAlias
from steam_backlog_enforcer._actions import allowed_app_ids
from steam_backlog_enforcer._enforce_loop import get_all_owned_app_ids
from steam_backlog_enforcer.config import Config, State, load_snapshot
from steam_backlog_enforcer.enforcer import (
enforce_allowed_game,
send_notification,
)
from steam_backlog_enforcer.config import State, load_snapshot
from steam_backlog_enforcer.game_install import (
_echo,
install_game,
is_game_installed,
uninstall_other_games,
)
from steam_backlog_enforcer.hltb import (
fetch_hltb_confidence_cached,
@ -25,9 +18,9 @@ from steam_backlog_enforcer.hltb import (
load_hltb_polls_cache,
save_hltb_cache,
)
from steam_backlog_enforcer.library_hider import try_hide_other_games
from steam_backlog_enforcer.scanning import pick_next_game
from steam_backlog_enforcer.steam_api import GameInfo, SteamAPIClient
if TYPE_CHECKING:
from steam_backlog_enforcer.steam_api import GameInfo
_REASSIGN_REFRESH_LIMIT = 50
_SKIP_DAYS = 7
@ -181,153 +174,28 @@ def _refresh_uncached_shortlist_hours(
hltb_cache.update(refreshed)
def _finalize_completion(
config: Config,
state: State,
game_name: str,
app_id: int,
) -> None:
"""Mark game complete, pick next, hide non-assigned games, notify."""
_echo(f"\n COMPLETED: {game_name}!")
state.finished_app_ids.append(app_id)
snapshot_data = load_snapshot()
_echo("\nPicking next game...")
if not snapshot_data:
_echo(" No snapshot found. Run 'scan' first.")
state.current_app_id = None
state.current_game_name = ""
state.save()
return
games = [GameInfo.from_snapshot(d) for d in snapshot_data]
hltb_cache = load_hltb_cache()
skip = set(state.finished_app_ids) | state.active_skipped_ids()
_refresh_uncached_shortlist_hours(games, hltb_cache, skip)
_apply_cached_hours_to_games(games, hltb_cache)
pick_next_game(games, state, config, on_select=_prompt_keep_or_skip)
if state.current_app_id is None:
_echo(" No more games to assign!")
return
owned_ids = get_all_owned_app_ids(config)
if owned_ids:
hidden, skipped = try_hide_other_games(owned_ids, allowed_app_ids(state))
if skipped is not None:
_echo(f"\n Library hiding: skipped ({skipped})")
elif hidden > 0:
_echo(f"\n Library: hid {hidden} games")
if not is_game_installed(state.current_app_id):
logger.info(
"Assigned game still missing after library reconciliation; "
"re-triggering install"
)
_echo(
"\n Assigned game still missing after library reconciliation; "
"re-triggering install..."
)
_echo(f"\n Auto-installing {state.current_game_name}...")
install_game(
state.current_app_id,
state.current_game_name,
config.steam_id,
use_steam_protocol=True,
)
send_notification(
"Game Complete!",
f"Finished {game_name}! Now playing: {state.current_game_name}",
)
_echo(f"\nAll done! Go play {state.current_game_name}!")
_MOVED_TO_CMD_DONE_FINALIZE = frozenset(
{
"_enforce_on_done",
"_finalize_completion",
"cmd_done",
}
)
def _enforce_on_done(config: Config, state: State) -> None:
"""Run a single enforcement pass during the 'done' command.
# Whatever the re-exported name turns out to be -- a function, a class or
# a constant. Aliased so the annotation is a name rather than a bare Any.
_Reexport: TypeAlias = Any
Kills unauthorized game processes, uninstalls unauthorized games,
and ensures the assigned game is installed.
def __getattr__(name: str) -> _Reexport:
"""Re-export the names that moved to :mod:`_cmd_done_finalize`.
Deferred rather than imported at the top because _cmd_done_finalize imports
back from this module, so a module-level import would be circular.
"""
if state.current_app_id is None:
return
if config.kill_unauthorized_games:
violations = enforce_allowed_game(
allowed_app_ids(state),
kill_unauthorized=True,
)
for pid, app_id in violations:
_echo(f" Killed unauthorized game: AppID={app_id} (PID={pid})")
if config.uninstall_other_games:
count = uninstall_other_games(allowed_app_ids(state))
if count:
_echo(f" Uninstalled {count} unauthorized game(s)")
if not is_game_installed(state.current_app_id):
_echo(f" Re-installing {state.current_game_name}...")
install_game(
state.current_app_id,
state.current_game_name,
config.steam_id,
use_steam_protocol=True,
)
# Reconcile library: hide non-assigned games and unhide the assigned one.
# Without this, an interrupted earlier completion can leave the new
# assigned game hidden and stale games visible.
owned_ids = get_all_owned_app_ids(config)
if owned_ids:
hidden, skipped = try_hide_other_games(owned_ids, allowed_app_ids(state))
if skipped is not None:
_echo(f" Library hiding: skipped ({skipped})")
elif hidden > 0:
_echo(f" Library: hid {hidden} games")
def cmd_done(config: Config, state: State) -> None:
"""Check completion, pick next game, uninstall & hide.
All-in-one command for after finishing a game:
1. Verify 100% achievements on Steam.
2. Pick the next game (shortest HLTB leisure+dlc time).
3. Uninstall all non-assigned games.
4. Hide all non-assigned games in the Steam library.
5. Install the newly assigned game.
"""
if state.current_app_id is None:
_echo("No game currently assigned. Run 'scan' first.")
return
client = SteamAPIClient(config.steam_api_key, config.steam_id)
game_name = state.current_game_name
app_id = state.current_app_id
_echo(f"Checking {game_name} (AppID={app_id})...")
game = client.refresh_single_game(app_id, game_name)
if game is None:
_echo(" Could not fetch achievement data from Steam.")
return
_echo(
f" Progress: {game.unlocked_achievements}/{game.total_achievements}"
f" ({game.completion_pct:.1f}%)"
)
hltb_cache = load_hltb_cache()
hours = hltb_cache.get(app_id, -1.0)
if hours < 0:
hltb_cache = fetch_hltb_times_cached([(app_id, game_name)])
hours = hltb_cache.get(app_id, -1.0)
if hours > 0:
_echo(f" HLTB leisure+dlc estimate: {hours:.1f} hours")
_report_assigned_confidence(app_id, state)
if not game.is_complete:
remaining = game.total_achievements - game.unlocked_achievements
_echo(f"\n NOT COMPLETE: {remaining} achievements remaining. Keep going!")
_enforce_on_done(config, state)
return
_finalize_completion(config, state, game_name, app_id)
if name in _MOVED_TO_CMD_DONE_FINALIZE:
module = importlib.import_module("steam_backlog_enforcer._cmd_done_finalize")
return getattr(module, name)
msg = f"module {__name__!r} has no attribute {name!r}"
raise AttributeError(msg)

192
steam_backlog_enforcer/_cmd_done_finalize.py Normal file
View File

@ -0,0 +1,192 @@
"""Finalising a completed game and re-enforcing afterwards.
Split out of :mod:`steam_backlog_enforcer._cmd_done` to keep both files
under the 250-line cap.
"""
from __future__ import annotations
import logging
from steam_backlog_enforcer._actions import allowed_app_ids
from steam_backlog_enforcer._cmd_done import (
_apply_cached_hours_to_games,
_prompt_keep_or_skip,
_refresh_uncached_shortlist_hours,
_report_assigned_confidence,
)
from steam_backlog_enforcer._enforce_loop import get_all_owned_app_ids
from steam_backlog_enforcer.config import Config, State, load_snapshot
from steam_backlog_enforcer.enforcer import (
enforce_allowed_game,
send_notification,
)
from steam_backlog_enforcer.game_install import (
_echo,
install_game,
is_game_installed,
uninstall_other_games,
)
from steam_backlog_enforcer.hltb import (
fetch_hltb_times_cached,
load_hltb_cache,
)
from steam_backlog_enforcer.library_hider import try_hide_other_games
from steam_backlog_enforcer.scanning import pick_next_game
from steam_backlog_enforcer.steam_api import GameInfo, SteamAPIClient
_REASSIGN_REFRESH_LIMIT = 50
_SKIP_DAYS = 7
logger = logging.getLogger(__name__)
def _finalize_completion(
config: Config,
state: State,
game_name: str,
app_id: int,
) -> None:
"""Mark game complete, pick next, hide non-assigned games, notify."""
_echo(f"\n COMPLETED: {game_name}!")
state.finished_app_ids.append(app_id)
snapshot_data = load_snapshot()
_echo("\nPicking next game...")
if not snapshot_data:
_echo(" No snapshot found. Run 'scan' first.")
state.current_app_id = None
state.current_game_name = ""
state.save()
return
games = [GameInfo.from_snapshot(d) for d in snapshot_data]
hltb_cache = load_hltb_cache()
skip = set(state.finished_app_ids) | state.active_skipped_ids()
_refresh_uncached_shortlist_hours(games, hltb_cache, skip)
_apply_cached_hours_to_games(games, hltb_cache)
pick_next_game(games, state, config, on_select=_prompt_keep_or_skip)
if state.current_app_id is None:
_echo(" No more games to assign!")
return
owned_ids = get_all_owned_app_ids(config)
if owned_ids:
hidden, skipped = try_hide_other_games(owned_ids, allowed_app_ids(state))
if skipped is not None:
_echo(f"\n Library hiding: skipped ({skipped})")
elif hidden > 0:
_echo(f"\n Library: hid {hidden} games")
if not is_game_installed(state.current_app_id):
logger.info(
"Assigned game still missing after library reconciliation; "
"re-triggering install"
)
_echo(
"\n Assigned game still missing after library reconciliation; "
"re-triggering install..."
)
_echo(f"\n Auto-installing {state.current_game_name}...")
install_game(
state.current_app_id,
state.current_game_name,
config.steam_id,
use_steam_protocol=True,
)
send_notification(
"Game Complete!",
f"Finished {game_name}! Now playing: {state.current_game_name}",
)
_echo(f"\nAll done! Go play {state.current_game_name}!")
def _enforce_on_done(config: Config, state: State) -> None:
"""Run a single enforcement pass during the 'done' command.
Kills unauthorized game processes, uninstalls unauthorized games,
and ensures the assigned game is installed.
"""
if state.current_app_id is None:
return
if config.kill_unauthorized_games:
violations = enforce_allowed_game(
allowed_app_ids(state),
kill_unauthorized=True,
)
for pid, app_id in violations:
_echo(f" Killed unauthorized game: AppID={app_id} (PID={pid})")
if config.uninstall_other_games:
count = uninstall_other_games(allowed_app_ids(state))
if count:
_echo(f" Uninstalled {count} unauthorized game(s)")
if not is_game_installed(state.current_app_id):
_echo(f" Re-installing {state.current_game_name}...")
install_game(
state.current_app_id,
state.current_game_name,
config.steam_id,
use_steam_protocol=True,
)
# Reconcile library: hide non-assigned games and unhide the assigned one.
# Without this, an interrupted earlier completion can leave the new
# assigned game hidden and stale games visible.
owned_ids = get_all_owned_app_ids(config)
if owned_ids:
hidden, skipped = try_hide_other_games(owned_ids, allowed_app_ids(state))
if skipped is not None:
_echo(f" Library hiding: skipped ({skipped})")
elif hidden > 0:
_echo(f" Library: hid {hidden} games")
def cmd_done(config: Config, state: State) -> None:
"""Check completion, pick next game, uninstall & hide.
All-in-one command for after finishing a game:
1. Verify 100% achievements on Steam.
2. Pick the next game (shortest HLTB leisure+dlc time).
3. Uninstall all non-assigned games.
4. Hide all non-assigned games in the Steam library.
5. Install the newly assigned game.
"""
if state.current_app_id is None:
_echo("No game currently assigned. Run 'scan' first.")
return
client = SteamAPIClient(config.steam_api_key, config.steam_id)
game_name = state.current_game_name
app_id = state.current_app_id
_echo(f"Checking {game_name} (AppID={app_id})...")
game = client.refresh_single_game(app_id, game_name)
if game is None:
_echo(" Could not fetch achievement data from Steam.")
return
_echo(
f" Progress: {game.unlocked_achievements}/{game.total_achievements}"
f" ({game.completion_pct:.1f}%)"
)
hltb_cache = load_hltb_cache()
hours = hltb_cache.get(app_id, -1.0)
if hours < 0:
hltb_cache = fetch_hltb_times_cached([(app_id, game_name)])
hours = hltb_cache.get(app_id, -1.0)
if hours > 0:
_echo(f" HLTB leisure+dlc estimate: {hours:.1f} hours")
_report_assigned_confidence(app_id, state)
if not game.is_complete:
remaining = game.total_achievements - game.unlocked_achievements
_echo(f"\n NOT COMPLETE: {remaining} achievements remaining. Keep going!")
_enforce_on_done(config, state)
return
_finalize_completion(config, state, game_name, app_id)

95
steam_backlog_enforcer/_hltb_cached.py Normal file
View File

@ -0,0 +1,95 @@
"""Cached batch lookup of HowLongToBeat times.
Split out of :mod:`steam_backlog_enforcer.hltb` to keep both files
under the 250-line cap.
"""
from __future__ import annotations
import logging
import time
from steam_backlog_enforcer._hltb_types import (
ProgressCb,
_HLTBExtras,
load_hltb_cache,
load_hltb_count_comp_cache,
load_hltb_leisure_100h_cache,
load_hltb_polls_cache,
load_hltb_rush_cache,
save_hltb_cache,
)
from steam_backlog_enforcer.hltb import fetch_hltb_times
logger = logging.getLogger(__name__)
# ──────────────────────────────────────────────────────────────
# Confidence-only batch fetch (no leisure/DLC detail pages)
# ──────────────────────────────────────────────────────────────
def fetch_hltb_times_cached(
games: list[tuple[int, str]],
progress_cb: ProgressCb | None = None,
) -> dict[int, float]:
"""Fetch HLTB times, using disk cache for already-known games.
Args:
games: list of (app_id, name) tuples to look up.
progress_cb: optional callback(done, total, found, game_name).
Returns: dict mapping app_id -> completionist_hours.
"""
cache = load_hltb_cache()
polls = load_hltb_polls_cache()
extras = _HLTBExtras(
count_comp=load_hltb_count_comp_cache(),
rush=load_hltb_rush_cache(),
leisure_100h=load_hltb_leisure_100h_cache(),
)
uncached = [(app_id, name) for app_id, name in games if app_id not in cache]
if uncached:
logger.info(
"Fetching HLTB data for %d uncached games (%d cached)...",
len(uncached),
len(games) - len(uncached),
)
t0 = time.monotonic()
fetch_hltb_times(
uncached,
cache=cache,
polls=polls,
progress_cb=progress_cb,
extras=extras,
)
elapsed = time.monotonic() - t0
# Final save.
save_hltb_cache(cache, polls, extras)
found = sum(1 for aid, _ in uncached if cache.get(aid, -1) > 0)
rate = len(uncached) / elapsed if elapsed > 0 else 0
logger.info(
"HLTB fetch done: %d/%d found in %.1fs (%.0f games/s)",
found,
len(uncached),
elapsed,
rate,
)
else:
logger.info("All %d games found in HLTB cache.", len(games))
return cache
_REEXPORTED_TYPES = frozenset({"HLTB_BASE_URL"})
_MOVED_TO_HLTB_CONFIDENCE = frozenset(
{
"fetch_hltb_confidence_cached",
"fetch_hltb_detail_missing",
"get_hltb_submit_url",
}
)

187
steam_backlog_enforcer/_hltb_confidence.py Normal file
View File

@ -0,0 +1,187 @@
"""Cached HowLongToBeat confidence lookups.
Split out of :mod:`steam_backlog_enforcer.hltb` to keep both files
under the 250-line cap.
"""
from __future__ import annotations
import logging
import time
from steam_backlog_enforcer._hltb_types import (
HLTB_BASE_URL,
ProgressCb,
_HLTBExtras,
load_hltb_cache,
load_hltb_count_comp_cache,
load_hltb_game_id_cache,
load_hltb_leisure_100h_cache,
load_hltb_polls_cache,
load_hltb_rush_cache,
save_hltb_cache,
)
from steam_backlog_enforcer.hltb import (
fetch_hltb_confidence,
fetch_hltb_times,
)
logger = logging.getLogger(__name__)
# ──────────────────────────────────────────────────────────────
# Confidence-only batch fetch (no leisure/DLC detail pages)
# ──────────────────────────────────────────────────────────────
def fetch_hltb_confidence_cached(
games: list[tuple[int, str]],
progress_cb: ProgressCb | None = None,
) -> dict[int, float]:
"""Fetch HLTB search-level confidence data, using disk cache for known IDs."""
cache = load_hltb_cache()
polls = load_hltb_polls_cache()
count_comp = load_hltb_count_comp_cache()
uncached = [(app_id, name) for app_id, name in games if app_id not in cache]
if uncached:
logger.info(
"Fetching HLTB confidence for %d uncached games (%d cached)...",
len(uncached),
len(games) - len(uncached),
)
t0 = time.monotonic()
fetch_hltb_confidence(
uncached,
cache=cache,
polls=polls,
progress_cb=progress_cb,
count_comp=count_comp,
)
elapsed = time.monotonic() - t0
save_hltb_cache(cache, polls, _HLTBExtras(count_comp=count_comp))
found = sum(1 for aid, _ in uncached if cache.get(aid, -1) > 0)
rate = len(uncached) / elapsed if elapsed > 0 else 0
logger.info(
"HLTB confidence fetch done: %d/%d found in %.1fs (%.0f games/s)",
found,
len(uncached),
elapsed,
rate,
)
else:
logger.info("All %d games found in HLTB cache.", len(games))
return cache
def fetch_hltb_detail_missing(
games: list[tuple[int, str]],
progress_cb: ProgressCb | None = None,
) -> int:
"""Fetch HLTB detail (rush + leisure) for games that are missing it.
Also backfills ``hltb_game_id`` for any game that already has rush/leisure
data but whose HLTB game ID was never stored (e.g. from an old cache).
Games with both rush data and a game_id are skipped entirely. For the
rest, temporarily removes them from the hours cache so ``fetch_hltb_times``
will visit their detail pages. Restores prior hours for any game the
re-fetch doesn't find.
Args:
games: list of (app_id, name) tuples to check.
progress_cb: optional progress callback.
Returns:
Number of games that now have rush-hour data after the fetch.
"""
rush = load_hltb_rush_cache()
game_id_cache = load_hltb_game_id_cache()
missing_rush = [
(app_id, name) for app_id, name in games if rush.get(app_id, -1) <= 0
]
# Also re-search games that have rush data but no HLTB game ID yet so the
# direct URL can be shown in stats output.
missing_id_only = [
(app_id, name)
for app_id, name in games
if rush.get(app_id, -1) > 0 and game_id_cache.get(app_id, 0) == 0
]
missing = missing_rush + missing_id_only
if not missing:
return 0
cache = load_hltb_cache()
polls = load_hltb_polls_cache()
extras = _HLTBExtras(
count_comp=load_hltb_count_comp_cache(),
rush=rush,
leisure_100h=load_hltb_leisure_100h_cache(),
hltb_game_id=game_id_cache,
)
# Remove from hours cache so fetch_hltb_times will visit the detail page.
prior_hours: dict[int, float] = {}
for app_id, _ in missing:
prior_hours[app_id] = cache.pop(app_id, -1.0)
n_rush = len(missing_rush)
n_id = len(missing_id_only)
if n_rush and n_id:
logger.info(
"Fetching HLTB detail for %d games missing rush/leisure data"
" + %d games missing game ID...",
n_rush,
n_id,
)
elif n_rush:
logger.info(
"Fetching HLTB detail for %d games missing rush/leisure data...", n_rush
)
else:
logger.info("Backfilling HLTB game ID for %d game(s)...", n_id)
t0 = time.monotonic()
fetch_hltb_times(
missing,
cache=cache,
polls=polls,
progress_cb=progress_cb,
extras=extras,
)
elapsed = time.monotonic() - t0
# Restore prior hours for games the detail fetch didn't re-find.
for app_id, old_hours in prior_hours.items():
if old_hours > 0 and cache.get(app_id, -1.0) <= 0:
cache[app_id] = old_hours
save_hltb_cache(cache, polls, extras)
fetched = sum(1 for app_id, _ in missing_rush if extras.rush.get(app_id, -1) > 0)
rate = len(missing) / elapsed if elapsed > 0 else 0
logger.info(
"HLTB detail fetch done: %d/%d got rush data in %.1fs (%.0f games/s)",
fetched,
len(missing_rush),
elapsed,
rate,
)
return fetched
def get_hltb_submit_url(game_name: str) -> str | None:
"""Look up a game on HLTB and return its submit page URL.
Args:
game_name: Name of the game to search for.
Returns:
URL like ``https://howlongtobeat.com/submit/game/12345``,
or ``None`` if the game wasn't found.
"""
results = fetch_hltb_times([(0, game_name)])
if results and results[0].hltb_game_id:
return f"{HLTB_BASE_URL}/submit/game/{results[0].hltb_game_id}"
return None

166
steam_backlog_enforcer/_manual_pick_lifecycle.py Normal file
View File

@ -0,0 +1,166 @@
"""Grace periods, abandonment and the status payload for manual picks.
Split out of :mod:`steam_backlog_enforcer._actions` to keep both files
under the 250-line cap.
"""
from __future__ import annotations
from datetime import datetime, timedelta, timezone
from typing import TYPE_CHECKING, Any
from steam_backlog_enforcer._actions import (
find_manual_pick,
is_manual_pick_locked,
)
from steam_backlog_enforcer._allowed_games import (
active_manual_picks,
allowed_games,
)
from steam_backlog_enforcer._total_block import get_total_block_status
from steam_backlog_enforcer.game_install import get_installed_games, is_protected_app
from steam_backlog_enforcer.store_blocker import is_store_blocked
# Marks these two as an intentional re-export (they're imported from
# _allowed_games above, not defined here) -- without this, mypy's
# --no-implicit-reexport flags every downstream `from _actions import
# active_manual_picks` as an error. Everything else in this file is
# accessible normally; this only needs to list names imported-then-reexported.
__all__ = [
"active_manual_picks",
"allowed_games",
]
if TYPE_CHECKING:
from steam_backlog_enforcer.config import State
# Mistake-correction window: for this many days after a manual pick the user
# may still back out via ``abandon-pick``. Deliberately short — it exists to
# undo a wrong pick, not to serve as an escape hatch from the lock.
MANUAL_GRACE_DAYS = 4
# How long an abandoned pick stays out of the auto-assignment pool, so that
# ``scan`` does not immediately hand back the game the user just rejected.
ABANDON_COOLDOWN_DAYS = 30
def manual_pick_grace_remaining(state: State, app_id: int) -> float | None:
"""Return days left in *app_id*'s grace window, or ``None``.
``None`` means "no grace window applies": *app_id* is not an active manual
pick, or its timestamp is missing/malformed. A value <= 0 means the window
has closed.
Args:
state: The loaded enforcer state.
app_id: The manually-picked app id to measure.
Returns:
Fractional days remaining before the pick can no longer be abandoned,
or ``None`` when the question does not apply.
"""
pick = find_manual_pick(state, app_id)
if pick is None or not pick.get("started_at"):
return None
try:
started = datetime.fromisoformat(pick["started_at"])
except ValueError:
return None
deadline = started + timedelta(days=MANUAL_GRACE_DAYS)
return (deadline - datetime.now(timezone.utc)).total_seconds() / 86400
def can_abandon_manual_pick(state: State, app_id: int) -> bool:
"""Return ``True`` if *app_id* is still inside its grace window.
Args:
state: The loaded enforcer state.
app_id: The manually-picked app id to check.
Returns:
Whether ``abandon_manual_pick`` would be accepted right now.
"""
remaining = manual_pick_grace_remaining(state, app_id)
return remaining is not None and remaining > 0
def abandon_manual_pick(state: State, app_id: int) -> bool:
"""Drop one manual pick and persist ``state``, if still inside its grace window.
Only the named pick is dropped: any other active pick keeps its own lock
and deadline. The abandoned app id goes onto the existing skip cooldown so
``scan`` will not hand the same game straight back. State-only: like
``apply_manual_pick`` this performs no uninstall/hide cascade, so the MCP
server can call it without touching the filesystem.
Args:
state: The enforcer state to mutate and save.
app_id: The manually-picked app id to back out of.
Returns:
``True`` if the pick was abandoned, ``False`` if its grace window has
closed (or it is not an active pick), in which case ``state`` is
untouched.
"""
if not can_abandon_manual_pick(state, app_id):
return False
state.skip_for_days(app_id, ABANDON_COOLDOWN_DAYS)
state.manual_picks = [
p for p in active_manual_picks(state) if p.get("app_id") != app_id
]
# The abandoned pick may also have been the current assignment; hand the
# assignment to a surviving pick so the enforcer keeps guarding it, or
# clear it so 'scan' can reassign.
if state.current_app_id == app_id:
survivor = state.manual_picks[-1] if state.manual_picks else None
state.current_app_id = survivor["app_id"] if survivor else None
state.current_game_name = survivor["game_name"] if survivor else ""
state.save()
return True
def status_payload(state: State) -> dict[str, Any]:
"""Build the structured status snapshot that ``cmd_status`` renders as text.
Pure data, no stdout — safe for the MCP server. Reads only ``State`` and the
stdout-free leaf helpers; never constructs ``Config`` and never exposes the
Steam API key.
Args:
state: The loaded enforcer state.
Returns:
A JSON-ready dict describing the current enforcement status.
"""
total_block = get_total_block_status()
installed = get_installed_games()
real_games = [(aid, name) for aid, name in installed if not is_protected_app(aid)]
assigned_installed = (
any(aid == state.current_app_id for aid, _ in installed)
if state.current_app_id
else None
)
return {
"current_app_id": state.current_app_id,
"current_game_name": state.current_game_name or None,
"finished_count": len(state.finished_app_ids),
"store_blocked": is_store_blocked(),
"installed_count": len(real_games),
"assigned_game_installed": assigned_installed,
"total_block": {
"active": total_block.active,
"days_remaining": round(total_block.days_remaining, 1),
"until": total_block.until.isoformat() if total_block.until else None,
},
"manual_pick_locked": is_manual_pick_locked(state),
"manual_picks": [
{
**pick,
"grace_days_left": manual_pick_grace_remaining(state, pick["app_id"]),
}
for pick in active_manual_picks(state)
],
}

208
steam_backlog_enforcer/_whitelist.py
View File

@ -2,20 +2,14 @@
from __future__ import annotations
import array
from collections import Counter
import contextlib
import fcntl
import json
import importlib
import logging
import math
import re
import shutil
import subprocess
import time
from typing import TYPE_CHECKING, cast
from typing import TYPE_CHECKING, Any, TypeAlias
from steam_backlog_enforcer.config import CONFIG_DIR, _atomic_write
from steam_backlog_enforcer.config import CONFIG_DIR
if TYPE_CHECKING:
from pathlib import Path
@ -119,180 +113,34 @@ def validate_reason(reason: str) -> str | None:
_FS_IOC_GETFLAGS = 0x80086601
_FS_IMMUTABLE_FL = 0x00000010
_MOVED_TO_WHITELIST_LOCKING = frozenset(
{
"_append_audit_log",
"_immutable_flag_is",
"_load_approved",
"_save_approved",
"_try_set_immutable",
"add_pending_exception",
"get_approved_exception_ids",
"lock_enforcement_files",
"unlock_for_write",
}
)
def _immutable_flag_is(path: Path, *, immutable: bool) -> bool:
"""Report whether *path*'s immutable bit already equals *immutable*.
Reading the flag is an ioctl, which costs no process; ``chattr`` costs a
fork and an exec. Since ``lock_enforcement_files`` runs on every
enforce-loop iteration, asking first turns a steady ~0.65 execs/second
into none at all for a flag that essentially never changes.
# Whatever the re-exported name turns out to be -- a function, a class or
# a constant. Aliased so the annotation is a name rather than a bare Any.
_Reexport: TypeAlias = Any
Fails closed: any uncertainty at all — filesystem without flag support,
ioctl refused, file unreadable — reports False so the caller still shells
out to chattr. A wrong True would silently stop enforcing immutability.
Args:
path: File to inspect.
immutable: The state to compare against.
def __getattr__(name: str) -> _Reexport:
"""Re-export the names that moved to :mod:`_whitelist_locking`.
Returns:
True only if the flag could be read and already matches.
Deferred rather than imported at the top because _whitelist_locking imports
back from this module, so a module-level import would be circular.
"""
buf = array.array("L", [0])
try:
with path.open("rb") as handle:
# mutate_flag defaults to True, so buf receives the flags in place.
fcntl.ioctl(handle.fileno(), _FS_IOC_GETFLAGS, buf)
except (OSError, ValueError):
return False
return bool(buf[0] & _FS_IMMUTABLE_FL) is immutable
def _try_set_immutable(path: Path, *, immutable: bool) -> None:
"""Silently attempt to set or clear the immutable flag on *path*.
This is a best-effort operation — it fails silently if chattr is not
available, the process lacks the required capability, or the filesystem
does not support the flag. When the flag already holds the requested
value, no subprocess is spawned at all.
Args:
path: File to modify.
immutable: True to set +i, False to clear -i.
"""
if not path.exists():
return
if _immutable_flag_is(path, immutable=immutable):
return
chattr = shutil.which("chattr")
if chattr is None:
return
flag = "+i" if immutable else "-i"
with contextlib.suppress(OSError, subprocess.TimeoutExpired):
subprocess.run(
[chattr, flag, str(path)],
capture_output=True,
check=False,
timeout=5,
)
def lock_enforcement_files(config_file: Path) -> None:
"""Apply chattr +i to enforcement-critical config files.
Called at the end of each enforce-loop iteration. Requires that the
daemon is running as root (or has CAP_LINUX_IMMUTABLE).
Args:
config_file: Path to the main config.json.
"""
_try_set_immutable(config_file, immutable=True)
_try_set_immutable(APPROVED_EXCEPTIONS_FILE, immutable=True)
def unlock_for_write(path: Path) -> None:
"""Clear the immutable flag before writing *path*.
Args:
path: File to unlock.
"""
_try_set_immutable(path, immutable=False)
# ──────────────────────────────────────────────────────────────
# Persistence helpers
# ──────────────────────────────────────────────────────────────
def _load_approved() -> list[dict[str, object]]:
"""Load approved exception entries from disk."""
if not APPROVED_EXCEPTIONS_FILE.exists():
return []
try:
data: object = json.loads(APPROVED_EXCEPTIONS_FILE.read_text(encoding="utf-8"))
if isinstance(data, list):
return cast("list[dict[str, object]]", data)
except (json.JSONDecodeError, OSError, ValueError):
pass
return []
def _save_approved(entries: list[dict[str, object]]) -> None:
"""Persist approved exception entries to disk."""
unlock_for_write(APPROVED_EXCEPTIONS_FILE)
_atomic_write(APPROVED_EXCEPTIONS_FILE, json.dumps(entries, indent=2) + "\n")
_try_set_immutable(APPROVED_EXCEPTIONS_FILE, immutable=True)
def _append_audit_log(app_id: int, reason: str, event: str) -> None:
"""Append one line to the append-only audit log.
Each line has the format::
ISO-TIMESTAMP | EVENT | app_id=NNN | reason='...'
Args:
app_id: Steam application ID involved.
reason: Justification text supplied by the user.
event: Short event label such as ``REQUESTED`` or ``APPROVED``.
"""
timestamp = time.strftime("%Y-%m-%dT%H:%M:%SZ", time.gmtime())
line = f"{timestamp} | {event} | app_id={app_id} | reason={reason!r}\n"
EXCEPTION_AUDIT_LOG.parent.mkdir(parents=True, exist_ok=True)
with EXCEPTION_AUDIT_LOG.open("a", encoding="utf-8") as fh:
fh.write(line)
# ──────────────────────────────────────────────────────────────
# Public API
# ──────────────────────────────────────────────────────────────
def add_pending_exception(app_id: int, reason: str) -> str:
"""Add a whitelist exception for *app_id* immediately.
The entry becomes active right away (no cooldown). Returns a
human-readable status message.
Args:
app_id: Steam application ID to add.
reason: Validated justification text (must pass :func:`validate_reason`).
Returns:
Human-readable confirmation message.
Raises:
ValueError: If the reason fails validation or the ID is already approved.
"""
err = validate_reason(reason)
if err is not None:
raise ValueError(err)
approved = _load_approved()
if any(int(e["app_id"]) == app_id for e in approved):
msg = f"AppID {app_id} is already in the approved exceptions list."
raise ValueError(msg)
now = time.time()
approved.append(
{
"app_id": app_id,
"reason": reason,
"approved_at": now,
}
)
_save_approved(approved)
_append_audit_log(app_id, reason, "APPROVED")
return f"Exception approved for AppID {app_id}. Active immediately. Reason logged."
def get_approved_exception_ids() -> frozenset[int]:
"""Return the frozenset of currently approved exception app IDs.
Returns:
Frozenset of approved app IDs.
"""
approved = _load_approved()
return frozenset(int(e["app_id"]) for e in approved)
if name in _MOVED_TO_WHITELIST_LOCKING:
module = importlib.import_module("steam_backlog_enforcer._whitelist_locking")
return getattr(module, name)
msg = f"module {__name__!r} has no attribute {name!r}"
raise AttributeError(msg)

229
steam_backlog_enforcer/_whitelist_locking.py Normal file
View File

@ -0,0 +1,229 @@
"""Making enforcement files immutable so they cannot be edited mid-block.
Split out of :mod:`steam_backlog_enforcer._whitelist` to keep both files
under the 250-line cap.
"""
from __future__ import annotations
import array
import contextlib
import fcntl
import json
import logging
import shutil
import subprocess
import time
from typing import TYPE_CHECKING, cast
from steam_backlog_enforcer._whitelist import (
_FS_IMMUTABLE_FL,
_FS_IOC_GETFLAGS,
validate_reason,
)
from steam_backlog_enforcer.config import CONFIG_DIR, _atomic_write
if TYPE_CHECKING:
from pathlib import Path
logger = logging.getLogger(__name__)
# ──────────────────────────────────────────────────────────────
# File paths (patched in tests via conftest)
# ──────────────────────────────────────────────────────────────
APPROVED_EXCEPTIONS_FILE: Path = CONFIG_DIR / "approved_exceptions.json"
EXCEPTION_AUDIT_LOG: Path = CONFIG_DIR / "exception_audit.log"
# ──────────────────────────────────────────────────────────────
# Constants
# ──────────────────────────────────────────────────────────────
_MIN_REASON_WORDS: int = 5
_MIN_REASON_LENGTH: int = 25
_MIN_ENTROPY: float = 3.0
# Reject runs of the same character longer than this (e.g. "aaaa").
_MAX_CHAR_RUN: int = 3
# ──────────────────────────────────────────────────────────────
# Reason validation
# ──────────────────────────────────────────────────────────────
def _immutable_flag_is(path: Path, *, immutable: bool) -> bool:
"""Report whether *path*'s immutable bit already equals *immutable*.
Reading the flag is an ioctl, which costs no process; ``chattr`` costs a
fork and an exec. Since ``lock_enforcement_files`` runs on every
enforce-loop iteration, asking first turns a steady ~0.65 execs/second
into none at all for a flag that essentially never changes.
Fails closed: any uncertainty at all — filesystem without flag support,
ioctl refused, file unreadable — reports False so the caller still shells
out to chattr. A wrong True would silently stop enforcing immutability.
Args:
path: File to inspect.
immutable: The state to compare against.
Returns:
True only if the flag could be read and already matches.
"""
buf = array.array("L", [0])
try:
with path.open("rb") as handle:
# mutate_flag defaults to True, so buf receives the flags in place.
fcntl.ioctl(handle.fileno(), _FS_IOC_GETFLAGS, buf)
except (OSError, ValueError):
return False
return bool(buf[0] & _FS_IMMUTABLE_FL) is immutable
def _try_set_immutable(path: Path, *, immutable: bool) -> None:
"""Silently attempt to set or clear the immutable flag on *path*.
This is a best-effort operation — it fails silently if chattr is not
available, the process lacks the required capability, or the filesystem
does not support the flag. When the flag already holds the requested
value, no subprocess is spawned at all.
Args:
path: File to modify.
immutable: True to set +i, False to clear -i.
"""
if not path.exists():
return
if _immutable_flag_is(path, immutable=immutable):
return
chattr = shutil.which("chattr")
if chattr is None:
return
flag = "+i" if immutable else "-i"
with contextlib.suppress(OSError, subprocess.TimeoutExpired):
subprocess.run(
[chattr, flag, str(path)],
capture_output=True,
check=False,
timeout=5,
)
def lock_enforcement_files(config_file: Path) -> None:
"""Apply chattr +i to enforcement-critical config files.
Called at the end of each enforce-loop iteration. Requires that the
daemon is running as root (or has CAP_LINUX_IMMUTABLE).
Args:
config_file: Path to the main config.json.
"""
_try_set_immutable(config_file, immutable=True)
_try_set_immutable(APPROVED_EXCEPTIONS_FILE, immutable=True)
def unlock_for_write(path: Path) -> None:
"""Clear the immutable flag before writing *path*.
Args:
path: File to unlock.
"""
_try_set_immutable(path, immutable=False)
# ──────────────────────────────────────────────────────────────
# Persistence helpers
# ──────────────────────────────────────────────────────────────
def _load_approved() -> list[dict[str, object]]:
"""Load approved exception entries from disk."""
if not APPROVED_EXCEPTIONS_FILE.exists():
return []
try:
data: object = json.loads(APPROVED_EXCEPTIONS_FILE.read_text(encoding="utf-8"))
if isinstance(data, list):
return cast("list[dict[str, object]]", data)
except (json.JSONDecodeError, OSError, ValueError):
pass
return []
def _save_approved(entries: list[dict[str, object]]) -> None:
"""Persist approved exception entries to disk."""
unlock_for_write(APPROVED_EXCEPTIONS_FILE)
_atomic_write(APPROVED_EXCEPTIONS_FILE, json.dumps(entries, indent=2) + "\n")
_try_set_immutable(APPROVED_EXCEPTIONS_FILE, immutable=True)
def _append_audit_log(app_id: int, reason: str, event: str) -> None:
"""Append one line to the append-only audit log.
Each line has the format::
ISO-TIMESTAMP | EVENT | app_id=NNN | reason='...'
Args:
app_id: Steam application ID involved.
reason: Justification text supplied by the user.
event: Short event label such as ``REQUESTED`` or ``APPROVED``.
"""
timestamp = time.strftime("%Y-%m-%dT%H:%M:%SZ", time.gmtime())
line = f"{timestamp} | {event} | app_id={app_id} | reason={reason!r}\n"
EXCEPTION_AUDIT_LOG.parent.mkdir(parents=True, exist_ok=True)
with EXCEPTION_AUDIT_LOG.open("a", encoding="utf-8") as fh:
fh.write(line)
# ──────────────────────────────────────────────────────────────
# Public API
# ──────────────────────────────────────────────────────────────
def add_pending_exception(app_id: int, reason: str) -> str:
"""Add a whitelist exception for *app_id* immediately.
The entry becomes active right away (no cooldown). Returns a
human-readable status message.
Args:
app_id: Steam application ID to add.
reason: Validated justification text (must pass :func:`validate_reason`).
Returns:
Human-readable confirmation message.
Raises:
ValueError: If the reason fails validation or the ID is already approved.
"""
err = validate_reason(reason)
if err is not None:
raise ValueError(err)
approved = _load_approved()
if any(int(e["app_id"]) == app_id for e in approved):
msg = f"AppID {app_id} is already in the approved exceptions list."
raise ValueError(msg)
now = time.time()
approved.append(
{
"app_id": app_id,
"reason": reason,
"approved_at": now,
}
)
_save_approved(approved)
_append_audit_log(app_id, reason, "APPROVED")
return f"Exception approved for AppID {app_id}. Active immediately. Reason logged."
def get_approved_exception_ids() -> frozenset[int]:
"""Return the frozenset of currently approved exception app IDs.
Returns:
Frozenset of approved app IDs.
"""
approved = _load_approved()
return frozenset(int(e["app_id"]) for e in approved)

247
steam_backlog_enforcer/hltb.py
View File

@ -13,8 +13,9 @@ Fetches leisure completionist hour estimates from howlongtobeat.com with:
from __future__ import annotations
import asyncio
import importlib
import logging
import time
from typing import Any, TypeAlias
import aiohttp
@ -26,18 +27,10 @@ from steam_backlog_enforcer._hltb_search import (
_SearchCtx,
)
from steam_backlog_enforcer._hltb_types import (
HLTB_BASE_URL,
MAX_CONCURRENT,
HLTBResult,
ProgressCb,
_HLTBExtras,
load_hltb_cache,
load_hltb_count_comp_cache,
load_hltb_game_id_cache,
load_hltb_leisure_100h_cache,
load_hltb_polls_cache,
load_hltb_rush_cache,
save_hltb_cache,
)
logger = logging.getLogger(__name__)
@ -176,209 +169,39 @@ def fetch_hltb_confidence(
)
def fetch_hltb_times_cached(
games: list[tuple[int, str]],
progress_cb: ProgressCb | None = None,
) -> dict[int, float]:
"""Fetch HLTB times, using disk cache for already-known games.
_REEXPORTED = {
"HLTB_BASE_URL": "steam_backlog_enforcer._hltb_types",
"MAX_CONCURRENT": "steam_backlog_enforcer._hltb_types",
"HLTBResult": "steam_backlog_enforcer._hltb_types",
"ProgressCb": "steam_backlog_enforcer._hltb_types",
"_HLTBExtras": "steam_backlog_enforcer._hltb_types",
"load_hltb_cache": "steam_backlog_enforcer._hltb_types",
"load_hltb_count_comp_cache": "steam_backlog_enforcer._hltb_types",
"load_hltb_game_id_cache": "steam_backlog_enforcer._hltb_types",
"load_hltb_leisure_100h_cache": "steam_backlog_enforcer._hltb_types",
"load_hltb_polls_cache": "steam_backlog_enforcer._hltb_types",
"load_hltb_rush_cache": "steam_backlog_enforcer._hltb_types",
"save_hltb_cache": "steam_backlog_enforcer._hltb_types",
"fetch_hltb_times_cached": "steam_backlog_enforcer._hltb_cached",
"fetch_hltb_confidence_cached": "steam_backlog_enforcer._hltb_confidence",
"fetch_hltb_detail_missing": "steam_backlog_enforcer._hltb_confidence",
"get_hltb_submit_url": "steam_backlog_enforcer._hltb_confidence",
}
Args:
games: list of (app_id, name) tuples to look up.
progress_cb: optional callback(done, total, found, game_name).
Returns: dict mapping app_id -> completionist_hours.
# Whatever the re-exported name turns out to be -- a function, a class or
# a constant. Aliased so the annotation is a name rather than a bare Any.
_Reexport: TypeAlias = Any
def __getattr__(name: str) -> _Reexport:
"""Serve names that moved out of this module when it was split.
Deferred via importlib rather than imported at the top: the modules below
import back from here, so a module-level import would be circular.
"""
cache = load_hltb_cache()
polls = load_hltb_polls_cache()
extras = _HLTBExtras(
count_comp=load_hltb_count_comp_cache(),
rush=load_hltb_rush_cache(),
leisure_100h=load_hltb_leisure_100h_cache(),
)
uncached = [(app_id, name) for app_id, name in games if app_id not in cache]
if uncached:
logger.info(
"Fetching HLTB data for %d uncached games (%d cached)...",
len(uncached),
len(games) - len(uncached),
)
t0 = time.monotonic()
fetch_hltb_times(
uncached,
cache=cache,
polls=polls,
progress_cb=progress_cb,
extras=extras,
)
elapsed = time.monotonic() - t0
# Final save.
save_hltb_cache(cache, polls, extras)
found = sum(1 for aid, _ in uncached if cache.get(aid, -1) > 0)
rate = len(uncached) / elapsed if elapsed > 0 else 0
logger.info(
"HLTB fetch done: %d/%d found in %.1fs (%.0f games/s)",
found,
len(uncached),
elapsed,
rate,
)
else:
logger.info("All %d games found in HLTB cache.", len(games))
return cache
def fetch_hltb_confidence_cached(
games: list[tuple[int, str]],
progress_cb: ProgressCb | None = None,
) -> dict[int, float]:
"""Fetch HLTB search-level confidence data, using disk cache for known IDs."""
cache = load_hltb_cache()
polls = load_hltb_polls_cache()
count_comp = load_hltb_count_comp_cache()
uncached = [(app_id, name) for app_id, name in games if app_id not in cache]
if uncached:
logger.info(
"Fetching HLTB confidence for %d uncached games (%d cached)...",
len(uncached),
len(games) - len(uncached),
)
t0 = time.monotonic()
fetch_hltb_confidence(
uncached,
cache=cache,
polls=polls,
progress_cb=progress_cb,
count_comp=count_comp,
)
elapsed = time.monotonic() - t0
save_hltb_cache(cache, polls, _HLTBExtras(count_comp=count_comp))
found = sum(1 for aid, _ in uncached if cache.get(aid, -1) > 0)
rate = len(uncached) / elapsed if elapsed > 0 else 0
logger.info(
"HLTB confidence fetch done: %d/%d found in %.1fs (%.0f games/s)",
found,
len(uncached),
elapsed,
rate,
)
else:
logger.info("All %d games found in HLTB cache.", len(games))
return cache
def fetch_hltb_detail_missing(
games: list[tuple[int, str]],
progress_cb: ProgressCb | None = None,
) -> int:
"""Fetch HLTB detail (rush + leisure) for games that are missing it.
Also backfills ``hltb_game_id`` for any game that already has rush/leisure
data but whose HLTB game ID was never stored (e.g. from an old cache).
Games with both rush data and a game_id are skipped entirely. For the
rest, temporarily removes them from the hours cache so ``fetch_hltb_times``
will visit their detail pages. Restores prior hours for any game the
re-fetch doesn't find.
Args:
games: list of (app_id, name) tuples to check.
progress_cb: optional progress callback.
Returns:
Number of games that now have rush-hour data after the fetch.
"""
rush = load_hltb_rush_cache()
game_id_cache = load_hltb_game_id_cache()
missing_rush = [
(app_id, name) for app_id, name in games if rush.get(app_id, -1) <= 0
]
# Also re-search games that have rush data but no HLTB game ID yet so the
# direct URL can be shown in stats output.
missing_id_only = [
(app_id, name)
for app_id, name in games
if rush.get(app_id, -1) > 0 and game_id_cache.get(app_id, 0) == 0
]
missing = missing_rush + missing_id_only
if not missing:
return 0
cache = load_hltb_cache()
polls = load_hltb_polls_cache()
extras = _HLTBExtras(
count_comp=load_hltb_count_comp_cache(),
rush=rush,
leisure_100h=load_hltb_leisure_100h_cache(),
hltb_game_id=game_id_cache,
)
# Remove from hours cache so fetch_hltb_times will visit the detail page.
prior_hours: dict[int, float] = {}
for app_id, _ in missing:
prior_hours[app_id] = cache.pop(app_id, -1.0)
n_rush = len(missing_rush)
n_id = len(missing_id_only)
if n_rush and n_id:
logger.info(
"Fetching HLTB detail for %d games missing rush/leisure data"
" + %d games missing game ID...",
n_rush,
n_id,
)
elif n_rush:
logger.info(
"Fetching HLTB detail for %d games missing rush/leisure data...", n_rush
)
else:
logger.info("Backfilling HLTB game ID for %d game(s)...", n_id)
t0 = time.monotonic()
fetch_hltb_times(
missing,
cache=cache,
polls=polls,
progress_cb=progress_cb,
extras=extras,
)
elapsed = time.monotonic() - t0
# Restore prior hours for games the detail fetch didn't re-find.
for app_id, old_hours in prior_hours.items():
if old_hours > 0 and cache.get(app_id, -1.0) <= 0:
cache[app_id] = old_hours
save_hltb_cache(cache, polls, extras)
fetched = sum(1 for app_id, _ in missing_rush if extras.rush.get(app_id, -1) > 0)
rate = len(missing) / elapsed if elapsed > 0 else 0
logger.info(
"HLTB detail fetch done: %d/%d got rush data in %.1fs (%.0f games/s)",
fetched,
len(missing_rush),
elapsed,
rate,
)
return fetched
def get_hltb_submit_url(game_name: str) -> str | None:
"""Look up a game on HLTB and return its submit page URL.
Args:
game_name: Name of the game to search for.
Returns:
URL like ``https://howlongtobeat.com/submit/game/12345``,
or ``None`` if the game wasn't found.
"""
results = fetch_hltb_times([(0, game_name)])
if results and results[0].hltb_game_id:
return f"{HLTB_BASE_URL}/submit/game/{results[0].hltb_game_id}"
return None
home = _REEXPORTED.get(name)
if home is not None:
return getattr(importlib.import_module(home), name)
msg = f"module {__name__!r} has no attribute {name!r}"
raise AttributeError(msg)

14
steam_backlog_enforcer/tests/conftest.py
View File

@ -135,13 +135,21 @@ def _isolate_filesystem(tmp_path: Path) -> Iterator[None]:
tmp_path / "fake_home" / ".config" / "CSDSteamBuild",
),
),
# Whitelist exception files (_whitelist module-level constants)
# Derived from CONFIG_DIR at import time, so it captures the REAL path
# before this fixture redirects CONFIG_DIR. Without it the suite
# rewrites the user's owned-games cache.
patch(
"steam_backlog_enforcer._whitelist.APPROVED_EXCEPTIONS_FILE",
"steam_backlog_enforcer._owned_apps_cache._OWNED_IDS_CACHE_FILE",
fake_config / "owned_app_ids_cache.json",
),
# Whitelist exception files (module-level constants; these live in
# _whitelist_locking, which is where the code that reads them is)
patch(
"steam_backlog_enforcer._whitelist_locking.APPROVED_EXCEPTIONS_FILE",
fake_config / "approved_exceptions.json",
),
patch(
"steam_backlog_enforcer._whitelist.EXCEPTION_AUDIT_LOG",
"steam_backlog_enforcer._whitelist_locking.EXCEPTION_AUDIT_LOG",
fake_config / "exception_audit.log",
),
# _enforce_loop imports CONFIG_FILE directly; patch the local binding so

2
steam_backlog_enforcer/tests/test_actions_status_payload.py
View File

@ -37,7 +37,7 @@ class TestStatusPayload:
) -> object:
"""Test patch leaves."""
return patch.multiple(
"steam_backlog_enforcer._actions",
"steam_backlog_enforcer._manual_pick_lifecycle",
get_total_block_status=lambda: total_block,
get_installed_games=lambda: installed,
is_store_blocked=lambda: store_blocked,

11
steam_backlog_enforcer/tests/test_hltb_confidence_helpers.py
View File

@ -15,6 +15,7 @@ from steam_backlog_enforcer.hltb import (
)
PKG = "steam_backlog_enforcer.hltb"
_CONF = "steam_backlog_enforcer._hltb_confidence"
class _DummySession:
@ -116,11 +117,11 @@ class TestConfidenceHelpers:
def test_fetch_hltb_confidence_cached_all_cached_skips_fetch(self) -> None:
"""Test fetch hltb confidence cached all cached skips fetch."""
with (
patch(f"{PKG}.load_hltb_cache", return_value={1: 12.0}),
patch(f"{PKG}.load_hltb_polls_cache", return_value={1: 30}),
patch(f"{PKG}.load_hltb_count_comp_cache", return_value={1: 200}),
patch(f"{PKG}.fetch_hltb_confidence") as mock_fetch,
patch(f"{PKG}.save_hltb_cache") as mock_save,
patch(f"{_CONF}.load_hltb_cache", return_value={1: 12.0}),
patch(f"{_CONF}.load_hltb_polls_cache", return_value={1: 30}),
patch(f"{_CONF}.load_hltb_count_comp_cache", return_value={1: 200}),
patch(f"{_CONF}.fetch_hltb_confidence") as mock_fetch,
patch(f"{_CONF}.save_hltb_cache") as mock_save,
):
result = fetch_hltb_confidence_cached([(1, "Game")])
assert result == {1: 12.0}

2
steam_backlog_enforcer/tests/test_hltb_fetch_hltb_detail_missing.py
View File

@ -14,7 +14,7 @@ from steam_backlog_enforcer.hltb import (
if TYPE_CHECKING:
from steam_backlog_enforcer._hltb_types import _HLTBExtras
PKG = "steam_backlog_enforcer.hltb"
PKG = "steam_backlog_enforcer._hltb_confidence"
class _DummySession:

42
steam_backlog_enforcer/tests/test_hltb_part2.py
View File

@ -18,6 +18,8 @@ if TYPE_CHECKING:
from steam_backlog_enforcer._hltb_types import _HLTBExtras
PKG = "steam_backlog_enforcer.hltb"
_CACHED = "steam_backlog_enforcer._hltb_cached"
_CONF = "steam_backlog_enforcer._hltb_confidence"
class _DummySession:
@ -38,7 +40,7 @@ class TestFetchHltbTimesCached:
def test_all_cached(self) -> None:
"""Test all cached."""
with (
patch(f"{PKG}.load_hltb_cache", return_value={440: 50.0}),
patch(f"{_CACHED}.load_hltb_cache", return_value={440: 50.0}),
):
result = fetch_hltb_times_cached([(440, "TF2")])
assert result == {440: 50.0}
@ -46,10 +48,10 @@ class TestFetchHltbTimesCached:
def test_uncached_games_fetched(self) -> None:
"""Test uncached games fetched."""
with (
patch(f"{PKG}.load_hltb_cache", return_value={440: 50.0}),
patch(f"{PKG}.fetch_hltb_times") as mock_fetch,
patch(f"{PKG}.save_hltb_cache") as mock_save,
patch(f"{PKG}.time.monotonic", side_effect=[0.0, 2.0]),
patch(f"{_CACHED}.load_hltb_cache", return_value={440: 50.0}),
patch(f"{_CACHED}.fetch_hltb_times") as mock_fetch,
patch(f"{_CACHED}.save_hltb_cache") as mock_save,
patch(f"{_CACHED}.time.monotonic", side_effect=[0.0, 2.0]),
):
# fetch_hltb_times modifies cache in-place
def add_to_cache(
@ -80,10 +82,10 @@ class TestFetchHltbTimesCached:
"""Test uncached with progress cb."""
cb = MagicMock()
with (
patch(f"{PKG}.load_hltb_cache", return_value={}),
patch(f"{PKG}.fetch_hltb_times") as mock_fetch,
patch(f"{PKG}.save_hltb_cache"),
patch(f"{PKG}.time.monotonic", side_effect=[0.0, 1.0]),
patch(f"{_CACHED}.load_hltb_cache", return_value={}),
patch(f"{_CACHED}.fetch_hltb_times") as mock_fetch,
patch(f"{_CACHED}.save_hltb_cache"),
patch(f"{_CACHED}.time.monotonic", side_effect=[0.0, 1.0]),
):
mock_fetch.return_value = []
result = fetch_hltb_times_cached(
@ -95,10 +97,10 @@ class TestFetchHltbTimesCached:
def test_uncached_zero_elapsed(self) -> None:
"""Covers the elapsed == 0 branch for rate calculation."""
with (
patch(f"{PKG}.load_hltb_cache", return_value={}),
patch(f"{PKG}.fetch_hltb_times") as mock_fetch,
patch(f"{PKG}.save_hltb_cache"),
patch(f"{PKG}.time.monotonic", side_effect=[5.0, 5.0]),
patch(f"{_CACHED}.load_hltb_cache", return_value={}),
patch(f"{_CACHED}.fetch_hltb_times") as mock_fetch,
patch(f"{_CACHED}.save_hltb_cache"),
patch(f"{_CACHED}.time.monotonic", side_effect=[5.0, 5.0]),
):
mock_fetch.return_value = []
fetch_hltb_times_cached([(440, "TF2")])
@ -106,10 +108,10 @@ class TestFetchHltbTimesCached:
def test_found_count(self) -> None:
"""Covers the found count in logging."""
with (
patch(f"{PKG}.load_hltb_cache", return_value={}),
patch(f"{PKG}.fetch_hltb_times") as mock_fetch,
patch(f"{PKG}.save_hltb_cache"),
patch(f"{PKG}.time.monotonic", side_effect=[0.0, 3.0]),
patch(f"{_CACHED}.load_hltb_cache", return_value={}),
patch(f"{_CACHED}.fetch_hltb_times") as mock_fetch,
patch(f"{_CACHED}.save_hltb_cache"),
patch(f"{_CACHED}.time.monotonic", side_effect=[0.0, 3.0]),
):
def add_found(
@ -151,13 +153,13 @@ class TestGetHltbSubmitUrl:
similarity=1.0,
hltb_game_id=12345,
)
with patch(f"{PKG}.fetch_hltb_times", return_value=[mock_result]):
with patch(f"{_CONF}.fetch_hltb_times", return_value=[mock_result]):
url = get_hltb_submit_url("TF2")
assert url == f"{HLTB_BASE_URL}/submit/game/12345"
def test_not_found_empty(self) -> None:
"""Test not found empty."""
with patch(f"{PKG}.fetch_hltb_times", return_value=[]):
with patch(f"{_CONF}.fetch_hltb_times", return_value=[]):
url = get_hltb_submit_url("Unknown Game")
assert url is None
@ -170,6 +172,6 @@ class TestGetHltbSubmitUrl:
similarity=1.0,
hltb_game_id=0,
)
with patch(f"{PKG}.fetch_hltb_times", return_value=[mock_result]):
with patch(f"{_CONF}.fetch_hltb_times", return_value=[mock_result]):
url = get_hltb_submit_url("TF2")
assert url is None

2
steam_backlog_enforcer/tests/test_main_done.py
View File

@ -9,7 +9,7 @@ from steam_backlog_enforcer.main import (
from steam_backlog_enforcer.steam_api import GameInfo
PKG = "steam_backlog_enforcer.main"
CMD_DONE_PKG = "steam_backlog_enforcer._cmd_done"
CMD_DONE_PKG = "steam_backlog_enforcer._cmd_done_finalize"
class TestCmdDone:

2
steam_backlog_enforcer/tests/test_main_part2.py
View File

@ -10,7 +10,7 @@ from steam_backlog_enforcer._cmd_done import (
)
from steam_backlog_enforcer.config import Config, State
CMD_DONE_PKG = "steam_backlog_enforcer._cmd_done"
CMD_DONE_PKG = "steam_backlog_enforcer._cmd_done_finalize"
PKG = "steam_backlog_enforcer.main"

2
steam_backlog_enforcer/tests/test_main_part2b.py
View File

@ -10,7 +10,7 @@ from steam_backlog_enforcer._cmd_done import (
from steam_backlog_enforcer.config import Config, State
from steam_backlog_enforcer.steam_api import GameInfo
CMD_DONE_PKG = "steam_backlog_enforcer._cmd_done"
CMD_DONE_PKG = "steam_backlog_enforcer._cmd_done_finalize"
PKG = "steam_backlog_enforcer.main"

7
steam_backlog_enforcer/tests/test_main_part2c.py
View File

@ -11,7 +11,10 @@ from steam_backlog_enforcer._cmd_done import (
from steam_backlog_enforcer.config import Config, State
from steam_backlog_enforcer.steam_api import GameInfo
CMD_DONE_PKG = "steam_backlog_enforcer._cmd_done"
CMD_DONE_PKG = "steam_backlog_enforcer._cmd_done_finalize"
# _refresh_uncached_shortlist_hours stayed in _cmd_done, so its
# fetch_hltb_times_cached resolves there, not in the finalize module.
_CMD_DONE = "steam_backlog_enforcer._cmd_done"
PKG = "steam_backlog_enforcer.main"
@ -59,7 +62,7 @@ class TestFinalizeCompletionRetries:
patch(f"{CMD_DONE_PKG}.load_snapshot", return_value=snap),
patch(f"{CMD_DONE_PKG}.load_hltb_cache", return_value={2: 20.05}),
patch(
f"{CMD_DONE_PKG}.fetch_hltb_times_cached",
f"{_CMD_DONE}.fetch_hltb_times_cached",
return_value={3: 18.81},
) as mock_fetch_hltb,
patch(f"{CMD_DONE_PKG}.pick_next_game", side_effect=capture_pick),

4
steam_backlog_enforcer/tests/test_whitelist.py
View File

@ -164,7 +164,7 @@ class TestTrySetImmutable:
# A fresh tmp file is already mutable, so without this the call
# would (correctly) short-circuit before reaching chattr.
patch(
"steam_backlog_enforcer._whitelist._immutable_flag_is",
"steam_backlog_enforcer._whitelist_locking._immutable_flag_is",
return_value=False,
),
patch("subprocess.run") as mock_run,
@ -181,7 +181,7 @@ class TestTrySetImmutable:
target.write_text("data", encoding="utf-8")
with (
patch(
"steam_backlog_enforcer._whitelist._immutable_flag_is",
"steam_backlog_enforcer._whitelist_locking._immutable_flag_is",
return_value=True,
),
patch("shutil.which") as mock_which,

8
steam_backlog_enforcer/tests/test_whitelist_immutable_flag_is.py
View File

@ -148,7 +148,7 @@ class TestLockAndUnlock:
with (
patch(
"steam_backlog_enforcer._whitelist.APPROVED_EXCEPTIONS_FILE",
"steam_backlog_enforcer._whitelist_locking.APPROVED_EXCEPTIONS_FILE",
approved,
),
patch("shutil.which", return_value="/usr/bin/chattr"),
@ -168,7 +168,7 @@ class TestLockAndUnlock:
# Pretend the file really is immutable; a fresh tmp file is not,
# and unlocking an already-mutable file is now a no-op.
patch(
"steam_backlog_enforcer._whitelist._immutable_flag_is",
"steam_backlog_enforcer._whitelist_locking._immutable_flag_is",
return_value=False,
),
patch("subprocess.run") as mock_run,
@ -198,7 +198,7 @@ class TestPersistence:
bad = tmp_path / "approved.json"
bad.write_text("{{broken", encoding="utf-8")
with patch(
"steam_backlog_enforcer._whitelist.APPROVED_EXCEPTIONS_FILE",
"steam_backlog_enforcer._whitelist_locking.APPROVED_EXCEPTIONS_FILE",
bad,
):
assert _load_approved() == []
@ -208,7 +208,7 @@ class TestPersistence:
bad = tmp_path / "approved.json"
bad.write_text('"just a string"', encoding="utf-8")
with patch(
"steam_backlog_enforcer._whitelist.APPROVED_EXCEPTIONS_FILE",
"steam_backlog_enforcer._whitelist_locking.APPROVED_EXCEPTIONS_FILE",
bad,
):
assert _load_approved() == []

4
steam_backlog_enforcer/tests/test_whitelist_immutable_flag_is_append_audit_log.py
View File

@ -66,7 +66,7 @@ class TestAppendAuditLog:
"""Test audit log written."""
log_file = tmp_path / "audit.log"
with patch(
"steam_backlog_enforcer._whitelist.EXCEPTION_AUDIT_LOG",
"steam_backlog_enforcer._whitelist_locking.EXCEPTION_AUDIT_LOG",
log_file,
):
_append_audit_log(440, "some reason", "REQUESTED")
@ -79,7 +79,7 @@ class TestAppendAuditLog:
"""Test audit log appends."""
log_file = tmp_path / "audit.log"
with patch(
"steam_backlog_enforcer._whitelist.EXCEPTION_AUDIT_LOG",
"steam_backlog_enforcer._whitelist_locking.EXCEPTION_AUDIT_LOG",
log_file,
):
_append_audit_log(440, "first", "REQUESTED")