import json from typing import List, Optional from psycopg2.extras import NumericRange from db import db from utils import list_utils from models.marketing_accounts import MarketingAccount, MarketingAccountSource from utils.snowflake.google_data.territories_mapping import google_territories_mapping from utils.snowflake.google_data.models import ( GRASCampaignsQueryResult, GRASCampaignNetworkType, GRASCampaignUrl, LinkfireLink, ) from constants.labels_constants import TEST_UK_LABEL_ID, TEST_US_LABEL_ID class GoogleImporterUtils: def map_types(self, types: Optional[str]) -> Optional[List[GRASCampaignNetworkType]]: if types: return [GRASCampaignNetworkType(type["main"], type.get("secondary", type["main"])) for type in json.loads(types)] else: return None def map_links(self, links: Optional[str]) -> Optional[List[GRASCampaignUrl]]: if links: return [self.__map_link(link) for link in json.loads(links) if link.get("final_urls")] else: return None def map_linkfire_links(self, links: Optional[str]) -> Optional[List[LinkfireLink]]: if links: return [ LinkfireLink(link_id=link.get("linkfire_id"), link_url=link.get("linkfire_url")) for link in json.loads(links) if link.get("linkfire_id") ] else: return None def __map_link(self, link) -> GRASCampaignUrl: urls = link["final_urls"] return GRASCampaignUrl(title=link.get("display_url"), links=json.loads(urls)) def map_artists_external_ids(self, item: GRASCampaignsQueryResult) -> List[str]: artists = [] if item.artists: artists = json.loads(item.artists) elif item.alt_artists_by_x: artists = json.loads(item.alt_artists_by_x) elif item.alt_artists_by_plus: artists = json.loads(item.alt_artists_by_plus) elif item.alt_artists_by_comma: artists = json.loads(item.alt_artists_by_comma) ids = [artist.get("member_id", artist["artist_id"]) for artist in artists if artist.get("artist_id")] return [f"GRAS_{id}" for id in ids] def marketing_accounts(self) -> List[str]: accounts = ( db.session.query(MarketingAccount.external_id) .filter( MarketingAccount.source == MarketingAccountSource.GOOGLE.value, MarketingAccount.label_id.notin_([TEST_UK_LABEL_ID, TEST_US_LABEL_ID]) ) .all() ) return [item[0] for item in accounts] class GoogleWorkerUtils: all_ages = 0, 65 all_genders = [1, 2, 3] global_territory_id = 0 max_links = 5 source_name = "google" def build_age_range(self, age_ranges: List[str]) -> NumericRange: if not age_ranges or (len(age_ranges) == 1 and age_ranges[0] == "Undetermined"): return NumericRange(self.all_ages[0], self.all_ages[1], "[]") ranges = map(self.__map_range_from, filter(lambda item: item != "Undetermined", age_ranges)) result = [] for begin, end in sorted(ranges): if result and result[-1][1] >= begin - 1: result[-1][1] = max(result[-1][1], end) else: result.append([begin, end]) return NumericRange(result[0][0], result[-1][1], "[]") def __map_range_from(self, age_string: str) -> (int, int): if age_string == "65 or more": return 65, 65 values = age_string.split("-") return int(values[0]), int(values[1]) def build_gender_ids(self, genders: List[str]) -> Optional[List[int]]: if not genders: return None return list(map(self.__map_gender_from, genders)) def __map_gender_from(self, gender_string: str) -> (int, int): if gender_string == "Male": return 1 if gender_string == "Female": return 2 else: return 3 def build_objective(self, bidding_type: Optional[str]) -> int: objective_map = { "Maximize Conversions": 3, # Conversions "Maximize Conversion Value": 3, # Conversions "Maximize clicks": 10, # Clicks "cpc": 10, # Clicks "cpv": 17, # Video Views "Target CPA": 3, # Conversions "Target ROAS": 3, # Conversions "cpm": 11, # Impressions "Target search page location": 16, # Traffic "Target Outranking Share": 10, # Clicks "unknown": 10, # Clicks } return objective_map.get(bidding_type, 10) def build_campaign_links(self, links: Optional[List[GRASCampaignUrl]]) -> List[str]: links_values = [link.links[0] for link in links or []] return list(set(links_values))[: self.max_links] def get_campaign_platforms(self, types: List[GRASCampaignNetworkType]): if not types: return [] return self.__build_platforms_by_gras_types(types) def __build_platforms_by_gras_types(self, types: List[GRASCampaignNetworkType]): if len(types) == 1: mapping = { # Search Ads - Google - Paid ("Search Network", "Search partners"): ["Search Ads", ["Google"]], # Search Ads - Youtube - Paid ("Search Network", "Google search"): ["Search Ads", ["Youtube"]], # Search Ads - Google - Paid ("YouTube Search", "YouTube Search"): ["Search Ads", ["Google"]], # Social Ads - Youtube - Paid ("YouTube Videos", "YouTube Videos"): ["Social Ads", ["Youtube"]], # Display Ads - Google - Paid ("Display Network", "Display Network"): ["Display Ads", ["Google"]], # Search Ads - Google + Youtube - Paid ("Cross-network", "Cross-network"): ["Search Ads", ["Google", "Youtube"]], } return mapping.get((types[0].main, types[0].secondary), "") if list_utils.exists(types, lambda x: x.main == "YouTube Videos"): return ["Social Ads", ["Youtube"]] # Social Ads - Youtube - Paid if list_utils.exists(types, lambda x: x.main == "Display Network"): if list_utils.exists(types, lambda x: x.main == "YouTube Search"): return ["Display Ads", ["Google", "Youtube"]] # Display Ads - Google - Paid return ["Display Ads", ["Google"]] # Display Ads - Google - Paid if list_utils.exists(types, lambda x: x.main == "YouTube Search"): return ["Search Ads", ["Google", "Youtube"]] # Search Ads - Google + Youtube - Paid return ["Search Ads", ["Google"]] # Search Ads - Google - Paid def map_territories(self, territories: Optional[List[int]]) -> List[int]: if not territories: return [self.global_territory_id] territories_ids = [] for territory in territories: territory_id = google_territories_mapping.get(territory) if territory_id: territories_ids.append(territory_id) else: print(f"Territory with id {territory} not found") if not territories_ids: return [self.global_territory_id] return territories_ids