From 18d1ea2aac1c60381f22c6bf3f9a2f1e1c066bb3 Mon Sep 17 00:00:00 2001 From: Kris <2893855659@qq.com> Date: Wed, 13 May 2026 16:16:02 +0800 Subject: [PATCH] =?UTF-8?q?refactor(twitch):=20=E9=87=8D=E6=9E=84Twitch?= =?UTF-8?q?=E9=80=82=E9=85=8D=E5=99=A8=EF=BC=8C=E6=96=B0=E5=A2=9EHelix=20A?= =?UTF-8?q?PI=E6=94=AF=E6=8C=81=E4=B8=8E=E5=8A=9F=E8=83=BD=E4=BC=98?= =?UTF-8?q?=E5=8C=96?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 本次提交对Twitch适配器进行了全面升级与优化： 1. 修复UTF8截断逻辑，避免越界访问 2. 重构群聊策略配置，标准化mention相关规则 3. 新增消息缓存管理器，支持通过消息ID查询已发送消息 4. 更新配置schema，新增prefer_helix_send开关和deprecated策略自动转换 5. 新增CLEARMSG和ROOMSTATE IRC消息解析，补充事件订阅支持 6. 优化令牌刷新逻辑，增加重试机制与退避策略 7. 新增Helix API聊天消息发送、删除和公告功能 8. 扩展事件订阅类型，新增直播状态、频道更新等系统事件 9. 新增reply、delete_message、announcement等动作支持，完善操作能力 10. 重构流式发送逻辑，新增进度指示器和配置项 11. 优化重连策略，增加指数退避与计数重置 --- .../yuxi/channels/adapters/twitch/actions.py | 125 ++++++++++++++-- .../yuxi/channels/adapters/twitch/adapter.py | 124 ++++++++++++--- .../channels/adapters/twitch/auth_provider.py | 56 ++++--- .../channels/adapters/twitch/config_schema.py | 10 +- .../adapters/twitch/eventsub_client.py | 141 +++++++++++++++++- .../yuxi/channels/adapters/twitch/helix.py | 92 +++++++++++- .../channels/adapters/twitch/normalizer.py | 51 ++++++- .../adapters/twitch/outbound_cache.py | 13 +- .../yuxi/channels/adapters/twitch/send.py | 1 + .../yuxi/channels/adapters/twitch/session.py | 7 +- 10 files changed, 551 insertions(+), 69 deletions(-) diff --git a/backend/package/yuxi/channels/adapters/twitch/actions.py b/backend/package/yuxi/channels/adapters/twitch/actions.py index 36449f72..636a4a39 100644 --- a/backend/package/yuxi/channels/adapters/twitch/actions.py +++ b/backend/package/yuxi/channels/adapters/twitch/actions.py @@ -5,8 +5,6 @@ from typing import Any from yuxi.channels.models import ChannelIdentity, ChannelResponse, DeliveryResult from yuxi.channels.protocols.actions import ActionContext -_ACTION_HANDLERS: dict[str, str] = {} - def _build_response(ctx: ActionContext, adapter: Any, content_key: str = "content") -> ChannelResponse | None: content = ctx.get(content_key, ctx.get("media_url", "")) @@ -26,18 +24,97 @@ async def handle_action(ctx: ActionContext, adapter: Any) -> DeliveryResult: response = _build_response(ctx, adapter, "content") if response is None: return DeliveryResult(success=False, error="missing chat_id or content") + if adapter.config.get("prefer_helix_send", False): + result = await adapter._send_via_helix(adapter.format_outbound(response)["target"], response.content) + if result.success: + return result return await adapter.send(response) + if ctx.action == "send_media": response = _build_response(ctx, adapter, "media_url") if response is None: return DeliveryResult(success=False, error="missing chat_id or media_url") return await adapter.send_media(ctx.chat_id, "url", response.content) + if ctx.action == "reply": + chat_id = ctx.get("chat_id", "") + content = ctx.get("content", "") + reply_to_msg_id = ctx.get("reply_to_msg_id", "") + if not chat_id or not content: + return DeliveryResult(success=False, error="missing chat_id or content") + if not reply_to_msg_id: + return DeliveryResult(success=False, error="missing reply_to_msg_id") + result = await adapter._send_via_helix(chat_id, content, reply_msg_id=reply_to_msg_id) + if result.success: + return result + prefixed = f"@{ctx.get('reply_to_username', 'user')} {content}" + response = ChannelResponse( + identity=ChannelIdentity( + channel_id="twitch", + channel_type=adapter.channel_type, + channel_user_id="", + channel_chat_id=chat_id, + ), + content=prefixed, + ) + return await adapter.send(response) + + if ctx.action in ("unsend", "delete_message"): + chat_id = ctx.get("chat_id", "") + message_id = ctx.get("message_id", "") + if not chat_id or not adapter._helix: + return DeliveryResult(success=False, error="missing chat_id or helix not available") + if not message_id: + cached = adapter._outbound_cache.find_by_message_id(ctx.get("content", "")) + if cached: + message_id = cached.get("message_id", "") + if not message_id: + return DeliveryResult(success=False, error="missing message_id for message deletion") + + broadcaster_id = await adapter._resolve_broadcaster_id(chat_id) + if not broadcaster_id: + return DeliveryResult(success=False, error="broadcaster_not_found") + + moderator_id = adapter._bot_user_id + if not moderator_id: + return DeliveryResult(success=False, error="bot_user_id_unknown") + + success = await adapter._helix.delete_chat_message( + broadcaster_id=broadcaster_id, + moderator_id=moderator_id, + message_id=message_id, + ) + if success: + return DeliveryResult(success=True, metadata={"messageId": message_id}) + return DeliveryResult(success=False, error="helix_delete_failed") + + if ctx.action == "announcement": + chat_id = ctx.get("chat_id", "") + content = ctx.get("content", "") + color = ctx.get("color", "primary") + if not chat_id or not content or not adapter._helix: + return DeliveryResult(success=False, error="missing chat_id or content or helix not available") + + broadcaster_id = await adapter._resolve_broadcaster_id(chat_id) + if not broadcaster_id: + return DeliveryResult(success=False, error="broadcaster_not_found") + + moderator_id = adapter._bot_user_id + if not moderator_id: + return DeliveryResult(success=False, error="bot_user_id_unknown") + + success = await adapter._helix.send_chat_announcement( + broadcaster_id=broadcaster_id, + moderator_id=moderator_id, + message=content, + color=color, + ) + if success: + return DeliveryResult(success=True) + return DeliveryResult(success=False, error="announcement_failed") + unsupported_hints: dict[str, str] = { - "reply": "Twitch IRC does not support message replies", "edit": "Twitch IRC does not support message editing", - "unsend": "Twitch IRC does not support message deletion", - "delete_message": "Twitch IRC does not support message deletion", "edit_message": "Twitch IRC does not support message editing", "reactions": "Twitch IRC does not support reactions", "send_reaction": "Twitch IRC does not support reactions", @@ -51,7 +128,7 @@ async def handle_action(ctx: ActionContext, adapter: Any) -> DeliveryResult: def supports_action(action: str) -> bool: - return action in {"send", "send_media"} + return action in {"send", "send_media", "reply", "unsend", "delete_message", "announcement"} def describe_message_tool() -> dict[str, Any]: @@ -70,6 +147,37 @@ def describe_message_tool() -> dict[str, Any]: "media_url": {"type": "string", "description": "Media URL to send"}, }, }, + "reply": { + "description": "Reply to a message in a Twitch channel via Helix API", + "parameters": { + "chat_id": {"type": "string", "description": "Channel name (e.g. #channel)"}, + "content": {"type": "string", "description": "Reply text content"}, + "reply_to_msg_id": {"type": "string", "description": "Message ID to reply to"}, + "reply_to_username": {"type": "string", "description": "Username being replied to (for IRC fallback)"}, + }, + }, + "unsend": { + "description": "Delete a message from a Twitch channel via Helix API", + "parameters": { + "chat_id": {"type": "string", "description": "Channel name (e.g. #channel)"}, + "message_id": {"type": "string", "description": "Message ID to delete"}, + }, + }, + "delete_message": { + "description": "Delete a message from a Twitch channel via Helix API", + "parameters": { + "chat_id": {"type": "string", "description": "Channel name (e.g. #channel)"}, + "message_id": {"type": "string", "description": "Message ID to delete"}, + }, + }, + "announcement": { + "description": "Send a colored announcement to a Twitch channel", + "parameters": { + "chat_id": {"type": "string", "description": "Channel name (e.g. #channel)"}, + "content": {"type": "string", "description": "Announcement text"}, + "color": {"type": "string", "description": "Announcement color: blue/green/orange/purple/primary"}, + }, + }, } @@ -86,19 +194,16 @@ def resolve_execution_mode(action: str) -> str: def get_action_stats() -> dict[str, list[str]]: return { - "implemented": ["send", "send_media"], + "implemented": ["send", "send_media", "reply", "unsend", "delete_message", "announcement"], "planned": [], "unsupported": [ - "reply", "edit", - "unsend", "reactions", "polls", "native_commands", "pin", "unpin", "send_reaction", - "delete_message", "edit_message", ], } diff --git a/backend/package/yuxi/channels/adapters/twitch/adapter.py b/backend/package/yuxi/channels/adapters/twitch/adapter.py index d79b860a..e80f59a4 100644 --- a/backend/package/yuxi/channels/adapters/twitch/adapter.py +++ b/backend/package/yuxi/channels/adapters/twitch/adapter.py @@ -10,11 +10,11 @@ from typing import Any, ClassVar from yuxi.channels.base import BaseChannelAdapter from yuxi.channels.capabilities import ChannelCapabilities -from yuxi.channels.meta import ChannelMeta -from yuxi.channels.infra.circuit_breaker import CircuitBreaker, CircuitBreakerOpenError from yuxi.channels.exceptions import ( ChannelAuthenticationError, ) +from yuxi.channels.infra.circuit_breaker import CircuitBreaker, CircuitBreakerOpenError +from yuxi.channels.meta import ChannelMeta from yuxi.channels.models import ( ChannelMessage, ChannelResponse, @@ -26,28 +26,38 @@ from yuxi.channels.models import ( from yuxi.channels.registry import register_builtin_adapter from yuxi.utils.logging_config import logger +from .actions import ( + describe_message_tool, + extract_target_from_args, + handle_action, + resolve_execution_mode, +) +from .actions import ( + supports_action as _supports_action, +) from .auth_provider import create_auth_provider from .deduplicator import MessageDeduplicator from .eventsub_client import EventSubListener from .helix import HelixClient -from .irc_parser import parse_badges, parse_irc_line, extract_nick +from .irc_parser import extract_nick, parse_badges, parse_irc_line from .markdown_utils import strip_twitch_markdown +from .normalizer import normalize_irc_message, resolve_mentions from .outbound_cache import OutboundCacheManager from .pairing import PairingStore, check_pairing_policy -from .actions import ( - handle_action, - supports_action as _supports_action, - describe_message_tool, - extract_target_from_args, - resolve_execution_mode, -) -from .normalizer import normalize_irc_message, resolve_mentions from .probe import get_app_access_token, validate_token from .rate_limiter import RateLimiter -from .send import format_privmsg_line, format_action_line, format_pong, format_cap_req, find_utf8_cut +from .send import find_utf8_cut, format_action_line, format_cap_req, format_pong, format_privmsg_line from .session import check_allowed_roles, check_group_policy from .token_utils import ensure_oauth_prefix +PROGRESS_BAR_WIDTH = 8 + + +def _progress_bar(text_len_hint: int, max_chars: int) -> str: + ratio = min(text_len_hint / max(max_chars, 1), 1.0) + filled = int(ratio * PROGRESS_BAR_WIDTH) + return "▓" * filled + "░" * (PROGRESS_BAR_WIDTH - filled) + @register_builtin_adapter class TwitchAdapter(BaseChannelAdapter): @@ -63,15 +73,15 @@ class TwitchAdapter(BaseChannelAdapter): delivery_mode: ClassVar[str] = "direct" capabilities = ChannelCapabilities( - chat_types=["group"], + chat_types=["group", "direct"], supports_markdown=False, supports_streaming=True, streaming_modes=["off", "block"], text_chunk_limit=500, - max_media_size_mb=0, # IRC protocol does not support media upload; URLs sent as text - reply=False, + max_media_size_mb=0, + reply=True, edit=False, - unsend=False, + unsend=True, reactions=False, polls=False, native_commands=False, @@ -102,9 +112,11 @@ class TwitchAdapter(BaseChannelAdapter): self._bot_user_id: str | None = None self._bot_username: str | None = None + self._broadcaster_ids: dict[str, str] = {} self._irc_task: asyncio.Task | None = None self._reconnect_task: asyncio.Task | None = None + self._reconnect_attempts: int = 0 self._deduplicator = MessageDeduplicator() @@ -275,6 +287,14 @@ class TwitchAdapter(BaseChannelAdapter): if not content.strip(): return DeliveryResult(success=True, metadata={"messageId": "skipped"}) + prefer_helix = self.config.get("prefer_helix_send", False) + if prefer_helix: + target = self.format_outbound(response)["target"] + result = await self._send_via_helix(target, content) + if result.success: + return result + logger.info("[Twitch] Helix send failed, falling back to IRC PRIVMSG") + async def _do_send(): target = self.format_outbound(response)["target"] chunks = self._split_irc_text(content, target) @@ -291,8 +311,19 @@ class TwitchAdapter(BaseChannelAdapter): if not chunk.strip(): return DeliveryResult(success=True, metadata={"messageId": "skipped"}) - coalesce_min_chars = self.config.get("stream_coalesce_min_chars", 30) - coalesce_max_delay_ms = self.config.get("stream_coalesce_max_delay_ms", 0) + stream_config = self.config.get("streaming", {}) + block_cfg = stream_config.get("block", {}) if isinstance(stream_config, dict) else {} + + coalesce_min_chars = self.config.get( + "stream_coalesce_min_chars", + block_cfg.get("coalesce_min_chars", 30), + ) + coalesce_max_delay_ms = self.config.get( + "stream_coalesce_max_delay_ms", + block_cfg.get("coalesce_idle_ms", 0), + ) + + show_progress = stream_config.get("progress_indicator", True) buffer_key = f"{chat_id}:{msg_id}" if coalesce_max_delay_ms > 0 and not finished: @@ -307,12 +338,12 @@ class TwitchAdapter(BaseChannelAdapter): buf["chunks"].append(chunk) total = "".join(buf["chunks"]) if len(total.encode("utf-8")) >= coalesce_min_chars: - return await self._send_with_protection(self._flush_stream_buffer, buffer_key) + return await self._send_with_protection(self._flush_stream_buffer, buffer_key, progress=show_progress) if buf["task"] is None or buf["task"].done(): buf["task"] = asyncio.create_task(self._delayed_flush(buffer_key, coalesce_max_delay_ms / 1000.0)) return DeliveryResult(success=True) elif finished and buffer_key in self._stream_buffers: - return await self._send_with_protection(self._flush_stream_buffer, buffer_key) + return await self._send_with_protection(self._flush_stream_buffer, buffer_key, progress=show_progress) else: if finished: @@ -330,8 +361,10 @@ class TwitchAdapter(BaseChannelAdapter): silent = self.config.get("silent", False) fmt = format_action_line if silent else format_privmsg_line prefix_len = len(f"PRIVMSG {chat_id} :") - suffix = "…" - text = chunk[: 510 - prefix_len - len(suffix.encode("utf-8"))] + suffix + max_chars = 500 - prefix_len + suffix = _progress_bar(len(chunk), max_chars) if show_progress else "…" + suffix_len = len(suffix.encode("utf-8")) + text = chunk[: max_chars - suffix_len] + suffix if not await self._rate_limiter.acquire(): return DeliveryResult(success=False, error="rate_limit_exceeded") self._writer.write(fmt(chat_id, text).encode("utf-8") + b"\r\n") @@ -344,7 +377,7 @@ class TwitchAdapter(BaseChannelAdapter): await asyncio.sleep(delay_sec) await self._flush_stream_buffer(buffer_key) - async def _flush_stream_buffer(self, buffer_key: str) -> DeliveryResult: + async def _flush_stream_buffer(self, buffer_key: str, progress: bool = True) -> DeliveryResult: buf = self._stream_buffers.pop(buffer_key, None) if buf is None: return DeliveryResult(success=True) @@ -385,6 +418,45 @@ class TwitchAdapter(BaseChannelAdapter): def _record_outbound(self, target: str, content: str) -> None: self._outbound_cache.record(target, content) + async def _resolve_broadcaster_id(self, channel: str) -> str | None: + channel_name = channel.lstrip("#").lower() + if channel_name in self._broadcaster_ids: + return self._broadcaster_ids[channel_name] + + if not self._helix: + return None + + user = await self._helix.get_user_by_name(channel_name) + if user: + self._broadcaster_ids[channel_name] = user["id"] + return user["id"] + return None + + async def _send_via_helix(self, target: str, content: str, reply_msg_id: str | None = None) -> DeliveryResult: + if not self._helix: + return DeliveryResult(success=False, error="helix_not_available") + + broadcaster_id = await self._resolve_broadcaster_id(target) + if not broadcaster_id: + return DeliveryResult(success=False, error="broadcaster_not_found") + + sender_id = self._bot_user_id + if not sender_id: + return DeliveryResult(success=False, error="bot_user_id_unknown") + + result = await self._helix.send_chat_message( + broadcaster_id=broadcaster_id, + sender_id=sender_id, + message=content, + reply_parent_msg_id=reply_msg_id, + ) + if result is None: + return DeliveryResult(success=False, error="helix_send_failed") + + message_id = result.get("message_id", "") + self._outbound_cache.record(target, content, message_id) + return DeliveryResult(success=True, metadata={"messageId": message_id}) + def get_outbound_cache(self) -> list[dict[str, Any]]: return self._outbound_cache.get_all() @@ -838,10 +910,14 @@ class TwitchAdapter(BaseChannelAdapter): await self._refresh_token_if_needed() - await asyncio.sleep(3) + delay = min(3 * (2**self._reconnect_attempts), 120) + self._reconnect_attempts += 1 + logger.info(f"[Twitch] IRC reconnecting in {delay}s (attempt {self._reconnect_attempts})") + await asyncio.sleep(delay) try: await self._connect_irc() self._status = ChannelStatus.CONNECTED + self._reconnect_attempts = 0 logger.info("[Twitch] IRC reconnected") except asyncio.CancelledError: self._status = ChannelStatus.DISCONNECTED diff --git a/backend/package/yuxi/channels/adapters/twitch/auth_provider.py b/backend/package/yuxi/channels/adapters/twitch/auth_provider.py index eda20fe0..7912d8dd 100644 --- a/backend/package/yuxi/channels/adapters/twitch/auth_provider.py +++ b/backend/package/yuxi/channels/adapters/twitch/auth_provider.py @@ -1,5 +1,6 @@ from __future__ import annotations +import asyncio import time from abc import ABC, abstractmethod from typing import Any @@ -74,26 +75,41 @@ class RefreshingAuthProvider(AuthProvider): async def _refresh(self) -> bool: if not self._client_id or not self._client_secret or not self._refresh_token: return False - try: - new_tokens = await refresh_access_token(self._client_id, self._client_secret, self._refresh_token) - if new_tokens is None: - return False - new_access = new_tokens.get("access_token", "") - new_refresh = new_tokens.get("refresh_token", "") - expires_in = new_tokens.get("expires_in", 0) - if new_access: - self._access_token = new_access - if new_refresh: - self._refresh_token = new_refresh - if expires_in: - self._token_obtained_at = time.time() - self._token_expires_in = expires_in - self._token_expires_at = self._token_obtained_at + expires_in - 300 - logger.info(f"[TwitchAuth] Token refreshed, expires in {expires_in}s") - return True - except Exception as e: - logger.error(f"[TwitchAuth] Token refresh error: {e}") - return False + + max_retries = 3 + for attempt in range(max_retries): + try: + new_tokens = await refresh_access_token(self._client_id, self._client_secret, self._refresh_token) + if new_tokens is None: + if attempt < max_retries - 1: + delay = 2**attempt + logger.warning(f"[TwitchAuth] Token refresh attempt {attempt + 1} failed, retrying in {delay}s") + await asyncio.sleep(delay) + continue + return False + new_access = new_tokens.get("access_token", "") + new_refresh = new_tokens.get("refresh_token", "") + expires_in = new_tokens.get("expires_in", 0) + if new_access: + self._access_token = new_access + if new_refresh: + self._refresh_token = new_refresh + if expires_in: + self._token_obtained_at = time.time() + self._token_expires_in = expires_in + self._token_expires_at = self._token_obtained_at + expires_in - 300 + logger.info(f"[TwitchAuth] Token refreshed, expires in {expires_in}s") + return True + except Exception as e: + if attempt < max_retries - 1: + delay = 2**attempt + logger.warning( + f"[TwitchAuth] Token refresh error (attempt {attempt + 1}): {e}, retrying in {delay}s" + ) + await asyncio.sleep(delay) + else: + logger.error(f"[TwitchAuth] Token refresh failed after {max_retries} attempts: {e}") + return False def _is_expired(self) -> bool: if self._token_expires_at is None: diff --git a/backend/package/yuxi/channels/adapters/twitch/config_schema.py b/backend/package/yuxi/channels/adapters/twitch/config_schema.py index 7cf9629a..89e51972 100644 --- a/backend/package/yuxi/channels/adapters/twitch/config_schema.py +++ b/backend/package/yuxi/channels/adapters/twitch/config_schema.py @@ -32,6 +32,7 @@ class TwitchAccountSchema(BaseModel): silent: bool = False dm_policy: str = "pairing" probe_timeout_ms: int = 10000 + prefer_helix_send: bool = False class TwitchConfigSchema(BaseModel): @@ -41,7 +42,7 @@ class TwitchConfigSchema(BaseModel): client_secret: str = "" refresh_token: str = "" channels: list[str] = [] - group_policy: str = Field(default="open", pattern=r"^(open|allowlist|disabled|mention_only)$") + group_policy: str = Field(default="open", pattern=r"^(open|allowlist|disabled|mention|mention_only)$") require_mention: bool = True allowedRoles: list[str] = [] group_allow_from: list[str] = [] @@ -59,6 +60,7 @@ class TwitchConfigSchema(BaseModel): silent: bool = False dm_policy: str = "pairing" probe_timeout_ms: int = 10000 + prefer_helix_send: bool = False accounts: dict[str, TwitchAccountSchema] = {} defaultAccount: str = "" @@ -82,7 +84,7 @@ def validate_twitch_config(config: dict[str, Any]) -> list[str]: errors.append("at least one channel is required in 'channels' list") group_policy = config.get("group_policy", "open") - valid_policies = {"open", "allowlist", "disabled", "mention_only"} + valid_policies = {"open", "allowlist", "disabled", "mention", "mention_only"} if group_policy not in valid_policies: errors.append(f"group_policy value invalid: '{group_policy}', valid values: {valid_policies}") @@ -146,6 +148,10 @@ def super_refine_twitch_config(config: dict[str, Any]) -> dict[str, Any]: logger.info("[TwitchConfig] Converting legacy 'allowall' to 'open'") refined["group_policy"] = "open" + if group_policy == "mention_only": + logger.info("[TwitchConfig] 'mention_only' is deprecated, use 'mention' instead (auto-converted)") + refined["group_policy"] = "mention" + group_allow_from = refined.get("group_allow_from", refined.get("groupAllowFrom", [])) if group_policy == "open" and ("*" in group_allow_from): logger.info("[TwitchConfig] group_policy is 'open', wildcard in allow_from is redundant") diff --git a/backend/package/yuxi/channels/adapters/twitch/eventsub_client.py b/backend/package/yuxi/channels/adapters/twitch/eventsub_client.py index 68e790ec..bf18c539 100644 --- a/backend/package/yuxi/channels/adapters/twitch/eventsub_client.py +++ b/backend/package/yuxi/channels/adapters/twitch/eventsub_client.py @@ -1,7 +1,7 @@ from __future__ import annotations import asyncio -from datetime import datetime, UTC +from datetime import UTC, datetime from typing import Any import aiohttp @@ -41,6 +41,11 @@ class EventSubListener: {"type": "channel.unban", "version": "1"}, {"type": "channel.moderator.add", "version": "1"}, {"type": "channel.moderator.remove", "version": "1"}, + {"type": "stream.online", "version": "1"}, + {"type": "stream.offline", "version": "1"}, + {"type": "channel.update", "version": "2"}, + {"type": "channel.chat.clear", "version": "1"}, + {"type": "channel.chat.message_delete", "version": "1"}, ] def __init__( @@ -231,6 +236,18 @@ class EventSubListener: if subscription_type == "channel.moderator.remove": return self._normalize_mod_remove(event, broadcaster_id) + if subscription_type == "stream.online": + return self._normalize_stream_online(event, broadcaster_id) + if subscription_type == "stream.offline": + return self._normalize_stream_offline(event, broadcaster_id) + + if subscription_type == "channel.update": + return self._normalize_channel_update(event, broadcaster_id) + if subscription_type == "channel.chat.clear": + return self._normalize_chat_clear(event, broadcaster_id) + if subscription_type == "channel.chat.message_delete": + return self._normalize_chat_message_delete(event, broadcaster_id) + return None @staticmethod @@ -527,3 +544,125 @@ class EventSubListener: metadata={"event": "moderator_remove"}, timestamp=datetime.now(UTC), ) + + @staticmethod + def _normalize_stream_online(event: dict, broadcaster_id: str) -> ChannelMessage: + broadcaster_name = event.get("broadcaster_user_name", "") + stream_type = event.get("type", "live") + started_at = event.get("started_at", "") + + return ChannelMessage( + identity=ChannelIdentity( + channel_id="twitch", + channel_type=ChannelType.TWITCH, + channel_user_id=broadcaster_id, + channel_chat_id=f"#broadcaster_{broadcaster_id}", + channel_message_id=f"eventsub:stream_online:{event.get('id')}", + ), + event_type=EventType.SYSTEM_EVENT, + message_type=MessageType.TEXT, + chat_type=ChatType.GROUP, + content=f"🔴 {broadcaster_name} went live!", + metadata={ + "event": "stream_online", + "type": stream_type, + "started_at": started_at, + }, + timestamp=datetime.now(UTC), + ) + + @staticmethod + def _normalize_stream_offline(event: dict, broadcaster_id: str) -> ChannelMessage: + broadcaster_name = event.get("broadcaster_user_name", "") + + return ChannelMessage( + identity=ChannelIdentity( + channel_id="twitch", + channel_type=ChannelType.TWITCH, + channel_user_id=broadcaster_id, + channel_chat_id=f"#broadcaster_{broadcaster_id}", + channel_message_id=f"eventsub:stream_offline:{event.get('id')}", + ), + event_type=EventType.SYSTEM_EVENT, + message_type=MessageType.TEXT, + chat_type=ChatType.GROUP, + content=f"⚫ {broadcaster_name} went offline", + metadata={ + "event": "stream_offline", + }, + timestamp=datetime.now(UTC), + ) + + @staticmethod + def _normalize_channel_update(event: dict, broadcaster_id: str) -> ChannelMessage: + broadcaster_name = event.get("broadcaster_user_name", "") + title = event.get("title", "") + category_name = event.get("category_name", "") + language = event.get("language", "") + + return ChannelMessage( + identity=ChannelIdentity( + channel_id="twitch", + channel_type=ChannelType.TWITCH, + channel_user_id=broadcaster_id, + channel_chat_id=f"#broadcaster_{broadcaster_id}", + channel_message_id=f"eventsub:channel_update:{event.get('id')}", + ), + event_type=EventType.SYSTEM_EVENT, + message_type=MessageType.TEXT, + chat_type=ChatType.GROUP, + content=f"📺 {broadcaster_name} updated channel: {title}", + metadata={ + "event": "channel_update", + "title": title, + "category": category_name, + "language": language, + }, + timestamp=datetime.now(UTC), + ) + + @staticmethod + def _normalize_chat_clear(event: dict, broadcaster_id: str) -> ChannelMessage: + return ChannelMessage( + identity=ChannelIdentity( + channel_id="twitch", + channel_type=ChannelType.TWITCH, + channel_user_id="twitch_system", + channel_chat_id=f"#broadcaster_{broadcaster_id}", + channel_message_id=f"eventsub:chat_clear:{event.get('id')}", + ), + event_type=EventType.SYSTEM_EVENT, + message_type=MessageType.TEXT, + chat_type=ChatType.GROUP, + content=f"🧹 Chat cleared in #{event.get('broadcaster_user_name', '')}", + metadata={ + "event": "chat_clear", + }, + timestamp=datetime.now(UTC), + ) + + @staticmethod + def _normalize_chat_message_delete(event: dict, broadcaster_id: str) -> ChannelMessage: + target_user_name = event.get("target_user_name", "") + target_user_id = event.get("target_user_id", "") + message_id = event.get("message_id", "") + + return ChannelMessage( + identity=ChannelIdentity( + channel_id="twitch", + channel_type=ChannelType.TWITCH, + channel_user_id=target_user_id or "twitch_system", + channel_chat_id=f"#broadcaster_{broadcaster_id}", + channel_message_id=f"eventsub:msg_delete:{event.get('id')}", + ), + event_type=EventType.MESSAGE_DELETED, + message_type=MessageType.TEXT, + chat_type=ChatType.GROUP, + content="", + metadata={ + "event": "message_delete", + "target_user_name": target_user_name, + "target_message_id": message_id, + }, + timestamp=datetime.now(UTC), + ) diff --git a/backend/package/yuxi/channels/adapters/twitch/helix.py b/backend/package/yuxi/channels/adapters/twitch/helix.py index 2bf0570c..0bbc7c03 100644 --- a/backend/package/yuxi/channels/adapters/twitch/helix.py +++ b/backend/package/yuxi/channels/adapters/twitch/helix.py @@ -46,16 +46,15 @@ class HelixClient: } async def _request_with_backoff(self, method: str, url: str, **kwargs) -> aiohttp.ClientResponse | None: - last_status: int | None = None for attempt in range(self.MAX_RETRIES): try: async with self._ensure_session.request(method, url, **kwargs) as resp: if resp.status not in self.RETRY_STATUSES or attempt == self.MAX_RETRIES - 1: return resp - last_status = resp.status except (TimeoutError, aiohttp.ClientError) as e: if attempt == self.MAX_RETRIES - 1: - raise + logger.error(f"Helix {method} {url} connection failed after {self.MAX_RETRIES} retries: {e}") + return None logger.warning(f"Helix {method} {url} attempt {attempt + 1} failed: {e}") delay = min(self.BASE_BACKOFF * (2**attempt), self.MAX_BACKOFF) @@ -190,3 +189,90 @@ class HelixClient: except aiohttp.ClientError as e: logger.error(f"EventSub delete subscription error: {e}") return False + + async def send_chat_message( + self, + broadcaster_id: str, + sender_id: str, + message: str, + reply_parent_msg_id: str | None = None, + ) -> dict[str, Any] | None: + url = f"{self.BASE_URL}/chat/messages" + body: dict[str, Any] = { + "broadcaster_id": broadcaster_id, + "sender_id": sender_id, + "message": message, + } + if reply_parent_msg_id: + body["reply_parent_msg_id"] = reply_parent_msg_id + + resp = await self._request_with_backoff( + "POST", + url, + headers={**self._headers(), "Content-Type": "application/json"}, + json=body, + ) + if resp is None: + return None + try: + if resp.status == 200: + data = await resp.json() + return data.get("data", [{}])[0] if data.get("data") else None + logger.error(f"Helix send_chat_message failed: {resp.status} {await resp.text()}") + return None + except aiohttp.ClientError as e: + logger.error(f"Helix send_chat_message connection error: {e}") + return None + + async def delete_chat_message( + self, + broadcaster_id: str, + moderator_id: str, + message_id: str, + ) -> bool: + url = f"{self.BASE_URL}/chat/messages" + params = { + "broadcaster_id": broadcaster_id, + "moderator_id": moderator_id, + "message_id": message_id, + } + resp = await self._request_with_backoff("DELETE", url, headers=self._headers(), params=params) + if resp is None: + return False + try: + return resp.status in (200, 204) + except aiohttp.ClientError as e: + logger.error(f"Helix delete_chat_message error: {e}") + return False + + async def send_chat_announcement( + self, + broadcaster_id: str, + moderator_id: str, + message: str, + color: str = "primary", + ) -> bool: + url = f"{self.BASE_URL}/chat/announcements" + valid_colors = {"blue", "green", "orange", "purple", "primary"} + if color not in valid_colors: + color = "primary" + + body = { + "broadcaster_id": broadcaster_id, + "moderator_id": moderator_id, + "message": message, + "color": color, + } + resp = await self._request_with_backoff( + "POST", + url, + headers={**self._headers(), "Content-Type": "application/json"}, + json=body, + ) + if resp is None: + return False + try: + return resp.status in (200, 204) + except aiohttp.ClientError as e: + logger.error(f"Helix send_chat_announcement error: {e}") + return False diff --git a/backend/package/yuxi/channels/adapters/twitch/normalizer.py b/backend/package/yuxi/channels/adapters/twitch/normalizer.py index 0cd0a251..3dbf89f9 100644 --- a/backend/package/yuxi/channels/adapters/twitch/normalizer.py +++ b/backend/package/yuxi/channels/adapters/twitch/normalizer.py @@ -1,6 +1,6 @@ from __future__ import annotations -from datetime import datetime, UTC +from datetime import UTC, datetime from typing import Any from yuxi.channels.models import ( @@ -16,9 +16,9 @@ from yuxi.channels.models import ( from .irc_parser import ( ParsedIRCMessage, extract_nick, + make_irc_message_id, parse_badges, parse_emotes, - make_irc_message_id, ) _FILTERED_NOTICE_MSG_IDS = frozenset( @@ -49,7 +49,7 @@ _EVENT_MAP: dict[str, str] = { "announcement": "announcement", } -_SKIP_COMMANDS = frozenset({"JOIN", "PART", "HOSTTARGET", "ROOMSTATE", "USERSTATE"}) +_SKIP_COMMANDS = frozenset({"JOIN", "PART", "HOSTTARGET", "USERSTATE"}) def _make_identity( @@ -178,6 +178,27 @@ def normalize_irc_message(parsed: ParsedIRCMessage) -> ChannelMessage | None: timestamp=datetime.now(UTC), ) + # ---- CLEARMSG: 单条消息删除 ---- + if command == "CLEARMSG": + channel = parsed.params[0] if parsed.params else "" + target_msg_id = tags.get("target-msg-id", "") + login = tags.get("login", "") + target_user_id = tags.get("target-user-id", "") + + return ChannelMessage( + identity=_make_identity(target_user_id or "twitch_system", channel, target_msg_id), + event_type=EventType.MESSAGE_DELETED, + message_type=MessageType.TEXT, + chat_type=ChatType.GROUP, + content="", + metadata={ + "event": "clearmsg", + "target_msg_id": target_msg_id, + "login": login, + }, + timestamp=datetime.now(UTC), + ) + # ---- WHISPER: 私信 ---- if command == "WHISPER": user_id = tags.get("user-id") or extract_nick(parsed.prefix) @@ -204,6 +225,30 @@ def normalize_irc_message(parsed: ParsedIRCMessage) -> ChannelMessage | None: timestamp=_make_timestamp(tags.get("tmi-sent-ts")), ) + # ---- ROOMSTATE: 房间状态变更 ---- + if command == "ROOMSTATE": + channel = parsed.params[0] if parsed.params else "" + room_state_flags = { + "emote_only": tags.get("emote-only") == "1", + "subs_only": tags.get("subs-only") == "1", + "followers_only": tags.get("followers-only", "-1"), + "slow": tags.get("slow", "0"), + "r9k": tags.get("r9k") == "1", + } + + return ChannelMessage( + identity=_make_identity("twitch_system", channel), + event_type=EventType.SYSTEM_EVENT, + message_type=MessageType.TEXT, + chat_type=ChatType.GROUP, + content="Room state changed", + metadata={ + "event": "roomstate", + "room_state": room_state_flags, + }, + timestamp=datetime.now(UTC), + ) + # ---- NOTICE: 系统通知 ---- if command == "NOTICE": channel = parsed.params[0] if parsed.params else "" diff --git a/backend/package/yuxi/channels/adapters/twitch/outbound_cache.py b/backend/package/yuxi/channels/adapters/twitch/outbound_cache.py index 003992b4..61c4ff31 100644 --- a/backend/package/yuxi/channels/adapters/twitch/outbound_cache.py +++ b/backend/package/yuxi/channels/adapters/twitch/outbound_cache.py @@ -10,16 +10,25 @@ class OutboundCacheManager: self._cache: OrderedDict[str, dict[str, Any]] = OrderedDict() self._max_size = max_size - def record(self, target: str, content: str) -> None: + def record(self, target: str, content: str, message_id: str | None = None) -> str: entry = { "channel": target, "content": content, "timestamp": time.time(), } - cache_key = f"{target}:{len(self._cache)}" + cache_key = message_id or f"{target}:{len(self._cache)}" + if message_id: + entry["message_id"] = message_id self._cache[cache_key] = entry while len(self._cache) > self._max_size: self._cache.popitem(last=False) + return cache_key def get_all(self) -> list[dict[str, Any]]: return list(self._cache.values()) + + def find_by_message_id(self, message_id: str) -> dict[str, Any] | None: + for entry in self._cache.values(): + if entry.get("message_id") == message_id: + return entry + return None diff --git a/backend/package/yuxi/channels/adapters/twitch/send.py b/backend/package/yuxi/channels/adapters/twitch/send.py index f97cb6f0..39940344 100644 --- a/backend/package/yuxi/channels/adapters/twitch/send.py +++ b/backend/package/yuxi/channels/adapters/twitch/send.py @@ -18,6 +18,7 @@ def format_cap_req(capabilities: list[str]) -> str: def find_utf8_cut(encoded: bytes, byte_limit: int) -> int: + byte_limit = min(byte_limit, len(encoded)) cut = byte_limit while cut > 0 and (encoded[cut - 1] & 0xC0) == 0x80: cut -= 1 diff --git a/backend/package/yuxi/channels/adapters/twitch/session.py b/backend/package/yuxi/channels/adapters/twitch/session.py index 66906a39..8424a5df 100644 --- a/backend/package/yuxi/channels/adapters/twitch/session.py +++ b/backend/package/yuxi/channels/adapters/twitch/session.py @@ -97,11 +97,10 @@ def check_group_policy( return True case "disabled": return False - case "mention_only": - require_mention = config.get("require_mention", True) - if not require_mention: - return True + case "mention": return bool(bot_name) and bot_name in content.lower() + case "mention_only": + return not config.get("require_mention", True) or (bool(bot_name) and bot_name in content.lower()) case "allowlist": allowlist = config.get("group_allow_from", []) channels_config = config.get("channels_config", {})