diff --git a/src/constants/investment_columns.py b/src/constants/investment_columns.py index f56bf54..a919a8f 100644 --- a/src/constants/investment_columns.py +++ b/src/constants/investment_columns.py @@ -20,7 +20,6 @@ COLUMN_ORDER = [ "PROV_OTHER_NPI", "PROV_OTHER_NAME_FULL", "PROV_INFO_JSON", - "PROV_INFO_JSON_FORMATTED", "EFFECTIVE_DT", "AARETE_DERIVED_EFFECTIVE_DT", "TERMINATION_DT", diff --git a/src/pipelines/clients/bcbs_promise/file_processing.py b/src/pipelines/clients/bcbs_promise/file_processing.py index d3c3c8c..5ae30a3 100644 --- a/src/pipelines/clients/bcbs_promise/file_processing.py +++ b/src/pipelines/clients/bcbs_promise/file_processing.py @@ -238,7 +238,7 @@ def run_one_to_one_prompts( # Add HSC's N/A if field doesn't exist yet one_to_one_results[key] = value - one_to_one_results = tin_npi_funcs.merge_provider_info_with_hybrid_smart_chunking( + one_to_one_results = tin_npi_funcs.merge_provider_info_with_one_to_one( one_to_one_results, filename ) diff --git a/src/pipelines/clients/clover/file_processing.py b/src/pipelines/clients/clover/file_processing.py index b9e3886..a72fc01 100644 --- a/src/pipelines/clients/clover/file_processing.py +++ b/src/pipelines/clients/clover/file_processing.py @@ -244,7 +244,7 @@ def run_one_to_one_prompts( # Add HSC's N/A if field doesn't exist yet one_to_one_results[key] = value - one_to_one_results = tin_npi_funcs.merge_provider_info_with_hybrid_smart_chunking( + one_to_one_results = tin_npi_funcs.merge_provider_info_with_one_to_one( one_to_one_results, filename ) diff --git a/src/pipelines/saas/file_processing.py b/src/pipelines/saas/file_processing.py index 03c8cba..550f416 100644 --- a/src/pipelines/saas/file_processing.py +++ b/src/pipelines/saas/file_processing.py @@ -228,7 +228,8 @@ def run_one_to_one_prompts( # Add HSC's N/A if field doesn't exist yet one_to_one_results[key] = value - one_to_one_results = tin_npi_funcs.merge_provider_info_with_hybrid_smart_chunking( + + one_to_one_results = tin_npi_funcs.merge_provider_info_with_one_to_one( one_to_one_results, filename ) diff --git a/src/pipelines/saas/prompts/prompt_calls.py b/src/pipelines/saas/prompts/prompt_calls.py index c8dcb99..d0c213d 100644 --- a/src/pipelines/saas/prompts/prompt_calls.py +++ b/src/pipelines/saas/prompts/prompt_calls.py @@ -929,13 +929,12 @@ def prompt_provider_info( Raises: ValueError: If the response from the language model cannot be parsed as valid JSON. """ - chunk = text_dict[page_num] - + 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( - chunk, provider_fields.print_prompt_dict(), payer_name + page_text, provider_fields.print_prompt_dict(), payer_name ) llm_answer_raw = llm_utils.invoke_claude( prompt, @@ -955,15 +954,12 @@ def prompt_provider_info( providers = _parser(llm_answer_raw) logging.debug(f"Parsed provider info on page {page_num}: {providers}") # Ensure we have a list of providers (not a dict or other type) - if isinstance(providers, dict): - # Single provider returned as dict - wrap in list - providers = [providers] - elif not isinstance(providers, list): + if not isinstance(providers, list): # If not a list or dict, create a default list with error info logging.warning( f"Warning: Unexpected format in provider info response: {type(providers)}" ) - providers = [{"TIN": [], "NPI": [], "NAME": ["PARSING_ERROR"]}] + providers = [{"TIN": "N/A", "NPI": "N/A", "NAME": "PARSING_ERROR"}] # VALIDATION LAYER - Cross-check with regex findings page_text = text_dict[page_num] @@ -975,4 +971,4 @@ def prompt_provider_info( logging.error(f"Error parsing JSON response from LLM: {str(e)}") logging.error(f"Raw response: {llm_answer_raw}") # Return a default value - return [{"TIN": [], "NPI": [], "NAME": ["PARSING_ERROR"]}] \ No newline at end of file + return [{"TIN": "N/A", "NPI": "N/A", "NAME": "N/A"}] \ No newline at end of file diff --git a/src/pipelines/saas/test.py b/src/pipelines/saas/test.py deleted file mode 100644 index 3b7f57e..0000000 --- a/src/pipelines/saas/test.py +++ /dev/null @@ -1,12 +0,0 @@ - -import src.pipelines.shared.extraction.tin_npi_funcs as tin_npi_funcs - - -one_to_one_results = {'PROV_INFO_JSON': [{'TIN': '971098746', 'NPI': 'UNKNOWN', 'NAME': 'Midwest Medical Equipment Solutions, Inc'}], 'PROV_INFO_JSON_FORMATTED': '{"TIN": "971098746", "NPI": "UNKNOWN", "NAME": "Midwest Medical Equipment Solutions, Inc"}', 'TIN_EXTRACTION_QUALITY': '1 exact, 0 OCR-corrected', 'NPI_EXTRACTION_QUALITY': '0 exact, 0 OCR-corrected', 'FILENAME_TIN': 'N/A', 'PAYER_STATE': ['UNKNOWN'], 'TERMINATION_DT': 'N/A', 'AUTO_RENEWAL_TERM': 'N/A', 'CONTRACT_AMENDMENT_NUM': 'UNKNOWN', 'CONTRACT_TITLE': 'MEDICARE PRODUCT ATTACHMENT COMPENSATION SCHEDULE EXHIBIT ANCILLARY SERVICES DURABLE MEDICAL EQUIPMENT', 'PAYER_NAME': 'IlliniCare Health', 'PROVIDER_NAME': 'Midwest Medical Equipment Solutions, Inc.', 'NUM_AVAILABLE_SIGNATORY_LINE_COUNT': '0', 'PROVIDER_STATE': ['IL'], 'AUTO_RENEWAL_IND': 'N', 'EFFECTIVE_DT': 'January 1, 2018', 'AARETE_DERIVED_EFFECTIVE_DT': '2018/01/01'} - -one_to_one_results = tin_npi_funcs.merge_provider_info_with_hybrid_smart_chunking(one_to_one_results, filename="") - -print("\nFinal: ", one_to_one_results) - - - diff --git a/src/pipelines/shared/extraction/one_to_one_funcs.py b/src/pipelines/shared/extraction/one_to_one_funcs.py index 0b0318c..2329157 100644 --- a/src/pipelines/shared/extraction/one_to_one_funcs.py +++ b/src/pipelines/shared/extraction/one_to_one_funcs.py @@ -196,26 +196,6 @@ def run_full_context_fields( contract_text, full_context_fields, constants, filename ) - if "PROV_GROUP_NAME_FULL" in full_context_answers_dict: - full_context_answers_dict["PROV_INFO_JSON"] = json.dumps( - [ - { - "TIN": "UNKNOWN", - "NPI": "UNKNOWN", - "NAME": full_context_answers_dict["PROV_GROUP_NAME_FULL"], - } - ] - ) - full_context_answers_dict["PROV_INFO_JSON_FORMATTED"] = json.dumps( - [ - { - "TIN": "UNKNOWN", - "NPI": "UNKNOWN", - "NAME": full_context_answers_dict["PROV_GROUP_NAME_FULL"], - } - ] - ) - # Normalize PAYER_STATE to two-letter abbreviation if "PAYER_STATE" in full_context_answers_dict: full_context_answers_dict["PAYER_STATE"] = ( diff --git a/src/pipelines/shared/extraction/tin_npi_funcs.py b/src/pipelines/shared/extraction/tin_npi_funcs.py index cce1f2e..9c5d93b 100644 --- a/src/pipelines/shared/extraction/tin_npi_funcs.py +++ b/src/pipelines/shared/extraction/tin_npi_funcs.py @@ -10,6 +10,70 @@ from src.pipelines.saas.prompts import prompt_calls from src.prompts.fieldset import Field, FieldSet +def run_provider_info_fields( + contract_text: str, 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. + Args: + contract_text (str): The full text of the contract to process. + text_dict (dict): A dictionary containing text data, organized by pages + filename (str): The name of the file being processed, used for logging or reference. + payer_name (str): The name of the payer extracted from the full context answers. + This is not being used right now, but it might be useful for filtering providers later. + Returns: + list[dict]: A cleaned and standardized list containing provider information extracted + from the contract text. + """ + one_to_one_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) + + 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" + ) + + # 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" + + return one_to_one_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}") + + return all_tins, all_npis, tin_matches, npi_matches + + def get_all_matches(text: str, pattern: str) -> list[str]: """ Extracts all unique matches of a given pattern from the provided text. @@ -138,77 +202,6 @@ def attempt_ocr_correction(candidate: str, identifier_type: str) -> tuple[str, b return candidate, False -def chunk_on_matches(matches: list[str], text_dict: dict) -> str: - """ - Extracts and concatenates text from a dictionary of pages where any of the specified matches are found. - - Args: - matches (list[str]): A list of strings to search for in the text. - text_dict (dict): A dictionary where keys are identifiers (e.g., page numbers) and values are strings of text. - - Returns: - str: A single string containing the concatenated text from all pages where at least one match is found, - separated by newline characters. - """ - valid_pages = [] - for page_text in text_dict.values(): - if any(match in page_text for match in matches): - valid_pages.append(page_text) - return "\n".join(valid_pages) - - -def clean_provider_info(prov_info_json: list[dict]) -> list[dict]: - """Cleans provider information by standardizing Name/TIN/NPI formats. - - Args: - provider_info (list[dict]): A list of provider information dictionaries. Each - dictionary should be keyed by TIN, NPI, NAME - - Returns: - list: Cleaned provider information - """ - prov_info_json_clean = [] - for provider in prov_info_json: - cleaned_provider = {} - - # ---- Clean TIN (digits only, 9 chars) ---- - tin = provider.get("TIN") - if tin: - tin_str = str(tin).replace("-", "").replace(".", "") - if tin_str.isdigit() and len(tin_str) == 9: - cleaned_provider["TIN"] = tin_str - else: - cleaned_provider["TIN"] = "UNKNOWN" - else: - cleaned_provider["TIN"] = "UNKNOWN" - - # ---- Clean NPI (digits only, 10 chars) ---- - npi = provider.get("NPI") - if npi: - npi_str = str(npi).replace("-", "").replace(".", "") - if npi_str.isdigit() and len(npi_str) == 10: - cleaned_provider["NPI"] = npi_str - else: - cleaned_provider["NPI"] = "UNKNOWN" - else: - cleaned_provider["NPI"] = "UNKNOWN" - - # ---- Clean NAME (list) ---- - name = provider.get("NAME") - if name: - name_str = " ".join(str(name).split()).strip().rstrip(".") - if name_str: - cleaned_provider["NAME"] = name_str - else: - cleaned_provider["NAME"] = "UNKNOWN" - else: - cleaned_provider["NAME"] = "UNKNOWN" - - prov_info_json_clean.append(cleaned_provider) - - return prov_info_json_clean - - def normalize_to_lists(provider: dict): tins = provider.get("TIN") @@ -234,28 +227,28 @@ def normalize_to_lists(provider: dict): def deunknown_and_deduplicate(group_tins, group_npis, group_names, other_tins, other_npis, other_names): # De-Unknown GROUP fields - group_tins = [v for v in group_tins if v != "UNKNOWN"] - group_npis = [v for v in group_npis if v != "UNKNOWN"] - group_names = [v for v in group_names if v != "UNKNOWN"] + 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)] # 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 v != "UNKNOWN"] + 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 v != "UNKNOWN"] + 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 v != "UNKNOWN"] + 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_hybrid_smart_chunking(one_to_one_results: dict, filename: str): +def merge_provider_info_with_one_to_one(one_to_one_results: dict, filename: str): """ Merges provider_info into one_to_one_results. @@ -268,14 +261,11 @@ def merge_provider_info_with_hybrid_smart_chunking(one_to_one_results: dict, fil dict: Updated one_to_one_results with merged values from provider_info. """ # Extract group and non-group providers - group_tins = set() - group_npis = set() - group_names = set() - other_tins = [] - other_npis = [] - other_names = [] + 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] @@ -292,14 +282,12 @@ def merge_provider_info_with_hybrid_smart_chunking(one_to_one_results: dict, fil found_match = False for provider_dict in prov_info_json_list: - print("Provider dict: ", provider_dict) tins, npis, names = normalize_to_lists(provider_dict) name_matches = prompt_calls.provider_name_match_check( names, provider_name, filename ) # Returns Boolean - print("Name matches: ", name_matches) if name_matches: found_match = True @@ -338,18 +326,12 @@ def merge_provider_info_with_hybrid_smart_chunking(one_to_one_results: dict, fil for p in prov_info_json_list if "NO_IDENTIFIERS_FOUND" not in (p.get("NAME", [])) ] - print("Before dedup: ", prov_info_json_list) # Deduplicate provider_list by NAME, keeping records with actual TIN/NPI values prov_info_json_list = deduplicate_providers_by_name(prov_info_json_list) - print("After dedup: ", 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 - one_to_one_results["PROV_INFO_JSON_FORMATTED"] = "\n".join( - [json.dumps(provider) for provider in 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 [] @@ -363,77 +345,8 @@ def merge_provider_info_with_hybrid_smart_chunking(one_to_one_results: dict, fil return one_to_one_results -def reimbursement_tin_npi( - exhibit_dict: dict[str, str], - exhibit_page: str, - reimbursement_level_fields: FieldSet, - filename: str, -) -> tuple[dict, FieldSet]: - """ - Extracts Reimbursement-Level TIN and NPI values from the given exhibit chunk using an LLM. - - Args: - exhibit_dict (dict[str, str]): Dictionary containing the exhibit text with page numbers as keys. - exhibit_page (str): The page number containing reimbursement details. - reimbursement_level_fields (FieldSet): The FieldSet object to store extracted field values. - filename (str): The name of the file being processed. - - Returns: - tuple[dict, FieldSet]: A dictionary containing extracted TIN and NPI values, and an - updated FieldSet with additional fields if multiple values are found. - Fields Added: REIMB_PROV_NAME, - REIMB_PROV_TIN, - REIMB_PROV_NPI - """ - - def get_list(field_name, filtered_providers): - # Get all values and filter out invalid ones - all_values = [ - provider.get(field_name, "N/A") for provider in filtered_providers - ] - valid_values = [ - val - for val in all_values - if not string_utils.is_empty(val) and val != "UNKNOWN" - ] - return list(set(valid_values)) # Remove duplicates - - answer_dict = {} - - # Extract all providers from the exhibit chunk - providers = prompt_calls.prompt_provider_info( - text_dict=exhibit_dict, page_num=exhibit_page, filename=filename - ) - - # Clean, standardize, and deduplicate extracted providers - filtered_providers = deduplicate_providers(clean_provider_info(providers)) - - field_configs = [ - ("REIMB_PROV_TIN", "TIN"), - ("REIMB_PROV_NPI", "NPI"), - ("REIMB_PROV_NAME", "NAME"), - ] - - # Always add providers to reimbursement_level fields for proper term attribution - for field_name, provider_key in field_configs: - if len(filtered_providers) > 0: - field = Field.load_from_file(config.FIELD_JSON_PATH, field_name) - value_list = get_list(provider_key, filtered_providers) - - if len(value_list) == 0: - answer_dict[field_name] = "N/A" - else: - field.update_valid_values(value_list) - reimbursement_level_fields.add_field(field) - else: - # If no providers are found, set default values - answer_dict[field_name] = "N/A" - - return answer_dict, reimbursement_level_fields - - def deduplicate_providers( - provider_info: list[dict], + prov_info_json: list[dict], ) -> list[dict]: # TODO: add unit test """Deduplicates provider information based on TIN, NPI, and NAME fields. This function checks for duplicates in the provider information list by comparing @@ -451,31 +364,19 @@ def deduplicate_providers( valid_providers = [] # Deduplicate if all three fields match - for provider in provider_info: - tins = provider.get("TIN") - npis = provider.get("NPI") - names = provider.get("NAME") + for prov_info_dict in prov_info_json: + tin = prov_info_dict.get("TIN") + npi = prov_info_dict.get("NPI") + name = prov_info_dict.get("NAME") - # # Normalize missing values - tins = tins if isinstance(tins, list) else [] - npis = npis if isinstance(npis, list) else [] - names = names if isinstance(names, list) else [] # Skip providers where all critical identification fields are unknown - if ( - (not tins or tins == "UNKNOWN") - and (not npis or npis == "UNKNOWN") - and (not names or names == "UNKNOWN") - ): + if (string_utils.is_empty(tin) and string_utils.is_empty(npi) and string_utils.is_empty(name)): continue - key = ( - tuple(tins), - tuple(npis), - tuple(names), - ) + key = (tin, npi, name) if key not in seen: seen.add(key) - valid_providers.append(provider) + valid_providers.append(prov_info_dict) return valid_providers @@ -526,10 +427,7 @@ def flag_missed_prov_info(page_text, providers, page_num, filename): 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: - # if there are no TINs or NPIs, we need to extract signature pages - relevant_pages = list( - string_utils.extract_signature_page(text_dict, filename).keys() - ) + return [] else: # Identify relevant pages containing either NPIs or TINs (with or without hyphen formatting, e.g., 68-0640053) relevant_pages = [ @@ -555,10 +453,10 @@ def get_relevant_pages(all_tins, all_npis, text_dict, filename): def get_prov_info_json(relevant_pages, text_dict, filename, payer_name): if not relevant_pages: - prov_info_json_clean = [ + prov_info_json = [ { - "TIN": "UNKNOWN", - "NPI": "UNKNOWN", + "TIN": "N/A", + "NPI": "N/A", "NAME": "NO_IDENTIFIERS_FOUND", } ] @@ -571,41 +469,12 @@ def get_prov_info_json(relevant_pages, text_dict, filename, payer_name): ) prov_info_json.extend(page_provider_info) - print("Before Clean: ", prov_info_json) - - # Clean and standardize results - prov_info_json_clean = clean_provider_info(prov_info_json) - print("After Clean: ", prov_info_json_clean) - # Deduplicate providers based on TIN and NPI - prov_info_json_clean = deduplicate_providers(prov_info_json_clean) - print("After dedup: ", prov_info_json_clean) - - return prov_info_json_clean + prov_info_json = deduplicate_providers(prov_info_json) -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", - ) + return prov_info_json - 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}") - - return all_tins, all_npis, tin_matches, npi_matches def get_tin_extraction_quality(tin_matches, npi_matches, filename): # Log OCR corrections for monitoring @@ -627,52 +496,6 @@ def get_tin_extraction_quality(tin_matches, npi_matches, filename): return exact_tins, exact_npis, ocr_corrected_tins, ocr_corrected_npis -def run_provider_info_fields( - contract_text: str, 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. - Args: - contract_text (str): The full text of the contract to process. - text_dict (dict): A dictionary containing text data, organized by pages - filename (str): The name of the file being processed, used for logging or reference. - payer_name (str): The name of the payer extracted from the full context answers. - This is not being used right now, but it might be useful for filtering providers later. - Returns: - list[dict]: A cleaned and standardized list containing provider information extracted - from the contract text. - """ - - 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) - - prov_info_json_clean = get_prov_info_json(relevant_pages, text_dict, filename, payer_name) - - # Create columns - one_to_one_results = {} - one_to_one_results["PROV_INFO_JSON"] = prov_info_json_clean - one_to_one_results["PROV_INFO_JSON_FORMATTED"] = "\n".join( - [json.dumps(provider) for provider in prov_info_json_clean] - ) - 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" - ) - - # 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" - - print("run_provider_info_fields: ", one_to_one_results) - return one_to_one_results - - def deduplicate_providers_by_name(provider_list: list[dict]) -> list[dict]: """ Deduplicates a list of provider dictionaries by NAME, preferring records with actual TIN/NPI values. diff --git a/src/prompts/prompt_templates.py b/src/prompts/prompt_templates.py index 595a2e3..afb0b48 100644 --- a/src/prompts/prompt_templates.py +++ b/src/prompts/prompt_templates.py @@ -1453,8 +1453,8 @@ Briefly explain your reasoning. After your explanation, include the heading "FIN - In your response, include ONLY ONE SINGLE properly formatted JSON array, after your explanation - Each provider must be an object within this array - All provider objects must have these exact keys: "TIN", "NPI", "NAME" -- Use null (not strings like "null" or "N/A") for missing values -- 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 null for those fields. +- Use "N/A" for any missing values +- 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/tests/test_tin_npi_funcs.py b/src/tests/test_tin_npi_funcs.py index b46ff5a..897fb4e 100644 --- a/src/tests/test_tin_npi_funcs.py +++ b/src/tests/test_tin_npi_funcs.py @@ -30,42 +30,6 @@ class TestTinNpiFuncs: # Test no matches assert tin_npi_funcs.get_all_matches("No TINs here", pattern) == [] - def test_chunk_on_matches(self): - text_dict = { - "1": "Page with match1", - "2": "Page without matches", - "3": "Page with match2", - } - matches = ["match1", "match2"] - result = tin_npi_funcs.chunk_on_matches(matches, text_dict) - assert "match1" in result - assert "match2" in result - assert "without matches" not in result - - def test_clean_provider_info(self): - providers = [ - { - "TIN": "12-345.6789", - "NPI": "12.34567890", - "NAME": "Test Provider", - "IS_GROUP": "Y", - }, - { - "TIN": "", - "NPI": "invalid", - "NAME": "", - "IS_GROUP": "N", - }, - ] - result = tin_npi_funcs.clean_provider_info(providers) - - assert result[0]["TIN"] == "123456789" - assert result[0]["NPI"] == "1234567890" - assert result[0]["NAME"] == "Test Provider" - - assert result[1]["TIN"] == "UNKNOWN" - assert result[1]["NPI"] == "UNKNOWN" - assert result[1]["NAME"] == "UNKNOWN" @patch("src.pipelines.saas.prompts.prompt_calls.provider_name_match_check") def test_merge_provider_info_with_hybrid_smart_chunking( @@ -97,7 +61,7 @@ class TestTinNpiFuncs: ), } - result = tin_npi_funcs.merge_provider_info_with_hybrid_smart_chunking( + result = tin_npi_funcs.merge_provider_info_with_one_to_one( one_to_one_results, "test.pdf" ) @@ -142,7 +106,6 @@ class TestTinNpiFuncs: ) assert "PROV_INFO_JSON" in results - assert "PROV_INFO_JSON_FORMATTED" in results assert isinstance(results["PROV_INFO_JSON"], str) def test_get_all_matches_with_ocr_exact_matches(self): diff --git a/src/utils/qa_qc_utils.py b/src/utils/qa_qc_utils.py index a808afc..2b836aa 100644 --- a/src/utils/qa_qc_utils.py +++ b/src/utils/qa_qc_utils.py @@ -678,7 +678,7 @@ QC_PROV_OTHER_TIN = "PROV_OTHER_TIN" QC_REIMB_PROV_TIN = "REIMB_PROV_TIN" # JSON columns that should not be scanned for hallucinations -QC_JSON_OK_COLS = {"PROV_INFO_JSON", "PROV_INFO_JSON_FORMATTED"} +QC_JSON_OK_COLS = {"PROV_INFO_JSON"} # Date pattern strings for extraction QC_DATE_PATTERNS = [ diff --git a/src/utils/string_utils.py b/src/utils/string_utils.py index 6cfca22..0af298b 100644 --- a/src/utils/string_utils.py +++ b/src/utils/string_utils.py @@ -895,7 +895,6 @@ def normalize_one_to_one_answers_dict(answers_dict: dict[str, str | list]) -> di """ DO_NOT_NORMALIZE = [ "PROV_INFO_JSON", - "PROV_INFO_JSON_FORMATTED", "PAYER_STATE", "PROVIDER_STATE" ]