From 4f53b5286edec3f43cd2e45c686d0ca79120057d Mon Sep 17 00:00:00 2001 From: ppanchigar Date: Thu, 16 Apr 2026 15:49:44 -0500 Subject: [PATCH] Revert premature merge of bugfix/retire_stale_client_file_processing PR #959 was merged into dev without approval. This reverts commits 5e143c10, 63f32c41, 849aa626, and 927abcae to restore dev to its pre-merge state. The changes will be re-submitted via a new PR after proper review. --- src/config.py | 8 - .../clients/bcbs_promise/file_processing.py | 520 +++++++++++++++++ .../bcbs_promise/prompts/prompt_calls.py | 2 +- src/pipelines/clients/chc/file_processing.py | 46 ++ .../clients/clover/file_processing.py | 524 ++++++++++++++++++ .../clients/clover/prompts/prompt_calls.py | 4 +- src/pipelines/runner.py | 410 +++++--------- src/pipelines/saas/file_processing.py | 52 -- src/pipelines/saas/main.py | 474 +++++++++++++++- .../postprocessing/postprocessing_funcs.py | 4 - 10 files changed, 1701 insertions(+), 343 deletions(-) create mode 100644 src/pipelines/clients/bcbs_promise/file_processing.py create mode 100644 src/pipelines/clients/chc/file_processing.py create mode 100644 src/pipelines/clients/clover/file_processing.py diff --git a/src/config.py b/src/config.py index d4b55f6..6e816fc 100644 --- a/src/config.py +++ b/src/config.py @@ -160,14 +160,6 @@ CLOVER_PROMPTS_PATH = "src/prompts/clover_prompts.json" BCBS_PROMISE_PROMPTS_PATH = "src/prompts/bcbs_promise_prompts.json" CHC_PROMPTS_PATH = "src/prompts/chc_prompts.json" -# Maps client name → path to its extra one-to-one prompt JSON. -# Used by run_one_to_one_prompts to load client-specific fields alongside SaaS fields. -CLIENT_PROMPTS_MAP = { - "clover": CLOVER_PROMPTS_PATH, - "bcbs_promise": BCBS_PROMISE_PROMPTS_PATH, - "chc": CHC_PROMPTS_PATH, -} - RAG_RETRIEVAL_QUESTIONS_PATH = "src/prompts/rag_retrieval_questions.json" # Client-specific args diff --git a/src/pipelines/clients/bcbs_promise/file_processing.py b/src/pipelines/clients/bcbs_promise/file_processing.py new file mode 100644 index 0000000..74a98b7 --- /dev/null +++ b/src/pipelines/clients/bcbs_promise/file_processing.py @@ -0,0 +1,520 @@ +import concurrent.futures +import logging +from typing import TYPE_CHECKING, Optional + +import pandas as pd +import src.codes.code_funcs as code_funcs +from src.pipelines.shared.preprocessing import ( + preprocessing_funcs, + preprocess, + hybrid_smart_chunking_funcs, +) +from src.pipelines.shared.postprocessing import ( + aarete_derived, + postprocess, + postprocessing_funcs, +) +from src.pipelines.shared.extraction import ( + dynamic_funcs, + one_to_n_funcs, + one_to_one_funcs, + row_funcs, + tin_npi_funcs, + exhibit_funcs, +) +from src.pipelines.shared.extraction.exhibit_funcs import Exhibit +from src.utils import io_utils, logging_utils, string_utils, timing_utils +from src.constants.constants import Constants +from src import config +from src.prompts.fieldset import FieldSet +from src.utils.string_utils import datetime_str + +if TYPE_CHECKING: + from src.pipelines.shared.extraction.page_funcs import Page + + +def process_file(file_object, constants: Constants, run_timestamp): + filename, contract_text = file_object + + # Set per-file logging context: + # With this, all logging calls in this thread will now route to logs/{filename}.log + # This includes logging from all called functions (preprocess, one_to_n_funcs, etc.) + logging_utils.set_current_file(filename) + logging.info(f"{datetime_str()} Processing {filename}...") + + # Set default values + dynamic_one_to_one_fields = FieldSet() + process_one_to_n = config.FIELDS in ["all", "one_to_n"] + process_one_to_one = config.FIELDS in ["all", "one_to_one"] + # Initialize default fallback for final results + final_results = pd.DataFrame([{"FILE_NAME": filename}]) + + ################## PREPROCESS ################## + with timing_utils.timed_block("preprocess", context=filename): + contract_text = preprocess.clean_text(contract_text) + text_dict, top_sheet_dict = preprocess.split_text(contract_text) + text_dict, header, footer = preprocess.find_headers_and_footers(text_dict) + + # ONE TO N PROCESSING + one_to_n_results = pd.DataFrame() # Initialize empty DataFrame for one_to_n_results + if process_one_to_n: + # Use split_text_with_pages() to get Page objects with table splitting + # Note: split_text_with_pages() internally calls split_text() which applies headers/footers + # But we need to apply headers/footers first, so we'll create pages_dict from the cleaned text_dict + from src.pipelines.shared.extraction.page_funcs import Page + from src.pipelines.shared.preprocessing import preprocessing_funcs as prep_funcs + + # Create pages_dict from text_dict (headers/footers already applied) + pages_dict = prep_funcs.split_large_tables(text_dict) + + with timing_utils.timed_block("one_to_n_exhibit_chunking", context=filename): + # Use pages_dict for exhibit chunking (preferred) or fall back to text_dict + exhibit_chunk_mapping, all_exhibit_headers = ( + preprocess.one_to_n_exhibit_chunking( + pages_dict=pages_dict, + text_dict=text_dict, + EXHIBIT_HEADER_MARKERS=constants.EXHIBIT_HEADER_MARKERS, + filename=filename, + ) + ) + logging.info(f"{datetime_str()} Preprocessing Complete - {filename}") + + one_to_n_results = pd.DataFrame([{"FILE_NAME": filename}]) # Initialize here + + if string_utils.contains_reimbursement(contract_text): + logging.info( + f"{datetime_str()} Starting One-to-N extraction for {len(exhibit_chunk_mapping)} exhibits - {filename}" + ) + with timing_utils.timed_block("one_to_n_extraction", context=filename): + ( + one_to_n_results, + dynamic_one_to_one_fields, + first_reimbursement_page, + ) = run_one_to_n_prompts( + pages_dict=pages_dict, + text_dict=text_dict, + exhibit_chunk_mapping=exhibit_chunk_mapping, + all_exhibit_headers=all_exhibit_headers, + constants=constants, + filename=filename, + ) + if not one_to_n_results.empty: + one_to_n_results["FILE_NAME"] = filename + with timing_utils.timed_block( + "one_to_n_generate_reimb_ids", context=filename + ): + one_to_n_results = postprocessing_funcs.generate_reimb_ids( + one_to_n_results + ) + logging.info(f"{datetime_str()} One to N Complete - {filename}") + else: + first_reimbursement_page = "1" + logging.info( + f"{datetime_str()} No Reimbursement Found, Skipping - {filename}" + ) + + final_results = one_to_n_results # Set as final results + else: + first_reimbursement_page = "1" + logging.info( + f"{datetime_str()} Fields not configured for One to N, Skipping - {filename}" + ) + + # ONE TO ONE PROCESSING + if process_one_to_one: + with timing_utils.timed_block("one_to_one_extraction", context=filename): + one_to_one_results = run_one_to_one_prompts( + filename, + contract_text, + text_dict, + top_sheet_dict, + dynamic_one_to_one_fields, + first_reimbursement_page, + constants, + ) + one_to_one_results["FILE_NAME"] = filename + logging.info(f"{datetime_str()} One to One Complete - {filename}") + + # Decide how to handle one_to_one results + with timing_utils.timed_block( + "merge_one_to_one_into_one_to_n", context=filename + ): + if not one_to_n_results.empty: + # BOTH processed - merge into one_to_n + final_results = row_funcs.merge_one_to_one_into_one_to_n( + one_to_n_results, one_to_one_results, constants + ) + else: + # ONLY one_to_one processed - convert dict to DataFrame + final_results = pd.DataFrame([one_to_one_results]) + else: + logging.info( + f"{datetime_str()} Fields not configured for One to One, Skipping - {filename}" + ) + + # APPLY CODES IF ONE_TO_N WAS PROCESSED + if not one_to_n_results.empty: + with timing_utils.timed_block("code_processing", context=filename): + results_with_code = code_funcs.code_breakout(final_results, constants) + final_results = code_funcs.grouper_breakout(results_with_code) + logging.info(f"{datetime_str()} Codes Complete - {filename}") + + # POSTPROCESS + with timing_utils.timed_block("postprocess", context=filename): + 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(cc_df, filename, run_timestamp, "individual_cc") + else: + io_utils.write_local(cc_df, filename, "", "individual_cc") + + logging.info(f"{datetime_str()} Writing Complete - {filename}") + + return cc_df, dashboard_df + + +def run_one_to_one_prompts( + filename: str, + contract_text: str, + text_dict: dict[str, str], + top_sheet_dict, + dynamic_one_to_one_fields: FieldSet, + first_reimbursement_page: str, + constants: Constants, +): + + ################## INITIALIZE FIELDS ################## + one_to_one_fields = FieldSet( + relationship="one_to_one", file_path=config.FIELD_JSON_PATH + ).combine(dynamic_one_to_one_fields) + + ################## INITIALIZE BCBS PROMISE FIELDS ################## + bcbs_promise_fields = FieldSet( + relationship="one_to_one", file_path=config.BCBS_PROMISE_PROMPTS_PATH + ) + # Combine bcbs_promise fields with one_to_one_fields for HSC processing + hsc_fields = one_to_one_fields.combine(bcbs_promise_fields) + + ################## RUN PROVIDER INFO ################## + with timing_utils.timed_block("one_to_one.provider_info", context=filename): + one_to_one_results = tin_npi_funcs.run_provider_info_fields( + text_dict, filename, payer_name="" + ) + + ################## RUN HYBRID SMART CHUNKED PROMPTS ################## + # RAG function loads retrieval questions internally + # hsc_fields includes both investment_prompts.json AND bcbs_promise_prompts.json fields + with timing_utils.timed_block("one_to_one.hybrid_smart_chunking", context=filename): + hybrid_smart_chunked_answers_dict = ( + hybrid_smart_chunking_funcs.run_hybrid_smart_chunked_fields( + hsc_fields, constants, contract_text, filename, text_dict + ) + ) + ################## COMBINE ANSWERS ################## + for key, value in hybrid_smart_chunked_answers_dict.items(): + if not string_utils.is_empty(value): + one_to_one_results[key] = value + elif key not in one_to_one_results: + # Add HSC's N/A if field doesn't exist yet + one_to_one_results[key] = value + + one_to_one_results = tin_npi_funcs.merge_provider_info_with_one_to_one( + one_to_one_results, filename + ) + + ################## DERIVE BCBS PROMISE FIELDS ################## + # Derive OFFSET_INDICATOR from OFFSET_TERM + offset_term = one_to_one_results.get("OFFSET_TERM", "N/A") + if not string_utils.is_empty(offset_term): + one_to_one_results["OFFSET_INDICATOR"] = "Y" + else: + one_to_one_results["OFFSET_INDICATOR"] = "N" + + ################## ADD AD FIELDS ################ + one_to_one_results = one_to_one_funcs.get_aarete_derived_dates( + one_to_one_results, text_dict, filename + ) + + ################## ADD SIGNATURE COUNT ################## + one_to_one_results = one_to_one_funcs.add_signature_count( + one_to_one_results, contract_text + ) + + ################## Crosswalk Fields ################## + one_to_one_results = aarete_derived.get_crosswalk_fields( + [one_to_one_results], constants + ) + + ################## Fill NA Mapping ################## + one_to_one_results = aarete_derived.fill_na_mapping(one_to_one_results) + + return one_to_one_results[0] + + +def process_page_reimbursements( + exhibit: Exhibit, + page_num: str, + pages_dict: Optional[dict[str, "Page"]] = None, + text_dict: Optional[dict[str, str]] = None, + exhibit_page_nums: Optional[list[str]] = None, + constants: Optional[Constants] = None, + filename: Optional[str] = None, +): + """ + STEP 1 HELPER: Extract reimbursements from a single page (without exhibit-level fields). + Runs: reimbursement_level → carveout_and_special_case → lesser_of_distribution → breakout + + Args: + exhibit: Exhibit object containing the page + page_num: Page identifier (can be "27" or "27.1" for sub-pages) + pages_dict: Dictionary mapping page numbers to Page objects (preferred) + text_dict: Dictionary mapping page numbers to text strings (for backward compatibility) + exhibit_page_nums: List of page numbers in the exhibit (for backward compatibility) + constants: Constants object + filename: Name of the file being processed + """ + # Get page text - prefer using exhibit.get_page_text() if available + if exhibit.pages is not None: + page_text = exhibit.get_page_text(page_num) + exhibit_page_nums = exhibit.exhibit_page_nums + elif pages_dict is not None: + # Handle sub-pages: "27.1" means get sub-page "1" from page "27" + if "." in page_num and not page_num.startswith("."): + base_page, sub_page = page_num.rsplit(".", 1) + if base_page in pages_dict: + page_text = pages_dict[base_page].get_text(sub_page) + else: + page_text = "" + elif page_num in pages_dict: + page_text = pages_dict[page_num].get_text() + else: + page_text = "" + exhibit_page_nums = ( + exhibit.exhibit_page_nums + if exhibit_page_nums is None + else exhibit_page_nums + ) + elif text_dict is not None: + page_text = text_dict.get(page_num, "") + exhibit_page_nums = ( + exhibit_page_nums + if exhibit_page_nums is not None + else exhibit.exhibit_page_nums + ) + else: + raise ValueError( + "Either pages_dict, text_dict, or exhibit.pages must be provided" + ) + + # Simplify exhibit text - prefer using exhibit object + exhibit_text_simplified = preprocessing_funcs.simplify_exhibit( + exhibit=exhibit, + current_page_num=page_num, + ) + + ############################### Reimbursement Primary ############################### + reimbursement_level_answers = one_to_n_funcs.reimbursement_level( + page_text, constants, filename + ) + + if not reimbursement_level_answers: + return [], [] + + ################################ Carveouts and Special Case ################################ + reimbursement_level_answers, special_case_answers = ( + one_to_n_funcs.carveout_and_special_case( + reimbursement_level_answers, constants, filename + ) + ) + + ############################### Lesser of Distribution ############################### + # Note: We run lesser_of without dynamic fields in Step 1 + reimbursement_level_answers = one_to_n_funcs.lesser_of_distribution( + reimbursement_level_answers, + exhibit_text_simplified, + page_num, + constants, + filename, + exhibit, # ← Added this parameter + ) + + ################################ Get Breakouts ############################### + reimbursement_level_answers, special_case_answers = one_to_n_funcs.breakout( + reimbursement_level_answers, + special_case_answers, + filename, + constants, + ) + + ################################ Assign REIMB_PAGE ############################### + for answer_dict in reimbursement_level_answers: + answer_dict["REIMB_PAGE"] = page_num + + return reimbursement_level_answers, special_case_answers + + +def run_one_to_n_prompts( + pages_dict: Optional[dict[str, "Page"]] = None, + text_dict: Optional[dict[str, str]] = None, + exhibit_chunk_mapping: Optional[dict[str, list[str]]] = None, + all_exhibit_headers: Optional[dict[str, str]] = None, + constants: Optional[Constants] = None, + filename: Optional[str] = None, +): + """ + 3-STEP APPROACH (per exhibit): + Step 1: Extract reimbursements for exhibit pages (parallel within exhibit) + Step 2: Run exhibit_level when exhibit has reimbursements + Step 3: Run dynamic_assignment and combine results + + Args: + pages_dict: Dictionary mapping page numbers to Page objects (preferred) + text_dict: Dictionary mapping page numbers to text strings (for backward compatibility) + exhibit_chunk_mapping: Dictionary mapping exhibit pages to lists of page numbers + all_exhibit_headers: Dictionary mapping exhibit pages to their headers + constants: Constants object + filename: Name of the file being processed + """ + if exhibit_chunk_mapping is None: + exhibit_chunk_mapping = {} + if all_exhibit_headers is None: + all_exhibit_headers = {} + + total_exhibits = len(exhibit_chunk_mapping) + logging.debug( + f"{datetime_str()} Processing {total_exhibits} exhibits with NEW 3-step approach - {filename}" + ) + + # Create Exhibit objects in order (maintains exhibit chain via prev_exhibit) + exhibits = exhibit_funcs.get_exhibit_list( + pages_dict=pages_dict, + text_dict=text_dict, + exhibit_chunk_mapping=exhibit_chunk_mapping, + all_exhibit_headers=all_exhibit_headers, + ) # <- Returns list[Exhibit] + + # Process exhibits serially; within each exhibit, process pages in parallel + total_pages = sum(len(exhibit.exhibit_page_nums) for exhibit in exhibits) + completed_pages = 0 + one_to_n_results = [] + first_reimbursement_page = "1" + + for exhibit in exhibits: + exhibit_pages = exhibit.exhibit_page_nums + if exhibit_pages: + with concurrent.futures.ThreadPoolExecutor( + max_workers=min(len(exhibit_pages), 20) + ) as executor: + page_futures = { + executor.submit( + process_page_reimbursements, + exhibit, + page_num, + pages_dict, + text_dict, + exhibit_pages, + constants, + filename, + ): page_num + for page_num in exhibit_pages + } + + for future in concurrent.futures.as_completed(page_futures): + try: + page_num = page_futures[future] + reimbursement_rows, special_case_rows = future.result() + exhibit.add_reimbursement_rows( + reimbursement_rows, special_case_rows + ) + completed_pages += 1 + if completed_pages % 20 == 0 or completed_pages == total_pages: + logging.debug( + f"{datetime_str()} Page progress: {completed_pages}/{total_pages} - {filename}" + ) + except Exception as e: + logging.error(f"Error processing page: {str(e)}") + + if not exhibit.has_reimbursements: + continue + + # STEP 2: exhibit_level for this exhibit + exhibit_level_answers, dynamic_reimbursement_fields = ( + one_to_n_funcs.exhibit_level( + exhibit.exhibit_text, + exhibit.exhibit_header, + exhibit.exhibit_page, + constants, + filename, + ) + ) + exhibit.set_exhibit_level_data( + exhibit_level_answers, dynamic_reimbursement_fields + ) + logging.debug( + f"Exhibit Level Answers for {filename}, Page {exhibit.exhibit_page}: {exhibit_level_answers}" + ) + + # STEP 3: dynamic assignment & combine for this exhibit + # Get dynamic fields from previous exhibit if needed (for future use) + if exhibit.dynamic_reimbursement_fields: + dynamic_fields = exhibit.dynamic_reimbursement_fields + elif ( + exhibit.prev_exhibit + and not exhibit.prev_exhibit.has_reimbursements + and exhibit.get_previous_exhibit_dynamic_fields().fields + ): + dynamic_fields = exhibit.get_previous_exhibit_dynamic_fields() + else: + dynamic_fields = None + + # Run dynamic assignment for ALL reimbursement rows in this exhibit + # Note: dynamic_assignment expects a list of rows and returns a list of rows + reimbursement_rows_with_dynamic = exhibit.reimbursement_rows + if exhibit.reimbursement_rows and dynamic_fields is not None: + reimbursement_rows_with_dynamic = dynamic_funcs.dynamic_assignment( + exhibit.reimbursement_rows, + dynamic_fields, + pages_dict, + text_dict, + exhibit, + constants, + filename, + ) + + # Combine reimbursement rows with exhibit-level answers + combined_rows = row_funcs.combine_one_to_n( + exhibit.exhibit_text, + reimbursement_rows_with_dynamic, + exhibit.special_case_rows, + exhibit.exhibit_level_answers, + filename, + ) + + # Run cleaning + combined_rows = one_to_n_funcs.one_to_n_cleaning( + combined_rows, exhibit.exhibit_text, constants, filename + ) + + exhibit.final_rows = combined_rows + one_to_n_results.extend(combined_rows) + + # Track first reimbursement page + if first_reimbursement_page == "1" and exhibit.has_reimbursements: + first_reimbursement_page = exhibit.exhibit_page + + logging.debug( + f"{datetime_str()} STEP 3 COMPLETE: Combined {len(one_to_n_results)} total rows - {filename}" + ) + + ################## Add N/A Dynamic or Exhibit to One-to-One ################## + dynamic_one_to_one_fields = dynamic_funcs.get_dynamic_one_to_one_fields( + one_to_n_results, constants + ) + + ################## CONVERT TO DF ################## + one_to_n_df = pd.DataFrame(one_to_n_results) + + return one_to_n_df, dynamic_one_to_one_fields, first_reimbursement_page diff --git a/src/pipelines/clients/bcbs_promise/prompts/prompt_calls.py b/src/pipelines/clients/bcbs_promise/prompts/prompt_calls.py index 41f636d..9ebeaa8 100644 --- a/src/pipelines/clients/bcbs_promise/prompts/prompt_calls.py +++ b/src/pipelines/clients/bcbs_promise/prompts/prompt_calls.py @@ -50,4 +50,4 @@ def validate_reimbursements_for_llm(answer_dict: dict[str, str], filename: str) f"LLM response for reimbursement validation in {filename}:\n{llm_response}" ) final_answer = _parser(llm_response) - return "YES" in final_answer + return final_answer == "YES" diff --git a/src/pipelines/clients/chc/file_processing.py b/src/pipelines/clients/chc/file_processing.py new file mode 100644 index 0000000..636b2b5 --- /dev/null +++ b/src/pipelines/clients/chc/file_processing.py @@ -0,0 +1,46 @@ +import logging + +import pandas as pd +from src.pipelines.shared.preprocessing import preprocess +from src.pipelines.clients.chc.extraction import extract_claims_coding_section +from src.utils import io_utils, logging_utils, timing_utils +from src.constants.constants import Constants +from src import config +from src.utils.string_utils import datetime_str + + +def process_file(file_object, _constants: Constants, run_timestamp): + """Process a single file for CHC client - extracts CHC-specific fields only.""" + filename, contract_text = file_object + + # Set per-file logging context + logging_utils.set_current_file(filename) + logging.info(f"{datetime_str()} Processing {filename}...") + + ################## PREPROCESS ################## + with timing_utils.timed_block("preprocess", context=filename): + contract_text = preprocess.clean_text(contract_text) + + ################## EXTRACT CHC FIELDS ################## + with timing_utils.timed_block("chc_fields", context=filename): + claims_coding_section = extract_claims_coding_section(contract_text) + + # Build result + result = { + "FILE_NAME": filename, + "CLAIMS_CODING_EDITING_DETERMINATIONS": claims_coding_section or "N/A", + } + + final_df = pd.DataFrame([result]) + logging.info(f"{datetime_str()} Extraction 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") + else: + io_utils.write_local(final_df, filename, "", "individual") + + logging.info(f"{datetime_str()} Writing Complete - {filename}") + + return final_df diff --git a/src/pipelines/clients/clover/file_processing.py b/src/pipelines/clients/clover/file_processing.py new file mode 100644 index 0000000..222ea2b --- /dev/null +++ b/src/pipelines/clients/clover/file_processing.py @@ -0,0 +1,524 @@ +import concurrent.futures +import logging +from typing import TYPE_CHECKING, Optional + +import pandas as pd +import src.codes.code_funcs as code_funcs +from src.pipelines.shared.preprocessing import ( + preprocessing_funcs, + preprocess, + hybrid_smart_chunking_funcs, +) +from src.pipelines.shared.postprocessing import ( + aarete_derived, + postprocess, + postprocessing_funcs, +) +from src.pipelines.shared.extraction import ( + dynamic_funcs, + one_to_n_funcs, + one_to_one_funcs, + row_funcs, + tin_npi_funcs, + exhibit_funcs, +) +from src.pipelines.shared.extraction.exhibit_funcs import Exhibit +from src.utils import io_utils, logging_utils, string_utils, timing_utils +from src.constants.constants import Constants +from src import config +from src.prompts.fieldset import FieldSet +from src.utils.string_utils import datetime_str + +if TYPE_CHECKING: + from src.pipelines.shared.extraction.page_funcs import Page + + +def process_file(file_object, constants: Constants, run_timestamp): + filename, contract_text = file_object + + # Set per-file logging context: + # With this, all logging calls in this thread will now route to logs/{filename}.log + # This includes logging from all called functions (preprocess, one_to_n_funcs, etc.) + logging_utils.set_current_file(filename) + logging.info(f"{datetime_str()} Processing {filename}...") + + # Set default values + dynamic_one_to_one_fields = FieldSet() + process_one_to_n = config.FIELDS in ["all", "one_to_n"] + process_one_to_one = config.FIELDS in ["all", "one_to_one"] + # Initialize default fallback for final results + final_results = pd.DataFrame([{"FILE_NAME": filename}]) + + ################## PREPROCESS ################## + with timing_utils.timed_block("preprocess", context=filename): + contract_text = preprocess.clean_text(contract_text) + text_dict, top_sheet_dict = preprocess.split_text(contract_text) + text_dict, header, footer = preprocess.find_headers_and_footers(text_dict) + + # ONE TO N PROCESSING + one_to_n_results = pd.DataFrame() # Initialize empty DataFrame for one_to_n_results + if process_one_to_n: + # Use split_text_with_pages() to get Page objects with table splitting + # Note: split_text_with_pages() internally calls split_text() which applies headers/footers + # But we need to apply headers/footers first, so we'll create pages_dict from the cleaned text_dict + from src.pipelines.shared.extraction.page_funcs import Page + from src.pipelines.shared.preprocessing import preprocessing_funcs as prep_funcs + + # Create pages_dict from text_dict (headers/footers already applied) + pages_dict = prep_funcs.split_large_tables(text_dict) + + with timing_utils.timed_block("one_to_n_exhibit_chunking", context=filename): + # Use pages_dict for exhibit chunking (preferred) or fall back to text_dict + exhibit_chunk_mapping, all_exhibit_headers = ( + preprocess.one_to_n_exhibit_chunking( + pages_dict=pages_dict, + text_dict=text_dict, + EXHIBIT_HEADER_MARKERS=constants.EXHIBIT_HEADER_MARKERS, + filename=filename, + ) + ) + logging.info(f"{datetime_str()} Preprocessing Complete - {filename}") + + one_to_n_results = pd.DataFrame([{"FILE_NAME": filename}]) # Initialize here + + if string_utils.contains_reimbursement(contract_text): + logging.info( + f"{datetime_str()} Starting One-to-N extraction for {len(exhibit_chunk_mapping)} exhibits - {filename}" + ) + with timing_utils.timed_block("one_to_n_extraction", context=filename): + ( + one_to_n_results, + dynamic_one_to_one_fields, + first_reimbursement_page, + ) = run_one_to_n_prompts( + pages_dict=pages_dict, + text_dict=text_dict, + exhibit_chunk_mapping=exhibit_chunk_mapping, + all_exhibit_headers=all_exhibit_headers, + constants=constants, + filename=filename, + ) + if not one_to_n_results.empty: + one_to_n_results["FILE_NAME"] = filename + with timing_utils.timed_block( + "one_to_n_generate_reimb_ids", context=filename + ): + one_to_n_results = postprocessing_funcs.generate_reimb_ids( + one_to_n_results + ) + logging.info(f"{datetime_str()} One to N Complete - {filename}") + else: + first_reimbursement_page = "1" + logging.info( + f"{datetime_str()} No Reimbursement Found, Skipping - {filename}" + ) + + final_results = one_to_n_results # Set as final results + else: + first_reimbursement_page = "1" + logging.info( + f"{datetime_str()} Fields not configured for One to N, Skipping - {filename}" + ) + + # ONE TO ONE PROCESSING + if process_one_to_one: + with timing_utils.timed_block("one_to_one_extraction", context=filename): + one_to_one_results = run_one_to_one_prompts( + filename, + contract_text, + text_dict, + top_sheet_dict, + dynamic_one_to_one_fields, + first_reimbursement_page, + constants, + ) + one_to_one_results["FILE_NAME"] = filename + logging.info(f"{datetime_str()} One to One Complete - {filename}") + + # Decide how to handle one_to_one results + with timing_utils.timed_block( + "merge_one_to_one_into_one_to_n", context=filename + ): + if not one_to_n_results.empty: + # BOTH processed - merge into one_to_n + final_results = row_funcs.merge_one_to_one_into_one_to_n( + one_to_n_results, one_to_one_results, constants + ) + else: + # ONLY one_to_one processed - convert dict to DataFrame + final_results = pd.DataFrame([one_to_one_results]) + else: + logging.info( + f"{datetime_str()} Fields not configured for One to One, Skipping - {filename}" + ) + + # APPLY CODES IF ONE_TO_N WAS PROCESSED + if not one_to_n_results.empty: + with timing_utils.timed_block("code_processing", context=filename): + results_with_code = code_funcs.code_breakout(final_results, constants) + final_results = code_funcs.grouper_breakout(results_with_code) + logging.info(f"{datetime_str()} Codes Complete - {filename}") + + # POSTPROCESS + with timing_utils.timed_block("postprocess", context=filename): + 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(cc_df, filename, run_timestamp, "individual_cc") + else: + io_utils.write_local(cc_df, filename, "", "individual_cc") + + logging.info(f"{datetime_str()} Writing Complete - {filename}") + + return cc_df, dashboard_df + + +def run_one_to_one_prompts( + filename: str, + contract_text: str, + text_dict: dict[str, str], + top_sheet_dict, + dynamic_one_to_one_fields: FieldSet, + first_reimbursement_page: str, + constants: Constants, +): + + ################## INITIALIZE FIELDS ################## + one_to_one_fields = FieldSet( + relationship="one_to_one", file_path=config.FIELD_JSON_PATH + ).combine(dynamic_one_to_one_fields) + + ################## INITIALIZE CLOVER FIELDS ################## + clover_fields = FieldSet( + relationship="one_to_one", file_path=config.CLOVER_PROMPTS_PATH + ) + + ################## RUN PROVIDER INFO ################## + with timing_utils.timed_block("one_to_one.provider_info", context=filename): + one_to_one_results = tin_npi_funcs.run_provider_info_fields( + text_dict, filename, payer_name="" + ) + + ################## RUN HYBRID SMART CHUNKED PROMPTS ################## + # RAG function loads retrieval questions internally and matches with investment_prompts.json + with timing_utils.timed_block("one_to_one.hybrid_smart_chunking", context=filename): + hybrid_smart_chunked_answers_dict = ( + hybrid_smart_chunking_funcs.run_hybrid_smart_chunked_fields( + one_to_one_fields, constants, contract_text, filename, text_dict + ) + ) + + ################## RUN CLOVER PROMPTS ################## + with timing_utils.timed_block("one_to_one.clover_fields", context=filename): + clover_answers_dict = one_to_one_funcs.run_full_context_fields( + clover_fields, + contract_text, + text_dict, + first_reimbursement_page, + constants, + filename, + ) + ################## COMBINE ANSWERS ################## + for key, value in hybrid_smart_chunked_answers_dict.items(): + if not string_utils.is_empty(value): + one_to_one_results[key] = value + elif key not in one_to_one_results: + # Add HSC's N/A if field doesn't exist yet + one_to_one_results[key] = value + + one_to_one_results = tin_npi_funcs.merge_provider_info_with_one_to_one( + one_to_one_results, filename + ) + + # Add clover fields (Claims Timely Filing Days, Appeals Timely Filing Days, etc.) + for key, value in clover_answers_dict.items(): + one_to_one_results[key] = value + + ################## ADD AD FIELDS ################ + one_to_one_results = one_to_one_funcs.get_aarete_derived_dates( + one_to_one_results, text_dict, filename + ) + + ################## ADD SIGNATURE COUNT ################## + one_to_one_results = one_to_one_funcs.add_signature_count( + one_to_one_results, contract_text + ) + + ################## Crosswalk Fields ################## + one_to_one_results = aarete_derived.get_crosswalk_fields( + [one_to_one_results], constants + ) + + ################## Fill NA Mapping ################## + one_to_one_results = aarete_derived.fill_na_mapping(one_to_one_results) + + return one_to_one_results[0] + + +def process_page_reimbursements( + exhibit: Exhibit, + page_num: str, + pages_dict: Optional[dict[str, "Page"]] = None, + text_dict: Optional[dict[str, str]] = None, + exhibit_page_nums: Optional[list[str]] = None, + constants: Optional[Constants] = None, + filename: Optional[str] = None, +): + """ + STEP 1 HELPER: Extract reimbursements from a single page (without exhibit-level fields). + Runs: reimbursement_level → carveout_and_special_case → lesser_of_distribution → breakout + + Args: + exhibit: Exhibit object containing the page + page_num: Page identifier (can be "27" or "27.1" for sub-pages) + pages_dict: Dictionary mapping page numbers to Page objects (preferred) + text_dict: Dictionary mapping page numbers to text strings (for backward compatibility) + exhibit_page_nums: List of page numbers in the exhibit (for backward compatibility) + constants: Constants object + filename: Name of the file being processed + """ + # Get page text - prefer using exhibit.get_page_text() if available + if exhibit.pages is not None: + page_text = exhibit.get_page_text(page_num) + exhibit_page_nums = exhibit.exhibit_page_nums + elif pages_dict is not None: + # Handle sub-pages: "27.1" means get sub-page "1" from page "27" + if "." in page_num and not page_num.startswith("."): + base_page, sub_page = page_num.rsplit(".", 1) + if base_page in pages_dict: + page_text = pages_dict[base_page].get_text(sub_page) + else: + page_text = "" + elif page_num in pages_dict: + page_text = pages_dict[page_num].get_text() + else: + page_text = "" + exhibit_page_nums = ( + exhibit.exhibit_page_nums + if exhibit_page_nums is None + else exhibit_page_nums + ) + elif text_dict is not None: + page_text = text_dict.get(page_num, "") + exhibit_page_nums = ( + exhibit_page_nums + if exhibit_page_nums is not None + else exhibit.exhibit_page_nums + ) + else: + raise ValueError( + "Either pages_dict, text_dict, or exhibit.pages must be provided" + ) + + # Simplify exhibit text - prefer using exhibit object + exhibit_text_simplified = preprocessing_funcs.simplify_exhibit( + exhibit=exhibit, + current_page_num=page_num, + ) + + ############################### Reimbursement Primary ############################### + reimbursement_level_answers = one_to_n_funcs.reimbursement_level( + page_text, constants, filename + ) + + if not reimbursement_level_answers: + return [], [] + + ################################ Carveouts and Special Case ################################ + reimbursement_level_answers, special_case_answers = ( + one_to_n_funcs.carveout_and_special_case( + reimbursement_level_answers, constants, filename + ) + ) + + ############################### Lesser of Distribution ############################### + # Note: We run lesser_of without dynamic fields in Step 1 + reimbursement_level_answers = one_to_n_funcs.lesser_of_distribution( + reimbursement_level_answers, + exhibit_text_simplified, + page_num, + constants, + filename, + exhibit, # ← Added this parameter + ) + + ################################ Get Breakouts ############################### + reimbursement_level_answers, special_case_answers = one_to_n_funcs.breakout( + reimbursement_level_answers, + special_case_answers, + filename, + constants, + ) + + ################################ Assign REIMB_PAGE ############################### + for answer_dict in reimbursement_level_answers: + answer_dict["REIMB_PAGE"] = page_num + + return reimbursement_level_answers, special_case_answers + + +def run_one_to_n_prompts( + pages_dict: Optional[dict[str, "Page"]] = None, + text_dict: Optional[dict[str, str]] = None, + exhibit_chunk_mapping: Optional[dict[str, list[str]]] = None, + all_exhibit_headers: Optional[dict[str, str]] = None, + constants: Optional[Constants] = None, + filename: Optional[str] = None, +): + """ + 3-STEP APPROACH (per exhibit): + Step 1: Extract reimbursements for exhibit pages (parallel within exhibit) + Step 2: Run exhibit_level when exhibit has reimbursements + Step 3: Run dynamic_assignment and combine results + + Args: + pages_dict: Dictionary mapping page numbers to Page objects (preferred) + text_dict: Dictionary mapping page numbers to text strings (for backward compatibility) + exhibit_chunk_mapping: Dictionary mapping exhibit pages to lists of page numbers + all_exhibit_headers: Dictionary mapping exhibit pages to their headers + constants: Constants object + filename: Name of the file being processed + """ + if exhibit_chunk_mapping is None: + exhibit_chunk_mapping = {} + if all_exhibit_headers is None: + all_exhibit_headers = {} + + total_exhibits = len(exhibit_chunk_mapping) + logging.debug( + f"{datetime_str()} Processing {total_exhibits} exhibits with NEW 3-step approach - {filename}" + ) + + # Create Exhibit objects in order (maintains exhibit chain via prev_exhibit) + exhibits = exhibit_funcs.get_exhibit_list( + pages_dict=pages_dict, + text_dict=text_dict, + exhibit_chunk_mapping=exhibit_chunk_mapping, + all_exhibit_headers=all_exhibit_headers, + ) # <- Returns list[Exhibit] + + # Process exhibits serially; within each exhibit, process pages in parallel + total_pages = sum(len(exhibit.exhibit_page_nums) for exhibit in exhibits) + completed_pages = 0 + one_to_n_results = [] + first_reimbursement_page = "1" + + for exhibit in exhibits: + exhibit_pages = exhibit.exhibit_page_nums + if exhibit_pages: + with concurrent.futures.ThreadPoolExecutor( + max_workers=min(len(exhibit_pages), 20) + ) as executor: + page_futures = { + executor.submit( + process_page_reimbursements, + exhibit, + page_num, + pages_dict, + text_dict, + exhibit_pages, + constants, + filename, + ): page_num + for page_num in exhibit_pages + } + + for future in concurrent.futures.as_completed(page_futures): + try: + page_num = page_futures[future] + reimbursement_rows, special_case_rows = future.result() + exhibit.add_reimbursement_rows( + reimbursement_rows, special_case_rows + ) + completed_pages += 1 + if completed_pages % 20 == 0 or completed_pages == total_pages: + logging.debug( + f"{datetime_str()} Page progress: {completed_pages}/{total_pages} - {filename}" + ) + except Exception as e: + logging.error(f"Error processing page: {str(e)}") + + if not exhibit.has_reimbursements: + continue + + # STEP 2: exhibit_level for this exhibit + exhibit_level_answers, dynamic_reimbursement_fields = ( + one_to_n_funcs.exhibit_level( + exhibit.exhibit_text, + exhibit.exhibit_header, + exhibit.exhibit_page, + constants, + filename, + ) + ) + exhibit.set_exhibit_level_data( + exhibit_level_answers, dynamic_reimbursement_fields + ) + logging.debug( + f"Exhibit Level Answers for {filename}, Page {exhibit.exhibit_page}: {exhibit_level_answers}" + ) + + # STEP 3: dynamic assignment & combine for this exhibit + # Get dynamic fields from previous exhibit if needed (for future use) + if exhibit.dynamic_reimbursement_fields: + dynamic_fields = exhibit.dynamic_reimbursement_fields + elif ( + exhibit.prev_exhibit + and not exhibit.prev_exhibit.has_reimbursements + and exhibit.get_previous_exhibit_dynamic_fields().fields + ): + dynamic_fields = exhibit.get_previous_exhibit_dynamic_fields() + else: + dynamic_fields = None + + # Run dynamic assignment for ALL reimbursement rows in this exhibit + # Note: dynamic_assignment expects a list of rows and returns a list of rows + reimbursement_rows_with_dynamic = exhibit.reimbursement_rows + if exhibit.reimbursement_rows and dynamic_fields is not None: + reimbursement_rows_with_dynamic = dynamic_funcs.dynamic_assignment( + exhibit.reimbursement_rows, + dynamic_fields, + pages_dict, + text_dict, + exhibit, + constants, + filename, + ) + + # Combine reimbursement rows with exhibit-level answers + combined_rows = row_funcs.combine_one_to_n( + exhibit.exhibit_text, + reimbursement_rows_with_dynamic, + exhibit.special_case_rows, + exhibit.exhibit_level_answers, + filename, + ) + + # Run cleaning + combined_rows = one_to_n_funcs.one_to_n_cleaning( + combined_rows, exhibit.exhibit_text, constants, filename + ) + + exhibit.final_rows = combined_rows + one_to_n_results.extend(combined_rows) + + # Track first reimbursement page + if first_reimbursement_page == "1" and exhibit.has_reimbursements: + first_reimbursement_page = exhibit.exhibit_page + + logging.debug( + f"{datetime_str()} STEP 3 COMPLETE: Combined {len(one_to_n_results)} total rows - {filename}" + ) + + ################## Add N/A Dynamic or Exhibit to One-to-One ################## + dynamic_one_to_one_fields = dynamic_funcs.get_dynamic_one_to_one_fields( + one_to_n_results, constants + ) + + ################## CONVERT TO DF ################## + one_to_n_df = pd.DataFrame(one_to_n_results) + + return one_to_n_df, dynamic_one_to_one_fields, first_reimbursement_page diff --git a/src/pipelines/clients/clover/prompts/prompt_calls.py b/src/pipelines/clients/clover/prompts/prompt_calls.py index 5813b4a..77a7f3c 100644 --- a/src/pipelines/clients/clover/prompts/prompt_calls.py +++ b/src/pipelines/clients/clover/prompts/prompt_calls.py @@ -49,5 +49,5 @@ def validate_reimbursements_for_llm(answer_dict: dict[str, str], filename: str) logging.debug( f"LLM response for reimbursement validation in {filename}:\n{llm_response}" ) - final_answer = _parser(llm_response) - return "YES" in final_answer + final_answer = _parser(llm_response).strip().upper() + return final_answer == "YES" diff --git a/src/pipelines/runner.py b/src/pipelines/runner.py index c0c8152..be85ddc 100644 --- a/src/pipelines/runner.py +++ b/src/pipelines/runner.py @@ -1,9 +1,8 @@ """ Common Pipeline Runner -Canonical entry point for the Doczy pipeline. Handles shared orchestration -(DTC, QC/QA, cache warming, duplicate detection, timing) and routes -to client-specific file_processing implementations via the resolver shim. +Handles shared orchestration (DTC, QC/QA, cache warming) and routes +to client-specific file_processing implementations. Usage: from src.pipelines.runner import main @@ -14,16 +13,12 @@ Usage: import concurrent.futures import logging import os -import random -import time import traceback import warnings from datetime import datetime import pandas as pd -random.seed(42) - # Suppress Pydantic protected namespace warning from langchain-aws warnings.filterwarnings("ignore", message=".*protected namespace.*") @@ -32,12 +27,7 @@ import src.utils.io_utils as io_utils import src.utils.llm_utils as llm_utils import src.utils.logging_utils as logging_utils import src.utils.usage_tracking as usage_tracking -import src.utils.timing_utils as timing_utils -import src.utils.duplicate_detection as duplicate_detection from src.pipelines.runtime_context import set_active_client -from src.pipelines.shared.extraction import one_to_one_funcs -from src.pipelines.shared.postprocessing import postprocessing_funcs -from src.constants.investment_columns import FIELD_FORMAT_MAPPING from src.constants.constants import Constants from src import config from src.core.registry import get_registry @@ -45,8 +35,6 @@ from src.parent_child import main as parent_child_main from src.document_classification import main as dtc_main from src.qc_qa.pipeline import ( generate_statistics, - generate_tin_contract_summary, - generate_tin_statistics, run_qc_qa_pipeline, save_qc_qa_outputs, ) @@ -164,39 +152,22 @@ def main(client: str = "saas", testing=False, test_params={}): # Windows paths cannot contain ':'; use '-' in time component run_timestamp = datetime.now().strftime(f"run_%Y%m%d_%H-%M_{config.BATCH_ID}") - # Track overall pipeline time - pipeline_start = time.time() - # Read Constants - with timing_utils.timed_block("read_constants", log_level="INFO"): - constants = Constants() + constants = Constants() # Read and process input - with timing_utils.timed_block("read_input", log_level="INFO"): - if not testing: - input_dict = io_utils.read_input() - max_workers = config.MAX_WORKERS - else: - input_dict = io_utils.read_input(test_params["local_input_dir"]) - max_workers = test_params["max_workers"] - if "input_files" in test_params: - input_dict = { - k: v - for k, v in input_dict.items() - if k in test_params["input_files"] - } + if not testing: + input_dict = io_utils.read_input() + max_workers = config.MAX_WORKERS + else: + input_dict = io_utils.read_input(test_params["local_input_dir"]) + max_workers = test_params["max_workers"] + if "input_files" in test_params: + input_dict = { + k: v for k, v in input_dict.items() if k in test_params["input_files"] + } - logging.info(f"Total Input Files : {len(input_dict)}") - - # Detect and handle duplicate contracts - with timing_utils.timed_block("duplicate_detection", log_level="INFO"): - duplicate_groups = duplicate_detection.find_duplicate_groups(input_dict) - original_files_to_process, file_status_map = ( - duplicate_detection.get_files_to_process(input_dict, duplicate_groups) - ) - logging.debug( - f"duplicate_groups={len(duplicate_groups)}, original_files={len(original_files_to_process)}, file_status_map length={len(file_status_map)}" - ) + logging.debug(f"Total Input Files : {len(input_dict)}") # ========== COMMON: Document Type Classification ========== if config.PERFORM_DTC: @@ -212,169 +183,97 @@ def main(client: str = "saas", testing=False, test_params={}): # ========== COMMON: Warm prompt caches ========== logging.info("Warming prompt caches before parallel processing...") - with timing_utils.timed_block("warm_prompt_caches", log_level="INFO"): - try: - cacheable_instructions = prompt_templates.get_cacheable_instructions() - for cache_key, instruction in cacheable_instructions.items(): - llm_utils.warm_prompt_cache( - instruction=instruction, - cache_key=cache_key, - model_id="sonnet_latest", - ) - logging.info( - f"Successfully warmed {len(cacheable_instructions)} prompt caches" + try: + cacheable_instructions = prompt_templates.get_cacheable_instructions() + for cache_key, instruction in cacheable_instructions.items(): + llm_utils.warm_prompt_cache( + instruction=instruction, cache_key=cache_key, model_id="sonnet_latest" ) - except Exception as e: - logging.warning(f"Error warming caches (will proceed anyway): {str(e)}") + logging.debug( + f"Successfully warmed {len(cacheable_instructions)} prompt caches" + ) + except Exception as e: + logging.warning(f"Error warming caches (will proceed anyway): {str(e)}") # ========== CLIENT-SPECIFIC: File Processing ========== successful_results_cc = [] successful_results_dashboard = [] error_results = [] + with concurrent.futures.ThreadPoolExecutor(max_workers=max_workers) as executor: + futures = [ + executor.submit( + safe_process_file, item, constants, run_timestamp, file_processing + ) + for item in input_dict.items() + ] - logging.info(f"Starting parallel file processing with {max_workers} workers...") - with timing_utils.timed_block("parallel_file_processing", log_level="INFO"): - with concurrent.futures.ThreadPoolExecutor(max_workers=max_workers) as executor: - # Only process original files; duplicates will have results copied after - futures = [ - executor.submit( - safe_process_file, item, constants, run_timestamp, file_processing - ) - for item in input_dict.items() - if item[0] in original_files_to_process - ] - - # collect results as they complete - completed_count = 0 - for future in concurrent.futures.as_completed(futures): - try: - cc_result, dashboard_result = future.result() - if "error" in cc_result.columns: - error_results.append(cc_result) - else: - 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: - logging.info( - f"Processed {completed_count}/{len(futures)} files..." - ) - except Exception as e: - logging.error(f"Error processing future: {str(e)}") - error_df = pd.DataFrame([{"error": str(e)}]) - error_results.append(error_df) + for future in concurrent.futures.as_completed(futures): + try: + cc_result, dashboard_result = future.result() + if "error" in cc_result.columns: + error_results.append(cc_result) + else: + 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_df = pd.DataFrame([{"error": str(e)}]) + error_results.append(error_df) # Combine results - with timing_utils.timed_block("concat_results", log_level="INFO"): - 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() + 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_CC = pd.DataFrame() FINAL_RESULT_DF_DASHBOARD = pd.DataFrame() + else: + FINAL_RESULT_DF_CC = pd.DataFrame() + FINAL_RESULT_DF_DASHBOARD = pd.DataFrame() - if not FINAL_RESULT_DF_CC.empty: - FINAL_RESULT_DF_CC = one_to_one_funcs.add_aarete_derived_payer_name( - FINAL_RESULT_DF_CC, config.STATE_FLAG - ) - FINAL_RESULT_DF_CC = one_to_one_funcs.add_aarete_derived_provider_name( - FINAL_RESULT_DF_CC, config.STATE_FLAG - ) - if not FINAL_RESULT_DF_DASHBOARD.empty: - FINAL_RESULT_DF_DASHBOARD = one_to_one_funcs.add_aarete_derived_payer_name( - FINAL_RESULT_DF_DASHBOARD, config.STATE_FLAG - ) - FINAL_RESULT_DF_DASHBOARD = ( - one_to_one_funcs.add_aarete_derived_provider_name( - FINAL_RESULT_DF_DASHBOARD, config.STATE_FLAG - ) - ) - - FINAL_RESULT_DF_CC = postprocessing_funcs.reorder_columns( - FINAL_RESULT_DF_CC, FIELD_FORMAT_MAPPING - ) - FINAL_RESULT_DF_DASHBOARD = postprocessing_funcs.reorder_columns( - FINAL_RESULT_DF_DASHBOARD, FIELD_FORMAT_MAPPING - ) - - if len(error_results) > 0: - ERROR_RESULT_DF = pd.concat(error_results, ignore_index=True) - else: - ERROR_RESULT_DF = pd.DataFrame() - - # Copy results from original files to their duplicates - logging.debug( - f"Before copy: file_status_map length={len(file_status_map)}, file_status_map keys={list(file_status_map.keys())[:5]}..." - ) - logging.debug( - f"Before copy: FINAL_RESULT_DF_CC shape={FINAL_RESULT_DF_CC.shape}, rows={len(FINAL_RESULT_DF_CC)}" - ) - - if len(file_status_map) > 0: - with timing_utils.timed_block("copy_duplicate_results", log_level="INFO"): - logging.info( - f"Calling duplicate_copy_results with {len(file_status_map)} file statuses and {len(FINAL_RESULT_DF_CC)} result rows" - ) - duplicate_detection.duplicate_copy_results( - FINAL_RESULT_DF_CC, file_status_map, "FILE_NAME" - ) - logging.debug( - f"After copy: FINAL_RESULT_DF_CC shape={FINAL_RESULT_DF_CC.shape}, rows={len(FINAL_RESULT_DF_CC)}" - ) - if not FINAL_RESULT_DF_DASHBOARD.empty: - duplicate_detection.duplicate_copy_results( - FINAL_RESULT_DF_DASHBOARD, file_status_map, "FILE_NAME" - ) + 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 ========== + # 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 on CC results...") logging.info("=" * 80) - with timing_utils.timed_block("qc_qa_validation", log_level="INFO"): - try: - 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) - - validated_cc_df = run_qc_qa_pipeline( - FINAL_RESULT_DF_CC.copy(), stats_df=None + try: + # 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..." ) - logging.info("QC/QA validation on CC results completed successfully") + 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_cc_df) - tin_stats_df = generate_tin_statistics(validated_cc_df) - tin_contract_summary_df = generate_tin_contract_summary(validated_cc_df) - logging.info("QC/QA statistics generated successfully") + validated_cc_df = run_qc_qa_pipeline( + FINAL_RESULT_DF_CC.copy(), stats_df=None + ) + logging.debug("QC/QA validation on CC results completed successfully") - save_qc_qa_outputs( - validated_df=validated_cc_df, - stats_df=qc_qa_stats_df, - run_timestamp=run_timestamp, - write_to_s3=config.WRITE_TO_S3, - tin_stats_df=tin_stats_df, - tin_contract_summary_df=tin_contract_summary_df, - ) - except Exception as 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 CC results...") + # Generate validation statistics from CC version + logging.debug("Generating QC/QA statistics...") + qc_qa_stats_df = generate_statistics(validated_cc_df) + logging.debug("QC/QA statistics generated successfully") + except Exception as 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 CC results...") logging.info("=" * 80) # ========== COMMON: Output & Parent-Child ========== @@ -383,68 +282,63 @@ def main(client: str = "saas", testing=False, test_params={}): config.BATCH_ID, run_timestamp ) - with timing_utils.timed_block("write_outputs", log_level="INFO"): - if config.WRITE_TO_S3: - doczy_output_for_pc = io_utils.write_s3( - FINAL_RESULT_DF_CC, "", run_timestamp, "cc_results_full" + if config.WRITE_TO_S3: + # Write CC version + doczy_output_for_pc = io_utils.write_s3( + 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", ) - 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", - ) - 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 error file + if not ERROR_RESULT_DF.empty: + io_utils.write_s3(ERROR_RESULT_DF, "", run_timestamp, "error") - 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") + # 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: - doczy_output_for_pc = io_utils.write_local( - FINAL_RESULT_DF_CC, "", run_timestamp, "cc_results_full" + if not per_file_usage_df.empty and not batch_summary_df.empty: + logging.debug("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_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", ) - 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", - ) - 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 error file + if not ERROR_RESULT_DF.empty: + io_utils.write_local(ERROR_RESULT_DF, "", run_timestamp, "error") - 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" - ) + # 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_CC.empty: @@ -452,36 +346,28 @@ def main(client: str = "saas", testing=False, test_params={}): "Skipping Parent-Child mapping: No successful results to process (FINAL_RESULT_DF_CC is empty)" ) else: - logging.info( + logging.debug( 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"): - pc_output_dir = os.path.join( - config.CONSOLIDATED_OUTPUT_DIRECTORY, - run_timestamp, - "parent-child", + 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" ) - _, 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...") + logging.debug("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") - - # Print timing summary - pipeline_duration = time.time() - pipeline_start - logging.info("=" * 80) - logging.info(f"PIPELINE COMPLETE - Total time: {pipeline_duration:.2f}s") - logging.info("=" * 80) - timing_utils.print_hierarchical_timing_summary() else: return FINAL_RESULT_DF_CC, FINAL_RESULT_DF_DASHBOARD, ERROR_RESULT_DF diff --git a/src/pipelines/saas/file_processing.py b/src/pipelines/saas/file_processing.py index 5996bc5..fec9b02 100644 --- a/src/pipelines/saas/file_processing.py +++ b/src/pipelines/saas/file_processing.py @@ -23,10 +23,6 @@ from src.pipelines.shared.extraction import ( tin_npi_funcs, exhibit_funcs, ) -from src.pipelines.runtime_context import get_active_client -from src.pipelines.vendors.shared.extraction import ( - run_full_context_fields as vendor_run_full_context_fields, -) from src.pipelines.shared.extraction.exhibit_funcs import Exhibit, ExhibitChunk from src.utils import io_utils, logging_utils, string_utils, timing_utils from src.constants.constants import Constants @@ -195,30 +191,6 @@ def run_one_to_one_prompts( relationship="one_to_one", file_path=config.FIELD_JSON_PATH ).combine(dynamic_one_to_one_fields) - ################## LOAD CLIENT-SPECIFIC FIELDS ################## - client_name = get_active_client() - client_prompts_path = config.CLIENT_PROMPTS_MAP.get(client_name) - client_full_context_fields = None - - if client_prompts_path: - client_fields = FieldSet( - relationship="one_to_one", file_path=client_prompts_path - ) - if client_fields.contains_fields(): - # smart_chunked fields (e.g. BCBS OFFSET_TERM) are combined into HSC fields - smart_chunked = client_fields.filter(field_type="smart_chunked") - if smart_chunked.contains_fields(): - one_to_one_fields = one_to_one_fields.combine(smart_chunked) - - # full_context fields (e.g. Clover's 12 fields) are extracted separately after HSC - full_context = client_fields.filter(field_type="full_context") - if full_context.contains_fields(): - client_full_context_fields = full_context - - logging.info( - f"Loaded {len(client_fields.fields)} client-specific fields for {client_name}" - ) - # Filter fields if specific_fields is provided if specific_fields is not None: one_to_one_fields = one_to_one_fields.filter_by_names(specific_fields) @@ -257,30 +229,6 @@ def run_one_to_one_prompts( # Add HSC's N/A if field doesn't exist yet one_to_one_results[key] = value - ################## RUN CLIENT FULL-CONTEXT FIELDS ################## - # Client-specific full_context fields (e.g. Clover's 12 audit/filing/CA fields) - if client_full_context_fields is not None: - with timing_utils.timed_block( - f"one_to_one.{client_name}_full_context", context=filename - ): - fc_answers = vendor_run_full_context_fields( - contract_text=contract_text, - fields=client_full_context_fields, - filename=filename, - usage_prefix=client_name.upper(), - constants=constants, - ) - one_to_one_results.update(fc_answers) - - ################## DERIVE CLIENT-SPECIFIC FIELDS ################## - # BCBS Promise: derive OFFSET_INDICATOR from OFFSET_TERM - if "OFFSET_TERM" in one_to_one_results: - offset_term = one_to_one_results.get("OFFSET_TERM", "N/A") - if not string_utils.is_empty(offset_term): - one_to_one_results["OFFSET_INDICATOR"] = "Y" - else: - one_to_one_results["OFFSET_INDICATOR"] = "N" - ################## RUN PROVIDER INFO ################## if run_provider_info: with timing_utils.timed_block("one_to_one.provider_info", context=filename): diff --git a/src/pipelines/saas/main.py b/src/pipelines/saas/main.py index 9d79c3c..057c690 100644 --- a/src/pipelines/saas/main.py +++ b/src/pipelines/saas/main.py @@ -1,25 +1,471 @@ -""" -Pipeline entry point — delegates to the common runner. +import concurrent.futures +import logging +import os +import random +import time +import traceback +from datetime import datetime -This module exists for backward compatibility. All orchestration logic -lives in src.pipelines.runner. Calling main() here determines the -client from config.CLIENT and delegates to runner.main(). -""" +import pandas as pd +random.seed(42) + +import src.pipelines.saas.file_processing as file_processing +from src.pipelines.runtime_context import set_active_client +from src.pipelines.shared.extraction import one_to_one_funcs +from src.pipelines.shared.postprocessing import postprocessing_funcs +from src.constants.investment_columns import FIELD_FORMAT_MAPPING +import src.prompts.prompt_templates as prompt_templates +import src.utils.io_utils as io_utils +import src.utils.llm_utils as llm_utils +import src.utils.logging_utils as logging_utils +import src.utils.usage_tracking as usage_tracking +import src.utils.timing_utils as timing_utils +import src.utils.duplicate_detection as duplicate_detection +from src.constants.constants import Constants from src import config -from src.pipelines import runner +from src.parent_child import main as parent_child_main +from src.document_classification import main as dtc_main +from src.qc_qa.pipeline import ( + generate_statistics, + generate_tin_contract_summary, + generate_tin_statistics, + run_qc_qa_pipeline, + save_qc_qa_outputs, +) -def _resolve_client() -> str: - """Determine the client name from CLI config.""" - if config.CLIENT and len(config.CLIENT) == 1 and config.CLIENT[0] != "None": - return config.CLIENT[0] - return "saas" +def safe_process_file(item, constants, run_timestamp): + """Process a single file with comprehensive error handling. + + This function wraps the main file processing logic to ensure that: + - Individual file failures don't crash the entire batch + - Detailed error information is captured for debugging + + Args: + item: Tuple of (filename, contract_text) representing the file to process + constants: Constants object containing configuration and processing rules + run_timestamp: Timestamp string for output file naming and tracking + + Returns: + 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. + 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: + 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 + error_type = type(e).__name__ + error_message = str(e) + full_traceback = traceback.format_exc() + + logging.error(f"Error processing file {file_id}") + logging.error(f"Error type: {error_type}") + logging.error(f"Error Message: {error_message}") + + logging.error(f"Full traceback:\n{full_traceback}") + + error_df = pd.DataFrame( + [ + { + "FILE_NAME": file_id, + "error": error_message.replace(",", ";").replace( + "\n", " " + ), # Replace problematic characters for CSV compatibility + "error_type": error_type, + "traceback": full_traceback.replace(",", ";").replace( + "\n", " | " + ), # keep line breaks as separators + } + ] + ) # 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={}): - client = _resolve_client() - return runner.main(client=client, testing=testing, test_params=test_params) + set_active_client("saas") + + # Check if batch_id is specified + if not config.BATCH_ID: + raise ValueError( + "batch_id is required. Please specify batch_id in your command" + ) + + # Check if max_workers is not negative + if config.MAX_WORKERS < 0: + raise ValueError( + f"Invalid configuration: MAX_WORKERS must be non-negative (got {config.MAX_WORKERS})" + ) + + # Set up per-file logging (creates separate log files for each document) + logging_utils.setup_per_file_logging() + + # Windows paths cannot contain ':'; use '-' in time component + run_timestamp = datetime.now().strftime(f"run_%Y%m%d_%H-%M_{config.BATCH_ID}") + + # Track overall pipeline time + pipeline_start = time.time() + + # Read Constants + with timing_utils.timed_block("read_constants", log_level="INFO"): + constants = Constants() + + # Read and process input + with timing_utils.timed_block("read_input", log_level="INFO"): + if not testing: + input_dict = io_utils.read_input() + max_workers = config.MAX_WORKERS + else: + input_dict = io_utils.read_input(test_params["local_input_dir"]) + max_workers = test_params["max_workers"] + if "input_files" in test_params: + input_dict = { + k: v + for k, v in input_dict.items() + if k in test_params["input_files"] + } + + logging.info(f"Total Input Files : {len(input_dict)}") + + # Detect and handle duplicate contracts + with timing_utils.timed_block("duplicate_detection", log_level="INFO"): + duplicate_groups = duplicate_detection.find_duplicate_groups(input_dict) + original_files_to_process, file_status_map = ( + duplicate_detection.get_files_to_process(input_dict, duplicate_groups) + ) + logging.debug( + f"Saas main: duplicate_groups={len(duplicate_groups)}, original_files={len(original_files_to_process)}, file_status_map length={len(file_status_map)}" + ) + + # Run document type classification if enabled - DTC mode only, then exit + if config.PERFORM_DTC: + logging.info("=" * 80) + logging.info("DOCUMENT TYPE CLASSIFICATION MODE - Running DTC and exiting") + logging.info("=" * 80) + + # run DTC classification + dtc_main.main(input_dict, run_timestamp) + + logging.info( + "DTC classification complete. Exiting (investment pipeline not run)." + ) + logging.info("=" * 80) + return + + # Warm prompt caches before parallel processing + # This ensures the cache is registered before multiple threads try to use it + logging.info("Warming prompt caches before parallel processing...") + with timing_utils.timed_block("warm_prompt_caches", log_level="INFO"): + try: + cacheable_instructions = prompt_templates.get_cacheable_instructions() + for cache_key, instruction in cacheable_instructions.items(): + llm_utils.warm_prompt_cache( + instruction=instruction, + cache_key=cache_key, + model_id="sonnet_latest", + ) + logging.info( + f"Successfully warmed {len(cacheable_instructions)} prompt caches" + ) + except Exception as e: + 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_cc = [] + successful_results_dashboard = [] + error_results = [] + + logging.info(f"Starting parallel file processing with {max_workers} workers...") + with timing_utils.timed_block("parallel_file_processing", log_level="INFO"): + with concurrent.futures.ThreadPoolExecutor(max_workers=max_workers) as executor: + # Each submitted task will set its own logging context in file_processing.process_file() + # We use an executor to process files concurrently; use submit() instead of map() + # so that we can catch exceptions and continue processing other files + # Only process original files; duplicates will have results copied after + futures = [ + executor.submit(safe_process_file, item, constants, run_timestamp) + for item in input_dict.items() + if item[0] in original_files_to_process + ] + + # collect results as they complete + completed_count = 0 + for future in concurrent.futures.as_completed(futures): + try: + cc_result, dashboard_result = future.result() + if "error" in cc_result.columns: + error_results.append(cc_result) + else: + 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 + logging.info( + f"Processed {completed_count}/{len(futures)} files..." + ) + except ( + Exception + ) 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_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_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 not FINAL_RESULT_DF_CC.empty: + FINAL_RESULT_DF_CC = one_to_one_funcs.add_aarete_derived_payer_name( + FINAL_RESULT_DF_CC, config.STATE_FLAG + ) # add aarete derived payer name to cc version + FINAL_RESULT_DF_CC = one_to_one_funcs.add_aarete_derived_provider_name( + FINAL_RESULT_DF_CC, config.STATE_FLAG + ) # add aarete derived provider name to cc version + if not FINAL_RESULT_DF_DASHBOARD.empty: + FINAL_RESULT_DF_DASHBOARD = one_to_one_funcs.add_aarete_derived_payer_name( + FINAL_RESULT_DF_DASHBOARD, config.STATE_FLAG + ) # add aarete derived payer name to dashboard version + FINAL_RESULT_DF_DASHBOARD = ( + one_to_one_funcs.add_aarete_derived_provider_name( + FINAL_RESULT_DF_DASHBOARD, config.STATE_FLAG + ) + ) # add aarete derived provider name to dashboard version + + FINAL_RESULT_DF_CC = postprocessing_funcs.reorder_columns( + FINAL_RESULT_DF_CC, FIELD_FORMAT_MAPPING + ) # Reorder columns in CC version only + FINAL_RESULT_DF_DASHBOARD = postprocessing_funcs.reorder_columns( + FINAL_RESULT_DF_DASHBOARD, FIELD_FORMAT_MAPPING + ) # Reorder columns in dashboard version only + + if len(error_results) > 0: + ERROR_RESULT_DF = pd.concat(error_results, ignore_index=True) + else: + ERROR_RESULT_DF = pd.DataFrame() + + # Copy results from original files to their duplicates + logging.debug( + f"Before copy: file_status_map length={len(file_status_map)}, file_status_map keys={list(file_status_map.keys())[:5]}..." + ) + logging.debug( + f"Before copy: FINAL_RESULT_DF_CC shape={FINAL_RESULT_DF_CC.shape}, rows={len(FINAL_RESULT_DF_CC)}" + ) + + if len(file_status_map) > 0: + with timing_utils.timed_block("copy_duplicate_results", log_level="INFO"): + logging.info( + f"Calling duplicate_copy_results with {len(file_status_map)} file statuses and {len(FINAL_RESULT_DF_CC)} result rows" + ) + duplicate_detection.duplicate_copy_results( + FINAL_RESULT_DF_CC, file_status_map, "FILE_NAME" + ) + logging.debug( + f"After copy: FINAL_RESULT_DF_CC shape={FINAL_RESULT_DF_CC.shape}, rows={len(FINAL_RESULT_DF_CC)}" + ) + if not FINAL_RESULT_DF_DASHBOARD.empty: + duplicate_detection.duplicate_copy_results( + FINAL_RESULT_DF_DASHBOARD, file_status_map, "FILE_NAME" + ) + + # 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 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_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) + + # 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) + tin_stats_df = generate_tin_statistics(validated_cc_df) + tin_contract_summary_df = generate_tin_contract_summary(validated_cc_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_cc_df, + stats_df=qc_qa_stats_df, + run_timestamp=run_timestamp, + write_to_s3=config.WRITE_TO_S3, + tin_stats_df=tin_stats_df, + tin_contract_summary_df=tin_contract_summary_df, + ) + except Exception as 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 CC results...") + logging.info("=" * 80) + + if not testing: + # Get usage tracking dataframes + per_file_usage_df, batch_summary_df = usage_tracking.get_usage_dataframes( + config.BATCH_ID, run_timestamp + ) + + 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_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...") + 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_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_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}" + ) + with timing_utils.timed_block("parent_child_mapping", log_level="INFO"): + client = ( + config.CLIENT[0] + if config.CLIENT + and len(config.CLIENT) == 1 + and config.CLIENT[0] != "None" + else None + ) + 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: + 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") + + # Print timing summary + pipeline_duration = time.time() - pipeline_start + logging.info("=" * 80) + logging.info(f"PIPELINE COMPLETE - Total time: {pipeline_duration:.2f}s") + logging.info("=" * 80) + timing_utils.print_hierarchical_timing_summary() + else: + return FINAL_RESULT_DF_CC, FINAL_RESULT_DF_DASHBOARD, ERROR_RESULT_DF if __name__ == "__main__": diff --git a/src/pipelines/shared/postprocessing/postprocessing_funcs.py b/src/pipelines/shared/postprocessing/postprocessing_funcs.py index da86ec2..b258626 100644 --- a/src/pipelines/shared/postprocessing/postprocessing_funcs.py +++ b/src/pipelines/shared/postprocessing/postprocessing_funcs.py @@ -511,10 +511,6 @@ def reorder_columns( # Reorder columns to match column_order final_column_order = [col for col in column_order if col in df_copy.columns] - # Append any extra columns not in column_order (e.g. client-specific fields) - extra_columns = [col for col in df_copy.columns if col not in column_order] - final_column_order.extend(extra_columns) - # Return the DataFrame with reordered columns return df_copy[final_column_order]