From bb8ade9e2be91755c652cd764aeb6aea0e2d48e0 Mon Sep 17 00:00:00 2001 From: Kris <2893855659@qq.com> Date: Thu, 21 May 2026 11:05:40 +0800 Subject: [PATCH] =?UTF-8?q?feat(channel):=20=E6=B7=BB=E5=8A=A0=E4=BA=AC?= =?UTF-8?q?=E4=B8=9C=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 新增京东（JD）渠道扩展，支持在 Yuxi 平台中集成京东客服渠道。 包含以下功能模块： - client: 京东 API 客户端封装 - config: 渠道配置管理 - gateway: SSE/WebSocket 网关接入 - webhook: Webhook 事件处理 - outbound: 外发消息管理 - pairing: 用户配对与绑定 - security: 安全校验 - signature: 请求签名验证 - crypto: 加解密处理 - dedupe: 消息去重 - monitor: 渠道状态监控 - status: 会话状态管理 - session: 会话管理 - business: 业务逻辑处理 - types: 类型定义 --- .../yuxi/channel/extensions/jd/__init__.py | 403 ++++++++++++++++++ .../yuxi/channel/extensions/jd/business.py | 200 +++++++++ .../yuxi/channel/extensions/jd/client.py | 83 ++++ .../yuxi/channel/extensions/jd/config.py | 102 +++++ .../yuxi/channel/extensions/jd/crypto.py | 28 ++ .../yuxi/channel/extensions/jd/dedupe.py | 38 ++ .../yuxi/channel/extensions/jd/errors.py | 20 + .../yuxi/channel/extensions/jd/format.py | 19 + .../yuxi/channel/extensions/jd/gateway.py | 239 +++++++++++ .../yuxi/channel/extensions/jd/monitor.py | 63 +++ .../yuxi/channel/extensions/jd/outbound.py | 147 +++++++ .../yuxi/channel/extensions/jd/pairing.py | 57 +++ .../yuxi/channel/extensions/jd/plugin.json | 27 ++ .../yuxi/channel/extensions/jd/security.py | 36 ++ .../yuxi/channel/extensions/jd/session.py | 128 ++++++ .../yuxi/channel/extensions/jd/signature.py | 79 ++++ .../yuxi/channel/extensions/jd/status.py | 83 ++++ .../yuxi/channel/extensions/jd/types.py | 57 +++ .../yuxi/channel/extensions/jd/webhook.py | 205 +++++++++ 19 files changed, 2014 insertions(+) create mode 100644 backend/package/yuxi/channel/extensions/jd/__init__.py create mode 100644 backend/package/yuxi/channel/extensions/jd/business.py create mode 100644 backend/package/yuxi/channel/extensions/jd/client.py create mode 100644 backend/package/yuxi/channel/extensions/jd/config.py create mode 100644 backend/package/yuxi/channel/extensions/jd/crypto.py create mode 100644 backend/package/yuxi/channel/extensions/jd/dedupe.py create mode 100644 backend/package/yuxi/channel/extensions/jd/errors.py create mode 100644 backend/package/yuxi/channel/extensions/jd/format.py create mode 100644 backend/package/yuxi/channel/extensions/jd/gateway.py create mode 100644 backend/package/yuxi/channel/extensions/jd/monitor.py create mode 100644 backend/package/yuxi/channel/extensions/jd/outbound.py create mode 100644 backend/package/yuxi/channel/extensions/jd/pairing.py create mode 100644 backend/package/yuxi/channel/extensions/jd/plugin.json create mode 100644 backend/package/yuxi/channel/extensions/jd/security.py create mode 100644 backend/package/yuxi/channel/extensions/jd/session.py create mode 100644 backend/package/yuxi/channel/extensions/jd/signature.py create mode 100644 backend/package/yuxi/channel/extensions/jd/status.py create mode 100644 backend/package/yuxi/channel/extensions/jd/types.py create mode 100644 backend/package/yuxi/channel/extensions/jd/webhook.py diff --git a/backend/package/yuxi/channel/extensions/jd/__init__.py b/backend/package/yuxi/channel/extensions/jd/__init__.py new file mode 100644 index 00000000..673a7038 --- /dev/null +++ b/backend/package/yuxi/channel/extensions/jd/__init__.py @@ -0,0 +1,403 @@ +import logging + +from yuxi.channel.capabilities import ChannelCapabilities +from yuxi.channel.context import ChannelContext +from yuxi.channel.extensions.base import BaseChannelPlugin +from yuxi.channel.extensions.jd.business import JDBusiness +from yuxi.channel.extensions.jd.client import JOSClient +from yuxi.channel.extensions.jd.config import JDConfig +from yuxi.channel.extensions.jd.dedupe import MessageDeduplicator +from yuxi.channel.extensions.jd.gateway import JDGateway +from yuxi.channel.extensions.jd.monitor import JDMonitor +from yuxi.channel.extensions.jd.outbound import JDOutbound +from yuxi.channel.extensions.jd.pairing import JDPairing +from yuxi.channel.extensions.jd.security import JDSecurity +from yuxi.channel.extensions.jd.session import JDSession +from yuxi.channel.extensions.jd.status import JDStatus +from yuxi.channel.extensions.jd.webhook import set_plugin +from yuxi.channel.plugins.registry import ChannelPluginRegistry +from yuxi.channel.protocols import ChannelAccountSnapshot + +logger = logging.getLogger(__name__) + + +class JDPlugin(BaseChannelPlugin): + id = "jd" + name = "京东/咚咚" + order = 90 + label = "京东咚咚 (Webhook + JOS API)" + aliases = ["jd", "dongdong", "京东", "京麦"] + resolve_reply_to_mode = "off" + + def __init__(self): + super().__init__() + self._config_adapter = JDConfig() + self._gateway: JDGateway | None = None + self._outbound: JDOutbound | None = None + self._status: JDStatus | None = None + self._security: JDSecurity | None = None + self._pairing = JDPairing() + self._business: JDBusiness | None = None + self._session: JDSession | None = None + self._deduplicator = MessageDeduplicator() + self._monitor = JDMonitor() + + @property + def capabilities(self) -> ChannelCapabilities: + return ChannelCapabilities( + chat_types=["direct"], + message_types=["text", "image", "link", "product_card"], + reactions=False, + typing_indicator=False, + threads=False, + edit=False, + unsend=True, + reply=False, + media=True, + effects=False, + native_commands=False, + polls=False, + group_management=False, + streaming=False, + streaming_mode=None, + block_streaming=False, + tts=None, + ) + + def list_account_ids(self, config: dict | None = None) -> list[str]: + return self._config_adapter.list_account_ids(config) + + async def resolve_account(self, account_id: str = "default") -> dict: + account = self._config_adapter.resolve_account(account_id) + return { + "account_id": account.account_id, + "app_key": account.app_key, + "app_secret": account.app_secret, + "shop_id": account.shop_id, + "access_token": account.access_token, + "refresh_token": account.refresh_token, + "dm_policy": account.dm_policy, + "allow_from": account.allow_from, + } + + def is_configured(self, account: dict | None = None) -> bool: + return self._config_adapter.is_configured(account) + + def config_schema(self) -> dict: + return self._config_adapter.config_schema() + + async def start(self, ctx) -> object: + self._gateway = JDGateway() + result = await self._gateway.start(ctx) + self._outbound = JDOutbound(self._gateway) + self._status = JDStatus(self._gateway) + + account = self._config_adapter.resolve_account() + self._security = JDSecurity(account) + + client = JOSClient( + app_key=account.app_key, + app_secret=account.app_secret, + shop_id=account.shop_id, + ) + self._business = JDBusiness(client, account.shop_id) + self._session = JDSession(client) + await self._session.start() + + set_plugin(self) + return result + + async def stop(self, ctx) -> None: + if self._session: + await self._session.stop() + self._session = None + if self._business: + if hasattr(self._business, "_client") and self._business._client: + await self._business._client.close() + self._business = None + if self._gateway: + await self._gateway.stop(ctx) + self._gateway = None + self._outbound = None + self._status = None + + async def on_config_changed(self, prev_cfg: dict, next_cfg: dict, account_id: str) -> None: + if prev_cfg != next_cfg: + logger.info("JD config changed, reloading...") + ctx = ChannelContext(channel_type="jd", account_id=account_id, config=next_cfg) + await self.stop(ctx) + await self.start(ctx) + + 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, + chat_id: str | None = None, + ) -> None: + if not content: + return + + if chat_id is None and self._gateway: + chat_id = self._gateway.get_cached_chat_id(target_id) + + if self._outbound: + await self._outbound.send_text( + target_id, content, reply_to_id=reply_to_id, thread_id=thread_id, account_id=account_id, chat_id=chat_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, + chat_id: str | None = None, + ) -> None: + if chat_id is None and self._gateway: + chat_id = self._gateway.get_cached_chat_id(target_id) + + if self._outbound: + await self._outbound.send_media( + target_id, media_url, media_type, reply_to_id=reply_to_id, thread_id=thread_id, chat_id=chat_id + ) + + async def send_link( + self, + target_id: str, + url: str, + text: str, + *, + chat_id: str | None = None, + ) -> None: + if chat_id is None and self._gateway: + chat_id = self._gateway.get_cached_chat_id(target_id) + + if self._outbound: + await self._outbound.send_link(target_id, url, text, chat_id=chat_id) + + async def send_product_card( + self, + target_id: str, + url: str, + pid: str, + *, + chat_id: str | None = None, + ) -> None: + if chat_id is None and self._gateway: + chat_id = self._gateway.get_cached_chat_id(target_id) + + if self._outbound: + await self._outbound.send_product_card(target_id, url, pid, chat_id=chat_id) + + def build_system_prompt(self, context) -> str | None: + return ( + "你是京东店铺的 AI 客服助手，当前对话通过京东咚咚平台进行。\n" + "\n" + "重要约束：\n" + "- 你只能回复纯文本消息，无法直接发送链接/按钮/富媒体格式（但可使用推荐工具发送商品链接和卡片）\n" + "- 请使用礼貌、专业的客服语气\n" + "- 回复简洁清晰，避免过长的段落（用户使用手机阅读）\n" + "- 你可以直接查询订单状态、物流信息、商品详情和用户评价\n" + "- 你可以通过工具向用户发送商品推荐链接或商品卡片\n" + "- 对于复杂售后问题，建议转接人工客服处理\n" + "\n" + "你可以帮助用户：\n" + "- 回答商品相关咨询（规格、价格、优惠活动、用户评价）\n" + "- 查询订单状态（提供订单号即可）\n" + "- 查询物流信息（提供订单号即可）\n" + "- 了解售后/退款进度\n" + "- 根据用户需求推荐商品并发送链接/卡片" + ) + + @property + def channel_format_instructions(self) -> str | None: + return ( + "京东咚咚支持纯文本消息（最大 4096 字符）、图片消息、链接消息和商品卡片消息。" + "请移除 Markdown 格式（加粗、斜体、代码块），转为纯文本发送。" + "图片通过 URL 发送，链接和商品推荐通过专用工具发送。" + "用户使用手机阅读，请保持回复简洁。" + ) + + async def probe(self, account: dict | None = None) -> bool: + if self._status: + return await self._status.probe(account) + return False + + def build_summary(self, snapshot: object) -> dict: + if self._status: + return self._status.build_summary(snapshot) + return {} + + def build_account_snapshot( + self, + account: dict, + config: dict, + runtime: ChannelAccountSnapshot | None = None, + probe_result: object | None = None, + audit: object | None = None, + ) -> ChannelAccountSnapshot: + if self._status: + return self._status.build_account_snapshot(account, config, runtime, probe_result, audit) + return ChannelAccountSnapshot(account_id="default") + + def resolve_dm_policy(self) -> dict: + if self._security: + return {"mode": self._security.resolve_dm_policy(), "allow_from": []} + return {"mode": "open", "allow_from": []} + + async def check_allowlist(self, peer_id: str, channel_type: str) -> bool: + if self._security: + return self._security.check_allowlist(peer_id, channel_type) + return True + + async def generate_code(self, peer_id: str) -> str: + code = self._pairing.generate_code(peer_id) + return code or "" + + async def verify_code(self, peer_id: str, code: str) -> bool: + return self._pairing.verify(peer_id, code) + + def get_agent_tools(self) -> list: + return [ + { + "name": "jd_query_order", + "description": "查询京东店铺订单。可按订单号精确查询，或按时间范围搜索最近的订单列表。", + "parameters": { + "type": "object", + "properties": { + "order_id": {"type": "string", "description": "订单号（精确查询）"}, + "start_time": {"type": "string", "description": "查询起始时间，格式 YYYY-MM-DD"}, + "end_time": {"type": "string", "description": "查询结束时间，格式 YYYY-MM-DD"}, + }, + }, + }, + { + "name": "jd_query_logistics", + "description": "查询京东订单的物流轨迹信息。需要提供订单号。", + "parameters": { + "type": "object", + "properties": { + "order_id": {"type": "string", "description": "订单号"}, + }, + "required": ["order_id"], + }, + }, + { + "name": "jd_query_product", + "description": "查询京东店铺商品的规格、价格等信息。需要提供商品 SKU ID。", + "parameters": { + "type": "object", + "properties": { + "sku_id": {"type": "string", "description": "商品 SKU ID（可从商品链接中获取）"}, + }, + "required": ["sku_id"], + }, + }, + { + "name": "jd_query_refund", + "description": "查询京东订单的售后/退款申请状态。需要提供订单号。", + "parameters": { + "type": "object", + "properties": { + "order_id": {"type": "string", "description": "订单号"}, + }, + "required": ["order_id"], + }, + }, + { + "name": "jd_query_comments", + "description": "查询京东店铺商品的好评率、用户评价等信息。需要提供商品 SKU ID。", + "parameters": { + "type": "object", + "properties": { + "sku_id": {"type": "string", "description": "商品 SKU ID"}, + }, + "required": ["sku_id"], + }, + }, + { + "name": "jd_send_product_recommend", + "description": "向用户发送商品推荐。可选择发送商品链接消息或商品卡片消息。", + "parameters": { + "type": "object", + "properties": { + "user_id": {"type": "string", "description": "用户咚咚 ID"}, + "url": {"type": "string", "description": "商品详情页链接"}, + "text": {"type": "string", "description": "链接描述文字（仅 link 类型需要）"}, + "sku_id": {"type": "string", "description": "商品 SKU ID（仅 card 类型需要）"}, + "msg_type": { + "type": "string", + "enum": ["link", "card"], + "description": "消息类型：link=文字链接, card=商品卡片", + }, + }, + "required": ["user_id", "url", "msg_type"], + }, + }, + { + "name": "jd_evaluate_cs", + "description": "邀请用户对当前客服服务进行满意度评价。需要评估分数 1-5 分。", + "parameters": { + "type": "object", + "properties": { + "chat_id": {"type": "string", "description": "当前会话 chat_id"}, + "score": {"type": "integer", "description": "评价分数 1-5"}, + }, + "required": ["chat_id", "score"], + }, + }, + ] + + async def execute_agent_tool(self, tool_name: str, params: dict, context: dict) -> dict: + if self._business is None: + return {"success": False, "error": "JD 业务模块未初始化"} + + token = self._gateway.get_token() if self._gateway else None + if not token: + return {"success": False, "error": "JD Token 不可用"} + + if tool_name == "jd_query_order": + order_id = params.get("order_id", "") + if order_id: + return await self._business.get_order_detail(order_id, token) + else: + return await self._business.query_order( + start_time=params.get("start_time"), + end_time=params.get("end_time"), + access_token=token, + ) + + if tool_name == "jd_query_logistics": + return await self._business.query_logistics(params.get("order_id", ""), token) + + if tool_name == "jd_query_product": + return await self._business.query_product(params.get("sku_id", ""), token) + + if tool_name == "jd_query_refund": + return await self._business.query_refund(params.get("order_id", ""), token) + + if tool_name == "jd_query_comments": + return await self._business.query_comments(params.get("sku_id", ""), access_token=token) + + if tool_name == "jd_send_product_recommend": + user_id = params.get("user_id", "") + msg_type = params.get("msg_type", "link") + url = params.get("url", "") + if msg_type == "link": + await self.send_link(user_id, url, params.get("text", "")) + elif msg_type == "card": + await self.send_product_card(user_id, url, params.get("sku_id", "")) + return {"success": True, "msg_type": msg_type} + + if tool_name == "jd_evaluate_cs": + return await self._business.evaluate_cs(params.get("chat_id", ""), params.get("score", 5), token) + + return {"success": False, "error": f"未知工具: {tool_name}"} + + +ChannelPluginRegistry.register(JDPlugin()) diff --git a/backend/package/yuxi/channel/extensions/jd/business.py b/backend/package/yuxi/channel/extensions/jd/business.py new file mode 100644 index 00000000..8a407c1e --- /dev/null +++ b/backend/package/yuxi/channel/extensions/jd/business.py @@ -0,0 +1,200 @@ +import logging + +from yuxi.channel.extensions.jd.client import JOSClient + +logger = logging.getLogger(__name__) + + +class JDBusiness: + def __init__(self, client: JOSClient, shop_id: str): + self._client = client + self._shop_id = shop_id + + async def query_order( + self, + order_id: str | None = None, + start_time: str | None = None, + end_time: str | None = None, + page: int = 1, + page_size: int = 10, + access_token: str = "", + ) -> dict: + biz_params = { + "optional_fields": "orderInfo", + "page": str(page), + "page_size": str(page_size), + } + if order_id: + biz_params["order_id"] = order_id + if start_time: + biz_params["start_date"] = start_time + if end_time: + biz_params["end_date"] = end_time + + try: + data = await self._client.call("jingdong.pop.order.search", biz_params, access_token) + return {"success": True, "data": self._flatten_order_result(data)} + except Exception as e: + logger.error("JD order query failed: %s", e) + return {"success": False, "error": str(e)} + + async def get_order_detail(self, order_id: str, access_token: str = "") -> dict: + biz_params = { + "order_id": order_id, + "optional_fields": "orderInfo,paymentInfo,consigneeInfo", + } + try: + data = await self._client.call("jingdong.pop.order.get", biz_params, access_token) + return {"success": True, "data": self._flatten_order_detail(data)} + except Exception as e: + logger.error("JD order detail failed: %s", e) + return {"success": False, "error": str(e)} + + async def query_logistics(self, order_id: str, access_token: str = "") -> dict: + biz_params = {"order_id": order_id} + + try: + data = await self._client.call("jingdong.etms.trace.get", biz_params, access_token) + trace_result = data.get("jingdong_etms_trace_get_responce", {}).get("traceResult", {}) + traces = trace_result.get("traceList", []) if isinstance(trace_result, dict) else [] + return { + "success": True, + "data": { + "order_id": order_id, + "waybill_code": trace_result.get("waybillCode", ""), + "traces": traces, + }, + } + except Exception as e: + logger.error("JD logistics query failed: %s", e) + return {"success": False, "error": str(e)} + + async def query_product(self, sku_id: str, access_token: str = "") -> dict: + biz_params = { + "sku": sku_id, + "queryExts": "warebase,wareimage", + } + + try: + data = await self._client.call("jingdong.ware.product.get", biz_params, access_token) + ware_result = data.get("jingdong_ware_product_get_responce", {}).get("product", {}) + if isinstance(ware_result, dict): + return { + "success": True, + "data": { + "sku_id": ware_result.get("sku", ""), + "name": ware_result.get("name", ""), + "brand_name": ware_result.get("brandName", ""), + "category": ware_result.get("category", ""), + "introduction": ware_result.get("introduction", ""), + "weight": ware_result.get("weight", ""), + "sale_date": ware_result.get("saleDate", ""), + "upc": ware_result.get("upc", ""), + "image_url": ware_result.get("imageUrl", ""), + }, + } + return {"success": False, "error": "未找到商品信息"} + except Exception as e: + logger.error("JD product query failed: %s", e) + return {"success": False, "error": str(e)} + + async def query_product_price(self, sku_ids: list[str], access_token: str = "") -> dict: + biz_params = {"sku": ",".join(sku_ids)} + + try: + data = await self._client.call("jingdong.ware.price.get", biz_params, access_token) + price_list = data.get("jingdong_ware_price_get_responce", {}).get("priceList", []) + return {"success": True, "data": price_list} + except Exception as e: + logger.error("JD product price query failed: %s", e) + return {"success": False, "error": str(e)} + + async def query_refund(self, order_id: str, access_token: str = "") -> dict: + biz_params = {"order_id": order_id} + try: + data = await self._client.call("jingdong.pop.afs.soa.refundapply.get", biz_params, access_token) + result = data.get("jingdong_pop_afs_soa_refundapply_get_responce", {}).get("queryResult", {}) + return {"success": True, "data": result} + except Exception as e: + logger.error("JD refund query failed: %s", e) + return {"success": False, "error": str(e)} + + async def set_cs_status(self, status: str, access_token: str = "") -> dict: + biz_params = {"status": status} + try: + await self._client.call("jingdong.im.setStatus", biz_params, access_token) + return {"success": True, "status": status} + except Exception as e: + logger.error("JD set cs status failed: %s", e) + return {"success": False, "error": str(e)} + + async def get_cs_status(self, access_token: str = "") -> dict: + try: + data = await self._client.call("jingdong.im.getStatus", {}, access_token) + return {"success": True, "data": data} + except Exception as e: + logger.error("JD get cs status failed: %s", e) + return {"success": False, "error": str(e)} + + async def get_user_info(self, user_id: str, access_token: str = "") -> dict: + biz_params = {"user_id": user_id} + + try: + data = await self._client.call("jingdong.im.getUserInfo", biz_params, access_token) + user_info = data.get("jingdong_im_getUserInfo_responce", {}).get("result", {}) + return {"success": True, "data": user_info if isinstance(user_info, dict) else {}} + except Exception as e: + logger.error("JD get user info failed: %s", e) + return {"success": False, "error": str(e)} + + async def query_comments(self, sku_id: str, page: int = 1, page_size: int = 10, access_token: str = "") -> dict: + biz_params = {"sku": sku_id, "page": str(page), "pageSize": str(page_size)} + + try: + data = await self._client.call("jingdong.comment.products.get", biz_params, access_token) + result = data.get("jingdong_comment_products_get_responce", {}).get("result", {}) + return {"success": True, "data": result} + except Exception as e: + logger.error("JD comment query failed: %s", e) + return {"success": False, "error": str(e)} + + async def evaluate_cs(self, chat_id: str, score: int, access_token: str = "") -> dict: + biz_params = {"chat_id": chat_id, "score": str(score)} + try: + await self._client.call("jingdong.im.evaluate", biz_params, access_token) + return {"success": True, "score": score} + except Exception as e: + logger.error("JD evaluate cs failed: %s", e) + return {"success": False, "error": str(e)} + + def _flatten_order_result(self, data: dict) -> list[dict]: + result = data.get("jingdong_pop_order_search_responce", {}).get("searchorderinforesult", {}) + order_list = result.get("orderInfoList", []) + if isinstance(order_list, dict): + order_list = [order_list] + return [ + { + "order_id": o.get("orderId", ""), + "order_state": o.get("orderState", ""), + "order_total_price": o.get("orderTotalPrice", ""), + "order_create_time": o.get("orderCreateTime", ""), + "order_payment_time": o.get("orderPaymentTime", ""), + } + for o in order_list + ] + + def _flatten_order_detail(self, data: dict) -> dict: + result = data.get("jingdong_pop_order_get_responce", {}).get("orderDetailInfo", {}) + order_info = result.get("orderInfo", {}) if isinstance(result, dict) else {} + return { + "order_id": order_info.get("orderId", ""), + "order_state": order_info.get("orderState", ""), + "order_total_price": order_info.get("orderTotalPrice", ""), + "freight_price": order_info.get("freightPrice", ""), + "seller_discount": order_info.get("sellerDiscount", ""), + "order_create_time": order_info.get("orderCreateTime", ""), + "order_payment_time": order_info.get("orderPaymentTime", ""), + "consignee_name": order_info.get("consigneeInfo", {}).get("fullname", ""), + "consignee_mobile": order_info.get("consigneeInfo", {}).get("mobile", ""), + "consignee_address": order_info.get("consigneeInfo", {}).get("fullAddress", ""), + } diff --git a/backend/package/yuxi/channel/extensions/jd/client.py b/backend/package/yuxi/channel/extensions/jd/client.py new file mode 100644 index 00000000..815a5a2c --- /dev/null +++ b/backend/package/yuxi/channel/extensions/jd/client.py @@ -0,0 +1,83 @@ +import asyncio +import logging + +import httpx + +from yuxi.channel.extensions.jd.signature import build_signed_params +from yuxi.channel.extensions.jd.types import JOS_API_GATEWAY, JOS_MAX_RETRIES, JOS_RETRY_BASE_DELAY, OutboundResult + +logger = logging.getLogger(__name__) + + +class JOSClient: + def __init__(self, app_key: str, app_secret: str, shop_id: str): + self._app_key = app_key + self._app_secret = app_secret + self._shop_id = shop_id + self._http = httpx.AsyncClient(timeout=15.0) + + async def close(self): + await self._http.aclose() + + async def call(self, method: str, biz_params: dict, access_token: str) -> dict: + params = build_signed_params( + method=method, + biz_params=biz_params, + app_key=self._app_key, + app_secret=self._app_secret, + access_token=access_token, + ) + + for attempt in range(JOS_MAX_RETRIES): + try: + resp = await self._http.post(JOS_API_GATEWAY, data=params) + resp.raise_for_status() + data = resp.json() + + error_response = data.get("error_response") + if error_response: + code = error_response.get("code") + msg = error_response.get("zh_desc", "Unknown error") + logger.error("JOS API error [%s]: %s", code, msg) + raise ValueError(f"JOS API error {code}: {msg}") + + return data + + except (httpx.TimeoutException, httpx.NetworkError) as e: + logger.warning("JOS API retry %d/%d: %s", attempt + 1, JOS_MAX_RETRIES, e) + if attempt == JOS_MAX_RETRIES - 1: + raise + await asyncio.sleep(JOS_RETRY_BASE_DELAY * (2**attempt)) + + raise RuntimeError("JOS API max retries exceeded") + + async def send_msg( + self, + chat_id: str, + to_id: str, + msg_type: int, + content: str, + access_token: str, + ) -> OutboundResult: + biz_params = { + "chat_id": chat_id, + "from_id": self._shop_id, + "to_id": to_id, + "msg_type": msg_type, + "content": content, + } + + try: + data = await self.call("jingdong.im.sendMsg", biz_params, access_token) + result = data.get("jingdong_im_sendMsg_responce", {}).get("result", {}) + return OutboundResult( + message_id=result.get("msg_id", ""), + success=True, + ) + except Exception as e: + return OutboundResult( + message_id="", + success=False, + error_code="JOS_ERROR", + error_msg=str(e), + ) diff --git a/backend/package/yuxi/channel/extensions/jd/config.py b/backend/package/yuxi/channel/extensions/jd/config.py new file mode 100644 index 00000000..abb96ca0 --- /dev/null +++ b/backend/package/yuxi/channel/extensions/jd/config.py @@ -0,0 +1,102 @@ +import os + +from yuxi.channel.extensions.jd.types import JDAccount + + +class JDConfig: + ENV_MAP = { + "app_key": "JD_APP_KEY", + "app_secret": "JD_APP_SECRET", + "shop_id": "JD_SHOP_ID", + "dm_policy": "JD_DM_POLICY", + } + + def list_account_ids(self, config: dict | None = None) -> list[str]: + channels = (config or {}).get("channels", {}) + jd_config = channels.get("jd", {}) + accounts = jd_config.get("accounts", {}) + if accounts: + return list(accounts.keys()) + if self._env_or_config("app_key", config): + return ["default"] + return [] + + def resolve_account(self, account_id: str = "default", config: dict | None = None) -> JDAccount: + channels = (config or {}).get("channels", {}) + jd_config = channels.get("jd", {}) + accounts = jd_config.get("accounts", {}) + account_data = accounts.get(account_id, {}) + + return JDAccount( + account_id=account_id, + app_key=account_data.get("app_key") or self._env_or_config("app_key") or "", + app_secret=account_data.get("app_secret") or self._env_or_config("app_secret") or "", + shop_id=account_data.get("shop_id") or self._env_or_config("shop_id") or "", + access_token=account_data.get("access_token"), + refresh_token=account_data.get("refresh_token"), + webhook_token=account_data.get("webhook_token"), + dm_policy=account_data.get("dm_policy") or self._env_or_config("dm_policy") or "pairing", + webhook_sign_algorithm=account_data.get("webhook_sign_algorithm", "hmac-sha256"), + allow_from=account_data.get("allow_from", []), + ) + + def is_configured(self, account: dict | None = None) -> bool: + if account: + return bool(account.get("app_key") and account.get("app_secret") and account.get("shop_id")) + return bool( + self._env_or_config("app_key") and self._env_or_config("app_secret") and self._env_or_config("shop_id") + ) + + def _env_or_config(self, key: str, config: dict | None = None) -> str | None: + env_key = self.ENV_MAP.get(key, "") + if env_key: + val = os.getenv(env_key) + if val: + return val + return None + + def config_schema(self) -> dict: + return { + "$schema": "https://json-schema.org/draft-07/schema#", + "type": "object", + "title": "京东/咚咚渠道配置", + "properties": { + "appKey": { + "type": "string", + "title": "App Key", + "description": "京东开放平台应用 App Key", + }, + "appSecret": { + "type": "string", + "title": "App Secret", + "x-ui-password": True, + "description": "京东开放平台应用 App Secret", + }, + "shopId": { + "type": "string", + "title": "店铺 ID", + "description": "京东店铺 ID（用于标识发送方）", + }, + "webhookToken": { + "type": "string", + "title": "Webhook Token", + "x-ui-password": True, + "description": "咚咚消息推送签名验证 Token（如配置）", + }, + "dmPolicy": { + "type": "string", + "title": "DM 策略", + "enum": ["pairing", "allowlist", "open", "disabled"], + "default": "pairing", + "description": "私聊安全策略：pairing=配对码, allowlist=白名单, open=完全开放", + }, + "webhookSignAlgorithm": { + "type": "string", + "title": "Webhook 签名算法", + "enum": ["hmac-sha256", "md5"], + "default": "hmac-sha256", + "description": "咚咚 Webhook 消息推送签名算法", + }, + }, + "required": ["appKey", "appSecret", "shopId"], + } diff --git a/backend/package/yuxi/channel/extensions/jd/crypto.py b/backend/package/yuxi/channel/extensions/jd/crypto.py new file mode 100644 index 00000000..63c05ee8 --- /dev/null +++ b/backend/package/yuxi/channel/extensions/jd/crypto.py @@ -0,0 +1,28 @@ +import base64 +import hashlib +import json +import logging + +logger = logging.getLogger(__name__) + + +def decrypt_jd_message(encrypted_body: str, app_secret: str) -> dict | None: + try: + key = hashlib.md5(app_secret.encode("utf-8")).hexdigest()[:16].encode("utf-8") + raw = base64.b64decode(encrypted_body) + + iv = raw[:16] + ciphertext = raw[16:] + + from Crypto.Cipher import AES + + cipher = AES.new(key, AES.MODE_CBC, iv) + decrypted = cipher.decrypt(ciphertext) + + pad_len = decrypted[-1] + decrypted = decrypted[:-pad_len] + + return json.loads(decrypted.decode("utf-8")) + except Exception: + logger.warning("JD message AES decrypt failed") + return None diff --git a/backend/package/yuxi/channel/extensions/jd/dedupe.py b/backend/package/yuxi/channel/extensions/jd/dedupe.py new file mode 100644 index 00000000..ed7065b4 --- /dev/null +++ b/backend/package/yuxi/channel/extensions/jd/dedupe.py @@ -0,0 +1,38 @@ +import time +from collections import OrderedDict + + +class MessageDeduplicator: + MAX_SIZE = 10000 + TTL_SECONDS = 300 + + def __init__(self, max_size: int = MAX_SIZE, ttl: int = TTL_SECONDS): + self._cache: OrderedDict[str, float] = OrderedDict() + self._max_size = max_size + self._ttl = ttl + + def is_duplicate(self, msg_id: str) -> bool: + if not msg_id: + return False + + self._evict_expired() + + if msg_id in self._cache: + return True + + self._cache[msg_id] = time.time() + self._cache.move_to_end(msg_id) + + while len(self._cache) > self._max_size: + self._cache.popitem(last=False) + + return False + + def _evict_expired(self): + now = time.time() + expired = [k for k, ts in self._cache.items() if now - ts > self._ttl] + for k in expired: + del self._cache[k] + + def reset(self): + self._cache.clear() diff --git a/backend/package/yuxi/channel/extensions/jd/errors.py b/backend/package/yuxi/channel/extensions/jd/errors.py new file mode 100644 index 00000000..f103259d --- /dev/null +++ b/backend/package/yuxi/channel/extensions/jd/errors.py @@ -0,0 +1,20 @@ +from enum import StrEnum + + +class JDErrorCode(StrEnum): + TOKEN_INVALID = "JD_TOKEN_INVALID" + TOKEN_EXPIRED = "JD_TOKEN_EXPIRED" + JOS_API_ERROR = "JD_JOS_API_ERROR" + NETWORK_ERROR = "JD_NETWORK_ERROR" + SIGNATURE_ERROR = "JD_SIGNATURE_ERROR" + DECRYPT_ERROR = "JD_DECRYPT_ERROR" + WEBHOOK_REJECTED = "JD_WEBHOOK_REJECTED" + RATE_LIMITED = "JD_RATE_LIMITED" + + +class JDError(Exception): + def __init__(self, code: JDErrorCode, message: str, detail: dict | None = None): + self.code = code + self.message = message + self.detail = detail or {} + super().__init__(message) diff --git a/backend/package/yuxi/channel/extensions/jd/format.py b/backend/package/yuxi/channel/extensions/jd/format.py new file mode 100644 index 00000000..9e9ceaad --- /dev/null +++ b/backend/package/yuxi/channel/extensions/jd/format.py @@ -0,0 +1,19 @@ +import re + +from yuxi.channel.extensions.jd.types import JD_TEXT_MAX_LENGTH + + +def remove_markdown(text: str) -> str: + text = re.sub(r"\*\*(.+?)\*\*", r"\1", text) + text = re.sub(r"\*(.+?)\*", r"\1", text) + text = re.sub(r"`(.+?)`", r"\1", text) + text = re.sub(r"\[(.+?)\]\(.+?\)", r"\1", text) + text = re.sub(r"^\s*[-*+]\s+", "• ", text, flags=re.MULTILINE) + text = re.sub(r"^\s*\d+\.\s+", "", text, flags=re.MULTILINE) + return text.strip() + + +def truncate(text: str, max_chars: int = JD_TEXT_MAX_LENGTH) -> str: + if len(text) > max_chars: + return text[: max_chars - 3] + "..." + return text diff --git a/backend/package/yuxi/channel/extensions/jd/gateway.py b/backend/package/yuxi/channel/extensions/jd/gateway.py new file mode 100644 index 00000000..d96ae98d --- /dev/null +++ b/backend/package/yuxi/channel/extensions/jd/gateway.py @@ -0,0 +1,239 @@ +import asyncio +import json +import logging +import os +import time + +import httpx + +from yuxi.channel.extensions.jd.config import JDConfig +from yuxi.channel.extensions.jd.types import JDAccount, JOSToken + +logger = logging.getLogger(__name__) + +OAUTH_TOKEN_URL = "https://oauth.jd.com/oauth/token" +TOKEN_REFRESH_MARGIN = 300 +REFRESH_INTERVAL = 1800 +RETRY_MAX = 3 +RETRY_BASE_DELAY = 2 + +_TOKEN_STORE_DIR = os.path.join(os.path.dirname(__file__), ".token_cache") + + +class JDGateway: + def __init__(self): + self._account: JDAccount | None = None + self._queue: asyncio.Queue | None = None + self._cancel_event = asyncio.Event() + self._access_token: str | None = None + self._token_obj: JOSToken | None = None + self._token_expires_at: float = 0 + self._refresh_task: asyncio.Task | None = None + self._http: httpx.AsyncClient | None = None + self._running = False + self._token_lock = asyncio.Lock() + self._last_chat_id: dict[str, str] = {} + + async def start(self, ctx) -> object: + config = JDConfig() + account = config.resolve_account(ctx.account_id if hasattr(ctx, "account_id") else "default") + + if not account.is_configured(): + logger.warning("JD account not configured, skipping start") + return {"running": False, "reason": "not-configured"} + + self._account = account + self._queue = asyncio.Queue(maxsize=1000) + self._cancel_event.clear() + self._http = httpx.AsyncClient(timeout=15.0) + + try: + await self._restore_or_fetch_token() + self._refresh_task = asyncio.create_task(self._token_refresh_loop()) + self._running = True + logger.info( + "JD gateway started: account=%s shop=%s token_expires=%d", + account.account_id, + account.shop_id, + int(self._token_expires_at - time.time()) if self._token_expires_at else 0, + ) + return { + "running": True, + "account_id": account.account_id, + "shop_id": account.shop_id, + "queue": self._queue, + } + except Exception: + logger.exception("JD gateway failed to start") + await self.stop(ctx) + raise + + async def stop(self, ctx) -> None: + self._running = False + self._cancel_event.set() + + if self._refresh_task and not self._refresh_task.done(): + self._refresh_task.cancel() + try: + await self._refresh_task + except asyncio.CancelledError: + pass + + if self._http: + await self._http.aclose() + self._http = None + + self._queue = None + self._last_chat_id.clear() + logger.info("JD gateway stopped") + + @property + def account(self) -> JDAccount | None: + return self._account + + @property + def queue(self) -> asyncio.Queue | None: + return self._queue + + def get_token(self, account_id: str = "default") -> str | None: + if self._access_token and time.time() < self._token_expires_at: + return self._access_token + return None + + def get_account(self, account_id: str = "default") -> JDAccount | None: + return self._account + + def cache_chat_id(self, user_id: str, chat_id: str): + self._last_chat_id[user_id] = chat_id + + def get_cached_chat_id(self, user_id: str) -> str | None: + return self._last_chat_id.get(user_id) + + async def _restore_or_fetch_token(self): + account = self._account + + loaded = await self._load_token_from_disk() + if loaded and loaded.get("expires_at", 0) > time.time() + TOKEN_REFRESH_MARGIN: + self._access_token = loaded["access_token"] + self._token_expires_at = loaded["expires_at"] + if loaded.get("refresh_token"): + self._account.refresh_token = loaded["refresh_token"] + logger.info("JD token restored from disk") + return + + if account.access_token and account.token_expires_at: + if account.token_expires_at > time.time() + TOKEN_REFRESH_MARGIN: + self._access_token = account.access_token + self._token_expires_at = account.token_expires_at + return + + if account.refresh_token: + try: + await self._refresh_token() + except Exception as e: + logger.error("JD token refresh failed on start: %s", e) + + if not self._access_token: + logger.warning("JD account %s has no valid access token", account.account_id) + + async def _token_refresh_loop(self): + while not self._cancel_event.is_set(): + try: + await asyncio.sleep(REFRESH_INTERVAL) + if self._cancel_event.is_set(): + return + + remaining = self._token_expires_at - time.time() + if remaining < TOKEN_REFRESH_MARGIN: + logger.info("JD token refresh triggered: remaining=%ds", int(remaining)) + await self._refresh_token() + + except asyncio.CancelledError: + break + except Exception as e: + logger.error("JD token refresh loop error: %s", e) + + async def _refresh_token(self): + async with self._token_lock: + if not self._account.refresh_token: + return + + params = { + "grant_type": "refresh_token", + "client_id": self._account.app_key, + "client_secret": self._account.app_secret, + "refresh_token": self._account.refresh_token, + } + + for attempt in range(RETRY_MAX): + try: + resp = await self._http.post(OAUTH_TOKEN_URL, data=params) + resp.raise_for_status() + data = resp.json() + + expires_in = data.get("expires_in", 7200) + self._access_token = data["access_token"] + self._token_expires_at = time.time() + expires_in - TOKEN_REFRESH_MARGIN + + new_refresh = data.get("refresh_token", "") + if new_refresh: + self._account.refresh_token = new_refresh + + self._token_obj = JOSToken( + access_token=data["access_token"], + refresh_token=data.get("refresh_token", self._account.refresh_token or ""), + expires_at=time.time() + expires_in, + uid=data.get("uid", ""), + user_nick=data.get("user_nick", ""), + ) + + await self._save_token_to_disk() + + logger.info( + "JD token refreshed: expires_in=%ds uid=%s", + expires_in, + data.get("uid", ""), + ) + return + + except Exception as e: + logger.warning("JD token refresh attempt %d failed: %s", attempt + 1, e) + if attempt == RETRY_MAX - 1: + raise + delay = RETRY_BASE_DELAY * (2**attempt) + await asyncio.sleep(delay) + + async def _save_token_to_disk(self): + if not self._token_obj: + return + try: + os.makedirs(_TOKEN_STORE_DIR, exist_ok=True) + token_file = os.path.join(_TOKEN_STORE_DIR, f"{self._account.shop_id}.json") + with open(token_file, "w") as f: + json.dump( + { + "access_token": self._access_token, + "refresh_token": self._account.refresh_token, + "expires_at": self._token_expires_at, + "uid": self._token_obj.uid, + "user_nick": self._token_obj.user_nick, + }, + f, + ) + except Exception as e: + logger.warning("Failed to save JD token to disk: %s", e) + + async def _load_token_from_disk(self) -> dict | None: + if not self._account: + return None + token_file = os.path.join(_TOKEN_STORE_DIR, f"{self._account.shop_id}.json") + if not os.path.exists(token_file): + return None + try: + with open(token_file) as f: + data = json.load(f) + if data.get("expires_at", 0) > time.time(): + return data + except Exception as e: + logger.warning("Failed to load JD token from disk: %s", e) + return None diff --git a/backend/package/yuxi/channel/extensions/jd/monitor.py b/backend/package/yuxi/channel/extensions/jd/monitor.py new file mode 100644 index 00000000..39e7b30d --- /dev/null +++ b/backend/package/yuxi/channel/extensions/jd/monitor.py @@ -0,0 +1,63 @@ +import json +import logging +from datetime import datetime, UTC + +from yuxi.channel.extensions.jd.types import InboundJDMessage +from yuxi.channel.message.models import MessageType, PeerInfo, UnifiedMessage +from yuxi.channel.routing.models import PeerKind + +logger = logging.getLogger(__name__) + + +class JDMonitor: + def convert_to_unified(self, msg: InboundJDMessage) -> UnifiedMessage: + msg_type = MessageType.TEXT + if msg.msg_type == 2: + msg_type = MessageType.IMAGE + + content = self._extract_content(msg) + + return UnifiedMessage( + msg_id=msg.msg_id, + channel_type="jd", + account_id="default", + content=content, + message_type=msg_type, + sender=PeerInfo( + id=msg.from_id, + kind=PeerKind.DIRECT, + display_name=msg.from_id, + ), + timestamp=self._parse_timestamp(msg.msg_time), + raw_payload=msg.raw, + body_for_agent=content, + metadata={ + "chat_id": msg.chat_id, + "to_id": msg.to_id, + "from_type": msg.from_type, + "extend": msg.extend, + }, + ) + + def _extract_content(self, msg: InboundJDMessage) -> str: + if msg.msg_type in (2, 3, 4): + try: + data = json.loads(msg.content) + if msg.msg_type == 2: + return data.get("url", msg.content) + if msg.msg_type == 3: + return f"{data.get('text', '')}\n{data.get('url', '')}" + if msg.msg_type == 4: + return f"[商品卡片] {data.get('pid', '')}\n{data.get('url', '')}" + except (json.JSONDecodeError, TypeError): + return msg.content + return msg.content + + def _parse_timestamp(self, msg_time: str) -> datetime | None: + try: + dt = datetime.strptime(msg_time, "%Y-%m-%d %H:%M:%S") + return dt.replace(tzinfo=UTC) + except (ValueError, TypeError): + import time + + return datetime.fromtimestamp(time.time(), tz=UTC) diff --git a/backend/package/yuxi/channel/extensions/jd/outbound.py b/backend/package/yuxi/channel/extensions/jd/outbound.py new file mode 100644 index 00000000..ca19c30f --- /dev/null +++ b/backend/package/yuxi/channel/extensions/jd/outbound.py @@ -0,0 +1,147 @@ +import json +import logging + +from yuxi.channel.extensions.jd.client import JOSClient +from yuxi.channel.extensions.jd.types import JD_TEXT_MAX_LENGTH + +logger = logging.getLogger(__name__) + + +class JDOutbound: + def __init__(self, gateway=None): + self._gateway = gateway + self._client: JOSClient | None = 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, + chat_id: str | None = None, + ) -> None: + if not content: + return + + if len(content) > JD_TEXT_MAX_LENGTH: + content = content[: JD_TEXT_MAX_LENGTH - 3] + "..." + + await self._send_msg(target_id, content, msg_type=1, chat_id=chat_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, + chat_id: str | None = None, + ) -> None: + if media_type != "image": + logger.warning("JD only supports image, got media_type=%s", media_type) + return + + img_content = json.dumps({"url": media_url}, ensure_ascii=False) + await self._send_msg(target_id, img_content, msg_type=2, chat_id=chat_id) + + async def send_image(self, to_user: str, media_url: str, *, chat_id: str | None = None) -> bool: + img_content = json.dumps({"url": media_url}, ensure_ascii=False) + return await self._send_msg(to_user, img_content, msg_type=2, chat_id=chat_id) + + async def send_link( + self, + target_id: str, + url: str, + text: str, + *, + chat_id: str | None = None, + ) -> bool: + link_content = json.dumps({"url": url, "text": text}, ensure_ascii=False) + return await self._send_msg(target_id, link_content, msg_type=3, chat_id=chat_id) + + async def send_product_card( + self, + target_id: str, + url: str, + pid: str, + *, + chat_id: str | None = None, + ) -> bool: + card_content = json.dumps({"url": url, "pid": pid}, ensure_ascii=False) + return await self._send_msg(target_id, card_content, msg_type=4, chat_id=chat_id) + + async def _send_msg(self, to_user: str, content: str, msg_type: int = 1, *, chat_id: str | None = None) -> bool: + gateway = self._gateway + if gateway is None or not (token := gateway.get_token()) or not (account := gateway.account): + logger.error("No JD gateway/token/account available") + return False + + if self._client is None: + self._client = JOSClient( + app_key=account.app_key, + app_secret=account.app_secret, + shop_id=account.shop_id, + ) + + biz_params = { + "chat_id": chat_id or "", + "from_id": account.shop_id, + "to_id": to_user, + "msg_type": msg_type, + "content": content, + } + + try: + data = await self._client.call("jingdong.im.sendMsg", biz_params, token) + result = data.get("jingdong_im_sendMsg_responce", {}).get("result", {}) + logger.debug("JD message sent: msg_id=%s to=%s", result.get("msg_id", ""), to_user) + return True + except Exception: + logger.exception("JD send_msg failed for user %s", to_user) + return False + + async def recall_msg(self, msg_id: str, chat_id: str | None = None) -> bool: + gateway = self._gateway + if gateway is None or not (token := gateway.get_token()) or not (account := gateway.account): + logger.error("No JD gateway/token/account available for recall") + return False + + if self._client is None: + self._client = JOSClient( + app_key=account.app_key, + app_secret=account.app_secret, + shop_id=account.shop_id, + ) + + biz_params = {"msg_id": msg_id, "chat_id": chat_id or ""} + try: + await self._client.call("jingdong.im.recallMsg", biz_params, token) + logger.debug("JD message recalled: msg_id=%s", msg_id) + return True + except Exception: + logger.exception("JD recall msg failed: msg_id=%s", msg_id) + return False + + def chunker(self, text: str, max_chars: int = JD_TEXT_MAX_LENGTH) -> list[str]: + if len(text) <= max_chars: + return [text] + + chunks = [] + while text: + if len(text) <= max_chars: + chunks.append(text) + break + split_at = text.rfind("\n", 0, max_chars) + if split_at == -1: + split_at = max_chars + chunks.append(text[:split_at]) + text = text[split_at:].lstrip() + + return chunks + + async def close(self): + if self._client: + await self._client.close() + self._client = None diff --git a/backend/package/yuxi/channel/extensions/jd/pairing.py b/backend/package/yuxi/channel/extensions/jd/pairing.py new file mode 100644 index 00000000..adb63a0f --- /dev/null +++ b/backend/package/yuxi/channel/extensions/jd/pairing.py @@ -0,0 +1,57 @@ +import random +import time + +CODE_LENGTH = 6 +CODE_TTL_SECONDS = 600 +PENDING_MAX = 3 +RATE_LIMIT_SECONDS = 3600 + + +class JDPairing: + def __init__(self): + self._pending: dict[str, dict] = {} + self._user_last_code_at: dict[str, float] = {} + + def generate_code(self, user_id: str) -> str | None: + now = time.time() + + if user_id in self._user_last_code_at: + elapsed = now - self._user_last_code_at[user_id] + if elapsed < RATE_LIMIT_SECONDS: + return None + + if len(self._pending) >= PENDING_MAX: + oldest = min(self._pending.values(), key=lambda p: p["created_at"]) + if now - oldest["created_at"] < CODE_TTL_SECONDS: + return None + + expired_user = next(k for k, v in self._pending.items() if v == oldest) + del self._pending[expired_user] + + code = _generate_code() + self._pending[user_id] = {"code": code, "created_at": now} + self._user_last_code_at[user_id] = now + return code + + def verify(self, user_id: str, code: str) -> bool: + entry = self._pending.get(user_id) + if not entry: + return False + + if time.time() - entry["created_at"] > CODE_TTL_SECONDS: + del self._pending[user_id] + return False + + if entry["code"] != code.upper().strip(): + return False + + del self._pending[user_id] + return True + + def get_pending_count(self) -> int: + return len(self._pending) + + +def _generate_code() -> str: + no_ambiguous = "ABCDEFGHJKLMNPQRSTUVWXYZ23456789" + return "".join(random.choices(no_ambiguous, k=CODE_LENGTH)) diff --git a/backend/package/yuxi/channel/extensions/jd/plugin.json b/backend/package/yuxi/channel/extensions/jd/plugin.json new file mode 100644 index 00000000..6141f047 --- /dev/null +++ b/backend/package/yuxi/channel/extensions/jd/plugin.json @@ -0,0 +1,27 @@ +{ + "id": "jd", + "name": "京东/咚咚", + "version": "0.3.0", + "description": "京东电商平台咚咚 IM 客服渠道，通过咚咚消息推送 Webhook + 京东开放平台 JOS API 实现双向消息收发，支持订单/物流/商品查询", + "author": "ForcePilot", + "order": 90, + "dependencies": ["httpx"], + "capabilities": { + "chat_types": ["direct"], + "message_types": ["text", "image", "link", "product_card"], + "reactions": false, + "typing_indicator": false, + "threads": false, + "edit": false, + "unsend": true, + "reply": false, + "media": true, + "effects": false, + "native_commands": false, + "polls": false, + "group_management": false, + "streaming": false + }, + "enabled": true, + "python_requires": ">=3.12" +} diff --git a/backend/package/yuxi/channel/extensions/jd/security.py b/backend/package/yuxi/channel/extensions/jd/security.py new file mode 100644 index 00000000..bb44c205 --- /dev/null +++ b/backend/package/yuxi/channel/extensions/jd/security.py @@ -0,0 +1,36 @@ +import logging + +from yuxi.channel.extensions.jd.types import JDAccount + +logger = logging.getLogger(__name__) + + +class JDSecurity: + VALID_POLICIES = {"open", "pairing", "allowlist", "disabled"} + + def __init__(self, account: JDAccount | None = None): + self._account = account + if account and account.dm_policy in self.VALID_POLICIES: + self._policy = account.dm_policy + else: + self._policy = "open" + self._allowlist: set[str] = set(account.allow_from) if account else set() + + def resolve_dm_policy(self) -> str: + return self._policy + + def check_allowlist(self, user_id: str, channel_type: str = "jd") -> bool: + if self._policy == "open": + return True + if self._policy == "disabled": + return False + if not self._allowlist: + logger.warning("allowlist is empty with policy=%s", self._policy) + return False + return user_id in self._allowlist + + def add_to_allowlist(self, user_id: str): + self._allowlist.add(user_id) + + def remove_from_allowlist(self, user_id: str): + self._allowlist.discard(user_id) diff --git a/backend/package/yuxi/channel/extensions/jd/session.py b/backend/package/yuxi/channel/extensions/jd/session.py new file mode 100644 index 00000000..533aac5a --- /dev/null +++ b/backend/package/yuxi/channel/extensions/jd/session.py @@ -0,0 +1,128 @@ +import asyncio +import logging +from datetime import datetime, timedelta, UTC + +from yuxi.channel.extensions.jd.client import JOSClient + +logger = logging.getLogger(__name__) + + +class SessionEntry: + __slots__ = ("chat_id", "user_id", "messages", "turn_count", "created_at", "updated_at", "status") + + def __init__(self, chat_id: str, user_id: str): + self.chat_id = chat_id + self.user_id = user_id + self.messages: list[dict] = [] + self.turn_count = 0 + self.created_at = datetime.now(UTC) + self.updated_at = datetime.now(UTC) + self.status = "active" + + +class JDSession: + MAX_HISTORY = 50 + SESSION_TIMEOUT_SECONDS = 1800 + CLEANUP_INTERVAL = 600 + + def __init__(self, client: JOSClient | None = None): + self._sessions: dict[str, SessionEntry] = {} + self._client = client + self._cleanup_task: asyncio.Task | None = None + + async def start(self): + self._cleanup_task = asyncio.create_task(self._cleanup_loop()) + + async def stop(self): + if self._cleanup_task: + self._cleanup_task.cancel() + try: + await self._cleanup_task + except asyncio.CancelledError: + pass + self._cleanup_task = None + + def add_message(self, chat_id: str, user_id: str, msg: dict): + session = self._sessions.get(chat_id) + if session is None: + session = SessionEntry(chat_id, user_id) + self._sessions[chat_id] = session + + session.messages.append({**msg, "_timestamp": datetime.now(UTC).isoformat()}) + if len(session.messages) > self.MAX_HISTORY: + session.messages.pop(0) + + if session.status != "active": + session.status = "active" + session.turn_count = 1 + else: + session.turn_count += 1 + + session.updated_at = datetime.now(UTC) + + def get_history(self, chat_id: str, limit: int = 20) -> list[dict]: + session = self._sessions.get(chat_id) + if session is None: + return [] + return session.messages[-limit:] + + def get_session_context(self, chat_id: str) -> dict | None: + session = self._sessions.get(chat_id) + if session is None: + return None + return { + "chat_id": session.chat_id, + "user_id": session.user_id, + "turn_count": session.turn_count, + "status": session.status, + "created_at": session.created_at.isoformat(), + "updated_at": session.updated_at.isoformat(), + "message_count": len(session.messages), + } + + def close_session(self, chat_id: str): + session = self._sessions.get(chat_id) + if session: + session.status = "closed" + session.updated_at = datetime.now(UTC) + + def clear_session(self, chat_id: str): + self._sessions.pop(chat_id, None) + + async def fetch_remote_history( + self, chat_id: str, to_id: str, access_token: str, page: int = 1, page_size: int = 20 + ) -> list[dict]: + if self._client is None: + return [] + + biz_params = { + "chat_id": chat_id, + "to_id": to_id, + "page": str(page), + "page_size": str(page_size), + } + + try: + data = await self._client.call("jingdong.im.getChatLog", biz_params, access_token) + messages = data.get("jingdong_im_getChatLog_responce", {}).get("result", {}).get("messages", []) + return messages if isinstance(messages, list) else [] + except Exception: + logger.exception("Failed to fetch JD chat history for %s", chat_id) + return [] + + async def _cleanup_loop(self): + while True: + try: + await asyncio.sleep(self.CLEANUP_INTERVAL) + self._expire_sessions() + except asyncio.CancelledError: + break + except Exception: + logger.exception("JD session cleanup loop error") + + def _expire_sessions(self): + cutoff = datetime.now(UTC) - timedelta(seconds=self.SESSION_TIMEOUT_SECONDS) + expired = [chat_id for chat_id, session in self._sessions.items() if session.updated_at < cutoff] + for chat_id in expired: + session = self._sessions.pop(chat_id) + logger.debug("JD session expired: chat_id=%s turns=%d", session.chat_id, session.turn_count) diff --git a/backend/package/yuxi/channel/extensions/jd/signature.py b/backend/package/yuxi/channel/extensions/jd/signature.py new file mode 100644 index 00000000..bc652f8b --- /dev/null +++ b/backend/package/yuxi/channel/extensions/jd/signature.py @@ -0,0 +1,79 @@ +import base64 +import hashlib +import hmac +import json +import time +from urllib.parse import quote + + +def sign_jos_request(params: dict, app_secret: str) -> str: + sorted_keys = sorted(params.keys()) + + sign_str = "" + for key in sorted_keys: + raw_value = str(params[key]) + encoded = quote(raw_value, safe="") + sign_str += key + encoded + + full_str = app_secret + sign_str + app_secret + signature = hmac.new(app_secret.encode("utf-8"), full_str.encode("utf-8"), hashlib.sha256).hexdigest().upper() + + return signature + + +def build_signed_params( + method: str, + biz_params: dict, + app_key: str, + app_secret: str, + access_token: str, +) -> dict: + params = { + "method": method, + "app_key": app_key, + "access_token": access_token, + "timestamp": time.strftime("%Y-%m-%d %H:%M:%S"), + "format": "json", + "v": "2.0", + "sign_method": "hmac", + } + + biz_json = json.dumps(biz_params, ensure_ascii=False, separators=(",", ":")) + params["360buy_param"] = biz_json + + sign = sign_jos_request(params, app_secret) + params["sign"] = sign + + return params + + +def verify_webhook_signature( + raw_body: bytes, + signature_header: str, + timestamp_header: str, + app_secret: str, + tolerance_seconds: int = 300, + algorithm: str = "hmac-sha256", +) -> bool: + try: + request_time = int(timestamp_header) + current_time = int(time.time()) + if abs(current_time - request_time) > tolerance_seconds: + return False + except (ValueError, TypeError): + return False + + if algorithm == "md5": + sign_str = app_secret + raw_body.decode("utf-8") + app_secret + expected = hashlib.md5(sign_str.encode("utf-8")).hexdigest().upper() + return signature_header.upper() == expected + + sign_str = raw_body + timestamp_header.encode("utf-8") + computed = hmac.new(app_secret.encode("utf-8"), sign_str, hashlib.sha256).digest() + + try: + expected = base64.b64decode(signature_header) + except Exception: + return False + + return hmac.compare_digest(computed, expected) diff --git a/backend/package/yuxi/channel/extensions/jd/status.py b/backend/package/yuxi/channel/extensions/jd/status.py new file mode 100644 index 00000000..d4f8fb3a --- /dev/null +++ b/backend/package/yuxi/channel/extensions/jd/status.py @@ -0,0 +1,83 @@ +import logging + +from yuxi.channel.protocols import ChannelAccountSnapshot + +logger = logging.getLogger(__name__) + + +class JDStatus: + def __init__(self, gateway=None): + self._gateway = gateway + + async def probe(self, account: dict | None = None) -> bool: + gateway = self._gateway + if gateway is None: + return False + + token = gateway.get_token("default") + if not token: + return False + + return True + + def build_summary(self, snapshot: object) -> dict: + if not isinstance(snapshot, ChannelAccountSnapshot): + return {} + return { + "account_id": snapshot.account_id, + "running": snapshot.running, + "connected": snapshot.connected, + "health_state": snapshot.health_state, + } + + def build_account_snapshot( + self, + account: dict, + config: dict, + runtime: ChannelAccountSnapshot | None = None, + probe_result: object | None = None, + audit: object | None = None, + ) -> ChannelAccountSnapshot: + gateway = self._gateway + token = gateway.get_token("default") if gateway else None + + shop_id = account.get("shop_id", "") + display = f"jd-{shop_id}" if shop_id else "jd-default" + + return ChannelAccountSnapshot( + account_id=account.get("account_id", "default"), + name=account.get("name", display), + enabled=account.get("enabled", True), + configured=bool(account.get("app_key") and account.get("app_secret") and account.get("shop_id")), + status_state="linked" if token else "not-linked", + running=getattr(runtime, "running", False) if runtime else False, + connected=bool(token), + last_message_at=getattr(runtime, "last_message_at", None) if runtime else None, + health_state="ok" if probe_result else "unknown", + dm_policy=account.get("dm_policy", "pairing"), + ) + + def collect_status_issues(self, accounts: list[ChannelAccountSnapshot]) -> list: + issues = [] + for acc in accounts: + if not acc.configured: + issues.append( + { + "channel": "jd", + "account_id": acc.account_id, + "kind": "not-configured", + "message": "JD credentials not configured", + "fix": "Set JD_APP_KEY, JD_APP_SECRET, JD_SHOP_ID environment variables", + } + ) + elif not acc.connected: + issues.append( + { + "channel": "jd", + "account_id": acc.account_id, + "kind": "not-connected", + "message": "JD gateway token not available", + "fix": "Check JD OAuth token configuration and gateway status", + } + ) + return issues diff --git a/backend/package/yuxi/channel/extensions/jd/types.py b/backend/package/yuxi/channel/extensions/jd/types.py new file mode 100644 index 00000000..65ae23c8 --- /dev/null +++ b/backend/package/yuxi/channel/extensions/jd/types.py @@ -0,0 +1,57 @@ +from dataclasses import dataclass, field +from typing import Literal + +JD_TEXT_MAX_LENGTH = 4096 +JOS_API_GATEWAY = "https://api.jd.com/routerjson" +JOS_MAX_RETRIES = 3 +JOS_RETRY_BASE_DELAY = 1.0 + + +@dataclass +class JDAccount: + account_id: str + app_key: str + app_secret: str + shop_id: str + access_token: str | None = None + refresh_token: str | None = None + token_expires_at: float | None = None + webhook_token: str | None = None + dm_policy: str = "pairing" + webhook_sign_algorithm: str = "hmac-sha256" + allow_from: list[str] = field(default_factory=list) + + def is_configured(self) -> bool: + return bool(self.app_key and self.app_secret and self.shop_id) + + +@dataclass +class InboundJDMessage: + msg_id: str + msg_type: Literal[1, 2, 3, 4] = 1 + content: str = "" + from_id: str = "" + to_id: str = "" + chat_id: str = "" + msg_time: str = "" + from_type: str = "1" + to_type: str = "2" + extend: str = "{}" + raw: dict = field(default_factory=dict) + + +@dataclass +class OutboundResult: + message_id: str + success: bool + error_code: str | None = None + error_msg: str | None = None + + +@dataclass +class JOSToken: + access_token: str + refresh_token: str + expires_at: float + uid: str = "" + user_nick: str = "" diff --git a/backend/package/yuxi/channel/extensions/jd/webhook.py b/backend/package/yuxi/channel/extensions/jd/webhook.py new file mode 100644 index 00000000..a76045d7 --- /dev/null +++ b/backend/package/yuxi/channel/extensions/jd/webhook.py @@ -0,0 +1,205 @@ +from __future__ import annotations + +import asyncio +import json +import logging +from typing import TYPE_CHECKING + +from fastapi import APIRouter, Request +from fastapi.responses import Response + +from yuxi.channel.extensions.jd.crypto import decrypt_jd_message +from yuxi.channel.extensions.jd.format import remove_markdown +from yuxi.channel.extensions.jd.outbound import JDOutbound +from yuxi.channel.extensions.jd.security import JDSecurity +from yuxi.channel.extensions.jd.signature import verify_webhook_signature +from yuxi.channel.extensions.jd.types import InboundJDMessage + +if TYPE_CHECKING: + from yuxi.channel.extensions.jd import JDPlugin + +logger = logging.getLogger(__name__) + +router = APIRouter(prefix="/webhook/jd", tags=["jd"]) + +_plugin: JDPlugin | None = None + + +def set_plugin(plugin: JDPlugin) -> None: + global _plugin + _plugin = plugin + + +def get_plugin() -> JDPlugin | None: + return _plugin + + +@router.post("/callback") +async def jd_message_callback(request: Request): + plugin = _plugin + if plugin is None: + logger.warning("JD plugin not initialized, rejecting webhook") + return Response(content='{"code":-1}', media_type="application/json", status_code=503) + + gw = plugin._gateway + account = gw.account if gw else None + + raw_body = await request.body() + + if account and account.app_secret: + signature = request.headers.get("X-JD-Sign", "") + timestamp = request.headers.get("X-JD-Timestamp", "") + sign_algorithm = account.webhook_sign_algorithm or "hmac-sha256" + if not verify_webhook_signature(raw_body, signature, timestamp, account.app_secret, algorithm=sign_algorithm): + logger.warning("JD webhook signature verification failed (algorithm=%s)", sign_algorithm) + return Response(content='{"code":-1}', media_type="application/json", status_code=403) + + try: + data = json.loads(raw_body) + except json.JSONDecodeError as e: + logger.error("JD webhook JSON parse error: %s", e) + return Response(content='{"code":-1}', media_type="application/json", status_code=400) + + if account and account.app_secret and data.get("encrypt"): + decrypted = decrypt_jd_message(data.get("encrypt_jd_param_json", ""), account.app_secret) + if decrypted: + data = decrypted + else: + logger.warning("JD webhook AES decrypt failed") + return Response(content='{"code":-1}', media_type="application/json", status_code=400) + + messages = data.get("messages", []) + if not messages: + event_type = data.get("type") + if event_type in ("system", "event"): + logger.info("JD webhook event: type=%s", event_type) + if event_type == "system": + await _handle_system_event(data) + return Response(content='{"code":0}', media_type="application/json", status_code=200) + + deduplicator = plugin._deduplicator + pairing = plugin._pairing + monitor = plugin._monitor + + for msg_data in messages: + msg_id = msg_data.get("msg_id", "") + if not msg_id or deduplicator.is_duplicate(msg_id): + continue + + from_type = msg_data.get("from_type", "1") + if from_type == "2": + continue + + content = msg_data.get("content", "") + from_id = msg_data.get("from_id", "") + chat_id = msg_data.get("chat_id", "") + + if not content or not account: + continue + + if gw and chat_id: + gw.cache_chat_id(from_id, chat_id) + + security = plugin._security + if security is None: + security = JDSecurity(account) + + policy = security.resolve_dm_policy() + + if policy == "disabled": + continue + + if policy == "pairing": + if not security.check_allowlist(from_id): + if content.strip().startswith("配对 "): + code_input = content.strip()[3:].strip() + if pairing.verify(from_id, code_input): + security.add_to_allowlist(from_id) + await _send_jd_text(plugin, from_id, "配对成功！现在可以开始对话了。", chat_id=chat_id) + else: + await _send_jd_text( + plugin, from_id, "配对码无效或已过期，请重新发送消息获取配对码。", chat_id=chat_id + ) + else: + code = pairing.generate_code(from_id) + if code: + await _send_jd_text( + plugin, + from_id, + f"首次对话需要验证身份，请输入以下配对码：\n\n配对 {code}\n\n（配对码有效期 10 分钟）", + chat_id=chat_id, + ) + else: + await _send_jd_text(plugin, from_id, "配对请求过于频繁，请稍后再试。", chat_id=chat_id) + continue + + if policy == "allowlist": + if not security.check_allowlist(from_id): + continue + + content = remove_markdown(content) + + msg = InboundJDMessage( + msg_id=msg_id, + msg_type=msg_data.get("msg_type", 1), + content=content, + from_id=from_id, + to_id=msg_data.get("to_id", ""), + chat_id=chat_id, + msg_time=msg_data.get("msg_time", ""), + from_type=msg_data.get("from_type", "1"), + to_type=msg_data.get("to_type", "2"), + extend=msg_data.get("extend", "{}"), + raw=msg_data, + ) + + unified = monitor.convert_to_unified(msg) + + from yuxi.channel.runtime.manager import gateway + + processor = getattr(gateway, "_processor", None) if gateway else None + if processor is None: + logger.warning("Message processor not available, cannot dispatch JD message") + continue + + asyncio.create_task( + _dispatch_to_agent(processor, unified), + name=f"jd-dispatch-{from_id}", + ) + + return Response(content='{"code":0}', media_type="application/json", status_code=200) + + +async def _handle_system_event(data: dict): + event_subtype = data.get("subtype", "") + chat_id = data.get("chat_id", "") + if event_subtype == "session_close": + logger.info("JD session closed: chat_id=%s", chat_id) + elif event_subtype == "user_enter": + logger.info("JD user entered: chat_id=%s", chat_id) + else: + logger.info("JD system event: subtype=%s chat_id=%s", event_subtype, chat_id) + + +async def _dispatch_to_agent(processor, msg) -> None: + try: + await asyncio.wait_for(processor.process(msg), timeout=120.0) + except TimeoutError: + logger.error("Agent response timeout for JD user %s", msg.sender.id) + except Exception: + logger.exception("Failed to process JD message for user %s", msg.sender.id) + + +async def _send_jd_text(plugin, to_user: str, content: str, *, chat_id: str | None = None) -> None: + gw = plugin._gateway + if gw is None: + logger.warning("No JD gateway available for sending text") + return + + outbound = JDOutbound(gw) + try: + await outbound.send_text(to_user, content, chat_id=chat_id) + except Exception: + logger.exception("Failed to send JD text to %s", to_user) + finally: + await outbound.close()