diff --git a/fieldExtraction/crosswalk/crosswalk_utils.py b/fieldExtraction/crosswalk/crosswalk_utils.py index 1332178..32e069f 100644 --- a/fieldExtraction/crosswalk/crosswalk_utils.py +++ b/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 diff --git a/fieldExtraction/crosswalk/test_crosswalk_utils.py b/fieldExtraction/crosswalk/test_crosswalk_utils.py index d70b895..3b3fd64 100644 --- a/fieldExtraction/crosswalk/test_crosswalk_utils.py +++ b/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"] \ No newline at end of file diff --git a/fieldExtraction/src/utils/string_utils.py b/fieldExtraction/src/utils/string_utils.py index 0ada330..2e2aac1 100644 --- a/fieldExtraction/src/utils/string_utils.py +++ 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 \ No newline at end of file