Wrangling and reading several xlsx files from sftp server and concatenating them into a single dataframe

Question

Please find below my code. Any recommendations/suggestions on how to improve it and make it more readable will be hugely appreciated. I've tried to comment it as much as possible so everyone can understand what my purpose was in every part of the code, but please feel free to ask me for more clarifications.

# Libraries
import pandas as pd
from datetime import datetime as dt, time as ti, timedelta as td
import io
import numpy as np
# Custom library with some auxiliary functions
import jtp_aux as aux

dict_attempts = {
    'Agency1': r'/path1/subpath/',
    'Agency2': r'/path2/subpath2/',
    'Agency3': r'/path3/subpath3/',
    'Agency4': r'/path4/subpath4/'
}


def fetch_attempts(attempts_):
   # Connect to sftp server via method defined in custom library
    sftp_ = aux.sftp_connect()

   # Initialise the dataframe which I will use to concatenate the files read from the sftp server
    out = []

   # Get the files between today at midnight and 1 week ago at midnight
    today_midnight = dt.combine(dt.today(), ti.min)
    last_week_midnight = today_midnight - td(days=7)
    for agency_, path_ in attempts_.items():
        for file in sftp_.listdir_attr(path_):
            mtime = dt.fromtimestamp(file.st_mtime)
            if (last_week_midnight <= mtime) and (mtime <= today_midnight):
                with sftp_.open(path_ + file.filename) as fp:
                    logger.info(path_ + file.filename)
                    bytes_p = fp.read()
                    file_obj = io.BytesIO(bytes_p)
                    fp_aux = pd.read_excel(
                        file_obj
                    )
                    fp_aux.dropna(axis=0, how='all', inplace=True)  # Delete the rows with all NaNs as they were causing issues
                    fp_aux.dropna(axis=1, how='all', inplace=True)  # Delete the columns with all NaNs as they were causing issues
                    # Need to insert the agency name and the execution date as this info does not appear in the files
                    fp_aux.insert(0, 'AGENCY', agency_)
                    fp_aux.insert(0, 'EXEC_DATE', today_midnight)

                    # Since the files have different column names (although same number of columns) need to standardise the 
            # column names to concatenate the dataframes.
                    fp_aux.set_axis(
                        ['EXEC_DATE', 'AGENCY', 'CALL_DATE', 'CALL_TIME', 'CALL_TYPE',
                         'CALL_DIRECTION', 'LIVE_FINAL', 'CACONT_ACC',
                         'PHONE_NUMBER', 'DESCRIPTION', 'CONTACTS',
                         'ATTEMPTS', 'RPC',
                         'AGENT', 'DOM_SME'
                         ], axis=1, inplace=True)

                    # If the file is valid, append it
                    out.append(fp_aux)
                    logger.info(path_)

                    # Note: not sure if I should handle here the errors in the files and pass to the following file without
                    # making the program crash.

    # Concatenate the obtained dataframes into a sigle dataframe
    df_out = pd.concat(out)

    # Need to replace the strings 'nan' with actual nulls for inserting the data into a DB
    df_out = df_out.replace({np.nan: ''})

    # The date is provided in string format in the xlsxs so I make sure it's converted to datetime
    df_out['CALL_DATE'] = pd.to_datetime(df_out['CALL_DATE'], dayfirst=True)

    # This commented part is one I use to debug the program. If the error happens after this line, I do not want to reexecute the whole
    # script but save the dataframe in a pickle file so I can take it from here.

    # df2pickle(df_out, r'output.pckl')
    # df_out = pickle2df(r'output.pckl')


    # I need to format this column to 12 digit string with leading 0 if needed. Example: from 1234 to '000000001234'
    df_out['CACONT_ACC'] = df_out['CACONT_ACC'].map('{:0>12}'.format).astype(str).\
        str.slice(0, 11)

    # These columns need to be numeric so I force them to be so.
    cols_number = ['CONTACTS', 'ATTEMPTS', 'RPC']
    df_out[cols_number] = df_out[cols_number].apply(pd.to_numeric, errors='coerce').\
        fillna(0, downcast='infer')

    # Convert these columns to string.
    cols_string = ['AGENCY', 'CALL_TIME', 'CALL_TYPE', 'CALL_DIRECTION',
                   'LIVE_FINAL', 'PHONE_NUMBER', 'DESCRIPTION', 'AGENT', 'DOM_SME']
    cols_trunc10 = ['CALL_TYPE', 'CALL_DIRECTION', 'LIVE_FINAL']
    cols_trunc50 = ['AGENCY', 'PHONE_NUMBER', 'AGENT', 'DOM_SME']
    df_out[cols_string] = df_out[cols_string].astype(str)

    # Truncate the string fields not to break the DB
    df_out[cols_trunc10] = df_out[cols_trunc10].astype(str).apply(lambda x: x.str[:10])
    df_out[cols_trunc50] = df_out[cols_trunc50].astype(str).apply(lambda x: x.str[:50])

    return df_out


