feat(telegram-adapter): 实现完整的Telegram适配器基础代码

新增了Telegram适配器的全套基础模块,包括:
1.  核心适配器入口与会话工具
2.  账号管理、认证与配置系统
3.  连接相关的轮询、Webhook、更新偏移管理
4.  话题路由、管理与缓存系统
5.  消息反抖动、超时配置与工具类
6.  响应式UI与命令交互系统
7.  反应表情与通知系统
8.  审批与安全审计模块
9.  健康检查与状态监控
10. 贴纸缓存与视觉工具
11. 流式响应与协作功能
12. 群组迁移与目标归一化处理
This commit is contained in:
Kris 2026-05-12 00:49:52 +08:00
parent 3ad218e537
commit 71fec609bb
81 changed files with 7203 additions and 0 deletions

View File

@ -0,0 +1,3 @@
from yuxi.channels.adapters.telegram.adapter import TelegramAdapter
__all__ = ["TelegramAdapter"]

View File

@ -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

View File

@ -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())

View File

@ -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)

View File

@ -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

View File

@ -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

View File

@ -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"

File diff suppressed because it is too large Load Diff

View File

@ -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}")

View File

@ -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,
}

View File

@ -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)

View File

@ -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]

View File

@ -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},
)

View File

@ -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

View File

@ -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,
},
}

View File

@ -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]

View File

@ -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"
)

View File

@ -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()

View File

@ -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

View File

@ -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,
}

View File

@ -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()]

View File

@ -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

View File

@ -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,
}

View File

@ -0,0 +1,110 @@
from __future__ import annotations
import re
from html import escape
_ESCAPE_CHARS = {"&": "&amp;", "<": "&lt;", ">": "&gt;"}
_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"<pre>{safe}</pre>")
elif placeholder in result:
inner = code_text.strip("`")
safe = escape(inner, quote=False)
result = result.replace(placeholder, f"<code>{safe}</code>")
return result
def _convert_bold(text: str) -> str:
return re.sub(r"\*\*(.+?)\*\*", r"<b>\1</b>", text)
def _convert_italic(text: str) -> str:
text = re.sub(r"(?<!\*)\*([^*\n]+?)\*(?!\*)", r"<i>\1</i>", text)
text = re.sub(r"(?<!_)_([^_\n]+?)_(?!_)", r"<i>\1</i>", text)
return text
def _convert_strikethrough(text: str) -> str:
return re.sub(r"~~(.+?)~~", r"<s>\1</s>", text)
def _convert_underline(text: str) -> str:
return re.sub(r"__(\w.*?\w)__", r"<u>\1</u>", text)
def _convert_links(text: str) -> str:
return re.sub(r"\[(.+?)\]\((.+?)\)", r'<a href="\2">\1</a>', text)
def _convert_spoiler(text: str) -> str:
return re.sub(r"\|\|(.+?)\|\|", r"<tg-spoiler>\1</tg-spoiler>", text)
def _sanitize_html(text: str) -> str:
result = []
depth = 0
for ch in text:
if ch == "<":
depth += 1
elif ch == ">":
depth = max(0, depth - 1)
elif depth == 0:
if ch in _ESCAPE_CHARS:
result.append(_ESCAPE_CHARS[ch])
continue
result.append(ch)
return "".join(result)

View File

@ -0,0 +1,50 @@
from __future__ import annotations
from typing import Any
async def run_doctor(adapter) -> dict[str, Any]:
results = {
"status": "ok",
"checks": [],
"warnings": [],
"errors": [],
}
if adapter._application is None:
results["errors"].append("Application not initialized")
results["status"] = "error"
return results
try:
bot_info = await adapter._application.bot.get_me()
results["checks"].append(f"Bot connected as @{bot_info.username} (ID: {bot_info.id})")
except Exception as e:
results["errors"].append(f"Bot connection failed: {e}")
results["status"] = "error"
return results
try:
webhook_info = await adapter._application.bot.get_webhook_info()
if webhook_info.url:
results["checks"].append(f"Webhook set to {webhook_info.url}")
else:
results["checks"].append("Webhook disabled (likely polling mode)")
if webhook_info.pending_update_count and webhook_info.pending_update_count > 0:
results["warnings"].append(f"{webhook_info.pending_update_count} pending updates")
except Exception as e:
results["warnings"].append(f"Webhook info check failed: {e}")
dm_policy = adapter.config.get("dm_policy", "pairing")
group_policy = adapter.config.get("group_policy", "allowlist")
results["checks"].append(f"DM policy: {dm_policy}")
results["checks"].append(f"Group policy: {group_policy}")
if dm_policy == "allowlist":
allow_count = len(adapter.config.get("allow_from", []))
if allow_count == 0:
results["warnings"].append("DM allowlist is empty with allowlist policy")
else:
results["checks"].append(f"DM allowlist has {allow_count} entries")
return results

View File

@ -0,0 +1,72 @@
from __future__ import annotations
import time
from typing import Any
class StatusIssueCollector:
def __init__(self):
self._issues: list[dict[str, Any]] = []
self._last_scan: float = 0.0
def add_issue(self, severity: str, issue_type: str, message: str, metadata: dict[str, Any] | None = None) -> None:
self._issues.append(
{
"severity": severity,
"type": issue_type,
"message": message,
"metadata": metadata or {},
"timestamp": time.time(),
}
)
def get_issues(self, since_scan: bool = False, max_count: int = 50) -> list[dict[str, Any]]:
if since_scan:
result = [i for i in self._issues if i["timestamp"] > self._last_scan]
else:
result = self._issues
self._last_scan = time.time()
return result[-max_count:]
def clear(self) -> None:
self._issues.clear()
def has_critical(self) -> bool:
return any(i["severity"] == "error" for i in self._issues[-10:])
async def collect_telegram_status_issues(adapter) -> list[dict[str, Any]]:
issues = []
if adapter._status.value in ("error", "disconnected"):
issues.append(
{
"severity": "error",
"type": "adapter_status",
"message": f"Adapter status is {adapter._status.value}",
}
)
if adapter._application is None:
issues.append(
{
"severity": "error",
"type": "no_application",
"message": "Application not initialized",
}
)
try:
webhook_info = await adapter._application.bot.get_webhook_info()
if webhook_info.pending_update_count and webhook_info.pending_update_count > 10:
issues.append(
{
"severity": "warning",
"type": "pending_updates",
"message": f"{webhook_info.pending_update_count} pending updates",
}
)
except Exception:
pass
return issues

View File

@ -0,0 +1,167 @@
from __future__ import annotations
from typing import Any, Literal
from telegram import Bot, InlineKeyboardButton, InlineKeyboardMarkup
ButtonStyle = Literal["default", "primary", "danger"]
ButtonScope = Literal["off", "dm", "group", "all", "allowlist"]
def _style_to_flag_url(style: ButtonStyle) -> str | None:
if style == "primary":
return "tg://btn_primary"
if style == "danger":
return "tg://btn_danger"
return None
def build_inline_keyboard(
buttons: list[dict[str, Any]] | None,
scope: ButtonScope = "all",
chat_type: str = "direct",
allowlist_chats: set[str] | None = None,
chat_id: str = "",
) -> InlineKeyboardMarkup | None:
if not buttons:
return None
match scope:
case "off":
return None
case "dm":
if chat_type != "direct":
return None
case "group":
if chat_type == "direct":
return None
case "allowlist":
allowlist_chats = allowlist_chats or set()
if chat_id not in allowlist_chats:
return None
rows = []
for row in buttons:
if isinstance(row, list):
row_buttons = []
for btn in row:
if isinstance(btn, dict):
ib = InlineKeyboardButton(
text=btn.get("text", ""),
callback_data=btn.get("callback_data", ""),
url=btn.get("url"),
)
row_buttons.append(ib)
if row_buttons:
rows.append(row_buttons)
elif isinstance(row, dict):
btn = row
ib = InlineKeyboardButton(
text=btn.get("text", ""),
callback_data=btn.get("callback_data", ""),
url=btn.get("url"),
)
rows.append([ib])
return InlineKeyboardMarkup(rows) if rows else None
async def send_inline_buttons(
bot: Bot,
chat_id: str,
text: str,
buttons: list[dict[str, Any]] | None = None,
scope: ButtonScope = "all",
chat_type: str = "direct",
allowlist_chats: set[str] | None = None,
**kwargs,
) -> Any:
payload = {"chat_id": chat_id, "text": text}
payload.update(kwargs)
markup = build_inline_keyboard(
buttons,
scope=scope,
chat_type=chat_type,
allowlist_chats=allowlist_chats,
chat_id=chat_id,
)
if markup:
payload["reply_markup"] = markup
return await bot.send_message(**payload)
def build_model_selector_buttons(models: list[str], current_model: str = "") -> list[dict[str, Any]]:
buttons = []
for model in models:
prefix = "\u2705 " if model == current_model else ""
buttons.append(
{
"text": f"{prefix}{model}",
"callback_data": f"model:{model}",
}
)
return buttons
def build_exec_approval_buttons(approval_id: str) -> list[dict[str, Any]]:
return [
[
{"text": "\u2705 \u6279\u51c6", "callback_data": f"exec:approve:{approval_id}"},
{"text": "\u274c \u62d2\u7edd", "callback_data": f"exec:reject:{approval_id}"},
]
]
def build_provider_browse_keyboard(
providers: list[str],
current_provider: str = "",
prefix: str = "provider",
) -> list[list[dict[str, Any]]]:
rows = []
for provider in providers:
label = f"\u2705 {provider}" if provider == current_provider else provider
rows.append([{"text": label, "callback_data": f"{prefix}:{provider}"}])
return rows
def build_paginated_command_keyboard(
commands: list[dict[str, Any]],
page: int = 0,
page_size: int = 8,
prefix: str = "cmd",
) -> list[list[dict[str, Any]]]:
total_pages = (len(commands) + page_size - 1) // page_size if page_size > 0 else 1
start = page * page_size
end = min(start + page_size, len(commands))
page_commands = commands[start:end]
rows = []
row_count = max(2, min(4, len(page_commands)))
cmds_per_row = (len(page_commands) + row_count - 1) // row_count
for i in range(row_count):
row = []
for j in range(cmds_per_row):
idx = i * cmds_per_row + j
if idx < len(page_commands):
cmd = page_commands[idx]
row.append(
{
"text": cmd.get("label", cmd.get("command", "")),
"callback_data": f"{prefix}:{cmd.get('command', '')}",
}
)
if row:
rows.append(row)
if total_pages > 1:
nav = []
if page > 0:
nav.append({"text": "\u25c0 \u4e0a\u4e00\u9875", "callback_data": f"{prefix}:page:{page - 1}"})
nav.append({"text": f"{page + 1}/{total_pages}", "callback_data": f"{prefix}:page:info"})
if page < total_pages - 1:
nav.append({"text": "\u4e0b\u4e00\u9875 \u25b6", "callback_data": f"{prefix}:page:{page + 1}"})
rows.append(nav)
return rows

View File

@ -0,0 +1,92 @@
from __future__ import annotations
import asyncio
import hashlib
import os
from typing import Any
from yuxi.utils.logging_config import logger
class PollingLease:
def __init__(self, config: dict[str, Any] | None = None):
cfg = config or {}
self._token_hash = hashlib.sha256(cfg.get("bot_token", "").encode()).hexdigest()[:8]
self._instance_id = os.urandom(6).hex()
self._lease_ttl = cfg.get("polling_lease_ttl", 30)
self._acquired = False
self._renew_task: asyncio.Task | None = None
@property
def is_acquired(self) -> bool:
return self._acquired
async def acquire(self) -> bool:
try:
from yuxi.storage.redis import get_redis
redis = get_redis()
except ImportError:
logger.error(
"[Telegram] Redis not available - lease disabled. "
"Multiple instances may poll simultaneously, causing duplicate message processing."
)
self._acquired = True
return True
lease_key = f"telegram:polling_lease:{self._token_hash}"
try:
acquired = await redis.set(lease_key, self._instance_id, nx=True, ex=self._lease_ttl)
except Exception as e:
logger.error(
f"[Telegram] Redis lease check failed: {e}. "
"Proceeding without lease - multiple instances may poll simultaneously."
)
self._acquired = True
return True
if acquired:
self._acquired = True
self._renew_task = asyncio.create_task(self._renew_loop(lease_key))
logger.info(f"[Telegram] Polling lease acquired (instance={self._instance_id}, ttl={self._lease_ttl}s)")
return True
logger.debug("[Telegram] Polling lease held by another instance")
return False
async def release(self) -> None:
self._acquired = False
if self._renew_task and not self._renew_task.done():
self._renew_task.cancel()
try:
await self._renew_task
except asyncio.CancelledError:
pass
self._renew_task = None
try:
from yuxi.storage.redis import get_redis
redis = get_redis()
lease_key = f"telegram:polling_lease:{self._token_hash}"
await redis.delete(lease_key)
except Exception:
pass
async def _renew_loop(self, key: str) -> None:
try:
from yuxi.storage.redis import get_redis
redis = get_redis()
except ImportError:
return
while self._acquired:
await asyncio.sleep(self._lease_ttl * 0.5)
try:
await redis.expire(key, self._lease_ttl)
except asyncio.CancelledError:
break
except Exception:
pass

