from flask import jsonify, request from analytics import config from analytics.api import app from analytics.constants.access import ACCESS_ANALYTICS from analytics.features import ( is_insights_line_soundcloud_collection_as_active_enabled, is_insights_published_max_available_date_enabled, is_insights_transfer_product_ownership_enabled, ) from analytics.handler_utils import _get_query_string_params from analytics.logic.permissions import get_permission_values from analytics.logic.product_summary import get_product_summary from analytics.logic.product_timeseries import ( get_product_growth_periods_bulk, get_product_timeseries, ) from analytics.logic.streams import get_aggregated_streams from analytics.validation.access import verify_profile @app.route(config.PRODUCT_TIME_SERIES_PATH) @verify_profile(access=ACCESS_ANALYTICS) def product_timeseries(product_id): """Return timeseries for a product. Route: GET /product//timeseries Path params: product_id (str): Product ID. Query params: start_date (str): Start date YYYY-MM-DD. end_date (str): End date YYYY-MM-DD. type (str): Timeseries type. Values: PRODUCT_STREAMS, PRODUCT_STREAMS_BY_STORE, PRODUCT_STREAMS_BY_COUNTRY, PRODUCT_STREAMS_BY_TRACK, PRODUCT_STREAMS_BY_SOS, PRODUCT_STREAMS_BY_SOS_V2, PRODUCT_STREAMS_BY_SOS_DETAILED, PRODUCT_DOWNLOADS, PRODUCT_DOWNLOADS_BY_COUNTRY, PRODUCT_DOWNLOADS_BY_STORE, TRACK_DOWNLOADS, TRACK_DOWNLOADS_BY_COUNTRY, TRACK_DOWNLOADS_BY_STORE. resolution (str): Time resolution. date_period (str): Date period. multi_product (str): "true" or "false" (default "false"). countries (list[str]): Country codes to filter by. distributors (list[str]): Distributor names. store_ids (list[int]): Store IDs to filter by. Required for PRODUCT_STREAMS_BY_SOS_DETAILED. Supported: 1 (Apple), 187 (Amazon), 286 (Spotify), 453 (YouTube). stream_sources (list[str]): SOS stream-source names to filter by (e.g. "discovery", "albumpage"). Only used with PRODUCT_STREAMS_BY_SOS_DETAILED. Values are lowercased. ids (list[str]): IDs to filter aggregation by. Snowflake tables: - V_STREAMS_BY_PRODUCT_TRACK_FEED_DISTRIBUTOR_DAILY - V_STREAMS_BY_PRODUCT_TRACK_COUNTRY_FEED_DISTRIBUTOR_DAILY - DOWNLOADS_BY_PRODUCT_FEED_DISTRIBUTOR_DAILY - DOWNLOADS_BY_PRODUCT_COUNTRY_FEED_DISTRIBUTOR_DAILY - DOWNLOADS_BY_PRODUCT_TRACK_FEED_DISTRIBUTOR_DAILY - DOWNLOADS_BY_PRODUCT_TRACK_COUNTRY_FEED_DISTRIBUTOR_DAILY - MULTI_PRIMARY_PRODUCTS - PRIMARY_PRODUCTS JOINs: - DIM_RELEASE - DIM_TRACK_CLEAN_MV Returns: JSON with timeseries items. """ path_params = {"product_id": product_id} query_params = _get_query_string_params( [ "start_date", "end_date", "multi_product", "type", "resolution", "date_period", ], path_params, ) query_params = { **query_params, "countries": request.args.getlist("countries"), "distributors": request.args.getlist("distributors"), "store_ids": request.args.getlist("store_ids"), "multi_product": request.args.get( "multi_product", default=False, type=lambda v: v.lower() == "true" ), "ids": request.args.getlist("ids"), "stream_sources": [s.lower() for s in request.args.getlist("stream_sources")], "transfer_product_ownership_enabled": is_insights_transfer_product_ownership_enabled(), "line_soundcloud_collection_as_active_enabled": is_insights_line_soundcloud_collection_as_active_enabled(), } permissions = get_permission_values() try: timeseries = get_product_timeseries(query_params, permissions) except ValueError as e: return jsonify({"error": str(e)}), 400 response_body = {"items": timeseries} return jsonify(response_body), 200 @app.route(config.PRODUCT_GROWTH_PERIODS_BULK_PATH) def product_bulk_growth_periods(): """Return growth periods for multiple products. Route: GET /product/growth-periods-bulk Query params: product_id (list[str]): Product IDs. countries (list[str]): Country codes to filter by. distributors (list[str]): Distributor names. store_ids (list[int]): Store IDs to filter by. Snowflake tables: - METRICS_BY_PRODUCT_FEED_DISTRIBUTOR_ROLLUP - METRICS_BY_PRODUCT_COUNTRY_FEED_DISTRIBUTOR_ROLLUP JOINs: - DIM_FEED (store_ids subquery) Returns: JSON with growth period data per product. """ query_params = { "product_ids": request.args.getlist("product_id"), "countries": request.args.getlist("countries"), "distributors": request.args.getlist("distributors"), "store_ids": request.args.getlist("store_ids"), "transfer_product_ownership_enabled": is_insights_transfer_product_ownership_enabled(), } permissions = get_permission_values() growth_periods = get_product_growth_periods_bulk(query_params, permissions) return jsonify(growth_periods), 200 @app.route(config.PRODUCT_SUMMARY_PATH) @verify_profile(access=ACCESS_ANALYTICS) def product_summary(product_id): """Return summary for a product. Route: GET /product//summary Path params: product_id (str): Product ID. Query params: start_date (str): Start date YYYY-MM-DD. end_date (str): End date YYYY-MM-DD. type (str): Summary type. Values: TOTAL, COUNTRY, SOS, SOS_DETAILED, STORE, TRACK. resolution (str): Time resolution. multi_product (str): "true" or "false" (default "false"). countries (list[str]): Country codes to filter by. distributors (list[str]): Distributor names. store_ids (list[int]): Store IDs to filter by. Required for SOS_DETAILED. Supported: 1 (Apple), 187 (Amazon), 286 (Spotify), 453 (YouTube). stream_sources (list[str]): SOS stream-source names to filter by (e.g. "discovery", "albumpage"). Only used with SOS_DETAILED. Values are lowercased. ids (list[str]): IDs to filter aggregation by. limit (int): Max results. offset (int): Pagination offset. order_by (str): Sort field. order_dir (str): Sort direction. Snowflake tables: - V_STREAMS_BY_PRODUCT_TRACK_FEED_DISTRIBUTOR_DAILY - V_STREAMS_BY_PRODUCT_TRACK_COUNTRY_FEED_DISTRIBUTOR_DAILY - DOWNLOADS_BY_PRODUCT_FEED_DISTRIBUTOR_DAILY - DOWNLOADS_BY_PRODUCT_COUNTRY_FEED_DISTRIBUTOR_DAILY - DOWNLOADS_BY_PRODUCT_TRACK_FEED_DISTRIBUTOR_DAILY - DOWNLOADS_BY_PRODUCT_TRACK_COUNTRY_FEED_DISTRIBUTOR_DAILY - MULTI_PRIMARY_PRODUCTS - PRIMARY_PRODUCTS JOINs: - DIM_RELEASE Returns: JSON with summary items. """ path_params = {"product_id": product_id} query_params = _get_query_string_params( [ "start_date", "end_date", "multi_product", "type", "resolution", "limit", "offset", "order_by", "order_dir", ], path_params, ) query_params = { **query_params, "countries": request.args.getlist("countries"), "distributors": request.args.getlist("distributors"), "store_ids": request.args.getlist("store_ids"), "summary": True, "multi_product": request.args.get( "multi_product", default=False, type=lambda v: v.lower() == "true" ), "ids": request.args.getlist("ids"), "stream_sources": [s.lower() for s in request.args.getlist("stream_sources")], "transfer_product_ownership_enabled": is_insights_transfer_product_ownership_enabled(), "line_soundcloud_collection_as_active_enabled": is_insights_line_soundcloud_collection_as_active_enabled(), } permissions = get_permission_values() try: summary = get_product_summary(query_params, permissions) except ValueError as e: return jsonify({"error": str(e)}), 400 response_body = {"items": summary} return jsonify(response_body), 200 @app.route(config.PRODUCT_AGGREGATED_STREAMS_PATH) @verify_profile(access=ACCESS_ANALYTICS) def product_aggregated_streams(product_id): """Return aggregated streams for a product. Route: GET /product//aggregated-streams Path params: product_id (str): Product ID. Query params: days_back (int): Days to look back (default 28, max 28). top_size (int): Number of top items (default 5). dimension (str): Aggregation dimension (required). Values: SOS, STORE, COUNTRY. order_by (str): Sort field (default "streams7Days"). countries (list[str]): Country codes to filter by. store_ids (list[int]): Store IDs to filter by. Snowflake tables: dimension=COUNTRY or STORE: - METRICS_BY_PRODUCT_COUNTRY_FEED_DISTRIBUTOR_ROLLUP - V_STREAMS_BY_PRODUCT_TRACK_FEED_DISTRIBUTOR_DAILY - V_STREAMS_BY_PRODUCT_TRACK_COUNTRY_FEED_DISTRIBUTOR_DAILY - DATA_AVAILABILITY_BY_STORE_DISTRIBUTOR_DAILY dimension=SOS: - V_STREAMS_BY_PRODUCT_TRACK_FEED_DISTRIBUTOR_DAILY - V_STREAMS_BY_PRODUCT_TRACK_COUNTRY_FEED_DISTRIBUTOR_DAILY - DATA_AVAILABILITY_BY_STORE_DISTRIBUTOR_DAILY JOINs: - DATA_AVAILABILITY_GET_MAX_AVAILABLE_STREAMING_STORES_DATE (_PUBLISHED twin when insights_published_max_available_date is on) Returns: JSON with aggregated streams by dimension. """ if int(request.args.get("days_back", 28)) > 28: raise Exception( "days_back for aggregated streams cannot be greater than 28 days" ) path_params = {"product_id": product_id} query_params = { "days_back": abs(int(str(request.args.get("days_back", 28)))), "top_size": request.args.get("top_size", 5), "dimension": request.args.get("dimension").upper(), "order_by": request.args.get("order_by", "streams7Days"), "countries": request.args.getlist("countries"), "store_ids": list(map(int, request.args.getlist("store_ids"))), } query_params.update(path_params) query_params[ "transfer_product_ownership_enabled" ] = is_insights_transfer_product_ownership_enabled() query_params[ "line_soundcloud_collection_as_active_enabled" ] = is_insights_line_soundcloud_collection_as_active_enabled() query_params[ "published_max_available_date_enabled" ] = is_insights_published_max_available_date_enabled() permissions = get_permission_values() aggregated_streams = get_aggregated_streams(query_params, permissions) response_body = {**aggregated_streams} return jsonify(response_body), 200