From c56270c9519b39aefa7eb0993083f9a2d8a23b8f Mon Sep 17 00:00:00 2001 From: Sha Brown Date: Mon, 13 Oct 2025 18:54:10 +0000 Subject: [PATCH] Merged in feature/field_type_config (pull request #731) Feature/field type config * Added field type procesing conditional logic * Updated field type naming convention in config.py; Updated date_postprocessing to avoid column key missing issue * Cleaned up code; Move merging functionality to one to one processing block * Merged main into feature/field_type_config Approved-by: Katon Minhas --- fieldExtraction/src/config.py | 2 +- .../src/investment/file_processing.py | 98 ++++++++++++------- .../src/investment/postprocessing_funcs.py | 21 ++-- 3 files changed, 75 insertions(+), 46 deletions(-) diff --git a/fieldExtraction/src/config.py b/fieldExtraction/src/config.py index 996eae9..e18d90e 100644 --- a/fieldExtraction/src/config.py +++ b/fieldExtraction/src/config.py @@ -104,7 +104,7 @@ PER_FILE_LOG_DIR = get_arg_value("per_file_log_dir", "logs") ENABLE_RUNTIME_ROTATION = False # Set to True to enable multi-runtime failover # Field args -FIELDS = get_arg_value("fields", "abc") # Valid: ac, b, abc +FIELDS = get_arg_value("fields", "all") # Valid: one_to_one, one_to_n, all FIELD_JSON_PATH = "src/prompts/investment_prompts.json" # Client-specific args diff --git a/fieldExtraction/src/investment/file_processing.py b/fieldExtraction/src/investment/file_processing.py index 30ca59f..5bcf880 100644 --- a/fieldExtraction/src/investment/file_processing.py +++ b/fieldExtraction/src/investment/file_processing.py @@ -36,49 +36,71 @@ def process_file(file_object, constants: Constants, run_timestamp): text_dict = preprocess.clean_tables( text_dict, constants.EXHIBIT_HEADER_MARKERS, filename ) - exhibit_dict, all_exhibit_headers = preprocess.one_to_n_exhibit_chunking( - text_dict, constants.EXHIBIT_HEADER_MARKERS, filename - ) - logging.info(f"{datetime_str()} Preprocessing Complete - {filename}") - ################## ONE TO N ################## - if string_utils.contains_reimbursement(contract_text): - one_to_n_results, dynamic_one_to_one_fields = run_one_to_n_prompts( - filename, exhibit_dict, all_exhibit_headers, constants - ) # Return df - one_to_n_results["FILE_NAME"] = filename - one_to_n_results = postprocessing_funcs.generate_reimb_ids( - one_to_n_results - ) # Add reimb_id - logging.info(f"{datetime_str()} One to N Complete - {filename}") + # Set default values + dynamic_one_to_one_fields = FieldSet() + + process_one_to_n = config.FIELDS in ['all', 'one_to_n'] + process_one_to_one = config.FIELDS in ['all', 'one_to_one'] + + # Initialize default fallback for final results + final_results = pd.DataFrame([{"FILE_NAME": filename}]) + + # ONE TO N PROCESSING + if process_one_to_n: + exhibit_dict, all_exhibit_headers = preprocess.one_to_n_exhibit_chunking( + text_dict, constants.EXHIBIT_HEADER_MARKERS, filename + ) + logging.info(f"{datetime_str()} Preprocessing Complete - {filename}") + + one_to_n_results = pd.DataFrame([{"FILE_NAME": filename}]) # Initialize here + + if string_utils.contains_reimbursement(contract_text): + one_to_n_results, dynamic_one_to_one_fields = run_one_to_n_prompts( + filename, exhibit_dict, all_exhibit_headers, constants + ) + one_to_n_results["FILE_NAME"] = filename + one_to_n_results = postprocessing_funcs.generate_reimb_ids(one_to_n_results) + logging.info(f"{datetime_str()} One to N Complete - {filename}") + else: + logging.info(f"{datetime_str()} No Reimbursement Found, Skipping - {filename}") + + final_results = one_to_n_results # Set as final results else: - one_to_n_results = pd.DataFrame([{"FILE_NAME": filename}]) - dynamic_one_to_one_fields = FieldSet() - logging.info(f"{datetime_str()} No One to N Found, Skipping - {filename}") + logging.info(f"{datetime_str()} Fields not configured for One to N, Skipping - {filename}") - ################## ONE TO ONE ################## - one_to_one_results = run_one_to_one_prompts( - filename, - contract_text, - text_dict, - top_sheet_dict, - dynamic_one_to_one_fields, - constants, - ) # Return dict - one_to_one_results["FILE_NAME"] = filename - logging.info(f"{datetime_str()} One to One Complete - {filename}") + # ONE TO ONE PROCESSING + if process_one_to_one: + one_to_one_results = run_one_to_one_prompts( + filename, + contract_text, + text_dict, + top_sheet_dict, + dynamic_one_to_one_fields, + constants, + ) + one_to_one_results["FILE_NAME"] = filename + logging.info(f"{datetime_str()} One to One Complete - {filename}") + + # Decide how to handle one_to_one results + if process_one_to_n: + # BOTH processed - merge into one_to_n + final_results = row_funcs.merge_one_to_one_into_one_to_n( + one_to_n_results, one_to_one_results, constants + ) + else: + # ONLY one_to_one processed - convert dict to DataFrame + final_results = pd.DataFrame([one_to_one_results]) + else: + logging.info(f"{datetime_str()} Fields not configured for One to One, Skipping - {filename}") - ################## MERGE ################## - merged_results = row_funcs.merge_one_to_one_into_one_to_n( - one_to_n_results, one_to_one_results, constants - ) + # APPLY CODES IF ONE_TO_N WAS PROCESSED + if process_one_to_n: + results_with_code = code_funcs.code_breakout(final_results, constants) + final_results = code_funcs.grouper_breakout(results_with_code) + logging.info(f"{datetime_str()} Codes Complete - {filename}") - ################## CODES ################## - results_with_code = code_funcs.code_breakout(merged_results, constants) - final_results = code_funcs.grouper_breakout(results_with_code) - logging.info(f"{datetime_str()} Codes Complete - {filename}") - - ################## POSTPROCESS ################## + # POSTPROCESS final_df = postprocess.postprocess(final_results, constants) logging.info(f"{datetime_str()} Postprocessing Complete - {filename}") diff --git a/fieldExtraction/src/investment/postprocessing_funcs.py b/fieldExtraction/src/investment/postprocessing_funcs.py index 032e483..841a380 100644 --- a/fieldExtraction/src/investment/postprocessing_funcs.py +++ b/fieldExtraction/src/investment/postprocessing_funcs.py @@ -173,14 +173,21 @@ def date_postprocess(df, field_json_path): if field.field_name in df.columns: df[field.field_name] = list(map(date_postprocessing, df[field.field_name])) - # Derived termination date - df["AARETE_DERIVED_TERMINATION_DT"] = list( - map( - invoke_derived_term_date, - df["AARETE_DERIVED_EFFECTIVE_DT"], - df["TERMINATION_DT"], + # Check if required columns exist when creating "AARETE_DERIVED_TERMINATION_DT" + # Avoid error in "one_to_n" processing + required_cols = ["AARETE_DERIVED_EFFECTIVE_DT", "TERMINATION_DT"] + if all(col in df.columns for col in required_cols): + df["AARETE_DERIVED_TERMINATION_DT"] = list( + map( + invoke_derived_term_date, + df["AARETE_DERIVED_EFFECTIVE_DT"], + df["TERMINATION_DT"], + ) ) - ) + else: + missing = [col for col in required_cols if col not in df.columns] + logging.info(f"Skipping AARETE_DERIVED_TERMINATION_DT creation. Missing: {missing}") + df["AARETE_DERIVED_TERMINATION_DT"] = pd.NA return df