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.

353 lines
11 KiB

  1. #!/usr/bin/env python3
  2. """
  3. Command line utility for parsing detailed Steam stat information.
  4. """
  5. import csv
  6. import json
  7. import pathlib
  8. import struct
  9. import time
  10. import urllib.error
  11. import urllib.parse
  12. import urllib.request
  13. import vdf
  14. # Output folder names
  15. FOLDER_ACH = 'achievements'
  16. FOLDER_GLOBAL = 'globalstats'
  17. FOLDER_STAT = 'stats'
  18. # Stats data types
  19. STAT_TYPE_INT = 1
  20. STAT_TYPE_FLOAT = 2
  21. STAT_TYPE_AVGRATE = 3
  22. STAT_TYPE_BITS = 4
  23. # Helpers to pack and unpack binary data
  24. STRUCT_INT32 = struct.Struct('<i')
  25. STRUCT_INT64 = struct.Struct('<q')
  26. STRUCT_FLOAT = struct.Struct('<f')
  27. STRUCT_DOUBLE = struct.Struct('<d')
  28. # Functions to retrieve data from the Steam client's local filesystem cache
  29. def load_schema(base_dir, appid):
  30. path_name = pathlib.Path(base_dir, 'appcache', 'stats')
  31. schema_file = 'UserGameStatsSchema_{}.bin'.format(appid)
  32. with path_name.joinpath(schema_file).open('rb') as fp:
  33. return vdf.binary_loads(fp.read())
  34. def load_cache(base_dir, appid, userid=None):
  35. if userid is None:
  36. return {'cache': {'crc': 0, 'PendingChanges': 0}}
  37. path_name = pathlib.Path(base_dir, 'appcache', 'stats')
  38. cache_file = 'UserGameStats_{}_{}.bin'.format(userid, appid)
  39. with path_name.joinpath(cache_file).open('rb') as fp:
  40. return vdf.binary_loads(fp.read())
  41. def get_global_stat_names(appid, schema):
  42. global_stat_names = []
  43. for stat in schema[appid]['stats'].values():
  44. if stat.get('aggregated', '0') == '1':
  45. global_stat_names.append(stat['name'])
  46. return global_stat_names
  47. def get_local_stats(base_dir, appid, userid=None):
  48. """
  49. Parses Steam's appcache for local stat values.
  50. In addition to fetching the stats for the given userid, global stats
  51. will be set to their default value.
  52. Parameters
  53. ----------
  54. base_dir : str
  55. Location of Steam install directory
  56. appid: str
  57. The game to fetch stats for
  58. userid : str
  59. The user to fetch stats for (if None, default stat values are used)
  60. Returns
  61. -------
  62. dict
  63. The data contained in the response
  64. """
  65. local_stats = {FOLDER_ACH: {}, FOLDER_GLOBAL: {}, FOLDER_STAT: {}}
  66. schema = load_schema(base_dir, appid)
  67. cache = load_cache(base_dir, appid, userid)
  68. def _stat_default(stat):
  69. mappers = {
  70. STAT_TYPE_INT: int,
  71. STAT_TYPE_FLOAT: float,
  72. STAT_TYPE_AVGRATE: float,
  73. STAT_TYPE_BITS: int
  74. }
  75. return mappers[stat['type_int']](stat.get('Default', '0'))
  76. for stat in schema[appid]['stats'].values():
  77. # Get default and cached stat values
  78. default = _stat_default(stat)
  79. if stat['id'] in cache['cache']:
  80. data = cache['cache'][stat['id']]['data']
  81. # data is read as int32 - convert to float if applicable
  82. if isinstance(default, float):
  83. data = STRUCT_FLOAT.unpack(STRUCT_INT32.pack(data))[0]
  84. else:
  85. data = default
  86. if stat['type_int'] == STAT_TYPE_BITS:
  87. # Store name:achieved of each bit in FOLDER_ACH
  88. for bit in stat['bits'].values():
  89. local_stats[FOLDER_ACH][bit['name']] = (data >> bit['bit']) & 1
  90. else:
  91. # Store name:data in FOLDER_STAT
  92. # If aggregated, also store name:default in FOLDER_GLOBAL
  93. local_stats[FOLDER_STAT][stat['name']] = data
  94. if stat.get('aggregated', '0') == '1':
  95. local_stats[FOLDER_GLOBAL][stat['name']] = default
  96. return local_stats
  97. # Functions to retrieve data from the Steam Web API
  98. def webapi_get(interface, method, version, params):
  99. """
  100. Fetches data using Steam's public Web API.
  101. Methods that use HTTP POST, or that are only available to
  102. publishers, cannot be called through this function.
  103. Parameters
  104. ----------
  105. interface : str
  106. The interface associated with the method call
  107. method: str
  108. The name of the method to call
  109. version : int
  110. The version number associated with the method call
  111. params : dict
  112. The arguments (if any) to pass to the method
  113. Returns
  114. -------
  115. bytes
  116. The data contained in the response
  117. """
  118. base_url = 'https://api.steampowered.com/'
  119. base_url += '{}/{}/v{}/'.format(interface, method, version)
  120. if len(params) > 0:
  121. base_url += '?{}'.format(urllib.parse.urlencode(params))
  122. with urllib.request.urlopen(base_url) as buf:
  123. return buf.read()
  124. def get_global_stats(appid, names):
  125. """
  126. Fetches the cumulative totals for global stats.
  127. Parameters
  128. ----------
  129. appid : str
  130. The game to fetch stats for
  131. names: list
  132. The names of the stats to fetch totals for
  133. Returns
  134. -------
  135. dict
  136. The data contained in the response
  137. """
  138. interface = 'ISteamUserStats'
  139. method = 'GetGlobalStatsForGame'
  140. version = 1
  141. global_stats = {}
  142. # Build parameters
  143. count = len(names)
  144. params = {'appid': appid, 'count': count}
  145. for i, name in enumerate(names):
  146. params['name[{}]'.format(i)] = name
  147. try:
  148. resp = webapi_get(interface, method, version, params)
  149. except urllib.error.HTTPError:
  150. pass
  151. else:
  152. resp = json.loads(resp.decode('utf-8')).get('response', {})
  153. if resp.get('result', 0) == 1:
  154. for stat_name, stat_data in resp['globalstats'].items():
  155. global_stats[stat_name] = stat_data['total']
  156. return global_stats
  157. def get_global_stats_history(appid, names, startdate, enddate):
  158. """
  159. Fetches the daily totals for global stats in the time range given.
  160. The GetGlobalStatsForGame method is limited to retrieving 60 days
  161. of daily totals per call. In the future, this implementation may
  162. be changed to split time ranges of more than 60 days into multiple
  163. API calls.
  164. Parameters
  165. ----------
  166. appid : str
  167. The game to fetch stats for
  168. names: list
  169. The names of the stats to fetch totals for
  170. startdate: int
  171. Unix timestamp for the start of the time range
  172. enddate: int
  173. Unix timestamp for the end of the time range
  174. Returns
  175. -------
  176. dict
  177. The data contained in the response
  178. """
  179. interface = 'ISteamUserStats'
  180. method = 'GetGlobalStatsForGame'
  181. version = 1
  182. global_stats = {}
  183. # Build parameters
  184. count = len(names)
  185. params = {'appid': appid, 'count': count, 'startdate': startdate, 'enddate': enddate}
  186. for i, name in enumerate(names):
  187. params['name[{}]'.format(i)] = name
  188. try:
  189. resp = webapi_get(interface, method, version, params)
  190. except urllib.error.HTTPError as e:
  191. print('Fetch failed for names ' + names)
  192. print(e)
  193. else:
  194. resp = json.loads(resp.decode('utf-8')).get('response', {})
  195. if resp.get('result', 0) == 1:
  196. for stat_name, stat_data in resp['globalstats'].items():
  197. global_stats[stat_name] = stat_data.get('history')
  198. else:
  199. print('Fetch failed for names ' + names)
  200. print(resp)
  201. return global_stats
  202. # Functions to dump raw data
  203. def dump_raw_stats(stats, target_dir=None):
  204. """
  205. Writes raw stats to files in target_dir.
  206. Parameters
  207. ----------
  208. stats : dict
  209. The stats to dump
  210. target_dir : str, optional
  211. Directory to store stats in (defaults to working
  212. directory if None).
  213. """
  214. if target_dir is None:
  215. target_dir = pathlib.Path.cwd()
  216. # Dump stats to target_dir
  217. for folder in stats:
  218. fp = pathlib.Path(target_dir, folder)
  219. fp.mkdir(parents=True, exist_ok=True)
  220. for statname in stats[folder].keys():
  221. data = stats[folder][statname]
  222. # 4 bytes for GetStat, 8 bytes for GetGlobalStat
  223. if folder == FOLDER_GLOBAL:
  224. mapper = STRUCT_DOUBLE if isinstance(data, float) else STRUCT_INT64
  225. else:
  226. mapper = STRUCT_FLOAT if isinstance(data, float) else STRUCT_INT32
  227. with fp.joinpath(statname).open('wb') as fh:
  228. fh.write(mapper.pack(data))
  229. def dump_history(data, startdate, enddate):
  230. dates = [d for d in range(int(startdate), int(enddate) + 1, 86400)]
  231. with open('history.csv', 'w', newline='') as csvfile:
  232. csvwriter = csv.writer(csvfile)
  233. csvwriter.writerow([''] + dates)
  234. for stat in data:
  235. f = [0] * len(dates)
  236. if data[stat] is not None:
  237. for record in data[stat]:
  238. f[dates.index(record['date'])] = int(record['total'])
  239. csvwriter.writerow([stat] + f)
  240. def dump_stat_defaults(appid, schema):
  241. type_labels = {
  242. STAT_TYPE_INT: 'int',
  243. STAT_TYPE_FLOAT: 'float',
  244. STAT_TYPE_AVGRATE: 'avgrate'
  245. }
  246. with open('stats.txt', 'w') as fp:
  247. for stat in schema[appid]['stats'].values():
  248. s_type = type_labels.get(stat['type_int'])
  249. if s_type is None:
  250. continue
  251. s_default = stat.get('Default', '0')
  252. fp.write('{}={}={}\n'.format(stat['name'], s_type, s_default))
  253. # Program interface
  254. def main():
  255. import sys
  256. if len(sys.argv) > 2:
  257. # Test writing default stat values
  258. schema = load_schema(sys.argv[1], sys.argv[2])
  259. dump_stat_defaults(sys.argv[2], schema)
  260. # Test fetching local player stats
  261. if len(sys.argv) > 3:
  262. stats = get_local_stats(sys.argv[1], sys.argv[2], sys.argv[3])
  263. else:
  264. stats = get_local_stats(sys.argv[1], sys.argv[2])
  265. # Test fetching global stats Web API use
  266. global_stat_names = get_global_stat_names(sys.argv[2], schema)
  267. global_stats = get_global_stats(sys.argv[2], global_stat_names)
  268. for stat_name, stat_data in global_stats.items():
  269. if stat_data is None:
  270. continue
  271. prev_data = stats[FOLDER_GLOBAL][stat_name]
  272. if isinstance(prev_data, float):
  273. stats[FOLDER_GLOBAL][stat_name] = float(stat_data)
  274. else:
  275. stats[FOLDER_GLOBAL][stat_name] = int(stat_data)
  276. # Test dumping raw stats
  277. dump_raw_stats(stats)
  278. else:
  279. print('No input specified.')
  280. def main_history():
  281. import sys
  282. if len(sys.argv) > 4:
  283. schema = load_schema(sys.argv[1], sys.argv[2])
  284. global_stat_names = get_global_stat_names(sys.argv[2], schema)
  285. # Make multiple calls to get daily totals
  286. global_stat_hist = {}
  287. def divide_names(l, n):
  288. for i in range(0, len(l), n):
  289. yield l[i:i+n]
  290. for partial_names in divide_names(global_stat_names, 4):
  291. time.sleep(3)
  292. global_stats = get_global_stats_history(sys.argv[2], partial_names, sys.argv[3], sys.argv[4])
  293. global_stat_hist.update(global_stats)
  294. # Organized in CSV file for easier spreadsheet import
  295. dump_history(global_stat_hist, sys.argv[3], sys.argv[4])
  296. else:
  297. print('No input specified.')
  298. if __name__ == '__main__':
  299. # Dump player stats and global stat totals. Current usage:
  300. # steam_stat_utils.py path/to/steam appid userid
  301. # where userid is the signed-in user's SteamID.
  302. main()
  303. # Fetch global stat history. Current usage:
  304. # steam_stat_utils.py path/to/steam appid startdate enddate
  305. # where startdate and enddate are integer timestamps.
  306. #main_history()