diff --git a/python_pkg/steam_backlog_enforcer/hltb.py b/python_pkg/steam_backlog_enforcer/hltb.py
index 5e87522..a7a287d 100644
--- a/python_pkg/steam_backlog_enforcer/hltb.py
+++ b/python_pkg/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/python_pkg/steam_backlog_enforcer/main.py b/python_pkg/steam_backlog_enforcer/main.py
index b42ff7c..7685e32 100644
--- a/python_pkg/steam_backlog_enforcer/main.py
+++ b/python_pkg/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/python_pkg/steam_backlog_enforcer/scanning.py b/python_pkg/steam_backlog_enforcer/scanning.py
index 3bf4fd0..5614f19 100644
--- a/python_pkg/steam_backlog_enforcer/scanning.py
+++ b/python_pkg/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/python_pkg/steam_backlog_enforcer/tests/test_hltb.py b/python_pkg/steam_backlog_enforcer/tests/test_hltb.py
index f93686e..9068efd 100644
--- a/python_pkg/steam_backlog_enforcer/tests/test_hltb.py
+++ b/python_pkg/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