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

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

更新于20250425：之前由于api接口更新，导致该方案失效。感谢Colin提供的参考代码，帮我把问题修复，现在又可以使用了！Colin的代码要比我的更为成熟完善一些，github链接已附在文章末尾参考资料中，欢迎大家学习Star。

  Switch是我目前的主力游戏平台，在实现了Steam游戏数据采集以后，一直有实现Switch数据采集的执念。相较于可以使用Steam官方的API接口，任天堂官方并没有向普通用户开放接口，使得Switch游戏数据的获取极为困难。但因为Jump等一些应用平台上都上线了Switch账户绑定与游戏时长统计的功能，使我相信一定是有办法可以实现Switch数据采集的。
  起初想在任天堂官网找到相关的开发文档，但是并没有收获，反倒是发现目前网上的一些可行方案都是社区通过逆向工程解析出的Web API，不过流程相当复杂。好在最终找到了一篇基于python实现的文章，得以将流程跑通。
  因为我都是使用日服账号进行游戏，并开通了任天堂会员以同步游戏数据到云服务器，所以我的所有游玩记录都集中在日服账号上。本文计划获取自己Switch日服账户中所有游戏的信息以及各时段的游玩时长记录，并存储到个人数据中心当中。

封装接口

  Switch接口调用的流程如下：

  由于接口调用流程步骤较多且复杂，我将其封装为三个函数方便后期使用：

  • NS_GetSessionToken：获取一个有效期两年的session_token
  • NS_GetAccessToken：基于session_token获取一个有效期15分钟的access_token
  • NS_GetPlayHistory：基于access_token获取游戏记录信息

  直接运行SwitchWebAPI.py调用NS_GetSessionToken函数，该函数首先会返回一个授权页面链接，用户访问该链接登陆自己的NS账号，然后复制出授权重定向链接，粘贴回命令行窗口并回车输入，最终函数会返回session_token。

  将第一步获取的session_token保存好，则可反复调用剩下两个接口获取游戏记录。

SwitchWebAPI.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
import requests
import json
import base64
import hashlib
import re
import sys
import os

def NS_GetSessionToken(client_id,ua):

    '''Logs in to a Nintendo Account and returns a session_token.'''
    session = requests.Session()

    auth_code_verifier = base64.urlsafe_b64encode(os.urandom(32))
    auth_cv_hash = hashlib.sha256()
    auth_cv_hash.update(auth_code_verifier.replace(b"=", b""))
    auth_code_challenge = base64.urlsafe_b64encode(auth_cv_hash.digest())

    app_head = {
        'Host':                      'accounts.nintendo.com',
        'Connection':                'keep-alive',
        'Cache-Control':             'max-age=0',
        'Upgrade-Insecure-Requests': '1',
        'User-Agent':                'Mozilla/5.0 (Nintendo Switch; WebApplet) AppleWebKit/609.4 (KHTML, like Gecko) NF/6.0.2.15.4 NintendoBrowser/5.1.0.22433',
        'Accept':                    'text/html,application/xhtml+xml,application/xml;q=0.9,image/webp,*/*;q=0.8',
        'DNT':                       '1',
        'Accept-Language':           'zh-CN,zh;q=0.9,en;q=0.8',
        'Accept-Encoding':           'gzip, deflate, br',
    }
    body = {
        'state':                               '',
        'redirect_uri':                        'npf{}://auth'.format(client_id),
        'client_id':                           client_id,
        'scope':                               'openid user user.mii user.email user.links[].id',
        'response_type':                       'session_token_code',
        'session_token_code_challenge':        auth_code_challenge.replace(b"=", b"").decode('utf-8'),
        'session_token_code_challenge_method': 'S256',
        'theme':                               'login_form'
    }

    url = 'https://accounts.nintendo.com/connect/1.0.0/authorize'
    r = session.get(url, headers=app_head, params=body)

    post_login = r.history[0].url

    print("\nMake sure you have fully read the \"Cookie generation\" section of the readme before proceeding. To manually input a cookie instead, enter \"skip\" at the prompt below.")
    print("\nNavigate to this URL in your browser:")
    print(post_login)
    print("Log in, right click the \"Select this account\" button, copy the link address, and paste it below:")
    while True:
        try:
            use_account_url = input("")
            if use_account_url == "skip":
                return "skip"
            session_token_code = re.search('session_token_code=([^&]+)', use_account_url)

            # get session tocken
            app_head = {
                'User-Agent':      ua,
                'Accept-Language': 'en-US',
                'Accept':          'application/json',
                'Content-Type':    'application/x-www-form-urlencoded',
                'Host':            'accounts.nintendo.com',
                'Connection':      'Keep-Alive',
                'Accept-Encoding': 'gzip'
            }

            body = {
                'client_id':                   client_id,
                'session_token_code':          session_token_code.group(1),
                'session_token_code_verifier': auth_code_verifier.replace(b"=", b"").decode('utf-8')
            }

            url = 'https://accounts.nintendo.com/connect/1.0.0/api/session_token'

            r = session.post(url, headers=app_head, data=body)
            print(r)
            session_token = json.loads(r.text)["session_token"]

            return session_token
        except KeyboardInterrupt:
            print("\nBye!")
            sys.exit(1)
        except AttributeError:
            print("Malformed URL. Please try again, or press Ctrl+C to exit.")
            print("URL:", end=' ')
        except KeyError:  # session_token not found
            print(
                "\nThe URL has expired. Please log out and back into your Nintendo Account and try again.")
            sys.exit(1)


