如何使用Python+Sockets实现Twitch IRC频道封禁/超时日志采集

Twitch的TMI IRC服务默认仅推送基础聊天消息，要获取 moderation 类事件（封禁、超时等）需要主动请求对应权限并解析专属事件命令，具体调整如下：

核心调整点 #

• 首先在发送NICK命令后、JOIN频道前，主动请求Twitch专属IRC capabilities：
  需申请twitch.tv/tags（获取事件元数据，比如超时时长、操作人、封禁原因等）和twitch.tv/commands（接收 moderation 事件通知）两个权限
• 修正PING响应逻辑，原固定匹配的写法容错率极低，只要PING内容有变动就会导致心跳失败断开连接
• 移除循环内的time.sleep(0.5)，该操作会导致消息堆积、丢包，影响事件接收
• 新增CLEARCHAT命令解析逻辑，Twitch的封禁/超时事件全部通过该命令推送：
  • 带ban-duration标签且值大于0：对应timeout事件，值为超时秒数
  • 不带ban-duration标签或值为0：对应永久ban事件

修改后可运行代码 #

import socket
import time

# 配置项自行补充对应值
SERVER = "irc.twitch.tv"
PORT = 6667
PASSWORD = "oauth:你的oauth密钥"
NICKNAME = "你的机器人用户名"
CHANNEL = "#目标频道名" # 多个频道用英文逗号分隔即可，如#频道1,#频道2

IRC = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
IRC.connect((SERVER, PORT))
IRC.send(f"PASS {PASSWORD}\n".encode('utf-8'))
IRC.send(f"NICK {NICKNAME}\n".encode('utf-8'))
# 新增：请求Twitch moderation事件权限
IRC.send(b"CAP REQ :twitch.tv/tags twitch.tv/commands\n")
IRC.send(f"JOIN {CHANNEL}\n".encode('utf-8'))

# 缓冲区处理TCP粘包/拆包问题
buffer = ""
while True:
    buffer += IRC.recv(1024).decode('utf-8')
    # 逐行处理完整消息
    while "\r\n" in buffer:
        line, buffer = buffer.split("\r\n", 1)
        # 处理心跳保活
        if line.startswith("PING"):
            IRC.send(f"PONG {line.split(' ',1)[1]}\r\n".encode('utf-8'))
            continue
        # 处理封禁/超时事件
        if "CLEARCHAT" in line:
            # 解析事件标签
            tags_part = line.split(" CLEARCHAT ",1)[0]
            tags = {}
            for tag in tags_part.lstrip('@').split(';'):
                if '=' in tag:
                    k, v = tag.split('=',1)
                    tags[k] = v
            # 解析被操作用户名
            target_user = line.split(':',2)[-1]
            # 区分事件类型输出日志
            if 'ban-duration' in tags:
                duration = int(tags['ban-duration'])
                print(f"[TIMEOUT] 用户{target_user}被超时{duration}秒，原因：{tags.get('ban-reason','无')}")
            else:
                print(f"[BAN] 用户{target_user}被永久封禁，原因：{tags.get('ban-reason','无')}")
        # 普通聊天消息可在此处继续处理
        # else:
        #     print(line)

多频道适配 #

要同时采集多个频道日志的话，仅需要在JOIN参数中将所有目标频道名用英文逗号拼接即可，格式为#频道1,#频道2,#频道3，所有频道的事件都会统一推送，解析时可从CLEARCHAT命令的频道参数中提取归属频道做分类存储。

内容的提问来源于stack exchange，提问作者kaizer