"""Application Handlers. Requests are redirected to handlers, which are responsible for getting information from the URL and passing it down to the logic layer. The way each layer talks to each other is through Response objects which defines the type status of the data and the data itself. Please note: the Orchard uses the term handlers over views as convention for clarity See: oto.response for more details. """ import logging from itertools import groupby from flask import request from playlist.api import app from playlist.constants.handler_constants import ( CURATOR_COUNTRIES, FILTER_BY_TOP_10_APPLE_MUSIC_MARKETS, PAGINATION_QUERY_STRING_PARAMS, STORE_IDS, STOREFRONT_ENABLED, ) from playlist.constants.stores import APPLE_MUSIC from playlist.features import ( is_filter_incorrect_playlists_enabled, is_insights_playlist_page_apple_music_playlists_enabled, is_insights_playlist_page_hourly_playlists_enabled, is_insights_playlist_v2_current_past_enabled, is_insights_primary_playlist_type_enabled, is_insights_transfer_product_ownership_enabled, ) from playlist.queries.constants import HIGHWATERMARK_DATE from playlist.queries.fetch_queries import ( fetch_bulk_playlist_analytics, fetch_bulk_playlist_analytics_timeseries, fetch_bulk_playlist_metadata, fetch_bulk_playlist_placements_company_brands, fetch_placement, fetch_placement_metrics_by_country, fetch_placement_metrics_by_store_playlisttype, fetch_placement_position_time_series, fetch_placement_streams_time_series, fetch_placements, fetch_placements_by_global_participant_id, fetch_placements_by_isrc, fetch_placements_by_product, fetch_placements_by_store_playlist_id, fetch_placements_by_store_playlist_id_on_date, fetch_placements_count, fetch_playlist_available_dates, fetch_playlist_demographics, fetch_playlist_demographics_by_country, fetch_playlist_ids, fetch_recent_placements, fetch_total_vs_playlist_streams_time_series, get_date_range, ) from playlist.services.ows_permissions import get_permissions from playlist.utils.handler_utils import ( format_response, get_query_boolean_params, get_query_string_list_params, get_query_string_params, parallel, separate_playlists_by_storefront, ) logger = logging.getLogger(__name__) def _map_country_to_storefront(query_params: dict) -> dict: """Map 'country' parameter to 'storefront' for Apple Music playlists. For Apple Music (store_id = 1), the storefront is a required identifier to distinguish between different regional versions of the same playlist. This helper accepts both 'storefront' and 'country' parameters, mapping 'country' to 'storefront' when dealing with Apple Music. For other stores (e.g., Spotify), 'country' remains a separate filter parameter. Args: query_params: Query parameters dict Returns: Updated query_params with country mapped to storefront for Apple Music """ store_id = query_params.get("store_id") # For Apple Music, map country to storefront if storefront not already set if ( store_id == APPLE_MUSIC and "country" in query_params and query_params["country"] and "storefront" not in query_params ): query_params["storefront"] = query_params["country"] return query_params @app.route("/product//placements", methods=["GET"]) def get_placements_by_product(product_id): """Get product placements by product_id.""" permissions = get_permissions(request) path_params = {"product_id": product_id} query_string_keys = [ "sort_key", "sort_direction", "past_placements_only", "playlist_appearances", ] query_params = get_query_string_params(query_string_keys) query_params = get_query_string_list_params( { "streams_country": "stream_countries", "store_id": "store_ids", "playlist_type": "playlist_types", "curator_country": "curator_countries", "distributor": "distributors", }, query_params, ) # update query params with storefront_enabled to return non-grouped AM playlists as unique records query_params = get_query_boolean_params([STOREFRONT_ENABLED], query_params) pagination_params = get_query_string_params(PAGINATION_QUERY_STRING_PARAMS) params = { **path_params, **query_params, **pagination_params, "use_primary_playlist_type": (is_insights_primary_playlist_type_enabled()), FILTER_BY_TOP_10_APPLE_MUSIC_MARKETS: should_filter_by_top_10_apple_music_markets( query_params ), } placements, total_count = fetch_placements_by_product(params, permissions) return format_response({"placements": placements, "total_count": total_count}) @app.route("/global-participant//placements", methods=["GET"]) def get_placements_by_global_participant_id(global_participant_id): """Get global participant placements by id.""" permissions = get_permissions(request) path_params = {"global_participant_id": global_participant_id} query_params = get_query_string_params( [ "sort_key", "sort_direction", "past_placements_only", "playlist_appearances", ] ) query_params = get_query_string_list_params( { "streams_country": "stream_countries", "store_id": "store_ids", "playlist_type": "playlist_types", "curator_country": "curator_countries", "distributor": "distributors", }, query_params, ) # update query params with storefront_enabled to return non-grouped AM playlists as unique records query_params = get_query_boolean_params([STOREFRONT_ENABLED], query_params) pagination_params = get_query_string_params(PAGINATION_QUERY_STRING_PARAMS) params = { **path_params, **query_params, **pagination_params, "use_primary_playlist_type": (is_insights_primary_playlist_type_enabled()), FILTER_BY_TOP_10_APPLE_MUSIC_MARKETS: should_filter_by_top_10_apple_music_markets( query_params ), } placements, total_count = fetch_placements_by_global_participant_id( params, permissions ) return format_response({"placements": placements, "total_count": total_count}) @app.route("/placements_by_isrc", methods=["GET"]) def get_placements_by_isrc(): """Get sound recording placements by isrc.""" permissions = get_permissions(request) query_params = get_query_string_params( [ "isrc", "sort_key", "sort_direction", "past_placements_only", "min_followers", "playlist_appearances", ] ) query_params = get_query_string_list_params( { "streams_country": "stream_countries", "store_id": "store_ids", "playlist_type": "playlist_types", "curator_country": "curator_countries", "distributor": "distributors", }, query_params, ) # update query params with storefront_enabled to return non-grouped AM playlists as unique records query_params = get_query_boolean_params([STOREFRONT_ENABLED], query_params) pagination_params = get_query_string_params(PAGINATION_QUERY_STRING_PARAMS) params = { **query_params, **pagination_params, "use_primary_playlist_type": (is_insights_primary_playlist_type_enabled()), "insights_playlist_v2_current_past_enabled": ( is_insights_playlist_v2_current_past_enabled() ), FILTER_BY_TOP_10_APPLE_MUSIC_MARKETS: should_filter_by_top_10_apple_music_markets( query_params ), } placements, total_count = fetch_placements_by_isrc(params, permissions) return format_response({"placements": placements, "total_count": total_count}) @app.route("/playlist//placements", methods=["GET", "POST"]) def get_placements_by_store_playlist_id(store_playlist_id): """Get all track placements for a specific playlist (tracklist). Args: store_playlist_id: The playlist identifier specific to the store Query params: store_id: Required - The streaming service ID (e.g., 1 for Apple Music, 286 for Spotify) storefront: Optional - For Apple Music, the country storefront code sort_key: Optional - Field to sort by sort_direction: Optional - ASC or DESC limit: Optional - Number of results per page offset: Optional - Pagination offset include_hourly: Optional - Boolean flag to include hourly playlists. If not provided, defaults to None and the feature flag is checked internally. Other standard placement filters Returns: JSON response with placements list and total_count """ permissions = get_permissions(request) path_params = {"store_playlist_id": store_playlist_id} query_string_keys = [ "store_id", "storefront", "country", "sort_key", "sort_direction", "past_placements_only", "min_followers", "playlist_appearances", ] query_params = get_query_string_params(query_string_keys) query_params = _map_country_to_storefront(query_params) # Parse include_hourly as a boolean query parameter boolean_params = get_query_boolean_params(["include_hourly"], query_params) include_hourly = boolean_params.get("include_hourly") # For Apple Music playlists: gate access behind feature flags # NOTE: This is a simple access gate while Apple Music support is being developed. # The handling logic for Apple Music storefronts exists but is not yet fully validated # for production use. When both flags are disabled, return empty results without explanation. if query_params.get("store_id") == "1": hourly_flag = ( include_hourly if include_hourly is not None else is_insights_playlist_page_hourly_playlists_enabled() ) if not ( hourly_flag and is_insights_playlist_page_apple_music_playlists_enabled() ): return format_response( { "placements": [], "placeholder_placements": [], "total_count": 0, } ) # For POST requests, check if filters are provided in JSON body has_json_filters = False if request.method == "POST": data = request.get_json() if data and any( data.get(k) for k in [ "streams_country", "playlist_type", "curator_country", "distributor", ] ): # Extract filters from JSON body if provided if data.get("streams_country"): query_params["stream_countries"] = data.get("streams_country") if data.get("playlist_type"): query_params["playlist_types"] = data.get("playlist_type") if data.get("curator_country"): query_params["curator_countries"] = data.get("curator_country") if data.get("distributor"): query_params["distributors"] = data.get("distributor") has_json_filters = True # Always parse multi-valued query params from query string (unless POST provided them in JSON) if not has_json_filters: query_params = get_query_string_list_params( { "streams_country": "stream_countries", "playlist_type": "playlist_types", "curator_country": "curator_countries", "distributor": "distributors", }, query_params, ) # update query params with storefront_enabled to return non-grouped AM playlists as unique records query_params = get_query_boolean_params( [STOREFRONT_ENABLED, "insights_playlist_page_hide_compilation_art"], query_params, ) # For Apple Music: ensure curator_countries is set to preserve position data. # NOTE: Apple Music playlists have per-storefront variants (28 countries available). # Each storefront variant has different tracks/positions and only contains streams from that storefront. # We auto-populate curator_countries from storefront or stream_countries to ensure the SQL # uses per-country aggregation tables (which preserve track-specific positions). # If no country filters provided, the query falls back to top 10 markets filter. # This logic is still being refined and validated for production use. if query_params.get("store_id") == "1": # Only auto-populate if curator_countries was not provided at all # (not in query_params), to avoid overriding user-provided empty lists if "curator_countries" not in query_params: # Derive curator_countries from storefront or stream_countries if query_params.get("storefront"): query_params["curator_countries"] = [query_params["storefront"]] elif query_params.get("stream_countries"): query_params["curator_countries"] = query_params["stream_countries"] pagination_params = get_query_string_params(PAGINATION_QUERY_STRING_PARAMS) # Default playlist_appearances to 'current' when feature flag is enabled # to ensure consistent behavior when not explicitly specified if is_insights_playlist_v2_current_past_enabled(): if ( "playlist_appearances" not in query_params or query_params["playlist_appearances"] is None ): query_params["playlist_appearances"] = "current" params = { **path_params, **query_params, **pagination_params, "use_primary_playlist_type": (is_insights_primary_playlist_type_enabled()), "insights_playlist_v2_current_past_enabled": ( is_insights_playlist_v2_current_past_enabled() ), "insights_playlist_page_hourly_playlists_enabled": ( include_hourly if include_hourly is not None else is_insights_playlist_page_hourly_playlists_enabled() ), FILTER_BY_TOP_10_APPLE_MUSIC_MARKETS: should_filter_by_top_10_apple_music_markets( query_params ), } logger.info( f"[COMPILATION_DEBUG] Params for playlist tracklist: " f"store_playlist_id={params.get('store_playlist_id')}, " f"insights_playlist_page_hide_compilation_art={params.get('insights_playlist_page_hide_compilation_art')}, " f"all_params_keys={list(params.keys())}" ) placements, placeholder_placements, total_count = ( fetch_placements_by_store_playlist_id(params, permissions) ) return format_response( { "placements": placements, "placeholder_placements": placeholder_placements, "total_count": total_count, } ) @app.route("/playlist//dates/", methods=["GET"]) def get_playlist_available_dates(store_playlist_id): """Get all available dates for a playlist's tracklist history. Args: store_playlist_id: The playlist identifier from the streaming service Query params: store_id: Required - The streaming service ID (e.g., 1 for Apple Music, 286 for Spotify) storefront: Optional - For Apple Music, the country storefront code Returns: JSON response with list of date strings in YYYY-MM-DD format """ path_params = {"store_playlist_id": store_playlist_id} query_params = get_query_string_params(["store_id", "storefront", "country"]) query_params = _map_country_to_storefront(query_params) # Auto-enable storefront_enabled for Apple Music when storefront is provided # This ensures the query uses the market-specific historical table if query_params.get("store_id") == "1" and query_params.get("storefront"): query_params[STOREFRONT_ENABLED] = True params = {**path_params, **query_params} # Validate supported stores - only Chartmetric-enabled stores supported_stores = { "1", "286", "453", "348", "187", } # Apple Music, Spotify, YouTube, Deezer, Amazon store_id = params.get("store_id") if store_id not in supported_stores: return format_response( error=f"Unsupported store_id '{store_id}'. Only stores {', '.join(sorted(supported_stores))} are supported for historical tracklist dates.", status_code=400, ) dates = fetch_playlist_available_dates(params) return format_response({"dates": dates}) @app.route("/playlist//placements/on-date", methods=["GET"]) def get_placements_by_store_playlist_id_on_date(store_playlist_id): """Get historical tracklist for a specific playlist at a given date. Reconstructs a playlist's tracklist as it existed at a specific point in time by querying the placement_position_events table. Args: store_playlist_id: The playlist identifier from the streaming service Query params: store_id: Required - The streaming service ID (e.g., 1 for Apple Music, 286 for Spotify) target_date: Required - The date to retrieve the tracklist for (YYYY-MM-DD format) storefront: Optional - For Apple Music, the country storefront code limit: Optional - Number of results per page offset: Optional - Pagination offset Returns: JSON response with placements list and total_count """ permissions = get_permissions(request) path_params = {"store_playlist_id": store_playlist_id} query_string_keys = [ "store_id", "storefront", "country", "target_date", ] query_params = get_query_string_params(query_string_keys) query_params = _map_country_to_storefront(query_params) query_params = get_query_string_list_params( { "streams_country": "stream_countries", "curator_country": "curator_countries", }, query_params, ) # update query params with storefront_enabled to return non-grouped AM playlists as unique records query_params = get_query_boolean_params( [STOREFRONT_ENABLED, "insights_playlist_page_hide_compilation_art"], query_params, ) # Auto-enable storefront_enabled for Apple Music when storefront is provided # This ensures the query uses the market-specific historical table if query_params.get("store_id") == "1" and query_params.get("storefront"): query_params[STOREFRONT_ENABLED] = True pagination_params = get_query_string_params(PAGINATION_QUERY_STRING_PARAMS) params = { **path_params, **query_params, **pagination_params, "use_primary_playlist_type": (is_insights_primary_playlist_type_enabled()), } # Validate that the target_date exists in available dates before running expensive query target_date = query_params.get("target_date") if target_date: # Fetch available dates for this playlist date_params = { "store_playlist_id": store_playlist_id, "store_id": query_params.get("store_id"), "storefront_enabled": query_params.get(STOREFRONT_ENABLED, False), } # Only include storefront if it's provided if query_params.get("storefront"): date_params["storefront"] = query_params.get("storefront") # Check for force_refresh parameter to bypass cache force_refresh = request.args.get("force_refresh", "false").lower() == "true" placements, placeholder_placements, total_count = ( fetch_placements_by_store_playlist_id_on_date( params, permissions, force_refresh=force_refresh ) ) return format_response( { "placements": placements, "placeholder_placements": placeholder_placements, "total_count": total_count, } ) @app.route("/placements", methods=["GET"]) def get_placements(): permissions = get_permissions(request) query_params = get_query_string_params( [ "isrc", "global_participant_id", "product_id", "sort_key", "sort_direction", "past_placements_only", "min_followers", "playlist_appearances", ] ) query_params = get_query_string_list_params( { "streams_country": "stream_countries", "store_id": "store_ids", "playlist_type": "playlist_types", "curator_country": "curator_countries", "distributor": "distributors", }, query_params, ) # update query params with storefront_enabled to return non-grouped AM playlists as unique records query_params = get_query_boolean_params([STOREFRONT_ENABLED], query_params) pagination_params = get_query_string_params(PAGINATION_QUERY_STRING_PARAMS) params = { **query_params, **pagination_params, "filter_incorrect_playlists": is_filter_incorrect_playlists_enabled(), "insights_playlist_v2_current_past_enabled": ( is_insights_playlist_v2_current_past_enabled() ), "use_primary_playlist_type": (is_insights_primary_playlist_type_enabled()), FILTER_BY_TOP_10_APPLE_MUSIC_MARKETS: should_filter_by_top_10_apple_music_markets( query_params ), } placements, total_count = fetch_placements(params, permissions) return format_response({"placements": placements, "total_count": total_count}) @app.route("/placements_count", methods=["GET"]) def get_placements_count(): permissions = get_permissions(request) query_params = get_query_string_params( [ "isrc", "global_participant_id", "product_id", "min_followers", "playlist_appearances", ] ) query_params = get_query_string_list_params( { "streams_country": "stream_countries", "store_id": "store_ids", "playlist_type": "playlist_types", "curator_country": "curator_countries", "distributor": "distributors", }, query_params, ) # FF checks query_params["filter_incorrect_playlists"] = is_filter_incorrect_playlists_enabled() query_params["use_primary_playlist_type"] = ( is_insights_primary_playlist_type_enabled() ) total_count = fetch_placements_count(query_params, permissions) return format_response({"total_count": total_count}) @app.route("/playlist//placement/", methods=["GET"]) def get_placement(store_playlist_id, isrc): path_params = {"store_playlist_id": store_playlist_id, "isrc": isrc} permissions = get_permissions(request) query_params = get_query_string_params( ["store_id", "storefront", "country"], path_params ) query_params = _map_country_to_storefront(query_params) query_params = get_query_string_list_params( { "streams_country": "stream_countries", "distributor": "distributors", }, query_params, ) placement = fetch_placement(query_params, permissions) return format_response({"placement": placement}) @app.route("/placements/recent", methods=["GET"]) def get_recent_placements(): permissions = get_permissions(request) query_params = get_query_string_params( ["min_followers", "sort_key", "sort_direction"] ) query_params = get_query_string_list_params( { "distributor": "distributors", "label_id": "filter_label_ids", "subaccount_id": "filter_subaccount_ids", "global_participant_id": "global_participant_ids", "store_id": STORE_IDS, "playlist_type": "playlist_types", "curator_country": CURATOR_COUNTRIES, }, query_params, ) # update query params with storefront_enabled to return non-grouped AM playlists as unique records query_params = get_query_boolean_params([STOREFRONT_ENABLED], query_params) # FFs checks query_params["use_primary_playlist_type"] = ( is_insights_primary_playlist_type_enabled() ) pagination_params = get_query_string_params(PAGINATION_QUERY_STRING_PARAMS) params = { **query_params, **pagination_params, FILTER_BY_TOP_10_APPLE_MUSIC_MARKETS: should_filter_by_top_10_apple_music_markets( query_params ), } placements, total_count = fetch_recent_placements(params, permissions) return format_response({"placements": placements, "total_count": total_count}) @app.route("/placements/breakdown", methods=["GET"]) def get_placement_metrics_by_store_playlisttype(): permissions = get_permissions(request) query_params = get_query_string_params(["isrc", "global_participant_id"]) query_params = get_query_string_list_params( {"distributor": "distributors"}, query_params ) placements_breakdown = fetch_placement_metrics_by_store_playlisttype( query_params, permissions ) return format_response(placements_breakdown) def _get_date_params(): start_date = request.args.get("start_date", HIGHWATERMARK_DATE) days = int(request.args.get("days", "-7")) return {"start_date": start_date, "days": days} @app.route("/playlist//placement//positions", methods=["GET"]) def get_placement_position_time_series(store_playlist_id, isrc): path_params = {"store_playlist_id": store_playlist_id, "isrc": isrc} date_params = _get_date_params() date_params = get_date_range("positions", **date_params) query_params = get_query_string_params(["store_id", "storefront", "country"]) query_params = _map_country_to_storefront(query_params) params = {**path_params, **date_params, **query_params} position_time_series = fetch_placement_position_time_series(params) return format_response({"positions": position_time_series}) @app.route("/playlist//placement//streams", methods=["GET"]) def get_placement_streams_time_series(store_playlist_id, isrc): path_params = {"store_playlist_id": store_playlist_id, "isrc": isrc} permissions = get_permissions(request) date_params = _get_date_params() date_params = get_date_range("streams", **date_params) query_params = get_query_string_params(["store_id", "storefront", "country"]) query_params = _map_country_to_storefront(query_params) query_params = get_query_string_list_params( {"streams_country": "stream_countries"}, query_params, ) params = { **path_params, **query_params, **date_params, "transfer_product_ownership_enabled": ( is_insights_transfer_product_ownership_enabled() ), } streams_time_series = fetch_placement_streams_time_series(params, permissions) data = { "streams": streams_time_series, } return format_response(data) @app.route("/placements/total_vs_playlist_streams_by_store", methods=["GET"]) def get_total_vs_playlist_streams_by_store(): permissions = get_permissions(request) path_params = get_query_string_params(["isrc", "global_participant_id"]) date_params = _get_date_params() query_params = get_query_string_list_params( {"streams_country": "stream_countries", "store_id": "store_ids"}, get_date_range("streams", **date_params), ) params = { **query_params, **path_params, "transfer_product_ownership_enabled": ( is_insights_transfer_product_ownership_enabled() ), } fetch_result = fetch_total_vs_playlist_streams_time_series(params, permissions) if not fetch_result: return format_response({"all_stores_aggregation": {}, "stores": []}) ( aggregate_result, total_vs_playlist_streams_by_store, ) = fetch_result return format_response( { "all_stores_aggregation": aggregate_result, "stores": total_vs_playlist_streams_by_store, } ) @app.route( "/playlist//placement//breakdown-by-country", methods=["GET"], ) def get_placement_metrics_by_country(store_playlist_id, isrc): path_params = { "store_playlist_id": store_playlist_id, "isrc": isrc, } permissions = get_permissions(request) query_params = get_query_string_params( [ "playlist_appearances", "sort_key", "sort_direction", "store_id", ] ) query_params = get_query_string_list_params( {"streams_country": "stream_countries", "distributor": "distributors"}, query_params, ) params = { **path_params, **query_params, "insights_playlist_v2_current_past_enabled": ( is_insights_playlist_v2_current_past_enabled() ), } country_breakdown_result = fetch_placement_metrics_by_country(params, permissions) return format_response({"countries": country_breakdown_result}) def _drop_stream_countries_key(p): playlist = p.copy() playlist.pop("stream_countries", None) return playlist def _group_timeseries_by_dimension(rows): """Group flattened timeseries rows by playlist, reconstructing the dimensions array. The SQL returns one row per (playlist, dimension_value) to avoid Snowflake's 16MB limit. This function groups them back into the nested structure expected by the API. """ from collections import defaultdict grouped = defaultdict( lambda: { "dimensions": [], "row_number": None, "store_id": None, "store_playlist_id": None, "storefront": None, } ) for row in rows: key = ( row["row_number"], row["store_id"], row["store_playlist_id"], row.get("storefront"), ) if grouped[key]["row_number"] is None: grouped[key]["row_number"] = row["row_number"] grouped[key]["store_id"] = row["store_id"] grouped[key]["store_playlist_id"] = row["store_playlist_id"] grouped[key]["storefront"] = row.get("storefront") if row.get("dimension_value") is not None: grouped[key]["dimensions"].append( { "dimension_value": row["dimension_value"], "streams_array": row.get("streams_array", []), "listeners_array": row.get("listeners_array", []), } ) return list(grouped.values()) def _fetch_analytics_for_stream_countries( stream_countries_and_playlists, by_dimension=None, order_by="STREAMS_28_DAYS", order_dir="DESC", ): stream_countries = stream_countries_and_playlists[0] playlists_to_request = stream_countries_and_playlists[1] playlists_to_request_without_stream_countries = list( map(_drop_stream_countries_key, playlists_to_request) ) additional_query_params = { "stream_countries": stream_countries, "transfer_product_ownership_enabled": ( is_insights_transfer_product_ownership_enabled() ), } if by_dimension is not None: additional_query_params["by_dimension"] = by_dimension additional_query_params["order_by"] = order_by additional_query_params["order_dir"] = order_dir return _fetch_playlist_data_separated_by_storefront( playlists_to_request_without_stream_countries, fetch_bulk_playlist_analytics, additional_query_params, ) @app.route("/playlist/analytics-bulk-timeseries", methods=["POST"]) def get_playlists_analytics_timeseries(): """Return timeseries streams analytics data for list of playlist ids. Body example: { "playlists": [ { "store_playlist_id": "37i9dQZF1DX2apWzyECwyZ", "store_id": "286", "storefront": "AD" } ], "stream_countries: ["US"] "start_date": "2025-01-01", "end_date": "2025-12-31" "by_dimension": "MARKET" # Optional } Returns: flask.Response with playlist analytics timeseries """ data = request.get_json() playlists = data.get("playlists", []) stream_countries = data.get("stream_countries", []) start_date = data.get("start_date", None) end_date = data.get("end_date", None) by_dimension = data.get("by_dimension", None) order_by = data.get("order_by", "activity_date") order_dir = data.get("order_dir", "ASC") for i, playlist in enumerate(playlists): playlist["row_number"] = i if len(playlists) == 0: return format_response({}, 400, "Missing playlists") additional_query_params = { "start_date": start_date, "end_date": end_date, "stream_countries": stream_countries, "by_dimension": by_dimension, "order_by": order_by, "order_dir": order_dir, } playlist_analytics_by_streams_timeseries = ( _fetch_playlist_data_separated_by_storefront( playlists, fetch_bulk_playlist_analytics_timeseries, additional_query_params, ) ) # When by_dimension is MARKET, the SQL returns one row per (playlist, dimension_value) # We need to group these back into a nested structure if by_dimension == "MARKET" and playlist_analytics_by_streams_timeseries: playlist_analytics_by_streams_timeseries = _group_timeseries_by_dimension( playlist_analytics_by_streams_timeseries ) playlist_analytics_ordered = sorted( playlist_analytics_by_streams_timeseries or [], key=lambda pa: pa["row_number"] ) for pa in playlist_analytics_ordered: pa.pop("row_number", None) return format_response({"playlists": playlist_analytics_ordered}) @app.route("/playlist/analytics-bulk", methods=["POST"]) def get_playlists_analytics(): """Return analytics data for list of playlist ids. Body example: { "playlists": [ { "store_playlist_id": "37i9dQZF1DX2apWzyECwyZ", "store_id": "286", "storefront": "AD", "stream_countries: ["US"] } ], "by_dimension": "MARKET" # Optional } Returns: flask.Response with playlist analytics """ data = request.get_json() playlists = data.get("playlists", []) by_dimension = data.get("by_dimension", None) order_by = data.get("order_by", "STREAMS_28_DAYS") order_dir = data.get("order_dir", "DESC") for i, playlist in enumerate(playlists): playlist["row_number"] = i if len(playlists) == 0: return format_response({}, 400, "Missing playlists") playlists_by_stream_countries = groupby( playlists, lambda p: p.get("stream_countries") ) playlist_analytics_by_streams_country = ( _fetch_analytics_for_stream_countries( item, by_dimension=by_dimension, order_by=order_by, order_dir=order_dir ) for item in playlists_by_stream_countries ) playlist_analytics = [] for pa in playlist_analytics_by_streams_country: playlist_analytics.extend(pa) playlist_analytics_ordered = sorted( playlist_analytics, key=lambda pa: pa["row_number"] ) for pa in playlist_analytics_ordered: pa.pop("row_number", None) return format_response({"playlists": playlist_analytics_ordered}) @app.route("/playlist-metadata-bulk", methods=["POST"]) def get_playlist_metadata_bulk(): data = request.get_json() playlists = data.get("playlists", []) insights_spotify_playlist_apollo_metadata_overrides_enabled = data.get( "insights_spotify_playlist_apollo_metadata_overrides_enabled", False ) use_primary_playlist_type = is_insights_primary_playlist_type_enabled() if len(playlists) == 0: return format_response({}, 400, "Missing playlists") metadata_playlists = _fetch_playlist_data_separated_by_storefront( playlists, fetch_bulk_playlist_metadata, { "use_primary_playlist_type": use_primary_playlist_type, "insights_spotify_playlist_apollo_metadata_overrides_enabled": ( insights_spotify_playlist_apollo_metadata_overrides_enabled ), }, ) return format_response({"playlists": metadata_playlists}) @app.route("/playlist-ids", methods=["GET"]) def get_playlist_ids(): """Get all distinct playlist IDs from the priority playlists table. Returns a list of all available playlist IDs with their store IDs. Optionally includes hourly playlist IDs and non-priority Spotify playlist IDs based on feature flags or query parameters. This endpoint is optimized for fast loading and caching by frontend applications. Query params: force_refresh: Optional - Boolean flag to bypass Redis cache and fetch fresh data. Accepts: true/false, 1/0, yes/no, y/n, on/off (case-insensitive) include_hourly: Optional - Boolean flag to include hourly playlists. If not provided, defaults to None and the feature flag is checked internally. include_non_priority: Optional - Boolean flag to include non-priority playlists. Non-priority playlists are Spotify playlists (store_id=286) from v_playlists_by_playlist that are not in priority_playlists or hourly_playlists. If not provided, defaults to None and the feature flag is checked internally. store_ids: Optional - Comma-separated list of store IDs to filter by (e.g., "1,286"). If provided, only playlists from these stores will be returned. Returns: JSON response with list of playlist objects containing: - store_playlist_id: The playlist identifier - store_id: The store/platform identifier (integer) """ query_params = get_query_boolean_params( ["force_refresh", "include_hourly", "include_non_priority"] ) force_refresh = query_params.get("force_refresh", False) include_hourly = query_params.get("include_hourly") include_non_priority = query_params.get("include_non_priority") # Parse store_ids as comma-separated integers store_ids_param = request.args.get("store_ids") store_ids = None if store_ids_param: try: store_ids = [int(sid.strip()) for sid in store_ids_param.split(",")] except ValueError: return format_response( {}, 400, "Invalid store_ids format. Must be comma-separated integers." ) # For include_apple_music, still check the feature flags # because it requires BOTH hourly and apple music flags to be enabled include_apple_music = ( include_hourly if include_hourly is not None else is_insights_playlist_page_hourly_playlists_enabled() ) and is_insights_playlist_page_apple_music_playlists_enabled() playlist_ids = fetch_playlist_ids( force_refresh=force_refresh, include_hourly=include_hourly, include_non_priority=include_non_priority, include_apple_music=include_apple_music, store_ids=store_ids, ) return format_response({"playlists": playlist_ids}) @app.route("/playlist/placement/company-brand", methods=["POST"]) def get_company_brand_for_playlist_placement(): """Return the company_brand for list of playlist placements. Currently this is for employee only, we will expand to other non-employees later. Body example: { "playlist_placements": [ { "store_playlist_id": "37i9dQZF1DX2apWzyECwyZ", "store_id": "286", "storefront": "AD", "isrc": "USUM72005928" } ] } Returns: flask.Response with playlist placement company_brand data """ data = request.get_json() placements = data.get("playlist_placements", []) for i, placement in enumerate(placements): placement["row_number"] = i if len(placements) == 0: return format_response({}, 400, "Missing placements") playlist_placements_company_brands = _fetch_playlist_data_separated_by_storefront( placements, fetch_bulk_playlist_placements_company_brands, {} ) playlist_placements_ordered = sorted( playlist_placements_company_brands, key=lambda pa: pa["row_number"] ) for pp in playlist_placements_ordered: pp.pop("row_number", None) return format_response({"playlist_placements": playlist_placements_ordered}) @app.route("/playlist//demographics/", methods=["GET"]) def get_playlist_demographics(store_playlist_id): """Get demographics data for a playlist. Args: store_playlist_id: The playlist identifier from the streaming service Query params: store_id: Required - The streaming service ID start_date: Optional - Start date for the date range (YYYY-MM-DD format) end_date: Optional - End date for the date range (YYYY-MM-DD format) storefront: Optional - For Apple Music, the country storefront code by_country: Optional - Boolean flag to return demographics broken down by country Returns: JSON response with demographics data """ from datetime import datetime, timedelta from playlist.queries.fetch_queries import get_max_available_streaming_date path_params = {"store_playlist_id": store_playlist_id} query_params = get_query_string_params( ["store_id", "start_date", "end_date", "storefront", "country"] ) streams_countries_param = get_query_string_params(["stream_countries"]).get( "stream_countries" ) query_params = _map_country_to_storefront(query_params) query_params = get_query_boolean_params(["by_country"], query_params) query_params["stream_countries"] = ( streams_countries_param.split(",") if streams_countries_param else [] ) params = {**path_params, **query_params} # Validate required store_id if not params.get("store_id"): return format_response( error="store_id is required", status_code=400, ) # Default to last 28 days if no date range provided if not params.get("start_date") and not params.get("end_date"): max_date_str = get_max_available_streaming_date() max_date = datetime.strptime(max_date_str, "%Y-%m-%d").date() params["end_date"] = max_date_str params["start_date"] = (max_date - timedelta(days=27)).strftime("%Y-%m-%d") by_country = params.pop("by_country", False) if by_country: demographics = fetch_playlist_demographics_by_country(params) else: demographics = fetch_playlist_demographics(params) return format_response({"demographics": demographics}) def _fetch_playlist_data_separated_by_storefront(playlists, fetch_func, query_params): ( playlists_with_storefront, playlists_without_storefront, ) = separate_playlists_by_storefront(playlists) query_params_with_storefront = { "playlists": playlists_with_storefront, "storefront_defined": True, **query_params, } query_params_without_storefront = { "playlists": playlists_without_storefront, "storefront_defined": False, **query_params, } permissions = get_permissions(request) playlist_data = [] if len(playlists_without_storefront) > 0 and len(playlists_with_storefront) > 0: requests = { "without_storefront": { "func": fetch_func, "args": (query_params_without_storefront, permissions), }, "with_storefront": { "func": fetch_func, "args": (query_params_with_storefront, permissions), }, } result = parallel(requests) playlists_storefront_not_provided = result.message["without_storefront"] playlists_storefront_provided = result.message["with_storefront"] playlist_data = ( playlists_storefront_provided + playlists_storefront_not_provided ) elif len(playlists_without_storefront) > 0: playlist_data = fetch_func(query_params_without_storefront, permissions) elif len(playlists_with_storefront) > 0: playlist_data = fetch_func(query_params_with_storefront, permissions) return playlist_data def should_filter_by_top_10_apple_music_markets(query_params) -> bool: """ Check if results should be filtered by top 10 Apple Music markets. Returns True if storefront_enabled is True, Apply Music id is in store_ids, curator_countries is an empty list, otherwise False. :param query_params: list of query parameters :return: boolean value """ return ( query_params.get(STOREFRONT_ENABLED, False) and APPLE_MUSIC in query_params.get(STORE_IDS, []) and not query_params.get(CURATOR_COUNTRIES, []) )