zod-backend/base/upload_file.py

"""
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