FileCleaning.py

Go to the documentation of this file.

import os
import sys
import json
import boto3
import traceback
from datetime import datetime
from dotenv import load_dotenv
f_path = __file__
index = f_path.find("upload_module/")
f_path = f_path[:index+len("upload_module/")]
sys.path.insert(1, f_path+"common")
from DownloadFiles import download_file_or_folder
from UploadFilesS3 import clean_files
from Tools import gen_file_list_drive, gen_file_list_event, send_notification
 
load_dotenv()
 
def get_path(local_path):
    """
    Adds the last folder downloaded to root path
 
    Args:
        local_path (str): Root path where is saved files
    Returns : The actual folder where files where downloaded
    """
    path_child = os.popen("ls {}".format(local_path)).read()
    local_path_full = local_path+path_child
    local_path_full = local_path_full.strip("\n")
    local_path_renamed = local_path_full.replace(" ", "_")
    os.rename(local_path_full, local_path_renamed)
    if os.path.isdir(local_path_renamed):
        return local_path_renamed+"/"
    return local_path
 
def download_local_file(event, local_path):
    """Downloads files uploaded directly via Otto Light interface in a temporal folder
 
    Args:
        event (dict): Contains the bucket and folder info where file was uploaded
        local_path (str): Root path where is saved files
    Returns : Nothing
    """
    # Access keys for AWS
    ACCESS_ID, ACCESS_KEY = os.environ.get("AWS_KEY_ID"), os.environ.get("AWS_KEY_SECRET")
    # Connection for AWS using boto3
    s3_client = boto3.client("s3", aws_access_key_id=ACCESS_ID, aws_secret_access_key=ACCESS_KEY)
    bucket = event["bucket"]
    s3_path = event["path"]
    file = event["file"]
    if type(file) == list and len(file) > 0:
        uploaded_files = local_path+"uploaded_files/"
        if not os.path.exists(uploaded_files):
            os.mkdir(uploaded_files)
        for i in file:
            s3_client.download_file(bucket, s3_path+"/"+i, uploaded_files+i)
    else:
        s3_client.download_file(bucket, s3_path+"/"+file, local_path+file)
 
 
 
def run_file_cleaning(event):
    """
    Transform Gdrive link to ID drive to be downloaded by Google API and execute all procedure
 
    Args:
        event (dict): Shared link of the folder/file to download
    Returns : The actual folder where files where downloaded
    """
    _ROOT_DIR = os.path.expanduser('~')
    path = _ROOT_DIR+"/GDrive_downloads/"
    os.makedirs(path, exist_ok=True)
    webhook_url = event["on_finish"]
    try:
        if event["drive_link"] != "":
            drive_link = event["drive_link"]
            if "folder" in drive_link:
            # Example: https://drive.google.com/drive/folders/1l6A3zjNGw6qWJQ-mkeep4W4ooCw7uy6W?usp=share_link
                drive_link = drive_link.split("folders/")[-1].split("?usp")[0]
            elif "file" in drive_link:
            # Example: https://drive.google.com/file/d/1o2q6RFa6wi7XMSJcZJhLtr-pNJxWQsVl/view?usp=share_link
                drive_link = drive_link.split("d/")[-1].split("/")[0]
            elif "spreadsheets" in drive_link:
                drive_link = drive_link.split("spreadsheets/d/")[-1].split("/")[0]
            download_file_or_folder(drive_link, path)
        else:  
            download_local_file(event, path)
        local_path = get_path(path)
        print(local_path)
        path_output, files = clean_files(event, local_path)
        os.system("rm -r {}*".format(path))
        print(f"Files uploaded to {path_output}")
        # if event["file"] == "" or event["file"] == []:
        #     files = gen_file_list_drive(path_output)
        # else:
        #     files = gen_file_list_event(event, path_output)
        # send_notification(event["module"], webhook_url, "Successful", files=files)
        return path_output, files
    except Exception as e:
        raise Exception("An error occurred: {}".format(e))
        # print("An error occurred: {}".format(e))
        # # print(sys.exc_info()[2])
        # # print(traceback.format_exc())
        # m = "{}\n{}".format(sys.exc_info()[2], traceback.format_exc())
        # send_notification(event["module"], webhook_url, "Failed", error=m)