from typing import List, Callable import snowflake.connector from snowflake.connector import SnowflakeConnection import config from utils.snowflake.metrics.schema import ( MetricsFilters, MetricsQueryResult, TotalMetricsQueryResult, MetricsDetailsQueryResult, AccountProvider, ) from utils.snowflake.metrics.queries.metrics_queries import MetricsQueries from utils.snowflake.metrics.queries.metrics_details_queries.metrics_campaign_details import ( MetricsCampaignDetailsQueries, ) from utils.snowflake.metrics.queries.metrics_details_queries.metrics_artist_details import MetricsArtistDetailsQueries from utils.snowflake.metrics.queries.metrics_details_queries.metrics_project_details import MetricsProjectDetailsQueries MAX_METRICS_INDEX = 37 MAX_TOTAL_METRICS_INDEX = 74 class PerformanceMetricsFetcher: project_campaigns_data: list project_artists_data: list providers_data: List[AccountProvider] global_territory_id = 0 def connector(self) -> SnowflakeConnection: return snowflake.connector.connect( user=config.SNOWFLAKE_USER, password=config.SNOWFLAKE_PASS, account=config.SNOWFLAKE_ACCOUNT, database=config.ADS_SNOWFLAKE_DB, schema=config.ADS_SNOWFLAKE_SCHEMA, warehouse=config.SNOWFLAKE_WAREHOUSE, ) def __init__( self, project_campaigns: list, project_artists: list, providers_data: List[AccountProvider], include_artists: bool = True, ): self.project_artists_data = project_artists self.project_campaigns_data = project_campaigns self.providers_data = providers_data self.include_artists = include_artists def __default_data_validation(self): return self.project_campaigns_data is not None and self.project_artists_data is not None def __campaigns_data_validation(self): return self.project_campaigns_data is not None def __fetch_metrics( self, metrics_query: str, validation: Callable = None ) -> (List[MetricsQueryResult], TotalMetricsQueryResult): validation = validation or self.__default_data_validation if not validation(): return [], None with self.connector() as con: if self.project_campaigns_data: data = con.cursor().execute(metrics_query).fetchall() if len(data) != 0: total_results = TotalMetricsQueryResult._make(data[0][MAX_METRICS_INDEX:MAX_TOTAL_METRICS_INDEX]) return [MetricsQueryResult._make(result[0:MAX_METRICS_INDEX]) for result in data], total_results return [], None def __fetch_entity_details_with_query(self, query: str, mapper: Callable = None) -> List[MetricsQueryResult]: if not self.project_campaigns_data or (not self.project_artists_data and self.include_artists): return [] with self.connector() as con: data = con.cursor().execute(query).fetchall() items = [MetricsDetailsQueryResult._make(result) for result in data] if mapper: return list(map(mapper, items)) return items def fetch_projects_metrics_data( self, filters: MetricsFilters ) -> (List[MetricsQueryResult], TotalMetricsQueryResult): if not self.project_campaigns_data: return [], None query = MetricsQueries.fetch_projects_metrics_query(filters, self.project_campaigns_data) return self.__fetch_metrics(query) # Exactly the same logic as for projects. This method is only for semantic def fetch_phases_metrics_data(self, filters: MetricsFilters) -> (List[MetricsQueryResult], TotalMetricsQueryResult): return self.fetch_projects_metrics_data(filters) def fetch_artists_metrics_data( self, filters: MetricsFilters ) -> (List[MetricsQueryResult], TotalMetricsQueryResult): if not self.project_campaigns_data or not self.project_artists_data: return [], None query = MetricsQueries.fetch_artists_metrics_query( filters, self.project_campaigns_data, self.project_artists_data ) return self.__fetch_metrics(query) def fetch_campaign_metrics_data( self, filters: MetricsFilters ) -> (List[MetricsQueryResult], TotalMetricsQueryResult): if not self.project_campaigns_data: return [], None query = MetricsQueries.fetch_campaigns_metrics_query(filters, self.project_campaigns_data) return self.__fetch_metrics(query, self.__campaigns_data_validation) def fetch_project_details_by_platforms(self, entity_id: int, filters: MetricsFilters): query = MetricsProjectDetailsQueries.fetch_details_by_platform(entity_id, filters, self.project_campaigns_data) return self.__fetch_entity_details_with_query(query) def fetch_project_details_by_campaigns(self, entity_id: int, filters: MetricsFilters): query = MetricsProjectDetailsQueries.fetch_details_by_campaign(entity_id, filters, self.project_campaigns_data) return self.__fetch_entity_details_with_query(query) def fetch_project_details_by_categories(self, entity_id: int, filters: MetricsFilters): query = MetricsProjectDetailsQueries.fetch_details_by_category(entity_id, filters, self.project_campaigns_data) return self.__fetch_entity_details_with_query(query) def fetch_project_details_by_objective(self, entity_id: int, filters: MetricsFilters): query = MetricsProjectDetailsQueries.fetch_details_by_objective(entity_id, filters, self.project_campaigns_data) return self.__fetch_entity_details_with_query(query) def fetch_project_details_by_goal(self, entity_id: int, filters: MetricsFilters): query = MetricsProjectDetailsQueries.fetch_details_by_goal(entity_id, filters, self.project_campaigns_data) return self.__fetch_entity_details_with_query(query) def fetch_project_details_by_type(self, entity_id: int, filters: MetricsFilters): query = MetricsProjectDetailsQueries.fetch_details_by_type(entity_id, filters, self.project_campaigns_data) return self.__fetch_entity_details_with_query(query) def fetch_project_details_by_country(self, entity_id: int, filters: MetricsFilters): query = MetricsProjectDetailsQueries.fetch_details_by_country(entity_id, filters, self.project_campaigns_data) return self.__fetch_entity_details_with_query(query) def fetch_project_details_by_day(self, entity_id: int, filters: MetricsFilters): query = MetricsProjectDetailsQueries.fetch_details_by_day(entity_id, filters, self.project_campaigns_data) return self.__fetch_entity_details_with_query(query) def fetch_project_details_by_week(self, entity_id: int, filters: MetricsFilters): query = MetricsProjectDetailsQueries.fetch_details_by_week(entity_id, filters, self.project_campaigns_data) return self.__fetch_entity_details_with_query(query) def fetch_project_details_by_month(self, entity_id: int, filters: MetricsFilters): query = MetricsProjectDetailsQueries.fetch_details_by_month(entity_id, filters, self.project_campaigns_data) return self.__fetch_entity_details_with_query(query) def fetch_project_details_by_chartweek(self, entity_id: int, filters: MetricsFilters): query = MetricsProjectDetailsQueries.fetch_details_by_chartweek(entity_id, filters, self.project_campaigns_data) return self.__fetch_entity_details_with_query(query) def fetch_project_details_by_genders(self, entity_id: int, filters: MetricsFilters): query = MetricsProjectDetailsQueries.fetch_details_by_gender(entity_id, filters, self.project_campaigns_data) return self.__fetch_entity_details_with_query(query) def fetch_project_details_by_age(self, entity_id: int, filters: MetricsFilters): query = MetricsProjectDetailsQueries.fetch_details_by_age(entity_id, filters, self.project_campaigns_data) return self.__fetch_entity_details_with_query(query) def fetch_project_details_by_ad_sets(self, entity_id: int, filters: MetricsFilters): query = MetricsProjectDetailsQueries.fetch_details_by_ad_set(entity_id, filters, self.project_campaigns_data) return self.__fetch_entity_details_with_query(query) def fetch_project_details_by_providers(self, entity_id: int, filters: MetricsFilters): query = MetricsProjectDetailsQueries.fetch_details_by_provider( entity_id, filters, self.project_campaigns_data, self.providers_data ) return self.__fetch_entity_details_with_query(query) def fetch_artist_details_by_projects(self, entity_id: int, filters: MetricsFilters): query = MetricsArtistDetailsQueries.fetch_details_by_project( entity_id, filters, self.project_artists_data, self.project_campaigns_data ) return self.__fetch_entity_details_with_query(query) def fetch_artist_details_by_campaigns(self, entity_id: int, filters: MetricsFilters): query = MetricsArtistDetailsQueries.fetch_details_by_campaign( entity_id, filters, self.project_artists_data, self.project_campaigns_data ) return self.__fetch_entity_details_with_query(query) def fetch_artist_details_by_ad_sets(self, entity_id: int, filters: MetricsFilters): query = MetricsArtistDetailsQueries.fetch_details_by_ad_set( entity_id, filters, self.project_artists_data, self.project_campaigns_data ) return self.__fetch_entity_details_with_query(query) def fetch_artist_details_by_platforms(self, entity_id: int, filters: MetricsFilters): query = MetricsArtistDetailsQueries.fetch_details_by_platform( entity_id, filters, self.project_artists_data, self.project_campaigns_data ) return self.__fetch_entity_details_with_query(query) def fetch_artist_details_by_category(self, entity_id: int, filters: MetricsFilters): query = MetricsArtistDetailsQueries.fetch_details_by_category( entity_id, filters, self.project_artists_data, self.project_campaigns_data ) return self.__fetch_entity_details_with_query(query) def fetch_artist_details_by_objective(self, entity_id: int, filters: MetricsFilters): query = MetricsArtistDetailsQueries.fetch_details_by_objective( entity_id, filters, self.project_artists_data, self.project_campaigns_data ) return self.__fetch_entity_details_with_query(query) def fetch_artist_details_by_goal(self, entity_id: int, filters: MetricsFilters): query = MetricsArtistDetailsQueries.fetch_details_by_goal( entity_id, filters, self.project_artists_data, self.project_campaigns_data ) return self.__fetch_entity_details_with_query(query) def fetch_artist_details_by_type(self, entity_id: int, filters: MetricsFilters): query = MetricsArtistDetailsQueries.fetch_details_by_type( entity_id, filters, self.project_artists_data, self.project_campaigns_data ) return self.__fetch_entity_details_with_query(query) def fetch_artist_details_by_country(self, entity_id: int, filters: MetricsFilters): query = MetricsArtistDetailsQueries.fetch_details_by_country( entity_id, filters, self.project_artists_data, self.project_campaigns_data ) return self.__fetch_entity_details_with_query(query) def fetch_artist_details_by_day(self, entity_id: int, filters: MetricsFilters): query = MetricsArtistDetailsQueries.fetch_details_by_day( entity_id, filters, self.project_artists_data, self.project_campaigns_data ) return self.__fetch_entity_details_with_query(query) def fetch_artist_details_by_month(self, entity_id: int, filters: MetricsFilters): query = MetricsArtistDetailsQueries.fetch_details_by_month( entity_id, filters, self.project_artists_data, self.project_campaigns_data ) return self.__fetch_entity_details_with_query(query) def fetch_artist_details_by_week(self, entity_id: int, filters: MetricsFilters): query = MetricsArtistDetailsQueries.fetch_details_by_week( entity_id, filters, self.project_artists_data, self.project_campaigns_data ) return self.__fetch_entity_details_with_query(query) def fetch_artist_details_by_chartweek(self, entity_id: int, filters: MetricsFilters): query = MetricsArtistDetailsQueries.fetch_details_by_chartweek( entity_id, filters, self.project_artists_data, self.project_campaigns_data ) return self.__fetch_entity_details_with_query(query) def fetch_artist_details_by_genders(self, entity_id: int, filters: MetricsFilters): query = MetricsArtistDetailsQueries.fetch_details_by_gender( entity_id, filters, self.project_artists_data, self.project_campaigns_data ) return self.__fetch_entity_details_with_query(query) def fetch_artist_details_by_provider(self, entity_id: int, filters: MetricsFilters): query = MetricsArtistDetailsQueries.fetch_details_by_provider( entity_id, filters, self.project_artists_data, self.project_campaigns_data, self.providers_data ) return self.__fetch_entity_details_with_query(query) def fetch_artist_details_by_age(self, entity_id: int, filters: MetricsFilters): query = MetricsArtistDetailsQueries.fetch_details_by_age( entity_id, filters, self.project_artists_data, self.project_campaigns_data ) return self.__fetch_entity_details_with_query(query) def fetch_campaigns_details_by_country(self, entity_id: int, filters: MetricsFilters): query = MetricsCampaignDetailsQueries.fetch_details_by_country(entity_id, filters, self.project_campaigns_data) return self.__fetch_entity_details_with_query(query) def fetch_campaigns_details_by_age(self, entity_id: int, filters: MetricsFilters): query = MetricsCampaignDetailsQueries.fetch_details_by_age(entity_id, filters, self.project_campaigns_data) return self.__fetch_entity_details_with_query(query) def fetch_campaigns_details_by_gender(self, entity_id: int, filters: MetricsFilters): query = MetricsCampaignDetailsQueries.fetch_details_by_gender(entity_id, filters, self.project_campaigns_data) return self.__fetch_entity_details_with_query(query) def fetch_campaigns_details_by_platform(self, entity_id: int, filters: MetricsFilters): query = MetricsCampaignDetailsQueries.fetch_details_by_platform(entity_id, filters, self.project_campaigns_data) return self.__fetch_entity_details_with_query(query) def fetch_campaigns_details_by_provider(self, entity_id: int, filters: MetricsFilters): query = MetricsCampaignDetailsQueries.fetch_details_by_provider( entity_id, filters, self.project_campaigns_data, self.providers_data ) return self.__fetch_entity_details_with_query(query) def fetch_campaigns_details_by_goal(self, entity_id: int, filters: MetricsFilters): query = MetricsCampaignDetailsQueries.fetch_details_by_goal(entity_id, filters, self.project_campaigns_data) return self.__fetch_entity_details_with_query(query) def fetch_campaigns_details_by_ad_set(self, entity_id: int, filters: MetricsFilters): query = MetricsCampaignDetailsQueries.fetch_details_by_ad_set(entity_id, filters, self.project_campaigns_data) return self.__fetch_entity_details_with_query(query) def fetch_campaigns_details_by_day(self, entity_id: int, filters: MetricsFilters): query = MetricsCampaignDetailsQueries.fetch_details_by_day(entity_id, filters, self.project_campaigns_data) return self.__fetch_entity_details_with_query(query) def fetch_campaigns_details_by_week(self, entity_id: int, filters: MetricsFilters): query = MetricsCampaignDetailsQueries.fetch_details_by_week(entity_id, filters, self.project_campaigns_data) return self.__fetch_entity_details_with_query(query) def fetch_campaigns_details_by_chartweek(self, entity_id: int, filters: MetricsFilters): query = MetricsCampaignDetailsQueries.fetch_details_by_chartweek( entity_id, filters, self.project_campaigns_data ) return self.__fetch_entity_details_with_query(query) def fetch_campaigns_details_by_month(self, entity_id: int, filters: MetricsFilters): query = MetricsCampaignDetailsQueries.fetch_details_by_month(entity_id, filters, self.project_campaigns_data) return self.__fetch_entity_details_with_query(query)