Use list comprehension

Author: eric-forte-elastic

detection_rules/index_mappings.py

@@ -160,25 +160,30 @@ def get_simulated_index_template_mappings(elastic_client: Elasticsearch, name: s
 
 
 def prune_mappings_of_unsupported_types(
-    integration: str, stream: str, stream_mappings: dict[str, Any], log: Callable[[str], None]
+    integration: str | None, index: str | None, stream_mappings: dict[str, Any], log: Callable[[str], None]
 ) -> dict[str, Any]:
     """Prune fields with unsupported types (ES|QL) from the provided mappings."""
+    debug_str = integration if integration is not None else index
     nested_multifields = find_nested_multifields(stream_mappings)
     for field in nested_multifields:
-        field_name = str(field).split(".fields.")[0].replace(".", ".properties.") + ".fields"
+        parts = str(field).split(".fields.")[0].split(".")
+        base_name = ".properties.".join(parts)
+        field_name = f"{base_name}.fields"
         log(
-            f"Warning: Nested multi-field `{field}` found in `{integration}-{stream}`. "
+            f"Warning: Nested multi-field `{field}` found in `{debug_str}`. "
             f"Removing parent field from schema for ES|QL validation."
         )
         delete_nested_key_from_dict(stream_mappings, field_name)
     nested_flattened_fields = find_flattened_fields_with_subfields(stream_mappings)
     for field in nested_flattened_fields:
         # Remove both .fields and .properties entries for flattened fields
         # .properties entries can occur when being merged with non-ecs or custom schemas
-        field_name = str(field).split(".fields.")[0].replace(".", ".properties.") + ".fields"
-        property_name = str(field).split(".fields.")[0].replace(".", ".properties.") + ".properties"
+        parts = str(field).split(".fields.")[0].split(".")
+        base_name = ".properties.".join(parts)
+        field_name = f"{base_name}.fields"
+        property_name = f"{base_name}.properties"
         log(
-            f"Warning: flattened field `{field}` found in `{integration}-{stream}` with sub fields. "
+            f"Warning: flattened field `{field}` found in `{debug_str}` with sub fields. "
             f"Removing parent field from schema for ES|QL validation."
         )
         delete_nested_key_from_dict(stream_mappings, field_name)
@@ -226,7 +231,7 @@ def prepare_integration_mappings(  # noqa: PLR0913
         for stream in package_schema:
             flat_schema = package_schema[stream]
             stream_mappings = flat_schema_to_index_mapping(flat_schema)
-            stream_mappings = prune_mappings_of_unsupported_types(integration, stream, stream_mappings, log)
+            stream_mappings = prune_mappings_of_unsupported_types(f"{integration}-{stream}", None, stream_mappings, log)
             utils.combine_dicts(integration_mappings, deepcopy(stream_mappings))
             index_lookup[f"{integration}-{stream}"] = stream_mappings
 
@@ -309,7 +314,7 @@ def get_filtered_index_schema(  # noqa: PLR0913
         # Need to use a merge here to not overwrite existing fields
         utils.combine_dicts(base, deepcopy(non_ecs_mapping.get(match, {})))
         utils.combine_dicts(base, deepcopy(custom_mapping.get(match, {})))
-        filtered_index_lookup[match] = prune_mappings_of_unsupported_types("index", match, base, log)
+        filtered_index_lookup[match] = prune_mappings_of_unsupported_types(None, match, base, log)
         utils.combine_dicts(combined_mappings, deepcopy(base))
 
     # Reduce the index lookup to only the matched indices (remote/Kibana schema validation source of truth)
@@ -495,8 +500,7 @@ def prepare_mappings(  # noqa: PLR0913
     # and also at a per index level as custom schemas can override non-ecs fields and/or indices
     non_ecs_schema = ecs.flatten(non_ecs_schema)
     non_ecs_schema = utils.convert_to_nested_schema(non_ecs_schema)
-    non_ecs_schema = prune_mappings_of_unsupported_types("non-ecs", "non-ecs", non_ecs_schema, log)
-    non_ecs_mapping = prune_mappings_of_unsupported_types("non-ecs", "non-ecs", non_ecs_mapping, log)
+    non_ecs_schema = prune_mappings_of_unsupported_types(None, "non-ecs", non_ecs_schema, log)
 
     # Load custom schema and convert to index mapping format (nested schema)
     custom_mapping: dict[str, Any] = {}
@@ -506,7 +510,6 @@ def prepare_mappings(  # noqa: PLR0913
         index_mapping = ecs.flatten(index_mapping)
         index_mapping = utils.convert_to_nested_schema(index_mapping)
         custom_mapping.update({index: index_mapping})
-    custom_mapping = prune_mappings_of_unsupported_types("custom", "custom", custom_mapping, log)
 
     # Load ECS in an index mapping format (nested schema)
     current_version = Version.parse(load_current_package_version(), optional_minor_and_patch=True)