diff --git a/backend/python/app/api/channel_webhook.py b/backend/python/app/api/channel_webhook.py new file mode 100644 index 0000000..e62f2ec --- /dev/null +++ b/backend/python/app/api/channel_webhook.py @@ -0,0 +1,47 @@ +""" +Channel Webhook 入口 — /app/api/channel/ +任务 #935 · 2026-08-09 + +- Telegram 走 POST JSON +- secret_token 校验(头部 X-Telegram-Bot-Api-Secret-Token) +- 其他渠道后续扩展(飞书 / 钉钉加密回调 / 企业微信 AES 解密各不同,放各 adapter 自己) +""" +from __future__ import annotations +import hashlib +import hmac +import logging +from typing import Any + +from fastapi import APIRouter, HTTPException, Request + +from app.core.channel_adapter import ChannelRouter + +log = logging.getLogger(__name__) + +router = APIRouter(prefix="/app/api/channel", tags=["channel-webhook"]) + + +@router.post("/{channel_id}") +async def channel_webhook(channel_id: str, request: Request) -> dict: + r = ChannelRouter.shared() + if channel_id not in r.channels: + raise HTTPException(404, f"channel not registered: {channel_id}") + adapter = r._adapters.get(channel_id) # noqa: SLF001 (内部访问 OK) + raw = await request.body() + # secret 校验(各家 header 名不同,Telegram: X-Telegram-Bot-Api-Secret-Token) + expected_secret = getattr(adapter, "webhook_secret", "") + if expected_secret: + secret_header = request.headers.get("x-telegram-bot-api-secret-token", "") + if not secret_header or not hmac.compare_digest(secret_header, expected_secret): + raise HTTPException(401, "invalid secret") + try: + payload: Any = await request.json() + except Exception: + raise HTTPException(400, "invalid json") + # 调对应 adapter 的 receive_update(Telegram 是同步 await;其他渠道可异步) + if hasattr(adapter, "receive_update"): + import asyncio + await adapter.receive_update(payload) + else: + raise HTTPException(501, "channel has no receive_update impl") + return {"ok": True} \ No newline at end of file diff --git a/backend/python/app/core/channel_adapter.py b/backend/python/app/core/channel_adapter.py new file mode 100644 index 0000000..16fabeb --- /dev/null +++ b/backend/python/app/core/channel_adapter.py @@ -0,0 +1,214 @@ +""" +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() \ No newline at end of file diff --git a/backend/python/app/core/telegram_channel.py b/backend/python/app/core/telegram_channel.py new file mode 100644 index 0000000..4519dbf --- /dev/null +++ b/backend/python/app/core/telegram_channel.py @@ -0,0 +1,185 @@ +""" +Telegram 渠道实现 — ChannelAdapter 的 POC 实例 +任务 #935 · 2026-08-09 + +依赖:httpx(项目已用)。无 NoneBot2 / python-telegram-bot,纯 httpx 直调 Bot API。 +- Bot API:https://core.telegram.org/bots/api +- 单消息 4096 字符上限(自动分段) +- Webhook 注册:setWebhook +""" +from __future__ import annotations +import asyncio +import logging +from typing import Any + +import httpx + +from app.core.channel_adapter import ChannelAdapter, IncomingMessage + +log = logging.getLogger(__name__) + +TELEGRAM_API_BASE = "https://api.telegram.org" + + +class TelegramChannel(ChannelAdapter): + """Telegram Bot 渠道。 + + 启动: + - 调 setWebhook,把 https:///app/api/channel/telegram 注册给 Bot + - 调 deleteWebhook 清掉旧 webhook,避免双收 + + 出站: + - sendMessage API,长度超过 4096 自动分段 + - MarkdownV2 特殊字符需转义;POC 暂不启用 parse_mode(由 LLM 输出纯文本更安全) + + 入站: + - webhook 端点接收 update,转 IncomingMessage,交给 router.dispatch + """ + + channel_id = "telegram" + + def __init__( + self, + *, + bot_token: str, + webhook_url: str, + webhook_secret: str = "", + ) -> None: + self.bot_token = bot_token + self.webhook_url = webhook_url + self.webhook_secret = webhook_secret + self._api = f"{TELEGRAM_API_BASE}/bot{bot_token}" + self._client: httpx.AsyncClient | None = None + self._running = False + + async def start(self) -> None: + self._client = httpx.AsyncClient(timeout=15.0) + # 1) 清掉旧 webhook,避免双收 + try: + await self._client.post(f"{self._api}/deleteWebhook", json={"drop_pending_updates": True}) + except Exception as e: + log.warning("[telegram] deleteWebhook failed: %s", e) + # 2) 注册新 webhook + payload: dict[str, Any] = { + "url": self.webhook_url, + "allowed_updates": ["message"], + "drop_pending_updates": False, + } + if self.webhook_secret: + payload["secret_token"] = self.webhook_secret + try: + r = await self._client.post(f"{self._api}/setWebhook", json=payload) + j = r.json() if r.content else {} + if not j.get("ok"): + raise RuntimeError(f"setWebhook failed: {j}") + log.info("[telegram] webhook registered: %s", self.webhook_url) + except Exception as e: + log.warning("[telegram] start failed: %s", e, exc_info=True) + self._running = True + + async def stop(self) -> None: + self._running = False + if self._client: + try: + await self._client.post(f"{self._api}/deleteWebhook", json={}) + except Exception: + pass + await self._client.aclose() + self._client = None + + async def send(self, *, user_id: str, content: str) -> None: + """出站单条。user_id 在这里是 chat_id。""" + if not self._client: + log.warning("[telegram] send called but client not initialized") + return + # Telegram 单消息 4096 字符上限,长文本按 4000 分段(留 buffer) + chunk_size = 4000 + for i in range(0, max(len(content), 1), chunk_size): + piece = content[i:i + chunk_size] + try: + r = await self._client.post( + f"{self._api}/sendMessage", + json={"chat_id": user_id, "text": piece}, + ) + if not r.is_success: + log.warning("[telegram] sendMessage failed: %d %s", r.status_code, r.text[:200]) + except Exception as e: + log.warning("[telegram] send failed: %s", e, exc_info=True) + + async def receive_update(self, payload: dict) -> None: + """webhook 端点调用此方法。""" + msg = payload.get("message") or payload.get("edited_message") + if not msg: + return + chat = msg.get("chat") or {} + chat_type = chat.get("type", "private") + is_group = chat_type in ("group", "supergroup") + text = msg.get("text") or msg.get("caption") or "" + # 去除 /command 后传给 LLM + if text.startswith("/"): + parts = text.split(maxsplit=1) + cmd = parts[0].split("@")[0] # 去除 @botname 后缀 + if cmd == "/bind": + # /bind 由 binding_router 处理,不在这里 + await self._handle_bind(chat_id=str(chat.get("id")), args=parts[1] if len(parts) > 1 else "") + return + text = parts[1] if len(parts) > 1 else "" + if not text: + return + + incoming = IncomingMessage( + channel="telegram", + platform_user_id=str(chat.get("id")), + text=text, + is_group=is_group, + group_id=str(chat.get("id")) if is_group else None, + raw=msg, + ) + if not self.router: + log.warning("[telegram] router not bound; drop") + return + try: + reply = await self.router.dispatch(incoming) + except Exception as e: + log.exception("[telegram] dispatch failed: %s", e) + reply = "(系统繁忙,请稍后再试)" + await self.send(user_id=incoming.platform_user_id, content=reply) + + async def _handle_bind(self, *, chat_id: str, args: str) -> None: + """/bind — 解析 token,写 im_channel_routes,绑定 platform_user_id ↔ user_id。""" + from sqlalchemy.orm import Session + from app.core.db import SessionLocal + from app.models import IMChannelRoute, User + + token = (args or "").strip() + if not token: + await self.send(user_id=chat_id, content="用法:/bind <你在七虎AI后台生成的绑定码>") + return + db: Session = SessionLocal() + try: + # token = base64(user_id) 简化方案;生产可换 JWT + 有效期 + import base64 + try: + user_id = base64.urlsafe_b64decode(token.encode()).decode() + except Exception: + await self.send(user_id=chat_id, content="绑定码无效") + return + user = db.query(User).filter(User.id == user_id).first() + if user is None: + await self.send(user_id=chat_id, content="绑定码无效或用户不存在") + return + existing = db.query(IMChannelRoute).filter( + IMChannelRoute.channel == "telegram", + IMChannelRoute.platform_user_id == chat_id, + ).first() + if existing: + existing.internal_user_id = user.id + else: + db.add(IMChannelRoute(channel="telegram", platform_user_id=chat_id, internal_user_id=user.id)) + db.commit() + await self.send(user_id=chat_id, content=f"✅ 绑定成功!你现在可以与我对话。") + except Exception as e: + log.exception("[telegram.bind] failed: %s", e) + await self.send(user_id=chat_id, content="绑定失败,请稍后再试") + finally: + db.close() \ No newline at end of file diff --git a/backend/python/main.py b/backend/python/main.py index f326625..9084dea 100644 --- a/backend/python/main.py +++ b/backend/python/main.py @@ -39,14 +39,15 @@ from datetime import datetime sys.path.insert(0, os.path.dirname(os.path.dirname(os.path.abspath(__file__)))) -from app.api import auth, chat, knowledge, market, token_usage, payment, profile, notifications, assets, evolution, skills_api +from app.api import auth, chat, knowledge, market, market_products, token_usage, payment, profile, notifications, assets, evolution, skills_api from app.api.bug_audit import router as bug_audit_router from app.api.auth_verify import router as auth_verify_router from app.api.billing_webhook import router as billing_webhook_router from app.casdoor_auth import router as casdoor_router -from app.api.admin import dashboard, users, models, settings, payment_admin, services, membership_admin, ads, market_admin, logs, skills_admin, agents_admin, knowledge_admin, workflow_admin, storage_admin, automation_admin, mcp_admin, developer_admin, apps_admin, distro_admin, code_exec_admin, privacy_admin, im_bot_admin, promotion_admin, achievement_admin, tenant_admin, billing_admin, analytics_admin, observability_admin +from app.api.admin import dashboard, users, models, settings, payment_admin, services, membership_admin, ads, market_admin, market_products_admin, logs, skills_admin, agents_admin, knowledge_admin, workflow_admin, storage_admin, automation_admin, mcp_admin, developer_admin, apps_admin, distro_admin, code_exec_admin, privacy_admin, im_bot_admin, promotion_admin, achievement_admin, tenant_admin, billing_admin, analytics_admin, observability_admin from app.api.im_bot_user import router as im_bot_user_router from app.api.im_bot_platform import router as im_bot_platform_router +from app.api.channel_webhook import router as channel_webhook_router from app.api.portal import user_portal from app.api.admin.developer_tools_admin import router as developer_tools_router from app.api.admin.mcp_market_admin import router as mcp_market_router @@ -194,6 +195,31 @@ async def lifespan(app: FastAPI): # 启动个人微信接收线程(之前仅在扫码确认时启动,重启后丢失) try: from app.api.im_bot_user import start_all_wechat_receivers + # === #935 IM Bot ChannelRouter (Telegram / 飞书 / 企微 etc.) === + try: + from app.core.channel_adapter import ChannelRouter + import os + router = ChannelRouter.shared() + if os.getenv("TELEGRAM_BOT_TOKEN"): + from app.core.telegram_channel import TelegramChannel + router.register(TelegramChannel( + bot_token=os.environ["TELEGRAM_BOT_TOKEN"], + webhook_url=os.getenv("TELEGRAM_WEBHOOK_URL", ""), + webhook_secret=os.getenv("TELEGRAM_WEBHOOK_SECRET", ""), + )) + # 启动所有渠道(non-fatal) + import asyncio + try: + loop = asyncio.get_event_loop() + if loop.is_running(): + asyncio.ensure_future(router.start_all()) + else: + loop.run_until_complete(router.start_all()) + except Exception as e: + print(f"[main] channel router start_all failed (non-fatal): {e}") + except Exception as _e: + print(f"[main] channel router init failed (non-fatal): {_e}") + start_all_wechat_receivers() except Exception as e: logger.warning(f"[IM个人微信] 启动恢复失败(不影响正常服务): {e}") @@ -247,6 +273,7 @@ app.include_router(auth.router, prefix="/app/api/auth", tags=["认证"]) app.include_router(chat.router, prefix="/app/api/chat", tags=["AI聊天"]) app.include_router(knowledge.router, prefix="/app/api/knowledge", tags=["知识库"]) app.include_router(market.router, prefix="/app/api/market", tags=["应用市场"]) +app.include_router(market_products.router, prefix="/app/api/market", tags=["产品市场"]) app.include_router(token_usage.router, prefix="/app/api/token", tags=["Token驾驶舱"]) app.include_router(payment.router, prefix="/app/api/payment", tags=["付费中心"]) app.include_router(profile.router, prefix="/app/api/profile", tags=["个人中心"]) @@ -265,6 +292,7 @@ app.include_router(services.router, prefix="/app/api/admin/services", tags=["后 app.include_router(membership_admin.router, prefix="/app/api/admin/membership", tags=["后台-会员管理"]) app.include_router(ads.router, prefix="/app/api/admin/ads", tags=["后台-广告管理"]) app.include_router(market_admin.router, prefix="/app/api/admin/market", tags=["后台-应用审核"]) +app.include_router(market_products_admin.router, prefix="/app/api/admin/market-products", tags=["后台-产品审核"]) app.include_router(logs.router, prefix="/app/api/admin/logs", tags=["后台-日志风控"]) app.include_router(skills_admin.router, prefix="/app/api/admin/skills", tags=["后台-技能管理"]) @@ -285,6 +313,7 @@ app.include_router(privacy_admin.router, prefix="/app/api/admin/privacy", tags=[ app.include_router(im_bot_admin.router, prefix="/app/api/admin/im-bot", tags=["im-bot"]) app.include_router(im_bot_user_router, prefix="/app/api/im/wechat", tags=["im-wechat-user"]) app.include_router(im_bot_platform_router, prefix="/app/api/im/platform", tags=["im-platform-user"]) +app.include_router(channel_webhook_router, prefix="", tags=["channel-webhook"]) app.include_router(promotion_admin.router, prefix="/app/api/admin/promotion", tags=["promotion"]) app.include_router(tenant_admin.router, prefix="/app/api/admin/tenant", tags=["tenant"]) app.include_router(billing_admin.router, prefix="/app/api/admin/billing", tags=["billing"])