From 48b609e1a388699517c7d311882bc5a8b7e67f8d Mon Sep 17 00:00:00 2001 From: Krzysztof kuhy Rudnicki Date: Thu, 28 May 2026 07:02:48 +0200 Subject: [PATCH] =?UTF-8?q?steam=5Fbacklog=5Fenforcer:=20fix=20stats=20com?= =?UTF-8?q?mand=20=E2=80=94=20show=20real=20Rush/Leisure/Worst=20data?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Four bugs fixed: - HLTB search returned 0 results for ~87 games with special chars (™, ®, &, standalone -, (Legacy), RHCP, etc.) — add _sanitize_search_name() and extend _build_search_variants() with Steam-suffix and edition stripping - fetch_hltb_detail_missing returned immediately because `app_id not in rush` was always False (all keys present with -1) — fix to `rush.get(id,-1) <= 0` - save_hltb_cache overwrote rush/leisure on confidence-only partial saves — now reads existing cache and preserves data when extras dicts are empty - _filter_qualifying_games excluded 57 games with stale snapshot hours (-1) even though HLTB hours cache had valid data — add cache fallback Result: stats shows Rush 64,670h / Leisure 136,807h / Worst 228,594h for all 785 qualifying games with full rush+leisure detail. Co-Authored-By: Claude Sonnet 4.6 --- steam_backlog_enforcer/_hltb_detail.py | 75 +- steam_backlog_enforcer/_hltb_search.py | 105 ++- steam_backlog_enforcer/_hltb_types.py | 102 ++- .../_scanning_confidence.py | 7 +- steam_backlog_enforcer/_stats.py | 350 +++++++++ steam_backlog_enforcer/config.py | 2 + steam_backlog_enforcer/hltb.py | 97 ++- steam_backlog_enforcer/main.py | 2 + steam_backlog_enforcer/protondb.py | 17 +- steam_backlog_enforcer/scanning.py | 3 + .../tests/test_hltb_detail.py | 113 ++- .../tests/test_hltb_part2.py | 145 +++- .../tests/test_hltb_search.py | 20 + .../tests/test_hltb_search_part2.py | 18 + .../tests/test_polls_tracking.py | 36 +- steam_backlog_enforcer/tests/test_protondb.py | 21 +- .../tests/test_scanning_part3.py | 11 + steam_backlog_enforcer/tests/test_stats.py | 701 ++++++++++++++++++ 18 files changed, 1761 insertions(+), 64 deletions(-) create mode 100644 steam_backlog_enforcer/_stats.py create mode 100644 steam_backlog_enforcer/tests/test_stats.py diff --git a/steam_backlog_enforcer/_hltb_detail.py b/steam_backlog_enforcer/_hltb_detail.py index 4e3ed45..9137ef2 100644 --- a/steam_backlog_enforcer/_hltb_detail.py +++ b/steam_backlog_enforcer/_hltb_detail.py @@ -17,6 +17,7 @@ from python_pkg.steam_backlog_enforcer._hltb_types import ( MAX_CONCURRENT, HLTBResult, ProgressCb, + _HLTBExtras, save_hltb_cache, ) @@ -70,6 +71,28 @@ def _platform_comp_high_candidates(game_data: dict[str, Any]) -> list[int]: return candidates +def _extract_comp_100_avg_and_high(game_data: dict[str, Any]) -> tuple[float, float]: + """Extract (average comp_100, high comp_100) from game detail data. + + Returns hours as floats: (avg_hours, high_hours). Returns (-1, -1) when + insufficient data is present. The average is ``comp_100`` (seconds) from + ``game[0]``; the high is ``comp_100_h``. + """ + games = game_data.get("game", []) + if not isinstance(games, list) or not games: + return -1, -1 + if not isinstance(games[0], dict): + return -1, -1 + + base = games[0] + avg_s = _as_positive_int(base.get("comp_100", 0)) + high_s = _as_positive_int(base.get("comp_100_h", 0)) + + avg_h = round(avg_s / 3600, 2) if avg_s > 0 else -1 + high_h = round(high_s / 3600, 2) if high_s > 0 else avg_h + return avg_h, high_h + + def _extract_base_leisure_hours(game_data: dict[str, Any]) -> float: """Extract base-game leisure hours from game detail data. @@ -186,22 +209,46 @@ async def _fetch_detail_one( return None +def _process_game_detail( + game_data: dict[str, Any], + dlc_rels: list[tuple[int, float]], + dlc_hours_by_id: dict[int, float], +) -> tuple[float, float, float]: + """Return (leisure_hours, rush_hours, leisure_100h) for one game's detail data.""" + leisure = _extract_leisure_hours(game_data) + if leisure > 0: + leisure = _apply_dlc_leisure_overrides(leisure, dlc_rels, dlc_hours_by_id) + + avg_h, high_h = _extract_comp_100_avg_and_high(game_data) + rush_h = -1.0 + if avg_h > 0: + dlc_rush = sum(fh for _, fh in dlc_rels if fh > 0) + rush_h = round(avg_h + dlc_rush, 2) + + l100 = -1.0 + if high_h > 0: + l100 = _apply_dlc_leisure_overrides(high_h, dlc_rels, dlc_hours_by_id) + + return leisure, rush_h, l100 + + async def _fetch_leisure_times( search_results: list[HLTBResult], cache: dict[int, float], polls: dict[int, int], progress_cb: ProgressCb | None, - count_comp: dict[int, int] | None = None, + extras: _HLTBExtras | None = None, ) -> None: """Fetch leisure times from game detail pages for all search results. Updates ``cache`` in-place with leisure hours (including DLC time). - The ``polls`` and ``count_comp`` mappings are forwarded to - :func:`save_hltb_cache` so the on-disk cache keeps confidence metrics - captured during the search step. + Also populates ``extras.rush`` (avg comp_100 + DLC) and + ``extras.leisure_100h`` (comp_100_h + DLC leisure). + The ``polls`` and ``extras.count_comp`` are forwarded to + :func:`save_hltb_cache` so confidence metrics persist. """ - if count_comp is None: - count_comp = {} + if extras is None: + extras = _HLTBExtras() valid = [r for r in search_results if r.hltb_game_id > 0] if not valid: @@ -231,22 +278,24 @@ async def _fetch_leisure_times( for r, game_data in zip(valid, details, strict=False): done += 1 if game_data is not None: - leisure = _extract_leisure_hours(game_data) + dlc_rels = dlc_relationships_by_app.get(r.app_id, []) + leisure, rush_h, l100 = _process_game_detail( + game_data, dlc_rels, dlc_hours_by_id + ) if leisure > 0: - leisure = _apply_dlc_leisure_overrides( - leisure, - dlc_relationships_by_app.get(r.app_id, []), - dlc_hours_by_id, - ) r.completionist_hours = leisure cache[r.app_id] = leisure found += 1 + if rush_h > 0: + extras.rush[r.app_id] = rush_h + if l100 > 0: + extras.leisure_100h[r.app_id] = l100 if progress_cb is not None: progress_cb(done, total, found, r.game_name) if not done % _SAVE_INTERVAL: - save_hltb_cache(cache, polls, count_comp) + save_hltb_cache(cache, polls, extras) def _collect_dlc_relationships( diff --git a/steam_backlog_enforcer/_hltb_search.py b/steam_backlog_enforcer/_hltb_search.py index 2ec5b95..b23c09b 100644 --- a/steam_backlog_enforcer/_hltb_search.py +++ b/steam_backlog_enforcer/_hltb_search.py @@ -26,6 +26,7 @@ from python_pkg.steam_backlog_enforcer._hltb_types import ( HLTBResult, ProgressCb, _AuthInfo, + _HLTBExtras, save_hltb_cache, ) @@ -98,11 +99,30 @@ def _similarity(a: str, b: str) -> float: return SequenceMatcher(None, a.lower(), b.lower()).ratio() +_TM_RE = re.compile("[™®©\uff0a]") +_COLON_WORD_RE = re.compile(r":(\s|$)") +_STANDALONE_PUNCT_RE = re.compile(r"^[-/|]$") +_AMP_RE = re.compile(r"\s*&\s*") + + +def _sanitize_search_name(name: str) -> str: + """Strip HLTB-breaking characters from a game name for searchTerms. + + Removes trademark/copyright symbols, colons at word-end, standalone + punctuation tokens (dash, slash, pipe), and replaces & with 'and'. + """ + cleaned = _TM_RE.sub("", name) + cleaned = _AMP_RE.sub(" and ", cleaned) + cleaned = _COLON_WORD_RE.sub(" ", cleaned) + tokens = [t for t in cleaned.split() if not _STANDALONE_PUNCT_RE.match(t)] + return " ".join(tokens) + + def _build_search_payload(game_name: str, auth: _AuthInfo | None = None) -> str: """Build the JSON POST body for an HLTB search.""" payload: dict[str, Any] = { "searchType": "games", - "searchTerms": game_name.split(), + "searchTerms": _sanitize_search_name(game_name).split(), "searchPage": 1, "size": 20, "searchOptions": { @@ -134,13 +154,66 @@ def _build_search_payload(game_name: str, auth: _AuthInfo | None = None) -> str: return json.dumps(payload) +_STEAM_SUFFIX_RE = re.compile( + r"\s+(?:\(Legacy\)|\(Classic\)|\(beta\)|\(Remastered\)|Legacy|Classic|RHCP" + r"|\(Phase\s+\d+\))\s*$", + re.IGNORECASE, +) + + def _build_search_variants(game_name: str) -> list[str]: - """Return fallback search terms for one Steam game title.""" + """Return fallback search terms for one Steam game title. + + Tries progressively simplified names so HLTB search finds a result even + when the Steam title contains edition suffixes, Steam-only labels, or + subtitle decorators that HLTB does not index under. + + Order matters: most-specific first, then stripped-down fallbacks. + Simplifications are chained so e.g. "Foo - Bar Edition" → + "Foo - Bar" → "Foo" and "Foo - Bar Edition" → "Foo Edition" → "Foo". + """ base = game_name.strip() - variants = [base] + seen: set[str] = set() + variants: list[str] = [] + + def _add(name: str) -> None: + s = name.strip() + if s and s not in seen: + seen.add(s) + variants.append(s) + + _add(base) + + # Strip Steam-only labels that HLTB never uses + no_steam = _STEAM_SUFFIX_RE.sub("", base).strip() + _add(no_steam) + + # Strip trailing year "(YYYY)" no_year = re.sub(r"\s*\(\d{4}\)$", "", base).strip() - if no_year and no_year != base: - variants.append(no_year) + _add(no_year) + + # Strip " - subtitle" portion (e.g. "Brothers - A Tale of Two Sons" → "Brothers") + no_subtitle = re.sub(r"\s+-\s+.*$", "", base).strip() + _add(no_subtitle) + # Also strip edition from the subtitle-stripped name. + # e.g. "Rocksmith 2014 Edition - Remastered" → "Rocksmith 2014 Edition" + # → "Rocksmith 2014" + if no_subtitle != base: + _add(re.sub(r"\s+\w+\s+Edition\s*$", "", no_subtitle, flags=re.IGNORECASE)) + _add(re.sub(r"\s+Edition\s*$", "", no_subtitle, flags=re.IGNORECASE)) + + # Strip "GOTY Edition" / "Gold Edition" / "Definitive Edition" etc. from base + no_edition = re.sub(r"\s+\w+\s+Edition\s*$", "", base, flags=re.IGNORECASE).strip() + _add(no_edition) + + # Strip just " Edition" at end from base + no_bare_edition = re.sub(r"\s+Edition\s*$", "", base, flags=re.IGNORECASE).strip() + _add(no_bare_edition) + + # Strip ": subtitle" portion (e.g. "Batman: Arkham Asylum" → "Batman") + no_colon_sub = re.sub(r"\s*:.*$", "", base).strip() + _add(no_colon_sub) + return variants @@ -322,6 +395,7 @@ class _SearchCtx: counter: dict[str, int] = field(default_factory=dict) total: int = 0 progress_cb: ProgressCb | None = None + hltb_game_id: dict[int, int] = field(default_factory=dict) async def _search_one( @@ -358,6 +432,8 @@ async def _search_one( ctx.cache[app_id] = result.completionist_hours ctx.polls[app_id] = result.comp_100_count ctx.count_comp[app_id] = result.count_comp + if result.hltb_game_id > 0: + ctx.hltb_game_id[app_id] = result.hltb_game_id ctx.counter["found"] += 1 else: ctx.cache[app_id] = -1 @@ -369,7 +445,11 @@ async def _search_one( # Incremental save every _SAVE_INTERVAL lookups. if not done % _SAVE_INTERVAL: - save_hltb_cache(ctx.cache, ctx.polls, ctx.count_comp) + save_hltb_cache( + ctx.cache, + ctx.polls, + _HLTBExtras(count_comp=ctx.count_comp, hltb_game_id=ctx.hltb_game_id), + ) # Report progress. if ctx.progress_cb is not None: @@ -383,9 +463,12 @@ async def _fetch_batch( cache: dict[int, float], polls: dict[int, int], progress_cb: ProgressCb | None, - count_comp: dict[int, int] | None = None, + extras: _HLTBExtras | None = None, ) -> list[HLTBResult]: """Fetch HLTB data for a batch of games using one shared session.""" + if extras is None: + extras = _HLTBExtras() + # 1. Discover the search URL (sync, one-time). search_url = _get_hltb_search_url() logger.info("HLTB search URL: %s", search_url) @@ -419,9 +502,6 @@ async def _fetch_batch( counter = {"done": 0, "found": 0} total = len(games) - if count_comp is None: - count_comp = {} - connector = aiohttp.TCPConnector( limit=MAX_CONCURRENT, keepalive_timeout=30, @@ -436,11 +516,12 @@ async def _fetch_batch( headers=headers, cache=cache, polls=polls, - count_comp=count_comp, + count_comp=extras.count_comp, auth=auth, counter=counter, total=total, progress_cb=progress_cb, + hltb_game_id=extras.hltb_game_id, ) tasks = [ _search_one( @@ -465,7 +546,7 @@ async def _fetch_batch( cache, polls, progress_cb=None, - count_comp=count_comp, + extras=extras, ) return search_results diff --git a/steam_backlog_enforcer/_hltb_types.py b/steam_backlog_enforcer/_hltb_types.py index d569041..4dc5e14 100644 --- a/steam_backlog_enforcer/_hltb_types.py +++ b/steam_backlog_enforcer/_hltb_types.py @@ -45,6 +45,32 @@ class HLTBResult: hltb_game_id: int = 0 comp_100_count: int = 0 count_comp: int = 0 + rush_hours: float = -1 + leisure_100h: float = -1 + + +class _HLTBExtras: + """Mutable accumulator for HLTB data beyond the core hours cache. + + Passed through the fetch pipeline so callers stay within the 5-arg limit. + """ + + def __init__( + self, + count_comp: dict[int, int] | None = None, + rush: dict[int, float] | None = None, + leisure_100h: dict[int, float] | None = None, + hltb_game_id: dict[int, int] | None = None, + ) -> None: + """Initialize with optional pre-populated dicts.""" + self.count_comp: dict[int, int] = count_comp if count_comp is not None else {} + self.rush: dict[int, float] = rush if rush is not None else {} + self.leisure_100h: dict[int, float] = ( + leisure_100h if leisure_100h is not None else {} + ) + self.hltb_game_id: dict[int, int] = ( + hltb_game_id if hltb_game_id is not None else {} + ) @dataclass @@ -64,7 +90,10 @@ def _read_raw_cache() -> dict[int, dict[str, Any]]: "": { "hours": , "polls": , - "count_comp": + "count_comp": , + "rush_hours": , + "leisure_100h": , + "hltb_game_id": } } @@ -88,10 +117,20 @@ def _read_raw_cache() -> dict[int, dict[str, Any]]: "hours": float(v.get("hours", -1)), "polls": int(v.get("polls", 0)), "count_comp": int(v.get("count_comp", 0)), + "rush_hours": float(v.get("rush_hours", -1)), + "leisure_100h": float(v.get("leisure_100h", -1)), + "hltb_game_id": int(v.get("hltb_game_id", 0)), } else: try: - out[aid] = {"hours": float(v), "polls": 0, "count_comp": 0} + out[aid] = { + "hours": float(v), + "polls": 0, + "count_comp": 0, + "rush_hours": -1, + "leisure_100h": -1, + "hltb_game_id": 0, + } except (TypeError, ValueError): continue return out @@ -121,19 +160,70 @@ def load_hltb_count_comp_cache() -> dict[int, int]: return {aid: v["count_comp"] for aid, v in _read_raw_cache().items()} +def load_hltb_rush_cache() -> dict[int, float]: + """Load the rush-hours (avg comp_100 + DLC) portion of the HLTB cache. + + Returns: dict mapping app_id -> rush_hours (-1 = not yet computed). + """ + return {aid: v["rush_hours"] for aid, v in _read_raw_cache().items()} + + +def load_hltb_leisure_100h_cache() -> dict[int, float]: + """Load the leisure-100h (comp_100_h + DLC) portion of the HLTB cache. + + Returns: dict mapping app_id -> leisure_100h (-1 = not yet computed). + """ + return {aid: v["leisure_100h"] for aid, v in _read_raw_cache().items()} + + +def load_hltb_game_id_cache() -> dict[int, int]: + """Load the HLTB game ID portion of the cache. + + Returns: dict mapping app_id -> hltb_game_id (0 = not yet looked up). + """ + return {aid: v["hltb_game_id"] for aid, v in _read_raw_cache().items()} + + def save_hltb_cache( cache: dict[int, float], polls: dict[int, int] | None = None, - count_comp: dict[int, int] | None = None, + extras: _HLTBExtras | None = None, ) -> None: - """Save the HLTB cache to disk, including confidence metrics.""" + """Save the HLTB cache to disk, including confidence and stats metrics.""" polls = polls or {} - count_comp = count_comp or {} + if extras is None: + extras = _HLTBExtras() + # Preserve existing per-game data when the caller didn't populate the maps. + # A partial save (e.g. confidence-only) must not clobber rush/leisure/game-id + # data that a prior detail fetch already wrote. + needs_existing = ( + not extras.hltb_game_id or not extras.rush or not extras.leisure_100h + ) + if needs_existing: + existing = _read_raw_cache() + game_id_map: dict[int, int] = extras.hltb_game_id or { + aid: v["hltb_game_id"] for aid, v in existing.items() + } + rush_map: dict[int, float] = extras.rush or { + aid: v["rush_hours"] for aid, v in existing.items() if v["rush_hours"] > 0 + } + leisure_map: dict[int, float] = extras.leisure_100h or { + aid: v["leisure_100h"] + for aid, v in existing.items() + if v["leisure_100h"] > 0 + } + else: + game_id_map = extras.hltb_game_id + rush_map = extras.rush + leisure_map = extras.leisure_100h out = { str(aid): { "hours": hours, "polls": polls.get(aid, 0), - "count_comp": count_comp.get(aid, 0), + "count_comp": extras.count_comp.get(aid, 0), + "rush_hours": rush_map.get(aid, -1), + "leisure_100h": leisure_map.get(aid, -1), + "hltb_game_id": game_id_map.get(aid, 0), } for aid, hours in cache.items() } diff --git a/steam_backlog_enforcer/_scanning_confidence.py b/steam_backlog_enforcer/_scanning_confidence.py index e46775e..225e8b7 100644 --- a/steam_backlog_enforcer/_scanning_confidence.py +++ b/steam_backlog_enforcer/_scanning_confidence.py @@ -6,6 +6,7 @@ import logging from typing import TYPE_CHECKING from python_pkg.steam_backlog_enforcer._hltb_types import ( + _HLTBExtras, load_hltb_cache, load_hltb_count_comp_cache, load_hltb_polls_cache, @@ -105,7 +106,7 @@ def _refresh_candidate_confidence_batch( cache.pop(aid, None) polls.pop(aid, None) count_comp.pop(aid, None) - save_hltb_cache(cache, polls, count_comp) + save_hltb_cache(cache, polls, _HLTBExtras(count_comp=count_comp)) fetch_hltb_confidence_cached(names) @@ -115,7 +116,9 @@ def _refresh_candidate_confidence_batch( for aid, old_hours in prior_hours.items(): if old_hours > 0 and refreshed_hours.get(aid, -1) <= 0: refreshed_hours[aid] = old_hours - save_hltb_cache(refreshed_hours, refreshed_polls, refreshed_count_comp) + save_hltb_cache( + refreshed_hours, refreshed_polls, _HLTBExtras(count_comp=refreshed_count_comp) + ) for game in refresh_slice: game.comp_100_count = refreshed_polls.get(game.app_id, 0) diff --git a/steam_backlog_enforcer/_stats.py b/steam_backlog_enforcer/_stats.py new file mode 100644 index 0000000..5e7a184 --- /dev/null +++ b/steam_backlog_enforcer/_stats.py @@ -0,0 +1,350 @@ +"""Backlog completion-time statistics for Steam Backlog Enforcer.""" + +from __future__ import annotations + +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 python_pkg.steam_backlog_enforcer._hltb_types import ( + HLTB_BASE_URL, + load_hltb_cache, + load_hltb_game_id_cache, + load_hltb_leisure_100h_cache, + load_hltb_rush_cache, +) +from python_pkg.steam_backlog_enforcer._scanning_confidence import ( + _apply_cached_confidence_to_candidates, + _confidence_fail_reasons, + _refresh_candidate_confidence_batch, +) +from python_pkg.steam_backlog_enforcer.config import load_snapshot +from python_pkg.steam_backlog_enforcer.game_install import _echo +from python_pkg.steam_backlog_enforcer.hltb import fetch_hltb_detail_missing +from python_pkg.steam_backlog_enforcer.protondb import ( + ProtonDBRating, + fetch_protondb_ratings, +) +from python_pkg.steam_backlog_enforcer.steam_api import GameInfo + +if TYPE_CHECKING: + from python_pkg.steam_backlog_enforcer.config import Config, State + +logger = logging.getLogger(__name__) + +_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 _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") + if example.hltb_game_id > 0: + _echo(f" HLTB: {HLTB_BASE_URL}/game/{example.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, + total_hours: float, + missing: int, + total_games: int, +) -> None: + """Print a single time-scenario block.""" + _echo(f"\n {label}") + if total_hours <= 0: + _echo(" No data available.") + return + + missing_note = ( + f" ({missing}/{total_games} games had no data, hours underestimated)" + if missing + else "" + ) + _echo(f" Total: {total_hours:,.1f} h{missing_note}") + for daily in _HOURS_PER_DAY_PRESETS: + estimate = _format_completion_date(total_hours, daily) + _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`` should be the count of 100%-complete games in the library + snapshot (``sum(1 for g in games if g.is_complete)``), not the enforcer's + own ``finished_app_ids`` list, which misses games completed outside the + enforcer flow. + """ + _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") + _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 cmd_stats(_config: Config, state: State) -> None: + """Display backlog completion-time statistics. + + Filters games by the same HLTB-confidence and Linux-compatibility rules + used when picking the next game. Auto-fetches missing rush/leisure detail + data before printing. Shows four scenarios: + + 1. At your current pace (games finished per day since enforcement started). + 2. Rush — avg comp_100 + DLC completion time per HLTB. + 3. Leisure — comp_100_h (slowest 100 %) + DLC leisure per HLTB. + 4. Worst — absolute maximum recorded time (any category) per HLTB. + """ + snapshot = load_snapshot() + if snapshot is None: + _echo("No snapshot found. Run 'scan' first.") + return + + games = [GameInfo.from_snapshot(d) for d in snapshot] + # Count all 100%-achievement games in library (more accurate than + # finished_app_ids, which only tracks enforcer-assigned completions). + games_done = sum(1 for g in games if g.is_complete) + + qualified, hltb_skip, linux_skip, no_data_skip = _filter_qualifying_games( + games, state + ) + if _ensure_rush_data(qualified): + # Re-filter picks up updated rush/leisure caches; ProtonDB is now cached. + qualified, hltb_skip, linux_skip, no_data_skip = _filter_qualifying_games( + games, state + ) + total_q = len(qualified) + + _echo(f"\n{'═' * 70}") + _echo(" BACKLOG COMPLETION ESTIMATES") + _echo(f"{'═' * 70}") + _echo(f"\n Qualifying games: {total_q}") + if hltb_skip: + _echo(f" HLTB-skipped: {hltb_skip} (confidence too low)") + if linux_skip: + _echo(f" Linux-skipped: {linux_skip} (poor ProtonDB rating)") + if no_data_skip: + _echo(f" No-data-skipped: {no_data_skip} (no HLTB hours at all)") + + missing_rush_final = sum(1 for e in qualified if e.rush_hours <= 0) + if missing_rush_final: + _echo( + f"\n Note: {missing_rush_final}/{total_q} games still missing" + " rush/leisure data (HLTB search may not have matched them)." + ) + elif total_q: + _echo( + f"\n Detail data: rush + leisure available for all {total_q}" + " qualifying games." + ) + + if state.current_app_id: + _echo( + f"\n Current game: {state.current_game_name} (excluded from totals)" + ) + _echo(f" Finished games: {games_done} (excluded from totals)") + + _echo(f"\n{_LINE}") + _print_pace_scenario(state, total_q, games_done) + + worst_total, worst_missing = _sum_hours(qualified, "worst_hours") + rush_total, rush_missing = _sum_hours(qualified, "rush_hours") + leisure_total, leisure_missing = _sum_hours(qualified, "leisure_100h") + + _echo(f"\n{_LINE}") + _print_scenario( + "2. RUSH (avg comp_100 + DLC — typical fast completionist)", + rush_total, + rush_missing, + total_q, + ) + + _echo(f"\n{_LINE}") + _print_scenario( + "3. LEISURE (comp_100_h + DLC — slow/comfortable 100 %)", + leisure_total, + leisure_missing, + total_q, + ) + + _echo(f"\n{_LINE}") + _print_scenario( + "4. WORST CASE (max recorded time, any category, + DLC)", + worst_total, + worst_missing, + total_q, + ) + _print_worst_example(qualified) + + _echo(f"\n{_LINE}\n") diff --git a/steam_backlog_enforcer/config.py b/steam_backlog_enforcer/config.py index 79c84fb..4425180 100644 --- a/steam_backlog_enforcer/config.py +++ b/steam_backlog_enforcer/config.py @@ -87,6 +87,8 @@ class State: current_game_name: str = "" finished_app_ids: list[int] = field(default_factory=list) skipped_until: dict[str, str] = field(default_factory=dict) + enforcement_started_at: str = "" + """ISO-8601 UTC timestamp set on the first game assignment.""" """Map of ``str(app_id)`` → ISO-8601 UTC timestamp when the skip expires. Games in this map are excluded from auto-assignment until the timestamp diff --git a/steam_backlog_enforcer/hltb.py b/steam_backlog_enforcer/hltb.py index 24ed965..16f8a84 100644 --- a/steam_backlog_enforcer/hltb.py +++ b/steam_backlog_enforcer/hltb.py @@ -30,9 +30,13 @@ from python_pkg.steam_backlog_enforcer._hltb_types import ( 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, ) @@ -125,7 +129,7 @@ def fetch_hltb_times( cache: dict[int, float] | None = None, polls: dict[int, int] | None = None, progress_cb: ProgressCb | None = None, - count_comp: dict[int, int] | None = None, + extras: _HLTBExtras | None = None, ) -> list[HLTBResult]: """Synchronous wrapper: fetch HLTB times for games.""" if not games: @@ -134,10 +138,14 @@ def fetch_hltb_times( cache = {} if polls is None: polls = {} - if count_comp is None: - count_comp = {} return asyncio.run( - _fetch_batch(games, cache, polls, progress_cb, count_comp=count_comp) + _fetch_batch( + games, + cache, + polls, + progress_cb, + extras=extras, + ) ) @@ -182,7 +190,11 @@ def fetch_hltb_times_cached( """ cache = load_hltb_cache() polls = load_hltb_polls_cache() - count_comp = load_hltb_count_comp_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: @@ -197,12 +209,12 @@ def fetch_hltb_times_cached( cache=cache, polls=polls, progress_cb=progress_cb, - count_comp=count_comp, + extras=extras, ) elapsed = time.monotonic() - t0 # Final save. - save_hltb_cache(cache, polls, count_comp) + 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 @@ -245,7 +257,7 @@ def fetch_hltb_confidence_cached( ) elapsed = time.monotonic() - t0 - save_hltb_cache(cache, polls, count_comp) + 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 @@ -262,6 +274,75 @@ def fetch_hltb_confidence_cached( 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. + + Games already in the rush cache are skipped. 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() + missing = [(app_id, name) for app_id, name in games if rush.get(app_id, -1) <= 0] + 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=load_hltb_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) + + logger.info( + "Fetching HLTB detail for %d games missing rush/leisure data...", + len(missing), + ) + 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 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), + 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. diff --git a/steam_backlog_enforcer/main.py b/steam_backlog_enforcer/main.py index b926549..fa9ec87 100644 --- a/steam_backlog_enforcer/main.py +++ b/steam_backlog_enforcer/main.py @@ -13,6 +13,7 @@ from python_pkg.steam_backlog_enforcer._enforce_loop import ( get_all_owned_app_ids, ) from python_pkg.steam_backlog_enforcer._hltb_types import load_hltb_cache +from python_pkg.steam_backlog_enforcer._stats import cmd_stats from python_pkg.steam_backlog_enforcer._whitelist import ( WHITELIST_COOLDOWN_SECONDS, add_pending_exception, @@ -397,6 +398,7 @@ COMMANDS: dict[str, tuple[str, Callable[[Config, State], object]]] = { "setup": ("Run first-time setup", cmd_setup), "done": ("Finish game, open HLTB, pick next", cmd_done), "pick": ("Manually pick your next game from candidates", cmd_pick), + "stats": ("Show backlog completion-time estimates", cmd_stats), } # Extra commands with non-standard arg handling (shown in help but not in COMMANDS). diff --git a/steam_backlog_enforcer/protondb.py b/steam_backlog_enforcer/protondb.py index cdb29a7..fc52100 100644 --- a/steam_backlog_enforcer/protondb.py +++ b/steam_backlog_enforcer/protondb.py @@ -124,8 +124,12 @@ async def _fetch_one( session: aiohttp.ClientSession, sem: asyncio.Semaphore, app_id: int, -) -> ProtonDBRating: - """Fetch a single game's ProtonDB rating.""" +) -> ProtonDBRating | None: + """Fetch a single game's ProtonDB rating. + + Returns None on network/server errors (not cached, will retry next run). + Returns ProtonDBRating with empty tier on HTTP 404 (no ProtonDB data). + """ url = _PROTONDB_API.format(app_id=app_id) async with sem: try: @@ -142,9 +146,9 @@ async def _fetch_one( confidence=data.get("confidence", ""), total_reports=data.get("total", 0), ) - except (aiohttp.ClientError, asyncio.TimeoutError, OSError): - logger.warning("ProtonDB fetch failed for AppID=%d", app_id) - return ProtonDBRating(app_id=app_id) + except (aiohttp.ClientError, asyncio.TimeoutError, OSError) as e: + logger.warning("ProtonDB fetch failed for AppID=%d: %s", app_id, e) + return None # Don't cache transient failures — retry next run. async def _fetch_batch(app_ids: list[int]) -> list[ProtonDBRating]: @@ -152,7 +156,8 @@ async def _fetch_batch(app_ids: list[int]) -> list[ProtonDBRating]: sem = asyncio.Semaphore(MAX_CONCURRENT) async with aiohttp.ClientSession() as session: tasks = [_fetch_one(session, sem, aid) for aid in app_ids] - return await asyncio.gather(*tasks) + results = await asyncio.gather(*tasks) + return [r for r in results if r is not None] def _rating_to_dict(r: ProtonDBRating) -> dict[str, Any]: diff --git a/steam_backlog_enforcer/scanning.py b/steam_backlog_enforcer/scanning.py index ce18006..0d4c995 100644 --- a/steam_backlog_enforcer/scanning.py +++ b/steam_backlog_enforcer/scanning.py @@ -2,6 +2,7 @@ from __future__ import annotations +from datetime import datetime, timezone import logging import time from typing import TYPE_CHECKING, Any @@ -229,6 +230,8 @@ def _assign_chosen_game( """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)" diff --git a/steam_backlog_enforcer/tests/test_hltb_detail.py b/steam_backlog_enforcer/tests/test_hltb_detail.py index c36af86..8f3b22d 100644 --- a/steam_backlog_enforcer/tests/test_hltb_detail.py +++ b/steam_backlog_enforcer/tests/test_hltb_detail.py @@ -15,12 +15,18 @@ from python_pkg.steam_backlog_enforcer._hltb_detail import ( _as_positive_int, _collect_dlc_relationships, _extract_base_leisure_hours, + _extract_comp_100_avg_and_high, _extract_dlc_relationships, _fetch_detail_one, _fetch_dlc_leisure_hours, _fetch_leisure_times, + _process_game_detail, +) +from python_pkg.steam_backlog_enforcer._hltb_types import ( + _SAVE_INTERVAL, + HLTBResult, + _HLTBExtras, ) -from python_pkg.steam_backlog_enforcer._hltb_types import _SAVE_INTERVAL, HLTBResult class TestInternalHelpers: @@ -183,6 +189,76 @@ class TestInternalHelpers: assert asyncio.run(_run()) == {} +class TestExtractComp100AvgAndHigh: + """Tests for _extract_comp_100_avg_and_high.""" + + def test_returns_minus_one_for_empty_game_list(self) -> None: + assert _extract_comp_100_avg_and_high({"game": []}) == (-1, -1) + + def test_returns_minus_one_for_non_list_game(self) -> None: + assert _extract_comp_100_avg_and_high({"game": "bad"}) == (-1, -1) + + def test_returns_minus_one_when_game0_not_dict(self) -> None: + assert _extract_comp_100_avg_and_high({"game": [42]}) == (-1, -1) + + def test_returns_avg_and_high(self) -> None: + data: dict[str, Any] = {"game": [{"comp_100": 7200, "comp_100_h": 10800}]} + avg_h, high_h = _extract_comp_100_avg_and_high(data) + assert avg_h == round(7200 / 3600, 2) + assert high_h == round(10800 / 3600, 2) + + def test_high_falls_back_to_avg_when_zero(self) -> None: + data: dict[str, Any] = {"game": [{"comp_100": 7200, "comp_100_h": 0}]} + avg_h, high_h = _extract_comp_100_avg_and_high(data) + assert avg_h == round(7200 / 3600, 2) + assert high_h == avg_h + + def test_avg_zero_returns_minus_one_avg(self) -> None: + data: dict[str, Any] = {"game": [{"comp_100": 0, "comp_100_h": 0}]} + avg_h, high_h = _extract_comp_100_avg_and_high(data) + assert avg_h == -1 + assert high_h == -1 + + +class TestProcessGameDetail: + """Tests for _process_game_detail.""" + + def test_returns_leisure_rush_and_l100(self) -> None: + data: dict[str, Any] = { + "game": [{"comp_100_h": 10800, "comp_100": 7200}], + "relationships": [], + } + leisure, rush_h, l100 = _process_game_detail(data, [], {}) + assert leisure == round(10800 / 3600, 2) + assert rush_h == round(7200 / 3600, 2) + assert l100 == round(10800 / 3600, 2) + + def test_negative_leisure_when_no_data(self) -> None: + leisure, rush_h, l100 = _process_game_detail({"game": []}, [], {}) + assert leisure == -1 + assert rush_h == -1.0 + assert l100 == -1.0 + + def test_rush_includes_dlc_fallback(self) -> None: + data: dict[str, Any] = { + "game": [{"comp_100": 7200, "comp_100_h": 0}], + "relationships": [], + } + dlc_rels = [(99, 1.5)] + _leisure, rush_h, _l100 = _process_game_detail(data, dlc_rels, {}) + assert rush_h == round(7200 / 3600 + 1.5, 2) + + def test_l100_uses_dlc_override(self) -> None: + data: dict[str, Any] = { + "game": [{"comp_100_h": 10800, "comp_100": 7200}], + "relationships": [], + } + dlc_rels = [(77, 2.0)] + dlc_hours_by_id = {77: 3.0} + _leisure, _rush_h, l100 = _process_game_detail(data, dlc_rels, dlc_hours_by_id) + assert l100 == round(10800 / 3600 + (3.0 - 2.0), 2) + + class _FakeTextResponse: """Async context manager mimicking aiohttp response for text.""" @@ -441,8 +517,34 @@ class TestFetchLeisureTimes: assert cache[1289310] == expected assert results[0].completionist_hours == expected - def test_with_explicit_count_comp(self) -> None: - """Pass a non-None count_comp to cover the False branch of the None check.""" + def test_extras_populated_with_rush_and_l100(self) -> None: + """rush_h and l100 are stored in extras when game has comp_100 data.""" + results = [ + HLTBResult( + app_id=440, + game_name="TF2", + completionist_hours=50.0, + similarity=1.0, + hltb_game_id=12345, + ), + ] + game_data: dict[str, Any] = { + "game": [{"comp_100_h": 10800, "comp_100": 7200}], + "relationships": [], + } + cache: dict[int, float] = {} + extras = _HLTBExtras(count_comp={440: 5}) + with patch( + "python_pkg.steam_backlog_enforcer._hltb_detail._fetch_detail_one", + new_callable=AsyncMock, + return_value=game_data, + ): + asyncio.run(_fetch_leisure_times(results, cache, {}, None, extras=extras)) + assert extras.rush[440] == round(7200 / 3600, 2) + assert extras.leisure_100h[440] == round(10800 / 3600, 2) + + def test_with_explicit_extras(self) -> None: + """Pass a pre-populated _HLTBExtras to cover the non-None extras branch.""" results = [ HLTBResult( app_id=440, @@ -457,12 +559,11 @@ class TestFetchLeisureTimes: "relationships": [], } cache: dict[int, float] = {} + extras = _HLTBExtras(count_comp={440: 5}) with patch( "python_pkg.steam_backlog_enforcer._hltb_detail._fetch_detail_one", new_callable=AsyncMock, return_value=game_data, ): - asyncio.run( - _fetch_leisure_times(results, cache, {}, None, count_comp={440: 5}) - ) + asyncio.run(_fetch_leisure_times(results, cache, {}, None, extras=extras)) assert cache[440] == 1.0 diff --git a/steam_backlog_enforcer/tests/test_hltb_part2.py b/steam_backlog_enforcer/tests/test_hltb_part2.py index a3df9e8..683ef25 100644 --- a/steam_backlog_enforcer/tests/test_hltb_part2.py +++ b/steam_backlog_enforcer/tests/test_hltb_part2.py @@ -3,6 +3,7 @@ from __future__ import annotations import asyncio +from typing import TYPE_CHECKING from unittest.mock import MagicMock, patch from typing_extensions import Self @@ -14,10 +15,14 @@ from python_pkg.steam_backlog_enforcer.hltb import ( _fetch_batch_confidence_only, fetch_hltb_confidence, fetch_hltb_confidence_cached, + fetch_hltb_detail_missing, fetch_hltb_times_cached, get_hltb_submit_url, ) +if TYPE_CHECKING: + from python_pkg.steam_backlog_enforcer._hltb_types import _HLTBExtras + PKG = "python_pkg.steam_backlog_enforcer.hltb" @@ -44,14 +49,14 @@ class TestFetchHltbTimesCached: cache: dict[int, float] | None = None, polls: dict[int, int] | None = None, progress_cb: object = None, - count_comp: dict[int, int] | None = None, + extras: _HLTBExtras | None = None, ) -> list[object]: if cache is not None: cache[730] = 20.0 if polls is not None: polls[730] = 0 - if count_comp is not None: - count_comp[730] = 0 + if extras is not None: + extras.count_comp[730] = 0 return [] mock_fetch.side_effect = add_to_cache @@ -102,7 +107,7 @@ class TestFetchHltbTimesCached: cache: dict[int, float] | None = None, polls: dict[int, int] | None = None, progress_cb: object = None, - count_comp: dict[int, int] | None = None, + extras: _HLTBExtras | None = None, ) -> list[object]: if cache is not None: cache[440] = 50.0 @@ -110,9 +115,9 @@ class TestFetchHltbTimesCached: if polls is not None: polls[440] = 5 polls[730] = 0 - if count_comp is not None: - count_comp[440] = 15 - count_comp[730] = 0 + if extras is not None: + extras.count_comp[440] = 15 + extras.count_comp[730] = 0 return [] mock_fetch.side_effect = add_found @@ -233,3 +238,129 @@ class TestConfidenceHelpers: assert result == {1: 12.0} mock_fetch.assert_not_called() mock_save.assert_not_called() + + +class TestFetchHltbDetailMissing: + """Tests for fetch_hltb_detail_missing.""" + + def test_no_missing_returns_zero(self) -> None: + """All games in rush cache → early return without fetching.""" + with ( + patch(f"{PKG}.load_hltb_rush_cache", return_value={440: 15.0}), + patch(f"{PKG}.fetch_hltb_times") as mock_fetch, + ): + result = fetch_hltb_detail_missing([(440, "TF2")]) + assert result == 0 + mock_fetch.assert_not_called() + + def test_fetches_missing_and_returns_count(self) -> None: + """Games not in rush cache are fetched; returns count with rush data.""" + + def add_rush( + _games: object, + cache: dict[int, float] | None = None, + polls: dict[int, int] | None = None, + progress_cb: object = None, + extras: _HLTBExtras | None = None, + ) -> list[object]: + if extras is not None: + extras.rush[730] = 10.0 + if cache is not None: + cache[730] = 25.0 + return [] + + with ( + patch(f"{PKG}.load_hltb_rush_cache", return_value={440: 15.0}), + patch(f"{PKG}.load_hltb_cache", return_value={730: 20.0}), + patch(f"{PKG}.load_hltb_polls_cache", return_value={}), + patch(f"{PKG}.load_hltb_count_comp_cache", return_value={}), + patch(f"{PKG}.load_hltb_leisure_100h_cache", return_value={}), + patch(f"{PKG}.load_hltb_game_id_cache", return_value={}), + patch(f"{PKG}.fetch_hltb_times", side_effect=add_rush), + patch(f"{PKG}.save_hltb_cache") as mock_save, + patch(f"{PKG}.time.monotonic", side_effect=[0.0, 2.0]), + ): + result = fetch_hltb_detail_missing([(440, "TF2"), (730, "CS")]) + assert result == 1 + mock_save.assert_called_once() + + def test_restores_prior_hours_when_not_refound(self) -> None: + """Hours are restored when re-fetch finds nothing for the game.""" + saved: dict[int, float] = {} + + def capture_save( + cache: dict[int, float], + _polls: object, + _extras: object = None, + ) -> None: + saved.update(cache) + + with ( + patch(f"{PKG}.load_hltb_rush_cache", return_value={}), + patch(f"{PKG}.load_hltb_cache", return_value={730: 20.0}), + patch(f"{PKG}.load_hltb_polls_cache", return_value={}), + patch(f"{PKG}.load_hltb_count_comp_cache", return_value={}), + patch(f"{PKG}.load_hltb_leisure_100h_cache", return_value={}), + patch(f"{PKG}.load_hltb_game_id_cache", return_value={}), + patch(f"{PKG}.fetch_hltb_times"), # no-op, cache stays empty + patch(f"{PKG}.save_hltb_cache", side_effect=capture_save), + patch(f"{PKG}.time.monotonic", side_effect=[0.0, 1.0]), + ): + fetch_hltb_detail_missing([(730, "CS")]) + assert saved[730] == 20.0 + + def test_does_not_restore_when_refound(self) -> None: + """Prior hours are NOT restored when re-fetch successfully finds game.""" + + def add_hours_and_rush( + _games: object, + cache: dict[int, float] | None = None, + polls: dict[int, int] | None = None, + progress_cb: object = None, + extras: _HLTBExtras | None = None, + ) -> list[object]: + if cache is not None: + cache[730] = 30.0 + if extras is not None: + extras.rush[730] = 12.0 + return [] + + saved: dict[int, float] = {} + + def capture_save( + cache: dict[int, float], + _polls: object, + _extras: object = None, + ) -> None: + saved.update(cache) + + with ( + patch(f"{PKG}.load_hltb_rush_cache", return_value={}), + patch(f"{PKG}.load_hltb_cache", return_value={730: 20.0}), + patch(f"{PKG}.load_hltb_polls_cache", return_value={}), + patch(f"{PKG}.load_hltb_count_comp_cache", return_value={}), + patch(f"{PKG}.load_hltb_leisure_100h_cache", return_value={}), + patch(f"{PKG}.load_hltb_game_id_cache", return_value={}), + patch(f"{PKG}.fetch_hltb_times", side_effect=add_hours_and_rush), + patch(f"{PKG}.save_hltb_cache", side_effect=capture_save), + patch(f"{PKG}.time.monotonic", side_effect=[0.0, 1.0]), + ): + result = fetch_hltb_detail_missing([(730, "CS")]) + assert result == 1 + assert saved[730] == 30.0 + + def test_zero_elapsed_rate(self) -> None: + """Covers the elapsed == 0 branch in the rate calculation.""" + with ( + patch(f"{PKG}.load_hltb_rush_cache", return_value={}), + patch(f"{PKG}.load_hltb_cache", return_value={}), + patch(f"{PKG}.load_hltb_polls_cache", return_value={}), + patch(f"{PKG}.load_hltb_count_comp_cache", return_value={}), + patch(f"{PKG}.load_hltb_leisure_100h_cache", return_value={}), + patch(f"{PKG}.load_hltb_game_id_cache", return_value={}), + patch(f"{PKG}.fetch_hltb_times"), + patch(f"{PKG}.save_hltb_cache"), + patch(f"{PKG}.time.monotonic", side_effect=[5.0, 5.0]), + ): + result = fetch_hltb_detail_missing([(730, "CS")]) + assert result == 0 diff --git a/steam_backlog_enforcer/tests/test_hltb_search.py b/steam_backlog_enforcer/tests/test_hltb_search.py index a3777ab..15c9872 100644 --- a/steam_backlog_enforcer/tests/test_hltb_search.py +++ b/steam_backlog_enforcer/tests/test_hltb_search.py @@ -84,6 +84,26 @@ class TestSearchOne: assert result is not None assert result.app_id == 440 + def test_found_without_game_id(self) -> None: + """Found result with hltb_game_id=0 does not populate ctx.hltb_game_id.""" + resp = _FakeResponse( + 200, + { + "data": [ + { + "game_name": "TF2", + "game_alias": "", + "comp_100": 180000, + "game_id": 0, + } + ], + }, + ) + ctx = _make_ctx(_make_session(resp)) + result = asyncio.run(_search_one(asyncio.Semaphore(1), ctx, 440, "TF2")) + assert result is not None + assert 440 not in ctx.hltb_game_id + def test_not_found(self) -> None: resp = _FakeResponse(200, {"data": []}) ctx = _make_ctx(_make_session(resp)) diff --git a/steam_backlog_enforcer/tests/test_hltb_search_part2.py b/steam_backlog_enforcer/tests/test_hltb_search_part2.py index 3891a37..15e0d0e 100644 --- a/steam_backlog_enforcer/tests/test_hltb_search_part2.py +++ b/steam_backlog_enforcer/tests/test_hltb_search_part2.py @@ -14,6 +14,7 @@ from python_pkg.steam_backlog_enforcer._hltb_detail import ( _parse_game_page, ) from python_pkg.steam_backlog_enforcer._hltb_search import ( + _build_search_variants, _fetch_batch, _pick_best_hltb_entry, ) @@ -305,3 +306,20 @@ class TestExtractLeisureHours: "relationships": "not-a-list", } assert _extract_leisure_hours(data) == 1.0 + + +class TestBuildSearchVariants: + """Tests for _build_search_variants.""" + + def test_subtitle_with_edition_strips_edition_from_subtitle_part(self) -> None: + # "Rocksmith 2014 Edition - Remastered" → no_subtitle = "Rocksmith 2014 Edition" + # (which != base), so lines 201-202 also add "Rocksmith" and "Rocksmith 2014" + variants = _build_search_variants("Rocksmith 2014 Edition - Remastered") + assert "Rocksmith 2014 Edition" in variants + assert "Rocksmith 2014" in variants + assert "Rocksmith" in variants + + def test_no_subtitle_skips_edition_strip(self) -> None: + # No " - " → no_subtitle == base → lines 201-202 are not executed + variants = _build_search_variants("Portal 2") + assert "Portal 2" in variants diff --git a/steam_backlog_enforcer/tests/test_polls_tracking.py b/steam_backlog_enforcer/tests/test_polls_tracking.py index 9c2c40f..3014b51 100644 --- a/steam_backlog_enforcer/tests/test_polls_tracking.py +++ b/steam_backlog_enforcer/tests/test_polls_tracking.py @@ -9,8 +9,10 @@ from unittest.mock import patch from python_pkg.steam_backlog_enforcer import _cmd_done from python_pkg.steam_backlog_enforcer._hltb_types import ( HLTBResult, + _HLTBExtras, load_hltb_cache, load_hltb_count_comp_cache, + load_hltb_game_id_cache, load_hltb_polls_cache, save_hltb_cache, ) @@ -67,9 +69,18 @@ class TestCacheSchema: patch(f"{_TYPES}.HLTB_CACHE_FILE", cache_file), patch(f"{_TYPES}.CONFIG_DIR", tmp_path), ): - save_hltb_cache({440: 10.5}, {440: 7}, {440: 20}) + save_hltb_cache({440: 10.5}, {440: 7}, _HLTBExtras(count_comp={440: 20})) data = json.loads(cache_file.read_text(encoding="utf-8")) - assert data == {"440": {"hours": 10.5, "polls": 7, "count_comp": 20}} + assert data == { + "440": { + "hours": 10.5, + "polls": 7, + "count_comp": 20, + "rush_hours": -1, + "leisure_100h": -1, + "hltb_game_id": 0, + } + } def test_save_without_polls_defaults_zero(self, tmp_path: Path) -> None: cache_file = tmp_path / "hltb_cache.json" @@ -79,7 +90,26 @@ class TestCacheSchema: ): save_hltb_cache({440: 10.5}) data = json.loads(cache_file.read_text(encoding="utf-8")) - assert data == {"440": {"hours": 10.5, "polls": 0, "count_comp": 0}} + assert data == { + "440": { + "hours": 10.5, + "polls": 0, + "count_comp": 0, + "rush_hours": -1, + "leisure_100h": -1, + "hltb_game_id": 0, + } + } + + def test_load_game_id_cache(self, tmp_path: Path) -> None: + """load_hltb_game_id_cache returns the hltb_game_id portion of the cache.""" + cache_file = tmp_path / "hltb_cache.json" + with ( + patch(f"{_TYPES}.HLTB_CACHE_FILE", cache_file), + patch(f"{_TYPES}.CONFIG_DIR", tmp_path), + ): + save_hltb_cache({440: 10.5}, extras=_HLTBExtras(hltb_game_id={440: 99})) + assert load_hltb_game_id_cache() == {440: 99} class TestHltbResultPolls: diff --git a/steam_backlog_enforcer/tests/test_protondb.py b/steam_backlog_enforcer/tests/test_protondb.py index 55f2069..7e46438 100644 --- a/steam_backlog_enforcer/tests/test_protondb.py +++ b/steam_backlog_enforcer/tests/test_protondb.py @@ -222,7 +222,7 @@ class TestFetchOne: sem = asyncio.Semaphore(1) result = asyncio.run(_fetch_one(mock_session, sem, 440)) - assert result.tier == "" + assert result is None class TestFetchBatch: @@ -239,6 +239,25 @@ class TestFetchBatch: assert len(result) == 1 assert result[0].tier == "gold" + def test_filters_none_results(self) -> None: + """Network failures (None) are filtered out of the batch result.""" + rating = ProtonDBRating(app_id=440, tier="gold") + + async def mock_fetch_one( + _session: aiohttp.ClientSession, + _sem: asyncio.Semaphore, + app_id: int, + ) -> ProtonDBRating | None: + return rating if app_id == 440 else None + + with patch( + "python_pkg.steam_backlog_enforcer.protondb._fetch_one", + side_effect=mock_fetch_one, + ): + result = asyncio.run(_fetch_batch([440, 999])) + assert len(result) == 1 + assert result[0].app_id == 440 + class TestFetchProtondbRatings: """Tests for fetch_protondb_ratings.""" diff --git a/steam_backlog_enforcer/tests/test_scanning_part3.py b/steam_backlog_enforcer/tests/test_scanning_part3.py index 3041dce..c07dc89 100644 --- a/steam_backlog_enforcer/tests/test_scanning_part3.py +++ b/steam_backlog_enforcer/tests/test_scanning_part3.py @@ -401,3 +401,14 @@ class TestPickNextGameSequential: pick_next_game([g1], state, config, on_select=lambda _g: True) assert state.current_app_id is None assert any("No assignable games" in line for line in echoed) + + def test_enforcement_started_at_not_overwritten_when_set(self) -> None: + """enforcement_started_at is preserved when already populated.""" + g1 = _game(app_id=1, name="G1", hours=5.0) + config = Config(steam_api_key="k", steam_id="i") + existing_ts = "2024-01-01T00:00:00+00:00" + state = State(enforcement_started_at=existing_ts) + with self._common_patches([]): + pick_next_game([g1], state, config, on_select=lambda _g: True) + assert state.current_app_id == 1 + assert state.enforcement_started_at == existing_ts diff --git a/steam_backlog_enforcer/tests/test_stats.py b/steam_backlog_enforcer/tests/test_stats.py new file mode 100644 index 0000000..8a09158 --- /dev/null +++ b/steam_backlog_enforcer/tests/test_stats.py @@ -0,0 +1,701 @@ +"""Tests for _stats module — 100% branch coverage.""" + +from __future__ import annotations + +from datetime import datetime, timedelta, timezone +from unittest.mock import patch + +from python_pkg.steam_backlog_enforcer._stats import ( + _ensure_rush_data, + _filter_qualifying_games, + _format_completion_date, + _GameTimes, + _print_pace_scenario, + _print_scenario, + _print_worst_example, + _sum_hours, + cmd_stats, +) +from python_pkg.steam_backlog_enforcer.config import Config, State +from python_pkg.steam_backlog_enforcer.protondb import ProtonDBRating +from python_pkg.steam_backlog_enforcer.steam_api import GameInfo + +_PKG = "python_pkg.steam_backlog_enforcer._stats" + + +def _game( + app_id: int = 1, + name: str = "G", + hours: float = 10.0, + total: int = 10, + unlocked: int = 0, +) -> GameInfo: + return GameInfo( + app_id=app_id, + name=name, + total_achievements=total, + unlocked_achievements=unlocked, + playtime_minutes=60, + completionist_hours=hours, + comp_100_count=5, + count_comp=20, + ) + + +def _unplayable_rating(app_id: int) -> ProtonDBRating: + return ProtonDBRating(app_id=app_id, tier="borked") + + +class TestFilterQualifyingGames: + """Tests for _filter_qualifying_games.""" + + def _run( + self, + games: list[GameInfo], + state: State, + rush_cache: dict[int, float] | None = None, + leisure_cache: dict[int, float] | None = None, + game_id_cache: dict[int, int] | None = None, + ) -> tuple[list[_GameTimes], int, int, int]: + with ( + patch(f"{_PKG}.load_hltb_rush_cache", return_value=rush_cache or {}), + patch( + f"{_PKG}.load_hltb_leisure_100h_cache", + return_value=leisure_cache or {}, + ), + patch( + f"{_PKG}.load_hltb_game_id_cache", + return_value=game_id_cache or {}, + ), + patch(f"{_PKG}._apply_cached_confidence_to_candidates"), + patch(f"{_PKG}._refresh_candidate_confidence_batch"), + patch(f"{_PKG}._confidence_fail_reasons", return_value=[]), + patch(f"{_PKG}.fetch_protondb_ratings", return_value={}), + ): + return _filter_qualifying_games(games, state) + + def test_current_app_id_excluded(self) -> None: + state = State(current_app_id=1) + g1 = _game(app_id=1) + g2 = _game(app_id=2) + qualified, _, _, _ = self._run([g1, g2], state) + ids = [e.game.app_id for e in qualified] + assert 1 not in ids + assert 2 in ids + + def test_no_current_app_id_branch(self) -> None: + """current_app_id is None — the exclude.add branch is not taken.""" + state = State(current_app_id=None) + g = _game(app_id=3) + qualified, _, _, _ = self._run([g], state) + assert len(qualified) == 1 + + def test_finished_app_ids_excluded(self) -> None: + state = State() + state.finished_app_ids = [1] + g1 = _game(app_id=1) + g2 = _game(app_id=2) + qualified, _, _, _ = self._run([g1, g2], state) + assert all(e.game.app_id != 1 for e in qualified) + + def test_complete_games_excluded(self) -> None: + """Games where is_complete is True are excluded from candidates.""" + state = State() + complete = _game(app_id=1, total=5, unlocked=5) + incomplete = _game(app_id=2, total=5, unlocked=0) + qualified, _, _, _ = self._run([complete, incomplete], state) + assert len(qualified) == 1 + assert qualified[0].game.app_id == 2 + + def test_low_confidence_counts_hltb_skipped(self) -> None: + state = State() + g = _game(app_id=1) + with ( + patch(f"{_PKG}.load_hltb_rush_cache", return_value={}), + patch(f"{_PKG}.load_hltb_leisure_100h_cache", return_value={}), + patch(f"{_PKG}.load_hltb_game_id_cache", return_value={}), + patch(f"{_PKG}._apply_cached_confidence_to_candidates"), + patch(f"{_PKG}._refresh_candidate_confidence_batch"), + patch(f"{_PKG}._confidence_fail_reasons", return_value=["low"]), + patch(f"{_PKG}.fetch_protondb_ratings", return_value={}), + ): + qualified, hltb_skip, _, _ = _filter_qualifying_games([g], state) + assert hltb_skip == 1 + assert len(qualified) == 0 + + def test_no_candidates_skips_protondb_call(self) -> None: + """When confidence filters all out, fetch_protondb_ratings is not called.""" + state = State() + g = _game(app_id=1) + with ( + patch(f"{_PKG}.load_hltb_rush_cache", return_value={}), + patch(f"{_PKG}.load_hltb_leisure_100h_cache", return_value={}), + patch(f"{_PKG}.load_hltb_game_id_cache", return_value={}), + patch(f"{_PKG}._apply_cached_confidence_to_candidates"), + patch(f"{_PKG}._refresh_candidate_confidence_batch"), + patch(f"{_PKG}._confidence_fail_reasons", return_value=["low"]), + patch(f"{_PKG}.fetch_protondb_ratings") as mock_proton, + ): + _filter_qualifying_games([g], state) + mock_proton.assert_not_called() + + def test_unplayable_rating_counts_linux_skipped(self) -> None: + state = State() + g = _game(app_id=1) + ratings = {1: _unplayable_rating(1)} + with ( + patch(f"{_PKG}.load_hltb_rush_cache", return_value={}), + patch(f"{_PKG}.load_hltb_leisure_100h_cache", return_value={}), + patch(f"{_PKG}.load_hltb_game_id_cache", return_value={}), + patch(f"{_PKG}._apply_cached_confidence_to_candidates"), + patch(f"{_PKG}._refresh_candidate_confidence_batch"), + patch(f"{_PKG}._confidence_fail_reasons", return_value=[]), + patch(f"{_PKG}.fetch_protondb_ratings", return_value=ratings), + ): + qualified, _, linux_skip, _ = _filter_qualifying_games([g], state) + assert linux_skip == 1 + assert len(qualified) == 0 + + def test_no_data_counts_no_data_skipped(self) -> None: + """Game with all -1 hours is counted as no_data_skipped.""" + state = State() + g = _game(app_id=1, hours=-1.0) + qualified, _, _, no_data_skip = self._run([g], state) + assert no_data_skip == 1 + assert len(qualified) == 0 + + def test_worst_hours_positive_when_completionist_hours_positive(self) -> None: + state = State() + g = _game(app_id=1, hours=25.0) + qualified, _, _, _ = self._run([g], state, rush_cache={1: 10.0}) + assert qualified[0].worst_hours == 25.0 + + def test_worst_hours_from_leisure_when_completionist_zero(self) -> None: + """worst_hours falls back to leisure_100h when completionist_hours is zero.""" + state = State() + g = _game(app_id=1, hours=0.0) + qualified, _, _, _ = self._run( + [g], state, rush_cache={1: 5.0}, leisure_cache={1: 6.0} + ) + assert qualified[0].worst_hours == 6.0 + + def test_worst_hours_is_max_when_leisure_exceeds_completionist(self) -> None: + """worst_hours is max(completionist, leisure_100h) when leisure is higher.""" + state = State() + g = _game(app_id=1, hours=25.0) + qualified, _, _, _ = self._run( + [g], state, rush_cache={1: 10.0}, leisure_cache={1: 40.0} + ) + assert qualified[0].worst_hours == 40.0 + + def test_worst_hours_negative_when_all_zero(self) -> None: + """worst_hours = -1 when both completionist_hours and leisure_100h are zero.""" + state = State() + g = _game(app_id=1, hours=0.0) + qualified, _, _, _ = self._run([g], state, rush_cache={1: 5.0}) + assert qualified[0].worst_hours == -1 + + def test_rush_and_leisure_from_cache(self) -> None: + state = State() + g = _game(app_id=1, hours=30.0) + qualified, _, _, _ = self._run( + [g], state, rush_cache={1: 12.0}, leisure_cache={1: 40.0} + ) + assert qualified[0].rush_hours == 12.0 + assert qualified[0].leisure_100h == 40.0 + + def test_missing_cache_entry_defaults_to_minus_one(self) -> None: + state = State() + g = _game(app_id=1, hours=20.0) + qualified, _, _, _ = self._run([g], state) + assert qualified[0].rush_hours == -1 + assert qualified[0].leisure_100h == -1 + + def test_only_rush_nonzero_qualifies(self) -> None: + """Game qualifies if only rush_hours is positive (worst <= 0, leisure <= 0).""" + state = State() + g = _game(app_id=1, hours=-1.0) + qualified, _, _, no_data_skip = self._run([g], state, rush_cache={1: 8.0}) + assert no_data_skip == 0 + assert len(qualified) == 1 + + def test_game_id_populated_from_cache(self) -> None: + """hltb_game_id is taken from game_id_cache.""" + state = State() + g = _game(app_id=1, hours=20.0) + qualified, _, _, _ = self._run([g], state, game_id_cache={1: 57514}) + assert qualified[0].hltb_game_id == 57514 + + def test_game_id_defaults_to_zero_when_not_in_cache(self) -> None: + """hltb_game_id defaults to 0 when not in cache.""" + state = State() + g = _game(app_id=1, hours=20.0) + qualified, _, _, _ = self._run([g], state) + assert qualified[0].hltb_game_id == 0 + + +class TestSumHours: + """Tests for _sum_hours.""" + + def _make_entry(self, worst: float, rush: float, leisure: float) -> _GameTimes: + return _GameTimes( + game=_game(), worst_hours=worst, rush_hours=rush, leisure_100h=leisure + ) + + def test_empty_list(self) -> None: + total, missing = _sum_hours([], "worst_hours") + assert total == 0.0 + assert missing == 0 + + def test_all_positive(self) -> None: + entries = [ + self._make_entry(10.0, 8.0, 12.0), + self._make_entry(20.0, 15.0, 25.0), + ] + total, missing = _sum_hours(entries, "worst_hours") + assert total == 30.0 + assert missing == 0 + + def test_some_negative(self) -> None: + entries = [ + self._make_entry(10.0, -1.0, 12.0), + self._make_entry(-1.0, 8.0, 25.0), + ] + total, missing = _sum_hours(entries, "worst_hours") + assert total == 10.0 + assert missing == 1 + + def test_all_negative(self) -> None: + entries = [self._make_entry(-1.0, -1.0, -1.0)] + total, missing = _sum_hours(entries, "rush_hours") + assert total == 0.0 + assert missing == 1 + + +class TestFormatCompletionDate: + """Tests for _format_completion_date.""" + + def test_zero_hours_returns_na(self) -> None: + assert _format_completion_date(0.0, 4.0) == "N/A" + + def test_negative_hours_returns_na(self) -> None: + assert _format_completion_date(-5.0, 4.0) == "N/A" + + def test_zero_daily_hours_returns_na(self) -> None: + assert _format_completion_date(100.0, 0.0) == "N/A" + + def test_negative_daily_hours_returns_na(self) -> None: + assert _format_completion_date(100.0, -1.0) == "N/A" + + def test_normal_returns_days_and_date(self) -> None: + result = _format_completion_date(40.0, 4.0) + # 40 / 4 = 10 days + assert result.startswith("10 days (") + assert ")" in result + + +class TestPrintScenario: + """Tests for _print_scenario.""" + + def test_no_data_prints_no_data_message(self) -> None: + echoed: list[str] = [] + with patch(f"{_PKG}._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: + echoed: list[str] = [] + with patch(f"{_PKG}._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) + + def test_with_data_and_missing(self) -> None: + echoed: list[str] = [] + with patch(f"{_PKG}._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) + + +class TestPrintPaceScenario: + """Tests for _print_pace_scenario.""" + + def test_no_start_date(self) -> None: + state = State() + echoed: list[str] = [] + with patch(f"{_PKG}._echo", side_effect=lambda *a, **_: echoed.append(a[0])): + _print_pace_scenario(state, 10, 0) + assert any("No start date recorded" in s for s in echoed) + + def test_invalid_start_date(self) -> None: + state = State(enforcement_started_at="not-a-date") + echoed: list[str] = [] + with patch(f"{_PKG}._echo", side_effect=lambda *a, **_: echoed.append(a[0])): + _print_pace_scenario(state, 10, 0) + assert any("Invalid enforcement_started_at" in s for s in echoed) + + def test_no_games_finished(self) -> None: + started = datetime.now(timezone.utc) - timedelta(days=30) + state = State(enforcement_started_at=started.isoformat()) + echoed: list[str] = [] + with patch(f"{_PKG}._echo", side_effect=lambda *a, **_: echoed.append(a[0])): + _print_pace_scenario(state, 10, 0) + assert any("No games finished yet" in s for s in echoed) + + def test_normal_pace(self) -> None: + started = datetime.now(timezone.utc) - timedelta(days=60) + state = State(enforcement_started_at=started.isoformat()) + echoed: list[str] = [] + with patch(f"{_PKG}._echo", side_effect=lambda *a, **_: echoed.append(a[0])): + _print_pace_scenario(state, 5, 3) + assert any("Pace:" in s for s in echoed) + assert any("Est. complete:" in s for s in echoed) + + +class TestCmdStats: + """Tests for cmd_stats.""" + + def _config(self) -> Config: + return Config(steam_api_key="k", steam_id="i") + + def test_no_snapshot(self) -> None: + echoed: list[str] = [] + state = State() + with ( + patch(f"{_PKG}.load_snapshot", return_value=None), + patch(f"{_PKG}._echo", side_effect=lambda *a, **_: echoed.append(a[0])), + ): + cmd_stats(self._config(), state) + assert any("No snapshot found" in s for s in echoed) + + def _snapshot_game(self, app_id: int = 1, hours: float = 20.0) -> dict[str, object]: + return { + "app_id": app_id, + "name": f"Game{app_id}", + "total_achievements": 10, + "unlocked_achievements": 0, + "playtime_minutes": 60, + "completionist_hours": hours, + "comp_100_count": 5, + "count_comp": 20, + } + + def _run_cmd_stats( + self, + state: State, + hltb_skip: int = 0, + linux_skip: int = 0, + no_data_skip: int = 0, + ) -> list[str]: + snapshot = [self._snapshot_game()] + game = GameInfo.from_snapshot(snapshot[0]) + entry = _GameTimes( + game=game, worst_hours=20.0, rush_hours=15.0, leisure_100h=25.0 + ) + echoed: list[str] = [] + with ( + patch(f"{_PKG}.load_snapshot", return_value=snapshot), + patch( + f"{_PKG}._filter_qualifying_games", + return_value=([entry], hltb_skip, linux_skip, no_data_skip), + ), + patch(f"{_PKG}._echo", side_effect=lambda *a, **_: echoed.append(a[0])), + patch(f"{_PKG}._print_pace_scenario"), + patch(f"{_PKG}._print_scenario"), + ): + cmd_stats(self._config(), state) + return echoed + + def test_with_no_current_game(self) -> None: + state = State() + echoed = self._run_cmd_stats(state) + assert any("Qualifying games" in s for s in echoed) + assert not any("Current game:" in s for s in echoed) + + def test_with_current_game(self) -> None: + state = State(current_app_id=42, current_game_name="Hollow Knight") + echoed = self._run_cmd_stats(state) + assert any("Current game:" in s and "Hollow Knight" in s for s in echoed) + + def test_hltb_skipped_shown(self) -> None: + state = State() + echoed = self._run_cmd_stats(state, hltb_skip=3) + assert any("HLTB-skipped" in s for s in echoed) + + def test_linux_skipped_shown(self) -> None: + state = State() + echoed = self._run_cmd_stats(state, linux_skip=2) + assert any("Linux-skipped" in s for s in echoed) + + def test_no_data_skipped_shown(self) -> None: + state = State() + echoed = self._run_cmd_stats(state, no_data_skip=1) + assert any("No-data-skipped" in s for s in echoed) + + def test_zero_skips_not_shown(self) -> None: + state = State() + echoed = self._run_cmd_stats(state) + assert not any("HLTB-skipped" in s for s in echoed) + assert not any("Linux-skipped" in s for s in echoed) + assert not any("No-data-skipped" in s for s in echoed) + + def test_finished_games_count_uses_snapshot_complete(self) -> None: + """'Finished games' count uses snapshot is_complete, not finished_app_ids.""" + state = State() + # finished_app_ids has 1 entry, but snapshot has 2 complete games — count = 2. + state.finished_app_ids = [99] + snapshot_complete = { + **self._snapshot_game(app_id=2), + "unlocked_achievements": 10, + } + snapshot = [self._snapshot_game(app_id=1), snapshot_complete] + game = GameInfo.from_snapshot(self._snapshot_game()) + entry = _GameTimes( + game=game, worst_hours=20.0, rush_hours=15.0, leisure_100h=25.0 + ) + echoed: list[str] = [] + with ( + patch(f"{_PKG}.load_snapshot", return_value=snapshot), + patch( + f"{_PKG}._filter_qualifying_games", + return_value=([entry], 0, 0, 0), + ), + patch(f"{_PKG}._echo", side_effect=lambda *a, **_: echoed.append(a[0])), + patch(f"{_PKG}._print_pace_scenario"), + patch(f"{_PKG}._print_scenario"), + ): + cmd_stats(self._config(), state) + assert any("Finished games" in s and "1" in s for s in echoed) + + def test_detail_data_complete_message_shown(self) -> None: + """'Detail data: ...' shown when all qualifying games have rush hours.""" + state = State() + echoed = self._run_cmd_stats(state) + # entry has rush_hours=15.0 > 0, so missing_rush_final == 0 and total_q == 1 + assert any("Detail data" in s for s in echoed) + + def test_note_missing_rush_shown_when_rush_absent(self) -> None: + """'Note: X games still missing...' shown when rush_hours <= 0 after fetch.""" + state = State() + snapshot = [self._snapshot_game()] + game = GameInfo.from_snapshot(snapshot[0]) + entry = _GameTimes( + game=game, worst_hours=20.0, rush_hours=-1.0, leisure_100h=-1.0 + ) + echoed: list[str] = [] + with ( + patch(f"{_PKG}.load_snapshot", return_value=snapshot), + patch( + f"{_PKG}._filter_qualifying_games", + return_value=([entry], 0, 0, 0), + ), + patch(f"{_PKG}._ensure_rush_data", return_value=False), + patch(f"{_PKG}._echo", side_effect=lambda *a, **_: echoed.append(a[0])), + patch(f"{_PKG}._print_pace_scenario"), + patch(f"{_PKG}._print_scenario"), + patch(f"{_PKG}._print_worst_example"), + ): + cmd_stats(self._config(), state) + assert any("still missing" in s for s in echoed) + + def test_no_detail_message_when_no_qualifying_games(self) -> None: + """Neither 'Note' nor 'Detail data' shown when qualified list is empty.""" + state = State() + snapshot = [self._snapshot_game()] + echoed: list[str] = [] + with ( + patch(f"{_PKG}.load_snapshot", return_value=snapshot), + patch( + f"{_PKG}._filter_qualifying_games", + return_value=([], 0, 0, 0), + ), + patch(f"{_PKG}._ensure_rush_data", return_value=False), + patch(f"{_PKG}._echo", side_effect=lambda *a, **_: echoed.append(a[0])), + patch(f"{_PKG}._print_pace_scenario"), + patch(f"{_PKG}._print_scenario"), + patch(f"{_PKG}._print_worst_example"), + ): + cmd_stats(self._config(), state) + assert not any("Detail data" in s for s in echoed) + assert not any("still missing" in s for s in echoed) + + def test_refilter_called_when_ensure_rush_data_returns_true(self) -> None: + """_filter_qualifying_games called twice when _ensure_rush_data returns True.""" + state = State() + snapshot = [self._snapshot_game()] + game = GameInfo.from_snapshot(snapshot[0]) + entry = _GameTimes( + game=game, worst_hours=20.0, rush_hours=15.0, leisure_100h=25.0 + ) + filter_calls: list[int] = [] + + def count_filter( + _games: object, _state: object + ) -> tuple[list[_GameTimes], int, int, int]: + filter_calls.append(1) + return [entry], 0, 0, 0 + + with ( + patch(f"{_PKG}.load_snapshot", return_value=snapshot), + patch(f"{_PKG}._filter_qualifying_games", side_effect=count_filter), + patch(f"{_PKG}._ensure_rush_data", return_value=True), + patch(f"{_PKG}._echo"), + patch(f"{_PKG}._print_pace_scenario"), + patch(f"{_PKG}._print_scenario"), + patch(f"{_PKG}._print_worst_example"), + ): + cmd_stats(self._config(), state) + assert len(filter_calls) == 2 + + def test_games_done_passed_to_pace_from_snapshot_complete(self) -> None: + """_print_pace_scenario receives is_complete count from snapshot.""" + state = State() + # Snapshot: 1 complete game (unlocked=total=10), 1 incomplete. + snapshot_complete = { + **self._snapshot_game(app_id=2), + "unlocked_achievements": 10, + } + snapshot = [self._snapshot_game(app_id=1), snapshot_complete] + game = GameInfo.from_snapshot(self._snapshot_game()) + entry = _GameTimes( + game=game, worst_hours=20.0, rush_hours=15.0, leisure_100h=25.0 + ) + captured: dict[str, int] = {} + + def capture_pace(_state: object, _remaining: object, games_done: int) -> None: + captured["games_done"] = games_done + + with ( + patch(f"{_PKG}.load_snapshot", return_value=snapshot), + patch( + f"{_PKG}._filter_qualifying_games", + return_value=([entry], 0, 0, 0), + ), + patch(f"{_PKG}._echo"), + patch(f"{_PKG}._print_pace_scenario", side_effect=capture_pace), + patch(f"{_PKG}._print_scenario"), + patch(f"{_PKG}._print_worst_example"), + ): + cmd_stats(self._config(), state) + assert captured["games_done"] == 1 + + +class TestEnsureRushData: + """Tests for _ensure_rush_data.""" + + def _entry(self, rush: float) -> _GameTimes: + return _GameTimes( + game=_game(), worst_hours=10.0, rush_hours=rush, leisure_100h=5.0 + ) + + def test_empty_qualified_returns_false(self) -> None: + with patch(f"{_PKG}.fetch_hltb_detail_missing") as mock_fetch: + result = _ensure_rush_data([]) + assert result is False + mock_fetch.assert_not_called() + + def test_all_have_rush_returns_false(self) -> None: + entries = [self._entry(10.0), self._entry(5.0)] + with patch(f"{_PKG}.fetch_hltb_detail_missing") as mock_fetch: + result = _ensure_rush_data(entries) + assert result is False + mock_fetch.assert_not_called() + + def test_missing_rush_fetches_and_returns_true(self) -> None: + entries = [self._entry(-1.0)] + with ( + patch(f"{_PKG}.fetch_hltb_detail_missing") as mock_fetch, + patch(f"{_PKG}._echo"), + ): + result = _ensure_rush_data(entries) + assert result is True + mock_fetch.assert_called_once() + + +class TestPrintWorstExample: + """Tests for _print_worst_example.""" + + def test_empty_list_does_nothing(self) -> None: + echoed: list[str] = [] + with patch(f"{_PKG}._echo", side_effect=lambda *a, **_: echoed.append(a[0])): + _print_worst_example([]) + assert echoed == [] + + def test_example_with_rush_and_leisure(self) -> None: + entry = _GameTimes( + game=_game(name="Portal"), + worst_hours=15.0, + rush_hours=5.0, + leisure_100h=20.0, + ) + echoed: list[str] = [] + with patch(f"{_PKG}._echo", side_effect=lambda *a, **_: echoed.append(a[0])): + _print_worst_example([entry]) + assert any("Portal" in s for s in echoed) + assert any("Rush" in s for s in echoed) + assert any("Leisure" in s for s in echoed) + + def test_example_without_rush(self) -> None: + entry = _GameTimes( + game=_game(name="X"), worst_hours=15.0, rush_hours=-1.0, leisure_100h=20.0 + ) + echoed: list[str] = [] + with patch(f"{_PKG}._echo", side_effect=lambda *a, **_: echoed.append(a[0])): + _print_worst_example([entry]) + assert not any("Rush" in s for s in echoed) + assert any("Leisure" in s for s in echoed) + + def test_example_without_leisure(self) -> None: + entry = _GameTimes( + game=_game(name="Y"), worst_hours=15.0, rush_hours=5.0, leisure_100h=-1.0 + ) + echoed: list[str] = [] + with patch(f"{_PKG}._echo", side_effect=lambda *a, **_: echoed.append(a[0])): + _print_worst_example([entry]) + assert any("Rush" in s for s in echoed) + assert not any("Leisure" in s for s in echoed) + + def test_hltb_search_url_shown_when_no_game_id(self) -> None: + """Falls back to search URL when hltb_game_id is 0.""" + entry = _GameTimes( + game=_game(name="Portal 2"), + worst_hours=15.0, + rush_hours=-1.0, + leisure_100h=-1.0, + ) + echoed: list[str] = [] + with patch(f"{_PKG}._echo", side_effect=lambda *a, **_: echoed.append(a[0])): + _print_worst_example([entry]) + assert any("howlongtobeat.com" in s and "Portal+2" in s for s in echoed) + + def test_hltb_direct_link_shown_when_game_id_known(self) -> None: + """Direct HLTB game link shown when hltb_game_id is populated.""" + entry = _GameTimes( + game=_game(name="Devil May Cry 5"), + worst_hours=186.0, + rush_hours=50.0, + leisure_100h=186.0, + hltb_game_id=57514, + ) + echoed: list[str] = [] + with patch(f"{_PKG}._echo", side_effect=lambda *a, **_: echoed.append(a[0])): + _print_worst_example([entry]) + assert any("howlongtobeat.com/game/57514" in s for s in echoed) + assert not any("?q=" in s for s in echoed) + + def test_entries_with_zero_worst_hours_excluded_from_examples(self) -> None: + """Games with worst_hours <= 0 are not selected as the example.""" + bad = _GameTimes( + game=_game(name="Skip"), worst_hours=0.0, rush_hours=-1.0, leisure_100h=-1.0 + ) + good = _GameTimes( + game=_game(name="Pick"), + worst_hours=10.0, + rush_hours=-1.0, + leisure_100h=-1.0, + ) + echoed: list[str] = [] + with patch(f"{_PKG}._echo", side_effect=lambda *a, **_: echoed.append(a[0])): + _print_worst_example([bad, good]) + assert any("Pick" in s for s in echoed) + assert not any("Skip" in s for s in echoed)