|
|
|
|
@ -1,12 +1,11 @@ |
|
|
|
|
import os |
|
|
|
|
import pandas as pd |
|
|
|
|
from pandas import DataFrame |
|
|
|
|
from datetime import datetime as dt, timedelta |
|
|
|
|
from datetime import datetime as dt |
|
|
|
|
import json |
|
|
|
|
import re |
|
|
|
|
from pathlib import Path |
|
|
|
|
import time |
|
|
|
|
import numpy as np |
|
|
|
|
from glob import glob |
|
|
|
|
from logging import debug, DEBUG, basicConfig, warn |
|
|
|
|
|
|
|
|
|
# V3.1 | 01/19/23 |
|
|
|
|
@ -26,7 +25,7 @@ class ILReport: |
|
|
|
|
It makes it easier to add new reports to the workflow and to make it more clear where |
|
|
|
|
the reports are coming from. It also helps with tracking reports that may not be ready yet. |
|
|
|
|
""" |
|
|
|
|
def __init__(self, location: str, extraction_function: str, output_location: str = None): |
|
|
|
|
def __init__(self, location: str, extraction_function, output_location: str = None): |
|
|
|
|
debug(f"ILReport:\n\tLocation: {location}\n\tExtract Function: {extraction_function}\n\tOutput Location: {output_location}") |
|
|
|
|
# The location where the InfoLease report is stored |
|
|
|
|
self.location = location |
|
|
|
|
@ -55,17 +54,61 @@ class ILReport: |
|
|
|
|
if dataframe.empty: |
|
|
|
|
warn(f"ILReport: resulting dataframe was empty! Exiting with None.") |
|
|
|
|
return None |
|
|
|
|
self._append_to_consolidated_report(dataframe, settings["consolidatedBasePath"]) |
|
|
|
|
return dataframe |
|
|
|
|
|
|
|
|
|
def append_to_consolidated_report(self, output_dataframe: DataFrame): |
|
|
|
|
def _append_to_consolidated_report(self, dataframe_to_append: DataFrame, base_path: str): |
|
|
|
|
""" |
|
|
|
|
Add's the reports dataframe to the current months consolidated report or creates one if |
|
|
|
|
it already exists |
|
|
|
|
|
|
|
|
|
The month folder is typically 2 directories above the output location |
|
|
|
|
Year is the folder abover that^ |
|
|
|
|
""" |
|
|
|
|
# Decide the sheet name based on the save_location_name |
|
|
|
|
# We only add certain types to the consolidated report |
|
|
|
|
if re.search("(?i)ach", self.location) != None: |
|
|
|
|
sheet_name = "ACH" |
|
|
|
|
elif re.search("(?i)progpay_ber", self.location) != None: |
|
|
|
|
sheet_name = "CHECKS LIVE" |
|
|
|
|
elif re.search("(?i)vmcc", self.location) != None: |
|
|
|
|
sheet_name = "CREDIT CARDS" |
|
|
|
|
elif re.search("(?i)lockbox", self.location) != None: |
|
|
|
|
sheet_name = "LOCKBOX" |
|
|
|
|
elif re.search("(?i)epay", self.location) != None: |
|
|
|
|
sheet_name = "PAY BY PHONE" |
|
|
|
|
elif re.search("(?i)wires", self.location) != None: |
|
|
|
|
sheet_name = "WIRES" |
|
|
|
|
else: |
|
|
|
|
return None |
|
|
|
|
|
|
|
|
|
current_date: list(str) = dt.now().strftime("%Y.%m.%d").split('.') |
|
|
|
|
report_name = f"{dt.now().strftime('%B')}_ConsolidatedReport.xlsx" |
|
|
|
|
debug(f"Consolidated Reports {report_name} | {self.output_location} | {self.x_method} | {current_date}") |
|
|
|
|
year = current_date[0] |
|
|
|
|
month = current_date[1] |
|
|
|
|
year_month = f"{year}.{month}" |
|
|
|
|
|
|
|
|
|
save_path = f"{base_path}/{year}/{year_month}/{report_name}" |
|
|
|
|
# Check if the current month has a consolidated report |
|
|
|
|
month_summary_file: list(str) = glob(save_path) |
|
|
|
|
if len(month_summary_file) == 0: |
|
|
|
|
debug(f"Consolidated Report | No monthly summary file!\n\tCreating: {save_path}") |
|
|
|
|
# No file exists yet |
|
|
|
|
# Create it and add the current month |
|
|
|
|
with pd.ExcelWriter(save_path) as writer: |
|
|
|
|
debug(f"Consolidated Report | {sheet_name}: Saving data as: {report_name}") |
|
|
|
|
dataframe_to_append.to_excel(writer, index=False, sheet_name=sheet_name) |
|
|
|
|
else: |
|
|
|
|
# We need to read the dataframe in the current monthly report |
|
|
|
|
# Check that we are not adding matching data |
|
|
|
|
# Save the new report |
|
|
|
|
current_data: DataFrame = pd.read_excel(month_summary_file[0], sheet_name=sheet_name) |
|
|
|
|
new_data_len = len(dataframe_to_append) |
|
|
|
|
cur_first_col = current_data.iloc[len(current_data)-new_data_len:,0].to_list() |
|
|
|
|
new_first_col = dataframe_to_append.iloc[:,0].to_list() |
|
|
|
|
if cur_first_col == new_first_col: |
|
|
|
|
debug(f"Consolidated Report | Data is same as previous! Skipping!") |
|
|
|
|
return None |
|
|
|
|
# We need to find the start cols (where the new data should go) |
|
|
|
|
with pd.ExcelWriter(save_path, engine='openpyxl', mode='a',if_sheet_exists="overlay") as writer: |
|
|
|
|
debug(f"Consolidated Report | {sheet_name}: Saving data as: {report_name}") |
|
|
|
|
dataframe_to_append.to_excel(writer, index=False, sheet_name=sheet_name,startrow=len(current_data),header=False) |
|
|
|
|
|
|
|
|
|
|
|
|
|
|
def create_line_divider(breakage_list: list): |
|
|
|
|
|
Griffiths Lott
3 years ago