From 3331da2a8ce46c155ed7a1a566a138775d347734 Mon Sep 17 00:00:00 2001 From: Mayank Aamseek Date: Fri, 6 Mar 2026 20:40:20 +0000 Subject: [PATCH] Merged in bugfix/prov_info_json_fixes (pull request #899) Bugfix/prov info json fixes * fix: robust PROV_INFO_JSON sanitization and TIN backfill logic json_utils: - Add sanitize_prov_info_json with layered parsing (JSON, literal_eval, empty-value-after-colon fix, best-effort dict extraction). - Add _normalize_prov_entries and _prov_value_to_str for uniform str-valued output; flatten list values, strip TIN hyphens. - format_prov_info_json now delegates to sanitize_prov_info_json. postprocessing_funcs: - Add fill_prov_info_tin_from_filename_tin for TIN backfill. - Add validate_and_reformat_date (pipe-wrapped, datetime strings). - Add format_as_json_list (pipe-delimited, comma-separated, quote stripping). postprocess: - Integrate new postprocessing helpers into pipeline flow. postprocess_existing_output: - Support CSV and Excel input, configurable paths, fillna for CSV. tests: - Add test_json_parsers.py for PROV_INFO_JSON parsing coverage. - Add test_postprocess.py for date/list formatting and default_ind. * Merge branch 'DAIP2-1947-tin-and-prov-info-json-issues' into bugfix/prov_info_json_fixes * Merged DEV into bugfix/prov_info_json_fixes * Strip out unused functionality * Update filename_tin functionality * Add docstring * Update filename_tin cleaning in PROV_INFO_JSON * test prep * black format * missing function added * Merge branch 'DEV' into bugfix/prov_info_json_fixes * Black format * Strip unused functions * Strip unused code * update unit tests * Merge branch 'DEV' into bugfix/prov_info_json_fixes * Update test * Simplify process * handle list of group names * Resolve run_provider_info_field call * Merge branch 'DEV' into bugfix/prov_info_json_fixes * Correct type hints * Fix unit tests * Fix unit test * Remove redundant postprocessing_funcs Approved-by: Katon Minhas --- src/constants/regex_patterns.py | 1 + .../clients/bcbs_promise/file_processing.py | 2 +- .../clients/clover/file_processing.py | 2 +- src/pipelines/saas/file_processing.py | 19 +- src/pipelines/saas/prompts/prompt_calls.py | 19 +- .../shared/extraction/tin_npi_funcs.py | 521 ++++++------------ .../shared/postprocessing/postprocess.py | 8 +- .../postprocessing/postprocessing_funcs.py | 95 +++- src/prompts/prompt_templates.py | 22 +- src/scripts/postprocess_existing_output.py | 5 +- src/testbed/testbed_utils.py | 2 +- src/tests/test_json_parsers.py | 120 ++++ src/tests/test_postprocess.py | 50 +- src/tests/test_prompt_calls.py | 21 - src/tests/test_string_utils.py | 6 + src/tests/test_tin_npi_funcs.py | 106 +--- src/utils/json_utils.py | 80 +++ src/utils/string_utils.py | 18 + 18 files changed, 549 insertions(+), 548 deletions(-) diff --git a/src/constants/regex_patterns.py b/src/constants/regex_patterns.py index ccb816a..1147574 100644 --- a/src/constants/regex_patterns.py +++ b/src/constants/regex_patterns.py @@ -10,6 +10,7 @@ TRIPLE_BACKTICK_PATTERN = r"```([^`]+)```" # NPI_PATTERN = r"\b\d{10}\b" TIN_PATTERN = r"\b(?:\d[\s\-.]*){9}\b" FILE_NAME_TIN_PATTERN = r"(? dict[str, str]: def prompt_provider_info( - text_dict: dict, page_num: str, filename: str, payer_name: str -) -> list: + text_dict: dict, + page_num: str, + filename: str, + payer_name: str, + provider_fields: FieldSet, +) -> list[dict[str, str]]: """ Extracts provider information from a specified page in a document using a language model. @@ -1101,9 +1105,6 @@ def prompt_provider_info( ValueError: If the response from the language model cannot be parsed as valid JSON. """ page_text = text_dict[page_num] - provider_fields = FieldSet( - file_path=config.FIELD_JSON_PATH, field_type="provider_info" - ) prompt, _parser = prompt_templates.TIN_NPI_TEMPLATE( page_text, provider_fields.print_prompt_dict(), payer_name ) @@ -1136,12 +1137,8 @@ def prompt_provider_info( ) providers = [{"TIN": "N/A", "NPI": "N/A", "NAME": "PARSING_ERROR"}] - # Normalization is already done in the JSON parser - - # VALIDATION LAYER - Cross-check with regex findings - page_text = text_dict[page_num] - tin_npi_funcs.flag_missed_prov_info(page_text, providers, page_num, filename) - + for provider_dict in providers: + provider_dict["FROM_FILENAME"] = "N" return providers except ValueError as e: # Handle JSON parsing error diff --git a/src/pipelines/shared/extraction/tin_npi_funcs.py b/src/pipelines/shared/extraction/tin_npi_funcs.py index ba6df7e..18966fe 100644 --- a/src/pipelines/shared/extraction/tin_npi_funcs.py +++ b/src/pipelines/shared/extraction/tin_npi_funcs.py @@ -1,18 +1,18 @@ import json +from typing import Any import logging import re import src.constants.regex_patterns as regex_patterns import src.config as config import src.prompts.prompt_templates as prompt_templates -from src.utils import llm_utils, string_utils +from src.utils import llm_utils, string_utils, json_utils from src.pipelines.saas.prompts import prompt_calls from src.prompts.fieldset import Field, FieldSet +import pandas as pd -def run_provider_info_fields( - contract_text: str, text_dict: dict, filename: str, payer_name: str -): +def run_provider_info_fields(text_dict: dict, filename: str, payer_name: str): """ Extracts provider information from a contract text using regex-based identifier extraction, processes the relevant text chunks, and cleans the results. @@ -26,71 +26,62 @@ def run_provider_info_fields( list[dict]: A cleaned and standardized list containing provider information extracted from the contract text. """ - one_to_one_results = {} + prov_info_results = {} - all_tins, all_npis, tin_matches, npi_matches = get_all_tins_and_npis(contract_text) - - exact_tins, exact_npis, ocr_corrected_tins, ocr_corrected_npis = ( - get_tin_extraction_quality(tin_matches, npi_matches, filename) - ) - - relevant_pages = get_relevant_pages(all_tins, all_npis, text_dict, filename) + # Regex to collect all tins and npis + all_tins, all_npis, relevant_pages = get_all_tins_and_npis(text_dict) + # Prompt to get prov_info_json prov_info_json = get_prov_info_json(relevant_pages, text_dict, filename, payer_name) - - # Create columns - one_to_one_results["PROV_INFO_JSON"] = prov_info_json - one_to_one_results["TIN_EXTRACTION_QUALITY"] = ( - f"{len(exact_tins)} exact, {len(ocr_corrected_tins)} OCR-corrected" - ) - one_to_one_results["NPI_EXTRACTION_QUALITY"] = ( - f"{len(exact_npis)} exact, {len(ocr_corrected_npis)} OCR-corrected" - ) + prov_info_results["PROV_INFO_JSON"] = prov_info_json # Add FILENAME_TIN - filename_tin = get_all_matches(filename, regex_patterns.FILE_NAME_TIN_PATTERN) - one_to_one_results["FILENAME_TIN"] = filename_tin[0] if filename_tin else "N/A" + prov_info_results = add_filename_tin(prov_info_results, filename) - return one_to_one_results + return prov_info_results -def get_all_tins_and_npis(contract_text): - tin_matches = get_all_matches_with_ocr( - contract_text, - regex_patterns.TIN_PATTERN, - regex_patterns.TIN_OCR_PATTERN, - identifier_type="TIN", - ) - npi_matches = get_all_matches_with_ocr( - contract_text, - regex_patterns.NPI_PATTERN, - regex_patterns.NPI_OCR_PATTERN, - identifier_type="NPI", - ) - # Extract just the values from the (value, match_quality) tuples for downstream processing - all_tins = [tin for tin, quality in tin_matches] - all_npis = [npi for npi, quality in npi_matches] - logging.debug(f"Extracted TINs: {all_tins}") - logging.debug(f"Extracted NPIs: {all_npis}") +def get_all_tins_and_npis(text_dict: dict[str, str]): + all_tins = [] + all_npis = [] + relevant_pages = [] + seen_tins = set() + seen_npis = set() - return all_tins, all_npis, tin_matches, npi_matches + for page_key, page_text in text_dict.items(): + tin_matches = get_all_matches_with_ocr( + page_text, + regex_patterns.TIN_PATTERN, + regex_patterns.TIN_OCR_PATTERN, + identifier_type="TIN", + ) + npi_matches = get_all_matches_with_ocr( + page_text, + regex_patterns.NPI_PATTERN, + regex_patterns.NPI_OCR_PATTERN, + identifier_type="NPI", + ) + # Extract just the values from the (value, match_quality) tuples for downstream processing + page_tins = [tin for tin, quality in tin_matches] + page_npis = [npi for npi, quality in npi_matches] -def get_all_matches(text: str, pattern: str) -> list[str]: - """ - Extracts all unique matches of a given pattern from the provided text. + # Track pages that had at least one match + if page_tins or page_npis: + relevant_pages.append(page_key) - Args: - text (str): The input text to search for matches. - pattern (str): The regular expression pattern to match against the text. + # Add matches to all lists, avoiding duplicates + for tin in page_tins: + if tin not in seen_tins: + all_tins.append(tin) + seen_tins.add(tin) - Returns: - list[str]: A list of unique matches found in the text. If the pattern is None or empty, returns an empty list. - """ - if string_utils.is_empty(pattern): # check if pattern is None or empty - return [] - matches = re.findall(pattern, text) - return list(set(matches)) # Remove duplicates + for npi in page_npis: + if npi not in seen_npis: + all_npis.append(npi) + seen_npis.add(npi) + + return all_tins, all_npis, relevant_pages def get_all_matches_with_ocr( @@ -107,6 +98,7 @@ def get_all_matches_with_ocr( Returns: list[tuple]: A list of (value, match_quality) tuples where match_quality is "EXACT" or "OCR_CORRECTED". """ + results = [] seen_values = set() @@ -153,6 +145,23 @@ def get_all_matches_with_ocr( return results +def get_all_matches(text: str, pattern: str) -> list[str]: + """ + Extracts all unique matches of a given pattern from the provided text. + + Args: + text (str): The input text to search for matches. + pattern (str): The regular expression pattern to match against the text. + + Returns: + list[str]: A list of unique matches found in the text. If the pattern is None or empty, returns an empty list. + """ + if string_utils.is_empty(pattern): # check if pattern is None or empty + return [] + matches = re.findall(pattern, text) + return list(set(matches)) # Remove duplicates + + def attempt_ocr_correction(candidate: str, identifier_type: str) -> tuple[str, bool]: """Attempt OCR correction on a candidate string @@ -204,165 +213,25 @@ def attempt_ocr_correction(candidate: str, identifier_type: str) -> tuple[str, b return candidate, False -def normalize_to_lists(provider: dict): - - tins = provider.get("TIN") - npis = provider.get("NPI") - names = provider.get("NAME") - - # Normalize to lists: if string, convert to list to avoid character splitting - if isinstance(tins, str): - tins = [tins] - elif not isinstance(tins, list): - tins = [] - if isinstance(npis, str): - npis = [npis] - elif not isinstance(npis, list): - npis = [] - if isinstance(names, str): - names = [names] - elif not isinstance(names, list): - names = [] - - return tins, npis, names - - -def deunknown_and_deduplicate( - group_tins, group_npis, group_names, other_tins, other_npis, other_names +def get_prov_info_json( + relevant_pages: list[str], text_dict: dict[str, str], filename: str, payer_name: str ): - # De-Unknown GROUP fields - group_tins = [v for v in group_tins if not string_utils.is_empty(v)] - group_npis = [v for v in group_npis if not string_utils.is_empty(v)] - group_names = [v for v in group_names if not string_utils.is_empty(v)] + if not relevant_pages: + return [] - # Deduplicate while preserving order and remove any "UNKNOWN" entries and overlaps between group and other - group_tins = list(dict.fromkeys(group_tins)) - other_tins = list(dict.fromkeys(other_tins)) - other_tins = [ - v for v in other_tins if v not in group_tins and not string_utils.is_empty(v) - ] - - group_npis = list(dict.fromkeys(group_npis)) - other_npis = list(dict.fromkeys(other_npis)) - other_npis = [ - v for v in other_npis if v not in group_npis and not string_utils.is_empty(v) - ] - - group_names = list(dict.fromkeys(group_names)) - other_names = list(dict.fromkeys(other_names)) - other_names = [ - v for v in other_names if v not in group_names and not string_utils.is_empty(v) - ] - - return group_tins, group_npis, group_names, other_tins, other_npis, other_names - - -def merge_provider_info_with_one_to_one(one_to_one_results: dict, filename: str): - """ - Merges provider_info into one_to_one_results. - - Args: - one_to_one_results (dict): Dictionary of one-to-one field results. - provider_info (list[dict]): List of dictionaries containing provider information. - Each dictionary is keyed by TIN, NPI, NAME - - Returns: - dict: Updated one_to_one_results with merged values from provider_info. - """ - # Extract group and non-group providers - group_tins, group_npis, group_names = set(), set(), set() - other_tins, other_npis, other_names = [], [], [] - - provider_name = one_to_one_results.get("PROVIDER_NAME") # Can be List or string - - # Normalize provider_name to list: if string, convert to list to avoid character splitting - if isinstance(provider_name, str): - provider_name = [provider_name] - elif not isinstance(provider_name, list): - provider_name = [] - - prov_info_json = one_to_one_results["PROV_INFO_JSON"] # Returns list - prov_info_json_list = ( - json.loads(prov_info_json) - if isinstance(prov_info_json, str) - else prov_info_json + prov_info_json = [] + provider_fields = FieldSet( + file_path=config.FIELD_JSON_PATH, field_type="provider_info" ) - # Track if we found any name matches - found_match = False - for provider_dict in prov_info_json_list: - - # Normalize to lists for downstream processing, but keep original format in provider_dict - tins, npis, names = normalize_to_lists(provider_dict) - - # Convert lists to strings for provider_name_match_check - names_str = ", ".join(names) if names else "" - provider_name_str = ", ".join(provider_name) if provider_name else "" - - name_matches = prompt_calls.provider_name_match_check( - names_str, provider_name_str, filename - ) # Returns Boolean - - if name_matches: - found_match = True - provider_dict["IS_GROUP"] = "Y" - for tin in tins: - group_tins.add(tin) - for npi in npis: - group_npis.add(npi) - for name in names: - group_names.add(name) - else: - provider_dict["IS_GROUP"] = "N" - for tin in tins: - other_tins.append(tin) - for npi in npis: - other_npis.append(npi) - for name in names: - other_names.append(name) - - # If provider_name is not null and no match was found, add a new record - if provider_name and not found_match: - new_provider = { - "NAME": provider_name, - "TIN": [], - "NPI": [], - "IS_GROUP": "Y", - } - prov_info_json_list.append(new_provider) - for name in provider_name: - group_names.add(name) - - group_tins, group_npis, group_names, other_tins, other_npis, other_names = ( - deunknown_and_deduplicate( - group_tins, group_npis, group_names, other_tins, other_npis, other_names + for page_num in relevant_pages: + page_provider_info = prompt_calls.prompt_provider_info( + text_dict, page_num, filename, payer_name, provider_fields ) - ) + prov_info_json.extend(page_provider_info) - # Filter out records with NO_IDENTIFIERS_FOUND - prov_info_json_list = [ - p - for p in prov_info_json_list - if "NO_IDENTIFIERS_FOUND" not in (p.get("NAME", [])) - ] - - # Deduplicate provider_list by NAME, keeping records with actual TIN/NPI values - prov_info_json_list = deduplicate_providers_by_name(prov_info_json_list) - - # Reconstruct PROV_INFO_JSON and PROV_INFO_JSON_FORMATTED with IS_GROUP flags - one_to_one_results["PROV_INFO_JSON"] = prov_info_json_list - - # Merge group provider information into one_to_one_results - one_to_one_results["PROV_GROUP_TIN"] = group_tins if group_tins else [] - one_to_one_results["PROV_GROUP_NPI"] = group_npis if group_npis else [] - one_to_one_results["PROV_GROUP_NAME_FULL"] = group_names if group_names else [] - # Merge other provider information into one_to_one_results - one_to_one_results["PROV_OTHER_TIN"] = other_tins if other_tins else [] - one_to_one_results["PROV_OTHER_NPI"] = other_npis if other_npis else [] - one_to_one_results["PROV_OTHER_NAME_FULL"] = other_names if other_names else [] - - return one_to_one_results + return prov_info_json def deduplicate_providers( @@ -405,162 +274,98 @@ def deduplicate_providers( return valid_providers -def flag_missed_prov_info(page_text, providers, page_num, filename): - # Get all TINs and NPIs found via regex on this page - regex_tins = [ - tin - for tin, _ in get_all_matches_with_ocr( - page_text, - regex_patterns.TIN_PATTERN, - regex_patterns.TIN_OCR_PATTERN, - identifier_type="TIN", - ) - ] - regex_npis = [ - npi - for npi, _ in get_all_matches_with_ocr( - page_text, - regex_patterns.NPI_PATTERN, - regex_patterns.NPI_OCR_PATTERN, - identifier_type="NPI", - ) - ] - - # Collect all TINs/NPIs that LLM found - llm_found_tins = [] - llm_found_npis = [] - for provider in providers: - tin = provider["TIN"] - if tin and tin != "UNKNOWN": - llm_found_tins.append(tin.replace("-", "").replace(".", "")) - - npi = provider["NPI"] - if npi and npi != "UNKNOWN": - llm_found_npis.append(npi.replace("-", "").replace(".", "")) - - # Note: regex_tins and regex_npis are already normalized in get_all_matches_with_ocr - missed_tins = [tin for tin in regex_tins if tin not in llm_found_tins] - missed_npis = [npi for npi in regex_npis if npi not in llm_found_npis] - if missed_tins or missed_npis: - # WARN ONLY if there are missed identifiers - logging.debug( - f"LLM missed identifiers on page {page_num} of {filename}. Missed TINs: {missed_tins}, Missed NPIs: {missed_npis}" - ) - - -def get_relevant_pages(all_tins, all_npis, text_dict, filename): - # If there are no identifiers, return early with default values - if not all_tins and not all_npis: - return [] - else: - # Identify relevant pages containing either NPIs or TINs (with or without hyphen formatting, e.g., 68-0640053) - relevant_pages = [ - page_num - for page_num, page_text in text_dict.items() - if any( - re.sub(r"[\s-]", "", tin) in re.sub(r"[\s-]", "", page_text) - for tin in all_tins - ) - or any( - re.sub(r"[\s-]", "", npi) in re.sub(r"[\s-]", "", page_text) - for npi in all_npis - ) - ] - # Stopgap for when the TIN is written in the header or footer of every page (or nearly every page) - if ( - len(relevant_pages) >= len(text_dict) - 1 - and (len(all_tins) == 1 or len(all_tins) == 0) - and (len(all_npis) == 1 or len(all_npis) == 0) - ): - relevant_pages = [relevant_pages[0]] - return relevant_pages - - -def get_prov_info_json(relevant_pages, text_dict, filename, payer_name): - if not relevant_pages: - prov_info_json = [ - { - "TIN": "N/A", - "NPI": "N/A", - "NAME": "NO_IDENTIFIERS_FOUND", - } - ] - else: - # Process each page separately and combine results - prov_info_json = [] - for page_num in relevant_pages: - page_provider_info = prompt_calls.prompt_provider_info( - text_dict, page_num, filename, payer_name - ) - prov_info_json.extend(page_provider_info) - - # Deduplicate providers based on TIN and NPI - prov_info_json = deduplicate_providers(prov_info_json) - - return prov_info_json - - -def get_tin_extraction_quality(tin_matches, npi_matches, filename): - # Log OCR corrections for monitoring - exact_tins = [tin for tin, quality in tin_matches if quality == "EXACT"] - ocr_corrected_tins = [ - tin for tin, quality in tin_matches if quality == "OCR_CORRECTED" - ] - - exact_npis = [npi for npi, quality in npi_matches if quality == "EXACT"] - ocr_corrected_npis = [ - npi for npi, quality in npi_matches if quality == "OCR_CORRECTED" - ] - logging.debug( - f"TIN extraction summary for {filename}: {len(exact_tins)} exact, {len(ocr_corrected_tins)} OCR-corrected" - ) - logging.debug( - f"NPI extraction summary for {filename}: {len(exact_npis)} exact, {len(ocr_corrected_npis)} OCR-corrected" - ) - return exact_tins, exact_npis, ocr_corrected_tins, ocr_corrected_npis - - -def deduplicate_providers_by_name(provider_list: list[dict]) -> list[dict]: +def add_filename_tin(one_to_one_results: dict, filename: str) -> dict: """ - Deduplicates a list of provider dictionaries by NAME, preferring records with actual TIN/NPI values. - + Extract TIN (Taxpayer Identification Number) from filename and add to results. + Attempts to extract TIN from the filename using a standard regex pattern. + If no matches are found, retries with a relaxed pattern. Extracted TINs are + added to the results dictionary and appended to PROV_INFO_JSON with additional + metadata indicating extraction from filename. Args: - provider_list (list[dict]): List of provider dictionaries with NAME, TIN, NPI keys. - + one_to_one_results (dict): Dictionary containing extraction results and + PROV_INFO_JSON list to be updated with TIN information. + filename (str): The filename string to extract TIN from. Returns: - list[dict]: Deduplicated list with one entry per unique NAME, preferring non-UNKNOWN values. + dict: Updated one_to_one_results dictionary with FILENAME_TIN key and + PROV_INFO_JSON entries populated with extracted TIN data, or unchanged + dictionary if no TINs were found. """ - seen_names = set() - deduplicated_list = [] - # Sort so providers with actual TIN/NPI values come first - sorted_providers = sorted( - provider_list, - key=lambda p: ( - not p.get("TIN") or p.get("TIN") == ["UNKNOWN"], - not p.get("NPI") or p.get("NPI") == ["UNKNOWN"], - ), + + filename_tin_list = get_all_matches(filename, regex_patterns.FILE_NAME_TIN_PATTERN) + + if not filename_tin_list: + filename_tin_list = get_all_matches( + filename, regex_patterns.FILE_NAME_TIN_PATTERN_RELAXED + ) + + if not filename_tin_list: + return one_to_one_results + + one_to_one_results["FILENAME_TIN"] = ( + filename_tin_list if filename_tin_list else "N/A" ) - for provider in sorted_providers: - names = provider.get("NAME", []) - # Normalize to list: if string, convert to list to avoid character iteration - if isinstance(names, str): - names = [names] - elif not isinstance(names, list): - names = [] + # Add to PROV_INFO_JSON + for filename_tin in filename_tin_list: + one_to_one_results["PROV_INFO_JSON"].append( + { + "TIN": "".join(c for c in str(filename_tin) if c.isdigit()), + "NPI": "N/A", + "NAME": "N/A", + "FROM_FILENAME": "Y", + } + ) - # Check if any name was already seen - is_duplicate = False - for name in names: - if name in seen_names: - is_duplicate = True - break + return one_to_one_results - if is_duplicate: - continue - # Mark all names as seen - for name in names: - seen_names.add(name) - deduplicated_list.append(provider) - return deduplicated_list +def add_group_and_other(one_to_one_results: dict[str, Any], filename: str): + + provider_name = one_to_one_results.get("PROVIDER_NAME") # Can be List or string + prov_info_json = one_to_one_results["PROV_INFO_JSON"] + + if string_utils.is_empty(provider_name): + return one_to_one_results + + group_tins, group_npis, group_names = set(), set(), set() + other_tins, other_npis, other_names = set(), set(), set() + + for prov_info_dict in prov_info_json: + tin, npi, name = ( + prov_info_dict.get("TIN"), + prov_info_dict.get("NPI"), + prov_info_dict.get("NAME"), + ) + name_matches = prompt_calls.provider_name_match_check( + name, provider_name, filename + ) + + if name_matches: + prov_info_dict["IS_GROUP"] = "Y" + if not string_utils.is_empty(tin): + group_tins.add(tin) + if not string_utils.is_empty(npi): + group_npis.add(npi) + if isinstance(provider_name, list): + group_names.update(provider_name) + else: + group_names.add(provider_name) + else: + prov_info_dict["IS_GROUP"] = "N" + if not string_utils.is_empty(tin): + other_tins.add(tin) + if not string_utils.is_empty(npi): + other_npis.add(npi) + if not string_utils.is_empty(name): + other_names.add(name) + + # Merge group provider information into one_to_one_results + one_to_one_results["PROV_GROUP_TIN"] = list(group_tins) + one_to_one_results["PROV_GROUP_NPI"] = list(group_npis) + one_to_one_results["PROV_GROUP_NAME_FULL"] = list(group_names) + # Merge other provider information into one_to_one_results + one_to_one_results["PROV_OTHER_TIN"] = list(other_tins) + one_to_one_results["PROV_OTHER_NPI"] = list(other_npis) + one_to_one_results["PROV_OTHER_NAME_FULL"] = list(other_names) + + return one_to_one_results diff --git a/src/pipelines/shared/postprocessing/postprocess.py b/src/pipelines/shared/postprocessing/postprocess.py index 542a106..b2dc19a 100644 --- a/src/pipelines/shared/postprocessing/postprocess.py +++ b/src/pipelines/shared/postprocessing/postprocess.py @@ -7,7 +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 +from src.utils import timing_utils, string_utils def postprocess(df, constants: Constants): @@ -87,7 +87,7 @@ def standard_postprocess(df, constants: Constants): df[col] = df[col].apply(postprocessing_funcs.normalize_indicator_field) # Normalize _IND fields and clean up TIN/NPI fields if "TIN" in col or "NPI" in col: - df[col] = df[col].apply(postprocessing_funcs.remove_hyphens) + df[col] = df[col].apply(string_utils.remove_hyphens) # Apply the flatten_singleton_string_list function to the 'CPT' column if "_CD" in col and "CPT" not in col: df[col] = df[col].apply(postprocessing_funcs.flatten_singleton_string_list) @@ -215,6 +215,10 @@ def contract_config_postprocess(df, constants: Constants): cc_df["BILL_TYPE_CD_DESC"] = cc_df["BILL_TYPE_CD_DESC"].apply( postprocessing_funcs.format_as_json_list ) + if "PROV_INFO_JSON" in cc_df.columns: + cc_df["PROV_INFO_JSON"] = cc_df["PROV_INFO_JSON"].apply( + postprocessing_funcs.format_prov_info_json + ) return cc_df diff --git a/src/pipelines/shared/postprocessing/postprocessing_funcs.py b/src/pipelines/shared/postprocessing/postprocessing_funcs.py index 281310c..db9cb8e 100644 --- a/src/pipelines/shared/postprocessing/postprocessing_funcs.py +++ b/src/pipelines/shared/postprocessing/postprocessing_funcs.py @@ -1,12 +1,15 @@ import ast import hashlib +import json import logging import os import re from datetime import datetime + import pandas as pd + from src.utils import string_utils -import json +from src.utils import json_utils # Determine the base directory for the project BASE_DIR = os.path.dirname(os.path.dirname(os.path.dirname(os.path.abspath(__file__)))) @@ -221,24 +224,6 @@ def normalize_currency(value: str) -> str: return text -def remove_hyphens(value): - """ - Remove hyphens from the input value. - - Args: - value : The input might be list or string - Returns: - str: The cleaned value with hyphens removed. - """ - if value is None: - return "" - if isinstance(value, str): - return value.replace("-", "") - elif isinstance(value, list): - return [v.replace("-", "") if isinstance(v, str) else v for v in value] - return value - - def flatten_singleton_string_list(list_str: str) -> str: """Take in a list within a string, and if it has a single element, return that element. If it has multiple elements, return the list as a string. @@ -270,6 +255,9 @@ def validate_and_reformat_date(date_str: str) -> str: """ Validate if a string is in YYYY/MM/DD format or reformat it to YYYY/MM/DD if possible. + Strips leading/trailing pipes (e.g. |01/01/2022|) before parsing so date-only + content is validated and reformatted. + Args: date_str (str): The input date string. @@ -279,13 +267,25 @@ def validate_and_reformat_date(date_str: str) -> str: if not isinstance(date_str, str): return date_str + # Strip pipes and whitespace so |01/01/2022| becomes 01/01/2022 before parsing. + date_str = date_str.strip().strip("|").strip() + if not date_str: + return "" + try: # Check if the date is already in YYYY/MM/DD format datetime.strptime(date_str, "%Y/%m/%d") return date_str except ValueError: - # Attempt to reformat the date from known formats - known_formats = ["%Y-%m-%d", "%m/%d/%Y", "%d-%b-%Y", "%d/%m/%Y"] + # Attempt to reformat the date from known formats (date-only and datetime). + known_formats = [ + "%Y-%m-%d", + "%m/%d/%Y", + "%d-%b-%Y", + "%d/%m/%Y", + "%Y-%m-%d %H:%M:%S", + "%Y-%m-%d %H:%M:%S.%f", + ] for fmt in known_formats: try: return datetime.strptime(date_str, fmt).strftime("%Y/%m/%d") @@ -1211,6 +1211,14 @@ def _is_empty_list_placeholder(item) -> bool: return s in ("[]", "['[]']", '["[]"]') or s == "" +def _strip_wrapping_single_quotes(s: str) -> str: + """Remove surrounding single quotes so \"'NV'\" becomes \"NV\".""" + s = s.strip() + if len(s) >= 2 and s[0] == "'" and s[-1] == "'": + return s[1:-1] + return s + + def format_as_json_list(val): """ Format a value as a JSON list string. Returns blank for empty lists @@ -1225,9 +1233,15 @@ def format_as_json_list(val): continue # If item is a list itself, extend (flatten) if isinstance(i, list): - flattened.extend([str(x).strip() for x in i if x is not None]) + flattened.extend( + [ + _strip_wrapping_single_quotes(str(x).strip()) + for x in i + if x is not None and not _is_empty_list_placeholder(x) + ] + ) else: - flattened.append(str(i).strip()) + flattened.append(_strip_wrapping_single_quotes(str(i).strip())) return json.dumps(flattened) if flattened else "" if pd.isna(val) or val == "" or val == "[]": @@ -1242,7 +1256,7 @@ def format_as_json_list(val): parsed = json.loads(text) if isinstance(parsed, list): filtered = [ - str(i).strip() + _strip_wrapping_single_quotes(str(i).strip()) for i in parsed if i is not None and not _is_empty_list_placeholder(i) ] @@ -1250,9 +1264,36 @@ def format_as_json_list(val): except (json.JSONDecodeError, ValueError): text = text.strip("[]") # Strip brackets if it's a "fake" list like ['A'] - # 3. Handle comma-separated strings & Clean special characters - # This regex keeps letters, numbers, commas, and spaces - clean_text = re.sub(r"[^a-zA-Z0-9, ]+", "", text) + # 3. Pipe-delimited string (e.g. CHIP|MMC from crosswalk/derived fields) + if "|" in text: + items = [ + _strip_wrapping_single_quotes(i.strip()) + for i in text.split("|") + if i.strip() + ] + return json.dumps(items) if items else "" + + # 4. Handle comma-separated or single string; keep only letters, digits, comma, + # space, apostrophe, hyphen (e.g. "Children's / Medicaid-Medicare (MM)" -> + # "Children's Medicaid-Medicare MM") + clean_text = re.sub(r"[^a-zA-Z0-9, ' -]+", "", text) items = [i.strip() for i in clean_text.split(",") if i.strip()] + items = [ + _strip_wrapping_single_quotes(re.sub(r"\s+", " ", i).strip()) + for i in items + if i + ] return json.dumps(items) if items else "" + + +def format_prov_info_json(val) -> str: + """ + Format PROV_INFO_JSON cell as a valid JSON string (list of objects). + + Delegates to json_utils.format_prov_info_json after normalizing pandas NaN. + Returns "[]" for empty; otherwise valid JSON string. + """ + if isinstance(val, float) and pd.isna(val): + val = None + return json_utils.format_prov_info_json(val) diff --git a/src/prompts/prompt_templates.py b/src/prompts/prompt_templates.py index 05b49a2..6ef8c92 100644 --- a/src/prompts/prompt_templates.py +++ b/src/prompts/prompt_templates.py @@ -1590,20 +1590,12 @@ def TIN_NPI_TEMPLATE_INSTRUCTION() -> str: Analyze the following healthcare payer-provider contract text and extract provider entities - these are healthcare providers who deliver services, NOT payers/insurance companies who reimburse for services. IMPORTANT DISTINCTION: -- Providers: Physicians, physician groups, hospitals, clinics who DELIVER healthcare +- Providers: Physicians, physician groups, hospitals, clinics who DELIVER healthcare, CEOs and other signatories that may sign on behalf of any of the previous entities - Payers: Insurance companies, health plans who PAY for healthcare services -CRITICAL: If a provider has multiple TINs or NPIs, include ALL of them as a JSON list in the same provider record. Do NOT create separate records for the same provider entity. - -Examples: -- If you see "NPI: 1083087266 and 1396117412", return "NPI": ["1083087266", "1396117412"] -- If you see "Tax ID: 12-3456789, also 98-7654321", return "TIN": ["123456789", "987654321"] -- If there is only one TIN or NPI, still return it as a string (not a list): "TIN": "123456789" - -Briefly explain your reasoning. After your explanation, include the heading "FINAL PROVIDER INFO:" followed by a properly formatted JSON array with your final output. - [EXTRACTION INSTRUCTIONS] -- First, analyze the text thoroughly and identify all provider entities +- First, analyze the text thoroughly and identify all Provider Entities +- A Provider Entity is considered valid as long as either a TIN, NPI, or NAME is present. If any one or two of those fields are missing, use "N/A" for those fields. - Return TIN and NPIs that are found on the page even if they aren't explicitly labeled as such. - If the table contains multiple name columns (e.g. 'System Name' and 'Facility Name'), YOU MUST extract the specific 'Facility Name' or 'Hospital Name' column. - IMPORTANT VALIDATION REQUIREMENTS: @@ -1612,13 +1604,13 @@ Briefly explain your reasoning. After your explanation, include the heading "FIN * CAREFULLY double-check the digit count for each TIN and NPI before including them in the response [FORMATTING INSTRUCTIONS] -- In your response, include ONLY ONE SINGLE properly formatted JSON List-of-Dicts-of-Str, after your explanation -- Each provider must be an object (DICT) within this array +- Briefly explain your reasoning. After your explanation, include the heading "FINAL PROVIDER INFO:" followed by a properly formatted JSON array with your final output +- In your response, include ONLY ONE SINGLE properly formatted JSON object. CRITICAL: The object MUST be of the form **dict[list[str, str]]** +- Each associated TIN-NPI-Name combination must be an object (DICT) within this array - All provider objects must have these exact keys: "TIN", "NPI", "NAME" -- ALl dict values must be strings - if there are multiple different TINs or NPIs for an identical NAME, put them in different dictionaries. +- ALl dict values must be strings - if there are multiple different NAMEs or NPIs for an identical TIN, put them in different dictionaries. - Use "N/A" for any missing values - Strip ALL non-digit characters from any found TIN or NPI (e.g., remove hyphens, spaces, parentheses). -- A provider record is considered valid as long as either a TIN, NPI, or NAME is present. If any one or two of those fields are missing, use "N/A" for those fields. - Your final JSON array must begin with an opening bracket '[' and end with a closing bracket ']' - Make sure the JSON array is correctly formatted with commas between objects - Do not include any JSON notation elsewhere in your response diff --git a/src/scripts/postprocess_existing_output.py b/src/scripts/postprocess_existing_output.py index fa7e017..3379d12 100644 --- a/src/scripts/postprocess_existing_output.py +++ b/src/scripts/postprocess_existing_output.py @@ -1,3 +1,7 @@ +import json +import zipfile +from pathlib import Path + import pandas as pd from src.constants.constants import Constants from src.pipelines.shared.postprocessing.postprocess import postprocess @@ -11,7 +15,6 @@ print(f"Reading Excel file: {input_excel_path}") df = pd.read_csv(input_excel_path, dtype=str) print(f"Loaded {len(df)} rows from Excel") -# Initialize Constants constants = Constants() # Process the DataFrame diff --git a/src/testbed/testbed_utils.py b/src/testbed/testbed_utils.py index 77c32f7..76eb7da 100644 --- a/src/testbed/testbed_utils.py +++ b/src/testbed/testbed_utils.py @@ -926,7 +926,7 @@ def testbed_postprocess(df): df[col] = df[col].apply(postprocessing_funcs.normalize_indicator_field) # Normalize _IND fields and clean up TIN/NPI fields if "TIN" in col or "NPI" in col: - df[col] = df[col].apply(postprocessing_funcs.remove_hyphens) + df[col] = df[col].apply(string_utils.remove_hyphens) # Apply the flatten_singleton_string_list function if "_CD" in col and "CPT" not in col: df[col] = df[col].apply(postprocessing_funcs.flatten_singleton_string_list) diff --git a/src/tests/test_json_parsers.py b/src/tests/test_json_parsers.py index c7859d5..e9b183b 100644 --- a/src/tests/test_json_parsers.py +++ b/src/tests/test_json_parsers.py @@ -5,10 +5,16 @@ Tests the new JSON parsing functions that replace pipe-delimited output format. Also tests field-aware normalization functionality. """ +import json + import pytest + from src.utils.json_utils import ( + format_prov_info_json, parse_json_dict, parse_json_list, + serialize_prov_info_json, + format_prov_info_json, ) from src.constants.investment_columns import FIELD_FORMAT_MAPPING @@ -514,3 +520,117 @@ class TestFieldAwareNormalization: result = parse_json_dict(output) keys = list(result.keys()) assert keys == ["A", "B", "C"] + + +class TestSerializeProvInfoJson: + """Tests for serialize_prov_info_json (list of provider dicts to JSON string).""" + + def test_empty_list_returns_empty_array_string(self): + """Empty list returns '[]'.""" + assert serialize_prov_info_json([]) == "[]" + + def test_single_provider_serialized(self): + """Single provider dict is serialized as (str, str) JSON.""" + data = [{"NAME": ["A"], "TIN": ["123"], "NPI": []}] + result = serialize_prov_info_json(data) + parsed = json.loads(result) + assert len(parsed) == 1 + assert parsed[0]["NAME"] == "A" + assert parsed[0]["TIN"] == "123" + assert parsed[0]["NPI"] == "N/A" + + def test_none_values_become_na_string(self): + """None values in dict become "N/A" string in output.""" + data = [{"NAME": ["A"], "TIN": None, "NPI": []}] + result = serialize_prov_info_json(data) + parsed = json.loads(result) + assert parsed[0]["TIN"] == "N/A" + + def test_nan_values_become_na_string(self): + """Float NaN values become "N/A" string in output.""" + data = [{"NAME": ["A"], "TIN": float("nan")}] + result = serialize_prov_info_json(data) + parsed = json.loads(result) + assert parsed[0]["TIN"] == "N/A" + + def test_na_placeholder_filtered(self): + """N/A and '[]' placeholders filtered from lists; output is string.""" + data = [{"NAME": ["A"], "TIN": ["123", "N/A", "[]"]}] + result = serialize_prov_info_json(data) + parsed = json.loads(result) + assert parsed[0]["TIN"] == "123" + + def test_non_dict_items_skipped(self): + """Non-dict items in list are skipped; output (str, str).""" + data = [{"NAME": ["A"]}, "not a dict", {"TIN": ["99"]}] + result = serialize_prov_info_json(data) + parsed = json.loads(result) + assert len(parsed) == 2 + assert parsed[0]["NAME"] == "A" + assert parsed[1]["TIN"] == "99" + + def test_multiple_values_joined_as_string(self): + """List values are joined to a single string.""" + data = [{"NAME": ["First", "Last"], "TIN": ["11", "22"]}] + result = serialize_prov_info_json(data) + parsed = json.loads(result) + assert parsed[0]["NAME"] == "First, Last" + assert parsed[0]["TIN"] == "11, 22" + + +class TestFormatProvInfoJson: + """Tests for format_prov_info_json (cell value to final JSON string).""" + + def test_none_returns_empty_array(self): + """None returns '[]'.""" + assert format_prov_info_json(None) == "[]" + + def test_list_serialized(self): + """List of dicts is serialized as (str, str).""" + data = [{"NAME": ["X"], "TIN": ["1"]}] + result = format_prov_info_json(data) + assert result.startswith("[") and result.endswith("]") + parsed = json.loads(result) + assert len(parsed) == 1 and parsed[0]["TIN"] == "1" + + def test_json_string_parsed_and_serialized(self): + """JSON string is parsed and re-serialized to (str, str).""" + s = '[{"NAME": ["B"], "TIN": null}]' + result = format_prov_info_json(s) + parsed = json.loads(result) + assert len(parsed) == 1 + assert parsed[0]["TIN"] == "N/A" + + def test_empty_string_returns_empty_array(self): + """Empty or placeholder string returns '[]'.""" + assert format_prov_info_json("") == "[]" + assert format_prov_info_json("[]") == "[]" + + def test_invalid_string_returns_empty_array(self): + """Invalid string returns '[]'.""" + assert format_prov_info_json("not json") == "[]" + + +class TestProvInfoJsonUtils: + """Tests for PROV_INFO_JSON parsing, serializing, and formatting utilities.""" + + def test_serialize_prov_info_json(self): + """Test serialization logic handling lists, None, and empty values.""" + + input_list = [{"TIN": ["123", "456"], "NAME": None, "NPI": "N/A", "OTHER": []}] + # Should join arrays, replace None/Empty with "N/A" + expected = '[{"TIN": "123, 456", "NAME": "N/A", "NPI": "N/A", "OTHER": "N/A"}]' + + assert serialize_prov_info_json(input_list) == expected + + def test_format_prov_info_json(self): + """Test formatting wrapper handles floats, strings, and lists correctly.""" + + assert format_prov_info_json(None) == "[]" + assert format_prov_info_json(float("nan")) == "[]" + + input_data = [{"TIN": ["123"], "NAME": None}] + expected = '[{"TIN": "123", "NAME": "N/A"}]' + + assert format_prov_info_json(input_data) == expected + assert format_prov_info_json('[{"TIN": ["123"], "NAME": null}]') == expected diff --git a/src/tests/test_postprocess.py b/src/tests/test_postprocess.py index 97e3a87..195fa7f 100644 --- a/src/tests/test_postprocess.py +++ b/src/tests/test_postprocess.py @@ -19,7 +19,6 @@ from src.pipelines.shared.postprocessing.postprocessing_funcs import ( normalize_cpt_fields, normalize_indicator_field, process_patient_age_range, - remove_hyphens, remove_redundant_reimb_info, rename_columns, standardize_reimb_method_and_fee_schedule, @@ -60,12 +59,6 @@ class TestPostprocessFunctions(unittest.TestCase): self.assertEqual(format_rate_fields_with_commas(1234.567), "1,234.57") self.assertEqual(format_rate_fields_with_commas(1000), "1,000.00") - def test_remove_hyphens(self): - self.assertEqual(remove_hyphens("123-45-6789"), "123456789") - self.assertEqual(remove_hyphens("123-456"), "123456") - self.assertEqual(remove_hyphens(None), "") - self.assertEqual(remove_hyphens(""), "") - def test_flatten_singleton_string_list(self): self.assertEqual(flatten_singleton_string_list("['123']"), "123") self.assertEqual(flatten_singleton_string_list("['123', '456']"), "123, 456") @@ -84,6 +77,37 @@ class TestPostprocessFunctions(unittest.TestCase): self.assertEqual(format_as_json_list(["Valid"]), '["Valid"]') self.assertEqual(format_as_json_list(["A", "B"]), '["A", "B"]') + def test_format_as_json_list_pipe_delimited(self): + """Test format_as_json_list converts pipe-delimited strings to JSON list.""" + self.assertEqual(format_as_json_list("CHIP|MMC"), '["CHIP", "MMC"]') + self.assertEqual(format_as_json_list("A|B|C"), '["A", "B", "C"]') + self.assertEqual(format_as_json_list(" Single "), '["Single"]') + + def test_format_as_json_list_clean_special_keep_apostrophe_hyphen(self): + """Test format_as_json_list keeps letters, digits, comma, space, apostrophe, hyphen only.""" + self.assertEqual( + format_as_json_list("Medicare-Medicaid (MM)"), + '["Medicare-Medicaid MM"]', + ) + self.assertEqual( + format_as_json_list("Medicare-Medicaid Plan (MMP)"), + '["Medicare-Medicaid Plan MMP"]', + ) + self.assertEqual( + format_as_json_list("Children's / Medicaid-Medicare (MM)"), + '["Children\'s Medicaid-Medicare MM"]', + ) + self.assertEqual( + format_as_json_list("Plan (MMP), Other (X)"), + '["Plan MMP", "Other X"]', + ) + + def test_format_as_json_list_strips_wrapping_single_quotes(self): + """Test format_as_json_list turns [\"'NV'\"] into [\"NV\"] for state fields.""" + self.assertEqual(format_as_json_list(["'NV'"]), '["NV"]') + self.assertEqual(format_as_json_list("[\"'NV'\"]"), '["NV"]') + self.assertEqual(format_as_json_list(["'NV'", "'CA'"]), '["NV", "CA"]') + def test_rename_columns(self): """Tests the rename_columns function to ensure it correctly renames specified columns. @@ -231,6 +255,18 @@ class TestPostprocessFunctions(unittest.TestCase): self.assertEqual(validate_and_reformat_date("01/15/2023"), "2023/01/15") self.assertEqual(validate_and_reformat_date("15-Jan-2023"), "2023/01/15") + # Pipe-wrapped date (e.g. from extraction) should be stripped and reformatted + self.assertEqual(validate_and_reformat_date("|01/01/2022|"), "2022/01/01") + self.assertEqual(validate_and_reformat_date("|2023/06/15|"), "2023/06/15") + + # Datetime with time component should output date-only YYYY/MM/DD + self.assertEqual( + validate_and_reformat_date("2022-01-01 00:00:00"), "2022/01/01" + ) + self.assertEqual( + validate_and_reformat_date("2023-06-15 12:30:00.123456"), "2023/06/15" + ) + # Test invalid formats - should return the original string self.assertEqual(validate_and_reformat_date("Invalid date"), "Invalid date") self.assertEqual(validate_and_reformat_date("01-15"), "01-15") diff --git a/src/tests/test_prompt_calls.py b/src/tests/test_prompt_calls.py index 07381ec..798246f 100644 --- a/src/tests/test_prompt_calls.py +++ b/src/tests/test_prompt_calls.py @@ -837,27 +837,6 @@ class TestPromptCalls(unittest.TestCase): # ==================== PROVIDER PROMPTS ==================== - @patch("src.utils.llm_utils.invoke_claude") - def test_prompt_provider_info_returns_list_of_dicts(self, mock_invoke): - """Test that prompt_provider_info returns list of provider dicts.""" - # Simulate LLM returning provider info - mock_invoke.return_value = """[ - {"TIN": "123456789", "NPI": "987654321", "NAME": "Provider Name"} - ]""" - - text_dict = {"1": "Page text with provider info"} - result = prompt_calls.prompt_provider_info( - text_dict, "1", self.filename, "Test Payer" - ) - - # Should return list of dicts - self.assertIsInstance(result, list) - self.assertGreater(len(result), 0) - self.assertIsInstance(result[0], dict) - self.assertIn("TIN", result[0]) - self.assertIn("NPI", result[0]) - self.assertIn("NAME", result[0]) - @patch("src.utils.llm_utils.invoke_claude") def test_provider_name_match_check_returns_boolean(self, mock_invoke): """Test that provider_name_match_check returns a boolean.""" diff --git a/src/tests/test_string_utils.py b/src/tests/test_string_utils.py index 6765d6c..ab3d531 100644 --- a/src/tests/test_string_utils.py +++ b/src/tests/test_string_utils.py @@ -301,6 +301,12 @@ class TestStringUtils(unittest.TestCase): self.assertEqual(result[0]["SERVICE_TERM"], "Inpatient services") self.assertIn("Exhibit [[", result[0]["REIMB_TERM"]) + def test_remove_hyphens(self): + self.assertEqual(string_utils.remove_hyphens("123-45-6789"), "123456789") + self.assertEqual(string_utils.remove_hyphens("123-456"), "123456") + self.assertEqual(string_utils.remove_hyphens(None), "") + self.assertEqual(string_utils.remove_hyphens(""), "") + if __name__ == "__main__": unittest.main() diff --git a/src/tests/test_tin_npi_funcs.py b/src/tests/test_tin_npi_funcs.py index 9cb6405..559bfb4 100644 --- a/src/tests/test_tin_npi_funcs.py +++ b/src/tests/test_tin_npi_funcs.py @@ -6,6 +6,7 @@ import src.constants.regex_patterns as regex_patterns import pytest import src.pipelines.shared.extraction.tin_npi_funcs as tin_npi_funcs from src.prompts.fieldset import FieldSet +from src import config class TestTinNpiFuncs: @@ -30,82 +31,6 @@ class TestTinNpiFuncs: # Test no matches assert tin_npi_funcs.get_all_matches("No TINs here", pattern) == [] - @patch("src.pipelines.saas.prompts.prompt_calls.provider_name_match_check") - def test_merge_provider_info_with_hybrid_smart_chunking( - self, mock_name_match_check - ): - """Test merge_provider_info_with_hybrid_smart_chunking with name matching logic""" - - # Mock the name matching to return True only when names actually match - # The function compares provider_name (from provider dict) with hybrid_smart_chunking_provider_group_name (from PROVIDER_NAME) - def name_match_side_effect( - provider_name, hybrid_smart_chunking_provider_group_name, filename - ): - # Convert lists to strings for comparison (join if list, use as-is if string) - if isinstance(provider_name, list): - provider_name_str = ", ".join(provider_name) if provider_name else "" - else: - provider_name_str = str(provider_name) if provider_name else "" - - if isinstance(hybrid_smart_chunking_provider_group_name, list): - hybrid_name_str = ( - ", ".join(hybrid_smart_chunking_provider_group_name) - if hybrid_smart_chunking_provider_group_name - else "" - ) - else: - hybrid_name_str = ( - str(hybrid_smart_chunking_provider_group_name) - if hybrid_smart_chunking_provider_group_name - else "" - ) - - # Check if names match (either exact match or one contains the other) - # Return True only if "Group Provider" is in provider_name_str (from provider dict) - # and matches hybrid_name_str (from PROVIDER_NAME) - return ( - "Group Provider" in provider_name_str - and "Group Provider" in hybrid_name_str - ) - - mock_name_match_check.side_effect = name_match_side_effect - - # PROV_INFO_JSON is now a list, not a JSON string - one_to_one_results = { - "PROVIDER_NAME": "Group Provider", - "PROV_INFO_JSON": [ - { - "TIN": "123456789", - "NPI": "1234567890", - "NAME": "Group Provider", - }, - { - "TIN": "987654321", - "NPI": "9876543210", - "NAME": "Other Provider", - }, - ], - } - - result = tin_npi_funcs.merge_provider_info_with_one_to_one( - one_to_one_results, "test.pdf" - ) - - # Check that IS_GROUP flags were added correctly - # PROV_INFO_JSON is now a list, not a JSON string - prov_info = result["PROV_INFO_JSON"] - assert ( - len(prov_info) == 2 - ), f"Expected 2 providers, got {len(prov_info)}: {prov_info}" - assert prov_info[0]["IS_GROUP"] == "Y" - assert prov_info[1]["IS_GROUP"] == "N" - - # Check group and other fields - now lists - assert result["PROV_GROUP_TIN"] == ["123456789"] - assert result["PROV_OTHER_TIN"] == ["987654321"] - assert "Group Provider" in result["PROV_GROUP_NAME_FULL"] - assert "Other Provider" in result["PROV_OTHER_NAME_FULL"] - @patch("src.utils.llm_utils.invoke_claude") def test_get_provider_info(self, mock_invoke_claude, sample_text_dict): """Test prompt_provider_info function (renamed from get_provider_info).""" @@ -115,8 +40,15 @@ class TestTinNpiFuncs: [{"TIN": "123456789", "NPI": "1234567890", "NAME": "Test Provider"}] ) + provider_fields = FieldSet( + file_path=config.FIELD_JSON_PATH, field_type="provider_info" + ) result = prompt_calls.prompt_provider_info( - sample_text_dict, "1", "test.pdf", payer_name="Test Payer" + sample_text_dict, + "1", + "test.pdf", + payer_name="Test Payer", + provider_fields=provider_fields, ) assert len(result) == 1 @@ -132,10 +64,8 @@ class TestTinNpiFuncs: ] ) - contract_text = "Contract with TIN 12-3456789" - results = tin_npi_funcs.run_provider_info_fields( - contract_text, sample_text_dict, "test.pdf", "Test Payer" + sample_text_dict, "test.pdf", "Test Payer" ) assert "PROV_INFO_JSON" in results @@ -274,17 +204,9 @@ class TestTinNpiFuncs: contract_text = "Contract with TIN 12-3456789 and corrupted TIN 98765432O" results = tin_npi_funcs.run_provider_info_fields( - contract_text, sample_text_dict, "test.pdf", "Test Pyer" + sample_text_dict, "test.pdf", "Test Pyer" ) - # Check that quality tracking fields are present - assert "TIN_EXTRACTION_QUALITY" in results - assert "NPI_EXTRACTION_QUALITY" in results - - # Check format of quality strings - assert "exact" in results["TIN_EXTRACTION_QUALITY"] - assert "OCR-corrected" in results["TIN_EXTRACTION_QUALITY"] - def test_ocr_regex_patterns(self): """Test that OCR regex patterns match expected cases""" @@ -331,9 +253,5 @@ class TestTinNpiFuncs: contract_text = "Contract with corrupted TIN 12345678O and NPI 123456789O" results = tin_npi_funcs.run_provider_info_fields( - contract_text, sample_text_dict, "test.pdf", "Test Payer" + sample_text_dict, "test.pdf", "Test Payer" ) - - # Should find OCR-corrected values - assert "0 exact, 1 OCR-corrected" in results["TIN_EXTRACTION_QUALITY"] - assert "0 exact, 1 OCR-corrected" in results["NPI_EXTRACTION_QUALITY"] diff --git a/src/utils/json_utils.py b/src/utils/json_utils.py index dba8947..e140e9a 100644 --- a/src/utils/json_utils.py +++ b/src/utils/json_utils.py @@ -6,12 +6,18 @@ from raw LLM output, replacing the deprecated pipe-delimited format. The parsers can optionally normalize field values based on the field format mapping configuration to ensure consistent output formats. + +Also provides PROV_INFO_JSON-specific helpers for parsing/serializing provider +info list-of-dicts from DataFrame cells and for output formatting. """ +import ast import json import logging import re +import ast from typing import Any +import math from src.utils.formatting_utils import normalize_dict_fields @@ -255,3 +261,77 @@ def parse_json_list( f"No valid JSON list found in LLM output. " f"Output preview: {raw_llm_output[:200]}..." ) + + +def _prov_value_to_str(v: Any) -> str: + """Convert a single provider field value to string for (str, str) dict output.""" + if v is None or (isinstance(v, float) and v != v): + return "N/A" + if isinstance(v, list): + parts = [ + str(x).strip() + for x in v + if x is not None and str(x).strip() not in ("N/A", "[]") + ] + return ", ".join(parts) if parts else "N/A" + s = str(v).strip() + return s if s and s not in ("N/A", "[]") else "N/A" + + +def serialize_prov_info_json(prov_list: list) -> str: + """ + Serialize a list of provider dicts to a valid JSON string. + + Output is list of dicts with (str, str) only: TIN, NPI, NAME, IS_GROUP etc. + are strings (e.g. "111562701", "N/A", "Northwell Health"). List values + are joined with ", "; empty/None become "N/A". + + Args: + prov_list: List of provider dicts (e.g. from PROV_INFO_JSON). + + Returns: + str: JSON string of the form '[{"TIN": "...", "NAME": "..."}, ...]'. + """ + out = [] + for p in prov_list: + if not isinstance(p, dict): + continue + clean = {str(k): _prov_value_to_str(v) for k, v in p.items()} + out.append(clean) + return json.dumps(out) + + +def format_prov_info_json(val: Any) -> str: + """ + Format a PROV_INFO_JSON cell value as a valid JSON string. + + Accepts None, str, or list. Returns "[]" for empty/unparseable; otherwise + returns a JSON array string. Handles pandas NaN values. + + Args: + val: Cell value (None, str, list of provider dicts, or float NaN). + + Returns: + str: JSON string "[...]" or "[]". + """ + # Handle pandas NaN + if isinstance(val, float): + if math.isnan(val): + return "[]" + if val is None: + return "[]" + if isinstance(val, str): + text = val.strip() + if not text or text in ("[]", "''", '""'): + return "[]" + if text.startswith("[") and text.endswith("]"): + try: + parsed = json.loads(text) + if isinstance(parsed, list): + return serialize_prov_info_json(parsed) + except json.JSONDecodeError: + pass + return "[]" + if isinstance(val, list): + return serialize_prov_info_json(val) + return "[]" diff --git a/src/utils/string_utils.py b/src/utils/string_utils.py index d11fb10..9099d5f 100644 --- a/src/utils/string_utils.py +++ b/src/utils/string_utils.py @@ -1094,6 +1094,24 @@ def token_signature(text: str) -> tuple: return tuple(sorted(tokens)) +def remove_hyphens(value): + """ + Remove hyphens from the input value. + + Args: + value : The input might be list or string + Returns: + str: The cleaned value with hyphens removed. + """ + if value is None: + return "" + if isinstance(value, str): + return value.replace("-", "") + elif isinstance(value, list): + return [v.replace("-", "") if isinstance(v, str) else v for v in value] + return value + + def keyword_match(text: str, keywords: list) -> bool: """ Check if text contains any of the keywords (case-insensitive).