jacob-mathison/doczyai-pipelines
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Activity

Initial operationalized script

Browse Source
This commit is contained in:
Katon Minhas
2024-06-24 21:15:01 -07:00
committed by Michael McGuinness
parent 44cc674381
commit 41bcd7f372
2 changed files with 10 additions and 40 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src/file_processing.py
+3 -17
View File
@@ -40,11 +40,6 @@ def process_file(file_object):
filename, contract_text = file_object
if config.VERBOSE: print(f"Processing {filename}...")
################## CREATE OUTPUT DIRECTORIES ##################
base_filename = os.path.splitext(filename)[0].strip()
output_dir = os.path.join(config.OUTPUT_DIRECTORY, base_filename)
os.makedirs(output_dir, exist_ok=True)
################## PREPROCESS ##################
contract_text = preprocess.clean_newlines(contract_text)
contract_text = preprocess.clean_billed_charges(contract_text)
@@ -54,29 +49,20 @@ def process_file(file_object):
################## RUN TOP DOWN ##################
td_results = prompt_funcs.run_top_down(filename, text_dict) # Returns list of dictionaries for each page
pd.DataFrame(td_results).to_csv(os.path.join(output_dir, config.TD_RESULTS_NAME), index=False)
print(f"Top Down Complete - {filename}")
################## RUN BOTTOM UP ##################
bu_results = prompt_funcs.run_bottom_up(filename, text_dict) # Returns list of dictionaries
pd.DataFrame(bu_results).to_csv(os.path.join(output_dir, config.BU_RESULTS_NAME), index=False)
print(f"Bottom Up Complete - {filename}")
################## COMBINE TOP DOWN AND BOTTOM UP ##################
combined_results = merge_funcs.merge_results(td_results, bu_results, text_dict)
print(f"TD/BU Merge Complete - {filename} ")
################## WRITE UNPROCESSED OUTPUT ##################
combined_df = pd.DataFrame(combined_results)
combined_df.to_csv(os.path.join(output_dir, config.UNPROCESSED_RESULTS_NAME), index=False)
################## RUN POSTPROCESSING ##################
# combined_df = combined_df.applymap(postprocessingfuncs.sanitize_value) # Deprecated
combined_df = pd.DataFrame(combined_results)
post_processed_combined_df = postprocess.postprocess_results(combined_df)
post_processed_combined_df.to_excel(os.path.join(output_dir, config.PROCESSED_RESULTS_NAME), index=False)
print(f"Postprocessing Complete - {filename} ")
print(f"Output Complete - {filename} ")
return post_processed_combined_df
src/main.py
+7 -23
View File
@@ -14,6 +14,7 @@ Functional Overview:
# Imports
import concurrent.futures
import traceback
import pandas as pd
import utils
import config
@@ -29,31 +30,14 @@ def main():
# already_processed = [s.split('.txt_results')[0]+'.txt' for s in os.listdir('results/')]
# input_dict = {key : input_dict[key] for key in input_dict.keys() if key not in already_processed}
def process_item(item):
key, value = item
try:
file_processing.process_file(item)
#prompt_funcs.bottom_up(item)
except Exception as e:
# Print the error message and traceback
print(f"Error processing item {key}: {e}")
traceback.print_exc()
with concurrent.futures.ThreadPoolExecutor(max_workers=config.MAX_WORKERS) as executor:
# Process each item individually
futures = [executor.submit(process_item, item) for item in input_dict.items()]
# Wait for all futures to complete
for future in concurrent.futures.as_completed(futures):
# Retrieve any exceptions raised by the thread
try:
future.result()
except Exception as e:
print(f"Error: {e}")
all_results = []
for item in input_dict.items():
file_result = file_processing.process_file(item)
all_results.append(file_result)
# Write Consolidated Output
if config.WRITE_OUTPUT:
utils.consolidate_csvs(config.OUTPUT_DIRECTORY, config.OUTPUT_CSV_PATH)
all_results_df = pd.concat(all_results).reset_index(drop=True)
if __name__ == "__main__":