From 904ab9a6c2505ce2e9573e1961337f02ddb04b31 Mon Sep 17 00:00:00 2001 From: Kris <2893855659@qq.com> Date: Thu, 21 May 2026 11:46:45 +0800 Subject: [PATCH] =?UTF-8?q?feat(channel):=20=E6=B7=BB=E5=8A=A0=E6=B7=98?= =?UTF-8?q?=E5=AE=9D=E6=B8=A0=E9=81=93=E6=89=A9=E5=B1=95?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 新增淘宝（Taobao）渠道扩展，支持在 Yuxi 平台中集成淘宝电商客服渠道。 包含以下功能模块： - client: 淘宝 API 客户端封装 - config: 渠道配置管理 - gateway: SSE/WebSocket 网关接入 - webhook: Webhook 事件处理 - outbound: 外发消息管理 - message: 消息处理 - pairing: 用户配对与绑定 - security: 安全校验 - crypto: 加解密处理 - status: 会话状态管理 - tools: Agent 工具集成 - types: 类型定义 --- .../channel/extensions/taobao/__init__.py | 286 ++++++++++++++++++ .../yuxi/channel/extensions/taobao/client.py | 197 ++++++++++++ .../yuxi/channel/extensions/taobao/config.py | 109 +++++++ .../yuxi/channel/extensions/taobao/crypto.py | 63 ++++ .../yuxi/channel/extensions/taobao/gateway.py | 105 +++++++ .../yuxi/channel/extensions/taobao/message.py | 85 ++++++ .../channel/extensions/taobao/outbound.py | 89 ++++++ .../yuxi/channel/extensions/taobao/pairing.py | 72 +++++ .../channel/extensions/taobao/plugin.json | 31 ++ .../channel/extensions/taobao/security.py | 50 +++ .../yuxi/channel/extensions/taobao/status.py | 64 ++++ .../yuxi/channel/extensions/taobao/tools.py | 257 ++++++++++++++++ .../yuxi/channel/extensions/taobao/types.py | 76 +++++ .../yuxi/channel/extensions/taobao/webhook.py | 127 ++++++++ 14 files changed, 1611 insertions(+) create mode 100644 backend/package/yuxi/channel/extensions/taobao/__init__.py create mode 100644 backend/package/yuxi/channel/extensions/taobao/client.py create mode 100644 backend/package/yuxi/channel/extensions/taobao/config.py create mode 100644 backend/package/yuxi/channel/extensions/taobao/crypto.py create mode 100644 backend/package/yuxi/channel/extensions/taobao/gateway.py create mode 100644 backend/package/yuxi/channel/extensions/taobao/message.py create mode 100644 backend/package/yuxi/channel/extensions/taobao/outbound.py create mode 100644 backend/package/yuxi/channel/extensions/taobao/pairing.py create mode 100644 backend/package/yuxi/channel/extensions/taobao/plugin.json create mode 100644 backend/package/yuxi/channel/extensions/taobao/security.py create mode 100644 backend/package/yuxi/channel/extensions/taobao/status.py create mode 100644 backend/package/yuxi/channel/extensions/taobao/tools.py create mode 100644 backend/package/yuxi/channel/extensions/taobao/types.py create mode 100644 backend/package/yuxi/channel/extensions/taobao/webhook.py diff --git a/backend/package/yuxi/channel/extensions/taobao/__init__.py b/backend/package/yuxi/channel/extensions/taobao/__init__.py new file mode 100644 index 00000000..04501b44 --- /dev/null +++ b/backend/package/yuxi/channel/extensions/taobao/__init__.py @@ -0,0 +1,286 @@ +import time +from collections import defaultdict + +from yuxi.channel.capabilities import ChannelCapabilities +from yuxi.channel.extensions.base import BaseChannelPlugin +from yuxi.channel.extensions.taobao.config import TaobaoConfigAdapter +from yuxi.channel.extensions.taobao.gateway import TaobaoGateway +from yuxi.channel.extensions.taobao.outbound import MAX_TEXT_LENGTH, send_media, send_text, set_global_client +from yuxi.channel.extensions.taobao.pairing import TaobaoPairing +from yuxi.channel.extensions.taobao.security import TaobaoSecurity +from yuxi.channel.extensions.taobao.status import TaobaoStatus +from yuxi.channel.extensions.taobao.tools import TaobaoTools +from yuxi.channel.protocols import SessionResolution +from yuxi.channel.plugins.registry import ChannelPluginRegistry + +DEDUPE_TTL_SECONDS = 300 + + +class TaobaoPlugin(BaseChannelPlugin): + id = "taobao" + name = "淘宝/千牛" + order = 120 + label = "淘宝/千牛 (TOP API)" + aliases = ["taobao", "qianniu", "千牛", "淘宝客服"] + resolve_reply_to_mode = "off" + + def __init__(self): + self._config_adapter = TaobaoConfigAdapter() + self._gateway = TaobaoGateway() + self._status = TaobaoStatus() + self._security = TaobaoSecurity() + self._pairing = TaobaoPairing() + self._tools: TaobaoTools | None = None + self._dedupe: dict[str, dict[str, float]] = defaultdict(dict) + + @property + def capabilities(self) -> ChannelCapabilities: + return ChannelCapabilities( + chat_types=["direct"], + message_types=["text", "image", "rich"], + reactions=False, + typing_indicator=False, + threads=False, + edit=False, + unsend=False, + reply=False, + media=True, + effects=False, + native_commands=False, + polls=False, + group_management=False, + streaming=True, + streaming_mode="block", + block_streaming=True, + block_streaming_chunk_min_chars=800, + block_streaming_chunk_max_chars=1200, + block_streaming_coalesce_min_chars=400, + block_streaming_coalesce_max_chars=800, + ) + + def list_account_ids(self, config: dict) -> list[str]: + return self._config_adapter.list_account_ids(config) + + async def resolve_account(self, account_id: str) -> dict: + account = self._config_adapter.resolve_account(account_id) + self._security.load_config(account_id, account) + return account + + def is_configured(self, account: dict) -> bool: + return self._config_adapter.is_configured(account) + + def is_enabled(self, account: dict) -> bool: + return self._config_adapter.is_enabled(account) + + def disabled_reason(self, account: dict) -> str: + return self._config_adapter.disabled_reason(account) + + def describe_account(self, account: dict) -> dict: + return self._config_adapter.describe_account(account) + + def config_schema(self) -> dict: + return self._config_adapter.config_schema() + + async def start(self, ctx) -> object: + result = await self._gateway.start(ctx) + if result.get("client"): + set_global_client(result["client"]) + self._tools = TaobaoTools(result["client"]) + return result + + async def stop(self, ctx) -> None: + await self._gateway.stop(ctx) + set_global_client(None) + self._tools = None + + async def send_text( + self, + target_id: str, + content: str, + *, + reply_to_id: str | None = None, + thread_id: str | None = None, + account_id: str | None = None, + ) -> None: + content = content[:MAX_TEXT_LENGTH] + await send_text(target_id, content, reply_to_id=reply_to_id, thread_id=thread_id, account_id=account_id) + + async def send_media( + self, + target_id: str, + media_url: str, + media_type: str, + reply_to_id: str | None = None, + thread_id: str | None = None, + ) -> None: + await send_media(target_id, media_url, media_type, reply_to_id=reply_to_id, thread_id=thread_id) + + async def probe(self, account: dict) -> bool: + return await self._status.probe(account) + + def build_summary(self, snapshot: object) -> dict: + return self._status.build_summary(snapshot) + + def build_account_snapshot( + self, + account: dict, + config: dict, + runtime: object | None = None, + probe_result: object | None = None, + audit: object | None = None, + ): + return self._status.build_account_snapshot(account, config, runtime, probe_result, audit) + + async def check_allowlist(self, peer_id: str, channel_type: str) -> bool: + account = await self._config_adapter.resolve_account("default") + return await self._security.check_allowlist(peer_id, channel_type, account=account) + + def resolve_dm_policy(self) -> dict: + return self._security.resolve_dm_policy() + + def collect_warnings(self, config: dict, account_id: str | None = None, account: dict | None = None) -> list[str]: + return self._security.collect_warnings(config, account_id, account) + + async def generate_code(self, peer_id: str) -> str: + return await self._pairing.generate_code(peer_id) + + async def verify_code(self, peer_id: str, code: str) -> bool: + return await self._pairing.verify_code(peer_id, code) + + def normalize_allow_entry(self, entry: str) -> str: + return self._pairing.normalize_allow_entry(entry) + + async def notify_approval(self, config: dict, peer_id: str, account_id: str | None = None) -> None: + await self._pairing.notify_approval(config, peer_id, account_id) + + @property + def channel_format_instructions(self) -> str | None: + return ( + "你正在通过淘宝千牛客服渠道与买家沟通。" + "消息最多 2000 字符，超长会自动分段发送。" + "可以使用纯文本和 Markdown 进行格式化。" + "如果消息中包含订单信息（order_id），可以调用订单查询工具获取详情。" + ) + + @property + def streaming_mode(self) -> str: + return "block" + + @property + def block_streaming_enabled(self) -> bool: + return True + + def get_agent_tools(self) -> list: + if self._tools is None: + return [] + return self._tools.get_agent_tools() + + async def execute_agent_tool(self, tool_name: str, params: dict, context: dict) -> dict: + if self._tools is None: + return {"success": False, "error": "Tools not initialized"} + return await self._tools.execute_agent_tool(tool_name, params, context) + + def resolve_session(self, msg: object) -> SessionResolution: + buyer_nick = None + if hasattr(msg, "raw") and isinstance(msg.raw, dict): + buyer_nick = msg.raw.get("buyer_nick", "") or msg.raw.get("from_user", "") + if not buyer_nick and hasattr(msg, "sender") and msg.sender: + buyer_nick = msg.sender.id + + if not buyer_nick: + return SessionResolution(kind="direct", conversation_id="taobao:unknown") + + return SessionResolution( + kind="direct", + conversation_id=f"taobao:dm:{buyer_nick}", + label=buyer_nick, + ) + + def build_system_prompt(self, context) -> str | None: + account_name = getattr(context, "account_name", "") or "淘宝店铺" + return ( + f"你是{account_name}的 AI 客服助手，通过千牛工作台与买家沟通。\n\n" + "你必须遵守以下规则：\n" + "1. 回复简洁专业，可使用 Markdown 进行排版\n" + "2. 买家询问订单/物流/退款时，主动使用工具查询并给出准确答案\n" + "3. 遇到退货退款请求，按照淘宝平台规则引导买家操作\n" + "4. 不得承诺商家未授权的事项（如免单、赔偿金额等）\n" + "5. 遇到自身无法处理的问题，告知买家可转接人工客服\n" + "6. 敏感问题（投诉、纠纷、法律风险）立即建议转人工\n" + "7. 不引导买家到站外联系，不索要买家密码/验证码" + ) + + async def on_config_changed(self, prev_cfg: dict, next_cfg: dict, account_id: str) -> None: + if prev_cfg == next_cfg: + return + + import logging as _logging + + _logger = _logging.getLogger(__name__) + _logger.info("[taobao:%s] Config changed, reloading...", account_id) + + from yuxi.channel.context import ChannelContext + + ctx = ChannelContext( + channel_type="taobao", + account_id=account_id, + config=next_cfg, + ) + await self._gateway.stop(ctx) + result = await self._gateway.start(ctx) + if result.get("client"): + set_global_client(result["client"]) + self._tools = TaobaoTools(result["client"]) + + _logger.info("[taobao:%s] Config reloaded", account_id) + + async def on_message_sending(self, msg: object, content: str) -> str | None: + if not content: + return content + return content[:MAX_TEXT_LENGTH] + + async def on_message_received(self, msg: object) -> object | None: + return msg + + def classify_error(self, error: BaseException) -> object: + from yuxi.channel.extensions.taobao.client import TaobaoAPIError + from yuxi.channel.protocols import ClassifiedError, ErrorSeverity + + if isinstance(error, TaobaoAPIError): + sub_code = error.sub_code + if sub_code in ("isv.invalid-parameter", "isv.missing-parameter"): + return ClassifiedError(severity=ErrorSeverity.FATAL, original_error=error, error_message=str(error)) + if sub_code in ("isp.top-remote-connection-timeout", "isp.service-unavailable"): + return ClassifiedError(severity=ErrorSeverity.RETRYABLE, original_error=error, error_message=str(error)) + if sub_code in ("isv.token-invalid", "isv.token-expired"): + return ClassifiedError(severity=ErrorSeverity.FATAL, original_error=error, error_message=str(error)) + return ClassifiedError(severity=ErrorSeverity.RETRYABLE, original_error=error, error_message=str(error)) + + from yuxi.channel.errors import classify_error as _classify_error + + return _classify_error(error) + + def is_duplicate(self, key: str) -> bool: + now = time.monotonic() + account_dedupe = self._dedupe["default"] + if key in account_dedupe and now - account_dedupe[key] < DEDUPE_TTL_SECONDS: + return True + account_dedupe[key] = now + cutoff = now - DEDUPE_TTL_SECONDS + self._dedupe["default"] = {k: v for k, v in account_dedupe.items() if v >= cutoff} + return False + + def mark_seen(self, key: str) -> None: + self._dedupe["default"][key] = time.monotonic() + + def sanitize_text(self, text: str) -> str: + if not text: + return "" + text = text.replace("\\n", "\n").replace("\\t", "\t") + return text[:MAX_TEXT_LENGTH] + + def markdown_to_native(self, md_text: str) -> str: + return md_text + + +ChannelPluginRegistry.register(TaobaoPlugin()) diff --git a/backend/package/yuxi/channel/extensions/taobao/client.py b/backend/package/yuxi/channel/extensions/taobao/client.py new file mode 100644 index 00000000..970ce566 --- /dev/null +++ b/backend/package/yuxi/channel/extensions/taobao/client.py @@ -0,0 +1,197 @@ +import hashlib +import hmac +import json +import logging +import time + +import httpx + +logger = logging.getLogger(__name__) + +GATEWAY_URL = "https://eco.taobao.com/router/rest" +SANDBOX_GATEWAY_URL = "https://gw.api.tbsandbox.com/router/rest" + + +class TaobaoAPIError(Exception): + def __init__(self, code: str, msg: str, sub_code: str = "", sub_msg: str = ""): + self.code = code + self.msg = msg + self.sub_code = sub_code + self.sub_msg = sub_msg + super().__init__(f"[{code}] {msg}" + (f" ({sub_code}: {sub_msg})" if sub_code else "")) + + +class TaobaoClient: + def __init__( + self, + app_key: str, + app_secret: str, + sign_method: str = "md5", + sandbox: bool = False, + timeout: int = 10, + ): + self.app_key = app_key + self.app_secret = app_secret + self.sign_method = sign_method + self.gateway = SANDBOX_GATEWAY_URL if sandbox else GATEWAY_URL + self.timeout = timeout + + def _sign(self, params: dict) -> str: + sorted_keys = sorted(k for k in params if k != "sign" and params[k] is not None) + raw = "".join(f"{k}{params[k]}" for k in sorted_keys) + raw = f"{self.app_secret}{raw}{self.app_secret}" + + if self.sign_method == "hmac-sha256": + return ( + hmac.new( + self.app_secret.encode("utf-8"), + raw.encode("utf-8"), + hashlib.sha256, + ) + .hexdigest() + .upper() + ) + return hashlib.md5(raw.encode("utf-8")).hexdigest().upper() + + def _build_payload(self, method: str, params: dict, session: str | None = None) -> dict: + payload = { + "method": method, + "app_key": self.app_key, + "timestamp": time.strftime("%Y-%m-%d %H:%M:%S"), + "format": "json", + "v": "2.0", + "sign_method": self.sign_method, + **params, + } + if session: + payload["session"] = session + payload["sign"] = self._sign(payload) + return payload + + async def execute( + self, + method: str, + params: dict | None = None, + session: str | None = None, + ) -> dict: + payload = self._build_payload(method, params or {}, session) + + async with httpx.AsyncClient(timeout=self.timeout) as http: + resp = await http.post(self.gateway, data=payload) + resp.raise_for_status() + data = resp.json() + + if "error_response" in data: + err = data["error_response"] + raise TaobaoAPIError( + code=err.get("code", "unknown"), + msg=err.get("msg", ""), + sub_code=err.get("sub_code", ""), + sub_msg=err.get("sub_msg", ""), + ) + + return data + + async def get_token(self, code: str, redirect_uri: str) -> dict: + return await self.execute( + "taobao.top.auth.token.create", + {"code": code, "redirect_uri": redirect_uri}, + ) + + async def refresh_token(self, refresh_token: str) -> dict: + return await self.execute( + "taobao.top.auth.token.refresh", + {"refresh_token": refresh_token}, + ) + + async def send_customer_message( + self, + to_user: str, + content: str, + msg_type: int = 0, + media_id: str | None = None, + session: str | None = None, + ) -> dict: + context = {"text": content} if msg_type == 0 else {"media_id": media_id or ""} + return await self.execute( + "taobao.openim.custmsg.push", + { + "to_user": to_user, + "msg_type": msg_type, + "context": json.dumps(context, ensure_ascii=False), + }, + session=session, + ) + + async def get_trade_fullinfo(self, tid: int, session: str) -> dict: + return await self.execute( + "taobao.trade.fullinfo.get", + { + "tid": tid, + "fields": "tid,type,status,payment,orders,logistics", + }, + session=session, + ) + + async def get_logistics_trace(self, tid: int, session: str) -> dict: + return await self.execute( + "taobao.logistics.trace.search", + {"tid": tid}, + session=session, + ) + + async def get_userservice(self, user_id: str, session: str) -> dict: + return await self.execute( + "taobao.openim.userservice.get", + {"user_id": user_id}, + session=session, + ) + + async def qianniu_sellerorder_query(self, buyer_nick: str, session: str) -> dict: + return await self.execute( + "taobao.qianniu.sellerorder.buyer.query", + {"buyer_nick": buyer_nick}, + session=session, + ) + + async def qianniu_refund_get(self, refund_id: str, session: str) -> dict: + return await self.execute( + "taobao.qianniu.refund.get", + {"refund_id": refund_id}, + session=session, + ) + + async def qianniu_buyer_tag_get(self, buyer_nick: str, session: str) -> dict: + return await self.execute( + "taobao.qianniu.buyer.tag.get", + {"buyer_nick": buyer_nick}, + session=session, + ) + + async def qianniu_coupon_buyer_get(self, buyer_nick: str, session: str) -> dict: + return await self.execute( + "taobao.qianniu.coupon.buyer.get", + {"buyer_nick": buyer_nick}, + session=session, + ) + + async def qianniu_cloudkefu_forward(self, buyer_nick: str, to_nick: str, session: str) -> dict: + return await self.execute( + "taobao.qianniu.cloudkefu.forward", + {"buyer_nick": buyer_nick, "to_nick": to_nick}, + session=session, + ) + + async def get_item(self, item_id: int, session: str) -> dict: + return await self.execute( + "taobao.item.get", + {"item_id": item_id, "fields": "title,price,pic_url,desc"}, + session=session, + ) + + async def get_chatlogs(self, user_id: str, begin: str, end: str, session: str) -> dict: + return await self.execute( + "taobao.openim.chatlogs.get", + {"user_id": user_id, "begin": begin, "end": end}, + session=session, + ) diff --git a/backend/package/yuxi/channel/extensions/taobao/config.py b/backend/package/yuxi/channel/extensions/taobao/config.py new file mode 100644 index 00000000..82cd47a6 --- /dev/null +++ b/backend/package/yuxi/channel/extensions/taobao/config.py @@ -0,0 +1,109 @@ +import os + +from yuxi.channel.extensions.taobao.types import TaobaoAccount + + +class TaobaoConfigAdapter: + def __init__(self): + self._config: dict = {} + self._accounts: dict = {} + + def list_account_ids(self, config: dict) -> list[str]: + self._config = config + self._accounts = config.get("accounts", {}) + if self._accounts: + return list(self._accounts.keys()) + return ["default"] + + def resolve_account(self, account_id: str) -> dict: + raw = self._accounts.get(account_id, {}) + if not raw: + raw = {"account_id": account_id} + return self.build_account(account_id, raw).to_dict() + + def is_configured(self, account: dict) -> bool: + acc = self._to_account(account) + return acc.is_configured() + + def is_enabled(self, account: dict) -> bool: + return account.get("enabled", True) + + def disabled_reason(self, account: dict) -> str: + if not account.get("enabled", True): + return "账户已禁用" + return "" + + def describe_account(self, account: dict) -> dict: + acc = self._to_account(account) + return { + "account_id": acc.account_id, + "name": acc.name or acc.seller_nick or acc.account_id, + "configured": acc.is_configured(), + "has_token": acc.has_valid_token(), + } + + def build_account(self, account_id: str, raw: dict) -> TaobaoAccount: + return TaobaoAccount( + account_id=account_id, + app_key=raw.get("app_key", os.getenv("TAOBAO_APP_KEY", "")), + app_secret=raw.get("app_secret", os.getenv("TAOBAO_APP_SECRET", "")), + sign_method=raw.get("sign_method", "md5"), + access_token=raw.get("access_token", ""), + refresh_token=raw.get("refresh_token", ""), + callback_url=raw.get("callback_url", os.getenv("TAOBAO_CALLBACK_URL", "")), + rsa_public_key=raw.get("rsa_public_key", ""), + rsa_private_key=raw.get("rsa_private_key", ""), + encrypt_messages=raw.get("encrypt_messages", False), + seller_nick=raw.get("seller_nick", ""), + name=raw.get("name", ""), + dm_policy=raw.get("dm_policy", "open"), + allow_from=raw.get("allow_from", []), + timeout=raw.get("timeout", 10), + sandbox=raw.get("sandbox", False), + ) + + def _to_account(self, account: dict) -> TaobaoAccount: + account_id = account.get("account_id", "default") + if account.get("app_key"): + return self.build_account(account_id, account) + return TaobaoAccount(account_id=account_id) + + def config_schema(self) -> dict: + return { + "type": "object", + "properties": { + "accounts": { + "type": "object", + "description": "多商家账号配置", + "additionalProperties": { + "type": "object", + "properties": { + "app_key": {"type": "string", "description": "淘宝开放平台 AppKey"}, + "app_secret": {"type": "string", "description": "AppSecret"}, + "sign_method": { + "type": "string", + "enum": ["md5", "hmac-sha256"], + "default": "md5", + }, + "callback_url": {"type": "string", "description": "Webhook 回调地址"}, + "rsa_public_key": {"type": "string", "description": "淘宝分配的 RSA 公钥"}, + "rsa_private_key": {"type": "string", "description": "ISV RSA 私钥"}, + "encrypt_messages": {"type": "boolean", "default": False}, + "seller_nick": {"type": "string", "description": "商家淘宝昵称"}, + "dm_policy": { + "type": "string", + "enum": ["open", "pairing", "allowlist", "disabled"], + }, + "allow_from": { + "type": "array", + "items": {"type": "string"}, + "description": "允许交互的买家 nick 列表", + }, + "timeout": {"type": "integer", "default": 10}, + "sandbox": {"type": "boolean", "default": False}, + }, + "required": ["app_key", "app_secret"], + }, + } + }, + } diff --git a/backend/package/yuxi/channel/extensions/taobao/crypto.py b/backend/package/yuxi/channel/extensions/taobao/crypto.py new file mode 100644 index 00000000..02c8beb4 --- /dev/null +++ b/backend/package/yuxi/channel/extensions/taobao/crypto.py @@ -0,0 +1,63 @@ +import base64 +import hashlib +import hmac as hmac_module +import logging + +logger = logging.getLogger(__name__) + +try: + from Crypto.Cipher import AES, PKCS1_v1_5 + from Crypto.PublicKey import RSA + + _CRYPTO_AVAILABLE = True +except ImportError: + _CRYPTO_AVAILABLE = False + + +def verify_top_sign(params: dict, secret: str, expected_sign: str, sign_method: str = "md5") -> bool: + sorted_keys = sorted(k for k in params if k != "sign" and params[k] is not None) + raw = "".join(f"{k}{params[k]}" for k in sorted_keys) + raw = f"{secret}{raw}{secret}" + + if sign_method == "hmac-sha256": + computed = hmac_module.new(secret.encode("utf-8"), raw.encode("utf-8"), hashlib.sha256).hexdigest().upper() + else: + computed = hashlib.md5(raw.encode("utf-8")).hexdigest().upper() + + return computed == expected_sign + + +def decrypt_message( + encrypt_data: str, + encrypt_key: str, + sign: str, + timestamp: str, + app_secret: str, + rsa_private_key_pem: str, +) -> str | None: + if not _CRYPTO_AVAILABLE: + logger.error("pycryptodome not installed, cannot decrypt TOP message") + return None + + try: + rsa_key = RSA.import_key(rsa_private_key_pem) + cipher_rsa = PKCS1_v1_5.new(rsa_key) + session_key = cipher_rsa.decrypt(base64.b64decode(encrypt_key), None) + if session_key is None: + logger.error("RSA decrypt session key failed") + return None + + expected_sign = hashlib.sha256( + f"{timestamp}{app_secret}{session_key.decode('utf-8')}{encrypt_data}".encode() + ).hexdigest() + if sign != expected_sign: + logger.error("Message sign verification failed") + return None + + cipher_aes = AES.new(session_key[:16], AES.MODE_CBC, session_key[:16]) + plain = cipher_aes.decrypt(base64.b64decode(encrypt_data)) + pad_len = plain[-1] + return plain[:-pad_len].decode("utf-8") + except Exception as e: + logger.exception("Message decryption failed: %s", e) + return None diff --git a/backend/package/yuxi/channel/extensions/taobao/gateway.py b/backend/package/yuxi/channel/extensions/taobao/gateway.py new file mode 100644 index 00000000..2e73b137 --- /dev/null +++ b/backend/package/yuxi/channel/extensions/taobao/gateway.py @@ -0,0 +1,105 @@ +import asyncio +import logging +import time +from datetime import datetime, timedelta + +from yuxi.channel.extensions.taobao.client import TaobaoClient +from yuxi.channel.extensions.taobao.config import TaobaoConfigAdapter +from yuxi.channel.extensions.taobao.types import TaobaoAccount + +logger = logging.getLogger(__name__) + +_current_gateway: TaobaoGateway | None = None + + +class TaobaoGateway: + def __init__(self): + self._running = False + self._account: TaobaoAccount | None = None + self._client: TaobaoClient | None = None + self._token_refresh_task: asyncio.Task | None = None + self._message_queue: asyncio.Queue | None = None + self._last_message_at: float | None = None + + async def start(self, ctx) -> dict: + global _current_gateway + + account = _resolve_account(ctx) + if not account.is_configured(): + logger.warning("Taobao account %s not configured, skipping", account.account_id) + return {"running": False, "reason": "not-configured", "account_id": account.account_id} + + self._account = account + self._client = TaobaoClient( + app_key=account.app_key, + app_secret=account.app_secret, + sign_method=account.sign_method, + sandbox=account.sandbox, + timeout=account.timeout, + ) + + self._message_queue = asyncio.Queue(maxsize=1000) + _current_gateway = self + + if account.has_valid_token(): + self._token_refresh_task = asyncio.create_task(self._token_refresh_loop()) + + self._running = True + logger.info("Taobao gateway started for account %s", account.account_id) + return {"running": True, "account_id": account.account_id, "client": self._client} + + async def stop(self, ctx) -> None: + self._running = False + + if self._token_refresh_task: + self._token_refresh_task.cancel() + self._token_refresh_task = None + + self._message_queue = None + self._client = None + self._account = None + + logger.info("Taobao gateway stopped") + + async def _token_refresh_loop(self): + while self._running: + await asyncio.sleep(600) + if not self._account or not self._account.refresh_token: + continue + try: + resp = await self._client.refresh_token(self._account.refresh_token) + token_data = resp.get("top_auth_token_create_response", {}).get("token_result", {}) + if token_data: + self._account.access_token = token_data.get("access_token", "") + self._account.refresh_token = token_data.get("refresh_token", "") + expires_in = token_data.get("expires_in", 0) + self._account.token_expires_at = datetime.now() + timedelta(seconds=expires_in) + logger.info("Taobao token refreshed for %s", self._account.account_id) + except asyncio.CancelledError: + break + except Exception as e: + logger.error("Taobao token refresh failed for %s: %s", self._account.account_id, e) + + async def enqueue_message(self, unified_msg: dict) -> None: + if self._message_queue is not None: + self._last_message_at = time.monotonic() + await self._message_queue.put(unified_msg) + + def get_message_queue(self) -> asyncio.Queue | None: + return self._message_queue + + def get_last_message_at(self) -> float | None: + return self._last_message_at + + +def _resolve_account(ctx) -> TaobaoAccount: + config = getattr(ctx, "config", {}) if ctx else {} + accounts = config.get("accounts", {}) + account_id = getattr(ctx, "account_id", "default") + raw = accounts.get(account_id, {}) + adapter = TaobaoConfigAdapter() + return adapter.build_account(account_id, raw) + + +def get_current_gateway() -> TaobaoGateway | None: + return _current_gateway diff --git a/backend/package/yuxi/channel/extensions/taobao/message.py b/backend/package/yuxi/channel/extensions/taobao/message.py new file mode 100644 index 00000000..57d8cee0 --- /dev/null +++ b/backend/package/yuxi/channel/extensions/taobao/message.py @@ -0,0 +1,85 @@ +import logging + +from yuxi.channel.extensions.taobao.types import InboundTaobaoMessage + +logger = logging.getLogger(__name__) + + +def parse_to_unified_message(raw: dict) -> dict | None: + msg_type = str(raw.get("msg_type", raw.get("type", "text"))) + + content_map = { + "text": lambda r: r.get("content", "") or r.get("text", ""), + "0": lambda r: r.get("content", "") or r.get("text", ""), + "image": lambda r: "[图片消息]", + "1": lambda r: "[图片消息]", + "order": lambda r: _format_order_message(r), + "trade": lambda r: _format_order_message(r), + "system": lambda r: "[系统消息]", + "ordering": lambda r: _format_ordering_message(r), + "item": lambda r: _format_item_card(r), + "coupon": lambda r: _format_coupon_card(r), + } + + handler = content_map.get(msg_type, lambda r: r.get("content", "") or r.get("text", "")) + content = handler(raw) + + buyer_nick = raw.get("buyer_nick", "") or raw.get("from_user", "") or raw.get("from_id", "") + msg_id = raw.get("msg_id", "") or raw.get("id", "") or str(hash(f"{buyer_nick}{content}")) + + if not buyer_nick or not content: + return None + + unified = { + "sender": {"id": buyer_nick, "name": buyer_nick}, + "content": content, + "message_id": msg_id, + "channel_type": "taobao", + "chat_type": "direct", + "timestamp": raw.get("timestamp"), + "raw": raw, + } + + order_id = raw.get("order_id") or raw.get("trade_id") or raw.get("tid") + if order_id: + unified["order_id"] = str(order_id) + + return unified + + +def _format_order_message(raw: dict) -> str: + order_id = raw.get("order_id", raw.get("trade_id", "")) + status = raw.get("status", raw.get("order_status", "")) + buyer = raw.get("buyer_nick", "") + return f"[订单消息] 订单ID: {order_id}, 状态: {status}, 买家: {buyer}" + + +def _format_ordering_message(raw: dict) -> str: + item_name = raw.get("item_title", raw.get("item_name", "未知商品")) + price = raw.get("price", "") + return f"[商品卡片] {item_name}" + (f" ¥{price}" if price else "") + + +def _format_item_card(raw: dict) -> str: + item_name = raw.get("item_title", raw.get("item_name", "未知商品")) + price = raw.get("price", "") + return f"[商品推荐] {item_name}" + (f" ¥{price}" if price else "") + + +def _format_coupon_card(raw: dict) -> str: + coupon_name = raw.get("coupon_name", "优惠券") + amount = raw.get("amount", raw.get("denomination", "")) + return f"[优惠券] {coupon_name}" + (f" ¥{amount}" if amount else "") + + +def parse_raw_to_taobao_message(raw: dict) -> InboundTaobaoMessage: + return InboundTaobaoMessage( + msg_id=raw.get("msg_id", raw.get("id", "")), + buyer_nick=raw.get("buyer_nick", raw.get("from_user", "")), + seller_nick=raw.get("seller_nick", ""), + content=raw.get("content", raw.get("text", "")), + msg_type=raw.get("msg_type", "text"), + order_id=raw.get("order_id") or raw.get("trade_id"), + item_id=raw.get("item_id"), + raw_payload=raw, + ) diff --git a/backend/package/yuxi/channel/extensions/taobao/outbound.py b/backend/package/yuxi/channel/extensions/taobao/outbound.py new file mode 100644 index 00000000..bbde47d9 --- /dev/null +++ b/backend/package/yuxi/channel/extensions/taobao/outbound.py @@ -0,0 +1,89 @@ +import logging + +from yuxi.channel.extensions.taobao.client import TaobaoAPIError, TaobaoClient + +logger = logging.getLogger(__name__) + +MAX_TEXT_LENGTH = 2000 + +_GLOBAL_CLIENT: TaobaoClient | None = None + + +def set_global_client(client: TaobaoClient | None) -> None: + global _GLOBAL_CLIENT + _GLOBAL_CLIENT = client + + +def get_client() -> TaobaoClient | None: + return _GLOBAL_CLIENT + + +async def send_text( + target_id: str, + content: str, + *, + reply_to_id: str | None = None, + thread_id: str | None = None, + account_id: str | None = None, + session: str | None = None, +) -> None: + client = _GLOBAL_CLIENT + if client is None: + logger.error("Taobao client not initialized") + return + + try: + resp = await client.send_customer_message( + to_user=target_id, + content=content[:MAX_TEXT_LENGTH], + msg_type=0, + session=session, + ) + logger.debug("Taobao message sent to %s: %s", target_id, resp) + except TaobaoAPIError as e: + logger.error("Taobao send_text failed to %s: %s", target_id, e) + + +async def send_media( + target_id: str, + media_url: str, + media_type: str, + *, + reply_to_id: str | None = None, + thread_id: str | None = None, + session: str | None = None, +) -> None: + client = _GLOBAL_CLIENT + if client is None: + logger.error("Taobao client not initialized") + return + + msg_type = _media_type_to_int(media_type) + try: + await client.send_customer_message( + to_user=target_id, + content="", + msg_type=msg_type, + media_id=media_url, + session=session, + ) + except TaobaoAPIError as e: + logger.error("Taobao send_media failed to %s: %s", target_id, e) + + +def _media_type_to_int(media_type: str) -> int: + mapping = {"image": 1, "file": 4} + return mapping.get(media_type, 1) + + +def chunk_text(text: str, limit: int = MAX_TEXT_LENGTH) -> list[str]: + chunks = [] + while len(text) > limit: + split_at = text.rfind("\n", 0, limit) + if split_at == -1: + split_at = limit + chunks.append(text[:split_at]) + text = text[split_at:].lstrip("\n") + if text: + chunks.append(text) + return chunks diff --git a/backend/package/yuxi/channel/extensions/taobao/pairing.py b/backend/package/yuxi/channel/extensions/taobao/pairing.py new file mode 100644 index 00000000..2bb8f6b9 --- /dev/null +++ b/backend/package/yuxi/channel/extensions/taobao/pairing.py @@ -0,0 +1,72 @@ +import logging +import random +import string +import time +from collections import defaultdict + +logger = logging.getLogger(__name__) + +CODE_LENGTH = 6 +CODE_TTL_SECONDS = 600 + + +class TaobaoPairing: + id_label = "buyer_nick" + + def __init__(self): + self._codes: dict[str, dict[str, tuple[str, float]]] = defaultdict(dict) + + async def generate_code(self, peer_id: str, account_id: str = "default") -> str: + code = "".join(random.choices(string.digits, k=CODE_LENGTH)) + self._codes[account_id][code] = (peer_id, time.monotonic()) + self._cleanup_expired(account_id) + logger.info("Pairing code generated for %s:%s: %s", account_id, peer_id, code) + return code + + async def verify_code(self, peer_id: str, code: str, account_id: str = "default") -> bool: + self._cleanup_expired(account_id) + + account_codes = self._codes.get(account_id, {}) + entry = account_codes.get(code) + if entry is None: + return False + + stored_peer_id, created_at = entry + if time.monotonic() - created_at > CODE_TTL_SECONDS: + account_codes.pop(code, None) + return False + + if stored_peer_id != peer_id: + return False + + account_codes.pop(code, None) + logger.info("Pairing code verified for %s:%s", account_id, peer_id) + return True + + def normalize_allow_entry(self, entry: str) -> str: + return entry.strip().lower() + + async def notify_approval(self, config: dict, peer_id: str, account_id: str | None = None) -> None: + logger.info("Pairing approval notification for %s:%s", account_id, peer_id) + try: + from yuxi.channel.extensions.taobao.outbound import get_client + + client = get_client() + if client is None: + logger.warning("Cannot notify approval: client not initialized") + return + + await client.send_customer_message( + to_user=peer_id, + content="您的配对已通过审批，现在可以开始与我对话了！请问有什么可以帮助您的？", + msg_type=0, + ) + except Exception as e: + logger.error("Failed to send approval notification to %s: %s", peer_id, e) + + def _cleanup_expired(self, account_id: str): + now = time.monotonic() + account_codes = self._codes.get(account_id, {}) + expired = [k for k, (_, created) in account_codes.items() if now - created > CODE_TTL_SECONDS] + for k in expired: + account_codes.pop(k, None) diff --git a/backend/package/yuxi/channel/extensions/taobao/plugin.json b/backend/package/yuxi/channel/extensions/taobao/plugin.json new file mode 100644 index 00000000..6e5cfdb8 --- /dev/null +++ b/backend/package/yuxi/channel/extensions/taobao/plugin.json @@ -0,0 +1,31 @@ +{ + "id": "taobao", + "name": "淘宝/千牛", + "version": "0.1.0", + "label": "淘宝/千牛 (TOP API)", + "aliases": ["taobao", "qianniu", "千牛", "淘宝客服"], + "description": "淘宝千牛客服消息渠道插件，基于 TOP OpenAPI 实现消息收发", + "author": "ForcePilot Team", + "order": 120, + "dependencies": ["httpx"], + "capabilities": { + "chat_types": ["direct"], + "message_types": ["text", "image", "rich"], + "reactions": false, + "typing_indicator": false, + "threads": false, + "edit": false, + "unsend": false, + "reply": false, + "media": true, + "effects": false, + "native_commands": false, + "polls": false, + "group_management": false, + "streaming": true, + "streaming_mode": "block", + "block_streaming": true + }, + "enabled": true, + "python_requires": ">=3.12" +} \ No newline at end of file diff --git a/backend/package/yuxi/channel/extensions/taobao/security.py b/backend/package/yuxi/channel/extensions/taobao/security.py new file mode 100644 index 00000000..9ef61eb7 --- /dev/null +++ b/backend/package/yuxi/channel/extensions/taobao/security.py @@ -0,0 +1,50 @@ +import logging +from collections import defaultdict + +logger = logging.getLogger(__name__) + + +class TaobaoSecurity: + def __init__(self): + self._allowlists: dict[str, set[str]] = defaultdict(set) + + def load_config(self, account_id: str, account: dict) -> None: + allow_from = account.get("allow_from", []) + self._allowlists[account_id] = {entry.strip().lower() for entry in allow_from if entry and entry.strip()} + + async def check_allowlist(self, peer_id: str, channel_type: str, account_id: str | None = None, account: dict | None = None) -> bool: + if account is None: + return True + + policy = account.get("dm_policy", "open") + + if policy == "disabled": + return False + + if policy in ("allowlist", "pairing"): + if account_id: + allow_set = self._allowlists.get(account_id, set()) + if not allow_set: + allow_from = account.get("allow_from", []) + allow_set = {entry.strip().lower() for entry in allow_from if entry and entry.strip()} + return peer_id in allow_set + return peer_id in account.get("allow_from", []) + + return True + + def resolve_dm_policy(self, account_id: str | None = None, account: dict | None = None) -> dict: + if account is None: + return {"mode": "open", "allow_from": []} + return { + "mode": account.get("dm_policy", "open"), + "allow_from": account.get("allow_from", []), + } + + def collect_warnings(self, config: dict, account_id: str | None = None, account: dict | None = None) -> list[str]: + warnings = [] + acc = account or {} + if acc.get("dm_policy") == "open": + warnings.append("DM 策略为 'open'，任何人可发送消息") + if not acc.get("callback_url"): + warnings.append("未配置 Webhook 回调 URL，无法接收消息") + return warnings diff --git a/backend/package/yuxi/channel/extensions/taobao/status.py b/backend/package/yuxi/channel/extensions/taobao/status.py new file mode 100644 index 00000000..ca9a81d4 --- /dev/null +++ b/backend/package/yuxi/channel/extensions/taobao/status.py @@ -0,0 +1,64 @@ +import logging + +from yuxi.channel.protocols import ChannelAccountSnapshot + +logger = logging.getLogger(__name__) + + +class TaobaoStatus: + async def probe(self, account: dict) -> bool: + app_key = account.get("app_key", "") + app_secret = account.get("app_secret", "") + if not app_key or not app_secret: + return False + + from yuxi.channel.extensions.taobao.client import TaobaoClient + + client = TaobaoClient( + app_key=app_key, + app_secret=app_secret, + sign_method=account.get("sign_method", "md5"), + sandbox=account.get("sandbox", False), + ) + + try: + resp = await client.execute("taobao.time.get") + return "time_get_response" in resp + except Exception as e: + logger.warning("Taobao probe failed for %s: %s", account.get("account_id", "default"), e) + return False + + def build_summary(self, snapshot: object) -> dict: + if isinstance(snapshot, ChannelAccountSnapshot): + return { + "channel": "taobao", + "account_id": snapshot.account_id, + "configured": snapshot.configured, + "connected": snapshot.connected, + "last_message_at": snapshot.last_message_at, + "last_error": snapshot.last_error, + } + return {} + + def build_account_snapshot( + self, + account: dict, + config: dict, + runtime: ChannelAccountSnapshot | None = None, + probe_result: object | None = None, + audit: object | None = None, + ) -> ChannelAccountSnapshot: + acc_id = account.get("account_id", "default") + configured = bool(account.get("app_key") and account.get("app_secret")) + + return ChannelAccountSnapshot( + account_id=acc_id, + name=account.get("name", account.get("seller_nick", acc_id)), + enabled=account.get("enabled", True), + configured=configured, + connected=probe_result if isinstance(probe_result, bool) else False, + last_message_at=runtime.last_message_at if runtime else None, + last_error=getattr(runtime, "last_error", None) if runtime else None, + dm_policy=account.get("dm_policy", "open"), + allow_from=account.get("allow_from", []), + ) diff --git a/backend/package/yuxi/channel/extensions/taobao/tools.py b/backend/package/yuxi/channel/extensions/taobao/tools.py new file mode 100644 index 00000000..79071582 --- /dev/null +++ b/backend/package/yuxi/channel/extensions/taobao/tools.py @@ -0,0 +1,257 @@ +from __future__ import annotations + +import logging +from datetime import datetime, timedelta + +logger = logging.getLogger(__name__) + +TRADE_STATUS_MAP = { + "WAIT_BUYER_PAY": "等待买家付款", + "WAIT_SELLER_SEND_GOODS": "等待卖家发货", + "WAIT_BUYER_CONFIRM_GOODS": "等待买家确认收货", + "TRADE_FINISHED": "交易完成", + "TRADE_CLOSED": "交易关闭", + "TRADE_CLOSED_BY_TAOBAO": "交易被淘宝关闭", +} + + +class TaobaoTools: + def __init__(self, client): + self._client = client + + async def query_order(self, order_id: str, session: str | None = None) -> str: + try: + data = await self._client.get_trade_fullinfo(int(order_id), session) + trade = data.get("trade_fullinfo_get_response", {}).get("trade", {}) + + status = TRADE_STATUS_MAP.get(trade.get("status", ""), trade.get("status", "未知")) + payment = trade.get("payment", "未知") + buyer_nick = trade.get("buyer_nick", "") + orders = trade.get("orders", {}).get("order", []) + + lines = [ + f"📦 订单 {order_id}", + f"状态: {status}", + f"金额: ¥{payment}", + f"买家: {buyer_nick}", + ] + if orders: + lines.append("商品:") + for o in orders if isinstance(orders, list) else [orders]: + lines.append(f" - {o.get('title', '未知')} ×{o.get('num', 1)} ¥{o.get('price', 0)}") + return "\n".join(lines) + except Exception as e: + logger.exception("Query order failed: %s", order_id) + return f"查询订单 {order_id} 失败: {e}" + + async def query_logistics(self, order_id: str, session: str | None = None) -> str: + try: + data = await self._client.get_logistics_trace(int(order_id), session) + resp = data.get("logistics_trace_search_response", {}) + traces = resp.get("trace_list", resp.get("traces", [])) + + if not traces: + return f"📦 订单 {order_id} 暂无物流信息" + + lines = [f"📦 订单 {order_id} 物流轨迹:"] + for t in (traces if isinstance(traces, list) else [traces])[:10]: + lines.append(f" [{t.get('time', '')}] {t.get('status_desc', t.get('desc', ''))}") + return "\n".join(lines) + except Exception as e: + logger.exception("Query logistics failed: %s", order_id) + return f"查询物流 {order_id} 失败: {e}" + + async def query_item(self, item_id: str, session: str | None = None) -> str: + try: + data = await self._client.get_item(int(item_id), session) + item = data.get("item_get_response", {}).get("item", {}) + + title = item.get("title", "未知商品") + price = item.get("price", "") + desc = item.get("desc", "")[:200] + + lines = [f"🛒 {title}"] + if price: + lines.append(f"价格: ¥{price}") + if desc: + lines.append(f"描述: {desc}") + return "\n".join(lines) + except Exception as e: + logger.exception("Query item failed: %s", item_id) + return f"查询商品 {item_id} 失败: {e}" + + async def query_refund(self, refund_id: str, session: str | None = None) -> str: + try: + data = await self._client.qianniu_refund_get(refund_id, session) + refund = data.get("qianniu_refund_get_response", {}) + + status = refund.get("status", "未知") + amount = refund.get("refund_amount", refund.get("amount", "未知")) + reason = refund.get("refund_reason", refund.get("reason", "未知")) + + return f"🔙 退款 {refund_id}\n状态: {status}\n退款金额: ¥{amount}\n申请原因: {reason}" + except Exception as e: + logger.exception("Query refund failed: %s", refund_id) + return f"查询退款 {refund_id} 失败: {e}" + + async def query_buyer_tags(self, buyer_nick: str, session: str | None = None) -> str: + try: + data = await self._client.qianniu_buyer_tag_get(buyer_nick, session) + tags = data.get("qianniu_buyer_tag_get_response", {}).get("tags", []) + + if not tags: + return f"买家 {buyer_nick} 暂无标签" + + tag_names = [t.get("tag_name", str(t)) for t in tags] if isinstance(tags, list) else [str(tags)] + return f"买家 {buyer_nick} 标签: {', '.join(tag_names)}" + except Exception as e: + logger.exception("Query buyer tags failed: %s", buyer_nick) + return f"查询买家标签失败: {e}" + + async def transfer_human(self, buyer_nick: str, reason: str = "", session: str | None = None) -> str: + try: + if not hasattr(self._client, "qianniu_cloudkefu_forward"): + return "转人工功能暂不可用（千牛客服转发 API 未接入）" + + await self._client.qianniu_cloudkefu_forward( + buyer_nick=buyer_nick, + to_nick="", # 不指定分配到哪个客服，由分流系统自动分配 + session=session, + ) + return f"已将买家 {buyer_nick} 转接人工客服" + (f"，原因: {reason}" if reason else "") + except Exception as e: + logger.exception("Transfer to human failed: %s", buyer_nick) + return f"转人工失败: {e}" + + async def query_chat_history(self, buyer_nick: str, session: str | None = None) -> str: + end = datetime.now().strftime("%Y-%m-%d %H:%M:%S") + begin = (datetime.now() - timedelta(hours=24)).strftime("%Y-%m-%d %H:%M:%S") + try: + data = await self._client.get_chatlogs(buyer_nick, begin, end, session) + logs_data = data.get("openim_chatlogs_get_response", {}).get("chatlogs", {}) + logs = logs_data.get("chatlog", []) + + if not logs: + return "无近24小时聊天记录" + + lines = ["近24小时聊天记录:"] + for log in (logs if isinstance(logs, list) else [logs])[:10]: + role = "买家" if log.get("role") == "buyer" else "客服" + msg = str(log.get("message", ""))[:100] + lines.append(f" [{log.get('time', '')}] {role}: {msg}") + return "\n".join(lines) + except Exception as e: + logger.exception("Query chat history failed: %s", buyer_nick) + return f"查询聊天记录失败: {e}" + + def get_agent_tools(self) -> list[dict]: + return [ + { + "name": "taobao_query_order", + "description": "查询淘宝订单详情，获取订单状态、金额、买家、商品列表等信息", + "parameters": { + "type": "object", + "properties": { + "order_id": {"type": "string", "description": "淘宝订单号(tid)"}, + }, + "required": ["order_id"], + }, + "handler": self.query_order, + }, + { + "name": "taobao_query_logistics", + "description": "查询淘宝订单物流轨迹", + "parameters": { + "type": "object", + "properties": { + "order_id": {"type": "string", "description": "淘宝订单号(tid)"}, + }, + "required": ["order_id"], + }, + "handler": self.query_logistics, + }, + { + "name": "taobao_query_item", + "description": "查询淘宝商品详情，包括标题、价格、描述等", + "parameters": { + "type": "object", + "properties": { + "item_id": {"type": "string", "description": "淘宝商品ID(num_iid)"}, + }, + "required": ["item_id"], + }, + "handler": self.query_item, + }, + { + "name": "taobao_query_refund", + "description": "查询淘宝退款/售后状态和详情", + "parameters": { + "type": "object", + "properties": { + "refund_id": {"type": "string", "description": "退款单号(refund_id)"}, + }, + "required": ["refund_id"], + }, + "handler": self.query_refund, + }, + { + "name": "taobao_query_buyer_tags", + "description": "查询买家标签（VIP/黑名单等）", + "parameters": { + "type": "object", + "properties": { + "buyer_nick": {"type": "string", "description": "买家淘宝昵称"}, + }, + "required": ["buyer_nick"], + }, + "handler": self.query_buyer_tags, + }, + { + "name": "taobao_transfer_human", + "description": "将当前会话转接人工客服", + "parameters": { + "type": "object", + "properties": { + "buyer_nick": {"type": "string", "description": "买家淘宝昵称"}, + "reason": { + "type": "string", + "description": "转接原因（可选）", + }, + }, + "required": ["buyer_nick"], + }, + "handler": self.transfer_human, + }, + { + "name": "taobao_query_chat_history", + "description": "查询与指定买家的近24小时聊天记录", + "parameters": { + "type": "object", + "properties": { + "buyer_nick": {"type": "string", "description": "买家淘宝昵称"}, + }, + "required": ["buyer_nick"], + }, + "handler": self.query_chat_history, + }, + ] + + async def execute_agent_tool(self, tool_name: str, params: dict, context: dict) -> dict: + tool_map = { + "taobao_query_order": self.query_order, + "taobao_query_logistics": self.query_logistics, + "taobao_query_item": self.query_item, + "taobao_query_refund": self.query_refund, + "taobao_query_buyer_tags": self.query_buyer_tags, + "taobao_transfer_human": self.transfer_human, + "taobao_query_chat_history": self.query_chat_history, + } + handler = tool_map.get(tool_name) + if handler is None: + return {"success": False, "error": f"Unknown tool: {tool_name}"} + try: + session = context.get("session") or context.get("access_token") + result = await handler(**params, session=session) + return {"success": True, "result": result} + except Exception as e: + return {"success": False, "error": str(e)} diff --git a/backend/package/yuxi/channel/extensions/taobao/types.py b/backend/package/yuxi/channel/extensions/taobao/types.py new file mode 100644 index 00000000..e91c2915 --- /dev/null +++ b/backend/package/yuxi/channel/extensions/taobao/types.py @@ -0,0 +1,76 @@ +from dataclasses import dataclass, field +from datetime import datetime, timedelta + + +@dataclass +class TaobaoAccount: + account_id: str = "default" + app_key: str = "" + app_secret: str = "" + sign_method: str = "md5" + + access_token: str = "" + refresh_token: str = "" + token_expires_at: datetime | None = None + + callback_url: str = "" + + rsa_public_key: str = "" + rsa_private_key: str = "" + encrypt_messages: bool = False + + seller_nick: str = "" + name: str = "" + + dm_policy: str = "open" + allow_from: list[str] = field(default_factory=list) + + timeout: int = 10 + max_retries: int = 3 + sandbox: bool = False + + def is_configured(self) -> bool: + return bool(self.app_key and self.app_secret and self.callback_url) + + def is_token_expired(self) -> bool: + if not self.token_expires_at: + return True + return datetime.now() > self.token_expires_at - timedelta(minutes=10) + + def has_valid_token(self) -> bool: + return bool(self.access_token) and not self.is_token_expired() + + def to_dict(self) -> dict: + return { + "account_id": self.account_id, + "app_key": self.app_key, + "app_secret": self.app_secret, + "sign_method": self.sign_method, + "access_token": self.access_token, + "refresh_token": self.refresh_token, + "callback_url": self.callback_url, + "rsa_public_key": self.rsa_public_key, + "rsa_private_key": self.rsa_private_key, + "encrypt_messages": self.encrypt_messages, + "seller_nick": self.seller_nick, + "name": self.name, + "dm_policy": self.dm_policy, + "allow_from": self.allow_from, + "timeout": self.timeout, + "sandbox": self.sandbox, + "enabled": True, + } + + +@dataclass +class InboundTaobaoMessage: + msg_id: str + buyer_nick: str + seller_nick: str = "" + content: str = "" + msg_type: str = "text" + order_id: str | None = None + trade_id: str | None = None + item_id: str | None = None + timestamp: datetime | None = None + raw_payload: dict = field(default_factory=dict) diff --git a/backend/package/yuxi/channel/extensions/taobao/webhook.py b/backend/package/yuxi/channel/extensions/taobao/webhook.py new file mode 100644 index 00000000..52874da9 --- /dev/null +++ b/backend/package/yuxi/channel/extensions/taobao/webhook.py @@ -0,0 +1,127 @@ +import json +import logging +import time +from collections import defaultdict + +from fastapi import APIRouter, Request, Response + +from yuxi.channel.extensions.taobao.config import TaobaoConfigAdapter +from yuxi.channel.extensions.taobao.crypto import decrypt_message, verify_top_sign +from yuxi.channel.extensions.taobao.gateway import get_current_gateway +from yuxi.channel.extensions.taobao.message import parse_to_unified_message + +logger = logging.getLogger(__name__) + +router = APIRouter(prefix="/webhook/taobao", tags=["taobao"]) + +_config_adapter = TaobaoConfigAdapter() + +_dedupe_sets: dict[str, dict[str, float]] = defaultdict(dict) + +DEDUPE_TTL_SECONDS = 300 + + +def _get_config(account_id: str = "default"): + config = _config_adapter.resolve_account(account_id) + return _config_adapter.build_account(account_id, config) + + +def _dedupe_check(account_id: str, msg_id: str) -> bool: + now = time.monotonic() + dedupe = _dedupe_sets[account_id] + if msg_id in dedupe and now - dedupe[msg_id] < DEDUPE_TTL_SECONDS: + return True + dedupe[msg_id] = now + cutoff = now - DEDUPE_TTL_SECONDS + _dedupe_sets[account_id] = {k: v for k, v in dedupe.items() if v >= cutoff} + return False + + +@router.get("/callback") +async def verify_callback_default(request: Request): + return await _verify_callback(request, "default") + + +@router.get("/account/{account_id}/callback") +async def verify_callback_account(request: Request, account_id: str): + return await _verify_callback(request, account_id) + + +async def _verify_callback(request: Request, account_id: str): + config = _get_config(account_id) + if not config.is_configured(): + return Response(status_code=503, content="not configured") + + params = dict(request.query_params) + sign = params.pop("sign", "") + computed_ok = verify_top_sign(params, config.app_secret, sign, config.sign_method) + if computed_ok: + return Response(content="success") + return Response(status_code=403, content="sign error") + + +@router.post("/callback") +async def handle_callback_default(request: Request): + return await _handle_callback(request, "default") + + +@router.post("/account/{account_id}/callback") +async def handle_callback_account(request: Request, account_id: str): + return await _handle_callback(request, account_id) + + +async def _handle_callback(request: Request, account_id: str): + gateway = get_current_gateway() + if gateway is None: + return Response(status_code=503, content="gateway not started") + + queue = gateway.get_message_queue() + if queue is None: + return Response(status_code=503, content="gateway not started") + + config = _get_config(account_id) + if not config.is_configured(): + return Response(status_code=503, content="not configured") + + try: + body = await request.body() + content_type = request.headers.get("content-type", "") + + if "application/x-www-form-urlencoded" in content_type or "multipart/form-data" in content_type: + form = await request.form() + params = {k: v for k, v in form.items()} + else: + params = json.loads(body.decode("utf-8")) + + sign = params.pop("sign", "") + if not verify_top_sign(params, config.app_secret, sign, config.sign_method): + return Response(status_code=403, content="sign verification failed") + + message_data = params + if config.encrypt_messages and "encrypt_data" in params: + decrypted = decrypt_message( + encrypt_data=params.get("encrypt_data", ""), + encrypt_key=params.get("encrypt_key", ""), + sign=params.get("encrypt_sign", ""), + timestamp=params.get("timestamp", ""), + app_secret=config.app_secret, + rsa_private_key_pem=config.rsa_private_key, + ) + if decrypted is None: + return Response(status_code=400, content="decrypt failed") + message_data = json.loads(decrypted) + + msg_id = message_data.get("msg_id", "") or message_data.get("id", "") + if msg_id and _dedupe_check(account_id, msg_id): + logger.debug("Duplicate message ignored: %s", msg_id) + return Response(content="ok") + + unified = parse_to_unified_message(message_data) + if unified: + await gateway.enqueue_message(unified) + + return Response(content="ok") + + except Exception: + logger.exception("Taobao webhook processing error") + return Response(status_code=500, content="internal error")