e093995368
主要改进: - Agent 增强: 订单查询、售后支持、客服路由等功能优化 - 新增语言检测和 Token 管理模块 - 改进 Chatwoot webhook 处理和用户标识 - MCP 服务器增强: 订单 MCP 和 Strapi MCP 功能扩展 - 新增商城客户端、知识库、缓存和同步模块 - 添加多语言提示词系统 (YAML) - 完善项目结构: 整理文档、脚本和测试文件 - 新增调试和测试工具脚本 Co-Authored-By: Claude Sonnet 4.5 <noreply@anthropic.com>
106 lines
3.1 KiB
Python
106 lines
3.1 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_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()
|