def NS_GetAccessToken(client_id,session_token):
    session = requests.Session()

    body = '{"client_id":"' + client_id + '","session_token":"' + session_token + \
        '","grant_type":"urn:ietf:params:oauth:grant-type:jwt-bearer-session-token"}'
    url = 'https://accounts.nintendo.com/connect/1.0.0/api/token'

    r = session.post(
        url, headers={'Content-Type': 'application/json'}, data=body)
    access_token = json.loads(r.text)

    return access_token
 
def NS_GetPlayHistory(access_token,ua):
    session = requests.Session()

    url = 'https://news-api.entry.nintendo.co.jp/api/v1.1/users/me/play_histories'
    header = {
        'Authorization': access_token['token_type'] + ' ' + access_token['access_token'],
        'User-Agent': ua,
    }
    r = session.get(url, headers=header)
    history=json.loads(r.text)

    return history

if __name__ =="__main__":

    client_id = "5c38e31cd085304b"
    ua = 'com.nintendo.znej/1.13.0 (Android/7.1.2)'

    print(NS_GetSessionToken(client_id,ua))

NS_GetPlayHistory返回数据格式

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
{
    "playHistories": [
        {
            "titleId": "01001B300B9BE000",
            "titleName": "ディアブロ III エターナルコレクション",
            "deviceType": "HAC",
            "imageUrl": "https://atum-img-lp1.cdn.nintendo.net/i/c/21425b81f8d642fdbc461e7acd11e32b_256",
            "lastUpdatedAt": "2023-01-30T16:01:19+09:00",
            "firstPlayedAt": "2022-12-10T14:56:31+09:00",
            "lastPlayedAt": "2023-01-30T01:08:10+09:00",
            "totalPlayedDays": 13,
            "totalPlayedMinutes": 2967
        },
        ...
    ],
    "hiddenTitleList": [],
    "recentPlayHistories": [
        {
            "playedDate": "2023-01-30T00:00:00+09:00",
            "dailyPlayHistories": [
                {
                    "titleId": "01001B300B9BE000",
                    "titleName": "ディアブロ III エターナルコレクション",
                    "imageUrl": "https://atum-img-lp1.cdn.nintendo.net/i/c/21425b81f8d642fdbc461e7acd11e32b_256",
                    "totalPlayedMinutes": 16
                }
            ]
        },
        ...
    ],
    "lastUpdatedAt": "2023-01-30T16:01:19+09:00"
}

  playHistories包含了NS账户下所有游戏的历史统计记录，recentPlayHistories包含近七天每一天的游戏时长统计记录。
  需要注意的一点是，日服账号获取的返回数据中，时间是东九区时间，要转换为东八区北京时间需要减一个小时。

