|
- #!/usr/bin/env python3
- """
- Command line utility for parsing detailed Steam stat information.
- """
- import csv
- 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 load_schema(base_dir, appid):
- path_name = pathlib.Path(base_dir, 'appcache', 'stats')
- schema_file = 'UserGameStatsSchema_{}.bin'.format(appid)
- with path_name.joinpath(schema_file).open('rb') as fp:
- return vdf.binary_loads(fp.read())
-
- def load_cache(base_dir, appid, userid=None):
- if userid is None:
- return {'cache': {'crc': 0, 'PendingChanges': 0}}
- path_name = pathlib.Path(base_dir, 'appcache', 'stats')
- cache_file = 'UserGameStats_{}_{}.bin'.format(userid, appid)
- with path_name.joinpath(cache_file).open('rb') as fp:
- return vdf.binary_loads(fp.read())
-
- def get_global_stat_names(appid, schema):
- global_stat_names = []
- for stat in schema[appid]['stats'].values():
- if stat.get('aggregated', '0') == '1':
- global_stat_names.append(stat['name'])
- return global_stat_names
-
- 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: {}}
- schema = load_schema(base_dir, appid)
- cache = load_cache(base_dir, appid, userid)
-
- 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'))
-
- 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 = {}
-
- # 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 as e:
- print('Fetch failed for names ' + names)
- print(e)
- 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')
- else:
- print('Fetch failed for names ' + names)
- print(resp)
-
- 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))
-
- def dump_history(data, startdate, enddate):
- dates = [d for d in range(int(startdate), int(enddate) + 1, 86400)]
- with open('history.csv', 'w', newline='') as csvfile:
- csvwriter = csv.writer(csvfile)
- csvwriter.writerow([''] + dates)
- for stat in data:
- f = [0] * len(dates)
- if data[stat] is not None:
- for record in data[stat]:
- f[dates.index(record['date'])] = int(record['total'])
- csvwriter.writerow([stat] + f)
-
- def dump_stat_defaults(appid, schema):
- type_labels = {
- STAT_TYPE_INT: 'int',
- STAT_TYPE_FLOAT: 'float',
- STAT_TYPE_AVGRATE: 'avgrate'
- }
- with open('stats.txt', 'w') as fp:
- for stat in schema[appid]['stats'].values():
- s_type = type_labels.get(stat['type_int'])
- if s_type is None:
- continue
- s_default = stat.get('Default', '0')
- fp.write('{}={}={}\n'.format(stat['name'], s_type, s_default))
-
- # Program interface
-
- def main():
- import sys
- if len(sys.argv) > 2:
- # Test writing default stat values
- schema = load_schema(sys.argv[1], sys.argv[2])
- dump_stat_defaults(sys.argv[2], schema)
- # 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 = get_global_stat_names(sys.argv[2], schema)
- 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:
- schema = load_schema(sys.argv[1], sys.argv[2])
- global_stat_names = get_global_stat_names(sys.argv[2], schema)
- # 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):
- time.sleep(3)
- global_stats = get_global_stats_history(sys.argv[2], partial_names, sys.argv[3], sys.argv[4])
- global_stat_hist.update(global_stats)
- # Organized in CSV file for easier spreadsheet import
- dump_history(global_stat_hist, sys.argv[3], sys.argv[4])
- else:
- print('No input specified.')
-
- if __name__ == '__main__':
- # Dump player stats and global stat totals. Current usage:
- # steam_stat_utils.py path/to/steam appid userid
- # where userid is the signed-in user's SteamID.
- main()
- # Fetch global stat history. Current usage:
- # steam_stat_utils.py path/to/steam appid startdate enddate
- # where startdate and enddate are integer timestamps.
- #main_history()
|