Add Steam library import via OpenID
All checks were successful
ci/woodpecker/push/woodpecker Pipeline was successful

- Steam OpenID flow: user authenticates with Steam, we get their Steam ID
- Server-side API key fetches their owned games with playtime
- Import page shows full library, marks already-imported games
- Imported games land in backlog as GAMES items with hours_played set
- STEAM_API_KEY env var plumbed into both prod and dev containers

Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>
This commit is contained in:
2026-03-31 22:38:28 +03:00
parent 8f34d9388f
commit a4c31bf40b
6 changed files with 382 additions and 1 deletions

50
backlogger/steam.py Normal file
View File

@@ -0,0 +1,50 @@
import re
import requests
from urllib.parse import urlencode
STEAM_OPENID_URL = 'https://steamcommunity.com/openid/login'
STEAM_API_BASE = 'https://api.steampowered.com'
def build_auth_url(return_to, realm):
params = {
'openid.ns': 'http://specs.openid.net/auth/2.0',
'openid.mode': 'checkid_setup',
'openid.return_to': return_to,
'openid.realm': realm,
'openid.identity': 'http://specs.openid.net/auth/2.0/identifier_select',
'openid.claimed_id': 'http://specs.openid.net/auth/2.0/identifier_select',
}
return f"{STEAM_OPENID_URL}?{urlencode(params)}"
def verify_and_get_steam_id(params):
"""Verify OpenID assertion with Steam. Returns steam64 id string or None."""
verify_params = {k: v for k, v in params.items()}
verify_params['openid.mode'] = 'check_authentication'
try:
resp = requests.post(STEAM_OPENID_URL, data=verify_params, timeout=10)
resp.raise_for_status()
except requests.RequestException:
return None
if 'is_valid:true' not in resp.text:
return None
claimed_id = params.get('openid.claimed_id', '')
match = re.search(r'/openid/id/(\d+)$', claimed_id)
return match.group(1) if match else None
def get_owned_games(api_key, steam_id):
"""Return list of games sorted by playtime desc. Raises on API error."""
url = f"{STEAM_API_BASE}/IPlayerService/GetOwnedGames/v1/"
params = {
'key': api_key,
'steamid': steam_id,
'include_appinfo': 'true',
'include_played_free_games': 'true',
'format': 'json',
}
resp = requests.get(url, params=params, timeout=15)
resp.raise_for_status()
games = resp.json().get('response', {}).get('games', [])
return sorted(games, key=lambda g: g.get('playtime_forever', 0), reverse=True)