个人数据中心：获取Steam游戏记录

发表于 更新于 分类于 ， ， 阅读次数： Valine：
本文字数： 10k 阅读时长 ≈ 9 分钟

  由于Steam官方提供了Web API接口和文档，因此我们可以很方便的获取自己的游戏记录，并存储到个人数据中心当中。
  本文计划获取自己Steam游戏库中所有游戏的信息以及各时段的游玩时长记录。

获取Token和Steam ID

  要想使用Steam Web API，首先需要准备Token和Steam ID。

  Token的申请页面：https://steamcommunity.com/dev/apikey
  进入申请页面并登录账号以后就可以看到自己的Token和域名。

  打开Steam客户端，点击右上角头像进入“账户明细”就可以查看到自己的Steam ID。

封装API

  对常用的两个接口进行封装，API中除Token和Steam ID两个参数以外，其余参数解释如下：

  • format=json：Steam提供多种数据返回格式，这里设置为json的形式返回。
  • include_appinfo=1：返回游戏的详细信息。
  • include_played_free_games=1：返回的游戏中包括免费游戏。

  更多API和参数的说明请参照官方文档。

SteamWebAPI.py
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
import requests
import json

def Steam_GetOwnedGames(tocken,steamid):
    url = "http://api.steampowered.com/IPlayerService/GetOwnedGames/v0001/?key=%s&steamid=%s&format=json&include_appinfo=1&include_played_free_games=1" % (tocken,steamid)

    payload={}
    headers = {}

    response = requests.request("GET", url, headers=headers, data=payload)

    # print(response.text)

    res=json.loads(response.text)

    return res

def Steam_GetRecentlyPlayedGames(tocken,steamid):
    url = "http://api.steampowered.com/IPlayerService/GetRecentlyPlayedGames/v0001/?key=%s&steamid=%s&format=json" % (tocken,steamid)

    payload={}
    headers = {}

    response = requests.request("GET", url, headers=headers, data=payload)

    # print(response.text)

    res=json.loads(response.text)

    return res

GetOwnedGames返回数据格式

  该接口返回用户已拥有的游戏。

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
{
    "response": {
        "game_count": 381,
        "games": [
            {
                "appid": 10,
                "name": "Counter-Strike",
                "playtime_forever": 0,
                "img_icon_url": "6b0312cda02f5f777efa2f3318c307ff9acafbb5",
                "playtime_windows_forever": 0,
                "playtime_mac_forever": 0,
                "playtime_linux_forever": 0,
                "rtime_last_played": 0,
                "content_descriptorids": [
                    2,
                    5
                ]
            },
            ...
        ]
    }
}
字段 说明
appid 游戏id
name 游戏名称
playtime_forever 全平台累计游戏时长（分）
playtime_windows_forever windows上的游戏时长（分）
playtime_mac_forever mac上的游戏时长（分）
playtime_linux_forever linux上的游戏时长（分）
rtime_last_played 最近一次游戏时间（结束时间），timestamp格式
img_icon_url 游戏图标文件名

  rtime_last_played在使用时可以使用SQL的FROM_UNIXTIME(rtime_last_played)转换为datetime格式。
  可以通过下方等链接来获取游戏图标:
  http://media.steampowered.com/steamcommunity/public/images/apps/{appid}/{img_icon_url}.jpg

GetRecentlyPlayedGames返回数据格式

  该接口返回用户最近两周游玩的游戏记录，但并不能精确到具体时间。

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
{
    "response": {
        "total_count": 2,
        "games": [
            {
                "appid": 270880,
                "name": "American Truck Simulator",
                "playtime_2weeks": 865,
                "playtime_forever": 6672,
                "img_icon_url": "2aec8af4b4b23cb6054d9aea06194621cdd66945",
                "playtime_windows_forever": 0,
                "playtime_mac_forever": 6672,
                "playtime_linux_forever": 0
            },
            ...
        ]
    }
}

采集游戏数据

设计数据表

  我设计了一张维表和一张事实表来分别记录游戏的维度信息和游玩记录。

dim_steam_owned_game

  该表用于存储游戏维度信息，格式与Steam接口返回的格式一致。

字段 数据类型 备注
appid varchar(255) 游戏id
game_name varchar(255) 游戏名称
img_icon_url varchar(255) 游戏图标文件名
playtime_forever int 全平台累计游戏时长（分）
playtime_windows_forever int windows上的游戏时长（分）
playtime_mac_forever int mac上的游戏时长（分）
playtime_linux_forever int linux上的游戏时长（分）
rtime_last_played int 最近一次游戏时间（结束时间），timestamp格式

dwd_steam_game_played_record

  该表用于存储游戏的游玩记录。

