steam_stat_utils/steam_stat_utils.py

#!/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()