From 0d9f9314e9fb3267b2a86c2faf0d8a1e5d43615c Mon Sep 17 00:00:00 2001 From: Venkat Date: Tue, 3 Feb 2026 14:55:05 +0000 Subject: [PATCH] updated tin npi funcs and page funcs --- src/pipelines/shared/extraction/page_funcs.py | 10 +- .../shared/extraction/tin_npi_funcs.py | 170 +++++++++--------- 2 files changed, 95 insertions(+), 85 deletions(-) diff --git a/src/pipelines/shared/extraction/page_funcs.py b/src/pipelines/shared/extraction/page_funcs.py index 6ca26fc..d052639 100644 --- a/src/pipelines/shared/extraction/page_funcs.py +++ b/src/pipelines/shared/extraction/page_funcs.py @@ -299,7 +299,7 @@ def parse_table_rows(table_text: str, page_num: str = "") -> tuple[list[str], in if num_columns == 0: num_columns = len(row) # Join list items with a separator - row_str = " | ".join(str(item) for item in row) + row_str = [str(item) for item in row] rows.append(row_str) else: # Non-list row - log warning but include it @@ -360,6 +360,12 @@ def split_table_by_cell_count( chunks = [] for i in range(0, len(table_rows), max_rows): chunk = table_rows[i : i + max_rows] - chunks.append("\n".join(chunk)) + chunk_strs = [] + for row in chunk: + if isinstance(row, list): + chunk_strs.append(" ".join(str(cell) for cell in row)) + else: + chunk_strs.append(str(row)) + chunks.append("\n".join(chunk_strs)) return chunks diff --git a/src/pipelines/shared/extraction/tin_npi_funcs.py b/src/pipelines/shared/extraction/tin_npi_funcs.py index b965070..767e0ce 100644 --- a/src/pipelines/shared/extraction/tin_npi_funcs.py +++ b/src/pipelines/shared/extraction/tin_npi_funcs.py @@ -239,34 +239,35 @@ def merge_provider_info_with_hybrid_smart_chunking(one_to_one_results, filename) 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 [] name_matches = prompt_calls.provider_name_match_check( - provider_name, provider.get("NAME", "UNKNOWN"), filename + names, provider_name, filename ) # Returns Boolean if name_matches: found_match = True provider["IS_GROUP"] = "Y" - 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"]) + 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["IS_GROUP"] = "N" - 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"]) + other_tins.extend(tins) + other_npis.extend(npis) + other_names.extend(names) # 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": "UNKNOWN", - "NPI": "UNKNOWN", + "TIN": [], + "NPI": [], "IS_GROUP": "Y", } prov_info_json_list.append(new_provider) @@ -279,27 +280,21 @@ 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("|".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_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_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_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_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 + 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"] # Filter out records with NO_IDENTIFIERS_FOUND prov_info_json_list = [ - p for p in prov_info_json_list if p.get("NAME") != "NO_IDENTIFIERS_FOUND" + p for p in prov_info_json_list if "NO_IDENTIFIERS_FOUND" not in (p.get("NAME") or []) ] # Deduplicate provider_list by NAME, keeping records with actual TIN/NPI values @@ -312,17 +307,13 @@ def merge_provider_info_with_hybrid_smart_chunking(one_to_one_results, filename) ) # Merge group provider information into one_to_one_results - 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 "" - ) + 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"] = "|".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 "" - ) + 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 @@ -415,30 +406,31 @@ 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 ( - ( - 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" - ) + (not tins or tins == ["UNKNOWN"]) + and (not npis or npis == ["UNKNOWN"]) + and (not names or names == ["UNKNOWN"]) ): continue key = ( - provider.get("TIN", ""), - provider.get("NPI", ""), - provider.get("NAME", ""), + tuple(tins), + tuple(npis), + tuple(names), ) + if key not in seen: seen.add(key) valid_providers.append(provider) + return valid_providers @@ -495,7 +487,7 @@ def get_provider_info( logging.warning( f"Warning: Unexpected format in provider info response: {type(providers)}" ) - providers = [{"TIN": "UNKNOWN", "NPI": "UNKNOWN", "NAME": "PARSING_ERROR"}] + providers = [{"TIN": [], "NPI": [], "NAME": ["PARSING_ERROR"]}] # VALIDATION LAYER - Cross-check with regex findings page_text = text_dict[page_num] @@ -524,19 +516,13 @@ def get_provider_info( llm_found_tins = [] llm_found_npis = [] for provider in providers: - if provider.get("TIN") and provider["TIN"] != "UNKNOWN": - # Normalize for comparison (remove hyphens/dots) - raw_tins = [tin.strip() for tin in provider["TIN"]] - 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"]] - normalized_npis = [ - npi.replace("-", "").replace(".", "") for npi in raw_npis - ] - llm_found_npis.extend(normalized_npis) + 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(".", "")) # 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] @@ -553,7 +539,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": "UNKNOWN", "NPI": "UNKNOWN", "NAME": "PARSING_ERROR"}] + return [{"TIN": [], "NPI": [], "NAME": ["PARSING_ERROR"]}] def run_provider_info_fields( @@ -643,9 +629,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: @@ -695,14 +681,32 @@ 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 - for provider in sorted( + sorted_providers = sorted( provider_list, - key=lambda p: (p.get("TIN") == "UNKNOWN", p.get("NPI") == "UNKNOWN"), - ): - name_list = provider.get("NAME") - for name in name_list: - if name not in seen_names: - seen_names.add(name) - deduplicated_list.append(provider) - continue + 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: + seen_names.add(name) + deduplicated_list.append(provider) return deduplicated_list