You can not select more than 25 topics
Topics must start with a letter or number, can include dashes ('-') and can be up to 35 characters long.
87 lines
3.5 KiB
87 lines
3.5 KiB
import os
|
|
import pandas as pd
|
|
from datetime import datetime as dt, timedelta
|
|
import sys, getopt
|
|
import re
|
|
from pathlib import Path
|
|
import time
|
|
from pprint import pprint as prt
|
|
|
|
contract_number_regex = "\d{3}-\d{7}-\d{3}"
|
|
|
|
def create_line_divider(breakage_list: list):
|
|
"""
|
|
This allows for the creation of a custom data extractor
|
|
Breakage list defines the split points that will be used for the line
|
|
Example
|
|
Given breakage_list [10, 20, 30]
|
|
using slot_num 0 in the resulting extract_line_slot will yield
|
|
characters 0 - 10 from the string.
|
|
Slot 1 would give characters 10 - 20
|
|
"""
|
|
def extract_line_slot(slot_num : int, line_string: str, debug : bool = False):
|
|
"""
|
|
Pulls data from a line/string using break points defined by the
|
|
parent function.
|
|
ONLY USE THIS FUNCTION THROUGH CREATION USING 'create_line_extractor'
|
|
Will automatically convert numbers to floats
|
|
"""
|
|
assert(slot_num < len(breakage_list)+1)
|
|
low_range = 0 if slot_num == 0 else breakage_list[slot_num-1]
|
|
high_range = len(line_string) if slot_num == len(breakage_list) else breakage_list[slot_num]
|
|
data = line_string[low_range:high_range].strip().replace(",", "")
|
|
try: data = float(data)
|
|
except: pass
|
|
if debug:
|
|
print(f"Slot num: {slot_num} | Low: {low_range} | High: {high_range} | Data: {data}")
|
|
return data
|
|
return extract_line_slot
|
|
|
|
|
|
def ach(report: str, save_name: str):
|
|
|
|
lines = report.splitlines()
|
|
extracted_data_dict = {
|
|
"ContractNumber" : [],
|
|
"CustomerName" : [],
|
|
"BankCode" : [],
|
|
"BankNumber": [],
|
|
"AccountNumber" : [],
|
|
"Payment" : [],
|
|
"Batch": [],
|
|
"Lessor": [],
|
|
"PaymentDate": [],
|
|
}
|
|
columns = list(extracted_data_dict.keys())
|
|
batches = {
|
|
"batch_num": [],
|
|
"payment_date": [],
|
|
"lessor": [],
|
|
"count": [],
|
|
"total": []
|
|
}
|
|
|
|
data_extractor = create_line_divider([19,57,67,82,104])
|
|
bank_number_regex = "\d{9}"
|
|
batch_num_regex = "BATCH \d{4} TOTAL"
|
|
for line in enumerate(lines):
|
|
if (re.search(contract_number_regex, line[1]) != None) & (re.search(bank_number_regex, line[1]) != None):
|
|
[extracted_data_dict[columns[c]].append(data_extractor(c, line[1])) for c in range(0, len(columns)-3)]
|
|
if re.search(batch_num_regex, line[1]) != None:
|
|
batches["batch_num"].append(line[1][96:101])
|
|
batches["payment_date"].append(lines[line[0]+2][114:125])
|
|
batches["lessor"].append(extracted_data_dict["ContractNumber"][-1][0:3])
|
|
batches["total"].append(float(line[1][107:125].strip().replace(",", "")))
|
|
batches["count"].append(float(lines[line[0]+6][107:125].strip().replace(",", "")))
|
|
[extracted_data_dict["Batch"].append(batches["batch_num"][-1]) for _ in range(0, (len(extracted_data_dict["BankCode"]) - len(extracted_data_dict["Batch"])))]
|
|
[extracted_data_dict["Lessor"].append(batches["lessor"][-1]) for _ in range(0, (len(extracted_data_dict["BankCode"]) - len(extracted_data_dict["Lessor"])))]
|
|
[extracted_data_dict["PaymentDate"].append(batches["payment_date"][-1]) for _ in range(0, (len(extracted_data_dict["BankCode"]) - len(extracted_data_dict["PaymentDate"])))]
|
|
|
|
dataframe = pd.DataFrame(extracted_data_dict)
|
|
|
|
return dataframe
|
|
|
|
with open("/config/workspace/LEAF/IL Extract SRC/2022.05.04_ACH_C") as rep_file:
|
|
report = rep_file.read()
|
|
|
|
prt(ach(report, "ACH_TESTING.xlsx")) |