diff --git a/src/codes/code_funcs.py b/src/codes/code_funcs.py index a64b34d..10aee0e 100644 --- a/src/codes/code_funcs.py +++ b/src/codes/code_funcs.py @@ -510,7 +510,7 @@ def fill_bill_type( llm_answer_final = _parser(llm_answer_raw) # Second attempt: Service term + reimbursement term + exhibit context (if first attempt failed and context available) - if not llm_answer_final and exhibit_text: + if string_utils.is_empty(llm_answer_final) and exhibit_text: prompt, _parser = prompt_templates.FILL_BILL_TYPE( service, valid_bill_type, @@ -534,7 +534,7 @@ def fill_bill_type( bill_codes, bill_descs = [], [] for description in llm_answer_final: if description in BILL_TYPE_REVERSE_MAPPING: - bill_codes.append(BILL_TYPE_REVERSE_MAPPING[description]) + bill_codes += BILL_TYPE_REVERSE_MAPPING[description] bill_descs.append(description) if bill_codes: diff --git a/src/pipelines/saas/prompts/prompt_calls.py b/src/pipelines/saas/prompts/prompt_calls.py index 51ae68b..3cac1f9 100644 --- a/src/pipelines/saas/prompts/prompt_calls.py +++ b/src/pipelines/saas/prompts/prompt_calls.py @@ -481,6 +481,7 @@ def prompt_dynamic(text: str, field_prompts, filename): cache=True, instruction=prompt_templates.EXHIBIT_LEVEL_INSTRUCTION(), ) # Returns dictionary of lists + logging.debug(f"Dynamic answer for {filename}: {llm_answer_raw}") llm_answer_final = _parser(llm_answer_raw) return llm_answer_final diff --git a/src/pipelines/saas/test.py b/src/pipelines/saas/test.py deleted file mode 100644 index cd13581..0000000 --- a/src/pipelines/saas/test.py +++ /dev/null @@ -1,19 +0,0 @@ -from src.pipelines.shared.postprocessing import aarete_derived -from src.constants.constants import Constants -from src.utils import io_utils -from src.pipelines.saas.prompts import prompt_calls -from src.pipelines.shared.extraction.one_to_n_funcs import reimbursement_level - -# constants = Constants() - -service_term = "Covered Services provided to KIDSfirst Members" -reimb_term = "One hundred percent (100%) of the prevailing yearly and current Medicaid fee schedule for the State of Texas or the Participating Ancillary Services Provider's usual and customary charge, whichever is less." -exhibit_title = "Attachment B SECTION 1 - PCHP MEDICAID STAR HMO SERVICES" -filename = "" - - -lesser_of_answer_dict = prompt_calls.prompt_lesser_of_check( - service_term, reimb_term, exhibit_title, filename -) - -print(lesser_of_answer_dict) diff --git a/src/pipelines/shared/extraction/tin_npi_funcs.py b/src/pipelines/shared/extraction/tin_npi_funcs.py index 8d63b47..6c98c62 100644 --- a/src/pipelines/shared/extraction/tin_npi_funcs.py +++ b/src/pipelines/shared/extraction/tin_npi_funcs.py @@ -2,12 +2,16 @@ import json import logging import re +from typing import Optional + 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.pipelines.saas.prompts import prompt_calls +from src.constants.delimiters import Delimiter from src.prompts.fieldset import Field, FieldSet +from difflib import SequenceMatcher def get_all_matches(text: str, pattern: str) -> list[str]: @@ -171,38 +175,34 @@ def clean_provider_info(provider_info: list[dict]) -> list[dict]: for provider in provider_info: cleaned_provider = {} - # ---- Clean TIN (list, digits only, 9 chars) ---- - cleaned_tins = [] - if isinstance(provider.get("TIN"), list): - for tin in provider["TIN"]: - if not tin: - continue - tin_str = str(tin).replace("-", "").replace(".", "") - if tin_str.isdigit() and len(tin_str) == 9: - cleaned_tins.append(tin_str) - cleaned_provider["TIN"] = cleaned_tins if cleaned_tins else ["UNKNOWN"] + # Clean TIN - remove hyphens and dots, ensure it's 9 digits + if "TIN" in provider and provider["TIN"]: + cleaned_provider["TIN"] = provider["TIN"].replace("-", "").replace(".", "") + # Validate that TIN only contains digits or "|" + if not all(c.isdigit() or c == "|" for c in cleaned_provider["TIN"]): + cleaned_provider["TIN"] = "UNKNOWN" + else: + cleaned_provider["TIN"] = "UNKNOWN" - # ---- Clean NPI (list, digits only, 10 chars) ---- - cleaned_npis = [] - if isinstance(provider.get("NPI"), list): - for npi in provider["NPI"]: - if not npi: - continue - npi_str = str(npi).replace("-", "").replace(".", "") - if npi_str.isdigit() and len(npi_str) == 10: - cleaned_npis.append(npi_str) - cleaned_provider["NPI"] = cleaned_npis if cleaned_npis else ["UNKNOWN"] + # Clean NPI - remove hyphens and dots, ensure it's 10 digits + if "NPI" in provider and provider["NPI"]: + cleaned_provider["NPI"] = provider["NPI"].replace("-", "").replace(".", "") + # Validate that NPI only contains digits or "|" + if not all(c.isdigit() or c == "|" for c in cleaned_provider["NPI"]): + cleaned_provider["NPI"] = "UNKNOWN" + else: + cleaned_provider["NPI"] = "UNKNOWN" - # ---- Clean NAME (list) ---- - cleaned_names = [] - if isinstance(provider.get("NAME"), list): - for name in provider["NAME"]: - if not name: - continue - name_str = " ".join(str(name).split()).strip().rstrip(".") - if name_str: - cleaned_names.append(name_str) - cleaned_provider["NAME"] = cleaned_names if cleaned_names else ["UNKNOWN"] + # Clean Name - remove extra spaces and ensure it's not empty + if "NAME" in provider and provider["NAME"]: + cleaned_provider["NAME"] = ( + " ".join(provider["NAME"].split()).strip().rstrip(".") + ) + # If name is empty after cleaning, set to "UNKNOWN" + if not cleaned_provider["NAME"]: + cleaned_provider["NAME"] = "UNKNOWN" + else: + cleaned_provider["NAME"] = "UNKNOWN" cleaned_providers.append(cleaned_provider) return cleaned_providers @@ -220,6 +220,7 @@ def merge_provider_info_with_hybrid_smart_chunking(one_to_one_results, filename) Returns: dict: Updated one_to_one_results with merged values from provider_info. """ + # Extract group and non-group providers group_tins = set() group_npis = set() @@ -228,72 +229,48 @@ def merge_provider_info_with_hybrid_smart_chunking(one_to_one_results, filename) 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 = [] - - provider_info = one_to_one_results[ - "PROV_INFO_JSON" - ] # Returns string representation of list - prov_info_json_list = ( + provider_name = one_to_one_results.get("PROVIDER_NAME") + provider_info = one_to_one_results["PROV_INFO_JSON"] + provider_list = ( json.loads(provider_info) if isinstance(provider_info, str) else provider_info ) + # Track if we found any name matches found_match = False - for provider in prov_info_json_list: - # Ensure list format (FORMAT FIX ONLY) - tins = provider.get("TIN") or [] - npis = provider.get("NPI") or [] - names = provider.get("NAME") or [] - - # 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 = [] - + for provider in provider_list: name_matches = prompt_calls.provider_name_match_check( - names, provider_name, filename - ) # Returns Boolean + provider_name, provider.get("NAME", "UNKNOWN"), filename + ) if name_matches: found_match = True provider["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) + if provider.get("TIN"): + group_tins.add(provider["TIN"]) + if provider.get("NPI"): + group_npis.add(provider["NPI"]) + if provider.get("NAME"): + group_names.add(provider["NAME"]) else: provider["IS_GROUP"] = "N" - other_tins.extend(tins) - other_npis.extend(npis) - other_names.extend(names) + if provider.get("TIN"): + other_tins.append(provider["TIN"]) + if provider.get("NPI"): + other_npis.append(provider["NPI"]) + if provider.get("NAME"): + other_names.append(provider["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": [], + "TIN": "UNKNOWN", + "NPI": "UNKNOWN", "IS_GROUP": "Y", } - prov_info_json_list.append(new_provider) - for name in provider_name: - group_names.add(name) + provider_list.append(new_provider) + group_names.add(provider_name) # De-Unknown GROUP fields group_tins = [v for v in group_tins if v != "UNKNOWN"] @@ -301,42 +278,50 @@ def merge_provider_info_with_hybrid_smart_chunking(one_to_one_results, filename) group_names = [v for v in group_names if v != "UNKNOWN"] # 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"] + group_tins = list(dict.fromkeys("|".join(group_tins).split("|"))) + other_tins = list(dict.fromkeys("|".join(other_tins).split("|"))) + other_tins = [ + tin for tin in other_tins if tin not in group_tins and tin != "UNKNOWN" + ] # remove any group TINs from other TINs - 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"] + group_npis = list(dict.fromkeys("|".join(group_npis).split("|"))) + other_npis = list(dict.fromkeys("|".join(other_npis).split("|"))) + other_npis = [ + npi for npi in other_npis if npi not in group_npis and npi != "UNKNOWN" + ] # remove any group NPIs from other NPIs - 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"] + group_names = list(dict.fromkeys("|".join(group_names).split("|"))) + other_names = list(dict.fromkeys("|".join(other_names).split("|"))) + other_names = [ + name for name in other_names if name not in group_names and name != "UNKNOWN" + ] # remove any group Names from other Names # 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") or []) + provider_list = [ + p for p in provider_list if p.get("NAME") != "NO_IDENTIFIERS_FOUND" ] # Deduplicate provider_list by NAME, keeping records with actual TIN/NPI values - prov_info_json_list = deduplicate_providers_by_name(prov_info_json_list) + provider_list = deduplicate_providers_by_name(provider_list) # Reconstruct PROV_INFO_JSON and PROV_INFO_JSON_FORMATTED with IS_GROUP flags - one_to_one_results["PROV_INFO_JSON"] = json.dumps(prov_info_json_list) + one_to_one_results["PROV_INFO_JSON"] = json.dumps(provider_list) one_to_one_results["PROV_INFO_JSON_FORMATTED"] = "\n".join( - [json.dumps(provider) for provider in prov_info_json_list] + [json.dumps(provider) for provider in provider_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 [] + one_to_one_results["PROV_GROUP_TIN"] = "|".join(group_tins) if group_tins else "" + one_to_one_results["PROV_GROUP_NPI"] = "|".join(group_npis) if group_npis else "" + one_to_one_results["PROV_GROUP_NAME_FULL"] = ( + "|".join(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 [] + one_to_one_results["PROV_OTHER_TIN"] = "|".join(other_tins) if other_tins else "" + one_to_one_results["PROV_OTHER_NPI"] = "|".join(other_npis) if other_npis else "" + one_to_one_results["PROV_OTHER_NAME_FULL"] = ( + "|".join(other_names) if other_names else "" + ) return one_to_one_results @@ -429,31 +414,30 @@ def deduplicate_providers( # Deduplicate if all three fields match for provider in provider_info: - tins = provider.get("TIN", []) - npis = provider.get("NPI", []) - names = provider.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"]) + ( + string_utils.is_empty(provider.get("TIN", "")) + or provider.get("TIN", "") == "UNKNOWN" + ) + and ( + string_utils.is_empty(provider.get("NPI", "")) + or provider.get("NPI", "") == "UNKNOWN" + ) + and ( + string_utils.is_empty(provider.get("NAME", "")) + or provider.get("NAME", "") == "UNKNOWN" + ) ): continue key = ( - tuple(tins), - tuple(npis), - tuple(names), + provider.get("TIN", ""), + provider.get("NPI", ""), + provider.get("NAME", ""), ) - if key not in seen: seen.add(key) valid_providers.append(provider) - return valid_providers @@ -510,7 +494,7 @@ def get_provider_info( logging.warning( f"Warning: Unexpected format in provider info response: {type(providers)}" ) - providers = [{"TIN": [], "NPI": [], "NAME": ["PARSING_ERROR"]}] + providers = [{"TIN": "UNKNOWN", "NPI": "UNKNOWN", "NAME": "PARSING_ERROR"}] # VALIDATION LAYER - Cross-check with regex findings page_text = text_dict[page_num] @@ -539,13 +523,19 @@ def get_provider_info( llm_found_tins = [] llm_found_npis = [] for provider in providers: - for tin in provider["TIN"] or []: - if tin and tin != "UNKNOWN": - llm_found_tins.append(tin.replace("-", "").replace(".", "")) - - for npi in provider["NPI"] or []: - if npi and npi != "UNKNOWN": - llm_found_npis.append(npi.replace("-", "").replace(".", "")) + if provider.get("TIN") and provider["TIN"] != "UNKNOWN": + # Normalize for comparison (remove hyphens/dots) + raw_tins = [tin.strip() for tin in provider["TIN"].split("|")] + normalized_tins = [ + tin.replace("-", "").replace(".", "") for tin in raw_tins + ] + llm_found_tins.extend(normalized_tins) + if provider.get("NPI") and provider["NPI"] != "UNKNOWN": + raw_npis = [npi.strip() for npi in provider["NPI"].split("|")] + normalized_npis = [ + npi.replace("-", "").replace(".", "") for npi in raw_npis + ] + llm_found_npis.extend(normalized_npis) # 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] @@ -562,7 +552,7 @@ def get_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"]}] + return [{"TIN": "UNKNOWN", "NPI": "UNKNOWN", "NAME": "PARSING_ERROR"}] def run_provider_info_fields( @@ -652,9 +642,9 @@ def run_provider_info_fields( if not relevant_pages: deduplicated_provider_info = [ { - "TIN": ["UNKNOWN"], - "NPI": ["UNKNOWN"], - "NAME": ["NO_IDENTIFIERS_FOUND"], + "TIN": "UNKNOWN", + "NPI": "UNKNOWN", + "NAME": "NO_IDENTIFIERS_FOUND", } ] else: @@ -704,32 +694,12 @@ def deduplicate_providers_by_name(provider_list: list[dict]) -> list[dict]: seen_names = set() deduplicated_list = [] # Sort so providers with actual TIN/NPI values come first - sorted_providers = sorted( + for provider in sorted( provider_list, - key=lambda p: ( - not p.get("TIN") or p.get("TIN") == ["UNKNOWN"], - not p.get("NPI") or p.get("NPI") == ["UNKNOWN"], - ), - ) - for provider in sorted_providers: - names = provider.get("NAME", []) - - # Skip invalid NAME values - if not isinstance(names, list) or not names or names == ["UNKNOWN"]: - continue - - # Check if any name was already seen - is_duplicate = False - for name in names: - if name in seen_names: - is_duplicate = True - break - - if is_duplicate: - continue - - # Mark all names as seen - for name in names: + key=lambda p: (p.get("TIN") == "UNKNOWN", p.get("NPI") == "UNKNOWN"), + ): + name = provider.get("NAME") + if name not in seen_names: seen_names.add(name) - deduplicated_list.append(provider) + deduplicated_list.append(provider) return deduplicated_list