Update call_connection_manager.py

call_connection_manager.py

@@ -3,269 +3,141 @@
 #
 # SPDX-License-Identifier: BSD 2-Clause License
 #
-"""call_connection_manager.py.
-
-Manages customer/operator relationships and call routing for voice bots.
-Provides mapping between customers and operators, and functions for retrieving
-contact information. Also includes call state management.
-"""
-
 import json
 import os
+import time
 from typing import Any, Dict, List, Optional
-
 from loguru import logger
 
-
 class CallFlowState:
-    """State for tracking call flow operations and state transitions."""
-
     def __init__(self):
-        # Operator-related state
         self.dialed_operator = False
         self.operator_connected = False
         self.current_operator_index = 0
         self.operator_dialout_settings = []
         self.summary_finished = False
-
-        # Voicemail detection state
         self.voicemail_detected = False
         self.human_detected = False
         self.voicemail_message_left = False
-
-        # Call termination state
         self.call_terminated = False
         self.participant_left_early = False
+        self.silence_prompt_count = 0  # Track silence prompts
 
-    # Operator-related methods
     def set_operator_dialed(self):
-        """Mark that an operator has been dialed."""
         self.dialed_operator = True
 
     def set_operator_connected(self):
-        """Mark that an operator has connected to the call."""
         self.operator_connected = True
-        # Summary is not finished when operator first connects
         self.summary_finished = False
 
     def set_operator_disconnected(self):
-        """Handle operator disconnection."""
         self.operator_connected = False
         self.summary_finished = False
 
     def set_summary_finished(self):
-        """Mark the summary as finished."""
         self.summary_finished = True
 
     def set_operator_dialout_settings(self, settings):
-        """Set the list of operator dialout settings to try."""
         self.operator_dialout_settings = settings
         self.current_operator_index = 0
 
     def get_current_dialout_setting(self):
-        """Get the current operator dialout setting to try."""
-        if not self.operator_dialout_settings or self.current_operator_index >= len(
-            self.operator_dialout_settings
-        ):
+        if not self.operator_dialout_settings or self.current_operator_index >= len(self.operator_dialout_settings):
             return None
         return self.operator_dialout_settings[self.current_operator_index]
 
     def move_to_next_operator(self):
-        """Move to the next operator in the list."""
         self.current_operator_index += 1
         return self.get_current_dialout_setting()
 
-    # Voicemail detection methods
     def set_voicemail_detected(self):
-        """Mark that a voicemail system has been detected."""
         self.voicemail_detected = True
         self.human_detected = False
 
     def set_human_detected(self):
-        """Mark that a human has been detected (not voicemail)."""
         self.human_detected = True
         self.voicemail_detected = False
 
     def set_voicemail_message_left(self):
-        """Mark that a voicemail message has been left."""
         self.voicemail_message_left = True
 
-    # Call termination methods
     def set_call_terminated(self):
-        """Mark that the call has been terminated by the bot."""
         self.call_terminated = True
 
     def set_participant_left_early(self):
-        """Mark that a participant left the call early."""
         self.participant_left_early = True
 
+    def increment_silence_prompts(self):
+        self.silence_prompt_count += 1
 
-class SessionManager:
-    """Centralized management of session IDs and state for all call participants."""
+    def reset_silence_prompts(self):
+        self.silence_prompt_count = 0
 
+class SessionManager:
     def __init__(self):
-        # Track session IDs of different participant types
-        self.session_ids = {
-            "operator": None,
-            "customer": None,
-            "bot": None,
-            # Add other participant types as needed
-        }
-
-        # References for easy access in processors that need mutable containers
-        self.session_id_refs = {
-            "operator": [None],
-            "customer": [None],
-            "bot": [None],
-            # Add other participant types as needed
-        }
-
-        # State object for call flow
+        self.session_ids = {"operator": None, "customer": None, "bot": None}
+        self.session_id_refs = {"operator": [None], "customer": [None], "bot": [None]}
         self.call_flow_state = CallFlowState()
 
     def set_session_id(self, participant_type, session_id):
-        """Set the session ID for a specific participant type.
-
-        Args:
-            participant_type: Type of participant (e.g., "operator", "customer", "bot")
-            session_id: The session ID to set
-        """
         if participant_type in self.session_ids:
             self.session_ids[participant_type] = session_id
-
-            # Also update the corresponding reference if it exists
             if participant_type in self.session_id_refs:
                 self.session_id_refs[participant_type][0] = session_id
 
     def get_session_id(self, participant_type):
