Python 实现抖音直播间弹幕实时监控与关键词提取

# WebSocket客户端库
pip install websockets
# 中文分词与关键词提取
pip install jieba
# 数据处理与存储
pip install pandas
# 协程异步处理
pip install asyncio

import asyncio
import websockets
import json
import jieba
from collections import Counter
import pandas as pd
import time
from datetime import datetime

# 全局配置
ROOM_ID = "你的直播间ID"  # 替换为目标直播间ID
SAVE_PATH = "douyin_danmu.csv"  # 弹幕保存路径
KEYWORD_TOP_N = 10  # 提取前10个高频关键词
# 停用词列表（过滤无意义词汇）
STOP_WORDS = {"的", "了", "是", "我", "你", "在", "吗", "吧", "啊", "哦", "呀", "这个", "那个"}

# 存储实时弹幕数据
danmu_data = []

def save_danmu_to_csv(data):
    """将弹幕数据保存到CSV文件"""
    df = pd.DataFrame(data)
    df.to_csv(SAVE_PATH, mode="a", index=False, header=not pd.io.common.path_exists(SAVE_PATH))

def extract_keywords(content_list):
    """
    对弹幕内容进行分词并提取高频关键词
    :param content_list: 弹幕内容列表
    :return: 高频关键词及词频
    """
    words = []
    for content in content_list:
        # 精确模式分词
        seg_list = jieba.lcut(content.strip())
        # 过滤停用词、空字符、单字符
        for word in seg_list:
            if word not in STOP_WORDS and len(word) > 1 and word.strip():
                words.append(word)
    # 统计词频并取TOP N
    word_counter = Counter(words)
    top_keywords = word_counter.most_common(KEYWORD_TOP_N)
    return top_keywords

async def receive_danmu():
    """WebSocket连接抖音服务器，实时接收弹幕数据"""
    # 抖音直播弹幕WebSocket接口（适配最新协议）
    uri = f"wss://webcast3-ws-web-lf.bytedance.com/webcast/im/push/v2/?room_id={ROOM_ID}"
    
    headers = {
        "User-Agent": "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36"
    }

    try:
        async with websockets.connect(uri, extra_headers=headers) as websocket:
            print(f"✅ 成功连接直播间：{ROOM_ID}，开始实时监控弹幕...")
            while True:
                # 接收服务器推送的二进制数据
                response = await websocket.recv()
                try:
                    # 解码数据并解析JSON
                    data = json.loads(response.decode("utf-8", errors="ignore"))
                    
                    # 筛选弹幕消息类型
                    if "messages" in data:
                        for msg in data["messages"]:
                            if msg.get("method") == "WebcastChatMessage":
                                # 提取核心弹幕字段
                                payload = json.loads(msg.get("payload", "{}"))
                                user_name = payload.get("user", {}).get("nickname", "匿名用户")
                                content = payload.get("content", "")
                                send_time = datetime.now().strftime("%Y-%m-%d %H:%M:%S")
                                
                                # 组装弹幕数据
                                danmu = {
                                    "发送时间": send_time,
                                    "用户昵称": user_name,
                                    "弹幕内容": content
                                }
                                danmu_data.append(danmu)
                                # 实时打印弹幕
                                print(f"[{send_time}] {user_name}：{content}")
                                
                                # 每接收10条弹幕保存一次文件
                                if len(danmu_data) % 10 == 0:
                                    save_danmu_to_csv(danmu_data)
                                    danmu_data.clear()
                except:
                    # 忽略非弹幕格式数据
                    continue
    except Exception as e:
        print(f"❌ 连接失败：{str(e)}，请检查直播间ID是否正确")

async def keyword_monitor():
    """实时提取并打印高频关键词"""
    while True:
        # 每5秒分析一次关键词
        await asyncio.sleep(5)
        # 读取最新的弹幕数据
        try:
            df = pd.read_csv(SAVE_PATH)
            content_list = df["弹幕内容"].tolist()
            top_keywords = extract_keywords(content_list)
            print("\n" + "="*50)
            print(f"📊 实时高频关键词（{datetime.now().strftime('%H:%M:%S')}）：")
            for word, count in top_keywords:
                print(f"🔹 {word}：{count}次")
            print("="*50 + "\n")
        except:
            print("⏳ 暂无足够弹幕数据，继续监控...")

async def main():
    """主函数：并发运行弹幕接收与关键词提取"""
    task1 = asyncio.create_task(receive_danmu())
    task2 = asyncio.create_task(keyword_monitor())
    await asyncio.gather(task1, task2)

if __name__ == "__main__":
    # 启动异步任务
    asyncio.run(main())