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

@@ -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

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}")
 

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