-        """Get the session ID for a specific participant type.
-
-        Args:
-            participant_type: Type of participant (e.g., "operator", "customer", "bot")
-
-        Returns:
-            The session ID or None if not set
-        """
         return self.session_ids.get(participant_type)
 
     def get_session_id_ref(self, participant_type):
-        """Get the mutable reference for a specific participant type.
-
-        Args:
-            participant_type: Type of participant (e.g., "operator", "customer", "bot")
-
-        Returns:
-            A mutable list container holding the session ID or None if not available
-        """
         return self.session_id_refs.get(participant_type)
 
     def is_participant_type(self, session_id, participant_type):
-        """Check if a session ID belongs to a specific participant type.
-
-        Args:
-            session_id: The session ID to check
-            participant_type: Type of participant (e.g., "operator", "customer", "bot")
-
-        Returns:
-            True if the session ID matches the participant type, False otherwise
-        """
         return self.session_ids.get(participant_type) == session_id
 
     def reset_participant(self, participant_type):
-        """Reset the state for a specific participant type.
-
-        Args:
-            participant_type: Type of participant (e.g., "operator", "customer", "bot")
-        """
         if participant_type in self.session_ids:
             self.session_ids[participant_type] = None
-
             if participant_type in self.session_id_refs:
                 self.session_id_refs[participant_type][0] = None
-
-            # Additional reset actions for specific participant types
             if participant_type == "operator":
                 self.call_flow_state.set_operator_disconnected()
 
-
 class CallConfigManager:
-    """Manages customer/operator relationships and call routing."""
-
     def __init__(self, body_data: Dict[str, Any] = None):
-        """Initialize with optional body data.
-
-        Args:
-            body_data: Optional dictionary containing request body data
-        """
         self.body = body_data or {}
-
-        # Get environment variables with fallbacks
-        self.dial_in_from_number = os.getenv("DIAL_IN_FROM_NUMBER", "+10000000001")
-        self.dial_out_to_number = os.getenv("DIAL_OUT_TO_NUMBER", "+10000000002")
-        self.operator_number = os.getenv("OPERATOR_NUMBER", "+10000000003")
-
-        # Initialize maps with dynamic values
+        self.dial_in_from_number = os.environ.get("HF_DIAL_IN_FROM_NUMBER", "+10000000001")
+        self.dial_out_to_number = os.environ.get("HF_DIAL_OUT_TO_NUMBER", "+10000000002")
+        self.operator_number = os.environ.get("HF_OPERATOR_NUMBER", "+10000000003")
         self._initialize_maps()
         self._build_reverse_lookup_maps()
 
     def _initialize_maps(self):
-        """Initialize the customer and operator maps with environment variables."""
-        # Maps customer names to their contact information
         self.CUSTOMER_MAP = {
-            "Dominic": {
-                "phoneNumber": self.dial_in_from_number,  # I have two phone numbers, one for dialing in and one for dialing out. I give myself a separate name for each.
-            },
-            "Stewart": {
-                "phoneNumber": self.dial_out_to_number,
-            },
-            "James": {
-                "phoneNumber": "+10000000000",
-                "callerId": "james-caller-id-uuid",
-                "sipUri": "sip:[email protected]",
-            },
-            "Sarah": {
-                "sipUri": "sip:[email protected]",
-            },
-            "Michael": {
-                "phoneNumber": "+16505557890",
-                "callerId": "michael-caller-id-uuid",
-            },
+            "Dominic": {"phoneNumber": self.dial_in_from_number},
+            "Stewart": {"phoneNumber": self.dial_out_to_number},
+            "James": {"phoneNumber": "+10000000000", "callerId": "james-caller-id-uuid", "sipUri": "sip:[email protected]"},
+            "Sarah": {"sipUri": "sip:[email protected]"},
+            "Michael": {"phoneNumber": "+16505557890", "callerId": "michael-caller-id-uuid"},
         }
-
-        # Maps customer names to their assigned operator names
         self.CUSTOMER_TO_OPERATOR_MAP = {
-            "Dominic": ["Yunyoung", "Maria"],  # Try Yunyoung first, then Maria
+            "Dominic": ["Yunyoung", "Maria"],
             "Stewart": "Yunyoung",
             "James": "Yunyoung",
             "Sarah": "Jennifer",
             "Michael": "Paul",
-            # Default mapping to ensure all customers have an operator
             "Default": "Yunyoung",
         }
