#!/usr/bin/env python3 # SPDX-FileCopyrightText: 2026 Jason Doyle # SPDX-License-Identifier: MIT # # Permission is hereby granted, free of charge, to any person obtaining a copy # of this software and associated documentation files (the "Software"), to deal # in the Software without restriction, including without limitation the rights # to use, copy, modify, merge, publish, distribute, sublicense, and/or sell # copies of the Software, and to permit persons to whom the Software is # furnished to do so, subject to the following conditions: # # The above copyright notice and this permission notice shall be included in # all copies or substantial portions of the Software. # # THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR # IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY, # FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE # AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER # LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM, # OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE # SOFTWARE. from __future__ import annotations import hashlib import json import re import time import unicodedata from datetime import datetime, timedelta, timezone from pathlib import Path from typing import Any import requests from jsonschema import Draft202012Validator, FormatChecker ROOT = Path(__file__).resolve().parent ANNOTATIONS_PATH = ROOT / "public-case-study-annotations.json" SCHEMA_PATH = ROOT / "game-discovery-profile.schema.json" SNAPSHOT_PATH = ROOT / "public-case-study-source-snapshot.json" PROFILES_PATH = ROOT / "public-case-study-profiles.json" PROFILE_SCHEMA_URL = ( "https://jasondoyle.ie/whitepapers/game-discovery-is-an-evidence-problem/" "game-discovery-profile.schema.json" ) USER_AGENT = "jasondoyle-whitepaper-case-study/0.1" HTTP = requests.Session() HTTP.headers.update({"User-Agent": USER_AGENT}) REVIEW_PARAMS = { "json": 1, "language": "english", "purchase_type": "steam", "filter": "all", "review_type": "all", "day_range": 365, "num_per_page": 100, "filter_offtopic_activity": 1, } CATEGORY_SINGLE_PLAYER = 2 CATEGORY_ONLINE_COOP = 38 CATEGORY_LOCAL_COOP = 39 def utc_now() -> str: return ( datetime.now(timezone.utc) .replace(microsecond=0) .isoformat() .replace("+00:00", "Z") ) def sha256_bytes(value: bytes) -> str: return "sha256:" + hashlib.sha256(value).hexdigest() def sha256_text(value: str) -> str: return sha256_bytes(value.encode("utf-8")) def canonical_json(value: Any) -> str: return json.dumps( value, ensure_ascii=True, separators=(",", ":"), sort_keys=True, ) def fetch_json( url: str, params: dict[str, Any], ) -> tuple[dict[str, Any], str, str, str, str]: last_error: Exception | None = None for attempt in range(4): try: started_at = utc_now() response = HTTP.get(url, params=params, timeout=45) response.raise_for_status() completed_at = utc_now() raw = response.content return ( json.loads(raw), sha256_bytes(raw), response.url, started_at, completed_at, ) except Exception as exc: last_error = exc if attempt == 3: break time.sleep(1.5 * (attempt + 1)) raise RuntimeError(f"Failed to fetch {url}: {last_error}") def normalise_review(value: str) -> str: without_bbcode = re.sub(r"\[/?[^\]]+\]", " ", value) collapsed = " ".join(without_bbcode.split()) return ( unicodedata.normalize("NFKD", collapsed) .encode("ascii", "ignore") .decode("ascii") ) def review_source_uri(appid: int) -> str: return f"https://store.steampowered.com/appreviews/{appid}" def metadata_evidence( appid: int, retrieved_at: str, digest: str, request_url: str, suffix: str, ) -> dict[str, Any]: return { "evidence_id": f"steam-appdetails-{appid}-{suffix}", "source_type": "developer_declaration", "source_uri": f"https://store.steampowered.com/app/{appid}/", "source_snapshot_digest": digest, "retrieved_at": retrieved_at, "method": ( "Normalised from Valve-hosted appdetails data. The endpoint is " "undocumented and treated as an unstable extraction transport." ), "method_version": "public-case-study-0.1", "simulated": False, "x_extensions": { "transport_url": request_url, "documentation_status": "undocumented", }, } def functional_claim( appid: int, release_id: str, target_platforms: list[str], property_name: str, value: bool, metadata_retrieved_at: str, digest: str, request_url: str, ) -> dict[str, Any]: suffix = property_name.replace(".", "-") claim_target_platforms = target_platforms if property_name.startswith("target_platform."): claim_target_platforms = [property_name.split(".", 1)[1]] return { "claim_id": f"steam-{appid}-{suffix}", "property": property_name, "claim_class": "functional_fact", "asserted_by": "jasondoyle.ie:public-steam-case-study", "value_status": "known", "value": value, "applicability": { "release_ids": [release_id], "storefronts": ["steam"], "target_platforms": claim_target_platforms, "valid_from": metadata_retrieved_at, }, "evidence": [ metadata_evidence( appid, metadata_retrieved_at, digest, request_url, suffix, ) ], "privacy": {"release_mode": "public"}, "status": "active", } def query_evidence( appid: int, release_id: str, review_pages: list[dict[str, Any]], selected_count: int, ) -> dict[str, Any]: digest_input = canonical_json( [page["response_sha256"] for page in review_pages] ) return { "evidence_id": f"steam-review-query-{appid}", "source_type": "player_report", "source_uri": review_source_uri(appid), "source_snapshot_digest": sha256_text(digest_input), "retrieved_at": max(page["retrieved_at"] for page in review_pages), "method": ( "Human review of the configured excerpts found no sufficient " "support for a stronger claim." ), "method_version": "public-case-study-0.1", "sample_size": selected_count, "simulated": False, "x_extensions": { "release_id": release_id, "query_parameters": REVIEW_PARAMS, "human_checked": True, }, } def build_review_evidence( appid: int, review: dict[str, Any], ) -> dict[str, Any]: evidence_key = review["review_evidence_key"] return { "evidence_id": f"steam-review-{evidence_key.split(':', 1)[1][:16]}", "source_type": "player_report", "source_uri": review_source_uri(appid), "source_snapshot_digest": review["page_response_sha256"], "retrieved_at": review["retrieved_at"], "method": "Human-checked excerpt from the documented Steam review API.", "method_version": "public-case-study-0.1", "sample_size": 1, "simulated": False, "x_extensions": { "review_evidence_key": evidence_key, "human_checked": True, }, } def experiential_claim( annotation: dict[str, Any], appid: int, release_id: str, target_platforms: list[str], generated_at: str, selected_reviews: dict[str, dict[str, Any]], review_pages: list[dict[str, Any]], metadata_digest: str, metadata_url: str, metadata_retrieved_at: str, ) -> dict[str, Any]: value_status = annotation["value_status"] review_fingerprints = annotation["review_fingerprints"] evidence = [ build_review_evidence(appid, selected_reviews[fingerprint]) for fingerprint in review_fingerprints ] if not evidence and value_status == "not_applicable": evidence = [ metadata_evidence( appid, metadata_retrieved_at, metadata_digest, metadata_url, annotation["property"].replace(".", "-"), ) ] elif not evidence: evidence = [ query_evidence( appid, release_id, review_pages, len(selected_reviews), ) ] claim: dict[str, Any] = { "claim_id": ( f"steam-{appid}-" + annotation["property"].replace(".", "-") ), "property": annotation["property"], "claim_class": "population_experience", "asserted_by": "jasondoyle.ie:public-steam-case-study", "value_status": value_status, "applicability": { "release_ids": [release_id], "storefronts": ["steam"], "target_platforms": target_platforms, "languages": ["en"], "cohort": "Configured human-checked Steam review excerpts", "valid_from": generated_at, }, "evidence": evidence, "uncertainty": [ { "kind": "applicability", "description": ( annotation["rationale"] + " This case-study annotation is not a prevalence estimate." ), } ], "privacy": {"release_mode": "public"}, "status": annotation["status"], } if value_status == "known": claim["value"] = annotation["value"] return claim def collect_reviews( appid: int, configured_reviews: list[dict[str, Any]], ) -> tuple[dict[str, dict[str, Any]], list[dict[str, Any]]]: wanted = { review["review_fingerprint"]: review for review in configured_reviews } found: dict[str, dict[str, Any]] = {} pages: list[dict[str, Any]] = [] cursor = "*" cutoff = datetime.now(timezone.utc) - timedelta( days=REVIEW_PARAMS["day_range"] ) for _ in range(20): params = dict(REVIEW_PARAMS) params["cursor"] = cursor payload, digest, _request_url, started_at, retrieved_at = fetch_json( review_source_uri(appid), params, ) reviews = payload.get("reviews", []) page = { "request_url": review_source_uri(appid), "cursor_sha256": sha256_text(cursor), "response_sha256": digest, "review_count": len(reviews), "request_started_at": started_at, "retrieved_at": retrieved_at, } pages.append(page) for review in reviews: normalised = normalise_review(review.get("review", "")) fingerprint = sha256_text(normalised) if fingerprint not in wanted or fingerprint in found: continue created_at = datetime.fromtimestamp( review["timestamp_created"], timezone.utc, ) if created_at < cutoff: raise ValueError( "Configured review fell outside the fixed " f"{REVIEW_PARAMS['day_range']}-day window: " f"{fingerprint[:23]}" ) excerpt = wanted[fingerprint]["excerpt"] if excerpt.lower() not in normalised.lower(): raise ValueError( "Configured excerpt no longer matches review fingerprint " f"{fingerprint[:23]}" ) found[fingerprint] = { "review_evidence_key": fingerprint, "excerpt": excerpt, "page_response_sha256": digest, "retrieved_at": retrieved_at, } if len(found) == len(wanted): break if not reviews: break cursor = payload.get("cursor", "") time.sleep(0.25) missing = sorted(set(wanted) - set(found)) if missing: shortened = [value[:23] for value in missing] raise ValueError( f"Missing configured review fingerprints for {appid}: {shortened}" ) return found, pages def main() -> None: annotations_raw = ANNOTATIONS_PATH.read_bytes() annotations = json.loads(annotations_raw) schema = json.loads(SCHEMA_PATH.read_text(encoding="utf-8")) validator = Draft202012Validator( schema, format_checker=FormatChecker(), ) case_study_started_at = utc_now() snapshot_date = case_study_started_at[:10].replace("-", "") source_games: list[dict[str, Any]] = [] profiles: list[dict[str, Any]] = [] for game in annotations["games"]: appid = int(game["appid"]) ( metadata_payload, metadata_digest, metadata_url, metadata_started_at, metadata_retrieved_at, ) = fetch_json( "https://store.steampowered.com/api/appdetails", {"appids": appid, "cc": "US", "l": "english"}, ) wrapper = metadata_payload.get(str(appid), {}) if not wrapper.get("success"): raise ValueError(f"Steam appdetails failed for {appid}") metadata = wrapper["data"] if metadata.get("name") != game["expected_name"]: raise ValueError( f"Expected {game['expected_name']} for {appid}, " f"received {metadata.get('name')}" ) ( player_payload, player_digest, player_url, player_started_at, player_retrieved_at, ) = fetch_json( ( "https://api.steampowered.com/" "ISteamUserStats/GetNumberOfCurrentPlayers/v1/" ), {"appid": appid}, ) player_count = int(player_payload["response"]["player_count"]) selected_reviews, review_pages = collect_reviews( appid, game["reviews"], ) profile_generated_at = utc_now() category_ids = { int(category["id"]) for category in metadata.get("categories", []) } target_platforms = [ name for name, supported in { "windows": metadata.get("platforms", {}).get("windows", False), "macos": metadata.get("platforms", {}).get("mac", False), "linux": metadata.get("platforms", {}).get("linux", False), }.items() if supported ] release_id = f"steam-public-snapshot-{appid}-{snapshot_date}" edition_id = f"urn:case-study:steam-edition:{appid}" price_overview = metadata.get("price_overview") offer: dict[str, Any] = { "seller_id": "steam", "offer_id": f"steam-{appid}-us", "region": "US", "availability": "available", "requires_external_launcher_status": "unknown", "grants_edition_ids": [edition_id], "purchase_url": f"https://store.steampowered.com/app/{appid}/", } if price_overview: offer["price"] = { "amount_minor": int(price_overview["final"]), "currency": price_overview["currency"], } claims = [ functional_claim( appid, release_id, target_platforms, "target_platform.windows", "windows" in target_platforms, metadata_retrieved_at, metadata_digest, metadata_url, ), functional_claim( appid, release_id, target_platforms, "target_platform.macos", "macos" in target_platforms, metadata_retrieved_at, metadata_digest, metadata_url, ), functional_claim( appid, release_id, target_platforms, "target_platform.linux", "linux" in target_platforms, metadata_retrieved_at, metadata_digest, metadata_url, ), functional_claim( appid, release_id, target_platforms, "controller.full_support", metadata.get("controller_support") == "full", metadata_retrieved_at, metadata_digest, metadata_url, ), functional_claim( appid, release_id, target_platforms, "mode.single_player", CATEGORY_SINGLE_PLAYER in category_ids, metadata_retrieved_at, metadata_digest, metadata_url, ), functional_claim( appid, release_id, target_platforms, "multiplayer.online_coop", CATEGORY_ONLINE_COOP in category_ids, metadata_retrieved_at, metadata_digest, metadata_url, ), functional_claim( appid, release_id, target_platforms, "multiplayer.local_coop", CATEGORY_LOCAL_COOP in category_ids, metadata_retrieved_at, metadata_digest, metadata_url, ), ] claims.append( { "claim_id": f"steam-{appid}-current-players", "property": "population.current_players", "claim_class": "aggregate_observation", "asserted_by": "jasondoyle.ie:public-steam-case-study", "value_status": "known", "value": player_count, "unit": "players", "applicability": { "release_ids": [release_id], "storefronts": ["steam"], "target_platforms": target_platforms, "valid_from": player_started_at, "valid_until": player_retrieved_at, }, "evidence": [ { "evidence_id": f"steam-current-players-{appid}", "source_type": "marketplace_aggregate", "source_uri": player_url, "source_snapshot_digest": player_digest, "retrieved_at": player_retrieved_at, "observation_window": { "start": player_started_at, "end": player_retrieved_at, }, "method": ( "Steam GetNumberOfCurrentPlayers point-in-time " "count." ), "method_version": "steam-web-api-v1", "sample_size": player_count, "simulated": False, } ], "uncertainty": [ { "kind": "applicability", "description": ( "Point-in-time connected Steam users only; no " "region, mode or offline-play breakdown." ), } ], "privacy": {"release_mode": "public"}, "status": "active", } ) for annotation in game["experiential_claims"]: claims.append( experiential_claim( annotation, appid, release_id, target_platforms, profile_generated_at, selected_reviews, review_pages, metadata_digest, metadata_url, metadata_retrieved_at, ) ) profile = { "$schema": PROFILE_SCHEMA_URL, "profile_version": "0.1-draft", "generated_at": profile_generated_at, "issuer": { "id": "jasondoyle.ie:public-steam-case-study", "role": "third_party", }, "game_identity": { "canonical_work_id": f"urn:case-study:steam-work:{appid}", "edition_id": edition_id, "storefront_id": "steam", "storefront_product_id": str(appid), "release_id": release_id, }, "offer": offer, "claims": claims, "x_extensions": { "display_name": metadata["name"], "source_snapshot": SNAPSHOT_PATH.name, "case_study_scope": "public-data demonstration", }, } errors = sorted(validator.iter_errors(profile), key=lambda e: list(e.path)) if errors: details = "\n".join( f"{list(error.path)}: {error.message}" for error in errors ) raise ValueError(f"Profile validation failed for {appid}:\n{details}") profiles.append(profile) source_games.append( { "appid": appid, "name": metadata["name"], "store_url": f"https://store.steampowered.com/app/{appid}/", "metadata": { "request_url": metadata_url, "documentation_status": "undocumented", "response_sha256": metadata_digest, "request_started_at": metadata_started_at, "retrieved_at": metadata_retrieved_at, "normalised": { "platforms": metadata.get("platforms", {}), "controller_support": metadata.get("controller_support"), "category_ids": sorted(category_ids), "price_overview": price_overview, }, }, "current_players": { "request_url": player_url, "response_sha256": player_digest, "player_count": player_count, "request_started_at": player_started_at, "retrieved_at": player_retrieved_at, }, "reviews": { "documentation_url": ( "https://partner.steamgames.com/doc/store/getreviews" ), "query_parameters": REVIEW_PARAMS, "pages": review_pages, "selected": [ selected_reviews[review["review_fingerprint"]] for review in game["reviews"] ], }, } ) generated_at = utc_now() source_snapshot = { "snapshot_version": "0.1", "generated_at": generated_at, "annotations_sha256": sha256_bytes(annotations_raw), "privacy_note": ( "No Steam account identifiers or recommendation IDs are retained. " "The short public excerpts and review fingerprints may still be " "linkable to the underlying public reviews and are not anonymised." ), "games": source_games, } profile_set = { "profile_set_version": "0.1", "generated_at": generated_at, "profile_schema": PROFILE_SCHEMA_URL, "source_snapshot": SNAPSHOT_PATH.name, "profiles": profiles, } SNAPSHOT_PATH.write_text( json.dumps(source_snapshot, ensure_ascii=True, indent=2) + "\n", encoding="utf-8", newline="\n", ) PROFILES_PATH.write_text( json.dumps(profile_set, ensure_ascii=True, indent=2) + "\n", encoding="utf-8", newline="\n", ) print( f"Wrote {len(profiles)} profiles and " f"{sum(len(game['reviews']['selected']) for game in source_games)} " "selected review excerpts" ) if __name__ == "__main__": main()