利用Python脚本模拟Steam直播间观众数量

1. 项目简介

steam-broadcast 是一个用于提升 Steam 直播间观众数量的 Python 脚本。其核心原理是模拟多个虚拟观众进入指定直播间，并通过定期发送心跳请求来保持在线状态。

GitHub 仓库地址：steam-broadcast

在本篇文章中，我们将深入解析 broadcast.py 的实现思路，并详细剖析其关键逻辑。

2. broadcast.py 实现解析

2.1 代码的核心逻辑

该脚本的主要流程包括：

1. 获取 User-Agent：伪造不同的浏览器标识，以降低被检测的风险。
2. 获取 session_id：访问 Steam 直播间页面，解析 session_id。
3. 获取直播间核心信息：通过 API 请求获取 broadcast_id 和 viewer_token。
4. 获取直播间观众数：定期请求 API，获取直播间当前的观众数量。
5. 维持心跳：模拟观众定期向 Steam 服务器发送心跳包，保持在线状态。
6. 多线程模拟多个观众：通过 threading 模块实现并发，创建多个线程模拟不同观众。
7. 使用代理池（可选）：避免单一 IP 触发风控，提高稳定性。

2.2 代码拆解

2.2.1 获取 session_id

# 进入直播间并获取 session_id
def get_session_id(user_agent: str) -> str:
    session_id = ""
    url = f"{STEAM_COMMUNITY_URL}/broadcast/watch/{STEAM_ID}"
    headers = {
        "User-Agent": user_agent,
        'Referer': STEAM_COMMUNITY_URL
    }
    response = requests.get(url, headers=headers)
    if response is None:
        logging.error(f"获取 session_id 异常：{response}")
        return ""
    match = re.search(r'g_sessionID\s*=\s*"([^"]+)"', response.text)
    if match:
        session_id = match.group(1)
    return session_id

该函数首先访问 Steam 直播间页面，并解析返回的 HTML，提取 session_id。

2.2.2 获取 broadcast_id 和 viewer_token

# 获取直播间核心信息
def get_broadcast_mpd(session_id: str, user_agent: str) -> tuple:
    url = f"{STEAM_COMMUNITY_URL}/broadcast/getbroadcastmpd"
    params = {
        "steamid": STEAM_ID,
        "broadcastid": "0",
        "viewertoken": "",
        "watchlocation": "5",
        "sessionid": session_id
    }
    headers = {
        'Referer': f"{STEAM_COMMUNITY_URL}/broadcast/watch/{STEAM_ID}",
        "User-Agent": user_agent
    }
    response = requests.get(url, headers=headers, params=params)
    if response is None or response.text is None or response.status_code == 403:
        logging.error(f"请求 response 异常：{response}")
        return "", ""
    try:
        data = json.loads(response.text)
    except ValueError:
        logging.error(f"请求 response 异常：{response}")
        return "", ""
    broadcast_id = data["broadcastid"]
    viewer_token = data["viewertoken"]
    return broadcast_id, viewer_token

2.2.3 发送心跳请求

# 轮询心跳
def heartbeat(broadcast_id: str, viewer_token: str, user_agent: str) -> None:
    data = {
        "steamid": STEAM_ID,
        "broadcastid": broadcast_id,
        "viewertoken": viewer_token
    }
    url = f"{STEAM_COMMUNITY_URL}/broadcast/heartbeat/"
    headers = {
        'Referer': f"{STEAM_COMMUNITY_URL}/broadcast/watch/{STEAM_ID}",
        "User-Agent": user_agent
    }
    count = 1
    error_count = 0
    while count <= HEARTBEAT_COUNT:
        time.sleep(random.randint(25, 35))  # 避免固定间隔触发风控
        try:
            response = requests.post(url, headers=headers, data=data)
            logging.debug("发送心跳响应：%s" % response)
            if response.status_code == 403:
                error_count += 1
        except Exception as e:
            logging.error("发送心跳异常：%s" % e)
            error_count += 1
            if error_count >= 5:
                logging.error("心跳异常次数达到 5 次，准备停止当前线程")
                break
        count += 1
    logging.info(f"当前线程心跳结束")

2.2.4 多线程模拟观众

# 入口函数
def main():
    count = 1
    while count < VIEWER_COUNT:
        logging.info(f"当前在线观众数：{CURREN_VIEWER_COUNT}")
        logging.info(f"创建新线程 Thread-{count}")
        t = threading.Thread(target=step, name=f'Thread-{count}')
        t.start()
        time.sleep(random.randint(3, 5))
        count += 1

3. 关键点总结

1. 模拟观众请求：伪造 User-Agent，访问直播间并解析 session_id。
2. API 交互：获取 broadcast_id 和 viewer_token，用于后续的心跳维持。
3. 维持在线状态：定期发送心跳请求，确保观众不会掉线。
4. 多线程并发：通过 threading 创建多个线程，以模拟多个观众进入直播间。
5. 使用代理池（可选）：可以集成 requests 的代理功能，降低单 IP 访问风险。
6. 随机化心跳间隔：避免被服务器检测出固定时间间隔的心跳模式。

4. 可能的优化方向

• 代理池支持：可使用 requests 代理池减少 IP 被封的风险。
• 分布式架构：可结合 Celery 或 Docker 实现更大规模的模拟观众。
• 更智能的心跳策略：根据直播间真实观众波动情况调整心跳频率。
• Web 界面控制：开发一个简单的 Web 界面，允许用户动态调整参数。

5. 结语

steam-broadcast 通过模拟多个观众访问直播间，并定期发送心跳请求，有效提高直播间的人气。该脚本简单易用，并且可扩展性较强，可以根据需求调整参数或优化代码逻辑。如果你对 Steam 直播系统感兴趣，可以深入研究其 API 机制，甚至开发更高级的自动化工具。

6.免责声明

本项目仅供学习、测试与个人娱乐用途，请勿用于任何违反 Steam 相关服务条款的行为。作者不对使用本项目造成的任何后果负责。