-
-        # Maps operator names to their contact details
         self.OPERATOR_CONTACT_MAP = {
-            "Paul": {
-                "phoneNumber": "+12345678904",
-                "callerId": "paul-caller-id-uuid",
-            },
-            "Yunyoung": {
-                "phoneNumber": self.operator_number,  # Dials out to my other phone number.
-            },
-            "Maria": {
-                "sipUri": "sip:[email protected]",
-            },
+            "Paul": {"phoneNumber": "+12345678904", "callerId": "paul-caller-id-uuid"},
+            "Yunyoung": {"phoneNumber": self.operator_number},
+            "Maria": {"sipUri": "sip:[email protected]"},
             "Jennifer": {"phoneNumber": "+14155559876", "callerId": "jennifer-caller-id-uuid"},
-            "Default": {
-                "phoneNumber": self.operator_number,  # Use the operator number as default
-            },
+            "Default": {"phoneNumber": self.operator_number},
         }
 
     def _build_reverse_lookup_maps(self):
-        """Build reverse lookup maps for phone numbers and SIP URIs to customer names."""
         self._PHONE_TO_CUSTOMER_MAP = {}
         self._SIP_TO_CUSTOMER_MAP = {}
-
         for customer_name, contact_info in self.CUSTOMER_MAP.items():
             if "phoneNumber" in contact_info:
                 self._PHONE_TO_CUSTOMER_MAP[contact_info["phoneNumber"]] = customer_name
@@ -274,158 +146,59 @@ class CallConfigManager:
 
     @classmethod
     def from_json_string(cls, json_string: str):
-        """Create a CallRoutingManager from a JSON string.
-
-        Args:
-            json_string: JSON string containing body data
-
-        Returns:
-            CallRoutingManager instance with parsed data
-
-        Raises:
-            json.JSONDecodeError: If JSON string is invalid
-        """
         body_data = json.loads(json_string)
         return cls(body_data)
 
     def find_customer_by_contact(self, contact_info: str) -> Optional[str]:
-        """Find customer name from a contact identifier (phone number or SIP URI).
-
-        Args:
-            contact_info: The contact identifier (phone number or SIP URI)
-
-        Returns:
-            The customer name or None if not found
-        """
-        # Check if it's a phone number
         if contact_info in self._PHONE_TO_CUSTOMER_MAP:
             return self._PHONE_TO_CUSTOMER_MAP[contact_info]
-
-        # Check if it's a SIP URI
         if contact_info in self._SIP_TO_CUSTOMER_MAP:
             return self._SIP_TO_CUSTOMER_MAP[contact_info]
-
         return None
 
     def get_customer_name(self, phone_number: str) -> Optional[str]:
-        """Get customer name from their phone number.
-
-        Args:
-            phone_number: The customer's phone number
-
-        Returns:
-            The customer name or None if not found
-        """
-        # Note: In production, this would likely query a database
         return self.find_customer_by_contact(phone_number)
 
     def get_operators_for_customer(self, customer_name: Optional[str]) -> List[str]:
-        """Get the operator name(s) assigned to a customer.
-
-        Args:
-            customer_name: The customer's name
-
-        Returns:
-            List of operator names (single item or multiple)
-        """
-        # Note: In production, this would likely query a database
         if not customer_name or customer_name not in self.CUSTOMER_TO_OPERATOR_MAP:
             return ["Default"]
-
         operators = self.CUSTOMER_TO_OPERATOR_MAP[customer_name]
-        # Convert single string to list for consistency
         if isinstance(operators, str):
             return [operators]
         return operators
 
     def get_operator_dialout_settings(self, operator_name: str) -> Dict[str, str]:
-        """Get an operator's dialout settings from their name.
-
-        Args:
-            operator_name: The operator's name
-
-        Returns:
-            Dictionary with dialout settings for the operator
-        """
-        # Note: In production, this would likely query a database
         return self.OPERATOR_CONTACT_MAP.get(operator_name, self.OPERATOR_CONTACT_MAP["Default"])
 
-    def get_dialout_settings_for_caller(
-        self, from_number: Optional[str] = None
-    ) -> List[Dict[str, str]]:
-        """Determine the appropriate operator dialout settings based on caller's number.
-
-        This method uses the caller's number to look up the customer name,
-        then finds the assigned operators for that customer, and returns
-        an array of operator dialout settings to try in sequence.
-
-        Args:
-            from_number: The caller's phone number (from dialin_settings)
-
-        Returns:
-            List of operator dialout settings to try
-        """
+    def get_dialout_settings_for_caller(self, from_number: Optional[str] = None) -> List[Dict[str, str]]:
         if not from_number:
-            # If we don't have dialin settings, use the Default operator
             return [self.get_operator_dialout_settings("Default")]
-
-        # Get customer name from phone number
         customer_name = self.get_customer_name(from_number)
