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

Bring _stats.py and scanning.py under the 250-line cap

Browse Source 
Leaf-set extraction only, so every new import is one-way and no cycle is
possible:

  _stats.py    523 -> 233   + _stats_types (the shared _GameTimes dataclass,
                              imported by parent and helpers alike),
                              _stats_gathering, _stats_display
  scanning.py  521 -> 248   + _scanning_candidates, _scanning_assign,
                              _scanning_tampering

_GameTimes and the four moved scanning helpers are re-exported from their
original modules via __all__ (mypy --no-implicit-reexport needs them listed),
so callers and tests keep their existing import paths.

Test patch targets were retargeted per file from the pytest failure list —
the target is the module whose source now contains the function under test.
Note TestPrintScenario and TestDoCheck did NOT move, so they keep the
original target while the rest of their files change.

1344 tests, 100.00% coverage. ./run.sh status and ./run.sh stats verified
against the live install; service at 0 restarts.

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 16:00:54 +02:00
parent 965abcaeaa
commit dd480f3208
24 changed files with 876 additions and 694 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

142
steam_backlog_enforcer/_scanning_assign.py Normal file
View File

@ -0,0 +1,142 @@
"""Interactive pick prompt and assignment of the chosen game.
Split out of :mod:`steam_backlog_enforcer.scanning` to keep both files under
the 250-line cap. Leaf helpers: nothing here calls back into ``scanning``.
"""
from __future__ import annotations
from datetime import datetime, timezone
import logging
from typing import TYPE_CHECKING
from steam_backlog_enforcer._scanning_candidates import (
_pick_next_shortest_candidate,
_sort_key,
)
from steam_backlog_enforcer._scanning_confidence import (
_apply_cached_confidence_to_candidates,
_report_poll_confidence,
)
from steam_backlog_enforcer.game_install import (
_echo,
install_game,
is_game_installed,
uninstall_other_games,
)
if TYPE_CHECKING:
from collections.abc import Callable
from steam_backlog_enforcer.config import Config, State
from steam_backlog_enforcer.steam_api import GameInfo
logger = logging.getLogger(__name__)
_NO_CONF_MSG = (
"\nNo assignable games found "
"(HLTB confidence thresholds: comp_100 polls>=3, "
"count_comp>=15, sum>=18)."
)
def _prompt_user_pick(qualified: list[GameInfo]) -> int:
"""Present numbered list, return 0-based index of user's choice."""
for i, g in enumerate(qualified, 1):
hours_str = (
f" (~{g.completionist_hours:.1f}h)" if g.completionist_hours > 0 else ""
)
_echo(f" {i}. {g.name} (AppID={g.app_id}){hours_str}")
while True:
raw = input("Select game number: ")
try:
idx = int(raw)
except ValueError:
_echo(f"Invalid input: {raw!r}")
continue
if idx < 1 or idx > len(qualified):
_echo(f"Out of range: {idx}")
continue
return idx - 1
def _assign_chosen_game(
chosen: GameInfo,
games: list[GameInfo],
state: State,
config: Config,
) -> None:
"""Save assignment, announce it, and handle install/uninstall."""
state.current_app_id = chosen.app_id
state.current_game_name = chosen.name
if not state.enforcement_started_at:
state.enforcement_started_at = datetime.now(timezone.utc).isoformat()
state.save()
hours_str = (
f" (~{chosen.completionist_hours:.1f}h leisure+dlc)"
if chosen.completionist_hours > 0
else ""
)
_echo(f"\n>>> ASSIGNED: {chosen.name} (AppID={chosen.app_id}){hours_str}")
_echo(
f" Progress: {chosen.unlocked_achievements}/{chosen.total_achievements}"
f" ({chosen.completion_pct:.1f}%)"
)
_report_poll_confidence(chosen, games, state)
if config.uninstall_other_games:
count = uninstall_other_games(chosen.app_id)
if count:
_echo(f"\n Uninstalled {count} non-assigned games")
if not is_game_installed(chosen.app_id):
_echo(f"\n Auto-installing {chosen.name}...")
install_game(
chosen.app_id, chosen.name, config.steam_id, use_steam_protocol=True
)
def _pick_next_game_sequential(
games: list[GameInfo],
state: State,
config: Config,
on_select: Callable[[GameInfo], bool],
) -> None:
"""Pick the next-shortest playable game, asking the user per candidate.
``on_select`` is called with each prospective pick. Returning ``True``
accepts the assignment; returning ``False`` records a 7-day skip on
``state`` for that game and the next candidate is evaluated.
"""
while True:
skip = set(state.finished_app_ids) | state.active_skipped_ids()
candidates = [g for g in games if not g.is_complete and g.app_id not in skip]
if not candidates:
_echo(_NO_CONF_MSG)
state.current_app_id = None
state.current_game_name = ""
state.save()
return
candidates.sort(key=_sort_key)
_apply_cached_confidence_to_candidates(candidates)
chosen, confidence_skipped, linux_skipped = _pick_next_shortest_candidate(
candidates
)
if chosen is None:
_echo(
_NO_CONF_MSG
if confidence_skipped > 0 and linux_skipped == 0
else "\nNo playable games left (all have poor ProtonDB ratings)!"
)
state.current_app_id = None
state.current_game_name = ""
state.save()
return
if not on_select(chosen):
state.skip_for_days(chosen.app_id, 7)
state.save()
_echo(f"\n Skipped {chosen.name} for 7 days; picking next...")
continue
_assign_chosen_game(chosen, games, state, config)
return

149
steam_backlog_enforcer/_scanning_candidates.py Normal file
View File

@ -0,0 +1,149 @@
"""Candidate selection helpers for ``scan`` and ``pick_next_game``.
Split out of :mod:`steam_backlog_enforcer.scanning` to keep both files under
the 250-line cap. Leaf helpers: nothing here calls back into ``scanning``.
"""
from __future__ import annotations
import logging
from typing import TYPE_CHECKING
from steam_backlog_enforcer._scanning_confidence import (
_candidate_passes_hltb_confidence,
)
from steam_backlog_enforcer.game_install import _echo
from steam_backlog_enforcer.protondb import (
ProtonDBRating,
fetch_protondb_ratings,
)
if TYPE_CHECKING:
from steam_backlog_enforcer.steam_api import GameInfo
logger = logging.getLogger(__name__)
_PROTONDB_BATCH_SIZE = 20
_PICK_LIST_SIZE = 10
def _sort_key(g: GameInfo) -> tuple[int, float]:
"""Sort by known HLTB time (shortest first), then unknown games."""
if g.completionist_hours > 0:
return (0, g.completionist_hours)
return (1, g.name.lower().encode().hex().__hash__())
def _pick_playable_candidate(
candidates: list[GameInfo],
) -> GameInfo | None:
"""Return the first candidate with an acceptable ProtonDB rating.
Checks candidates in batches (sorted by HLTB hours, shortest first).
Games rated silver-or-worse, or gold-trending-down, are skipped.
"""
offset = 0
while offset < len(candidates):
batch = candidates[offset : offset + _PROTONDB_BATCH_SIZE]
app_ids = [g.app_id for g in batch]
ratings = fetch_protondb_ratings(app_ids)
for game in batch:
rating = ratings.get(game.app_id, ProtonDBRating(app_id=game.app_id))
if rating.is_playable:
if offset > 0 or game is not batch[0]:
_echo(
f" Skipped {offset + batch.index(game)} game(s) "
f"with poor Linux compatibility"
)
return game
logger.info(
"Skipping %s (AppID=%d): ProtonDB %s (trending %s)",
game.name,
game.app_id,
rating.tier,
rating.trending_tier,
)
offset += _PROTONDB_BATCH_SIZE
return None
def _collect_qualified_candidates(
candidates: list[GameInfo],
) -> tuple[list[GameInfo], int, int]:
"""Collect up to _PICK_LIST_SIZE playable, HLTB-confident candidates."""
qualified: list[GameInfo] = []
confidence_skipped = 0
linux_skipped = 0
for game in candidates:
if len(qualified) >= _PICK_LIST_SIZE:
break
if not _candidate_passes_hltb_confidence(game):
confidence_skipped += 1
continue
playable = _pick_playable_candidate([game])
if playable is not None:
qualified.append(playable)
else:
linux_skipped += 1
return qualified, confidence_skipped, linux_skipped
def _pick_next_shortest_candidate(
candidates: list[GameInfo],
) -> tuple[GameInfo | None, int, int]:
"""Pick next game by checking confidence one candidate at a time.
The list must be pre-sorted by desired priority (shortest first).
"""
confidence_skipped = 0
linux_skipped = 0
for game in candidates:
if not _candidate_passes_hltb_confidence(game):
confidence_skipped += 1
continue
# Reuse existing ProtonDB compatibility gate for one candidate.
playable = _pick_playable_candidate([game])
if playable is not None:
if linux_skipped > 0:
_echo(
f" Skipped {linux_skipped} game(s) with poor Linux compatibility"
)
return playable, confidence_skipped, linux_skipped
linux_skipped += 1
if linux_skipped > 0:
_echo(f" Skipped {linux_skipped} game(s) with poor Linux compatibility")
return None, confidence_skipped, linux_skipped
def _collect_top_candidates(
candidates: list[GameInfo],
n: int = 3,
) -> tuple[list[GameInfo], int, int]:
"""Collect up to n candidates that pass the Linux compatibility gate.
Args:
candidates: Pre-sorted list of candidate games.
n: Maximum number of qualified games to collect.
Returns:
Tuple of (qualified_list, conf_skipped, linux_skipped).
"""
qualified: list[GameInfo] = []
linux_skipped = 0
for game in candidates:
if len(qualified) >= n:
break
playable = _pick_playable_candidate([game])
if playable is not None:
qualified.append(playable)
else:
linux_skipped += 1
if linux_skipped > 0:
_echo(f" Skipped {linux_skipped} game(s) with poor Linux compatibility")
return qualified, 0, linux_skipped

