pipecat/examples/phone-chatbot/bot_runner_helpers.py

# bot_runner_helpers.py
from typing import Any, Dict, Optional

from bot_constants import (
    DEFAULT_CALLTRANSFER_MODE,
    DEFAULT_DIALIN_EXAMPLE,
    DEFAULT_SPEAK_SUMMARY,
    DEFAULT_STORE_SUMMARY,
    DEFAULT_TEST_IN_PREBUILT,
)
from call_connection_manager import CallConfigManager

# ----------------- Configuration Helpers ----------------- #


def determine_room_capabilities(config_body: Optional[Dict[str, Any]] = None) -> Dict[str, bool]:
    """Determine room capabilities based on the configuration.

    This function examines the configuration to determine which capabilities
    the Daily room should have enabled.

    Args:
        config_body: Configuration dictionary that determines room capabilities

    Returns:
        Dictionary of capability flags
    """
    capabilities = {
        "enable_dialin": False,
        "enable_dialout": False,
        # Add more capabilities here in the future as needed
    }

    if not config_body:
        return capabilities

    # Check for dialin capability
    capabilities["enable_dialin"] = "dialin_settings" in config_body

    # Check for dialout capability - needed for outbound calls or transfers
    has_dialout_settings = "dialout_settings" in config_body

    # Check if there's a transfer to an operator configured
    has_call_transfer = "call_transfer" in config_body

    # Enable dialout if any condition requires it
    capabilities["enable_dialout"] = has_dialout_settings or has_call_transfer

    return capabilities


def ensure_dialout_settings_array(body: Dict[str, Any]) -> Dict[str, Any]:
    """Ensures dialout_settings is an array of objects.

    Args:
        body: The configuration dictionary

    Returns:
        Updated configuration with dialout_settings as an array
    """
    if "dialout_settings" in body:
        # Convert to array if it's not already one
        if not isinstance(body["dialout_settings"], list):
            body["dialout_settings"] = [body["dialout_settings"]]

    return body


def ensure_prompt_config(body: Dict[str, Any]) -> Dict[str, Any]:
    """Ensures the body has appropriate prompts settings, but doesn't add defaults.

    Only makes sure the prompt section exists, allowing the bot script to handle defaults.

    Args:
        body: The configuration dictionary

    Returns:
        Updated configuration with prompt settings section
    """
    if "prompts" not in body:
        body["prompts"] = []
    return body


def create_call_transfer_settings(body: Dict[str, Any]) -> Dict[str, Any]:
    """Create call transfer settings based on configuration and customer mapping.

    Args:
        body: The configuration dictionary

    Returns:
        Call transfer settings dictionary
    """
    # Default transfer settings
    transfer_settings = {
        "mode": DEFAULT_CALLTRANSFER_MODE,
        "speakSummary": DEFAULT_SPEAK_SUMMARY,
        "storeSummary": DEFAULT_STORE_SUMMARY,
        "testInPrebuilt": DEFAULT_TEST_IN_PREBUILT,
    }

    # If call_transfer already exists, merge the defaults with the existing settings
    # This ensures all required fields exist while preserving user-specified values
    if "call_transfer" in body:
        existing_settings = body["call_transfer"]
        # Update defaults with existing settings (existing values will override defaults)
        for key, value in existing_settings.items():
            transfer_settings[key] = value
    else:
        # No existing call_transfer - check if we have dialin settings for customer lookup
        if "dialin_settings" in body:
            # Create a temporary routing manager just for customer lookup
            call_config_manager = CallConfigManager(body)

            # Get caller info
            caller_info = call_config_manager.get_caller_info()
            from_number = caller_info.get("caller_number")

            if from_number:
                # Get customer name from phone number
                customer_name = call_config_manager.get_customer_name(from_number)

                # If we know the customer name, add it to the config for the bot to use
                if customer_name:
                    transfer_settings["customerName"] = customer_name

    return transfer_settings


def create_simple_dialin_settings(body: Dict[str, Any]) -> Dict[str, Any]:
    """Create simple dialin settings based on configuration.

    Args:
        body: The configuration dictionary

    Returns:
        Simple dialin settings dictionary
    """
    # Default simple dialin settings
    simple_dialin_settings = {
        "testInPrebuilt": DEFAULT_TEST_IN_PREBUILT,
    }

    # If simple_dialin already exists, merge the defaults with the existing settings
    if "simple_dialin" in body:
        existing_settings = body["simple_dialin"]
        # Update defaults with existing settings (existing values will override defaults)
        for key, value in existing_settings.items():
            simple_dialin_settings[key] = value

    return simple_dialin_settings


def create_simple_dialout_settings(body: Dict[str, Any]) -> Dict[str, Any]:
    """Create simple dialout settings based on configuration.

    Args:
        body: The configuration dictionary

    Returns:
        Simple dialout settings dictionary
    """
    # Default simple dialout settings
    simple_dialout_settings = {
        "testInPrebuilt": DEFAULT_TEST_IN_PREBUILT,
    }

    # If simple_dialout already exists, merge the defaults with the existing settings
    if "simple_dialout" in body:
        existing_settings = body["simple_dialout"]
        # Update defaults with existing settings (existing values will override defaults)
        for key, value in existing_settings.items():
            simple_dialout_settings[key] = value

    return simple_dialout_settings


async def process_dialin_request(data: Dict[str, Any]) -> Dict[str, Any]:
    """Process incoming dial-in request data to create a properly formatted body.

    Converts camelCase fields received from webhook to snake_case format
    for internal consistency across the codebase.

    Args:
        data: Raw dialin data from webhook

    Returns:
        Properly formatted configuration with snake_case keys
    """
    # Create base body with dialin settings
    body = {
        "dialin_settings": {
            "to": data.get("To", ""),
            "from": data.get("From", ""),
            "call_id": data.get("callId", data.get("CallSid", "")),  # Convert to snake_case
            "call_domain": data.get("callDomain", ""),  # Convert to snake_case
        }
    }

    # Use the global default to determine which example to run for dialin webhooks
    example = DEFAULT_DIALIN_EXAMPLE

    # Configure the bot based on the example
    if example == "call_transfer":
        # Create call transfer settings
        body["call_transfer"] = create_call_transfer_settings(body)
    elif example == "simple_dialin":
        # Create simple dialin settings
        body["simple_dialin"] = create_simple_dialin_settings(body)

    return body