View File

@ -0,0 +1,367 @@
from __future__ import annotations
from typing import Any
TELEGRAM_MESSAGE_ACTIONS: dict[str, dict[str, Any]] = {
"send": {
"status": "implemented",
"description": "发送消息到指定聊天",
"api": "sendMessage",
},
"sendAttachment": {
"status": "implemented",
"description": "发送图片/视频/音频/文件/语音 附件消息",
"api": "sendPhoto / sendVideo / sendAudio / sendDocument / sendVoice",
},
"sendMediaGroup": {
"status": "implemented",
"description": "发送媒体组消息(多张图片/视频批量发送)",
"api": "sendMediaGroup",
},
"sendLocation": {
"status": "implemented",
"description": "发送位置信息",
"api": "sendLocation",
},
"sendContact": {
"status": "implemented",
"description": "发送联系人信息",
"api": "sendContact",
},
"sendPoll": {
"status": "implemented",
"description": "发送投票",
"api": "sendPoll",
},
"sendSticker": {
"status": "implemented",
"description": "发送贴纸",
"api": "sendSticker",
},
"sendDice": {
"status": "implemented",
"description": "发送骰子/飞镖等表情互动",
"api": "sendDice",
},
"sendInvoice": {
"status": "unsupported",
"description": "发送支付账单",
"api": "sendInvoice",
},
"sendGame": {
"status": "unsupported",
"description": "发送HTML5游戏",
"api": "sendGame",
},
"reply": {
"status": "implemented",
"description": "回复指定消息(通过 reply_to_message_id 参数)",
"api": "sendMessage + reply_to_message_id",
},
"forward": {
"status": "implemented",
"description": "转发消息",
"api": "forwardMessage",
},
"copy": {
"status": "implemented",
"description": "复制消息到其他聊天(不转发展示来源)",
"api": "copyMessage",
},
"edit": {
"status": "implemented",
"description": "编辑已发送的文本消息",
"api": "editMessageText",
},
"editCaption": {
"status": "implemented",
"description": "编辑已发送媒体的标题",
"api": "editMessageCaption",
},
"editMedia": {
"status": "unsupported",
"description": "替换已发送的媒体附件内容",
"api": "editMessageMedia",
},
"editInline": {
"status": "implemented",
"description": "更新 Inline Keyboard 按钮",
"api": "editMessageReplyMarkup",
},
"delete": {
"status": "implemented",
"description": "删除消息(Bot 能删除 48h 内的消息)",
"api": "deleteMessage",
},
"pin": {
"status": "implemented",
"description": "置顶消息(需要置顶权限)",
"api": "pinChatMessage",
},
"unpin": {
"status": "implemented",
"description": "取消置顶消息",
"api": "unpinChatMessage",
},
"unpinAll": {
"status": "implemented",
"description": "取消所有置顶消息",
"api": "unpinAllChatMessages",
},
"react": {
"status": "implemented",
"description": "发送emoji 反应(Bot API 7.0+)",
"api": "setMessageReaction",
},
"typing": {
"status": "implemented",
"description": "显示输入状态(打字/上传文件等)",
"api": "sendChatAction",
},
"thread": {
"status": "implemented",
"description": "创建论坛主题(Forum Topic)",
"api": "createForumTopic",
},
"closeThread": {
"status": "implemented",
"description": "关闭 Forum Topic",
"api": "closeForumTopic",
},
"reopenThread": {
"status": "implemented",
"description": "重新打开已关闭的 Forum Topic",
"api": "reopenForumTopic",
},
"deleteThread": {
"status": "implemented",
"description": "删除 Forum Topic",
"api": "deleteForumTopic",
},
"streamChunk": {
"status": "implemented",
"description": "流式输出消息(通过消息编辑逐步展示)",
"api": "editMessageText (streaming)",
},
"inlineButtons": {
"status": "implemented",
"description": "发送内联按钮(InlineKeyboardMarkup)",
"api": "InlineKeyboardMarkup + CallbackQuery",
},
"replyMarkup": {
"status": "implemented",
"description": "发送自定义回复键盘",
"api": "ReplyKeyboardMarkup",
},
"forceReply": {
"status": "implemented",
"description": "强制用户回复",
"api": "ForceReply",
},
"addParticipant": {
"status": "unsupported",
"description": "添加成员到群组(Bot API 不支持)",
"api": "N/A",
},
"removeParticipant": {
"status": "unsupported",
"description": "从群组移除成员(Bot API 不支持)",
"api": "N/A",
},
"kick": {
"status": "implemented",
"description": "踢出群成员(先封禁再解封实现)",
"api": "banChatMember + unbanChatMember",
},
"ban": {
"status": "implemented",
"description": "封禁用户",
"api": "banChatMember",
},
"unban": {
"status": "implemented",
"description": "解封用户",
"api": "unbanChatMember",
},
"mute": {
"status": "implemented",
"description": "禁言群成员",
"api": "restrictChatMember (can_send_messages=False)",
},
"unmute": {
"status": "implemented",
"description": "解除禁言",
"api": "restrictChatMember (can_send_messages=True)",
},
"promoteAdmin": {
"status": "implemented",
"description": "提升成员为管理员",
"api": "promoteChatMember",
},
"demoteAdmin": {
"status": "implemented",
"description": "撤销管理员权限",
"api": "promoteChatMember (reset)",
},
"setPermissions": {
"status": "implemented",
"description": "设置群权限",
"api": "setChatPermissions",
},
"setChatPhoto": {
"status": "implemented",
"description": "设置群头像",
"api": "setChatPhoto",
},
"deleteChatPhoto": {
"status": "implemented",
"description": "删除群头像",
"api": "deleteChatPhoto",
},
"setTitle": {
"status": "implemented",
"description": "设置群标题",
"api": "setChatTitle",
},
"setDescription": {
"status": "implemented",
"description": "设置群描述",
"api": "setChatDescription",
},
"leave": {
"status": "implemented",
"description": "Bot 退出群聊",
"api": "leaveChat",
},
"inviteLink": {
"status": "implemented",
"description": "创建/导出群邀请链接",
"api": "exportChatInviteLink / createChatInviteLink",
},
"revokeInviteLink": {
"status": "unsupported",
"description": "撤销邀请链接",
"api": "revokeChatInviteLink",
},
"broadcast": {
"status": "unsupported",
"description": "群发消息(Telegram Bot API 不直接支持)",
"api": "N/A",
},
"getMe": {
"status": "implemented",
"description": "获取 Bot 基本信息",
"api": "getMe",
},
"getChat": {
"status": "implemented",
"description": "获取聊天基本信息",
"api": "getChat",
},
"getChatAdmins": {
"status": "implemented",
"description": "获取群组管理员列表",
"api": "getChatAdministrators",
},
"getChatMemberCount": {
"status": "implemented",
"description": "获取聊天成员数量",
"api": "getChatMemberCount",
},
"getChatMember": {
"status": "implemented",
"description": "获取指定成员信息",
"api": "getChatMember",
},
"downloadFile": {
"status": "implemented",
"description": "下载文件",
"api": "getFile + file_path",
},
"uploadFile": {
"status": "implemented",
"description": "上传文件",
"api": "sendDocument (file upload)",
},
}
_TELEGRAM_ACTION_MAPPING: dict[str, str] = {
"send": "send",
"sendAttachment": "send_attachment",
"sendMediaGroup": "send_media_group",
"sendLocation": "send_location",
"sendContact": "send_contact",
"sendPoll": "create_poll",
"sendSticker": "sticker",
"sendDice": "send_dice",
"reply": "reply",
"forward": "forward",
"copy": "copy",
"edit": "edit",
"editCaption": "edit_caption",
"editInline": "edit_inline",
"delete": "delete",
"pin": "pin",
"unpin": "unpin",
"unpinAll": "unpin_all",
"react": "react",
"typing": "typing",
"thread": "thread_create",
"closeThread": "close_thread",
"reopenThread": "reopen_thread",
"deleteThread": "delete_thread",
"streamChunk": "stream",
"addParticipant": "add_participant",
"removeParticipant": "remove_participant",
"kick": "kick",
"ban": "ban",
"unban": "unban",
"mute": "mute",
"unmute": "unmute",
"setTitle": "rename_group",
"setChatPhoto": "set_group_icon",
"deleteChatPhoto": "delete_chat_photo",
"setDescription": "set_chat_description",
"leave": "leave_group",
"broadcast": "broadcast",
"inviteLink": "invite_link",
"getMe": "get_me",
"getChat": "channel_info",
"getChatAdmins": "role_info",
"getChatMemberCount": "channel_list",
"getChatMember": "member_info",
"downloadFile": "download_file",
"uploadFile": "upload_file",
"setPermissions": "permissions",
"promoteAdmin": "role_add",
"demoteAdmin": "role_remove",
"createEvent": "event_create",
}
def register_telegram_actions() -> None:
from yuxi.channels.message_actions import ActionDeclaration, ActionRegistry, ActionStatus, MessageAction
declarations: dict[MessageAction, ActionDeclaration] = {}
for action_name, info in TELEGRAM_MESSAGE_ACTIONS.items():
mapped = _TELEGRAM_ACTION_MAPPING.get(action_name)
if mapped is None:
continue
try:
action_enum = MessageAction(mapped)
except ValueError:
continue
status_str = info.get("status", "unsupported")
status = {
"implemented": ActionStatus.SUPPORTED,
"planned": ActionStatus.PARTIAL,
"unsupported": ActionStatus.UNSUPPORTED,
}.get(status_str, ActionStatus.UNSUPPORTED)
declarations[action_enum] = ActionDeclaration(
action=action_enum,
status=status,
reason=info.get("description", ""),
impl=info.get("api", ""),
)
ActionRegistry.register_channel_actions("telegram", declarations)

View File

@ -0,0 +1,79 @@
from __future__ import annotations
import asyncio
from typing import Any
from telegram.error import (
BadRequest,
Conflict,
Forbidden,
InvalidToken,
NetworkError,
RetryAfter,
TelegramError,
TimedOut,
)
RECOVERABLE_ERROR_TYPES = (TimedOut, NetworkError, OSError, asyncio.TimeoutError)
NON_RECOVERABLE_ERROR_TYPES = (InvalidToken, Forbidden, BadRequest, Conflict)
def is_recoverable_network_error(error: Exception) -> bool:
if isinstance(error, RECOVERABLE_ERROR_TYPES):
return True
if isinstance(error, RetryAfter):
return True
return False
def is_safe_to_retry_send(exc: Exception) -> bool:
if isinstance(exc, NON_RECOVERABLE_ERROR_TYPES):
return False
if isinstance(exc, RECOVERABLE_ERROR_TYPES):
return True
if isinstance(exc, RetryAfter):
return True
if isinstance(exc, TelegramError):
error_msg = str(exc).lower()
if "too many requests" in error_msg or "flood" in error_msg:
return True
if "chat not found" in error_msg or "bot was kicked" in error_msg:
return False
return False
def classify_telegram_error(exc: Exception) -> dict[str, Any]:
if isinstance(exc, RetryAfter):
return {
"category": "rate_limited",
"recoverable": True,
"retry_after": exc.retry_after,
}
if isinstance(exc, InvalidToken):
return {"category": "auth_error", "recoverable": False}
if isinstance(exc, Forbidden):
error_msg = str(exc).lower()
if "blocked" in error_msg:
return {"category": "blocked_by_user", "recoverable": False}
return {"category": "forbidden", "recoverable": False}
if isinstance(exc, BadRequest):
error_msg = str(exc).lower()
if "chat not found" in error_msg:
return {"category": "chat_not_found", "recoverable": False}
if "message not found" in error_msg:
return {"category": "message_not_found", "recoverable": False}
return {"category": "bad_request", "recoverable": False}
if isinstance(exc, (TimedOut, NetworkError, asyncio.TimeoutError, OSError)):
return {"category": "network_error", "recoverable": True}
return {"category": "unknown", "recoverable": False}
def get_retry_delay(exc: Exception, attempt: int, min_delay: float = 1.0, max_delay: float = 60.0) -> float:
if isinstance(exc, RetryAfter):
return exc.retry_after
if isinstance(exc, TelegramError):
error_msg = str(exc).lower()
if "too many requests" in error_msg or "flood" in error_msg:
return min(max_delay, min_delay * (2 ** (attempt + 1)))
return min(max_delay, min_delay * (2**attempt))

View File

