diff --git a/steam_backlog_enforcer/hltb.py b/steam_backlog_enforcer/hltb.py index 5e87522..a7a287d 100644 --- a/steam_backlog_enforcer/hltb.py +++ b/steam_backlog_enforcer/hltb.py @@ -1,11 +1,13 @@ """HowLongToBeat integration for estimating game completion times. -Fetches completionist hour estimates from howlongtobeat.com with: +Fetches leisure completionist hour estimates from howlongtobeat.com with: - direct API calls (bypassing the slow howlongtobeatpy per-request setup) - single shared aiohttp session for all requests - concurrent requests with configurable concurrency - live progress reporting via callback - incremental disk-cache saves so crashes don't lose work +- leisure time (upper-bound play time) from individual game pages +- DLC time aggregation (base game + all DLC leisure times combined) """ from __future__ import annotations @@ -17,6 +19,7 @@ from difflib import SequenceMatcher from http import HTTPStatus import json import logging +import re import time from typing import Any @@ -47,6 +50,15 @@ class HLTBResult: hltb_game_id: int = 0 +@dataclass +class _AuthInfo: + """HLTB API authentication details.""" + + token: str + hp_key: str = "" + hp_val: str = "" + + HLTB_BASE_URL = "https://howlongtobeat.com" @@ -107,11 +119,11 @@ def _get_hltb_search_url() -> str: return "https://howlongtobeat.com/api/finder" -async def _get_auth_token( +async def _get_auth_info( search_url: str, session: aiohttp.ClientSession, -) -> str | None: - """Fetch the HLTB auth token (one GET request).""" +) -> _AuthInfo | None: + """Fetch the HLTB auth token and honeypot key/val (one GET request).""" init_url = search_url + "/init" ts = int(time.time() * 1000) headers = { @@ -129,7 +141,13 @@ async def _get_auth_token( if resp.status == HTTPStatus.OK: data = await resp.json() token: str | None = data.get("token") - return token + if token is None: + return None + return _AuthInfo( + token=token, + hp_key=data.get("hpKey", ""), + hp_val=data.get("hpVal", ""), + ) except (aiohttp.ClientError, asyncio.TimeoutError): logger.warning("Failed to get HLTB auth token") return None @@ -140,39 +158,40 @@ def _similarity(a: str, b: str) -> float: return SequenceMatcher(None, a.lower(), b.lower()).ratio() -def _build_search_payload(game_name: str) -> str: +def _build_search_payload(game_name: str, auth: _AuthInfo | None = None) -> str: """Build the JSON POST body for an HLTB search.""" - return json.dumps( - { - "searchType": "games", - "searchTerms": game_name.split(), - "searchPage": 1, - "size": 20, - "searchOptions": { - "games": { - "userId": 0, - "platform": "", - "sortCategory": "popular", - "rangeCategory": "main", - "rangeTime": {"min": 0, "max": 0}, - "gameplay": { - "perspective": "", - "flow": "", - "genre": "", - "difficulty": "", - }, - "rangeYear": {"max": "", "min": ""}, - "modifier": "", + payload: dict[str, Any] = { + "searchType": "games", + "searchTerms": game_name.split(), + "searchPage": 1, + "size": 20, + "searchOptions": { + "games": { + "userId": 0, + "platform": "", + "sortCategory": "popular", + "rangeCategory": "main", + "rangeTime": {"min": 0, "max": 0}, + "gameplay": { + "perspective": "", + "flow": "", + "genre": "", + "difficulty": "", }, - "users": {"sortCategory": "postcount"}, - "lists": {"sortCategory": "follows"}, - "filter": "", - "sort": 0, - "randomizer": 0, + "rangeYear": {"max": "", "min": ""}, + "modifier": "", }, - "useCache": True, - } - ) + "users": {"sortCategory": "postcount"}, + "lists": {"sortCategory": "follows"}, + "filter": "", + "sort": 0, + "randomizer": 0, + }, + "useCache": True, + } + if auth and auth.hp_key: + payload[auth.hp_key] = auth.hp_val + return json.dumps(payload) def _pick_best_hltb_entry( @@ -187,17 +206,21 @@ def _pick_best_hltb_entry( """ if not candidates: return None - if len(candidates) == 1: - return candidates[0] + + # Prefer base games over DLC entries when both are present. + non_dlc = [c for c in candidates if str(c[0].get("game_type", "")).lower() != "dlc"] + usable = non_dlc or candidates + if len(usable) == 1: + return usable[0] lower = search_name.lower() - for entry, sim in candidates: + for entry, sim in usable: entry_name = (entry.get("game_name") or "").lower() if entry_name.startswith((lower + ":", lower + " -")): return entry, sim # Fall back to highest similarity. - return max(candidates, key=lambda x: x[1]) + return max(usable, key=lambda x: x[1]) # ────────────────────────────────────────────────────────────── @@ -213,6 +236,7 @@ class _SearchCtx: search_url: str headers: dict[str, str] cache: dict[int, float] + auth: _AuthInfo | None = None counter: dict[str, int] = field(default_factory=dict) total: int = 0 progress_cb: ProgressCb | None = None @@ -227,7 +251,7 @@ async def _search_one( """Search HLTB for one game via direct POST, update cache.""" async with sem: result: HLTBResult | None = None - payload = _build_search_payload(name) + payload = _build_search_payload(name, ctx.auth) try: async with ctx.session.post( ctx.search_url, @@ -241,13 +265,18 @@ async def _search_one( for entry in data.get("data", []): entry_name = entry.get("game_name", "") entry_alias = entry.get("game_alias", "") or "" + is_dlc = str(entry.get("game_type", "")).lower() == "dlc" sim = max( _similarity(name, entry_name), _similarity(name, entry_alias), ) - is_full_edition = entry_name.lower().startswith( - lower_name + ":" - ) or entry_name.lower().startswith(lower_name + " -") + is_full_edition = ( + (not is_dlc) + and entry_name.lower().startswith(lower_name + ":") + ) or ( + (not is_dlc) + and entry_name.lower().startswith(lower_name + " -") + ) if sim >= MIN_SIMILARITY or is_full_edition: comp_100 = entry.get("comp_100", 0) if comp_100 and comp_100 > 0: @@ -287,6 +316,246 @@ async def _search_one( return result +# ────────────────────────────────────────────────────────────── +# Leisure time + DLC fetching from game detail pages +# ────────────────────────────────────────────────────────────── + +_NEXT_DATA_RE = re.compile( + r'', +) + + +def _parse_game_page(html: str) -> dict[str, Any] | None: + """Extract game data dict from a HLTB game page's __NEXT_DATA__.""" + match = _NEXT_DATA_RE.search(html) + if not match: + return None + try: + data = json.loads(match.group(1)) + result: dict[str, Any] = data["props"]["pageProps"]["game"]["data"] + except (json.JSONDecodeError, KeyError, TypeError): + return None + else: + return result + + +def _as_positive_int(value: object) -> int: + """Convert HLTB numeric JSON values to a positive int, or 0 when invalid.""" + if isinstance(value, int): + return max(0, value) + if isinstance(value, float): + int_value = int(value) + return max(0, int_value) + if isinstance(value, str): + try: + int_value = int(value) + return max(0, int_value) + except ValueError: + return 0 + return 0 + + +def _extract_base_leisure_hours(game_data: dict[str, Any]) -> float: + """Extract base-game leisure hours from game detail data.""" + games = game_data.get("game", []) + if not isinstance(games, list) or not games: + return -1 + if not isinstance(games[0], dict): + return -1 + + base = games[0] + leisure_s = _as_positive_int(base.get("comp_100_h", 0)) + if leisure_s <= 0: + leisure_s = _as_positive_int(base.get("comp_100", 0)) + if leisure_s <= 0: + return -1 + + return round(leisure_s / 3600, 2) + + +def _extract_dlc_relationships(game_data: dict[str, Any]) -> list[tuple[int, float]]: + """Extract DLC relationship IDs and fallback hours from detail data.""" + relationships = game_data.get("relationships", []) + if not isinstance(relationships, list): + return [] + + dlcs: list[tuple[int, float]] = [] + for rel in relationships: + if not isinstance(rel, dict): + continue + if str(rel.get("game_type", "")).lower() != "dlc": + continue + dlc_id = _as_positive_int(rel.get("game_id", 0)) + fallback_comp_100 = _as_positive_int(rel.get("comp_100", 0)) + if fallback_comp_100 > 0: + fallback_hours = round(fallback_comp_100 / 3600, 2) + else: + fallback_hours = 0.0 + dlcs.append((dlc_id, fallback_hours)) + + return dlcs + + +def _extract_leisure_hours(game_data: dict[str, Any]) -> float: + """Compute total leisure hours: base game + all DLCs. + + Uses ``comp_100_h`` (leisure completionist) from the game detail page. + Falls back to ``comp_100`` (average completionist) if leisure unavailable. + Also sums leisure time from any DLC listed in ``relationships``. + """ + base_hours = _extract_base_leisure_hours(game_data) + if base_hours <= 0: + return -1 + + total_hours = base_hours + + # Add DLC leisure times from relationships. + for _dlc_id, fallback_hours in _extract_dlc_relationships(game_data): + total_hours += fallback_hours + + return round(total_hours, 2) + + +async def _fetch_detail_one( + sem: asyncio.Semaphore, + session: aiohttp.ClientSession, + hltb_game_id: int, +) -> dict[str, Any] | None: + """Fetch a single HLTB game detail page and parse its data.""" + async with sem: + url = f"{HLTB_BASE_URL}/game/{hltb_game_id}" + headers = { + "User-Agent": ( + "Mozilla/5.0 (X11; Linux x86_64; rv:136.0) Gecko/20100101 Firefox/136.0" + ), + "accept": "text/html", + "referer": "https://howlongtobeat.com/", + } + try: + async with session.get(url, headers=headers) as resp: + if resp.status == HTTPStatus.OK: + html = await resp.text() + return _parse_game_page(html) + except (aiohttp.ClientError, asyncio.TimeoutError) as exc: + logger.debug( + "HLTB detail fetch failed for game_id=%d: %s", + hltb_game_id, + exc, + ) + return None + + +async def _fetch_leisure_times( + search_results: list[HLTBResult], + cache: dict[int, float], + progress_cb: ProgressCb | None, +) -> None: + """Fetch leisure times from game detail pages for all search results. + + Updates ``cache`` in-place with leisure hours (including DLC time). + """ + valid = [r for r in search_results if r.hltb_game_id > 0] + if not valid: + return + + timeout = aiohttp.ClientTimeout(total=30, sock_read=20) + sem = asyncio.Semaphore(MAX_CONCURRENT) + connector = aiohttp.TCPConnector( + limit=MAX_CONCURRENT, + keepalive_timeout=30, + ) + + total = len(valid) + done = 0 + found = 0 + + async with aiohttp.ClientSession( + timeout=timeout, + connector=connector, + ) as session: + coros = [_fetch_detail_one(sem, session, r.hltb_game_id) for r in valid] + details = await asyncio.gather(*coros) + + dlc_relationships_by_app, dlc_ids = _collect_dlc_relationships(valid, details) + dlc_hours_by_id = await _fetch_dlc_leisure_hours(sem, session, dlc_ids) + + for r, game_data in zip(valid, details, strict=False): + done += 1 + if game_data is not None: + leisure = _extract_leisure_hours(game_data) + 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 progress_cb is not None: + progress_cb(done, total, found, r.game_name) + + if done % _SAVE_INTERVAL == 0: + save_hltb_cache(cache) + + +def _collect_dlc_relationships( + valid: list[HLTBResult], + details: list[dict[str, Any] | None], +) -> tuple[dict[int, list[tuple[int, float]]], list[int]]: + """Collect DLC relationship IDs for all base-game detail responses.""" + by_app: dict[int, list[tuple[int, float]]] = {} + unique_dlc_ids: set[int] = set() + + for result, game_data in zip(valid, details, strict=False): + if game_data is None: + continue + dlc_rels = _extract_dlc_relationships(game_data) + by_app[result.app_id] = dlc_rels + for dlc_id, _fallback_hours in dlc_rels: + if dlc_id > 0: + unique_dlc_ids.add(dlc_id) + + return by_app, sorted(unique_dlc_ids) + + +async def _fetch_dlc_leisure_hours( + sem: asyncio.Semaphore, + session: aiohttp.ClientSession, + dlc_ids: list[int], +) -> dict[int, float]: + """Fetch leisure hours for each DLC game id.""" + if not dlc_ids: + return {} + + coros = [_fetch_detail_one(sem, session, dlc_id) for dlc_id in dlc_ids] + dlc_details = await asyncio.gather(*coros) + + dlc_hours_by_id: dict[int, float] = {} + for dlc_id, dlc_data in zip(dlc_ids, dlc_details, strict=False): + if dlc_data is None: + continue + dlc_leisure = _extract_base_leisure_hours(dlc_data) + if dlc_leisure > 0: + dlc_hours_by_id[dlc_id] = dlc_leisure + return dlc_hours_by_id + + +def _apply_dlc_leisure_overrides( + base_hours: float, + dlc_rels: list[tuple[int, float]], + dlc_hours_by_id: dict[int, float], +) -> float: + """Replace fallback DLC hours with detailed leisure hours when available.""" + adjusted = base_hours + for dlc_id, fallback_hours in dlc_rels: + dlc_leisure = dlc_hours_by_id.get(dlc_id, -1.0) + if dlc_leisure > 0: + adjusted += dlc_leisure - fallback_hours + return round(adjusted, 2) + + async def _fetch_batch( games: list[tuple[int, str]], cache: dict[int, float], @@ -299,24 +568,27 @@ async def _fetch_batch( timeout = aiohttp.ClientTimeout(total=20, sock_read=15) - # 2. Get auth token (separate session — avoids reuse issues). + # 2. Get auth info (separate session — avoids reuse issues). async with aiohttp.ClientSession(timeout=timeout) as init_session: - token = await _get_auth_token(search_url, init_session) - if token is None: - logger.warning("Could not get HLTB auth token, aborting fetch.") + auth = await _get_auth_info(search_url, init_session) + if auth is None: + logger.warning("Could not get HLTB auth info, aborting fetch.") return [] logger.info("HLTB auth token acquired.") # 3. Build shared headers for all search requests. - headers = { + headers: dict[str, str] = { "content-type": "application/json", "accept": "*/*", "User-Agent": ( "Mozilla/5.0 (X11; Linux x86_64; rv:136.0) Gecko/20100101 Firefox/136.0" ), "referer": "https://howlongtobeat.com/", - "x-auth-token": token, + "x-auth-token": auth.token, } + if auth.hp_key: + headers["x-hp-key"] = auth.hp_key + headers["x-hp-val"] = auth.hp_val # 4. Fire all searches through a single persistent session. sem = asyncio.Semaphore(MAX_CONCURRENT) @@ -336,6 +608,7 @@ async def _fetch_batch( search_url=search_url, headers=headers, cache=cache, + auth=auth, counter=counter, total=total, progress_cb=progress_cb, @@ -351,7 +624,16 @@ async def _fetch_batch( ] results = await asyncio.gather(*tasks) - return [r for r in results if r is not None] + search_results = [r for r in results if r is not None] + + # 5. Fetch leisure times + DLC from game detail pages. + logger.info( + "Fetching leisure times for %d games from detail pages...", + len(search_results), + ) + await _fetch_leisure_times(search_results, cache, progress_cb=None) + + return search_results def fetch_hltb_times( diff --git a/steam_backlog_enforcer/main.py b/steam_backlog_enforcer/main.py index b42ff7c..7685e32 100644 --- a/steam_backlog_enforcer/main.py +++ b/steam_backlog_enforcer/main.py @@ -398,7 +398,7 @@ def cmd_done(config: Config, state: State) -> None: All-in-one command for after finishing a game: 1. Verify 100% achievements on Steam. - 2. Pick the next game (shortest HLTB 100% time). + 2. Pick the next game (shortest HLTB leisure+dlc time). 3. Uninstall all non-assigned games. 4. Hide all non-assigned games in the Steam library. 5. Install the newly assigned game. @@ -428,7 +428,7 @@ def cmd_done(config: Config, state: State) -> None: hltb_cache = fetch_hltb_times_cached([(app_id, game_name)]) hours = hltb_cache.get(app_id, -1.0) if hours > 0: - _echo(f" HLTB 100% estimate: {hours:.1f} hours") + _echo(f" HLTB leisure+dlc estimate: {hours:.1f} hours") if _try_reassign_shorter_game(hltb_cache, app_id, hours, state, config): return diff --git a/steam_backlog_enforcer/scanning.py b/steam_backlog_enforcer/scanning.py index 3bf4fd0..5614f19 100644 --- a/steam_backlog_enforcer/scanning.py +++ b/steam_backlog_enforcer/scanning.py @@ -178,7 +178,7 @@ def pick_next_game(games: list[GameInfo], state: State, config: Config) -> None: hours_str = "" if chosen.completionist_hours > 0: - hours_str = f" (~{chosen.completionist_hours:.1f}h to 100%)" + hours_str = f" (~{chosen.completionist_hours:.1f}h leisure+dlc)" _echo(f"\n>>> ASSIGNED: {chosen.name} (AppID={chosen.app_id}){hours_str}") _echo( f" Progress: {chosen.unlocked_achievements}/{chosen.total_achievements}" diff --git a/steam_backlog_enforcer/tests/test_hltb.py b/steam_backlog_enforcer/tests/test_hltb.py index f93686e..9068efd 100644 --- a/steam_backlog_enforcer/tests/test_hltb.py +++ b/steam_backlog_enforcer/tests/test_hltb.py @@ -12,10 +12,21 @@ from typing_extensions import Self from python_pkg.steam_backlog_enforcer.hltb import ( HLTBResult, + _apply_dlc_leisure_overrides, + _as_positive_int, + _AuthInfo, _build_search_payload, + _collect_dlc_relationships, + _extract_base_leisure_hours, + _extract_dlc_relationships, + _extract_leisure_hours, _fetch_batch, - _get_auth_token, + _fetch_detail_one, + _fetch_dlc_leisure_hours, + _fetch_leisure_times, + _get_auth_info, _get_hltb_search_url, + _parse_game_page, _pick_best_hltb_entry, _search_one, _SearchCtx, @@ -107,10 +118,27 @@ class TestGetHltbSearchUrl: assert url == "https://howlongtobeat.com/api/finder" -class TestGetAuthToken: - """Tests for _get_auth_token.""" +class TestGetAuthInfo: + """Tests for _get_auth_info.""" def test_success(self) -> None: + mock_resp = AsyncMock() + mock_resp.status = 200 + mock_resp.json = AsyncMock( + return_value={"token": "abc123", "hpKey": "ign_x", "hpVal": "ff"} + ) + mock_resp.__aenter__ = AsyncMock(return_value=mock_resp) + mock_resp.__aexit__ = AsyncMock(return_value=False) + + mock_session = MagicMock() + mock_session.get = MagicMock(return_value=mock_resp) + + result = asyncio.run( + _get_auth_info("https://howlongtobeat.com/api/finder", mock_session) + ) + assert result == _AuthInfo("abc123", "ign_x", "ff") + + def test_success_no_hp(self) -> None: mock_resp = AsyncMock() mock_resp.status = 200 mock_resp.json = AsyncMock(return_value={"token": "abc123"}) @@ -121,9 +149,24 @@ class TestGetAuthToken: mock_session.get = MagicMock(return_value=mock_resp) result = asyncio.run( - _get_auth_token("https://howlongtobeat.com/api/finder", mock_session) + _get_auth_info("https://howlongtobeat.com/api/finder", mock_session) ) - assert result == "abc123" + assert result == _AuthInfo("abc123") + + def test_no_token_key(self) -> None: + mock_resp = AsyncMock() + mock_resp.status = 200 + mock_resp.json = AsyncMock(return_value={"notoken": True}) + mock_resp.__aenter__ = AsyncMock(return_value=mock_resp) + mock_resp.__aexit__ = AsyncMock(return_value=False) + + mock_session = MagicMock() + mock_session.get = MagicMock(return_value=mock_resp) + + result = asyncio.run( + _get_auth_info("https://howlongtobeat.com/api/finder", mock_session) + ) + assert result is None def test_non_200(self) -> None: mock_resp = AsyncMock() @@ -135,7 +178,7 @@ class TestGetAuthToken: mock_session.get = MagicMock(return_value=mock_resp) result = asyncio.run( - _get_auth_token("https://howlongtobeat.com/api/finder", mock_session) + _get_auth_info("https://howlongtobeat.com/api/finder", mock_session) ) assert result is None @@ -147,7 +190,7 @@ class TestGetAuthToken: mock_session.get = MagicMock(return_value=ctx) result = asyncio.run( - _get_auth_token("https://howlongtobeat.com/api/finder", mock_session) + _get_auth_info("https://howlongtobeat.com/api/finder", mock_session) ) assert result is None @@ -174,6 +217,18 @@ class TestBuildSearchPayload: assert data["searchType"] == "games" assert data["searchTerms"] == ["Half-Life", "2"] + def test_with_auth(self) -> None: + auth = _AuthInfo("t", "ign_x", "ff") + payload = _build_search_payload("TF2", auth=auth) + data = json.loads(payload) + assert data["ign_x"] == "ff" + + def test_with_auth_no_hp_key(self) -> None: + auth = _AuthInfo("t") + payload = _build_search_payload("TF2", auth=auth) + data = json.loads(payload) + assert "" not in data + class TestPickBestHltbEntry: """Tests for _pick_best_hltb_entry.""" @@ -211,6 +266,21 @@ class TestPickBestHltbEntry: assert result is not None assert result[1] == 0.9 + def test_prefers_non_dlc_when_available(self) -> None: + base: dict[str, Any] = { + "game_name": "Helltaker", + "game_type": "game", + "comp_100": 6846, + } + dlc: dict[str, Any] = { + "game_name": "Helltaker - Bonus Chapter: Examtaker", + "game_type": "dlc", + "comp_100": 4075, + } + result = _pick_best_hltb_entry("Helltaker", [(dlc, 0.95), (base, 0.8)]) + assert result is not None + assert result[0]["game_type"] == "game" + class _FakeResponse: """Async context manager mimicking aiohttp response.""" @@ -409,14 +479,14 @@ class TestSearchOne: class TestFetchBatchHltb: """Tests for _fetch_batch (the hltb version).""" - def test_no_token(self) -> None: + def test_no_auth(self) -> None: with ( patch( "python_pkg.steam_backlog_enforcer.hltb._get_hltb_search_url", return_value="https://example.com", ), patch( - "python_pkg.steam_backlog_enforcer.hltb._get_auth_token", + "python_pkg.steam_backlog_enforcer.hltb._get_auth_info", new_callable=AsyncMock, return_value=None, ), @@ -424,16 +494,17 @@ class TestFetchBatchHltb: results = asyncio.run(_fetch_batch([(440, "TF2")], {}, None)) assert results == [] - def test_with_token(self) -> None: + def test_with_auth(self) -> None: + auth = _AuthInfo("token123", "ign_x", "ff") with ( patch( "python_pkg.steam_backlog_enforcer.hltb._get_hltb_search_url", return_value="https://example.com", ), patch( - "python_pkg.steam_backlog_enforcer.hltb._get_auth_token", + "python_pkg.steam_backlog_enforcer.hltb._get_auth_info", new_callable=AsyncMock, - return_value="token123", + return_value=auth, ), patch( "python_pkg.steam_backlog_enforcer.hltb._search_one", @@ -443,28 +514,540 @@ class TestFetchBatchHltb: game_name="TF2", completionist_hours=50.0, similarity=1.0, + hltb_game_id=12345, ), ), + patch( + "python_pkg.steam_backlog_enforcer.hltb._fetch_leisure_times", + new_callable=AsyncMock, + ), ): results = asyncio.run(_fetch_batch([(440, "TF2")], {}, None)) assert len(results) == 1 - def test_filters_none_results(self) -> None: + def test_with_auth_no_hp(self) -> None: + auth = _AuthInfo("tok123") with ( patch( "python_pkg.steam_backlog_enforcer.hltb._get_hltb_search_url", return_value="https://example.com", ), patch( - "python_pkg.steam_backlog_enforcer.hltb._get_auth_token", + "python_pkg.steam_backlog_enforcer.hltb._get_auth_info", new_callable=AsyncMock, - return_value="token123", + return_value=auth, ), patch( "python_pkg.steam_backlog_enforcer.hltb._search_one", new_callable=AsyncMock, return_value=None, ), + patch( + "python_pkg.steam_backlog_enforcer.hltb._fetch_leisure_times", + new_callable=AsyncMock, + ), ): results = asyncio.run(_fetch_batch([(440, "TF2")], {}, None)) assert results == [] + + def test_filters_none_results(self) -> None: + auth = _AuthInfo("tok123") + with ( + patch( + "python_pkg.steam_backlog_enforcer.hltb._get_hltb_search_url", + return_value="https://example.com", + ), + patch( + "python_pkg.steam_backlog_enforcer.hltb._get_auth_info", + new_callable=AsyncMock, + return_value=auth, + ), + patch( + "python_pkg.steam_backlog_enforcer.hltb._search_one", + new_callable=AsyncMock, + return_value=None, + ), + patch( + "python_pkg.steam_backlog_enforcer.hltb._fetch_leisure_times", + new_callable=AsyncMock, + ), + ): + results = asyncio.run(_fetch_batch([(440, "TF2")], {}, None)) + assert results == [] + + +class TestParseGamePage: + """Tests for _parse_game_page.""" + + def test_valid_html(self) -> None: + game_data: dict[str, Any] = { + "game": [{"comp_100_h": 21243, "comp_100": 6800}], + "relationships": [], + } + next_data = { + "props": {"pageProps": {"game": {"data": game_data}}}, + } + html = ( + '" + ) + assert _parse_game_page(html) == game_data + + def test_no_script_tag(self) -> None: + assert _parse_game_page("") is None + + def test_bad_json(self) -> None: + html = '' + assert _parse_game_page(html) is None + + def test_missing_keys(self) -> None: + html = ( + '' + ) + assert _parse_game_page(html) is None + + +class TestExtractLeisureHours: + """Tests for _extract_leisure_hours.""" + + def test_leisure_time_only(self) -> None: + data: dict[str, Any] = { + "game": [{"comp_100_h": 21243, "comp_100": 6800}], + "relationships": [], + } + assert _extract_leisure_hours(data) == round(21243 / 3600, 2) + + def test_leisure_with_dlc(self) -> None: + data: dict[str, Any] = { + "game": [{"comp_100_h": 21243, "comp_100": 6800}], + "relationships": [ + {"game_type": "dlc", "comp_100": 12298}, + {"game_type": "dlc", "comp_100": 3600}, + ], + } + assert _extract_leisure_hours(data) == round((21243 + 12298 + 3600) / 3600, 2) + + def test_fallback_to_comp_100(self) -> None: + data: dict[str, Any] = { + "game": [{"comp_100": 7200}], + "relationships": [], + } + assert _extract_leisure_hours(data) == round(7200 / 3600, 2) + + def test_no_game_data(self) -> None: + assert _extract_leisure_hours({"game": [], "relationships": []}) == -1 + + def test_zero_leisure(self) -> None: + data: dict[str, Any] = { + "game": [{"comp_100_h": 0, "comp_100": 0}], + "relationships": [], + } + assert _extract_leisure_hours(data) == -1 + + def test_no_game_key(self) -> None: + assert _extract_leisure_hours({"relationships": []}) == -1 + + def test_non_dlc_relationship_ignored(self) -> None: + data: dict[str, Any] = { + "game": [{"comp_100_h": 3600}], + "relationships": [ + {"game_type": "game", "comp_100": 9999}, + {"game_type": "dlc", "comp_100": 1800}, + ], + } + assert _extract_leisure_hours(data) == round((3600 + 1800) / 3600, 2) + + def test_dlc_zero_comp_100_skipped(self) -> None: + data: dict[str, Any] = { + "game": [{"comp_100_h": 3600}], + "relationships": [ + {"game_type": "dlc", "comp_100": 0}, + ], + } + assert _extract_leisure_hours(data) == round(3600 / 3600, 2) + + def test_negative_leisure(self) -> None: + data: dict[str, Any] = { + "game": [{"comp_100_h": -1, "comp_100": -1}], + "relationships": [], + } + assert _extract_leisure_hours(data) == -1 + + def test_string_numeric_fields(self) -> None: + data: dict[str, Any] = { + "game": [{"comp_100_h": "7200", "comp_100": "3600"}], + "relationships": [{"game_type": "dlc", "game_id": "1", "comp_100": "1800"}], + } + assert _extract_leisure_hours(data) == round((7200 + 1800) / 3600, 2) + + def test_bad_string_falls_back_to_comp_100(self) -> None: + data: dict[str, Any] = { + "game": [{"comp_100_h": "bad", "comp_100": "3600"}], + "relationships": [], + } + assert _extract_leisure_hours(data) == 1.0 + + def test_relationships_not_list(self) -> None: + data: dict[str, Any] = { + "game": [{"comp_100_h": 3600}], + "relationships": "not-a-list", + } + assert _extract_leisure_hours(data) == 1.0 + + +class TestInternalHelpers: + """Tests for internal helper coverage.""" + + def test_as_positive_int_float(self) -> None: + assert _as_positive_int(1.9) == 1 + + def test_as_positive_int_invalid_type(self) -> None: + assert _as_positive_int(object()) == 0 + + def test_extract_base_leisure_non_dict_game(self) -> None: + data: dict[str, Any] = {"game": [123]} + assert _extract_base_leisure_hours(data) == -1 + + def test_extract_dlc_relationships_skips_non_dict(self) -> None: + data: dict[str, Any] = { + "relationships": [ + "bad", + {"game_type": "dlc", "game_id": 7, "comp_100": 3600}, + ], + } + assert _extract_dlc_relationships(data) == [(7, 1.0)] + + def test_collect_dlc_relationships_ignores_non_positive_id(self) -> None: + valid = [ + HLTBResult( + app_id=1, + game_name="Game", + completionist_hours=1.0, + similarity=1.0, + hltb_game_id=123, + ) + ] + details: list[dict[str, Any] | None] = [ + { + "relationships": [ + {"game_type": "dlc", "game_id": 0, "comp_100": 3600}, + ] + } + ] + by_app, ids = _collect_dlc_relationships(valid, details) + assert by_app[1] == [(0, 1.0)] + assert ids == [] + + def test_apply_dlc_leisure_overrides(self) -> None: + adjusted = _apply_dlc_leisure_overrides( + base_hours=6.0, + dlc_rels=[(10, 1.0), (11, 2.0)], + dlc_hours_by_id={10: 3.0}, + ) + assert adjusted == 8.0 + + def test_fetch_dlc_leisure_hours_empty(self) -> None: + async def _run() -> dict[int, float]: + async with aiohttp.ClientSession() as session: + return await _fetch_dlc_leisure_hours(asyncio.Semaphore(1), session, []) + + assert asyncio.run(_run()) == {} + + def test_fetch_dlc_leisure_hours_skips_none_data(self) -> None: + async def _run() -> dict[int, float]: + async with aiohttp.ClientSession() as session: + with patch( + "python_pkg.steam_backlog_enforcer.hltb._fetch_detail_one", + new_callable=AsyncMock, + return_value=None, + ): + return await _fetch_dlc_leisure_hours( + asyncio.Semaphore(1), + session, + [1], + ) + + assert asyncio.run(_run()) == {} + + def test_fetch_dlc_leisure_hours_skips_non_positive_leisure(self) -> None: + bad_dlc_data: dict[str, Any] = { + "game": [{"comp_100_h": 0, "comp_100": 0}], + "relationships": [], + } + + async def _run() -> dict[int, float]: + async with aiohttp.ClientSession() as session: + with patch( + "python_pkg.steam_backlog_enforcer.hltb._fetch_detail_one", + new_callable=AsyncMock, + return_value=bad_dlc_data, + ): + return await _fetch_dlc_leisure_hours( + asyncio.Semaphore(1), + session, + [1], + ) + + assert asyncio.run(_run()) == {} + + +class _FakeTextResponse: + """Async context manager mimicking aiohttp response for text.""" + + def __init__(self, status: int, text: str = "") -> None: + self.status = status + self._text = text + + async def __aenter__(self) -> Self: + return self + + async def __aexit__(self, *args: object) -> None: + pass + + async def text(self) -> str: + return self._text + + +class TestFetchDetailOne: + """Tests for _fetch_detail_one.""" + + def test_success(self) -> None: + game_data: dict[str, Any] = { + "game": [{"comp_100_h": 21243}], + "relationships": [], + } + next_data = {"props": {"pageProps": {"game": {"data": game_data}}}} + html = ( + '" + ) + resp = _FakeTextResponse(200, html) + session = MagicMock() + session.get = MagicMock(return_value=resp) + result = asyncio.run(_fetch_detail_one(asyncio.Semaphore(1), session, 12345)) + assert result == game_data + + def test_non_200(self) -> None: + resp = _FakeTextResponse(404) + session = MagicMock() + session.get = MagicMock(return_value=resp) + result = asyncio.run(_fetch_detail_one(asyncio.Semaphore(1), session, 12345)) + assert result is None + + def test_client_error(self) -> None: + ctx = AsyncMock() + ctx.__aenter__ = AsyncMock(side_effect=aiohttp.ClientError) + ctx.__aexit__ = AsyncMock(return_value=False) + session = MagicMock() + session.get = MagicMock(return_value=ctx) + result = asyncio.run(_fetch_detail_one(asyncio.Semaphore(1), session, 12345)) + assert result is None + + def test_parse_failure(self) -> None: + resp = _FakeTextResponse(200, "no script") + session = MagicMock() + session.get = MagicMock(return_value=resp) + result = asyncio.run(_fetch_detail_one(asyncio.Semaphore(1), session, 12345)) + assert result is None + + +class TestFetchLeisureTimes: + """Tests for _fetch_leisure_times.""" + + def test_updates_cache(self) -> None: + 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": 21243}], + "relationships": [], + } + cache: dict[int, float] = {} + with patch( + "python_pkg.steam_backlog_enforcer.hltb._fetch_detail_one", + new_callable=AsyncMock, + return_value=game_data, + ): + asyncio.run(_fetch_leisure_times(results, cache, None)) + assert cache[440] == round(21243 / 3600, 2) + assert results[0].completionist_hours == round(21243 / 3600, 2) + + def test_no_valid_results(self) -> None: + results = [ + HLTBResult( + app_id=440, + game_name="TF2", + completionist_hours=50.0, + similarity=1.0, + hltb_game_id=0, + ), + ] + cache: dict[int, float] = {} + asyncio.run(_fetch_leisure_times(results, cache, None)) + assert cache == {} + + def test_empty_results(self) -> None: + cache: dict[int, float] = {} + asyncio.run(_fetch_leisure_times([], cache, None)) + assert cache == {} + + def test_detail_returns_none(self) -> None: + results = [ + HLTBResult( + app_id=440, + game_name="TF2", + completionist_hours=50.0, + similarity=1.0, + hltb_game_id=12345, + ), + ] + cache: dict[int, float] = {} + with patch( + "python_pkg.steam_backlog_enforcer.hltb._fetch_detail_one", + new_callable=AsyncMock, + return_value=None, + ): + asyncio.run(_fetch_leisure_times(results, cache, None)) + assert cache == {} + assert results[0].completionist_hours == 50.0 + + def test_negative_leisure(self) -> None: + 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": [], "relationships": []} + cache: dict[int, float] = {} + with patch( + "python_pkg.steam_backlog_enforcer.hltb._fetch_detail_one", + new_callable=AsyncMock, + return_value=game_data, + ): + asyncio.run(_fetch_leisure_times(results, cache, None)) + assert cache == {} + assert results[0].completionist_hours == 50.0 + + def test_with_progress_cb(self) -> None: + 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": 3600}], + "relationships": [], + } + cache: dict[int, float] = {} + cb = MagicMock() + with patch( + "python_pkg.steam_backlog_enforcer.hltb._fetch_detail_one", + new_callable=AsyncMock, + return_value=game_data, + ): + asyncio.run(_fetch_leisure_times(results, cache, cb)) + cb.assert_called_once() + + def test_save_interval(self) -> None: + """Trigger the _SAVE_INTERVAL branch in leisure fetching.""" + from python_pkg.steam_backlog_enforcer.hltb import _SAVE_INTERVAL + + results = [ + HLTBResult( + app_id=i, + game_name=f"Game{i}", + completionist_hours=1.0, + similarity=1.0, + hltb_game_id=i + 1000, + ) + for i in range(_SAVE_INTERVAL) + ] + game_data: dict[str, Any] = { + "game": [{"comp_100_h": 3600}], + "relationships": [], + } + cache: dict[int, float] = {} + with ( + patch( + "python_pkg.steam_backlog_enforcer.hltb._fetch_detail_one", + new_callable=AsyncMock, + return_value=game_data, + ), + patch( + "python_pkg.steam_backlog_enforcer.hltb.save_hltb_cache" + ) as mock_save, + ): + asyncio.run(_fetch_leisure_times(results, cache, None)) + mock_save.assert_called_once() + + def test_dlc_detail_overrides_relationship_fallback(self) -> None: + results = [ + HLTBResult( + app_id=1289310, + game_name="Helltaker", + completionist_hours=1.0, + similarity=1.0, + hltb_game_id=78118, + ), + ] + base_data: dict[str, Any] = { + "game": [{"comp_100_h": 21243, "comp_100": 6846}], + "relationships": [{"game_type": "dlc", "game_id": 92236, "comp_100": 4075}], + } + dlc_data: dict[str, Any] = { + "game": [{"comp_100_h": 12298, "comp_100": 4075}], + "relationships": [], + } + cache: dict[int, float] = {} + with patch( + "python_pkg.steam_backlog_enforcer.hltb._fetch_detail_one", + new_callable=AsyncMock, + side_effect=[base_data, dlc_data], + ): + asyncio.run(_fetch_leisure_times(results, cache, None)) + + expected = round((21243 + 12298) / 3600, 2) + assert cache[1289310] == expected + assert results[0].completionist_hours == expected + + def test_missing_dlc_detail_keeps_relationship_fallback(self) -> None: + results = [ + HLTBResult( + app_id=1289310, + game_name="Helltaker", + completionist_hours=1.0, + similarity=1.0, + hltb_game_id=78118, + ), + ] + base_data: dict[str, Any] = { + "game": [{"comp_100_h": 21243, "comp_100": 6846}], + "relationships": [{"game_type": "dlc", "game_id": 92236, "comp_100": 4075}], + } + cache: dict[int, float] = {} + with patch( + "python_pkg.steam_backlog_enforcer.hltb._fetch_detail_one", + new_callable=AsyncMock, + side_effect=[base_data, None], + ): + asyncio.run(_fetch_leisure_times(results, cache, None)) + + expected = round((21243 + 4075) / 3600, 2) + assert cache[1289310] == expected + assert results[0].completionist_hours == expected