Kris
061cc7076b
新增 IRC 渠道的完整扩展实现，包括客户端连接管理、消息收发与流式处理、安全策略与配对验证、消息去重与规范化、状态监控与健康探测、配置模型与账户管理、错误处理等模块。
117 lines
3.4 KiB
Python
117 lines
3.4 KiB
Python
import re
|
|
from dataclasses import dataclass, field
|
|
|
|
|
|
@dataclass
|
|
class ParsedIrcLine:
|
|
raw: str
|
|
prefix: str | None
|
|
command: str
|
|
params: list[str]
|
|
trailing: str | None
|
|
tags: dict[str, str | bool] = field(default_factory=dict)
|
|
|
|
|
|
@dataclass
|
|
class ParsedIrcPrefix:
|
|
nick: str | None = None
|
|
user: str | None = None
|
|
host: str | None = None
|
|
server: str | None = None
|
|
|
|
|
|
CONTROL_CHARS = set(range(0x00, 0x20)) | {0x7F}
|
|
|
|
|
|
def parse_irc_line(line: str) -> ParsedIrcLine:
|
|
raw = line.strip()
|
|
tags: dict[str, str | bool] = {}
|
|
if raw.startswith("@"):
|
|
space_idx = raw.find(" ")
|
|
if space_idx == -1:
|
|
return ParsedIrcLine(raw=raw, prefix=None, command="", params=[], trailing=None)
|
|
tags_str = raw[1:space_idx]
|
|
raw = raw[space_idx + 1 :]
|
|
for tag_pair in tags_str.split(";"):
|
|
if "=" in tag_pair:
|
|
key, val = tag_pair.split("=", 1)
|
|
tags[key] = val
|
|
else:
|
|
tags[tag_pair] = True
|
|
|
|
prefix = None
|
|
if raw.startswith(":"):
|
|
space_idx = raw.find(" ")
|
|
if space_idx == -1:
|
|
return ParsedIrcLine(raw=line.strip(), prefix=None, command=raw[1:], params=[], trailing=None, tags=tags)
|
|
prefix = raw[1:space_idx]
|
|
raw = raw[space_idx + 1 :]
|
|
|
|
trail_idx = raw.find(" :")
|
|
trailing = None
|
|
if trail_idx != -1:
|
|
trailing = raw[trail_idx + 2 :]
|
|
raw = raw[:trail_idx]
|
|
|
|
parts = raw.split()
|
|
command = parts[0].upper() if parts else ""
|
|
params = parts[1:] if len(parts) > 1 else []
|
|
|
|
return ParsedIrcLine(raw=line.strip(), prefix=prefix, command=command, params=params, trailing=trailing, tags=tags)
|
|
|
|
|
|
def parse_irc_prefix(prefix: str) -> ParsedIrcPrefix:
|
|
if "!" in prefix or "@" in prefix:
|
|
nick, rest = prefix.split("!", 1) if "!" in prefix else (prefix.split("@", 1)[0], prefix)
|
|
user = None
|
|
host = None
|
|
if "!" in prefix:
|
|
user_host = prefix.split("!", 1)[1]
|
|
if "@" in user_host:
|
|
user, host = user_host.split("@", 1)
|
|
else:
|
|
user = user_host
|
|
elif "@" in prefix:
|
|
host = prefix.split("@", 1)[1]
|
|
return ParsedIrcPrefix(nick=nick, user=user, host=host)
|
|
return ParsedIrcPrefix(server=prefix)
|
|
|
|
|
|
def sanitize_irc_text(text: str) -> str:
|
|
text = text.encode("utf-8", errors="surrogateescape").decode("unicode_escape")
|
|
text = re.sub(r"\r?\n", " ", text)
|
|
text = "".join(c for c in text if ord(c) not in CONTROL_CHARS)
|
|
return text.strip()
|
|
|
|
|
|
def split_irc_text(text: str, max_chars: int = 350) -> list[str]:
|
|
chunks = []
|
|
remaining = text
|
|
while len(remaining) > max_chars:
|
|
split_at = remaining.rfind(" ", 0, max_chars)
|
|
if split_at < max_chars * 0.5:
|
|
split_at = max_chars
|
|
chunks.append(remaining[:split_at].strip())
|
|
remaining = remaining[split_at:].strip()
|
|
if remaining:
|
|
chunks.append(remaining)
|
|
return chunks
|
|
|
|
|
|
CTCP_RE = re.compile(r"\x01([^\x01]*)\x01")
|
|
|
|
|
|
def parse_ctcp(text: str) -> tuple[str | None, str | None]:
|
|
m = CTCP_RE.fullmatch(text.strip())
|
|
if not m:
|
|
return None, None
|
|
inner = m.group(1).strip()
|
|
parts = inner.split(" ", 1)
|
|
cmd = parts[0].upper()
|
|
args = parts[1] if len(parts) > 1 else None
|
|
return cmd, args
|
|
|
|
|
|
def strip_ctcp(text: str) -> str:
|
|
text = CTCP_RE.sub(r"\1", text)
|
|
return text.strip() |