jacob-mathison/doczyai-pipelines
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Activity

Retire stale client file_processing forks, unify pipeline entry point

Browse Source 
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.
This commit is contained in:
ppanchigar
2026-04-14 10:39:51 -05:00
parent 96967d14f3
commit 5e143c10f0
9 changed files with 380 additions and 1697 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src/config.py
+8
View File
@@ -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
src/pipelines/clients/bcbs_promise/file_processing.py
-520
View File
@@ -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
src/pipelines/clients/chc/file_processing.py
-46
View File
@@ -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
src/pipelines/clients/clover/file_processing.py
-524
View File
@@ -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
src/pipelines/runner.py
+152 -28
View File
@@ -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,10 +174,15 @@ 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()
# 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
@@ -164,10 +191,22 @@ def main(client: str = "saas", testing=False, test_params={}):
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"]
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,13 +222,16 @@ 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"
instruction=instruction,
cache_key=cache_key,
model_id="sonnet_latest",
)
logging.debug(
logging.info(
f"Successfully warmed {len(cacheable_instructions)} prompt caches"
)
except Exception as e:
@@ -199,14 +241,21 @@ def main(client: str = "saas", testing=False, test_params={}):
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:
# 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()
@@ -219,12 +268,19 @@ def main(client: str = "saas", testing=False, test_params={}):
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
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:
@@ -237,24 +293,69 @@ def main(client: str = "saas", testing=False, test_params={}):
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"
)
# ========== 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:
# 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..."
@@ -264,12 +365,22 @@ def main(client: str = "saas", testing=False, test_params={}):
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("QC/QA validation on CC results completed successfully")
# Generate validation statistics from CC version
logging.debug("Generating QC/QA statistics...")
logging.info("Generating QC/QA statistics...")
qc_qa_stats_df = generate_statistics(validated_cc_df)
logging.debug("QC/QA statistics generated successfully")
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(
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()}")
@@ -282,12 +393,11 @@ 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:
# 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
@@ -298,26 +408,29 @@ def main(client: str = "saas", testing=False, test_params={}):
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")
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...")
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")
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
@@ -328,17 +441,20 @@ def main(client: str = "saas", testing=False, test_params={}):
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")
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,9 +462,10 @@ 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}"
)
with timing_utils.timed_block("parent_child_mapping", log_level="INFO"):
pc_output_dir = os.path.join(
config.CONSOLIDATED_OUTPUT_DIRECTORY,
run_timestamp,
@@ -365,9 +482,16 @@ def main(client: str = "saas", testing=False, test_params={}):
)
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
src/pipelines/saas/file_processing.py
+70
View File
@@ -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):
src/pipelines/saas/main.py
+16 -459
View File
@@ -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"
client = _resolve_client()
print(
f"[DEBUG ROUTING] main.py: resolved client='{client}' from config.CLIENT={config.CLIENT}, delegating to runner.main()"
)
# 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
return runner.main(client=client, testing=testing, test_params=test_params)
if __name__ == "__main__":
src/pipelines/shared/file_processing.py
+7
View File
@@ -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)
src/pipelines/shared/prompts/prompt_calls.py
+7
View File
@@ -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)