@ -0,0 +1,86 @@
from __future__ import annotations
from typing import Any
from telegram import Bot
from telegram.error import TelegramError
from yuxi.utils.logging_config import logger
async def create_poll(
bot: Bot,
chat_id: str,
question: str,
options: list[str],
is_anonymous: bool = True,
allows_multiple_answers: bool = False,
poll_type: str = "regular",
open_period: int | None = None,
close_date: int | None = None,
visibility: str = "public",
**kwargs,
) -> Any:
try:
poll_kwargs = {
"chat_id": chat_id,
"question": question,
"options": options,
"is_anonymous": is_anonymous,
"allows_multiple_answers": allows_multiple_answers,
"type": poll_type,
"open_period": open_period,
"close_date": close_date,
**kwargs,
}
if visibility == "quiz" and poll_type == "quiz":
poll_kwargs["type"] = "quiz"
return await bot.send_poll(**poll_kwargs)
except TelegramError as e:
logger.error(f"[Telegram] Failed to create poll: {e}")
raise
async def stop_poll(
bot: Bot,
chat_id: str,
message_id: int,
) -> Any:
try:
return await bot.stop_poll(
chat_id=chat_id,
message_id=message_id,
)
except TelegramError as e:
logger.error(f"[Telegram] Failed to stop poll: {e}")
raise
async def create_quiz(
bot: Bot,
chat_id: str,
question: str,
options: list[str],
correct_option_id: int = 0,
explanation: str | None = None,
is_anonymous: bool = True,
open_period: int | None = None,
close_date: int | None = None,
**kwargs,
) -> Any:
try:
return await bot.send_poll(
chat_id=chat_id,
question=question,
options=options,
type="quiz",
correct_option_id=correct_option_id,
explanation=explanation or "",
is_anonymous=is_anonymous,
open_period=open_period,
close_date=close_date,
**kwargs,
)
except TelegramError as e:
logger.error(f"[Telegram] Failed to create quiz: {e}")
raise

View File

@ -0,0 +1,40 @@
from __future__ import annotations
from typing import Any
from telegram import Bot
from telegram.error import TelegramError
from yuxi.utils.logging_config import logger
async def probe_bot(bot: Bot) -> dict[str, Any]:
try:
bot_info = await bot.get_me()
return {
"id": bot_info.id,
"username": bot_info.username,
"first_name": bot_info.first_name,
"last_name": bot_info.last_name,
"can_join_groups": bot_info.can_join_groups,
"can_read_all_group_messages": bot_info.can_read_all_group_messages,
"supports_inline_queries": bot_info.supports_inline_queries,
}
except TelegramError as e:
logger.error(f"[Telegram] Bot probe failed: {e}")
raise
async def probe_webhook(bot: Bot) -> dict[str, Any]:
try:
info = await bot.get_webhook_info()
return {
"url": info.url,
"has_custom_certificate": info.has_custom_certificate,
"pending_update_count": info.pending_update_count,
"last_error_date": info.last_error_date,
"last_error_message": info.last_error_message,
}
except TelegramError as e:
logger.error(f"[Telegram] Webhook probe failed: {e}")
raise

View File

@ -0,0 +1,59 @@
from __future__ import annotations
from typing import Any
REACTION_VARIANTS: dict[str, list[str]] = {
"like": ["\u2764\ufe0f", "\U0001f44d", "\U0001f90d", "\U0001f499", "\U0001f49c"],
"ack": ["\U0001f44d", "\u2705", "\U0001f44c"],
"processing": ["\u23f3", "\U0001f916", "\U0001f4ac"],
"done": ["\u2705", "\U0001f44d", "\U0001f389"],
"error": ["\u274c", "\u26a0\ufe0f", "\U0001f6ab"],
"thinking": ["\U0001f4ad", "\U0001f9e0", "\U0001f914"],
"searching": ["\U0001f50d", "\U0001f50e", "\U0001f4da"],
"generating": ["\u2728", "\U0001f4dd", "\U0001f3a8"],
}
def get_reaction_variant(action: str, index: int = 0) -> str:
variants = REACTION_VARIANTS.get(action, ["\u2764\ufe0f"])
if index < 0:
return variants[0]
return variants[index % len(variants)]
def get_all_reaction_actions() -> list[str]:
return list(REACTION_VARIANTS.keys())
def resolve_reaction_emoji(action_or_emoji: str) -> str:
if action_or_emoji in REACTION_VARIANTS:
return get_reaction_variant(action_or_emoji, 0)
return action_or_emoji
class ReactionNotificationTriggerFilter:
def __init__(self, config: dict[str, Any] | None = None):
cfg = config or {}
react_cfg = cfg.get("reaction_notifications", {})
if isinstance(react_cfg, dict):
self._mode = react_cfg.get("mode", "off")
self._whitelist = react_cfg.get("whitelist", [])
else:
self._mode = str(react_cfg) if react_cfg else "off"
self._whitelist = []
def should_notify(self, action: str, chat_id: str | None = None) -> bool:
if self._mode == "off":
return False
if self._mode == "all":
return True
if self._mode == "own":
return True
if self._mode == "whitelist" and chat_id:
return chat_id in self._whitelist
return False
def notify_reaction(self, action: str) -> str | None:
if not self.should_notify(action):
return None
return get_reaction_variant(action, 0)

View File

@ -0,0 +1,31 @@
from __future__ import annotations
from typing import Any, Literal
ReactionLevel = Literal["minimal", "ack", "off"]
class ReactionLevelController:
def __init__(self, config: dict[str, Any] | None = None):
cfg = config or {}
self._level: ReactionLevel = cfg.get("reaction_level", "minimal")
def should_react(self, event_type: str) -> bool:
if self._level == "off":
return False
if self._level == "ack":
return event_type in ("message.received",)
return True
def get_ack_emoji(self) -> str:
return "\u2705"
def get_error_emoji(self) -> str:
return "\u274c"
def get_processing_emoji(self) -> str:
return "\u23f3"
def get_reaction_level(config: dict[str, Any]) -> str:
return config.get("reaction_level", "minimal")

View File

@ -0,0 +1,21 @@
from __future__ import annotations
from typing import Any, Literal
ReactionNotificationMode = Literal["off", "own", "all"]
class ReactionNotifications:
def __init__(self, config: dict[str, Any] | None = None):
cfg = config or {}
self._mode: ReactionNotificationMode = cfg.get("reaction_notifications", "all")
def should_notify(self, bot_username: str, reactor_username: str) -> bool:
if self._mode == "off":
return False
if self._mode == "own":
return reactor_username == bot_username
return True
def get_mode(self) -> str:
return self._mode

View File

@ -0,0 +1,17 @@
from __future__ import annotations
STATUS_REACTION_VARIANTS: dict[str, list[str]] = {
"ok": ["\u2705", "\u2714\ufe0f", "\u2b50"],
"error": ["\u274c", "\u26a0\ufe0f", "\u203c\ufe0f"],
"processing": ["\u23f3", "\ud83d\udce2", "\ud83d\udc40"],
"thinking": ["\ud83e\udd14", "\ud83d\udcad", "\ud83e\udde0"],
"success": ["\ud83c\udf89", "\ud83c\udfc6", "\ud83d\udc4f"],
"warning": ["\u26a0\ufe0f", "\ud83d\uded1", "\ud83d\udea8"],
}
def get_status_reaction(status: str, index: int = 0) -> str | None:
variants = STATUS_REACTION_VARIANTS.get(status, [])
if not variants:
return None
return variants[index % len(variants)]

View File

@ -0,0 +1,155 @@
from __future__ import annotations
import hashlib
import secrets
import time
from typing import Any
from .security_audit import audit_security_policy
from .group_migration import GroupMigrationHandler
from .audit_membership import MembershipAuditor
class TelegramSecurityPolicy:
def __init__(self, config: dict[str, Any] | None = None):
cfg = config or {}
self._dm_policy = cfg.get("dm_policy", "pairing")
self._dm_allow_from = set(cfg.get("allow_from", []))
self._group_policy = cfg.get("group_policy", "allowlist")
self._group_allow_from = set(cfg.get("group_allow_from", []))
self._groups_config = cfg.get("groups", {})
self._pairing_code = cfg.get("pairing_code") or secrets.token_hex(6)
self._paired_users: set[str] = set(cfg.get("paired_users", []))
self._pairing_code_hash = hashlib.sha256(self._pairing_code.encode()).hexdigest()
self._error_policy = cfg.get("error_policy", "off")
self._error_cooldown_ms = cfg.get("error_cooldown_ms", 5000)
self._error_cooldowns: dict[str, float] = {}
self._error_counts: dict[str, int] = {}
def check_dm_policy(self, user_id: str) -> tuple[bool, str]:
match self._dm_policy:
case "open":
return True, ""
case "disabled":
return False, "DM is disabled"
case "pairing":
prefix = f"tg:{user_id}"
if prefix in self._paired_users:
return True, ""
return False, f"User {user_id} not paired. Use pairing code to connect."
case "allowlist":
prefix = f"tg:{user_id}"
if prefix in self._dm_allow_from:
return True, ""
return False, f"User {user_id} not in DM allowlist"
case _:
return False, f"Unknown DM policy: {self._dm_policy}"
def check_group_policy(self, chat_id: str, user_id: str, thread_id: str | None = None) -> tuple[bool, str]:
match self._group_policy:
case "open":
return True, ""
case "disabled":
return False, "Group messaging is disabled"
case "allowlist":
prefix = f"tg:{user_id}"
if prefix in self._group_allow_from:
return True, ""
chat_config = self._groups_config.get(chat_id, {})
per_group_allow = set(chat_config.get("allow_from", []))
if prefix in per_group_allow:
return True, ""
if thread_id:
topics_config = chat_config.get("topics", {})
topic_config = topics_config.get(thread_id, {})
per_topic_allow = set(topic_config.get("allow_from", []))
if prefix in per_topic_allow:
return True, ""
return False, f"User {user_id} not in group allowlist"
case _:
return False, f"Unknown group policy: {self._group_policy}"
def check_topic_policy(self, chat_id: str, thread_id: str, user_id: str) -> tuple[bool, str]:
chat_config = self._groups_config.get(chat_id, {})
topics_config = chat_config.get("topics", {})
topic_config = topics_config.get(thread_id, {})
topic_policy = topic_config.get("policy", self._group_policy)
topic_allow_from = set(topic_config.get("allow_from", []))
match topic_policy:
case "open":
return True, ""
case "disabled":
return False, "Topic messaging is disabled"
case "allowlist":
prefix = f"tg:{user_id}"
if prefix in topic_allow_from:
return True, ""
return False, f"User {user_id} not in topic allowlist"
case _:
return self.check_group_policy(chat_id, user_id, thread_id)
def get_group_config(self, chat_id: str) -> dict[str, Any]:
default_config = self._groups_config.get("*", {})
specific_config = self._groups_config.get(chat_id, {})
return {**default_config, **specific_config}
def get_topic_config(self, chat_id: str, thread_id: str) -> dict[str, Any]:
chat_config = self._groups_config.get(chat_id, {})
topics_config = chat_config.get("topics", {})
return topics_config.get(thread_id, {})
def check_require_mention(self, chat_id: str) -> bool:
group_config = self.get_group_config(chat_id)
return group_config.get("require_mention", True)
def get_error_policy(self) -> str:
return self._error_policy
def should_send_error(self, chat_id: str) -> bool:
if self._error_policy == "off":
return False
now = time.monotonic()
cooldown_key = f"err:{chat_id}"
if cooldown_key in self._error_cooldowns:
elapsed = (now - self._error_cooldowns[cooldown_key]) * 1000
if elapsed < self._error_cooldown_ms:
return False
self._error_cooldowns[cooldown_key] = now
self._error_counts[chat_id] = self._error_counts.get(chat_id, 0) + 1
return True
def reset_error_cooldown(self, chat_id: str) -> None:
self._error_cooldowns.pop(f"err:{chat_id}", None)
def verify_pairing_code(self, code: str) -> bool:
code_hash = hashlib.sha256(code.encode()).hexdigest()
return secrets.compare_digest(code_hash, self._pairing_code_hash)
def pair_user(self, user_id: str, code: str) -> tuple[bool, str]:
if not self.verify_pairing_code(code):
return False, "Invalid pairing code"
prefix = f"tg:{user_id}"
if prefix in self._paired_users:
return True, "Already paired"
self._paired_users.add(prefix)
return True, "Paired successfully"
def unpair_user(self, user_id: str) -> bool:
prefix = f"tg:{user_id}"
if prefix in self._paired_users:
self._paired_users.discard(prefix)
return True
return False
__all__ = [
"TelegramSecurityPolicy",
"audit_security_policy",
"GroupMigrationHandler",
"MembershipAuditor",
]

View File

