From 31736aef74eef121ced1c6afdd7b3e375062b177 Mon Sep 17 00:00:00 2001 From: Kris <2893855659@qq.com> Date: Wed, 13 May 2026 16:15:39 +0800 Subject: [PATCH] =?UTF-8?q?refactor(telegram-adapter):=20=E6=95=B4?= =?UTF-8?q?=E7=90=86=E5=AF=BC=E5=85=A5=E9=A1=BA=E5=BA=8F=E5=B9=B6=E5=AE=9E?= =?UTF-8?q?=E7=8E=B0=E5=A4=9A=E8=B4=A6=E5=8F=B7=E7=AE=A1=E7=90=86=E5=A2=9E?= =?UTF-8?q?=E5=BC=BA?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 1. 整理多个文件的导入顺序，统一模块导入规范 2. 为多账号管理器添加线程安全的账号守卫和生命周期钩子 3. 新增账号增删、启停和配置变更的完整操作方法 4. 重构轮询偏移存储，使用适配器状态替代本地缓存 5. 优化BotCommandScope的参数构建逻辑 --- .../telegram/accounts/multi_account.py | 143 ++++++++++++++++++ .../channels/adapters/telegram/adapter.py | 53 ++++--- .../adapters/telegram/network_errors.py | 1 - .../adapters/telegram/security/__init__.py | 4 +- .../yuxi/channels/adapters/telegram/send.py | 3 +- .../adapters/telegram/setup/setup_wizard.py | 2 +- .../adapters/telegram/vision/sticker_cache.py | 1 + .../yuxi/channels/adapters/telegram/voice.py | 1 - 8 files changed, 180 insertions(+), 28 deletions(-) diff --git a/backend/package/yuxi/channels/adapters/telegram/accounts/multi_account.py b/backend/package/yuxi/channels/adapters/telegram/accounts/multi_account.py index 625ba9e2..517513cf 100644 --- a/backend/package/yuxi/channels/adapters/telegram/accounts/multi_account.py +++ b/backend/package/yuxi/channels/adapters/telegram/accounts/multi_account.py @@ -1,13 +1,50 @@ from __future__ import annotations +import asyncio +from collections.abc import Callable from typing import Any +AccountEventHandler = Callable[[str, dict[str, Any]], Any] + + +class AccountGuard: + def __init__(self): + self._lock = asyncio.Lock() + self._pending: dict[str, bool] = {} + + async def acquire(self, account_id: str) -> bool: + async with self._lock: + if account_id in self._pending: + return False + self._pending[account_id] = True + return True + + async def release(self, account_id: str) -> None: + async with self._lock: + self._pending.pop(account_id, None) + + @property + def active_count(self) -> int: + return len(self._pending) + + def has_pending(self, account_id: str) -> bool: + return account_id in self._pending + class MultiAccountManager: def __init__(self, config: dict[str, Any] | None = None): cfg = config or {} self._accounts: dict[str, dict[str, Any]] = {} self._primary_account_id: str | None = None + self._guard = AccountGuard() + + self._lifecycle_hooks: dict[str, list[AccountEventHandler]] = { + "on_account_added": [], + "on_account_removed": [], + "on_account_enabled": [], + "on_account_disabled": [], + "on_account_config_changed": [], + } accounts_list = cfg.get("accounts", []) if isinstance(accounts_list, list): @@ -73,3 +110,109 @@ class MultiAccountManager: def is_multi_account(self) -> bool: return len(self._accounts) > 1 + + def register_hook(self, event: str, handler: AccountEventHandler) -> None: + if event in self._lifecycle_hooks: + self._lifecycle_hooks[event].append(handler) + + async def _fire_hook(self, event: str, account_id: str, account_data: dict[str, Any]) -> None: + handlers = self._lifecycle_hooks.get(event, []) + for handler in handlers: + try: + result = handler(account_id, account_data) + if asyncio.iscoroutine(result): + await result + except Exception: + pass + + async def add_account(self, account_data: dict[str, Any]) -> bool: + account_id = account_data.get("id") + if not account_id: + return False + if not await self._guard.acquire(account_id): + return False + try: + if account_id in self._accounts: + return False + self._accounts[account_id] = { + "id": account_id, + "token": account_data.get("token") or account_data.get("bot_token", ""), + "token_file": account_data.get("token_file", ""), + "label": account_data.get("label", f"account-{account_id}"), + "api_root": account_data.get("api_root", ""), + "proxy": account_data.get("proxy", ""), + } + if self._primary_account_id is None: + self._primary_account_id = account_id + await self._fire_hook("on_account_added", account_id, self._accounts[account_id]) + return True + finally: + await self._guard.release(account_id) + + async def remove_account(self, account_id: str) -> bool: + if not await self._guard.acquire(account_id): + return False + try: + if account_id not in self._accounts: + return False + removed = self._accounts.pop(account_id) + if self._primary_account_id == account_id: + self._primary_account_id = next(iter(self._accounts), None) + await self._fire_hook("on_account_removed", account_id, removed) + return True + finally: + await self._guard.release(account_id) + + async def on_account_config_changed(self, account_id: str, new_config: dict[str, Any]) -> None: + if account_id not in self._accounts: + return + if not await self._guard.acquire(account_id): + return + try: + account = self._accounts[account_id] + old_config = dict(account) + account.update( + { + "token": new_config.get("token", account.get("token", "")), + "token_file": new_config.get("token_file", account.get("token_file", "")), + "label": new_config.get("label", account.get("label", "")), + "api_root": new_config.get("api_root", account.get("api_root", "")), + "proxy": new_config.get("proxy", account.get("proxy", "")), + } + ) + await self._fire_hook( + "on_account_config_changed", + account_id, + { + "old": old_config, + "new": dict(account), + }, + ) + finally: + await self._guard.release(account_id) + + async def enable_account(self, account_id: str) -> bool: + if not await self._guard.acquire(account_id): + return False + try: + account = self._accounts.get(account_id) + if not account: + return False + account["enabled"] = True + await self._fire_hook("on_account_enabled", account_id, account) + return True + finally: + await self._guard.release(account_id) + + async def disable_account(self, account_id: str) -> bool: + if not await self._guard.acquire(account_id): + return False + try: + account = self._accounts.get(account_id) + if not account: + return False + account["enabled"] = False + await self._fire_hook("on_account_disabled", account_id, account) + return True + finally: + await self._guard.release(account_id) diff --git a/backend/package/yuxi/channels/adapters/telegram/adapter.py b/backend/package/yuxi/channels/adapters/telegram/adapter.py index e6e29def..18ceaa97 100644 --- a/backend/package/yuxi/channels/adapters/telegram/adapter.py +++ b/backend/package/yuxi/channels/adapters/telegram/adapter.py @@ -7,15 +7,17 @@ from collections.abc import AsyncIterator from typing import Any, Literal from telegram import BotCommand, BotCommandScope, Update -from telegram.error import TelegramError, InvalidToken, RetryAfter, Forbidden, BadRequest, TimedOut, NetworkError +from telegram.error import BadRequest, Forbidden, InvalidToken, NetworkError, RetryAfter, TelegramError, TimedOut from telegram.ext import Application, ApplicationBuilder, MessageHandler, filters from yuxi.channels.base import BaseChannelAdapter -from yuxi.channels.infra.circuit_breaker import CircuitBreaker, CircuitBreakerOpenError +from yuxi.channels.capabilities import ChannelCapabilities, TTSCapabilities, TTSVoiceCapabilities from yuxi.channels.exceptions import ( ChannelAuthenticationError, ChannelNotConnectedError, ) +from yuxi.channels.infra.circuit_breaker import CircuitBreaker, CircuitBreakerOpenError +from yuxi.channels.meta import ChannelMeta from yuxi.channels.models import ( Attachment, ChannelIdentity, @@ -31,26 +33,24 @@ from yuxi.channels.models import ( MessageType, ) from yuxi.channels.registry import register_builtin_adapter -from yuxi.utils.datetime_utils import utc_now_naive -from yuxi.utils.logging_config import logger - -from .send import send_with_retry, send_stream_edit -from yuxi.channels.sdk.chunking import ChunkConfig, chunk_message as sdk_chunk_message +from yuxi.channels.sdk.chunking import ChunkConfig +from yuxi.channels.sdk.chunking import chunk_message as sdk_chunk_message from yuxi.channels.sdk.format import markdown_to_html as sdk_markdown_to_html from yuxi.channels.sdk.format import strip_html_tags as sdk_strip_html_tags -from yuxi.channels.capabilities import ChannelCapabilities, TTSVoiceCapabilities, TTSCapabilities -from yuxi.channels.meta import ChannelMeta +from yuxi.utils.datetime_utils import utc_now_naive +from yuxi.utils.logging_config import logger from .accounts.token import resolve_bot_token from .connect.allowed_updates import get_allowed_updates from .connect.polling_liveness import PollingLiveness from .connect.update_offset_store import UpdateOffsetStore +from .debounce import MessageDebouncer from .reactions.reaction_level import ReactionLevelController from .reactions.reaction_notifications import ReactionNotifications -from .tools.sendchataction_401_backoff import SendChatAction401Backoff -from .tools.request_timeouts import resolve_global_timeout -from .debounce import MessageDebouncer +from .send import send_stream_edit, send_with_retry from .thread_persistence import ThreadPersistence +from .tools.request_timeouts import resolve_global_timeout +from .tools.sendchataction_401_backoff import SendChatAction401Backoff from .topics.auto_topic_label import suggest_icon_color from .topics.topic_name_cache import TopicNameCache @@ -71,11 +71,14 @@ def _build_command_scope(scope_config: str | dict) -> BotCommandScope | None: scope_type = scope_config.get("type", "default") if scope_type == "default": return None - return BotCommandScope( - type=scope_type, - chat_id=scope_config.get("chat_id"), - user_id=scope_config.get("user_id"), - ) + kwargs = {"type": scope_type} + chat_id = scope_config.get("chat_id") + if chat_id is not None: + kwargs["chat_id"] = chat_id + user_id = scope_config.get("user_id") + if user_id is not None: + kwargs["user_id"] = user_id + return BotCommandScope(**kwargs) return None @@ -207,9 +210,10 @@ class TelegramAdapter(BaseChannelAdapter): dns_order = network_cfg.get("dnsResultOrder") if dns_order: - from telegram.request import HTTPXRequest import socket + from telegram.request import HTTPXRequest + httpx_kwargs: dict[str, Any] = {} if dns_order == "ipv4first": httpx_kwargs["socket_options"] = [(socket.AF_INET, None)] @@ -534,7 +538,7 @@ class TelegramAdapter(BaseChannelAdapter): if not self._application: return DeliveryResult(success=False, error="Application not initialized") try: - from telegram import InputMediaPhoto, InputMediaVideo, InputMediaAudio, InputMediaDocument + from telegram import InputMediaAudio, InputMediaDocument, InputMediaPhoto, InputMediaVideo input_media = [] for item in media: @@ -968,6 +972,7 @@ class TelegramAdapter(BaseChannelAdapter): chat_type = ChatType.GUILD_CHANNEL event_type = EventType.MESSAGE_RECEIVED + metadata_extra = {} if msg.new_chat_members: event_type = EventType.MEMBER_JOINED elif msg.left_chat_member: @@ -983,8 +988,6 @@ class TelegramAdapter(BaseChannelAdapter): elif msg.forum_topic_reopened: event_type = EventType.SYSTEM_EVENT metadata_extra = {"forum_topic_reopened": True} - else: - metadata_extra = {} content = msg.text or msg.caption or "" message_type = MessageType.TEXT @@ -1670,6 +1673,7 @@ class TelegramAdapter(BaseChannelAdapter): self._liveness.record_activity(update.update_id) self._offset_store.set_offset(self._token_hash, update.update_id) + await self.state_set("polling_offset", update.update_id, namespace="connection") if update.callback_query: try: @@ -1739,7 +1743,11 @@ class TelegramAdapter(BaseChannelAdapter): poll_interval = self.config.get("poll_interval", 1.0) - saved_offset = self._offset_store.get_offset(self._token_hash) + saved_offset = await self.state_get("polling_offset", namespace="connection") or 0 + if saved_offset <= 0: + saved_offset = self._offset_store.get_offset(self._token_hash) + else: + self._offset_store.set_offset(self._token_hash, saved_offset) if saved_offset > 0: logger.info(f"[Telegram] Resuming from saved offset: {saved_offset}") @@ -1781,6 +1789,7 @@ class TelegramAdapter(BaseChannelAdapter): status = self._liveness.get_status() if status["last_update_id"] > 0: self._offset_store.set_offset(self._token_hash, status["last_update_id"]) + await self.state_set("polling_offset", status["last_update_id"], namespace="connection") self._polling_task = asyncio.create_task(_poll()) asyncio.create_task(_monitor_liveness()) diff --git a/backend/package/yuxi/channels/adapters/telegram/network_errors.py b/backend/package/yuxi/channels/adapters/telegram/network_errors.py index 15d69421..c7fa3533 100644 --- a/backend/package/yuxi/channels/adapters/telegram/network_errors.py +++ b/backend/package/yuxi/channels/adapters/telegram/network_errors.py @@ -14,7 +14,6 @@ from telegram.error import ( TimedOut, ) - RECOVERABLE_ERROR_TYPES = (TimedOut, NetworkError, OSError, asyncio.TimeoutError) NON_RECOVERABLE_ERROR_TYPES = (InvalidToken, Forbidden, BadRequest, Conflict) diff --git a/backend/package/yuxi/channels/adapters/telegram/security/__init__.py b/backend/package/yuxi/channels/adapters/telegram/security/__init__.py index fdb9f18f..759d4b0f 100644 --- a/backend/package/yuxi/channels/adapters/telegram/security/__init__.py +++ b/backend/package/yuxi/channels/adapters/telegram/security/__init__.py @@ -5,9 +5,9 @@ import secrets import time from typing import Any -from .security_audit import audit_security_policy -from .group_migration import GroupMigrationHandler from .audit_membership import MembershipAuditor +from .group_migration import GroupMigrationHandler +from .security_audit import audit_security_policy class TelegramSecurityPolicy: diff --git a/backend/package/yuxi/channels/adapters/telegram/send.py b/backend/package/yuxi/channels/adapters/telegram/send.py index bb768953..0c657fc5 100644 --- a/backend/package/yuxi/channels/adapters/telegram/send.py +++ b/backend/package/yuxi/channels/adapters/telegram/send.py @@ -8,7 +8,8 @@ from telegram import Bot, ReplyParameters from telegram.error import BadRequest, NetworkError, RetryAfter, TelegramError, TimedOut from telegram.ext import Application -from yuxi.channels.sdk.format import markdown_to_html, strip_html_tags as strip_all_tags +from yuxi.channels.sdk.format import markdown_to_html +from yuxi.channels.sdk.format import strip_html_tags as strip_all_tags from yuxi.utils.logging_config import logger diff --git a/backend/package/yuxi/channels/adapters/telegram/setup/setup_wizard.py b/backend/package/yuxi/channels/adapters/telegram/setup/setup_wizard.py index 948ea343..6828ab49 100644 --- a/backend/package/yuxi/channels/adapters/telegram/setup/setup_wizard.py +++ b/backend/package/yuxi/channels/adapters/telegram/setup/setup_wizard.py @@ -2,8 +2,8 @@ from __future__ import annotations from typing import Any -from yuxi.utils.logging_config import logger from yuxi.channels.adapters.telegram.setup.setup_core import probe_token +from yuxi.utils.logging_config import logger async def run_setup_wizard(config_dir: str | None = None) -> dict[str, Any]: diff --git a/backend/package/yuxi/channels/adapters/telegram/vision/sticker_cache.py b/backend/package/yuxi/channels/adapters/telegram/vision/sticker_cache.py index 406a0141..7907866d 100644 --- a/backend/package/yuxi/channels/adapters/telegram/vision/sticker_cache.py +++ b/backend/package/yuxi/channels/adapters/telegram/vision/sticker_cache.py @@ -4,6 +4,7 @@ from collections import OrderedDict from typing import Any from yuxi.utils.logging_config import logger + from .sticker_cache_store import StickerCacheStore diff --git a/backend/package/yuxi/channels/adapters/telegram/voice.py b/backend/package/yuxi/channels/adapters/telegram/voice.py index 1e6e5f9f..d55d91cc 100644 --- a/backend/package/yuxi/channels/adapters/telegram/voice.py +++ b/backend/package/yuxi/channels/adapters/telegram/voice.py @@ -8,7 +8,6 @@ from telegram.error import TelegramError from yuxi.utils.logging_config import logger - _VOICE_COMPATIBLE_FORMATS = frozenset({"ogg", "opus", "flac", "wav", "m4a", "mp3", "aac"})