-
-        # Get operator names assigned to this customer
         operator_names = self.get_operators_for_customer(customer_name)
-
-        # Get dialout settings for each operator
         return [self.get_operator_dialout_settings(name) for name in operator_names]
 
     def get_caller_info(self) -> Dict[str, Optional[str]]:
-        """Get caller and dialed numbers from dialin settings in the body.
-
-        Returns:
-            Dictionary containing caller_number and dialed_number
-        """
         raw_dialin_settings = self.body.get("dialin_settings")
         if not raw_dialin_settings:
             return {"caller_number": None, "dialed_number": None}
-
-        # Handle different case variations
         dialed_number = raw_dialin_settings.get("To") or raw_dialin_settings.get("to")
         caller_number = raw_dialin_settings.get("From") or raw_dialin_settings.get("from")
-
         return {"caller_number": caller_number, "dialed_number": dialed_number}
 
     def get_caller_number(self) -> Optional[str]:
-        """Get the caller's phone number from dialin settings in the body.
-
-        Returns:
-            The caller's phone number or None if not available
-        """
         return self.get_caller_info()["caller_number"]
 
     async def start_dialout(self, transport, dialout_settings=None):
-        """Helper function to start dialout using the provided settings or from body.
-
-        Args:
-            transport: The transport instance to use for dialout
-            dialout_settings: Optional override for dialout settings
-
-        Returns:
-            None
-        """
-        # Use provided settings or get from body
         settings = dialout_settings or self.get_dialout_settings()
         if not settings:
             logger.warning("No dialout settings available")
             return
-
         for setting in settings:
             if "phoneNumber" in setting:
                 logger.info(f"Dialing number: {setting['phoneNumber']}")
                 if "callerId" in setting:
                     logger.info(f"with callerId: {setting['callerId']}")
-                    await transport.start_dialout(
-                        {"phoneNumber": setting["phoneNumber"], "callerId": setting["callerId"]}
-                    )
+                    await transport.start_dialout({"phoneNumber": setting["phoneNumber"], "callerId": setting["callerId"]})
                 else:
                     logger.info("with no callerId")
                     await transport.start_dialout({"phoneNumber": setting["phoneNumber"]})
@@ -436,59 +209,27 @@ class CallConfigManager:
                 logger.warning(f"Unknown dialout setting format: {setting}")
 
     def get_dialout_settings(self) -> Optional[List[Dict[str, Any]]]:
-        """Extract dialout settings from the body.
-
-        Returns:
-            List of dialout setting objects or None if not present
-        """
-        # Check if we have dialout settings
         if "dialout_settings" in self.body:
             dialout_settings = self.body["dialout_settings"]
-
-            # Convert to list if it's an object (for backward compatibility)
             if isinstance(dialout_settings, dict):
                 return [dialout_settings]
             elif isinstance(dialout_settings, list):
                 return dialout_settings
-
         return None
 
     def get_dialin_settings(self) -> Optional[Dict[str, Any]]:
-        """Extract dialin settings from the body.
-
-        Handles both camelCase and snake_case variations of fields for backward compatibility,
-        but normalizes to snake_case for internal usage.
-
-        Returns:
-            Dictionary containing dialin settings or None if not present
-        """
         raw_dialin_settings = self.body.get("dialin_settings")
         if not raw_dialin_settings:
             return None
-
-        # Normalize dialin settings to handle different case variations
-        # Prioritize snake_case (call_id, call_domain) but fall back to camelCase (callId, callDomain)
         dialin_settings = {
             "call_id": raw_dialin_settings.get("call_id") or raw_dialin_settings.get("callId"),
-            "call_domain": raw_dialin_settings.get("call_domain")
-            or raw_dialin_settings.get("callDomain"),
+            "call_domain": raw_dialin_settings.get("call_domain") or raw_dialin_settings.get("callDomain"),
             "to": raw_dialin_settings.get("to") or raw_dialin_settings.get("To"),
             "from": raw_dialin_settings.get("from") or raw_dialin_settings.get("From"),
         }
-
         return dialin_settings
 
-    # Bot prompt helper functions - no defaults provided, just return what's in the body
-
     def get_prompt(self, prompt_name: str) -> Optional[str]:
-        """Retrieve the prompt text for a given prompt name.
-
-        Args:
-            prompt_name: The name of the prompt to retrieve.
-
-        Returns:
-            The prompt string corresponding to the provided name, or None if not configured.
-        """
         prompts = self.body.get("prompts", [])
         for prompt in prompts:
             if prompt.get("name") == prompt_name:
