From 59cd13cf84dbc46266b1b73a7d66f6138368c7d1 Mon Sep 17 00:00:00 2001 From: Kris <2893855659@qq.com> Date: Tue, 12 May 2026 00:50:10 +0800 Subject: [PATCH] =?UTF-8?q?feat(twitch):=20=E5=AE=9E=E7=8E=B0=E5=AE=8C?= =?UTF-8?q?=E6=95=B4=E7=9A=84Twitch=E8=81=8A=E5=A4=A9=E9=80=82=E9=85=8D?= =?UTF-8?q?=E5=99=A8=E6=A8=A1=E5=9D=97?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 新增Twitch IRC协议相关的全套实现，包括： 1. 基础工具类：令牌处理、消息格式化、速率限制、消息去重 2. 核心适配器组件：IRC解析器、消息归一化、外发消息处理 3. API客户端：Helix API封装、认证提供者 4. 配置与部署：配置校验、设置向导 5. 辅助功能：配对管理、健康检查、目标解析等 --- .../yuxi/channels/adapters/twitch/__init__.py | 3 + .../yuxi/channels/adapters/twitch/actions.py | 104 ++ .../yuxi/channels/adapters/twitch/adapter.py | 935 ++++++++++++++++++ .../channels/adapters/twitch/auth_provider.py | 127 +++ .../channels/adapters/twitch/config_schema.py | 190 ++++ .../channels/adapters/twitch/deduplicator.py | 48 + .../adapters/twitch/eventsub_client.py | 529 ++++++++++ .../yuxi/channels/adapters/twitch/helix.py | 192 ++++ .../channels/adapters/twitch/irc_parser.py | 142 +++ .../adapters/twitch/markdown_utils.py | 55 ++ .../channels/adapters/twitch/normalizer.py | 233 +++++ .../yuxi/channels/adapters/twitch/outbound.py | 91 ++ .../yuxi/channels/adapters/twitch/pairing.py | 112 +++ .../yuxi/channels/adapters/twitch/probe.py | 89 ++ .../channels/adapters/twitch/rate_limiter.py | 62 ++ .../yuxi/channels/adapters/twitch/resolver.py | 47 + .../yuxi/channels/adapters/twitch/send.py | 17 + .../yuxi/channels/adapters/twitch/session.py | 110 +++ .../yuxi/channels/adapters/twitch/setup.py | 400 ++++++++ .../channels/adapters/twitch/token_utils.py | 12 + 20 files changed, 3498 insertions(+) create mode 100644 backend/package/yuxi/channels/adapters/twitch/__init__.py create mode 100644 backend/package/yuxi/channels/adapters/twitch/actions.py create mode 100644 backend/package/yuxi/channels/adapters/twitch/adapter.py create mode 100644 backend/package/yuxi/channels/adapters/twitch/auth_provider.py create mode 100644 backend/package/yuxi/channels/adapters/twitch/config_schema.py create mode 100644 backend/package/yuxi/channels/adapters/twitch/deduplicator.py create mode 100644 backend/package/yuxi/channels/adapters/twitch/eventsub_client.py create mode 100644 backend/package/yuxi/channels/adapters/twitch/helix.py create mode 100644 backend/package/yuxi/channels/adapters/twitch/irc_parser.py create mode 100644 backend/package/yuxi/channels/adapters/twitch/markdown_utils.py create mode 100644 backend/package/yuxi/channels/adapters/twitch/normalizer.py create mode 100644 backend/package/yuxi/channels/adapters/twitch/outbound.py create mode 100644 backend/package/yuxi/channels/adapters/twitch/pairing.py create mode 100644 backend/package/yuxi/channels/adapters/twitch/probe.py create mode 100644 backend/package/yuxi/channels/adapters/twitch/rate_limiter.py create mode 100644 backend/package/yuxi/channels/adapters/twitch/resolver.py create mode 100644 backend/package/yuxi/channels/adapters/twitch/send.py create mode 100644 backend/package/yuxi/channels/adapters/twitch/session.py create mode 100644 backend/package/yuxi/channels/adapters/twitch/setup.py create mode 100644 backend/package/yuxi/channels/adapters/twitch/token_utils.py diff --git a/backend/package/yuxi/channels/adapters/twitch/__init__.py b/backend/package/yuxi/channels/adapters/twitch/__init__.py new file mode 100644 index 00000000..b56cbeda --- /dev/null +++ b/backend/package/yuxi/channels/adapters/twitch/__init__.py @@ -0,0 +1,3 @@ +from yuxi.channels.adapters.twitch.adapter import TwitchAdapter + +__all__ = ["TwitchAdapter"] diff --git a/backend/package/yuxi/channels/adapters/twitch/actions.py b/backend/package/yuxi/channels/adapters/twitch/actions.py new file mode 100644 index 00000000..36449f72 --- /dev/null +++ b/backend/package/yuxi/channels/adapters/twitch/actions.py @@ -0,0 +1,104 @@ +from __future__ import annotations + +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", "")) + if not ctx.chat_id or not content: + return None + identity = ChannelIdentity( + channel_id="twitch", + channel_type=adapter.channel_type, + channel_user_id="", + channel_chat_id=ctx.chat_id, + ) + return ChannelResponse(identity=identity, content=content) + + +async def handle_action(ctx: ActionContext, adapter: Any) -> DeliveryResult: + if ctx.action == "send": + response = _build_response(ctx, adapter, "content") + if response is None: + return DeliveryResult(success=False, error="missing chat_id or content") + 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) + + 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", + "polls": "Twitch IRC does not support polls", + "native_commands": "Twitch IRC does not support native commands", + "pin": "Twitch IRC does not support message pinning", + "unpin": "Twitch IRC does not support message unpinning", + } + hint = unsupported_hints.get(ctx.action, f"unknown action: {ctx.action}") + return DeliveryResult(success=False, error=hint) + + +def supports_action(action: str) -> bool: + return action in {"send", "send_media"} + + +def describe_message_tool() -> dict[str, Any]: + return { + "send": { + "description": "Send a text message to a Twitch channel", + "parameters": { + "chat_id": {"type": "string", "description": "Channel name (e.g. #channel)"}, + "content": {"type": "string", "description": "Text content to send"}, + }, + }, + "send_media": { + "description": "Send a media URL to a Twitch channel (sent as text)", + "parameters": { + "chat_id": {"type": "string", "description": "Channel name (e.g. #channel)"}, + "media_url": {"type": "string", "description": "Media URL to send"}, + }, + }, + } + + +def extract_target_from_args(args: dict[str, Any]) -> dict | None: + chat_id = args.get("chat_id") + if chat_id: + return {"chat_id": chat_id} + return None + + +def resolve_execution_mode(action: str) -> str: + return "send" + + +def get_action_stats() -> dict[str, list[str]]: + return { + "implemented": ["send", "send_media"], + "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 new file mode 100644 index 00000000..01331203 --- /dev/null +++ b/backend/package/yuxi/channels/adapters/twitch/adapter.py @@ -0,0 +1,935 @@ +from __future__ import annotations + +import asyncio +import os +import ssl +import time +import uuid +from collections import OrderedDict +from collections.abc import AsyncIterator +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.models import ( + ChannelMessage, + ChannelResponse, + ChannelStatus, + ChannelType, + DeliveryResult, + HealthStatus, +) +from yuxi.channels.registry import register_builtin_adapter +from yuxi.utils.logging_config import logger + +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 .markdown_utils import strip_twitch_markdown +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, refresh_access_token, validate_token +from .rate_limiter import RateLimiter +from .send import format_privmsg_line, format_action_line, format_pong, format_cap_req +from .session import check_allowed_roles, check_group_policy +from .token_utils import ensure_oauth_prefix + + +@register_builtin_adapter +class TwitchAdapter(BaseChannelAdapter): + channel_id: ClassVar[str] = "twitch" + channel_type: ClassVar[ChannelType] = ChannelType.TWITCH + + text_chunk_limit: ClassVar[int] = 500 + supports_markdown: ClassVar[bool] = False + supports_media: ClassVar[bool] = False + supports_streaming: ClassVar[bool] = True + streaming_modes: ClassVar[list[str]] = ["off", "block"] + webhook_path: ClassVar[str | None] = None + delivery_mode: ClassVar[str] = "direct" + + capabilities = ChannelCapabilities( + chat_types=["group"], + 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, + edit=False, + unsend=False, + reactions=False, + polls=False, + native_commands=False, + block_streaming=True, + pin=False, + unpin=False, + ) + meta = ChannelMeta(id="twitch", label="Twitch", aliases=["twitch-chat"]) + + KNOWN_USER_LIMIT = (20, 30) + MOD_VIP_LIMIT = (100, 30) + + def __init__(self, config: dict[str, Any] | None = None): + super().__init__(config) + self._status = ChannelStatus.DISCONNECTED + + self._reader: asyncio.StreamReader | None = None + self._writer: asyncio.StreamWriter | None = None + + self._eventsub: EventSubListener | None = None + self._helix: HelixClient | None = None + + self._irc_connected = False + self._joined_channels: set[str] = set() + + self._rate_limiter: RateLimiter | None = None + self._circuit_breaker = CircuitBreaker(failure_threshold=5) + + self._bot_user_id: str | None = None + self._bot_username: str | None = None + + self._irc_task: asyncio.Task | None = None + self._reconnect_task: asyncio.Task | None = None + + self._deduplicator = MessageDeduplicator() + + self._token_expires_at: float | None = None + self._token_obtained_at: float | None = None + self._token_expires_in: int = 0 + + self._accounts: dict[str, dict[str, Any]] = {} + self._active_account_id: str = "" + self._resolve_accounts_config() + + self._pairing_store = PairingStore() + self._pairing_enabled = self.config.get("pairing_enabled", False) + + self._token_source: str = "none" + + self._outbound_cache: OrderedDict[str, dict[str, Any]] = OrderedDict() + self._outbound_cache_max = 500 + self._started_at = time.time() + + self._stream_buffers: dict[str, dict[str, Any]] = {} + + async def connect(self) -> None: + if self._status == ChannelStatus.CONNECTED: + return + + self._status = ChannelStatus.CONNECTING + logger.info(f"[Twitch] Starting channel '{self.channel_id}'") + + client_id = self.config.get("client_id", "") + access_token = ( + self.config.get("access_token") + or os.environ.get("TWITCH_ACCESS_TOKEN", "") + or os.environ.get("OPENCLAW_TWITCH_ACCESS_TOKEN", "") + ) + bot_username = self.config.get("bot_username", "").lower() + + if self.config.get("access_token"): + self._token_source = "config" + elif os.environ.get("TWITCH_ACCESS_TOKEN", ""): + self._token_source = "env" + else: + self._token_source = "none" + + if not client_id or not access_token or not bot_username: + raise ChannelAuthenticationError("Missing client_id, access_token, or bot_username") + + user_info = await validate_token(client_id, access_token) + if user_info is None: + raise ChannelAuthenticationError("Invalid Twitch access token") + + self._bot_user_id = user_info.get("id", "") + self._bot_username = user_info.get("login", bot_username) + logger.info(f"[Twitch] Bot verified: {self._bot_username} (ID: {self._bot_user_id})") + + expires_in = self.config.get("token_expires_in", 0) + 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"[Twitch] Token expires in {expires_in}s, will refresh at {expires_in - 300}s before expiry") + + self._helix = HelixClient(client_id, access_token) + await self._helix.start() + + self._rate_limiter = RateLimiter( + limit=self.config.get("rate_limit", 20), + window=self.config.get("rate_window", 30), + mod_limit=self.config.get("mod_rate_limit", 100), + ) + + irc_task = asyncio.create_task(self._connect_irc()) + eventsub_task = asyncio.create_task(self._connect_eventsub()) + + await irc_task + + try: + await eventsub_task + except Exception as e: + logger.warning(f"[Twitch] EventSub failed (non-fatal): {e}") + + self._status = ChannelStatus.CONNECTED + eventsub_status = self._eventsub.connected if self._eventsub else False + logger.info(f"[Twitch] Connected: IRC={self._irc_connected}, EventSub={eventsub_status}") + + async def disconnect(self) -> None: + if self._status == ChannelStatus.DISCONNECTED: + return + + logger.info(f"[Twitch] Stopping channel '{self.channel_id}'") + self._status = ChannelStatus.DISCONNECTED + self._irc_connected = False + + if self._irc_task and not self._irc_task.done(): + self._irc_task.cancel() + try: + await self._irc_task + except asyncio.CancelledError: + pass + self._irc_task = None + + if self._reconnect_task and not self._reconnect_task.done(): + self._reconnect_task.cancel() + try: + await self._reconnect_task + except asyncio.CancelledError: + pass + self._reconnect_task = None + + if self._writer: + try: + self._writer.write(b"QUIT :ForcePilot shutting down\r\n") + await self._writer.drain() + self._writer.close() + await self._writer.wait_closed() + except Exception: + pass + self._writer = None + self._reader = None + + if self._eventsub: + await self._eventsub.disconnect() + self._eventsub = None + + if self._helix: + await self._helix.close() + self._helix = None + + async def send(self, response: ChannelResponse) -> DeliveryResult: + if not self._irc_connected or not self._writer: + return DeliveryResult(success=False, error="not_connected") + + await self._refresh_token_if_needed() + + content = self._maybe_strip_markdown(response.content) + + response_prefix = self.config.get("response_prefix", "") + if response_prefix and not content.startswith(response_prefix): + content = f"{response_prefix} {content}".rstrip() + + if not content.strip(): + return DeliveryResult(success=True, metadata={"messageId": "skipped"}) + + async def _do_send(): + target = self.format_outbound(response)["target"] + chunks = self._split_irc_text(content, target) + return await self._send_irc_chunks(target, chunks) + + try: + return await self._circuit_breaker.call(_do_send) + except CircuitBreakerOpenError: + return DeliveryResult(success=False, error="circuit_breaker_open") + except Exception as e: + refreshed = await self._refresh_token_if_needed() + if not refreshed: + return DeliveryResult(success=False, error=f"unknown:{str(e)}") + try: + return await self._circuit_breaker.call(_do_send) + except CircuitBreakerOpenError: + return DeliveryResult(success=False, error="circuit_breaker_open") + except Exception as e2: + return DeliveryResult(success=False, error=f"retry_failed:{str(e2)}") + + async def send_stream_chunk(self, chat_id: str, msg_id: str, chunk: str, finished: bool) -> DeliveryResult: + if not self._irc_connected or not self._writer: + return DeliveryResult(success=False, error="not_connected") + + chunk = self._maybe_strip_markdown(chunk) + + 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) + silent = self.config.get("silent", False) + fmt = format_action_line if silent else format_privmsg_line + + try: + buffer_key = f"{chat_id}:{msg_id}" + if coalesce_max_delay_ms > 0 and not finished: + buf = self._stream_buffers.get(buffer_key) + if buf is None: + buf = { + "chat_id": chat_id, + "chunks": [], + "task": None, + } + self._stream_buffers[buffer_key] = buf + buf["chunks"].append(chunk) + total = "".join(buf["chunks"]) + if len(total.encode("utf-8")) >= coalesce_min_chars: + return await self._flush_stream_buffer(buffer_key) + 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._flush_stream_buffer(buffer_key) + else: + if finished: + chunks = self._split_irc_text(chunk, chat_id) + return await self._send_irc_chunks(chat_id, chunks) + else: + chunk_bytes = len(chunk.encode("utf-8")) + if chunk_bytes < coalesce_min_chars: + return DeliveryResult(success=True) + prefix_len = len(f"PRIVMSG {chat_id} :") + suffix = "…" + text = chunk[: 510 - prefix_len - len(suffix.encode("utf-8"))] + 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") + await self._writer.drain() + return DeliveryResult(success=True) + except Exception as e: + return DeliveryResult(success=False, error=f"unknown:{str(e)}") + + async def _delayed_flush(self, buffer_key: str, delay_sec: float) -> None: + await asyncio.sleep(delay_sec) + await self._flush_stream_buffer(buffer_key) + + async def _flush_stream_buffer(self, buffer_key: str) -> DeliveryResult: + buf = self._stream_buffers.pop(buffer_key, None) + if buf is None: + return DeliveryResult(success=True) + content = "".join(buf.get("chunks", [])) + if not content.strip(): + return DeliveryResult(success=True) + chunks = self._split_irc_text(content, buf["chat_id"]) + return await self._send_irc_chunks(buf["chat_id"], chunks) + + async def send_media(self, chat_id: str, media_type: str, data: Any) -> DeliveryResult: + if not self._irc_connected or not self._writer: + return DeliveryResult(success=False, error="not_connected") + + url = str(data) if data else "" + if not url: + return DeliveryResult(success=False, error="unknown:empty_media_url") + + text = url + + async def _do_send(): + chunks = self._split_irc_text(text, chat_id) + return await self._send_irc_chunks(chat_id, chunks) + + 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=f"unknown:{str(e)}") + + async def _send_irc_chunks(self, target: str, chunks: list[str]) -> DeliveryResult: + content = "".join(chunks) + silent = self.config.get("silent", False) + fmt = format_action_line if silent else format_privmsg_line + for chunk in chunks: + if not await self._rate_limiter.acquire(): + return DeliveryResult(success=False, error="rate_limit_exceeded") + self._writer.write(fmt(target, chunk).encode("utf-8") + b"\r\n") + await self._writer.drain() + self._record_outbound(target, content) + return DeliveryResult(success=True) + + def _record_outbound(self, target: str, content: str) -> None: + entry = { + "channel": target, + "content": content, + "timestamp": time.time(), + } + cache_key = f"{target}:{len(self._outbound_cache)}" + self._outbound_cache[cache_key] = entry + while len(self._outbound_cache) > self._outbound_cache_max: + self._outbound_cache.popitem(last=False) + + def _audit_log(self, event: str, **kwargs: Any) -> None: + logger.info(f"[Twitch|AUDIT] {event} | " + " | ".join(f"{k}={v}" for k, v in kwargs.items())) + + def get_outbound_cache(self) -> list[dict[str, Any]]: + return list(self._outbound_cache.values()) + + def _maybe_strip_markdown(self, text: str) -> str: + if self.config.get("strip_markdown", True): + return strip_twitch_markdown(text) + return text + + def resolve_markdown_table_mode(self) -> str: + return self.config.get("markdown_table_mode", "text") + + async def receive(self) -> AsyncIterator[ChannelMessage]: + return + yield # type: ignore[misc] # Twitch 浣跨敤 IRC push + EventSub push 妯″紡锛屼笉閫氳繃 receive 杞 + + def normalize_inbound(self, raw: Any) -> ChannelMessage: + return normalize_irc_message(raw) + + def format_outbound(self, response: ChannelResponse) -> dict[str, Any]: + target = response.identity.channel_chat_id + if not target: + joined = list(self._joined_channels) + if joined: + target = joined[0] + else: + channels = self.config.get("channels", []) + if channels: + ch = channels[0] + target = f"#{ch.lstrip('#')}" if not ch.startswith("#") else ch + return { + "command": "PRIVMSG", + "target": target, + "text": response.content, + } + + def resolve_implicit_target(self, mode: str = "implicit") -> str | None: + if mode == "heartbeat" and self._joined_channels: + return next(iter(self._joined_channels)) + joined = list(self._joined_channels) + if joined: + return joined[0] + channels = self.config.get("channels", []) + if channels: + ch = channels[0] + return f"#{ch.lstrip('#')}" if not ch.startswith("#") else ch + return None + + async def health_check(self) -> HealthStatus: + start = time.monotonic() + issues: list[str] = [] + + if not self._irc_connected: + issues.append("IRC not connected") + if self._writer and self._writer.is_closing(): + issues.append("IRC connection closing") + + if self._eventsub and not self._eventsub.connected: + issues.append("EventSub not connected") + + token = self.config.get("access_token", "") + client_id = self.config.get("client_id", "") + if token and client_id: + try: + user_info = await validate_token(client_id, token) + if user_info is None: + issues.append("Token invalid") + except Exception as e: + issues.append(f"Helix API unreachable: {e}") + + status_issues = self.collect_status_issues() + issues.extend(status_issues) + + latency_ms = (time.monotonic() - start) * 1000 + + if issues: + return HealthStatus(status="unhealthy", last_error="; ".join(issues), latency_ms=latency_ms) + return HealthStatus( + status="healthy", + latency_ms=latency_ms, + metadata={ + "irc_connected": self._irc_connected, + "eventsub_connected": self._eventsub.connected if self._eventsub else False, + "joined_channels": list(self._joined_channels), + "token_expires_in": self._token_expires_in, + "token_expires_at": self._token_expires_at, + "token_source": self._token_source, + }, + ) + + def collect_status_issues(self) -> list[str]: + issues: list[str] = [] + + client_id = self.config.get("client_id", "") + access_token = ( + self.config.get("access_token", "") + or os.environ.get("TWITCH_ACCESS_TOKEN", "") + or os.environ.get("OPENCLAW_TWITCH_ACCESS_TOKEN", "") + ) + if not client_id or not access_token: + issues.append("not_configured: missing client_id or access_token") + + channels = self.config.get("channels", []) + if not channels: + issues.append("no_channels_configured") + + group_policy = self.config.get("group_policy", "open") + if group_policy == "allowlist": + allowlist = self.config.get("group_allow_from", []) + channels_config = self.config.get("channels_config", {}) + has_allow = bool(allowlist) or any( + cfg.get("allow_from") for cfg in channels_config.values() if isinstance(cfg, dict) + ) + if not has_allow: + issues.append("allowlist_policy_with_empty_allow_from") + + allowed_roles = self.config.get("allowedRoles", []) + if "all" in allowed_roles and group_policy == "allowlist": + issues.append("allowed_roles_all_with_allowlist_policy_conflict") + + uptime = time.time() - self._started_at + if uptime > 7 * 86400: + issues.append(f"long_running: uptime {uptime / 86400:.1f} days") + + if self._status != ChannelStatus.CONNECTED: + issues.append(f"channel_not_connected: status={self._status}") + + if not self._irc_connected: + issues.append("irc_disconnected") + + return issues + + def _resolve_accounts_config(self) -> None: + raw_accounts = self.config.get("accounts", {}) + if not isinstance(raw_accounts, dict) or not raw_accounts: + self._accounts = {} + return + + base_overrides = {k: v for k, v in self.config.items() if k != "accounts"} + for account_id, account_conf in raw_accounts.items(): + if not isinstance(account_conf, dict): + continue + merged = {**account_conf} + for key in ( + "client_id", + "client_secret", + "rate_limit", + "rate_window", + "irc_host", + "irc_port", + "group_policy", + "require_mention", + "allowedRoles", + "group_allow_from", + "channels_config", + ): + if key in base_overrides and key not in merged: + merged[key] = base_overrides[key] + self._accounts[account_id] = merged + + default_account = self.config.get("defaultAccount", "") + if default_account and default_account in self._accounts: + self._active_account_id = default_account + elif "default" in self._accounts: + self._active_account_id = "default" + elif self._accounts: + self._active_account_id = next(iter(self._accounts)) + + logger.info( + f"[Twitch] Multi-account mode: {len(self._accounts)} accounts, " + f"active={self._active_account_id}, ids={list(self._accounts.keys())}" + ) + + def list_account_ids(self) -> list[str]: + if not self._accounts: + return ["default"] + return list(self._accounts.keys()) + + async def resolve_account_context(self, account_id: str | None = None) -> dict[str, Any] | None: + if not self._accounts: + return None + + resolved_id = account_id or self._active_account_id + if not resolved_id or resolved_id not in self._accounts: + return None + + account_conf = self._accounts[resolved_id] + self._active_account_id = resolved_id + return account_conf + + def _get_active_account_config(self) -> dict[str, Any]: + if self._accounts and self._active_account_id: + account_conf = self._accounts.get(self._active_account_id, {}) + merged = {**account_conf} + for key in ( + "client_id", + "client_secret", + "rate_limit", + "rate_window", + "irc_host", + "irc_port", + "group_policy", + "require_mention", + "allowedRoles", + "group_allow_from", + "channels_config", + ): + if key in self.config and self.config[key]: + merged[key] = self.config[key] + return merged + return self.config + + def get_account_snapshot(self, account_id: str | None = None) -> dict[str, Any]: + accounts = self._accounts if self._accounts else {"default": self.config} + resolved_id = account_id or self._active_account_id or next(iter(accounts), "default") + account_conf = accounts.get(resolved_id, {}) + return { + "account_id": resolved_id, + "bot_username": account_conf.get("bot_username", ""), + "channels": account_conf.get("channels", []), + "active": resolved_id == self._active_account_id, + "irc_connected": self._irc_connected, + "eventsub_connected": self._eventsub.connected if self._eventsub else False, + } + + def supports_action(self, action: str) -> bool: + return _supports_action(action) + + def describe_message_tool(self) -> dict: + return describe_message_tool() + + async def handle_action(self, ctx) -> DeliveryResult: + return await handle_action(ctx, self) + + def extract_target_from_args(self, args: dict) -> dict | None: + return extract_target_from_args(args) + + def resolve_execution_mode(self, action: str) -> str: + return resolve_execution_mode(action) + + async def _connect_irc(self) -> None: + host = self.config.get("irc_host", "irc.chat.twitch.tv") + port = self.config.get("irc_port", 6697) + token = self.config.get("access_token", "") + username = self.config.get("bot_username", "").lower() + + ctx = ssl.create_default_context() + self._reader, self._writer = await asyncio.open_connection( + host=host, + port=port, + ssl=ctx, + ) + + self._send_line(f"PASS {ensure_oauth_prefix(token)}") + self._send_line(f"NICK {username}") + self._send_line(format_cap_req(["twitch.tv/tags", "twitch.tv/commands", "twitch.tv/membership"])) + + await self._wait_for_welcome() + + channels = self.config.get("channels", []) + for channel in channels: + channel_name = channel if channel.startswith("#") else f"#{channel}" + self._send_line(f"JOIN {channel_name}") + self._joined_channels.add(channel_name.lower()) + + self._irc_connected = True + logger.info(f"[Twitch] IRC connected as {username}, joined {len(channels)} channels") + + if self._pairing_enabled and channels: + first_channel = list(self._joined_channels)[0] + + def _on_pairing_approve(user_name: str, channel: str) -> None: + if self._writer and self._irc_connected: + try: + notify_channel = channel or first_channel + text = f"{user_name} has been approved to interact!" + self._writer.write(format_privmsg_line(notify_channel, text).encode("utf-8") + b"\r\n") + logger.info(f"[Twitch] Pairing approval notification sent for {user_name}") + except Exception as e: + logger.warning(f"[Twitch] Failed to send pairing notification: {e}") + + self._pairing_store.set_on_approve(_on_pairing_approve, first_channel) + + self._irc_task = asyncio.create_task(self._irc_message_loop()) + + async def _wait_for_welcome(self) -> None: + deadline = time.monotonic() + 30 + while time.monotonic() < deadline: + remaining = deadline - time.monotonic() + if remaining <= 0: + raise ChannelAuthenticationError("IRC welcome timeout") + try: + line = await asyncio.wait_for(self._reader.readline(), timeout=remaining) + except TimeoutError: + raise ChannelAuthenticationError("IRC welcome timeout") + if not line: + raise ChannelAuthenticationError("IRC connection closed before welcome") + raw = line.decode("utf-8", errors="replace").rstrip("\r\n") + if " 001 " in raw or " 376 " in raw or " 372 " in raw: + continue + if " 002 " in raw or " 003 " in raw or " 004 " in raw: + return + if "NOTICE" in raw and "Login" in raw: + return + raise ChannelAuthenticationError("IRC welcome timeout") + + async def _irc_message_loop(self) -> None: + IRC_READ_TIMEOUT = 360.0 + while self._irc_connected and self._reader: + try: + line = await asyncio.wait_for(self._reader.readline(), timeout=IRC_READ_TIMEOUT) + except TimeoutError: + logger.warning("[Twitch] IRC read timeout (no data for 360s), reconnecting") + self._irc_connected = False + self._start_reconnect_task() + break + + try: + if not line: + logger.warning("[Twitch] IRC connection closed") + self._irc_connected = False + break + + raw_line = line.decode("utf-8", errors="replace").rstrip("\r\n") + if not raw_line: + continue + + if raw_line.startswith("PING"): + token_str = raw_line[5:].lstrip(":") + self._send_line(format_pong(token_str)) + continue + + parsed = parse_irc_line(raw_line) + if not parsed.command: + continue + + if parsed.command == "RECONNECT": + logger.warning("[Twitch] Reconnect requested by server") + self._start_reconnect_task() + continue + + if parsed.command == "PRIVMSG": + channel = parsed.params[0] if parsed.params else "" + user_id = parsed.tags.get("user-id") or extract_nick(parsed.prefix) + content = parsed.trailing + + if self._bot_user_id and user_id == self._bot_user_id: + continue + + if self._pairing_enabled and not check_pairing_policy(self._pairing_store, user_id): + display_name = parsed.tags.get("display-name", user_id) + self._pairing_store.add_pending(user_id, display_name, channel) + logger.info(f"[Twitch] Unpaired user {display_name} ({user_id}) in {channel}") + self._audit_log("pairing_request", user_id=user_id, display_name=display_name, channel=channel) + continue + + policy_result = check_group_policy(self.config, channel, user_id, content) + if policy_result is False: + self._audit_log("policy_rejected", user_id=user_id, channel=channel, reason="group_policy") + continue + + badges = parse_badges(parsed.tags.get("badges", "")) + if check_allowed_roles(self.config, channel, badges) is False: + self._audit_log("policy_rejected", user_id=user_id, channel=channel, reason="invalid_role") + continue + + if "bits" in parsed.tags or ( + "badges" in parsed.tags + and any(badge in parsed.tags.get("badges", "") for badge in ("broadcaster", "moderator", "vip")) + ): + if self._rate_limiter: + await self._rate_limiter.switch_to_mod() + + message_id = parsed.tags.get("id") or parsed.tags.get("tmi-sent-ts") or str(uuid.uuid4()) + if message_id and self._deduplicator.is_duplicate(message_id): + continue + + msg = normalize_irc_message(parsed) + if msg: + if self._bot_username and msg.mentions: + msg.mentions = resolve_mentions(msg.content, self._bot_username) + await self._handle_message(msg) + + except asyncio.CancelledError: + break + except Exception as e: + logger.error(f"[Twitch] IRC message loop error: {e}") + await asyncio.sleep(1) + + async def _connect_eventsub(self) -> None: + client_id = self.config.get("client_id", "") + client_secret = self.config.get("client_secret", "") + + if not client_secret: + logger.warning("[Twitch] No client_secret, skipping EventSub") + return + + app_token = await get_app_access_token(client_id, client_secret) + if not app_token: + logger.warning("[Twitch] Failed to get app access token, skipping EventSub") + return + + channels = self.config.get("channels", []) + broadcaster_ids: list[str] = [] + for ch in channels: + ch_name = ch.lstrip("#") + user = await self._helix.get_user_by_name(ch_name) if self._helix else None + if user: + broadcaster_ids.append(user["id"]) + + if not broadcaster_ids: + logger.warning("[Twitch] No broadcaster IDs resolved, skipping EventSub") + return + + self._eventsub = EventSubListener( + helix=self._helix, + app_access_token=app_token, + broadcaster_ids=broadcaster_ids, + ) + self._eventsub.on_event(self._handle_message) + await self._eventsub.connect() + + async def _reconnect_irc(self) -> None: + self._status = ChannelStatus.RECONNECTING + self._irc_connected = False + if self._writer: + try: + self._writer.close() + except Exception: + pass + self._writer = None + self._reader = None + + await self._refresh_token_if_needed() + + await asyncio.sleep(3) + try: + await self._connect_irc() + self._status = ChannelStatus.CONNECTED + logger.info("[Twitch] IRC reconnected") + except asyncio.CancelledError: + self._status = ChannelStatus.DISCONNECTED + raise + except Exception as e: + self._status = ChannelStatus.ERROR + logger.error(f"[Twitch] IRC reconnection failed: {e}") + + def _send_line(self, line: str) -> None: + if self._writer: + self._writer.write(line.encode("utf-8") + b"\r\n") + + def _start_reconnect_task(self) -> None: + async def _reconnect_with_timeout(): + try: + await asyncio.wait_for(self._reconnect_irc(), timeout=60.0) + except TimeoutError: + logger.error("[Twitch] IRC reconnection timed out after 60s") + self._status = ChannelStatus.ERROR + except asyncio.CancelledError: + pass + except Exception as e: + logger.error(f"[Twitch] IRC reconnect task failed: {e}") + + self._reconnect_task = asyncio.create_task(_reconnect_with_timeout()) + + async def _refresh_token_if_needed(self) -> bool: + if self._token_expires_at is not None and time.time() < self._token_expires_at: + return True + + client_id = self.config.get("client_id", "") + client_secret = self.config.get("client_secret", "") + refresh_token = self.config.get("refresh_token", "") + + if not client_id or not client_secret or not refresh_token: + return False + + try: + new_tokens = await refresh_access_token(client_id, client_secret, refresh_token) + if new_tokens is None: + logger.warning("[Twitch] Token refresh failed") + return False + + new_access_token = new_tokens.get("access_token", "") + new_refresh_token = new_tokens.get("refresh_token", "") + expires_in = new_tokens.get("expires_in", 0) + + if new_access_token: + self.config["access_token"] = new_access_token + if new_refresh_token: + self.config["refresh_token"] = new_refresh_token + 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 + self.config["token_expires_in"] = expires_in + logger.info(f"[Twitch] Token refreshed, expires in {expires_in}s") + + if self._helix: + self._helix._access_token = new_access_token + + self._token_source = "refreshed" + self._audit_log("token_refresh", expires_in=str(expires_in)) + logger.info("[Twitch] Access token refreshed") + return True + except Exception as e: + logger.error(f"[Twitch] Token refresh error: {e}") + return False + + @staticmethod + def _split_irc_text(text: str, target: str) -> list[str]: + prefix_len = len(f"PRIVMSG {target} :") + available = max(200, 510 - prefix_len) + if available < 1: + available = 200 + + chunks: list[str] = [] + if not text: + return [""] + remaining = text + + while remaining: + encoded = remaining.encode("utf-8") + if len(encoded) <= available: + chunks.append(remaining) + break + + cut = _find_utf8_cut(encoded, available) + split_pos = len(encoded[:cut].decode("utf-8", errors="replace")) + + nl = remaining.rfind("\n", 0, split_pos) + if nl > split_pos * 0.5: + split_pos = nl + 1 + else: + sp = remaining.rfind(" ", 0, split_pos) + if sp > split_pos * 0.5: + split_pos = sp + 1 + + chunk = remaining[:split_pos].rstrip() + chunks.append(chunk) + remaining = remaining[split_pos:].lstrip() + + return chunks + + +def _find_utf8_cut(encoded: bytes, byte_limit: int) -> int: + cut = byte_limit + while cut > 0 and (encoded[cut - 1] & 0xC0) == 0x80: + cut -= 1 + while cut > 0 and (encoded[cut - 1] & 0xC0) == 0xC0: + cut -= 1 + if cut == 0: + cut = max(1, byte_limit) + return cut diff --git a/backend/package/yuxi/channels/adapters/twitch/auth_provider.py b/backend/package/yuxi/channels/adapters/twitch/auth_provider.py new file mode 100644 index 00000000..eda20fe0 --- /dev/null +++ b/backend/package/yuxi/channels/adapters/twitch/auth_provider.py @@ -0,0 +1,127 @@ +from __future__ import annotations + +import time +from abc import ABC, abstractmethod +from typing import Any + +from yuxi.utils.logging_config import logger + +from .probe import refresh_access_token, validate_token + + +class AuthProvider(ABC): + @abstractmethod + async def get_access_token(self) -> str | None: ... + + @abstractmethod + async def get_client_id(self) -> str: ... + + @abstractmethod + async def validate(self) -> bool: ... + + +class StaticAuthProvider(AuthProvider): + def __init__(self, client_id: str, access_token: str): + self._client_id = client_id + self._access_token = access_token + + async def get_access_token(self) -> str | None: + return self._access_token or None + + async def get_client_id(self) -> str: + return self._client_id + + async def validate(self) -> bool: + if not self._client_id or not self._access_token: + return False + user_info = await validate_token(self._client_id, self._access_token) + return user_info is not None + + +class RefreshingAuthProvider(AuthProvider): + def __init__( + self, + client_id: str, + client_secret: str, + access_token: str = "", + refresh_token: str = "", + ): + self._client_id = client_id + self._client_secret = client_secret + self._access_token = access_token + self._refresh_token = refresh_token + self._token_expires_at: float | None = None + self._token_obtained_at: float | None = None + self._token_expires_in: int = 0 + + async def get_access_token(self) -> str | None: + if self._is_expired(): + refreshed = await self._refresh() + if not refreshed: + logger.warning("[TwitchAuth] Token expired and refresh failed") + return None + return self._access_token or None + + async def get_client_id(self) -> str: + return self._client_id + + async def validate(self) -> bool: + if not self._client_id or not self._access_token: + return False + user_info = await validate_token(self._client_id, self._access_token) + return user_info is not None + + 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 + + def _is_expired(self) -> bool: + if self._token_expires_at is None: + return False + return time.time() >= self._token_expires_at + + @property + def expires_at(self) -> float | None: + return self._token_expires_at + + @property + def expires_in(self) -> int: + return self._token_expires_in + + +def create_auth_provider(config: dict[str, Any]) -> AuthProvider: + client_id = config.get("client_id", "") + client_secret = config.get("client_secret", "") + access_token = config.get("access_token", "") + refresh_token = config.get("refresh_token", "") + + if client_secret and refresh_token: + logger.info("[TwitchAuth] Using RefreshingAuthProvider (auto-refresh enabled)") + return RefreshingAuthProvider( + client_id=client_id, + client_secret=client_secret, + access_token=access_token, + refresh_token=refresh_token, + ) + logger.info("[TwitchAuth] Using StaticAuthProvider (manual token rotation)") + return StaticAuthProvider(client_id=client_id, access_token=access_token) diff --git a/backend/package/yuxi/channels/adapters/twitch/config_schema.py b/backend/package/yuxi/channels/adapters/twitch/config_schema.py new file mode 100644 index 00000000..7cf9629a --- /dev/null +++ b/backend/package/yuxi/channels/adapters/twitch/config_schema.py @@ -0,0 +1,190 @@ +from __future__ import annotations + +from typing import Any + +from pydantic import BaseModel, Field + +from yuxi.utils.logging_config import logger + + +class TwitchAccountSchema(BaseModel): + bot_username: str = "" + access_token: str = "" + client_id: str = "" + client_secret: str = "" + refresh_token: str = "" + channels: list[str] = [] + group_policy: str = "open" + require_mention: bool = True + allowedRoles: list[str] = [] + group_allow_from: list[str] = [] + channels_config: dict[str, Any] = {} + pairing_enabled: bool = False + rate_limit: int = 20 + rate_window: int = 30 + mod_rate_limit: int = 100 + irc_host: str = "irc.chat.twitch.tv" + irc_port: int = 6697 + strip_markdown: bool = True + response_prefix: str = "" + stream_coalesce_min_chars: int = 30 + stream_coalesce_max_delay_ms: int = 0 + silent: bool = False + dm_policy: str = "pairing" + probe_timeout_ms: int = 10000 + + +class TwitchConfigSchema(BaseModel): + bot_username: str = "" + access_token: str = "" + client_id: str = "" + client_secret: str = "" + refresh_token: str = "" + channels: list[str] = [] + group_policy: str = Field(default="open", pattern=r"^(open|allowlist|disabled|mention_only)$") + require_mention: bool = True + allowedRoles: list[str] = [] + group_allow_from: list[str] = [] + channels_config: dict[str, Any] = {} + pairing_enabled: bool = False + rate_limit: int = Field(default=20, ge=1) + rate_window: int = Field(default=30, ge=1) + mod_rate_limit: int = Field(default=100, ge=1) + irc_host: str = "irc.chat.twitch.tv" + irc_port: int = 6697 + strip_markdown: bool = True + response_prefix: str = "" + stream_coalesce_min_chars: int = 30 + stream_coalesce_max_delay_ms: int = 0 + silent: bool = False + dm_policy: str = "pairing" + probe_timeout_ms: int = 10000 + accounts: dict[str, TwitchAccountSchema] = {} + defaultAccount: str = "" + + +def validate_twitch_config(config: dict[str, Any]) -> list[str]: + errors: list[str] = [] + + bot_username = config.get("bot_username", "") + access_token = config.get("access_token", "") + client_id = config.get("client_id", "") + + if not client_id: + errors.append("client_id is required") + if not access_token: + errors.append("access_token is required") + if not bot_username: + errors.append("bot_username is required") + + channels = config.get("channels", []) + if not channels: + 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"} + if group_policy not in valid_policies: + errors.append(f"group_policy value invalid: '{group_policy}', valid values: {valid_policies}") + + group_allow_from = config.get("group_allow_from", []) + if group_policy == "allowlist" and not group_allow_from: + channels_config = config.get("channels_config", {}) + has_per_channel = any(cfg.get("allow_from") for cfg in channels_config.values() if isinstance(cfg, dict)) + if not has_per_channel: + errors.append("group_policy is 'allowlist' but group_allow_from is empty") + + allowed_roles = config.get("allowedRoles", []) + valid_roles = {"moderator", "owner", "vip", "subscriber", "all"} + for role in allowed_roles: + if role.lower() not in valid_roles: + errors.append(f"allowedRoles contains invalid role: '{role}', valid values: {valid_roles}") + + if "all" in allowed_roles and group_policy == "allowlist": + errors.append("allowedRoles contains 'all' but group_policy is 'allowlist' — this combination is redundant") + + rate_limit = config.get("rate_limit", 20) + rate_window = config.get("rate_window", 30) + if not isinstance(rate_limit, int) or rate_limit < 1: + errors.append(f"rate_limit must be a positive integer, got: {rate_limit}") + if not isinstance(rate_window, int) or rate_window < 1: + errors.append(f"rate_window must be a positive integer, got: {rate_window}") + + irc_port = config.get("irc_port", 6697) + if not isinstance(irc_port, int) or irc_port < 1 or irc_port > 65535: + errors.append(f"irc_port must be 1-65535, got: {irc_port}") + + accounts = config.get("accounts", {}) + if isinstance(accounts, dict): + for acct_id, acct_cfg in accounts.items(): + if not isinstance(acct_cfg, dict): + continue + if not acct_cfg.get("bot_username"): + errors.append(f"account '{acct_id}' missing bot_username") + if not acct_cfg.get("access_token"): + errors.append(f"account '{acct_id}' missing access_token") + + default_account = config.get("defaultAccount", "") + if default_account and isinstance(accounts, dict) and default_account not in accounts: + errors.append(f"defaultAccount '{default_account}' not found in accounts") + + dm_policy = config.get("dm_policy", "pairing") + if dm_policy not in ("open", "pairing"): + errors.append(f"dm_policy value invalid: '{dm_policy}', valid values: open, pairing") + + probe_timeout = config.get("probe_timeout_ms", 10000) + if not isinstance(probe_timeout, int) or probe_timeout < 1000: + errors.append(f"probe_timeout_ms must be >= 1000, got: {probe_timeout}") + + return errors + + +def super_refine_twitch_config(config: dict[str, Any]) -> dict[str, Any]: + refined = dict(config) + + group_policy = refined.get("group_policy", refined.get("groupPolicy", "open")) + if group_policy == "allowall": + logger.info("[TwitchConfig] Converting legacy 'allowall' to 'open'") + refined["group_policy"] = "open" + + 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") + + access_token = refined.get("access_token", "") + if access_token and not access_token.startswith("oauth:"): + refined["access_token"] = f"oauth:{access_token}" + + channels = refined.get("channels", []) + if channels: + refined["channels"] = [ch.lstrip("#@").strip().lower() if isinstance(ch, str) else ch for ch in channels] + + accounts = refined.get("accounts", {}) + if isinstance(accounts, dict): + for acct_id, acct_cfg in list(accounts.items()): + if not isinstance(acct_cfg, dict): + continue + acct_token = acct_cfg.get("access_token", "") + if acct_token and not acct_token.startswith("oauth:"): + acct_cfg["access_token"] = f"oauth:{acct_token}" + acct_channels = acct_cfg.get("channels", []) + if acct_channels: + acct_cfg["channels"] = [ + ch.lstrip("#@").strip().lower() if isinstance(ch, str) else ch for ch in acct_channels + ] + for field in ( + "client_id", + "client_secret", + "rate_limit", + "rate_window", + "irc_host", + "irc_port", + "group_policy", + "require_mention", + "allowedRoles", + "group_allow_from", + "channels_config", + ): + if field not in acct_cfg and field in refined: + acct_cfg[field] = refined[field] + + return refined diff --git a/backend/package/yuxi/channels/adapters/twitch/deduplicator.py b/backend/package/yuxi/channels/adapters/twitch/deduplicator.py new file mode 100644 index 00000000..c0620a86 --- /dev/null +++ b/backend/package/yuxi/channels/adapters/twitch/deduplicator.py @@ -0,0 +1,48 @@ +from __future__ import annotations + +import time +from collections import OrderedDict +from threading import Lock + + +class MessageDeduplicator: + TTL_SECONDS = 60.0 + + def __init__(self, max_size: int = 1000): + self._max_size = max_size + self._cache: OrderedDict[str, float] = OrderedDict() + self._lock = Lock() + + def is_duplicate(self, message_id: str) -> bool: + now = time.monotonic() + with self._lock: + if message_id in self._cache: + ts = self._cache[message_id] + if now - ts < self.TTL_SECONDS: + return True + del self._cache[message_id] + + self._cache[message_id] = now + self._cache.move_to_end(message_id) + + if len(self._cache) > self._max_size: + self._cleanup_expired(now) + + return False + + def mark_seen(self, message_id: str) -> None: + with self._lock: + self._cache[message_id] = time.monotonic() + self._cache.move_to_end(message_id) + + if len(self._cache) > self._max_size: + self._cleanup_expired(time.monotonic()) + + def _cleanup_expired(self, now: float) -> None: + expired = [k for k, ts in self._cache.items() if now - ts >= self.TTL_SECONDS] + for k in expired: + del self._cache[k] + + def __len__(self) -> int: + with self._lock: + return len(self._cache) diff --git a/backend/package/yuxi/channels/adapters/twitch/eventsub_client.py b/backend/package/yuxi/channels/adapters/twitch/eventsub_client.py new file mode 100644 index 00000000..68e790ec --- /dev/null +++ b/backend/package/yuxi/channels/adapters/twitch/eventsub_client.py @@ -0,0 +1,529 @@ +from __future__ import annotations + +import asyncio +from datetime import datetime, UTC +from typing import Any + +import aiohttp + +from yuxi.channels.models import ( + ChannelIdentity, + ChannelMessage, + ChannelType, + ChatType, + EventType, + MessageType, +) +from yuxi.utils.logging_config import logger + +from .helix import HelixClient + + +class EventSubListener: + WS_URL = "wss://eventsub.wss.twitch.tv/ws" + INITIAL_RECONNECT_DELAY = 5 + MAX_RECONNECT_DELAY = 120 + KEEPALIVE_TIMEOUT = 15 + + SUBSCRIPTION_TYPES: list[dict[str, str]] = [ + {"type": "channel.follow", "version": "2"}, + {"type": "channel.subscribe", "version": "1"}, + {"type": "channel.subscription.message", "version": "1"}, + {"type": "channel.subscription.gift", "version": "1"}, + {"type": "channel.cheer", "version": "1"}, + {"type": "channel.raid", "version": "1"}, + {"type": "channel.channel_points_custom_reward_redemption.add", "version": "1"}, + {"type": "channel.channel_points_custom_reward_redemption.update", "version": "1"}, + {"type": "channel.hype_train.begin", "version": "1"}, + {"type": "channel.hype_train.progress", "version": "1"}, + {"type": "channel.hype_train.end", "version": "1"}, + {"type": "channel.ban", "version": "1"}, + {"type": "channel.unban", "version": "1"}, + {"type": "channel.moderator.add", "version": "1"}, + {"type": "channel.moderator.remove", "version": "1"}, + ] + + def __init__( + self, + helix: HelixClient, + app_access_token: str, + broadcaster_ids: list[str], + ): + self._helix = helix + self._app_token = app_access_token + self._broadcaster_ids = broadcaster_ids + self._subscription_ids: list[str] = [] + self._session: aiohttp.ClientSession | None = None + self._ws: aiohttp.ClientWebSocketResponse | None = None + self._session_id: str = "" + self._connected = False + self._running = False + self._task: asyncio.Task | None = None + self._on_message: Any = None + + @property + def connected(self) -> bool: + return self._connected + + def on_event(self, handler) -> None: + self._on_message = handler + + async def connect(self) -> None: + self._running = True + self._task = asyncio.create_task(self._run_loop()) + + async def disconnect(self) -> None: + self._running = False + if self._task and not self._task.done(): + self._task.cancel() + try: + await self._task + except asyncio.CancelledError: + pass + self._task = None + + if self._subscription_ids: + await self._cleanup_subscriptions() + + if self._ws: + try: + await self._ws.close() + except Exception: + pass + if self._session: + try: + await self._session.close() + except Exception: + pass + self._connected = False + + async def _cleanup_subscriptions(self) -> None: + ids = self._subscription_ids[:] + self._subscription_ids.clear() + for sub_id in ids: + try: + await self._helix.delete_eventsub_subscription(sub_id) + except Exception as e: + logger.warning(f"EventSub failed to delete subscription {sub_id}: {e}") + + async def _run_loop(self) -> None: + delay = self.INITIAL_RECONNECT_DELAY + while self._running: + try: + await self._connect_ws() + delay = self.INITIAL_RECONNECT_DELAY + except asyncio.CancelledError: + break + except Exception as e: + logger.warning(f"EventSub error, reconnecting in {delay}s: {e}") + self._connected = False + await asyncio.sleep(delay) + delay = min(delay * 2, self.MAX_RECONNECT_DELAY) + + async def _connect_ws(self) -> None: + self._session = aiohttp.ClientSession() + try: + self._ws = await self._session.ws_connect(self.WS_URL) + + welcome = await self._ws.receive_json() + payload = welcome.get("payload", {}) + self._session_id = payload.get("session", {}).get("id", "") + logger.info(f"EventSub WebSocket connected, session: {self._session_id}") + + await self._subscribe_all() + self._connected = True + + await self._message_loop() + finally: + self._connected = False + if self._session: + await self._session.close() + self._session = None + + async def _subscribe_all(self) -> None: + self._subscription_ids.clear() + for broadcaster_id in self._broadcaster_ids: + for sub_def in self.SUBSCRIPTION_TYPES: + body = { + "type": sub_def["type"], + "version": sub_def["version"], + "condition": {"broadcaster_user_id": broadcaster_id}, + "transport": { + "method": "websocket", + "session_id": self._session_id, + }, + } + sub_id = await self._helix.create_eventsub_subscription(body) + if sub_id: + self._subscription_ids.append(sub_id) + + async def _message_loop(self) -> None: + while self._running and self._ws: + try: + msg = await asyncio.wait_for( + self._ws.receive_json(), + timeout=self.KEEPALIVE_TIMEOUT, + ) + except asyncio.CancelledError: + raise + except TimeoutError: + logger.warning("EventSub keepalive timeout, reconnecting") + break + except Exception as e: + logger.warning(f"EventSub WebSocket error: {e}") + break + + message_type = msg.get("metadata", {}).get("message_type", "") + + if message_type == "session_keepalive": + continue + + if message_type == "session_reconnect": + logger.warning("EventSub reconnect requested") + break + + if message_type == "notification": + payload = msg.get("payload", {}) + subscription = payload.get("subscription", {}) + event = payload.get("event", {}) + + if not event or not subscription: + continue + + channel_msg = self._normalize_event( + { + "subscription": subscription, + "event": event, + } + ) + if channel_msg and self._on_message: + await self._on_message(channel_msg) + + def _normalize_event(self, event_data: dict[str, Any]) -> ChannelMessage | None: + event = event_data.get("event", {}) + subscription_type = event_data.get("subscription", {}).get("type", "") + + broadcaster_id = event.get("broadcaster_user_id", "") + + if subscription_type == "channel.follow": + return self._normalize_follow(event, broadcaster_id) + if subscription_type.startswith("channel.subscribe"): + return self._normalize_subscription(event, broadcaster_id) + if subscription_type == "channel.subscription.message": + return self._normalize_subscription_message(event, broadcaster_id) + if subscription_type == "channel.subscription.gift": + return self._normalize_subscription_gift(event, broadcaster_id) + if subscription_type == "channel.cheer": + return self._normalize_cheer(event, broadcaster_id) + if subscription_type == "channel.raid": + return self._normalize_raid(event, broadcaster_id) + if subscription_type.startswith("channel.channel_points"): + return self._normalize_channel_points(event, broadcaster_id) + if subscription_type.startswith("channel.hype_train"): + return self._normalize_hype_train(event, broadcaster_id) + + if subscription_type == "channel.ban": + return self._normalize_ban(event, broadcaster_id) + if subscription_type == "channel.unban": + return self._normalize_unban(event, broadcaster_id) + if subscription_type == "channel.moderator.add": + return self._normalize_mod_add(event, broadcaster_id) + if subscription_type == "channel.moderator.remove": + return self._normalize_mod_remove(event, broadcaster_id) + + return None + + @staticmethod + def _normalize_follow(event: dict, broadcaster_id: str) -> ChannelMessage: + user_id = event.get("user_id", "") + user_name = event.get("user_name", "") + return ChannelMessage( + identity=ChannelIdentity( + channel_id="twitch", + channel_type=ChannelType.TWITCH, + channel_user_id=user_id, + channel_chat_id=f"#broadcaster_{broadcaster_id}", + channel_message_id=f"eventsub:follow:{event.get('id')}", + ), + event_type=EventType.SYSTEM_EVENT, + message_type=MessageType.TEXT, + chat_type=ChatType.GROUP, + content=f"❤️ {user_name} followed the channel!", + metadata={"event": "follow"}, + timestamp=datetime.now(UTC), + ) + + @staticmethod + def _normalize_subscription(event: dict, broadcaster_id: str) -> ChannelMessage: + user_id = event.get("user_id", "") + user_name = event.get("user_name", "") + tier = event.get("tier", "1000") + return ChannelMessage( + identity=ChannelIdentity( + channel_id="twitch", + channel_type=ChannelType.TWITCH, + channel_user_id=user_id, + channel_chat_id=f"#broadcaster_{broadcaster_id}", + channel_message_id=f"eventsub:sub:{event.get('id')}", + ), + event_type=EventType.SYSTEM_EVENT, + message_type=MessageType.TEXT, + chat_type=ChatType.GROUP, + content=f"🎉 {user_name} subscribed at Tier {tier}!", + metadata={"event": "subscription", "tier": tier}, + timestamp=datetime.now(UTC), + ) + + @staticmethod + def _normalize_subscription_message(event: dict, broadcaster_id: str) -> ChannelMessage: + user_id = event.get("user_id", "") + user_name = event.get("user_name", "") + tier = event.get("tier", "1000") + cumulative_months = event.get("cumulative_months", 0) + message_content = event.get("message", {}).get("text", "") + return ChannelMessage( + identity=ChannelIdentity( + channel_id="twitch", + channel_type=ChannelType.TWITCH, + channel_user_id=user_id, + channel_chat_id=f"#broadcaster_{broadcaster_id}", + channel_message_id=f"eventsub:resub:{event.get('id')}", + ), + event_type=EventType.SYSTEM_EVENT, + message_type=MessageType.TEXT, + chat_type=ChatType.GROUP, + content=f"🔁 {user_name} resubscribed for {cumulative_months} months! Message: {message_content}", + metadata={ + "event": "resubscription", + "tier": tier, + "cumulative_months": cumulative_months, + "message": message_content, + }, + timestamp=datetime.now(UTC), + ) + + @staticmethod + def _normalize_subscription_gift(event: dict, broadcaster_id: str) -> ChannelMessage: + user_id = event.get("user_id", "") + user_name = event.get("user_name", "") + total = event.get("total", 1) + tier = event.get("tier", "1000") + return ChannelMessage( + identity=ChannelIdentity( + channel_id="twitch", + channel_type=ChannelType.TWITCH, + channel_user_id=user_id, + channel_chat_id=f"#broadcaster_{broadcaster_id}", + channel_message_id=f"eventsub:gift:{event.get('id')}", + ), + event_type=EventType.SYSTEM_EVENT, + message_type=MessageType.TEXT, + chat_type=ChatType.GROUP, + content=f"🎁 {user_name} gifted {total} Tier {tier} subs!", + metadata={"event": "subscription_gift", "tier": tier, "total": total}, + timestamp=datetime.now(UTC), + ) + + @staticmethod + def _normalize_cheer(event: dict, broadcaster_id: str) -> ChannelMessage: + user_id = event.get("user_id", "") + user_name = event.get("user_name", "") + bits = event.get("bits", 0) + return ChannelMessage( + identity=ChannelIdentity( + channel_id="twitch", + channel_type=ChannelType.TWITCH, + channel_user_id=user_id, + channel_chat_id=f"#broadcaster_{broadcaster_id}", + channel_message_id=f"eventsub:cheer:{event.get('id')}", + ), + event_type=EventType.SYSTEM_EVENT, + message_type=MessageType.TEXT, + chat_type=ChatType.GROUP, + content=f"💎 {user_name} cheered {bits} Bits!", + metadata={ + "event": "cheer", + "bits": bits, + "message": event.get("message", ""), + }, + timestamp=datetime.now(UTC), + ) + + @staticmethod + def _normalize_raid(event: dict, broadcaster_id: str) -> ChannelMessage: + from_broadcaster_name = event.get("from_broadcaster_user_name", "") + from_broadcaster_id = event.get("from_broadcaster_user_id", "") + viewers = event.get("viewers", 0) + return ChannelMessage( + identity=ChannelIdentity( + channel_id="twitch", + channel_type=ChannelType.TWITCH, + channel_user_id=from_broadcaster_id, + channel_chat_id=f"#broadcaster_{broadcaster_id}", + channel_message_id=f"eventsub:raid:{event.get('id')}", + ), + event_type=EventType.SYSTEM_EVENT, + message_type=MessageType.TEXT, + chat_type=ChatType.GROUP, + content=f"🚀 {from_broadcaster_name} raided with {viewers} viewers!", + metadata={ + "event": "raid", + "from_broadcaster_name": from_broadcaster_name, + "from_broadcaster_id": from_broadcaster_id, + "viewers": viewers, + }, + timestamp=datetime.now(UTC), + ) + + @staticmethod + def _normalize_channel_points(event: dict, broadcaster_id: str) -> ChannelMessage: + user_id = event.get("user_id", "") + user_name = event.get("user_name", "") + reward = event.get("reward", {}) + status = event.get("status", "unfulfilled") + return ChannelMessage( + identity=ChannelIdentity( + channel_id="twitch", + channel_type=ChannelType.TWITCH, + channel_user_id=user_id, + channel_chat_id=f"#broadcaster_{broadcaster_id}", + channel_message_id=f"eventsub:points:{event.get('id')}", + ), + event_type=EventType.SYSTEM_EVENT, + message_type=MessageType.TEXT, + chat_type=ChatType.GROUP, + content=event.get("user_input", ""), + metadata={ + "event": "channel_points", + "reward_title": reward.get("title", ""), + "reward_cost": reward.get("cost", 0), + "reward_id": reward.get("id", ""), + "status": status, + "user_name": user_name, + }, + timestamp=datetime.now(UTC), + ) + + @staticmethod + def _normalize_hype_train(event: dict, broadcaster_id: str) -> ChannelMessage: + level = event.get("level", 1) + progress = event.get("progress", 0) + goal = event.get("goal", 0) + total = event.get("total", 0) + 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:hype_train:{event.get('id')}", + ), + event_type=EventType.SYSTEM_EVENT, + message_type=MessageType.TEXT, + chat_type=ChatType.GROUP, + content=f"🚂 Hype Train Level {level}! ({progress}/{goal})", + metadata={ + "event": "hype_train", + "level": level, + "total": total, + "progress": progress, + "goal": goal, + "top_contributions": event.get("top_contributions", []), + }, + timestamp=datetime.now(UTC), + ) + + @staticmethod + def _normalize_ban(event: dict, broadcaster_id: str) -> ChannelMessage: + user_name = event.get("user_name", "") + user_id = event.get("user_id", "") + moderator_name = event.get("moderator_user_name", "") + reason = event.get("reason", "") + expires_at = event.get("ends_at", "") + permanent = event.get("is_permanent", False) + duration = "permanent" if permanent else f"until {expires_at}" + return ChannelMessage( + identity=ChannelIdentity( + channel_id="twitch", + channel_type=ChannelType.TWITCH, + channel_user_id=user_id, + channel_chat_id=f"#broadcaster_{broadcaster_id}", + channel_message_id=f"eventsub:ban:{event.get('id')}", + ), + event_type=EventType.SYSTEM_EVENT, + message_type=MessageType.TEXT, + chat_type=ChatType.GROUP, + content=f"🚫 {user_name} banned by {moderator_name} ({duration})" + (f": {reason}" if reason else ""), + metadata={ + "event": "ban", + "reason": reason, + "permanent": permanent, + "expires_at": expires_at, + "moderator_name": moderator_name, + }, + timestamp=datetime.now(UTC), + ) + + @staticmethod + def _normalize_unban(event: dict, broadcaster_id: str) -> ChannelMessage: + user_name = event.get("user_name", "") + user_id = event.get("user_id", "") + moderator_name = event.get("moderator_user_name", "") + return ChannelMessage( + identity=ChannelIdentity( + channel_id="twitch", + channel_type=ChannelType.TWITCH, + channel_user_id=user_id, + channel_chat_id=f"#broadcaster_{broadcaster_id}", + channel_message_id=f"eventsub:unban:{event.get('id')}", + ), + event_type=EventType.SYSTEM_EVENT, + message_type=MessageType.TEXT, + chat_type=ChatType.GROUP, + content=f"✅ {user_name} unbanned by {moderator_name}", + metadata={ + "event": "unban", + "moderator_name": moderator_name, + }, + timestamp=datetime.now(UTC), + ) + + @staticmethod + def _normalize_mod_add(event: dict, broadcaster_id: str) -> ChannelMessage: + user_name = event.get("user_name", "") + user_id = event.get("user_id", "") + return ChannelMessage( + identity=ChannelIdentity( + channel_id="twitch", + channel_type=ChannelType.TWITCH, + channel_user_id=user_id, + channel_chat_id=f"#broadcaster_{broadcaster_id}", + channel_message_id=f"eventsub:mod_add:{event.get('id')}", + ), + event_type=EventType.SYSTEM_EVENT, + message_type=MessageType.TEXT, + chat_type=ChatType.GROUP, + content=f"🛡️ {user_name} is now a moderator", + metadata={"event": "moderator_add"}, + timestamp=datetime.now(UTC), + ) + + @staticmethod + def _normalize_mod_remove(event: dict, broadcaster_id: str) -> ChannelMessage: + user_name = event.get("user_name", "") + user_id = event.get("user_id", "") + return ChannelMessage( + identity=ChannelIdentity( + channel_id="twitch", + channel_type=ChannelType.TWITCH, + channel_user_id=user_id, + channel_chat_id=f"#broadcaster_{broadcaster_id}", + channel_message_id=f"eventsub:mod_remove:{event.get('id')}", + ), + event_type=EventType.SYSTEM_EVENT, + message_type=MessageType.TEXT, + chat_type=ChatType.GROUP, + content=f"⬇️ {user_name} moderator removed", + metadata={"event": "moderator_remove"}, + timestamp=datetime.now(UTC), + ) diff --git a/backend/package/yuxi/channels/adapters/twitch/helix.py b/backend/package/yuxi/channels/adapters/twitch/helix.py new file mode 100644 index 00000000..2bf0570c --- /dev/null +++ b/backend/package/yuxi/channels/adapters/twitch/helix.py @@ -0,0 +1,192 @@ +from __future__ import annotations + +import asyncio +from typing import Any + +import aiohttp + +from yuxi.utils.logging_config import logger + + +class HelixClient: + """Twitch Helix API 客户端""" + + BASE_URL = "https://api.twitch.tv/helix" + AUTH_URL = "https://id.twitch.tv/oauth2/token" + + MAX_RETRIES = 3 + BASE_BACKOFF = 1.0 + MAX_BACKOFF = 10.0 + RETRY_STATUSES = frozenset({401, 429, 500, 502, 503, 504}) + + def __init__(self, client_id: str, access_token: str): + self._client_id = client_id + self._access_token = access_token + self._session: aiohttp.ClientSession | None = None + + async def start(self) -> None: + if self._session is None or self._session.closed: + self._session = aiohttp.ClientSession() + + async def close(self) -> None: + if self._session and not self._session.closed: + await self._session.close() + self._session = None + + @property + def _ensure_session(self) -> aiohttp.ClientSession: + if self._session is None or self._session.closed: + raise RuntimeError("HelixClient session not started, call start() first") + return self._session + + def _headers(self) -> dict[str, str]: + return { + "Authorization": f"Bearer {self._access_token}", + "Client-Id": self._client_id, + } + + 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.warning(f"Helix {method} {url} attempt {attempt + 1} failed: {e}") + + delay = min(self.BASE_BACKOFF * (2**attempt), self.MAX_BACKOFF) + logger.info(f"Helix {method} {url} retrying in {delay:.1f}s (attempt {attempt + 2}/{self.MAX_RETRIES})") + await asyncio.sleep(delay) + + return None + + async def _get(self, path: str, **params: str) -> dict[str, Any] | None: + url = f"{self.BASE_URL}{path}" + resp = await self._request_with_backoff("GET", url, headers=self._headers(), params=params) + if resp is None: + return None + try: + if resp.status == 200: + return await resp.json() + if resp.status == 401: + logger.warning(f"Helix GET {path} returned 401 after {self.MAX_RETRIES} retries") + return None + if resp.status == 404: + return None + logger.error(f"Helix GET {path} failed: {resp.status} {await resp.text()}") + return None + except aiohttp.ClientError as e: + logger.error(f"Helix GET {path} connection error: {e}") + return None + + async def validate_token(self) -> dict[str, Any] | None: + data = await self._get("/users") + if data: + users = data.get("data", []) + return users[0] if users else None + return None + + async def get_user_by_name(self, username: str) -> dict[str, Any] | None: + data = await self._get("/users", login=username) + if data: + users = data.get("data", []) + return users[0] if users else None + return None + + async def get_user_by_id(self, user_id: str) -> dict[str, Any] | None: + data = await self._get("/users", id=user_id) + if data: + users = data.get("data", []) + return users[0] if users else None + return None + + async def get_channel_info(self, broadcaster_id: str) -> dict[str, Any] | None: + data = await self._get("/channels", broadcaster_id=broadcaster_id) + if data: + channels = data.get("data", []) + return channels[0] if channels else None + return None + + async def get_chat_badges(self, broadcaster_id: str) -> dict[str, Any] | None: + return await self._get("/chat/badges", broadcaster_id=broadcaster_id) + + async def get_global_chat_badges(self) -> dict[str, Any] | None: + return await self._get("/chat/badges/global") + + async def create_eventsub_subscription(self, payload: dict[str, Any]) -> str | None: + url = f"{self.BASE_URL}/eventsub/subscriptions" + resp = await self._request_with_backoff("POST", url, headers=self._headers(), json=payload) + if resp is None: + return None + try: + if resp.status in (200, 202): + data = await resp.json() + subs = data.get("data", []) + if subs: + return subs[0].get("id") + return None + if resp.status == 409: + return None + if resp.status == 429: + logger.warning("EventSub subscription rate limited") + return None + body = await resp.text() + logger.error(f"EventSub subscription failed ({resp.status}): {body}") + return None + except aiohttp.ClientError as e: + logger.error(f"EventSub subscription connection error: {e}") + return None + + async def get_app_access_token(self, client_secret: str) -> str | None: + params = { + "client_id": self._client_id, + "client_secret": client_secret, + "grant_type": "client_credentials", + } + resp = await self._request_with_backoff("POST", self.AUTH_URL, json=params) + if resp is None: + return None + try: + if resp.status == 200: + data = await resp.json() + return data.get("access_token") + logger.error(f"App access token request failed: {resp.status} {await resp.text()}") + return None + except aiohttp.ClientError as e: + logger.error(f"App access token connection error: {e}") + return None + + async def refresh_user_token(self, client_secret: str, refresh_token: str) -> dict[str, Any] | None: + params = { + "client_id": self._client_id, + "client_secret": client_secret, + "grant_type": "refresh_token", + "refresh_token": refresh_token, + } + resp = await self._request_with_backoff("POST", self.AUTH_URL, json=params) + if resp is None: + return None + try: + if resp.status == 200: + return await resp.json() + logger.error(f"Token refresh failed: {resp.status} {await resp.text()}") + return None + except aiohttp.ClientError as e: + logger.error(f"Token refresh connection error: {e}") + return None + + async def delete_eventsub_subscription(self, subscription_id: str) -> bool: + url = f"{self.BASE_URL}/eventsub/subscriptions" + params = {"id": subscription_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"EventSub delete subscription error: {e}") + return False diff --git a/backend/package/yuxi/channels/adapters/twitch/irc_parser.py b/backend/package/yuxi/channels/adapters/twitch/irc_parser.py new file mode 100644 index 00000000..43dcf872 --- /dev/null +++ b/backend/package/yuxi/channels/adapters/twitch/irc_parser.py @@ -0,0 +1,142 @@ +from __future__ import annotations + +import re +import time +from dataclasses import dataclass +from typing import Any + +_IRC_LINE_PATTERN = re.compile( + r"^(?:@(?P[^ ]+) )?(?::(?P[^ ]+) )?(?P[a-zA-Z]+|\d{3})" + r"(?: (?P.+?))?(?: :(?P.*))?$" +) + +_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}" diff --git a/backend/package/yuxi/channels/adapters/twitch/markdown_utils.py b/backend/package/yuxi/channels/adapters/twitch/markdown_utils.py new file mode 100644 index 00000000..e0949852 --- /dev/null +++ b/backend/package/yuxi/channels/adapters/twitch/markdown_utils.py @@ -0,0 +1,55 @@ +from __future__ import annotations + +import re + +_LINK_PATTERN = re.compile(r"\[([^\]]*)\]\([^)]*\)") +_IMG_PATTERN = re.compile(r"!\[([^\]]*)\]\([^)]*\)") +_BOLD_PATTERN = re.compile(r"\*\*(.+?)\*\*") +_BOLD_ALT_PATTERN = re.compile(r"__(.+?)__") +_ITALIC_PATTERN = re.compile(r"\*(.+?)\*") +_ITALIC_ALT_PATTERN = re.compile(r"_(.+?)_") +_STRIKETHROUGH_PATTERN = re.compile(r"~~(.+?)~~") +_FENCED_CODE_PATTERN = re.compile(r"```[^\n]*\n(.*?)```", re.DOTALL) +_INLINE_CODE_PATTERN = re.compile(r"`([^`]+)`") +_HEADING_PATTERN = re.compile(r"^#{1,6}\s+", re.MULTILINE) +_UNORDERED_LIST_PATTERN = re.compile(r"^[\-\*]\s+", re.MULTILINE) +_ORDERED_LIST_PATTERN = re.compile(r"^\d+\.\s+", re.MULTILINE) +_BLOCKQUOTE_PATTERN = re.compile(r"^>\s?", re.MULTILINE) +_HR_PATTERN = re.compile(r"^[\-\*_]{3,}\s*$", re.MULTILINE) +_MULTISPACE_PATTERN = re.compile(r"[ \t]+") +_MULTINEWLINE_PATTERN = re.compile(r"\n{3,}") + + +def strip_twitch_markdown(text: str) -> str: + if not text: + return text + + text = _IMG_PATTERN.sub("", text) + + text = _LINK_PATTERN.sub(r"\1", text) + + text = _FENCED_CODE_PATTERN.sub(r"\1", text) + + text = _HR_PATTERN.sub("", text) + + text = _BOLD_PATTERN.sub(r"\1", text) + text = _BOLD_ALT_PATTERN.sub(r"\1", text) + + text = _ITALIC_PATTERN.sub(r"\1", text) + text = _ITALIC_ALT_PATTERN.sub(r"\1", text) + + text = _STRIKETHROUGH_PATTERN.sub(r"\1", text) + + text = _INLINE_CODE_PATTERN.sub(r"\1", text) + + text = _HEADING_PATTERN.sub("", text) + + text = _ORDERED_LIST_PATTERN.sub("", text) + text = _UNORDERED_LIST_PATTERN.sub("", text) + + text = _BLOCKQUOTE_PATTERN.sub("", text) + + text = _MULTISPACE_PATTERN.sub(" ", text) + text = _MULTINEWLINE_PATTERN.sub("\n\n", text) + + return text.strip() diff --git a/backend/package/yuxi/channels/adapters/twitch/normalizer.py b/backend/package/yuxi/channels/adapters/twitch/normalizer.py new file mode 100644 index 00000000..0cd0a251 --- /dev/null +++ b/backend/package/yuxi/channels/adapters/twitch/normalizer.py @@ -0,0 +1,233 @@ +from __future__ import annotations + +from datetime import datetime, UTC +from typing import Any + +from yuxi.channels.models import ( + ChannelIdentity, + ChannelMessage, + ChannelType, + ChatType, + EventType, + MentionsInfo, + MessageType, +) + +from .irc_parser import ( + ParsedIRCMessage, + extract_nick, + parse_badges, + parse_emotes, + make_irc_message_id, +) + +_FILTERED_NOTICE_MSG_IDS = frozenset( + { + "subs_on", + "subs_off", + "emote_only_on", + "emote_only_off", + "slow_on", + "slow_off", + "followers_on", + "followers_off", + "r9k_on", + "r9k_off", + "host_on", + "host_off", + } +) + +_EVENT_MAP: dict[str, str] = { + "sub": "subscription", + "resub": "resubscription", + "subgift": "subscription_gift", + "anonsubgift": "anonymous_subscription_gift", + "submysterygift": "subscription_mystery_gift", + "raid": "raid", + "ritual": "ritual", + "announcement": "announcement", +} + +_SKIP_COMMANDS = frozenset({"JOIN", "PART", "HOSTTARGET", "ROOMSTATE", "USERSTATE"}) + + +def _make_identity( + channel_user_id: str, + channel_chat_id: str, + channel_message_id: str | None = None, +) -> ChannelIdentity: + return ChannelIdentity( + channel_id="twitch", + channel_type=ChannelType.TWITCH, + channel_user_id=channel_user_id, + channel_chat_id=channel_chat_id, + channel_message_id=channel_message_id, + ) + + +def _make_timestamp(tmi_sent_ts: str | None) -> datetime: + ts = int(tmi_sent_ts or "0") + if ts: + return datetime.fromtimestamp(ts / 1000, tz=UTC) + return datetime.now(UTC) + + +def resolve_mentions(content: str, bot_username: str | None = None) -> MentionsInfo: + mentioned: list[str] = [] + is_bot_mentioned = False + bot_lower = bot_username.lower() if bot_username else None + for word in content.split(): + if word.startswith("@"): + name = word[1:].rstrip(",.") + mentioned.append(name) + if bot_lower and name.lower() == bot_lower: + is_bot_mentioned = True + return MentionsInfo( + mentioned_user_ids=mentioned, + is_bot_mentioned=is_bot_mentioned, + ) + + +def normalize_irc_message(parsed: ParsedIRCMessage) -> ChannelMessage | None: + command = parsed.command + tags = parsed.tags + + # ---- PRIVMSG: 普通聊天消息 ---- + if command == "PRIVMSG": + channel = parsed.params[0] if parsed.params else "" + user_id = tags.get("user-id") or extract_nick(parsed.prefix) + message_id = tags.get("id") or tags.get("tmi-sent-ts") or make_irc_message_id(user_id) + + identity = _make_identity(user_id, channel, message_id) + + bits = int(tags.get("bits", "0")) + badges = parse_badges(tags.get("badges", "")) + emotes = parse_emotes(tags.get("emotes", ""), parsed.trailing) + + metadata: dict[str, Any] = { + "display_name": tags.get("display-name", user_id), + "color": tags.get("color"), + "badges": badges, + "emotes": emotes, + "bits": bits, + "is_mod": tags.get("mod") == "1", + "is_subscriber": tags.get("subscriber") == "1", + "is_turbo": tags.get("turbo") == "1", + "user_type": tags.get("user-type"), + "tmi_sent_ts": tags.get("tmi-sent-ts"), + } + + return ChannelMessage( + identity=identity, + event_type=EventType.MESSAGE_RECEIVED, + message_type=MessageType.TEXT, + chat_type=ChatType.GROUP, + content=parsed.trailing, + mentions=resolve_mentions(parsed.trailing), + metadata=metadata, + timestamp=_make_timestamp(tags.get("tmi-sent-ts")), + ) + + # ---- USERNOTICE: 订阅/赠礼/Raid/新人 ---- + if command == "USERNOTICE": + msg_id = tags.get("msg-id", "") + channel = parsed.params[0] if parsed.params else "" + user_id = tags.get("user-id", "system") + message_id = tags.get("id") or make_irc_message_id(user_id, tags.get("tmi-sent-ts")) + + identity = _make_identity(user_id, channel, message_id) + + metadata = { + "event": _EVENT_MAP.get(msg_id, msg_id), + "display_name": tags.get("display-name", user_id), + "msg_param_sub_plan": tags.get("msg-param-sub-plan"), + "msg_param_months": tags.get("msg-param-cumulative-months"), + "msg_param_recipient": tags.get("msg-param-recipient-user-name"), + "msg_param_gift_count": tags.get("msg-param-gift-months"), + "msg_param_viewer_count": tags.get("msg-param-viewerCount"), + "system_msg": tags.get("system-msg", parsed.trailing), + } + + return ChannelMessage( + identity=identity, + event_type=EventType.SYSTEM_EVENT, + message_type=MessageType.TEXT, + chat_type=ChatType.GROUP, + content=tags.get("system-msg", parsed.trailing), + metadata=metadata, + timestamp=_make_timestamp(tags.get("tmi-sent-ts")), + ) + + # ---- CLEARCHAT: 消息删除/用户 timeout ---- + if command == "CLEARCHAT": + channel = parsed.params[0] if parsed.params else "" + target_user = tags.get("target-user-id", "") or parsed.trailing + ban_duration = tags.get("ban-duration", "0") + + return ChannelMessage( + identity=_make_identity(target_user, channel), + event_type=EventType.MESSAGE_DELETED, + message_type=MessageType.TEXT, + chat_type=ChatType.GROUP, + content="", + metadata={ + "event": "clearchat", + "ban_duration_sec": int(ban_duration), + }, + timestamp=datetime.now(UTC), + ) + + # ---- WHISPER: 私信 ---- + if command == "WHISPER": + user_id = tags.get("user-id") or extract_nick(parsed.prefix) + message_id = tags.get("message-id") or make_irc_message_id(user_id) + identity = _make_identity(user_id, f"#whisper_{user_id}", message_id) + display_name = tags.get("display-name", user_id) + + metadata: dict[str, Any] = { + "display_name": display_name, + "color": tags.get("color"), + "badges": parse_badges(tags.get("badges", "")), + "emotes": parse_emotes(tags.get("emotes", ""), parsed.trailing), + "thread_id": tags.get("thread-id"), + } + + return ChannelMessage( + identity=identity, + event_type=EventType.MESSAGE_RECEIVED, + message_type=MessageType.TEXT, + chat_type=ChatType.DIRECT, + content=parsed.trailing, + mentions=resolve_mentions(parsed.trailing), + metadata=metadata, + timestamp=_make_timestamp(tags.get("tmi-sent-ts")), + ) + + # ---- NOTICE: 系统通知 ---- + if command == "NOTICE": + channel = parsed.params[0] if parsed.params else "" + msg_id = tags.get("msg-id", "") + + if msg_id in _FILTERED_NOTICE_MSG_IDS: + return None + + return ChannelMessage( + identity=_make_identity("twitch_system", channel), + event_type=EventType.SYSTEM_EVENT, + message_type=MessageType.TEXT, + chat_type=ChatType.GROUP, + content=parsed.trailing, + metadata={"event": "notice", "msg_id": msg_id}, + timestamp=datetime.now(UTC), + ) + + # ---- PING → 内部处理 ---- + if command == "PING": + return None + + # ---- JOIN/PART/HOSTTARGET/ROOMSTATE/USERSTATE → 内部状态 ---- + if command in _SKIP_COMMANDS: + return None + + return None diff --git a/backend/package/yuxi/channels/adapters/twitch/outbound.py b/backend/package/yuxi/channels/adapters/twitch/outbound.py new file mode 100644 index 00000000..b59eea87 --- /dev/null +++ b/backend/package/yuxi/channels/adapters/twitch/outbound.py @@ -0,0 +1,91 @@ +from __future__ import annotations + +from typing import Any + +from yuxi.channels.models import DeliveryResult + +from .send import format_privmsg_line, format_action_line + + +class TwitchOutboundAdapter: + def __init__(self, writer: Any, rate_limiter: Any, silent: bool = False): + self._writer = writer + self._rate_limiter = rate_limiter + self._silent = silent + + @property + def silent(self) -> bool: + return self._silent + + @silent.setter + def silent(self, value: bool) -> None: + self._silent = value + + def _fmt(self, target: str, text: str) -> str: + return format_action_line(target, text) if self._silent else format_privmsg_line(target, text) + + @staticmethod + def split_text(text: str, target: str, prefix_len: int | None = None) -> list[str]: + if prefix_len is None: + prefix_len = len(f"PRIVMSG {target} :") + available = max(200, 510 - prefix_len) + if available < 1: + available = 200 + + chunks: list[str] = [] + if not text: + return [""] + remaining = text + + while remaining: + encoded = remaining.encode("utf-8") + if len(encoded) <= available: + chunks.append(remaining) + break + + cut = _find_utf8_cut(encoded, available) + split_pos = len(encoded[:cut].decode("utf-8", errors="replace")) + + nl = remaining.rfind("\n", 0, split_pos) + if nl > split_pos * 0.5: + split_pos = nl + 1 + else: + sp = remaining.rfind(" ", 0, split_pos) + if sp > split_pos * 0.5: + split_pos = sp + 1 + + chunk = remaining[:split_pos].rstrip() + chunks.append(chunk) + remaining = remaining[split_pos:].lstrip() + + return chunks + + async def send_chunks(self, target: str, chunks: list[str]) -> DeliveryResult: + for chunk in chunks: + if self._rate_limiter and not await self._rate_limiter.acquire(): + return DeliveryResult(success=False, error="rate_limit_exceeded") + self._writer.write(self._fmt(target, chunk).encode("utf-8") + b"\r\n") + await self._writer.drain() + return DeliveryResult(success=True) + + async def send_text(self, target: str, text: str) -> DeliveryResult: + chunks = self.split_text(text, target) + return await self.send_chunks(target, chunks) + + async def send_single_line(self, target: str, text: str) -> DeliveryResult: + if self._rate_limiter and not await self._rate_limiter.acquire(): + return DeliveryResult(success=False, error="rate_limit_exceeded") + self._writer.write(self._fmt(target, text).encode("utf-8") + b"\r\n") + await self._writer.drain() + return DeliveryResult(success=True) + + +def _find_utf8_cut(encoded: bytes, byte_limit: int) -> int: + cut = byte_limit + while cut > 0 and (encoded[cut - 1] & 0xC0) == 0x80: + cut -= 1 + while cut > 0 and (encoded[cut - 1] & 0xC0) == 0xC0: + cut -= 1 + if cut == 0: + cut = max(1, byte_limit) + return cut diff --git a/backend/package/yuxi/channels/adapters/twitch/pairing.py b/backend/package/yuxi/channels/adapters/twitch/pairing.py new file mode 100644 index 00000000..3160cc6e --- /dev/null +++ b/backend/package/yuxi/channels/adapters/twitch/pairing.py @@ -0,0 +1,112 @@ +from __future__ import annotations + +import re +import time +from collections import OrderedDict +from threading import Lock +from typing import Any + +_PAIRING_PREFIX_PATTERN = re.compile(r"^(twitch:|user:)", re.IGNORECASE) + + +def strip_pairing_prefix(user_id: str) -> str: + return _PAIRING_PREFIX_PATTERN.sub("", user_id).strip() + + +class PairingStore: + MAX_PENDING = 500 + PENDING_TTL = 300.0 + + def __init__(self): + self._approved: dict[str, float] = {} + self._pending: OrderedDict[str, float] = OrderedDict() + self._lock = Lock() + self._on_approve: Any = None + self._on_approve_channel: str = "" + + def set_on_approve(self, callback: Any, channel: str = "") -> None: + self._on_approve = callback + self._on_approve_channel = channel + + def _normalize(self, user_id: str) -> str: + return strip_pairing_prefix(user_id) + + def is_approved(self, user_id: str) -> bool: + with self._lock: + return self._normalize(user_id) in self._approved + + def is_pending(self, user_id: str) -> bool: + with self._lock: + return self._normalize(user_id) in self._pending + + def add_pending(self, user_id: str, user_name: str, channel: str) -> None: + uid = self._normalize(user_id) + with self._lock: + self._pending[uid] = time.monotonic() + self._pending.move_to_end(uid) + self._evict_expired_pending() + + def approve(self, user_id: str, user_name: str = "") -> bool: + uid = self._normalize(user_id) + with self._lock: + if uid in self._pending: + del self._pending[uid] + self._approved[uid] = time.monotonic() + if self._on_approve: + try: + self._on_approve(user_name or uid, self._on_approve_channel) + except Exception: + pass + return True + + def reject(self, user_id: str) -> bool: + uid = self._normalize(user_id) + with self._lock: + if uid in self._pending: + del self._pending[uid] + return True + return False + + def remove_approval(self, user_id: str) -> bool: + uid = self._normalize(user_id) + with self._lock: + if uid in self._approved: + del self._approved[uid] + return True + return False + + def get_approved_users(self) -> list[str]: + with self._lock: + return list(self._approved.keys()) + + def get_pending_users(self) -> list[str]: + with self._lock: + self._evict_expired_pending() + return list(self._pending.keys()) + + def _evict_expired_pending(self) -> None: + now = time.monotonic() + expired = [uid for uid, ts in self._pending.items() if now - ts >= self.PENDING_TTL] + for uid in expired: + del self._pending[uid] + + def __len__(self) -> int: + with self._lock: + return len(self._approved) + + @property + def pending_count(self) -> int: + with self._lock: + return len(self._pending) + + +def check_pairing_policy( + pairings: PairingStore, + user_id: str, + content: str | None = None, +) -> bool: + return pairings.is_approved(user_id) + + +def format_pairing_notification(user_name: str, approval_code: str) -> str: + return f"{user_name} wants to interact. Approve with: !approve {approval_code}" diff --git a/backend/package/yuxi/channels/adapters/twitch/probe.py b/backend/package/yuxi/channels/adapters/twitch/probe.py new file mode 100644 index 00000000..596e9185 --- /dev/null +++ b/backend/package/yuxi/channels/adapters/twitch/probe.py @@ -0,0 +1,89 @@ +from __future__ import annotations + +import time +from typing import Any + +from yuxi.channels.models import HealthStatus + +from .helix import HelixClient + + +async def health_check(config: dict[str, Any], timeout_ms: int = 10000) -> HealthStatus: + start = time.monotonic() + issues: list[str] = [] + + token = config.get("access_token", "") + client_id = config.get("client_id", "") + + if not token or not client_id: + return HealthStatus( + status="unhealthy", + last_error="Missing client_id or access_token", + latency_ms=(time.monotonic() - start) * 1000, + ) + + helix = HelixClient(client_id, token) + try: + await helix.start() + user_info = await helix.validate_token() + if user_info is None: + issues.append("Token invalid or expired (401)") + except Exception as e: + issues.append(f"Helix API unreachable: {e}") + finally: + await helix.close() + + latency_ms = (time.monotonic() - start) * 1000 + probe_timeout_ms = config.get("probe_timeout_ms", timeout_ms) + if latency_ms > probe_timeout_ms: + issues.append(f"Probe timeout: {latency_ms:.0f}ms > {probe_timeout_ms}ms") + + if issues: + return HealthStatus( + status="unhealthy", + last_error="; ".join(issues), + latency_ms=latency_ms, + metadata={"timeout_ms": probe_timeout_ms}, + ) + return HealthStatus( + status="healthy", + latency_ms=latency_ms, + metadata={"api": "helix", "timeout_ms": probe_timeout_ms}, + ) + + +async def validate_token(client_id: str, access_token: str) -> dict[str, Any] | None: + helix = HelixClient(client_id, access_token) + try: + await helix.start() + return await helix.validate_token() + finally: + await helix.close() + + +async def get_broadcaster_id(client_id: str, access_token: str, channel_name: str) -> str | None: + helix = HelixClient(client_id, access_token) + try: + await helix.start() + user = await helix.get_user_by_name(channel_name) + return user["id"] if user else None + finally: + await helix.close() + + +async def refresh_access_token(client_id: str, client_secret: str, refresh_token: str) -> dict[str, Any] | None: + helix = HelixClient(client_id, "") + try: + await helix.start() + return await helix.refresh_user_token(client_secret, refresh_token) + finally: + await helix.close() + + +async def get_app_access_token(client_id: str, client_secret: str) -> str | None: + helix = HelixClient(client_id, "") + try: + await helix.start() + return await helix.get_app_access_token(client_secret) + finally: + await helix.close() diff --git a/backend/package/yuxi/channels/adapters/twitch/rate_limiter.py b/backend/package/yuxi/channels/adapters/twitch/rate_limiter.py new file mode 100644 index 00000000..b8f5efe7 --- /dev/null +++ b/backend/package/yuxi/channels/adapters/twitch/rate_limiter.py @@ -0,0 +1,62 @@ +from __future__ import annotations + +import asyncio +import time + + +class RateLimiter: + """Twitch 速率限制令牌桶 + + 默认 20msg/30s (known user)，Mod/VIP 切换至 100msg/30s。 + mod 模式 60 秒后自动切回 known 模式。 + """ + + MOD_TIMEOUT = 60.0 + + def __init__(self, limit: int = 20, window: float = 30.0, mod_limit: int = 100): + self._default_limit = limit + self._mod_limit = mod_limit + self._current_limit = limit + self._window = window + self._tokens = float(limit) + self._last_refill = time.monotonic() + self._mod_until: float | None = None + self._lock = asyncio.Lock() + + async def switch_to_mod(self) -> None: + async with self._lock: + if self._current_limit == self._mod_limit: + return + self._current_limit = self._mod_limit + self._tokens = float(self._mod_limit) + self._last_refill = time.monotonic() + self._mod_until = time.monotonic() + self.MOD_TIMEOUT + + async def switch_to_known(self) -> None: + async with self._lock: + self._current_limit = self._default_limit + self._tokens = float(self._default_limit) + self._last_refill = time.monotonic() + self._mod_until = None + + async def acquire(self) -> bool: + async with self._lock: + now = time.monotonic() + + if self._mod_until is not None and now >= self._mod_until: + self._current_limit = self._default_limit + self._mod_until = None + + elapsed = now - self._last_refill + refill = elapsed * (self._current_limit / self._window) + self._tokens = min(float(self._current_limit), self._tokens + refill) + self._last_refill = now + + if self._tokens >= 1.0: + self._tokens -= 1.0 + return True + return False + + @property + def available_tokens(self) -> float: + return self._tokens diff --git a/backend/package/yuxi/channels/adapters/twitch/resolver.py b/backend/package/yuxi/channels/adapters/twitch/resolver.py new file mode 100644 index 00000000..049ef744 --- /dev/null +++ b/backend/package/yuxi/channels/adapters/twitch/resolver.py @@ -0,0 +1,47 @@ +from __future__ import annotations + +from typing import Any + +from yuxi.utils.logging_config import logger + +from .helix import HelixClient + + +def _is_numeric_id(value: str) -> bool: + return value.isdigit() + + +async def resolve_twitch_target( + input_value: str, + client_id: str, + access_token: str, +) -> dict[str, Any] | None: + """将输入（用户名或用户 ID）解析为 Twitch 用户信息""" + value = input_value.strip().lstrip("@").lower() + + if not value: + return None + + if not client_id or not access_token: + logger.warning("resolve_twitch_target: missing credentials") + return None + + helix = HelixClient(client_id, access_token) + try: + await helix.start() + + if _is_numeric_id(value): + user = await helix.get_user_by_id(value) + if user: + return user + return None + + user = await helix.get_user_by_name(value) + if user: + return user + return None + except Exception as e: + logger.error(f"resolve_twitch_target error for '{input_value}': {e}") + return None + finally: + await helix.close() diff --git a/backend/package/yuxi/channels/adapters/twitch/send.py b/backend/package/yuxi/channels/adapters/twitch/send.py new file mode 100644 index 00000000..b4465049 --- /dev/null +++ b/backend/package/yuxi/channels/adapters/twitch/send.py @@ -0,0 +1,17 @@ +from __future__ import annotations + + +def format_privmsg_line(target: str, text: str) -> str: + return f"PRIVMSG {target} :{text}" + + +def format_action_line(target: str, text: str) -> str: + return f"PRIVMSG {target} :\x01ACTION {text}\x01" + + +def format_pong(token: str) -> str: + return f"PONG :{token}" + + +def format_cap_req(capabilities: list[str]) -> str: + return f"CAP REQ :{' '.join(capabilities)}" diff --git a/backend/package/yuxi/channels/adapters/twitch/session.py b/backend/package/yuxi/channels/adapters/twitch/session.py new file mode 100644 index 00000000..4c2cfd62 --- /dev/null +++ b/backend/package/yuxi/channels/adapters/twitch/session.py @@ -0,0 +1,110 @@ +from __future__ import annotations + +from typing import Any + +TWITCH_ROLES = frozenset({"moderator", "owner", "vip", "subscriber", "all"}) + + +def normalize_twitch_channel(channel: str) -> str: + return channel.lstrip("#@").strip().lower() + + +def resolve_channel_name(channel_chat_id: str) -> str: + return normalize_twitch_channel(channel_chat_id) + + +def resolve_route( + agent_id: str, + channel_name: str, +) -> str: + channel_name = normalize_twitch_channel(channel_name) + return f"agent:{agent_id}:twitch:channel:{channel_name}" + + +def resolve_agent_id_for_channel( + config: dict[str, Any], + channel_name: str, + default_agent_id: str | None = None, +) -> str: + channel_name = normalize_twitch_channel(channel_name) + channels_config = config.get("channels_config", {}) + channel_config = channels_config.get(channel_name, {}) + return channel_config.get("agent_id") or default_agent_id or "" + + +def check_require_mention( + config: dict[str, Any], + content: str, +) -> bool: + bot_name = config.get("bot_username", "").lower() + require_mention = config.get("require_mention", True) + if not require_mention: + return True + return bool(bot_name) and bot_name in content.lower() + + +def check_allowed_roles( + config: dict[str, Any], + channel: str, + badges: dict[str, str], +) -> bool: + allowed_roles: list[str] = config.get("allowedRoles", []) + + if "all" in allowed_roles: + return True + + channels_config = config.get("channels_config", {}) + channel_name = normalize_twitch_channel(channel) + per_channel_roles: list[str] = channels_config.get(channel_name, {}).get("allowedRoles", []) + if "all" in per_channel_roles: + return True + + effective_roles = list(dict.fromkeys(allowed_roles + per_channel_roles)) + if not effective_roles: + return True + + for role in effective_roles: + role_lower = role.lower() + if role_lower not in TWITCH_ROLES: + continue + if role_lower == "all": + return True + if role_lower == "owner" and "broadcaster" in badges: + return True + if role_lower == "moderator" and "moderator" in badges: + return True + if role_lower == "vip" and "vip" in badges: + return True + if role_lower == "subscriber" and ("subscriber" in badges or "founder" in badges): + return True + + return False + + +def check_group_policy( + config: dict[str, Any], + channel: str, + user_id: str, + content: str, +) -> bool: + channel_name = normalize_twitch_channel(channel) + group_policy = config.get("group_policy", "open") + bot_name = config.get("bot_username", "").lower() + + match group_policy: + case "open": + return True + case "disabled": + return False + case "mention_only": + require_mention = config.get("require_mention", True) + if not require_mention: + return True + return bool(bot_name) and bot_name in content.lower() + case "allowlist": + allowlist = config.get("group_allow_from", []) + channels_config = config.get("channels_config", {}) + per_channel_allow = channels_config.get(channel_name, {}).get("allow_from", []) + return bool(f"twitch:{user_id}" in allowlist or f"twitch:{user_id}" in per_channel_allow) + case _: + return False diff --git a/backend/package/yuxi/channels/adapters/twitch/setup.py b/backend/package/yuxi/channels/adapters/twitch/setup.py new file mode 100644 index 00000000..1cc72d57 --- /dev/null +++ b/backend/package/yuxi/channels/adapters/twitch/setup.py @@ -0,0 +1,400 @@ +from __future__ import annotations + +import json +import time +from dataclasses import dataclass, field +from pathlib import Path +from typing import Any + +from yuxi.utils.logging_config import logger + + +@dataclass +class SetupWizardStep: + step_id: str + title: str = "" + description: str = "" + required: bool = True + completed: bool = False + config_key: str = "" + config_value: Any = None + + +@dataclass +class SetupWizardState: + steps: list[SetupWizardStep] = field(default_factory=list) + current_step: int = 0 + + @property + def is_complete(self) -> bool: + return all(s.completed for s in self.steps if s.required) + + @property + def current(self) -> SetupWizardStep | None: + if 0 <= self.current_step < len(self.steps): + return self.steps[self.current_step] + return None + + def advance(self) -> None: + if self.current_step < len(self.steps): + self.steps[self.current_step].completed = True + self.current_step += 1 + + def to_config(self) -> dict[str, Any]: + config: dict[str, Any] = {} + for step in self.steps: + if step.config_key and step.config_value is not None: + config[step.config_key] = step.config_value + return config + + +def create_setup_wizard_steps(config: dict[str, Any] | None = None) -> list[SetupWizardStep]: + existing = config or {} + return [ + SetupWizardStep( + step_id="username", + title="Bot Username", + description="Enter your Twitch bot username (lowercase). " + "This must match the username associated with your Twitch application.", + required=True, + completed=bool(existing.get("bot_username")), + config_key="bot_username", + ), + SetupWizardStep( + step_id="token", + title="Access Token", + description="Enter your Twitch IRC access token (oauth:...) or leave blank to use " + "TWITCH_ACCESS_TOKEN / OPENCLAW_TWITCH_ACCESS_TOKEN env var. " + "Get it from https://twitchtokengenerator.com/ with scopes: " + "chat:read, chat:edit, channel:read:subscriptions, moderator:read:followers", + required=True, + completed=bool(existing.get("access_token")), + config_key="access_token", + ), + SetupWizardStep( + step_id="client_id", + title="Client ID", + description="Enter your Twitch application Client ID. " + "Create a Twitch application at https://dev.twitch.tv/console/apps", + required=True, + completed=bool(existing.get("client_id")), + config_key="client_id", + ), + SetupWizardStep( + step_id="client_secret", + title="Client Secret (Optional)", + description="Enter your Twitch application Client Secret for EventSub and token refresh. " + "Required for EventSub events (follows, subs, raids, etc.) and automatic token rotation.", + required=False, + completed=bool(existing.get("client_secret")), + config_key="client_secret", + ), + SetupWizardStep( + step_id="channels", + title="Channels", + description="Enter channels to join (comma-separated, e.g. channel1,channel2). " + "Do NOT include the # prefix — it will be added automatically.", + required=True, + completed=bool(existing.get("channels")), + config_key="channels", + ), + SetupWizardStep( + step_id="refresh_token", + title="Refresh Token (Optional)", + description="Enter a refresh token for automatic token rotation, or leave blank to skip. " + "Required scopes for token refresh: channel:read:subscriptions, chat:read, chat:edit", + required=False, + completed=bool(existing.get("refresh_token")), + config_key="refresh_token", + ), + SetupWizardStep( + step_id="dm_policy", + title="DM Policy", + description="DM/whisper policy: 'open' (allow all) or 'pairing' (require approval). Default: 'pairing'", + required=False, + completed=bool(existing.get("dm_policy")), + config_key="dm_policy", + ), + SetupWizardStep( + step_id="group_access", + title="Group Access Policy", + description="Group access policy: open / allowlist / disabled / mention_only. Default: 'open'. " + "'open' = respond to all messages; 'allowlist' = only respond to allowed users; " + "'mention_only' = only respond when @mentioned", + required=True, + completed=bool(existing.get("group_policy")), + config_key="group_policy", + ), + SetupWizardStep( + step_id="allowed_roles", + title="Allowed Roles", + description="Allowed roles for responses (comma-separated): moderator,owner,vip,subscriber,all. " + "Default: all", + required=False, + completed=bool(existing.get("allowedRoles")), + config_key="allowedRoles", + ), + SetupWizardStep( + step_id="silent", + title="Silent Mode (ACTION)", + description="Send messages as /me ACTION instead of normal PRIVMSG. " + "Type 'yes' to enable, or leave blank for default (no).", + required=False, + completed="silent" in existing, + config_key="silent", + ), + SetupWizardStep( + step_id="probe_timeout", + title="Probe Timeout (ms)", + description="Timeout for health probe API calls in milliseconds. Default: 10000 (10s). " + "Increase if your Twitch API calls are slow.", + required=False, + completed=bool(existing.get("probe_timeout_ms")), + config_key="probe_timeout_ms", + ), + ] + + +def setup_wizard_to_config(state: SetupWizardState) -> dict[str, Any]: + return state.to_config() + + +class TwitchSetupWizard: + def __init__(self, config: dict[str, Any] | None = None): + self._config = config or {} + self._state = SetupWizardState() + self._account_id: str = "" + self._disabled = False + self._build_steps() + + def _build_steps(self) -> None: + self._state.steps = create_setup_wizard_steps(self._config) + + def resolve_account_id(self, account_id: str | None = None) -> str: + self._account_id = account_id or "default" + logger.info(f"[TwitchSetup] Resolved account ID: {self._account_id}") + return self._account_id + + def prompt_username(self, username: str) -> bool: + if not username or not username.strip(): + logger.warning("[TwitchSetup] Bot username cannot be empty") + return False + self._config["bot_username"] = username.strip().lower() + self._mark_completed("username") + logger.info(f"[TwitchSetup] Bot username: {self._config['bot_username']}") + return True + + def prompt_token(self, access_token: str) -> bool: + token = access_token.strip() + if not token: + import os + + env_token = os.environ.get("TWITCH_ACCESS_TOKEN", "") or os.environ.get("OPENCLAW_TWITCH_ACCESS_TOKEN", "") + if env_token: + self._config["access_token"] = env_token + logger.info("[TwitchSetup] Using token from environment variable") + else: + logger.warning("[TwitchSetup] No access token provided and no env var found") + return False + else: + from .token_utils import ensure_oauth_prefix + + self._config["access_token"] = ensure_oauth_prefix(token) + self._mark_completed("token") + logger.info("[TwitchSetup] Access token configured") + return True + + def prompt_client_id(self, client_id: str) -> bool: + if not client_id or not client_id.strip(): + logger.warning("[TwitchSetup] Client ID cannot be empty") + return False + self._config["client_id"] = client_id.strip() + self._mark_completed("client_id") + logger.info(f"[TwitchSetup] Client ID: {self._config['client_id']}") + return True + + def prompt_client_secret(self, client_secret: str = "") -> bool: + if client_secret and client_secret.strip(): + self._config["client_secret"] = client_secret.strip() + logger.info("[TwitchSetup] Client secret configured") + else: + logger.info("[TwitchSetup] Client secret skipped (EventSub and token refresh disabled)") + self._mark_completed("client_secret") + return True + + def prompt_channels(self, channels_input: str | list[str]) -> bool: + if isinstance(channels_input, str): + channels = [c.strip().lower() for c in channels_input.split(",") if c.strip()] + else: + channels = channels_input + + if not channels: + logger.warning("[TwitchSetup] At least one channel is required") + return False + + self._config["channels"] = channels + self._mark_completed("channels") + logger.info(f"[TwitchSetup] Channels: {channels}") + return True + + def prompt_refresh_token(self, refresh_token: str = "") -> bool: + if refresh_token and refresh_token.strip(): + self._config["refresh_token"] = refresh_token.strip() + logger.info("[TwitchSetup] Refresh token configured") + else: + logger.info("[TwitchSetup] Refresh token skipped (manual rotation only)") + self._mark_completed("refresh_token") + return True + + def prompt_dm_policy(self, policy: str = "pairing") -> bool: + valid = {"open", "pairing"} + policy = policy.strip().lower() + if policy not in valid: + logger.warning(f"[TwitchSetup] Invalid DM policy '{policy}', must be one of {valid}") + return False + self._config["dm_policy"] = policy + self._mark_completed("dm_policy") + logger.info(f"[TwitchSetup] DM policy: {policy}") + return True + + def prompt_group_access(self, policy: str = "open") -> bool: + valid = {"open", "allowlist", "disabled", "mention_only"} + policy = policy.strip().lower() + if policy not in valid: + logger.warning(f"[TwitchSetup] Invalid group policy '{policy}', must be one of {valid}") + return False + self._config["group_policy"] = policy + self._mark_completed("group_access") + logger.info(f"[TwitchSetup] Group access policy: {policy}") + return True + + def prompt_allowed_roles(self, roles_input: str | list[str] = "all") -> bool: + if isinstance(roles_input, str): + roles = [r.strip().lower() for r in roles_input.split(",") if r.strip()] + else: + roles = roles_input + + valid_roles = {"moderator", "owner", "vip", "subscriber", "all"} + filtered = [r for r in roles if r in valid_roles] + if not filtered: + logger.warning("[TwitchSetup] No valid roles specified, defaulting to 'all'") + filtered = ["all"] + + self._config["allowedRoles"] = filtered + self._mark_completed("allowed_roles") + logger.info(f"[TwitchSetup] Allowed roles: {filtered}") + return True + + def prompt_silent(self, silent_input: str = "") -> bool: + silent = silent_input.strip().lower() in ("yes", "true", "1", "y", "on") + self._config["silent"] = silent + self._mark_completed("silent") + logger.info(f"[TwitchSetup] Silent mode (ACTION): {silent}") + return True + + def prompt_probe_timeout(self, timeout_input: str | int = 10000) -> bool: + if isinstance(timeout_input, str): + timeout_input = timeout_input.strip() + if not timeout_input: + self._config["probe_timeout_ms"] = 10000 + self._mark_completed("probe_timeout") + return True + try: + timeout = int(timeout_input) + except ValueError: + logger.warning(f"[TwitchSetup] Invalid probe timeout: {timeout_input}") + return False + else: + timeout = timeout_input + + if timeout < 1000: + logger.warning("[TwitchSetup] Probe timeout too small, minimum 1000ms") + return False + self._config["probe_timeout_ms"] = timeout + self._mark_completed("probe_timeout") + logger.info(f"[TwitchSetup] Probe timeout: {timeout}ms") + return True + + def finalize(self) -> dict[str, Any]: + if self._disabled: + return {"enabled": False} + validation = self._validate_config() + return { + "enabled": True, + "account_id": self._account_id or "default", + "config": dict(self._config), + "validation": validation, + "finalized_at": time.time(), + } + + def finalize_and_write(self, storage_path: str | None = None) -> dict[str, Any]: + result = self.finalize() + if storage_path: + try: + config_path = Path(storage_path) / "twitch_config.json" + config_path.parent.mkdir(parents=True, exist_ok=True) + config_path.write_text( + json.dumps(result["config"], indent=2, ensure_ascii=False), + encoding="utf-8", + ) + result["storage"] = {"path": str(config_path), "success": True} + logger.info(f"[TwitchSetup] Config written to {config_path}") + except Exception as e: + result["storage"] = {"path": str(storage_path), "success": False, "error": str(e)} + logger.error(f"[TwitchSetup] Failed to write config: {e}") + else: + result["storage"] = {"mode": "memory", "success": True} + return result + + def _validate_config(self) -> dict[str, Any]: + errors: list[str] = [] + warnings: list[str] = [] + + if not self._config.get("bot_username"): + errors.append("bot_username is required") + if not self._config.get("access_token"): + errors.append("access_token is required") + if not self._config.get("client_id"): + errors.append("client_id is required") + if not self._config.get("channels"): + errors.append("at least one channel is required") + + group_policy = self._config.get("group_policy", "open") + if group_policy not in {"open", "allowlist", "disabled", "mention_only"}: + errors.append(f"invalid group_policy: {group_policy}") + + dm_policy = self._config.get("dm_policy", "pairing") + if dm_policy not in {"open", "pairing"}: + warnings.append(f"unknown dm_policy '{dm_policy}', defaulting to 'pairing'") + + if not self._config.get("client_secret"): + warnings.append("no client_secret — EventSub and token refresh disabled") + + if not self._config.get("refresh_token"): + warnings.append("no refresh_token — manual token rotation required") + + return { + "valid": len(errors) == 0, + "errors": errors, + "warnings": warnings, + } + + def disable(self) -> dict[str, Any]: + self._disabled = True + logger.info("[TwitchSetup] Account disabled") + return {"enabled": False} + + def _mark_completed(self, step_id: str) -> None: + for step in self._state.steps: + if step.step_id == step_id: + step.completed = True + step.config_value = self._config.get(step.config_key) + break + + @property + def state(self) -> SetupWizardState: + return self._state + + @property + def config(self) -> dict[str, Any]: + return dict(self._config) diff --git a/backend/package/yuxi/channels/adapters/twitch/token_utils.py b/backend/package/yuxi/channels/adapters/twitch/token_utils.py new file mode 100644 index 00000000..6ceb90dd --- /dev/null +++ b/backend/package/yuxi/channels/adapters/twitch/token_utils.py @@ -0,0 +1,12 @@ +from __future__ import annotations + + +def ensure_oauth_prefix(token: str) -> str: + token = token.strip() + if token.startswith("oauth:"): + return token + return f"oauth:{token}" + + +def normalize_token(token: str) -> str: + return ensure_oauth_prefix(token)