Adds Enhanced Conversions Uploader

Change-Id: I816f77b42facc58f26fe362352ca4a3b9ad5baab

Author: astivi

megalist_dataflow/main.py

@@ -29,6 +29,7 @@
 from uploaders.google_ads.customer_match.user_id_uploader import GoogleAdsCustomerMatchUserIdUploaderDoFn
 from uploaders.google_ads.conversions.google_ads_offline_conversions_uploader import GoogleAdsOfflineUploaderDoFn
 from uploaders.google_ads.conversions.google_ads_ssd_uploader import GoogleAdsSSDUploaderDoFn
+from uploaders.google_ads.conversions.google_ads_enhanced_conversions_uploader import GoogleAdsEnhancedConversionsUploaderDoFn
 from uploaders.google_analytics.google_analytics_data_import_uploader import GoogleAnalyticsDataImportUploaderDoFn
 from uploaders.google_analytics.google_analytics_measurement_protocol import GoogleAnalyticsMeasurementProtocolUploaderDoFn
 from uploaders.google_analytics.google_analytics_user_list_uploader import GoogleAnalyticsUserListUploaderDoFn
@@ -58,6 +59,14 @@ def __init__(self, oauth_credentials, dataflow_options=None, hasher=None):
     def expand(self, executions):
         pass
 
+class GoogleAdsEnhancedConversionsStep(MegalistaStep):
+    def expand(self, executions):
+        return (
+            executions
+            | 'Load Data - Google Ads Enhanced Conversions' >> BatchesFromExecutions(DestinationType.ADS_ENHANCED_CONVERSION)
+            | 'Hash Users - Google Ads Enhanced Conversions' >> beam.Map(self._hasher.hash_users)
+            | 'Upload - Google Ads Enhanced Conversions' >> beam.ParDo(GoogleAdsEnhancedConversionsUploaderDoFn(self._oauth_credentials))
+        )
 
 class GoogleAdsSSDStep(MegalistaStep):
     def expand(self, executions):
@@ -181,7 +190,7 @@ def run(argv=None):
     oauth_credentials = OAuthCredentials(
         dataflow_options.client_id,
         dataflow_options.client_secret,
-        dataflow_options.developer_token,
+        dataflow_options.access_token,
         dataflow_options.refresh_token)
 
     sheets_config = SheetsConfig(oauth_credentials)
@@ -208,6 +217,7 @@ def run(argv=None):
             oauth_credentials, dataflow_options)
         executions | CampaignManagerConversionStep(oauth_credentials, dataflow_options)
         executions | AppsFlyerEventsStep(oauth_credentials, dataflow_options)
+        executions | GoogleAdsEnhancedConversionsStep(oauth_credentials, dataflow_options, AdsUserListPIIHashingMapper())
 
         # todo: update trix at the end
 

megalist_dataflow/mappers/ads_user_list_pii_hashing_mapper.py

@@ -38,11 +38,12 @@ def __init__(self):
 
     def _hash_user(self, user, hasher):
 
-        hashed = dict()
+        hashed = user.copy()
 
         try:
             if 'email' in user:
                 hashed['hashedEmail'] = hasher.hash_field(user['email'])
+                del hashed['email']
         except:
             self.logger.error("Error hashing email for user: %s" % user)
 
@@ -54,25 +55,32 @@ def _hash_user(self, user, hasher):
                     'countryCode': user['mailing_address_country'],
                     'zipCode': user['mailing_address_zip']
                 }
+                del hashed['mailing_address_first_name']
+                del hashed['mailing_address_last_name']
+                del hashed['mailing_address_country']
+                del hashed['mailing_address_zip']
         except:
             self.logger.error("Error hashing address for user: %s" % user)
 
         try:
             if 'phone' in user:
                 hashed['hashedPhoneNumber'] = hasher.hash_field(user['phone'])
+                del hashed['phone']
         except:
             self.logger.error("Error hashing phone for user: %s" % user)
 
         try:
             if 'mobile_device_id' in user:
                 hashed['mobileId'] = user['mobile_device_id']
+                del hashed['mobile_device_id']
         except:
             self.logger.error(
                 "Error hashing mobile_device_id for user: %s" % user)
 
         try:
             if 'user_id' in user:
                 hashed['userId'] = user['user_id']
+                del hashed['user_id']
         except:
             self.logger.error("Error hashing user_id for user: %s" % user)
 

megalist_dataflow/models/execution.py

@@ -13,7 +13,7 @@
 # limitations under the License.
 
 from enum import Enum
-from typing import Dict, List
+from typing import Dict, List, Union
 
 OK_STATUS = 'OK'
 
@@ -22,14 +22,15 @@ class DestinationType(Enum):
     CM_OFFLINE_CONVERSION, \
         ADS_OFFLINE_CONVERSION, \
         ADS_SSD_UPLOAD, \
