kris/ForcePilot
1
0
Fork 0
Code Issues Pull Requests Actions 1 Packages Projects Releases Wiki Activity

refactor(irc): 整理导入并修复代码风格

Browse Source 
1.  调整导入顺序和移除多余空行
2.  重构IRC NAMES命令的成员解析逻辑，正确剥离所有前缀
3.  更新配置schema：修改默认消息块大小为350字节，新增disableBlockStreaming配置项
4.  完善CTCP处理：新增ACTION支持，提取CTCP动作文本
5.  新增IRC线程上下文模拟器类
6.  新增SRV记录解析支持，自动解析IRC服务器域名
7.  新增多种IRC通知处理：ACCOUNT、AWAY、INVITE、CAP
8.  重构消息发送逻辑，添加熔断器和重试机制
9.  重写流式消息合并逻辑，新增智能合并缓冲
10. 扩展IRC CAP支持，新增多个常用扩展能力
11. 修复配置读取逻辑，适配新的配置结构
This commit is contained in:
Kris 2026-05-13 16:10:40 +08:00
parent 8f352d91bb
commit 4e07292a81
10 changed files with 350 additions and 109 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

3
backend/package/yuxi/channels/adapters/irc/__init__.py
View File

@ -7,7 +7,8 @@ from .accounts import (
)
from .adapter import IRCAdapter, IrcStatus
from .commands import sender_allowed_for_commands
from .config_schema import get_config_schema, get_schema_keys, validate_config as validate_config_schema
from .config_schema import get_config_schema, get_schema_keys
from .config_schema import validate_config as validate_config_schema
from .config_ui_hints import get_config_ui_hints
from .configured_state import has_configured_state, is_configured
from .directory import DirectoryGroup, DirectoryUser, list_groups, list_users

262
backend/package/yuxi/channels/adapters/irc/adapter.py
View File

