# By: Riasat Ullah # This file processes live call routing related requests. from constants import api_paths, component_names as cnm, label_names as lnm, pages, var_names from context_manager import configurations_context from django.core import exceptions from django.http import JsonResponse from django.shortcuts import redirect, render from django.views.decorators.http import require_http_methods from taskcallweb import settings from system_tests.test_data import test_data_live_call_routing from translators import label_translator as lt from utils import client_data_path, helpers, logging, s3 from validations import request_validator import configuration as configs import botocore.exceptions import filetype import json @require_http_methods(['GET', 'POST']) def get_live_call_routing_list(request): ''' GET: Returns the live call routing list page. POST: Gets the list of live call routing. :param request: Http request :return: GET: Http response | POST: JSON response ''' if request.method == 'GET': if request_validator.user_in_session(request): lang = request_validator.get_user_language(request) nav_bar_components = request_validator.get_nav_bar_components(request) has_view_perm, has_edit_perm = request_validator.get_session_permission( request, cnm.dis_com_live_call_routing, nav_bar_components ) if not has_view_perm: raise exceptions.PermissionDenied context = configurations_context.get_live_call_routing_list_context(lang, nav_bar_components) context[var_names.has_edit_permission] = has_edit_perm return render(request, pages.live_call_routing_list_page, context=context) else: helpers.set_session_redirect_page(request) return redirect(pages.login_url) elif request.method == 'POST': lang = request_validator.get_user_language(request) if request_validator.user_in_session(request): if settings.TEST_MODE: return JsonResponse(test_data_live_call_routing.call_routing_list, safe=False) else: body = dict() status, output = helpers.post_api_request(api_paths.live_call_routing_list, body, request, lang=lang) return JsonResponse(output, status=status, safe=False) else: return JsonResponse(lt.get_label(lnm.err_unauthorized_access, lang), status=401, safe=False) @require_http_methods(['GET', 'POST']) def get_live_call_routing_details(request, call_routing_ref_id=None): ''' GET: Returns the live call routing details page. POST: Gets the details of a live call routing. :param request: Http request :param call_routing_ref_id: the reference ID of the live call routing :return: JSON response ''' if request.method == 'GET': if request_validator.user_in_session(request): try: lang = request_validator.get_user_language(request) nav_bar_components = request_validator.get_nav_bar_components(request) component_features = request_validator.get_component_features(request) has_view_perm, has_edit_perm = request_validator.get_session_permission( request, cnm.dis_com_live_call_routing, nav_bar_components ) if not has_view_perm: raise exceptions.PermissionDenied context = configurations_context.get_live_call_routing_details_context(lang, nav_bar_components) context[var_names.context] = json.dumps({var_names.call_routing_ref_id: call_routing_ref_id}) context[var_names.has_edit_permission] = has_edit_perm context[var_names.has_call_back_permission] = \ True if cnm.feat_incidents_call_back in component_features else False return render(request, pages.live_call_routing_details_page, context=context) except ValueError as e: logging.exception(str(e)) return render(request, pages.live_call_routing_details_page) else: helpers.set_session_redirect_page(request) return redirect(pages.login_url) elif request.method == 'POST': lang = request_validator.get_user_language(request) if request_validator.user_in_session(request): if settings.TEST_MODE: return JsonResponse(test_data_live_call_routing.call_routing_details[call_routing_ref_id], safe=False) else: body = {var_names.call_routing_ref_id: call_routing_ref_id} status, output = helpers.post_api_request(api_paths.live_call_routing_details, body, request, lang=lang) return JsonResponse(output, status=status, safe=False) else: return JsonResponse(lt.get_label(lnm.err_unauthorized_access, lang), status=401, safe=False) @require_http_methods(['POST']) def create_live_call_routing(request): ''' Creates a live call routing specification. :param request: Http request :return: JSON response ''' if request.method == 'POST': lang = request_validator.get_user_language(request) if request_validator.user_in_session(request): body = json.loads(request.POST[var_names.data]) if settings.TEST_MODE: status, output = 200, 'new_ref_id' else: status, output = helpers.post_api_request(api_paths.live_call_routing_create, body, request, lang=lang) ################################### # Handle s3 audio file uploads here ################################### grt_name = body[var_names.greeting_audio_filename] end_name = body[var_names.ending_audio_filename] if status == 200 and (grt_name is not None or end_name is not None): # Prepare the new payload that will be sent to the rest server to update the audio location new_payload = {var_names.call_routing_ref_id: output} # Upload the greeting audio to s3 if var_names.greeting_audio in request.FILES: grt_bucket, grt_fldr, grt_key, grt_url = client_data_path.get_greeting_audio_path(output, grt_name) try: logging.info('Uploading greeting audio file - ' + grt_key) s3.delete_folder_files(grt_bucket, grt_fldr) s3.upload_media_file(grt_bucket, grt_key, request.FILES[var_names.greeting_audio]) # add the variables to the payload new_payload[var_names.greeting_audio_filename] = grt_name new_payload[var_names.greeting_audio_location] = grt_key new_payload[var_names.greeting_audio_url] = grt_url except (botocore.exceptions.ClientError, botocore.exceptions.ParamValidationError) as e: logging.exception(str(e)) logging.exception(lnm.msg_live_call_route_created_with_audio_issues) return JsonResponse(lt.get_label(lnm.msg_live_call_route_created_with_audio_issues, lang), status=207, safe=False) # Upload the ending audio to s3 if var_names.ending_audio in request.FILES: end_bucket, end_fldr, end_key, end_url = client_data_path.get_ending_audio_path(output, end_name) try: logging.info('Uploading ending audio file - ' + end_key) s3.delete_folder_files(end_bucket, end_fldr) s3.upload_media_file(end_bucket, end_key, request.FILES[var_names.ending_audio]) # add the variables to the payload new_payload[var_names.ending_audio_filename] = end_name new_payload[var_names.ending_audio_location] = end_key new_payload[var_names.ending_audio_url] = end_url except (botocore.exceptions.ClientError, botocore.exceptions.ParamValidationError) as e: logging.exception(str(e)) logging.exception(lnm.msg_live_call_route_created_with_audio_issues) return JsonResponse(lt.get_label(lnm.msg_live_call_route_created_with_audio_issues, lang), status=207, safe=False) # Update the audio file location information in the backend if len(new_payload) > 1 and not settings.TEST_MODE: helpers.post_api_request(api_paths.live_call_routing_add_audio_info, new_payload, request, lang=lang) return JsonResponse(output, status=status, safe=False) else: return JsonResponse(lt.get_label(lnm.err_unauthorized_access, lang), status=401, safe=False) @require_http_methods(['POST']) def edit_live_call_routing(request): ''' Edits a live call routing specification. :param request: Http request :return: JSON response ''' if request.method == 'POST': lang = request_validator.get_user_language(request) if request_validator.user_in_session(request): body = json.loads(request.POST[var_names.data]) if settings.TEST_MODE: status, output = 200, 'Success' else: status, output = helpers.post_api_request(api_paths.live_call_routing_edit, body, request, lang=lang) ################################### # Handle s3 audio file uploads here ################################### ref_id = body[var_names.call_routing_ref_id] grt_name = body[var_names.greeting_audio_filename] has_grt_changed = body[var_names.has_greeting_audio_changed] end_name = body[var_names.ending_audio_filename] has_end_changed = body[var_names.has_ending_audio_changed] if status == 200 and ((grt_name is not None and has_grt_changed) or (end_name is not None and has_end_changed)): # Prepare the new payload that will be sent to the rest server to update the audio location new_payload = {var_names.call_routing_ref_id: body[var_names.call_routing_ref_id]} # Upload the greeting audio to s3 if var_names.greeting_audio in request.FILES and has_grt_changed: grt_bucket, grt_fldr, grt_key, grt_url = client_data_path.get_greeting_audio_path(ref_id, grt_name) try: logging.info('Uploading greeting audio file - ' + grt_key) s3.delete_folder_files(grt_bucket, grt_fldr) s3.upload_media_file(grt_bucket, grt_key, request.FILES[var_names.greeting_audio]) # add the variables to the payload new_payload[var_names.greeting_audio_filename] = grt_name new_payload[var_names.greeting_audio_location] = grt_key new_payload[var_names.greeting_audio_url] = grt_url except (botocore.exceptions.ClientError, botocore.exceptions.ParamValidationError) as e: logging.exception(str(e)) logging.exception(lnm.msg_live_call_route_created_with_audio_issues) return JsonResponse(lt.get_label(lnm.msg_live_call_route_created_with_audio_issues, lang), status=207, safe=False) # Upload the ending audio to s3 if var_names.ending_audio in request.FILES and has_end_changed: end_bucket, end_fldr, end_key, end_url = client_data_path.get_ending_audio_path(ref_id, end_name) try: logging.info('Uploading ending audio file - ' + end_key) s3.delete_folder_files(end_bucket, end_fldr) s3.upload_media_file(end_bucket, end_key, request.FILES[var_names.ending_audio]) # add the variables to the payload new_payload[var_names.ending_audio_filename] = end_name new_payload[var_names.ending_audio_location] = end_key new_payload[var_names.ending_audio_url] = end_url except (botocore.exceptions.ClientError, botocore.exceptions.ParamValidationError) as e: logging.exception(str(e)) logging.exception(lnm.msg_live_call_route_created_with_audio_issues) return JsonResponse(lt.get_label(lnm.msg_live_call_route_created_with_audio_issues, lang), status=207, safe=False) # Update the audio file location information in the backend if len(new_payload) > 1 and not settings.TEST_MODE: helpers.post_api_request(api_paths.live_call_routing_add_audio_info, new_payload, request, lang=lang) return JsonResponse(output, status=status, safe=False) else: return JsonResponse(lt.get_label(lnm.err_unauthorized_access, lang), status=401, safe=False) @require_http_methods(['POST']) def delete_live_call_routing(request): ''' Deletes a live call routing specification. :param request: Http request :return: JSON response ''' if request.method == 'POST': lang = request_validator.get_user_language(request) if request_validator.user_in_session(request): body = json.loads(request.body.decode()) if settings.TEST_MODE: status, output = 200, {var_names.greeting_audio_location: None, var_names.ending_audio_location: None} else: status, output = helpers.post_api_request(api_paths.live_call_routing_delete, body, request, lang=lang) if status == 200: greet_loc = output[var_names.greeting_audio_location] end_loc = output[var_names.ending_audio_location] try: if greet_loc is not None: s3.delete_single_file(client_data_path.get_client_data_s3_bucket(), greet_loc) if end_loc is not None: s3.delete_single_file(client_data_path.get_client_data_s3_bucket(), end_loc) except (botocore.exceptions.ClientError, botocore.exceptions.ParamValidationError) as e: logging.exception(str(e)) logging.exception('Failed to delete live call routing audio files - greeting: ' + str(greet_loc) + ' | ending: ' + str(end_loc)) finally: output = lt.get_label(lnm.msg_live_call_route_deleted, lang) return JsonResponse(output, status=status, safe=False) else: return JsonResponse(lt.get_label(lnm.err_unauthorized_access, lang), status=401, safe=False) @require_http_methods(['POST']) def get_available_phone_numbers(request): ''' Get the list of available phone numbers. :param request: Http request :return: JSON response ''' if request.method == 'POST': lang = request_validator.get_user_language(request) if request_validator.user_in_session(request): body = json.loads(request.body.decode()) if settings.TEST_MODE: return JsonResponse(test_data_live_call_routing.searched_phone_numbers, safe=False) else: status, output = helpers.post_api_request(api_paths.live_call_routing_twilio_available_numbers, body, request, lang=lang) return JsonResponse(output, status=status, safe=False) else: return JsonResponse(lt.get_label(lnm.err_unauthorized_access, lang), status=401, safe=False) @require_http_methods(['POST']) def validate_live_call_routing_audio_files(request): ''' Validates that the live call routing audio files are in the correct format. :param request: Http request :return: JSON response ''' if request.method == 'POST': lang = request_validator.get_user_language(request) if request_validator.user_in_session(request): audio_files = request.FILES.getlist('files[]') for item in audio_files: file = filetype.guess(item) if file is not None and file.extension not in configs.LIVE_CALL_ROUTING_AUDIO_EXTENSIONS and\ file.mime not in configs.LIVE_CALL_ROUTING_AUDIO_MIME_TYPES: return JsonResponse(lt.get_label(lnm.err_live_call_route_audio_format, lang), status=422, safe=False) return JsonResponse(lt.get_label(lnm.msg_success, lang), safe=False) else: return JsonResponse(lt.get_label(lnm.err_unauthorized_access, lang), status=401, safe=False)