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 )