80
steam_backlog_enforcer/_scanning_tampering.py Normal file
View File

@ -0,0 +1,80 @@
"""Tampering detection for installed games.
Split out of :mod:`steam_backlog_enforcer.scanning` to keep both files under
the 250-line cap. Leaf helpers: nothing here calls back into ``scanning``.
"""
from __future__ import annotations
import logging
from typing import TYPE_CHECKING, Any
from steam_backlog_enforcer.config import load_snapshot
from steam_backlog_enforcer.enforcer import send_notification
from steam_backlog_enforcer.game_install import _echo
from steam_backlog_enforcer.steam_api import SteamAPIClient
if TYPE_CHECKING:
from steam_backlog_enforcer.config import Config, State
logger = logging.getLogger(__name__)
_TAMPER_CHECK_LIMIT = 3
def _check_game_tampering(
client: SteamAPIClient,
entry: dict[str, Any],
state: State,
) -> tuple[str, int, int] | None:
"""Check if a single game has unexpected achievement progress.
Args:
client: Steam API client.
entry: Snapshot entry for the game.
state: Current enforcer state.
Returns:
Tuple of (name, app_id, diff) if tampering detected, else None.
"""
app_id = entry["app_id"]
if app_id == state.current_app_id:
return None
if entry["unlocked_achievements"] >= entry["total_achievements"]:
return None
if entry.get("playtime_minutes", 0) <= 0:
return None
game = client.refresh_single_game(
app_id, entry["name"], entry.get("playtime_minutes", 0)
)
if game and game.unlocked_achievements > entry["unlocked_achievements"]:
diff = game.unlocked_achievements - entry["unlocked_achievements"]
return (entry["name"], app_id, diff)
return None
def detect_tampering(config: Config, state: State) -> None:
"""Check if achievements were unlocked on non-assigned games."""
old_snapshot = load_snapshot()
if old_snapshot is None:
return
client = SteamAPIClient(config.steam_api_key, config.steam_id)
# Quick check: only re-fetch a few random non-assigned games.
suspicious: list[tuple[str, int, int]] = []
for entry in old_snapshot:
result = _check_game_tampering(client, entry, state)
if result:
suspicious.append(result)
if len(suspicious) >= _TAMPER_CHECK_LIMIT:
break
if suspicious:
_echo("\n TAMPERING DETECTED:")
for name, app_id, diff in suspicious:
_echo(f" {name} (AppID={app_id}): +{diff} new achievements!")
send_notification(
"Tampering Detected!",
f"Achievements unlocked on {len(suspicious)} non-assigned games!",
)

328
steam_backlog_enforcer/_stats.py
View File

