def process()

in python/activation/main.py [0:0]

• 17 lines of code
• 2 McCabe index (conditional complexity)

  def process(self, element):
    """
    Transforms the output of the inference pipeline into a format suitable for sending to the Google Analytics 4 Measurement Protocol API.

    Args:
      element: A dictionary containing the output of the inference pipeline.

    Yields:
      A dictionary containing the Measurement Protocol payload.
    """
    # Removing bad shaping strings in client_id
    _client_id = element['client_id'].replace(r'<img onerror="_exploit_dom_xss(20007)', '')
    _client_id = element['client_id'].replace(r'<img onerror="_exploit_dom_xss(20023)', '')
    _client_id = element['client_id'].replace(r'<img onerror="_exploit_dom_xss(20013)', '')
    _client_id = element['client_id'].replace(r'<img onerror="_exploit_dom_xss(20010)', '')
    _client_id = element['client_id'].replace(r'q="><script>_exploit_dom_xss(40007)</script>', '')
    _client_id = element['client_id'].replace(r'q="><script>_exploit_dom_xss(40013)</script>', '')

    result = {}
    result['client_id'] = _client_id
    if element['user_id']:
      result['user_id'] = element['user_id']
    result['timestamp_micros'] = self.date_to_micro(element["inference_date"])
    result['non_personalized_ads'] = False
    result['consent'] = self.consent_obj
    result['user_properties'] = self.extract_user_properties(element)
    result['events'] = [self.extract_event(element)]

    yield result