jacob-mathison/doczyai-pipelines
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Activity

fixed clean provider info

Browse Source
This commit is contained in:
Venkat
2026-02-03 14:01:17 +00:00
parent 2aa68f75b9
commit 18c094568b
1 changed files with 30 additions and 26 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src/pipelines/shared/extraction/tin_npi_funcs.py
+30 -26
View File
@@ -171,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