Steam Web API实战：除了查库存，你还能用Python脚本自动追踪好友的游戏成就与时长-程序员充电站

Steam Web API实战：用Python构建游戏数据分析系统

Steam平台不仅是全球最大的数字游戏发行平台，更是一个隐藏着海量玩家行为数据的宝库。作为一名资深游戏开发者兼数据分析师，我发现许多技术爱好者仅仅将Steam Web API用于查询好友在线状态这种基础功能，却忽略了它真正的潜力。本文将带你深入探索如何利用Python构建一个完整的游戏数据分析系统，从基础的API调用到复杂的数据可视化，解锁Steam数据的更多可能性。

1. Steam Web API基础配置

在开始我们的数据探索之旅前，需要先完成一些基础配置工作。与大多数API服务不同，Steam的认证机制相对简单，但仍需注意几个关键点。

首先访问Steam Web API密钥申请页面，使用你的Steam账号登录后，只需填写一个域名（开发阶段可填写localhost）即可获得API密钥。这个32位的字符串将是我们所有请求的通行证。

获取好友的SteamID是另一个关键步骤。这里有个实用技巧：通过Steam社区资料页的URL可以直接提取用户的64位SteamID。例如在URLhttps://steamcommunity.com/profiles/76561197960434622中，末尾的数字就是我们要的ID。

# 示例：提取SteamID profile_url = "https://steamcommunity.com/profiles/76561197960434622" steam_id = profile_url.split('/')[-1] # 获取76561197960434622

注意：根据Steam的API使用条款，批量采集非好友用户数据可能违反服务协议，建议仅收集已同意共享数据的好友信息。

2. 核心API接口解析

Steam Web API提供了数十个功能各异的接口，我们重点分析几个对游戏数据分析最有价值的端点。

2.1 玩家游戏库接口

IPlayerService/GetOwnedGames接口能获取玩家拥有的所有游戏及其累计时长：

import requests def get_owned_games(api_key, steam_id): url = f"http://api.steampowered.com/IPlayerService/GetOwnedGames/v0001/?key={api_key}&steamid={steam_id}&include_appinfo=1&format=json" response = requests.get(url) return response.json() # 示例返回数据结构 { "response": { "game_count": 42, "games": [ { "appid": 292030, "name": "The Witcher 3: Wild Hunt", "playtime_forever": 327, "img_icon_url": "abcdefg123456" } ] } }

2.2 游戏成就接口

ISteamUserStats/GetPlayerAchievements可以查询玩家在特定游戏中的成就获取情况：

def get_achievements(api_key, steam_id, app_id): url = f"http://api.steampowered.com/ISteamUserStats/GetPlayerAchievements/v0001/?appid={app_id}&key={api_key}&steamid={steam_id}" response = requests.get(url) return response.json()

2.3 好友列表接口

ISteamUser/GetFriendList不仅能获取好友列表，还能了解他们的在线状态：

def get_friend_list(api_key, steam_id): url = f"http://api.steampowered.com/ISteamUser/GetFriendList/v0001/?key={api_key}&steamid={steam_id}&relationship=friend" response = requests.get(url) return response.json()

3. 构建游戏数据分析系统

有了这些基础接口，我们可以开始构建更复杂的数据分析功能。下面是一个完整的系统设计方案。

3.1 数据采集模块

首先设计一个稳定的数据采集模块，定期获取并存储好友游戏数据：

import sqlite3 from datetime import datetime def init_database(): conn = sqlite3.connect('steam_data.db') c = conn.cursor() c.execute('''CREATE TABLE IF NOT EXISTS game_records (id INTEGER PRIMARY KEY AUTOINCREMENT, steam_id TEXT, game_id INTEGER, game_name TEXT, playtime INTEGER, record_date TEXT)''') conn.commit() conn.close() def save_game_data(steam_id, game_data): conn = sqlite3.connect('steam_data.db') c = conn.cursor() today = datetime.now().strftime('%Y-%m-%d') for game in game_data['response']['games']: c.execute("INSERT INTO game_records (steam_id, game_id, game_name, playtime, record_date) VALUES (?, ?, ?, ?, ?)", (steam_id, game['appid'], game['name'], game['playtime_forever'], today)) conn.commit() conn.close()

3.2 数据分析模块

基于采集的数据，我们可以进行各种有趣的分析：

import pandas as pd import matplotlib.pyplot as plt def analyze_friend_games(friend_id): conn = sqlite3.connect('steam_data.db') df = pd.read_sql_query(f""" SELECT game_name, SUM(playtime) as total_time FROM game_records WHERE steam_id = '{friend_id}' GROUP BY game_name ORDER BY total_time DESC LIMIT 5 """, conn) conn.close() plt.figure(figsize=(10, 6)) plt.barh(df['game_name'], df['total_time']/60, color='skyblue') plt.xlabel('游戏时长(小时)') plt.title(f'好友{friend_id}的游戏时长TOP5') plt.tight_layout() plt.savefig(f'{friend_id}_top_games.png') return df

