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

Update fill_na_mapping

Browse Source
This commit is contained in:
Katon Minhas
2026-02-02 23:44:27 -05:00
parent 62605981c1
commit cdce35739b
3 changed files with 66 additions and 69 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src/pipelines/shared/extraction/tin_npi_funcs.py
+19 -19
View File
@@ -203,7 +203,6 @@ def clean_provider_info(provider_info: list[dict]) -> list[dict]:
cleaned_providers.append(cleaned_provider)
return cleaned_providers
def merge_provider_info_with_hybrid_smart_chunking(one_to_one_results, filename):
"""
Merges provider_info into one_to_one_results.
@@ -216,7 +215,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()
@@ -225,19 +223,18 @@ 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") # Returns List
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:
name_matches = prompt_calls.provider_name_match_check(
provider_name, provider.get("NAME", "UNKNOWN"), filename
)
) # Returns Boolean
if name_matches:
found_match = True
@@ -265,8 +262,9 @@ def merge_provider_info_with_hybrid_smart_chunking(one_to_one_results, filename)
"NPI": "UNKNOWN",
"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"]
@@ -293,17 +291,17 @@ def merge_provider_info_with_hybrid_smart_chunking(one_to_one_results, filename)
] # remove any group Names from other Names
# 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 p.get("NAME") != "NO_IDENTIFIERS_FOUND"
]
# 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
@@ -694,8 +692,10 @@ def deduplicate_providers_by_name(provider_list: list[dict]) -> list[dict]:
provider_list,
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)
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
return deduplicated_list
src/pipelines/shared/postprocessing/aarete_derived.py
+46 -40
View File
@@ -27,73 +27,79 @@ def fill_na_mapping(answer_dicts):
Returns:
list[dict]: List of answer dictionaries with filled/merged values
"""
print("Answer dicts for fill_na_mapping:", answer_dicts)
for answer_dict in answer_dicts:
# Collect all AARETE_DERIVED_LOB values from different sources
all_lob_values = set()
# Get existing AARETE_DERIVED_LOB values (if any)
existing_lob = answer_dict.get("AARETE_DERIVED_LOB", "")
if not string_utils.is_empty(existing_lob):
if "|" in existing_lob:
all_lob_values.update(
v.strip()
for v in existing_lob.split("|")
if v.strip() and v.strip() != "N/A"
)
elif existing_lob.strip() and existing_lob.strip() != "N/A":
all_lob_values.add(existing_lob.strip())
existing_lob_list = answer_dict.get("AARETE_DERIVED_LOB", "")
if not string_utils.is_empty(existing_lob_list):
for lob_val in existing_lob_list:
if "|" in existing_lob_list:
all_lob_values.update(
v.strip()
for v in lob_val.split("|")
if v.strip() and v.strip() != "N/A"
)
elif lob_val.strip() and lob_val.strip() != "N/A":
all_lob_values.add(lob_val)
# Get AARETE_DERIVED_LOB from AARETE_DERIVED_PROGRAM crosswalk (always check if PROGRAM exists)
program_lob_values = set()
if not string_utils.is_empty(answer_dict.get("AARETE_DERIVED_PROGRAM")):
program_filled_value = fill_na_from_field(
program_filled_value_list = fill_na_from_field(
answer_dict,
"AARETE_DERIVED_LOB",
"AARETE_DERIVED_PROGRAM",
"src/constants/mappings/crosswalk_program_lob.json",
)
print("program_filled_value: ", program_filled_value_list)
if (
not string_utils.is_empty(program_filled_value)
and program_filled_value != "N/A"
not string_utils.is_empty(program_filled_value_list) and "N/A" not in program_filled_value_list
):
if "|" in program_filled_value:
program_lob_values = set(
v.strip()
for v in program_filled_value.split("|")
if v.strip() and v.strip() != "N/A"
)
elif (
program_filled_value.strip()
and program_filled_value.strip() != "N/A"
):
program_lob_values = {program_filled_value.strip()}
all_lob_values.update(program_lob_values)
for program_lob_val in program_filled_value_list:
if "|" in program_lob_val:
program_lob_values = set(
v.strip()
for v in program_lob_val.split("|")
if v.strip() and v.strip() != "N/A"
)
elif (
program_lob_val.strip()
and program_lob_val.strip() != "N/A"
):
program_lob_values = {program_lob_val.strip()}
all_lob_values.update(program_lob_values)
# Get AARETE_DERIVED_LOB from PRODUCT crosswalk (always check if PRODUCT exists)
product_lob_values = set()
if not string_utils.is_empty(answer_dict.get("PRODUCT")):
product_filled_value = fill_na_from_field(
product_filled_value_list = fill_na_from_field(
answer_dict,
"AARETE_DERIVED_LOB",
"PRODUCT",
"src/constants/mappings/crosswalk_product_lob.json",
)
if (
not string_utils.is_empty(product_filled_value)
and product_filled_value != "N/A"
not string_utils.is_empty(product_filled_value_list)
and "N/A" not in product_filled_value_list
):
if "|" in product_filled_value:
product_lob_values = set(
v.strip()
for v in product_filled_value.split("|")
if v.strip() and v.strip() != "N/A"
)
elif (
product_filled_value.strip()
and product_filled_value.strip() != "N/A"
):
product_lob_values = {product_filled_value.strip()}
all_lob_values.update(product_lob_values)
for product_lob_val in product_filled_value_list:
if "|" in product_lob_val:
product_lob_values = set(
v.strip()
for v in product_filled_value_list.split("|")
if v.strip() and v.strip() != "N/A"
)
elif (
product_lob_val.strip()
and product_lob_val.strip() != "N/A"
):
product_lob_values = {product_lob_val.strip()}
all_lob_values.update(product_lob_values)
# Set the merged, deduplicated value
if all_lob_values:
src/utils/crosswalk_utils.py
+1 -10
View File
@@ -36,8 +36,6 @@ def apply_crosswalk(val: str, mapping: dict[str, str], default: str = "N/A") ->
values = string_utils.flatten_to_strings(parsed_val)
elif "," in val:
values = [v.strip() for v in val.split(",")] # Split by comma if present
elif "|" in val:
values = [v.strip() for v in val.split("|")] # Split by pipe if present
else:
values = [val.strip()]
except (ValueError, SyntaxError):
@@ -61,11 +59,4 @@ def apply_crosswalk(val: str, mapping: dict[str, str], default: str = "N/A") ->
# Convert to list and filter out empty values
final_mapped = [v for v in mapped_values if v and not string_utils.is_empty(v)]
if len(final_mapped) == 1:
result = final_mapped[0]
return result
elif len(final_mapped) > 1:
result = "|".join(final_mapped)
return result
else:
return "N/A"
return final_mapped