Revert
This commit is contained in:
Katon Minhas
@@ -2,16 +2,12 @@ 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]:
|
||||
@@ -175,34 +171,38 @@ def clean_provider_info(provider_info: list[dict]) -> list[dict]:
|
||||
for provider in provider_info:
|
||||
cleaned_provider = {}
|
||||
|
||||
# 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 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 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 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 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"
|
||||
# ---- 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"]
|
||||
|
||||
cleaned_providers.append(cleaned_provider)
|
||||
return cleaned_providers
|
||||
@@ -220,7 +220,6 @@ 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()
|
||||
@@ -229,48 +228,72 @@ 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")
|
||||
provider_info = one_to_one_results["PROV_INFO_JSON"]
|
||||
provider_list = (
|
||||
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 = (
|
||||
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 provider_list:
|
||||
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 = []
|
||||
|
||||
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",
|
||||
}
|
||||
provider_list.append(new_provider)
|
||||
group_names.add(provider_name)
|
||||
prov_info_json_list.append(new_provider)
|
||||
for name in provider_name:
|
||||
group_names.add(name)
|
||||
|
||||
# De-Unknown GROUP fields
|
||||
group_tins = [v for v in group_tins if v != "UNKNOWN"]
|
||||
@@ -278,50 +301,42 @@ 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
|
||||
provider_list = [
|
||||
p for p in provider_list if p.get("NAME") != "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 [])
|
||||
]
|
||||
|
||||
# Deduplicate provider_list by NAME, keeping records with actual TIN/NPI values
|
||||
provider_list = deduplicate_providers_by_name(provider_list)
|
||||
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"] = json.dumps(provider_list)
|
||||
one_to_one_results["PROV_INFO_JSON"] = json.dumps(prov_info_json_list)
|
||||
one_to_one_results["PROV_INFO_JSON_FORMATTED"] = "\n".join(
|
||||
[json.dumps(provider) for provider in provider_list]
|
||||
[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"] = "|".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
|
||||
|
||||
|
||||
@@ -414,30 +429,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
|
||||
|
||||
|
||||
@@ -494,7 +510,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]
|
||||
@@ -523,19 +539,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"].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)
|
||||
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]
|
||||
@@ -552,7 +562,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(
|
||||
@@ -642,9 +652,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:
|
||||
@@ -694,12 +704,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 = provider.get("NAME")
|
||||
if name not in seen_names:
|
||||
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)
|
||||
deduplicated_list.append(provider)
|
||||
return deduplicated_list
|
||||
|
||||
Reference in New Issue
Block a user