0b5d0a8086
主要改动： - 订单列表：使用 order_list 格式，展示 5 个订单（全部状态） - 订单详情：使用 order_detail 格式，优化价格和时间显示 - 物流信息：使用 logistics 格式，根据 track id 动态生成步骤 - 商品图片：从 orderProduct.imageUrl 字段获取 - 时间格式：统一为 YYYY-MM-DD HH:MM:SS - 多语言支持：amountLabel、orderTime 支持中英文 - 配置管理：新增 FRONTEND_URL 环境变量 - API 集成：改进 Mall API tracks 数据解析 - 认证优化：account_id 从 webhook 动态获取 Co-Authored-By: Claude Sonnet 4.5 <noreply@anthropic.com>
139 lines
4.3 KiB
Python
139 lines
4.3 KiB
Python
"""
|
||
Token Manager - 管理 JWT token 的获取和使用
|
||
|
||
支持从 Chatwoot contact custom_attributes 中获取用户的 JWT token
|
||
"""
|
||
from typing import Optional
|
||
from utils.logger import get_logger
|
||
|
||
logger = get_logger(__name__)
|
||
|
||
|
||
class TokenManager:
|
||
"""管理用户 JWT token"""
|
||
|
||
@staticmethod
|
||
def extract_token_from_sender(sender: Optional[dict]) -> Optional[str]:
|
||
"""从 sender 对象中提取 JWT token
|
||
|
||
支持从以下位置提取 token（按优先级排序）：
|
||
1. sender.jwt_token（根级别）
|
||
2. sender.mall_token（根级别）
|
||
3. sender.custom_attributes.jwt_token
|
||
4. sender.custom_attributes.mall_token
|
||
5. sender.custom_attributes.access_token
|
||
6. sender.custom_attributes.auth_token
|
||
7. sender.custom_attributes.token
|
||
|
||
Args:
|
||
sender: Chatwoot sender 对象（来自 conversation.meta.sender）
|
||
|
||
Returns:
|
||
JWT token 字符串，如果未找到则返回 None
|
||
"""
|
||
if not sender:
|
||
logger.debug("No sender provided")
|
||
return None
|
||
|
||
# 1. 优先从根级别获取 token
|
||
root_token = sender.get("jwt_token") or sender.get("mall_token")
|
||
if root_token:
|
||
logger.debug("JWT token found at sender root level")
|
||
logger.debug(f"Token prefix: {root_token[:20]}...")
|
||
return root_token
|
||
|
||
# 2. 从 custom_attributes 中获取 token
|
||
return TokenManager.extract_token_from_contact(sender)
|
||
|
||
@staticmethod
|
||
def extract_token_from_contact(contact: Optional[dict]) -> Optional[str]:
|
||
"""从 Chatwoot contact 中提取 JWT token
|
||
|
||
Args:
|
||
contact: Chatwoot contact 对象，包含 custom_attributes
|
||
|
||
Returns:
|
||
JWT token 字符串，如果未找到则返回 None
|
||
"""
|
||
if not contact:
|
||
logger.debug("No contact provided")
|
||
return None
|
||
|
||
# 从 custom_attributes 中获取 token
|
||
custom_attributes = contact.get("custom_attributes", {})
|
||
if not custom_attributes:
|
||
logger.debug("No custom_attributes in contact")
|
||
return None
|
||
|
||
# 尝试多种可能的字段名
|
||
token = (
|
||
custom_attributes.get("jwt_token") or
|
||
custom_attributes.get("mall_token") or
|
||
custom_attributes.get("access_token") or
|
||
custom_attributes.get("auth_token") or
|
||
custom_attributes.get("token")
|
||
)
|
||
|
||
if token:
|
||
logger.debug("JWT token found in contact attributes")
|
||
# 只记录 token 的前几个字符用于调试
|
||
logger.debug(f"Token prefix: {token[:20]}...")
|
||
else:
|
||
logger.debug("No JWT token found in contact custom_attributes")
|
||
|
||
return token
|
||
|
||
@staticmethod
|
||
def validate_token(token: str) -> bool:
|
||
"""验证 token 格式是否有效
|
||
|
||
Args:
|
||
token: JWT token 字符串
|
||
|
||
Returns:
|
||
True 如果 token 格式有效
|
||
"""
|
||
if not token or not isinstance(token, str):
|
||
return False
|
||
|
||
# JWT token 通常是 header.payload.signature 格式
|
||
parts = token.split(".")
|
||
if len(parts) != 3:
|
||
logger.warning("Invalid JWT token format")
|
||
return False
|
||
|
||
return True
|
||
|
||
@staticmethod
|
||
def get_token_from_context(context: dict, contact: Optional[dict] = None) -> Optional[str]:
|
||
"""从上下文或 contact 中获取 token
|
||
|
||
优先级：context > contact
|
||
|
||
Args:
|
||
context: 对话上下文
|
||
contact: Chatwoot contact 对象
|
||
|
||
Returns:
|
||
JWT token 或 None
|
||
"""
|
||
# 首先尝试从 context 中获取（可能之前的对话中已经获取）
|
||
token = context.get("user_token")
|
||
if token and TokenManager.validate_token(token):
|
||
logger.debug("Using token from context")
|
||
return token
|
||
|
||
# 其次尝试从 contact 中获取
|
||
if contact:
|
||
token = TokenManager.extract_token_from_contact(contact)
|
||
if token and TokenManager.validate_token(token):
|
||
logger.debug("Using token from contact")
|
||
return token
|
||
|
||
logger.debug("No valid JWT token found")
|
||
return None
|
||
|
||
|
||
# 全局 token 管理器
|
||
token_manager = TokenManager()
|