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
2026-02-23 20:27:31 +00:00
parent a4c5e80161
commit 14cb727967
8 changed files with 205 additions and 205 deletions
@@ -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 = (
@@ -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:
@@ -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

@@ -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(
@@ -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
@@ -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


@@ -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


@@ -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",