Files
doczyai-pipelines/src/utils/crosswalk_utils.py
T
Venkatakrishna Avula 6744c57f95 Merged in bugfix/DAIP2-1870-dynamic-issues (pull request #892)
Bugfix/DAIP2-1870 dynamic issues

* updated list format appending

* fill bill type from claim type cd

* Fixed remaining format fixes

* Merged DEV into bugfix/DAIP2-1870-dynamic-issues

* changed crosswalk mapping from PRODUCT to AARETE_DERIVED_PRODUCT

* Merged DEV into bugfix/DAIP2-1870-dynamic-issues

* updating LOB field values as well from PRODUCT AND PROGRAM

* updated apply crosswalk for reverse mapping

* Merged DEV into bugfix/DAIP2-1870-dynamic-issues

* added QHP and BHP

* removed lob reverse mapping

* Merged DEV into bugfix/DAIP2-1870-dynamic-issues

* Remove Bill Type/Claim Type fill

* Merged DEV into bugfix/DAIP2-1870-dynamic-issues

* Black format

* Merged DEV into bugfix/DAIP2-1870-dynamic-issues


Approved-by: Katon Minhas
2026-03-06 16:11:59 +00:00

85 lines
3.1 KiB
Python

import ast
import logging
import src.utils.string_utils as string_utils
def apply_crosswalk(
val: str | list, mapping: dict[str, str | list[str]], default: str = "N/A"
) -> list:
"""Apply a crosswalk to a value
Args:
val (str | list): Input value (can be string, list, or string representation of list)
mapping (dict[str, str | list[str]]): Mapping dictionary (supports both regular and reverse mappings)
default (str | None, optional): Default value; if None and val is not found in the mapping, val is returned. Defaults to "".
Returns:
list: Crosswalked value as a list
"""
# Handle None or blank values defensively
if val is None:
return default
# Store original for fallback
original_val = str(val).strip()
# IF empty after stripping, return default
if string_utils.is_empty(original_val):
return default if default else original_val
# Ensure val is a string
val = original_val
try:
if "[" in val and "]" in val:
# Try to parse as literal, but handle any structure returned
parsed_val = ast.literal_eval(val)
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):
# If parsing fails, treat val as a regular string
values = [val.strip()]
# Apply mapping with defensive string handling
mapped_values = set()
for v in values:
v_str = str(v).strip()
if v_str in mapping:
mapped_result = mapping[v_str]
if mapped_result and not string_utils.is_empty(mapped_result):
# Handle both string and list values in mapping (e.g., reverse mappings)
if isinstance(mapped_result, list):
for item in mapped_result:
if item and not string_utils.is_empty(item):
mapped_values.add(item)
else:
mapped_values.add(mapped_result)
else:
# Check if v_str is already a mapped value (in mapping.values())
# Handle both str and list[str] values
found_in_values = False
for mapped_val in mapping.values():
if isinstance(mapped_val, list):
if v_str in mapped_val:
if not string_utils.is_empty(v_str):
mapped_values.add(v_str)
found_in_values = True
break
elif v_str == mapped_val:
if not string_utils.is_empty(v_str):
mapped_values.add(v_str)
found_in_values = True
break
# 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)]
return final_mapped