Boris
a4c31bf40b
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>
51 lines
1.8 KiB
Python
51 lines
1.8 KiB
Python
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)
|