mirror of
http://88.130.71.182:3000/BlitTech/contexta_be.git
synced 2026-06-12 23:23:21 +00:00
63 lines
1.7 KiB
Python
63 lines
1.7 KiB
Python
import httpx
|
|
import logging
|
|
from datetime import datetime
|
|
from typing import Any, Dict, List, Optional
|
|
|
|
logger = logging.getLogger(__name__)
|
|
|
|
|
|
async def send_notification(
|
|
event_type: str,
|
|
data: Dict[str, Any],
|
|
webhook_url: Optional[str] = None,
|
|
) -> bool:
|
|
"""
|
|
Generic n8n notification sender.
|
|
Returns True if sent, False if not configured or failed.
|
|
"""
|
|
if not webhook_url:
|
|
return False
|
|
|
|
payload = {
|
|
"event": event_type,
|
|
"timestamp": datetime.utcnow().isoformat(),
|
|
**data,
|
|
}
|
|
|
|
try:
|
|
async with httpx.AsyncClient(timeout=10) as client:
|
|
response = await client.post(webhook_url, json=payload)
|
|
response.raise_for_status()
|
|
logger.info(f"n8n notification sent: {event_type}")
|
|
return True
|
|
except Exception as e:
|
|
logger.error(f"Failed to send n8n notification ({event_type}): {e}")
|
|
return False
|
|
|
|
|
|
async def send_handoff_notification(
|
|
chatbot_name: str,
|
|
owner_email: str,
|
|
conversation_history: List[dict],
|
|
trigger_message: str,
|
|
chatbot_id: str,
|
|
conversation_id: str,
|
|
webhook_url: Optional[str] = None,
|
|
) -> bool:
|
|
"""
|
|
Send a human handoff notification to the configured n8n webhook.
|
|
Returns True if sent, False if not configured or failed.
|
|
"""
|
|
return await send_notification(
|
|
event_type="handoff",
|
|
data={
|
|
"chatbot_name": chatbot_name,
|
|
"owner_email": owner_email,
|
|
"trigger_message": trigger_message,
|
|
"conversation_history": conversation_history[-10:],
|
|
"chatbot_id": chatbot_id,
|
|
"conversation_id": conversation_id,
|
|
},
|
|
webhook_url=webhook_url,
|
|
)
|