webhooks.py

import asyncio
import json
from datetime import datetime
from typing import List, Any

import requests
import websockets

from database.redis_db import (
    get_user_webhook_db,
    user_webhook_status_db,
    disable_user_webhook_db,
    enable_user_webhook_db,
    set_user_webhook_db,
)
from models.conversation import Conversation
from models.users import WebhookType
import database.notifications as notification_db
import database.users as users_db
from utils.notifications import send_notification
import logging

logger = logging.getLogger(__name__)


def _json_serialize_datetime(obj: Any) -> Any:
    """Helper function to recursively convert datetime objects to ISO format strings for JSON serialization"""
    if isinstance(obj, datetime):
        return obj.isoformat()
    elif isinstance(obj, dict):
        return {key: _json_serialize_datetime(value) for key, value in obj.items()}
    elif isinstance(obj, list):
        return [_json_serialize_datetime(item) for item in obj]
    else:
        return obj


def _add_speaker_names_to_payload(uid, payload: dict):
    """Add speaker_name to transcript segments in webhook payload."""
    segments = payload.get('transcript_segments', [])
    if not segments:
        return

    user_profile = users_db.get_user_profile(uid)
    user_name = user_profile.get('name') or 'User'

    person_ids = [seg.get('person_id') for seg in segments if seg.get('person_id')]
    people_map = {}
    if person_ids:
        people_data = users_db.get_people_by_ids(uid, list(set(person_ids)))
        people_map = {p['id']: p['name'] for p in people_data}

    for seg in segments:
        if seg.get('is_user'):
            seg['speaker_name'] = user_name
        elif seg.get('person_id') and seg['person_id'] in people_map:
            seg['speaker_name'] = people_map[seg['person_id']]
        else:
            seg['speaker_name'] = f"Speaker {seg.get('speaker_id', 0)}"


def conversation_created_webhook(uid, memory: Conversation):
    toggled = user_webhook_status_db(uid, WebhookType.memory_created)

    if toggled:
        webhook_url = get_user_webhook_db(uid, WebhookType.memory_created)
        if not webhook_url:
            return
        webhook_url += f'?uid={uid}'
        try:
            payload = memory.as_dict_cleaned_dates()
            _add_speaker_names_to_payload(uid, payload)
            payload = _json_serialize_datetime(payload)
            response = requests.post(
                webhook_url,
                json=payload,
                headers={'Content-Type': 'application/json'},
                timeout=30,
            )
            logger.info(f'memory_created_webhook: {webhook_url} {response.status_code}')
        except Exception as e:
            logger.error(f"Error sending memory created to developer webhook: {e}")
    else:
        return


def day_summary_webhook(uid, summary: str):
    toggled = user_webhook_status_db(uid, WebhookType.day_summary)
    if toggled:
        webhook_url = get_user_webhook_db(uid, WebhookType.day_summary)
        if not webhook_url:
            return
        webhook_url += f'?uid={uid}'
        try:
            response = requests.post(
                webhook_url,
                json={'summary': summary, 'uid': uid, 'created_at': datetime.now().isoformat()},
                headers={'Content-Type': 'application/json'},
                timeout=30,
            )
            logger.info(f'day_summary_webhook: {webhook_url} {response.status_code}')
        except Exception as e:
            logger.error(f"Error sending day summary to developer webhook: {e}")
    else:
        return


async def realtime_transcript_webhook(uid, segments: List[dict]):
    logger.info(f"realtime_transcript_webhook {uid}")
    toggled = user_webhook_status_db(uid, WebhookType.realtime_transcript)

    if toggled:
        webhook_url = get_user_webhook_db(uid, WebhookType.realtime_transcript)
        if not webhook_url:
            return
        webhook_url += f'?uid={uid}'
        try:
            response = await asyncio.to_thread(
                requests.post,
                webhook_url,
                json={'segments': segments, 'session_id': uid},
                headers={'Content-Type': 'application/json'},
                timeout=15,
            )
            logger.info(f'realtime_transcript_webhook: {webhook_url} {response.status_code}')
            if response.status_code == 200:
                response_data = response.json()
                if not response_data:
                    return
                message = response_data.get('message', '')
                if len(message) > 5:
                    send_webhook_notification(uid, message)
        except Exception as e:
            logger.error(f"Error sending realtime transcript to developer webhook: {e}")
    else:
        return


def get_audio_bytes_webhook_seconds(uid: str):
    toggled = user_webhook_status_db(uid, WebhookType.audio_bytes)
    if toggled:
        webhook_url = get_user_webhook_db(uid, WebhookType.audio_bytes)
        if not webhook_url:
            return
        parts = webhook_url.split(',')
        if len(parts) == 2:
            try:
                return int(parts[1])
            except ValueError:
                pass
        return 5
    else:
        return


async def send_audio_bytes_developer_webhook(uid: str, sample_rate: int, data: bytearray):
    logger.info(f"send_audio_bytes_developer_webhook {uid}")
    # TODO: add a lock, send shorter segments, validate regex.
    toggled = user_webhook_status_db(uid, WebhookType.audio_bytes)
    if toggled:
        webhook_url = get_user_webhook_db(uid, WebhookType.audio_bytes)
        webhook_url = webhook_url.split(',')[0]
        if not webhook_url:
            return
        webhook_url += f'?sample_rate={sample_rate}&uid={uid}'
        try:
            response = await asyncio.to_thread(
                requests.post,
                webhook_url, data=data, headers={'Content-Type': 'application/octet-stream'}, timeout=15
            )
            logger.info(f'send_audio_bytes_developer_webhook: {webhook_url} {response.status_code}')
        except Exception as e:
            logger.error(f"Error sending audio bytes to developer webhook: {e}")
    else:
        return


# continue?
async def connect_user_webhook_ws(sample_rate: int, language: str, preseconds: int = 0):
    uri = ''

    try:
        socket = await websockets.connect(uri, extra_headers={})
        await socket.send(json.dumps({}))

        async def on_message():
            try:
                async for message in socket:
                    response = json.loads(message)
            except websockets.exceptions.ConnectionClosedOK:
                logger.info("Speechmatics connection closed normally.")
            except Exception as e:
                logger.error(f"Error receiving from Speechmatics: {e}")
            finally:
                if not socket.closed:
                    await socket.close()
                    logger.info("Speechmatics WebSocket closed in on_message.")

        asyncio.create_task(on_message())
        return socket
    except Exception as e:
        logger.error(f"Exception in process_audio_speechmatics: {e}")
        raise


def webhook_first_time_setup(uid: str, wType: WebhookType) -> bool:
    res = False
    url = get_user_webhook_db(uid, wType)
    if url == '' or url == ',':
        disable_user_webhook_db(uid, wType)
        res = False
    else:
        enable_user_webhook_db(uid, wType)
        res = True
    return res


def send_webhook_notification(user_id: str, message: str):
    send_notification(user_id, "Webhook" + ' says', message)