From 14cb72796756ca8eb8fd2655e145a98af32fbdce Mon Sep 17 00:00:00 2001 From: Faizan Mohiuddin Date: Mon, 23 Feb 2026 20:27:31 +0000 Subject: [PATCH] Merged in improve-logging (pull request #886) Improve logging * Add timing blocks for comprehensive pipeline logging - Add timing_utils.timed_block() to key extraction functions in one_to_n_funcs.py: - exhibit_level sub-functions (prompt_exhibit_level, dynamic_primary, etc.) - reimbursement_level extraction and cleaning - breakout functions (methodology_breakout, special_case_breakout) - carveout_and_special_case parallel processing - one_to_n_cleaning sub-steps (crosswalk, lob_relationship, split_reimb_dates) - Add timing to dynamic_funcs.py: - dynamic_assignment parallel processing - Add timing to postprocess.py: - standard_postprocess, contract_config_postprocess, dashboard_postprocess - Add timing to preprocess.py: - exhibit_chunking sub-steps (get_exhibit_pages, link_exhibit_pages, chunk_by_exhibit) - Add info-level logging statements for completed operations with row/item counts * Standardize logging levels and remove deprecated code DAIP2-1798: Ensure logging statements are at the right level - Demote per-file processing status messages from info to debug in file_processing.py - Demote operational details (counts, timing) from info to debug in runner.py - Demote detailed extraction messages from info to debug across extraction modules - Keep only high-level milestone messages (section headers, major ops) at info level DAIP2-1796: Ensure console output matches content of logs/ - Add LOG_LEVEL config setting for console output (defaults to INFO) - Update logging_utils.py to use config.LOG_LEVEL for console handler DAIP2-1797: Remove deprecated functionality - Remove deprecated check_and_combine_exhibit_inheritance function from one_to_n_funcs.py - Remove deprecated WRITE_PC_TO_S3 config setting (now uses WRITE_TO_S3) - Remove deprecated MODEL_STATS/GLOBAL_STATS comment from config.py * Merged DEV into improve-logging * Format code for improved readability * Remove duplicate logger.propagate line * Merge remote-tracking branch 'origin/DEV' into improve-logging Approved-by: Siddhant Medar --- src/config.py | 10 +- src/pipelines/runner.py | 18 +- src/pipelines/saas/file_processing.py | 22 +- .../shared/extraction/dynamic_funcs.py | 14 +- .../shared/extraction/one_to_n_funcs.py | 261 ++++++++---------- .../shared/postprocessing/postprocess.py | 12 +- .../shared/preprocessing/preprocess.py | 66 +++-- src/utils/logging_utils.py | 7 +- 8 files changed, 205 insertions(+), 205 deletions(-) diff --git a/src/config.py b/src/config.py index 8e7a835..2fdb804 100644 --- a/src/config.py +++ b/src/config.py @@ -95,6 +95,11 @@ RUN_MODE = get_arg_value("run_mode", "ec2") # Valid: local or ec2 READ_MODE = get_arg_value("read_mode", "s3") # Valid: local or s3 BATCH_ID = get_arg_value("batch_id", None) # BATCH_ID is mandatory +# Logging settings +LOG_LEVEL = get_arg_value( + "log_level", "INFO" +).upper() # Standard logging level for console output + # Per-file logging settings (for debugging) ENABLE_PER_FILE_LOGGING = get_arg_value("enable_per_file_logging", "False") == "True" PER_FILE_LOG_LEVEL = get_arg_value( @@ -349,8 +354,6 @@ MAX_ROWS_PER_SPLIT = int(get_arg_value("max_rows_per_split", "70000")) ######################################## EC2 ######################################## -# DEPRECATED: MODEL_STATS and GLOBAL_STATS are no longer updated. - def assume_role_and_get_runtime(account_id, role_name): sts_client = boto3.client("sts") @@ -388,9 +391,6 @@ PERFORM_PARENT_CHILD_MAPPING = get_arg_value("perform_pc", "True") == "True" DOCZY_OUTPUT_FOR_PC = get_arg_value( "doczy_output_for_pc", "" ) # Pass either S3 URI path, or local path -# TODO: Remove WRITE_PC_TO_S3 after approval. Parent-child S3 upload now uses WRITE_TO_S3 -# via io_utils.upload_local_file_to_s3 in the pipeline. -WRITE_PC_TO_S3 = get_arg_value("write_pc_to_s3", "False") == "True" ############## DOCUMENT TYPE CLASSIFICATION SETTINGS ############## PERFORM_DTC = ( diff --git a/src/pipelines/runner.py b/src/pipelines/runner.py index d8f6c42..6f61eae 100644 --- a/src/pipelines/runner.py +++ b/src/pipelines/runner.py @@ -157,7 +157,7 @@ def main(client: str = "saas", testing=False, test_params={}): 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)}") + logging.debug(f"Total Input Files : {len(input_dict)}") # ========== COMMON: Document Type Classification ========== if config.PERFORM_DTC: @@ -179,7 +179,9 @@ def main(client: str = "saas", testing=False, test_params={}): llm_utils.warm_prompt_cache( instruction=instruction, cache_key=cache_key, model_id="sonnet_latest" ) - logging.info(f"Successfully warmed {len(cacheable_instructions)} prompt caches") + logging.debug( + f"Successfully warmed {len(cacheable_instructions)} prompt caches" + ) except Exception as e: logging.warning(f"Error warming caches (will proceed anyway): {str(e)}") @@ -252,12 +254,12 @@ 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.info("QC/QA validation on CC results completed successfully") + logging.debug("QC/QA validation on CC results completed successfully") # Generate validation statistics from CC version - logging.info("Generating QC/QA statistics...") + logging.debug("Generating QC/QA statistics...") qc_qa_stats_df = generate_statistics(validated_cc_df) - logging.info("QC/QA statistics generated successfully") + logging.debug("QC/QA statistics generated successfully") except Exception as e: logging.error(f"QC/QA validation on CC results failed: {str(e)}") logging.error(f"Full traceback:\n{traceback.format_exc()}") @@ -297,7 +299,7 @@ def main(client: str = "saas", testing=False, test_params={}): io_utils.write_s3(qc_qa_stats_df, "", run_timestamp, "qc_qa_stats") if not per_file_usage_df.empty and not batch_summary_df.empty: - logging.info("Exporting usage and cost tracking data to S3...") + logging.debug("Exporting usage and cost tracking data to S3...") io_utils.write_s3(per_file_usage_df, "", run_timestamp, "usage") io_utils.write_s3(batch_summary_df, "", run_timestamp, "usage_summary") else: @@ -334,7 +336,7 @@ def main(client: str = "saas", testing=False, test_params={}): "Skipping Parent-Child mapping: No successful results to process (FINAL_RESULT_DF_CC is empty)" ) else: - logging.info( + logging.debug( f"Starting Parent-Child mapping with {len(FINAL_RESULT_DF_CC)} records from: {doczy_output_for_pc}" ) pc_output_dir = os.path.join( @@ -353,7 +355,7 @@ def main(client: str = "saas", testing=False, test_params={}): ) if not per_file_usage_df.empty and not batch_summary_df.empty: - logging.info("Exporting usage and cost tracking data locally...") + logging.debug("Exporting usage and cost tracking data locally...") io_utils.write_local(per_file_usage_df, "", run_timestamp, "usage") io_utils.write_local(batch_summary_df, "", run_timestamp, "usage_summary") else: diff --git a/src/pipelines/saas/file_processing.py b/src/pipelines/saas/file_processing.py index d1aa09e..eb1e276 100644 --- a/src/pipelines/saas/file_processing.py +++ b/src/pipelines/saas/file_processing.py @@ -40,7 +40,7 @@ def process_file(file_object, constants: Constants, run_timestamp): # 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}...") + logging.debug(f"{datetime_str()} Processing {filename}...") # Set default values dynamic_one_to_one_fields = FieldSet() @@ -77,12 +77,12 @@ def process_file(file_object, constants: Constants, run_timestamp): filename=filename, ) ) - logging.info(f"{datetime_str()} Preprocessing Complete - {filename}") + logging.debug(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( + logging.debug( 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): @@ -109,17 +109,17 @@ def process_file(file_object, constants: Constants, run_timestamp): one_to_n_results = postprocessing_funcs.generate_reimb_ids( one_to_n_results ) - logging.info(f"{datetime_str()} One to N Complete - {filename}") + logging.debug(f"{datetime_str()} One to N Complete - {filename}") else: first_reimbursement_page = "1" - logging.info( + logging.debug( 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( + logging.debug( f"{datetime_str()} Fields not configured for One to N, Skipping - {filename}" ) @@ -139,7 +139,7 @@ def process_file(file_object, constants: Constants, run_timestamp): specific_fields=specific_fields, ) one_to_one_results["FILE_NAME"] = filename - logging.info(f"{datetime_str()} One to One Complete - {filename}") + logging.debug(f"{datetime_str()} One to One Complete - {filename}") # Decide how to handle one_to_one results with timing_utils.timed_block( @@ -154,7 +154,7 @@ def process_file(file_object, constants: Constants, run_timestamp): # ONLY one_to_one processed - convert dict to DataFrame final_results = pd.DataFrame([one_to_one_results]) else: - logging.info( + logging.debug( f"{datetime_str()} Fields not configured for One to One, Skipping - {filename}" ) @@ -163,12 +163,12 @@ def process_file(file_object, constants: Constants, run_timestamp): 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}") + logging.debug(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}") + logging.debug(f"{datetime_str()} Postprocessing Complete - {filename}") ################## WRITE INDIVIDUAL ################## with timing_utils.timed_block("write_individual", context=filename): @@ -177,7 +177,7 @@ def process_file(file_object, constants: Constants, run_timestamp): else: io_utils.write_local(cc_df, filename, "", "individual_cc") - logging.info(f"{datetime_str()} Writing Complete - {filename}") + logging.debug(f"{datetime_str()} Writing Complete - {filename}") return cc_df, dashboard_df diff --git a/src/pipelines/shared/extraction/dynamic_funcs.py b/src/pipelines/shared/extraction/dynamic_funcs.py index 942aa36..188e472 100644 --- a/src/pipelines/shared/extraction/dynamic_funcs.py +++ b/src/pipelines/shared/extraction/dynamic_funcs.py @@ -4,7 +4,7 @@ from typing import TYPE_CHECKING, Optional from src.pipelines.saas.prompts import prompt_calls from src.pipelines.shared.preprocessing import preprocessing_funcs -from src.utils import string_utils +from src.utils import string_utils, timing_utils from src.constants.constants import Constants from src import config from src.prompts import prompt_templates @@ -219,8 +219,16 @@ def dynamic_assignment( return updated max_workers = min(len(reimbursement_primary_answers), 8) - with concurrent.futures.ThreadPoolExecutor(max_workers=max_workers) as executor: - return list(executor.map(_assign_single, reimbursement_primary_answers)) + with timing_utils.timed_block( + "dynamic_assignment.parallel_processing", context=filename + ): + with concurrent.futures.ThreadPoolExecutor(max_workers=max_workers) as executor: + results = list(executor.map(_assign_single, reimbursement_primary_answers)) + + logging.debug( + f"Dynamic assignment complete: {len(results)} rows processed with {len(dynamic_reimbursement_fields.fields)} dynamic fields - {filename}" + ) + return results def add_one_to_one_field( diff --git a/src/pipelines/shared/extraction/one_to_n_funcs.py b/src/pipelines/shared/extraction/one_to_n_funcs.py index 3b6180f..ac37b3f 100644 --- a/src/pipelines/shared/extraction/one_to_n_funcs.py +++ b/src/pipelines/shared/extraction/one_to_n_funcs.py @@ -4,7 +4,7 @@ import logging from src.pipelines.shared.extraction import dynamic_funcs from src.pipelines.saas.prompts import prompt_calls from src.pipelines.shared.postprocessing import aarete_derived, postprocessing_funcs -from src.utils import llm_utils, string_utils +from src.utils import llm_utils, string_utils, timing_utils import src.prompts.prompt_templates as prompt_templates from src.constants.constants import Constants from src import config @@ -56,50 +56,63 @@ def exhibit_level( # True Exhibit Level - these apply to the entire Exhibit exhibit_level_answers = {} if exhibit_level_fields.contains_fields(): - exhibit_level_answers = prompt_calls.prompt_exhibit_level( - exhibit_text, exhibit_level_fields, constants, filename - ) + with timing_utils.timed_block( + "exhibit_level.prompt_exhibit_level", context=filename + ): + exhibit_level_answers = prompt_calls.prompt_exhibit_level( + exhibit_text, exhibit_level_fields, constants, filename + ) - exhibit_level_answers = prompt_calls.prompt_exhibit_level_breakout( - exhibit_level_answers, filename - ) + with timing_utils.timed_block( + "exhibit_level.prompt_exhibit_level_breakout", context=filename + ): + exhibit_level_answers = prompt_calls.prompt_exhibit_level_breakout( + exhibit_level_answers, filename + ) # Dynamic Primary (skip if filtering to specific fields that don't include dynamic_primary) dynamic_reimbursement_fields = FieldSet() if dynamic_primary_fields.contains_fields(): - exhibit_level_answers, dynamic_reimbursement_fields = ( - dynamic_funcs.dynamic_primary( - exhibit_text, - exhibit_level_answers, - constants, - filename, - dynamic_primary_fields, + with timing_utils.timed_block( + "exhibit_level.dynamic_primary", context=filename + ): + exhibit_level_answers, dynamic_reimbursement_fields = ( + dynamic_funcs.dynamic_primary( + exhibit_text, + exhibit_level_answers, + constants, + filename, + dynamic_primary_fields, + ) ) - ) # Dynamic Codes (skip if filtering to specific fields) if dynamic_code_fields.contains_fields(): - exhibit_level_answers, dynamic_reimbursement_fields = dynamic_funcs.dynamic( - exhibit_text, - exhibit_header, - exhibit_level_answers, - constants, - filename, - dynamic_code_fields, - dynamic_reimbursement_fields, - ) + with timing_utils.timed_block("exhibit_level.dynamic_codes", context=filename): + exhibit_level_answers, dynamic_reimbursement_fields = dynamic_funcs.dynamic( + exhibit_text, + exhibit_header, + exhibit_level_answers, + constants, + filename, + dynamic_code_fields, + dynamic_reimbursement_fields, + ) # Dynamic Reimb Info (skip if filtering to specific fields) if dynamic_reimb_info_fields.contains_fields(): - exhibit_level_answers, dynamic_reimbursement_fields = dynamic_funcs.dynamic( - exhibit_text, - exhibit_header, - exhibit_level_answers, - constants, - filename, - dynamic_reimb_info_fields, - dynamic_reimbursement_fields, - ) + with timing_utils.timed_block( + "exhibit_level.dynamic_reimb_info", context=filename + ): + exhibit_level_answers, dynamic_reimbursement_fields = dynamic_funcs.dynamic( + exhibit_text, + exhibit_header, + exhibit_level_answers, + constants, + filename, + dynamic_reimb_info_fields, + dynamic_reimbursement_fields, + ) # add exhibit identifiers exhibit_level_answers["EXHIBIT_PAGE"] = exhibit_page @@ -113,14 +126,20 @@ def reimbursement_level( ) -> list[dict[str, str]]: # Reimbursement Primary - reimbursement_primary_answers = prompt_calls.prompt_reimbursement_primary( - page_text, filename - ) + with timing_utils.timed_block( + "reimbursement_level.prompt_reimbursement_primary", context=filename + ): + reimbursement_primary_answers = prompt_calls.prompt_reimbursement_primary( + page_text, filename + ) if reimbursement_primary_answers: - reimbursement_primary_answers = clean_reimbursement_primary( - reimbursement_primary_answers, constants, filename - ) + with timing_utils.timed_block( + "reimbursement_level.clean_reimbursement_primary", context=filename + ): + reimbursement_primary_answers = clean_reimbursement_primary( + reimbursement_primary_answers, constants, filename + ) return reimbursement_primary_answers @@ -178,12 +197,14 @@ def breakout( - list[dict]: The processed reimbursement primary answers with breakout details. """ - reimbursement_level_answers = methodology_breakout( - reimbursement_level_answers, constants, filename - ) # Including base methodology breakout, grouper, and fee schedule + with timing_utils.timed_block("breakout.methodology_breakout", context=filename): + reimbursement_level_answers = methodology_breakout( + reimbursement_level_answers, constants, filename + ) # Including base methodology breakout, grouper, and fee schedule # Special Case Breakout - special_case_answers = special_case_breakout(special_case_answers, filename) + with timing_utils.timed_block("breakout.special_case_breakout", context=filename): + special_case_answers = special_case_breakout(special_case_answers, filename) return reimbursement_level_answers, special_case_answers @@ -269,30 +290,36 @@ def carveout_and_special_case( # Parallelize carveout processing if reimbursement_breakout_answers: - with concurrent.futures.ThreadPoolExecutor( - max_workers=min(len(reimbursement_breakout_answers), 5) - ) as executor: - futures = [ - executor.submit( - process_single_carveout, - answer_dict.copy(), - carveout_definitions, - special_case_definitions, - filename, - ) - for answer_dict in reimbursement_breakout_answers - ] + with timing_utils.timed_block( + "carveout_and_special_case.parallel_processing", context=filename + ): + with concurrent.futures.ThreadPoolExecutor( + max_workers=min(len(reimbursement_breakout_answers), 5) + ) as executor: + futures = [ + executor.submit( + process_single_carveout, + answer_dict.copy(), + carveout_definitions, + special_case_definitions, + filename, + ) + for answer_dict in reimbursement_breakout_answers + ] - for future in concurrent.futures.as_completed(futures): - try: - result_type, result_dict = future.result() - if result_type == "reimbursement": - reimbursement_level_answers.append(result_dict) - elif result_type == "special_case": - special_case_answers.append(result_dict) - except Exception as e: - logging.error(f"Error in carveout processing: {str(e)}") + for future in concurrent.futures.as_completed(futures): + try: + result_type, result_dict = future.result() + if result_type == "reimbursement": + reimbursement_level_answers.append(result_dict) + elif result_type == "special_case": + special_case_answers.append(result_dict) + except Exception as e: + logging.error(f"Error in carveout processing: {str(e)}") + logging.debug( + f"Carveout processing complete: {len(reimbursement_level_answers)} reimbursements, {len(special_case_answers)} special cases - {filename}" + ) return reimbursement_level_answers, special_case_answers @@ -795,16 +822,32 @@ def one_to_n_cleaning( all_exhibit_rows: list[dict], exhibit_text: str, constants: Constants, filename: str ): ################################ Crosswalk Fields ################################ - all_exhibit_rows = aarete_derived.get_crosswalk_fields(all_exhibit_rows, constants) + with timing_utils.timed_block( + "one_to_n_cleaning.crosswalk_fields", context=filename + ): + all_exhibit_rows = aarete_derived.get_crosswalk_fields( + all_exhibit_rows, constants + ) ################################ Determine LOB Relationship ################################ - all_exhibit_rows = get_lob_relationship(all_exhibit_rows, exhibit_text, filename) + with timing_utils.timed_block( + "one_to_n_cleaning.lob_relationship", context=filename + ): + all_exhibit_rows = get_lob_relationship( + all_exhibit_rows, exhibit_text, filename + ) ################################ Fill NA Mapping ################################ - all_exhibit_rows = aarete_derived.fill_na_mapping(all_exhibit_rows) + with timing_utils.timed_block( + "one_to_n_cleaning.fill_na_mapping", context=filename + ): + all_exhibit_rows = aarete_derived.fill_na_mapping(all_exhibit_rows) ################################ Split REIMB_DATES ################################ - all_exhibit_rows = split_reimb_dates(all_exhibit_rows, filename) + with timing_utils.timed_block( + "one_to_n_cleaning.split_reimb_dates", context=filename + ): + all_exhibit_rows = split_reimb_dates(all_exhibit_rows, filename) ################################ Normalize Single-Value Fields to Strings ################################ # REIMB_TERM, SERVICE_TERM should always be strings (not lists) @@ -933,83 +976,3 @@ def lesser_of_distribution( final_reimbursement_level_answers.append(answer_dict) return final_reimbursement_level_answers - - -# TODO: DELETE - Outdated. Replaced by Exhibit-based prev_exhibit inheritance in -# file_processing.run_one_to_n_prompts. Uses get_previous_exhibit_dynamic_fields() on -# Exhibit; requires running exhibit_level for exhibits without reimbursements (see docs). -def check_and_combine_exhibit_inheritance( - previous_exhibit: dict | None, - current_exhibit_page_nums: list[str], - current_exhibit_text: str, - current_dynamic_fields: FieldSet, -) -> tuple[bool, list[str], FieldSet]: - """ - Check if inheritance should happen and combine exhibits if needed. - - Conditions for inheritance: - 1. Previous exhibit exists - 2. Current exhibit's dynamic_reimbursement_fields is empty FieldSet - 3. Previous exhibit's dynamic_reimbursement_fields is NOT empty - 4. Previous exhibit had NO reimbursement rows - - When inheritance happens: - - Replace current dynamic_reimbursement_fields with previous exhibit's dynamic_reimbursement_fields - - Combine exhibit_page_nums (previous + current) to provide context through simplify_exhibit - - Args: - previous_exhibit: Dict with keys: 'exhibit_page_nums', 'dynamic_reimbursement_fields', 'had_reimbursements' - current_exhibit_page_nums: List of page numbers for current exhibit - current_exhibit_text: Full text of current exhibit (not used in return, but kept for consistency) - current_dynamic_fields: FieldSet of dynamic fields found in current exhibit - - Returns: - Tuple of: - - should_inherit: bool indicating if inheritance happened - - combined_page_nums: list[str] (combined if inheritance, original otherwise) - - updated_dynamic_fields: FieldSet (replaced with previous if inheritance, original otherwise) - """ - # Check if previous exhibit exists - if previous_exhibit is None: - return False, current_exhibit_page_nums, current_dynamic_fields - - # Check if current exhibit's dynamic_reimbursement_fields is empty - current_field_names = current_dynamic_fields.list_fields() - current_is_empty = len(current_field_names) == 0 - - if not current_is_empty: - return False, current_exhibit_page_nums, current_dynamic_fields - - # Check if previous exhibit's dynamic_reimbursement_fields is NOT empty - prev_dynamic_fields = previous_exhibit["dynamic_reimbursement_fields"] - prev_field_names = prev_dynamic_fields.list_fields() - prev_is_not_empty = len(prev_field_names) > 0 - - if not prev_is_not_empty: - return False, current_exhibit_page_nums, current_dynamic_fields - - # Check if previous exhibit had NO reimbursement rows - prev_no_reimbursements = not previous_exhibit.get("had_reimbursements", True) - - if not prev_no_reimbursements: - return False, current_exhibit_page_nums, current_dynamic_fields - - # All conditions met - inheritance should happen - # Combine exhibit_page_nums (previous + current) - # This ensures simplify_exhibit will include context from both exhibits - combined_page_nums = ( - previous_exhibit["exhibit_page_nums"] + current_exhibit_page_nums - ) - - # Replace current dynamic_reimbursement_fields with previous exhibit's - updated_dynamic_fields = prev_dynamic_fields - - # Minimal logging: inheritance activated, pages being merged, and fields inherited - prev_pages = ", ".join(previous_exhibit.get("exhibit_page_nums", [])) - current_pages = ", ".join(current_exhibit_page_nums) - fields_str = ", ".join(prev_field_names) - logging.debug( - f"Exhibit inheritance activated: merging pages {prev_pages} → {current_pages}, inheriting fields: {fields_str}" - ) - - return True, combined_page_nums, updated_dynamic_fields diff --git a/src/pipelines/shared/postprocessing/postprocess.py b/src/pipelines/shared/postprocessing/postprocess.py index 3d0d5ac..e3cb3a0 100644 --- a/src/pipelines/shared/postprocessing/postprocess.py +++ b/src/pipelines/shared/postprocessing/postprocess.py @@ -1,4 +1,5 @@ import json +import logging import pandas as pd @@ -6,6 +7,7 @@ import src.config as config from src.constants.constants import Constants from src.constants.investment_columns import FIELD_FORMAT_MAPPING from src.pipelines.shared.postprocessing import postprocessing_funcs +from src.utils import timing_utils def postprocess(df, constants: Constants): @@ -23,17 +25,21 @@ def postprocess(df, constants: Constants): dashboard_df is None when RUN_DASHBOARD_POSTPROCESSING is False. """ # Standard Postprocessing Steps for All Outputs - df = standard_postprocess(df, constants) + with timing_utils.timed_block("postprocess.standard_postprocess"): + df = standard_postprocess(df, constants) # Postprocessing Steps unique to Contract Config Output - cc_df = contract_config_postprocess(df, constants) + with timing_utils.timed_block("postprocess.contract_config_postprocess"): + cc_df = contract_config_postprocess(df, constants) # Postprocessing Steps unique to Dashboard Output (optional, off by default) if config.RUN_DASHBOARD_POSTPROCESSING: - dashboard_df = dashboard_postprocess(df, constants) + with timing_utils.timed_block("postprocess.dashboard_postprocess"): + dashboard_df = dashboard_postprocess(df, constants) else: dashboard_df = None + logging.debug(f"Postprocessing complete: {len(cc_df)} rows in CC output") return cc_df, dashboard_df diff --git a/src/pipelines/shared/preprocessing/preprocess.py b/src/pipelines/shared/preprocessing/preprocess.py index 8b666cc..7b9b2d8 100644 --- a/src/pipelines/shared/preprocessing/preprocess.py +++ b/src/pipelines/shared/preprocessing/preprocess.py @@ -3,6 +3,7 @@ from typing import TYPE_CHECKING from src import config from src.pipelines.shared.preprocessing import preprocessing_funcs +from src.utils import timing_utils if TYPE_CHECKING: from src.pipelines.shared.extraction.page_funcs import Page @@ -105,32 +106,53 @@ def one_to_n_exhibit_chunking( """ # Determine which dictionary to use if pages_dict is not None: - exhibit_pages, all_exhibit_headers = preprocessing_funcs.get_exhibit_pages( - pages_dict=pages_dict, - EXHIBIT_HEADER_MARKERS=EXHIBIT_HEADER_MARKERS, - filename=filename, - ) - exhibit_pages, all_exhibit_headers = preprocessing_funcs.link_exhibit_pages( - all_exhibit_headers, pages_dict=pages_dict, filename=filename - ) - exhibit_chunk_mapping = preprocessing_funcs.chunk_by_exhibit( - pages_dict=pages_dict, exhibit_pages=exhibit_pages - ) + with timing_utils.timed_block( + "exhibit_chunking.get_exhibit_pages", context=filename + ): + exhibit_pages, all_exhibit_headers = preprocessing_funcs.get_exhibit_pages( + pages_dict=pages_dict, + EXHIBIT_HEADER_MARKERS=EXHIBIT_HEADER_MARKERS, + filename=filename, + ) + with timing_utils.timed_block( + "exhibit_chunking.link_exhibit_pages", context=filename + ): + exhibit_pages, all_exhibit_headers = preprocessing_funcs.link_exhibit_pages( + all_exhibit_headers, pages_dict=pages_dict, filename=filename + ) + with timing_utils.timed_block( + "exhibit_chunking.chunk_by_exhibit", context=filename + ): + exhibit_chunk_mapping = preprocessing_funcs.chunk_by_exhibit( + pages_dict=pages_dict, exhibit_pages=exhibit_pages + ) elif text_dict is not None: - exhibit_pages, all_exhibit_headers = preprocessing_funcs.get_exhibit_pages( - text_dict=text_dict, - EXHIBIT_HEADER_MARKERS=EXHIBIT_HEADER_MARKERS, - filename=filename, - ) - exhibit_pages, all_exhibit_headers = preprocessing_funcs.link_exhibit_pages( - all_exhibit_headers, text_dict=text_dict, filename=filename - ) - exhibit_chunk_mapping = preprocessing_funcs.chunk_by_exhibit( - text_dict=text_dict, exhibit_pages=exhibit_pages - ) + with timing_utils.timed_block( + "exhibit_chunking.get_exhibit_pages", context=filename + ): + exhibit_pages, all_exhibit_headers = preprocessing_funcs.get_exhibit_pages( + text_dict=text_dict, + EXHIBIT_HEADER_MARKERS=EXHIBIT_HEADER_MARKERS, + filename=filename, + ) + with timing_utils.timed_block( + "exhibit_chunking.link_exhibit_pages", context=filename + ): + exhibit_pages, all_exhibit_headers = preprocessing_funcs.link_exhibit_pages( + all_exhibit_headers, text_dict=text_dict, filename=filename + ) + with timing_utils.timed_block( + "exhibit_chunking.chunk_by_exhibit", context=filename + ): + exhibit_chunk_mapping = preprocessing_funcs.chunk_by_exhibit( + text_dict=text_dict, exhibit_pages=exhibit_pages + ) else: raise ValueError("Either text_dict or pages_dict must be provided") + logging.debug( + f"Exhibit chunking complete: {len(exhibit_chunk_mapping)} exhibits identified - {filename}" + ) return exhibit_chunk_mapping, all_exhibit_headers diff --git a/src/utils/logging_utils.py b/src/utils/logging_utils.py index 9d638b5..4f78ee5 100644 --- a/src/utils/logging_utils.py +++ b/src/utils/logging_utils.py @@ -201,7 +201,9 @@ def setup_per_file_logging() -> None: sys.stdout.buffer, encoding="utf-8", errors="replace" ) console_handler = logging.StreamHandler(utf8_stdout) - console_handler.setLevel(logging.INFO) + # Use config.LOG_LEVEL for console output (defaults to INFO) + console_log_level = getattr(logging, config.LOG_LEVEL, logging.INFO) + console_handler.setLevel(console_log_level) console_formatter = logging.Formatter("%(asctime)s - %(levelname)s - %(message)s") console_handler.setFormatter(console_formatter) logger.addHandler(console_handler) @@ -209,9 +211,6 @@ def setup_per_file_logging() -> None: # Prevent propagation to avoid duplicate console output logger.propagate = False - # Prevent propagation to avoid duplicate console output - logger.propagate = False - # Suppress noisy third-party loggers that clutter output noisy_loggers = [ "botocore",