@ -0,0 +1,51 @@
from __future__ import annotations
from typing import Any
class MembershipAuditor:
def __init__(self, config: dict[str, Any] | None = None):
self._config = config or {}
self._membership_log: list[dict[str, Any]] = []
def record_join(self, chat_id: str, user_id: str, user_name: str) -> None:
self._membership_log.append(
{
"chat_id": chat_id,
"user_id": user_id,
"user_name": user_name,
"action": "join",
"timestamp": __import__("time").time(),
}
)
def record_leave(self, chat_id: str, user_id: str, user_name: str) -> None:
self._membership_log.append(
{
"chat_id": chat_id,
"user_id": user_id,
"user_name": user_name,
"action": "leave",
"timestamp": __import__("time").time(),
}
)
def record_kick(self, chat_id: str, user_id: str, by_user_id: str) -> None:
self._membership_log.append(
{
"chat_id": chat_id,
"user_id": user_id,
"by_user_id": by_user_id,
"action": "kick",
"timestamp": __import__("time").time(),
}
)
def get_chat_activity(self, chat_id: str, limit: int = 50) -> list[dict[str, Any]]:
return [e for e in self._membership_log if e["chat_id"] == chat_id][-limit:]
def get_user_activity(self, user_id: str, limit: int = 50) -> list[dict[str, Any]]:
return [e for e in self._membership_log if e["user_id"] == user_id][-limit:]
def clear(self) -> None:
self._membership_log.clear()

View File

@ -0,0 +1,31 @@
from __future__ import annotations
from typing import Any
from yuxi.utils.logging_config import logger
class GroupMigrationHandler:
def __init__(self):
self._migrations: dict[str, str] = {}
def record_migration(self, old_chat_id: str, new_chat_id: str) -> None:
self._migrations[old_chat_id] = new_chat_id
logger.info(f"[Telegram] Group migration recorded: {old_chat_id} -> {new_chat_id}")
def get_migrated_id(self, chat_id: str) -> str:
return self._migrations.get(chat_id, chat_id)
def has_migration(self, chat_id: str) -> bool:
return chat_id in self._migrations
def handle_migration_message(self, message) -> dict[str, Any] | None:
if message.migrate_to_chat_id:
old_id = str(message.chat.id)
new_id = str(message.migrate_to_chat_id)
self.record_migration(old_id, new_id)
return {"old_chat_id": old_id, "new_chat_id": new_id}
return None
def list_migrations(self) -> dict[str, str]:
return dict(self._migrations)

View File

@ -0,0 +1,78 @@
from __future__ import annotations
from typing import Any
def audit_security_policy(config: dict[str, Any]) -> dict[str, Any]:
issues = []
dm_policy = config.get("dm_policy", "pairing")
group_policy = config.get("group_policy", "allowlist")
if dm_policy == "open":
issues.append(
{
"severity": "warning",
"type": "dm_policy_open",
"message": "DM policy is set to 'open' - anyone can message the bot",
}
)
if group_policy == "open":
issues.append(
{
"severity": "warning",
"type": "group_policy_open",
"message": "Group policy is set to 'open' - bot responds to all groups",
}
)
if dm_policy == "disabled" and group_policy == "disabled":
issues.append(
{
"severity": "info",
"type": "all_disabled",
"message": "Both DM and group policies are disabled - bot won't respond to anyone",
}
)
allow_from = config.get("allow_from", [])
if dm_policy == "allowlist" and not allow_from:
issues.append(
{
"severity": "error",
"type": "empty_allowlist",
"message": "DM policy is 'allowlist' but allowlist is empty",
}
)
exec_approvals = config.get("exec_approvals", {})
if exec_approvals.get("enabled") and not exec_approvals.get("approvers"):
issues.append(
{
"severity": "error",
"type": "no_approvers",
"message": "Exec approvals enabled but no approvers configured",
}
)
return {
"issues_count": len(issues),
"issues": issues,
"dm_policy": dm_policy,
"group_policy": group_policy,
}
def audit_dm_access(config: dict[str, Any], recent_users: list[str]) -> dict[str, Any]:
dm_policy = config.get("dm_policy", "pairing")
allow_from = set(config.get("allow_from", []))
unknown_users = [u for u in recent_users if f"tg:{u}" not in allow_from]
return {
"dm_policy": dm_policy,
"allowed_users": len(allow_from),
"recent_unknown_users": len(unknown_users),
"unknown_users": unknown_users[:20],
}

View File

@ -0,0 +1,216 @@
from __future__ import annotations
import asyncio
import random
from typing import Any
from telegram import Bot, ReplyParameters
from telegram.error import BadRequest, NetworkError, RetryAfter, TelegramError, TimedOut
from telegram.ext import Application
from yuxi.channels.sdk.format import markdown_to_html, strip_html_tags as strip_all_tags
from yuxi.utils.logging_config import logger
async def send_with_retry(
bot: Bot,
chat_id: str,
payload: dict[str, Any],
config: dict[str, Any] | None = None,
) -> Any:
cfg = config or {}
retry_cfg = cfg.get("retry", {})
max_retries = retry_cfg.get("attempts", 3) if isinstance(retry_cfg, dict) else 3
min_delay_ms = retry_cfg.get("min_delay_ms", 400) if isinstance(retry_cfg, dict) else 400
max_delay_ms = retry_cfg.get("max_delay_ms", 30000) if isinstance(retry_cfg, dict) else 30000
jitter = retry_cfg.get("jitter", 0.1) if isinstance(retry_cfg, dict) else 0.1
last_exception = None
for attempt in range(max_retries):
try:
return await bot.send_message(**payload)
except RetryAfter as e:
await asyncio.sleep(e.retry_after)
last_exception = e
except BadRequest as e:
if "can't parse entities" in str(e).lower() and payload.get("parse_mode") == "HTML":
payload = dict(payload)
payload.pop("parse_mode", None)
if "text" in payload:
payload["text"] = strip_all_tags(payload["text"])
try:
return await bot.send_message(**payload)
except Exception:
raise
if (
"reply message not found" in str(e).lower() or "replied message not found" in str(e).lower()
) and payload.get("reply_to_message_id"):
payload = dict(payload)
payload.pop("reply_to_message_id", None)
payload.pop("reply_parameters", None)
logger.debug(f"[Telegram] Reply target not found, retrying without reply_to for chat {chat_id}")
return await bot.send_message(**payload)
raise
except (TimedOut, NetworkError, OSError) as e:
last_exception = e
delay = min(max_delay_ms / 1000, (min_delay_ms / 1000) * (2**attempt))
delay += random.uniform(0, delay * jitter)
logger.warning(f"[Telegram] Send retry {attempt + 1}/{max_retries} after {delay:.1f}s: {e}")
await asyncio.sleep(delay)
if last_exception:
raise last_exception
raise TelegramError("Send failed after retries")
async def send_stream_edit(
application: Application,
chat_id: str,
msg_id: str,
chunk: str,
config: dict[str, Any] | None = None,
) -> Any:
html = markdown_to_html(chunk)
try:
await application.bot.edit_message_text(
chat_id=chat_id,
message_id=int(msg_id),
text=html,
parse_mode="HTML",
)
except BadRequest as e:
if "can't parse entities" in str(e).lower():
clean_text = strip_all_tags(html)
try:
await application.bot.edit_message_text(
chat_id=chat_id,
message_id=int(msg_id),
text=clean_text,
)
except Exception:
pass
else:
logger.debug(f"[Telegram] Stream edit skipped (BadRequest): {e}")
except RetryAfter as e:
await asyncio.sleep(e.retry_after)
except Exception as e:
logger.debug(f"[Telegram] Stream edit error: {e}")
async def send_with_format_fallback(
bot: Bot,
chat_id: str,
text: str,
**kwargs,
) -> Any:
try:
return await bot.send_message(
chat_id=chat_id,
text=text,
parse_mode="HTML",
**kwargs,
)
except BadRequest as e:
if "can't parse entities" in str(e).lower():
clean_text = strip_all_tags(text)
logger.warning(f"[Telegram] HTML parse failed, falling back to plain text for chat {chat_id}")
return await bot.send_message(
chat_id=chat_id,
text=clean_text,
**kwargs,
)
raise
async def send_media_with_retry(
bot: Bot,
chat_id: str,
media_type: str,
media_data: Any,
caption: str | None = None,
config: dict[str, Any] | None = None,
force_document: bool = False,
) -> Any:
max_retries = 3
last_exception = None
for attempt in range(max_retries):
try:
kwargs = {}
if caption:
kwargs["caption"] = caption
kwargs["parse_mode"] = "HTML"
if media_type == "image":
if force_document:
return await bot.send_document(chat_id=chat_id, document=media_data, **kwargs)
return await bot.send_photo(chat_id=chat_id, photo=media_data, **kwargs)
elif media_type == "video":
return await bot.send_video(chat_id=chat_id, video=media_data, **kwargs)
elif media_type == "audio":
return await bot.send_audio(chat_id=chat_id, audio=media_data, **kwargs)
elif media_type == "file":
return await bot.send_document(chat_id=chat_id, document=media_data, **kwargs)
else:
raise ValueError(f"Unsupported media type: {media_type}")
except RetryAfter as e:
await asyncio.sleep(e.retry_after)
last_exception = e
except (TimedOut, NetworkError, OSError) as e:
last_exception = e
await asyncio.sleep(1.0 * (2**attempt))
if last_exception:
raise last_exception
raise TelegramError("Media send failed after retries")
def build_reply_parameters(
response: Any,
reply_mode: str,
) -> dict[str, Any] | None:
reply_to_id = None
if hasattr(response, "reply_to_message_id"):
reply_to_id = response.reply_to_message_id
elif isinstance(response, dict):
reply_to_id = response.get("reply_to_message_id")
if reply_mode == "off":
return None
if reply_mode == "first":
if reply_to_id:
return {"reply_to_message_id": reply_to_id, "message_thread_id": response.metadata.get("thread_id")}
return None
if reply_mode in ("all", "batched"):
if reply_to_id:
return {"reply_to_message_id": reply_to_id, "message_thread_id": response.metadata.get("thread_id")}
return None
return None
def build_quote_payload(response: Any) -> dict[str, Any] | None:
quote_text = None
if hasattr(response, "quote_text") and response.quote_text:
quote_text = response.quote_text
elif isinstance(response, dict) and response.get("quote_text"):
quote_text = response["quote_text"]
if not quote_text:
return None
reply_to_id = None
if hasattr(response, "reply_to_message_id"):
reply_to_id = response.reply_to_message_id
elif isinstance(response, dict):
reply_to_id = response.get("reply_to_message_id")
if not reply_to_id:
return None
params: dict[str, Any] = {
"message_id": int(reply_to_id),
}
if quote_text:
params["quote"] = quote_text[:256]
return {"reply_parameters": ReplyParameters(**params)}

View File

@ -0,0 +1,19 @@
from __future__ import annotations
from yuxi.channels.models import ChatType
def resolve_chat_type(telegram_chat_type: str) -> ChatType:
if telegram_chat_type == "private":
return ChatType.DIRECT
if telegram_chat_type in ("group", "supergroup"):
return ChatType.GROUP
if telegram_chat_type == "channel":
return ChatType.GUILD_CHANNEL
return ChatType.GROUP
def resolve_thread_key(chat_id: str, thread_id: str | None = None) -> str:
if thread_id:
return f"telegram:{chat_id}:topic:{thread_id}"
return f"telegram:{chat_id}"

View File

@ -0,0 +1,51 @@
from __future__ import annotations
from typing import Any
from telegram.ext import ApplicationBuilder
async def probe_token(token: str) -> dict[str, Any]:
try:
app = ApplicationBuilder().token(token).build()
bot_info = await app.bot.get_me()
await app.shutdown()
return {
"status": "ok",
"bot_id": bot_info.id,
"bot_username": bot_info.username,
"bot_name": f"{bot_info.first_name or ''} {bot_info.last_name or ''}".strip(),
"can_join_groups": bot_info.can_join_groups,
"can_read_all_group_messages": bot_info.can_read_all_group_messages,
"supports_inline_queries": bot_info.supports_inline_queries,
}
except Exception as e:
return {"status": "error", "message": str(e)}
def generate_config_schema() -> dict[str, Any]:
return {
"bot_token": {"type": "str", "required": True, "description": "Telegram Bot Token from @BotFather"},
"api_root": {"type": "str", "required": False, "description": "Custom API root URL"},
"proxy": {"type": "str", "required": False, "description": "HTTP proxy URL"},
"webhook_url": {"type": "str", "required": False, "description": "Webhook URL for receiving updates"},
"webhook_path": {"type": "str", "required": False, "default": "/telegram-webhook"},
"webhook_secret": {"type": "str", "required": False, "description": "Webhook secret token"},
"webhook_cert_path": {"type": "str", "required": False, "description": "Path to SSL certificate"},
"webhook_host": {"type": "str", "required": False, "default": "0.0.0.0"},
"webhook_port": {"type": "int", "required": False, "default": 8443},
"poll_interval": {"type": "float", "required": False, "default": 1.0},
"poll_timeout": {"type": "int", "required": False, "default": 10},
"allowed_updates": {"type": "list[str]", "required": False, "description": "List of update types to receive"},
"reaction_level": {"type": "str", "required": False, "default": "minimal", "enum": ["minimal", "ack", "off"]},
"reaction_notifications": {"type": "str", "required": False, "default": "all", "enum": ["off", "own", "all"]},
"silent": {"type": "bool", "required": False, "default": False, "description": "Send messages silently"},
"exec_approvals": {
"type": "dict",
"required": False,
"schema": {
"enabled": {"type": "bool", "default": False},
"approvers": {"type": "list[str]", "default": []},
},
},
}