+        ADS_ENHANCED_CONVERSION, \
         ADS_CUSTOMER_MATCH_CONTACT_INFO_UPLOAD, \
         ADS_CUSTOMER_MATCH_MOBILE_DEVICE_ID_UPLOAD, \
         ADS_CUSTOMER_MATCH_USER_ID_UPLOAD, \
         GA_USER_LIST_UPLOAD, \
         APPSFLYER_S2S_EVENTS, \
         GA_MEASUREMENT_PROTOCOL, \
         GA_DATA_IMPORT, \
-        GA_4_MEASUREMENT_PROTOCOL = range(11)
+        GA_4_MEASUREMENT_PROTOCOL = range(12)
 
     def __eq__(self, other):
         if other is None:
@@ -202,7 +203,7 @@ class Batch:
     def __init__(
             self,
             execution: Execution,
-            elements: List[Dict[str, str]]
+            elements: List[Dict[str, Union[str, Dict[str, str]]]]
     ):
         self._execution = execution
         self._elements = elements
@@ -212,7 +213,7 @@ def execution(self) -> Execution:
         return self._execution
 
     @property
-    def elements(self) -> List[Dict[str, str]]:
+    def elements(self) -> List[Dict[str, Union[str, Dict[str, str]]]]:
         return self._elements
 
     def __str__(self):

megalist_dataflow/uploaders/campaign_manager/campaign_manager_conversion_uploader_test.py

@@ -147,7 +147,5 @@ def test_error_on_api_call(mocker, uploader, caplog):
 
     uploader._do_process(Batch(execution, [{'gclid': '123'}]), time.time())
 
-    print (caplog.records)
-
     assert 'Error(s) inserting conversions:' in caplog.text
     assert '[123]: error_returned' in caplog.text

megalist_dataflow/uploaders/google_ads/conversions/google_ads_enhanced_conversions_uploader.py

@@ -0,0 +1,93 @@
+# Copyright 2021 Google LLC
+#
+# Licensed under the Apache License, Version 2.0 (the "License");
+# you may not use this file except in compliance with the License.
+# You may obtain a copy of the License at
+#
+#     https://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+
+
+import logging
+from typing import Dict, Any, Union, List
+from urllib.parse import quote
+
+import apache_beam as beam
+import requests
+import json
+
+from uploaders import utils
+from models.execution import DestinationType, Batch
+from models.oauth_credentials import OAuthCredentials
+
+
+class GoogleAdsEnhancedConversionsUploaderDoFn(beam.DoFn):
+    def __init__(self, oauth_credentials: OAuthCredentials):
+        super().__init__()
+        self._api_url = "https://www.google.com/ads/event/api/v1"
+        self._ua = "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/74.0.3729.169 Safari/537.36"
+        self._oauth_credentials = oauth_credentials
+
+    def _format_query_params(self, payload: Dict[str, Any]) -> str:
+        return "&".join([key + "=" + quote(str(value)) for key, value in payload.items() if value is not None])
+
+    def _get_access_token(self):
+        payload = {
+            'client_id': self._oauth_credentials.get_client_id(),
+            'client_secret': self._oauth_credentials.get_client_secret(),
+            'refresh_token': self._oauth_credentials.get_refresh_token(),
+            'grant_type': 'refresh_token'
+        }
+        response = requests.post(
+            url='https://oauth2.googleapis.com/token', data=payload)
+        return response.json()['access_token']
+
+    @utils.safe_process(logger=logging.getLogger("megalista.GoogleAdsEnhancedConversionsUploaderDoFn"))
+    def process(self, batch: Batch, **kwargs):
+        execution = batch.execution
+        rows = batch.elements
+        headers = {
+            "Authorization": f"Bearer {self._get_access_token()}"
+        }
+
+        for row in rows:
+            payload: Dict[str, Any] = {
+                'pii_data': {},
+                'user_agent': self._ua
+            }
+            if 'addressInfo' in row and isinstance(row['addressInfo'], dict):
+                address_info: Dict[str, str] = row['addressInfo']
+                payload['pii_data']['address'] = [{
+                    'hashed_first_name': address_info['hashedFirstName'],
+                    'hashed_last_name': address_info['hashedLastName'],
+                    'country': address_info['countryCode'],
+                    'postcode': address_info['zipCode']
+                }]
+            if 'hashedEmail' in row:
+                payload['pii_data']['hashed_email'] = row['hashedEmail']
+            if 'hashedPhoneNumber' in row:
+                payload['pii_data']['hashed_phone_number'] = row['hashedPhoneNumber']
+
+            query_params = {
+                'gclid': row['gclid'],
+                'conversion_time': row['conversion_time'],
+                'conversion_tracking_id': execution.destination.destination_metadata[1],
+                'label': execution.destination.destination_metadata[0],
+                'oid': row['oid'],
+                'value': row['value'],
+                'currency_code': execution.destination.destination_metadata[2]
+            }
+
+            url = f"{self._api_url}?{self._format_query_params(query_params)}"
+
+            response = requests.post(url=url, json=payload, headers=headers)
+            if response.status_code != 200:
+                logging.getLogger('megalista.GoogleAdsEnhancedConversionsUploaderDoFn').error(
+                    f"Error uploading Enhanced Conversion {response.status_code}: {response.json()}")
+
+        yield batch

