ReceivePath.py

Go to the documentation of this file.

import os
import boto3
import sys
import traceback
from os.path import join, dirname
from dotenv import load_dotenv
from pathlib import Path
from ErrorHandler import *
 
# dotenv_path = Path("src/importer/.env")
load_dotenv()
 
def replace_file(bucket, path, file, s3):
    """Removes blank spaces in the filename by underscores and update information in S3
 
    Args:
        bucket (str): bucket where parquet file is storage
        path (str): s3 path where parquet file is storage
        file (str): current filename
        s3 (boto3 obj): AWS client connection
    Returns: new filename (str)
    """
    new_filename = file.replace(" ", "_")
    print("File Renamed: {}".format(new_filename))
    response = s3.list_objects_v2(Bucket= bucket, Prefix = path+"/"+file)
    source_key = response["Contents"][0]["Key"]
    copy_source = {'Bucket': bucket, 'Key': path+"/"+file}
    s3.copy_object(Bucket = bucket, CopySource = copy_source, Key = path+"/"+new_filename)
    s3.delete_object(Bucket = bucket, Key = source_key)
    return new_filename
 
def receive_path(event, s3_client):
    """Reviews if file exists in s3 and takes actions if it doesn't or file extension is not allowed
 
    Args:
        event (dict): is a dictionary with all client and sales information
        s3_client (boto3 obj): AWS client connection
    Returns: (dict)
    """
    bucket_in = event['bucket'][0]
    path_in = event["path"][0]
    bucket_out = event["bucket"][1]
    path_out = event["path"][1]
    response_in = s3_client.list_objects(Bucket = bucket_in, Prefix=path_in)
    response_out = s3_client.list_objects(Bucket = bucket_out, Prefix=path_out)
    # Verifies if input path exist
    if "Contents" not in response_in:
        raise Exception ("Path invalid.")
    #Create a new folder for client if it does not exist
    if "Contents" not in response_out:
        s3_client.put_object(Bucket=bucket_out, Key=(path_out+'/'))
    possible_files=[".csv", ".tsv", ".xls", ".xlsx", ".txt", ".xml"]
    status = "OK"
    for file in event["files"]:
        try:
            file_extension = os.path.splitext(file["file"])[1]
            if file_extension not in possible_files:
                print("FILE {} NOT SUPPORTED.".format(file["file"]))
                # no_processed[file_db_id]={"file":file["file"], "error": error}
                raise FileNotSupported()
            s3_client.head_object(Bucket=bucket_in, Key=path_in+"/"+file["file"])
            if " " in file["file"]:
                file["file"] = replace_file(bucket_in, path_in, file["file"], s3_client)
        except Exception as e:
            print("File {} will be discarded due {}".format(file["file"], e))
            m = "{}\n{}".format(sys.exc_info()[2], traceback.format_exc())
            file_db_id = file["file_id"]
            event["files"].remove(file)
            eh  = ErrorHandler()
            error = eh.handle(e, m, file_db_id)
            status = "Warning. Some files can not be found"
    event["status"] = status
    return event
    # return {
    #         "status": status,
    #         "tag": event["tag"],
    #         "cat_gen": event["cat_gen"],
    #         "cat_match": event["cat_match"],
    #         "bucket": event['bucket'],
    #         "path": event['path'],
    #         "client_id": event["client_id"],
    #         "files": event["files"],
    #         'currency': event['currency']
    #         }