@ -15,6 +15,8 @@ from yuxi.channels.exceptions import (
ChannelAuthenticationError,
ChannelException,
)
from yuxi.channels.infra.circuit_breaker import CircuitBreaker, CircuitBreakerOpenError
from yuxi.channels.sdk.retry import with_retry, RetryConfig
from yuxi.channels.meta import ChannelMeta
from yuxi.channels.models import (
ChannelMessage,
@ -25,6 +27,7 @@ from yuxi.channels.models import (
HealthStatus,
)
from yuxi.channels.registry import register_builtin_adapter
from yuxi.channels.utils.streaming_fallback import SmartCoalesceBuffer
from yuxi.utils.datetime_utils import utc_now_naive
from yuxi.utils.logging_config import logger
@ -218,9 +221,23 @@ class IRCAdapter(BaseChannelAdapter):
)
if self._block_streaming_coalesce is None:
self._block_streaming_coalesce = {}
self._disable_block_streaming = self.config.get("disable_block_streaming", False)
streaming_cfg = self.config.get("streaming", {})
if isinstance(streaming_cfg, dict):
block_cfg = streaming_cfg.get("block", {})
if isinstance(block_cfg, dict) and block_cfg.get("coalesce"):
if "delay_ms" not in self._block_streaming_coalesce:
self._block_streaming_coalesce["delay_ms"] = block_cfg.get(
"coalesce_idle_ms", block_cfg.get("idle_ms", 1000)
)
if "min_chars" not in self._block_streaming_coalesce:
self._block_streaming_coalesce["min_chars"] = block_cfg.get(
"coalesce_min_chars", block_cfg.get("min_chars", 1500)
)
self._disable_block_streaming = self.config.get("disableBlockStreaming", False)
self._streaming_buffer: dict[str, list[str]] = {}
self._streaming_timer: dict[str, asyncio.Task | None] = {}
self._coalesce_buffers: dict[str, SmartCoalesceBuffer] = {}
self.security_config = IRCSecurityConfig(
dm_policy=self.dm_policy,
@ -256,6 +273,7 @@ class IRCAdapter(BaseChannelAdapter):
self._account_id = config.get("_account_id", "default")
self._extra_caps: list[str] = []
self._delivery_tracker: Any = None
self._circuit_breaker = CircuitBreaker(failure_threshold=5, recovery_timeout=60, channel_id="irc")
def _resolve_auto_join(self) -> list:
channels_env = _env_or("channels", "", self.config)
@ -505,24 +523,27 @@ class IRCAdapter(BaseChannelAdapter):
if self._status != ChannelStatus.CONNECTED:
return DeliveryResult(success=False, error="IRC not connected")
try:
chat_id = response.identity.channel_chat_id or self._nick
text = sanitize_outbound_text(response.content)
text = convert_markdown_tables(text, self._markdown_table_mode)
chat_id = response.identity.channel_chat_id or self._nick
text = sanitize_outbound_text(response.content)
text = convert_markdown_tables(text, self._markdown_table_mode)
reply_to_id = ""
if self._reply_enabled and response.reply_to_message_id:
reply_to_id = response.reply_to_message_id
reply_to_id = ""
if self._reply_enabled and response.reply_to_message_id:
reply_to_id = response.reply_to_message_id
await send_privmsg(
self._connection.send_line,
chat_id,
text,
nick=self._nick,
username=self._username,
server=self._server,
reply_to_id=reply_to_id,
chunker_mode=self._chunker_mode,
async def _do_send() -> DeliveryResult:
await with_retry(
lambda: send_privmsg(
self._connection.send_line,
chat_id,
text,
nick=self._nick,
username=self._username,
server=self._server,
reply_to_id=reply_to_id,
chunker_mode=self._chunker_mode,
),
config=RetryConfig(max_retries=2, base_delay=0.5, jitter=True),
)
await self._connection.drain()
self._last_outbound_at = time.monotonic()
@ -531,6 +552,10 @@ class IRCAdapter(BaseChannelAdapter):
self._sent_message_cache.put(msg_id, chat_id, text)
return DeliveryResult(success=True)
try:
return await self._circuit_breaker.call(_do_send)
except CircuitBreakerOpenError:
return DeliveryResult(success=False, error="Circuit breaker open")
except Exception as e:
return DeliveryResult(success=False, error=str(e))
@ -601,9 +626,20 @@ class IRCAdapter(BaseChannelAdapter):
if self._status != ChannelStatus.CONNECTED:
return DeliveryResult(success=False, error="IRC not connected")
if self._disable_block_streaming:
text = sanitize_outbound_text(chunk_text)
text = convert_markdown_tables(text, self._markdown_table_mode)
text = sanitize_outbound_text(chunk_text)
text = convert_markdown_tables(text, self._markdown_table_mode)
coalesce_config = self._block_streaming_coalesce
coalesce_enabled = (
bool(coalesce_config)
and isinstance(coalesce_config, dict)
and not self._disable_block_streaming
)
if coalesce_enabled:
return await self._send_coalesced(chat_id, message_id, text, finished, coalesce_config)
if self._disable_block_streaming or not coalesce_enabled:
await send_privmsg(
self._connection.send_line,
chat_id,
@ -617,81 +653,103 @@ class IRCAdapter(BaseChannelAdapter):
self._last_outbound_at = time.monotonic()
return DeliveryResult(success=True)
coalesce_config = self._block_streaming_coalesce
coalesce_delay = coalesce_config.get("delay_ms", 0) / 1000.0 if isinstance(coalesce_config, dict) else 0
try:
text = sanitize_outbound_text(chunk_text)
text = convert_markdown_tables(text, self._markdown_table_mode)
if coalesce_delay > 0 and not finished:
key = f"{chat_id}:{message_id}"
self._streaming_buffer.setdefault(key, []).append(text)
if self._streaming_timer.get(key):
self._streaming_timer[key].cancel() # type: ignore[union-attr]
self._streaming_timer[key] = asyncio.create_task(self._flush_coalesced(key, chat_id, coalesce_delay))
if finished:
final_text = text
else:
ellipsis = "\u2026"
ellipsis_bytes = len(ellipsis.encode("utf-8"))
chunk_bytes = text.encode("utf-8")
if len(chunk_bytes) < 30:
return DeliveryResult(success=True)
if coalesce_delay > 0 and finished:
key = f"{chat_id}:{message_id}"
buffered = self._streaming_buffer.pop(key, [])
timer = self._streaming_timer.pop(key, None)
if timer:
timer.cancel()
text = "".join(buffered) + text if buffered else text
if finished:
final_text = text
if len(chunk_bytes) + ellipsis_bytes > self.text_chunk_limit:
cut_bytes = self.text_chunk_limit - ellipsis_bytes
final_text = _truncate_utf8(chunk_bytes, cut_bytes) + ellipsis
else:
ellipsis = "\u2026"
ellipsis_bytes = len(ellipsis.encode("utf-8"))
chunk_bytes = text.encode("utf-8")
if len(chunk_bytes) < 30:
return DeliveryResult(success=True)
if len(chunk_bytes) + ellipsis_bytes > self.text_chunk_limit:
cut_bytes = self.text_chunk_limit - ellipsis_bytes
final_text = _truncate_utf8(chunk_bytes, cut_bytes) + ellipsis
else:
final_text = text + ellipsis
final_text = text + ellipsis
await send_privmsg(
self._connection.send_line,
chat_id,
final_text,
nick=self._nick,
username=self._username,
server=self._server,
chunker_mode=self._chunker_mode,
await send_privmsg(
self._connection.send_line,
chat_id,
final_text,
nick=self._nick,
username=self._username,
server=self._server,
chunker_mode=self._chunker_mode,
)
await self._connection.drain()
self._last_outbound_at = time.monotonic()
if finished and message_id:
self._sent_message_cache.put(message_id, chat_id, final_text)
return DeliveryResult(success=True)
async def _send_coalesced(
self,
chat_id: str,
message_id: str,
text: str,
finished: bool,
coalesce_config: dict,
) -> DeliveryResult:
key = f"{chat_id}:{message_id}"
coalesce_delay = coalesce_config.get("delay_ms", 1000) / 1000.0
min_chars = coalesce_config.get("min_chars", 1500)
if key not in self._coalesce_buffers:
self._coalesce_buffers[key] = SmartCoalesceBuffer(
min_chars=min_chars,
idle_ms=int(coalesce_delay * 1000),
max_wait_ms=5000,
)
await self._connection.drain()
self._last_outbound_at = time.monotonic()
if finished and message_id:
self._sent_message_cache.put(message_id, chat_id, final_text)
buf = self._coalesce_buffers[key]
if finished:
buf.feed(text)
result = buf.flush()
self._coalesce_buffers.pop(key, None)
self._streaming_timer.pop(key, None)
if result:
await self._send_irc_text(chat_id, result)
if message_id:
self._sent_message_cache.put(message_id, chat_id, result)
return DeliveryResult(success=True)
except Exception as e:
return DeliveryResult(success=False, error=str(e))
async def _flush_coalesced(self, key: str, chat_id: str, delay: float) -> None:
await asyncio.sleep(delay)
buffered = self._streaming_buffer.pop(key, None)
self._streaming_timer.pop(key, None)
if not buffered:
return
combined = "".join(buffered)
try:
await send_privmsg(
self._connection.send_line,
chat_id,
combined,
nick=self._nick,
username=self._username,
server=self._server,
chunker_mode=self._chunker_mode,
result = buf.feed(text)
if result:
self._coalesce_buffers.pop(key, None)
await self._send_irc_text(chat_id, result)
return DeliveryResult(success=True)
if key not in self._streaming_timer or (self._streaming_timer.get(key) and self._streaming_timer[key].done()):
if key in self._streaming_timer and self._streaming_timer[key] and not self._streaming_timer[key].done():
self._streaming_timer[key].cancel()
self._streaming_timer[key] = asyncio.create_task(
self._flush_coalesce_buffer(key, chat_id, coalesce_delay)
)
await self._connection.drain()
self._last_outbound_at = time.monotonic()
except Exception as e:
logger.warning(f"IRC coalesce flush failed: {e}")
return DeliveryResult(success=True)
async def _flush_coalesce_buffer(self, key: str, chat_id: str, delay: float) -> None:
await asyncio.sleep(delay)
buf = self._coalesce_buffers.pop(key, None)
self._streaming_timer.pop(key, None)
if buf is None:
return
flushed = buf.flush()
if flushed:
await self._send_irc_text(chat_id, flushed)
async def _send_irc_text(self, chat_id: str, text: str) -> None:
await send_privmsg(
self._connection.send_line,
chat_id,
text,
nick=self._nick,
username=self._username,
server=self._server,
chunker_mode=self._chunker_mode,
)
await self._connection.drain()
self._last_outbound_at = time.monotonic()
async def get_user_info(self, channel_user_id: str) -> dict[str, Any]:
return {"nick": resolve_channel_user_id(channel_user_id)}
@ -742,6 +800,10 @@ class IRCAdapter(BaseChannelAdapter):
channel = args[0] if args else ""
nick = extract_nick(prefix)
self._channel_runtime.handle_join(channel, nick)
if len(args) >= 3:
account = args[1] if len(args) > 1 else None
realname = args[2] if len(args) > 2 else None
logger.debug(f"IRC extended-join: {nick} -> {channel} (account={account}, realname={realname})")
elif command == "PART":
channel = args[0] if args else ""
@ -807,7 +869,17 @@ class IRCAdapter(BaseChannelAdapter):
def _send_cap_and_nickuser(self) -> None:
self._connection.send_line("CAP LS 302")
caps = ["server-time"]
caps = [
"server-time",
"account-tag",
"message-ids",
"account-notify",
"away-notify",
"extended-join",
"multi-prefix",
"invite-notify",
"cap-notify",
]
if self._use_sasl:
caps.append("sasl")
caps.extend(self._extra_caps)
@ -879,7 +951,17 @@ class IRCAdapter(BaseChannelAdapter):
logger.info(f"IRC joined channels: {self._channel_runtime.joined_channels}")
async def _on_reconnect_handler(self) -> None:
caps_to_request = ["server-time"]
caps_to_request = [
"server-time",
"account-tag",
"message-ids",
"account-notify",
"away-notify",
"extended-join",
"multi-prefix",
"invite-notify",
"cap-notify",
]
if self._use_sasl:
caps_to_request.append("sasl")
caps_to_request.extend(self._extra_caps)

10
backend/package/yuxi/channels/adapters/irc/channel_runtime.py
View File

@ -25,7 +25,15 @@ class ChannelRuntime:
def handle_names(self, channel: str, names_str: str) -> None:
channel_lower = channel.lower()
members = {name.lstrip("@+") for name in names_str.split() if name}
members = set()
for raw_name in names_str.split():
if not raw_name:
continue
name = raw_name
while name and name[0] in "@+%~&!":
name = name[1:]
if name:
members.add(name)
self._channel_members[channel_lower] = members
logger.debug(f"IRC NAMES {channel_lower}: {len(members)} members")

9
backend/package/yuxi/channels/adapters/irc/config_schema.py
View File

@ -2,7 +2,6 @@ from __future__ import annotations
from typing import Any
_CONFIG_SCHEMA: dict[str, dict[str, Any]] = {
"server": {
"type": "str",
@ -181,7 +180,7 @@ _CONFIG_SCHEMA: dict[str, dict[str, Any]] = {
"text_chunk_limit": {
"type": "int",
"required": False,
"default": 512,
"default": 350,
"description": "Maximum bytes per IRC message chunk",
},
"reply_enabled": {
@ -202,6 +201,12 @@ _CONFIG_SCHEMA: dict[str, dict[str, Any]] = {
"default": {},
"description": "Block streaming coalesce config (delay_ms)",
},
"disableBlockStreaming": {
"type": "bool",
"required": False,
"default": False,
"description": "Disable block streaming mode, send each chunk immediately",
},
"markdown.tableMode": {
"type": "str",
"required": False,

1
backend/package/yuxi/channels/adapters/irc/configured_state.py
View File

@ -2,5 +2,4 @@ from __future__ import annotations
from .accounts import has_configured_state, is_configured
__all__ = ["has_configured_state", "is_configured"]

34
backend/package/yuxi/channels/adapters/irc/connection.py
View File

@ -1,6 +1,7 @@
from __future__ import annotations
import asyncio
import ipaddress
import ssl
import time
from collections.abc import Callable
@ -9,6 +10,14 @@ from typing import Any
from yuxi.utils.logging_config import logger
def _is_ip_address(host: str) -> bool:
try:
ipaddress.ip_address(host)
return True
except ValueError:
return False
class IRCConnection:
PING_TIMEOUT = 30
RECONNECT_DELAY = 5
@ -26,6 +35,8 @@ class IRCConnection:
self._writer: asyncio.StreamWriter | None = None
self._keepalive_task: asyncio.Task | None = None
self._on_reconnect: Callable[[], None] | None = None
self._srv_host: str | None = None
self._srv_port: int | None = None
@property
def reader(self) -> asyncio.StreamReader | None:
@ -39,6 +50,19 @@ class IRCConnection:
self._on_reconnect = callback
async def connect(self) -> None:
host = self._server
port = self._port
if not self._proxy and not _is_ip_address(host):
srv_result = await resolve_srv_record(host)
if srv_result:
self._srv_host, self._srv_port = srv_result
logger.info(
f"IRC DNS SRV resolved: {host} -> {self._srv_host}:{self._srv_port}"
)
host = self._srv_host
port = self._srv_port
if self._proxy:
await self._connect_via_proxy()
elif self._use_tls:
@ -46,19 +70,19 @@ class IRCConnection:
ctx.check_hostname = True
ctx.verify_mode = ssl.CERT_REQUIRED
self._reader, self._writer = await asyncio.open_connection(
host=self._server,
port=self._port,
host=host,
port=port,
ssl=ctx,
server_hostname=self._server,
)
else:
self._reader, self._writer = await asyncio.open_connection(
host=self._server,
port=self._port,
host=host,
port=port,
)
self._reconnect_attempts = 0
self._last_pong = time.monotonic()
logger.info(f"IRC TCP connected: {self._server}:{self._port} (TLS={self._use_tls})")
logger.info(f"IRC TCP connected: {host}:{port} (TLS={self._use_tls})")
async def _connect_via_proxy(self) -> None:
proxy = self._proxy or {}

19
backend/package/yuxi/channels/adapters/irc/ctcp.py
View File

@ -10,9 +10,9 @@ async def handle_ctcp(
send_fn,
source_nick: str,
text: str,
) -> None:
) -> bool:
if not is_ctcp_message(text):
return
return False
command, args = parse_ctcp(text)
logger.debug(f"IRC CTCP from {source_nick}: {command} {args}")
@ -29,8 +29,21 @@ async def handle_ctcp(
reply = format_ctcp("TIME", f"{_time.time():.0f}")
send_raw_line(send_fn, f"NOTICE {source_nick} :{reply}")
elif command == "CLIENTINFO":
reply = format_ctcp("CLIENTINFO", "PING VERSION TIME")
reply = format_ctcp("CLIENTINFO", "PING VERSION TIME ACTION")
send_raw_line(send_fn, f"NOTICE {source_nick} :{reply}")
elif command == "ACTION":
return True
return False
def extract_ctcp_action(text: str) -> str | None:
if not is_ctcp_message(text):
return None
command, args = parse_ctcp(text)
if command == "ACTION":
return args
return None
def supports_ctcp() -> bool:

3
backend/package/yuxi/channels/adapters/irc/gateway.py
View File

@ -1,9 +1,8 @@
from __future__ import annotations
import asyncio
from typing import Any
import warnings
from typing import Any
from yuxi.utils.logging_config import logger

72
backend/package/yuxi/channels/adapters/irc/monitor.py
View File

@ -13,8 +13,8 @@ from .commands import (
has_control_command,
parse_command,
)
from .ctcp import handle_ctcp
from .protocol import extract_nick, is_ctcp_message, parse_irc_line, parse_irc_tags_dict
from .ctcp import extract_ctcp_action, handle_ctcp
from .protocol import extract_nick, is_ctcp_message, parse_irc_line, parse_irc_prefix, parse_irc_tags_dict
from .sanitize import sanitize_inbound_text
if TYPE_CHECKING:
@ -104,6 +104,22 @@ class IRCMonitor:
self._handle_monitor_reply(command, args)
continue
if command == "ACCOUNT":
self._handle_account_notify(prefix, args)
continue
if command == "AWAY":
self._handle_away_notify(prefix, args)
continue
if command == "INVITE":
self._handle_invite_notify(prefix, args)
continue
if command == "CAP":
self._handle_cap_notify(args)
continue
if command == "PRIVMSG":
if len(args) < 2:
continue
@ -118,8 +134,13 @@ class IRCMonitor:
continue
if is_ctcp_message(text):
await handle_ctcp(self._adapter.send_line, sender_nick, text)
continue
action_text = extract_ctcp_action(text)
if action_text is not None:
text = f"* {sender_nick} {action_text}"
logger.debug(f"IRC CTCP ACTION from {sender_nick}: {action_text}")
else:
await handle_ctcp(self._adapter.send_line, sender_nick, text)
continue
if sender_nick == self._adapter.nick:
self._handle_echo_message(tags, msgid, text)
@ -261,6 +282,49 @@ class IRCMonitor:
case "734":
logger.warning(f"MONITOR list full: {targets}")
def _handle_account_notify(self, prefix: str | None, args: list[str]) -> None:
if len(args) < 2:
return
nick = args[0]
account = args[1].lstrip(":")
if account == "*":
logger.debug(f"IRC account-notify: {nick} logged out")
else:
logger.debug(f"IRC account-notify: {nick} logged in as {account}")
def _handle_away_notify(self, prefix: str | None, args: list[str]) -> None:
if not args:
return
nick = extract_nick(prefix)
if len(args) >= 1 and args[0].startswith(":"):
away_msg = " ".join(args).lstrip(":")
logger.debug(f"IRC away-notify: {nick} is away ({away_msg})")
else:
logger.debug(f"IRC away-notify: {nick} is back")
def _handle_invite_notify(self, prefix: str | None, args: list[str]) -> None:
if len(args) < 2:
return
inviter = extract_nick(prefix)
target = args[0]
channel = args[1].lstrip(":")
logger.debug(f"IRC invite-notify: {inviter} invited {target} to {channel}")
def _handle_cap_notify(self, args: list[str]) -> None:
if len(args) < 3:
return
target = args[0]
subcmd = args[1]
cap_name = args[2].lstrip(":")
if subcmd == "NEW":
logger.info(f"IRC cap-notify: {target} added CAP {cap_name}")
elif subcmd == "DEL":
logger.info(f"IRC cap-notify: {target} removed CAP {cap_name}")
elif subcmd == "ACK":
pass
elif subcmd == "NAK":
logger.warning(f"IRC cap-notify: {target} NAK for {cap_name}")
def _add_dedup(self, msgid: str) -> None:
if len(self._dedup_window) >= self._dedup_max:
self._dedup_window.clear()

46
backend/package/yuxi/channels/adapters/irc/thread_simulator.py Normal file
View File

@ -0,0 +1,46 @@
from __future__ import annotations
from datetime import datetime
class IRCThreadSimulator:
def __init__(self, context_size: int = 10):
self._channel_contexts: dict[str, list[dict]] = {}
self._dm_contexts: dict[str, list[dict]] = {}
self._context_size = context_size
def add_message(self, channel: str, sender: str, content: str, is_dm: bool = False) -> None:
key = sender if is_dm else channel
contexts = self._dm_contexts if is_dm else self._channel_contexts
window = contexts.get(key)
if window is None:
window = []
contexts[key] = window
window.append(
{
"sender": sender,
"content": content,
"timestamp": datetime.now(),
}
)
if len(window) > self._context_size:
window.pop(0)
def get_context_text(self, channel: str, is_dm: bool = False) -> str:
key = channel if is_dm else channel
contexts = self._dm_contexts if is_dm else self._channel_contexts
window = contexts.get(key, [])
lines = [f"<{msg['sender']}> {msg['content']}" for msg in window]
return "\n".join(lines)
@staticmethod
def format_reply_prefix(target_nick: str) -> str:
return f"@{target_nick}: "
def clear_context(self, channel: str) -> None:
self._channel_contexts.pop(channel, None)
self._dm_contexts.pop(channel, None)