/ bot / transport_context.py
transport_context.py
  1  """Transport context for call reporting.
  2  
  3  This module provides a dataclass to capture transport-specific metadata
  4  (Daily WebRTC, Twilio telephony, local WebRTC) for end-of-call reports.
  5  """
  6  
  7  from dataclasses import dataclass
  8  from typing import Optional
  9  from urllib.parse import urlparse
 10  import uuid
 11  
 12  
 13  @dataclass
 14  class TransportContext:
 15      """Context for call reporting based on transport type.
 16  
 17      Attributes:
 18          transport: Transport type - "web", "daily", or "twilio"
 19          direction: Call direction - "inbound" (default)
 20          session_id: Unique session ID for recording filenames
 21          call_id: Unique call identifier
 22  
 23          # Daily-specific
 24          daily_room_name: Daily.co room name
 25  
 26          # Twilio-specific
 27          twilio_call_sid: Twilio Call SID (CA...)
 28          twilio_stream_sid: Twilio Media Stream SID (MZ...)
 29          twilio_direction: Twilio call direction
 30          twilio_phone_from: Caller phone number (E.164 format)
 31          twilio_phone_to: Receiving phone number (E.164 format)
 32      """
 33  
 34      transport: str  # "web", "daily", "twilio"
 35      direction: str  # "inbound"
 36      session_id: str  # For recording filenames
 37      call_id: str  # Unique call identifier
 38  
 39      # Daily-specific metadata
 40      daily_room_name: Optional[str] = None
 41  
 42      # Twilio-specific metadata
 43      twilio_call_sid: Optional[str] = None
 44      twilio_stream_sid: Optional[str] = None
 45      twilio_direction: Optional[str] = None
 46      twilio_phone_from: Optional[str] = None
 47      twilio_phone_to: Optional[str] = None
 48  
 49  
 50  def build_transport_context(
 51      transport_type: str,
 52      room_url: Optional[str] = None,
 53      call_data: Optional[dict] = None,
 54      direction: str = "inbound",
 55  ) -> TransportContext:
 56      """Build transport context from available data.
 57  
 58      Args:
 59          transport_type: "daily", "twilio", or "web"
 60          room_url: Daily room URL (for daily transport)
 61          call_data: Twilio call data from parse_telephony_websocket (for twilio transport)
 62          direction: Call direction, defaults to "inbound"
 63  
 64      Returns:
 65          TransportContext with appropriate IDs and metadata
 66      """
 67      if transport_type == "daily" and room_url:
 68          room_name = urlparse(room_url).path.strip("/")
 69          return TransportContext(
 70              transport="daily",
 71              direction=direction,
 72              session_id=room_name,
 73              call_id=room_name,
 74              daily_room_name=room_name,
 75          )
 76  
 77      if transport_type == "twilio" and call_data:
 78          call_sid = call_data.get("call_id", "")
 79          stream_sid = call_data.get("stream_id", "")
 80          body = call_data.get("body", {})
 81  
 82          # Use stream_sid for session_id, fallback to truncated call_sid
 83          session_id = stream_sid if stream_sid else call_sid[:16]
 84  
 85          return TransportContext(
 86              transport="twilio",
 87              direction=body.get("direction", direction),
 88              session_id=session_id,
 89              call_id=call_sid,
 90              twilio_call_sid=call_sid,
 91              twilio_stream_sid=stream_sid,
 92              twilio_direction=body.get("direction"),
 93              twilio_phone_from=body.get("phoneFrom"),
 94              twilio_phone_to=body.get("phoneTo"),
 95          )
 96  
 97      # Local WebRTC fallback
 98      unique_id = uuid.uuid4().hex[:12]
 99      return TransportContext(
100          transport="web",
101          direction=direction,
102          session_id=f"web-{unique_id}",
103          call_id=f"web-{unique_id}",
104      )