polylan/polylan-submitter
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

feat(games): add steam pages with steam price/reduc

Browse Source
This commit is contained in:
Legrems
2026-10-16 01:03:46 +02:00
parent 919e611895
commit 27d3c08a72
24 changed files with 779 additions and 6 deletions
Download Patch File Download Diff File Expand all files Collapse all files
polylan_submitter/games/admin.py
+24 -1
View File
@@ -8,6 +8,7 @@ from django_object_actions import (
from submissions.tasks import sync_event_game_scopes_task
from .models import EventGame, Game
from .tasks import refresh_steam_games_task
class EventGameInline(admin.TabularInline):
@@ -19,8 +20,30 @@ class EventGameInline(admin.TabularInline):
class GameAdmin(admin.ModelAdmin):
list_display = ["name", "steam_app_id", "updated_at"]
search_fields = ["name", "steam_app_id"]
readonly_fields = ["created_at", "updated_at"]
readonly_fields = [
"created_at",
"updated_at",
"steam_price_initial",
"steam_price_final",
"steam_discount_percent",
"steam_currency",
"steam_is_free",
"steam_short_description",
"steam_header_image",
"steam_info_updated_at",
]
inlines = [EventGameInline]
actions = ["refresh_steam_info"]
@admin.action(description="Refresh Steam store info")
def refresh_steam_info(self, request, queryset):
game_ids = list(
queryset.filter(steam_app_id__isnull=False).values_list("id", flat=True)
)
refresh_steam_games_task.enqueue(game_ids=game_ids, reschedule=False)
self.message_user(
request, f"Queued Steam store refresh for {len(game_ids)} game(s)."
)
@admin.register(EventGame)
polylan_submitter/games/api.py
+12 -1
View File
@@ -3,7 +3,8 @@ from typing import List
from ninja import Router
from .models import Game
from .schemas import GameOut
from .schemas import GameOut, SteamGameOut
from .tasks import schedule_steam_refresh
router = Router()
event_router = Router()
@@ -17,3 +18,13 @@ def list_games(request):
@event_router.get("/events/{event_path}/games", response=List[GameOut])
def list_event_games(request, event_path: str):
return Game.objects.enabled_for_event(request.event)
@event_router.get("/events/{event_path}/games/steam", response=List[SteamGameOut])
def list_event_steam_games(request, event_path: str):
"""The event's Steam games with their store info (price, discount, blurb)."""
# Safety net in case the periodic refresh loop isn't running (no-op otherwise).
schedule_steam_refresh()
return Game.objects.enabled_for_event(request.event).filter(
steam_app_id__isnull=False
)
polylan_submitter/games/management/__init__.py
View File
polylan_submitter/games/management/commands/__init__.py
View File
polylan_submitter/games/management/commands/schedule_steam_refresh.py
+13
View File
@@ -0,0 +1,13 @@
from django.core.management.base import BaseCommand
from games.tasks import schedule_steam_refresh
class Command(BaseCommand):
help = "Start the periodic Steam store refresh loop (no-op if already running)"
def handle(self, *args, **options):
if schedule_steam_refresh():
self.stdout.write(self.style.SUCCESS("Steam store refresh scheduled"))
else:
self.stdout.write("Steam store refresh already scheduled")
polylan_submitter/games/migrations/0010_game_steam_store_info.py
+52
View File
@@ -0,0 +1,52 @@
# Generated by Django 6.0.5 on 2026-10-15 22:27
from django.db import migrations, models
class Migration(migrations.Migration):
dependencies = [
("games", "0009_game_description"),
]
operations = [
migrations.AddField(
model_name="game",
name="steam_currency",
field=models.CharField(blank=True, default="", max_length=3),
),
migrations.AddField(
model_name="game",
name="steam_discount_percent",
field=models.PositiveSmallIntegerField(default=0),
),
migrations.AddField(
model_name="game",
name="steam_header_image",
field=models.URLField(blank=True, default="", max_length=500),
),
migrations.AddField(
model_name="game",
name="steam_info_updated_at",
field=models.DateTimeField(blank=True, null=True),
),
migrations.AddField(
model_name="game",
name="steam_is_free",
field=models.BooleanField(default=False),
),
migrations.AddField(
model_name="game",
name="steam_price_final",
field=models.PositiveIntegerField(blank=True, null=True),
),
migrations.AddField(
model_name="game",
name="steam_price_initial",
field=models.PositiveIntegerField(blank=True, null=True),
),
migrations.AddField(
model_name="game",
name="steam_short_description",
field=models.TextField(blank=True, default=""),
),
]
polylan_submitter/games/models.py
+11
View File
@@ -16,6 +16,17 @@ class Game(models.Model):
image = models.URLField(null=True, blank=True)
description = models.TextField(blank=True, default="")
# Steam store info, refreshed in the background (see games.tasks).
# Prices are in cents, in steam_currency.
steam_price_initial = models.PositiveIntegerField(null=True, blank=True)
steam_price_final = models.PositiveIntegerField(null=True, blank=True)
steam_discount_percent = models.PositiveSmallIntegerField(default=0)
steam_currency = models.CharField(max_length=3, blank=True, default="")
steam_is_free = models.BooleanField(default=False)
steam_short_description = models.TextField(blank=True, default="")
steam_header_image = models.URLField(max_length=500, blank=True, default="")
steam_info_updated_at = models.DateTimeField(null=True, blank=True)
created_at = models.DateTimeField(auto_now_add=True)
updated_at = models.DateTimeField(auto_now=True)
polylan_submitter/games/schemas.py
+60
View File
@@ -1,3 +1,4 @@
from datetime import datetime
from typing import Optional
from ninja import Schema
@@ -11,3 +12,62 @@ class GameOut(Schema):
custom_path: bool
image: Optional[str] = None
description: str
@staticmethod
def resolve_description(obj) -> str:
# The admin one wins, Steam games fall back to their store blurb
return obj.description or obj.steam_short_description
class SteamGameOut(Schema):
id: int
name: str
steam_app_id: int
store_url: str
header_image: str
short_description: str
is_free: bool
currency: str
price_initial: Optional[int] = None
price_final: Optional[int] = None
discount_percent: int
updated_at: Optional[datetime] = None
@staticmethod
def resolve_store_url(obj) -> str:
return f"https://store.steampowered.com/app/{obj.steam_app_id}"
@staticmethod
def resolve_header_image(obj) -> str:
return (
obj.steam_header_image
or f"https://cdn.akamai.steamstatic.com/steam/apps/{obj.steam_app_id}/header.jpg"
)
@staticmethod
def resolve_short_description(obj) -> str:
return obj.steam_short_description
@staticmethod
def resolve_is_free(obj) -> bool:
return obj.steam_is_free
@staticmethod
def resolve_currency(obj) -> str:
return obj.steam_currency
@staticmethod
def resolve_price_initial(obj) -> Optional[int]:
return obj.steam_price_initial
@staticmethod
def resolve_price_final(obj) -> Optional[int]:
return obj.steam_price_final
@staticmethod
def resolve_discount_percent(obj) -> int:
return obj.steam_discount_percent
@staticmethod
def resolve_updated_at(obj) -> Optional[datetime]:
return obj.steam_info_updated_at
polylan_submitter/games/services/__init__.py
View File
polylan_submitter/games/services/steam_store.py
+79
View File
@@ -0,0 +1,79 @@
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"
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 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)
polylan_submitter/games/tasks.py
+52
View File
@@ -0,0 +1,52 @@
from datetime import timedelta
from django.conf import settings
from django.core.cache import cache
from django.tasks import task
from django.utils import timezone
from .services.steam_store import refresh_steam_games
# Set while a periodic refresh is queued, so there's only ever one loop running,
# whatever the number of workers/containers trying to start it. It expires on
# its own if the loop dies (worker killed mid-job, Redis queue flushed, ...),
# letting the next schedule_steam_refresh() start a new one.
STEAM_REFRESH_GUARD_KEY = "games:steam_refresh:scheduled"
def steam_refresh_interval() -> timedelta:
return timedelta(minutes=settings.STEAM_STORE_REFRESH_MINUTES)
def _guard_timeout() -> int:
return int(steam_refresh_interval().total_seconds() * 2)
def schedule_steam_refresh() -> bool:
"""Start the periodic refresh loop right away, unless one is already queued.
Safe to call from anywhere, any number of times."""
if not cache.add(STEAM_REFRESH_GUARD_KEY, 1, timeout=_guard_timeout()):
return False
refresh_steam_games_task.enqueue()
return True
@task
def refresh_steam_games_task(
game_ids: list[int] | None = None, reschedule: bool = True
):
"""Refresh Steam store info, then re-enqueue itself every
STEAM_STORE_REFRESH_MINUTES (same self-rescheduling idea as
sync_event_game_scopes_task, no external scheduler needed).
reschedule=False is a one-off run (e.g. the admin action) that leaves the
periodic loop alone."""
try:
refresh_steam_games(game_ids)
finally:
if reschedule:
cache.set(STEAM_REFRESH_GUARD_KEY, 1, timeout=_guard_timeout())
refresh_steam_games_task.using(
run_after=timezone.now() + steam_refresh_interval()
).enqueue()
polylan_submitter/games/tests.py
+157
View File
@@ -0,0 +1,157 @@
from unittest import mock
import httpx
from django.core.cache import cache
from django.test import TestCase, override_settings
from accounts.models import CustomUser
from events.models import Event
from .models import EventGame, Game
from .services.steam_store import refresh_steam_games
from .tasks import refresh_steam_games_task, schedule_steam_refresh
LOCMEM_CACHE = {"default": {"BACKEND": "django.core.cache.backends.locmem.LocMemCache"}}
def steam_response(app_id, data, success=True):
return httpx.Response(
200,
json={str(app_id): {"success": success, "data": data}},
request=httpx.Request("GET", "https://store.steampowered.com"),
)
class RefreshSteamGamesTests(TestCase):
def setUp(self):
self.game = Game.objects.create(name="Noita", path="noita", steam_app_id=881100)
@mock.patch("games.services.steam_store.httpx.get")
def test_stores_price_and_discount(self, get):
get.return_value = steam_response(
881100,
{
"is_free": False,
"short_description": "Every pixel is simulated.",
"header_image": "https://example.com/header.jpg",
"price_overview": {
"currency": "CHF",
"initial": 1950,
"final": 780,
"discount_percent": 60,
},
},
)
refresh_steam_games()
self.game.refresh_from_db()
self.assertEqual(self.game.steam_price_initial, 1950)
self.assertEqual(self.game.steam_price_final, 780)
self.assertEqual(self.game.steam_discount_percent, 60)
self.assertEqual(self.game.steam_currency, "CHF")
self.assertEqual(self.game.steam_short_description, "Every pixel is simulated.")
self.assertIsNotNone(self.game.steam_info_updated_at)
@mock.patch("games.services.steam_store.httpx.get")
def test_free_game_has_no_price(self, get):
get.return_value = steam_response(881100, {"is_free": True})
refresh_steam_games()
self.game.refresh_from_db()
self.assertTrue(self.game.steam_is_free)
self.assertIsNone(self.game.steam_price_final)
@mock.patch("games.services.steam_store.httpx.get")
def test_failure_keeps_previous_info(self, get):
Game.objects.filter(pk=self.game.pk).update(steam_price_final=999)
get.side_effect = httpx.ConnectError("down")
refresh_steam_games()
self.game.refresh_from_db()
self.assertEqual(self.game.steam_price_final, 999)
@override_settings(CACHES=LOCMEM_CACHE)
class ScheduleSteamRefreshTests(TestCase):
def setUp(self):
cache.clear()
@mock.patch("games.tasks.refresh_steam_games_task")
def test_only_one_loop_is_scheduled(self, task):
self.assertTrue(schedule_steam_refresh())
self.assertFalse(schedule_steam_refresh())
task.enqueue.assert_called_once_with()
@mock.patch("games.tasks.refresh_steam_games")
def test_task_reschedules_itself_even_on_failure(self, refresh):
refresh.side_effect = RuntimeError("boom")
with mock.patch("games.tasks.refresh_steam_games_task") as patched:
with self.assertRaises(RuntimeError):
refresh_steam_games_task.call()
patched.using.return_value.enqueue.assert_called_once_with()
@mock.patch("games.tasks.refresh_steam_games")
def test_one_off_run_does_not_reschedule(self, refresh):
with mock.patch("games.tasks.refresh_steam_games_task") as patched:
refresh_steam_games_task.call(game_ids=[1], reschedule=False)
refresh.assert_called_once_with([1])
patched.using.assert_not_called()
@override_settings(CACHES=LOCMEM_CACHE)
class EventSteamGamesApiTests(TestCase):
def setUp(self):
self.event = Event.objects.create(name="Event", path="event")
noita = Game.objects.create(
name="Noita", path="noita", steam_app_id=881100, steam_price_final=780
)
clock = Game.objects.create(name="Clock", path="clock", custom_path=True)
EventGame.objects.create(event=self.event, game=noita)
EventGame.objects.create(event=self.event, game=clock)
user = CustomUser.objects.create_user(username="user", password="x")
self.client.force_login(user)
@mock.patch("games.api.schedule_steam_refresh")
def test_lists_only_steam_games_with_store_info(self, schedule):
response = self.client.get("/api/events/event/games/steam")
self.assertEqual(response.status_code, 200)
[game] = response.json()
self.assertEqual(game["name"], "Noita")
self.assertEqual(game["price_final"], 780)
self.assertEqual(game["store_url"], "https://store.steampowered.com/app/881100")
schedule.assert_called_once()
class EventGamesDescriptionTests(TestCase):
def setUp(self):
self.event = Event.objects.create(name="Event", path="event")
self.user = CustomUser.objects.create_user(username="user", password="x")
self.client.force_login(self.user)
def test_steam_description_is_fallback_for_admin_one(self):
steam = Game.objects.create(
name="Noita",
path="noita",
steam_app_id=881100,
steam_short_description="From Steam",
)
custom = Game.objects.create(
name="Opus",
path="opus",
steam_app_id=558990,
description="From admin",
steam_short_description="From Steam",
)
EventGame.objects.create(event=self.event, game=steam)
EventGame.objects.create(event=self.event, game=custom)
response = self.client.get("/api/events/event/games")
descriptions = {g["name"]: g["description"] for g in response.json()}
self.assertEqual(descriptions, {"Noita": "From Steam", "Opus": "From admin"})