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
+17 -460
View File
@@ -1,471 +1,28 @@
import concurrent.futures
import logging
import os
import random
import time
import traceback
from datetime import datetime
import pandas as pd
random.seed(42)
import src.pipelines.saas.file_processing as file_processing
from src.pipelines.runtime_context import set_active_client
from src.pipelines.shared.extraction import one_to_one_funcs
from src.pipelines.shared.postprocessing import postprocessing_funcs
from src.constants.investment_columns import FIELD_FORMAT_MAPPING
import src.prompts.prompt_templates as prompt_templates
import src.utils.io_utils as io_utils
import src.utils.llm_utils as llm_utils
import src.utils.logging_utils as logging_utils
import src.utils.usage_tracking as usage_tracking
import src.utils.timing_utils as timing_utils
import src.utils.duplicate_detection as duplicate_detection
from src.constants.constants import Constants
from src import config
from src.parent_child import main as parent_child_main
from src.document_classification import main as dtc_main
from src.qc_qa.pipeline import (
generate_statistics,
generate_tin_contract_summary,
generate_tin_statistics,
run_qc_qa_pipeline,
save_qc_qa_outputs,
)
def 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()
Pipeline entry point — delegates to the common runner.
logging.error(f"Error processing file {file_id}")
logging.error(f"Error type: {error_type}")
logging.error(f"Error Message: {error_message}")
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().
"""
logging.error(f"Full traceback:\n{full_traceback}")
from src import config
from src.pipelines import runner
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)