设计数据表

  Switch数据表的设计与Steam数据表基本一致。因为使用日区账号获取的游戏数据名称都为日文，一种解决方案为使用游戏id在港服eshop页面爬取游戏中文名称，不过该名称为繁体中文，且在港服没有上架的游戏无法获取名称。因为我Switch上的游戏不会很多，所以就手动维护了一个中文译名表，在查询时关联使用。

dim_switch_game_play_history

字段 数据类型 备注
titleId varchar(255) 游戏id
titleName varchar(255) 游戏名称
deviceType varchar(255) 设备型号
imageUrl varchar(255) 游戏图标链接
lastUpdatedAt datetime 最近一次服务器数据更新时间
firstPlayedAt datetime 第一次游玩时间（开始时间）
lastPlayedAt datetime 最近一次游玩时间（开始时间）
totalPlayedDays int 总游玩天数
totalPlayedMinutes int 总游戏时长（分）
create_time datetime 记录创建时间，默认值CURRENT_TIMESTAMP
update_time datetime 记录修改时间，默认值CURRENT_TIMESTAMP ON UPDATE CURRENT_TIMESTAMP

dwd_switch_game_played_record

字段 数据类型 备注
id int 记录id，自增主键
titleId varchar(255) 游戏id
titleName varchar(255) 游戏名称
lastPlayedAt datetime 最近一次游戏时间（开始时间）
playtime int 游戏时长（分）
create_time datetime 记录创建时间，默认值CURRENT_TIMESTAMP
update_time datetime 记录修改时间，默认值CURRENT_TIMESTAMP ON UPDATE CURRENT_TIMESTAMP

建表语句：