View File

@ -0,0 +1,57 @@
from __future__ import annotations
from typing import Any
from yuxi.utils.logging_config import logger
from yuxi.channels.adapters.telegram.setup.setup_core import probe_token
async def run_setup_wizard(config_dir: str | None = None) -> dict[str, Any]:
logger.info("[Telegram] Starting setup wizard")
token = input("Enter your Telegram Bot Token (from @BotFather): ").strip()
if not token:
return {"status": "cancelled", "message": "No token provided"}
logger.info("[Telegram] Verifying token...")
result = await probe_token(token)
if result["status"] != "ok":
logger.error(f"[Telegram] Token verification failed: {result.get('message', 'unknown error')}")
return {"status": "error", "message": result.get("message", "Token verification failed")}
logger.info(f"[Telegram] Bot verified: @{result['bot_username']} (ID: {result['bot_id']})")
mode = input("Choose mode - [p]olling or [w]ebhook (default: polling): ").strip().lower()
mode = mode if mode in ("p", "w") else "p"
config: dict[str, Any] = {
"bot_token": token,
"dm_policy": "pairing",
"group_policy": "allowlist",
}
if mode == "w":
webhook_url = input("Enter webhook URL (e.g. https://your-domain.com): ").strip()
if webhook_url:
config["webhook_url"] = webhook_url
config["webhook_path"] = input("Webhook path (default: /telegram-webhook): ").strip() or "/telegram-webhook"
secret = input("Webhook secret token (optional, press Enter to skip): ").strip()
if secret:
config["webhook_secret"] = secret
else:
poll_interval = input("Poll interval in seconds (default: 1.0): ").strip()
if poll_interval:
config["poll_interval"] = float(poll_interval)
proxies = input("Proxy URL (optional, press Enter to skip): ").strip()
if proxies:
config["proxy"] = proxies
logger.info(f"[Telegram] Setup complete for @{result['bot_username']}")
return {
"status": "ok",
"bot_username": result["bot_username"],
"bot_id": result["bot_id"],
"mode": "webhook" if mode == "w" else "polling",
"config": config,
}

View File

@ -0,0 +1,59 @@
from __future__ import annotations
from typing import Any
from yuxi.utils.logging_config import logger
class SetupContract:
EMPTY_GIF_CONTENT = b"R0lGODlhAQABAIAAAAAAAP///yH5BAEAAAAALAAAAAABAAEAAAIBRAA7"
RESPONSE_TIMEOUT_S = 3.0
PIPE_TERMINATE_ID = "telegram:pipe:terminate"
ACK_TIMEOUT_S = 1.0
def __init__(self, config: dict[str, Any] | None = None):
cfg = config or {}
self._bot_display_name: str | None = None
self._bot_username: str | None = None
self._setup_completed = False
async def initialize_setup(self, bot) -> bool:
try:
me = await bot.get_me()
self._bot_username = me.username or ""
self._bot_display_name = me.first_name or self._bot_username
logger.info(f"[Telegram] Setup initialized for @{self._bot_username}")
return True
except Exception as e:
logger.error(f"[Telegram] Setup initialization failed: {e}")
return False
@property
def bot_display_name(self) -> str:
return self._bot_display_name or "Telegram Bot"
@property
def bot_username(self) -> str | None:
return self._bot_username
@property
def setup_completed(self) -> bool:
return self._setup_completed
def mark_setup_completed(self) -> None:
self._setup_completed = True
def get_setup_status(self) -> dict[str, Any]:
return {
"initialized": self._setup_completed,
"bot_display_name": self._bot_display_name,
"bot_username": self._bot_username,
"response_timeout_s": self.RESPONSE_TIMEOUT_S,
"ack_timeout_s": self.ACK_TIMEOUT_S,
}
@staticmethod
def create_empty_gif() -> bytes:
import base64
return base64.b64decode(SetupContract.EMPTY_GIF_CONTENT)

View File

@ -0,0 +1,47 @@
from __future__ import annotations
from dataclasses import dataclass, field
@dataclass
class DraftStreamState:
chat_id: str
content: str = ""
draft_chunks: list[str] = field(default_factory=list)
final_message_id: str | None = None
is_complete: bool = False
class DraftStream:
def __init__(self, max_draft_chunks: int = 100):
self._states: dict[str, DraftStreamState] = {}
self._max_chunks = max_draft_chunks
def start_draft(self, chat_id: str) -> DraftStreamState:
state = DraftStreamState(chat_id=chat_id)
self._states[chat_id] = state
return state
def add_chunk(self, chat_id: str, chunk: str) -> DraftStreamState | None:
state = self._states.get(chat_id)
if not state:
return None
state.content += chunk
state.draft_chunks.append(chunk)
if len(state.draft_chunks) > self._max_chunks:
state.draft_chunks.pop(0)
return state
def finalize(self, chat_id: str, message_id: str) -> DraftStreamState | None:
state = self._states.get(chat_id)
if not state:
return None
state.final_message_id = message_id
state.is_complete = True
return state
def get_state(self, chat_id: str) -> DraftStreamState | None:
return self._states.get(chat_id)
def cleanup(self, chat_id: str) -> None:
self._states.pop(chat_id, None)

View File

@ -0,0 +1,84 @@
from __future__ import annotations
from dataclasses import dataclass, field
from enum import StrEnum
from typing import Any
class LaneType(StrEnum):
MAIN = "main"
REASONING = "reasoning"
TOOL_CALL = "tool_call"
ERROR = "error"
@dataclass
class LaneChunk:
lane: LaneType
content: str
metadata: dict[str, Any] = field(default_factory=dict)
class LaneDelivery:
def __init__(self, config: dict[str, Any] | None = None):
cfg = config or {}
self._reasoning_enabled = cfg.get("streaming", {}).get("reasoning_lane", False)
self._lane_separator = "\n\n---\n\n"
def is_reasoning_enabled(self) -> bool:
return self._reasoning_enabled
def extract_lane(self, content: str) -> LaneType:
if content.startswith("THINK:"):
return LaneType.REASONING
if content.startswith("TOOL:"):
return LaneType.TOOL_CALL
if content.startswith("ERROR:"):
return LaneType.ERROR
return LaneType.MAIN
def merge_lanes(self, chunks: list[LaneChunk]) -> str:
main_parts = []
reasoning_parts = []
for chunk in chunks:
if chunk.lane == LaneType.REASONING:
reasoning_parts.append(chunk.content)
else:
main_parts.append(chunk.content)
if reasoning_parts and main_parts:
return "".join(reasoning_parts) + self._lane_separator + "".join(main_parts)
elif reasoning_parts:
return "".join(reasoning_parts)
else:
return "".join(main_parts)
def split_lane_aware(self, content: str) -> list[LaneChunk]:
chunks = []
current_lane = LaneType.MAIN
current_content = ""
for line in content.split("\n"):
if line.startswith("THINK:"):
if current_content:
chunks.append(LaneChunk(lane=current_lane, content=current_content))
current_lane = LaneType.REASONING
current_content = line[6:] + "\n"
elif line.startswith("TOOL:"):
if current_content:
chunks.append(LaneChunk(lane=current_lane, content=current_content))
current_lane = LaneType.TOOL_CALL
current_content = line[5:] + "\n"
elif line.startswith("ERROR:"):
if current_content:
chunks.append(LaneChunk(lane=current_lane, content=current_content))
current_lane = LaneType.ERROR
current_content = line[6:] + "\n"
else:
current_content += line + "\n"
if current_content:
chunks.append(LaneChunk(lane=current_lane, content=current_content))
return chunks

View File

@ -0,0 +1,58 @@
from __future__ import annotations
from dataclasses import dataclass, field
from typing import Any
@dataclass
class ReasoningState:
chat_id: str
message_id: str | None = None
reasoning_text: str = ""
main_text: str = ""
is_complete: bool = False
metadata: dict[str, Any] = field(default_factory=dict)
class ReasoningLaneCoordinator:
def __init__(self):
self._states: dict[str, ReasoningState] = {}
def start_reasoning(self, chat_id: str) -> ReasoningState:
state = ReasoningState(chat_id=chat_id)
self._states[chat_id] = state
return state
def append_reasoning(self, chat_id: str, text: str) -> ReasoningState | None:
state = self._states.get(chat_id)
if not state:
return None
state.reasoning_text += text
return state
def append_main(self, chat_id: str, text: str) -> ReasoningState | None:
state = self._states.get(chat_id)
if not state:
return None
state.main_text += text
return state
def complete(self, chat_id: str) -> ReasoningState | None:
state = self._states.get(chat_id)
if state:
state.is_complete = True
return state
def get_state(self, chat_id: str) -> ReasoningState | None:
return self._states.get(chat_id)
def cleanup(self, chat_id: str) -> None:
self._states.pop(chat_id, None)
def get_combined_text(self, chat_id: str) -> str | None:
state = self._states.get(chat_id)
if not state:
return None
if state.reasoning_text and state.main_text:
return f"\ud83e\udde0 *思考过程:*\n{state.reasoning_text}\n\n---\n\n{state.main_text}"
return state.main_text or state.reasoning_text

View File

