这篇教程用一个本地 Python 示例，演示如何把抖音直播间的事件流整理成可统计的数据。它不负责抓取、登录、签名或绕过风控，而是处理你已经通过官方开放平台、授权回调或手动导出的 JSON 事件。采集和统计分开后，后续替换数据源会更安全，也更容易维护。
运行环境：Python 3.8 及以上，标准库即可。步骤一，新建文件 live_room_event_stats.py，把下面的代码完整粘贴进去。步骤二，在终端执行 python live_room_event_stats.py，会先用内置模拟事件输出统计结果。步骤三，如果要分析自己的数据，把事件保存为 events.json，数组中的每个对象至少包含 type 字段。
事件字段可以这样设计：comment 表示弹幕，包含 user 和 content；gift 表示礼物，包含 user、gift、count、price；like 表示点赞，包含 user 和 count；enter 表示进场，包含 user；online 表示在线人数快照，包含 count。时间戳可放在 ts 字段，当前脚本没有按时间窗口过滤，你可以后续在 analyze 中增加。
# live_room_event_stats.py
# 用途：对本地保存的直播间事件流做统计，不包含登录、签名、抓包或绕过风控逻辑。
import json
from collections import Counter
SAMPLE_EVENTS = [
{
'type': 'comment', 'user': '小明', 'content': '主播好', 'ts': 1710000001},
{
'type': 'comment', 'user': '小红', 'content': '这个怎么买', 'ts': 1710000002},
{
'type': 'gift', 'user': '阿强', 'gift': '小心心', 'count': 2, 'price': 1, 'ts': 1710000003},
{
'type': 'like', 'user': '路人甲', 'count': 10, 'ts': 1710000004},
{
'type': 'enter', 'user': '新观众', 'ts': 1710000005},
{
'type': 'comment', 'user': '小明', 'content': '已经下单', 'ts': 1710000006},
{
'type': 'gift', 'user': '小红', 'gift': '玫瑰', 'count': 1, 'price': 5, 'ts': 1710000007},
{
'type': 'online', 'count': 128, 'ts': 1710000008},
{
'type': 'comment', 'user': '阿强', 'content': '主播讲得清楚', 'ts': 1710000009},
{
'type': 'gift', 'user': '阿强', 'gift': '小心心', 'count': 1, 'price': 1, 'ts': 1710000010},
]
def load_events(path=None):
# 默认读取内置示例；如果传入 JSON 文件路径，则读取本地事件文件。
if not path:
return SAMPLE_EVENTS
with open(path, 'r', encoding='utf-8') as f:
return json.load(f)
def analyze(events):
result = {
'总事件数': len(events),
'弹幕数': 0,
'礼物事件数': 0,
'独立用户数': 0,
'礼物总值': 0,
'最高在线人数': 0,
'弹幕用户排行': [],
'礼物排行': [],
'热门弹幕词': [],
}
users = set()
comment_users = Counter()
gift_counter = Counter()
words = Counter()
for event in events:
etype = event.get('type')
user = event.get('user')
if user:
users.add(user)
if etype == 'comment':
result['弹幕数'] += 1
if user:
comment_users[user] += 1
content = event.get('content', '')
for word in content.replace('，', ' ').replace('。', ' ').split():
if len(word) >= 2:
words[word] += 1
elif etype == 'gift':
result['礼物事件数'] += 1
gift = event.get('gift', '未知礼物')
count = int(event.get('count', 1))
price = float(event.get('price', 0))
gift_counter[gift] += count
result['礼物总值'] += count * price
elif etype == 'online':
result['最高在线人数'] = max(result['最高在线人数'], int(event.get('count', 0)))
result['独立用户数'] = len(users)
result['弹幕用户排行'] = comment_users.most_common(5)
result['礼物排行'] = gift_counter.most_common(5)
result['热门弹幕词'] = words.most_common(5)
return result
def main():
events = load_events() # 改成 load_events('events.json') 可读取本地导出的事件流
stats = analyze(events)
print('直播间事件统计结果')
print('-' * 30)
for key, value in stats.items():
print(f'{key}: {value}')
if __name__ == '__main__':
main()
代码的核心是 load_events 和 analyze。load_events 默认返回 SAMPLE_EVENTS，传入路径后读取本地 JSON。analyze 遍历事件：评论统计弹幕数和评论用户排行，礼物累计数量和金额，在线事件记录最高在线人数，最后用 Counter 输出排行。main 里把 load_events() 改成 load_events('events.json') 就能切换到真实导出文件。
想继续扩展，可以加这些方向：按 ts 做分钟级聚合，输出每 10 秒的弹幕峰值；把礼物 price 换成平台约定的价值单位；从 WebSocket 回调里把消息转成同一套事件结构，再交给 analyze。要注意合规：只采集你有权限的数据，遵守平台协议，控制请求频率，不要尝试破解接口或绕过限制。统计脚本只做本地数据处理，不会主动访问抖音。
