Merged in feature/new_output_format (pull request #879)
Feature/new output format
* bugfix
* Merge branch 'DEV' into Optimize/DAIP2-1474-restructure-postprocess
* requested postprocessing changes
* prompt changes reverted
* fix pipeline issues
* fix pipeline issues
* fix pipeline issues
* fixed formatting
* fixed formatting
* Merge branch 'DEV' into Optimize/DAIP2-1474-restructure-postprocess
* Merge branch 'DEV' into Optimize/DAIP2-1474-restructure-postprocess
* save dashboard and cc output separately
* save dashboard output in s3
* pipeline error fixed
* json list through postprocessing
* Merged DEV into Optimize/DAIP2-1474-restructure-postprocess
* Merge remote-tracking branch 'origin/Optimize/DAIP2-1474-restructure-postprocess' into feature/new_output_format
* Restructure output file organization and add standard field sanitization
Output Structure Changes:
- Reorganize output files into hierarchical directory structure:
- full_outputs/cc_results/ for consolidated CC results
- full_outputs/dashboard_results/ for consolidated dashboard results
- full_outputs/ for error files
- automation_qa-qc/ for QC/QA validated results and statistics
- parent-child/ for parent-child mapping outputs
- tracking/ for usage and cost tracking data
- individual/ for per-file CC results (dashboard individual files removed)
- Update file naming conventions to match new structure
- Remove QC/QA processing for error files (error files are saved without validation)
Post-Processing Changes:
- Add standard N/A value cleaning: remove placeholder values (N/A, UNKNOWN, etc.)
when they are the only value in a cell (applies before CC/dashboard split)
- Normalize all _IND fields to contain only 'Y' or 'N' values (no blanks)
- Ensure standard cleaning runs before splitting into CC …
* Ran Black for CI
* Refactor file splitting logic and add comprehensive tests
- Refactored splitting logic in io_utils.py:
- Consolidated repeated splitting code into two focused helper functions:
- _write_local_split_files() for local file writing with splitting
- _write_s3_split_files() for S3 file writing with splitting
- Both helpers use shared split_dataframe_by_filename() function
- Added MAX_ROWS_PER_SPLIT configuration (default: 70000) in config.py
- Added comprehensive test coverage for splitting logic:
- Tests for split_dataframe_by_filename() with various scenarios
- Tests for local and S3 write operations with single and multiple splits
- Tests for cc_results_full, dashboard_results_full, and qc_qa_cc_full output types
- Fixed existing test failures (write_s3 error handling, path assertions)
- Improved code maintainability and readability
* Fix failing tests in test_postprocess.py
- Updated standard_postprocess tests to use actual columns from FIELD_FORMAT_MAPPING
(PAYER_NAME, CONTRACT_TITLE) instead of custom test columns that get dropped
- Added FILE_NAME column to all file structure test DataFrames (required for splitting logic)
- Added MAX_ROWS_PER_SPLIT mock configuration for splitting tests
- Fixed patch decorators for S3 tests to properly mock logging
All 41 tests now passing.
* Black for CI
* Blank [] and ['[]'] in output instead of displaying them
- Add placeholder patterns in clean_na_values for [], ['[]'], ["[]"]
- Update format_as_json_list to return blank for empty lists instead of []
- Filter out empty-list placeholder items from list values in format_as_json_list
- Add tests for clean_na_values empty list handling and format_as_json_list
Co-authored-by: Cursor <cursoragent@cursor.com>
* Merged DEV into feature/new_output_format
* Pipeline config, parent-child, dashboard, and runner fixes
- Parent-child: enable by default, write to run directory, S3 upload via io_utils
- Dashboard: optional (CC only by default), run_dashboard=True to enable
- Add io_utils.upload_local_file_to_s3 for centralized file uploads
- Parent-child returns (row_count, local_path); pipeline handles S3 upload
- Add TODO in config for WRITE_PC_TO_S3 removal after approval
- Fix indentation errors in runner.py
Co-authored-by: Cursor <cursoragent@cursor.com>
Approved-by: Katon Minhas
This commit is contained in:
Praneel Panchigar
committed by
Katon Minhas
Katon Minhas
parent
70215bbc69
commit
637d2dea1f
@@ -108,3 +108,4 @@ TEST*.xml
|
||||
# PRD Documents
|
||||
*.prd
|
||||
*prd.md
|
||||
.git.instructions.md
|
||||
|
||||
+12
-1
@@ -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"
|
||||
|
||||
@@ -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")
|
||||
|
||||
@@ -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(
|
||||
|
||||
@@ -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(
|
||||
|
||||
+131
-40
@@ -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__":
|
||||
|
||||
@@ -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(
|
||||
|
||||
+120
-46
@@ -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__":
|
||||
|
||||
@@ -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
|
||||
|
||||
@@ -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 ""
|
||||
|
||||
+262
-3
@@ -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)
|
||||
|
||||
@@ -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()
|
||||
|
||||
+414
-28
@@ -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.
|
||||
|
||||
|
||||
Reference in New Issue
Block a user