@ -0,0 +1,275 @@
from __future__ import annotations
import asyncio
import re
import time
from typing import Any
from telegram import Bot
from telegram.error import BadRequest, RetryAfter
from yuxi.channels.adapters.telegram.format import markdown_to_html, strip_all_tags
from yuxi.channels.adapters.telegram.stream.lane_delivery import LaneDelivery, LaneType
from yuxi.channels.adapters.telegram.stream.reasoning_lane_coordinator import ReasoningLaneCoordinator
from yuxi.utils.logging_config import logger
_STREAM_STORAGE: dict[str, dict] = {}
_STREAM_TTL_SECONDS = 300
_lane_delivery = LaneDelivery()
_reasoning_coordinator = ReasoningLaneCoordinator()
_BREAK_BOUNDARY_SENTENCE = re.compile(r"[。!?.!?\n]")
_BREAK_BOUNDARY_WORD = re.compile(r"\s")
def streaming_mode_from_config(config: dict[str, Any]) -> str:
streaming_cfg = config.get("streaming", {})
if isinstance(streaming_cfg, dict):
return streaming_cfg.get("mode", config.get("streaming_mode", "partial"))
if isinstance(streaming_cfg, str):
return streaming_cfg
return config.get("streaming_mode", "partial")
def _resolve_stream_config(config: dict[str, Any]) -> dict[str, Any]:
streaming_cfg = config.get("streaming", {})
if not isinstance(streaming_cfg, dict):
streaming_cfg = {}
preview_cfg = streaming_cfg.get("preview", {})
if not isinstance(preview_cfg, dict):
preview_cfg = {}
chunk_cfg = preview_cfg.get("chunk", {})
if not isinstance(chunk_cfg, dict):
chunk_cfg = {}
block_cfg = streaming_cfg.get("block", {})
if not isinstance(block_cfg, dict):
block_cfg = {}
return {
"preview_chunk_min_chars": chunk_cfg.get("min_chars", chunk_cfg.get("minChars", 50)),
"preview_chunk_max_chars": chunk_cfg.get("max_chars", chunk_cfg.get("maxChars", 200)),
"preview_chunk_break_preference": chunk_cfg.get(
"break_preference", chunk_cfg.get("breakPreference", "newline")
),
"preview_tool_progress": preview_cfg.get("tool_progress", preview_cfg.get("toolProgress", True)),
"block_enabled": block_cfg.get("enabled", True),
"block_coalesce": block_cfg.get("coalesce", False),
}
def _cleanup_expired(now: float) -> None:
expired = [key for key, state in _STREAM_STORAGE.items() if now - state.get("created_at", 0) > _STREAM_TTL_SECONDS]
for key in expired:
del _STREAM_STORAGE[key]
def _format_block(content: str, max_chars: int = 4096) -> str:
if len(content) <= max_chars:
return content
return content[: max_chars - 4] + "..."
def _format_preview_chunk(content: str, stream_cfg: dict[str, Any]) -> str:
min_chars = stream_cfg.get("preview_chunk_min_chars", 50)
max_chars = stream_cfg.get("preview_chunk_max_chars", 200)
break_pref = stream_cfg.get("preview_chunk_break_preference", "newline")
if len(content) <= max_chars:
return content
if break_pref == "any":
return content[:max_chars] + "..."
lo = max(min_chars, max_chars // 2)
hi = max_chars
if break_pref == "sentence":
pattern = _BREAK_BOUNDARY_SENTENCE
elif break_pref == "word":
pattern = _BREAK_BOUNDARY_WORD
else:
pattern = re.compile(r"\n")
candidates = [m.start() for m in pattern.finditer(content, lo, hi)]
if candidates:
return content[: candidates[-1] + 1].rstrip() + "..."
return content[:max_chars] + "..."
def _format_progress(content: str, state: dict) -> str:
counter = state.get("edit_counter", 0) + 1
state["edit_counter"] = counter
dots = "." * ((counter % 3) + 1)
if len(content) > 3800:
content = content[:3800]
return f"{content}\n\n\u23f3 {dots}"
def _format_tool_progress(content: str, stream_cfg: dict[str, Any]) -> str:
if not stream_cfg.get("preview_tool_progress", True):
return content
return f"\U0001f6e0\ufe0f {content}"
async def stream_send_chunked(
bot: Bot,
chat_id: str,
content: str,
message_id: str | None = None,
finished: bool = False,
config: dict[str, Any] | None = None,
) -> str | None:
config = config or {}
stream_cfg = _resolve_stream_config(config)
mode = streaming_mode_from_config(config)
edit_interval = config.get("stream_edit_interval_ms", 500) / 1000
stream_key = f"{chat_id}:{message_id}" if message_id else None
state = _STREAM_STORAGE.get(stream_key) if stream_key else None
_cleanup_expired(time.monotonic())
if finished and state:
del _STREAM_STORAGE[stream_key]
if state.get("message_id"):
try:
final_content = content
if mode == "progress":
final_content = _format_progress(content, state).rstrip("\n\u23f3 .")
elif mode == "block" and stream_cfg.get("block_coalesce"):
final_content = state.get("accumulated", "") + content
formatted = markdown_to_html(final_content)
await bot.edit_message_text(
chat_id=chat_id,
message_id=state["message_id"],
text=formatted,
parse_mode="HTML",
)
except BadRequest:
try:
clean = strip_all_tags(content)
await bot.edit_message_text(
chat_id=chat_id,
message_id=state["message_id"],
text=clean,
)
except Exception:
pass
if state.get("reasoning_state"):
combined = _reasoning_coordinator.get_combined_text(chat_id)
if combined and state["reasoning_state"] != stream_key:
_reasoning_coordinator.cleanup(chat_id)
return None
if not message_id:
try:
if mode == "block":
if stream_cfg.get("block_coalesce"):
preview = _format_preview_chunk(content, stream_cfg)
else:
preview = _format_block(content, 100) + "..." if len(content) > 100 else content
elif mode == "progress":
preview = content[:50] + "..." if len(content) > 50 else content
else:
preview = _format_preview_chunk(content, stream_cfg)
msg = await bot.send_message(chat_id=chat_id, text=preview)
now = time.monotonic()
_STREAM_STORAGE[f"{chat_id}:{str(msg.message_id)}"] = {
"message_id": msg.message_id,
"last_edit": now,
"counter": 0,
"edit_counter": 0,
"created_at": now,
"mode": mode,
"accumulated": "",
}
return str(msg.message_id)
except Exception:
logger.exception("[Telegram] Failed to create stream placeholder")
return None
if not state:
return message_id
if stream_cfg.get("block_coalesce"):
state["accumulated"] = (state.get("accumulated", "") + "\n" + content).strip()
return message_id
counter = state.get("counter", 0)
now = time.monotonic()
if counter < 3 and (now - state.get("last_edit", 0)) < edit_interval:
state["counter"] = counter + 1
return message_id
state["counter"] = 0
state["last_edit"] = now
display_content = content
if mode == "block":
display_content = _format_block(content)
elif mode == "progress":
display_content = _format_progress(content, state)
try:
formatted = markdown_to_html(display_content)
await bot.edit_message_text(
chat_id=chat_id,
message_id=state["message_id"],
text=formatted,
parse_mode="HTML",
)
except RetryAfter as e:
await asyncio.sleep(e.retry_after)
except BadRequest:
pass
except Exception:
logger.exception(f"[Telegram] Stream edit failed for {chat_id}")
return message_id
async def stream_send_with_lane(
bot: Bot,
chat_id: str,
content: str,
message_id: str | None = None,
finished: bool = False,
config: dict[str, Any] | None = None,
) -> str | None:
config = config or {}
lane_enabled = (
config.get("streaming", {}).get("lane_enabled", False) if isinstance(config.get("streaming"), dict) else False
)
if not lane_enabled:
return await stream_send_chunked(bot, chat_id, content, message_id, finished, config)
reasoning_enabled = _lane_delivery.is_reasoning_enabled()
if reasoning_enabled:
lane = _lane_delivery.extract_lane(content)
if lane == LaneType.REASONING:
state = _reasoning_coordinator.get_state(chat_id)
if not state:
_reasoning_coordinator.start_reasoning(chat_id)
_reasoning_coordinator.append_reasoning(chat_id, content)
if not message_id:
return await stream_send_chunked(bot, chat_id, content, None, False, config)
return message_id
if lane == LaneType.MAIN:
combined = _reasoning_coordinator.get_combined_text(chat_id)
if combined:
content = combined
if finished:
_reasoning_coordinator.complete(chat_id)
return await stream_send_chunked(bot, chat_id, content, message_id, finished, config)

View File

@ -0,0 +1,36 @@
from __future__ import annotations
def strip_internal_prefixes(chat_id: str) -> str:
cleaned = chat_id.strip()
prefixes = ["tg:", "tg://", "telegram:", "telegram://"]
for prefix in prefixes:
if cleaned.lower().startswith(prefix):
cleaned = cleaned[len(prefix) :]
break
if cleaned.lstrip("-").isdigit():
return str(int(cleaned))
return cleaned
def normalize_chat_id(chat_id: str | int) -> str:
return str(chat_id).removeprefix("-100")
def normalize_user_id(user_id: str | int) -> str:
return str(user_id)
def normalize_group_id(chat_id: str | int) -> str:
sid = str(chat_id)
if sid.startswith("-100"):
return sid[4:]
if sid.startswith("-"):
return sid[1:]
return sid
def format_chat_identifier(chat_id: str, chat_type: str, username: str | None = None) -> str:
if username:
return f"@{username}"
return normalize_group_id(chat_id)

View File

@ -0,0 +1,23 @@
from __future__ import annotations
from typing import Any
from yuxi.utils.logging_config import logger
async def handle_target_writeback(adapter, old_chat_id: str, new_chat_id: str) -> dict[str, Any]:
logger.info(f"[Telegram] Target writeback: {old_chat_id} -> {new_chat_id}")
config = adapter.config
monitored = config.get("monitored_chats", [])
if old_chat_id in monitored:
idx = monitored.index(old_chat_id)
monitored[idx] = new_chat_id
logger.info(f"[Telegram] Updated monitored_chats[{idx}]: {old_chat_id} -> {new_chat_id}")
groups = config.get("groups", {})
if old_chat_id in groups:
groups[new_chat_id] = groups.pop(old_chat_id)
logger.info(f"[Telegram] Updated groups config: {old_chat_id} -> {new_chat_id}")
return {"old_chat_id": old_chat_id, "new_chat_id": new_chat_id, "updated": True}

View File

@ -0,0 +1,48 @@
from __future__ import annotations
from typing import Any
def parse_telegram_target(target: str) -> dict[str, Any] | None:
target = target.strip()
if target.startswith("tg://"):
target = target[5:]
elif target.startswith("tg:"):
target = target[3:]
elif target.startswith("telegram://"):
target = target[11:]
elif target.startswith("telegram:"):
target = target[9:]
if target.lstrip("-").isdigit():
return {"chat_id": target, "type": "chat_id"}
if target.startswith("@"):
return {"username": target[1:], "type": "username"}
return None
def parse_messaging_target(target: str) -> dict[str, Any] | None:
parts = target.split("/")
if len(parts) >= 2:
chat_part = parse_telegram_target(parts[0])
if chat_part and parts[1].isdigit():
return {**chat_part, "thread_id": parts[1]}
return parse_telegram_target(target)
def normalize_telegram_chat_id(chat_id: str | int) -> str:
return str(chat_id).removeprefix("-100")
def normalize_telegram_target(target: str) -> str:
parsed = parse_telegram_target(target)
if not parsed:
return target
if "chat_id" in parsed:
return f"tg:{normalize_telegram_chat_id(parsed['chat_id'])}"
elif "username" in parsed:
return f"tg:@{parsed['username']}"
return target

View File

@ -0,0 +1,81 @@
from __future__ import annotations
import json
import time
from pathlib import Path
from typing import Any
from yuxi.utils.logging_config import logger
class ThreadPersistence:
def __init__(self, storage_path: str | None = None):
self._storage_path = storage_path
self._bindings: dict[str, dict[str, Any]] = {}
self._load()
def _load(self) -> None:
if not self._storage_path:
return
try:
path = Path(self._storage_path)
if path.exists():
self._bindings = json.loads(path.read_text(encoding="utf-8"))
logger.debug(f"[Telegram] Loaded {len(self._bindings)} thread bindings")
except Exception as e:
logger.warning(f"[Telegram] Failed to load thread bindings: {e}")
self._bindings = {}
def _save(self) -> None:
if not self._storage_path:
return
try:
path = Path(self._storage_path)
path.parent.mkdir(parents=True, exist_ok=True)
path.write_text(json.dumps(self._bindings, ensure_ascii=False, indent=2), encoding="utf-8")
except Exception as e:
logger.warning(f"[Telegram] Failed to save thread bindings: {e}")
def bind_channel_to_thread(
self,
channel_id: str,
thread_id: str,
metadata: dict[str, Any] | None = None,
) -> None:
entry = {
"channel_id": channel_id,
"thread_id": thread_id,
"bound_at": time.time(),
"metadata": metadata or {},
}
self._bindings[f"channel:{channel_id}"] = entry
self._bindings[f"thread:{thread_id}"] = entry
self._save()
logger.debug(f"[Telegram] Bound channel {channel_id} -> thread {thread_id}")
def get_thread_for_channel(self, channel_id: str) -> str | None:
binding = self._bindings.get(f"channel:{channel_id}")
return binding["thread_id"] if binding else None
def get_channel_for_thread(self, thread_id: str) -> str | None:
binding = self._bindings.get(f"thread:{thread_id}")
return binding["channel_id"] if binding else None
def unbind_channel(self, channel_id: str) -> None:
binding = self._bindings.pop(f"channel:{channel_id}", None)
if binding:
self._bindings.pop(f"thread:{binding['thread_id']}", None)
self._save()
def get_all_bindings(self) -> list[dict[str, Any]]:
seen = set()
result = []
for key, binding in self._bindings.items():
if key.startswith("channel:"):
result.append(binding)
seen.add(key)
return result
def clear(self) -> None:
self._bindings.clear()
self._save()

View File

@ -0,0 +1,125 @@
from __future__ import annotations
import asyncio
import time
from typing import Any
from yuxi.utils.logging_config import logger
class ThreadBinding:
def __init__(
self,
thread_key: str,
subagent_id: str | None = None,
acp_binding: str | None = None,
idle_timeout_ms: int = 86_400_000,
):
self.thread_key = thread_key
self.subagent_id = subagent_id
self.acp_binding = acp_binding
self.idle_timeout_ms = idle_timeout_ms
self.last_activity: float = time.monotonic()
self.created_at: float = time.monotonic()
def touch(self) -> None:
self.last_activity = time.monotonic()
def is_expired(self) -> bool:
elapsed = (time.monotonic() - self.last_activity) * 1000
return elapsed > self.idle_timeout_ms
class ThreadBindingManager:
def __init__(self, config: dict[str, Any] | None = None):
cfg = config or {}
bindings_cfg = cfg.get("thread_bindings", {})
self._enabled = bindings_cfg.get("enabled", False)
self._default_idle_timeout_ms = bindings_cfg.get("idle_timeout_ms", 86_400_000)
self._cleanup_interval = bindings_cfg.get("cleanup_interval", 300)
self._bindings: dict[str, ThreadBinding] = {}
self._cleanup_task: asyncio.Task | None = None
@property
def enabled(self) -> bool:
return self._enabled
def bind(
self,
thread_key: str,
subagent_id: str | None = None,
acp_binding: str | None = None,
idle_timeout_ms: int | None = None,
) -> ThreadBinding:
timeout = idle_timeout_ms or self._default_idle_timeout_ms
binding = ThreadBinding(
thread_key=thread_key,
subagent_id=subagent_id,
acp_binding=acp_binding,
idle_timeout_ms=timeout,
)
self._bindings[thread_key] = binding
logger.debug(f"[Telegram] Thread bound: {thread_key} -> subagent={subagent_id}")
return binding
def unbind(self, thread_key: str) -> bool:
if thread_key in self._bindings:
del self._bindings[thread_key]
logger.debug(f"[Telegram] Thread unbound: {thread_key}")
return True
return False
def get_binding(self, thread_key: str) -> ThreadBinding | None:
binding = self._bindings.get(thread_key)
if binding and binding.is_expired():
self.unbind(thread_key)
return None
return binding
def get_subagent_id(self, thread_key: str) -> str | None:
binding = self.get_binding(thread_key)
return binding.subagent_id if binding else None
def touch(self, thread_key: str) -> None:
binding = self._bindings.get(thread_key)
if binding:
binding.touch()
def cleanup_expired(self) -> int:
expired = [k for k, v in self._bindings.items() if v.is_expired()]
for k in expired:
self.unbind(k)
if expired:
logger.info(f"[Telegram] Cleaned up {len(expired)} expired thread bindings")
return len(expired)
async def start_cleanup_loop(self) -> None:
if not self._enabled:
return
async def _loop():
while True:
await asyncio.sleep(self._cleanup_interval)
self.cleanup_expired()
self._cleanup_task = asyncio.create_task(_loop())
async def stop_cleanup_loop(self) -> None:
if self._cleanup_task and not self._cleanup_task.done():
self._cleanup_task.cancel()
try:
await self._cleanup_task
except asyncio.CancelledError:
pass
self._cleanup_task = None
def get_all_bindings(self) -> dict[str, dict[str, Any]]:
return {
k: {
"thread_key": v.thread_key,
"subagent_id": v.subagent_id,
"acp_binding": v.acp_binding,
"idle_remaining_ms": max(0, int(v.idle_timeout_ms - (time.monotonic() - v.last_activity) * 1000)),
}
for k, v in self._bindings.items()
}

View File

@ -0,0 +1,51 @@
from __future__ import annotations
from typing import Any
POLL_TOOL_SCHEMA = {
"type": "object",
"properties": {
"question": {"type": "string", "description": "Poll question"},
"options": {
"type": "array",
"items": {"type": "string"},
"description": "Poll options (2-10 items)",
},
"is_anonymous": {"type": "boolean", "default": True},
"allows_multiple_answers": {"type": "boolean", "default": False},
"type": {"type": "string", "enum": ["regular", "quiz"], "default": "regular"},
"open_period": {"type": "integer", "description": "Seconds poll is open"},
"correct_option_id": {"type": "integer", "description": "Correct answer for quiz"},
},
"required": ["question", "options"],
}
LOCATION_TOOL_SCHEMA = {
"type": "object",
"properties": {
"latitude": {"type": "number", "description": "Latitude"},
"longitude": {"type": "number", "description": "Longitude"},
"live_period": {"type": "integer", "description": "Live location period in seconds"},
},
"required": ["latitude", "longitude"],
}
CONTACT_TOOL_SCHEMA = {
"type": "object",
"properties": {
"phone_number": {"type": "string", "description": "Phone number"},
"first_name": {"type": "string", "description": "First name"},
"last_name": {"type": "string", "description": "Last name (optional)"},
},
"required": ["phone_number", "first_name"],
}
def get_message_tool_schemas() -> dict[str, dict[str, Any]]:
return {
"send_poll": POLL_TOOL_SCHEMA,
"send_location": LOCATION_TOOL_SCHEMA,
"send_contact": CONTACT_TOOL_SCHEMA,
}

View File

@ -0,0 +1,41 @@
from __future__ import annotations
from typing import Any
DEFAULT_GLOBAL_TIMEOUT_SECONDS = 500
DEFAULT_TIMEOUTS: dict[str, float] = {
"send_message": 30.0,
"send_media": 60.0,
"edit_message": 15.0,
"delete_message": 10.0,
"get_me": 10.0,
"get_chat": 10.0,
"send_chat_action": 5.0,
"forward_message": 30.0,
"send_poll": 30.0,
"create_forum_topic": 15.0,
"set_webhook": 15.0,
"delete_webhook": 15.0,
}
def get_timeout(operation: str, config: dict[str, Any] | None = None) -> float:
cfg = config or {}
custom = cfg.get("request_timeouts", {}).get(operation)
if custom is not None:
return float(custom)
return DEFAULT_TIMEOUTS.get(operation, 30.0)
def get_timeout_config(config: dict[str, Any] | None = None) -> dict[str, float]:
cfg = config or {}
return {op: get_timeout(op, cfg) for op in DEFAULT_TIMEOUTS}
def resolve_global_timeout(config: dict[str, Any] | None = None) -> float:
cfg = config or {}
seconds = cfg.get("timeout_seconds")
if seconds is not None:
return float(seconds)
return float(DEFAULT_GLOBAL_TIMEOUT_SECONDS)

View File

@ -0,0 +1,38 @@
from __future__ import annotations
import time
from yuxi.utils.logging_config import logger
class SendChatAction401Backoff:
def __init__(self, max_backoff: float = 120.0, base_delay: float = 5.0):
self._backoff_until: float = 0.0
self._base_delay = base_delay
self._max_backoff = max_backoff
self._consecutive_failures: int = 0
async def execute(self, bot, chat_id: str, action: str = "typing") -> bool:
now = time.monotonic()
if now < self._backoff_until:
return False
try:
await bot.send_chat_action(chat_id=chat_id, action=action)
self._consecutive_failures = 0
return True
except Exception as e:
error_str = str(e).lower()
if "unauthorized" in error_str or "401" in error_str or "bot was blocked" in error_str:
self._consecutive_failures += 1
delay = min(self._base_delay * (2 ** (self._consecutive_failures - 1)), self._max_backoff)
self._backoff_until = now + delay
logger.warning(f"[Telegram] sendChatAction 401 backoff for {delay:.0f}s")
return False
def can_send(self) -> bool:
return time.monotonic() >= self._backoff_until
def reset(self) -> None:
self._backoff_until = 0.0
self._consecutive_failures = 0

View File

@ -0,0 +1,17 @@
from __future__ import annotations
def generate_sequential_key(channel_id: str, chat_id: str) -> str:
return f"{channel_id}:chat:{chat_id}"
def generate_debounce_key(channel_id: str, chat_id: str, user_id: str) -> str:
return f"{channel_id}:debounce:{chat_id}:{user_id}"
def generate_group_sequential_key(channel_id: str, group_id: str) -> str:
return f"{channel_id}:group:{group_id}"
def generate_topic_sequential_key(channel_id: str, chat_id: str, thread_id: str) -> str:
return f"{channel_id}:topic:{chat_id}:{thread_id}"

View File

@ -0,0 +1,47 @@
from __future__ import annotations
from typing import Any
class TopicRouter:
def __init__(self, config: dict[str, Any] | None = None):
cfg = config or {}
self._groups_config = cfg.get("groups", {})
self._default_agent_id = cfg.get("default_agent_id", "default")
def resolve_topic_agent(self, chat_id: str, thread_id: str) -> str:
chat_config = self._groups_config.get(chat_id, {})
topics_config = chat_config.get("topics", {})
topic_config = topics_config.get(thread_id)
if topic_config and "agent_id" in topic_config:
return topic_config["agent_id"]
group_agent = chat_config.get("agent_id")
return group_agent or self._default_agent_id
def resolve_group_agent(self, chat_id: str) -> str:
chat_config = self._groups_config.get(chat_id, {})
group_agent = chat_config.get("agent_id")
return group_agent or self._default_agent_id
def resolve_route(
self,
chat_id: str,
chat_type: str,
thread_id: str | None = None,
) -> str:
agent_id = self._resolve_agent(chat_id, chat_type, thread_id)
if chat_type == "private":
return f"agent:{agent_id}:telegram:direct:{chat_id}"
if thread_id:
return f"agent:{agent_id}:telegram:group:{chat_id}:topic:{thread_id}"
return f"agent:{agent_id}:telegram:group:{chat_id}"
def _resolve_agent(self, chat_id: str, chat_type: str, thread_id: str | None) -> str:
if chat_type == "private":
return self._default_agent_id
if thread_id:
return self.resolve_topic_agent(chat_id, thread_id)
return self.resolve_group_agent(chat_id)

View File

@ -0,0 +1,31 @@
from __future__ import annotations
import re
def generate_topic_label(content: str) -> str:
cleaned = content.strip()
if not cleaned:
return "New Topic"
first_line = cleaned.split("\n")[0].strip()
label = re.sub(r"[^\w\s\u4e00-\u9fff]", "", first_line)
label = re.sub(r"\s+", " ", label).strip()
max_chars = 30
if len(label) > max_chars:
label = label[:max_chars].rstrip() + "..."
return label or "New Topic"
def suggest_icon_color(content: str) -> int:
colors = [
0x6FB9F0,
0xFFD67E,
0xCB86DB,
0x8EEE98,
0xFF93B2,
0xFB6F5F,
]
return colors[hash(content) % len(colors)]

View File

@ -0,0 +1,35 @@
from __future__ import annotations
from yuxi.channels.models import DeliveryResult
async def manage_topic(
adapter,
chat_id: str,
action: str,
topic_id: str | None = None,
name: str | None = None,
icon_custom_emoji_id: str | None = None,
) -> DeliveryResult:
if action == "create":
if not name:
return DeliveryResult(success=False, error="Topic name is required for creation")
return await adapter.create_forum_topic(chat_id, name, icon_custom_emoji_id=icon_custom_emoji_id)
elif action == "edit":
if not topic_id:
return DeliveryResult(success=False, error="Topic ID is required for editing")
return await adapter.edit_forum_topic(chat_id, topic_id, name, icon_custom_emoji_id=icon_custom_emoji_id)
elif action == "close":
if not topic_id:
return DeliveryResult(success=False, error="Topic ID is required")
return await adapter.close_forum_topic(chat_id, topic_id)
elif action == "reopen":
if not topic_id:
return DeliveryResult(success=False, error="Topic ID is required")
return await adapter.reopen_forum_topic(chat_id, topic_id)
elif action == "delete":
if not topic_id:
return DeliveryResult(success=False, error="Topic ID is required")
return await adapter.delete_forum_topic(chat_id, topic_id)
else:
return DeliveryResult(success=False, error=f"Unknown topic action: {action}")

View File

@ -0,0 +1,34 @@
from __future__ import annotations
from collections import OrderedDict
from typing import Any
class TopicNameCache:
def __init__(self, max_size: int = 100):
self._cache: OrderedDict[str, dict[str, Any]] = OrderedDict()
self._max_size = max_size
def get(self, thread_id: str) -> dict[str, Any] | None:
if thread_id in self._cache:
self._cache.move_to_end(thread_id)
return self._cache[thread_id]
return None
def set(self, thread_id: str, data: dict[str, Any]) -> None:
if thread_id in self._cache:
self._cache.move_to_end(thread_id)
else:
if len(self._cache) >= self._max_size:
self._cache.popitem(last=False)
self._cache[thread_id] = data
def invalidate(self, thread_id: str) -> None:
self._cache.pop(thread_id, None)
def clear(self) -> None:
self._cache.clear()
def get_topic_name(self, thread_id: str) -> str:
data = self.get(thread_id)
return data.get("name", f"Topic {thread_id}") if data else f"Topic {thread_id}"

View File

@ -0,0 +1,47 @@
from __future__ import annotations
from typing import Any
from telegram import Bot, InlineKeyboardButton, InlineKeyboardMarkup
PAGE_SIZE = 5
async def send_command_ui(
bot: Bot,
chat_id: str,
commands: list[dict[str, Any]],
page: int = 0,
) -> Any:
total_pages = max(1, (len(commands) + PAGE_SIZE - 1) // PAGE_SIZE)
page = max(0, min(page, total_pages - 1))
start = page * PAGE_SIZE
page_commands = commands[start : start + PAGE_SIZE]
text = "\u231a **Available Commands**"
if total_pages > 1:
text += f" ({page + 1}/{total_pages})"
text += "\n\n"
for cmd in page_commands:
text += f"/{cmd['command']} - {cmd.get('description', '')}\n"
rows = []
if total_pages > 1:
nav_row = []
if page > 0:
nav_row.append(InlineKeyboardButton("\u25c0 Prev", callback_data=f"cmd_page:{page - 1}"))
nav_row.append(InlineKeyboardButton(f"{page + 1}/{total_pages}", callback_data="cmd_page:current"))
if page < total_pages - 1:
nav_row.append(InlineKeyboardButton("Next \u25b6", callback_data=f"cmd_page:{page + 1}"))
rows.append(nav_row)
reply_markup = InlineKeyboardMarkup(rows) if rows else None
return await bot.send_message(
chat_id=chat_id,
text=text,
parse_mode="HTML",
reply_markup=reply_markup,
)

View File

@ -0,0 +1,27 @@
from __future__ import annotations
from typing import Any
async def dispatch_tgcmd(adapter, chat_id: str, data: str, from_user_id: str) -> dict[str, Any] | None:
if not data.startswith("tgcmd:"):
return None
parts = data.split(":", 2)
if len(parts) < 3:
return None
command = parts[2]
handlers = {
"ping": lambda: {"action": "reply", "text": "pong!"},
"status": lambda: {"action": "reply", "text": f"Bot is online. Status: {adapter._status.value}"},
"help": lambda: {"action": "command_ui"},
"doctor": lambda: {"action": "doctor"},
}
handler = handlers.get(command)
if handler:
return handler()
return {"action": "unknown", "command": command}

View File

@ -0,0 +1,165 @@
from __future__ import annotations
import base64
import struct
from yuxi.utils.logging_config import logger
_IMAGE_FORMAT_MIME = {
b"\x89PNG": "image/png",
b"\xff\xd8": "image/jpeg",
b"GIF8": "image/gif",
b"RIFF": "image/webp",
b"BM": "image/bmp",
}
def _detect_image_format(data: bytes) -> str:
for magic, mime in _IMAGE_FORMAT_MIME.items():
if data[: len(magic)] == magic:
if magic == b"RIFF" and data[8:12] != b"WEBP":
continue
return mime
return "application/octet-stream"
def _extract_image_dimensions(data: bytes) -> tuple[int, int] | None:
mime = _detect_image_format(data)
try:
if mime == "image/png":
ihdr_idx = data.find(b"IHDR")
if ihdr_idx >= 0:
w, h = struct.unpack(">II", data[ihdr_idx + 4 : ihdr_idx + 12])
return w, h
elif mime == "image/jpeg":
idx = 2
while idx < len(data) - 9:
if data[idx : idx + 2] == b"\xff\xc0":
h, w = struct.unpack(">HH", data[idx + 5 : idx + 9])
return w, h
idx += struct.unpack(">H", data[idx + 2 : idx + 4])[0] + 2
elif mime == "image/gif":
w, h = struct.unpack("<HH", data[6:10])
return w, h
elif mime == "image/webp" and data[12:16] == b"VP8X":
w = int.from_bytes(data[24:27], "little") + 1
h = int.from_bytes(data[27:30], "little") + 1
return w, h
except (struct.error, ValueError, IndexError):
pass
return None
def _format_media_info(media_type: str, data: bytes) -> str:
size_kb = len(data) / 1024
if size_kb >= 1024:
size_str = f"{size_kb / 1024:.1f}MB"
else:
size_str = f"{size_kb:.0f}KB"
dims = _extract_image_dimensions(data)
type_label = {
"image": "Image",
"photo": "Photo",
"sticker": "Sticker",
"audio": "Audio",
"voice": "Voice Message",
"video": "Video",
"animation": "Animation",
"file": "File",
"document": "Document",
}.get(media_type, media_type.capitalize())
if dims:
return f"[{type_label}: {dims[0]}x{dims[1]}px, {size_str}]"
return f"[{type_label}: {size_str}]"
async def understand_media(
adapter,
file_id: str,
media_type: str,
vision_api_url: str | None = None,
) -> str | None:
try:
media_bytes = await adapter.download_media(file_id)
if media_type in ("image", "photo", "sticker"):
return await _describe_image(media_bytes, vision_api_url)
elif media_type in ("audio", "voice"):
return await _transcribe_audio(media_bytes, vision_api_url)
else:
return _format_media_info(media_type, media_bytes)
except Exception as e:
logger.error(f"[Telegram] Media understanding failed for {file_id}: {e}")
return None
async def _describe_image(image_bytes: bytes, api_url: str | None = None) -> str:
if not api_url:
return _format_media_info("image", image_bytes)
mime = _detect_image_format(image_bytes)
image_b64 = base64.b64encode(image_bytes).decode("utf-8")
try:
import aiohttp
payload = {
"model": "vision",
"messages": [
{
"role": "user",
"content": [
{"type": "text", "text": "Describe this image in detail."},
{"type": "image_url", "image_url": {"url": f"data:{mime};base64,{image_b64}"}},
],
}
],
"max_tokens": 300,
}
async with aiohttp.ClientSession() as session:
async with session.post(
api_url,
json=payload,
timeout=aiohttp.ClientTimeout(total=30),
) as resp:
if resp.status == 200:
result = await resp.json()
description = result.get("choices", [{}])[0].get("message", {}).get("content", "")
if description:
return description
except Exception as e:
logger.debug(f"[Telegram] Vision API failed: {e}")
return _format_media_info("image", image_bytes)
async def _transcribe_audio(audio_bytes: bytes, api_url: str | None = None) -> str:
if not api_url:
return _format_media_info("audio", audio_bytes)
try:
import aiohttp
audio_b64 = base64.b64encode(audio_bytes).decode("utf-8")
async with aiohttp.ClientSession() as session:
form = aiohttp.FormData()
form.add_field("audio_b64", audio_b64)
async with session.post(
api_url,
data=form,
timeout=aiohttp.ClientTimeout(total=60),
) as resp:
if resp.status == 200:
result = await resp.json()
text = result.get("text", "")
if text:
return text
except Exception as e:
logger.debug(f"[Telegram] Transcription API failed: {e}")
return _format_media_info("audio", audio_bytes)

View File

@ -0,0 +1,55 @@
from __future__ import annotations
from collections import OrderedDict
from typing import Any
from yuxi.utils.logging_config import logger
from .sticker_cache_store import StickerCacheStore
class StickerCache:
def __init__(self, max_size: int = 100):
self._cache: OrderedDict[str, dict[str, Any]] = OrderedDict()
self._max_size = max_size
def get(self, file_id: str) -> dict[str, Any] | None:
if file_id in self._cache:
self._cache.move_to_end(file_id)
return self._cache[file_id]
return None
def set(self, file_id: str, sticker_info: dict[str, Any]) -> None:
if file_id in self._cache:
self._cache.move_to_end(file_id)
else:
if len(self._cache) >= self._max_size:
self._cache.popitem(last=False)
self._cache[file_id] = sticker_info
def has(self, file_id: str) -> bool:
return file_id in self._cache
def cache_size(self) -> int:
return len(self._cache)
def clear(self) -> None:
self._cache.clear()
async def search_stickers(
query: str,
store: StickerCacheStore | None = None,
max_results: int = 10,
) -> list[dict[str, Any]]:
if store is None:
store = StickerCacheStore()
try:
results = store.search_by_description(query)
if len(results) > max_results:
results = results[:max_results]
logger.debug(f"[Telegram] Sticker search for '{query}': {len(results)} results")
return results
except Exception as e:
logger.error(f"[Telegram] Sticker search failed: {e}")
return []

View File

@ -0,0 +1,49 @@
from __future__ import annotations
import json
import os
from pathlib import Path
from typing import Any
DEFAULT_CACHE_PATH = Path("data/sticker_cache.json")
class StickerCacheStore:
def __init__(self, cache_path: str | Path | None = None):
self._path = Path(cache_path or DEFAULT_CACHE_PATH)
self._store: dict[str, dict[str, Any]] = {}
self._load()
def _load(self) -> None:
if self._path.exists():
try:
with open(self._path, encoding="utf-8") as f:
self._store = json.load(f)
except (json.JSONDecodeError, OSError):
self._store = {}
def _save(self) -> None:
self._path.parent.mkdir(parents=True, exist_ok=True)
with open(self._path, "w", encoding="utf-8") as f:
json.dump(self._store, f, ensure_ascii=False, indent=2)
def get(self, file_id: str) -> dict[str, Any] | None:
return self._store.get(file_id)
def set(self, file_id: str, data: dict[str, Any]) -> None:
self._store[file_id] = data
self._save()
def search_by_description(self, query: str) -> list[dict[str, Any]]:
results = []
query_lower = query.lower()
for file_id, info in self._store.items():
desc = info.get("description", "").lower()
if query_lower in desc:
results.append({"file_id": file_id, **info})
return results
def clear(self) -> None:
self._store.clear()
if self._path.exists():
os.remove(self._path)

View File

@ -0,0 +1,104 @@
from __future__ import annotations
import base64
import struct
from typing import Any
from yuxi.utils.logging_config import logger
_STICKER_FORMAT_MIME = {
"image/png": "PNG",
"image/webp": "WEBP",
"application/x-tgsticker": "TGS (Animated)",
"video/webm": "WEBM (Animated)",
}
def _detect_sticker_format(data: bytes) -> str:
if data[:4] == b"\x89PNG":
return "image/png"
if data[:4] == b"RIFF" and data[8:12] == b"WEBP":
return "image/webp"
if data[:4] == b"\x1a\x45\xdf\xa3":
return "video/webm"
return "application/x-tgsticker"
def _describe_sticker_heuristic(data: bytes) -> str:
mime = _detect_sticker_format(data)
fmt_name = _STICKER_FORMAT_MIME.get(mime, "Unknown")
size_bytes = len(data)
size_kb = size_bytes / 1024
if mime == "image/png":
try:
ihdr_idx = data.find(b"IHDR")
if ihdr_idx >= 0:
width, height = struct.unpack(">II", data[ihdr_idx + 4 : ihdr_idx + 12])
return f"Sticker ({fmt_name}, {width}x{height}px, {size_kb:.0f}KB)"
except (struct.error, ValueError):
pass
elif mime == "image/webp":
try:
if data[12:16] == b"VP8X":
width = int.from_bytes(data[24:27], "little") + 1
height = int.from_bytes(data[27:30], "little") + 1
return f"Sticker ({fmt_name}, {width}x{height}px, {size_kb:.0f}KB)"
except (ValueError, IndexError):
pass
return f"Sticker ({fmt_name}, {size_kb:.0f}KB)"
async def describe_sticker(
adapter,
file_id: str,
vision_api_url: str | None = None,
) -> str | None:
try:
image_bytes = await adapter.download_media(file_id)
image_b64 = base64.b64encode(image_bytes).decode("utf-8")
sticker_info: dict[str, Any] = {
"file_id": file_id,
"size_bytes": len(image_bytes),
"format": _detect_sticker_format(image_bytes),
}
if vision_api_url:
try:
import aiohttp
async with aiohttp.ClientSession() as session:
async with session.post(
vision_api_url,
json={"image_b64": image_b64, "task": "describe_sticker"},
timeout=aiohttp.ClientTimeout(total=30),
) as resp:
if resp.status == 200:
result = await resp.json()
description = result.get("description", "")
if description:
sticker_info["description"] = description
return description
except Exception:
logger.debug("[Telegram] Vision API call failed for sticker, using heuristic")
description = _describe_sticker_heuristic(image_bytes)
sticker_info["description"] = description
return description
except Exception as e:
logger.error(f"[Telegram] Failed to describe sticker {file_id}: {e}")
return None
async def batch_describe_stickers(
adapter,
file_ids: list[str],
vision_api_url: str | None = None,
) -> dict[str, str | None]:
results: dict[str, str | None] = {}
for file_id in file_ids:
results[file_id] = await describe_sticker(adapter, file_id, vision_api_url)
return results

View File

@ -0,0 +1,128 @@
from __future__ import annotations
import io
from typing import Any
from telegram import Bot
from telegram.error import TelegramError
from yuxi.utils.logging_config import logger
_VOICE_COMPATIBLE_FORMATS = frozenset({"ogg", "opus", "flac", "wav", "m4a", "mp3", "aac"})
def is_voice_compatible_audio(filename: str) -> bool:
ext = filename.rsplit(".", 1)[-1].lower() if "." in filename else ""
return ext in _VOICE_COMPATIBLE_FORMATS
async def send_voice_from_bytes(
bot: Bot,
chat_id: str,
audio_bytes: bytes,
filename: str = "voice.ogg",
duration: int | None = None,
caption: str | None = None,
**kwargs,
) -> Any:
try:
buf = io.BytesIO(audio_bytes)
buf.name = filename
return await bot.send_voice(
chat_id=chat_id,
voice=buf,
duration=duration,
caption=caption,
**kwargs,
)
except TelegramError as e:
logger.error(f"[Telegram] Failed to send voice: {e}")
raise
async def send_audio_from_bytes(
bot: Bot,
chat_id: str,
audio_bytes: bytes,
filename: str = "audio.mp3",
title: str | None = None,
performer: str | None = None,
duration: int | None = None,
caption: str | None = None,
**kwargs,
) -> Any:
try:
buf = io.BytesIO(audio_bytes)
buf.name = filename
return await bot.send_audio(
chat_id=chat_id,
audio=buf,
title=title,
performer=performer,
duration=duration,
caption=caption,
**kwargs,
)
except TelegramError as e:
logger.error(f"[Telegram] Failed to send audio: {e}")
raise
async def send_voice_or_audio(
bot: Bot,
chat_id: str,
audio_bytes: bytes,
filename: str = "audio.ogg",
as_voice: bool = False,
duration: int | None = None,
caption: str | None = None,
title: str | None = None,
performer: str | None = None,
**kwargs,
) -> Any:
if as_voice or is_voice_compatible_audio(filename):
return await send_voice_from_bytes(
bot,
chat_id,
audio_bytes,
filename=filename or "voice.ogg",
duration=duration,
caption=caption,
**kwargs,
)
return await send_audio_from_bytes(
bot,
chat_id,
audio_bytes,
filename=filename or "audio.mp3",
title=title,
performer=performer,
duration=duration,
caption=caption,
**kwargs,
)
async def send_video_note_from_bytes(
bot: Bot,
chat_id: str,
video_bytes: bytes,
filename: str = "video_note.mp4",
duration: int | None = None,
length: int | None = None,
**kwargs,
) -> Any:
try:
buf = io.BytesIO(video_bytes)
buf.name = filename
return await bot.send_video_note(
chat_id=chat_id,
video_note=buf,
duration=duration,
length=length,
**kwargs,
)
except TelegramError as e:
logger.error(f"[Telegram] Failed to send video note: {e}")
raise