From 71fec609bbf923166cdde1bc00059367ae2b7ad5 Mon Sep 17 00:00:00 2001 From: Kris <2893855659@qq.com> Date: Tue, 12 May 2026 00:49:52 +0800 Subject: [PATCH] =?UTF-8?q?feat(telegram-adapter):=20=E5=AE=9E=E7=8E=B0?= =?UTF-8?q?=E5=AE=8C=E6=95=B4=E7=9A=84Telegram=E9=80=82=E9=85=8D=E5=99=A8?= =?UTF-8?q?=E5=9F=BA=E7=A1=80=E4=BB=A3=E7=A0=81?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 新增了Telegram适配器的全套基础模块,包括: 1. 核心适配器入口与会话工具 2. 账号管理、认证与配置系统 3. 连接相关的轮询、Webhook、更新偏移管理 4. 话题路由、管理与缓存系统 5. 消息反抖动、超时配置与工具类 6. 响应式UI与命令交互系统 7. 反应表情与通知系统 8. 审批与安全审计模块 9. 健康检查与状态监控 10. 贴纸缓存与视觉工具 11. 流式响应与协作功能 12. 群组迁移与目标归一化处理 --- .../channels/adapters/telegram/__init__.py | 3 + .../adapters/telegram/accounts/__init__.py | 0 .../telegram/accounts/account_config.py | 44 + .../telegram/accounts/account_selection.py | 21 + .../adapters/telegram/accounts/accounts.py | 98 + .../telegram/accounts/multi_account.py | 75 + .../telegram/accounts/parallel_monitor.py | 57 + .../adapters/telegram/accounts/token.py | 48 + .../channels/adapters/telegram/adapter.py | 1867 +++++++++++++++++ .../adapters/telegram/approve/__init__.py | 0 .../telegram/approve/approval_callbacks.py | 99 + .../telegram/approve/approval_config.py | 44 + .../telegram/approve/approval_handler.py | 74 + .../approve/exec_approval_forwarding.py | 122 ++ .../telegram/approve/exec_approvals.py | 55 + .../channels/adapters/telegram/chunking.py | 128 ++ .../adapters/telegram/config_ui_hints.py | 364 ++++ .../adapters/telegram/connect/__init__.py | 0 .../telegram/connect/allowed_updates.py | 46 + .../telegram/connect/polling_liveness.py | 46 + .../telegram/connect/update_offset_store.py | 38 + .../adapters/telegram/connect/webhook_cert.py | 32 + .../channels/adapters/telegram/debounce.py | 73 + .../channels/adapters/telegram/directory.py | 110 + .../adapters/telegram/directory_contract.py | 100 + .../channels/adapters/telegram/dual_plugin.py | 105 + .../yuxi/channels/adapters/telegram/format.py | 110 + .../adapters/telegram/health/__init__.py | 0 .../adapters/telegram/health/doctor.py | 50 + .../adapters/telegram/health/status_issues.py | 72 + .../adapters/telegram/inline_buttons.py | 167 ++ .../yuxi/channels/adapters/telegram/lease.py | 92 + .../adapters/telegram/message_actions.py | 367 ++++ .../adapters/telegram/network_errors.py | 79 + .../yuxi/channels/adapters/telegram/poll.py | 86 + .../yuxi/channels/adapters/telegram/probe.py | 40 + .../adapters/telegram/reaction_variants.py | 59 + .../adapters/telegram/reactions/__init__.py | 0 .../telegram/reactions/reaction_level.py | 31 + .../reactions/reaction_notifications.py | 21 + .../reactions/status_reaction_variants.py | 17 + .../adapters/telegram/security/__init__.py | 155 ++ .../telegram/security/audit_membership.py | 51 + .../telegram/security/group_migration.py | 31 + .../telegram/security/security_audit.py | 78 + .../yuxi/channels/adapters/telegram/send.py | 216 ++ .../channels/adapters/telegram/session.py | 19 + .../adapters/telegram/setup/__init__.py | 0 .../adapters/telegram/setup/setup_core.py | 51 + .../adapters/telegram/setup/setup_wizard.py | 57 + .../adapters/telegram/setup_contract.py | 59 + .../adapters/telegram/stream/__init__.py | 0 .../adapters/telegram/stream/draft_stream.py | 47 + .../adapters/telegram/stream/lane_delivery.py | 84 + .../stream/reasoning_lane_coordinator.py | 58 + .../channels/adapters/telegram/streaming.py | 275 +++ .../adapters/telegram/targets/__init__.py | 0 .../adapters/telegram/targets/normalize.py | 36 + .../telegram/targets/target_writeback.py | 23 + .../adapters/telegram/targets/targets.py | 48 + .../adapters/telegram/thread_persistence.py | 81 + .../channels/adapters/telegram/threading.py | 125 ++ .../adapters/telegram/tools/__init__.py | 0 .../telegram/tools/message_tool_schema.py | 51 + .../telegram/tools/request_timeouts.py | 41 + .../tools/sendchataction_401_backoff.py | 38 + .../adapters/telegram/tools/sequential_key.py | 17 + .../adapters/telegram/topic_routing.py | 47 + .../adapters/telegram/topics/__init__.py | 0 .../telegram/topics/auto_topic_label.py | 31 + .../adapters/telegram/topics/topic_manager.py | 35 + .../telegram/topics/topic_name_cache.py | 34 + .../channels/adapters/telegram/ui/__init__.py | 0 .../adapters/telegram/ui/command_ui.py | 47 + .../telegram/ui/interactive_dispatch.py | 27 + .../adapters/telegram/vision/__init__.py | 0 .../telegram/vision/media_understanding.py | 165 ++ .../adapters/telegram/vision/sticker_cache.py | 55 + .../telegram/vision/sticker_cache_store.py | 49 + .../telegram/vision/sticker_vision.py | 104 + .../yuxi/channels/adapters/telegram/voice.py | 128 ++ 81 files changed, 7203 insertions(+) create mode 100644 backend/package/yuxi/channels/adapters/telegram/__init__.py create mode 100644 backend/package/yuxi/channels/adapters/telegram/accounts/__init__.py create mode 100644 backend/package/yuxi/channels/adapters/telegram/accounts/account_config.py create mode 100644 backend/package/yuxi/channels/adapters/telegram/accounts/account_selection.py create mode 100644 backend/package/yuxi/channels/adapters/telegram/accounts/accounts.py create mode 100644 backend/package/yuxi/channels/adapters/telegram/accounts/multi_account.py create mode 100644 backend/package/yuxi/channels/adapters/telegram/accounts/parallel_monitor.py create mode 100644 backend/package/yuxi/channels/adapters/telegram/accounts/token.py create mode 100644 backend/package/yuxi/channels/adapters/telegram/adapter.py create mode 100644 backend/package/yuxi/channels/adapters/telegram/approve/__init__.py create mode 100644 backend/package/yuxi/channels/adapters/telegram/approve/approval_callbacks.py create mode 100644 backend/package/yuxi/channels/adapters/telegram/approve/approval_config.py create mode 100644 backend/package/yuxi/channels/adapters/telegram/approve/approval_handler.py create mode 100644 backend/package/yuxi/channels/adapters/telegram/approve/exec_approval_forwarding.py create mode 100644 backend/package/yuxi/channels/adapters/telegram/approve/exec_approvals.py create mode 100644 backend/package/yuxi/channels/adapters/telegram/chunking.py create mode 100644 backend/package/yuxi/channels/adapters/telegram/config_ui_hints.py create mode 100644 backend/package/yuxi/channels/adapters/telegram/connect/__init__.py create mode 100644 backend/package/yuxi/channels/adapters/telegram/connect/allowed_updates.py create mode 100644 backend/package/yuxi/channels/adapters/telegram/connect/polling_liveness.py create mode 100644 backend/package/yuxi/channels/adapters/telegram/connect/update_offset_store.py create mode 100644 backend/package/yuxi/channels/adapters/telegram/connect/webhook_cert.py create mode 100644 backend/package/yuxi/channels/adapters/telegram/debounce.py create mode 100644 backend/package/yuxi/channels/adapters/telegram/directory.py create mode 100644 backend/package/yuxi/channels/adapters/telegram/directory_contract.py create mode 100644 backend/package/yuxi/channels/adapters/telegram/dual_plugin.py create mode 100644 backend/package/yuxi/channels/adapters/telegram/format.py create mode 100644 backend/package/yuxi/channels/adapters/telegram/health/__init__.py create mode 100644 backend/package/yuxi/channels/adapters/telegram/health/doctor.py create mode 100644 backend/package/yuxi/channels/adapters/telegram/health/status_issues.py create mode 100644 backend/package/yuxi/channels/adapters/telegram/inline_buttons.py create mode 100644 backend/package/yuxi/channels/adapters/telegram/lease.py create mode 100644 backend/package/yuxi/channels/adapters/telegram/message_actions.py create mode 100644 backend/package/yuxi/channels/adapters/telegram/network_errors.py create mode 100644 backend/package/yuxi/channels/adapters/telegram/poll.py create mode 100644 backend/package/yuxi/channels/adapters/telegram/probe.py create mode 100644 backend/package/yuxi/channels/adapters/telegram/reaction_variants.py create mode 100644 backend/package/yuxi/channels/adapters/telegram/reactions/__init__.py create mode 100644 backend/package/yuxi/channels/adapters/telegram/reactions/reaction_level.py create mode 100644 backend/package/yuxi/channels/adapters/telegram/reactions/reaction_notifications.py create mode 100644 backend/package/yuxi/channels/adapters/telegram/reactions/status_reaction_variants.py create mode 100644 backend/package/yuxi/channels/adapters/telegram/security/__init__.py create mode 100644 backend/package/yuxi/channels/adapters/telegram/security/audit_membership.py create mode 100644 backend/package/yuxi/channels/adapters/telegram/security/group_migration.py create mode 100644 backend/package/yuxi/channels/adapters/telegram/security/security_audit.py create mode 100644 backend/package/yuxi/channels/adapters/telegram/send.py create mode 100644 backend/package/yuxi/channels/adapters/telegram/session.py create mode 100644 backend/package/yuxi/channels/adapters/telegram/setup/__init__.py create mode 100644 backend/package/yuxi/channels/adapters/telegram/setup/setup_core.py create mode 100644 backend/package/yuxi/channels/adapters/telegram/setup/setup_wizard.py create mode 100644 backend/package/yuxi/channels/adapters/telegram/setup_contract.py create mode 100644 backend/package/yuxi/channels/adapters/telegram/stream/__init__.py create mode 100644 backend/package/yuxi/channels/adapters/telegram/stream/draft_stream.py create mode 100644 backend/package/yuxi/channels/adapters/telegram/stream/lane_delivery.py create mode 100644 backend/package/yuxi/channels/adapters/telegram/stream/reasoning_lane_coordinator.py create mode 100644 backend/package/yuxi/channels/adapters/telegram/streaming.py create mode 100644 backend/package/yuxi/channels/adapters/telegram/targets/__init__.py create mode 100644 backend/package/yuxi/channels/adapters/telegram/targets/normalize.py create mode 100644 backend/package/yuxi/channels/adapters/telegram/targets/target_writeback.py create mode 100644 backend/package/yuxi/channels/adapters/telegram/targets/targets.py create mode 100644 backend/package/yuxi/channels/adapters/telegram/thread_persistence.py create mode 100644 backend/package/yuxi/channels/adapters/telegram/threading.py create mode 100644 backend/package/yuxi/channels/adapters/telegram/tools/__init__.py create mode 100644 backend/package/yuxi/channels/adapters/telegram/tools/message_tool_schema.py create mode 100644 backend/package/yuxi/channels/adapters/telegram/tools/request_timeouts.py create mode 100644 backend/package/yuxi/channels/adapters/telegram/tools/sendchataction_401_backoff.py create mode 100644 backend/package/yuxi/channels/adapters/telegram/tools/sequential_key.py create mode 100644 backend/package/yuxi/channels/adapters/telegram/topic_routing.py create mode 100644 backend/package/yuxi/channels/adapters/telegram/topics/__init__.py create mode 100644 backend/package/yuxi/channels/adapters/telegram/topics/auto_topic_label.py create mode 100644 backend/package/yuxi/channels/adapters/telegram/topics/topic_manager.py create mode 100644 backend/package/yuxi/channels/adapters/telegram/topics/topic_name_cache.py create mode 100644 backend/package/yuxi/channels/adapters/telegram/ui/__init__.py create mode 100644 backend/package/yuxi/channels/adapters/telegram/ui/command_ui.py create mode 100644 backend/package/yuxi/channels/adapters/telegram/ui/interactive_dispatch.py create mode 100644 backend/package/yuxi/channels/adapters/telegram/vision/__init__.py create mode 100644 backend/package/yuxi/channels/adapters/telegram/vision/media_understanding.py create mode 100644 backend/package/yuxi/channels/adapters/telegram/vision/sticker_cache.py create mode 100644 backend/package/yuxi/channels/adapters/telegram/vision/sticker_cache_store.py create mode 100644 backend/package/yuxi/channels/adapters/telegram/vision/sticker_vision.py create mode 100644 backend/package/yuxi/channels/adapters/telegram/voice.py diff --git a/backend/package/yuxi/channels/adapters/telegram/__init__.py b/backend/package/yuxi/channels/adapters/telegram/__init__.py new file mode 100644 index 00000000..5ce3871c --- /dev/null +++ b/backend/package/yuxi/channels/adapters/telegram/__init__.py @@ -0,0 +1,3 @@ +from yuxi.channels.adapters.telegram.adapter import TelegramAdapter + +__all__ = ["TelegramAdapter"] diff --git a/backend/package/yuxi/channels/adapters/telegram/accounts/__init__.py b/backend/package/yuxi/channels/adapters/telegram/accounts/__init__.py new file mode 100644 index 00000000..e69de29b diff --git a/backend/package/yuxi/channels/adapters/telegram/accounts/account_config.py b/backend/package/yuxi/channels/adapters/telegram/accounts/account_config.py new file mode 100644 index 00000000..b5776eb7 --- /dev/null +++ b/backend/package/yuxi/channels/adapters/telegram/accounts/account_config.py @@ -0,0 +1,44 @@ +from __future__ import annotations + +import copy +from typing import Any + + +def merge_telegram_account_config( + base_config: dict[str, Any], + account_config: dict[str, Any], +) -> dict[str, Any]: + result = copy.deepcopy(base_config) + account_only = account_config.get("accounts", {}) + if not account_only: + return result + + for account_id, per_account in account_only.items(): + if account_id not in result.get("accounts", {}): + result.setdefault("accounts", {})[account_id] = {} + result["accounts"][account_id] = _deep_merge( + result["accounts"].get(account_id, {}), + per_account, + ) + + return result + + +def _deep_merge(base: dict, override: dict) -> dict: + result = copy.deepcopy(base) + for key, value in override.items(): + if key in result and isinstance(result[key], dict) and isinstance(value, dict): + result[key] = _deep_merge(result[key], value) + else: + result[key] = copy.deepcopy(value) + return result + + +def get_default_account(config: dict[str, Any]) -> str | None: + accounts = config.get("accounts", {}) + default = config.get("default_account") + if default and default in accounts: + return default + if accounts: + return next(iter(accounts)) + return None diff --git a/backend/package/yuxi/channels/adapters/telegram/accounts/account_selection.py b/backend/package/yuxi/channels/adapters/telegram/accounts/account_selection.py new file mode 100644 index 00000000..fa5f5dcd --- /dev/null +++ b/backend/package/yuxi/channels/adapters/telegram/accounts/account_selection.py @@ -0,0 +1,21 @@ +from __future__ import annotations + +from typing import Any + + +class AccountSelector: + def __init__(self, config: dict[str, Any] | None = None): + cfg = config or {} + self._default_account = cfg.get("default_account", "default") + + def select_account(self, chat_id: str | None = None, user_id: str | None = None) -> str: + return self._default_account + + def get_default(self) -> str: + return self._default_account + + def account_for_chat(self, chat_id: str) -> str: + return self._default_account + + def list_available(self, accounts: dict[str, Any]) -> list[str]: + return list(accounts.keys()) diff --git a/backend/package/yuxi/channels/adapters/telegram/accounts/accounts.py b/backend/package/yuxi/channels/adapters/telegram/accounts/accounts.py new file mode 100644 index 00000000..8edecf14 --- /dev/null +++ b/backend/package/yuxi/channels/adapters/telegram/accounts/accounts.py @@ -0,0 +1,98 @@ +from __future__ import annotations + +from typing import Any + +from .token import resolve_bot_token + + +class MultiAccountManager: + def __init__(self, configs: dict[str, dict[str, Any]] | None = None): + self._accounts: dict[str, dict[str, Any]] = configs or {} + self._active_accounts: dict[str, Any] = {} + + def add_account(self, account_id: str, config: dict[str, Any]) -> None: + self._accounts[account_id] = config + + def remove_account(self, account_id: str) -> bool: + if account_id in self._accounts: + del self._accounts[account_id] + self._active_accounts.pop(account_id, None) + return True + return False + + def get_account_config(self, account_id: str) -> dict[str, Any] | None: + return self._accounts.get(account_id) + + def list_accounts(self) -> list[dict[str, Any]]: + return [ + { + "account_id": aid, + "active": aid in self._active_accounts, + "config": cfg, + } + for aid, cfg in self._accounts.items() + ] + + def list_enabled_accounts(self) -> list[dict[str, Any]]: + return [ + { + "account_id": aid, + "active": aid in self._active_accounts, + "config": cfg, + } + for aid, cfg in self._accounts.items() + if cfg.get("enabled", True) + ] + + def resolve_default_account(self) -> dict[str, Any] | None: + default_id = None + for aid, cfg in self._accounts.items(): + if cfg.get("default", False): + default_id = aid + break + if default_id is None and self._accounts: + default_id = next(iter(self._accounts)) + if default_id: + return { + "account_id": default_id, + "active": default_id in self._active_accounts, + "config": self._accounts[default_id], + } + return None + + def find_token_owner_account_id(self, token: str) -> str | None: + for aid, cfg in self._accounts.items(): + account_token = resolve_bot_token(cfg, aid) + if account_token and account_token == token: + return aid + return None + + def detect_token_conflicts(self) -> list[dict[str, Any]]: + token_map: dict[str, list[str]] = {} + for aid, cfg in self._accounts.items(): + token = resolve_bot_token(cfg, aid) + if token: + token_map.setdefault(token, []).append(aid) + + conflicts = [] + for token, account_ids in token_map.items(): + if len(account_ids) > 1: + conflicts.append( + { + "token_hash": token[:12] + "...", + "account_ids": account_ids, + } + ) + return conflicts + + def set_active(self, account_id: str, adapter: Any) -> None: + self._active_accounts[account_id] = adapter + + def get_active(self, account_id: str) -> Any | None: + return self._active_accounts.get(account_id) + + def get_all_active(self) -> dict[str, Any]: + return dict(self._active_accounts) + + def deactivate(self, account_id: str) -> None: + self._active_accounts.pop(account_id, None) diff --git a/backend/package/yuxi/channels/adapters/telegram/accounts/multi_account.py b/backend/package/yuxi/channels/adapters/telegram/accounts/multi_account.py new file mode 100644 index 00000000..625ba9e2 --- /dev/null +++ b/backend/package/yuxi/channels/adapters/telegram/accounts/multi_account.py @@ -0,0 +1,75 @@ +from __future__ import annotations + +from typing import Any + + +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 + + accounts_list = cfg.get("accounts", []) + if isinstance(accounts_list, list): + for account in accounts_list: + if isinstance(account, dict): + token = account.get("token") or account.get("bot_token", "") + token_file = account.get("token_file", "") + if token or token_file: + account_id = account.get("id", str(len(self._accounts) + 1)) + self._accounts[account_id] = { + "id": account_id, + "token": token, + "token_file": token_file, + "label": account.get("label", f"account-{account_id}"), + "api_root": account.get("api_root", ""), + "proxy": account.get("proxy", ""), + } + if self._primary_account_id is None: + self._primary_account_id = account_id + + single_token = cfg.get("bot_token", "") or cfg.get("token", "") + single_token_file = cfg.get("token_file", "") + if (single_token or single_token_file) and not self._accounts: + self._accounts["default"] = { + "id": "default", + "token": single_token, + "token_file": single_token_file, + "label": "default", + "api_root": cfg.get("api_root", ""), + "proxy": cfg.get("proxy", ""), + } + self._primary_account_id = "default" + + @property + def account_count(self) -> int: + return len(self._accounts) + + @property + def primary_account_id(self) -> str | None: + return self._primary_account_id + + def get_account(self, account_id: str) -> dict[str, Any] | None: + return self._accounts.get(account_id) + + def get_all_accounts(self) -> list[dict[str, Any]]: + return list(self._accounts.values()) + + def get_token_for_account(self, account_id: str) -> str: + account = self._accounts.get(account_id) + if account: + token = account.get("token", "") + if token: + return token + token_file = account.get("token_file", "") + if token_file: + try: + from pathlib import Path + + return Path(token_file).read_text().strip() + except Exception: + pass + return "" + + def is_multi_account(self) -> bool: + return len(self._accounts) > 1 diff --git a/backend/package/yuxi/channels/adapters/telegram/accounts/parallel_monitor.py b/backend/package/yuxi/channels/adapters/telegram/accounts/parallel_monitor.py new file mode 100644 index 00000000..8fd08cd7 --- /dev/null +++ b/backend/package/yuxi/channels/adapters/telegram/accounts/parallel_monitor.py @@ -0,0 +1,57 @@ +from __future__ import annotations + +import asyncio +from typing import Any + +from yuxi.utils.logging_config import logger + + +class ParallelMonitor: + def __init__(self): + self._adapters: dict[str, Any] = {} + self._tasks: dict[str, asyncio.Task] = {} + self._running = False + + def register_adapter(self, account_id: str, adapter: Any) -> None: + self._adapters[account_id] = adapter + + def unregister_adapter(self, account_id: str) -> None: + self._adapters.pop(account_id, None) + task = self._tasks.pop(account_id, None) + if task and not task.done(): + task.cancel() + + async def start_all(self) -> None: + self._running = True + for account_id, adapter in self._adapters.items(): + task = asyncio.create_task(self._monitor_adapter(account_id, adapter)) + self._tasks[account_id] = task + logger.info(f"[Telegram] Parallel monitor started for {len(self._adapters)} accounts") + + async def stop_all(self) -> None: + self._running = False + for account_id, task in list(self._tasks.items()): + if not task.done(): + task.cancel() + try: + await task + except asyncio.CancelledError: + pass + self._tasks.clear() + logger.info("[Telegram] Parallel monitor stopped") + + async def _monitor_adapter(self, account_id: str, adapter: Any) -> None: + try: + while self._running: + try: + adapter_info = getattr(adapter, "_bot_info", None) + if adapter_info is None: + logger.warning(f"[Telegram] Account {account_id} not connected") + await asyncio.sleep(30) + except asyncio.CancelledError: + raise + except Exception as e: + logger.error(f"[Telegram] Monitor error for {account_id}: {e}") + await asyncio.sleep(5) + except asyncio.CancelledError: + pass diff --git a/backend/package/yuxi/channels/adapters/telegram/accounts/token.py b/backend/package/yuxi/channels/adapters/telegram/accounts/token.py new file mode 100644 index 00000000..d2958105 --- /dev/null +++ b/backend/package/yuxi/channels/adapters/telegram/accounts/token.py @@ -0,0 +1,48 @@ +from __future__ import annotations + +import os +from typing import Any + + +def resolve_bot_token(config: dict[str, Any], account_id: str = "default") -> str | None: + accounts = config.get("accounts", {}) + account_cfg = accounts.get(account_id, {}) + + token = account_cfg.get("bot_token") + if token: + return token + + token = config.get("bot_token") + if token: + return token + + token_file = account_cfg.get("token_file") or config.get("token_file") + if token_file and os.path.isfile(token_file): + with open(token_file) as f: + return f.read().strip() + + env_token = os.getenv("TELEGRAM_BOT_TOKEN") + if env_token: + return env_token + + env_prefix = os.getenv("TELEGRAM_BOT_TOKEN_PREFIX", "") + env_suffix = account_id.upper() + env_key = f"{env_prefix}_{env_suffix}" if env_prefix else f"TELEGRAM_BOT_TOKEN_{env_suffix}" + env_token = os.getenv(env_key) + if env_token: + return env_token + + return None + + +def resolve_token_source(config: dict[str, Any], account_id: str = "default") -> str: + accounts = config.get("accounts", {}) + account_cfg = accounts.get(account_id, {}) + + if account_cfg.get("bot_token"): + return f"accounts.{account_id}.bot_token" + if config.get("bot_token"): + return "config.bot_token" + if account_cfg.get("token_file") or config.get("token_file"): + return "token_file" + return "env.TELEGRAM_BOT_TOKEN" diff --git a/backend/package/yuxi/channels/adapters/telegram/adapter.py b/backend/package/yuxi/channels/adapters/telegram/adapter.py new file mode 100644 index 00000000..2d3e63e6 --- /dev/null +++ b/backend/package/yuxi/channels/adapters/telegram/adapter.py @@ -0,0 +1,1867 @@ +from __future__ import annotations + +import asyncio +import hashlib +import time +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.ext import Application, ApplicationBuilder, MessageHandler, filters + +from yuxi.channels.base import BaseChannelAdapter +from yuxi.channels.infra.circuit_breaker import CircuitBreaker, CircuitBreakerOpenError +from yuxi.channels.exceptions import ( + ChannelAuthenticationError, + ChannelNotConnectedError, +) +from yuxi.channels.models import ( + Attachment, + ChannelIdentity, + ChannelMessage, + ChannelResponse, + ChannelStatus, + ChannelType, + ChatType, + DeliveryResult, + EventType, + HealthStatus, + MentionsInfo, + 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.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 .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 .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 .thread_persistence import ThreadPersistence +from .topics.auto_topic_label import suggest_icon_color +from .topics.topic_name_cache import TopicNameCache + + +def _build_command_scope(scope_config: str | dict) -> BotCommandScope | None: + """Build a BotCommandScope from configuration.""" + if isinstance(scope_config, str): + if scope_config == "default": + return None + if scope_config == "all_private_chats": + return BotCommandScope(type="all_private_chats") + if scope_config == "all_group_chats": + return BotCommandScope(type="all_group_chats") + if scope_config == "all_chat_administrators": + return BotCommandScope(type="all_chat_administrators") + return None + if isinstance(scope_config, dict): + 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"), + ) + return None + + +@register_builtin_adapter(aliases=["tg"]) +class TelegramAdapter(BaseChannelAdapter): + channel_id = "telegram" + channel_type = ChannelType.TELEGRAM + + text_chunk_limit = 4096 + supports_markdown = False + supports_streaming = True + streaming_modes = ["off", "partial", "block", "progress"] + max_media_size_mb = 100 + + capabilities = ChannelCapabilities( + chat_types=["direct", "group", "channel", "thread"], + polls=True, + reactions=True, + edit=True, + unsend=True, + reply=True, + threads=True, + media=True, + native_commands=True, + pin=True, + unpin=True, + group_management=True, + supports_markdown=False, + supports_streaming=True, + streaming_modes=["off", "partial", "block", "progress"], + text_chunk_limit=4096, + max_media_size_mb=100, + tts=TTSCapabilities(voice=TTSVoiceCapabilities(synthesis_target="voice-note", enabled=True)), + block_streaming=True, + ) + meta = ChannelMeta( + id="telegram", + label="Telegram", + markdown_capable=False, + ) + + def __init__(self, config: dict[str, Any] | None = None): + super().__init__(config) + self._status = ChannelStatus.DISCONNECTED + self._enabled = self.config.get("enabled", True) + self._history_limit = self.config.get("history_limit", 50) + self._application: Application | None = None + self._monitor_mode: Literal["polling", "webhook"] = "polling" + self._bot_info: dict[str, Any] | None = None + self._token_hash: str = "" + self._circuit_breaker = CircuitBreaker(failure_threshold=5, recovery_timeout=60) + self._polling_task: asyncio.Task | None = None + self._dedup_set: dict[str, float] = {} + self._dedup_ttl = 300 + cfg_size = self.config.get("max_media_size_mb") + if cfg_size is not None: + self.max_media_size_mb = int(cfg_size) + + self._liveness = PollingLiveness(max_idle_seconds=self.config.get("polling_stall_threshold_ms", 120000) / 1000) + self._offset_store = UpdateOffsetStore() + self._reaction_level_ctrl = ReactionLevelController(self.config) + self._reaction_notifications = ReactionNotifications(self.config) + self._chat_action_backoff = SendChatAction401Backoff() + self._topic_name_cache = TopicNameCache() + self._sent_message_cache: dict[str, dict[str, Any]] = {} + self._sent_message_cache_ttl = self.config.get("sent_message_cache_ttl", 3600) + + self._silent_error_replies = self.config.get("silent_error_replies", False) + self._trusted_local_file_roots: list[str] = self.config.get("trusted_local_file_roots", []) + self._config_writes = self.config.get("config_writes", True) + self._commands_native = ( + self.config.get("commands", {}).get("native", "auto") + if isinstance(self.config.get("commands"), dict) + else "auto" + ) + self._commands_native_skills = ( + self.config.get("commands", {}).get("native_skills", "auto") + if isinstance(self.config.get("commands"), dict) + else "auto" + ) + + self._debouncer = MessageDebouncer( + max_entries=self.config.get("debounce_max_entries", 1000), + ttl_seconds=self.config.get("debounce_ttl_seconds", 300), + ) + self._thread_persistence = ThreadPersistence( + storage_path=self.config.get("thread_persistence_path"), + ) + + async def connect(self) -> None: + if self._status == ChannelStatus.CONNECTED: + return + + if not self._enabled: + self._status = ChannelStatus.DISABLED + logger.info( + f"[Telegram] Channel '{self.config.get('name', self.channel_id)}' is disabled, skipping connect" + ) + return + + self._status = ChannelStatus.CONNECTING + logger.info(f"[Telegram] Starting channel '{self.config.get('name', self.channel_id)}'") + + token = resolve_bot_token(self.config) + if not token: + raise ChannelAuthenticationError() + + builder = ApplicationBuilder().token(token) + api_root = self.config.get("api_root") + if api_root: + builder.base_url(f"{api_root.rstrip('/')}/bot") + proxy = self.config.get("proxy") + if proxy: + builder.proxy_url(proxy) + + connect_timeout = resolve_global_timeout(self.config) + builder.connect_timeout(connect_timeout) + builder.read_timeout(connect_timeout) + builder.write_timeout(connect_timeout) + + network_cfg = self.config.get("network") or {} + if network_cfg: + pool_size = network_cfg.get("connectionPoolSize", 1) + builder.connection_pool_size(pool_size) + http_ver = network_cfg.get("httpVersion", "1.1") + builder.http_version(http_ver) + + dns_order = network_cfg.get("dnsResultOrder") + if dns_order: + from telegram.request import HTTPXRequest + import socket + + httpx_kwargs: dict[str, Any] = {} + if dns_order == "ipv4first": + httpx_kwargs["socket_options"] = [(socket.AF_INET, None)] + elif dns_order == "ipv6first": + httpx_kwargs["socket_options"] = [(socket.AF_INET6, None)] + + if httpx_kwargs: + request = HTTPXRequest( + connection_pool_size=pool_size, + proxy_url=proxy, + connect_timeout=connect_timeout, + read_timeout=connect_timeout, + write_timeout=connect_timeout, + http_version=http_ver, + httpx_kwargs=httpx_kwargs, + ) + builder.request(request) + builder.get_updates_request(request) + + auto_select = network_cfg.get("autoSelectFamily", True) + logger.debug(f"[Telegram] Network: autoSelectFamily={auto_select}, dnsResultOrder={dns_order}") + + self._application = builder.concurrent_updates(False).build() + + try: + bot_info = await self._application.bot.get_me() + except InvalidToken: + raise ChannelAuthenticationError() + except (NetworkError, TimedOut) as e: + raise ChannelNotConnectedError() from e + + self._bot_info = { + "id": bot_info.id, + "username": bot_info.username, + "name": f"{bot_info.first_name} {bot_info.last_name or ''}".strip(), + } + self._token_hash = hashlib.sha256(token.encode()).hexdigest()[:8] + logger.info(f"[Telegram] Bot verified: @{bot_info.username} (ID: {bot_info.id})") + + await self._register_handlers() + await self._register_commands() + + webhook_url = self.config.get("webhook_url") + if webhook_url: + await self._start_webhook() + self._monitor_mode = "webhook" + else: + await self._start_polling() + self._monitor_mode = "polling" + + self._status = ChannelStatus.CONNECTED + logger.info(f"[Telegram] Channel started in {self._monitor_mode} mode") + + async def disconnect(self) -> None: + if self._status == ChannelStatus.DISCONNECTED: + return + + logger.info(f"[Telegram] Stopping channel '{self.config.get('name', self.channel_id)}'") + + if self._polling_task and not self._polling_task.done(): + self._polling_task.cancel() + try: + await self._polling_task + except asyncio.CancelledError: + pass + self._polling_task = None + + if self._application: + if self._monitor_mode == "polling": + try: + await self._application.updater.stop() + except Exception: + pass + try: + await self._application.stop() + await self._application.shutdown() + except Exception as e: + logger.warning(f"[Telegram] Error during shutdown: {e}") + self._application = None + + self._status = ChannelStatus.DISCONNECTED + + async def send(self, response: ChannelResponse) -> DeliveryResult: + if not self._application: + return DeliveryResult(success=False, error="Application not initialized") + + chat_id = response.identity.channel_chat_id + payload = self.format_outbound(response) + + async def _do_send(): + return await send_with_retry(self._application.bot, chat_id, payload, self.config) + + try: + result = await self._circuit_breaker.call(_do_send) + if result and hasattr(result, "message_id"): + now = time.monotonic() + self._sent_message_cache[str(result.message_id)] = { + "chat_id": chat_id, + "msg_id": str(result.message_id), + "cached_at": now, + } + expired = [ + k + for k, v in self._sent_message_cache.items() + if now - v["cached_at"] > self._sent_message_cache_ttl + ] + for k in expired: + del self._sent_message_cache[k] + return result + except CircuitBreakerOpenError: + return DeliveryResult(success=False, error="Circuit breaker open") + except (TelegramError, Exception) as e: + return DeliveryResult(success=False, error=str(e)) + + async def send_media(self, chat_id: str, media_type: str, data: Any) -> DeliveryResult: + if not self._application: + return DeliveryResult(success=False, error="Application not initialized") + + max_bytes = self.max_media_size_mb * 1024 * 1024 + if isinstance(data, (bytes, bytearray)) and len(data) > max_bytes: + return DeliveryResult(success=False, error=f"Media exceeds {self.max_media_size_mb}MB limit") + + _send_map = { + "image": self._application.bot.send_photo, + "video": self._application.bot.send_video, + "audio": self._application.bot.send_audio, + "file": self._application.bot.send_document, + "animation": self._application.bot.send_animation, + "sticker": self._application.bot.send_sticker, + } + send_func = _send_map.get(media_type) + if send_func is None: + return DeliveryResult(success=False, error=f"Unsupported media type: {media_type}") + + max_retries = 3 + for attempt in range(max_retries): + try: + msg = await send_func(chat_id=chat_id, **{media_type if media_type != "file" else "document": data}) + return DeliveryResult(success=True, message_id=str(msg.message_id)) + except RetryAfter as e: + await asyncio.sleep(e.retry_after) + except Forbidden: + return DeliveryResult(success=False, error="Bot blocked by user") + except (TimedOut, NetworkError, OSError): + if attempt == max_retries - 1: + return DeliveryResult(success=False, error="Media send failed after retries") + delay = 1.0 * (2**attempt) + await asyncio.sleep(delay) + except Exception as e: + return DeliveryResult(success=False, error=str(e)) + + return DeliveryResult(success=False, error="Media send failed after retries") + + async def edit_message(self, chat_id: str, msg_id: str, content: str) -> DeliveryResult: + if not self._application: + return DeliveryResult(success=False, error="Application not initialized") + + try: + html = sdk_markdown_to_html(content) + await self._application.bot.edit_message_text( + chat_id=chat_id, + message_id=int(msg_id), + text=html, + parse_mode="HTML", + ) + return DeliveryResult(success=True, message_id=msg_id) + except BadRequest: + try: + clean_text = sdk_strip_html_tags(sdk_markdown_to_html(content)) + await self._application.bot.edit_message_text( + chat_id=chat_id, + message_id=int(msg_id), + text=clean_text, + ) + return DeliveryResult(success=True, message_id=msg_id) + except Exception as e: + return DeliveryResult(success=False, error=str(e)) + except RetryAfter as e: + return DeliveryResult(success=False, error=f"Rate limited, retry after {e.retry_after}s") + except Exception as e: + return DeliveryResult(success=False, error=str(e)) + + async def send_stream_chunk(self, chat_id: str, msg_id: str, chunk: str, finished: bool) -> DeliveryResult: + if finished: + return await self.edit_message(chat_id, msg_id, chunk) + return await send_stream_edit(self._application, chat_id, msg_id, chunk, self.config) + + async def edit_message_caption(self, chat_id: str, msg_id: str, caption: str) -> DeliveryResult: + if not self._application: + return DeliveryResult(success=False, error="Application not initialized") + try: + html = sdk_markdown_to_html(caption) + await self._application.bot.edit_message_caption( + chat_id=chat_id, + message_id=int(msg_id), + caption=html, + parse_mode="HTML", + ) + return DeliveryResult(success=True, message_id=msg_id) + except BadRequest: + try: + clean = sdk_strip_html_tags(sdk_markdown_to_html(caption)) + await self._application.bot.edit_message_caption( + chat_id=chat_id, + message_id=int(msg_id), + caption=clean, + ) + return DeliveryResult(success=True, message_id=msg_id) + except Exception as e: + return DeliveryResult(success=False, error=str(e)) + except Exception as e: + return DeliveryResult(success=False, error=str(e)) + + async def edit_message_reply_markup(self, chat_id: str, msg_id: str, reply_markup: Any = None) -> DeliveryResult: + if not self._application: + return DeliveryResult(success=False, error="Application not initialized") + try: + await self._application.bot.edit_message_reply_markup( + chat_id=chat_id, + message_id=int(msg_id), + reply_markup=reply_markup, + ) + return DeliveryResult(success=True, message_id=msg_id) + except Exception as e: + return DeliveryResult(success=False, error=str(e)) + + async def delete_message(self, chat_id: str, msg_id: str) -> DeliveryResult: + if not self._application: + return DeliveryResult(success=False, error="Application not initialized") + try: + await self._application.bot.delete_message(chat_id=chat_id, message_id=int(msg_id)) + return DeliveryResult(success=True, message_id=msg_id) + except Forbidden: + return DeliveryResult(success=False, error="Bot lacks delete permission") + except Exception as e: + return DeliveryResult(success=False, error=str(e)) + + async def send_reaction(self, chat_id: str, msg_id: str, emoji: str) -> DeliveryResult: + if not self._application: + return DeliveryResult(success=False, error="Application not initialized") + if not self._reaction_level_ctrl.should_react("message.received"): + return DeliveryResult(success=True, message_id=msg_id, metadata={"reaction_skipped": True}) + try: + await self._application.bot.set_message_reaction( + chat_id=chat_id, + message_id=int(msg_id), + reaction=[{"type": "emoji", "emoji": emoji}], + ) + return DeliveryResult(success=True, message_id=msg_id) + except BadRequest as e: + if "REACTION_INVALID" in str(e): + return DeliveryResult(success=True, message_id=msg_id, metadata={"reaction_rejected": True}) + return DeliveryResult(success=False, error=str(e)) + except Exception as e: + return DeliveryResult(success=False, error=str(e)) + + async def send_chat_action(self, chat_id: str, action: str = "typing") -> DeliveryResult: + if not self._application: + return DeliveryResult(success=False, error="Application not initialized") + success = await self._chat_action_backoff.execute(self._application.bot, chat_id, action) + if success: + return DeliveryResult(success=True) + return DeliveryResult(success=False, error="sendChatAction skipped (backoff active)") + + async def pin_message(self, chat_id: str, msg_id: str, disable_notification: bool = False) -> DeliveryResult: + if not self._application: + return DeliveryResult(success=False, error="Application not initialized") + try: + await self._application.bot.pin_chat_message( + chat_id=chat_id, + message_id=int(msg_id), + disable_notification=disable_notification, + ) + return DeliveryResult(success=True, message_id=msg_id) + except Forbidden: + return DeliveryResult(success=False, error="Bot lacks pin permission") + except Exception as e: + return DeliveryResult(success=False, error=str(e)) + + async def unpin_message(self, chat_id: str, msg_id: str) -> DeliveryResult: + if not self._application: + return DeliveryResult(success=False, error="Application not initialized") + try: + await self._application.bot.unpin_chat_message( + chat_id=chat_id, + message_id=int(msg_id), + ) + return DeliveryResult(success=True, message_id=msg_id) + except Forbidden: + return DeliveryResult(success=False, error="Bot lacks unpin permission") + except Exception as e: + return DeliveryResult(success=False, error=str(e)) + + async def unpin_all_messages(self, chat_id: str) -> DeliveryResult: + if not self._application: + return DeliveryResult(success=False, error="Application not initialized") + try: + await self._application.bot.unpin_all_chat_messages(chat_id=chat_id) + return DeliveryResult(success=True) + except Forbidden: + return DeliveryResult(success=False, error="Bot lacks unpin permission") + except Exception as e: + return DeliveryResult(success=False, error=str(e)) + + async def forward_message(self, from_chat_id: str, to_chat_id: str, msg_id: str) -> DeliveryResult: + if not self._application: + return DeliveryResult(success=False, error="Application not initialized") + try: + result = await self._application.bot.forward_message( + chat_id=to_chat_id, + from_chat_id=from_chat_id, + message_id=int(msg_id), + ) + return DeliveryResult(success=True, message_id=str(result.message_id)) + except Forbidden: + return DeliveryResult(success=False, error="Bot lacks forward permission") + except Exception as e: + return DeliveryResult(success=False, error=str(e)) + + async def send_media_group(self, chat_id: str, media: list[dict[str, Any]]) -> DeliveryResult: + if not self._application: + return DeliveryResult(success=False, error="Application not initialized") + try: + from telegram import InputMediaPhoto, InputMediaVideo, InputMediaAudio, InputMediaDocument + + input_media = [] + for item in media: + media_type = item.get("type", "photo") + media_data = item.get("media", "") + caption = item.get("caption") + kwargs = {} + if caption: + kwargs["caption"] = caption + kwargs["parse_mode"] = "HTML" + if media_type == "photo": + input_media.append(InputMediaPhoto(media=media_data, **kwargs)) + elif media_type == "video": + input_media.append(InputMediaVideo(media=media_data, **kwargs)) + elif media_type == "audio": + input_media.append(InputMediaAudio(media=media_data, **kwargs)) + elif media_type == "document": + input_media.append(InputMediaDocument(media=media_data, **kwargs)) + + if not input_media: + return DeliveryResult(success=False, error="No valid media in group") + + messages = await self._application.bot.send_media_group( + chat_id=chat_id, + media=input_media, + ) + msg_ids = [str(m.message_id) for m in messages] + return DeliveryResult(success=True, message_id=msg_ids[0], metadata={"message_ids": msg_ids}) + except Exception as e: + return DeliveryResult(success=False, error=str(e)) + + async def send_location(self, chat_id: str, latitude: float, longitude: float) -> DeliveryResult: + if not self._application: + return DeliveryResult(success=False, error="Application not initialized") + try: + msg = await self._application.bot.send_location( + chat_id=chat_id, + latitude=latitude, + longitude=longitude, + ) + return DeliveryResult(success=True, message_id=str(msg.message_id)) + except Exception as e: + return DeliveryResult(success=False, error=str(e)) + + async def send_contact( + self, chat_id: str, phone_number: str, first_name: str, last_name: str = "" + ) -> DeliveryResult: + if not self._application: + return DeliveryResult(success=False, error="Application not initialized") + try: + msg = await self._application.bot.send_contact( + chat_id=chat_id, + phone_number=phone_number, + first_name=first_name, + last_name=last_name or None, + ) + return DeliveryResult(success=True, message_id=str(msg.message_id)) + except Exception as e: + return DeliveryResult(success=False, error=str(e)) + + async def create_forum_topic( + self, chat_id: str, name: str, icon_color: int | None = None, icon_custom_emoji_id: str | None = None + ) -> DeliveryResult: + if not self._application: + return DeliveryResult(success=False, error="Application not initialized") + try: + if not name or name.strip() == "": + name = "New Topic" + if icon_color is None: + icon_color = suggest_icon_color(name) + kwargs = {"name": name, "icon_color": icon_color} + if icon_custom_emoji_id is not None: + kwargs["icon_custom_emoji_id"] = icon_custom_emoji_id + result = await self._application.bot.create_forum_topic( + chat_id=chat_id, + **kwargs, + ) + thread_id = str(result.message_thread_id) + self._topic_name_cache.set( + thread_id, + { + "name": result.name, + "icon_color": icon_color, + }, + ) + return DeliveryResult( + success=True, + message_id=None, + metadata={ + "message_thread_id": thread_id, + "name": result.name, + }, + ) + except Exception as e: + return DeliveryResult(success=False, error=str(e)) + + async def send_dice(self, chat_id: str, emoji: str = "\U0001f3b2") -> DeliveryResult: + if not self._application: + return DeliveryResult(success=False, error="Application not initialized") + try: + msg = await self._application.bot.send_dice(chat_id=chat_id, emoji=emoji) + return DeliveryResult(success=True, message_id=str(msg.message_id)) + except Exception as e: + return DeliveryResult(success=False, error=str(e)) + + async def edit_forum_topic( + self, chat_id: str, topic_id: str, name: str | None = None, icon_custom_emoji_id: str | None = None + ) -> DeliveryResult: + if not self._application: + return DeliveryResult(success=False, error="Application not initialized") + try: + kwargs = {"message_thread_id": int(topic_id)} + if name is not None: + kwargs["name"] = name + if icon_custom_emoji_id is not None: + kwargs["icon_custom_emoji_id"] = icon_custom_emoji_id + await self._application.bot.edit_forum_topic(chat_id=chat_id, **kwargs) + return DeliveryResult(success=True) + except Exception as e: + return DeliveryResult(success=False, error=str(e)) + + async def close_forum_topic(self, chat_id: str, topic_id: str) -> DeliveryResult: + if not self._application: + return DeliveryResult(success=False, error="Application not initialized") + try: + await self._application.bot.close_forum_topic( + chat_id=chat_id, + message_thread_id=int(topic_id), + ) + return DeliveryResult(success=True) + except Exception as e: + return DeliveryResult(success=False, error=str(e)) + + async def reopen_forum_topic(self, chat_id: str, topic_id: str) -> DeliveryResult: + if not self._application: + return DeliveryResult(success=False, error="Application not initialized") + try: + await self._application.bot.reopen_forum_topic( + chat_id=chat_id, + message_thread_id=int(topic_id), + ) + return DeliveryResult(success=True) + except Exception as e: + return DeliveryResult(success=False, error=str(e)) + + async def delete_forum_topic(self, chat_id: str, topic_id: str) -> DeliveryResult: + if not self._application: + return DeliveryResult(success=False, error="Application not initialized") + try: + await self._application.bot.delete_forum_topic( + chat_id=chat_id, + message_thread_id=int(topic_id), + ) + return DeliveryResult(success=True) + except Exception as e: + return DeliveryResult(success=False, error=str(e)) + + async def kick_member(self, chat_id: str, user_id: str) -> DeliveryResult: + if not self._application: + return DeliveryResult(success=False, error="Application not initialized") + try: + await self._application.bot.ban_chat_member(chat_id=chat_id, user_id=int(user_id)) + await self._application.bot.unban_chat_member(chat_id=chat_id, user_id=int(user_id)) + return DeliveryResult(success=True) + except Forbidden: + return DeliveryResult(success=False, error="Bot lacks kick permission") + except Exception as e: + return DeliveryResult(success=False, error=str(e)) + + async def ban_member(self, chat_id: str, user_id: str, until_date: int | None = None) -> DeliveryResult: + if not self._application: + return DeliveryResult(success=False, error="Application not initialized") + try: + kwargs = {} + if until_date is not None: + kwargs["until_date"] = until_date + await self._application.bot.ban_chat_member(chat_id=chat_id, user_id=int(user_id), **kwargs) + return DeliveryResult(success=True) + except Forbidden: + return DeliveryResult(success=False, error="Bot lacks ban permission") + except Exception as e: + return DeliveryResult(success=False, error=str(e)) + + async def unban_member(self, chat_id: str, user_id: str) -> DeliveryResult: + if not self._application: + return DeliveryResult(success=False, error="Application not initialized") + try: + await self._application.bot.unban_chat_member(chat_id=chat_id, user_id=int(user_id)) + return DeliveryResult(success=True) + except Forbidden: + return DeliveryResult(success=False, error="Bot lacks unban permission") + except Exception as e: + return DeliveryResult(success=False, error=str(e)) + + async def mute_member(self, chat_id: str, user_id: str) -> DeliveryResult: + if not self._application: + return DeliveryResult(success=False, error="Application not initialized") + try: + from telegram import ChatPermissions + + await self._application.bot.restrict_chat_member( + chat_id=chat_id, + user_id=int(user_id), + permissions=ChatPermissions(can_send_messages=False), + ) + return DeliveryResult(success=True) + except Forbidden: + return DeliveryResult(success=False, error="Bot lacks mute permission") + except Exception as e: + return DeliveryResult(success=False, error=str(e)) + + async def unmute_member(self, chat_id: str, user_id: str) -> DeliveryResult: + if not self._application: + return DeliveryResult(success=False, error="Application not initialized") + try: + from telegram import ChatPermissions + + await self._application.bot.restrict_chat_member( + chat_id=chat_id, + user_id=int(user_id), + permissions=ChatPermissions( + can_send_messages=True, + can_send_media_messages=True, + can_send_polls=True, + can_send_other_messages=True, + can_add_web_page_previews=True, + ), + ) + return DeliveryResult(success=True) + except Forbidden: + return DeliveryResult(success=False, error="Bot lacks unmute permission") + except Exception as e: + return DeliveryResult(success=False, error=str(e)) + + async def set_chat_permissions(self, chat_id: str, permissions: dict[str, bool]) -> DeliveryResult: + if not self._application: + return DeliveryResult(success=False, error="Application not initialized") + try: + from telegram import ChatPermissions + + await self._application.bot.set_chat_permissions( + chat_id=chat_id, + permissions=ChatPermissions(**permissions), + ) + return DeliveryResult(success=True) + except Forbidden: + return DeliveryResult(success=False, error="Bot lacks permission management") + except Exception as e: + return DeliveryResult(success=False, error=str(e)) + + async def set_chat_photo(self, chat_id: str, photo: Any) -> DeliveryResult: + if not self._application: + return DeliveryResult(success=False, error="Application not initialized") + try: + await self._application.bot.set_chat_photo(chat_id=chat_id, photo=photo) + return DeliveryResult(success=True) + except Forbidden: + return DeliveryResult(success=False, error="Bot lacks permission to set chat photo") + except BadRequest as e: + return DeliveryResult(success=False, error=f"Invalid photo: {e}") + except Exception as e: + return DeliveryResult(success=False, error=str(e)) + + async def delete_chat_photo(self, chat_id: str) -> DeliveryResult: + if not self._application: + return DeliveryResult(success=False, error="Application not initialized") + try: + await self._application.bot.delete_chat_photo(chat_id=chat_id) + return DeliveryResult(success=True) + except Forbidden: + return DeliveryResult(success=False, error="Bot lacks permission to delete chat photo") + except Exception as e: + return DeliveryResult(success=False, error=str(e)) + + async def rename_group(self, chat_id: str, title: str) -> DeliveryResult: + if not self._application: + return DeliveryResult(success=False, error="Application not initialized") + try: + await self._application.bot.set_chat_title(chat_id=chat_id, title=title) + return DeliveryResult(success=True) + except Forbidden: + return DeliveryResult(success=False, error="Bot lacks permission to rename group") + except Exception as e: + return DeliveryResult(success=False, error=str(e)) + + async def set_chat_description(self, chat_id: str, description: str) -> DeliveryResult: + if not self._application: + return DeliveryResult(success=False, error="Application not initialized") + try: + await self._application.bot.set_chat_description(chat_id=chat_id, description=description) + return DeliveryResult(success=True) + except Forbidden: + return DeliveryResult(success=False, error="Bot lacks permission to set description") + except Exception as e: + return DeliveryResult(success=False, error=str(e)) + + async def promote_admin( + self, + chat_id: str, + user_id: str, + can_change_info: bool = False, + can_post_messages: bool = False, + can_edit_messages: bool = False, + can_delete_messages: bool = False, + can_invite_users: bool = False, + can_restrict_members: bool = False, + can_pin_messages: bool = False, + can_promote_members: bool = False, + can_manage_chat: bool = False, + can_manage_video_chats: bool = False, + can_manage_topics: bool = False, + is_anonymous: bool = False, + ) -> DeliveryResult: + if not self._application: + return DeliveryResult(success=False, error="Application not initialized") + try: + from telegram import ChatAdministratorRights + + rights = ChatAdministratorRights( + is_anonymous=is_anonymous, + can_manage_chat=can_manage_chat, + can_delete_messages=can_delete_messages, + can_manage_video_chats=can_manage_video_chats, + can_restrict_members=can_restrict_members, + can_promote_members=can_promote_members, + can_change_info=can_change_info, + can_invite_users=can_invite_users, + can_post_messages=can_post_messages, + can_edit_messages=can_edit_messages, + can_pin_messages=can_pin_messages, + can_manage_topics=can_manage_topics, + ) + await self._application.bot.promote_chat_member( + chat_id=chat_id, + user_id=int(user_id), + rights=rights, + ) + return DeliveryResult(success=True) + except Forbidden: + return DeliveryResult(success=False, error="Bot lacks promote permission") + except Exception as e: + return DeliveryResult(success=False, error=str(e)) + + async def demote_admin(self, chat_id: str, user_id: str) -> DeliveryResult: + if not self._application: + return DeliveryResult(success=False, error="Application not initialized") + try: + from telegram import ChatAdministratorRights + + await self._application.bot.promote_chat_member( + chat_id=chat_id, + user_id=int(user_id), + rights=ChatAdministratorRights(), + ) + return DeliveryResult(success=True) + except Forbidden: + return DeliveryResult(success=False, error="Bot lacks demote permission") + except Exception as e: + return DeliveryResult(success=False, error=str(e)) + + async def create_chat_invite_link( + self, chat_id: str, name: str | None = None, member_limit: int | None = None + ) -> DeliveryResult: + if not self._application: + return DeliveryResult(success=False, error="Application not initialized") + try: + kwargs = {} + if name is not None: + kwargs["name"] = name + if member_limit is not None: + kwargs["member_limit"] = member_limit + result = await self._application.bot.create_chat_invite_link(chat_id=chat_id, **kwargs) + return DeliveryResult( + success=True, + metadata={"invite_link": result.invite_link, "name": result.name}, + ) + except Forbidden: + return DeliveryResult(success=False, error="Bot lacks invite link permission") + except Exception as e: + return DeliveryResult(success=False, error=str(e)) + + async def export_chat_invite_link(self, chat_id: str) -> DeliveryResult: + if not self._application: + return DeliveryResult(success=False, error="Application not initialized") + try: + invite_link = await self._application.bot.export_chat_invite_link(chat_id=chat_id) + return DeliveryResult(success=True, metadata={"invite_link": invite_link}) + except Forbidden: + return DeliveryResult(success=False, error="Bot lacks invite link permission") + except Exception as e: + return DeliveryResult(success=False, error=str(e)) + + async def leave_chat(self, chat_id: str) -> DeliveryResult: + if not self._application: + return DeliveryResult(success=False, error="Application not initialized") + try: + await self._application.bot.leave_chat(chat_id=chat_id) + return DeliveryResult(success=True) + except Forbidden: + return DeliveryResult(success=False, error="Bot cannot leave this chat") + except Exception as e: + return DeliveryResult(success=False, error=str(e)) + + async def receive(self) -> AsyncIterator[ChannelMessage]: + if False: + yield # type: ignore[misc] + + def normalize_inbound(self, raw: Update) -> ChannelMessage: + if raw.message is None: + return self._make_passthrough_event(raw) + + msg = raw.message + chat = msg.chat + user = msg.from_user + + debounce_key = self._debouncer.build_debounce_key(raw.to_dict()) + if debounce_key and self._debouncer.is_duplicate(debounce_key): + return self._make_skip_event(raw) + + chat_id = str(chat.id) + user_id = str(user.id) if user else "unknown" + msg_id = str(msg.message_id) + + if chat.type == "channel": + chat_id = self._normalize_channel_id(chat_id) + + chat_type = ChatType.DIRECT + if chat.type in ("group", "supergroup"): + chat_type = ChatType.GROUP + elif chat.type == "channel": + chat_type = ChatType.GUILD_CHANNEL + + event_type = EventType.MESSAGE_RECEIVED + if msg.new_chat_members: + event_type = EventType.MEMBER_JOINED + elif msg.left_chat_member: + event_type = EventType.MEMBER_LEFT + elif msg.pinned_message: + event_type = EventType.SYSTEM_EVENT + elif msg.forum_topic_created: + event_type = EventType.SYSTEM_EVENT + metadata_extra = {"forum_topic_created": True, "forum_topic_name": msg.forum_topic_created.name} + elif msg.forum_topic_closed: + event_type = EventType.SYSTEM_EVENT + metadata_extra = {"forum_topic_closed": True} + 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 + if msg.text and self._is_bot_command(msg): + message_type = MessageType.COMMAND + + attachments = self._extract_attachments(msg) + if not content and attachments: + content = "(Media)" + + mentions = self._extract_mentions(msg) + extracted_urls = self._extract_urls(msg) + + metadata = { + "telegram_chat_type": chat.type, + "is_topic_message": msg.is_topic_message, + } + metadata.update(metadata_extra) + if msg.message_thread_id: + metadata["thread_id"] = str(msg.message_thread_id) + cached = self._topic_name_cache.get(str(msg.message_thread_id)) + if cached: + metadata["thread_name"] = cached.get("name", "") + if chat.username: + metadata["chat_username"] = chat.username + if chat.title: + metadata["chat_title"] = chat.title + if msg.reply_to_message: + metadata["reply_to_message_id"] = str(msg.reply_to_message.message_id) + + if msg.forward_origin: + fwd = msg.forward_origin + fwd_info = {"type": fwd.type} + if getattr(fwd, "sender_user", None): + fwd_info["sender_user_id"] = str(fwd.sender_user.id) + if getattr(fwd, "sender_chat", None): + fwd_info["sender_chat_id"] = str(fwd.sender_chat.id) + if fwd.sender_chat.title: + fwd_info["sender_chat_title"] = fwd.sender_chat.title + metadata["forward_info"] = fwd_info + + return ChannelMessage( + identity=ChannelIdentity( + channel_id=self.channel_id, + channel_type=self.channel_type, + channel_user_id=user_id, + channel_chat_id=chat_id, + channel_message_id=msg_id, + ), + event_type=event_type, + message_type=message_type, + chat_type=chat_type, + content=content, + attachments=attachments, + mentions=mentions, + extracted_urls=extracted_urls, + reply_to_message_id=metadata.get("reply_to_message_id"), + metadata=metadata, + timestamp=msg.date, + ) + + @staticmethod + def _is_bot_command(msg) -> bool: + try: + from telegram import MessageEntity + except ImportError: + return bool(msg.text and msg.text.startswith("/")) + if msg.entities: + return any(e.type == MessageEntity.BOT_COMMAND for e in msg.entities) + return bool(msg.text and msg.text.startswith("/")) + + def _extract_attachments(self, msg) -> list[Attachment]: + attachments = [] + + if msg.photo: + largest = msg.photo[-1] + attachments.append( + Attachment( + type="image", + file_id=largest.file_id, + mime_type="image/jpeg", + ) + ) + if msg.document: + attachments.append( + Attachment( + type="file", + file_id=msg.document.file_id, + filename=msg.document.file_name, + mime_type=msg.document.mime_type, + size_bytes=msg.document.file_size or 0, + ) + ) + if msg.video: + attachments.append( + Attachment( + type="video", + file_id=msg.video.file_id, + mime_type=msg.video.mime_type, + ) + ) + if msg.audio: + attachments.append( + Attachment( + type="audio", + file_id=msg.audio.file_id, + mime_type=msg.audio.mime_type, + ) + ) + if msg.voice: + attachments.append( + Attachment( + type="audio", + file_id=msg.voice.file_id, + mime_type=msg.voice.mime_type, + ) + ) + if msg.sticker: + attachments.append( + Attachment( + type="sticker", + file_id=msg.sticker.file_id, + ) + ) + if msg.animation: + attachments.append( + Attachment( + type="animation", + file_id=msg.animation.file_id, + filename=msg.animation.file_name, + mime_type=msg.animation.mime_type, + size_bytes=msg.animation.file_size or 0, + ) + ) + if msg.video_note: + attachments.append( + Attachment( + type="video_note", + file_id=msg.video_note.file_id, + size_bytes=msg.video_note.file_size or 0, + ) + ) + if msg.location: + attachments.append( + Attachment( + type="location", + metadata={ + "latitude": msg.location.latitude, + "longitude": msg.location.longitude, + }, + ) + ) + if msg.contact: + attachments.append( + Attachment( + type="contact", + metadata={ + "phone_number": msg.contact.phone_number, + "first_name": msg.contact.first_name, + "last_name": msg.contact.last_name, + "user_id": str(msg.contact.user_id) if msg.contact.user_id else None, + }, + ) + ) + if msg.dice: + attachments.append( + Attachment( + type="dice", + metadata={"emoji": msg.dice.emoji, "value": msg.dice.value}, + ) + ) + if msg.poll: + attachments.append( + Attachment( + type="poll", + metadata={ + "poll_id": msg.poll.id, + "question": msg.poll.question, + "options": [opt.text for opt in msg.poll.options], + }, + ) + ) + if msg.venue: + attachments.append( + Attachment( + type="venue", + metadata={ + "title": msg.venue.title, + "address": msg.venue.address, + "latitude": msg.venue.location.latitude, + "longitude": msg.venue.location.longitude, + }, + ) + ) + if msg.media_group_id: + for att in attachments: + if att.type in ("image", "video", "audio"): + att.metadata["media_group_id"] = msg.media_group_id + + return attachments + + def _extract_mentions(self, msg) -> MentionsInfo | None: + if not msg.entities or not msg.text: + return None + + mentioned_ids: list[str] = [] + bot_username = (self._bot_info or {}).get("username", "") + + for entity in msg.entities: + if entity.type == "mention": + mention_text = msg.text[entity.offset : entity.offset + entity.length] + mentioned_ids.append(mention_text) + elif entity.type == "text_mention" and entity.user: + mentioned_ids.append(str(entity.user.id)) + + if not mentioned_ids: + return None + + bot_mentioned = f"@{bot_username}" in mentioned_ids if bot_username else False + return MentionsInfo( + mentioned_user_ids=mentioned_ids, + is_bot_mentioned=bot_mentioned, + raw_text=msg.text, + ) + + def _extract_urls(self, msg) -> list[str]: + if not msg.entities or not msg.text: + return [] + + urls: list[str] = [] + for entity in msg.entities: + if entity.type == "url": + url = msg.text[entity.offset : entity.offset + entity.length] + urls.append(url) + elif entity.type == "text_link": + urls.append(entity.url) + return urls + + def _make_passthrough_event(self, update: Update) -> ChannelMessage: + if update.edited_message: + msg = update.edited_message + chat_id = str(msg.chat.id) if msg.chat else "unknown" + user_id = str(msg.from_user.id) if msg.from_user else "unknown" + return ChannelMessage( + identity=ChannelIdentity( + channel_id=self.channel_id, + channel_type=self.channel_type, + channel_user_id=user_id, + channel_chat_id=chat_id, + ), + event_type=EventType.MESSAGE_UPDATED, + message_type=MessageType.TEXT, + chat_type=ChatType.DIRECT, + content="(passthrough)", + ) + + if update.callback_query: + cb = update.callback_query + return ChannelMessage( + identity=ChannelIdentity( + channel_id=self.channel_id, + channel_type=self.channel_type, + channel_user_id=str(cb.from_user.id), + channel_chat_id=str(cb.message.chat.id) if cb.message else "unknown", + ), + event_type=EventType.CARD_ACTION, + message_type=MessageType.TEXT, + chat_type=ChatType.DIRECT, + content=cb.data or "", + ) + + if update.my_chat_member: + mcm = update.my_chat_member + chat_id = str(mcm.chat.id) + event = ( + EventType.BOT_ADDED + if mcm.new_chat_member.status in ("member", "administrator") + else EventType.BOT_REMOVED + ) + return ChannelMessage( + identity=ChannelIdentity( + channel_id=self.channel_id, + channel_type=self.channel_type, + channel_user_id=str(mcm.from_user.id), + channel_chat_id=chat_id, + ), + event_type=event, + message_type=MessageType.TEXT, + chat_type=ChatType.GROUP if mcm.chat.type in ("group", "supergroup") else ChatType.DIRECT, + content=f"Bot {mcm.new_chat_member.status} in chat {chat_id}", + metadata={"chat_type": mcm.chat.type, "new_status": mcm.new_chat_member.status}, + ) + + if update.chat_member: + cm = update.chat_member + chat_id = str(cm.chat.id) + event = ( + EventType.MEMBER_JOINED + if cm.new_chat_member.status in ("member", "administrator") + and cm.old_chat_member.status not in ("member", "administrator") + else EventType.MEMBER_LEFT + if cm.new_chat_member.status in ("left", "kicked") + else EventType.SYSTEM_EVENT + ) + return ChannelMessage( + identity=ChannelIdentity( + channel_id=self.channel_id, + channel_type=self.channel_type, + channel_user_id=str(cm.new_chat_member.user.id), + channel_chat_id=chat_id, + ), + event_type=event, + message_type=MessageType.TEXT, + chat_type=ChatType.GROUP if cm.chat.type in ("group", "supergroup") else ChatType.DIRECT, + content=f"Member status change in chat {chat_id}", + metadata={ + "chat_type": cm.chat.type, + "old_status": cm.old_chat_member.status, + "new_status": cm.new_chat_member.status, + }, + ) + + if update.poll: + p = update.poll + return ChannelMessage( + identity=ChannelIdentity( + channel_id=self.channel_id, + channel_type=self.channel_type, + channel_user_id="unknown", + channel_chat_id="unknown", + ), + event_type=EventType.SYSTEM_EVENT, + message_type=MessageType.POLL, + chat_type=ChatType.DIRECT, + content=f"Poll {p.id}: {p.question}", + metadata={"poll_id": p.id, "is_closed": p.is_closed}, + ) + + if update.poll_answer: + pa = update.poll_answer + return ChannelMessage( + identity=ChannelIdentity( + channel_id=self.channel_id, + channel_type=self.channel_type, + channel_user_id=str(pa.user.id) if pa.user else "unknown", + channel_chat_id="unknown", + ), + event_type=EventType.SYSTEM_EVENT, + message_type=MessageType.POLL, + chat_type=ChatType.DIRECT, + content=f"Poll answer for {pa.poll_id}", + metadata={"poll_id": pa.poll_id, "option_ids": pa.option_ids}, + ) + + if update.message_reaction: + mr = update.message_reaction + bot_username = (self._bot_info or {}).get("username", "") + reactor = mr.user + if not self._reaction_notifications.should_notify( + bot_username, f"@{reactor.username}" if reactor and reactor.username else "" + ): + return ChannelMessage( + identity=ChannelIdentity( + channel_id=self.channel_id, + channel_type=self.channel_type, + channel_user_id=str(mr.user.id) if mr.user else "unknown", + channel_chat_id=str(mr.chat.id), + ), + event_type=EventType.MESSAGE_RECEIVED, + message_type=MessageType.TEXT, + chat_type=ChatType.GROUP if mr.chat.type in ("group", "supergroup") else ChatType.DIRECT, + content="(reaction update - filtered)", + metadata={"_filtered": True}, + ) + return ChannelMessage( + identity=ChannelIdentity( + channel_id=self.channel_id, + channel_type=self.channel_type, + channel_user_id=str(mr.user.id) if mr.user else "unknown", + channel_chat_id=str(mr.chat.id), + ), + event_type=EventType.SYSTEM_EVENT, + message_type=MessageType.TEXT, + chat_type=ChatType.GROUP if mr.chat.type in ("group", "supergroup") else ChatType.DIRECT, + content="(reaction update)", + metadata={ + "message_id": str(mr.message_id), + "old_reaction": [str(r) for r in (mr.old_reaction or [])], + "new_reaction": [str(r) for r in (mr.new_reaction or [])], + }, + ) + + if update.message_reaction_count: + mrc = update.message_reaction_count + return ChannelMessage( + identity=ChannelIdentity( + channel_id=self.channel_id, + channel_type=self.channel_type, + channel_user_id="unknown", + channel_chat_id=str(mrc.chat.id), + ), + event_type=EventType.SYSTEM_EVENT, + message_type=MessageType.TEXT, + chat_type=ChatType.GROUP if mrc.chat.type in ("group", "supergroup") else ChatType.DIRECT, + content="(reaction count update)", + metadata={ + "message_id": str(mrc.message_id), + "reactions": [ + {"type": str(r.type), "emoji": getattr(r, "emoji", None), "count": r.total_count} + for r in (mrc.reactions or []) + ], + }, + ) + + if update.chat_join_request: + cjr = update.chat_join_request + return ChannelMessage( + identity=ChannelIdentity( + channel_id=self.channel_id, + channel_type=self.channel_type, + channel_user_id=str(cjr.from_user.id), + channel_chat_id=str(cjr.chat.id), + ), + event_type=EventType.SYSTEM_EVENT, + message_type=MessageType.TEXT, + chat_type=ChatType.GROUP, + content=f"Join request from user {cjr.from_user.id} in chat {cjr.chat.id}", + metadata={ + "chat_type": cjr.chat.type, + "chat_title": getattr(cjr.chat, "title", None), + "user_id": str(cjr.from_user.id), + "user_name": cjr.from_user.full_name, + "bio": getattr(cjr, "bio", None), + "invite_link": getattr(cjr, "invite_link", None), + }, + ) + + return ChannelMessage( + identity=ChannelIdentity( + channel_id=self.channel_id, + channel_type=self.channel_type, + channel_user_id="unknown", + channel_chat_id="unknown", + ), + event_type=EventType.MESSAGE_RECEIVED, + message_type=MessageType.TEXT, + chat_type=ChatType.DIRECT, + content="(unknown update)", + ) + + def format_outbound(self, response: ChannelResponse) -> dict[str, Any]: + html = sdk_markdown_to_html(response.content) + chunks = sdk_chunk_message(html, ChunkConfig(limit=self.text_chunk_limit)) + main_text = chunks[0] if chunks else html + + payload: dict[str, Any] = { + "chat_id": response.identity.channel_chat_id, + "text": main_text, + "parse_mode": "HTML", + } + + thread_id = response.metadata.get("thread_id") + if thread_id: + payload["message_thread_id"] = thread_id + + reply_mode = self.config.get("reply_to_mode", "first") + if reply_mode == "off": + pass + elif reply_mode in ("first", "all", "batched"): + if response.reply_to_message_id: + payload["reply_to_message_id"] = int(response.reply_to_message_id) + if reply_mode == "all" and response.identity.channel_message_id: + payload["reply_to_message_id"] = int(response.identity.channel_message_id) + + quote_text = response.metadata.get("quote_text") or self.config.get("quote_text") + if quote_text and response.reply_to_message_id: + try: + from telegram import ReplyParameters + + payload["reply_parameters"] = ReplyParameters( + message_id=int(response.reply_to_message_id), + quote=quote_text[:256], + ) + except ImportError: + pass + + if self.config.get("silent") or response.metadata.get("silent"): + payload["disable_notification"] = True + + link_preview = self.config.get("link_preview", True) + payload["link_preview_options"] = {"is_disabled": not link_preview} + + if response.message_type == MessageType.IMAGE and response.attachments: + force_doc = self.config.get("force_document", False) + payload.pop("text", None) + if force_doc: + payload["document"] = response.attachments[0].url or response.attachments[0].file_id + else: + payload["photo"] = response.attachments[0].url or response.attachments[0].file_id + payload["caption"] = main_text + elif response.message_type == MessageType.FILE and response.attachments: + payload.pop("text", None) + payload["document"] = response.attachments[0].url or response.attachments[0].file_id + payload["caption"] = main_text + + reply_markup = self._build_reply_markup(response) + if reply_markup is not None: + payload["reply_markup"] = reply_markup + + return payload + + def _build_reply_markup(self, response: ChannelResponse) -> Any: + markup_cfg = response.metadata.get("reply_markup") + if not markup_cfg: + return None + + markup_type = markup_cfg.get("type", "keyboard") + if markup_type == "force_reply": + from telegram import ForceReply + + return ForceReply( + input_field_placeholder=markup_cfg.get("input_field_placeholder", ""), + selective=markup_cfg.get("selective", False), + ) + + if markup_type == "keyboard": + from telegram import KeyboardButton, ReplyKeyboardMarkup + + keyboard_rows = markup_cfg.get("keyboard", []) + if not keyboard_rows: + return None + + rows = [] + for row in keyboard_rows: + buttons = [] + for btn in row: + if isinstance(btn, str): + buttons.append(KeyboardButton(text=btn)) + elif isinstance(btn, dict): + kwargs: dict[str, Any] = {"text": btn["text"]} + if btn.get("request_contact"): + kwargs["request_contact"] = True + if btn.get("request_location"): + kwargs["request_location"] = True + if btn.get("request_poll"): + from telegram import KeyboardButtonPollType + + kwargs["request_poll"] = KeyboardButtonPollType(type=btn["request_poll"].get("type")) + if btn.get("web_app"): + from telegram import WebAppInfo + + kwargs["web_app"] = WebAppInfo(url=btn["web_app"]["url"]) + buttons.append(KeyboardButton(**kwargs)) + rows.append(buttons) + + return ReplyKeyboardMarkup( + keyboard=rows, + resize_keyboard=markup_cfg.get("resize_keyboard", True), + one_time_keyboard=markup_cfg.get("one_time_keyboard", False), + input_field_placeholder=markup_cfg.get("input_field_placeholder", ""), + selective=markup_cfg.get("selective", False), + is_persistent=markup_cfg.get("is_persistent", True), + ) + + return None + + async def health_check(self) -> HealthStatus: + if not self._application: + if not self._enabled: + return HealthStatus(status="degraded", metadata={"reason": "channel_disabled"}) + return HealthStatus(status="unhealthy", last_error="Application not initialized") + + try: + bot_info = await self._application.bot.get_me() + liveness_status = self._liveness.get_status() + return HealthStatus( + status="degraded" if not liveness_status["healthy"] else "healthy", + metadata={ + "bot_id": bot_info.id, + "bot_username": bot_info.username, + "monitor_mode": self._monitor_mode, + "adapter_status": self._status.value, + "liveness": liveness_status, + "backoff_active": not self._chat_action_backoff.can_send(), + }, + last_connected_at=utc_now_naive(), + ) + except InvalidToken: + return HealthStatus(status="unhealthy", last_error="Invalid bot token") + except (NetworkError, TimedOut) as e: + return HealthStatus(status="degraded", last_error=str(e)) + except Exception as e: + return HealthStatus(status="unhealthy", last_error=str(e)) + + async def get_user_info(self, channel_user_id: str) -> dict[str, Any]: + if not self._application: + return {} + try: + chat = await self._application.bot.get_chat(chat_id=int(channel_user_id)) + return { + "id": chat.id, + "username": chat.username, + "first_name": chat.first_name, + "last_name": chat.last_name, + "type": chat.type, + } + except Exception: + return {} + + async def download_media(self, file_id: str) -> bytes: + if not self._application: + raise ChannelNotConnectedError() + tg_file = await self._application.bot.get_file(file_id) + max_bytes = self.max_media_size_mb * 1024 * 1024 + if tg_file.file_size and tg_file.file_size > max_bytes: + raise ValueError(f"File size {tg_file.file_size} bytes exceeds {self.max_media_size_mb}MB limit") + return await tg_file.download_as_bytearray() + + def is_local_file_path_trusted(self, file_path: str) -> bool: + if not self._trusted_local_file_roots: + return False + from pathlib import Path + + resolved = Path(file_path).resolve() + return any(resolved.is_relative_to(Path(root).resolve()) for root in self._trusted_local_file_roots) + + async def verify_webhook_signature(self, headers: dict, body: bytes) -> bool: + secret = self.config.get("webhook_secret", "") + if not secret: + return True + token = headers.get("X-Telegram-Bot-Api-Secret-Token", "") + return token == secret + + @property + def history_limit(self) -> int: + return self._history_limit + + @property + def enabled(self) -> bool: + return self._enabled + + @property + def silent_error_replies(self) -> bool: + return self._silent_error_replies + + @property + def trusted_local_file_roots(self) -> list[str]: + return self._trusted_local_file_roots + + @property + def config_writes(self) -> bool: + return self._config_writes + + def _get_commands_native(self) -> str: + commands_cfg = self.config.get("commands", {}) + if isinstance(commands_cfg, dict): + return commands_cfg.get("native", self._commands_native) + return self._commands_native + + def _get_commands_native_skills(self) -> str: + commands_cfg = self.config.get("commands", {}) + if isinstance(commands_cfg, dict): + return commands_cfg.get("native_skills", self._commands_native_skills) + return self._commands_native_skills + + async def _register_handlers(self) -> None: + if not self._application: + return + + async def handle_update(update: Update, _context): + if self._message_handler is None: + return + + update_key = f"u:{update.update_id}" + now = time.monotonic() + if update_key in self._dedup_set: + return + self._dedup_set[update_key] = now + expired = [k for k, v in self._dedup_set.items() if now - v > self._dedup_ttl] + for k in expired: + del self._dedup_set[k] + + self._liveness.record_activity(update.update_id) + self._offset_store.set_offset(self._token_hash, update.update_id) + + if update.callback_query: + try: + await update.callback_query.answer() + except Exception: + pass + + channel_msg = self.normalize_inbound(update) + await self._message_handler(channel_msg) + + self._application.add_handler(MessageHandler(filters.ALL, handle_update)) + + async def _register_commands(self) -> None: + if not self._application: + return + + native_commands = self._get_commands_native() + if native_commands == "never": + return + + commands_config = self.config.get("commands", []) + if isinstance(commands_config, dict): + cmd_entries = commands_config.get("commands", []) + elif isinstance(commands_config, list): + cmd_entries = commands_config + else: + cmd_entries = [] + + if not cmd_entries and native_commands != "always": + return + + if isinstance(commands_config[0], (str, int)): + scoped = False + scoped_configs = cmd_entries # type: ignore[arg-type] + elif isinstance(commands_config[0], dict): + scoped = True + scoped_configs = [commands_config] if "scope" in commands_config else commands_config + else: + scoped = False + scoped_configs = cmd_entries # type: ignore[arg-type] + + for entry in scoped_configs: + if scoped: + cmd_list = entry.get("commands", []) + scope_config = entry.get("scope", "default") + else: + cmd_list = [{"command": c, "description": c} if isinstance(c, str) else c for c in commands_config] + scope_config = "default" + break + + commands = [BotCommand(c["command"], c["description"]) for c in cmd_list] + scope = _build_command_scope(scope_config) + if scope: + await self._application.bot.set_my_commands(commands, scope=scope) + else: + await self._application.bot.set_my_commands(commands) + + async def _start_polling(self) -> None: + if not self._application: + return + + allowed_updates = get_allowed_updates(self.config) + + await self._application.bot.delete_webhook(drop_pending_updates=True) + await self._application.initialize() + await self._application.start() + + poll_interval = self.config.get("poll_interval", 1.0) + + saved_offset = self._offset_store.get_offset(self._token_hash) + if saved_offset > 0: + logger.info(f"[Telegram] Resuming from saved offset: {saved_offset}") + + async def _poll(): + poll_params: dict[str, Any] = { + "poll_interval": poll_interval, + "timeout": self.config.get("poll_timeout", 10), + "allowed_updates": allowed_updates, + } + if saved_offset > 0: + poll_params["offset"] = saved_offset + 1 + + try: + await self._application.updater.start_polling(**poll_params) + except asyncio.CancelledError: + pass + except Exception as e: + logger.error(f"[Telegram] Polling error, marking channel as error: {e}") + self._status = ChannelStatus.ERROR + + async def _monitor_liveness(): + check_interval = self.config.get("polling_liveness_interval", 60) + while True: + await asyncio.sleep(check_interval) + if not self._liveness.is_healthy(): + logger.warning( + f"[Telegram] Polling liveness check failed, " + f"idle for {self._liveness.get_status()['last_activity_ago']:.1f}s" + ) + if self._status == ChannelStatus.CONNECTED: + self._status = ChannelStatus.RECONNECTING + else: + if self._status == ChannelStatus.RECONNECTING: + self._status = ChannelStatus.CONNECTED + + async def _persist_offset(): + while True: + await asyncio.sleep(30) + status = self._liveness.get_status() + if status["last_update_id"] > 0: + self._offset_store.set_offset(self._token_hash, status["last_update_id"]) + + self._polling_task = asyncio.create_task(_poll()) + asyncio.create_task(_monitor_liveness()) + asyncio.create_task(_persist_offset()) + + async def _start_webhook(self) -> None: + if not self._application: + return + + webhook_url = self.config["webhook_url"] + webhook_path = self.config.get("webhook_path", "/telegram-webhook") + webhook_secret = self.config.get("webhook_secret", "") + allowed_updates = get_allowed_updates(self.config) + + full_url = f"{webhook_url.rstrip('/')}{webhook_path}" + + webhook_params: dict[str, Any] = { + "url": full_url, + "drop_pending_updates": True, + "allowed_updates": allowed_updates, + } + if webhook_secret: + webhook_params["secret_token"] = webhook_secret + + cert_path = self.config.get("webhook_cert_path") + if cert_path: + try: + with open(cert_path, "rb") as f: + webhook_params["certificate"] = f.read() + logger.info(f"[Telegram] Using custom SSL cert from {cert_path}") + except FileNotFoundError: + logger.warning(f"[Telegram] Webhook cert file not found: {cert_path}") + except Exception as e: + logger.warning(f"[Telegram] Failed to read webhook cert: {e}") + + await self._application.bot.set_webhook(**webhook_params) + + await self._application.initialize() + await self._application.start() + + logger.info(f"[Telegram] Webhook set to {full_url}") + + async def pre_connect(self) -> dict: + token = resolve_bot_token(self.config) + if not token: + return {"status": "error", "message": "Missing bot_token"} + + try: + builder = ApplicationBuilder().token(token) + api_root = self.config.get("api_root") + if api_root: + builder.base_url(f"{api_root.rstrip('/')}/bot") + app = builder.build() + bot_info = await app.bot.get_me() + await app.shutdown() + return { + "status": "ok", + "bot_id": bot_info.id, + "bot_username": bot_info.username, + } + except InvalidToken: + return {"status": "error", "message": "Invalid bot token (401 Unauthorized)"} + except Exception as e: + return {"status": "error", "message": str(e)} + + def _normalize_channel_id(self, chat_id: str) -> str: + if chat_id.startswith("-100"): + return chat_id[4:] + return chat_id + + def _make_skip_event(self, raw: Update) -> ChannelMessage: + msg = raw.message + chat = msg.chat if msg else None + chat_id = str(chat.id) if chat else "0" + return ChannelMessage( + identity=ChannelIdentity( + channel_id=self.channel_id, + channel_type=self.channel_type, + channel_user_id="", + channel_chat_id=chat_id, + channel_message_id="0", + ), + event_type=EventType.SYSTEM_EVENT, + message_type=MessageType.TEXT, + chat_type=ChatType.DIRECT, + content="", + metadata={"skip": True}, + ) diff --git a/backend/package/yuxi/channels/adapters/telegram/approve/__init__.py b/backend/package/yuxi/channels/adapters/telegram/approve/__init__.py new file mode 100644 index 00000000..e69de29b diff --git a/backend/package/yuxi/channels/adapters/telegram/approve/approval_callbacks.py b/backend/package/yuxi/channels/adapters/telegram/approve/approval_callbacks.py new file mode 100644 index 00000000..1b996618 --- /dev/null +++ b/backend/package/yuxi/channels/adapters/telegram/approve/approval_callbacks.py @@ -0,0 +1,99 @@ +from __future__ import annotations + +import re +from typing import Any + +from yuxi.channels.models import DeliveryResult +from yuxi.utils.logging_config import logger + +_REPLY_APPROVE_PATTERN = re.compile(r"^\s*(?:/approve|approve)\s+([\w-]+)\s*$", re.IGNORECASE) +_REPLY_REJECT_PATTERN = re.compile(r"^\s*(?:/reject|reject)\s+([\w-]+)\s*$", re.IGNORECASE) + + +async def resolve_approver( + adapter, + user_id: str, + config: dict[str, Any] | None = None, +) -> bool: + cfg = config or {} + exec_cfg = cfg.get("exec_approvals", {}) + approvers = exec_cfg.get("approvers", []) + + if not approvers: + return False + + clean_user = user_id.replace("tg:", "") + is_approver = clean_user in approvers + + logger.debug(f"[Telegram] Resolve approver {clean_user}: {'yes' if is_approver else 'no'}") + return is_approver + + +async def verify_exec_reply( + adapter, + user_id: str, + reply_text: str, + config: dict[str, Any] | None = None, +) -> dict[str, Any] | None: + cfg = config or {} + exec_cfg = cfg.get("exec_approvals", {}) + approvers = exec_cfg.get("approvers", []) + + if not approvers: + return None + + clean_user = user_id.replace("tg:", "") + if clean_user not in approvers: + logger.debug(f"[Telegram] Non-approver {clean_user} attempted exec reply") + return None + + approve_match = _REPLY_APPROVE_PATTERN.match(reply_text.strip()) + if approve_match: + return { + "action": "approve", + "approval_id": approve_match.group(1), + "approved_by": user_id, + } + + reject_match = _REPLY_REJECT_PATTERN.match(reply_text.strip()) + if reject_match: + return { + "action": "reject", + "approval_id": reject_match.group(1), + "rejected_by": user_id, + } + + return None + + +async def handle_approval_callback(adapter, callback_data: str, from_user_id: str) -> DeliveryResult | None: + if not callback_data.startswith("exec:approve:") and not callback_data.startswith("exec:reject:"): + return None + + parts = callback_data.split(":", 2) + action = parts[1] + approval_id = parts[2] + + if not hasattr(adapter, "_approval_handler"): + return DeliveryResult(success=False, error="No approval handler configured") + + handler = adapter._approval_handler + + if action == "approve": + metadata = handler.approve(approval_id) + if metadata is None: + return DeliveryResult(success=False, error=f"Approval {approval_id} not found") + return DeliveryResult( + success=True, + metadata={"approval_id": approval_id, "action": "approved", "approved_by": from_user_id, **metadata}, + ) + elif action == "reject": + metadata = handler.reject(approval_id) + if metadata is None: + return DeliveryResult(success=False, error=f"Approval {approval_id} not found") + return DeliveryResult( + success=True, + metadata={"approval_id": approval_id, "action": "rejected", "rejected_by": from_user_id, **metadata}, + ) + + return DeliveryResult(success=False, error=f"Unknown approval action: {action}") diff --git a/backend/package/yuxi/channels/adapters/telegram/approve/approval_config.py b/backend/package/yuxi/channels/adapters/telegram/approve/approval_config.py new file mode 100644 index 00000000..24f06cb6 --- /dev/null +++ b/backend/package/yuxi/channels/adapters/telegram/approve/approval_config.py @@ -0,0 +1,44 @@ +from __future__ import annotations + +from typing import Any + + +class ApprovalConfig: + def __init__(self, config: dict[str, Any] | None = None): + cfg = config or {} + exec_cfg = cfg.get("exec_approvals", {}) + self.enabled = exec_cfg.get("enabled", False) + self.approvers: list[str] = exec_cfg.get("approvers", []) + self.dm_approval_only = exec_cfg.get("dm_approval_only", True) + self.require_reason = exec_cfg.get("require_reason", False) + self.forward_to_approvers = exec_cfg.get("forward_to_approvers", False) + self.suppress_notifications = exec_cfg.get("suppress_notifications", True) + + def is_enabled(self) -> bool: + return self.enabled and len(self.approvers) > 0 + + def get_approvers(self) -> list[str]: + return self.approvers + + def is_approver(self, user_id: str) -> bool: + return f"tg:{user_id}" in self.approvers or user_id in self.approvers + + def is_authorized_sender(self, user_id: str, chat_id: str | None = None) -> bool: + if self.is_approver(user_id): + return True + if chat_id and chat_id in self.approvers: + return True + return False + + def should_forward_to_approvers(self) -> bool: + return self.forward_to_approvers and self.is_enabled() + + def get_approval_settings(self) -> dict[str, Any]: + return { + "enabled": self.enabled, + "approvers": self.approvers, + "dm_approval_only": self.dm_approval_only, + "require_reason": self.require_reason, + "forward_to_approvers": self.forward_to_approvers, + "suppress_notifications": self.suppress_notifications, + } diff --git a/backend/package/yuxi/channels/adapters/telegram/approve/approval_handler.py b/backend/package/yuxi/channels/adapters/telegram/approve/approval_handler.py new file mode 100644 index 00000000..f05f60de --- /dev/null +++ b/backend/package/yuxi/channels/adapters/telegram/approve/approval_handler.py @@ -0,0 +1,74 @@ +from __future__ import annotations + +import time +from typing import Any + +from yuxi.utils.logging_config import logger + + +class ApprovalHandler: + def __init__(self, config: dict[str, Any] | None = None): + cfg = config or {} + exec_cfg = cfg.get("exec_approvals", {}) + self._approval_timeout = exec_cfg.get("timeout", 300) + self._dm_approval_only = exec_cfg.get("dm_approval_only", True) + self._pending: dict[str, dict[str, Any]] = {} + + def track_approval( + self, + approval_id: str, + metadata: dict[str, Any], + approval_mode: str = "dm", + chat_id: str | None = None, + ) -> None: + self._pending[approval_id] = { + "metadata": metadata, + "status": "pending", + "mode": approval_mode, + "chat_id": chat_id, + "created_at": time.monotonic(), + } + logger.debug(f"[Telegram] Approval tracked: {approval_id} (mode={approval_mode})") + + def approve(self, approval_id: str, approved_by: str = "") -> dict[str, Any] | None: + entry = self._pending.get(approval_id) + if not entry: + return None + entry["status"] = "approved" + entry["approved_by"] = approved_by + entry["approved_at"] = time.monotonic() + logger.info(f"[Telegram] Approval {approval_id} approved by {approved_by}") + return entry["metadata"] + + def reject(self, approval_id: str, rejected_by: str = "") -> dict[str, Any] | None: + entry = self._pending.get(approval_id) + if not entry: + return None + entry["status"] = "rejected" + entry["rejected_by"] = rejected_by + entry["rejected_at"] = time.monotonic() + logger.info(f"[Telegram] Approval {approval_id} rejected by {rejected_by}") + return entry["metadata"] + + def get_status(self, approval_id: str) -> str: + entry = self._pending.get(approval_id) + return entry["status"] if entry else "unknown" + + def get_approval_info(self, approval_id: str) -> dict[str, Any] | None: + return self._pending.get(approval_id) + + def is_dm_mode(self, approval_id: str) -> bool: + entry = self._pending.get(approval_id) + if entry: + return entry.get("mode", "dm") == "dm" + return self._dm_approval_only + + def cleanup_expired(self, ttl: int | None = None) -> int: + ttl = ttl or self._approval_timeout + now = time.monotonic() + expired = [aid for aid, entry in self._pending.items() if now - entry.get("created_at", now) > ttl] + for aid in expired: + del self._pending[aid] + if expired: + logger.debug(f"[Telegram] Cleaned {len(expired)} expired approvals") + return len(expired) diff --git a/backend/package/yuxi/channels/adapters/telegram/approve/exec_approval_forwarding.py b/backend/package/yuxi/channels/adapters/telegram/approve/exec_approval_forwarding.py new file mode 100644 index 00000000..520e8cb2 --- /dev/null +++ b/backend/package/yuxi/channels/adapters/telegram/approve/exec_approval_forwarding.py @@ -0,0 +1,122 @@ +from __future__ import annotations + +import uuid +from typing import Any + +from yuxi.utils.logging_config import logger + + +class ExecApprovalForwarder: + def __init__(self, config: dict[str, Any] | None = None): + cfg = config or {} + exec_cfg = cfg.get("exec_approvals", {}) + self._approvers: list[str] = exec_cfg.get("approvers", []) + self._suppress_notifications = exec_cfg.get("suppress_notifications", True) + + async def forward_to_approvers( + self, + adapter, + execution_request: dict[str, Any], + source_chat_id: str, + source_user_id: str, + ) -> list[str]: + if not self._approvers: + logger.warning("[Telegram] No approvers configured for forwarding") + return [] + + approval_id = execution_request.get("approval_id", str(uuid.uuid4())[:8]) + action = execution_request.get("action", "unknown") + params_preview = self._format_params_preview(execution_request.get("params", {})) + + message = self._format_approval_request( + approval_id=approval_id, + action=action, + params_preview=params_preview, + requester=source_user_id, + source_chat=source_chat_id, + ) + + sent_ids: list[str] = [] + for approver in self._approvers: + approver_clean = approver.replace("tg:", "") + try: + msg = await adapter.bot.send_message( + chat_id=approver_clean, + text=message, + disable_notification=self._suppress_notifications, + ) + sent_ids.append(str(msg.message_id)) + logger.info(f"[Telegram] Approval {approval_id} forwarded to {approver}") + except Exception as e: + logger.error(f"[Telegram] Failed to forward approval to {approver}: {e}") + + return sent_ids + + async def send_approval_result( + self, + adapter, + approval_id: str, + result: str, + details: dict[str, Any] | None = None, + ) -> None: + result_emoji = {"approved": "\u2705", "rejected": "\u274c", "expired": "\u23f3"}.get(result, "\u2139\ufe0f") + message = f"{result_emoji} Approval *{approval_id}*: _{result.capitalize()}_" + + if details: + message += f"\n\n{self._format_params_preview(details)}" + + for approver in self._approvers: + approver_clean = approver.replace("tg:", "") + try: + await adapter.bot.send_message( + chat_id=approver_clean, + text=message, + parse_mode="Markdown", + disable_notification=True, + ) + except Exception: + pass + + def _format_approval_request( + self, + approval_id: str, + action: str, + params_preview: str, + requester: str, + source_chat: str, + ) -> str: + return ( + f"\u26a0\ufe0f *Exec Approval Required*\n\n" + f"*ID:* `{approval_id}`\n" + f"*Action:* {action}\n" + f"*Requester:* `{requester}`\n" + f"*Source Chat:* `{source_chat}`\n\n" + f"*Params:* {params_preview}\n\n" + f"_Reply /approve {approval_id} or /reject {approval_id}_" + ) + + def _format_params_preview(self, params: dict[str, Any]) -> str: + if not params: + return "_none_" + items = [] + for k, v in params.items(): + v_str = str(v) + if len(v_str) > 80: + v_str = v_str[:77] + "..." + items.append(f" `{k}`: {v_str}") + return "\n".join(items) + + +class ExecApprovalApprovers: + @staticmethod + def normalize_approvers(raw: list[str]) -> list[str]: + return [a.strip().replace("tg:", "") for a in raw if a.strip()] + + @staticmethod + def is_approver(user_id: str, approvers: list[str]) -> bool: + clean_user = user_id.replace("tg:", "") + return clean_user in approvers + + @staticmethod + def resolve_approver_chat_ids(approvers: list[str]) -> list[str]: + return [a.replace("tg:", "") for a in approvers] diff --git a/backend/package/yuxi/channels/adapters/telegram/approve/exec_approvals.py b/backend/package/yuxi/channels/adapters/telegram/approve/exec_approvals.py new file mode 100644 index 00000000..e55e5e02 --- /dev/null +++ b/backend/package/yuxi/channels/adapters/telegram/approve/exec_approvals.py @@ -0,0 +1,55 @@ +from __future__ import annotations + +from typing import Any + +from yuxi.channels.models import DeliveryResult + + +async def exec_approvals( + adapter, + chat_id: str, + command: str, + params: dict[str, Any] | None = None, +) -> DeliveryResult: + params = params or {} + approvers = adapter.config.get("exec_approvals", {}).get("approvers", []) + enabled = adapter.config.get("exec_approvals", {}).get("enabled", False) + + if not enabled: + return DeliveryResult(success=False, error="Exec approvals are not enabled") + + if not approvers: + return DeliveryResult(success=False, error="No approvers configured") + + approval_id = f"exec_{chat_id}_{command}" + + from yuxi.channels.adapters.telegram.inline_buttons import build_exec_approval_buttons + + buttons = build_exec_approval_buttons(approval_id) + + text = ( + f"\u26a0\ufe0f **Exec Approval Required**\n\n" + f"Command: `{command}`\n" + f"Requested by: {params.get('user_id', 'unknown')}\n\n" + ) + if params: + text += f"Params: {params}\n" + + if not adapter._application: + return DeliveryResult(success=False, error="Application not initialized") + + for approver in approvers: + try: + await adapter._application.bot.send_message( + chat_id=approver, + text=text, + parse_mode="HTML", + reply_markup=build_exec_approval_buttons(approval_id), + ) + except Exception: + continue + + return DeliveryResult( + success=True, + metadata={"approval_id": approval_id, "approvers": approvers}, + ) diff --git a/backend/package/yuxi/channels/adapters/telegram/chunking.py b/backend/package/yuxi/channels/adapters/telegram/chunking.py new file mode 100644 index 00000000..1021bf76 --- /dev/null +++ b/backend/package/yuxi/channels/adapters/telegram/chunking.py @@ -0,0 +1,128 @@ +from __future__ import annotations + +import re +from typing import Any, Literal + +ChunkMode = Literal["length", "newline", "paragraph"] + +TELEGRAM_MAX_TEXT_LENGTH = 4096 +_SENTENCE_BOUNDARY = re.compile(r"[。!?.!?\n]") + + +def _chunk_by_length(text: str, limit: int) -> list[str]: + chunks: list[str] = [] + start = 0 + while start < len(text): + end = min(start + limit, len(text)) + if end < len(text): + split_at = _find_split_point(text, limit, start, end) + chunks.append(text[start:split_at].strip()) + start = split_at + else: + chunks.append(text[start:end].strip()) + start = end + return chunks or [text] + + +def _chunk_by_newline(text: str, limit: int) -> list[str]: + chunks: list[str] = [] + lines = text.split("\n") + current = "" + for line in lines: + if len(line) > limit: + if current: + chunks.append(current) + sub_chunks = _chunk_by_length(line, limit) + if sub_chunks: + current = sub_chunks.pop() + chunks.extend(sub_chunks) + else: + current = "" + elif len(current) + len(line) + (1 if current else 0) <= limit: + current = f"{current}\n{line}" if current else line + else: + chunks.append(current) + current = line + if current: + chunks.append(current) + return chunks or [text] + + +def _chunk_by_paragraph(text: str, limit: int) -> list[str]: + if len(text) <= limit: + return [text] + + chunks: list[str] = [] + paragraphs = text.split("\n\n") + current = "" + + for para in paragraphs: + if len(current) + len(para) + 2 <= limit: + current = f"{current}\n\n{para}" if current else para + else: + if current: + chunks.append(current) + if len(para) > limit: + sub_chunks = _chunk_long_paragraph(para, limit) + if sub_chunks: + if sub_chunks[-1]: + current = sub_chunks.pop() + else: + sub_chunks.pop() + current = "" + chunks.extend(sub_chunks) + else: + current = "" + else: + current = para + + if current: + chunks.append(current) + return chunks or [text] + + +def chunk_text( + text: str, + limit: int = TELEGRAM_MAX_TEXT_LENGTH, + mode: ChunkMode = "paragraph", +) -> list[str]: + modes: dict[ChunkMode, callable] = { + "length": _chunk_by_length, + "newline": _chunk_by_newline, + "paragraph": _chunk_by_paragraph, + } + chunker = modes.get(mode, _chunk_by_paragraph) + return chunker(text, limit) + + +def chunk_mode_from_config(config: dict[str, Any] | None = None) -> ChunkMode: + cfg = config or {} + streaming_cfg = cfg.get("streaming", {}) + if isinstance(streaming_cfg, dict): + return streaming_cfg.get("chunk_mode", "paragraph") + return cfg.get("chunking_mode", "paragraph") + + +def _chunk_long_paragraph(text: str, limit: int) -> list[str]: + chunks: list[str] = [] + while len(text) > limit: + split_at = _find_split_point(text, limit, 0, limit) + chunks.append(text[:split_at].rstrip()) + text = text[split_at:].lstrip() + chunks.append(text) + return chunks + + +def _find_split_point(text: str, limit: int, lo: int = 0, hi: int | None = None) -> int: + look_start = max(lo, lo or limit // 2) + look_end = hi or limit + candidates = [m.start() for m in _SENTENCE_BOUNDARY.finditer(text, look_start, look_end)] + if candidates: + return candidates[-1] + 1 + newline = text.rfind("\n", look_start, look_end) + if newline != -1: + return newline + 1 + space = text.rfind(" ", look_start, look_end) + if space != -1: + return space + 1 + return look_end diff --git a/backend/package/yuxi/channels/adapters/telegram/config_ui_hints.py b/backend/package/yuxi/channels/adapters/telegram/config_ui_hints.py new file mode 100644 index 00000000..1f5a149b --- /dev/null +++ b/backend/package/yuxi/channels/adapters/telegram/config_ui_hints.py @@ -0,0 +1,364 @@ +from __future__ import annotations + +from typing import Any + + +def get_config_ui_hints() -> dict[str, Any]: + return { + "bot_token": { + "label": "Bot Token", + "description": "Telegram Bot Token from @BotFather", + "type": "password", + "sensitive": True, + "placeholder": "123456:ABC-DEF1234ghIkl-zyx57W2v1u123ew11", + "required": True, + }, + "token_file": { + "label": "Token File", + "description": "Path to a file containing the bot token (alternative to direct token config)", + "type": "text", + "placeholder": "/run/secrets/telegram_token", + }, + "api_root": { + "label": "API Root", + "description": "Custom Telegram Bot API server URL (for self-hosted Bot API)", + "type": "text", + "placeholder": "https://api.telegram.org", + }, + "proxy": { + "label": "Proxy URL", + "description": "HTTP/SOCKS5 proxy URL for Telegram API connections", + "type": "text", + "placeholder": "socks5://127.0.0.1:1080", + }, + "webhook_url": { + "label": "Webhook URL", + "description": "Public HTTPS URL for Telegram webhook (set to empty to use polling)", + "type": "text", + "placeholder": "https://your-domain.com/telegram/webhook", + }, + "webhook_secret": { + "label": "Webhook Secret", + "description": "Secret token to validate incoming webhook requests (X-Telegram-Bot-Api-Secret-Token)", + "type": "password", + "sensitive": True, + }, + "webhook_path": { + "label": "Webhook Path", + "description": "Custom webhook URL path segment", + "type": "text", + "default": "/telegram/webhook", + }, + "webhook_cert_path": { + "label": "Webhook SSL Certificate", + "description": "Path to a custom SSL certificate for webhook (for self-signed certs)", + "type": "text", + }, + "poll_interval": { + "label": "Poll Interval (seconds)", + "description": "Interval between polling requests in seconds", + "type": "number", + "default": 1.0, + }, + "poll_timeout": { + "label": "Poll Timeout (seconds)", + "description": "Long polling timeout in seconds", + "type": "number", + "default": 10, + }, + "polling_stall_threshold_ms": { + "label": "Polling Stall Threshold (ms)", + "description": "Maximum idle time in milliseconds before polling is considered stalled", + "type": "number", + "default": 120000, + }, + "timeout_seconds": { + "label": "API Timeout (seconds)", + "description": "Global timeout for all Telegram API requests", + "type": "number", + "default": 500, + }, + "dm_policy": { + "label": "DM Policy", + "description": "How to handle direct messages from users", + "type": "select", + "options": [ + {"value": "open", "label": "Open (anyone can DM)"}, + {"value": "pairing", "label": "Pairing (challenge code required)"}, + {"value": "allowlist", "label": "Allowlist (only listed users)"}, + {"value": "disabled", "label": "Disabled (reject all DMs)"}, + ], + "default": "pairing", + }, + "group_policy": { + "label": "Group Policy", + "description": "How to handle messages in group chats", + "type": "select", + "options": [ + {"value": "open", "label": "Open (respond to everyone)"}, + {"value": "allowlist", "label": "Allowlist (only listed users)"}, + {"value": "disabled", "label": "Disabled (ignore group messages)"}, + ], + "default": "allowlist", + }, + "allow_from": { + "label": "DM Allowlist", + "description": "Users allowed to DM the bot (user IDs, one per line)", + "type": "textarea", + }, + "group_allow_from": { + "label": "Group Allowlist", + "description": "Users allowed to trigger the bot in groups", + "type": "textarea", + }, + "pairing_code": { + "label": "Pairing Code", + "description": "Shared secret for pairing new users (leave empty to auto-generate)", + "type": "password", + "sensitive": True, + }, + "commands": { + "label": "Custom Commands", + "description": "Custom bot commands to register with Telegram (JSON array or scoped object)", + "type": "json", + }, + "commands.native": { + "label": "Native Commands", + "description": "Auto-register native bot commands with Telegram", + "type": "select", + "options": [ + {"value": "auto", "label": "Auto (register when configured)"}, + {"value": "always", "label": "Always (force register)"}, + {"value": "never", "label": "Never (do not register)"}, + ], + "default": "auto", + }, + "commands.nativeSkills": { + "label": "Native Skill Commands", + "description": "Auto-register native skill commands with Telegram", + "type": "select", + "options": [ + {"value": "auto", "label": "Auto"}, + {"value": "always", "label": "Always"}, + {"value": "never", "label": "Never"}, + ], + "default": "auto", + }, + "streaming": { + "label": "Streaming Mode", + "description": "How to deliver streaming responses", + "type": "select", + "options": [ + {"value": "off", "label": "Off (send when complete)"}, + {"value": "partial", "label": "Partial (edit message incrementally)"}, + {"value": "block", "label": "Block (coalesce chunks)"}, + {"value": "progress", "label": "Progress (show dots animation)"}, + ], + "default": "partial", + }, + "streaming.chunk_mode": { + "label": "Chunk Mode", + "description": "How to split long messages into chunks", + "type": "select", + "options": [ + {"value": "paragraph", "label": "Paragraph (split on double newline)"}, + {"value": "newline", "label": "Newline (split on line breaks)"}, + {"value": "length", "label": "Length (split at character limit)"}, + ], + "default": "paragraph", + }, + "streaming.block.enabled": { + "label": "Block Streaming Enabled", + "description": "Enable block-mode streaming with chunked preview delivery", + "type": "toggle", + "default": True, + }, + "streaming.block.coalesce": { + "label": "Block Coalesce", + "description": "Coalesce streamed block chunks and send a final delivery after completion", + "type": "toggle", + "default": False, + }, + "streaming.block.coalesce_delay_ms": { + "label": "Block Coalesce Delay (ms)", + "description": "Delay in milliseconds before flushing coalesced chunks", + "type": "number", + "default": 0, + }, + "streaming.preview.chunk.minChars": { + "label": "Preview Chunk Min Characters", + "description": "Minimum characters per preview chunk before sending", + "type": "number", + "default": 50, + }, + "streaming.preview.chunk.maxChars": { + "label": "Preview Chunk Max Characters", + "description": "Target maximum characters per preview chunk", + "type": "number", + "default": 200, + }, + "streaming.preview.chunk.breakPreference": { + "label": "Preview Chunk Break Preference", + "description": "Preferred character boundary for chunk splits", + "type": "select", + "options": [ + {"value": "newline", "label": "Newline"}, + {"value": "sentence", "label": "Sentence"}, + {"value": "word", "label": "Word"}, + {"value": "any", "label": "Any"}, + ], + "default": "newline", + }, + "streaming.preview.toolProgress": { + "label": "Show Tool Progress", + "description": "Display tool/progress activity in real-time preview messages", + "type": "toggle", + "default": True, + }, + "stream_edit_interval_ms": { + "label": "Stream Edit Interval (ms)", + "description": "Minimum interval between stream message edits in milliseconds", + "type": "number", + "default": 500, + }, + "streaming.lane_enabled": { + "label": "Lane Streaming", + "description": "Enable lane-based streaming with reasoning/main separation", + "type": "toggle", + "default": False, + }, + "reply_to_mode": { + "label": "Reply Mode", + "description": "When to reply to incoming messages", + "type": "select", + "options": [ + {"value": "off", "label": "Off (never reply)"}, + {"value": "first", "label": "First (reply to first message only)"}, + {"value": "all", "label": "All (reply to every message)"}, + {"value": "batched", "label": "Batched (reply to last in group)"}, + ], + "default": "first", + }, + "silent": { + "label": "Silent Mode", + "description": "Send all messages without notification sound", + "type": "toggle", + "default": False, + }, + "silent_error_replies": { + "label": "Silent Error Replies", + "description": "Send error replies without notification sound", + "type": "toggle", + "default": False, + }, + "link_preview": { + "label": "Link Preview", + "description": "Show link previews in messages", + "type": "toggle", + "default": True, + }, + "force_document": { + "label": "Force Document", + "description": "Send images as documents instead of photos", + "type": "toggle", + "default": False, + }, + "config_writes": { + "label": "Allow Config Writes", + "description": "Allow the Telegram channel to write configuration changes via events/commands", + "type": "toggle", + "default": True, + }, + "retry.attempts": { + "label": "Retry Attempts", + "description": "Maximum number of send retries", + "type": "number", + "default": 3, + }, + "retry.min_delay_ms": { + "label": "Retry Min Delay (ms)", + "description": "Minimum delay between retries in milliseconds", + "type": "number", + "default": 400, + }, + "retry.max_delay_ms": { + "label": "Retry Max Delay (ms)", + "description": "Maximum delay between retries in milliseconds", + "type": "number", + "default": 30000, + }, + "retry.jitter": { + "label": "Retry Jitter", + "description": "Random jitter factor for retry delays (0.0 to 1.0)", + "type": "number", + "default": 0.1, + }, + "max_media_size_mb": { + "label": "Max Media Size (MB)", + "description": "Maximum allowed media file size in megabytes", + "type": "number", + "default": 100, + }, + "text_chunk_limit": { + "label": "Text Chunk Limit", + "description": "Maximum characters per message chunk", + "type": "number", + "default": 4096, + }, + "history_limit": { + "label": "History Limit", + "description": "Number of recent messages to fetch as context", + "type": "number", + "default": 50, + }, + "trusted_local_file_roots": { + "label": "Trusted Local File Roots", + "description": "Allowed local file paths for self-hosted Bot API file access (one per line)", + "type": "textarea", + }, + "reaction_level": { + "label": "Reaction Level", + "description": "Emoji reaction feedback level", + "type": "select", + "options": [ + {"value": "off", "label": "Off (no reactions)"}, + {"value": "minimal", "label": "Minimal (ack only)"}, + {"value": "full", "label": "Full (all reactions)"}, + ], + "default": "minimal", + }, + "reaction_notifications": { + "label": "Reaction Notifications", + "description": "When to trigger notification reactions", + "type": "select", + "options": [ + {"value": "off", "label": "Off"}, + {"value": "own", "label": "Own messages only"}, + {"value": "all", "label": "All messages"}, + ], + "default": "off", + }, + "error_policy": { + "label": "Error Policy", + "description": "How to handle error responses to users", + "type": "select", + "options": [ + {"value": "off", "label": "Off (never send errors)"}, + {"value": "cooldown", "label": "Cooldown (throttled by interval)"}, + {"value": "full", "label": "Full (always send errors)"}, + ], + "default": "off", + }, + "error_cooldown_ms": { + "label": "Error Cooldown (ms)", + "description": "Minimum interval between error messages to the same chat", + "type": "number", + "default": 5000, + }, + "enabled": { + "label": "Enabled", + "description": "Enable or disable the Telegram channel", + "type": "toggle", + "default": True, + }, + } diff --git a/backend/package/yuxi/channels/adapters/telegram/connect/__init__.py b/backend/package/yuxi/channels/adapters/telegram/connect/__init__.py new file mode 100644 index 00000000..e69de29b diff --git a/backend/package/yuxi/channels/adapters/telegram/connect/allowed_updates.py b/backend/package/yuxi/channels/adapters/telegram/connect/allowed_updates.py new file mode 100644 index 00000000..faa6b969 --- /dev/null +++ b/backend/package/yuxi/channels/adapters/telegram/connect/allowed_updates.py @@ -0,0 +1,46 @@ +from __future__ import annotations + +from typing import Any + +DEFAULT_ALLOWED_UPDATES = [ + "message", + "edited_message", + "callback_query", + "my_chat_member", + "chat_member", + "message_reaction", + "message_reaction_count", + "poll", + "poll_answer", + "chat_join_request", +] + + +def get_allowed_updates(config: dict[str, Any] | None = None) -> list[str]: + cfg = config or {} + configured = cfg.get("allowed_updates") + if configured and isinstance(configured, list): + return configured + return DEFAULT_ALLOWED_UPDATES + + +def validate_update_types(updates: list[str]) -> list[str]: + valid = { + "message", + "edited_message", + "channel_post", + "edited_channel_post", + "inline_query", + "chosen_inline_result", + "callback_query", + "shipping_query", + "pre_checkout_query", + "poll", + "poll_answer", + "my_chat_member", + "chat_member", + "chat_join_request", + "message_reaction", + "message_reaction_count", + } + return [u for u in updates if u in valid] diff --git a/backend/package/yuxi/channels/adapters/telegram/connect/polling_liveness.py b/backend/package/yuxi/channels/adapters/telegram/connect/polling_liveness.py new file mode 100644 index 00000000..03f62897 --- /dev/null +++ b/backend/package/yuxi/channels/adapters/telegram/connect/polling_liveness.py @@ -0,0 +1,46 @@ +from __future__ import annotations + +import asyncio +import time +from typing import Any + +from yuxi.utils.logging_config import logger + + +class PollingLiveness: + def __init__(self, max_idle_seconds: float = 30.0): + self._max_idle = max_idle_seconds + self._last_activity: float = time.monotonic() + self._last_update_id: int = 0 + self._update_count: int = 0 + self._is_healthy: bool = True + + def record_activity(self, update_id: int) -> None: + self._last_activity = time.monotonic() + if update_id > self._last_update_id: + self._last_update_id = update_id + self._update_count += 1 + + def is_healthy(self) -> bool: + now = time.monotonic() + idle = now - self._last_activity + self._is_healthy = idle <= self._max_idle + return self._is_healthy + + def get_status(self) -> dict[str, Any]: + return { + "healthy": self.is_healthy(), + "last_activity_ago": time.monotonic() - self._last_activity, + "last_update_id": self._last_update_id, + "update_count": self._update_count, + } + + +async def start_liveness_monitor(liveness: PollingLiveness, check_interval: float = 10.0) -> None: + logger.info("[Telegram] Liveness monitor started") + while True: + await asyncio.sleep(check_interval) + if not liveness.is_healthy(): + logger.warning( + f"[Telegram] Polling liveness check failed, idle for {liveness.get_status()['last_activity_ago']:.1f}s" + ) diff --git a/backend/package/yuxi/channels/adapters/telegram/connect/update_offset_store.py b/backend/package/yuxi/channels/adapters/telegram/connect/update_offset_store.py new file mode 100644 index 00000000..688603bf --- /dev/null +++ b/backend/package/yuxi/channels/adapters/telegram/connect/update_offset_store.py @@ -0,0 +1,38 @@ +from __future__ import annotations + +import sqlite3 +from pathlib import Path + +DEFAULT_DB_PATH = Path("data/telegram_offset.db") + + +class UpdateOffsetStore: + def __init__(self, db_path: str | Path | None = None): + self._db_path = Path(db_path or DEFAULT_DB_PATH) + self._db_path.parent.mkdir(parents=True, exist_ok=True) + self._init_db() + + def _init_db(self) -> None: + with sqlite3.connect(str(self._db_path)) as conn: + conn.execute(""" + CREATE TABLE IF NOT EXISTS update_offsets ( + token_hash TEXT PRIMARY KEY, + offset INTEGER NOT NULL DEFAULT 0, + updated_at REAL NOT NULL DEFAULT 0 + ) + """) + + def get_offset(self, token_hash: str) -> int: + with sqlite3.connect(str(self._db_path)) as conn: + row = conn.execute("SELECT offset FROM update_offsets WHERE token_hash = ?", (token_hash,)).fetchone() + return row[0] if row else 0 + + def set_offset(self, token_hash: str, offset: int) -> None: + import time + + with sqlite3.connect(str(self._db_path)) as conn: + conn.execute( + "INSERT OR REPLACE INTO update_offsets (token_hash, offset, updated_at) VALUES (?, ?, ?)", + (token_hash, offset, time.time()), + ) + conn.commit() diff --git a/backend/package/yuxi/channels/adapters/telegram/connect/webhook_cert.py b/backend/package/yuxi/channels/adapters/telegram/connect/webhook_cert.py new file mode 100644 index 00000000..2799a010 --- /dev/null +++ b/backend/package/yuxi/channels/adapters/telegram/connect/webhook_cert.py @@ -0,0 +1,32 @@ +from __future__ import annotations + +from typing import Any + +from yuxi.utils.logging_config import logger + + +class WebhookCertManager: + def __init__(self, config: dict[str, Any] | None = None): + cfg = config or {} + self._cert_path = cfg.get("webhook_cert_path") + self._host = cfg.get("webhook_host", "0.0.0.0") + self._port = cfg.get("webhook_port", 8443) + + def has_cert(self) -> bool: + return bool(self._cert_path) + + def get_cert_path(self) -> str | None: + return self._cert_path + + def get_listen_config(self) -> dict[str, Any]: + return {"host": self._host, "port": self._port} + + def validate_cert(self) -> bool: + if not self._cert_path: + return False + import os + + if not os.path.exists(self._cert_path): + logger.error(f"[Telegram] Cert file not found: {self._cert_path}") + return False + return True diff --git a/backend/package/yuxi/channels/adapters/telegram/debounce.py b/backend/package/yuxi/channels/adapters/telegram/debounce.py new file mode 100644 index 00000000..a8e853c1 --- /dev/null +++ b/backend/package/yuxi/channels/adapters/telegram/debounce.py @@ -0,0 +1,73 @@ +from __future__ import annotations + +import hashlib +import time +from collections import OrderedDict +from typing import Any + +from yuxi.utils.logging_config import logger + + +class MessageDebouncer: + def __init__(self, max_entries: int = 1000, ttl_seconds: float = 300): + self._cache: OrderedDict[str, float] = OrderedDict() + self._max_entries = max_entries + self._ttl_seconds = ttl_seconds + + def _cleanup(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 build_debounce_key(self, update: dict[str, Any]) -> str | None: + message = update.get("message") or update.get("edited_message") or update.get("channel_post") + if not message: + return None + + chat_id = message.get("chat", {}).get("id", "") + message_id = message.get("message_id", "") + text = message.get("text", "") or message.get("caption", "") + + components = f"{chat_id}|{message_id}|{text[:100]}" + return hashlib.sha256(components.encode()).hexdigest()[:16] + + def is_duplicate(self, debounce_key: str) -> bool: + now = time.monotonic() + self._cleanup(now) + + if debounce_key in self._cache: + logger.debug(f"[Telegram] Debounced duplicate message: {debounce_key}") + return True + + if len(self._cache) >= self._max_entries: + self._cache.popitem(last=False) + + self._cache[debounce_key] = now + return False + + def clear_entry(self, debounce_key: str) -> None: + self._cache.pop(debounce_key, None) + + def cache_size(self) -> int: + return len(self._cache) + + +class ConnectTimeoutConfig: + def __init__(self, config: dict[str, Any] | None = None): + cfg = config or {} + self.connect_timeout = float(cfg.get("connect_timeout", 30)) + self.read_timeout = float(cfg.get("read_timeout", 60)) + self.write_timeout = float(cfg.get("write_timeout", 30)) + self.pool_timeout = float(cfg.get("pool_timeout", 30)) + + def get_request_kwargs(self) -> dict[str, Any]: + timeout_value = self.read_timeout + if hasattr(self, "connect_timeout"): + timeout_value = float(self.connect_timeout) + + return { + "connect_timeout": self.connect_timeout, + "read_timeout": self.read_timeout, + "write_timeout": self.write_timeout, + "pool_timeout": self.pool_timeout, + } diff --git a/backend/package/yuxi/channels/adapters/telegram/directory.py b/backend/package/yuxi/channels/adapters/telegram/directory.py new file mode 100644 index 00000000..a0321e99 --- /dev/null +++ b/backend/package/yuxi/channels/adapters/telegram/directory.py @@ -0,0 +1,110 @@ +from __future__ import annotations + +from typing import Any + +import httpx + +from yuxi.utils.logging_config import logger + +TELEGRAM_API_BASE = "https://api.telegram.org" + + +async def _telegram_api(token: str, method: str, params: dict | None = None) -> dict[str, Any]: + url = f"{TELEGRAM_API_BASE}/bot{token}/{method}" + async with httpx.AsyncClient(timeout=httpx.Timeout(15.0)) as client: + resp = await client.post(url, json=params or {}) + return resp.json() + + +async def list_peers(config: dict[str, Any], account_id: str = "default") -> list[dict[str, Any]]: + token = config.get("bot_token", "") + if not token: + logger.warning("[Telegram/Directory] No bot_token in config, cannot list peers") + return [] + + monitored_chats = config.get("monitored_chats", []) + results: list[dict[str, Any]] = [] + + for chat_id_str in monitored_chats: + try: + chat_id = int(chat_id_str) if chat_id_str.lstrip("-").isdigit() else chat_id_str + data = await _telegram_api(token, "getChat", {"chat_id": chat_id}) + if not data.get("ok"): + continue + + chat = data.get("result", {}) + ctype = chat.get("type", "") + + if ctype in ("private", "group", "supergroup"): + administrators = await _telegram_api(token, "getChatAdministrators", {"chat_id": chat_id}) + if administrators.get("ok"): + for admin in administrators.get("result", []): + user = admin.get("user", {}) + uid = user.get("id") + uname = user.get("first_name", "") or user.get("username", "") or str(uid) + results.append( + { + "id": f"tg:{uid}", + "name": uname, + "chat_id": str(chat_id), + "chat_title": chat.get("title", ""), + "is_admin": True, + } + ) + + member_count_data = await _telegram_api(token, "getChatMemberCount", {"chat_id": chat_id}) + if member_count_data.get("ok"): + count = member_count_data.get("result", 0) + if count > len([r for r in results if r.get("chat_id") == str(chat_id)]): + pass + + except Exception as e: + logger.warning(f"[Telegram/Directory] Failed to list peers for chat {chat_id_str}: {e}") + continue + + return results + + +async def list_groups(config: dict[str, Any], account_id: str = "default") -> list[dict[str, Any]]: + token = config.get("bot_token", "") + if not token: + logger.warning("[Telegram/Directory] No bot_token in config, cannot list groups") + return [] + + monitored_chats = config.get("monitored_chats", []) + results: list[dict[str, Any]] = [] + + for chat_id_str in monitored_chats: + try: + chat_id = int(chat_id_str) if chat_id_str.lstrip("-").isdigit() else chat_id_str + data = await _telegram_api(token, "getChat", {"chat_id": chat_id}) + if not data.get("ok"): + continue + + chat = data.get("result", {}) + ctype = chat.get("type", "") + + if ctype in ("group", "supergroup"): + count_data = await _telegram_api(token, "getChatMemberCount", {"chat_id": chat_id}) + member_count = count_data.get("result", 0) if count_data.get("ok") else 0 + + results.append( + { + "id": f"tg:group:{chat_id}", + "name": chat.get("title", str(chat_id)), + "type": ctype, + "member_count": member_count, + "description": chat.get("description", ""), + } + ) + except Exception as e: + logger.warning(f"[Telegram/Directory] Failed to list groups for chat {chat_id_str}: {e}") + continue + + return results + + +async def search_peers(config: dict[str, Any], query: str) -> list[dict[str, Any]]: + all_peers = await list_peers(config) + query_lower = query.lower() + return [p for p in all_peers if query_lower in p.get("name", "").lower() or query_lower in p.get("id", "").lower()] diff --git a/backend/package/yuxi/channels/adapters/telegram/directory_contract.py b/backend/package/yuxi/channels/adapters/telegram/directory_contract.py new file mode 100644 index 00000000..f35f6f58 --- /dev/null +++ b/backend/package/yuxi/channels/adapters/telegram/directory_contract.py @@ -0,0 +1,100 @@ +from __future__ import annotations + +from typing import Any + + +class DirectoryContract: + STORAGE_DIRS = [ + "bot_data", + "sessions", + "state", + "uploaded_files", + "downloaded_media", + "pending_approvals", + ] + STORAGE_EXT = "json" + + @staticmethod + def resolve_data_path(base_dir: str, sub_dir: str) -> str: + from pathlib import Path + + path = Path(base_dir) / sub_dir + path.mkdir(parents=True, exist_ok=True) + return str(path) + + @staticmethod + def resolve_json_path(base_dir: str, filename: str) -> str: + from pathlib import Path + + if not filename.endswith(".json"): + filename = f"{filename}.json" + return str(Path(base_dir) / filename) + + @staticmethod + def ensure_directories(base_dir: str) -> dict[str, str]: + result = {} + for sub_dir in DirectoryContract.STORAGE_DIRS: + result[sub_dir] = DirectoryContract.resolve_data_path(base_dir, sub_dir) + return result + + @staticmethod + def get_default_state_dir(config: dict[str, Any] | None = None) -> str: + cfg = config or {} + return cfg.get("state_dir", cfg.get("data_dir", "./data/telegram")) + + +class AckTimeoutGuard: + def __init__(self, timeout_s: float = 1.0): + self._timeout_s = timeout_s + + def wrap_callback(self, callback): + import asyncio + + async def guarded(*args, **kwargs): + try: + return await asyncio.wait_for(callback(*args, **kwargs), timeout=self._timeout_s) + except TimeoutError: + return {"ack_status": "timeout", "timeout_s": self._timeout_s} + except Exception as e: + return {"ack_status": "error", "error": str(e)} + + return guarded + + +class RespawnGuard: + def __init__(self, cooldown_s: float = 5.0): + self._cooldown_s = cooldown_s + self._last_spawn: float = 0 + self._spawn_count: int = 0 + + def can_spawn(self, now: float | None = None) -> bool: + import time + + now = now or time.monotonic() + if now - self._last_spawn < self._cooldown_s: + return False + self._last_spawn = now + self._spawn_count += 1 + return True + + def get_spawn_count(self) -> int: + return self._spawn_count + + +class RestartHandler: + def __init__(self): + self._restart_requested = False + self._restart_reason: str | None = None + + def request_restart(self, reason: str) -> None: + self._restart_requested = True + self._restart_reason = reason + + def is_restart_requested(self) -> bool: + return self._restart_requested + + def consume_restart_reason(self) -> str | None: + reason = self._restart_reason + self._restart_requested = False + self._restart_reason = None + return reason diff --git a/backend/package/yuxi/channels/adapters/telegram/dual_plugin.py b/backend/package/yuxi/channels/adapters/telegram/dual_plugin.py new file mode 100644 index 00000000..54442633 --- /dev/null +++ b/backend/package/yuxi/channels/adapters/telegram/dual_plugin.py @@ -0,0 +1,105 @@ +from __future__ import annotations + +from typing import Any + +from yuxi.utils.logging_config import logger + + +class TelegramAccountPlugin: + def __init__(self, config: dict[str, Any] | None = None): + self._config = config or {} + self._bot_id: str | None = None + self._bot_username: str | None = None + self._connected = False + self._connection_error: str | None = None + + @property + def bot_id(self) -> str | None: + return self._bot_id + + @property + def bot_username(self) -> str | None: + return self._bot_username + + @property + def connected(self) -> bool: + return self._connected + + @property + def connection_error(self) -> str | None: + return self._connection_error + + async def login(self, bot) -> dict[str, Any] | None: + try: + me = await bot.get_me() + self._bot_id = str(me.id) + self._bot_username = me.username or "" + self._connected = True + self._connection_error = None + + profile = { + "id": self._bot_id, + "username": self._bot_username, + "first_name": me.first_name, + "can_join_groups": getattr(me, "can_join_groups", None), + "can_read_all_group_messages": getattr(me, "can_read_all_group_messages", None), + "supports_inline_queries": getattr(me, "supports_inline_queries", None), + } + logger.info(f"[Telegram] Account logged in: @{self._bot_username} ({self._bot_id})") + return profile + except Exception as e: + self._connection_error = str(e) + logger.error(f"[Telegram] Account login failed: {e}") + return None + + def configure(self, config: dict[str, Any]) -> None: + self._config = config + + def get_account_info(self) -> dict[str, Any]: + return { + "bot_id": self._bot_id, + "bot_username": self._bot_username, + "connected": self._connected, + "error": self._connection_error, + } + + +class ReplyLoopPlugin: + def __init__(self, config: dict[str, Any] | None = None): + self._config = config or {} + self._running = False + self._processed_count = 0 + self._error_count = 0 + + @property + def running(self) -> bool: + return self._running + + @property + def processed_count(self) -> int: + return self._processed_count + + @property + def error_count(self) -> int: + return self._error_count + + def start(self) -> None: + self._running = True + logger.info("[Telegram] Reply loop started") + + def stop(self) -> None: + self._running = False + logger.info(f"[Telegram] Reply loop stopped (processed={self._processed_count}, errors={self._error_count})") + + def increment_processed(self) -> None: + self._processed_count += 1 + + def increment_error(self) -> None: + self._error_count += 1 + + def get_loop_info(self) -> dict[str, Any]: + return { + "running": self._running, + "processed_count": self._processed_count, + "error_count": self._error_count, + } diff --git a/backend/package/yuxi/channels/adapters/telegram/format.py b/backend/package/yuxi/channels/adapters/telegram/format.py new file mode 100644 index 00000000..f18ef40c --- /dev/null +++ b/backend/package/yuxi/channels/adapters/telegram/format.py @@ -0,0 +1,110 @@ +from __future__ import annotations + +import re +from html import escape + +_ESCAPE_CHARS = {"&": "&", "<": "<", ">": ">"} + +_TAG_PATTERN = r"<(?!/?(?:b|strong|i|em|u|ins|s|strike|del|code|pre|a|tg-spoiler|tg-emoji)\b)[^>]*>" +_TAG_PLACEHOLDER_PATTERN = re.compile(_TAG_PATTERN) + +_CODE_BLOCK_RE = re.compile(r"```(?:\w+)?\n.+?```", re.DOTALL) +_INLINE_CODE_RE = re.compile(r"`([^`\n]+?)`") + +_PLACEHOLDER_PREFIX = "\x00MDPLACEHOLDER" + + +def markdown_to_html(text: str) -> str: + code_blocks: dict[str, str] = {} + text = _protect_code_spans(text, _CODE_BLOCK_RE, code_blocks) + text = _protect_code_spans(text, _INLINE_CODE_RE, code_blocks) + + text = _convert_bold(text) + text = _convert_italic(text) + text = _convert_strikethrough(text) + text = _convert_underline(text) + text = _convert_links(text) + text = _convert_spoiler(text) + + text = _restore_code_spans(text, code_blocks) + text = _sanitize_html(text) + return text + + +def strip_all_tags(text: str) -> str: + return re.sub(r"<[^>]+>", "", text) + + +def strip_html_tags(text: str) -> str: + return strip_all_tags(text) + + +def _protect_code_spans(text: str, pattern: re.Pattern, storage: dict[str, str]) -> str: + result = [] + idx = 0 + for m in pattern.finditer(text): + result.append(text[idx : m.start()]) + placeholder = f"{_PLACEHOLDER_PREFIX}{len(storage)}" + storage[placeholder] = m.group(0) + result.append(placeholder) + idx = m.end() + result.append(text[idx:]) + return "".join(result) + + +def _restore_code_spans(text: str, storage: dict[str, str]) -> str: + result = text + for placeholder, code_text in storage.items(): + if placeholder.startswith(f"{_PLACEHOLDER_PREFIX}") and "```" in code_text: + inner = code_text[3:].rstrip("`").lstrip("\n") + lang_end = inner.find("\n") + inner = inner[lang_end + 1 :] if lang_end >= 0 else inner + safe = escape(inner, quote=False) + result = result.replace(placeholder, f"
{safe}")
+ elif placeholder in result:
+ inner = code_text.strip("`")
+ safe = escape(inner, quote=False)
+ result = result.replace(placeholder, f"{safe}")
+ return result
+
+
+def _convert_bold(text: str) -> str:
+ return re.sub(r"\*\*(.+?)\*\*", r"\1", text)
+
+
+def _convert_italic(text: str) -> str:
+ text = re.sub(r"(?\1", text)
+ text = re.sub(r"(?\1", text)
+ return text
+
+
+def _convert_strikethrough(text: str) -> str:
+ return re.sub(r"~~(.+?)~~", r"