Files
doczyai-pipelines/src/utils/crosswalk_utils.py
T

67 lines
2.2 KiB
Python

import ast
import logging
import src.utils.string_utils as string_utils
def apply_crosswalk(
val: str | list, mapping: dict[str, 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]): Mapping dictionary
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):
mapped_values.add(mapped_result)
elif v_str in mapping.values():
mapped_values.add(
v_str
) # If the value is already a mapped value, add it directly
# 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