diff --git a/.gitignore b/.gitignore index b09252d..8af674e 100644 --- a/.gitignore +++ b/.gitignore @@ -108,3 +108,4 @@ TEST*.xml # PRD Documents *.prd *prd.md +.git.instructions.md diff --git a/src/config.py b/src/config.py index 9910ef0..42e2501 100644 --- a/src/config.py +++ b/src/config.py @@ -343,6 +343,10 @@ MIN_TABLE_ROWS_TO_SPLIT = int(get_arg_value("min_table_rows_to_split", "20")) # Table splitting method (currently only "cells" is supported) TABLE_SPLITTING_METHOD = get_arg_value("table_splitting_method", "cells") +######################################## OUTPUT FILE SPLITTING SETTINGS ######################################## +# Maximum rows per split file (files are split to keep filenames together) +MAX_ROWS_PER_SPLIT = int(get_arg_value("max_rows_per_split", "70000")) + ######################################## EC2 ######################################## # DEPRECATED: MODEL_STATS and GLOBAL_STATS are no longer updated. @@ -380,10 +384,12 @@ DOCZY_PDF_FILES_BUCKET_S3_URL = f"https://{DOCZY_PDF_FILES_BUCKET_NAME}.s3.us-ea ENABLE_VISION = get_arg_value("enable_vision", "False") == "True" # False by default ############## PARENT CHILD CONFIG ############## -PERFORM_PARENT_CHILD_MAPPING = get_arg_value("perform_pc", "False") == "True" +PERFORM_PARENT_CHILD_MAPPING = get_arg_value("perform_pc", "True") == "True" DOCZY_OUTPUT_FOR_PC = get_arg_value( "doczy_output_for_pc", "" ) # Pass either S3 URI path, or local path +# TODO: Remove WRITE_PC_TO_S3 after approval. Parent-child S3 upload now uses WRITE_TO_S3 +# via io_utils.upload_local_file_to_s3 in the pipeline. WRITE_PC_TO_S3 = get_arg_value("write_pc_to_s3", "False") == "True" ############## DOCUMENT TYPE CLASSIFICATION SETTINGS ############## @@ -399,6 +405,11 @@ DTC_OUTPUT_FILE = ( ) DTC_JSON_OUTPUT_FOLDER = "dtc_json_results" # Folder for individual JSON results +############## DASHBOARD POSTPROCESSING ############## +# Run dashboard postprocessing (comma-separated list format). Default: CC only. +# Set to True via run_dashboard arg when dashboard output is needed. +RUN_DASHBOARD_POSTPROCESSING = get_arg_value("run_dashboard", "False") == "True" + ############## QC/QA VALIDATION SETTINGS ############## # Enable QC/QA validation in pipeline ENABLE_QC_QA = get_arg_value("enable_qc_qa", "True") == "True" diff --git a/src/parent_child/__main__.py b/src/parent_child/__main__.py index 6b6a257..a69e449 100644 --- a/src/parent_child/__main__.py +++ b/src/parent_child/__main__.py @@ -79,7 +79,11 @@ def write_pc_excel( return output_path -def main(doczy_output_for_pc: str, client: Optional[str] = None) -> int: +def main( + doczy_output_for_pc: str, + client: Optional[str] = None, + output_dir: Optional[str] = None, +) -> tuple[int, str]: """ Main entry point for parent-child mapping. @@ -88,15 +92,20 @@ def main(doczy_output_for_pc: str, client: Optional[str] = None) -> int: client: Optional client identifier. If provided, takes precedence over CLI argument and path-based detection. Supported values: bcbs, caresource, molina, cnc, clover_health + output_dir: Directory to write output. When provided by pipeline/batch + run, PC results go to the same run directory as other outputs. + When None (standalone run), uses outputs/parent_child/{run_timestamp}. Returns: - Number of rows processed + Tuple of (row_count, local_path) where local_path is the written Excel file. """ # Extract client name from input file for output naming client_name = extract_client_name(doczy_output_for_pc) - # Generate timestamp for this run - run_timestamp = f"run_{datetime.now().strftime('%Y%m%d_%H-%M')}_{client_name}" + # Pipeline/batch run passes output_dir; standalone uses outputs/parent_child + if output_dir is None: + run_timestamp = f"run_{datetime.now().strftime('%Y%m%d_%H-%M')}_{client_name}" + output_dir = os.path.join("outputs", "parent_child", run_timestamp) try: # Read input file @@ -145,21 +154,13 @@ def main(doczy_output_for_pc: str, client: Optional[str] = None) -> int: original_df_with_pc, pc_df = parent_child_mapping(cleaned_df, original_df) # Save detailed results - output_dir = os.path.join("outputs/parent_child", run_timestamp) pc_df1 = qc_main(pc_df, client) output_filename = f"{client_name}_generic_hierarchy_lineage_output.xlsx" - if config.WRITE_PC_TO_S3: - # Write locally first, then upload to S3 - write_pc_excel(df_read, pc_df1, output_dir, client_name) - # TODO: Add S3 upload logic if needed - logging.info(f"PC results saved: {run_timestamp}/{output_filename}") - else: - write_pc_excel(df_read, pc_df1, output_dir, client_name) - logging.info(f"PC results saved locally: {output_dir}/{output_filename}") + local_path = write_pc_excel(df_read, pc_df1, output_dir, client_name) - # TODO: check with dev - return pc_df.shape[0] if pc_df is not None else mapped_df.shape[0] + row_count = pc_df.shape[0] if pc_df is not None else mapped_df.shape[0] + return row_count, local_path except Exception as e: logging.error(f"Parent child mapping failed due to {e}", exc_info=True) raise @@ -173,7 +174,7 @@ if __name__ == "__main__": # Generate timestamp for this run run_timestamp = f"run_{datetime.now().strftime('%Y%m%d_%H-%M')}_{config.BATCH_ID}" if config.DOCZY_OUTPUT_FOR_PC: - results = main(config.DOCZY_OUTPUT_FOR_PC) - logging.info(f"PC Mapping complete. Processed {results} files.") + row_count, _ = main(config.DOCZY_OUTPUT_FOR_PC) + logging.info(f"PC Mapping complete. Processed {row_count} files.") else: logging.warning("No input file specified in config.DOCZY_OUTPUT_FOR_PC") diff --git a/src/pipelines/clients/bcbs_promise/file_processing.py b/src/pipelines/clients/bcbs_promise/file_processing.py index ba2477a..9753204 100644 --- a/src/pipelines/clients/bcbs_promise/file_processing.py +++ b/src/pipelines/clients/bcbs_promise/file_processing.py @@ -161,19 +161,19 @@ def process_file(file_object, constants: Constants, run_timestamp): # POSTPROCESS with timing_utils.timed_block("postprocess", context=filename): - final_df = postprocess.postprocess(final_results, constants) + cc_df, dashboard_df = postprocess.postprocess(final_results, constants) logging.info(f"{datetime_str()} Postprocessing Complete - {filename}") ################## WRITE INDIVIDUAL ################## with timing_utils.timed_block("write_individual", context=filename): if config.WRITE_TO_S3: - io_utils.write_s3(final_df, filename, run_timestamp, "individual") + io_utils.write_s3(cc_df, filename, run_timestamp, "individual_cc") else: - io_utils.write_local(final_df, filename, "", "individual") + io_utils.write_local(cc_df, filename, "", "individual_cc") logging.info(f"{datetime_str()} Writing Complete - {filename}") - return final_df + return cc_df, dashboard_df def run_one_to_one_prompts( diff --git a/src/pipelines/clients/clover/file_processing.py b/src/pipelines/clients/clover/file_processing.py index 04f3b2f..86ebb3d 100644 --- a/src/pipelines/clients/clover/file_processing.py +++ b/src/pipelines/clients/clover/file_processing.py @@ -161,19 +161,19 @@ def process_file(file_object, constants: Constants, run_timestamp): # POSTPROCESS with timing_utils.timed_block("postprocess", context=filename): - final_df = postprocess.postprocess(final_results, constants) + cc_df, dashboard_df = postprocess.postprocess(final_results, constants) logging.info(f"{datetime_str()} Postprocessing Complete - {filename}") ################## WRITE INDIVIDUAL ################## with timing_utils.timed_block("write_individual", context=filename): if config.WRITE_TO_S3: - io_utils.write_s3(final_df, filename, run_timestamp, "individual") + io_utils.write_s3(cc_df, filename, run_timestamp, "individual_cc") else: - io_utils.write_local(final_df, filename, "", "individual") + io_utils.write_local(cc_df, filename, "", "individual_cc") logging.info(f"{datetime_str()} Writing Complete - {filename}") - return final_df + return cc_df, dashboard_df def run_one_to_one_prompts( diff --git a/src/pipelines/runner.py b/src/pipelines/runner.py index f328a5c..b13de73 100644 --- a/src/pipelines/runner.py +++ b/src/pipelines/runner.py @@ -13,6 +13,7 @@ Usage: import concurrent.futures import importlib import logging +import os import traceback import warnings from datetime import datetime @@ -68,14 +69,19 @@ def safe_process_file(item, constants, run_timestamp, file_processing): file_processing: The file_processing module to use Returns: - pd.DataFrame: Either: - - DataFrame with extracted field results (success case) - - Single-row DataFrame with error details (failure case) + tuple: Either: + - (cc_df, dashboard_df) tuple with extracted field results (success case) + - (error_df, error_df) tuple with error details (failure case) + + Note: + Returns both CC and dashboard versions from file processing. """ file_id = item[0] if isinstance(item, (list, tuple)) and len(item) > 0 else item try: - result = file_processing.process_file(item, constants, run_timestamp) - return result + cc_df, dashboard_df = file_processing.process_file( + item, constants, run_timestamp + ) + return cc_df, dashboard_df except Exception as e: error_type = type(e).__name__ error_message = str(e) @@ -86,7 +92,7 @@ def safe_process_file(item, constants, run_timestamp, file_processing): logging.error(f"Error Message: {error_message}") logging.error(f"Full traceback:\n{full_traceback}") - return pd.DataFrame( + error_df = pd.DataFrame( [ { "FILE_NAME": file_id, @@ -96,6 +102,7 @@ def safe_process_file(item, constants, run_timestamp, file_processing): } ] ) + return error_df, error_df # Return tuple for consistency def main(client: str = "saas", testing=False, test_params={}): @@ -177,7 +184,8 @@ def main(client: str = "saas", testing=False, test_params={}): logging.warning(f"Error warming caches (will proceed anyway): {str(e)}") # ========== CLIENT-SPECIFIC: File Processing ========== - successful_results = [] + successful_results_cc = [] + successful_results_dashboard = [] error_results = [] with concurrent.futures.ThreadPoolExecutor(max_workers=max_workers) as executor: futures = [ @@ -189,49 +197,71 @@ def main(client: str = "saas", testing=False, test_params={}): for future in concurrent.futures.as_completed(futures): try: - results = future.result() - if "error" in results.columns: - error_results.append(results) + cc_result, dashboard_result = future.result() + if "error" in cc_result.columns: + error_results.append(cc_result) else: - successful_results.append(results) + successful_results_cc.append(cc_result) + if ( + config.RUN_DASHBOARD_POSTPROCESSING + and dashboard_result is not None + ): + successful_results_dashboard.append(dashboard_result) except Exception as e: logging.error(f"Error processing future: {str(e)}") - error_results.append(pd.DataFrame([{"error": str(e)}])) + error_df = pd.DataFrame([{"error": str(e)}]) + error_results.append(error_df) - # Combine results - if len(successful_results) > 0: - FINAL_RESULT_DF = pd.concat(successful_results) - else: - FINAL_RESULT_DF = pd.DataFrame() + # Combine results + if len(successful_results_cc) > 0: + FINAL_RESULT_DF_CC = pd.concat(successful_results_cc, ignore_index=True) + if successful_results_dashboard: + FINAL_RESULT_DF_DASHBOARD = pd.concat( + successful_results_dashboard, ignore_index=True + ) + else: + FINAL_RESULT_DF_DASHBOARD = pd.DataFrame() + else: + FINAL_RESULT_DF_CC = pd.DataFrame() + FINAL_RESULT_DF_DASHBOARD = pd.DataFrame() - if len(error_results) > 0: - ERROR_RESULT_DF = pd.concat(error_results) - else: - ERROR_RESULT_DF = pd.DataFrame() + if len(error_results) > 0: + ERROR_RESULT_DF = pd.concat(error_results, ignore_index=True) + else: + ERROR_RESULT_DF = pd.DataFrame() # ========== COMMON: QC/QA Validation ========== - if config.ENABLE_QC_QA and not FINAL_RESULT_DF.empty: + # Run QC/QA validation by default (preserves automatic behavior for single files and batches) + # QC/QA runs on CC version only (dashboard version and error files do not need QC/QA) + validated_cc_df = None + qc_qa_stats_df = None + + # Run QC/QA on CC version if we have successful results + if not FINAL_RESULT_DF_CC.empty: logging.info("=" * 80) - logging.info("Running QC/QA Validation Pipeline...") + logging.info("Running QC/QA Validation Pipeline on CC results...") logging.info("=" * 80) try: - validated_df = run_qc_qa_pipeline(FINAL_RESULT_DF.copy(), stats_df=None) - logging.info("QC/QA validation completed successfully") + # Ensure unique index before QC/QA (safety check) + if FINAL_RESULT_DF_CC.index.has_duplicates: + logging.warning( + "Detected duplicate indices in FINAL_RESULT_DF_CC, resetting index..." + ) + FINAL_RESULT_DF_CC = FINAL_RESULT_DF_CC.reset_index(drop=True) - logging.info("Generating QC/QA statistics...") - qc_qa_stats_df = generate_statistics(validated_df) - logging.info("QC/QA statistics generated successfully") - - save_qc_qa_outputs( - validated_df=validated_df, - stats_df=qc_qa_stats_df, - run_timestamp=run_timestamp, - write_to_s3=config.WRITE_TO_S3, + validated_cc_df = run_qc_qa_pipeline( + FINAL_RESULT_DF_CC.copy(), stats_df=None ) + logging.info("QC/QA validation on CC results completed successfully") + + # Generate validation statistics from CC version + logging.info("Generating QC/QA statistics...") + qc_qa_stats_df = generate_statistics(validated_cc_df) + logging.info("QC/QA statistics generated successfully") except Exception as e: - logging.error(f"QC/QA validation failed: {str(e)}") + logging.error(f"QC/QA validation on CC results failed: {str(e)}") logging.error(f"Full traceback:\n{traceback.format_exc()}") - logging.warning("Continuing with original unvalidated results...") + logging.warning("Continuing with original unvalidated CC results...") logging.info("=" * 80) # ========== COMMON: Output & Parent-Child ========== @@ -241,32 +271,93 @@ def main(client: str = "saas", testing=False, test_params={}): ) if config.WRITE_TO_S3: + # Write CC version doczy_output_for_pc = io_utils.write_s3( - FINAL_RESULT_DF, "", run_timestamp, "final" + FINAL_RESULT_DF_CC, "", run_timestamp, "cc_results_full" ) + # Write dashboard version (only when dashboard postprocessing is enabled) + if ( + config.RUN_DASHBOARD_POSTPROCESSING + and not FINAL_RESULT_DF_DASHBOARD.empty + ): + io_utils.write_s3( + FINAL_RESULT_DF_DASHBOARD, + "", + run_timestamp, + "dashboard_results_full", + ) + # Write error file if not ERROR_RESULT_DF.empty: io_utils.write_s3(ERROR_RESULT_DF, "", run_timestamp, "error") + # Write QC/QA outputs (only for CC version, not dashboard) + if validated_cc_df is not None: + io_utils.write_s3(validated_cc_df, "", run_timestamp, "qc_qa_cc_full") + if qc_qa_stats_df is not None: + io_utils.write_s3(qc_qa_stats_df, "", run_timestamp, "qc_qa_stats") + if not per_file_usage_df.empty and not batch_summary_df.empty: logging.info("Exporting usage and cost tracking data to S3...") io_utils.write_s3(per_file_usage_df, "", run_timestamp, "usage") io_utils.write_s3(batch_summary_df, "", run_timestamp, "usage_summary") else: + # Write CC version doczy_output_for_pc = io_utils.write_local( - FINAL_RESULT_DF, "", run_timestamp, "final" + FINAL_RESULT_DF_CC, "", run_timestamp, "cc_results_full" ) + # Write dashboard version (only when dashboard postprocessing is enabled) + if ( + config.RUN_DASHBOARD_POSTPROCESSING + and not FINAL_RESULT_DF_DASHBOARD.empty + ): + io_utils.write_local( + FINAL_RESULT_DF_DASHBOARD, + "", + run_timestamp, + "dashboard_results_full", + ) + # Write error file if not ERROR_RESULT_DF.empty: io_utils.write_local(ERROR_RESULT_DF, "", run_timestamp, "error") + # Write QC/QA outputs (only for CC version, not dashboard) + if validated_cc_df is not None: + io_utils.write_local( + validated_cc_df, "", run_timestamp, "qc_qa_cc_full" + ) + if qc_qa_stats_df is not None: + io_utils.write_local(qc_qa_stats_df, "", run_timestamp, "qc_qa_stats") + if config.PERFORM_PARENT_CHILD_MAPPING: - parent_child_main.main(doczy_output_for_pc, client=client) + if FINAL_RESULT_DF_CC.empty: + logging.warning( + "Skipping Parent-Child mapping: No successful results to process (FINAL_RESULT_DF_CC is empty)" + ) + else: + logging.info( + f"Starting Parent-Child mapping with {len(FINAL_RESULT_DF_CC)} records from: {doczy_output_for_pc}" + ) + pc_output_dir = os.path.join( + config.CONSOLIDATED_OUTPUT_DIRECTORY, + run_timestamp, + "parent-child", + ) + _, pc_local_path = parent_child_main.main( + doczy_output_for_pc, + client=client, + output_dir=pc_output_dir, + ) + if config.WRITE_TO_S3 and pc_local_path: + io_utils.upload_local_file_to_s3( + pc_local_path, run_timestamp, "parent-child" + ) if not per_file_usage_df.empty and not batch_summary_df.empty: logging.info("Exporting usage and cost tracking data locally...") io_utils.write_local(per_file_usage_df, "", run_timestamp, "usage") io_utils.write_local(batch_summary_df, "", run_timestamp, "usage_summary") else: - return FINAL_RESULT_DF, ERROR_RESULT_DF + return FINAL_RESULT_DF_CC, FINAL_RESULT_DF_DASHBOARD, ERROR_RESULT_DF if __name__ == "__main__": diff --git a/src/pipelines/saas/file_processing.py b/src/pipelines/saas/file_processing.py index 9ade81a..c9b43cc 100644 --- a/src/pipelines/saas/file_processing.py +++ b/src/pipelines/saas/file_processing.py @@ -167,19 +167,19 @@ def process_file(file_object, constants: Constants, run_timestamp): # POSTPROCESS with timing_utils.timed_block("postprocess", context=filename): - final_df = postprocess.postprocess(final_results, constants) + cc_df, dashboard_df = postprocess.postprocess(final_results, constants) logging.info(f"{datetime_str()} Postprocessing Complete - {filename}") ################## WRITE INDIVIDUAL ################## with timing_utils.timed_block("write_individual", context=filename): if config.WRITE_TO_S3: - io_utils.write_s3(final_df, filename, run_timestamp, "individual") + io_utils.write_s3(cc_df, filename, run_timestamp, "individual_cc") else: - io_utils.write_local(final_df, filename, "", "individual") + io_utils.write_local(cc_df, filename, "", "individual_cc") logging.info(f"{datetime_str()} Writing Complete - {filename}") - return final_df + return cc_df, dashboard_df def run_one_to_one_prompts( diff --git a/src/pipelines/saas/main.py b/src/pipelines/saas/main.py index 0a74473..1d103c0 100644 --- a/src/pipelines/saas/main.py +++ b/src/pipelines/saas/main.py @@ -1,5 +1,6 @@ import concurrent.futures import logging +import os import random import time import traceback @@ -40,20 +41,22 @@ def safe_process_file(item, constants, run_timestamp): run_timestamp: Timestamp string for output file naming and tracking Returns: - pd.DataFrame: Either: - - DataFrame with extracted field results (success case) - - Single-row DataFrame with error details (failure case) + tuple: Either: + - (cc_df, dashboard_df) tuple with extracted field results (success case) + - (error_df, error_df) tuple with error details (failure case) Note: - Always returns a DataFrame to maintain consistent output format for pd.concat. + Returns both CC and dashboard versions from file processing. With FAISS migration, no special resource cleanup is needed. """ file_id = ( item[0] if isinstance(item, (list, tuple)) and len(item) > 0 else item ) # unpack file_id from item (passed in as a tuple below) try: - result = file_processing.process_file(item, constants, run_timestamp) - return result + cc_df, dashboard_df = file_processing.process_file( + item, constants, run_timestamp + ) + return cc_df, dashboard_df except ( Exception ) as e: # When there's an issue with the processing inside the future @@ -67,7 +70,7 @@ def safe_process_file(item, constants, run_timestamp): logging.error(f"Full traceback:\n{full_traceback}") - return pd.DataFrame( + error_df = pd.DataFrame( [ { "FILE_NAME": file_id, @@ -81,6 +84,7 @@ def safe_process_file(item, constants, run_timestamp): } ] ) # Return a single-row dataframe so we can still concat it later + return error_df, error_df # Return tuple for consistency def main(testing=False, test_params={}): @@ -160,7 +164,8 @@ def main(testing=False, test_params={}): logging.warning(f"Error warming caches (will proceed anyway): {str(e)}") # Process files concurrently - each thread gets its own per-file logging context - successful_results = [] + successful_results_cc = [] + successful_results_dashboard = [] error_results = [] logging.info(f"Starting parallel file processing with {max_workers} workers...") @@ -178,11 +183,16 @@ def main(testing=False, test_params={}): completed_count = 0 for future in concurrent.futures.as_completed(futures): try: - results = future.result() - if "error" in results.columns: - error_results.append(results) + cc_result, dashboard_result = future.result() + if "error" in cc_result.columns: + error_results.append(cc_result) else: - successful_results.append(results) + successful_results_cc.append(cc_result) + if ( + config.RUN_DASHBOARD_POSTPROCESSING + and dashboard_result is not None + ): + successful_results_dashboard.append(dashboard_result) completed_count += 1 if completed_count % 5 == 0: # Log progress every 5 files @@ -194,54 +204,61 @@ def main(testing=False, test_params={}): ) as e: # When there's an issue with the future itself (not the processing inside the future) logging.error(f"Error processing future: {str(e)}") # Coerce this to the format expected by pd.concat (so a single-row dataframe) - error_results.append(pd.DataFrame([{"error": str(e)}])) + error_df = pd.DataFrame([{"error": str(e)}]) + error_results.append(error_df) # only concat if we have results with timing_utils.timed_block("concat_results", log_level="INFO"): - if len(successful_results) > 0: - FINAL_RESULT_DF = pd.concat(successful_results, ignore_index=True) + if len(successful_results_cc) > 0: + FINAL_RESULT_DF_CC = pd.concat(successful_results_cc, ignore_index=True) + if successful_results_dashboard: + FINAL_RESULT_DF_DASHBOARD = pd.concat( + successful_results_dashboard, ignore_index=True + ) + else: + FINAL_RESULT_DF_DASHBOARD = pd.DataFrame() else: - FINAL_RESULT_DF = pd.DataFrame() + FINAL_RESULT_DF_CC = pd.DataFrame() + FINAL_RESULT_DF_DASHBOARD = pd.DataFrame() if len(error_results) > 0: ERROR_RESULT_DF = pd.concat(error_results, ignore_index=True) else: ERROR_RESULT_DF = pd.DataFrame() - # Run QC/QA validation if enabled (creates separate validated copy) - if config.ENABLE_QC_QA and not FINAL_RESULT_DF.empty: + # Run QC/QA validation by default (preserves automatic behavior for single files and batches) + # QC/QA runs on CC version only (dashboard version and error files do not need QC/QA) + validated_cc_df = None + qc_qa_stats_df = None + + # Run QC/QA on CC version if we have successful results + if not FINAL_RESULT_DF_CC.empty: logging.info("=" * 80) - logging.info("Running QC/QA Validation Pipeline...") + logging.info("Running QC/QA Validation Pipeline on CC results...") logging.info("=" * 80) with timing_utils.timed_block("qc_qa_validation", log_level="INFO"): try: # Ensure unique index before QC/QA (safety check) - if FINAL_RESULT_DF.index.has_duplicates: + if FINAL_RESULT_DF_CC.index.has_duplicates: logging.warning( - "Detected duplicate indices in FINAL_RESULT_DF, resetting index..." + "Detected duplicate indices in FINAL_RESULT_DF_CC, resetting index..." ) - FINAL_RESULT_DF = FINAL_RESULT_DF.reset_index(drop=True) + FINAL_RESULT_DF_CC = FINAL_RESULT_DF_CC.reset_index(drop=True) - # Create a copy for validation - preserve original FINAL_RESULT_DF - validated_df = run_qc_qa_pipeline(FINAL_RESULT_DF.copy(), stats_df=None) - logging.info("QC/QA validation completed successfully") - - # Generate validation statistics - logging.info("Generating QC/QA statistics...") - qc_qa_stats_df = generate_statistics(validated_df) - logging.info("QC/QA statistics generated successfully") - - # Save QC/QA outputs (local + S3) - separate from main results - save_qc_qa_outputs( - validated_df=validated_df, - stats_df=qc_qa_stats_df, - run_timestamp=run_timestamp, - write_to_s3=config.WRITE_TO_S3, + # Create a copy for validation - preserve original FINAL_RESULT_DF_CC + validated_cc_df = run_qc_qa_pipeline( + FINAL_RESULT_DF_CC.copy(), stats_df=None ) + logging.info("QC/QA validation on CC results completed successfully") + + # Generate validation statistics from CC version + logging.info("Generating QC/QA statistics...") + qc_qa_stats_df = generate_statistics(validated_cc_df) + logging.info("QC/QA statistics generated successfully") except Exception as e: - logging.error(f"QC/QA validation failed: {str(e)}") + logging.error(f"QC/QA validation on CC results failed: {str(e)}") logging.error(f"Full traceback:\n{traceback.format_exc()}") - logging.warning("Continuing with original unvalidated results...") + logging.warning("Continuing with original unvalidated CC results...") logging.info("=" * 80) if not testing: @@ -252,15 +269,36 @@ def main(testing=False, test_params={}): with timing_utils.timed_block("write_outputs", log_level="INFO"): if config.WRITE_TO_S3: + # Write CC version doczy_output_for_pc = io_utils.write_s3( - FINAL_RESULT_DF, "", run_timestamp, "final" + FINAL_RESULT_DF_CC, "", run_timestamp, "cc_results_full" ) + # Write dashboard version (only when dashboard postprocessing is enabled) + if ( + config.RUN_DASHBOARD_POSTPROCESSING + and not FINAL_RESULT_DF_DASHBOARD.empty + ): + io_utils.write_s3( + FINAL_RESULT_DF_DASHBOARD, + "", + run_timestamp, + "dashboard_results_full", + ) + # Write error file if not ERROR_RESULT_DF.empty: io_utils.write_s3(ERROR_RESULT_DF, "", run_timestamp, "error") logging.warning( f"{len(ERROR_RESULT_DF)} files failed processing - error results written to S3" ) + # Write QC/QA outputs (only for CC version, not dashboard) + if validated_cc_df is not None: + io_utils.write_s3( + validated_cc_df, "", run_timestamp, "qc_qa_cc_full" + ) + if qc_qa_stats_df is not None: + io_utils.write_s3(qc_qa_stats_df, "", run_timestamp, "qc_qa_stats") + # Export usage and cost tracking data to S3 only if both dataframes have data if not per_file_usage_df.empty and not batch_summary_df.empty: logging.info("Exporting usage and cost tracking data to S3...") @@ -269,23 +307,46 @@ def main(testing=False, test_params={}): batch_summary_df, "", run_timestamp, "usage_summary" ) else: + # Write CC version doczy_output_for_pc = io_utils.write_local( - FINAL_RESULT_DF, "", run_timestamp, "final" + FINAL_RESULT_DF_CC, "", run_timestamp, "cc_results_full" ) + # Write dashboard version (only when dashboard postprocessing is enabled) + if ( + config.RUN_DASHBOARD_POSTPROCESSING + and not FINAL_RESULT_DF_DASHBOARD.empty + ): + io_utils.write_local( + FINAL_RESULT_DF_DASHBOARD, + "", + run_timestamp, + "dashboard_results_full", + ) + # Write error file if not ERROR_RESULT_DF.empty: io_utils.write_local(ERROR_RESULT_DF, "", run_timestamp, "error") logging.warning( f"{len(ERROR_RESULT_DF)} files failed processing - error results written locally" ) + # Write QC/QA outputs (only for CC version, not dashboard) + if validated_cc_df is not None: + io_utils.write_local( + validated_cc_df, "", run_timestamp, "qc_qa_cc_full" + ) + if qc_qa_stats_df is not None: + io_utils.write_local( + qc_qa_stats_df, "", run_timestamp, "qc_qa_stats" + ) + if config.PERFORM_PARENT_CHILD_MAPPING: - if FINAL_RESULT_DF.empty: + if FINAL_RESULT_DF_CC.empty: logging.warning( - "Skipping Parent-Child mapping: No successful results to process (FINAL_RESULT_DF is empty)" + "Skipping Parent-Child mapping: No successful results to process (FINAL_RESULT_DF_CC is empty)" ) else: logging.info( - f"Starting Parent-Child mapping with {len(FINAL_RESULT_DF)} records from: {doczy_output_for_pc}" + f"Starting Parent-Child mapping with {len(FINAL_RESULT_DF_CC)} records from: {doczy_output_for_pc}" ) with timing_utils.timed_block("parent_child_mapping", log_level="INFO"): client = ( @@ -295,7 +356,20 @@ def main(testing=False, test_params={}): and config.CLIENT[0] != "None" else None ) - parent_child_main.main(doczy_output_for_pc, client=client) + pc_output_dir = os.path.join( + config.CONSOLIDATED_OUTPUT_DIRECTORY, + run_timestamp, + "parent-child", + ) + _, pc_local_path = parent_child_main.main( + doczy_output_for_pc, + client=client, + output_dir=pc_output_dir, + ) + if config.WRITE_TO_S3 and pc_local_path: + io_utils.upload_local_file_to_s3( + pc_local_path, run_timestamp, "parent-child" + ) # Export usage and cost tracking data locally only if both dataframes have data if not per_file_usage_df.empty and not batch_summary_df.empty: @@ -310,7 +384,7 @@ def main(testing=False, test_params={}): logging.info("=" * 80) timing_utils.print_hierarchical_timing_summary() else: - return FINAL_RESULT_DF, ERROR_RESULT_DF + return FINAL_RESULT_DF_CC, FINAL_RESULT_DF_DASHBOARD, ERROR_RESULT_DF if __name__ == "__main__": diff --git a/src/pipelines/shared/postprocessing/postprocess.py b/src/pipelines/shared/postprocessing/postprocess.py index 88addaf..598fed6 100644 --- a/src/pipelines/shared/postprocessing/postprocess.py +++ b/src/pipelines/shared/postprocessing/postprocess.py @@ -1,3 +1,7 @@ +import json + +import pandas as pd + import src.config as config from src.constants.constants import Constants from src.constants.investment_columns import FIELD_FORMAT_MAPPING @@ -5,6 +9,35 @@ from src.pipelines.shared.postprocessing import postprocessing_funcs def postprocess(df, constants: Constants): + """ + Postprocess the extracted data into CC (Contract Config) and optionally Dashboard format. + + Dashboard postprocessing is skipped unless config.RUN_DASHBOARD_POSTPROCESSING is True. + + Args: + df: Raw extracted data DataFrame + constants: Constants object + + Returns: + tuple: (cc_df, dashboard_df) - CC-formatted and Dashboard-formatted DataFrames. + dashboard_df is None when RUN_DASHBOARD_POSTPROCESSING is False. + """ + # Standard Postprocessing Steps for All Outputs + df = standard_postprocess(df, constants) + + # Postprocessing Steps unique to Contract Config Output + cc_df = contract_config_postprocess(df, constants) + + # Postprocessing Steps unique to Dashboard Output (optional, off by default) + if config.RUN_DASHBOARD_POSTPROCESSING: + dashboard_df = dashboard_postprocess(df, constants) + else: + dashboard_df = None + + return cc_df, dashboard_df + + +def standard_postprocess(df, constants: Constants): if df.shape[0] == 0: return df @@ -23,9 +56,19 @@ def postprocess(df, constants: Constants): sort_columns.append("REIMB_LESSER_OF_ID") df = df.sort_values(by=sort_columns).reset_index(drop=True) - # Format rate fields + # Normalize currency fields and format rate fields if "REIMB_FEE_RATE" in df.columns: df["REIMB_FEE_RATE"] = df["REIMB_FEE_RATE"].apply( + postprocessing_funcs.normalize_currency + ) + df["REIMB_FEE_RATE"] = df["REIMB_FEE_RATE"].apply( + postprocessing_funcs.format_rate_fields_with_commas + ) + if "REIMB_CONVERSION_FACTOR" in df.columns: + df["REIMB_CONVERSION_FACTOR"] = df["REIMB_CONVERSION_FACTOR"].apply( + postprocessing_funcs.normalize_currency + ) + df["REIMB_CONVERSION_FACTOR"] = df["REIMB_CONVERSION_FACTOR"].apply( postprocessing_funcs.format_rate_fields_with_commas ) if "REIMB_PCT_RATE" in df.columns: @@ -90,4 +133,153 @@ def postprocess(df, constants: Constants): # Standardize output column order - this should ALWAYS be the final postprocessing step df = postprocessing_funcs.reorder_columns(df, FIELD_FORMAT_MAPPING) + df["FILE_NAME"] = df["FILE_NAME"].apply(postprocessing_funcs.standardize_file_name) + + # This will remove non-standard characters from every cell in the DataFrame + df = df.replace(r"[^ -~]+", "", regex=True) + + # Remove N/A values from all columns (standard cleaning for all outputs) + df = postprocessing_funcs.clean_na_values(df) + return df + + +def contract_config_postprocess(df, constants: Constants): + cc_df = df.copy() + + if cc_df.shape[0] == 0: + return cc_df + + # Contract Config specific postprocessing steps: + # Apply json formatting + cc_df["PAYER_STATE"] = cc_df["PAYER_STATE"].apply( + postprocessing_funcs.format_as_json_list + ) + cc_df["PROVIDER_STATE"] = cc_df["PROVIDER_STATE"].apply( + postprocessing_funcs.format_as_json_list + ) + cc_df["LOB"] = cc_df["LOB"].apply(postprocessing_funcs.format_as_json_list) + cc_df["AARETE_DERIVED_LOB"] = cc_df["AARETE_DERIVED_LOB"].apply( + postprocessing_funcs.format_as_json_list + ) + cc_df["NETWORK"] = cc_df["NETWORK"].apply(postprocessing_funcs.format_as_json_list) + cc_df["AARETE_DERIVED_NETWORK"] = cc_df["AARETE_DERIVED_NETWORK"].apply( + postprocessing_funcs.format_as_json_list + ) + cc_df["PRODUCT"] = cc_df["PRODUCT"].apply(postprocessing_funcs.format_as_json_list) + cc_df["AARETE_DERIVED_PRODUCT"] = cc_df["AARETE_DERIVED_PRODUCT"].apply( + postprocessing_funcs.format_as_json_list + ) + cc_df["PROGRAM"] = cc_df["PROGRAM"].apply(postprocessing_funcs.format_as_json_list) + cc_df["AARETE_DERIVED_PROGRAM"] = cc_df["AARETE_DERIVED_PROGRAM"].apply( + postprocessing_funcs.format_as_json_list + ) + cc_df["REIMB_PROV_TIN"] = cc_df["REIMB_PROV_TIN"].apply( + postprocessing_funcs.format_as_json_list + ) + cc_df["REIMB_PROV_NPI"] = cc_df["REIMB_PROV_NPI"].apply( + postprocessing_funcs.format_as_json_list + ) + cc_df["REIMB_PROV_NAME"] = cc_df["REIMB_PROV_NAME"].apply( + postprocessing_funcs.format_as_json_list + ) + cc_df["PROV_TAXONOMY_CD"] = cc_df["PROV_TAXONOMY_CD"].apply( + postprocessing_funcs.format_as_json_list + ) + cc_df["PROV_TAXONOMY_CD_DESC"] = cc_df["PROV_TAXONOMY_CD_DESC"].apply( + postprocessing_funcs.format_as_json_list + ) + cc_df["PROV_SPECIALTY_CD"] = cc_df["PROV_SPECIALTY_CD"].apply( + postprocessing_funcs.format_as_json_list + ) + cc_df["PROV_SPECIALTY_CD_DESC"] = cc_df["PROV_SPECIALTY_CD_DESC"].apply( + postprocessing_funcs.format_as_json_list + ) + cc_df["PLACE_OF_SERVICE_CD"] = cc_df["PLACE_OF_SERVICE_CD"].apply( + postprocessing_funcs.format_as_json_list + ) + cc_df["PLACE_OF_SERVICE_CD_DESC"] = cc_df["PLACE_OF_SERVICE_CD_DESC"].apply( + postprocessing_funcs.format_as_json_list + ) + cc_df["BILL_TYPE_CD"] = cc_df["BILL_TYPE_CD"].apply( + postprocessing_funcs.format_as_json_list + ) + cc_df["BILL_TYPE_CD_DESC"] = cc_df["BILL_TYPE_CD_DESC"].apply( + postprocessing_funcs.format_as_json_list + ) + + return cc_df + + +def dashboard_postprocess(df, constants: Constants): + dashboard_df = df.copy() + + if dashboard_df.shape[0] == 0: + return dashboard_df + + # Dashboard specific postprocessing steps can be added here + dashboard_df["PAYER_STATE"] = dashboard_df["PAYER_STATE"].apply( + lambda x: ", ".join(x) if isinstance(x, list) else x + ) + dashboard_df["PROVIDER_STATE"] = dashboard_df["PROVIDER_STATE"].apply( + lambda x: ", ".join(x) if isinstance(x, list) else x + ) + dashboard_df["LOB"] = dashboard_df["LOB"].apply( + lambda x: ", ".join(x) if isinstance(x, list) else x + ) + dashboard_df["AARETE_DERIVED_LOB"] = dashboard_df["AARETE_DERIVED_LOB"].apply( + lambda x: ", ".join(x) if isinstance(x, list) else x + ) + dashboard_df["NETWORK"] = dashboard_df["NETWORK"].apply( + lambda x: ", ".join(x) if isinstance(x, list) else x + ) + dashboard_df["AARETE_DERIVED_NETWORK"] = dashboard_df[ + "AARETE_DERIVED_NETWORK" + ].apply(lambda x: ", ".join(x) if isinstance(x, list) else x) + dashboard_df["PRODUCT"] = dashboard_df["PRODUCT"].apply( + lambda x: ", ".join(x) if isinstance(x, list) else x + ) + dashboard_df["AARETE_DERIVED_PRODUCT"] = dashboard_df[ + "AARETE_DERIVED_PRODUCT" + ].apply(lambda x: ", ".join(x) if isinstance(x, list) else x) + dashboard_df["PROGRAM"] = dashboard_df["PROGRAM"].apply( + lambda x: ", ".join(x) if isinstance(x, list) else x + ) + dashboard_df["AARETE_DERIVED_PROGRAM"] = dashboard_df[ + "AARETE_DERIVED_PROGRAM" + ].apply(lambda x: ", ".join(x) if isinstance(x, list) else x) + dashboard_df["REIMB_PROV_TIN"] = dashboard_df["REIMB_PROV_TIN"].apply( + lambda x: ", ".join(x) if isinstance(x, list) else x + ) + dashboard_df["REIMB_PROV_NPI"] = dashboard_df["REIMB_PROV_NPI"].apply( + lambda x: ", ".join(x) if isinstance(x, list) else x + ) + dashboard_df["REIMB_PROV_NAME"] = dashboard_df["REIMB_PROV_NAME"].apply( + lambda x: ", ".join(x) if isinstance(x, list) else x + ) + dashboard_df["PROV_TAXONOMY_CD"] = dashboard_df["PROV_TAXONOMY_CD"].apply( + lambda x: ", ".join(x) if isinstance(x, list) else x + ) + dashboard_df["PROV_TAXONOMY_CD_DESC"] = dashboard_df["PROV_TAXONOMY_CD_DESC"].apply( + lambda x: ", ".join(x) if isinstance(x, list) else x + ) + dashboard_df["PROV_SPECIALTY_CD"] = dashboard_df["PROV_SPECIALTY_CD"].apply( + lambda x: ", ".join(x) if isinstance(x, list) else x + ) + dashboard_df["PROV_SPECIALTY_CD_DESC"] = dashboard_df[ + "PROV_SPECIALTY_CD_DESC" + ].apply(lambda x: ", ".join(x) if isinstance(x, list) else x) + dashboard_df["PLACE_OF_SERVICE_CD"] = dashboard_df["PLACE_OF_SERVICE_CD"].apply( + lambda x: ", ".join(x) if isinstance(x, list) else x + ) + dashboard_df["PLACE_OF_SERVICE_CD_DESC"] = dashboard_df[ + "PLACE_OF_SERVICE_CD_DESC" + ].apply(lambda x: ", ".join(x) if isinstance(x, list) else x) + dashboard_df["BILL_TYPE_CD"] = dashboard_df["BILL_TYPE_CD"].apply( + lambda x: ", ".join(x) if isinstance(x, list) else x + ) + dashboard_df["BILL_TYPE_CD_DESC"] = dashboard_df["BILL_TYPE_CD_DESC"].apply( + lambda x: ", ".join(x) if isinstance(x, list) else x + ) + + return dashboard_df diff --git a/src/pipelines/shared/postprocessing/postprocessing_funcs.py b/src/pipelines/shared/postprocessing/postprocessing_funcs.py index aecfb66..8ed7a61 100644 --- a/src/pipelines/shared/postprocessing/postprocessing_funcs.py +++ b/src/pipelines/shared/postprocessing/postprocessing_funcs.py @@ -6,6 +6,7 @@ import re from datetime import datetime import pandas as pd from src.utils import string_utils +import json # Determine the base directory for the project BASE_DIR = os.path.dirname(os.path.dirname(os.path.dirname(os.path.abspath(__file__)))) @@ -40,16 +41,141 @@ def normalize_indicator_field(value: str) -> str: """ Normalize the indicator field value to 'Y' or 'N'. + Explicitly handles empty strings, None, NaN, and blank values by setting them to 'N'. + Args: value (str): The input value from an _IND field. Returns: str: 'Y' if the value is 'Y', otherwise 'N'. """ - if isinstance(value, str) and value.strip().upper() == "Y": - return "Y" + # Handle None, NaN, and empty values + if value is None or (isinstance(value, float) and pd.isna(value)): + return "N" + + # Handle empty strings and whitespace-only strings + if isinstance(value, str): + value_stripped = value.strip() + if not value_stripped: + return "N" + if value_stripped.upper() == "Y": + return "Y" + + # Everything else becomes 'N' return "N" +def clean_na_values(df: pd.DataFrame) -> pd.DataFrame: + """ + Remove cells containing ONLY "N/A", "['N/A']", "[]", "['[]']", "UNKNOWN", or similar placeholder values. + + This function removes placeholder values only when they are the sole value in a cell. + If a placeholder value is part of a list with other non-placeholder values, it is kept. + Empty list representations (e.g. "[]", "['[]']") are replaced with blank to avoid + clutter in the output. + + Args: + df (pd.DataFrame): Input DataFrame to clean. + + Returns: + pd.DataFrame: DataFrame with placeholder values removed from cells where they are the only value. + """ + if df.empty: + return df + + df_cleaned = df.copy() + + # Patterns to match placeholder values (case-insensitive) + # Include [] and ['[]'] so empty list representations are blanked instead of shown + placeholder_patterns = [ + r"^N/A$", + r"^NA$", + r"^NaN$", + r"^UNKNOWN$", + r"^None$", + r"^null$", + r"^\['N/A'\]$", + r"^\[\"N/A\"\]$", + r"^\[N/A\]$", + r"^\[\]$", # Empty list "[]" + r"^\['\[\]'\]$", # Python repr of list containing "[]" + r"^\[\"\[\]\"\]$", # JSON list containing "[]" + ] + + def is_placeholder_only(value): + """Check if value is only a placeholder.""" + # Handle None, NaN, and pd.NA + if value is None: + return True + try: + if pd.isna(value): + return True + except (TypeError, ValueError): + # pd.isna might raise TypeError for some types, continue checking + pass + + if isinstance(value, str): + value_stripped = value.strip() + if not value_stripped: + return True + + # Try to parse as JSON list first + if value_stripped.startswith("[") and value_stripped.endswith("]"): + try: + parsed_list = json.loads(value_stripped) + if isinstance(parsed_list, list): + # If list contains only placeholder values, consider it placeholder + if len(parsed_list) == 0: + return True + all_placeholders = True + for item in parsed_list: + item_str = str(item).strip() + is_placeholder = False + for pattern in placeholder_patterns: + if re.match(pattern, item_str, re.IGNORECASE): + is_placeholder = True + break + if not is_placeholder and item_str: + all_placeholders = False + break + return all_placeholders + except (json.JSONDecodeError, ValueError): + # If JSON parsing fails, continue with string matching + pass + + # Check against placeholder patterns for string values + for pattern in placeholder_patterns: + if re.match(pattern, value_stripped, re.IGNORECASE): + return True + + # Handle list types (Python lists) + if isinstance(value, list): + if len(value) == 0: + return True + # If list contains only placeholder values, consider it placeholder + all_placeholders = True + for item in value: + item_str = str(item).strip() + is_placeholder = False + for pattern in placeholder_patterns: + if re.match(pattern, item_str, re.IGNORECASE): + is_placeholder = True + break + if not is_placeholder and item_str: + all_placeholders = False + break + return all_placeholders + + return False + + # Apply cleaning to all columns + for col in df_cleaned.columns: + df_cleaned[col] = df_cleaned[col].apply( + lambda x: "" if is_placeholder_only(x) else x + ) + + return df_cleaned + + def format_rate_fields_with_commas(value: str) -> str: """ Format the rate with commas and two decimal places. @@ -69,6 +195,32 @@ def format_rate_fields_with_commas(value: str) -> str: return "" +def normalize_currency(value: str) -> str: + """ + Normalize currency-like strings by removing $ and commas. + + Args: + value (str): A string representing a currency value. + Returns: + str: Normalized numeric string or empty string if invalid. + """ + if string_utils.is_empty(value): + return "" + + if isinstance(value, (int, float)) and not pd.isna(value): + return str(value) + + if not isinstance(value, str): + return "" + + text = value.strip() + if text.upper() in {"N/A", "NA", "NAN"}: + return "" + + text = text.replace("$", "").replace(",", "").strip() + return text + + def remove_hyphens(value): """ Remove hyphens from the input value. @@ -238,7 +390,7 @@ def normalize_cpt_fields(value): else: result = [str(value).strip()] # Default case for other types - return str(result) # Return the list directly + return json.dumps(result) # Return JSON list string def generate_reimb_ids(df: pd.DataFrame) -> pd.DataFrame: @@ -509,6 +661,30 @@ def add_aarete_derived_amendment_num(df: pd.DataFrame) -> pd.DataFrame: pd.DataFrame: The updated DataFrame with the 'AARETE_DERIVED_AMENDMENT_NUM' column. """ + def derive_amendment_num(value): + if value is None or (isinstance(value, float) and pd.isna(value)): + return "0" + + if not isinstance(value, str): + value = str(value) + + text = value.strip() + if string_utils.is_empty(text): + return "0" + + lowered = text.lower() + if lowered in {"n/a", "na", "nan", "unknown", "unspecified", "none", "null"}: + return "0" + + if any(token in lowered for token in ["base", "original", "master"]): + return "N/A" + + numbers = re.findall(r"\b\d+\b", text) + if numbers: + return str(int(numbers[-1])) + + return "0" + if "CONTRACT_AMENDMENT_NUM" in df.columns: df["AARETE_DERIVED_AMENDMENT_NUM"] = ( df["CONTRACT_AMENDMENT_NUM"] @@ -517,6 +693,10 @@ def add_aarete_derived_amendment_num(df: pd.DataFrame) -> pd.DataFrame: .fillna(df["CONTRACT_AMENDMENT_NUM"]) ) + df["AARETE_DERIVED_AMENDMENT_NUM"] = df["CONTRACT_AMENDMENT_NUM"].apply( + derive_amendment_num + ) + return df @@ -861,6 +1041,24 @@ def add_aarete_derived_signatory_complete_ind(df: pd.DataFrame) -> pd.DataFrame: return df +def standardize_file_name(file_name: str) -> str: + """ + Standardizes the given file name by applying a series of regex replacements + to normalize terms related to time periods. + + Args: + file_name (str): The original file name string. + + Returns: + str: The standardized file name string. + """ + + # remove .txt from the end if present + file_name = re.sub(r"\.txt$", "", file_name, flags=re.IGNORECASE) + + return file_name + + def fill_claim_type_from_title(df: pd.DataFrame) -> pd.DataFrame: """ Fill empty AARETE_DERIVED_CLAIM_TYPE_CD values using two strategies: @@ -978,3 +1176,52 @@ def fill_claim_type_from_title(df: pd.DataFrame) -> pd.DataFrame: df["AARETE_DERIVED_CLAIM_TYPE_CD"] = df.apply(infer_from_title, axis=1) return df + + +def _is_empty_list_placeholder(item) -> bool: + """Return True if item is "[]", "['[]']", or similar empty-list representation.""" + if item is None or (isinstance(item, float) and pd.isna(item)): + return True + s = str(item).strip() + return s in ("[]", "['[]']", '["[]"]') or s == "" + + +def format_as_json_list(val): + """ + Format a value as a JSON list string. Returns blank for empty lists + instead of "[]" to avoid clutter in output. + """ + # 1. Handle actual lists or Nulls + if isinstance(val, list): + filtered = [ + str(i).strip() + for i in val + if i is not None and not _is_empty_list_placeholder(i) + ] + return json.dumps(filtered) if filtered else "" + if pd.isna(val) or val == "" or val == "[]": + return "" + + # 2. Cleanup & Parsing + text = str(val).strip() + + # Try to parse as JSON if it looks like a list + if text.startswith("[") and text.endswith("]"): + try: + parsed = json.loads(text) + if isinstance(parsed, list): + filtered = [ + str(i).strip() + for i in parsed + if i is not None and not _is_empty_list_placeholder(i) + ] + return json.dumps(filtered) if filtered else "" + except (json.JSONDecodeError, ValueError): + text = text.strip("[]") # Strip brackets if it's a "fake" list like ['A'] + + # 3. Handle comma-separated strings & Clean special characters + # This regex keeps letters, numbers, commas, and spaces + clean_text = re.sub(r"[^a-zA-Z0-9, ]+", "", text) + items = [i.strip() for i in clean_text.split(",") if i.strip()] + + return json.dumps(items) if items else "" diff --git a/src/tests/test_io_utils.py b/src/tests/test_io_utils.py index a94b427..ec5040a 100644 --- a/src/tests/test_io_utils.py +++ b/src/tests/test_io_utils.py @@ -14,6 +14,8 @@ from src.utils.io_utils import ( read_s3_csv, read_xlsb, remove_txt_extension, + split_dataframe_by_filename, + write_local, write_s3, ) @@ -373,7 +375,7 @@ class TestIOUtils: call_kwargs = mock_s3_client.put_object.call_args[1] assert ( call_kwargs["Key"] - == "test_batch_123/20240101120000/test_batch_123-ERRORS.csv" + == "test_batch_123/20240101120000/full_outputs/test_batch_123-ERRORS.csv" ) def test_write_s3_usage(self, mocker): @@ -516,6 +518,263 @@ class TestIOUtils: mock_to_csv.assert_called_once() call_args = mock_to_csv.call_args[0][0] - # Check output path includes outputs/qc_qa folder and correct filename - assert "outputs/qc_qa" in call_args + # Check output path includes automation_qa-qc folder and correct filename + assert "automation_qa-qc" in call_args assert "QC-QA-STATS.csv" in call_args + + # Tests for splitting logic + def test_split_dataframe_by_filename_empty(self): + """Test split_dataframe_by_filename with empty DataFrame.""" + df = pd.DataFrame() + result = split_dataframe_by_filename(df, 70000) + assert len(result) == 1 + assert result[0].empty + + def test_split_dataframe_by_filename_no_file_name_column(self): + """Test split_dataframe_by_filename raises error without FILE_NAME column.""" + df = pd.DataFrame({"col1": [1, 2], "col2": [3, 4]}) + with pytest.raises(ValueError, match="FILE_NAME"): + split_dataframe_by_filename(df, 70000) + + def test_split_dataframe_by_filename_single_file_small(self): + """Test split_dataframe_by_filename with single file under limit.""" + df = pd.DataFrame( + { + "FILE_NAME": ["file1.txt"] * 10, + "col1": range(10), + "col2": range(10, 20), + } + ) + result = split_dataframe_by_filename(df, 70000) + assert len(result) == 1 + assert len(result[0]) == 10 + assert all(result[0]["FILE_NAME"] == "file1.txt") + + def test_split_dataframe_by_filename_single_file_large(self): + """Test split_dataframe_by_filename with single file over limit.""" + # Create a DataFrame with one file that exceeds the limit + df = pd.DataFrame( + { + "FILE_NAME": ["large_file.txt"] * 80000, + "col1": range(80000), + } + ) + result = split_dataframe_by_filename(df, 70000) + # Should still be one split since we can't break up a filename + assert len(result) == 1 + assert len(result[0]) == 80000 + + def test_split_dataframe_by_filename_multiple_files_sorted(self): + """Test split_dataframe_by_filename sorts by row count ascending.""" + # Create files with different row counts + # small.txt: 10 rows total (5 + 5) + # medium.txt: 40 rows total (20 + 20) + # large.txt: 50 rows + df = pd.DataFrame( + { + "FILE_NAME": ( + ["small.txt"] * 5 + + ["medium.txt"] * 20 + + ["large.txt"] * 50 + + ["small.txt"] * 5 + + ["medium.txt"] * 20 + ), + "col1": range(100), + } + ) + result = split_dataframe_by_filename(df, 30) + # With max_rows=30: + # - small (10) fits in first split + # - medium (40) alone exceeds 30, so new split + # - large (50) alone exceeds 30, so new split + # Result: 3 splits + assert len(result) == 3 + # First split should have small (10 rows) + assert len(result[0]) == 10 + assert set(result[0]["FILE_NAME"].unique()) == {"small.txt"} + # Second split should have medium (40 rows) + assert len(result[1]) == 40 + assert set(result[1]["FILE_NAME"].unique()) == {"medium.txt"} + # Third split should have large (50 rows) + assert len(result[2]) == 50 + assert set(result[2]["FILE_NAME"].unique()) == {"large.txt"} + + def test_split_dataframe_by_filename_multiple_splits(self): + """Test split_dataframe_by_filename creates multiple splits correctly.""" + # Create multiple files that will need splitting + df = pd.DataFrame( + { + "FILE_NAME": ( + ["file1.txt"] * 40000 + + ["file2.txt"] * 40000 + + ["file3.txt"] * 40000 + ), + "col1": range(120000), + } + ) + result = split_dataframe_by_filename(df, 70000) + # Each file is 40k, so: file1 (40k) fits, file2 (40k) would make 80k > 70k, so separate + # So we should get 3 splits + assert len(result) == 3 + assert all(len(split) == 40000 for split in result) + + def test_write_local_cc_results_full_single_split(self, mocker, tmp_path): + """Test write_local with cc_results_full that doesn't need splitting.""" + mocker.patch("src.config.BATCH_ID", "test_batch") + mocker.patch("src.config.CONSOLIDATED_OUTPUT_DIRECTORY", str(tmp_path)) + mocker.patch("src.config.PERFORM_PARENT_CHILD_MAPPING", False) + mocker.patch("src.config.MAX_ROWS_PER_SPLIT", 70000) + + df = pd.DataFrame( + { + "FILE_NAME": ["file1.txt"] * 10, + "col1": range(10), + } + ) + result = write_local(df, "", "run_20240101_10-00_test", "cc_results_full") + + # Should create FULL file + expected_file = ( + tmp_path + / "run_20240101_10-00_test" + / "full_outputs" + / "cc_results" + / "test_batch-RESULTS-FULL.csv" + ) + assert expected_file.exists() + assert result is None # No parent-child mapping + + def test_write_local_cc_results_full_multiple_splits(self, mocker, tmp_path): + """Test write_local with cc_results_full that needs splitting.""" + mocker.patch("src.config.BATCH_ID", "test_batch") + mocker.patch("src.config.CONSOLIDATED_OUTPUT_DIRECTORY", str(tmp_path)) + mocker.patch("src.config.PERFORM_PARENT_CHILD_MAPPING", True) + mocker.patch("src.config.MAX_ROWS_PER_SPLIT", 30) + + # Create data that will need splitting + df = pd.DataFrame( + { + "FILE_NAME": ( + ["file1.txt"] * 20 + ["file2.txt"] * 20 + ["file3.txt"] * 20 + ), + "col1": range(60), + } + ) + result = write_local(df, "", "run_20240101_10-00_test", "cc_results_full") + + # Should create SPLIT files + output_dir = ( + tmp_path / "run_20240101_10-00_test" / "full_outputs" / "cc_results" + ) + split1_file = output_dir / "test_batch-RESULTS-SPLIT1.csv" + split2_file = output_dir / "test_batch-RESULTS-SPLIT2.csv" + + assert split1_file.exists() + assert split2_file.exists() + # Should return first split for parent-child mapping + assert result == str(split1_file) + + def test_write_s3_cc_results_full_single_split(self, mocker): + """Test write_s3 with cc_results_full that doesn't need splitting.""" + mock_s3_client = mocker.MagicMock() + mocker.patch("src.config.S3_CLIENT", mock_s3_client) + mocker.patch("src.config.BATCH_ID", "test_batch") + mocker.patch("src.config.S3_OUTPUT_BUCKET", "test-bucket") + mocker.patch("src.config.PERFORM_PARENT_CHILD_MAPPING", False) + mocker.patch("src.config.MAX_ROWS_PER_SPLIT", 70000) + mocker.patch("src.utils.io_utils.logging") + + df = pd.DataFrame( + { + "FILE_NAME": ["file1.txt"] * 10, + "col1": range(10), + } + ) + write_s3(df, "", "run_20240101_10-00_test", "cc_results_full") + + # Should upload FULL file + mock_s3_client.put_object.assert_called_once() + call_kwargs = mock_s3_client.put_object.call_args[1] + assert "full_outputs/cc_results" in call_kwargs["Key"] + assert "RESULTS-FULL.csv" in call_kwargs["Key"] + + def test_write_s3_cc_results_full_multiple_splits(self, mocker): + """Test write_s3 with cc_results_full that needs splitting.""" + mock_s3_client = mocker.MagicMock() + mocker.patch("src.config.S3_CLIENT", mock_s3_client) + mocker.patch("src.config.BATCH_ID", "test_batch") + mocker.patch("src.config.S3_OUTPUT_BUCKET", "test-bucket") + mocker.patch("src.config.PERFORM_PARENT_CHILD_MAPPING", True) + mocker.patch("src.config.MAX_ROWS_PER_SPLIT", 30) + mocker.patch("src.utils.io_utils.logging") + + # Create data that will need splitting + df = pd.DataFrame( + { + "FILE_NAME": ( + ["file1.txt"] * 20 + ["file2.txt"] * 20 + ["file3.txt"] * 20 + ), + "col1": range(60), + } + ) + result = write_s3(df, "", "run_20240101_10-00_test", "cc_results_full") + + # Should upload multiple SPLIT files + assert mock_s3_client.put_object.call_count >= 2 + # Check that SPLIT files were uploaded + call_keys = [ + call[1]["Key"] for call in mock_s3_client.put_object.call_args_list + ] + assert any("SPLIT1" in key for key in call_keys) + assert any("SPLIT2" in key for key in call_keys) + # Should return first split path for parent-child mapping + assert result is not None + assert "SPLIT1" in result + + def test_write_s3_dashboard_results_full_splits(self, mocker): + """Test write_s3 with dashboard_results_full that needs splitting.""" + mock_s3_client = mocker.MagicMock() + mocker.patch("src.config.S3_CLIENT", mock_s3_client) + mocker.patch("src.config.BATCH_ID", "test_batch") + mocker.patch("src.config.S3_OUTPUT_BUCKET", "test-bucket") + mocker.patch("src.config.MAX_ROWS_PER_SPLIT", 30) + mocker.patch("src.utils.io_utils.logging") + + df = pd.DataFrame( + { + "FILE_NAME": (["file1.txt"] * 20 + ["file2.txt"] * 20), + "col1": range(40), + } + ) + write_s3(df, "", "run_20240101_10-00_test", "dashboard_results_full") + + # Should upload multiple SPLIT files + assert mock_s3_client.put_object.call_count >= 2 + call_keys = [ + call[1]["Key"] for call in mock_s3_client.put_object.call_args_list + ] + assert any("dashboard-SPLIT" in key for key in call_keys) + + def test_write_s3_qc_qa_cc_full_splits(self, mocker): + """Test write_s3 with qc_qa_cc_full that needs splitting.""" + mock_s3_client = mocker.MagicMock() + mocker.patch("src.config.S3_CLIENT", mock_s3_client) + mocker.patch("src.config.BATCH_ID", "test_batch") + mocker.patch("src.config.S3_OUTPUT_BUCKET", "test-bucket") + mocker.patch("src.config.MAX_ROWS_PER_SPLIT", 30) + mocker.patch("src.utils.io_utils.logging") + + df = pd.DataFrame( + { + "FILE_NAME": (["file1.txt"] * 20 + ["file2.txt"] * 20), + "col1": range(40), + } + ) + write_s3(df, "", "run_20240101_10-00_test", "qc_qa_cc_full") + + # Should upload multiple SPLIT files + assert mock_s3_client.put_object.call_count >= 2 + call_keys = [ + call[1]["Key"] for call in mock_s3_client.put_object.call_args_list + ] + assert any("QC-QA-SPLIT" in key for key in call_keys) diff --git a/src/tests/test_postprocess.py b/src/tests/test_postprocess.py index e5e1e29..e5a5c3d 100644 --- a/src/tests/test_postprocess.py +++ b/src/tests/test_postprocess.py @@ -1,12 +1,17 @@ +import json +import os +import tempfile import unittest from unittest.mock import MagicMock, patch import pandas as pd import pytest from src.pipelines.shared.postprocessing.postprocessing_funcs import ( + clean_na_values, deduplicate_provider_columns, fill_claim_type_from_title, flatten_singleton_string_list, + format_as_json_list, format_rate_fields_with_commas, normalize_auto_renewal_term, normalize_cpt_fields, @@ -17,16 +22,33 @@ from src.pipelines.shared.postprocessing.postprocessing_funcs import ( rename_columns, validate_and_reformat_date, ) +from src.pipelines.shared.postprocessing.postprocess import standard_postprocess +from src.constants.constants import Constants class TestPostprocessFunctions(unittest.TestCase): def test_normalize_indicator_field(self): + """Test normalize_indicator_field with various inputs including blanks/None/NaN.""" self.assertEqual(normalize_indicator_field("Y"), "Y") self.assertEqual(normalize_indicator_field("y"), "Y") self.assertEqual(normalize_indicator_field("N"), "N") self.assertEqual(normalize_indicator_field(""), "N") self.assertEqual(normalize_indicator_field(None), "N") + # Test with whitespace-only strings + self.assertEqual(normalize_indicator_field(" "), "N") + self.assertEqual(normalize_indicator_field("\t\n"), "N") + # Test with NaN (float) + import numpy as np + + self.assertEqual(normalize_indicator_field(np.nan), "N") + self.assertEqual(normalize_indicator_field(float("nan")), "N") + # Test with pd.NA + self.assertEqual(normalize_indicator_field(pd.NA), "N") + # Test that non-Y values become N + self.assertEqual(normalize_indicator_field("X"), "N") + self.assertEqual(normalize_indicator_field("Yes"), "N") + self.assertEqual(normalize_indicator_field("1"), "N") def test_format_rate_fields_with_commas(self): # Test numeric values @@ -48,6 +70,17 @@ class TestPostprocessFunctions(unittest.TestCase): self.assertEqual(flatten_singleton_string_list("invalid"), "invalid") self.assertEqual(flatten_singleton_string_list(None), "") + def test_format_as_json_list_returns_blank_for_empty(self): + """Test format_as_json_list returns blank instead of [] for empty lists.""" + self.assertEqual(format_as_json_list([]), "") + self.assertEqual(format_as_json_list("[]"), "") + self.assertEqual(format_as_json_list(None), "") + self.assertEqual(format_as_json_list(""), "") + self.assertEqual(format_as_json_list([""]), "") + self.assertEqual(format_as_json_list(["[]"]), "") + self.assertEqual(format_as_json_list(["Valid"]), '["Valid"]') + self.assertEqual(format_as_json_list(["A", "B"]), '["A", "B"]') + def test_rename_columns(self): """Tests the rename_columns function to ensure it correctly renames specified columns. @@ -108,15 +141,15 @@ class TestPostprocessFunctions(unittest.TestCase): self.assertEqual(normalize_auto_renewal_term(None), "") def test_normalize_cpt_fields(self): - self.assertEqual(normalize_cpt_fields("[123, 456]"), "['123', '456']") - self.assertEqual(normalize_cpt_fields("123-456"), "['123-456']") + self.assertEqual(normalize_cpt_fields("[123, 456]"), '["123", "456"]') + self.assertEqual(normalize_cpt_fields("123-456"), '["123-456"]') self.assertEqual( normalize_cpt_fields("['T0000-T9999, S0000-S9999']"), - "['T0000-T9999', 'S0000-S9999']", + '["T0000-T9999", "S0000-S9999"]', ) - self.assertEqual(normalize_cpt_fields("123"), "['123']") + self.assertEqual(normalize_cpt_fields("123"), '["123"]') self.assertEqual(normalize_cpt_fields(None), "") - self.assertEqual(normalize_cpt_fields(123), "['123']") + self.assertEqual(normalize_cpt_fields(123), '["123"]') def test_process_patient_age_range(self): """Tests the process_patient_age_range function with various age range formats. @@ -662,6 +695,496 @@ class TestPostprocessFunctions(unittest.TestCase): # Empty list should be treated as empty and filled with mode or inferred assert result_df.loc[0, "AARETE_DERIVED_CLAIM_TYPE_CD"] == "M" + def test_clean_na_values_string_placeholders(self): + """Test clean_na_values removes placeholder strings when they're the only value.""" + # Test with string placeholders + input_df = pd.DataFrame( + { + "FIELD1": ["N/A", "Valid Value", "UNKNOWN", "NA"], + "FIELD2": ["Valid", "N/A", "Another Valid", "null"], + } + ) + + result_df = clean_na_values(input_df) + + # N/A, UNKNOWN, NA, null should be removed (become empty strings) + self.assertEqual(result_df.loc[0, "FIELD1"], "") + self.assertEqual(result_df.loc[1, "FIELD1"], "Valid Value") + self.assertEqual(result_df.loc[2, "FIELD1"], "") + self.assertEqual(result_df.loc[3, "FIELD1"], "") + + self.assertEqual(result_df.loc[0, "FIELD2"], "Valid") + self.assertEqual(result_df.loc[1, "FIELD2"], "") + self.assertEqual(result_df.loc[2, "FIELD2"], "Another Valid") + self.assertEqual(result_df.loc[3, "FIELD2"], "") + + def test_clean_na_values_json_list_placeholders(self): + """Test clean_na_values handles JSON list placeholders.""" + # Test with JSON string lists + input_df = pd.DataFrame( + { + "FIELD1": ['["N/A"]', '["Medicare", "N/A"]', '["N/A", "N/A"]'], + "FIELD2": ['["Valid"]', '["N/A"]', '["Medicare"]'], + } + ) + + result_df = clean_na_values(input_df) + + # '["N/A"]' should be removed (only placeholder) + self.assertEqual(result_df.loc[0, "FIELD1"], "") + # '["Medicare", "N/A"]' should be kept (has non-placeholder) + self.assertEqual(result_df.loc[1, "FIELD1"], '["Medicare", "N/A"]') + # '["N/A", "N/A"]' should be removed (all placeholders) + self.assertEqual(result_df.loc[2, "FIELD1"], "") + + self.assertEqual(result_df.loc[0, "FIELD2"], '["Valid"]') + self.assertEqual(result_df.loc[1, "FIELD2"], "") + self.assertEqual(result_df.loc[2, "FIELD2"], '["Medicare"]') + + def test_clean_na_values_empty_list_representations(self): + """Test clean_na_values blanks [] and ['[]'] instead of showing them.""" + input_df = pd.DataFrame( + { + "FIELD1": ["[]", "['[]']", '["[]"]', "Valid"], + "FIELD2": [["[]"], ["Valid"], [], "[]"], + } + ) + + result_df = clean_na_values(input_df) + + self.assertEqual(result_df.loc[0, "FIELD1"], "") + self.assertEqual(result_df.loc[1, "FIELD1"], "") + self.assertEqual(result_df.loc[2, "FIELD1"], "") + self.assertEqual(result_df.loc[3, "FIELD1"], "Valid") + + self.assertEqual(result_df.loc[0, "FIELD2"], "") + self.assertEqual(result_df.loc[1, "FIELD2"], ["Valid"]) + self.assertEqual(result_df.loc[2, "FIELD2"], "") + self.assertEqual(result_df.loc[3, "FIELD2"], "") + + def test_clean_na_values_python_lists(self): + """Test clean_na_values handles Python list types.""" + # Test with Python lists + input_df = pd.DataFrame( + { + "FIELD1": [["N/A"], ["Medicare", "N/A"], ["N/A", "N/A"]], + "FIELD2": [["Valid"], ["N/A"], ["Medicare"]], + } + ) + + result_df = clean_na_values(input_df) + + # ["N/A"] should be removed + self.assertEqual(result_df.loc[0, "FIELD1"], "") + # ["Medicare", "N/A"] should be kept + self.assertEqual(result_df.loc[1, "FIELD1"], ["Medicare", "N/A"]) + # ["N/A", "N/A"] should be removed + self.assertEqual(result_df.loc[2, "FIELD1"], "") + + self.assertEqual(result_df.loc[0, "FIELD2"], ["Valid"]) + self.assertEqual(result_df.loc[1, "FIELD2"], "") + self.assertEqual(result_df.loc[2, "FIELD2"], ["Medicare"]) + + def test_clean_na_values_none_and_nan(self): + """Test clean_na_values handles None and NaN values.""" + import numpy as np + + input_df = pd.DataFrame( + { + "FIELD1": [None, np.nan, pd.NA, "Valid"], + "FIELD2": ["", " ", "Valid", None], + } + ) + + result_df = clean_na_values(input_df) + + # None, NaN, pd.NA, empty strings should be removed (converted to empty strings) + self.assertEqual(result_df.loc[0, "FIELD1"], "") + self.assertEqual(result_df.loc[1, "FIELD1"], "") + # pd.NA should be converted to empty string + result_val = result_df.loc[2, "FIELD1"] + self.assertTrue( + result_val == "" or pd.isna(result_val), + f"Expected empty string or NA, got: {result_val}", + ) + self.assertEqual(result_df.loc[3, "FIELD1"], "Valid") + + self.assertEqual(result_df.loc[0, "FIELD2"], "") + self.assertEqual(result_df.loc[1, "FIELD2"], "") + self.assertEqual(result_df.loc[2, "FIELD2"], "Valid") + self.assertEqual(result_df.loc[3, "FIELD2"], "") + + def test_clean_na_values_empty_dataframe(self): + """Test clean_na_values handles empty DataFrame.""" + empty_df = pd.DataFrame() + result_df = clean_na_values(empty_df) + pd.testing.assert_frame_equal(result_df, empty_df) + + def test_clean_na_values_case_insensitive(self): + """Test clean_na_values is case-insensitive for placeholder matching.""" + input_df = pd.DataFrame( + { + "FIELD1": ["n/a", "N/A", "Na", "unknown", "UNKNOWN", "None", "NULL"], + } + ) + + result_df = clean_na_values(input_df) + + # All should be removed (case-insensitive matching) + for idx in range(len(result_df)): + self.assertEqual(result_df.loc[idx, "FIELD1"], "") + + def test_standard_postprocess_cleans_na_and_normalizes_ind(self): + """Test standard_postprocess removes N/A values and normalizes _IND fields.""" + constants = Constants() + input_df = pd.DataFrame( + { + "FILE_NAME": ["test1.pdf", "test2.pdf"], + "PAYER_NAME": [ + "N/A", + "Valid Value", + ], # Use actual column from FIELD_FORMAT_MAPPING + "AUTO_RENEWAL_IND": ["Y", ""], + "DSH_IND": [None, "N"], + "IME_IND": [" ", "y"], + "NETWORK": ['["N/A"]', '["Medicare"]'], + } + ) + + result_df = standard_postprocess(input_df, constants) + + # N/A should be removed (PAYER_NAME is in FIELD_FORMAT_MAPPING so it will be preserved) + self.assertEqual(result_df.loc[0, "PAYER_NAME"], "") + self.assertEqual(result_df.loc[1, "PAYER_NAME"], "Valid Value") + + # _IND fields should be normalized to Y or N + self.assertEqual(result_df.loc[0, "AUTO_RENEWAL_IND"], "Y") + self.assertEqual(result_df.loc[1, "AUTO_RENEWAL_IND"], "N") # Empty becomes N + + self.assertEqual(result_df.loc[0, "DSH_IND"], "N") # None becomes N + self.assertEqual(result_df.loc[1, "DSH_IND"], "N") + + self.assertEqual(result_df.loc[0, "IME_IND"], "N") # Whitespace becomes N + self.assertEqual(result_df.loc[1, "IME_IND"], "Y") # "y" becomes "Y" + + # JSON list with only N/A should be removed + self.assertEqual(result_df.loc[0, "NETWORK"], "") + # JSON list with valid value should be kept + self.assertEqual(result_df.loc[1, "NETWORK"], '["Medicare"]') + + def test_standard_postprocess_empty_dataframe(self): + """Test standard_postprocess handles empty DataFrame.""" + constants = Constants() + empty_df = pd.DataFrame() + result_df = standard_postprocess(empty_df, constants) + pd.testing.assert_frame_equal(result_df, empty_df) + + def test_standard_postprocess_cleans_na_without_ind_fields(self): + """Test standard_postprocess cleans N/A values even when no _IND fields are present.""" + constants = Constants() + input_df = pd.DataFrame( + { + "FILE_NAME": ["test1.pdf", "test2.pdf"], + "PAYER_NAME": [ + "N/A", + "Valid", + ], # Use actual column from FIELD_FORMAT_MAPPING + "CONTRACT_TITLE": [ + "UNKNOWN", + "Another Valid", + ], # Use actual column from FIELD_FORMAT_MAPPING + } + ) + + result_df = standard_postprocess(input_df, constants) + + # Should still clean N/A values (these columns are in FIELD_FORMAT_MAPPING so they will be preserved) + self.assertEqual(result_df.loc[0, "PAYER_NAME"], "") + self.assertEqual(result_df.loc[1, "PAYER_NAME"], "Valid") + self.assertEqual(result_df.loc[0, "CONTRACT_TITLE"], "") + self.assertEqual(result_df.loc[1, "CONTRACT_TITLE"], "Another Valid") + + +class TestOutputFileStructure(unittest.TestCase): + """Test that the new output file structure is generated correctly.""" + + def setUp(self): + """Create temporary directory for outputs.""" + self.temp_dir = tempfile.mkdtemp() + self.run_timestamp = "run_20250106_10-30_test" + self.batch_id = "test_batch" + + def tearDown(self): + """Clean up temporary directory.""" + import shutil + + shutil.rmtree(self.temp_dir, ignore_errors=True) + + @patch("src.utils.io_utils.config") + def test_write_local_cc_results_full_structure(self, mock_config): + """Test that cc_results_full creates correct directory structure.""" + mock_config.BATCH_ID = "test_batch" + mock_config.CONSOLIDATED_OUTPUT_DIRECTORY = self.temp_dir + mock_config.PERFORM_PARENT_CHILD_MAPPING = False + mock_config.MAX_ROWS_PER_SPLIT = 70000 + + from src.utils.io_utils import write_local + + # Add FILE_NAME column required for splitting logic + test_df = pd.DataFrame( + {"FILE_NAME": ["file1.txt", "file2.txt"], "col1": [1, 2], "col2": [3, 4]} + ) + + write_local(test_df, "", self.run_timestamp, "cc_results_full") + + # Check directory structure was created + expected_dir = os.path.join( + self.temp_dir, self.run_timestamp, "full_outputs", "cc_results" + ) + self.assertTrue(os.path.exists(expected_dir)) + + # Check file was created + expected_file = os.path.join(expected_dir, "test_batch-RESULTS-FULL.csv") + self.assertTrue(os.path.exists(expected_file)) + + @patch("src.utils.io_utils.config") + def test_write_local_dashboard_results_full_structure(self, mock_config): + """Test that dashboard_results_full creates correct directory structure.""" + mock_config.BATCH_ID = "test_batch" + mock_config.CONSOLIDATED_OUTPUT_DIRECTORY = self.temp_dir + mock_config.PERFORM_PARENT_CHILD_MAPPING = False + mock_config.MAX_ROWS_PER_SPLIT = 70000 + + from src.utils.io_utils import write_local + + # Add FILE_NAME column required for splitting logic + test_df = pd.DataFrame( + {"FILE_NAME": ["file1.txt", "file2.txt"], "col1": [1, 2], "col2": [3, 4]} + ) + + write_local(test_df, "", self.run_timestamp, "dashboard_results_full") + + # Check directory structure was created + expected_dir = os.path.join( + self.temp_dir, self.run_timestamp, "full_outputs", "dashboard_results" + ) + self.assertTrue(os.path.exists(expected_dir)) + + # Check file was created + expected_file = os.path.join(expected_dir, "test_batch-RESULTS-dashboard.csv") + self.assertTrue(os.path.exists(expected_file)) + + @patch("src.utils.io_utils.config") + def test_write_local_error_in_full_outputs(self, mock_config): + """Test that error files go to full_outputs/ directory.""" + mock_config.BATCH_ID = "test_batch" + mock_config.CONSOLIDATED_OUTPUT_DIRECTORY = self.temp_dir + mock_config.PERFORM_PARENT_CHILD_MAPPING = False + + from src.utils.io_utils import write_local + + test_df = pd.DataFrame({"error": ["Error 1", "Error 2"]}) + + write_local(test_df, "", self.run_timestamp, "error") + + # Check directory structure was created + expected_dir = os.path.join(self.temp_dir, self.run_timestamp, "full_outputs") + self.assertTrue(os.path.exists(expected_dir)) + + # Check file was created + expected_file = os.path.join(expected_dir, "test_batch-ERRORS.csv") + self.assertTrue(os.path.exists(expected_file)) + + @patch("src.utils.io_utils.config") + def test_write_local_qc_qa_cc_full_structure(self, mock_config): + """Test that qc_qa_cc_full creates correct directory structure.""" + mock_config.BATCH_ID = "test_batch" + mock_config.CONSOLIDATED_OUTPUT_DIRECTORY = self.temp_dir + mock_config.PERFORM_PARENT_CHILD_MAPPING = False + mock_config.MAX_ROWS_PER_SPLIT = 70000 + + from src.utils.io_utils import write_local + + # Add FILE_NAME column required for splitting logic + test_df = pd.DataFrame( + {"FILE_NAME": ["file1.txt", "file2.txt"], "col1": [1, 2], "col2": [3, 4]} + ) + + write_local(test_df, "", self.run_timestamp, "qc_qa_cc_full") + + # Check directory structure was created + expected_dir = os.path.join( + self.temp_dir, self.run_timestamp, "automation_qa-qc" + ) + self.assertTrue(os.path.exists(expected_dir)) + + # Check file was created + expected_file = os.path.join(expected_dir, "test_batch-RESULTS-QC-QA-FULL.csv") + self.assertTrue(os.path.exists(expected_file)) + + @patch("src.utils.io_utils.config") + def test_write_local_qc_qa_stats_structure(self, mock_config): + """Test that qc_qa_stats creates correct directory structure.""" + mock_config.BATCH_ID = "test_batch" + mock_config.CONSOLIDATED_OUTPUT_DIRECTORY = self.temp_dir + mock_config.PERFORM_PARENT_CHILD_MAPPING = False + + from src.utils.io_utils import write_local + + test_df = pd.DataFrame({"stat": ["value1", "value2"]}) + + write_local(test_df, "", self.run_timestamp, "qc_qa_stats") + + # Check directory structure was created + expected_dir = os.path.join( + self.temp_dir, self.run_timestamp, "automation_qa-qc" + ) + self.assertTrue(os.path.exists(expected_dir)) + + # Check file was created + expected_file = os.path.join(expected_dir, "test_batch-QC-QA-STATS.csv") + self.assertTrue(os.path.exists(expected_file)) + + @patch("src.utils.io_utils.config") + def test_write_local_parent_child_structure(self, mock_config): + """Test that parent_child creates correct directory structure and file path.""" + mock_config.BATCH_ID = "test_batch" + mock_config.CONSOLIDATED_OUTPUT_DIRECTORY = self.temp_dir + mock_config.PERFORM_PARENT_CHILD_MAPPING = False + + from src.utils.io_utils import write_local + + test_df = pd.DataFrame({"col1": [1, 2]}) + + # Mock to_csv on the DataFrame instance to avoid actual file I/O + test_df.to_csv = MagicMock() + + write_local(test_df, "", self.run_timestamp, "parent_child") + + # Verify to_csv was called on the DataFrame + test_df.to_csv.assert_called_once() + # Verify the call had index=False and quoting=1 + call_kwargs = test_df.to_csv.call_args[1] + self.assertEqual(call_kwargs.get("index"), False) + self.assertEqual(call_kwargs.get("quoting"), 1) + + # Verify the file path contains the expected components + call_args = test_df.to_csv.call_args[0] + file_path = call_args[0] if call_args else None + self.assertIsNotNone(file_path, "to_csv should be called with a file path") + self.assertIn( + "parent-child", + file_path, + f"File path should contain 'parent-child', got: {file_path}", + ) + self.assertIn( + "test_batch-PC.csv", + file_path, + f"File path should contain 'test_batch-PC.csv', got: {file_path}", + ) + + # Verify the directory structure would be correct + expected_dir = os.path.join(self.temp_dir, self.run_timestamp, "parent-child") + self.assertIn( + expected_dir, + file_path, + f"File path should contain expected directory {expected_dir}, got: {file_path}", + ) + + @patch("src.utils.io_utils.config") + @patch("src.utils.io_utils.logging") + def test_write_s3_cc_results_full_path(self, mock_logging, mock_config): + """Test that write_s3 uses correct S3 path for cc_results_full.""" + mock_config.BATCH_ID = "test_batch" + mock_config.S3_OUTPUT_BUCKET = "test-bucket" + mock_config.S3_CLIENT = MagicMock() + mock_config.S3_CLIENT.put_object = MagicMock() + mock_config.PERFORM_PARENT_CHILD_MAPPING = False + mock_config.MAX_ROWS_PER_SPLIT = 70000 + + from src.utils.io_utils import write_s3 + + # Add FILE_NAME column required for splitting logic + test_df = pd.DataFrame( + {"FILE_NAME": ["file1.txt", "file2.txt"], "col1": [1, 2]} + ) + + write_s3(test_df, "", self.run_timestamp, "cc_results_full") + + # Verify S3 put_object was called + mock_config.S3_CLIENT.put_object.assert_called_once() + + # Verify the S3 key/path + call_args = mock_config.S3_CLIENT.put_object.call_args + s3_key = call_args[1]["Key"] + expected_key = f"test_batch/{self.run_timestamp}/full_outputs/cc_results/test_batch-RESULTS-FULL.csv" + self.assertEqual(s3_key, expected_key) + + @patch("src.utils.io_utils.config") + @patch("src.utils.io_utils.logging") + def test_write_s3_dashboard_results_full_path(self, mock_logging, mock_config): + """Test that write_s3 uses correct S3 path for dashboard_results_full.""" + mock_config.BATCH_ID = "test_batch" + mock_config.S3_OUTPUT_BUCKET = "test-bucket" + mock_config.S3_CLIENT = MagicMock() + mock_config.S3_CLIENT.put_object = MagicMock() + mock_config.PERFORM_PARENT_CHILD_MAPPING = False + mock_config.MAX_ROWS_PER_SPLIT = 70000 + + from src.utils.io_utils import write_s3 + + # Add FILE_NAME column required for splitting logic + test_df = pd.DataFrame( + {"FILE_NAME": ["file1.txt", "file2.txt"], "col1": [1, 2]} + ) + + write_s3(test_df, "", self.run_timestamp, "dashboard_results_full") + + # Verify S3 put_object was called + mock_config.S3_CLIENT.put_object.assert_called_once() + + # Verify the S3 key/path + call_args = mock_config.S3_CLIENT.put_object.call_args + s3_key = call_args[1]["Key"] + expected_key = f"test_batch/{self.run_timestamp}/full_outputs/dashboard_results/test_batch-RESULTS-dashboard.csv" + self.assertEqual(s3_key, expected_key) + + @patch("src.utils.io_utils.config") + @patch("src.utils.io_utils.logging") + def test_write_s3_qc_qa_paths(self, mock_logging, mock_config): + """Test that write_s3 uses correct S3 paths for QC/QA outputs.""" + mock_config.BATCH_ID = "test_batch" + mock_config.S3_OUTPUT_BUCKET = "test-bucket" + mock_config.S3_CLIENT = MagicMock() + mock_config.S3_CLIENT.put_object = MagicMock() + mock_config.PERFORM_PARENT_CHILD_MAPPING = False + mock_config.MAX_ROWS_PER_SPLIT = 70000 + + from src.utils.io_utils import write_s3 + + # Add FILE_NAME column required for splitting logic + test_df = pd.DataFrame( + {"FILE_NAME": ["file1.txt", "file2.txt"], "col1": [1, 2]} + ) + + # Test qc_qa_cc_full + write_s3(test_df, "", self.run_timestamp, "qc_qa_cc_full") + call_args = mock_config.S3_CLIENT.put_object.call_args + s3_key = call_args[1]["Key"] + expected_key = f"test_batch/{self.run_timestamp}/automation_qa-qc/test_batch-RESULTS-QC-QA-FULL.csv" + self.assertEqual(s3_key, expected_key) + + # Test qc_qa_stats + mock_config.S3_CLIENT.put_object.reset_mock() + test_df_stats = pd.DataFrame( + {"col1": [1, 2]} + ) # qc_qa_stats doesn't need FILE_NAME + write_s3(test_df_stats, "", self.run_timestamp, "qc_qa_stats") + call_args = mock_config.S3_CLIENT.put_object.call_args + s3_key = call_args[1]["Key"] + expected_key = f"test_batch/{self.run_timestamp}/automation_qa-qc/test_batch-QC-QA-STATS.csv" + self.assertEqual(s3_key, expected_key) + if __name__ == "__main__": unittest.main() diff --git a/src/utils/io_utils.py b/src/utils/io_utils.py index 739f0dd..00eff0b 100644 --- a/src/utils/io_utils.py +++ b/src/utils/io_utils.py @@ -338,6 +338,163 @@ def read_xlsb( return df +def split_dataframe_by_filename( + df: pd.DataFrame, max_rows_per_split: int = 70000 +) -> list[pd.DataFrame]: + """ + Split a DataFrame into multiple DataFrames based on filename groups. + + This function ensures that all rows from the same filename stay together. + It sorts filenames by row count (ascending) and creates splits that are as + close to max_rows_per_split as possible without breaking up filename groups. + + Args: + df (pd.DataFrame): The DataFrame to split. Must have a 'FILE_NAME' column. + max_rows_per_split (int): Maximum number of rows per split file. Default: 70000. + + Returns: + list[pd.DataFrame]: List of DataFrames, each representing a split file. + + Raises: + ValueError: If DataFrame doesn't have 'FILE_NAME' column or is empty. + """ + if df.empty: + return [df] + + if "FILE_NAME" not in df.columns: + raise ValueError("DataFrame must have a 'FILE_NAME' column for splitting") + + # Count rows per filename + filename_counts = df["FILE_NAME"].value_counts().sort_values(ascending=True) + filename_order = filename_counts.index.tolist() + + # Create list to store split DataFrames + splits = [] + current_split_rows = [] + current_split_count = 0 + + for filename in filename_order: + filename_df = df[df["FILE_NAME"] == filename] + filename_row_count = len(filename_df) + + # If adding this filename would exceed the limit, start a new split + if ( + current_split_count + filename_row_count > max_rows_per_split + and current_split_rows + ): + # Create and save current split + current_split = pd.concat(current_split_rows, ignore_index=True) + splits.append(current_split) + # Start new split with current filename + current_split_rows = [filename_df] + current_split_count = filename_row_count + else: + # Add filename to current split + current_split_rows.append(filename_df) + current_split_count += filename_row_count + + # Add the last split if it has any rows + if current_split_rows: + final_split = pd.concat(current_split_rows, ignore_index=True) + splits.append(final_split) + + return splits if splits else [df] + + +def _write_local_split_files( + df: pd.DataFrame, + output_dir: str, + full_filename: str, + split_filename_template: str, + return_path_for_pc: bool = False, +) -> Optional[str]: + """ + Helper function to write split files locally. + + Args: + df: DataFrame to write (will be split if needed) + output_dir: Full output directory path + full_filename: Filename for single file (e.g., "{BATCH_ID}-RESULTS-FULL.csv") + split_filename_template: Template for split files (e.g., "{BATCH_ID}-RESULTS-SPLIT{idx}.csv") + return_path_for_pc: Whether to return path for parent-child mapping + + Returns: + Full file path string if return_path_for_pc is True, None otherwise + """ + os.makedirs(output_dir, exist_ok=True) + splits = split_dataframe_by_filename(df, config.MAX_ROWS_PER_SPLIT) + + if len(splits) == 1: + output_path = os.path.join(output_dir, full_filename) + splits[0].to_csv(output_path, index=False, quoting=1) + return output_path if return_path_for_pc else None + else: + # Multiple splits - write numbered files + for idx, split_df in enumerate(splits, start=1): + split_filename = split_filename_template.format(idx=idx) + split_output_path = os.path.join(output_dir, split_filename) + split_df.to_csv(split_output_path, index=False, quoting=1) + # For parent-child mapping, use the first split file + if return_path_for_pc: + first_split_filename = split_filename_template.format(idx=1) + return os.path.join(output_dir, first_split_filename) + return None + + +def _write_s3_split_files( + df: pd.DataFrame, + run_timestamp: str, + base_path: str, + full_filename: str, + split_filename_template: str, + return_path_for_pc: bool = False, +) -> Optional[str]: + """ + Helper function to write split files to S3. + + Args: + df: DataFrame to write (will be split if needed) + run_timestamp: Timestamp string for path + base_path: Base S3 path (e.g., "full_outputs/cc_results") + full_filename: Filename for single file (e.g., "{BATCH_ID}-RESULTS-FULL.csv") + split_filename_template: Template for split files (e.g., "{BATCH_ID}-RESULTS-SPLIT{idx}.csv") + return_path_for_pc: Whether to return path for parent-child mapping + + Returns: + S3 path string if return_path_for_pc is True, None otherwise + """ + splits = split_dataframe_by_filename(df, config.MAX_ROWS_PER_SPLIT) + + if len(splits) == 1: + output_path = f"{config.BATCH_ID}/{run_timestamp}/{base_path}/{full_filename}" + csv_buffer = splits[0].to_csv(index=False).encode() + config.S3_CLIENT.put_object( + Bucket=config.S3_OUTPUT_BUCKET, Key=output_path, Body=csv_buffer + ) + logging.info(f"Saved {base_path.split('/')[-1]} full file: {output_path}") + return output_path if return_path_for_pc else None + else: + # Multiple splits - upload each one + for idx, split_df in enumerate(splits, start=1): + split_filename = split_filename_template.format(idx=idx) + split_output_path = ( + f"{config.BATCH_ID}/{run_timestamp}/{base_path}/{split_filename}" + ) + csv_buffer = split_df.to_csv(index=False).encode() + config.S3_CLIENT.put_object( + Bucket=config.S3_OUTPUT_BUCKET, Key=split_output_path, Body=csv_buffer + ) + logging.info( + f"Saved {base_path.split('/')[-1]} split files ({len(splits)} files)" + ) + if return_path_for_pc: + first_split_filename = split_filename_template.format(idx=1) + return ( + f"{config.BATCH_ID}/{run_timestamp}/{base_path}/{first_split_filename}" + ) + return None + + def write_local( df: pd.DataFrame, filename: str, run_timestamp: str, output_type: str ) -> Optional[str]: @@ -386,18 +543,12 @@ def write_local( config.OUTPUT_DIRECTORY, f"{base_filename}_-RESULTS.csv" ) - return None - elif output_type == "parent_child": - output_dir = os.path.join(config.CONSOLIDATED_OUTPUT_DIRECTORY, run_timestamp) - os.makedirs(output_dir, exist_ok=True) - df.to_csv( - os.path.join(output_dir, f"{config.BATCH_ID}-Parent_Child.csv"), - index=False, - quoting=1, - ) return None elif output_type == "error": - output_dir = os.path.join(config.CONSOLIDATED_OUTPUT_DIRECTORY, run_timestamp) + # Error file goes in full_outputs/ folder + output_dir = os.path.join( + config.CONSOLIDATED_OUTPUT_DIRECTORY, run_timestamp, "full_outputs" + ) os.makedirs(output_dir, exist_ok=True) df.to_csv( os.path.join(output_dir, f"{config.BATCH_ID}-ERRORS.csv"), @@ -407,6 +558,74 @@ def write_local( if config.PERFORM_PARENT_CHILD_MAPPING: return os.path.join(output_dir, f"{config.BATCH_ID}-RESULTS.csv") return None + elif output_type == "cc_results_full": + # Full CC results (may be split into multiple files) + output_dir = os.path.join( + config.CONSOLIDATED_OUTPUT_DIRECTORY, + run_timestamp, + "full_outputs", + "cc_results", + ) + return _write_local_split_files( + df=df, + output_dir=output_dir, + full_filename=f"{config.BATCH_ID}-RESULTS-FULL.csv", + split_filename_template=f"{config.BATCH_ID}-RESULTS-SPLIT{{idx}}.csv", + return_path_for_pc=config.PERFORM_PARENT_CHILD_MAPPING, + ) + elif output_type == "dashboard_results_full": + # Full dashboard results (may be split into multiple files) + output_dir = os.path.join( + config.CONSOLIDATED_OUTPUT_DIRECTORY, + run_timestamp, + "full_outputs", + "dashboard_results", + ) + _write_local_split_files( + df=df, + output_dir=output_dir, + full_filename=f"{config.BATCH_ID}-RESULTS-dashboard.csv", + split_filename_template=f"{config.BATCH_ID}-RESULTS-dashboard-SPLIT{{idx}}.csv", + return_path_for_pc=False, + ) + return None + elif output_type == "qc_qa_cc_full": + # QC/QA CC full results (may be split into multiple files) + output_dir = os.path.join( + config.CONSOLIDATED_OUTPUT_DIRECTORY, run_timestamp, "automation_qa-qc" + ) + _write_local_split_files( + df=df, + output_dir=output_dir, + full_filename=f"{config.BATCH_ID}-RESULTS-QC-QA-FULL.csv", + split_filename_template=f"{config.BATCH_ID}-RESULTS-QC-QA-SPLIT{{idx}}.csv", + return_path_for_pc=False, + ) + return None + elif output_type == "qc_qa_error": + # QC/QA error results + output_dir = os.path.join( + config.CONSOLIDATED_OUTPUT_DIRECTORY, run_timestamp, "automation_qa-qc" + ) + os.makedirs(output_dir, exist_ok=True) + df.to_csv( + os.path.join(output_dir, f"{config.BATCH_ID}-ERRORS-QC-QA.csv"), + index=False, + quoting=1, + ) + return None + elif output_type == "parent_child": + # Parent-child output + output_dir = os.path.join( + config.CONSOLIDATED_OUTPUT_DIRECTORY, run_timestamp, "parent-child" + ) + os.makedirs(output_dir, exist_ok=True) + df.to_csv( + os.path.join(output_dir, f"{config.BATCH_ID}-PC.csv"), + index=False, + quoting=1, + ) + return None elif output_type == "usage": output_dir = os.path.join( config.CONSOLIDATED_OUTPUT_DIRECTORY, run_timestamp, "tracking" @@ -458,8 +677,10 @@ def write_local( ) return None elif output_type == "qc_qa_stats": - # QC/QA statistics output goes to outputs/qc_qa folder - output_dir = os.path.join(config.QC_QA_OUTPUT_DIRECTORY, run_timestamp) + # QC/QA statistics output goes to automation_qa-qc folder + output_dir = os.path.join( + config.CONSOLIDATED_OUTPUT_DIRECTORY, run_timestamp, "automation_qa-qc" + ) os.makedirs(output_dir, exist_ok=True) df.to_csv( os.path.join(output_dir, f"{config.BATCH_ID}-QC-QA-STATS.csv"), @@ -467,9 +688,22 @@ def write_local( quoting=1, ) return None + elif output_type == "individual_cc": + # Individual CC results + base_filename = os.path.splitext(filename)[0].strip() + output_dir = os.path.join( + config.CONSOLIDATED_OUTPUT_DIRECTORY, run_timestamp, "individual" + ) + os.makedirs(output_dir, exist_ok=True) + df.to_csv( + os.path.join(output_dir, f"{base_filename}-RESULTS-CC.csv"), + index=False, + quoting=1, + ) + return None else: logging.error( - f"Unknown output_type: {output_type}. Valid types are: 'final', 'individual', 'error', 'usage', 'usage_summary', 'dtc', 'dtc_summary', 'qc_qa_validated', 'qc_qa_stats'" + f"Unknown output_type: {output_type}. Valid types are: 'final', 'individual', 'error', 'usage', 'usage_summary', 'dtc', 'dtc_summary', 'qc_qa_validated', 'qc_qa_stats', 'cc_results_full', 'dashboard_results_full', 'qc_qa_cc_full', 'qc_qa_error', 'parent_child', 'individual_cc'" ) return None @@ -507,42 +741,144 @@ def write_s3( """ if output_type == "final": output_path = f"{config.BATCH_ID}/{run_timestamp}/{config.BATCH_ID}-RESULTS.csv" + csv_buffer = df.to_csv(index=False).encode() + try: + config.S3_CLIENT.put_object( + Bucket=config.S3_OUTPUT_BUCKET, Key=output_path, Body=csv_buffer + ) + except ClientError as e: + logging.error(f"Error uploading to S3: {e}") + return None elif output_type == "individual": output_path = ( f"{config.BATCH_ID}/{run_timestamp}/individual/{filename}-RESULTS.csv" ) + csv_buffer = df.to_csv(index=False).encode() + config.S3_CLIENT.put_object( + Bucket=config.S3_OUTPUT_BUCKET, Key=output_path, Body=csv_buffer + ) + elif output_type == "individual_cc": + output_path = ( + f"{config.BATCH_ID}/{run_timestamp}/individual/{filename}-RESULTS-CC.csv" + ) + csv_buffer = df.to_csv(index=False).encode() + config.S3_CLIENT.put_object( + Bucket=config.S3_OUTPUT_BUCKET, Key=output_path, Body=csv_buffer + ) elif output_type == "parent_child": output_path = ( - f"{config.BATCH_ID}/{run_timestamp}/parent_child/{filename}-RESULTS.csv" + f"{config.BATCH_ID}/{run_timestamp}/parent-child/{config.BATCH_ID}-PC.csv" + ) + csv_buffer = df.to_csv(index=False).encode() + config.S3_CLIENT.put_object( + Bucket=config.S3_OUTPUT_BUCKET, Key=output_path, Body=csv_buffer ) elif output_type == "error": - output_path = f"{config.BATCH_ID}/{run_timestamp}/{config.BATCH_ID}-ERRORS.csv" + output_path = f"{config.BATCH_ID}/{run_timestamp}/full_outputs/{config.BATCH_ID}-ERRORS.csv" + csv_buffer = df.to_csv(index=False).encode() + try: + config.S3_CLIENT.put_object( + Bucket=config.S3_OUTPUT_BUCKET, Key=output_path, Body=csv_buffer + ) + except ClientError as e: + logging.error(f"Error uploading to S3: {e}") + return None + elif output_type in ("cc_results_full", "dashboard_results_full", "qc_qa_cc_full"): + # Will be handled in try block using helper function + output_path = None + parent_child_path = None if output_type == "cc_results_full" else None + elif output_type == "qc_qa_error": + output_path = f"{config.BATCH_ID}/{run_timestamp}/automation_qa-qc/{config.BATCH_ID}-ERRORS-QC-QA.csv" + csv_buffer = df.to_csv(index=False).encode() + config.S3_CLIENT.put_object( + Bucket=config.S3_OUTPUT_BUCKET, Key=output_path, Body=csv_buffer + ) elif output_type == "usage": output_path = ( f"{config.BATCH_ID}/{run_timestamp}/tracking/{config.BATCH_ID}-USAGE.csv" ) + csv_buffer = df.to_csv(index=False).encode() + config.S3_CLIENT.put_object( + Bucket=config.S3_OUTPUT_BUCKET, Key=output_path, Body=csv_buffer + ) elif output_type == "usage_summary": output_path = f"{config.BATCH_ID}/{run_timestamp}/tracking/{config.BATCH_ID}-USAGE-SUMMARY.csv" + csv_buffer = df.to_csv(index=False).encode() + config.S3_CLIENT.put_object( + Bucket=config.S3_OUTPUT_BUCKET, Key=output_path, Body=csv_buffer + ) elif output_type == "dtc": output_path = f"{config.BATCH_ID}/{run_timestamp}/{config.BATCH_ID}-DTC.csv" + csv_buffer = df.to_csv(index=False).encode() + config.S3_CLIENT.put_object( + Bucket=config.S3_OUTPUT_BUCKET, Key=output_path, Body=csv_buffer + ) elif output_type == "dtc_summary": output_path = ( f"{config.BATCH_ID}/{run_timestamp}/{config.BATCH_ID}-DTC-SUMMARY.csv" ) - elif output_type == "qc_qa_validated": - output_path = f"{config.BATCH_ID}/{run_timestamp}/automation_qa-qc/{config.BATCH_ID}-QC-QA-VALIDATED.csv" - elif output_type == "qc_qa_stats": - output_path = f"{config.BATCH_ID}/{run_timestamp}/automation_qa-qc/{config.BATCH_ID}-QC-QA-STATS.csv" - else: - # Fallback path for unknown output types - output_path = f"{config.BATCH_ID}/{run_timestamp}/{config.BATCH_ID}-unknown-{output_type}.csv" - - csv_buffer = df.to_csv(index=False).encode() - try: + csv_buffer = df.to_csv(index=False).encode() config.S3_CLIENT.put_object( Bucket=config.S3_OUTPUT_BUCKET, Key=output_path, Body=csv_buffer ) - if output_type == "final": + elif output_type == "qc_qa_validated": + output_path = f"{config.BATCH_ID}/{run_timestamp}/automation_qa-qc/{config.BATCH_ID}-QC-QA-VALIDATED.csv" + csv_buffer = df.to_csv(index=False).encode() + config.S3_CLIENT.put_object( + Bucket=config.S3_OUTPUT_BUCKET, Key=output_path, Body=csv_buffer + ) + elif output_type == "qc_qa_stats": + output_path = f"{config.BATCH_ID}/{run_timestamp}/automation_qa-qc/{config.BATCH_ID}-QC-QA-STATS.csv" + csv_buffer = df.to_csv(index=False).encode() + config.S3_CLIENT.put_object( + Bucket=config.S3_OUTPUT_BUCKET, Key=output_path, Body=csv_buffer + ) + elif output_type == "individual_cc": + output_path = ( + f"{config.BATCH_ID}/{run_timestamp}/individual/{filename}-RESULTS-CC.csv" + ) + csv_buffer = df.to_csv(index=False).encode() + config.S3_CLIENT.put_object( + Bucket=config.S3_OUTPUT_BUCKET, Key=output_path, Body=csv_buffer + ) + else: + # Fallback path for unknown output types + output_path = f"{config.BATCH_ID}/{run_timestamp}/{config.BATCH_ID}-unknown-{output_type}.csv" + csv_buffer = df.to_csv(index=False).encode() + config.S3_CLIENT.put_object( + Bucket=config.S3_OUTPUT_BUCKET, Key=output_path, Body=csv_buffer + ) + + try: + # Handle split files for cc_results_full, dashboard_results_full, and qc_qa_cc_full + if output_type == "cc_results_full": + parent_child_path = _write_s3_split_files( + df=df, + run_timestamp=run_timestamp, + base_path="full_outputs/cc_results", + full_filename=f"{config.BATCH_ID}-RESULTS-FULL.csv", + split_filename_template=f"{config.BATCH_ID}-RESULTS-SPLIT{{idx}}.csv", + return_path_for_pc=config.PERFORM_PARENT_CHILD_MAPPING, + ) + elif output_type == "dashboard_results_full": + _write_s3_split_files( + df=df, + run_timestamp=run_timestamp, + base_path="full_outputs/dashboard_results", + full_filename=f"{config.BATCH_ID}-RESULTS-dashboard.csv", + split_filename_template=f"{config.BATCH_ID}-RESULTS-dashboard-SPLIT{{idx}}.csv", + return_path_for_pc=False, + ) + elif output_type == "qc_qa_cc_full": + _write_s3_split_files( + df=df, + run_timestamp=run_timestamp, + base_path="automation_qa-qc", + full_filename=f"{config.BATCH_ID}-RESULTS-QC-QA-FULL.csv", + split_filename_template=f"{config.BATCH_ID}-RESULTS-QC-QA-SPLIT{{idx}}.csv", + return_path_for_pc=False, + ) + elif output_type == "final": logging.info(f"Saved batch output file: {output_path}") elif output_type == "individual": logging.info(f"Saved individual output file: {output_path}") @@ -562,11 +898,16 @@ def write_s3( logging.info(f"Saved QC/QA validated data file: {output_path}") elif output_type == "qc_qa_stats": logging.info(f"Saved QC/QA statistics file: {output_path}") + elif output_type == "qc_qa_error": + logging.info(f"Saved QC/QA error file: {output_path}") + elif output_type == "individual_cc": + logging.info(f"Saved individual CC file: {output_path}") else: logging.info(f"Saved unknown output type file: {output_path}") - if config.PERFORM_PARENT_CHILD_MAPPING: - return f"s3://{config.S3_OUTPUT_BUCKET}/{output_path}" + if config.PERFORM_PARENT_CHILD_MAPPING and output_type == "cc_results_full": + if "parent_child_path" in locals() and parent_child_path: + return f"s3://{config.S3_OUTPUT_BUCKET}/{parent_child_path}" return None except ClientError as e: @@ -574,6 +915,51 @@ def write_s3( return None +def upload_local_file_to_s3( + local_path: str, + run_timestamp: str, + s3_subpath: str, + content_type: Optional[str] = None, +) -> Optional[str]: + """ + Upload a local file to S3 using the same path structure as other pipeline outputs. + + Args: + local_path: Full path to the local file. + run_timestamp: Run timestamp (e.g. run_20260212_13-53_test_batch). + s3_subpath: Subpath under run_timestamp (e.g. parent-child). + content_type: Optional MIME type. Inferred from extension if not provided. + + Returns: + S3 URI on success, None on failure. + """ + if not os.path.isfile(local_path): + logging.error(f"Cannot upload: file not found: {local_path}") + return None + filename = os.path.basename(local_path) + s3_key = f"{config.BATCH_ID}/{run_timestamp}/{s3_subpath}/{filename}" + if content_type is None and filename.lower().endswith(".xlsx"): + content_type = ( + "application/vnd.openxmlformats-officedocument.spreadsheetml.sheet" + ) + try: + with open(local_path, "rb") as f: + put_kwargs = { + "Bucket": config.S3_OUTPUT_BUCKET, + "Key": s3_key, + "Body": f.read(), + } + if content_type: + put_kwargs["ContentType"] = content_type + config.S3_CLIENT.put_object(**put_kwargs) + s3_uri = f"s3://{config.S3_OUTPUT_BUCKET}/{s3_key}" + logging.info(f"Uploaded file to S3: {s3_uri}") + return s3_uri + except ClientError as e: + logging.error(f"Failed to upload {local_path} to S3: {e}") + return None + + def save_result_to_json(filename, result, json_folder): """Save a result dictionary to an individual JSON file in the specified folder.