@ -2,45 +2,42 @@
from __future__ import annotations
from concurrent.futures import ThreadPoolExecutor, as_completed
from dataclasses import dataclass, field
from datetime import datetime, timedelta, timezone
import logging
import secrets
from typing import TYPE_CHECKING
from urllib.parse import quote_plus
from steam_backlog_enforcer._hltb_types import (
HLTB_BASE_URL,
_read_raw_cache,
load_hltb_cache,
load_hltb_game_id_cache,
load_hltb_leisure_100h_cache,
load_hltb_rush_cache,
)
from steam_backlog_enforcer._scanning_confidence import (
_apply_cached_confidence_to_candidates,
_confidence_fail_reasons,
_refresh_candidate_confidence_batch,
from steam_backlog_enforcer._stats_display import (
_format_completion_date,
_print_pace_scenario,
_print_worst_example,
_sum_hours,
)
from steam_backlog_enforcer._stats_gathering import (
_ensure_completed_rush_data,
_ensure_rush_data,
_filter_qualifying_games,
_refresh_recently_played_completions,
)
from steam_backlog_enforcer._stats_types import _GameTimes
from steam_backlog_enforcer._web_dataset import (
PaceVsHLTB,
compute_pace_vs_hltb,
count_complete_since_start,
)
from steam_backlog_enforcer.config import SNAPSHOT_FILE, load_snapshot
from steam_backlog_enforcer.config import load_snapshot
from steam_backlog_enforcer.game_install import _echo
from steam_backlog_enforcer.hltb import fetch_hltb_detail_missing
from steam_backlog_enforcer.protondb import (
ProtonDBRating,
fetch_protondb_ratings,
)
from steam_backlog_enforcer.steam_api import (
GameInfo,
SteamAPIClient,
SteamAPIError,
)
# _GameTimes lives in the _stats_types leaf so both this module and its
# helper modules can share it; it is re-exported here because callers (and
# tests) have always imported it from _stats. Without __all__, mypy's
# --no-implicit-reexport rejects those imports.
__all__ = ["_GameTimes"]
if TYPE_CHECKING:
from steam_backlog_enforcer.config import Config, State
@ -50,181 +47,6 @@ _HOURS_PER_DAY_PRESETS = (2.0, 4.0, 6.0, 8.0)
_LINE = "─" * 70
_HLTB_SEARCH_BASE = "https://howlongtobeat.com/?q="
@dataclass
class _GameTimes:
"""Per-game time estimates for stats display."""
game: GameInfo
worst_hours: float
rush_hours: float
leisure_100h: float
hltb_game_id: int = field(default=0)
def _filter_qualifying_games(
games: list[GameInfo],
state: State,
) -> tuple[list[_GameTimes], int, int, int]:
"""Return qualifying incomplete games with their time estimates.
Applies the same HLTB-confidence and Linux-compatibility filters as the
game picker. The current game and already-finished games are excluded.
Returns:
(qualified_list, hltb_skipped, linux_skipped, no_data_skipped)
"""
rush_cache = load_hltb_rush_cache()
leisure_100h_cache = load_hltb_leisure_100h_cache()
game_id_cache = load_hltb_game_id_cache()
hours_cache = load_hltb_cache()
exclude = set(state.finished_app_ids)
if state.current_app_id is not None:
exclude.add(state.current_app_id)
candidates = [g for g in games if not g.is_complete and g.app_id not in exclude]
_apply_cached_confidence_to_candidates(candidates)
_refresh_candidate_confidence_batch(candidates)
hltb_skipped = 0
linux_skipped = 0
no_data_skipped = 0
app_ids_to_check: list[int] = []
conf_ok: list[GameInfo] = []
for game in candidates:
if _confidence_fail_reasons(game):
hltb_skipped += 1
continue
conf_ok.append(game)
app_ids_to_check.append(game.app_id)
ratings: dict[int, ProtonDBRating] = {}
if app_ids_to_check:
ratings = fetch_protondb_ratings(app_ids_to_check)
qualified: list[_GameTimes] = []
for game in conf_ok:
rating = ratings.get(game.app_id, ProtonDBRating(app_id=game.app_id))
if not rating.is_playable:
linux_skipped += 1
continue
rush = rush_cache.get(game.app_id, -1)
leisure = leisure_100h_cache.get(game.app_id, -1)
# worst_hours = max of: snapshot completionist, HLTB hours cache (fallback
# when snapshot is stale/missing), and leisure_100h (slowest 100% time).
snap_hours = game.completionist_hours if game.completionist_hours > 0 else -1
cache_hours = hours_cache.get(game.app_id, -1)
worst_candidates = [v for v in (snap_hours, cache_hours, leisure) if v > 0]
worst = max(worst_candidates) if worst_candidates else -1
if worst <= 0 and rush <= 0 and leisure <= 0:
no_data_skipped += 1
continue
qualified.append(
_GameTimes(
game=game,
worst_hours=worst,
rush_hours=rush,
leisure_100h=leisure,
hltb_game_id=game_id_cache.get(game.app_id, 0),
)
)
return qualified, hltb_skipped, linux_skipped, no_data_skipped
def _ensure_rush_data(qualified: list[_GameTimes]) -> bool:
"""Auto-fetch rush/leisure detail for games that are missing it.
Returns True when a fetch was performed; the caller should then re-run
``_filter_qualifying_games`` to pick up the updated caches.
"""
total_q = len(qualified)
missing = sum(1 for e in qualified if e.rush_hours <= 0)
if not qualified or not missing:
return False
_echo(f"Fetching HLTB detail for {missing}/{total_q} games missing rush/leisure...")
game_pairs = [(e.game.app_id, e.game.name) for e in qualified]
fetch_hltb_detail_missing(game_pairs)
return True
def _ensure_completed_rush_data(games: list[GameInfo]) -> bool:
"""Fetch rush/leisure detail for completed games used for pace calibration.
Completed games aren't processed by ``_ensure_rush_data`` (which only
handles incomplete qualifying games), so this separate pass fills in
their rush/leisure data for ``compute_pace_vs_hltb``.
Returns True when at least one new fetch was performed.
"""
pairs = [
(g.app_id, g.name) for g in games if g.is_complete and g.playtime_minutes > 0
]
if not pairs:
return False
_echo(
f"Fetching HLTB detail for {len(pairs)} completed games (pace calibration)..."
)
fetched = fetch_hltb_detail_missing(pairs)
return fetched > 0
def _print_worst_example(entries: list[_GameTimes]) -> None:
"""Print a randomly selected example from the worst-case qualified games."""
examples = [e for e in entries if e.worst_hours > 0]
if not examples:
return
example = secrets.choice(examples)
_echo(f"\n Example game: {example.game.name!r}")
_echo(f" Worst case: {example.worst_hours:.1f} h")
if example.rush_hours > 0:
_echo(f" Rush: {example.rush_hours:.1f} h")
if example.leisure_100h > 0:
_echo(f" Leisure: {example.leisure_100h:.1f} h")
hltb_game_id = example.hltb_game_id
if hltb_game_id == 0:
# On-demand backfill: one search to get the HLTB game ID for this game.
fetch_hltb_detail_missing([(example.game.app_id, example.game.name)])
hltb_game_id = load_hltb_game_id_cache().get(example.game.app_id, 0)
if hltb_game_id > 0:
_echo(f" HLTB: {HLTB_BASE_URL}/game/{hltb_game_id}")
else:
_echo(f" HLTB: {_HLTB_SEARCH_BASE}{quote_plus(example.game.name)}")
def _sum_hours(entries: list[_GameTimes], attr: str) -> tuple[float, int]:
"""Sum a time attribute across entries; return (total_hours, missing_count).
Games where the attribute is ≤ 0 contribute 0 to the sum and are counted
in ``missing_count`` so the user knows the estimate may be an undercount.
"""
total = 0.0
missing = 0
for e in entries:
val: float = getattr(e, attr)
if val > 0:
total += val
else:
missing += 1
return round(total, 1), missing
def _format_completion_date(hours: float, daily_hours: float) -> str:
"""Return 'N days (YYYY-MM-DD)' for finishing hours at daily_hours per day."""
if hours <= 0 or daily_hours <= 0:
return "N/A"
days = int(hours / daily_hours)
target = datetime.now(timezone.utc) + timedelta(days=days)
return f"{days} days ({target.strftime('%Y-%m-%d')})"
def _print_scenario(
label: str,
@ -249,50 +71,6 @@ def _print_scenario(
_echo(f" @ {daily:.0f} h/day → {estimate}")
def _print_pace_scenario(state: State, remaining: int, games_done: int) -> None:
"""Print the pace-based completion estimate.
``games_done`` must be the count of games completed ON OR AFTER
``state.enforcement_started_at`` (use ``count_complete_since_start``).
Pre-enforcement completions inflate the rate and are excluded.
"""
_echo("\n 1. AT YOUR CURRENT PACE")
if not state.enforcement_started_at:
_echo(" No start date recorded.")
_echo(" Set enforcement_started_at in state.json (ISO-8601 UTC)")
_echo(" to enable this estimate.")
return
try:
started = datetime.fromisoformat(state.enforcement_started_at)
except ValueError:
_echo(f" Invalid enforcement_started_at: {state.enforcement_started_at!r}")
return
now = datetime.now(timezone.utc)
days_elapsed = max(1, (now - started).days)
if games_done == 0:
_echo(f" Started: {started.strftime('%Y-%m-%d')}")
_echo(" No games finished yet — pace cannot be estimated.")
return
rate = games_done / days_elapsed
_echo(f" Started: {started.strftime('%Y-%m-%d')}")
_echo(
f" Finished: {games_done} games in {days_elapsed} days "
"(since enforcement start)"
)
_echo(
f" Pace: {rate:.4f} games/day (1 game every {1 / rate:.1f} days)"
)
_echo(f" Remaining: {remaining} games")
days_to_go = int(remaining / rate)
finish = now + timedelta(days=days_to_go)
_echo(f" Est. complete: {days_to_go} days ({finish.strftime('%Y-%m-%d')})")
def _print_player_speed_scenario(
pace: PaceVsHLTB | None,
rush_total: float,
@ -344,74 +122,6 @@ def _print_player_speed_scenario(
_echo(f" @ {daily:.0f} h/day → {estimate}")
def _refresh_recently_played_completions(
games: list[GameInfo],
config: Config,
) -> list[GameInfo]:
"""Refresh achievement data for incomplete games played since last scan.
Makes 1 ``GetOwnedGames`` request + 1 ``GetPlayerAchievements`` per
recently-played incomplete game. Finds games newly completed since the
last ``scan`` without re-scanning the whole library.
Returns a new list with updated GameInfo objects for any game that was
played after the snapshot was written; all other games are unchanged.
"""
try:
snapshot_mtime = SNAPSHOT_FILE.stat().st_mtime
except OSError:
return games
try:
client = SteamAPIClient(config.steam_api_key, config.steam_id)
owned_raw = client.get_owned_games()
except SteamAPIError:
logger.debug("Steam API unavailable; skipping completion refresh.")
return games
last_played_map = {g["appid"]: g.get("rtime_last_played", 0) for g in owned_raw}
to_refresh = [
g
for g in games
if not g.is_complete and last_played_map.get(g.app_id, 0) > snapshot_mtime
]
if not to_refresh:
return games
_echo(
f"Refreshing {len(to_refresh)} recently-played game(s)"
" for up-to-date completion status..."
)
game_map = {g.app_id: g for g in games}
def _refresh_one(game: GameInfo) -> GameInfo:
achievements = client.get_achievement_details(game.app_id)
if not achievements:
return game
unlocked = sum(1 for a in achievements if a.achieved)
return GameInfo(
app_id=game.app_id,
name=game.name,
total_achievements=len(achievements),
unlocked_achievements=unlocked,
playtime_minutes=game.playtime_minutes,
achievements=achievements,
completionist_hours=game.completionist_hours,
comp_100_count=game.comp_100_count,
count_comp=game.count_comp,
)
with ThreadPoolExecutor(max_workers=20) as pool:
futures = {pool.submit(_refresh_one, g): g for g in to_refresh}
for future in as_completed(futures):
refreshed = future.result()
game_map[refreshed.app_id] = refreshed
return list(game_map.values())
def cmd_stats(_config: Config, state: State) -> None:
"""Display backlog completion-time statistics.

116
steam_backlog_enforcer/_stats_display.py Normal file
View File

@ -0,0 +1,116 @@
"""Formatting helpers for the ``stats`` command's printed report.
Split out of :mod:`steam_backlog_enforcer._stats` to keep both files under
the 250-line cap. Leaf helpers: nothing here calls back into ``_stats``.
"""
from __future__ import annotations
from datetime import datetime, timedelta, timezone
import secrets
from typing import TYPE_CHECKING
from urllib.parse import quote_plus
from steam_backlog_enforcer._hltb_types import (
HLTB_BASE_URL,
load_hltb_game_id_cache,
)
from steam_backlog_enforcer._stats_types import _HLTB_SEARCH_BASE, _GameTimes
from steam_backlog_enforcer.game_install import _echo
from steam_backlog_enforcer.hltb import fetch_hltb_detail_missing
if TYPE_CHECKING:
from steam_backlog_enforcer.config import State
def _print_worst_example(entries: list[_GameTimes]) -> None:
"""Print a randomly selected example from the worst-case qualified games."""
examples = [e for e in entries if e.worst_hours > 0]
if not examples:
return
example = secrets.choice(examples)
_echo(f"\n Example game: {example.game.name!r}")
_echo(f" Worst case: {example.worst_hours:.1f} h")
if example.rush_hours > 0:
_echo(f" Rush: {example.rush_hours:.1f} h")
if example.leisure_100h > 0:
_echo(f" Leisure: {example.leisure_100h:.1f} h")
hltb_game_id = example.hltb_game_id
if hltb_game_id == 0:
# On-demand backfill: one search to get the HLTB game ID for this game.
fetch_hltb_detail_missing([(example.game.app_id, example.game.name)])
hltb_game_id = load_hltb_game_id_cache().get(example.game.app_id, 0)
if hltb_game_id > 0:
_echo(f" HLTB: {HLTB_BASE_URL}/game/{hltb_game_id}")
else:
_echo(f" HLTB: {_HLTB_SEARCH_BASE}{quote_plus(example.game.name)}")
def _sum_hours(entries: list[_GameTimes], attr: str) -> tuple[float, int]:
"""Sum a time attribute across entries; return (total_hours, missing_count).
Games where the attribute is ≤ 0 contribute 0 to the sum and are counted
in ``missing_count`` so the user knows the estimate may be an undercount.
"""
total = 0.0
missing = 0
for e in entries:
val: float = getattr(e, attr)
if val > 0:
total += val
else:
missing += 1
return round(total, 1), missing
def _format_completion_date(hours: float, daily_hours: float) -> str:
"""Return 'N days (YYYY-MM-DD)' for finishing hours at daily_hours per day."""
if hours <= 0 or daily_hours <= 0:
return "N/A"
days = int(hours / daily_hours)
target = datetime.now(timezone.utc) + timedelta(days=days)
return f"{days} days ({target.strftime('%Y-%m-%d')})"
def _print_pace_scenario(state: State, remaining: int, games_done: int) -> None:
"""Print the pace-based completion estimate.
``games_done`` must be the count of games completed ON OR AFTER
``state.enforcement_started_at`` (use ``count_complete_since_start``).
Pre-enforcement completions inflate the rate and are excluded.
"""
_echo("\n 1. AT YOUR CURRENT PACE")
if not state.enforcement_started_at:
_echo(" No start date recorded.")
_echo(" Set enforcement_started_at in state.json (ISO-8601 UTC)")
_echo(" to enable this estimate.")
return
try:
started = datetime.fromisoformat(state.enforcement_started_at)
except ValueError:
_echo(f" Invalid enforcement_started_at: {state.enforcement_started_at!r}")
return
now = datetime.now(timezone.utc)
days_elapsed = max(1, (now - started).days)
if games_done == 0:
_echo(f" Started: {started.strftime('%Y-%m-%d')}")
_echo(" No games finished yet — pace cannot be estimated.")
return
rate = games_done / days_elapsed
_echo(f" Started: {started.strftime('%Y-%m-%d')}")
_echo(
f" Finished: {games_done} games in {days_elapsed} days "
"(since enforcement start)"
)
_echo(
f" Pace: {rate:.4f} games/day (1 game every {1 / rate:.1f} days)"
)
_echo(f" Remaining: {remaining} games")
days_to_go = int(remaining / rate)
finish = now + timedelta(days=days_to_go)
_echo(f" Est. complete: {days_to_go} days ({finish.strftime('%Y-%m-%d')})")

222
steam_backlog_enforcer/_stats_gathering.py Normal file
View File

@ -0,0 +1,222 @@
"""Game selection and HLTB/completion data top-ups for ``stats``.
Split out of :mod:`steam_backlog_enforcer._stats` to keep both files under
the 250-line cap. Leaf helpers: nothing here calls back into ``_stats``.
"""
from __future__ import annotations
from concurrent.futures import ThreadPoolExecutor, as_completed
import logging
from typing import TYPE_CHECKING
from steam_backlog_enforcer._hltb_types import (
load_hltb_cache,
load_hltb_game_id_cache,
load_hltb_leisure_100h_cache,
load_hltb_rush_cache,
)
from steam_backlog_enforcer._scanning_confidence import (
_apply_cached_confidence_to_candidates,
_confidence_fail_reasons,
_refresh_candidate_confidence_batch,
)
from steam_backlog_enforcer._stats_types import _GameTimes
from steam_backlog_enforcer.config import SNAPSHOT_FILE
from steam_backlog_enforcer.game_install import _echo
from steam_backlog_enforcer.hltb import fetch_hltb_detail_missing
from steam_backlog_enforcer.protondb import (
ProtonDBRating,
fetch_protondb_ratings,
)
from steam_backlog_enforcer.steam_api import (
GameInfo,
SteamAPIClient,
SteamAPIError,
)
if TYPE_CHECKING:
from steam_backlog_enforcer.config import Config, State
logger = logging.getLogger(__name__)
def _filter_qualifying_games(
games: list[GameInfo],
state: State,
) -> tuple[list[_GameTimes], int, int, int]:
"""Return qualifying incomplete games with their time estimates.
Applies the same HLTB-confidence and Linux-compatibility filters as the
game picker. The current game and already-finished games are excluded.
Returns:
(qualified_list, hltb_skipped, linux_skipped, no_data_skipped)
"""
rush_cache = load_hltb_rush_cache()
leisure_100h_cache = load_hltb_leisure_100h_cache()
game_id_cache = load_hltb_game_id_cache()
hours_cache = load_hltb_cache()
exclude = set(state.finished_app_ids)
if state.current_app_id is not None:
exclude.add(state.current_app_id)
candidates = [g for g in games if not g.is_complete and g.app_id not in exclude]
_apply_cached_confidence_to_candidates(candidates)
_refresh_candidate_confidence_batch(candidates)
hltb_skipped = 0
linux_skipped = 0
no_data_skipped = 0
app_ids_to_check: list[int] = []
conf_ok: list[GameInfo] = []
for game in candidates:
if _confidence_fail_reasons(game):
hltb_skipped += 1
continue
conf_ok.append(game)
app_ids_to_check.append(game.app_id)
ratings: dict[int, ProtonDBRating] = {}
if app_ids_to_check:
ratings = fetch_protondb_ratings(app_ids_to_check)
qualified: list[_GameTimes] = []
for game in conf_ok:
rating = ratings.get(game.app_id, ProtonDBRating(app_id=game.app_id))
if not rating.is_playable:
linux_skipped += 1
continue
rush = rush_cache.get(game.app_id, -1)
leisure = leisure_100h_cache.get(game.app_id, -1)
# worst_hours = max of: snapshot completionist, HLTB hours cache (fallback
# when snapshot is stale/missing), and leisure_100h (slowest 100% time).
snap_hours = game.completionist_hours if game.completionist_hours > 0 else -1
cache_hours = hours_cache.get(game.app_id, -1)
worst_candidates = [v for v in (snap_hours, cache_hours, leisure) if v > 0]
worst = max(worst_candidates) if worst_candidates else -1
if worst <= 0 and rush <= 0 and leisure <= 0:
no_data_skipped += 1
continue
qualified.append(
_GameTimes(
game=game,
worst_hours=worst,
rush_hours=rush,
leisure_100h=leisure,
hltb_game_id=game_id_cache.get(game.app_id, 0),
)
)
return qualified, hltb_skipped, linux_skipped, no_data_skipped
def _ensure_rush_data(qualified: list[_GameTimes]) -> bool:
"""Auto-fetch rush/leisure detail for games that are missing it.
Returns True when a fetch was performed; the caller should then re-run
``_filter_qualifying_games`` to pick up the updated caches.
"""
total_q = len(qualified)
missing = sum(1 for e in qualified if e.rush_hours <= 0)
if not qualified or not missing:
return False
_echo(f"Fetching HLTB detail for {missing}/{total_q} games missing rush/leisure...")
game_pairs = [(e.game.app_id, e.game.name) for e in qualified]
fetch_hltb_detail_missing(game_pairs)
return True
def _ensure_completed_rush_data(games: list[GameInfo]) -> bool:
"""Fetch rush/leisure detail for completed games used for pace calibration.
Completed games aren't processed by ``_ensure_rush_data`` (which only
handles incomplete qualifying games), so this separate pass fills in
their rush/leisure data for ``compute_pace_vs_hltb``.
Returns True when at least one new fetch was performed.
"""
pairs = [
(g.app_id, g.name) for g in games if g.is_complete and g.playtime_minutes > 0
]
if not pairs:
return False
_echo(
f"Fetching HLTB detail for {len(pairs)} completed games (pace calibration)..."
)
fetched = fetch_hltb_detail_missing(pairs)
return fetched > 0
def _refresh_recently_played_completions(
games: list[GameInfo],
config: Config,
) -> list[GameInfo]:
"""Refresh achievement data for incomplete games played since last scan.
Makes 1 ``GetOwnedGames`` request + 1 ``GetPlayerAchievements`` per
recently-played incomplete game. Finds games newly completed since the
last ``scan`` without re-scanning the whole library.
Returns a new list with updated GameInfo objects for any game that was
played after the snapshot was written; all other games are unchanged.
"""
try:
snapshot_mtime = SNAPSHOT_FILE.stat().st_mtime
except OSError:
return games
try:
client = SteamAPIClient(config.steam_api_key, config.steam_id)
owned_raw = client.get_owned_games()
except SteamAPIError:
logger.debug("Steam API unavailable; skipping completion refresh.")
return games
last_played_map = {g["appid"]: g.get("rtime_last_played", 0) for g in owned_raw}
to_refresh = [
g
for g in games
if not g.is_complete and last_played_map.get(g.app_id, 0) > snapshot_mtime
]
if not to_refresh:
return games
_echo(
f"Refreshing {len(to_refresh)} recently-played game(s)"
" for up-to-date completion status..."
)
game_map = {g.app_id: g for g in games}
def _refresh_one(game: GameInfo) -> GameInfo:
achievements = client.get_achievement_details(game.app_id)
if not achievements:
return game
unlocked = sum(1 for a in achievements if a.achieved)
return GameInfo(
app_id=game.app_id,
name=game.name,
total_achievements=len(achievements),
unlocked_achievements=unlocked,
playtime_minutes=game.playtime_minutes,
achievements=achievements,
completionist_hours=game.completionist_hours,
comp_100_count=game.comp_100_count,
count_comp=game.count_comp,
)
with ThreadPoolExecutor(max_workers=20) as pool:
futures = {pool.submit(_refresh_one, g): g for g in to_refresh}
for future in as_completed(futures):
refreshed = future.result()
game_map[refreshed.app_id] = refreshed
return list(game_map.values())

26
steam_backlog_enforcer/_stats_types.py Normal file
View File

@ -0,0 +1,26 @@
"""Shared value types for the ``stats`` command.
A leaf module: imported by :mod:`steam_backlog_enforcer._stats` and by its
helper modules, so it must not import either of them back.
"""
from __future__ import annotations
from dataclasses import dataclass, field
from typing import TYPE_CHECKING
if TYPE_CHECKING:
from steam_backlog_enforcer.steam_api import GameInfo
_HLTB_SEARCH_BASE = "https://howlongtobeat.com/?q="
@dataclass
class _GameTimes:
"""Per-game time estimates for stats display."""
game: GameInfo
worst_hours: float
rush_hours: float
leisure_100h: float
hltb_game_id: int = field(default=0)

331
steam_backlog_enforcer/scanning.py
View File

@ -2,20 +2,35 @@
from __future__ import annotations
from datetime import datetime, timezone
import logging
import time
from typing import TYPE_CHECKING, Any
from typing import TYPE_CHECKING
from steam_backlog_enforcer._hltb_types import (
load_hltb_count_comp_cache,
load_hltb_polls_cache,
)
from steam_backlog_enforcer._scanning_assign import (
_NO_CONF_MSG,
_assign_chosen_game,
_pick_next_game_sequential,
_prompt_user_pick,
)
from steam_backlog_enforcer._scanning_candidates import (
_collect_qualified_candidates,
_collect_top_candidates,
_pick_next_shortest_candidate,
_pick_playable_candidate,
_sort_key,
)
from steam_backlog_enforcer._scanning_confidence import (
_apply_cached_confidence_to_candidates,
_candidate_passes_hltb_confidence,
_report_poll_confidence,
)
from steam_backlog_enforcer._scanning_tampering import (
_check_game_tampering,
detect_tampering,
)
from steam_backlog_enforcer.config import (
Config,
State,
@ -27,25 +42,28 @@ from steam_backlog_enforcer.enforcer import (
)
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,
)
from steam_backlog_enforcer.protondb import (
ProtonDBRating,
fetch_protondb_ratings,
)
from steam_backlog_enforcer.steam_api import GameInfo, SteamAPIClient
if TYPE_CHECKING:
from collections.abc import Callable
# These helpers moved to the _scanning_* leaf modules, but callers and tests
# have always imported them from here. mypy's --no-implicit-reexport needs
# them listed explicitly.
__all__ = [
"_check_game_tampering",
"_collect_top_candidates",
"_pick_next_shortest_candidate",
"_pick_playable_candidate",
"detect_tampering",
]
logger = logging.getLogger(__name__)
_TAMPER_CHECK_LIMIT = 3
# ──────────────────────────────────────────────────────────────
# Scanning & game selection
@ -125,182 +143,6 @@ def do_scan(config: Config, state: State) -> list[GameInfo]:
# How many candidates to check per ProtonDB batch.
_PROTONDB_BATCH_SIZE = 20
def _pick_playable_candidate(
candidates: list[GameInfo],
) -> GameInfo | None:
"""Return the first candidate with an acceptable ProtonDB rating.
Checks candidates in batches (sorted by HLTB hours, shortest first).
Games rated silver-or-worse, or gold-trending-down, are skipped.
"""
offset = 0
while offset < len(candidates):
batch = candidates[offset : offset + _PROTONDB_BATCH_SIZE]
app_ids = [g.app_id for g in batch]
ratings = fetch_protondb_ratings(app_ids)
for game in batch:
rating = ratings.get(game.app_id, ProtonDBRating(app_id=game.app_id))
if rating.is_playable:
if offset > 0 or game is not batch[0]:
_echo(
f" Skipped {offset + batch.index(game)} game(s) "
f"with poor Linux compatibility"
)
return game
logger.info(
"Skipping %s (AppID=%d): ProtonDB %s (trending %s)",
game.name,
game.app_id,
rating.tier,
rating.trending_tier,
)
offset += _PROTONDB_BATCH_SIZE
return None
_PICK_LIST_SIZE = 10
_NO_CONF_MSG = (
"\nNo assignable games found "
"(HLTB confidence thresholds: comp_100 polls>=3, "
"count_comp>=15, sum>=18)."
)
def _sort_key(g: GameInfo) -> tuple[int, float]:
"""Sort by known HLTB time (shortest first), then unknown games."""
if g.completionist_hours > 0:
return (0, g.completionist_hours)
return (1, g.name.lower().encode().hex().__hash__())
def _collect_qualified_candidates(
candidates: list[GameInfo],
) -> tuple[list[GameInfo], int, int]:
"""Collect up to _PICK_LIST_SIZE playable, HLTB-confident candidates."""
qualified: list[GameInfo] = []
confidence_skipped = 0
linux_skipped = 0
for game in candidates:
if len(qualified) >= _PICK_LIST_SIZE:
break
if not _candidate_passes_hltb_confidence(game):
confidence_skipped += 1
continue
playable = _pick_playable_candidate([game])
if playable is not None:
qualified.append(playable)
else:
linux_skipped += 1
return qualified, confidence_skipped, linux_skipped
def _prompt_user_pick(qualified: list[GameInfo]) -> int:
"""Present numbered list, return 0-based index of user's choice."""
for i, g in enumerate(qualified, 1):
hours_str = (
f" (~{g.completionist_hours:.1f}h)" if g.completionist_hours > 0 else ""
)
_echo(f" {i}. {g.name} (AppID={g.app_id}){hours_str}")
while True:
raw = input("Select game number: ")
try:
idx = int(raw)
except ValueError:
_echo(f"Invalid input: {raw!r}")
continue
if idx < 1 or idx > len(qualified):
_echo(f"Out of range: {idx}")
continue
return idx - 1
def _assign_chosen_game(
chosen: GameInfo,
games: list[GameInfo],
state: State,
config: Config,
) -> None:
"""Save assignment, announce it, and handle install/uninstall."""
state.current_app_id = chosen.app_id
state.current_game_name = chosen.name
if not state.enforcement_started_at:
state.enforcement_started_at = datetime.now(timezone.utc).isoformat()
state.save()
hours_str = (
f" (~{chosen.completionist_hours:.1f}h leisure+dlc)"
if chosen.completionist_hours > 0
else ""
)
_echo(f"\n>>> ASSIGNED: {chosen.name} (AppID={chosen.app_id}){hours_str}")
_echo(
f" Progress: {chosen.unlocked_achievements}/{chosen.total_achievements}"
f" ({chosen.completion_pct:.1f}%)"
)
_report_poll_confidence(chosen, games, state)
if config.uninstall_other_games:
count = uninstall_other_games(chosen.app_id)
if count:
_echo(f"\n Uninstalled {count} non-assigned games")
if not is_game_installed(chosen.app_id):
_echo(f"\n Auto-installing {chosen.name}...")
install_game(
chosen.app_id, chosen.name, config.steam_id, use_steam_protocol=True
)
def _pick_next_game_sequential(
games: list[GameInfo],
state: State,
config: Config,
on_select: Callable[[GameInfo], bool],
) -> None:
"""Pick the next-shortest playable game, asking the user per candidate.
``on_select`` is called with each prospective pick. Returning ``True``
accepts the assignment; returning ``False`` records a 7-day skip on
``state`` for that game and the next candidate is evaluated.
"""
while True:
skip = set(state.finished_app_ids) | state.active_skipped_ids()
candidates = [g for g in games if not g.is_complete and g.app_id not in skip]
if not candidates:
_echo(_NO_CONF_MSG)
state.current_app_id = None
state.current_game_name = ""
state.save()
return
candidates.sort(key=_sort_key)
_apply_cached_confidence_to_candidates(candidates)
chosen, confidence_skipped, linux_skipped = _pick_next_shortest_candidate(
candidates
)
if chosen is None:
_echo(
_NO_CONF_MSG
if confidence_skipped > 0 and linux_skipped == 0
else "\nNo playable games left (all have poor ProtonDB ratings)!"
)
state.current_app_id = None
state.current_game_name = ""
state.save()
return
if not on_select(chosen):
state.skip_for_days(chosen.app_id, 7)
state.save()
_echo(f"\n Skipped {chosen.name} for 7 days; picking next...")
continue
_assign_chosen_game(chosen, games, state, config)
return
def pick_next_game(
@ -356,63 +198,6 @@ def pick_next_game(
_assign_chosen_game(qualified[idx], games, state, config)
def _pick_next_shortest_candidate(
candidates: list[GameInfo],
) -> tuple[GameInfo | None, int, int]:
"""Pick next game by checking confidence one candidate at a time.
The list must be pre-sorted by desired priority (shortest first).
"""
confidence_skipped = 0
linux_skipped = 0
for game in candidates:
if not _candidate_passes_hltb_confidence(game):
confidence_skipped += 1
continue
# Reuse existing ProtonDB compatibility gate for one candidate.
playable = _pick_playable_candidate([game])
if playable is not None:
if linux_skipped > 0:
_echo(
f" Skipped {linux_skipped} game(s) with poor Linux compatibility"
)
return playable, confidence_skipped, linux_skipped
linux_skipped += 1
if linux_skipped > 0:
_echo(f" Skipped {linux_skipped} game(s) with poor Linux compatibility")
return None, confidence_skipped, linux_skipped
def _collect_top_candidates(
candidates: list[GameInfo],
n: int = 3,
) -> tuple[list[GameInfo], int, int]:
"""Collect up to n candidates that pass the Linux compatibility gate.
Args:
candidates: Pre-sorted list of candidate games.
n: Maximum number of qualified games to collect.
Returns:
Tuple of (qualified_list, conf_skipped, linux_skipped).
"""
qualified: list[GameInfo] = []
linux_skipped = 0
for game in candidates:
if len(qualified) >= n:
break
playable = _pick_playable_candidate([game])
if playable is not None:
qualified.append(playable)
else:
linux_skipped += 1
if linux_skipped > 0:
_echo(f" Skipped {linux_skipped} game(s) with poor Linux compatibility")
return qualified, 0, linux_skipped
# ──────────────────────────────────────────────────────────────
# Checking & tampering detection
# ──────────────────────────────────────────────────────────────
@ -461,61 +246,3 @@ def do_check(config: Config, state: State) -> None:
# Tampering detection on snapshot.
detect_tampering(config, state)
def _check_game_tampering(
client: SteamAPIClient,
entry: dict[str, Any],
state: State,
) -> tuple[str, int, int] | None:
"""Check if a single game has unexpected achievement progress.
Args:
client: Steam API client.
entry: Snapshot entry for the game.
state: Current enforcer state.
Returns:
Tuple of (name, app_id, diff) if tampering detected, else None.
"""
app_id = entry["app_id"]
if app_id == state.current_app_id:
return None
if entry["unlocked_achievements"] >= entry["total_achievements"]:
return None
if entry.get("playtime_minutes", 0) <= 0:
return None
game = client.refresh_single_game(
app_id, entry["name"], entry.get("playtime_minutes", 0)
)
if game and game.unlocked_achievements > entry["unlocked_achievements"]:
diff = game.unlocked_achievements - entry["unlocked_achievements"]
return (entry["name"], app_id, diff)
return None
def detect_tampering(config: Config, state: State) -> None:
"""Check if achievements were unlocked on non-assigned games."""
old_snapshot = load_snapshot()
if old_snapshot is None:
return
client = SteamAPIClient(config.steam_api_key, config.steam_id)
# Quick check: only re-fetch a few random non-assigned games.
suspicious: list[tuple[str, int, int]] = []
for entry in old_snapshot:
result = _check_game_tampering(client, entry, state)
if result:
suspicious.append(result)
if len(suspicious) >= _TAMPER_CHECK_LIMIT:
break
if suspicious:
_echo("\n TAMPERING DETECTED:")
for name, app_id, diff in suspicious:
_echo(f" {name} (AppID={app_id}): +{diff} new achievements!")
send_notification(
"Tampering Detected!",
f"Achievements unlocked on {len(suspicious)} non-assigned games!",
)

8
steam_backlog_enforcer/tests/test_scanning.py
View File

@ -154,7 +154,7 @@ class TestPickPlayableCandidate:
game = _game(app_id=440, name="TF2")
with (
patch(
"steam_backlog_enforcer.scanning.fetch_protondb_ratings",
"steam_backlog_enforcer._scanning_candidates.fetch_protondb_ratings",
return_value={
440: ProtonDBRating(app_id=440, tier="gold"),
},
@ -171,7 +171,7 @@ class TestPickPlayableCandidate:
good = _game(app_id=2, name="Good")
with (
patch(
"steam_backlog_enforcer.scanning.fetch_protondb_ratings",
"steam_backlog_enforcer._scanning_candidates.fetch_protondb_ratings",
return_value={
1: ProtonDBRating(app_id=1, tier="borked"),
2: ProtonDBRating(app_id=2, tier="platinum"),
@ -188,7 +188,7 @@ class TestPickPlayableCandidate:
game = _game(app_id=1, name="Bad")
with (
patch(
"steam_backlog_enforcer.scanning.fetch_protondb_ratings",
"steam_backlog_enforcer._scanning_candidates.fetch_protondb_ratings",
return_value={
1: ProtonDBRating(app_id=1, tier="borked"),
},
@ -206,7 +206,7 @@ class TestPickPlayableCandidate:
game = _game(app_id=440, name="TF2")
with (
patch(
"steam_backlog_enforcer.scanning.fetch_protondb_ratings",
"steam_backlog_enforcer._scanning_candidates.fetch_protondb_ratings",
return_value={
440: ProtonDBRating(app_id=440, tier="platinum"),
},

34
steam_backlog_enforcer/tests/test_scanning_cached_confidence_overlay_avoids_refetch.py
View File

@ -58,16 +58,16 @@ class TestPickNextGameGroup2:
"steam_backlog_enforcer._scanning_confidence._refresh_candidate_confidence_batch"
) as mock_refresh_batch,
patch(
"steam_backlog_enforcer.scanning._pick_playable_candidate",
"steam_backlog_enforcer._scanning_candidates._pick_playable_candidate",
side_effect=lambda c: c[0] if c else None,
),
patch("steam_backlog_enforcer.scanning._echo"),
patch(
"steam_backlog_enforcer.scanning.is_game_installed",
"steam_backlog_enforcer._scanning_assign.is_game_installed",
return_value=True,
),
patch(
"steam_backlog_enforcer.scanning.uninstall_other_games",
"steam_backlog_enforcer._scanning_assign.uninstall_other_games",
return_value=0,
),
patch("builtins.input", return_value="1"),
@ -95,16 +95,16 @@ class TestPickNextGameGroup2:
with (
patch(
"steam_backlog_enforcer.scanning._pick_playable_candidate",
"steam_backlog_enforcer._scanning_candidates._pick_playable_candidate",
side_effect=playable_side_effect,
),
patch("steam_backlog_enforcer.scanning._echo"),
patch(
"steam_backlog_enforcer.scanning.is_game_installed",
"steam_backlog_enforcer._scanning_assign.is_game_installed",
return_value=True,
),
patch(
"steam_backlog_enforcer.scanning.uninstall_other_games",
"steam_backlog_enforcer._scanning_assign.uninstall_other_games",
return_value=0,
),
patch("builtins.input", return_value="1"),
@ -122,16 +122,16 @@ class TestPickNextGameGroup2:
state = State()
with (
patch(
"steam_backlog_enforcer.scanning._pick_playable_candidate",
"steam_backlog_enforcer._scanning_candidates._pick_playable_candidate",
side_effect=lambda c: c[0] if c else None,
),
patch("steam_backlog_enforcer.scanning._echo"),
patch(
"steam_backlog_enforcer.scanning.is_game_installed",
"steam_backlog_enforcer._scanning_assign.is_game_installed",
return_value=True,
),
patch(
"steam_backlog_enforcer.scanning.uninstall_other_games",
"steam_backlog_enforcer._scanning_assign.uninstall_other_games",
return_value=0,
),
patch("builtins.input", return_value="2"),
@ -147,19 +147,19 @@ class TestPickNextGameGroup2:
echoed: list[str] = []
with (
patch(
"steam_backlog_enforcer.scanning._pick_playable_candidate",
"steam_backlog_enforcer._scanning_candidates._pick_playable_candidate",
side_effect=lambda c: c[0] if c else None,
),
patch(
"steam_backlog_enforcer.scanning._echo",
"steam_backlog_enforcer._scanning_assign._echo",
side_effect=lambda *a, **_: echoed.append(a[0]),
),
patch(
"steam_backlog_enforcer.scanning.is_game_installed",
"steam_backlog_enforcer._scanning_assign.is_game_installed",
return_value=True,
),
patch(
"steam_backlog_enforcer.scanning.uninstall_other_games",
"steam_backlog_enforcer._scanning_assign.uninstall_other_games",
return_value=0,
),
patch("builtins.input", side_effect=["abc", "1"]),
@ -176,19 +176,19 @@ class TestPickNextGameGroup2:
echoed: list[str] = []
with (
patch(
"steam_backlog_enforcer.scanning._pick_playable_candidate",
"steam_backlog_enforcer._scanning_candidates._pick_playable_candidate",
side_effect=lambda c: c[0] if c else None,
),
patch(
"steam_backlog_enforcer.scanning._echo",
"steam_backlog_enforcer._scanning_assign._echo",
side_effect=lambda *a, **_: echoed.append(a[0]),
),
patch(
"steam_backlog_enforcer.scanning.is_game_installed",
"steam_backlog_enforcer._scanning_assign.is_game_installed",
return_value=True,
),
patch(
"steam_backlog_enforcer.scanning.uninstall_other_games",
"steam_backlog_enforcer._scanning_assign.uninstall_other_games",
return_value=0,
),
patch("builtins.input", side_effect=["99", "1"]),

4
steam_backlog_enforcer/tests/test_scanning_confidence_helpers.py
View File

@ -115,10 +115,10 @@ class TestConfidenceHelpers:
with (
patch(
"steam_backlog_enforcer.scanning._pick_playable_candidate",
"steam_backlog_enforcer._scanning_candidates._pick_playable_candidate",
side_effect=[None, good],
),
patch("steam_backlog_enforcer.scanning._echo") as mock_echo,
patch("steam_backlog_enforcer._scanning_candidates._echo") as mock_echo,
):
picked, skipped_low_conf, skipped_linux = _pick_next_shortest_candidate(
[bad, good],

16
steam_backlog_enforcer/tests/test_scanning_confidence_helpers_pick_next_shortest_candidate_no_echo_whe.py
View File

@ -41,10 +41,10 @@ class TestConfidenceHelpersGroup2:
good = _game(app_id=51, name="Good", hours=2.0)
with (
patch(
"steam_backlog_enforcer.scanning._pick_playable_candidate",
"steam_backlog_enforcer._scanning_candidates._pick_playable_candidate",
return_value=good,
),
patch("steam_backlog_enforcer.scanning._echo") as mock_echo,
patch("steam_backlog_enforcer._scanning_candidates._echo") as mock_echo,
):
picked, _skipped_low_conf, skipped_linux = _pick_next_shortest_candidate(
[good],
@ -62,7 +62,7 @@ class TestConfidenceHelpersGroup2:
patch(
"steam_backlog_enforcer._scanning_confidence._refresh_candidate_confidence"
),
patch("steam_backlog_enforcer.scanning._echo"),
patch("steam_backlog_enforcer._scanning_candidates._echo"),
):
picked, skipped_low_conf, skipped_linux = _pick_next_shortest_candidate(
[low_conf],
@ -76,10 +76,10 @@ class TestConfidenceHelpersGroup2:
g1 = _game(app_id=10, name="Borked", hours=1.0)
with (
patch(
"steam_backlog_enforcer.scanning._pick_playable_candidate",
"steam_backlog_enforcer._scanning_candidates._pick_playable_candidate",
return_value=None,
),
patch("steam_backlog_enforcer.scanning._echo") as mock_echo,
patch("steam_backlog_enforcer._scanning_candidates._echo") as mock_echo,
):
picked, _skipped_low_conf, skipped_linux = _pick_next_shortest_candidate(
[g1],
@ -100,7 +100,7 @@ class TestConfidenceHelpersGroup2:
"steam_backlog_enforcer.scanning.SteamAPIClient",
return_value=mock_client,
),
patch("steam_backlog_enforcer.scanning._echo"),
patch("steam_backlog_enforcer._scanning_candidates._echo"),
patch(
"steam_backlog_enforcer.scanning.send_notification",
),
@ -127,7 +127,7 @@ class TestConfidenceHelpersGroup2:
"steam_backlog_enforcer.scanning.SteamAPIClient",
return_value=mock_client,
),
patch("steam_backlog_enforcer.scanning._echo"),
patch("steam_backlog_enforcer._scanning_candidates._echo"),
patch(
"steam_backlog_enforcer.scanning.send_notification",
),
@ -150,7 +150,7 @@ class TestConfidenceHelpersGroup2:
"steam_backlog_enforcer.scanning.SteamAPIClient",
return_value=mock_client,
),
patch("steam_backlog_enforcer.scanning._echo"),
patch("steam_backlog_enforcer._scanning_candidates._echo"),
patch("steam_backlog_enforcer.scanning.detect_tampering"),
):
state = State(current_app_id=440, current_game_name="TF2")

2
steam_backlog_enforcer/tests/test_scanning_part2.py
View File

@ -11,7 +11,7 @@ from steam_backlog_enforcer.scanning import (
detect_tampering,
)
PKG = "steam_backlog_enforcer.scanning"
PKG = "steam_backlog_enforcer._scanning_tampering"
def _entry(

12
steam_backlog_enforcer/tests/test_scanning_part3.py
View File

@ -54,7 +54,7 @@ class TestPickNextGame:
side_effect=refresh_side_effect,
) as mock_refresh,
patch(
"steam_backlog_enforcer.scanning._pick_playable_candidate",
"steam_backlog_enforcer._scanning_candidates._pick_playable_candidate",
side_effect=lambda c: c[0] if c else None,
),
patch(
@ -66,11 +66,11 @@ class TestPickNextGame:
side_effect=lambda *a, **_: echoed.append(a[0]),
),
patch(
"steam_backlog_enforcer.scanning.is_game_installed",
"steam_backlog_enforcer._scanning_assign.is_game_installed",
return_value=True,
),
patch(
"steam_backlog_enforcer.scanning.uninstall_other_games",
"steam_backlog_enforcer._scanning_assign.uninstall_other_games",
return_value=0,
),
patch("builtins.input", return_value="1"),
@ -96,16 +96,16 @@ class TestPickNextGame:
"steam_backlog_enforcer._scanning_confidence._refresh_candidate_confidence_batch"
) as mock_refresh_batch,
patch(
"steam_backlog_enforcer.scanning._pick_playable_candidate",
"steam_backlog_enforcer._scanning_candidates._pick_playable_candidate",
side_effect=lambda c: c[0] if c else None,
),
patch("steam_backlog_enforcer.scanning._echo"),
patch(
"steam_backlog_enforcer.scanning.is_game_installed",
"steam_backlog_enforcer._scanning_assign.is_game_installed",
return_value=True,
),
patch(
"steam_backlog_enforcer.scanning.uninstall_other_games",
"steam_backlog_enforcer._scanning_assign.uninstall_other_games",
return_value=0,
),
patch("builtins.input", return_value="1"),

10
steam_backlog_enforcer/tests/test_scanning_part4.py
View File

@ -38,7 +38,7 @@ class TestCollectTopCandidates:
"""Returns at most n qualified candidates."""
games = [_game(app_id=i, name=f"G{i}", hours=float(i)) for i in range(1, 6)]
with patch(
"steam_backlog_enforcer.scanning._pick_playable_candidate",
"steam_backlog_enforcer._scanning_candidates._pick_playable_candidate",
side_effect=lambda c: c[0] if c else None,
):
qualified, conf_skip, linux_skip = _collect_top_candidates(games, n=3)
@ -53,10 +53,10 @@ class TestCollectTopCandidates:
g2 = _game(app_id=2, name="Good", hours=2.0)
with (
patch(
"steam_backlog_enforcer.scanning._pick_playable_candidate",
"steam_backlog_enforcer._scanning_candidates._pick_playable_candidate",
side_effect=lambda c: None if c[0].app_id == 1 else c[0],
),
patch("steam_backlog_enforcer.scanning._echo"),
patch("steam_backlog_enforcer._scanning_candidates._echo"),
):
qualified, conf_skip, linux_skip = _collect_top_candidates([g1, g2], n=10)
assert [g.app_id for g in qualified] == [2]
@ -75,10 +75,10 @@ class TestCollectTopCandidates:
g = _game(app_id=1, name="Good", hours=1.0)
with (
patch(
"steam_backlog_enforcer.scanning._pick_playable_candidate",
"steam_backlog_enforcer._scanning_candidates._pick_playable_candidate",
side_effect=lambda c: c[0] if c else None,
),
patch("steam_backlog_enforcer.scanning._echo") as mock_echo,
patch("steam_backlog_enforcer._scanning_candidates._echo") as mock_echo,
):
_collect_top_candidates([g], n=10)
mock_echo.assert_not_called()

28
steam_backlog_enforcer/tests/test_scanning_pick_next_game.py
View File

@ -41,17 +41,17 @@ class TestPickNextGame:
state = State()
with (
patch(
"steam_backlog_enforcer.scanning._pick_playable_candidate",
"steam_backlog_enforcer._scanning_candidates._pick_playable_candidate",
side_effect=lambda c: c[0] if c else None,
),
patch("steam_backlog_enforcer.scanning._echo"),
patch("steam_backlog_enforcer._scanning_confidence._echo"),
patch(
"steam_backlog_enforcer.scanning.is_game_installed",
"steam_backlog_enforcer._scanning_assign.is_game_installed",
return_value=True,
),
patch(
"steam_backlog_enforcer.scanning.uninstall_other_games",
"steam_backlog_enforcer._scanning_assign.uninstall_other_games",
return_value=0,
),
patch("builtins.input", return_value="1"),
@ -76,16 +76,16 @@ class TestPickNextGame:
state = State(finished_app_ids=[1])
with (
patch(
"steam_backlog_enforcer.scanning._pick_playable_candidate",
"steam_backlog_enforcer._scanning_candidates._pick_playable_candidate",
side_effect=lambda c: c[0] if c else None,
),
patch("steam_backlog_enforcer.scanning._echo"),
patch(
"steam_backlog_enforcer.scanning.is_game_installed",
"steam_backlog_enforcer._scanning_assign.is_game_installed",
return_value=True,
),
patch(
"steam_backlog_enforcer.scanning.uninstall_other_games",
"steam_backlog_enforcer._scanning_assign.uninstall_other_games",
return_value=0,
),
patch("builtins.input", return_value="1"),
@ -100,7 +100,7 @@ class TestPickNextGame:
state = State()
with (
patch(
"steam_backlog_enforcer.scanning._pick_playable_candidate",
"steam_backlog_enforcer._scanning_candidates._pick_playable_candidate",
return_value=None,
),
patch("steam_backlog_enforcer.scanning._echo"),
@ -115,17 +115,17 @@ class TestPickNextGame:
state = State()
with (
patch(
"steam_backlog_enforcer.scanning._pick_playable_candidate",
"steam_backlog_enforcer._scanning_candidates._pick_playable_candidate",
side_effect=lambda c: c[0] if c else None,
),
patch("steam_backlog_enforcer.scanning._echo"),
patch("steam_backlog_enforcer._scanning_confidence._echo"),
patch(
"steam_backlog_enforcer.scanning.uninstall_other_games",
"steam_backlog_enforcer._scanning_assign.uninstall_other_games",
return_value=2,
),
patch(
"steam_backlog_enforcer.scanning.is_game_installed",
"steam_backlog_enforcer._scanning_assign.is_game_installed",
return_value=True,
),
patch("builtins.input", return_value="1"),
@ -140,16 +140,18 @@ class TestPickNextGame:
state = State()
with (
patch(
"steam_backlog_enforcer.scanning._pick_playable_candidate",
"steam_backlog_enforcer._scanning_candidates._pick_playable_candidate",
side_effect=lambda c: c[0] if c else None,
),
patch("steam_backlog_enforcer.scanning._echo"),
patch("steam_backlog_enforcer._scanning_confidence._echo"),
patch(
"steam_backlog_enforcer.scanning.is_game_installed",
"steam_backlog_enforcer._scanning_assign.is_game_installed",
return_value=False,
),
patch("steam_backlog_enforcer.scanning.install_game") as mock_install,
patch(
"steam_backlog_enforcer._scanning_assign.install_game"
) as mock_install,
patch("builtins.input", return_value="1"),
):
pick_next_game([g1], state, config)

16
steam_backlog_enforcer/tests/test_scanning_pick_next_game_sequential.py
View File

@ -38,25 +38,25 @@ class TestPickNextGameSequential:
stack = contextlib.ExitStack()
stack.enter_context(
patch(
"steam_backlog_enforcer.scanning._pick_playable_candidate",
"steam_backlog_enforcer._scanning_candidates._pick_playable_candidate",
side_effect=lambda c: c[0] if c else None,
)
)
stack.enter_context(
patch(
"steam_backlog_enforcer.scanning._echo",
"steam_backlog_enforcer._scanning_assign._echo",
side_effect=lambda *a, **_: echoed.append(a[0]),
)
)
stack.enter_context(
patch(
"steam_backlog_enforcer.scanning.is_game_installed",
"steam_backlog_enforcer._scanning_assign.is_game_installed",
return_value=True,
)
)
stack.enter_context(
patch(
"steam_backlog_enforcer.scanning.uninstall_other_games",
"steam_backlog_enforcer._scanning_assign.uninstall_other_games",
return_value=0,
)
)
@ -113,11 +113,11 @@ class TestPickNextGameSequential:
echoed: list[str] = []
with (
patch(
"steam_backlog_enforcer.scanning._pick_playable_candidate",
"steam_backlog_enforcer._scanning_candidates._pick_playable_candidate",
return_value=None,
),
patch(
"steam_backlog_enforcer.scanning._echo",
"steam_backlog_enforcer._scanning_assign._echo",
side_effect=lambda *a, **_: echoed.append(a[0]),
),
patch("steam_backlog_enforcer.config._atomic_write"),
@ -136,11 +136,11 @@ class TestPickNextGameSequential:
echoed: list[str] = []
with (
patch(
"steam_backlog_enforcer.scanning._pick_playable_candidate",
"steam_backlog_enforcer._scanning_candidates._pick_playable_candidate",
side_effect=lambda c: c[0] if c else None,
),
patch(
"steam_backlog_enforcer.scanning._echo",
"steam_backlog_enforcer._scanning_assign._echo",
side_effect=lambda *a, **_: echoed.append(a[0]),
),
patch(

20
steam_backlog_enforcer/tests/test_scanning_pick_next_game_unknown_hours.py
View File

@ -41,16 +41,16 @@ class TestPickNextGameGroup2:
state = State()
with (
patch(
"steam_backlog_enforcer.scanning._pick_playable_candidate",
"steam_backlog_enforcer._scanning_candidates._pick_playable_candidate",
side_effect=lambda c: c[0] if c else None,
),
patch("steam_backlog_enforcer.scanning._echo"),
patch(
"steam_backlog_enforcer.scanning.is_game_installed",
"steam_backlog_enforcer._scanning_assign.is_game_installed",
return_value=True,
),
patch(
"steam_backlog_enforcer.scanning.uninstall_other_games",
"steam_backlog_enforcer._scanning_assign.uninstall_other_games",
return_value=0,
),
patch("builtins.input", return_value="1"),
@ -65,16 +65,16 @@ class TestPickNextGameGroup2:
state = State()
with (
patch(
"steam_backlog_enforcer.scanning._pick_playable_candidate",
"steam_backlog_enforcer._scanning_candidates._pick_playable_candidate",
side_effect=lambda c: c[0] if c else None,
),
patch("steam_backlog_enforcer.scanning._echo"),
patch(
"steam_backlog_enforcer.scanning.is_game_installed",
"steam_backlog_enforcer._scanning_assign.is_game_installed",
return_value=True,
),
patch(
"steam_backlog_enforcer.scanning.uninstall_other_games",
"steam_backlog_enforcer._scanning_assign.uninstall_other_games",
return_value=0,
),
patch("builtins.input", return_value="1"),
@ -95,7 +95,7 @@ class TestPickNextGameGroup2:
state = State()
with (
patch(
"steam_backlog_enforcer.scanning._pick_playable_candidate",
"steam_backlog_enforcer._scanning_candidates._pick_playable_candidate",
side_effect=lambda c: c[0] if c else None,
),
patch(
@ -107,11 +107,11 @@ class TestPickNextGameGroup2:
side_effect=lambda *a, **_: echoed.append(a[0]),
),
patch(
"steam_backlog_enforcer.scanning.is_game_installed",
"steam_backlog_enforcer._scanning_assign.is_game_installed",
return_value=True,
),
patch(
"steam_backlog_enforcer.scanning.uninstall_other_games",
"steam_backlog_enforcer._scanning_assign.uninstall_other_games",
return_value=0,
),
patch("builtins.input", return_value="1"),
@ -142,7 +142,7 @@ class TestPickNextGameGroup2:
side_effect=lambda *a, **_: echoed.append(a[0]),
),
patch(
"steam_backlog_enforcer.scanning._pick_playable_candidate",
"steam_backlog_enforcer._scanning_candidates._pick_playable_candidate",
return_value=None,
) as mock_pick,
):

2
steam_backlog_enforcer/tests/test_stats.py
View File

@ -12,7 +12,7 @@ from steam_backlog_enforcer.config import State
from steam_backlog_enforcer.protondb import ProtonDBRating
from steam_backlog_enforcer.steam_api import GameInfo
_PKG = "steam_backlog_enforcer._stats"
_PKG = "steam_backlog_enforcer._stats_gathering"
def _game(

2
steam_backlog_enforcer/tests/test_stats_ensure_rush_data.py
View File

@ -12,7 +12,7 @@ from steam_backlog_enforcer._stats import (
from steam_backlog_enforcer.protondb import ProtonDBRating
from steam_backlog_enforcer.steam_api import GameInfo
_PKG = "steam_backlog_enforcer._stats"
_PKG = "steam_backlog_enforcer._stats_gathering"
def _game(

2
steam_backlog_enforcer/tests/test_stats_print_worst_example.py
View File

@ -11,7 +11,7 @@ from steam_backlog_enforcer._stats import (
from steam_backlog_enforcer.protondb import ProtonDBRating
from steam_backlog_enforcer.steam_api import GameInfo
_PKG = "steam_backlog_enforcer._stats"
_PKG = "steam_backlog_enforcer._stats_display"
def _game(

2
steam_backlog_enforcer/tests/test_stats_refresh_recently_played_completions.py
View File

@ -8,7 +8,7 @@ from steam_backlog_enforcer.config import Config
from steam_backlog_enforcer.protondb import ProtonDBRating
from steam_backlog_enforcer.steam_api import GameInfo, SteamAPIError
_PKG = "steam_backlog_enforcer._stats"
_PKG = "steam_backlog_enforcer._stats_gathering"
def _game(

16
steam_backlog_enforcer/tests/test_stats_sum_hours.py
View File

@ -16,7 +16,9 @@ from steam_backlog_enforcer.config import State
from steam_backlog_enforcer.protondb import ProtonDBRating
from steam_backlog_enforcer.steam_api import GameInfo
_PKG = "steam_backlog_enforcer._stats"
_PKG = "steam_backlog_enforcer._stats_display"
# _print_scenario was not moved: it still lives in _stats.
_PKG_STATS = "steam_backlog_enforcer._stats"
def _game(
@ -118,14 +120,18 @@ class TestPrintScenario:
def test_no_data_prints_no_data_message(self) -> None:
"""Test no data prints no data message."""
echoed: list[str] = []
with patch(f"{_PKG}._echo", side_effect=lambda *a, **_: echoed.append(a[0])):
with patch(
f"{_PKG_STATS}._echo", side_effect=lambda *a, **_: echoed.append(a[0])
):
_print_scenario("2. RUSH", 0.0, 0, 5)
assert any("No data available" in s for s in echoed)
def test_with_data_no_missing(self) -> None:
"""Test with data no missing."""
echoed: list[str] = []
with patch(f"{_PKG}._echo", side_effect=lambda *a, **_: echoed.append(a[0])):
with patch(
f"{_PKG_STATS}._echo", side_effect=lambda *a, **_: echoed.append(a[0])
):
_print_scenario("2. RUSH", 100.0, 0, 5)
assert any("Total:" in s for s in echoed)
assert not any("had no data" in s for s in echoed)
@ -133,7 +139,9 @@ class TestPrintScenario:
def test_with_data_and_missing(self) -> None:
"""Test with data and missing."""
echoed: list[str] = []
with patch(f"{_PKG}._echo", side_effect=lambda *a, **_: echoed.append(a[0])):
with patch(
f"{_PKG_STATS}._echo", side_effect=lambda *a, **_: echoed.append(a[0])
):
_print_scenario("2. RUSH", 100.0, 2, 5)
assert any("had no data" in s for s in echoed)

2
steam_backlog_enforcer/tests/test_stats_unplayable_rating_counts_linux_skipped.py
View File

@ -12,7 +12,7 @@ from steam_backlog_enforcer.config import State
from steam_backlog_enforcer.protondb import ProtonDBRating
from steam_backlog_enforcer.steam_api import GameInfo
_PKG = "steam_backlog_enforcer._stats"
_PKG = "steam_backlog_enforcer._stats_gathering"
def _game(