# Logger
path_logger = r'log_attempts.log'
logger = aux.init_logger(path_logger)


def main():
    try:
        # Read Attempts from sftp server
        logger.info('START Read Attempts from sftp server')
        df_attempts = fetch_attempts(dict_attempts)
        logger.info('END Read Attempts from sftp server')
    
    except Exception as e:
        logger.error(e, exc_info=True)
        raise e


if __name__ == "__main__":
    main()

EDIT: sample data contained in an excel file in the sftp server.

Date        Time        Type    Direction   Live_Final  AccountID   Phone       Description Contact Attempt RPC Agent Name  DOM_SME
16/09/2021  08:29:13    Inbound Inbound     Final       55509110555 55538488555 Description 1   1   0   1   Agent1      DOM
16/09/2021  08:32:22    Inbound Inbound     Final       55591795555         Description 2   0   0   0   Agent2      
16/09/2021  08:33:10    Inbound Inbound     Final       55512508555 55591795555 Description 2   0   0   0   Agent2      
16/09/2021  08:35:28    Inbound Inbound     Final       55506159555 55532252555 Description 3   1   0   1   Agent2      DOM
16/09/2021  08:36:18    Inbound Inbound     Final       55512508555 55591795555 Description 3   1   0   1   Agent3      DOM
16/09/2021  08:37:12    Inbound Inbound     Final       55547003555 55525927555 Description 4   1   0   1   Agent3      DOM
16/09/2021  08:51:57    Inbound Inbound     Final       55574811555 55568501555 Description 2   0   0   0   Agent2      DOM

Answer 1

I don't think that these aliases:

from datetime import datetime as dt, time as ti, timedelta as td

are well-advised; just use the full type names.

Though I can't say for sure, this:

# Custom library with some auxiliary functions
import jtp_aux as aux

smells like a many-purpose utility library, and at the least it should be broken up to have better module names; at the most it may be wrappers that offer no additional value.

From the aux library, aux.sftp_connect() should return a context manager. If it does, fetch_attempts should wrap sftp_ in a with. If it doesn't, use a try ... finally: sftp_.close().

Delete comments like # Libraries that describe the obvious.

Why do variables like attempts_ have an underscore suffix? That should be removed.

These parens:

if (last_week_midnight <= mtime) and (mtime <= today_midnight):

are redundant and should be removed; also, convert to double-sided inequality form if last_week_midnight <= mtime <= today_midnight.

This:

logger.info(path_ + file.filename)

is not a good idea, since any % characters will not be escaped, and this is missing a descriptive string. Instead use e.g. logger.info('Remote path: %s%s', path_, file.filename).

This:

                bytes_p = fp.read()
                file_obj = io.BytesIO(bytes_p)
                fp_aux = pd.read_excel(
                    file_obj
                )

is inefficient and too complex. Instead, pass fp directly to read_excel.

About this:

                # Note: not sure if I should handle here the errors in the files and pass to the following file without
                # making the program crash.

that's a business logic decision you need to make; but overall, I would guess "yes". You're already using a good logger, so the approach could be that inside the innermost for loop, add a try; except for the appropriate exception type, and on the inside of the except, call logger.error('Path %s failed', path_, exc_info=True).

This:

df_out.replace({np.nan: ''})

should instead use fillna.

At the bottom, this:

    logger.error(e, exc_info=True)
    raise e

should replace the first argument with a less-redundant description like 'Program failed', and instead of raise e, just raise.