code: def parse_aggregations(
      es_aggregations: Dict[str, Any] = None
  ) -> Optional[Dict[str, Any]]:
      """Transforms elasticsearch raw aggregations into a more friendly structure"""
  
      if es_aggregations is None:
          return None
  
      def is_extended_stats_aggregation(aggr_data) -> bool:
          for metric in [
              "count",
              "min",
              "max",
              "avg",
              "sum",
              "variance",
              "std_deviation",
          ]:
              if metric not in aggr_data:
                  return False
          return True
  
      def parse_buckets(
          buckets: List[Dict[str, Any]]
      ) -> Dict[str, Any]:
          parsed = {}
          for bucket in buckets:
              key, key_as_string, doc_count, meta, _from, _to = (
                  bucket.pop("key", None),
                  bucket.pop("key_as_string", None),
                  bucket.pop("doc_count", 0),
                  bucket.pop("meta", None),
                  bucket.pop("from", None),
                  bucket.pop("to", None),
              )
              if len(bucket) == 1 and not (_from or _to):
                  k = [k for k in bucket][0]
                  parsed.update(
                      {
                          key
                          or k: parse_buckets(
                              bucket[k].get("buckets", [])
                          )
                      }
                  )
              elif len(bucket) > 1:
                  key_metrics = {}
                  for metric_key, metric in list(bucket.items()):
                      if "buckets" in metric:
                          key_metrics.update(
                              {
                                  metric_key: parse_buckets(
                                      metric.get("buckets", [])
                                  )
                              }
                          )
                      else:
                          metric_values = list(metric.values())
                          value = (
                              metric_values[0]
                              if len(metric_values) == 1
                              else metric
                          )
                          key_metrics[metric_key] = value
                  parsed.update({key: key_metrics})
              elif key is not None:
                  parsed.update({key_as_string or key: doc_count})
          return parsed
  
      result = {}
      for key, values in es_aggregations.items() or {}:
          if "buckets" in values:
              buckets = values["buckets"]
              if isinstance(buckets, dict):
                  buckets = [
                      {"key": k, **v} for k, v in buckets.items()
                  ]
              result[key] = parse_buckets(buckets)
          elif is_extended_stats_aggregation(values):
              result[key] = {
                  "argilla:stats": values
              }  # statistical aggregations
          else:
              result[key] = list(parse_buckets([values]).values())[0]
      return result
  
imports: from enum import Enum
  from typing import Any, Dict, List, Optional, Type, Union
  from argilla.server.commons.models import TaskStatus
  from argilla.server.daos.backend.mappings.helpers import mappings
  from argilla.server.pydantic_v1 import BaseModel
label_desc: Use image node_id 2 for calling a global function with example usage: parse_aggregations(es_aggregations) and returns: result, None, True, parsed, False