3.3 成就追踪与比较

游戏成就系统是玩家投入度的重要指标。我们可以设计一个成就比较功能：

def compare_achievements(player1_id, player2_id, app_id): p1_achs = get_achievements(API_KEY, player1_id, app_id) p2_achs = get_achievements(API_KEY, player2_id, app_id) p1_earned = [a['apiname'] for a in p1_achs['playerstats']['achievements'] if a['achieved']] p2_earned = [a['apiname'] for a in p2_achs['playerstats']['achievements'] if a['achieved']] unique_p1 = set(p1_earned) - set(p2_earned) unique_p2 = set(p2_earned) - set(p1_earned) print(f"玩家1独有的成就({len(unique_p1)}个): {unique_p1}") print(f"玩家2独有的成就({len(unique_p2)}个): {unique_p2}")

4. 高级应用：游戏社交网络分析

将上述功能组合起来，我们可以构建更复杂的分析应用。例如，分析你的Steam好友社交网络中的游戏偏好分布。

4.1 构建游戏偏好矩阵

首先收集所有好友的游戏数据：

def collect_friends_games(api_key, steam_id): friends = get_friend_list(api_key, steam_id) all_games = {} for friend in friends['friendslist']['friends']: friend_id = friend['steamid'] games = get_owned_games(api_key, friend_id) all_games[friend_id] = [g['appid'] for g in games['response']['games']] return all_games

4.2 计算游戏相似度

基于收集的数据，我们可以计算好友间的游戏相似度：

from collections import defaultdict def calculate_similarity(all_games): game_users = defaultdict(list) for user, games in all_games.items(): for game in games: game_users[game].append(user) similarity_matrix = defaultdict(dict) users = list(all_games.keys()) for i in range(len(users)): for j in range(i+1, len(users)): common_games = set(all_games[users[i]]) & set(all_games[users[j]]) total_games = set(all_games[users[i]]) | set(all_games[users[j]]) similarity = len(common_games) / len(total_games) if total_games else 0 similarity_matrix[users[i]][users[j]] = similarity similarity_matrix[users[j]][users[i]] = similarity return similarity_matrix

4.3 可视化社交网络

使用networkx库将相似度关系可视化：

import networkx as nx def visualize_social_network(similarity_matrix, threshold=0.3): G = nx.Graph() for user1 in similarity_matrix: G.add_node(user1) for user2 in similarity_matrix[user1]: if similarity_matrix[user1][user2] > threshold: G.add_edge(user1, user2, weight=similarity_matrix[user1][user2]) plt.figure(figsize=(12, 12)) pos = nx.spring_layout(G) nx.draw_networkx_nodes(G, pos, node_size=50) nx.draw_networkx_edges(G, pos, alpha=0.2) plt.title("Steam好友游戏偏好社交网络") plt.savefig('steam_social_network.png')

5. 自动化通知系统的进阶实现

原始的钉钉通知功能可以进一步扩展，实现更智能的游戏动态追踪。

5.1 游戏动态监控

设计一个监控好友特定游戏动态的系统：

def monitor_game_activity(api_key, friend_id, game_id, check_interval=3600): last_playtime = 0 while True: games = get_owned_games(api_key, friend_id) current_game = next((g for g in games['response']['games'] if g['appid'] == game_id), None) if current_game and current_game['playtime_forever'] > last_playtime: playtime_diff = current_game['playtime_forever'] - last_playtime message = f"你的好友最近玩了{current_game['name']}，新增时长：{playtime_diff}分钟" send_dingtalk_message(message) last_playtime = current_game['playtime_forever'] time.sleep(check_interval)

5.2 成就获取通知

当好友获得特定成就时发送通知：

def monitor_achievements(api_key, friend_id, game_id, achievement_names): known_achievements = set() while True: achievements = get_achievements(api_key, friend_id, game_id) new_achievements = [] for ach in achievements['playerstats']['achievements']: if ach['achieved'] and ach['apiname'] not in known_achievements: if ach['apiname'] in achievement_names: new_achievements.append(ach['name']) known_achievements.add(ach['apiname']) if new_achievements: message = f"你的好友获得了新成就：{', '.join(new_achievements)}" send_dingtalk_message(message) time.sleep(3600) # 每小时检查一次

在实际项目中，我发现将游戏数据分析结果与自动化通知结合，可以创造出许多有趣的场景。比如当好友的游戏时长突然激增时，可能意味着新DLC发布或周末游戏马拉松；而当多个好友同时开始玩某款游戏时，可能暗示着值得关注的新游戏趋势。