data/cli_dropbox.py

import os
import pathlib
from typing import List
import dropbox
from dropbox.exceptions import AuthError
from dropbox import DropboxOAuth2FlowNoRedirect

import pandas as pd
import streamlit as st
from dotenv import load_dotenv

load_dotenv()
DROPBOX_ACCESS_TOKEN = os.environ.get("DROPBOX_ACCESS_TOKEN")
DROPBOX_REFRESH_TOKEN = os.environ.get("DROPBOX_REFRESH_TOKEN")
DROPBOX_APP_KEY = os.environ.get("DROPBOX_APP_KEY")
DROPBOX_APP_SECRET = os.environ.get("DROPBOX_APP_SECRET")
pd.set_option('display.max_columns', None)


def obtain_refresh_token(app_key, app_secret):
    # Create a NoRedirectFlow object
    auth_flow = DropboxOAuth2FlowNoRedirect(app_key, app_secret, token_access_type='offline')

    # Generate the authorization URL
    authorize_url = auth_flow.start()

    # Print the authorization URL and instruct the user to visit it
    print("1. Go to:", authorize_url)
    print("2. Click 'Allow' (you might have to log in first)")
    print("3. Copy the authorization code.")

    # Prompt the user to enter the authorization code
    auth_code = input("Enter the authorization code here: ").strip()

    # Exchange the authorization code for an access token and refresh token
    auth_result = auth_flow.finish(auth_code)
    access_token = auth_result.access_token
    refresh_token = auth_result.refresh_token
    print(access_token)
    print('\n\n')
    print(refresh_token)
    return access_token, refresh_token

def dropbox_connect():
    try:
        dbx = dropbox.Dropbox(oauth2_access_token=DROPBOX_ACCESS_TOKEN, 
                              oauth2_refresh_token=DROPBOX_REFRESH_TOKEN, 
                              app_key=DROPBOX_APP_KEY,
                              app_secret=DROPBOX_APP_SECRET)
    except AuthError as err:
        st.error(f'Erreur: Impossible de se connecter à la base de données - {err}')
    return dbx

def dropbox_list_files(path):
    """Return a Pandas dataframe of files in a given Dropbox folder path in the Apps directory.
    """

    dbx = dropbox_connect()

    try:
        files = dbx.files_list_folder(path).entries
        files_list = []
        for file in files:
            if isinstance(file, dropbox.files.FileMetadata):
                metadata = {
                    'name': file.name,
                    'path_display': file.path_display,
                    'client_modified': file.client_modified,
                    'server_modified': file.server_modified
                }
                files_list.append(metadata)

        df = pd.DataFrame.from_records(files_list)
        return df.sort_values(by='server_modified', ascending=False)

    except Exception as e:
        print('Error getting list of files from Dropbox: ' + str(e))

def dropbox_download_file(dropbox_file_path, local_file_path, load_data: bool = True):
    """Download a file from Dropbox to the local machine."""

    try:
        dbx = dropbox_connect()
        
        metadata = dbx.files_get_metadata(dropbox_file_path)
        if metadata:
            metadata, result = dbx.files_download(path=dropbox_file_path)
            if load_data:
                with open(local_file_path, 'wb') as f:
                    f.write(result.content)
            return result.content

        else:
            print("Fichier non trouvé")
            return None
    except Exception as e:
        if e.error.is_path() and e.error.get_path().is_not_found():
            print("Fichier non trouvé")
        return None

