如何解决Python Selenium获取Twitch聊天元素列表150条长度上限问题
问题根因
Twitch 前端聊天区采用虚拟滚动机制,DOM 树中默认只会保留最近150条聊天消息,更早的消息会被自动从DOM中移除,因此你通过find_elements_by_class_name获取的元素列表长度上限就是150,当长度达到阈值后,新消息进入时旧消息被移除,列表长度保持不变,你的长度差判断逻辑自然无法识别到新消息。
可行解决方案
方案1:直接对接Twitch官方IRC接口(最稳定,优先推荐)
不用依赖Selenium抓DOM,Twitch本身开放了聊天消息的IRC推送接口,全量实时获取所有聊天消息,不存在漏消息、条数限制问题:
- 到Twitch开发者平台申请仅带聊天读取权限的OAuth认证令牌
- 用Python的
irc或websocket库连接Twitch IRC服务器,加入对应直播间的聊天频道即可实时接收所有消息推送
示例代码片段:
import irc.bot class TwitchBot(irc.bot.SingleServerIRCBot): def __init__(self, username, token, channel): server = 'irc.chat.twitch.tv' port = 6667 super().__init__([(server, port, token)], username, username) self.channel = '#' + channel def on_welcome(self, c, e): c.join(self.channel) def on_pubmsg(self, c, e): # 直接获取发送者、消息内容 author = e.source.split('!')[0] message = e.arguments[0] print(f"{author}: {message}") if __name__ == "__main__": bot = TwitchBot("你的Twitch用户名", "你的OAuth令牌", "目标直播间ID") bot.start()
方案2:基于消息唯一ID判断更新(不替换现有Selenium逻辑的最小改造成本)
Twitch每条聊天消息的DOM元素都带有唯一的data-id属性,数值随消息发送时间递增,你可以放弃长度判断逻辑,改为记录已处理的最新消息ID,每次扫描仅处理ID大于已记录值的新消息:
修改后的代码逻辑:
import time # 初始化最新消息ID为0 last_processed_id = 0 while True: chat_messages = driver.find_elements_by_class_name(chatbox_classname) for chat in chat_messages: # 获取当前消息的唯一ID msg_id = int(chat.get_attribute("data-id")) if msg_id > last_processed_id: # 提取作者和消息内容处理 author = chat.find_element_by_class_name("chat-author__display-name").text content = chat.find_element_by_class_name("text-fragment").text print(f"{author}: {content}") # 更新最新处理ID last_processed_id = msg_id # 加适当间隔避免请求太频繁 time.sleep(0.5)
该方案完全不受DOM中150条消息上限的影响,只要新消息的ID大于已处理值就会被识别。
方案3:注入MutationObserver监听DOM变动(无需轮询,效率更高)
直接通过Selenium在页面注入JS代码,监听聊天容器的子节点新增事件,只要有新消息插入DOM就立即把内容传回Python,不需要主动轮询扫描,也不会漏消息:
示例逻辑:
import time # 先获取聊天容器元素 chat_container = driver.find_element_by_class_name("chat-scrollable-area__message-container") # 注入MutationObserver监听逻辑 driver.execute_script(""" const observer = new MutationObserver((mutations) => { mutations.forEach(m => { m.addedNodes.forEach(node => { if (node.nodeType === 1 && node.classList.contains('""" + chatbox_classname + """')) { const author = node.querySelector('.chat-author__display-name')?.textContent || ''; const content = node.querySelector('.text-fragment')?.textContent || ''; // 把消息存入window对象供Python读取 window.newChatMessages = window.newChatMessages || []; window.newChatMessages.push({author, content}); } }) }) }); observer.observe(arguments[0], {childList: true, subtree: true}); """, chat_container) # 循环读取新消息 while True: new_messages = driver.execute_script(""" const msgs = window.newChatMessages || []; window.newChatMessages = []; return msgs; """) for msg in new_messages: print(f"{msg['author']}: {msg['content']}") time.sleep(0.5)
内容的提问来源于stack exchange,提问作者LtBoofy
相关产品推荐
相关产品推荐

