# By: Riasat Ullah # This file contains code for serving general info views. from constants import api_paths, label_names as lnm, var_names, pages, static_vars from context_manager import support_context from django.http import HttpResponse, JsonResponse from django.shortcuts import redirect, render from django.views.decorators.http import require_http_methods from pathlib import Path from taskcallweb import settings from translators import label_translator as lt from twilio.request_validator import RequestValidator from system_tests.test_data import test_data_external, test_data_incidents, test_data_groups, \ test_data_live_call_routing, test_data_organization, test_data_policies, test_data_services, \ test_data_status_dashboard, test_data_tags, test_data_teams, test_data_users, test_data_workflows from utils import client_data_path, file_storage, helpers, logging, s3 from validations import request_validator import json import os import re import requests @require_http_methods(['GET']) def get_component_list(request): ''' Get the list of components given the type of component(s). :param request: Http request :return: JSON response ''' if request.method == 'GET': lang = request_validator.get_user_language(request) if request_validator.user_in_session(request): list_type = request.GET.get('param') param_list = [x.replace('-', '_') for x in list_type.split(' ')] if settings.TEST_MODE: param_set = set(param_list) if param_set == {var_names.users}: return JsonResponse(test_data_users.users_list, safe=False) elif param_set == {var_names.routines}: return JsonResponse(test_data_policies.routines_basic_list, safe=False) elif param_set == {var_names.routines, var_names.users}: all_routines = test_data_policies.routines_basic_list + test_data_users.user_routines return JsonResponse(all_routines, safe=False) elif param_set == {var_names.policies}: return JsonResponse(test_data_policies.policies_basic_list, safe=False) elif param_set == {var_names.policies, var_names.users}: return JsonResponse(test_data_policies.policies_basic_list + test_data_users.users_list, safe=False) elif param_set == {var_names.services}: return JsonResponse(test_data_services.services_basic_list, safe=False) elif param_set == {var_names.business_services}: return JsonResponse(test_data_services.business_services_basic_list, safe=False) elif param_set == {var_names.response_sets}: return JsonResponse([], safe=False) elif param_set == {var_names.workflows}: return JsonResponse(test_data_workflows.incident_workflows_basic_list, safe=False) elif param_set == {var_names.teams}: return JsonResponse(test_data_teams.teams_basic_list, safe=False) elif param_set == {var_names.user_roles}: return JsonResponse(test_data_organization.allowed_user_roles, safe=False) elif param_set == {var_names.conference_bridges}: return JsonResponse(test_data_workflows.conference_bridges_basic_list, safe=False) elif param_set == {var_names.integrations}: return JsonResponse(test_data_services.integration_types, safe=False) elif param_set == {var_names.chat_integrations}: return JsonResponse(test_data_services.chat_integrations_basic_list, safe=False) elif param_set == {var_names.custom_action_integrations}: return JsonResponse(test_data_services.custom_action_integrations, safe=False) elif param_set == {var_names.recent_changes_integrations}: return JsonResponse(test_data_services.recent_changes_integration_types, safe=False) elif param_set == {var_names.incidents}: inc_basic = [[item[var_names.task_title], item[var_names.organization_instance_id], item[var_names.instance_id]] for item in test_data_incidents.incident_list] return JsonResponse(inc_basic, safe=False) elif param_set == {var_names.tags}: return JsonResponse([x[var_names.tag] for x in test_data_tags.tags_list], safe=False) elif param_set == {var_names.tags, var_names.policies}: return JsonResponse(test_data_tags.tags_list, safe=False) elif param_set == {var_names.tags, var_names.users}: return JsonResponse(test_data_tags.tags_list, safe=False) elif param_set == {var_names.live_call_routing}: return JsonResponse(test_data_live_call_routing.call_routing_basic_list, safe=False) elif param_set == {var_names.live_call_routing_countries}: return JsonResponse(test_data_live_call_routing.allowed_live_call_routing_countries, safe=False) elif param_set == {var_names.people}: return JsonResponse(test_data_groups.people_basic_list, safe=False) elif param_set == {var_names.groups}: return JsonResponse(test_data_groups.groups_basic_list, safe=False) elif param_set == {var_names.status_pages}: return JsonResponse(test_data_status_dashboard.status_pages_basic_list, safe=False) elif param_set == {var_names.external_sso}: return JsonResponse(test_data_external.external_sso_basic_list, safe=False) else: lang = request_validator.get_user_language(request) body = {var_names.data_type: param_list} status, output = helpers.post_api_request(api_paths.org_list_maker, 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']) def get_internal_image(request, file_name): ''' Gets an image that has been uploaded by TaskCall for internal purposes. :param request: Http request :param file_name: name of the file with file type extension :return: Http response ''' if request.method == 'GET': extension = file_name.split('.')[-1] file_type = extension if extension == 'jpg': file_type = 'jpeg' elif extension == 'svg': file_type = 'svg+xml' s3_path = '/'.join(['images', file_name]) img_bytes_file = s3.read_image(file_storage.S3_BUCKET_TASKCALL_PUBLIC_DATA, s3_path) return HttpResponse(img_bytes_file, content_type='image/' + file_type) @require_http_methods(['GET']) def get_js_file(request, file_name): ''' Serves js file from the static folder. :param request: Http request :param file_name: name of the file with file type extension :return: Http response ''' if request.method == 'GET': file_reader = open(os.path.join(str(Path(__file__).parents[1]), 'static', 'scripts', file_name), "r", encoding='utf-8') return HttpResponse(file_reader, content_type='text/javascript') @require_http_methods(['GET']) def get_css_file(request, file_name): ''' Serves a css file from the static folder. :param request: Http request :param file_name: name of the file with file type extension :return: Http response ''' if request.method == 'GET': file_reader = open(os.path.join(str(Path(__file__).parents[1]), 'static', 'css', file_name), "r", encoding='utf-8') return HttpResponse(file_reader, content_type='text/css') @require_http_methods(['GET']) def get_client_audio_file(request, folder_path): ''' Gets audio that has been uploaded by users. :param request: Http request :param folder_path: name of the file with file type extension :return: Http response ''' if request.method == 'GET': lang = request_validator.get_user_language(request) try: is_req_allowed = request_validator.user_in_session(request) if not is_req_allowed: twl_signature = request.headers.get(static_vars.x_twilio_signature) if twl_signature is not None: twl_acc_sid, twl_auth_token = helpers.get_twilio_authentication_details() twl_validator = RequestValidator(twl_auth_token) is_req_allowed = twl_validator.validate(request.build_absolute_uri(), request.GET, twl_signature) if is_req_allowed: file_bucket = file_storage.S3_BUCKET_TASKCALL_CLIENT_DATA[settings.REGION] s3_path = '/'.join([client_data_path.audio_folder, folder_path]) file_key = s3.get_first_file_key(file_bucket, s3_path) if file_key is not None: audio_bytes_file = s3.read_audio(file_bucket, file_key) content_type = 'audio/wav' if file_key.split('.')[-1] == 'wav' else 'audio/mpeg' return HttpResponse(audio_bytes_file, content_type=content_type) else: return HttpResponse(lnm.err_unknown_resource, status=404) else: return JsonResponse(lt.get_label(lnm.err_unauthorized_access, lang), status=401, safe=False) except OSError as e: logging.exception(str(e)) return JsonResponse(lt.get_label(lnm.err_system_error, lang), status=401, safe=False) @require_http_methods(['GET']) def get_client_image_file(request, folder_path): ''' Gets images that have been uploaded by users - ones that do not show the full file path with the filename. :param request: Http request :param folder_path: name of the file with file type extension :return: Http response ''' if request.method == 'GET': lang = request_validator.get_user_language(request) access_allowed = False if request_validator.user_in_session(request): access_allowed = True else: auth_check_body = {var_names.url: folder_path} header_params = { 'Accept': static_vars.content_type_json, 'Content-Type': static_vars.content_type_json, 'Accept-Language': lang, static_vars.authorization_attribute: request.headers.get(static_vars.authorization_attribute) } auth_check_response = requests.post(api_paths.user_check_file_access, headers=header_params, data=json.dumps(auth_check_body)) if auth_check_response.status_code == 200: access_allowed = True if access_allowed: extension, img_bytes_file = None, None if re.match(r"^[\w\-/]+\.(jpg|png|svg)$", folder_path): extension = folder_path.split('.')[-1] elif not re.match(r"^[\w\-/]+$", folder_path): return HttpResponse(lt.get_label(lnm.err_unknown_resource, lang), status=404) file_bucket = file_storage.S3_BUCKET_TASKCALL_CLIENT_DATA[settings.REGION] s3_path = '/'.join([client_data_path.images_folder, folder_path]) if extension is not None: img_bytes_file = s3.read_image(file_bucket, s3_path) else: file_key = s3.get_first_file_key(file_bucket, s3_path) if file_key is not None: img_bytes_file = s3.read_image(file_bucket, file_key) extension = file_key.split('.')[-1] if img_bytes_file is not None and extension is not None: file_type = extension if extension == 'jpg': file_type = 'jpeg' elif extension == 'svg': file_type = 'svg+xml' return HttpResponse(img_bytes_file, content_type=file_type) else: return HttpResponse(lt.get_label(lnm.err_unknown_resource, lang), status=404) else: return JsonResponse(lt.get_label(lnm.err_unauthorized_access, lang), status=401, safe=False) @require_http_methods(['GET']) def get_support_page(request): ''' Gets the support page. :param request: Http request :return: Http 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) context = support_context.get_support_context(lang, nav_bar_components) return render(request, pages.support_page, context=context) else: helpers.set_session_redirect_page(request) return redirect(pages.login_url) @require_http_methods(['POST']) def create_support_ticket(request): ''' Creates a new support ticket. :param request: HttpRequest :return: JsonResponse ''' 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('A new ticket has been created', safe=False) else: status, output = helpers.post_api_request( api_paths.support_ticket, 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)