Kris
59cd13cf84
新增Twitch IRC协议相关的全套实现，包括： 1. 基础工具类：令牌处理、消息格式化、速率限制、消息去重 2. 核心适配器组件：IRC解析器、消息归一化、外发消息处理 3. API客户端：Helix API封装、认证提供者 4. 配置与部署：配置校验、设置向导 5. 辅助功能：配对管理、健康检查、目标解析等
143 lines
3.7 KiB
Python
143 lines
3.7 KiB
Python
from __future__ import annotations
|
|
|
|
import re
|
|
import time
|
|
from dataclasses import dataclass
|
|
from typing import Any
|
|
|
|
_IRC_LINE_PATTERN = re.compile(
|
|
r"^(?:@(?P<tags>[^ ]+) )?(?::(?P<prefix>[^ ]+) )?(?P<command>[a-zA-Z]+|\d{3})"
|
|
r"(?: (?P<params>.+?))?(?: :(?P<trailing>.*))?$"
|
|
)
|
|
|
|
_PREFIX_PATTERN = re.compile(r"^([^!@]+)(?:!([^@]+))?(?:@(.+))?$")
|
|
|
|
|
|
@dataclass
|
|
class ParsedIRCMessage:
|
|
command: str
|
|
prefix: str | None
|
|
params: list[str]
|
|
trailing: str
|
|
tags: dict[str, str]
|
|
raw: str
|
|
|
|
|
|
def parse_irc_line(line: str) -> ParsedIRCMessage:
|
|
match = _IRC_LINE_PATTERN.match(line)
|
|
if match is None:
|
|
return ParsedIRCMessage(
|
|
command="",
|
|
prefix=None,
|
|
params=[],
|
|
trailing="",
|
|
tags={},
|
|
raw=line,
|
|
)
|
|
|
|
command = match.group("command").upper()
|
|
prefix = match.group("prefix") or None
|
|
|
|
params: list[str] = []
|
|
params_str = match.group("params") or ""
|
|
if params_str:
|
|
params.extend(params_str.split())
|
|
|
|
trailing = match.group("trailing") or ""
|
|
|
|
tags = _parse_tags(match.group("tags") or "")
|
|
|
|
return ParsedIRCMessage(
|
|
command=command,
|
|
prefix=prefix,
|
|
params=params,
|
|
trailing=trailing,
|
|
tags=tags,
|
|
raw=line,
|
|
)
|
|
|
|
|
|
def _parse_tags(tags_str: str) -> dict[str, str]:
|
|
if not tags_str:
|
|
return {}
|
|
|
|
tags: dict[str, str] = {}
|
|
for tag in tags_str.split(";"):
|
|
if "=" in tag:
|
|
key, value = tag.split("=", 1)
|
|
tags[key] = value
|
|
else:
|
|
tags[tag] = "1"
|
|
return tags
|
|
|
|
|
|
def parse_prefix(prefix: str | None) -> dict[str, str | None]:
|
|
if not prefix:
|
|
return {"nick": None, "user": None, "host": None}
|
|
|
|
m = _PREFIX_PATTERN.match(prefix)
|
|
if m is None:
|
|
return {"nick": prefix, "user": None, "host": None}
|
|
|
|
return {
|
|
"nick": m.group(1) or None,
|
|
"user": m.group(2) or None,
|
|
"host": m.group(3) or None,
|
|
}
|
|
|
|
|
|
def extract_nick(prefix: str | None) -> str:
|
|
if not prefix:
|
|
return "unknown"
|
|
return prefix.split("!")[0]
|
|
|
|
|
|
def parse_badges(badges_str: str) -> dict[str, str]:
|
|
if not badges_str:
|
|
return {}
|
|
result: dict[str, str] = {}
|
|
for badge in badges_str.split(","):
|
|
if "/" in badge:
|
|
name, version = badge.split("/", 1)
|
|
result[name] = version
|
|
else:
|
|
result[badge] = "1"
|
|
return result
|
|
|
|
|
|
def parse_emotes(emotes_str: str, message_text: str) -> list[dict[str, Any]]:
|
|
if not emotes_str:
|
|
return []
|
|
result: list[dict[str, Any]] = []
|
|
for emote_entry in emotes_str.split("/"):
|
|
if ":" not in emote_entry:
|
|
continue
|
|
emote_id, positions = emote_entry.split(":", 1)
|
|
for pos in positions.split(","):
|
|
if "-" not in pos:
|
|
continue
|
|
start_str, end_str = pos.split("-", 1)
|
|
start_idx, end_idx = int(start_str), int(end_str)
|
|
emote_text = message_text[start_idx : end_idx + 1] if message_text else ""
|
|
result.append(
|
|
{
|
|
"id": emote_id,
|
|
"text": emote_text,
|
|
"start": start_idx,
|
|
"end": end_idx,
|
|
"url": f"https://static-cdn.jtvnw.net/emoticons/v2/{emote_id}/default/dark/1.0",
|
|
}
|
|
)
|
|
return result
|
|
|
|
|
|
def build_identity_tags(**kwargs: str | int | None) -> str:
|
|
"""生成 Twitch IRC 消息中常用的 identity 标签字符串"""
|
|
parts = [f"{k}={v}" for k, v in kwargs.items() if v is not None]
|
|
return ";".join(parts)
|
|
|
|
|
|
def make_irc_message_id(user_id: str, tmi_sent_ts: str | None = None) -> str:
|
|
ts = tmi_sent_ts or str(int(time.time() * 1000))
|
|
return f"{user_id}:{ts}"
|