字段 数据类型 备注
id int 记录id，自增主键
appid varchar(255) 游戏id
game_name varchar(255) 游戏名称
playtime int 全平台累计游戏时长（分）
playtime_windows int windows上的游戏时长（分）
playtime_mac int mac上的游戏时长（分）
playtime_linux int linux上的游戏时长（分）
rtime_last_played int 最近一次游戏时间（结束时间），timestamp格式
create_time datetime 记录创建时间，默认值CURRENT_TIMESTAMP
update_time datetime 记录修改时间，默认值CURRENT_TIMESTAMP ON UPDATE CURRENT_TIMESTAMP

建表语句：

1
2
3
4
5
6
7
8
9
10
11
12
13
CREATE TABLE `dwd_steam_game_played_record` (
  `id` int NOT NULL AUTO_INCREMENT,
  `appid` varchar(255) COLLATE utf8mb4_general_ci DEFAULT NULL,
  `game_name` varchar(255) COLLATE utf8mb4_general_ci DEFAULT NULL,
  `playtime` int DEFAULT NULL,
  `playtime_windows` int DEFAULT NULL,
  `playtime_mac` int DEFAULT NULL,
  `playtime_linux` int DEFAULT NULL,
  `rtime_last_played` int DEFAULT NULL,
  `create_time` datetime DEFAULT CURRENT_TIMESTAMP,
  `update_time` datetime DEFAULT CURRENT_TIMESTAMP ON UPDATE CURRENT_TIMESTAMP,
  PRIMARY KEY (`id`)
) ENGINE=InnoDB AUTO_INCREMENT=1 DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_general_ci;

数据采集脚本

  Steam_GetOwnedGames函数负责从Steam服务器获取全量最新的游戏信息，然后覆盖dim_steam_owned_game表中的数据。

  SteamDA_GamePlayedRecord在Steam_GetOwnedGames执行完成之后执行，可重复执行。SteamDA_GamePlayedRecord通过执行数据查询SQL，以增量更新的形式，计算出最新的游戏记录并插入dwd_steam_game_played_record表中。
  第一次使用时需要从dim_steam_owned_game导入全量数据进行初始化，作为后续计算的基础，但该部分数据在统计时通常不使用。

SteamDA.py
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
import json
import time
from datetime import datetime,timedelta
import pandas as pd 
from sqlalchemy import create_engine,types
import pymysql

from SteamWebAPI import *

tocken="your_steam_tocken" # 你的steam_tocken
steamid="your_steam_id" # 你的steam_id

db_user="your_user_name" # 用于连接数据库的用户名
db_passwd="your_password" # 用于连接数据库的用户密码
db_dbname="your_database_name" # 数据库中待连接的库名
db_hostname="your_hostname" # 数据库主机地址

def SteamDA_OwnedGames(tocken,steamid):

    results=Steam_GetOwnedGames(tocken,steamid)

    # print(results["response"]["games"][0])

    appid=[]
    game_name=[]
    img_icon_url=[]
    playtime_forever=[]
    playtime_windows_forever=[]
    playtime_mac_forever=[]
    playtime_linux_forever=[]
    rtime_last_played=[]

    for i in results["response"]["games"]:
        appid.append(i["appid"])
        game_name.append(i["name"])
        img_icon_url.append(i["img_icon_url"])
        playtime_forever.append(i["playtime_forever"])
        playtime_windows_forever.append(i["playtime_windows_forever"])
        playtime_mac_forever.append(i["playtime_mac_forever"])
        playtime_linux_forever.append(i["playtime_linux_forever"])
        rtime_last_played.append(i["rtime_last_played"])

    df=pd.DataFrame({"appid":appid,"game_name":game_name,"img_icon_url":img_icon_url,"playtime_forever":playtime_forever,"playtime_windows_forever":playtime_windows_forever,"playtime_mac_forever":playtime_mac_forever,"playtime_linux_forever":playtime_linux_forever,"rtime_last_played":rtime_last_played})

    # print(df)

    con_engine = create_engine('mysql+pymysql://'+db_user+':'+db_passwd+'@'+db_hostname+':3306/'+db_dbname+'?charset=utf8')

    dtype={"appid":types.String(length=255),
            "game_name":types.String(length=255),
            "img_icon_url":types.String(length=255),
            "playtime_forever":types.Integer(),
            "playtime_windows_forever":types.Integer(),
            "playtime_mac_forever":types.Integer(),
            "playtime_linux_forever":types.Integer(),
            "rtime_last_played":types.Integer()
    }

    df.to_sql('dim_steam_owned_game', con_engine, dtype=dtype, if_exists='replace', index = False)

