wangliang/assistant
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

feat: 完善物流信息展示功能

Browse Source 
- 修复物流信息中 order_id 字段缺失的问题，确保按钮正常生成
- 添加 tracking_url 支持，新增"官网追踪"按钮
- "官网追踪"按钮在新标签页打开 (target: "_blank")
- 改进日志记录，添加完整的 payload 预览

Co-Authored-By: Claude Sonnet 4.5 <noreply@anthropic.com>
This commit is contained in:
wangliang
2026-01-26 13:15:47 +08:00
parent 0b5d0a8086
commit 9fe29ff3fe
2 changed files with 168 additions and 16 deletions
Download Patch File Download Diff File Expand all files Collapse all files

2
agent/agents/order.py
View File

@@ -815,8 +815,10 @@ def _parse_logistics_data(data: dict[str, Any]) -> dict[str, Any]:
)
return {
"order_id": mcp_result.get("order_id", ""), # 添加订单号
"carrier": mcp_result.get("courier", mcp_result.get("carrier", mcp_result.get("express_name", "未知"))),
"tracking_number": mcp_result.get("tracking_number") or "",
"tracking_url": mcp_result.get("tracking_url", mcp_result.get("trackingUrl", "")), # 添加追踪链接
"status": mcp_result.get("status"),
"estimated_delivery": mcp_result.get("estimatedDelivery"),
"timeline": mcp_result.get("timeline", [])

172
agent/integrations/chatwoot.py
View File

@@ -5,6 +5,7 @@ import json
from typing import Any, Optional
from dataclasses import dataclass
from enum import Enum
from contextlib import asynccontextmanager
import httpx
@@ -646,6 +647,7 @@ class ChatwootClient:
- status: 当前状态
- timeline: 物流轨迹列表
- order_id: 订单号（可选，用于生成链接）
- tracking_url: 官方追踪链接（可选，跳转到物流公司官网）
language: 语言代码（en, nl, de, es, fr, it, tr, zh），默认 en
Returns:
@@ -665,6 +667,7 @@ class ChatwootClient:
status = logistics_data.get("status", "")
timeline = logistics_data.get("timeline", [])
order_id = logistics_data.get("order_id", "")
tracking_url = logistics_data.get("tracking_url", "") # 添加追踪链接
# 获取最新物流信息（从 timeline 中提取 remark）
latest_log = ""
@@ -760,6 +763,23 @@ class ChatwootClient:
"url": f"{frontend_url}/logistic-tracking/{order_id}"
})
# 如果有 tracking_url，添加官方追踪按钮（在新标签页打开）
if tracking_url:
if language == "zh":
actions.append({
"text": "官网追踪",
"style": "default",
"url": tracking_url,
"target": "_blank"
})
else:
actions.append({
"text": "Official Tracking",
"style": "default",
"url": tracking_url,
"target": "_blank"
})
# 构建 content_attributes（logistics 格式）
content_attributes = {
"logisticsName": carrier,
@@ -772,17 +792,6 @@ class ChatwootClient:
"actions": actions
}
# 记录发送的数据（用于调试）
logger.info(
"Sending logistics info",
conversation_id=conversation_id,
carrier=carrier,
tracking_number=tracking_number,
current_step=current_step,
timeline_count=len(timeline) if timeline else 0,
language=language
)
# 发送 logistics 类型消息
client = await self._get_client()
@@ -792,6 +801,18 @@ class ChatwootClient:
"content_attributes": content_attributes
}
# 记录完整的发送 payload（用于调试）
logger.info(
"Sending logistics info",
conversation_id=conversation_id,
carrier=carrier,
tracking_number=tracking_number,
current_step=current_step,
timeline_count=len(timeline) if timeline else 0,
language=language,
payload_preview=json.dumps(payload, ensure_ascii=False, indent=2)
)
response = await client.post(
f"/conversations/{conversation_id}/messages",
json=payload
@@ -1014,6 +1035,135 @@ class ChatwootClient:
return response.json()
async def toggle_typing_status(
self,
conversation_id: int,
typing_status: str # "on" or "off"
) -> dict[str, Any]:
"""Toggle typing status for a conversation
显示/隐藏"正在输入..."状态指示器。Chatwoot 会根据用户设置的语言
自动显示对应文本（如"正在输入..."、"Typing..."等）。
Args:
conversation_id: Conversation ID
typing_status: Typing status, either "on" or "off"
Returns:
API response
Raises:
ValueError: If typing_status is not "on" or "off"
Example:
>>> # 开启 typing status
>>> await chatwoot.toggle_typing_status(123, "on")
>>> # 处理消息...
>>> # 关闭 typing status
>>> await chatwoot.toggle_typing_status(123, "off")
"""
if typing_status not in ["on", "off"]:
raise ValueError(
f"typing_status must be 'on' or 'off', got '{typing_status}'"
)
client = await self._get_client()
# 尝试使用 admin API 端点
response = await client.post(
f"/conversations/{conversation_id}/toggle_typing_status",
json={"typing_status": typing_status}
)
# 如果 admin API 不可用，返回空响应（兼容性处理）
try:
response.raise_for_status()
except httpx.HTTPStatusError as e:
logger.warning(
"Failed to toggle typing status via admin API",
conversation_id=conversation_id,
status=typing_status,
error=str(e)
)
# 尝试降级到 public API
return await self._toggle_typing_status_public(
conversation_id, typing_status
)
logger.debug(
"Typing status toggled",
conversation_id=conversation_id,
status=typing_status
)
# 尝试解析 JSON 响应，如果失败则返回空字典
try:
return response.json() if response.content else {}
except Exception:
return {}
async def _toggle_typing_status_public(
self,
conversation_id: int,
typing_status: str
) -> dict[str, Any]:
"""使用 public API 端点切换 typing status（降级方案）
Args:
conversation_id: Conversation ID
typing_status: Typing status, either "on" or "off"
Returns:
API response
"""
# TODO: 实现 public API 调用（需要 inbox_identifier 和 contact_identifier）
# 目前先记录日志并返回空响应
logger.info(
"Public API typing status not implemented, skipping",
conversation_id=conversation_id,
status=typing_status
)
return {}
@asynccontextmanager
async def typing_indicator(self, conversation_id: int):
"""Typing indicator 上下文管理器
自动管理 typing status 的开启和关闭。进入上下文时开启，
退出时自动关闭，即使发生异常也会确保关闭。
显示多语言支持：Chatwoot 会根据用户语言自动显示对应文本
（如中文"正在输入..."、英文"Typing..."等）。
Args:
conversation_id: Conversation ID
Yields:
None
Example:
>>> async with chatwoot.typing_indicator(conversation_id):
... # 处理消息或发送请求
... await process_message(...)
... await send_message(...)
>>> # 退出上下文时自动关闭 typing status
"""
try:
# 开启 typing status
await self.toggle_typing_status(conversation_id, "on")
yield
finally:
# 确保关闭 typing status（即使发生异常）
try:
await self.toggle_typing_status(conversation_id, "off")
except Exception as e:
# 记录警告但不抛出异常
logger.debug(
"Failed to disable typing status in context manager",
conversation_id=conversation_id,
error=str(e)
)
async def assign_agent(
self,
conversation_id: int,