""" 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()