""" This file used for file uploaded """ import datetime # python imports import logging import mimetypes import os import boto3 from django.core.files.storage import FileSystemStorage # django imports from django.utils.crypto import get_random_string from rest_framework.status import HTTP_400_BAD_REQUEST, HTTP_200_OK from base import constants from base.constants import NUMBER # local import from zod_bank.settings import base_settings as settings from zod_bank.settings.base_settings import BASE_DIR def image_upload(folder, file_name, data): """ Function to upload files :param folder:folder location string :param file_name:file_name without ext string :param data:data file obj :return:Dictionary """ status = HTTP_400_BAD_REQUEST img_name = None error = None try: s3_client = boto3.client('s3', aws_access_key_id=settings.AWS_ACCESS_KEY, aws_secret_access_key=settings.AWS_SECRET_ACCESS_KEY, region_name=settings.AWS_DEFAULT_REGION ) bucket_name = settings.BUCKET_NAME MEDIA_ROOT = os.path.join(BASE_DIR, 'media/tmp') fss = FileSystemStorage() file = fss.save('tmp/' + str(file_name), data) fss.url(file) tmp_file = os.path.join(MEDIA_ROOT, str(file_name)) s3_client.upload_file( tmp_file, bucket_name, folder + str(file_name), ExtraArgs={'ACL': 'public-read', 'ContentType': data.content_type} ) os.unlink(tmp_file) img_name = file_name status = HTTP_200_OK except Exception as e: error = e logging.error(e) return status, error, img_name def file_delete(folder, file_name): """ To delete common file :param folder: folder name str :param file_name: file_name string type """ status = HTTP_400_BAD_REQUEST error = None try: s3_client = boto3.client('s3', aws_access_key_id=settings.AWS_ACCESS_KEY, aws_secret_access_key=settings.AWS_SECRET_ACCESS_KEY, region_name=settings.AWS_DEFAULT_REGION ) s3_client.delete_object(Bucket=settings.BUCKET_NAME, Key=str(folder) + str(file_name)) status = HTTP_200_OK except Exception as e: error = e return status, error def get_aws_obj(folder, file_name): """ To get aws file obj :param folder: folder string type :param file_name: file_name string type """ status = HTTP_400_BAD_REQUEST obj = None try: s3_client = boto3.client('s3', aws_access_key_id=settings.AWS_ACCESS_KEY, aws_secret_access_key=settings.AWS_SECRET_ACCESS_KEY, region_name=settings.AWS_DEFAULT_REGION ) file_name = folder + file_name obj = s3_client.get_object(Bucket=settings.BUCKET_NAME, Key=file_name) status = HTTP_200_OK except Exception as e: logging.error(e) return status, obj def upload_image(post_data, folder): """ :param post_data: :param folder: string type :return: """ upload_obj = None # Check Post data if post_data: date_now = datetime.datetime.now() file_extension = os.path.splitext(str(post_data.name)) file_extension = file_extension[constants.NUMBER['one']].split(".")[constants.NUMBER['one']].lower() rand = get_random_string(NUMBER['twelve']) image_name = str(rand) + date_now.strftime("%s") + "." + file_extension upload_obj = image_upload(folder, image_name, post_data) return upload_obj def upload_voice_kit_image(post_data, folder, image_dir): """ :param post_data: :param folder: string type :param image_dir: image_dir :return: """ upload_obj = None # Check Post data if post_data: date_now = datetime.datetime.now() file_extension = os.path.splitext(str(post_data)) file_extension = file_extension[constants.NUMBER['one']].split(".")[constants.NUMBER['one']].lower() rand = get_random_string(NUMBER['twelve']) image_name = str(rand) + date_now.strftime("%s") + "." + file_extension upload_obj = voice_kit_image_upload(folder, image_name, post_data, image_dir) return upload_obj def voice_kit_image_upload(folder, file_name, data, image_dir): """ Function to upload files :param folder:folder location string :param file_name:file_name without ext string :param data:data file obj :return:Dictionary """ status = HTTP_400_BAD_REQUEST img_name = None error = None try: s3_client = boto3.client('s3', aws_access_key_id=settings.AWS_ACCESS_KEY, aws_secret_access_key=settings.AWS_SECRET_ACCESS_KEY, region_name=settings.AWS_DEFAULT_REGION ) bucket_name = settings.BUCKET_NAME MEDIA_ROOT = os.path.join(BASE_DIR, 'media/tmp') fss = FileSystemStorage() with open(image_dir+data, 'rb') as f: file = fss.save('tmp/' + str(file_name), f) fss.url(file) tmp_file = os.path.join(MEDIA_ROOT, str(file_name)) s3_client.upload_file( tmp_file, bucket_name, folder + str(file_name), ExtraArgs={'ACL': 'public-read', 'ContentType': mimetypes.guess_type(file_name)[0]} ) os.unlink(tmp_file) img_name = file_name status = HTTP_200_OK except Exception as e: error = e logging.error(e) return status, error, img_name