""" IM Bot Channel Adapter — 渠道插件化基础设施 任务 #935 · 2026-08-09 - ChannelAdapter: 抽象基类 - IncomingMessage: 跨渠道归一入站消息 - ChannelRouter: 单例路由,复用 _send_message_core + BufferAdapter(单引擎原则) 依赖:已存在于 app/core/ 的 capability_discovery + claw_adapters(BufferAdapter) """ from __future__ import annotations import asyncio import logging import os from abc import ABC, abstractmethod from dataclasses import dataclass from typing import Any, Awaitable, Callable, Dict, Optional log = logging.getLogger(__name__) @dataclass(frozen=True) class IncomingMessage: """跨渠道归一化入站消息。 ChannelPlugin.start() 启动后接收到 raw message,转成 IncomingMessage, 交给 ChannelRouter.dispatch,组装后回送给原渠道。 """ channel: str # "telegram" / "feishu" / "wecom" / ... platform_user_id: str # 渠道内用户/群 ID text: str # 正文(去 @-mention + 富文本转 markdown) is_group: bool = False # 群消息 group_id: Optional[str] = None at_me: bool = False # 群中是否 @ 机器人 attachments: Optional[list] = None # 图片/文件 URL(可选) raw: Any = None # 原始消息,留给高级用例 class ChannelAdapter(ABC): """IM 渠道插件契约。 实现要点: - start() 启动连接(扫码 / WebSocket / webhook / Bot) - send() 出站文本(注意各家消息长度上限,Telegram=4096 / 飞书=4000) - 收到入站消息时,内部组装为 IncomingMessage,转发给 self.router.dispatch """ @property @abstractmethod def channel_id(self) -> str: ... @abstractmethod async def start(self) -> None: ... @abstractmethod async def stop(self) -> None: ... @abstractmethod async def send(self, *, user_id: str, content: str) -> None: ... # 由 ChannelRouter 注入(register() 时反向 set) router: Optional["ChannelRouter"] = None class ChannelRouter: """单例路由。所有入站消息统一过 _send_message_core。""" _shared: Optional["ChannelRouter"] = None @classmethod def shared(cls) -> "ChannelRouter": if cls._shared is None: cls._shared = cls() return cls._shared def __init__(self) -> None: self._adapters: Dict[str, ChannelAdapter] = {} self._started: bool = False # 延迟导入(避免循环依赖) self._send_message_core = None self._SendMessageRequest = None self._BufferAdapter = None self._discover_skills = None self._discover_mcp = None self._Conversation = None self._User = None self._gen_uuid = None def register(self, adapter: ChannelAdapter) -> None: if self._started: raise RuntimeError("ChannelRouter already started; can't register more adapters") adapter.router = self self._adapters[adapter.channel_id] = adapter log.info("[router] registered channel: %s", adapter.channel_id) @property def channels(self) -> list[str]: return list(self._adapters.keys()) async def start_all(self) -> None: """main.py 的 startup 钩子调用。""" self._lazy_import() for a in self._adapters.values(): try: await a.start() except Exception as e: log.warning("[router] start %s failed: %s", a.channel_id, e, exc_info=True) self._started = True log.info("[router] started %d channels: %s", len(self._adapters), self.channels) async def stop_all(self) -> None: for a in self._adapters.values(): try: await a.stop() except Exception: pass self._started = False def _lazy_import(self) -> None: if self._send_message_core is not None: return from app.api.chat_service import _send_message_core from app.api.chat_utils import SendMessageRequest from app.core.claw_adapters import BufferAdapter from app.core.capability_discovery import discover_mcp_servers, discover_skills from app.models import Conversation, User from app.core.uuid_compat import gen_uuid # 项目内统一 UUID 生成 self._send_message_core = _send_message_core self._SendMessageRequest = SendMessageRequest self._BufferAdapter = BufferAdapter self._discover_skills = discover_skills self._discover_mcp = discover_mcp_servers self._Conversation = Conversation self._User = User self._gen_uuid = gen_uuid async def dispatch(self, incoming: IncomingMessage) -> str: """1 条入站 → 1 条回复。 复用单引擎:_send_message_core + BufferAdapter。 鉴权:从 im_channel_routes 解析 platform_user_id → internal User。 """ self._lazy_import() # 1. 解析内部 user(由 platform_user_id → user_id) from sqlalchemy.orm import Session from app.models import IMChannelRoute from app.core.db import SessionLocal db: Session = SessionLocal() try: route = db.query(IMChannelRoute).filter( IMChannelRoute.channel == incoming.channel, IMChannelRoute.platform_user_id == incoming.platform_user_id, ).first() if route is None: return "（请先在七虎AI平台绑定账号:发送 /bind <你的绑定码>）" user = db.query(self._User).filter(self._User.id == route.internal_user_id).first() if user is None: return "（绑定账号已失效,请重新绑定）" # 2. 能力发现 skills = [] try: skills = self._discover_skills(db, user.id, query=incoming.text, claw="im") or [] except Exception as e: log.warning("[router.disp] skill discovery: %s", e) mcp_servers = [] try: mcp_servers = self._discover_mcp(db, None, claw="im") or [] except Exception as e: log.warning("[router.disp] mcp discovery: %s", e) skill_ids = [ (s.get("id") if isinstance(s, dict) else getattr(s, "id", None)) for s in skills ] skill_ids = [s for s in skill_ids if s] mcp_ids = [ (m.get("id") if isinstance(m, dict) else getattr(m, "id", None)) for m in mcp_servers ] mcp_ids = [m for m in mcp_ids if m] # 3. 临时会话 conv = self._Conversation( id=self._gen_uuid(), user_id=user.id, title=f"IM-{incoming.channel}", source="im_bot", ) db.add(conv) db.commit() req = self._SendMessageRequest( content=incoming.text, skill_ids=skill_ids or None, mcp_ids=mcp_ids or None, use_tools=bool(skill_ids or mcp_ids), agent_ids=None, knowledge_base_ids=None, attachments=None, ) adapter = self._BufferAdapter(tenant_id=str(getattr(user, "tenant_id", "") or "")) await self._send_message_core(conv.id, req, db, user, stream=True, on_event=adapter) reply = adapter.assemble() return reply or "（生成失败,请稍后再试）" except Exception as e: log.exception("[router.disp] failed: %s", e) return f"（系统繁忙:{e}）" finally: db.close()