"""Logic for retrieving TADAS related data.""" from collections import OrderedDict from typing import Any, Dict, List, Mapping from analytics.constants import cache from analytics.queries.format import format_row from analytics.queries.tadas import ( RepOwners, TADASDataAvailability, TADASTrendByISRC, TADASTrends, ) from analytics.utils.cache import cache_in_redis FLAGS_CONFIG = OrderedDict( [ ( "S", [ { "name": "Spotify Collection Streams", "flag": "spotify_collection_flag", "lift": "spotify_collection_lift", "store_id": 286, }, { "name": "Spotify Lean Forward Streams", "flag": "spotify_lean_forward_flag", "lift": "spotify_lean_forward_lift", "store_id": 286, }, { "name": "TikTok Creations", "flag": "tiktok_creations_global_flag", "lift": "tiktok_creations_global_lift", "market": "GLOBAL", "store_id": 1202, }, { "name": "TikTok Creations", "flag": "tiktok_creations_country_flag", "lift": "tiktok_creations_country_lift", "store_id": 1202, }, { "name": "Spotify Search Streams", "flag": "spotify_search_flag", "lift": "spotify_search_lift", "store_id": 286, }, ], ), ( "ONE", [ { "name": "TikTok Views", "flag": "tiktok_views_global_flag", "lift": "tiktok_views_global_lift", "market": "GLOBAL", "store_id": 1202, }, { "name": "TikTok Views", "flag": "tiktok_views_country_flag", "lift": "tiktok_views_country_lift", "store_id": 1202, }, { "name": "Apple Music Lean Forward Streams", "flag": "apple_lean_forward_flag", "lift": "apple_lean_forward_lift", "store_id": 1, }, { "name": "Apple Music Search Streams", "flag": "apple_search_flag", "lift": "apple_search_lift", "store_id": 1, }, { "name": "TikTok Likes", "flag": "tiktok_likes_global_flag", "lift": "tiktok_likes_global_lift", "market": "GLOBAL", "store_id": 1202, }, { "name": "TikTok Likes", "flag": "tiktok_likes_country_flag", "lift": "tiktok_likes_country_lift", "store_id": 1202, }, ], ), ( "TWO", [ { "name": "Apple Music Overall streams", "flag": "apple_all_flag", "lift": "apple_all_lift", "store_id": 1, } ], ), ( "THREE", [ { "name": "Spotify Overall Streams", "flag": "spotify_all_flag", "lift": "spotify_all_lift", "store_id": 286, }, ], ), ] ) def get_trending_flags_for_row(row): trending_flags = [] for tier, flag_defs in FLAGS_CONFIG.items(): tier_flags = [] for flag_def in flag_defs: if row.get(flag_def["flag"]) is True: lift_value = row.get(flag_def["lift"], 0) tier_flags.append( { "name": flag_def["name"], "lift": lift_value, "tier": tier, "market": flag_def.get("market", row.get("market")), "store_id": flag_def["store_id"], } ) # Sort by lift value descending tier_flags.sort(key=lambda x: x["lift"], reverse=True) # Add flags from this tier to trending_flags trending_flags.extend(tier_flags) # If we have any flags from this tier and we have 3 or more total flags, # take top 3 and stop if tier_flags and len(trending_flags) >= 3: trending_flags = trending_flags[:3] break # If we have any flags from this tier but fewer than 3, continue to next tier # Only if we have NO flags from this tier, continue to next tier # Take top 3 flags across all tiers we processed return trending_flags[:3] def calculate_trending_flags(results: List[Dict]) -> List[Dict]: """Calculate trending flags for TADAS results. 1. Flags should be assessed first based on tier (S-3). 2. Within that tier, the flags with the 3 highest percent increase in the corresponding _lift column should be shown. 3. If there are no trending flags in a tier or under 3 flags in a tier, then the next tier should be evaluated for the highest percent increase flags to show. """ for row in results: row["trending_flags"] = get_trending_flags_for_row(row) # Remove all _flag and _lift columns from the row keys_to_remove = [k for k in row if k.endswith("_flag") or k.endswith("_lift")] for k in keys_to_remove: del row[k] return results @cache_in_redis(ttl=cache.FIVE_MINUTES) def get_tadas_trends( query_params: Mapping[str, Any], permissions: Mapping[str, Any], ) -> Dict: """Get TADAS Trends.""" query = TADASTrends({**query_params, **permissions}) result = [ format_row( data_point, array_fields=["company_brand_uuids", "rep_owner_label_ids"], ) for data_point in query.execute() ] if result: total_count = result[0]["total_count"] else: total_count = 0 result = calculate_trending_flags(result) return {"trends": result, "count": total_count} @cache_in_redis(ttl=cache.FIVE_MINUTES) def get_tadas_trend_globalsoundrecording_by_isrc( query_params: Mapping[str, Any], permissions: Mapping[str, Any], ) -> Dict: """Get TADAS Trend for GSR by ISRC.""" query = TADASTrendByISRC({**query_params, **permissions}) result = [ format_row( data_point, array_fields=["company_brand_uuids", "rep_owner_label_ids"], ) for data_point in query.execute() ] if result: total_count = result[0]["total_count"] else: total_count = 0 result = calculate_trending_flags(result) return {"trends": result, "count": total_count} @cache_in_redis(ttl=cache.FIVE_MINUTES) def get_tadas_data_availability( query_params: Mapping[str, Any], permissions: Mapping[str, Any], ) -> Dict: """Get TADAS Data Availability.""" query = TADASDataAvailability({**query_params, **permissions}) result = [format_row(data_point) for data_point in query.execute()] return result[0] @cache_in_redis(ttl=cache.FIVE_MINUTES) def get_rep_owners( query_params: Mapping[str, Any], permissions: Mapping[str, Any], ) -> List[Dict]: """Get the list of rep owners and the label ids that roll up to each.""" query = RepOwners({**query_params, **permissions}) return [ format_row(data_point, array_fields=["label_ids"]) for data_point in query.execute() ]