Compare commits
5 Commits
5067678a8c
...
86afb277e5
| Author | SHA1 | Date |
|---|---|---|
=
|
86afb277e5
|
3 years ago |
|
=
|
9e2d960f7e
|
3 years ago |
|
=
|
40c2a8a0df
|
3 years ago |
|
=
|
5caaf3d7ac
|
3 years ago |
|
=
|
a3905d118e
|
3 years ago |
|
Before Width: | Height: | Size: 1.1 KiB |
|
Before Width: | Height: | Size: 2.6 KiB |
|
Before Width: | Height: | Size: 477 B |
|
Before Width: | Height: | Size: 6.9 KiB |
|
Before Width: | Height: | Size: 18 KiB |
|
Before Width: | Height: | Size: 819 B |
|
Before Width: | Height: | Size: 512 B |
|
Before Width: | Height: | Size: 568 B |
|
Before Width: | Height: | Size: 1.2 KiB |
|
Before Width: | Height: | Size: 3.1 KiB |
@ -0,0 +1 @@ |
||||
name = "Test Name" |
||||
Binary file not shown.
@ -0,0 +1 @@ |
||||
{"debug": true, "consolidatedBasePath": ".", "defaultLocations": {"ach": "Z:/Business Solutions/Griff/Code/InfoLeaseExtract/InputFiles", "disp": "", "gl": "", "lb": "Z:/Business Solutions/Griff/Code/InfoLeaseExtract/InputFiles", "minv": "", "niv": "", "ren": "", "pymt": "Z:/Business Solutions/Griff/Code/InfoLeaseExtract/InputFiles", "uap": "", "pastdue": ""}} |
||||
@ -0,0 +1,242 @@ |
||||
from pathlib import Path |
||||
import re |
||||
from re import Pattern |
||||
import pandas as pd |
||||
from pandas import DataFrame, ExcelWriter, read_excel |
||||
from datetime import datetime as dt, timedelta |
||||
import logging |
||||
import il_reports as ilr |
||||
from dataclasses import dataclass |
||||
from typing import Callable |
||||
from tqdm import tqdm |
||||
from multiprocessing import Pool, cpu_count |
||||
import os |
||||
|
||||
TOP_PATH: Path = Path(r"\\leafnow.com\shared\Accounting\CASH APPS\2023") |
||||
|
||||
class LevelFilter(object): |
||||
def __init__(self, level): |
||||
self.__level = level |
||||
|
||||
def filter(self, logRecord): |
||||
return logRecord.levelno == self.__level |
||||
|
||||
def create_logger(logger_name: str = __name__, ): |
||||
logger = logging.getLogger(logger_name) |
||||
|
||||
log_folder = Path(r"\\leafnow.com\shared\Business Solutions\Griff\Code\InfoLeaseExtract\logs") |
||||
fail_handler = logging.FileHandler(Path(log_folder,"Fail_br.log"), 'w') |
||||
fail_handler.setLevel(logging.WARNING) |
||||
|
||||
info_handler = logging.FileHandler(Path(log_folder,"Info_br.log"), 'w') |
||||
info_handler.setLevel(logging.INFO) |
||||
info_handler.addFilter(LevelFilter(logging.INFO)) |
||||
|
||||
debug_handler = logging.FileHandler(Path(log_folder,"Debug_br.log"), 'w') |
||||
debug_handler.setLevel(logging.DEBUG) |
||||
debug_handler.addFilter(LevelFilter(logging.DEBUG)) |
||||
|
||||
s_handler = logging.StreamHandler() |
||||
s_handler.setLevel(logging.INFO) |
||||
|
||||
logger.addHandler(fail_handler) |
||||
logger.addHandler(info_handler) |
||||
logger.addHandler(debug_handler) |
||||
logger.addHandler(s_handler) |
||||
|
||||
return logger |
||||
|
||||
logger = create_logger() |
||||
|
||||
@dataclass |
||||
class ExtractInstruction: |
||||
input_regex: Pattern |
||||
sheet_name: str |
||||
extract_method: Callable |
||||
|
||||
@dataclass |
||||
class ReportFolder: |
||||
folder_name: Path |
||||
extraction_methods: list[ExtractInstruction] |
||||
|
||||
def extract_date_path(path: Path) -> Path: |
||||
date_pattern = re.compile(r'^\d{4}\.\d{2}$') |
||||
|
||||
for parent in path.parents: |
||||
if date_pattern.match(parent.name): |
||||
return parent |
||||
return None |
||||
|
||||
def append_to_consolidated_report( report_path: Path, report_df: DataFrame, sheet_name: str): |
||||
""" |
||||
""" |
||||
|
||||
report_month: Path = extract_date_path(report_path) |
||||
report_name: str = f"{str(report_month.name).replace('.','-')}_{sheet_name}_ConsolidatedReport.xlsx" |
||||
|
||||
|
||||
save_path = Path(r"\\leafnow.com\shared\Business Solutions\Griff\Code\InfoLeaseExtract\2023",report_name) |
||||
logger.debug(f"{save_path=}") |
||||
# Check if the current month has a consolidated report |
||||
|
||||
if not save_path.exists(): |
||||
logger.debug(f"Consolidated Report | No monthly summary file!\n\tCreating: {save_path}") |
||||
# No file exists yet |
||||
# Create it and add the current month |
||||
try: |
||||
with pd.ExcelWriter(save_path) as writer: |
||||
logger.debug(f"Consolidated Report | {sheet_name}: Saving data as: {report_name}") |
||||
report_df.to_excel(writer, index=False, sheet_name=sheet_name) |
||||
except Exception as e: |
||||
logger.error(f"Failed to create to consolidated report! {report_name} | {sheet_name} | {report_path} :\n{e}") |
||||
|
||||
else: |
||||
# We need to read the dataframe in the current monthly report |
||||
# Check that we are not adding matching data |
||||
# Save the new report |
||||
#FIXME: This is so hacky it's embaressing |
||||
try: |
||||
current_data_len = len(pd.read_excel(save_path,sheet_name=sheet_name)) |
||||
with pd.ExcelWriter(save_path, engine='openpyxl', mode='a',if_sheet_exists="overlay") as writer: |
||||
logger.debug(f"Consolidated Report | {sheet_name}: Saving data as: {report_name}") |
||||
report_df.to_excel(writer, index=False, sheet_name=sheet_name,startrow=current_data_len,header=False) |
||||
except Exception as e: |
||||
logger.error(f"Failed to append to consolidated report! {report_name} | {sheet_name} | {report_path} :\n{e}") |
||||
|
||||
def process_report(file: Path, extract_inst: ExtractInstruction) -> bool: |
||||
|
||||
try: |
||||
with open(str(file), errors="replace") as f: |
||||
report_str: str = f.read() |
||||
#logger.debug(f"{report_str}") |
||||
try: |
||||
df: DataFrame = extract_inst.extract_method(report_str, None) |
||||
if df.empty: |
||||
raise ValueError("Dataframe is empty!") |
||||
except Exception as e: |
||||
logger.warning(f"Failed to create report df: {extract_inst.sheet_name}:\n{e}") |
||||
return False |
||||
append_to_consolidated_report(file, df, extract_inst.sheet_name) |
||||
return True |
||||
except Exception as e: |
||||
logger.exception(f"could not process {file}:\n{e}") |
||||
return False |
||||
|
||||
def process_folder(folder: ReportFolder): |
||||
|
||||
# Search recurively through date directories |
||||
report_date: dt = dt(2023, 5, 1) |
||||
while report_date.date() < dt.now().date(): |
||||
logger.info(f"{folder.folder_name} | Processing date: {report_date}") |
||||
report_folder: Path = Path(TOP_PATH, |
||||
report_date.strftime("%Y.%m"), |
||||
report_date.strftime("%Y.%m.%d"), |
||||
folder.folder_name |
||||
) |
||||
logger.debug(f"report_folder: {report_folder}") |
||||
if report_folder.exists(): |
||||
for xi in folder.extraction_methods: |
||||
try: |
||||
files = report_folder.glob(f"*{xi.input_regex}*") |
||||
report_file: Path = next(files) |
||||
logger.debug(f"Report file: {report_file}") |
||||
except IndexError as ie: |
||||
logger.warning(f"No matching reports!: {ie}") |
||||
except Exception as e: |
||||
logger.debug(f"Could not get report_file: {report_folder.glob(f'*{xi.input_regex}*')} \n{e}") |
||||
continue |
||||
try: |
||||
success = process_report(report_file, xi) |
||||
if success: |
||||
logger.info(f"Report Processed: {report_file} | {xi.sheet_name}") |
||||
else: |
||||
logger.warning(f"Failed to process report: {report_file} | {xi.sheet_name}") |
||||
|
||||
except Exception as e: |
||||
logger.exception(f"Could not process report ({report_file}) :\n{e}") |
||||
continue |
||||
else: |
||||
logger.debug(f"Folder '{report_folder}' does not exist!") |
||||
report_date = report_date + timedelta(days=1) |
||||
logger.debug(f"Finished scanning {folder.folder_name}!") |
||||
|
||||
|
||||
def combine(): |
||||
|
||||
WORK_DIR = Path(r"\\leafnow.com\shared\Business Solutions\Griff\Code\InfoLeaseExtract\2023") |
||||
|
||||
REPORTS = [ |
||||
"ACH", |
||||
"CHECKS LIVE", |
||||
"CREDIT CARDS", |
||||
"PAY BY PHONE", |
||||
"WIRE", |
||||
"RETURNS ACH", |
||||
"RETURNS PORTAL" |
||||
] |
||||
|
||||
for i in range(1,6): |
||||
|
||||
month = f"2023-0{i}" |
||||
mcr: Path = Path(f"{month} Consolidated Report.xlsx") |
||||
print(f"Creating monthly consolidated report: {mcr}") |
||||
with ExcelWriter(Path(WORK_DIR, "Monthly", mcr), engine="xlsxwriter") as wrtr: |
||||
for r in REPORTS: |
||||
report_path: Path = Path(WORK_DIR, f"{month}_{r}_ConsolidatedReport.xlsx") |
||||
print(f"Report Path ({r}): {report_path}") |
||||
|
||||
rdf: DataFrame = read_excel(report_path, sheet_name=r) |
||||
|
||||
rdf.to_excel(wrtr, sheet_name=r, freeze_panes=(1,0), index=False) |
||||
|
||||
if __name__ == "__main__": |
||||
|
||||
|
||||
|
||||
|
||||
FOLDERS = [ |
||||
ReportFolder("ACH", [ |
||||
ExtractInstruction("_ACH_", "ACH", ilr.ach), |
||||
]), |
||||
ReportFolder("CHECKS LIVE", [ |
||||
ExtractInstruction("_PROGPAY_BER", "CHECKS LIVE", ilr.payment_transactions) |
||||
]), |
||||
ReportFolder("CREDIT CARDS", [ |
||||
ExtractInstruction("_VMCC_BER", "CREDIT CARDS", ilr.payment_transactions) |
||||
]), |
||||
ReportFolder("LOCKBOX", [ |
||||
ExtractInstruction("_LOCKBOX_\d+_", "LOCKBOX", ilr.lockbox) |
||||
]), |
||||
ReportFolder("PAY BY PHONE", [ |
||||
ExtractInstruction("_PBP_EPAY_DPS_BER", "PAY BY PHONE", ilr.payment_transactions) |
||||
]), |
||||
ReportFolder("RETURN REPORTING", [ |
||||
ExtractInstruction("_PBP_EPAY_RETURNS_BER", "RETURNS ACH", ilr.payment_transactions), |
||||
ExtractInstruction("_RETURNS_BER", "RETURNS PORTAL", ilr.payment_transactions)] |
||||
), |
||||
ReportFolder("WIRES", [ |
||||
ExtractInstruction("MTBWIRE_BER", "WIRE", ilr.payment_transactions) |
||||
]), |
||||
] |
||||
|
||||
process_folder(FOLDERS[0]) |
||||
# with Pool(cpu_count()) as pool: |
||||
# for folder in tqdm(pool.imap_unordered(process_folder,FOLDERS)): |
||||
# try: |
||||
# print(f"Completed!") |
||||
# except Exception as e: |
||||
# print(f"Failed to process\n {e}") |
||||
# continue |
||||
# for folder in tqdm(FOLDERS): |
||||
# try: |
||||
# process_folder(folder) |
||||
# print(f"Completed: {folder.folder_name}") |
||||
# except Exception as e: |
||||
# print(f"Failed to process {folder.folder_name} \n {e}") |
||||
# continue |
||||
# input("Complete!") |
||||
|
||||
|
||||
|
||||
combine() |
||||
|
||||
@ -0,0 +1,12 @@ |
||||
{ |
||||
"name": { |
||||
"report": "", |
||||
"excel": "" |
||||
}, |
||||
"relative_position": { |
||||
"rows": 0, |
||||
"col": 0 |
||||
}, |
||||
"length": 0, |
||||
"data_type": "int" |
||||
} |
||||
@ -0,0 +1,184 @@ |
||||
from typing import TypeAlias, TypeVar |
||||
from dataclasses import dataclass |
||||
from pathlib import Path |
||||
import pathlib as pl |
||||
from abc import ABC, abstractmethod, abstractproperty |
||||
from re import search, match, compile, Match, Pattern |
||||
from enum import Enum |
||||
|
||||
ColumnIndex: TypeAlias = int |
||||
Money: TypeAlias = float |
||||
|
||||
Numeric = TypeVar("Numeric", float, int) |
||||
|
||||
class Line(Enum): |
||||
Header: str |
||||
Data: str |
||||
Erroneous: str |
||||
Top: str |
||||
Bottom: str |
||||
|
||||
|
||||
|
||||
@dataclass |
||||
class RelativePosition: |
||||
""" |
||||
Coordinates for navigating from one point in a row to another |
||||
""" |
||||
rows: int |
||||
col: ColumnIndex |
||||
|
||||
@dataclass |
||||
class DataValue: |
||||
|
||||
position: RelativePosition |
||||
length : int |
||||
regex: Pattern |
||||
dtype: type |
||||
|
||||
def correct_line(self, adj_lines_since_header: int) -> bool: |
||||
""" |
||||
""" |
||||
return adj_lines_since_header % self.position.rows == 0 |
||||
|
||||
def _line_slice(self, line: Line.Data) -> str|None: |
||||
""" |
||||
Attempts to get the data from the line. |
||||
Returns string in correct postion or None if out of range. |
||||
""" |
||||
try: |
||||
start: int = self.position.col |
||||
end: int = start + self.length |
||||
line_slice: str = line[start:end] |
||||
except IndexError: |
||||
#TODO: Add logging |
||||
line_slice = None |
||||
finally: |
||||
return line_slice |
||||
|
||||
@staticmethod |
||||
def _to_float(number_str: str) -> float|None: |
||||
try: |
||||
f_value:float = float(number_str.replace(',','')) |
||||
return f_value |
||||
except: |
||||
return None |
||||
|
||||
def extract(self, line: Line.Data) -> type|None: |
||||
""" |
||||
""" |
||||
line_slice: str|None = self._line_slice(line) |
||||
if isinstance(line_slice, None): |
||||
return None |
||||
|
||||
value_match: Match|None = search(self.regex, line_slice) |
||||
if isinstance(value_match, None): |
||||
return None |
||||
|
||||
value_str: str = value_match.group() |
||||
|
||||
value_str.strip() |
||||
if self.dtype == int or self.dtype == float: |
||||
return self._to_float(value_str) |
||||
#TODO datetime |
||||
return value_str |
||||
|
||||
class DataSet: |
||||
|
||||
def __init__(self, config: dict) -> None: |
||||
self.r_name = config["naming"]["report"] |
||||
try: |
||||
self.e_name = config["naming"]["excel"] |
||||
except KeyError: |
||||
self.e_name = self.r_name |
||||
|
||||
self.data_value: DataValue = DataValue( |
||||
position = RelativePosition( |
||||
rows= config["relative_position"]["rows"], |
||||
col= config["relative_position"]["col"] |
||||
), |
||||
length = config["length"], |
||||
dtype = config["data_type"], |
||||
) |
||||
|
||||
def line_position(self, line: str) -> ColumnIndex|None: |
||||
""" |
||||
Searches a line for the report header for this dataset. |
||||
|
||||
Returns: |
||||
- ColumnIndex(int) | None: The column index of the matches end position |
||||
or None if no match was found |
||||
""" |
||||
header_match: Match|None = search(self.r_name, line) |
||||
return header_match.end() if isinstance(header_match, Match) else None |
||||
|
||||
@dataclass |
||||
class ReportConfig: |
||||
|
||||
file_extension: str |
||||
name: str |
||||
datasets: list[DataSet] |
||||
data_line_regexes: list[Pattern] |
||||
|
||||
|
||||
class ILReport(ABC): |
||||
|
||||
def __init__(self, file_path: Path, report_config: ReportConfig) -> None: |
||||
self.in_file_path: Path = file_path |
||||
self.line_gen = self._line_generator(file_path) |
||||
|
||||
self.config: ReportConfig = report_config |
||||
self.name = report_config.name |
||||
|
||||
self.line_type_history: list[Line] = [] |
||||
self.last_header_line: int|None = None |
||||
|
||||
self.data_dict: dict = { |
||||
header.e_name: [] |
||||
for header in self.config.datasets |
||||
} |
||||
|
||||
@staticmethod |
||||
def _line_generator(file_path: Path): |
||||
with open(file_path, 'r') as in_file: |
||||
line: str |
||||
for line in in_file.readlines(): |
||||
yield line |
||||
|
||||
def _add_line_history(self, line: Line, max_history: int = 10): |
||||
self.line_type_history.append(line) |
||||
while len(self.line_type_history) > max_history: |
||||
self.line_type_history.pop(0) |
||||
|
||||
def _is_header_line(self, line: str) -> bool: |
||||
""" |
||||
Checks whether a report line has data headers. |
||||
""" |
||||
regex: Pattern |
||||
for regex in self.config.data_line_regexes: |
||||
if isinstance(search(regex,line), Match): |
||||
return True |
||||
return False |
||||
|
||||
@abstractmethod |
||||
def _skip_line(self, line) -> bool: |
||||
""" |
||||
Tells whether we should skip this line |
||||
""" |
||||
|
||||
@abstractmethod |
||||
def _process_line(self): |
||||
""" |
||||
|
||||
""" |
||||
|
||||
@abstractmethod |
||||
def _process_dataline(self, dataline: Line.Data): |
||||
""" |
||||
""" |
||||
|
||||
# Search the row for a data set name, or list of data set names |
||||
# extract all the data until the next row |
||||
|
||||
if __name__ == "__main__": |
||||
datasets = [] |
||||
@ -1 +0,0 @@ |
||||
{"debug": false, "consolidatedBasePath": "leafnow.com/shared/cashapps", "defaultLocations": {"ach": "", "disp": "", "gl": "", "lb": "", "minv": "", "niv": "", "ren": "", "pymt": "", "uap": "", "pastdue": ""}} |
||||
Loading…
Reference in new issue