214 lines
7.6 KiB
Python
214 lines
7.6 KiB
Python
"""
|
||
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() |