def SteamDA_GamePlayedRecord():

    db = pymysql.connect(
        host=db_hostname, 
        port=3306,
        user=db_user,    #在这里输入用户名
        password=db_passwd,     #在这里输入密码
        charset='utf8mb4',
        database=db_dbname    #指定操作的数据库
        )

    cursor = db.cursor() #创建游标对象

    try:
        sql="""
        INSERT INTO dwd_steam_game_played_record 
        SELECT 
                NULL id,
                t1.appid,
                t1.game_name,
                COALESCE(t1.playtime_forever - t2.playtime,t1.playtime_forever) playtime,
                COALESCE(t1.playtime_windows_forever - t2.playtime_windows,t1.playtime_windows_forever) playtime_windows,
                COALESCE(t1.playtime_mac_forever - t2.playtime_mac,t1.playtime_mac_forever) playtime_mac,
                COALESCE(t1.playtime_linux_forever - t2.playtime_linux,t1.playtime_linux_forever) playtime_linux,
                t1.rtime_last_played,
                NOW() create_time,
                NOW() update_time 
        FROM (
        SELECT 
                    appid,
                    game_name,
                    playtime_forever,
                    playtime_windows_forever,
                    playtime_mac_forever,
                    playtime_linux_forever,
                    rtime_last_played
        FROM dim_steam_owned_game
        ) t1
        LEFT JOIN
        (
        SELECT 
                    appid,
                    SUM(playtime) playtime,
                    SUM(playtime_windows) playtime_windows,
                    SUM(playtime_mac) playtime_mac,
                    SUM(playtime_linux) playtime_linux,
                    MAX(rtime_last_played) rtime_last_played
        FROM dwd_steam_game_played_record
        GROUP BY appid,game_name
        )t2
        ON t1.appid=t2.appid
        WHERE t2.rtime_last_played IS NULL OR t1.rtime_last_played !=t2.rtime_last_played
        """
        # print(sql)
        cursor.execute(sql)
        db.commit()

    except Exception as e:
        print(e)
        db.rollback()  #回滚事务

    finally:
        cursor.close() 
        db.close()  #关闭数据库连接

if __name__ == "__main__":

    bt=time.time()
    print(datetime.now().strftime("%Y-%m-%d %H:%M:%S"),"开始采集Steam游戏数据")

    SteamDA_OwnedGames(tocken,steamid)
    SteamDA_GamePlayedRecord()

    print(datetime.now().strftime("%Y-%m-%d %H:%M:%S"),"Steam游戏数据采集结束",time.time()-bt)

任务调度

  因为访问Steam Web API需要科学上网，我的服务器没有配置代理，所以就只好放在本地电脑运行。目前并没有设置定时任务，而是在每次游戏结束以后等游戏记录上传云端再手动运行脚本进行采集。

无需科学上网，使用腾讯云函数进行任务部署及调度的方案请参照这篇文章

  为方便脚本执行，我使用快捷指令调用sh脚本，然后sh脚本调用数据采集脚本的形式执行。最后将快捷指令放到菜单栏，就可以很方便的在每次游戏结束后执行数据采集任务了。

sh脚本实现

crontab_run01.sh
1
2
3
4
DA01_HOME_PATH=/your_path/data_acquisition_local
DA01_PYTHON_PATH=/your_path/miniconda3/bin/python

$DA01_PYTHON_PATH "$DA01_HOME_PATH/SteamDA.py" >> "$DA01_HOME_PATH/crontab_out.log" 2>&1

快捷指令实现

快捷指令菜单栏效果

定时自动采集

  快捷指令配合Shortery这个软件可以实现定时自动执行。
  在Mac上不推荐使用crontab配置定时任务，因为电脑处于休眠状态时定时任务不会被触发。而Shortery的为防止定时任务在休眠中没有按时执行，可以配置为延迟到下次电脑唤醒时执行。

启动Steam时自动采集

  Shortery的触发器还支持在Steam应用启动或关闭时执行快捷指令。

报表效果展示

参考资料：

  相关代码已托管于GitHub，欢迎Star!

-------- 本文结束 感谢阅读 --------
相关文章
觉得文章写的不错的话，请我喝瓶怡宝吧！😀
  • 本文标题： 个人数据中心：获取Steam游戏记录
  • 本文作者： SiriYang
  • 创建时间： 2023年01月30日 - 10时01分
  • 修改时间： 2023年05月15日 - 21时05分
  • 本文链接： https://blog.siriyang.cn/posts/20230130105406id.html
  • 版权声明： 本博客所有文章除特别声明外，均采用 CC BY-NC-SA 4.0 许可协议。转载请注明出处！
0%