92 lines
3.0 KiB
Python
92 lines
3.0 KiB
Python
import logging
|
|
|
|
import httpx
|
|
from django.conf import settings
|
|
from django.utils import timezone
|
|
|
|
from ..models import Game
|
|
|
|
logger = logging.getLogger(__name__)
|
|
|
|
# Public store endpoint, no API key needed. Rate limited to ~200 calls / 5 min,
|
|
# which is plenty for a handful of games every few hours.
|
|
APP_DETAILS_URL = "https://store.steampowered.com/api/appdetails"
|
|
STORE_SEARCH_URL = "https://store.steampowered.com/api/storesearch/"
|
|
|
|
|
|
class SteamStoreError(Exception):
|
|
"""Raised when the Steam store didn't return usable details for an app."""
|
|
|
|
|
|
def fetch_app_details(steam_app_id: int) -> dict:
|
|
response = httpx.get(
|
|
APP_DETAILS_URL,
|
|
params={
|
|
"appids": steam_app_id,
|
|
"cc": settings.STEAM_STORE_COUNTRY,
|
|
"l": "en",
|
|
"filters": "basic,price_overview",
|
|
},
|
|
timeout=10,
|
|
)
|
|
response.raise_for_status()
|
|
|
|
entry = (response.json() or {}).get(str(steam_app_id)) or {}
|
|
if not entry.get("success"):
|
|
raise SteamStoreError(f"No store details for app {steam_app_id}")
|
|
|
|
return entry["data"]
|
|
|
|
|
|
def search_store(term: str) -> list[dict]:
|
|
"""Store search results for term, as [{"id": app id, "name": ...}, ...]."""
|
|
response = httpx.get(
|
|
STORE_SEARCH_URL,
|
|
params={"term": term, "cc": settings.STEAM_STORE_COUNTRY, "l": "english"},
|
|
timeout=10,
|
|
)
|
|
response.raise_for_status()
|
|
return (response.json() or {}).get("items") or []
|
|
|
|
|
|
def refresh_steam_game(game: Game) -> None:
|
|
"""Pull the current store info for a Steam game into its steam_* fields."""
|
|
data = fetch_app_details(game.steam_app_id)
|
|
price = data.get("price_overview") or {}
|
|
|
|
game.steam_is_free = bool(data.get("is_free"))
|
|
game.steam_price_initial = price.get("initial")
|
|
game.steam_price_final = price.get("final")
|
|
game.steam_discount_percent = price.get("discount_percent") or 0
|
|
game.steam_currency = price.get("currency") or ""
|
|
game.steam_short_description = data.get("short_description") or ""
|
|
game.steam_header_image = data.get("header_image") or ""
|
|
game.steam_info_updated_at = timezone.now()
|
|
game.save(
|
|
update_fields=[
|
|
"steam_is_free",
|
|
"steam_price_initial",
|
|
"steam_price_final",
|
|
"steam_discount_percent",
|
|
"steam_currency",
|
|
"steam_short_description",
|
|
"steam_header_image",
|
|
"steam_info_updated_at",
|
|
"updated_at",
|
|
]
|
|
)
|
|
|
|
|
|
def refresh_steam_games(game_ids: list[int] | None = None) -> None:
|
|
"""Refresh every Steam game (or just game_ids). A failing game keeps its
|
|
previous info and doesn't stop the others."""
|
|
games = Game.objects.filter(steam_app_id__isnull=False)
|
|
if game_ids is not None:
|
|
games = games.filter(id__in=game_ids)
|
|
|
|
for game in games:
|
|
try:
|
|
refresh_steam_game(game)
|
|
except (httpx.HTTPError, ValueError, SteamStoreError) as exc:
|
|
logger.warning("Steam store refresh failed for %s: %s", game, exc)
|