import requests import os import logging import json import re import uuid import hashlib from firebase.firebase import Firebase from google.ads.googleads.client import GoogleAdsClient from google.ads.googleads.errors import GoogleAdsException from google_auth_oauthlib.flow import Flow, InstalledAppFlow LOGGING_PREFIX = "[GG CTM Sync]: " fb = Firebase(host=os.environ.get("FIREBASE_HOST")) logging.basicConfig(level=logging.INFO) logger = logging.getLogger(__name__) def _sanitize_error(err): pass class GoogleCustomerMatch: def __init__(self): self.config_dict = { "developer_token": os.environ.get("GOOGLE_AD_DEVELOPER_TOKEN"), "client_id": os.environ.get("GOOGLE_AD_CLIENT_ID"), "client_secret": os.environ.get("GOOGLE_AD_CLIENT_SECRET"), "use_proto_plus": True } self.oauth2_url = "https://oauth2.googleapis.com" self.oauth2_redirect_url = "https://develop-c360-ydm-907324191702.asia-southeast1.run.app/" def fetch_refresh_token(self, code): try: data = { "code": code, "client_id": self.config_dict['client_id'], "client_secret": self.config_dict['client_secret'], "redirect_uri": self.oauth2_redirect_url, "grant_type": "authorization_code" } response = requests.post(f"{self.oauth2_url}/token", data=data) response_json = response.json() if response.status_code == 200: return { "status": "success", "refresh_token": response_json.get("refresh_token"), "access_token": response_json.get("access_token") } else: return { "status": "error", "message": response_json.get("error_description", response.text) } except Exception as e: return {"status": "error", "message": str(e)} def get_access_token(self, user_id): ref = fb.db.reference(f"users/{user_id}/ad_account/google_customer_match/") access_token = ref.child("access_token").get() if access_token: self._set_up_client(access_token) return access_token def _set_up_client(self, refresh_token): config_dict = self.config_dict config_dict['refresh_token'] = refresh_token client = GoogleAdsClient.load_from_dict(config_dict) #Set self.client = client self.config_dict = config_dict def _set_login_customer_id(self, customer_id): self.client.login_customer_id = customer_id self.client.customer_id = customer_id def list_accounts(self): customer_service = self.client.get_service("CustomerService") google_ads_service = self.client.get_service("GoogleAdsService") response = customer_service.list_accessible_customers() account_list = [] for resource_name in response.resource_names: customer_id = resource_name.split("/")[1] self.client.login_customer_id = customer_id # Query the 'customer' resource to get the name query = """ SELECT customer.id, customer.descriptive_name FROM customer LIMIT 1 """ try: search_response = google_ads_service.search( customer_id=customer_id, query=query ) for row in search_response: account_data = { "id": row.customer.id, "name": row.customer.descriptive_name or "Unnamed Account" } account_list.append(account_data) except GoogleAdsException as inner_ex: logging.info(f"Skipping ID {customer_id}: {inner_ex.error.code().name}") return account_list def search_open_user_lists(self, client_cid, upload_key_type): google_ads_service = self.client.get_service("GoogleAdsService") query = f""" SELECT user_list.id, user_list.name FROM user_list WHERE user_list.type = 'CRM_BASED' AND user_list.read_only = FALSE AND user_list.account_user_list_status = 'ENABLED' AND user_list.membership_status = 'OPEN' -- AND user_list.crm_based_user_list.upload_key_type = '{upload_key_type}' """ try: search_response = google_ads_service.search( customer_id=str(client_cid), query=query ) results = [] for row in search_response: results.append({ "id": row.user_list.id, "name": row.user_list.name }) return results except GoogleAdsException as ex: print(f"Request with ID '{ex.request_id}' failed with status " f"'{ex.error.code().name}' and includes the following errors:") for error in ex.failure.errors: print(f"\tError with message '{error.message}'.") if error.location: for field_path_element in error.location.field_path_elements: print(f"\t\tOn field: {field_path_element.field_name}") return [] def search_client_customers(self, client_cid): google_ads_service = self.client.get_service("GoogleAdsService") query = """ SELECT customer_client.client_customer, customer_client.hidden, customer_client.id, customer_client.level, customer_client.resource_name, customer_client.test_account, customer_client.descriptive_name, customer_client.manager, customer_client.status FROM customer_client WHERE customer_client.status NOT IN ('CANCELED', 'SUSPENDED') """ try: search_response = google_ads_service.search( customer_id=str(client_cid), query=query ) results = [] for row in search_response: client = row.customer_client results.append({ "client_customer": client.client_customer, "hidden": client.hidden, "id": client.id, "level": client.level, "resource_name": client.resource_name, "test_account": client.test_account, "descriptive_name": client.descriptive_name, "manager": client.manager, "status": client.status.name }) return results except GoogleAdsException as ex: print(f"Request with ID '{ex.request_id}' failed with status " f"'{ex.error.code().name}' and includes the following errors:") for error in ex.failure.errors: print(f"\tError with message '{error.message}'.") return [] def create_user_list(self, target_cid, new_list_name, new_list_description, upload_key_type, mobile_app_id=None): user_list_service = self.client.get_service("UserListService") user_list_operation = self.client.get_type("UserListOperation") user_list = user_list_operation.create user_list.name = new_list_name user_list.description = new_list_description user_list.membership_status = self.client.enums.UserListMembershipStatusEnum.OPEN user_list.membership_life_span = 540 # CRM Based User List user_list.crm_based_user_list.upload_key_type = self.client.enums.CustomerMatchUploadKeyTypeEnum[upload_key_type] user_list.crm_based_user_list.data_source_type = self.client.enums.UserListCrmDataSourceTypeEnum.FIRST_PARTY if mobile_app_id: user_list.crm_based_user_list.app_id = mobile_app_id try: response = user_list_service.mutate_user_lists( customer_id=str(target_cid), operations=[user_list_operation] ) return response.results[0].resource_name except GoogleAdsException as ex: print(f"Request with ID '{ex.request_id}' failed with status " f"'{ex.error.code().name}' and includes the following errors:") for error in ex.failure.errors: print(f"\tError with message '{error.message}'.") return None def create_data_job(self, target_cid, user_list_resource_name, consent_ad_user_data=None, consent_ad_personalization=None): offline_user_data_job_service = self.client.get_service("OfflineUserDataJobService") job = self.client.get_type("OfflineUserDataJob") job.type_ = self.client.enums.OfflineUserDataJobTypeEnum.CUSTOMER_MATCH_USER_LIST job.customer_match_user_list_metadata.user_list = user_list_resource_name if consent_ad_user_data or consent_ad_personalization: # Assuming consent_ad_user_data and consent_ad_personalization are strings like 'GRANTED' or 'DENIED' # We need to map them to the enum if they are passed as strings, or use them directly if they are enums. # For simplicity, let's assume they are passed as valid enum member names or None. if consent_ad_user_data: job.customer_match_user_list_metadata.consent.ad_user_data = self.client.enums.ConsentStatusEnum[consent_ad_user_data] if consent_ad_personalization: job.customer_match_user_list_metadata.consent.ad_personalization = self.client.enums.ConsentStatusEnum[consent_ad_personalization] try: response = offline_user_data_job_service.create_offline_user_data_job( customer_id=str(target_cid), job=job ) return response.resource_name except GoogleAdsException as ex: print(f"Request with ID '{ex.request_id}' failed with status " f"'{ex.error.code().name}' and includes the following errors:") for error in ex.failure.errors: print(f"\tError with message '{error.message}'.") return None def add_data_job_operations(self, offline_user_data_job_resource_name, user_identifiers): # user_identifiers should be a list of UserIdentifier objects or dicts that can be converted # However, the JS example passes `operations` directly. # In Python, we need to create OfflineUserDataJobOperation objects. offline_user_data_job_service = self.client.get_service("OfflineUserDataJobService") operations = [] for identifier in user_identifiers: op = self.client.get_type("OfflineUserDataJobOperation") # Create UserData object and add the identifier to it user_data = self.client.get_type("UserData") user_data.user_identifiers.append(identifier) op.create = user_data operations.append(op) # Construct the request request = self.client.get_type("AddOfflineUserDataJobOperationsRequest") request.resource_name = offline_user_data_job_resource_name request.enable_partial_failure = True request.operations = operations try: response = offline_user_data_job_service.add_offline_user_data_job_operations( request=request ) return response except GoogleAdsException as ex: print(f"Request with ID '{ex.request_id}' failed with status " f"'{ex.error.code().name}' and includes the following errors:") for error in ex.failure.errors: print(f"\tError with message '{error.message}'.") return None def run_job(self, offline_user_data_job_resource_name): offline_user_data_job_service = self.client.get_service("OfflineUserDataJobService") try: response = offline_user_data_job_service.run_offline_user_data_job( resource_name=offline_user_data_job_resource_name ) return response except GoogleAdsException as ex: print(f"Request with ID '{ex.request_id}' failed with status " f"'{ex.error.code().name}' and includes the following errors:") for error in ex.failure.errors: print(f"\tError with message '{error.message}'.") return None def normalize_and_hash(self, value): if not value: return None normalized = value.strip().lower() return hashlib.sha256(normalized.encode('utf-8')).hexdigest() def create_user_identifier(self, type, value): user_identifier = self.client.get_type("UserIdentifier") if type == 'email': user_identifier.hashed_email = self.normalize_and_hash(value) elif type == 'phone': user_identifier.hashed_phone_number = self.normalize_and_hash(value) elif type == 'mobile_id': user_identifier.mobile_id = value elif type == 'third_party_user_id': user_identifier.third_party_user_id = value elif type == 'address_info': # Value should be a dict with keys like 'first_name', 'last_name', 'country_code', 'postal_code' address_info = self.client.get_type("OfflineUserAddressInfo") if 'first_name' in value: address_info.hashed_first_name = self.normalize_and_hash(value['first_name']) if 'last_name' in value: address_info.hashed_last_name = self.normalize_and_hash(value['last_name']) if 'country_code' in value: address_info.country_code = value['country_code'] if 'postal_code' in value: address_info.postal_code = value['postal_code'] if 'street_address' in value: address_info.hashed_street_address = self.normalize_and_hash(value['street_address']) if 'city' in value: address_info.city = value['city'] if 'state' in value: address_info.state = value['state'] user_identifier.address_info = address_info return user_identifier