@@ -496,41 +237,21 @@ class CallConfigManager:
         return None
 
     def get_transfer_mode(self) -> Optional[str]:
-        """Get transfer mode from the body.
-
-        Returns:
-            Transfer mode string or None if not configured
-        """
         if "call_transfer" in self.body:
             return self.body["call_transfer"].get("mode")
         return None
 
     def get_speak_summary(self) -> Optional[bool]:
-        """Get speak summary from the body.
-
-        Returns:
-            Boolean indicating if summary should be spoken or None if not configured
-        """
         if "call_transfer" in self.body:
             return self.body["call_transfer"].get("speakSummary")
         return None
 
     def get_store_summary(self) -> Optional[bool]:
-        """Get store summary from the body.
-
-        Returns:
-            Boolean indicating if summary should be stored or None if not configured
-        """
         if "call_transfer" in self.body:
             return self.body["call_transfer"].get("storeSummary")
         return None
 
     def is_test_mode(self) -> bool:
-        """Check if running in test mode.
-
-        Returns:
-            Boolean indicating if test mode is enabled
-        """
         if "voicemail_detection" in self.body:
             return bool(self.body["voicemail_detection"].get("testInPrebuilt"))
         if "call_transfer" in self.body:
@@ -542,67 +263,46 @@ class CallConfigManager:
         return False
 
     def is_voicemail_detection_enabled(self) -> bool:
-        """Check if voicemail detection is enabled in the body.
-
-        Returns:
-            Boolean indicating if voicemail detection is enabled
-        """
         return bool(self.body.get("voicemail_detection"))
 
     def customize_prompt(self, prompt: str, customer_name: Optional[str] = None) -> str:
-        """Insert customer name into prompt template if available.
-
-        Args:
-            prompt: The prompt template containing optional {customer_name} placeholders
-            customer_name: Optional customer name to insert
-
-        Returns:
-            Customized prompt with customer name inserted
-        """
         if customer_name and prompt:
             return prompt.replace("{customer_name}", customer_name)
         return prompt
 
     def create_system_message(self, content: str) -> Dict[str, str]:
-        """Create a properly formatted system message.
-
-        Args:
-            content: The message content
-
-        Returns:
-            Dictionary with role and content for the system message
-        """
         return {"role": "system", "content": content}
 
     def create_user_message(self, content: str) -> Dict[str, str]:
-        """Create a properly formatted user message.
-
-        Args:
-            content: The message content
-
-        Returns:
-            Dictionary with role and content for the user message
-        """
         return {"role": "user", "content": content}
 
-    def get_customer_info_suffix(
-        self, customer_name: Optional[str] = None, preposition: str = "for"
-    ) -> str:
-        """Create a consistent customer info suffix.
-
-        Args:
-            customer_name: Optional customer name
-            preposition: Preposition to use before the name (e.g., "for", "to", "")
-
-        Returns:
-            String with formatted customer info suffix
-        """
+    def get_customer_info_suffix(self, customer_name: Optional[str] = None, preposition: str = "for") -> str:
         if not customer_name:
             return ""
-
-        # Add a space before the preposition if it's not empty
         space_prefix = " " if preposition else ""
-        # For non-empty prepositions, add a space after it
         space_suffix = " " if preposition else ""
-
         return f"{space_prefix}{preposition}{space_suffix}{customer_name}"
+
+    async def log_call_summary(self, start_time: float, session_manager: SessionManager, caller_number: str, dialed_number: str, customer_name: str):
+        """Log call summary to a file."""
+        end_time = time.time()
+        duration = end_time - start_time
+        silence_prompts = session_manager.call_flow_state.silence_prompt_count
+        summary = {
+            "call_start_time": time.strftime("%Y-%m-%d %H:%M:%S", time.localtime(start_time)),
+            "call_end_time": time.strftime("%Y-%m-%d %H:%M:%S", time.localtime(end_time)),
+            "duration_seconds": round(duration, 2),
+            "caller_number": caller_number,
+            "dialed_number": dialed_number,
+            "customer_name": customer_name,
+            "silence_prompts_triggered": silence_prompts,
+            "call_terminated_by_bot": session_manager.call_flow_state.call_terminated,
+            "participant_left_early": session_manager.call_flow_state.participant_left_early
+        }
+        log_file = "/home/user/call_summary.log"
+        try:
+            with open(log_file, "a") as f:
+                f.write(json.dumps(summary) + "\n")
+            logger.info(f"Call summary logged to {log_file}")
+        except Exception as e:
+            logger.error(f"Failed to log call summary: {e}")