megalist_dataflow/uploaders/google_ads/conversions/google_ads_enhanced_conversions_uploader_test.py

@@ -0,0 +1,64 @@
+"""Tests for megalist_dataflow.uploaders.google_ads.google_ads_enhanced_conversions_uploader."""
+
+from uploaders.google_ads.conversions import google_ads_enhanced_conversions_uploader
+from models.oauth_credentials import OAuthCredentials
+from apache_beam.options.value_provider import StaticValueProvider
+from models.execution import Batch, Execution, AccountConfig, Source, Destination, SourceType, DestinationType
+
+import requests_mock
+import pytest
+import logging
+
+
+@pytest.fixture
+def uploader():
+    credential_id = StaticValueProvider(str, 'id')
+    secret = StaticValueProvider(str, 'secret')
+    access = StaticValueProvider(str, 'access')
+    refresh = StaticValueProvider(str, 'refresh')
+    credentials = OAuthCredentials(credential_id, secret, access, refresh)
+    return google_ads_enhanced_conversions_uploader.GoogleAdsEnhancedConversionsUploaderDoFn(credentials)
+
+
+def test_request_with_all_info(uploader, mocker):
+    mocker.patch.object(uploader, '_get_access_token')
+    with requests_mock.Mocker() as m:
+        m.post(requests_mock.ANY, status_code=200)
+        account_config = AccountConfig(
+            "123-456-7890", False, "UA-123", "CM-123", "com.app")
+        source = Source('ecs', SourceType.BIG_QUERY, ['megalista', 'ecs'])
+        destination = Destination('ecd', DestinationType.ADS_ENHANCED_CONVERSION, [
+                                  'label', 'id', 'BRL'])
+        execution = Execution(account_config, source, destination)
+
+        element = {
+            'gclid': 'Cj0KCQjwj7v0BRDOARIsAGh37ipEIqtU82p7-VKMsS1Gu-jVAtfAAr0-hsWzLv4kiZ9EpLpZpJWhUaArEnEALw_wcB',
+            'conversion_time': '1615230348456000',
+            'oid': '1',
+            'value': '15000000',
+            'hashedEmail': 'b83c49e9841ea158035dc4ac2587c512c26e5a3dbc636d68389133b5e8a6c3ca',
+            'addressInfo': {
+                'hashedFirstName': '318b22d6258b32b559d9abf7b60a8db37937f68e5c5994abb5ba3cb9610405a4',
+                'hashedLastName': 'bd442dfc1a89a054a557a516ccec91b1430f13dc9157252e0bc683f75d82cd81',
+                'countryCode': 'Brasil',
+                'zipCode': '04304032'
+            }
+        }
+
+        next(uploader.process(Batch(execution, [element])))
+
+        assert m.call_count == 1
+        assert m.last_request.query == "gclid=cj0kcqjwj7v0brdoarisagh37ipeiqtu82p7-vkmss1gu-jvatfaar0-hswzlv4kiz9eplpzpjwhuaarenealw_wcb&conversion_time=1615230348456000&conversion_tracking_id=id&label=label&oid=1&value=15000000&currency_code=brl"
+        assert m.last_request.json() == {
+            'pii_data': {
+                'address': [
+                    {
+                        'hashed_first_name': '318b22d6258b32b559d9abf7b60a8db37937f68e5c5994abb5ba3cb9610405a4',
+                        'hashed_last_name': 'bd442dfc1a89a054a557a516ccec91b1430f13dc9157252e0bc683f75d82cd81',
+                        'country': 'Brasil', 'postcode': '04304032'
+                    }
+                ],
+                'hashed_email': 'b83c49e9841ea158035dc4ac2587c512c26e5a3dbc636d68389133b5e8a6c3ca'
+            },
+            'user_agent': 'Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/74.0.3729.169 Safari/537.36'
+        }

megalist_dataflow/uploaders/google_analytics/google_analytics_data_import_uploader.py

@@ -22,7 +22,7 @@
 from googleapiclient.http import MediaInMemoryUpload
 
 from uploaders import utils
-from models.execution import DestinationType, Batch
+from models.execution import DestinationType, Batch, Union
 
 
 class GoogleAnalyticsDataImportUploaderDoFn(beam.DoFn):
@@ -85,7 +85,7 @@ def process(self, batch: Batch, **kwargs):
                              ga_account_id, batch.elements)
 
     def _do_upload_data(self, web_property_id, data_import_name, ga_account_id,
-                        rows: List[Dict[str, str]]):
+                        rows: List[Dict[str, Union[str, Dict[str, str]]]]):
         analytics = self._get_analytics_service()
         data_sources = analytics.management().customDataSources().list(
             accountId=ga_account_id,

megalist_dataflow/uploaders/utils.py

@@ -55,7 +55,7 @@ def inner(*args, **kwargs):
                 return
             logger.info(f'Uploading {len(batch.elements)} rows...')
             try:
-                return func(*args, *kwargs)
+                return func(*args, **kwargs)
             except Exception as e:
                 logger.error(f'Error uploading data for :{batch.elements}')
                 logger.error(e, exc_info=True)