Command line utility for parsing detailed Steam stat information.
You can not select more than 25 topics Topics must start with a letter or number, can include dashes ('-') and can be up to 35 characters long.

301 lines
9.1 KiB

#!/usr/bin/env python3
"""
Command line utility for parsing detailed Steam stat information.
"""
import json
import pathlib
import struct
import time
import urllib.error
import urllib.parse
import urllib.request
import vdf
# Output folder names
FOLDER_ACH = 'achievements'
FOLDER_GLOBAL = 'globalstats'
FOLDER_STAT = 'stats'
# Stats data types
STAT_TYPE_INT = 1
STAT_TYPE_FLOAT = 2
STAT_TYPE_AVGRATE = 3
STAT_TYPE_BITS = 4
# Helpers to pack and unpack binary data
STRUCT_INT32 = struct.Struct('<i')
STRUCT_INT64 = struct.Struct('<q')
STRUCT_FLOAT = struct.Struct('<f')
STRUCT_DOUBLE = struct.Struct('<d')
# Functions to retrieve data from the Steam client's local filesystem cache
def _stat_default(stat):
mappers = {
STAT_TYPE_INT: int,
STAT_TYPE_FLOAT: float,
STAT_TYPE_AVGRATE: float,
STAT_TYPE_BITS: int
}
return mappers[stat['type_int']](stat.get('Default', '0'))
def get_local_stats(base_dir, appid, userid=None):
"""
Parses Steam's appcache for local stat values.
In addition to fetching the stats for the given userid, global stats
will be set to their default value.
Parameters
----------
base_dir : str
Location of Steam install directory
appid: str
The game to fetch stats for
userid : str
The user to fetch stats for (if None, default stat values are used)
Returns
-------
dict
The data contained in the response
"""
local_stats = {FOLDER_ACH: {}, FOLDER_GLOBAL: {}, FOLDER_STAT: {}}
path_name = pathlib.Path(base_dir, 'appcache', 'stats')
# Read in schema file
schema_file = 'UserGameStatsSchema_{}.bin'.format(appid)
with path_name.joinpath(schema_file).open('rb') as fh:
schema = vdf.binary_loads(fh.read())
# Read in cache file
if userid is not None:
cache_file = 'UserGameStats_{}_{}.bin'.format(userid, appid)
with path_name.joinpath(cache_file).open('rb') as fh:
cache = vdf.binary_loads(fh.read())
else:
cache = {'cache': {'crc': 0, 'PendingChanges': 0}}
for stat in schema[appid]['stats'].values():
# Get default and cached stat values
default = _stat_default(stat)
if stat['id'] in cache['cache']:
data = cache['cache'][stat['id']]['data']
# data is read as int32 - convert to float if applicable
if isinstance(default, float):
data = STRUCT_FLOAT.unpack(STRUCT_INT32.pack(data))[0]
else:
data = default
if stat['type_int'] == STAT_TYPE_BITS:
# Store name:achieved of each bit in FOLDER_ACH
for bit in stat['bits'].values():
local_stats[FOLDER_ACH][bit['name']] = (data >> bit['bit']) & 1
else:
# Store name:data in FOLDER_STAT
# If aggregated, also store name:default in FOLDER_GLOBAL
local_stats[FOLDER_STAT][stat['name']] = data
if stat.get('aggregated', '0') == '1':
local_stats[FOLDER_GLOBAL][stat['name']] = default
return local_stats
# Functions to retrieve data from the Steam Web API
def webapi_get(interface, method, version, params):
"""
Fetches data using Steam's public Web API.
Methods that use HTTP POST, or that are only available to
publishers, cannot be called through this function.
Parameters
----------
interface : str
The interface associated with the method call
method: str
The name of the method to call
version : int
The version number associated with the method call
params : dict
The arguments (if any) to pass to the method
Returns
-------
bytes
The data contained in the response
"""
base_url = 'https://api.steampowered.com/'
base_url += '{}/{}/v{}/'.format(interface, method, version)
if len(params) > 0:
base_url += '?{}'.format(urllib.parse.urlencode(params))
with urllib.request.urlopen(base_url) as buf:
return buf.read()
def get_global_stats(appid, names):
"""
Fetches the cumulative totals for global stats.
Parameters
----------
appid : str
The game to fetch stats for
names: list
The names of the stats to fetch totals for
Returns
-------
dict
The data contained in the response
"""
interface = 'ISteamUserStats'
method = 'GetGlobalStatsForGame'
version = 1
global_stats = {}
# Build parameters
count = len(names)
params = {'appid': appid, 'count': count}
for i, name in enumerate(names):
params['name[{}]'.format(i)] = name
try:
resp = webapi_get(interface, method, version, params)
except urllib.error.HTTPError:
pass
else:
resp = json.loads(resp.decode('utf-8')).get('response', {})
if resp.get('result', 0) == 1:
for stat_name, stat_data in resp['globalstats'].items():
global_stats[stat_name] = stat_data['total']
return global_stats
def get_global_stats_history(appid, names, startdate, enddate):
"""
Fetches the daily totals for global stats in the time range given.
The GetGlobalStatsForGame method is limited to retrieving 60 days
of daily totals per call. In the future, this implementation may
be changed to split time ranges of more than 60 days into multiple
API calls.
Parameters
----------
appid : str
The game to fetch stats for
names: list
The names of the stats to fetch totals for
startdate: int
Unix timestamp for the start of the time range
enddate: int
Unix timestamp for the end of the time range
Returns
-------
dict
The data contained in the response
"""
interface = 'ISteamUserStats'
method = 'GetGlobalStatsForGame'
version = 1
global_stats = {}
time.sleep(1)
# Build parameters
count = len(names)
params = {'appid': appid, 'count': count, 'startdate': startdate, 'enddate': enddate}
for i, name in enumerate(names):
params['name[{}]'.format(i)] = name
try:
resp = webapi_get(interface, method, version, params)
except urllib.error.HTTPError:
pass
else:
resp = json.loads(resp.decode('utf-8')).get('response', {})
if resp.get('result', 0) == 1:
for stat_name, stat_data in resp['globalstats'].items():
global_stats[stat_name] = stat_data.get('history')
return global_stats
# Functions to dump raw data
def dump_raw_stats(stats, target_dir=None):
"""
Writes raw stats to files in target_dir.
Parameters
----------
stats : dict
The stats to dump
target_dir : str, optional
Directory to store stats in (defaults to working
directory if None).
"""
if target_dir is None:
target_dir = pathlib.Path.cwd()
# Dump stats to target_dir
for folder in stats:
fp = pathlib.Path(target_dir, folder)
fp.mkdir(parents=True, exist_ok=True)
for statname in stats[folder].keys():
data = stats[folder][statname]
# 4 bytes for GetStat, 8 bytes for GetGlobalStat
if folder == FOLDER_GLOBAL:
mapper = STRUCT_DOUBLE if isinstance(data, float) else STRUCT_INT64
else:
mapper = STRUCT_FLOAT if isinstance(data, float) else STRUCT_INT32
with fp.joinpath(statname).open('wb') as fh:
fh.write(mapper.pack(data))
# Program interface
def main():
import sys
if len(sys.argv) > 2:
# Test fetching local player stats
if len(sys.argv) > 3:
stats = get_local_stats(sys.argv[1], sys.argv[2], sys.argv[3])
else:
stats = get_local_stats(sys.argv[1], sys.argv[2])
# Test fetching global stats Web API use
global_stat_names = [n for n in stats[FOLDER_GLOBAL].keys()]
global_stats = get_global_stats(sys.argv[2], global_stat_names)
for stat_name, stat_data in global_stats.items():
if stat_data is None:
continue
prev_data = stats[FOLDER_GLOBAL][stat_name]
if isinstance(prev_data, float):
stats[FOLDER_GLOBAL][stat_name] = float(stat_data)
else:
stats[FOLDER_GLOBAL][stat_name] = int(stat_data)
# Test dumping raw stats
dump_raw_stats(stats)
else:
print('No input specified.')
def main_history():
import sys
if len(sys.argv) > 4:
stats = get_local_stats(sys.argv[1], sys.argv[2])
global_stat_names = [n for n in stats[FOLDER_GLOBAL].keys()]
# Make multiple calls to get daily totals
global_stat_hist = {}
def divide_names(l, n):
for i in range(0, len(l), n):
yield l[i:i+n]
for partial_names in divide_names(global_stat_names, 4):
global_stats = get_global_stats_history(sys.argv[2], partial_names, sys.argv[3], sys.argv[4])
global_stat_hist.update(global_stats)
print(global_stat_hist)
else:
print('No input specified.')
if __name__ == '__main__':
main()