1
2
3
4
5
6
7
8
9
10
CREATE TABLE `dwd_switch_game_played_record` (
  `id` int NOT NULL AUTO_INCREMENT,
  `title_id` varchar(255) COLLATE utf8mb4_general_ci DEFAULT NULL,
  `title_name` varchar(255) COLLATE utf8mb4_general_ci DEFAULT NULL,
  `last_played_at` datetime DEFAULT NULL,
  `play_time` 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;

dim_switch_game_name_translate_man

字段 数据类型 说明
title_id varchar(255) 游戏id
jp_name varchar(255) 日服游戏名称
zh_name varchar(255) 中文游戏名称

建表语句：

1
2
3
4
5
6
CREATE TABLE `dim_switch_game_name_translate_man` (
  `title_id` varchar(100) CHARACTER SET utf8mb4 COLLATE utf8mb4_general_ci NOT NULL,
  `jp_name` varchar(100) CHARACTER SET utf8mb4 COLLATE utf8mb4_general_ci DEFAULT NULL,
  `zh_name` varchar(100) COLLATE utf8mb4_general_ci DEFAULT NULL,
  PRIMARY KEY (`title_id`)
) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_general_ci;

数据采集

数据采集脚本

  Swith游戏数据采集脚本的实现思路与Steam数据采集基本一致，要注意的是session_token有两年的有效期，且获取该tocken时使用的client_id一一对应。

SwitchDA.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 SwitchWebAPI import *

client_id = "5c38e31cd085304b" # 你生成session_token时使用的client_id
ua = 'com.nintendo.znej/1.13.0 (Android/7.1.2)'
session_token="your_session_token" #你的session_token

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

def SwitchDA_GamePlayHistory(client_id,session_token,ua):
    results=NS_GetPlayHistory(NS_GetAccessToken(client_id,session_token),ua)

    titleId=[]
    titleName=[]
    deviceType=[]
    imageUrl=[]
    lastUpdatedAt=[]
    firstPlayedAt=[]
    lastPlayedAt=[]
    totalPlayedDays=[]
    totalPlayedMinutes=[]

    for i in results["playHistories"]:
        titleId.append(i["titleId"])
        titleName.append(i["titleName"])
        deviceType.append(i["deviceType"])
        imageUrl.append(i["imageUrl"])
        lastUpdatedAt.append(i["lastUpdatedAt"])
        firstPlayedAt.append(i["firstPlayedAt"])
        lastPlayedAt.append(i["lastPlayedAt"])
        totalPlayedDays.append(i["totalPlayedDays"])
        totalPlayedMinutes.append(i["totalPlayedMinutes"])
    
    df=pd.DataFrame({"titleId":titleId,"titleName":titleName,"deviceType":deviceType,"imageUrl":imageUrl,"lastUpdatedAt":lastUpdatedAt,"firstPlayedAt":firstPlayedAt,"lastPlayedAt":lastPlayedAt,"totalPlayedDays":totalPlayedDays,"totalPlayedMinutes":totalPlayedMinutes})

    UTC9to8=lambda x: (datetime.strptime(x,"%Y-%m-%dT%H:%M:%S+09:00")-timedelta(hours=1)).strftime("%Y-%m-%d %H:%M:%S")

    df["lastUpdatedAt"]=df["lastUpdatedAt"].map(UTC9to8)
    df["firstPlayedAt"]=df["firstPlayedAt"].map(UTC9to8)
    df["lastPlayedAt"]=df["lastPlayedAt"].map(UTC9to8)

    # print(df)

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

    dtype={"titleId":types.String(length=255),
            "titleName":types.String(length=255),
            "deviceType":types.String(length=255),
            "imageUrl":types.String(length=255),
            "lastUpdatedAt":types.DateTime(),
            "firstPlayedAt":types.DateTime(),
            "lastPlayedAt":types.DateTime(),
            "totalPlayedDays":types.Integer(),
            "totalPlayedMinutes":types.Integer()
    }

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


def SwitchDA_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_switch_game_played_record 
        SELECT 
                NULL id,
                t1.titleId ,
                t1.titleName ,
                t1.lastPlayedAt,
                COALESCE(t1.totalPlayedMinutes  - t2.play_time,t1.totalPlayedMinutes) play_time,
                NOW() create_time,
                NOW() update_time 
        FROM (
        SELECT 
                    titleId ,
                    titleName ,
                    lastPlayedAt ,
                    totalPlayedMinutes
        FROM dim_switch_game_play_history
        ) t1
        LEFT JOIN
        (
        SELECT 
                    title_id ,
                    SUM(play_time) play_time,
                    MAX(last_played_at) last_played_at
        FROM dwd_switch_game_played_record
        GROUP BY title_id
        )t2
        ON t1.titleId=t2.title_id
        WHERE t2.last_played_at IS NULL OR t1.lastPlayedAt !=t2.last_played_at
        """
        # 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"),"开始采集Switch游戏数据")

    SwitchDA_GamePlayHistory(client_id,session_token,ua)
    SwitchDA_GamePlayedRecord()

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

任务调度

使用腾讯云函数进行任务部署及调度的方案请参照这篇文章

  调度方案与Steam数据采集一样，依然采用本地手动调度，通过快捷指令间接调用执行脚本。不过需要注意的是，Switch游戏数据上传到云服务器并没有Steam那么快，可能需要等几个小时才会同步。

sh脚本实现

1
2
3
4
DA02_HOME_PATH=/your_path/data_acquisition_local
DA02_PYTHON_PATH=/your_path/miniconda3/bin/python

$DA02_PYTHON_PATH "$DA02_HOME_PATH/SwitchDA.py" >> "$DA02_HOME_PATH/crontab_out.log" 2>&1

快捷指令实现

快捷指令菜单栏效果

定时自动采集

  配合Shortery实现定时任务，由于Switch服务器的数据更新有延迟，建议一天多采集几次数据。

报表效果展示

参考资料

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

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