Merged in bugfix/retire_stale_client_file_processing (pull request #961)
Return None for dashboard output when dashboard postprocessing is off
* Return None for dashboard output when dashboard postprocessing is off
FINAL_RESULT_DF_DASHBOARD was initialized as an empty DataFrame even
when RUN_DASHBOARD_POSTPROCESSING was False, causing downstream code
to needlessly process it (reorder_columns, etc). Now returns None
when dashboard is not requested, matching the postprocess() contract.
* Merged dev into bugfix/retire_stale_client_file_processing
* Merge dev (with revert) into feature branch
* Re-apply retire stale client file_processing changes
Revert of the revert (4f53b528) to restore the original changes
from the feature branch for proper PR review.
* Merge branch 'bugfix/retire_stale_client_file_processing' of bitbucket.org:aarete/doczy.ai into bugfix/retire_stale_client_file_processing
Approved-by: Katon Minhas
This commit is contained in:
Praneel Panchigar
committed by
Katon Minhas
Katon Minhas
parent
4f53b5286e
commit
a936e0ce05
@@ -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
|
||||
|
||||
@@ -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
|
||||
@@ -50,4 +50,4 @@ def validate_reimbursements_for_llm(answer_dict: dict[str, str], filename: str)
|
||||
f"LLM response for reimbursement validation in {filename}:\n{llm_response}"
|
||||
)
|
||||
final_answer = _parser(llm_response)
|
||||
return final_answer == "YES"
|
||||
return "YES" in final_answer
|
||||
|
||||
@@ -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
|
||||
@@ -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
|
||||
@@ -49,5 +49,5 @@ def validate_reimbursements_for_llm(answer_dict: dict[str, str], filename: str)
|
||||
logging.debug(
|
||||
f"LLM response for reimbursement validation in {filename}:\n{llm_response}"
|
||||
)
|
||||
final_answer = _parser(llm_response).strip().upper()
|
||||
return final_answer == "YES"
|
||||
final_answer = _parser(llm_response)
|
||||
return "YES" in final_answer
|
||||
|
||||
+153
-32
@@ -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,
|
||||
)
|
||||
@@ -152,10 +164,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 +181,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 +212,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 +231,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 +258,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:
|
||||
@@ -232,29 +278,81 @@ def main(client: str = "saas", testing=False, test_params={}):
|
||||
successful_results_dashboard, ignore_index=True
|
||||
)
|
||||
else:
|
||||
FINAL_RESULT_DF_DASHBOARD = pd.DataFrame()
|
||||
FINAL_RESULT_DF_DASHBOARD = None
|
||||
else:
|
||||
FINAL_RESULT_DF_CC = pd.DataFrame()
|
||||
FINAL_RESULT_DF_DASHBOARD = pd.DataFrame()
|
||||
FINAL_RESULT_DF_DASHBOARD = None
|
||||
|
||||
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 (
|
||||
FINAL_RESULT_DF_DASHBOARD is not None
|
||||
and 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
|
||||
)
|
||||
if FINAL_RESULT_DF_DASHBOARD is not None:
|
||||
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 (
|
||||
FINAL_RESULT_DF_DASHBOARD is not None
|
||||
and 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 +362,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,14 +390,13 @@ 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
|
||||
FINAL_RESULT_DF_DASHBOARD is not None
|
||||
and not FINAL_RESULT_DF_DASHBOARD.empty
|
||||
):
|
||||
io_utils.write_s3(
|
||||
@@ -298,28 +405,31 @@ 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
|
||||
FINAL_RESULT_DF_DASHBOARD is not None
|
||||
and not FINAL_RESULT_DF_DASHBOARD.empty
|
||||
):
|
||||
io_utils.write_local(
|
||||
@@ -328,17 +438,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 +459,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 +479,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
|
||||
|
||||
|
||||
@@ -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
|
||||
@@ -191,6 +195,30 @@ def run_one_to_one_prompts(
|
||||
relationship="one_to_one", file_path=config.FIELD_JSON_PATH
|
||||
).combine(dynamic_one_to_one_fields)
|
||||
|
||||
################## LOAD CLIENT-SPECIFIC FIELDS ##################
|
||||
client_name = get_active_client()
|
||||
client_prompts_path = config.CLIENT_PROMPTS_MAP.get(client_name)
|
||||
client_full_context_fields = None
|
||||
|
||||
if client_prompts_path:
|
||||
client_fields = FieldSet(
|
||||
relationship="one_to_one", file_path=client_prompts_path
|
||||
)
|
||||
if client_fields.contains_fields():
|
||||
# smart_chunked fields (e.g. BCBS OFFSET_TERM) are combined into HSC fields
|
||||
smart_chunked = client_fields.filter(field_type="smart_chunked")
|
||||
if smart_chunked.contains_fields():
|
||||
one_to_one_fields = one_to_one_fields.combine(smart_chunked)
|
||||
|
||||
# full_context fields (e.g. Clover's 12 fields) are extracted separately after HSC
|
||||
full_context = client_fields.filter(field_type="full_context")
|
||||
if full_context.contains_fields():
|
||||
client_full_context_fields = full_context
|
||||
|
||||
logging.info(
|
||||
f"Loaded {len(client_fields.fields)} client-specific fields for {client_name}"
|
||||
)
|
||||
|
||||
# Filter fields if specific_fields is provided
|
||||
if specific_fields is not None:
|
||||
one_to_one_fields = one_to_one_fields.filter_by_names(specific_fields)
|
||||
@@ -229,6 +257,30 @@ def run_one_to_one_prompts(
|
||||
# Add HSC's N/A if field doesn't exist yet
|
||||
one_to_one_results[key] = value
|
||||
|
||||
################## RUN CLIENT FULL-CONTEXT FIELDS ##################
|
||||
# Client-specific full_context fields (e.g. Clover's 12 audit/filing/CA fields)
|
||||
if client_full_context_fields is not None:
|
||||
with timing_utils.timed_block(
|
||||
f"one_to_one.{client_name}_full_context", context=filename
|
||||
):
|
||||
fc_answers = vendor_run_full_context_fields(
|
||||
contract_text=contract_text,
|
||||
fields=client_full_context_fields,
|
||||
filename=filename,
|
||||
usage_prefix=client_name.upper(),
|
||||
constants=constants,
|
||||
)
|
||||
one_to_one_results.update(fc_answers)
|
||||
|
||||
################## DERIVE CLIENT-SPECIFIC FIELDS ##################
|
||||
# BCBS Promise: derive OFFSET_INDICATOR from OFFSET_TERM
|
||||
if "OFFSET_TERM" in one_to_one_results:
|
||||
offset_term = one_to_one_results.get("OFFSET_TERM", "N/A")
|
||||
if not string_utils.is_empty(offset_term):
|
||||
one_to_one_results["OFFSET_INDICATOR"] = "Y"
|
||||
else:
|
||||
one_to_one_results["OFFSET_INDICATOR"] = "N"
|
||||
|
||||
################## RUN PROVIDER INFO ##################
|
||||
if run_provider_info:
|
||||
with timing_utils.timed_block("one_to_one.provider_info", context=filename):
|
||||
|
||||
+14
-460
@@ -1,471 +1,25 @@
|
||||
import concurrent.futures
|
||||
import logging
|
||||
import os
|
||||
import random
|
||||
import time
|
||||
import traceback
|
||||
from datetime import datetime
|
||||
"""
|
||||
Pipeline entry point — delegates to the common runner.
|
||||
|
||||
import pandas as pd
|
||||
This module exists for backward compatibility. All orchestration logic
|
||||
lives in src.pipelines.runner. Calling main() here determines the
|
||||
client from config.CLIENT and delegates to runner.main().
|
||||
"""
|
||||
|
||||
random.seed(42)
|
||||
|
||||
import src.pipelines.saas.file_processing as file_processing
|
||||
from src.pipelines.runtime_context import set_active_client
|
||||
from src.pipelines.shared.extraction import one_to_one_funcs
|
||||
from src.pipelines.shared.postprocessing import postprocessing_funcs
|
||||
from src.constants.investment_columns import FIELD_FORMAT_MAPPING
|
||||
import src.prompts.prompt_templates as prompt_templates
|
||||
import src.utils.io_utils as io_utils
|
||||
import src.utils.llm_utils as llm_utils
|
||||
import src.utils.logging_utils as logging_utils
|
||||
import src.utils.usage_tracking as usage_tracking
|
||||
import src.utils.timing_utils as timing_utils
|
||||
import src.utils.duplicate_detection as duplicate_detection
|
||||
from src.constants.constants import Constants
|
||||
from src import config
|
||||
from src.parent_child import main as parent_child_main
|
||||
from src.document_classification import main as dtc_main
|
||||
from src.qc_qa.pipeline import (
|
||||
generate_statistics,
|
||||
generate_tin_contract_summary,
|
||||
generate_tin_statistics,
|
||||
run_qc_qa_pipeline,
|
||||
save_qc_qa_outputs,
|
||||
)
|
||||
from src.pipelines import runner
|
||||
|
||||
|
||||
def safe_process_file(item, constants, run_timestamp):
|
||||
"""Process a single file with comprehensive error handling.
|
||||
|
||||
This function wraps the main file processing logic to ensure that:
|
||||
- Individual file failures don't crash the entire batch
|
||||
- Detailed error information is captured for debugging
|
||||
|
||||
Args:
|
||||
item: Tuple of (filename, contract_text) representing the file to process
|
||||
constants: Constants object containing configuration and processing rules
|
||||
run_timestamp: Timestamp string for output file naming and tracking
|
||||
|
||||
Returns:
|
||||
tuple: Either:
|
||||
- (cc_df, dashboard_df) tuple with extracted field results (success case)
|
||||
- (error_df, error_df) tuple with error details (failure case)
|
||||
|
||||
Note:
|
||||
Returns both CC and dashboard versions from file processing.
|
||||
With FAISS migration, no special resource cleanup is needed.
|
||||
"""
|
||||
file_id = (
|
||||
item[0] if isinstance(item, (list, tuple)) and len(item) > 0 else item
|
||||
) # unpack file_id from item (passed in as a tuple below)
|
||||
try:
|
||||
cc_df, dashboard_df = file_processing.process_file(
|
||||
item, constants, run_timestamp
|
||||
)
|
||||
return cc_df, dashboard_df
|
||||
except (
|
||||
Exception
|
||||
) as e: # When there's an issue with the processing inside the future
|
||||
error_type = type(e).__name__
|
||||
error_message = str(e)
|
||||
full_traceback = traceback.format_exc()
|
||||
|
||||
logging.error(f"Error processing file {file_id}")
|
||||
logging.error(f"Error type: {error_type}")
|
||||
logging.error(f"Error Message: {error_message}")
|
||||
|
||||
logging.error(f"Full traceback:\n{full_traceback}")
|
||||
|
||||
error_df = pd.DataFrame(
|
||||
[
|
||||
{
|
||||
"FILE_NAME": file_id,
|
||||
"error": error_message.replace(",", ";").replace(
|
||||
"\n", " "
|
||||
), # Replace problematic characters for CSV compatibility
|
||||
"error_type": error_type,
|
||||
"traceback": full_traceback.replace(",", ";").replace(
|
||||
"\n", " | "
|
||||
), # keep line breaks as separators
|
||||
}
|
||||
]
|
||||
) # Return a single-row dataframe so we can still concat it later
|
||||
return error_df, error_df # Return tuple for consistency
|
||||
def _resolve_client() -> str:
|
||||
"""Determine the client name from CLI config."""
|
||||
if config.CLIENT and len(config.CLIENT) == 1 and config.CLIENT[0] != "None":
|
||||
return config.CLIENT[0]
|
||||
return "saas"
|
||||
|
||||
|
||||
def main(testing=False, test_params={}):
|
||||
set_active_client("saas")
|
||||
|
||||
# Check if batch_id is specified
|
||||
if not config.BATCH_ID:
|
||||
raise ValueError(
|
||||
"batch_id is required. Please specify batch_id in your command"
|
||||
)
|
||||
|
||||
# Check if max_workers is not negative
|
||||
if config.MAX_WORKERS < 0:
|
||||
raise ValueError(
|
||||
f"Invalid configuration: MAX_WORKERS must be non-negative (got {config.MAX_WORKERS})"
|
||||
)
|
||||
|
||||
# Set up per-file logging (creates separate log files for each document)
|
||||
logging_utils.setup_per_file_logging()
|
||||
|
||||
# Windows paths cannot contain ':'; use '-' in time component
|
||||
run_timestamp = datetime.now().strftime(f"run_%Y%m%d_%H-%M_{config.BATCH_ID}")
|
||||
|
||||
# Track overall pipeline time
|
||||
pipeline_start = time.time()
|
||||
|
||||
# Read Constants
|
||||
with timing_utils.timed_block("read_constants", log_level="INFO"):
|
||||
constants = Constants()
|
||||
|
||||
# Read and process input
|
||||
with timing_utils.timed_block("read_input", log_level="INFO"):
|
||||
if not testing:
|
||||
input_dict = io_utils.read_input()
|
||||
max_workers = config.MAX_WORKERS
|
||||
else:
|
||||
input_dict = io_utils.read_input(test_params["local_input_dir"])
|
||||
max_workers = test_params["max_workers"]
|
||||
if "input_files" in test_params:
|
||||
input_dict = {
|
||||
k: v
|
||||
for k, v in input_dict.items()
|
||||
if k in test_params["input_files"]
|
||||
}
|
||||
|
||||
logging.info(f"Total Input Files : {len(input_dict)}")
|
||||
|
||||
# Detect and handle duplicate contracts
|
||||
with timing_utils.timed_block("duplicate_detection", log_level="INFO"):
|
||||
duplicate_groups = duplicate_detection.find_duplicate_groups(input_dict)
|
||||
original_files_to_process, file_status_map = (
|
||||
duplicate_detection.get_files_to_process(input_dict, duplicate_groups)
|
||||
)
|
||||
logging.debug(
|
||||
f"Saas main: duplicate_groups={len(duplicate_groups)}, original_files={len(original_files_to_process)}, file_status_map length={len(file_status_map)}"
|
||||
)
|
||||
|
||||
# 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
|
||||
client = _resolve_client()
|
||||
return runner.main(client=client, testing=testing, test_params=test_params)
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
|
||||
@@ -511,6 +511,10 @@ def reorder_columns(
|
||||
# Reorder columns to match column_order
|
||||
final_column_order = [col for col in column_order if col in df_copy.columns]
|
||||
|
||||
# Append any extra columns not in column_order (e.g. client-specific fields)
|
||||
extra_columns = [col for col in df_copy.columns if col not in column_order]
|
||||
final_column_order.extend(extra_columns)
|
||||
|
||||
# Return the DataFrame with reordered columns
|
||||
return df_copy[final_column_order]
|
||||
|
||||
|
||||
Reference in New Issue
Block a user