""" Views for the call APIs. """ import json import logging import os import pytz from typing import Any, cast from django.conf import settings from django.utils import timezone from django_filters.rest_framework import DjangoFilterBackend from rest_framework import viewsets, status from rest_framework.decorators import action from rest_framework.response import Response from rest_framework.permissions import IsAuthenticated, AllowAny from apps.appointments.serializers import BookAppointmentInputSerializer from apps.appointments.services import WebhookEventService, AppointmentService from apps.companies.constants import CompanyType, BotName from apps.userprofile.models import UserProfile from apps.appointments.models import Appointment from apps.customers.models import Customer from apps.companies.models import Company, CompanyBotSettings from apps.companies.utils import ( get_company_working_info, check_sales_time_by_phone, get_rabeeca_working_info, ) from apps.calls.repositories import MessageRepository from apps.calls.constants import TRANSFER_STATUS, SENTIMENT, BotType from apps.calls.filters import CallFilter, CallOrderingFilter from apps.calls.models import Call, CallActivity from apps.calls.pagination import CallLimitOffsetPagination from apps.calls.utils import get_bot_name from apps.calls.tasks import ( process_booking_intent, process_catch_phrases_for_call ) from apps.calls.serializers import ( CallDetailSerializer, ReadCallSerializer, ) from apps.calls.services import ( RecordingService, StorageService, SmsService, EmailService, NotificationService, get_caller_name_by_phone_number, get_twilio_client, ) from apps.calls.mixins import ( UserPerformanceMixin, TransferPercentageMixin, CallSummaryMixin, MonthlyStatsMixin, TopCallersMixin, HourlyCallsMixin, DailyCallsMixin, CallSearchMixin, ) logger = logging.getLogger(__name__) class ServiceCallViewSet( CallSearchMixin, TransferPercentageMixin, UserPerformanceMixin, CallSummaryMixin, MonthlyStatsMixin, TopCallersMixin, HourlyCallsMixin, DailyCallsMixin, viewsets.ModelViewSet ): """View for manage call APIs.""" serializer_class = CallDetailSerializer queryset = Call.objects.all() permission_classes = [IsAuthenticated] pagination_class = CallLimitOffsetPagination filter_backends = [DjangoFilterBackend, CallOrderingFilter] filterset_class = CallFilter ordering_fields = [ 'created_at', 'caller_name', 'from_number', 'transfer_user__username', 'call_time_category', 'sentiment', 'transfer_status', 'duration', 'cost' ] ordering = ['-created_at'] def get_queryset(self): user = self.request.user qs = self.queryset if user.is_superuser: pass elif user.active_company: try: if user.active_company.company_type == CompanyType.DEALERSHIP: qs = qs.filter( dealership=user.active_company ) else: qs = qs.filter(company=user.active_company) except Company.DoesNotExist: return self.queryset.none() else: return self.queryset.none() if self.action == "list": qs = qs.filter( bot_type=BotType.SERVICE_BOT.value ) qs = self.with_sort_name(qs) qs = self.with_sort_advisor_name(qs) qs = self.apply_call_search(qs) return qs def get_serializer_class(self): """Return the serializer class for request.""" if self.action == 'list': return CallDetailSerializer return self.serializer_class def perform_create(self, serializer): """Create a new call.""" serializer.save() @action(detail=False, methods=['get'], url_path='read') def read_call(self, request): """Process booking intent for a call and mark as read.""" serializer = ReadCallSerializer( data=request.query_params, context={'request': request} ) if not serializer.is_valid(): return Response( serializer.errors, status=status.HTTP_400_BAD_REQUEST ) call_id = serializer.validated_data['id'] user = request.user if user.is_superuser: call = Call.objects.get(id=call_id) current_time = timezone.now() call.read_by = user call.read_at = current_time CallActivity.objects.create( call=call, user=request.user, action='READ', performed_at=current_time, ) call.save() return Response({ "detail": "Call marked as read from super admin.", }, status=status.HTTP_200_OK) if user.active_company.company_type == CompanyType.DEALERSHIP: call = Call.objects.get(id=call_id, dealership=user.active_company) else: call = Call.objects.get(id=call_id, company=user.active_company) company_timezone = pytz.timezone(user.active_company.timezone) current_time_utc = timezone.now() current_time_company = current_time_utc.astimezone(company_timezone) call.read_by = user call.read_at = current_time_company CallActivity.objects.create( call=call, user=request.user, action='READ', performed_at=current_time_company, ) call.save() return Response({ "detail": "Call marked as read.", }, status=status.HTTP_200_OK) @action(detail=True, methods=['get'], url_path='get-recording') def download_recording(self, request, pk=None): if pk: logger.info(pk) call = self.get_object() recording_service = RecordingService(get_twilio_client()) storage_service = StorageService() filename = storage_service.build_call_audio_filename(call) relative_path, filepath = storage_service.call_recording_local_path(filename) from utils.feature_flags import is_feature_enabled if is_feature_enabled("new_audio_stream"): if not os.path.exists(filepath): recording = recording_service.get_latest_call_recording( call.twilio_call_sid ) response = recording_service.download(recording.sid) storage_service.save_call_recording_stream(filepath, response) from django.http import FileResponse from django.core.files import File response = FileResponse( File(open(filepath, "rb")), content_type="audio/mpeg", ) response["Accept-Ranges"] = "bytes" return response if os.path.exists(filepath): return Response( {"recording_url": storage_service.call_recording_public_url( request, relative_path )} ) recording = recording_service.get_latest_call_recording( call.twilio_call_sid ) response = recording_service.download(recording.sid) storage_service.save_call_recording_stream(filepath, response) return Response( {"recording_url": storage_service.call_recording_public_url( request, relative_path)} ) ## ========================= WEB HOOKS =========================== @action(detail=False, methods=['post'], url_path='record-offtime-message', permission_classes=[AllowAny], authentication_classes=[] ) def record_offtime_message(self, request): """Handle incoming webhook requests.""" webhook_data = request.data call_data = webhook_data.get('call', {}) customer_phone = call_data.get('from_number') company_phone = call_data.get('to_number') twilio_call_sid = call_data.get( 'telephony_identifier', {} ).get('twilio_call_sid') logger.info("=============================== Record Off-time Message request.data starts ===================================") logger.info(f"Webhook Data: {webhook_data}") args = webhook_data.get('args', {}) advisor_id = args.get('advisor_id') advisor_name = args.get('advisor_name') customer_name = args.get('customer_name') customer_note = args.get('customer_note') logger.info(f"type of advisor_id: {type(advisor_id)}") logger.info(f"advisor_id: {advisor_id}") logger.info(f"advisor_name: {advisor_name}") logger.info(f"customer_name: {customer_name}") logger.info(f"customer_note: {customer_note}") logger.info("=============================== Getting Company Bot Settings ===================================") company_bot_settings = CompanyBotSettings.objects.filter( phone_number=company_phone ).first() company = company_bot_settings.company if company_bot_settings else None if not company: return Response({ "detail": "Company not found" }, status=status.HTTP_404_NOT_FOUND) user = company.users.filter(id=advisor_id, is_active=True).first() if not user: return Response({ "detail": "User not found" }, status=status.HTTP_404_NOT_FOUND) logger.info("=============================== Creating User Message ===================================") MessageRepository.create_message( recipient=user, company=company, customer_name=customer_name, customer_number=customer_phone, subject=f"Off-time Message from {customer_name}", body=customer_note, twilio_call_sid=twilio_call_sid, ) logger.info("=============================== User Message created successfully ===================================") logger.info("=============================== Recording Off-time Message ===================================") return Response( {"message": "Offtime message recorded successfully"}, status=status.HTTP_200_OK ) @action( detail=False, methods=['post'], url_path='send-appointment-email', permission_classes=[AllowAny], authentication_classes=[] ) def send_appointment_email(self, request): """Handle incoming webhook requests.""" event = WebhookEventService.log_event( request, "retell", "Send Appointment Email", ) webhook_data = request.data call_data = webhook_data.get('call', {}) # customer_phone = call_data.get('from_number') company_phone = call_data.get('to_number') # logger.info("=============================== Send Appointment Email request.data starts ===================================") # logger.info(f"Webhook Data: {webhook_data}") # args = webhook_data.get('args', {}) # booking_datetime = args.get('booking_datetime') # customer_name = args.get('customer_name') # service = args.get('service') # logger.info(f"booking_datetime: {booking_datetime}") # logger.info(f"customer_name: {customer_name}") # logger.info(f"service: {service}") logger.info("=============================== Getting Company Bot Settings ===================================") company_bot_settings = CompanyBotSettings.objects.filter( phone_number=company_phone ).first() company = company_bot_settings.company if company_bot_settings else None if not company: if event: event.mark_failed("Company not found") return Response({ "detail": "Company not found" }, status=status.HTTP_404_NOT_FOUND) logger.info("=============================== Getting BDC number ===================================") ## Get bdc number from company bdc_number = company.bdc_number if not bdc_number: if event: event.mark_failed("BDC number not found") return Response({ "detail": "BDC number not found" }, status=status.HTTP_404_NOT_FOUND) # logger.info("=============================== Sending Email and SMS ===================================") # EmailService.send_booking_email( # bdc_number, # customer_name, # booking_datetime, # service, # customer_phone # ) # SmsService.send_booking_sms( # bdc_number, # customer_name, # booking_datetime, # service, # customer_phone # ) # logger.info("=============================== Email and SMS sent successfully ===================================") try: serializer = BookAppointmentInputSerializer(data=request.data) serializer.is_valid(raise_exception=True) AppointmentService.book_service_appointment(serializer.validated_data) if event: event.mark_success() except Exception as exc: if event: event.mark_failed(f"Failed in creation of Appointment Object {exc}") logger.warning(f"Failed in creation of Appointment Object {exc}") pass return Response( {"message": "Email and SMS sent successfully"}, status=status.HTTP_200_OK ) @action(detail=False, methods=['post'], url_path='log-call-webhook', permission_classes=[AllowAny], authentication_classes=[] ) def log_call_webhook(self, request): """Handle incoming webhook requests.""" webhook_data = request.data event = webhook_data.get('event') if event == 'call_analyzed': call_data = webhook_data.get('call', {}) customer_phone = call_data.get('from_number') company_phone = call_data.get('to_number') is_call_transferred = (call_data.get('disconnection_reason') == 'call_transfer') logger.info("=-=-=-=-=-=-=-=Call Data=-=-=-=-=-=-=-=") logger.info(f"Customer Phone: {customer_phone}") logger.info(f"Company Phone: {company_phone}") logger.info(f"Is Call Transferred: {is_call_transferred}") logger.info(f"Call Data: {call_data}") logger.info("=-=-=-=-=-=-=-=Call Data=-=-=-=-=-=-=-=") company_bot_settings = CompanyBotSettings.objects.filter( phone_number=company_phone ).first() company = ( company_bot_settings.company if company_bot_settings else None ) bot_type = None if company_bot_settings.bot_name == BotName.hazel: bot_type = BotType.SALES_BOT elif company_bot_settings.bot_name == BotName.maya: bot_type = BotType.SERVICE_BOT elif company_bot_settings.bot_name == BotName.rebecca: bot_type = BotType.DEALERSHIP_GROUP_BOT transfer_destination_number = None transcript_with_tool_calls = call_data.get( 'transcript_with_tool_calls', [] ) for item in transcript_with_tool_calls: if ( item.get('role') == 'tool_call_invocation' and 'transfer_call' in (item.get('name') or '').lower() ): arguments = item.get('arguments', '{}') args_data = json.loads(arguments) transfer_destination_number = args_data.get('number') if transfer_destination_number is None: transfer_destination_number = args_data.get('transfer_destination_number') logger.info(f"Transfer Destination Number 1: " f"{args_data.get('number')}") logger.info(f"Transfer Destination Number: " f"{transfer_destination_number}") user = None if transfer_destination_number: user_profile = UserProfile.objects.filter( phone_number=transfer_destination_number ).first() user = user_profile.user if user_profile else None dealership = None if bot_type == BotType.DEALERSHIP_GROUP_BOT: if transfer_destination_number: dealership = Company.objects.filter( contact_phone=transfer_destination_number, company_type=CompanyType.DEALERSHIP, ).first() transfer_status = TRANSFER_STATUS.NOT_TRANSFERRED.value if transfer_destination_number and is_call_transferred: transfer_status = TRANSFER_STATUS.SUCCESSFUL.value elif transfer_destination_number and not is_call_transferred: transfer_status = TRANSFER_STATUS.FAILED.value sentiment_str = call_data.get( 'call_analysis', {} ).get('user_sentiment', '') if isinstance(sentiment_str, tuple) and sentiment_str: sentiment_str = sentiment_str[0] sentiment_str = str(sentiment_str).strip("'") sentiment = SENTIMENT.NEGATIVE.value if sentiment_str.lower() == 'positive': sentiment = SENTIMENT.POSITIVE.value elif sentiment_str.lower() == 'neutral': sentiment = SENTIMENT.NEUTRAL.value twilio_call_sid = call_data.get( 'telephony_identifier', {} ).get('twilio_call_sid') call_analysis = call_data.get("call_analysis", {}) custom_analysis = call_analysis.get("custom_analysis_data", {}) customer_name = custom_analysis.get("customer_name") call_time_category = custom_analysis.get("call_time_category") clean_customer_name = ( customer_name.strip() if isinstance(customer_name, str) and customer_name.strip() else None ) logger.info(f"Customer Name: {customer_name}") logger.info(f"Clean Customer Phone: {clean_customer_name}") call, call_created = Call.objects.update_or_create( call_id=call_data.get('call_id'), defaults={ 'from_number': customer_phone, 'to_number': company_phone, 'caller_name': clean_customer_name, 'company': company, 'dealership': dealership, 'transfer_number': transfer_destination_number, 'call_time_category': call_time_category, 'transfer_user': user, 'transfer_status': transfer_status, 'bot_type': bot_type, 'status': call_data.get('call_status'), 'transcript': call_data.get('transcript'), 'sentiment': sentiment, 'summary': call_data.get( 'call_analysis', {} ).get('call_summary'), 'duration': call_data.get( 'call_cost', {} ).get('total_duration_seconds') or 0, 'cost': call_data.get( 'call_cost', {} ).get('combined_cost') or 0, 'twilio_call_sid': twilio_call_sid, } ) process_catch_phrases_for_call.delay(call.id) appointment = Appointment.objects.filter( twilio_call_sid=twilio_call_sid ).first() if appointment: appointment.call = call appointment.save() customer_defaults = { 'last_call_at': timezone.now(), 'last_call_id': call.id, } if clean_customer_name: customer_defaults['name'] = cast(Any, customer_name.strip()) customer, created = Customer.objects.get_or_create( phone=customer_phone, company=company, defaults=customer_defaults ) if created: customer.name_extracted = get_caller_name_by_phone_number( customer_phone ) customer.save(update_fields=["name_extracted"]) if call_created and transfer_destination_number is not None: process_booking_intent.apply_async( kwargs={"call_id": call.id, "use_speaker": True}, countdown=30, ) if transfer_status == TRANSFER_STATUS.FAILED.value: # Sends notifications to advisor that he have missed the call from customer name along with phone number # Sends notifications to all the chief admins and service managers of that company NotificationService.send_missed_call_notification( call=call, user=user, company=company, customer_name=clean_customer_name, customer_phone=customer_phone ) call_action = "created" if call_created else "updated" logger.info(f"Call {call_created} " f"successfully: {call.call_id}") return Response({ 'message': f'Call {call_action} successfully', 'call_id': call.call_id, 'action': call_action }, status=status.HTTP_200_OK) else: logger.info(f"Call webhook for {event} " f"received successfully") return Response({ 'message': f'Call webhook for {event} ' f'received successfully', 'event': event, 'data': webhook_data }, status=status.HTTP_200_OK) @action( methods=['post'], detail=False, url_path='inbound-dynamic-variables-webhook', permission_classes=[AllowAny], authentication_classes=[] ) def inbound_dynamic_variables_webhook(self, request): """Handle incoming webhook requests.""" to_number = None if (request.data.get('event') == 'call_inbound' and 'call_inbound' in request.data): to_number = request.data['call_inbound'].get('to_number') logger.info(f"Extracted company_number: {to_number}") bot_name = get_bot_name(to_number) if not bot_name: return Response( {"detail": "This bot not configured in our system."}, status=status.HTTP_400_BAD_REQUEST ) override_agent_id = None dynamic_variables = None if bot_name.lower() == BotName.hazel: sales_info = check_sales_time_by_phone(to_number) is_holiday = sales_info['is_holiday'] is_office_hours = sales_info['is_office_hours'] dynamic_variables = {k: str(v) for k, v in sales_info.items()} if is_holiday: override_agent_id = settings.LYRA_AGENT_ID # Lyra elif is_office_hours is False: override_agent_id = settings.OLIVIA_AGENT_ID # Olivia elif bot_name.lower() == BotName.maya: company_info = get_company_working_info(to_number) is_holiday = company_info['is_holiday'] is_office_hours = company_info['is_office_hours'] if is_holiday: override_agent_id = settings.LYRA_AGENT_ID # Lyra elif is_office_hours is False: override_agent_id = settings.OLIVIA_AGENT_ID # Olivia dynamic_variables = {k: str(v) for k, v in company_info.items()} elif bot_name.lower() == BotName.rebecca: rabeeca_info = get_rabeeca_working_info(to_number) dynamic_variables = {k: str(v) for k, v in rabeeca_info.items()} if dynamic_variables is not None: call_inbound_payload = { "dynamic_variables": dynamic_variables } if override_agent_id is not None: call_inbound_payload["override_agent_id"] = cast(Any, override_agent_id) return Response({ "call_inbound": call_inbound_payload }, status=status.HTTP_200_OK) return Response( {"detail": "This bot not configured in our system."}, status=status.HTTP_400_BAD_REQUEST ) @action(detail=False, methods=['post'], url_path='log-transfer-call-webhook', permission_classes=[AllowAny], authentication_classes=[] ) def log_transfer_call_webhook(self, request): """Handle incoming webhook requests.""" logger.info(f"Log Transfer Call Webhook received - Data: {request.data}") logger.info(f"Log Transfer Call Webhook - Headers: {dict(request.headers)}") logger.info(f"Log Transfer Call Webhook - Method: {request.method}") return Response({'message': 'Webhook received successfully'}, status=status.HTTP_200_OK)