From 5e143c10f0838d82b788b7a269796512bd7ef508 Mon Sep 17 00:00:00 2001 From: ppanchigar Date: Tue, 14 Apr 2026 10:39:51 -0500 Subject: [PATCH] Retire stale client file_processing forks, unify pipeline entry point Delete BCBS/Clover/CHC client file_processing.py files (1090 lines of 96% stale drift with 3 crash points). Recover the 4% real business logic (BCBS OFFSET_TERM, Clover 12 full_context fields) into the shared pipeline via config-driven field loading. Make runner.py the canonical pipeline entry point by merging all production features from main.py (timing, duplicate detection, aarete derived fields, TIN statistics, column reordering). Reduce main.py to a thin wrapper that delegates to runner.main(). Fix the root cause of all client overrides being dead in production: set_active_client is now called with the correct client name instead of being hardcoded to "saas". Both resolver shims (file_processing and prompt_calls) now route correctly. Includes temporary [DEBUG ROUTING] print statements at 6 routing decision points for E2E verification. Cherry-pick this commit to restore debug instrumentation for future testing. E2E verified: BCBS Promise and Clover both pass with correct routing. --- src/config.py | 8 + .../clients/bcbs_promise/file_processing.py | 520 ----------------- src/pipelines/clients/chc/file_processing.py | 46 -- .../clients/clover/file_processing.py | 524 ------------------ src/pipelines/runner.py | 420 +++++++++----- src/pipelines/saas/file_processing.py | 70 +++ src/pipelines/saas/main.py | 475 +--------------- src/pipelines/shared/file_processing.py | 7 + src/pipelines/shared/prompts/prompt_calls.py | 7 + 9 files changed, 380 insertions(+), 1697 deletions(-) delete mode 100644 src/pipelines/clients/bcbs_promise/file_processing.py delete mode 100644 src/pipelines/clients/chc/file_processing.py delete mode 100644 src/pipelines/clients/clover/file_processing.py diff --git a/src/config.py b/src/config.py index 6e816fc..d4b55f6 100644 --- a/src/config.py +++ b/src/config.py @@ -160,6 +160,14 @@ 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 deleted file mode 100644 index 74a98b7..0000000 --- a/src/pipelines/clients/bcbs_promise/file_processing.py +++ /dev/null @@ -1,520 +0,0 @@ -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/chc/file_processing.py b/src/pipelines/clients/chc/file_processing.py deleted file mode 100644 index 636b2b5..0000000 --- a/src/pipelines/clients/chc/file_processing.py +++ /dev/null @@ -1,46 +0,0 @@ -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 deleted file mode 100644 index 222ea2b..0000000 --- a/src/pipelines/clients/clover/file_processing.py +++ /dev/null @@ -1,524 +0,0 @@ -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/runner.py b/src/pipelines/runner.py index be85ddc..7191bc5 100644 --- a/src/pipelines/runner.py +++ b/src/pipelines/runner.py @@ -1,8 +1,9 @@ """ Common Pipeline Runner -Handles shared orchestration (DTC, QC/QA, cache warming) and routes -to client-specific file_processing implementations. +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. Usage: from src.pipelines.runner import main @@ -13,12 +14,16 @@ 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.*") @@ -27,7 +32,12 @@ 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 @@ -35,6 +45,8 @@ 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, ) @@ -57,10 +69,16 @@ def get_file_processing_module(client: str): if registry.is_vendor(client): from src.pipelines.vendors.shared.generic_processor import VendorProcessor + print( + f"[DEBUG ROUTING] runner.get_file_processing_module: client='{client}' is a vendor, using VendorProcessor" + ) return VendorProcessor(client) # Set active client so shared resolver can select function-level overrides. set_active_client(client) + print( + f"[DEBUG ROUTING] runner.get_file_processing_module: set_active_client('{client}'), using shared resolver shim" + ) from src.pipelines.shared import file_processing return file_processing @@ -141,9 +159,13 @@ def main(client: str = "saas", testing=False, test_params={}): # Ensure dynamic fallback resolution targets the requested client. set_active_client(client) + print( + f"[DEBUG ROUTING] runner.main: received client='{client}', set_active_client called" + ) # Get client-specific file processing module file_processing = get_file_processing_module(client) + print(f"[DEBUG ROUTING] runner.main: file_processing module = {file_processing}") logging.info(f"Using {client} pipeline") # Set up per-file logging @@ -152,22 +174,39 @@ 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 - constants = Constants() + with timing_utils.timed_block("read_constants", log_level="INFO"): + constants = Constants() # Read and process input - 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"] - } + 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.debug(f"Total Input Files : {len(input_dict)}") + 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)}" + ) # ========== COMMON: Document Type Classification ========== if config.PERFORM_DTC: @@ -183,97 +222,169 @@ def main(client: str = "saas", testing=False, test_params={}): # ========== COMMON: Warm prompt caches ========== logging.info("Warming prompt caches before parallel processing...") - 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" + 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" ) - 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)}") + 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() - ] - 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) + 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) # 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 - ) + 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() - 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, ignore_index=True) - else: - ERROR_RESULT_DF = 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" + ) # ========== 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) - 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..." + 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 ) - FINAL_RESULT_DF_CC = FINAL_RESULT_DF_CC.reset_index(drop=True) + logging.info("QC/QA validation on CC results completed 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") + 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") - # 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...") + 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) # ========== COMMON: Output & Parent-Child ========== @@ -282,63 +393,68 @@ def main(client: str = "saas", testing=False, test_params={}): config.BATCH_ID, run_timestamp ) - 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", + 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" ) - # Write error file - if not ERROR_RESULT_DF.empty: - io_utils.write_s3(ERROR_RESULT_DF, "", run_timestamp, "error") + 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 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 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.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 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" ) - # Write error file - if not ERROR_RESULT_DF.empty: - io_utils.write_local(ERROR_RESULT_DF, "", run_timestamp, "error") + 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 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 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: @@ -346,28 +462,36 @@ 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.debug( + 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" + 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_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.debug("Exporting usage and cost tracking data locally...") + 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 diff --git a/src/pipelines/saas/file_processing.py b/src/pipelines/saas/file_processing.py index 7e28bdc..58a99f3 100644 --- a/src/pipelines/saas/file_processing.py +++ b/src/pipelines/saas/file_processing.py @@ -23,6 +23,10 @@ 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 @@ -186,6 +190,39 @@ 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 + print( + f"[DEBUG ROUTING] run_one_to_one_prompts: active_client='{client_name}', client_prompts_path={client_prompts_path}" + ) + + 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) + print( + f"[DEBUG ROUTING] run_one_to_one_prompts: combined {len(smart_chunked.fields)} smart_chunked client fields into HSC: {[f.field_name for f in smart_chunked.fields]}" + ) + + # 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 + print( + f"[DEBUG ROUTING] run_one_to_one_prompts: queued {len(full_context.fields)} full_context client fields for extraction: {[f.field_name for f in full_context.fields]}" + ) + + 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) @@ -224,6 +261,39 @@ 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: + print( + f"[DEBUG ROUTING] run_one_to_one_prompts: extracting {len(client_full_context_fields.fields)} full_context fields for '{client_name}'" + ) + 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, + ) + print( + f"[DEBUG ROUTING] run_one_to_one_prompts: full_context extraction complete, got {len(fc_answers)} answers: {list(fc_answers.keys())}" + ) + 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" + print( + f"[DEBUG ROUTING] run_one_to_one_prompts: OFFSET_TERM='{offset_term[:80]}...', derived OFFSET_INDICATOR='{one_to_one_results['OFFSET_INDICATOR']}'" + ) + ################## 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 057c690..a24984a 100644 --- a/src/pipelines/saas/main.py +++ b/src/pipelines/saas/main.py @@ -1,471 +1,28 @@ -import concurrent.futures -import logging -import os -import random -import time -import traceback -from datetime import datetime +""" +Pipeline entry point — delegates to the common runner. -import pandas as pd +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(). +""" -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.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, -) +from src.pipelines import runner -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 _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 main(testing=False, 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)}" + client = _resolve_client() + print( + f"[DEBUG ROUTING] main.py: resolved client='{client}' from config.CLIENT={config.CLIENT}, delegating to runner.main()" ) - - # 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 + return runner.main(client=client, testing=testing, test_params=test_params) if __name__ == "__main__": diff --git a/src/pipelines/shared/file_processing.py b/src/pipelines/shared/file_processing.py index 750b8a2..8477705 100644 --- a/src/pipelines/shared/file_processing.py +++ b/src/pipelines/shared/file_processing.py @@ -39,7 +39,14 @@ class ClientResolver: if client_name != "saas": client_module = cls.load_module(cls.get_client_module_name(client_name)) if client_module is not None and hasattr(client_module, name): + print( + f"[DEBUG ROUTING] file_processing resolver: '{name}' → client '{client_name}' override" + ) return getattr(client_module, name) + else: + print( + f"[DEBUG ROUTING] file_processing resolver: '{name}' → SaaS fallthrough (client='{client_name}', no override found)" + ) return getattr(saas_module, name) diff --git a/src/pipelines/shared/prompts/prompt_calls.py b/src/pipelines/shared/prompts/prompt_calls.py index 52bdc32..c08431d 100644 --- a/src/pipelines/shared/prompts/prompt_calls.py +++ b/src/pipelines/shared/prompts/prompt_calls.py @@ -39,7 +39,14 @@ class ClientResolver: if client_name != "saas": client_module = cls.load_module(cls.get_client_module_name(client_name)) if client_module is not None and hasattr(client_module, name): + print( + f"[DEBUG ROUTING] prompt_calls resolver: '{name}' → client '{client_name}' override" + ) return getattr(client_module, name) + else: + print( + f"[DEBUG ROUTING] prompt_calls resolver: '{name}' → SaaS fallthrough (client='{client_name}', no override found)" + ) return getattr(saas_module, name)