Merged in hotfix/defensive-crosswalk (pull request #632)

Hotfix/defensive crosswalk handling

* Enhance crosswalk functionality with flatten_to_strings helper and improve value handling in apply_crosswalk

* Improve apply_crosswalk function to return original value as fallback when no mapping is found

* Enhance apply_crosswalk function to handle empty values and improve defensive programming tests

* move flatten_to_strings helper function to string_utils


Approved-by: Katon Minhas

fieldExtraction/crosswalk/crosswalk_utils.py

@@ -170,7 +170,6 @@ class CrosswalkBuilder:
         # Then join the lists with commas
         return {target: ", ".join(sources) for target, sources in reverse_dict.items()}
 
-
 def apply_crosswalk(
     val: str, mapping: dict[str, str], default: str = ""
 ) -> str:
@@ -179,25 +178,58 @@ def apply_crosswalk(
     Args:
         val (str): Input value
         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 None.
+        default (str | None, optional): Default value; if None and val is not found in the mapping, val is returned. Defaults to "".
 
     Returns:
         str: Crosswalked value
     """
-    if "[" in val and "]" in val:
-        values = ast.literal_eval(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]
+    # Handle None or blank values defensively
+    if val is None:
+        return default
+    
+    # Store original for fallback
+    original_val = str(val).strip()
 
-    mapped_values = {mapping.get(v.strip(), default) for v in values}
-    mapped_values.update({v for v in values if v in mapping.values()})
-    mapped_values = [v for v in mapped_values if not string_utils.is_empty(v)]
+    # 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
 
-    if len(mapped_values) == 1:
-        return list(mapped_values)[0]
+    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)]
+
+    if len(final_mapped) == 1:
+        return final_mapped[0]
+    elif len(final_mapped) > 1:
+        return "|".join(final_mapped)
     else:
-        return "|".join(mapped_values)  # Join back into a single string
+        return default if default else original_val

fieldExtraction/crosswalk/test_crosswalk_utils.py

@@ -141,3 +141,83 @@ def test_create_reverse_mapping():
 
     reversed_mapping = builder.create_reverse_mapping()
     assert reversed_mapping == {"1": "A, B", "2": "C, D", "3": "E"}
+
+
+# Defensive programming tests
+def test_apply_crosswalk_edge_cases():
+    """Test defensive programming features in apply_crosswalk"""
+    mapping = {"A": "1", "B": "2", "C": "3"}
+    
+    # Test None and empty values
+    assert apply_crosswalk(None, mapping, default="default") == "default"
+    assert apply_crosswalk("", mapping, default="default") == "default"
+    assert apply_crosswalk("   ", mapping, default="default") == "default"  # Empty after strip, WITH default
+    assert apply_crosswalk("   ", mapping) == ""  # Empty after strip, no default
+    
+    # Test non-string inputs (should return original as string)
+    assert apply_crosswalk(123, mapping) == "123"  # No default = return original
+    assert apply_crosswalk(123, mapping, default="default") == "default"  # With default
+    
+    # Test list-like string input
+    assert apply_crosswalk("['A']", mapping) == "1"  # String representation of list
+    
+    # Test comma-separated values
+    result = apply_crosswalk("A,B", mapping)
+    assert result in ["1|2", "2|1"]  # Order may vary due to set
+    
+    # Test pipe-separated values
+    result = apply_crosswalk("A|B", mapping)
+    assert result in ["1|2", "2|1"]
+    
+    # Test list string format
+    result = apply_crosswalk("['A', 'B']", mapping)
+    assert result in ["1|2", "2|1"]
+    
+
+def test_apply_crosswalk_nested_structures():
+    """Test handling of nested data structures"""
+    mapping = {"A": "1", "B": "2", "C": "3"}
+    
+    # Test nested lists
+    assert apply_crosswalk("[['A'], ['B']]", mapping) in ["1|2", "2|1"]
+    
+    # Test mixed nested structures
+    assert apply_crosswalk("['A', ['B', 'C']]", mapping) in ["1|2|3", "1|3|2", "2|1|3", "2|3|1", "3|1|2", "3|2|1"]
+
+
+def test_apply_crosswalk_reverse_mapping():
+    """Test when value is already in mapping.values()"""
+    mapping = {"A": "1", "B": "2", "C": "3"}
+    
+    # Value already mapped should be preserved
+    assert apply_crosswalk("1", mapping) == "1"
+    assert apply_crosswalk("2", mapping) == "2"
+
+
+def test_apply_crosswalk_empty_results():
+    """Test when mapping results in empty values"""
+    mapping = {"A": "", "B": "2", "C": ""}  # Some empty mappings
+    
+    # Should return non-empty results only
+    assert apply_crosswalk("A,B", mapping) == "2"
+    assert apply_crosswalk("A,C", mapping, default="default") == "default"
+
+
+def test_flatten_to_strings():
+    """Test the flatten_to_strings helper function"""
+    from crosswalk.crosswalk_utils import flatten_to_strings
+    
+    # Test simple cases
+    assert flatten_to_strings("A") == ["A"]
+    assert flatten_to_strings(["A", "B"]) == ["A", "B"]
+    assert flatten_to_strings(("A", "B")) == ["A", "B"]
+    
+    # Test nested structures
+    assert flatten_to_strings([["A"], ["B"]]) == ["A", "B"]
+    assert flatten_to_strings([["A", "B"], "C"]) == ["A", "B", "C"]
+    
+    # Test mixed types
+    assert flatten_to_strings([1, ["2", 3]]) == ["1", "2", "3"]
+    
+    # Test with whitespace
+    assert flatten_to_strings([" A ", [" B "]]) == ["A", "B"]

fieldExtraction/src/utils/string_utils.py

@@ -587,3 +587,24 @@ def normalize_state_to_abbreviation(state_value: str) -> str:
 
     # If no match found, return original value
     return state_value
+
+def flatten_to_strings(items) -> list[str]:
+    """Helper function to flatten and stringify any nested structure
+
+    Args:
+        items (Any): The input items; can be a list, tuple, or single value
+
+    Returns:
+        list[str]: A flattened list of stringified items
+    """
+    result = []
+    if isinstance(items, (list, tuple)):
+        for item in items:
+            if isinstance(item, (list, tuple)):
+                result.extend(flatten_to_strings(item))
+            else:
+                result.append(str(item).strip())
+    else:
+        result.append(str(items).strip())
+    
+    return result