""" Chatwoot Webhook Handler """ import hmac import hashlib from typing import Any, Optional from fastapi import APIRouter, Request, HTTPException, BackgroundTasks from pydantic import BaseModel from config import settings from core.graph import process_message from integrations.chatwoot import get_chatwoot_client, ConversationStatus from utils.cache import get_cache_manager from utils.logger import get_logger from utils.token_manager import TokenManager logger = get_logger(__name__) router = APIRouter(prefix="/webhooks", tags=["webhooks"]) # ============ Webhook Payload Models ============ class WebhookSender(BaseModel): """Webhook sender information""" id: Optional[int] = None name: Optional[str] = None email: Optional[str] = None type: Optional[str] = None # "contact" or "user" class WebhookMessage(BaseModel): """Webhook message content""" id: int content: Optional[str] = None message_type: str # "incoming" or "outgoing" content_type: Optional[str] = None private: bool = False sender: Optional[WebhookSender] = None class WebhookConversation(BaseModel): """Webhook conversation information""" id: int inbox_id: int status: str account_id: Optional[int] = None # Chatwoot may not always include this contact_inbox: Optional[dict] = None messages: Optional[list] = None additional_attributes: Optional[dict] = None can_reply: Optional[bool] = None channel: Optional[str] = None meta: Optional[dict] = None # Contains sender info including custom_attributes class WebhookContact(BaseModel): """Webhook contact information""" id: int name: Optional[str] = None email: Optional[str] = None phone_number: Optional[str] = None custom_attributes: Optional[dict] = None class ChatwootWebhookPayload(BaseModel): """Chatwoot webhook payload structure""" event: str id: Optional[int] = None content: Optional[str] = None message_type: Optional[str] = None content_type: Optional[str] = None private: Optional[bool] = False conversation: Optional[WebhookConversation] = None sender: Optional[WebhookSender] = None contact: Optional[WebhookContact] = None account: Optional[dict] = None # ============ Signature Verification ============ def verify_webhook_signature(payload: bytes, signature: str) -> bool: """Verify Chatwoot webhook signature Args: payload: Raw request body signature: X-Chatwoot-Signature header value Returns: True if signature is valid """ # TODO: Re-enable signature verification after configuring Chatwoot properly # For now, skip verification to test webhook functionality logger.debug("Skipping webhook signature verification for testing") return True if not settings.chatwoot_webhook_secret: logger.warning("Webhook secret not configured, skipping verification") return True if not signature: logger.warning("No signature provided in request") return True expected = hmac.new( settings.chatwoot_webhook_secret.encode(), payload, hashlib.sha256 ).hexdigest() return hmac.compare_digest(expected, signature) # ============ Message Processing ============ async def handle_incoming_message(payload: ChatwootWebhookPayload, cookie_token: str = None) -> None: """Process incoming message from Chatwoot Args: payload: Webhook payload cookie_token: User token from request cookies """ conversation = payload.conversation if not conversation: logger.warning("No conversation in payload") return conversation_id = str(conversation.id) content = payload.content if not content: logger.debug("Empty message content, skipping") return # Get user/contact info contact = payload.contact or payload.sender user_id = str(contact.id) if contact else "unknown" # Log webhook payload structure for debugging logger.info( "Webhook payload structure", payload_event=payload.event, has_conversation=bool(conversation), has_contact=bool(contact), contact_type=type(contact).__name__ if contact else None, conversation_id=conversation_id ) # Log full payload keys for debugging payload_dict = payload.model_dump() logger.info( "Full webhook payload keys", keys=list(payload_dict.keys()), conversation_keys=list(payload_dict.get('conversation', {}).keys()) if payload_dict.get('conversation') else [], contact_keys=list(payload_dict.get('contact', {}).keys()) if payload_dict.get('contact') else [], sender_keys=list(payload_dict.get('sender', {}).keys()) if payload_dict.get('sender') else [] ) # Get account_id from payload (top-level account object) # Chatwoot webhook includes account info at the top level account_obj = payload.account account_id = str(account_obj.get("id")) if account_obj else "1" # 优先使用 Cookie 中的 token user_token = cookie_token # 如果 Cookie 中没有,尝试从多个来源提取 token if not user_token: # 1. 尝试从 contact/custom_attributes 获取 if contact: logger.info( "Checking contact for token", contact_id=contact.id if contact else None, contact_type=type(contact).__name__ ) # 只有 WebhookContact 才有 custom_attributes if hasattr(contact, 'custom_attributes'): logger.info( "Checking contact custom_attributes", has_custom_attributes=bool(contact.custom_attributes) ) custom_attrs = contact.custom_attributes or {} logger.info( "Contact custom_attributes", keys=list(custom_attrs.keys()) if custom_attrs else [], has_jwt_token='jwt_token' in custom_attrs if custom_attrs else False, has_mall_token='mall_token' in custom_attrs if custom_attrs else False ) contact_dict = {"custom_attributes": custom_attrs} user_token = TokenManager.extract_token_from_contact(contact_dict) logger.debug("Extracted token from contact", has_token=bool(user_token)) else: logger.debug("Contact type is WebhookSender, no custom_attributes available") # 2. 尝试从 conversation.meta.sender.custom_attributes 获取(Chatwoot SDK setUser 设置的位置) if not user_token and conversation: logger.debug("Conversation object type", type=str(type(conversation))) if hasattr(conversation, 'model_dump'): conv_dict = conversation.model_dump() logger.debug("Conversation dict keys", keys=list(conv_dict.keys())) logger.debug("Has meta", has_meta='meta' in conv_dict) meta_sender = conv_dict.get('meta', {}).get('sender', {}) logger.info( "Conversation meta.sender", has_sender=bool(meta_sender), sender_keys=list(meta_sender.keys()) if meta_sender else [], has_custom_attributes=bool(meta_sender.get('custom_attributes')) if meta_sender else False ) if meta_sender.get('custom_attributes'): logger.info("Found custom_attributes in meta.sender", keys=list(meta_sender['custom_attributes'].keys())) user_token = TokenManager.extract_token_from_contact({'custom_attributes': meta_sender['custom_attributes']}) logger.info("Token found in conversation.meta.sender.custom_attributes", token_prefix=user_token[:20] if user_token else None) if user_token: logger.info( "JWT token found", user_id=user_id, source="cookie" if cookie_token else "contact", token_prefix=user_token[:20] if user_token else None ) else: logger.warning( "No JWT token found from any source", user_id=user_id, cookie_token_exists=bool(cookie_token), contact_type=type(contact).__name__ if contact else None ) logger.info( "Processing incoming message", conversation_id=conversation_id, user_id=user_id, has_token=bool(user_token), message_length=len(content) ) # Load conversation context from cache cache = get_cache_manager() await cache.connect() context = await cache.get_context(conversation_id) or {} history = await cache.get_messages(conversation_id) # Add token to context if available if user_token: context["user_token"] = user_token try: # Process message through agent workflow final_state = await process_message( conversation_id=conversation_id, user_id=user_id, account_id=account_id, message=content, history=history, context=context, user_token=user_token ) # Get response response = final_state.get("response") if not response: response = "抱歉,我暂时无法处理您的请求。请稍后重试或联系人工客服。" # Send response to Chatwoot # Create client with correct account_id from webhook from integrations.chatwoot import ChatwootClient chatwoot = ChatwootClient(account_id=int(account_id)) await chatwoot.send_message( conversation_id=conversation.id, content=response ) await chatwoot.close() # Handle human handoff if final_state.get("requires_human"): await chatwoot.update_conversation_status( conversation_id=conversation.id, status=ConversationStatus.OPEN ) # Add label for routing await chatwoot.add_labels( conversation_id=conversation.id, labels=["needs_human", final_state.get("intent", "unknown")] ) # Update cache await cache.add_message(conversation_id, "user", content) await cache.add_message(conversation_id, "assistant", response) # Save context new_context = final_state.get("context", {}) new_context["last_intent"] = final_state.get("intent") await cache.set_context(conversation_id, new_context) logger.info( "Message processed successfully", conversation_id=conversation_id, intent=final_state.get("intent"), requires_human=final_state.get("requires_human") ) except Exception as e: logger.error( "Message processing failed", conversation_id=conversation_id, error=str(e) ) # Send error response chatwoot = get_chatwoot_client() await chatwoot.send_message( conversation_id=conversation.id, content="抱歉,处理您的消息时遇到了问题。我们的客服团队将尽快为您服务。" ) # Transfer to human await chatwoot.update_conversation_status( conversation_id=conversation.id, status=ConversationStatus.OPEN ) async def handle_conversation_created(payload: ChatwootWebhookPayload) -> None: """Handle new conversation created Args: payload: Webhook payload """ conversation = payload.conversation if not conversation: return conversation_id = str(conversation.id) logger.info( "New conversation created", conversation_id=conversation_id ) # Initialize conversation context cache = get_cache_manager() await cache.connect() context = { "created": True, "inbox_id": conversation.inbox_id } # Add contact info to context contact = payload.contact if contact: context["contact_name"] = contact.name context["contact_email"] = contact.email if contact.custom_attributes: context.update(contact.custom_attributes) await cache.set_context(conversation_id, context) # Send welcome message chatwoot = get_chatwoot_client() await chatwoot.send_message( conversation_id=conversation.id, content="您好!我是 AI 智能助手,很高兴为您服务。请问有什么可以帮您的?\n\n您可以询问我关于订单、商品、售后等问题,我会尽力为您解答。" ) async def handle_conversation_status_changed(payload: ChatwootWebhookPayload) -> None: """Handle conversation status change Args: payload: Webhook payload """ conversation = payload.conversation if not conversation: return conversation_id = str(conversation.id) new_status = conversation.status logger.info( "Conversation status changed", conversation_id=conversation_id, status=new_status ) # If resolved, clean up context if new_status == "resolved": cache = get_cache_manager() await cache.connect() await cache.delete_context(conversation_id) await cache.clear_messages(conversation_id) # ============ Webhook Endpoint ============ @router.post("/chatwoot") async def chatwoot_webhook( request: Request, background_tasks: BackgroundTasks ): """Chatwoot webhook endpoint Receives events from Chatwoot and processes them asynchronously. """ # Get raw body for signature verification body = await request.body() # 尝试从请求 Cookie 中获取用户 Token user_token = request.cookies.get("token") # 从 Cookie 读取 token if user_token: logger.info("User token found in request cookies", token_prefix=user_token[:20]) else: logger.debug("No token found in request cookies (this is expected for webhook requests)") # Verify signature signature = request.headers.get("X-Chatwoot-Signature", "") if not verify_webhook_signature(body, signature): logger.warning("Invalid webhook signature") raise HTTPException(status_code=401, detail="Invalid signature") # Parse payload try: payload = ChatwootWebhookPayload.model_validate_json(body) except Exception as e: logger.error("Failed to parse webhook payload", error=str(e)) raise HTTPException(status_code=400, detail="Invalid payload") event = payload.event logger.debug(f"Webhook received: {event}") # Filter out bot's own messages if payload.message_type == "outgoing": return {"status": "ignored", "reason": "outgoing message"} # Filter private messages if payload.private: return {"status": "ignored", "reason": "private message"} # Route by event type if event == "message_created": # Only process incoming messages from contacts if payload.message_type == "incoming": background_tasks.add_task(handle_incoming_message, payload, user_token) elif event == "conversation_created": background_tasks.add_task(handle_conversation_created, payload) elif event == "conversation_status_changed": background_tasks.add_task(handle_conversation_status_changed, payload) elif event == "conversation_updated": # Handle other conversation updates if needed pass return {"status": "accepted", "event": event}