def dropbox_upload_file(local_path, local_filename, dropbox_data_path, dropbox_file_name: str = None):
    """Upload a file from the local machine to a path in the Dropbox app directory.

    Args:
        local_path (str): The path to the local file.
        local_file (str): The name of the local file.
        dropbox_file_path (str): The path to the file in the Dropbox app directory.

    Example:
        dropbox_upload_file('.', 'test.csv', '/stuff/test.csv')

    Returns:
        meta: The Dropbox file metadata.
    """
    try:
        dbx = dropbox_connect()

        local_file_path = os.path.join(local_path, local_filename)
        dropbox_file_name = local_filename if not dropbox_file_name else dropbox_file_name
        dropbox_file_path = os.path.join(dropbox_data_path, dropbox_file_name)
   
        with open(local_file_path,"rb") as f:
            if not dropbox_check_path_exists(dropbox_data_path):
                dropbox_create_folder(os.path.dirname(dropbox_data_path))
            
            meta = dbx.files_upload(f.read(), dropbox_file_path, mode=dropbox.files.WriteMode("overwrite"))
            print("File uploaded successfully!")
            return meta
    except Exception as e:
        print('Error uploading file to Dropbox: ' + str(e))

def dropbox_upload_bytefile(dropbox_data_path, dropbox_file_name: str, bytes):
    dropbox_file_path = os.path.join(dropbox_data_path, dropbox_file_name)

    try:
        dbx = dropbox_connect()
        if not dropbox_check_path_exists(dropbox_data_path):
            dropbox_create_folder(os.path.dirname(dropbox_data_path))
        meta = dbx.files_upload(bytes.getvalue(), dropbox_file_path, mode=dropbox.files.WriteMode("overwrite"))
        print("File uploaded successfully!")
        return meta
    except Exception as e:
         print('Error uploading file to Dropbox: ' + str(e))

def dropbox_create_folder(dropbox_folder_path):
    """Create a folder in the Dropbox app directory.

    Args:
        dropbox_folder_path (str): The path to the folder in the Dropbox app directory.

    Example:
        dropbox_create_folder('/stuff')

    Returns:
        meta: The Dropbox folder metadata.
    """

    try:
        dbx = dropbox_connect()

        dbx.files_create_folder_v2(dropbox_folder_path)
        print("Folder created successfully!")
    except dropbox.exceptions.ApiError as e:
        if e.error.is_path() and e.error.get_path().is_conflict():
            print("Folder already exists!")
        elif e.error.is_path() and e.error.get_path().is_not_found():
            # Split the folder path into individual components
            path_components = dropbox_folder_path.split("/")[1:]
            current_path = ""
            for component in path_components:
                current_path += "/" + component
                try:
                    dbx.files_create_folder_v2(current_path)
                    print(f"Created folder: {current_path}")
                except dropbox.exceptions.ApiError as e:
                    print(f"Error creating folder: {e}")
        else:
            print(f"Error creating folder: {e}")

def dropbox_check_path_exists(dropbox_folder_path):
    """Check if a folder exists in the Dropbox app directory.

    Args:
        dropbox_folder_path (str): The path to the folder in the Dropbox app directory.

    Example:
        dropbox_check_path_exists('/stuff')

    Returns:
        meta: The Dropbox folder metadata.
    """

    try:
        dbx = dropbox_connect()

        dbx.files_get_metadata(dropbox_folder_path)
        return True
    except dropbox.exceptions.ApiError as e:
        if e.error.is_path() and e.error.get_path().is_not_found():
            return False
        else:
            print(f"Error checking if folder exists: {e}", dropbox_folder_path)
            return False

@st.cache_data
def dropbox_load_config_files(dropbox_datapath: str, local_datapath: str, excel_sources: List[dict]):
    for key, value in excel_sources.items():
        dropbox_download_file(os.path.join(dropbox_datapath, excel_sources[key]['path']), os.path.join(local_datapath, excel_sources[key]['path']))

# obtain_refresh_token(DROPBOX_APP_KEY, DROPBOX_APP_SECRET)
# df = dropbox_list_files('/SEC_IND_GTP2023_OUTPUT')
# dropbox_create_folder('/SEC_IND_GTP2023_OUTPUT/TEST/FIT')
# dropbox_create_folder('/SEC_IND_GTP2023_OZUTPUT/TEST/FIT')
# print(df)