"""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 define 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. """ from flask import g, jsonify, request from oto import response from oto.adaptors.flask import flaskify from analytics import config from analytics.api import app from analytics.constants.access import ACCESS_ANALYTICS from analytics.constants.account import ACCOUNT_TO_FIN_LABEL_MAPPING from analytics.constants.distributors import DISTRIBUTORS, DISTRIBUTORS_ORCH_AWAL from analytics.constants.pagination import DEFAULT_OFFSET from analytics.constants.parameters import ALL_TIME 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_csv_list, _get_global_filters, _parse_int_list, filter_non_empty_items, user_has_full_access, ) from analytics.logic import ( aggregate_streams, channel_metrics, channel_top_videos, channel_traffic_sources, data_availability, demographics, feed, highwatermark, market_ranks, product, product_metadata, product_metrics, sound_recording_metadata, stores, streams, streams_breakdown, streams_bulk, streams_by_store, tadas, top_accounts_metrics, top_channel_metrics, top_countries_channels, top_countries_videos, top_markets, top_metrics, top_sound_recording_families, top_sound_recordings, top_video_metrics, top_video_traffic_sources, ugc_video_metrics, video_metrics, ) from analytics.logic.permissions import get_permission_values from analytics.validation import access @app.route(config.HEALTH_CHECK_PATH) def health(): """Check the health of the application. Route: GET /hello/ Snowflake tables: None. Returns: JSON with status indicator. """ return jsonify({"status": "ok"}) @app.errorhandler(500) def exception_handler(error): """Handle error when uncaught exception is raised. Default exception handler. Note: Exception will also be sent to Sentry if config.SENTRY is set. Returns: flask.Response: A 500 response with JSON 'code' & 'message' payload. """ message = ( "The server encountered an internal error " "and was unable to complete your request." ) g.log.exception(error) return flaskify(response.create_fatal_response(message)) @app.route(config.FEEDS_PATH) @access.verify_profile(access=ACCESS_ANALYTICS) def get_feeds(): """Return feeds available to the user, and their details. Route: GET /feeds Snowflake tables: - DATA_AVAILABILITY_BY_STORE_DAILY JOINs: - DATA_AVAILABILITY_SKIPS_SAVES_BY_FEED_SUMMARY Returns: JSON with feeds and their details. """ return flaskify(feed.get_feeds()) @app.route(config.FEEDS_V2_PATH) @access.verify_profile(access=ACCESS_ANALYTICS) def get_feeds_v2(): """Return feeds with outages and watermarks. Route: GET /feeds/v2 Snowflake tables: - DATA_AVAILABILITY_BY_STORE_DISTRIBUTOR_DAILY JOINs: - DIM_FEED Returns: JSON with feeds, outages, and watermarks. """ return flaskify(feed.get_feeds_v2()) @app.route(config.STREAMS_BREAKDOWN_PATH) @access.verify_profile(access=ACCESS_ANALYTICS) def get_streams_breakdown(isrc): """Return source-of-streams breakdown for a sound recording. Route: GET /sound-recording//streams-breakdown Path params: isrc (str): ISRC identifier. Query params: country_code (list[str]): Country codes to filter by. store_ids (list[int]): Store IDs to filter by. start_date (str): Start date YYYY-MM-DD. end_date (str): End date YYYY-MM-DD. distributors (str): Comma-separated distributor names. Snowflake tables: - V_STREAMS_BY_TRACK_COUNTRY_FEED_DISTRIBUTOR_DAILY JOINs: - DIM_RELEASE - LABEL_PARTICIPANT_PARTICIPATED_IN_ORCHARD_PRODUCT (permissions subquery) Returns: JSON with source-of-streams breakdown. """ countries, store_ids, start_date, end_date, distributors = _get_global_filters() query_params = { "isrc": isrc, "country_ids": countries, "store_ids": store_ids, "start_date": start_date, "end_date": end_date, "distributors": distributors, "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() return flaskify(streams_breakdown.get_streams_breakdown(query_params, permissions)) @app.route(config.STREAMS_BY_STORE_PATH) @access.verify_profile(access=ACCESS_ANALYTICS) def get_streams_by_store(isrc): """Return sound-recording streams by store for a given isrc. Route: GET /sound-recording//streams-by-store Path params: isrc (str): ISRC identifier. Query params: country_code (list[str]): Country codes to filter by. store_ids (list[int]): Store IDs to filter by. start_date (str): Start date YYYY-MM-DD. end_date (str): End date YYYY-MM-DD. distributors (str): Comma-separated distributor names. Snowflake tables: - V_STREAMS_BY_TRACK_FEED_DISTRIBUTOR_DAILY - V_STREAMS_BY_TRACK_COUNTRY_FEED_DISTRIBUTOR_DAILY JOINs: - DIM_RELEASE - LABEL_PARTICIPANT_PARTICIPATED_IN_ORCHARD_PRODUCT (permissions subquery) Returns: JSON with streams by store timeseries. """ countries, store_ids, start_date, end_date, distributors = _get_global_filters() query_params = { "isrc": isrc, "country_ids": countries, "store_ids": store_ids, "start_date": start_date, "end_date": end_date, "distributors": distributors, "transfer_product_ownership_enabled": is_insights_transfer_product_ownership_enabled(), } permissions = get_permission_values() return flaskify(streams_by_store.get_streams_by_store(query_params, permissions)) @app.route(config.STREAMS_ALL_PATH) @access.verify_profile(access=ACCESS_ANALYTICS) def get_streams_all(isrc): """Return sound-recording streams timeseries for a given isrc. Route: GET /sound-recording//streams-all Path params: isrc (str): ISRC identifier. Query params: country_code (list[str]): Country codes to filter by. store_ids (list[int]): Store IDs to filter by. start_date (str): Start date YYYY-MM-DD. end_date (str): End date YYYY-MM-DD. distributors (str): Comma-separated distributor names. Snowflake tables: - V_STREAMS_BY_TRACK_FEED_DISTRIBUTOR_DAILY - V_STREAMS_BY_TRACK_COUNTRY_FEED_DISTRIBUTOR_DAILY JOINs: - DIM_RELEASE - LABEL_PARTICIPANT_PARTICIPATED_IN_ORCHARD_PRODUCT (permissions subquery) Returns: JSON with streams timeseries. """ countries, store_ids, start_date, end_date, distributors = _get_global_filters() query_params = { "isrc": isrc, "country_ids": countries, "store_ids": store_ids, "start_date": start_date, "end_date": end_date, "distributors": distributors, "transfer_product_ownership_enabled": is_insights_transfer_product_ownership_enabled(), } permissions = get_permission_values() return flaskify(streams.get_streams_all(query_params, permissions)) @app.route(config.PRODUCT_METRICS_PATH) @access.verify_profile(access=ACCESS_ANALYTICS) def get_product_metrics(): """Return product streams metrics. Route: GET /product-metrics Query params: country_code (list[str]): Country codes to filter by. distributors (str): Comma-separated distributor names. global_participant_ids (list[str]): Participant IDs. parent_company (list[str]): Parent company names. company_brand (list[str]): Company brand names. service_tier (str): Service tier filter. label_ids (list[int]): Label IDs. subaccount_ids (list[int]): Subaccount IDs. fin_label_ids (list[str]): Financial label IDs. upper_profit_centers (list[str]): Upper profit center IDs. order_by (str): Sort field (default "streams_7_days"). order_dir (str): Sort direction (default "DESC"). limit (int): Max results (default 25). offset (int): Pagination offset (default 0). multi_product (str): "true" or "false" (default "false"). Snowflake tables: - METRICS_BY_PRODUCT_FEED_DISTRIBUTOR_ROLLUP - METRICS_BY_PRODUCT_COUNTRY_FEED_DISTRIBUTOR_ROLLUP - METRICS_BY_PRODUCT_PARTICIPANT_FEED_DISTRIBUTOR_ROLLUP - METRICS_BY_PRODUCT_PARTICIPANT_COUNTRY_FEED_DISTRIBUTOR_ROLLUP - METRICS_BY_MULTI_PRODUCT_FEED_DISTRIBUTOR_ROLLUP - V_METRICS_BY_MULTI_PRODUCT_COUNTRY_FEED_DISTRIBUTOR_ROLLUP - METRICS_BY_TRACK_FEED_DISTRIBUTOR_ROLLUP - V_METRICS_BY_TRACK_COUNTRY_FEED_DISTRIBUTOR_ROLLUP JOINs: - DIM_RELEASE - MAPPINGS_FINANCIAL_LABEL_ID_TO_PRODUCT_IDS - GLOBAL_PARTICIPANT_BY_PRODUCT_ID_BY_ISRC - PRIMARY_PRODUCTS - VENDOR_COMPANY_BRAND_PARENT_COMPANY_SERVICE_TIER_VIEW (filter subquery) - MAPPINGS_UPPER_PROFIT_CENTER_TO_PRODUCT_IDS (filter subquery) Returns: JSON with product metrics. """ distributors = request.args.get("distributors", DISTRIBUTORS).split(",") parent_companies = filter_non_empty_items( request.args.getlist("parent_company") ) or filter_non_empty_items(request.args.getlist("parent_companies")) company_brands = filter_non_empty_items( request.args.getlist("company_brand") ) or filter_non_empty_items(request.args.getlist("company_brands")) query_params = { "distributors": distributors, "country_ids": _get_csv_list("country_code"), "global_participant_ids": request.args.getlist("global_participant_ids"), "parent_companies": parent_companies, "company_brands": company_brands, "service_tier": request.args.get("service_tier"), "label_ids": _parse_int_list(request, "label_ids"), "subaccount_ids": _parse_int_list(request, "subaccount_ids"), "fin_label_ids": request.args.getlist("fin_label_ids"), "upper_profit_center_ids": request.args.getlist("upper_profit_centers"), "order_by": request.args.get("order_by", "streams_7_days"), "order_dir": request.args.get("order_dir", "DESC").upper(), "limit": request.args.get("limit", 25, type=int), "offset": request.args.get("offset", 0, type=int), "multi_product": request.args.get("multi_product", "false") == "true", "transfer_product_ownership_enabled": is_insights_transfer_product_ownership_enabled(), } permissions = get_permission_values() return flaskify(product_metrics.get_product_metrics(query_params, permissions)) @app.route(config.PRODUCT_PATH) @access.verify_profile(access=ACCESS_ANALYTICS) def get_product(product_id): """Return product details for a given product_id. Route: GET /product/ Path params: product_id (str): Product ID. Query params: distributors (str): Comma-separated distributor names. Snowflake tables: - V_STREAMS_BY_PRODUCT_TRACK_FEED_DISTRIBUTOR_DAILY - STREAMS_BY_TRACK_FEED_DISTRIBUTOR_ROLLUP - DATA_AVAILABILITY_BY_STORE_DAILY JOINs: - DIM_RELEASE - DATA_AVAILABILITY_SKIPS_SAVES_BY_FEED_SUMMARY Returns: JSON with product details and track streams. """ distributors = request.args.get("distributors", DISTRIBUTORS).split(",") query_params = { "product_id": product_id, "distributors": distributors, "transfer_product_ownership_enabled": is_insights_transfer_product_ownership_enabled(), } permissions = get_permission_values() return flaskify(product.get_product(query_params, permissions)) @app.route(config.PRODUCT_AGGREGATE_STREAMS_PATH) @access.verify_profile(access=ACCESS_ANALYTICS) def get_product_aggregate_streams(product_id): """Return aggregate streams for a given product_id. Route: GET /product//aggregate-streams Path params: product_id (str): Product ID. Query params: distributors (str): Comma-separated distributor names. multi_product (str): "true" or "false" (default "false"). country_code (str): Country code to filter by. Can be provided multiple times. Snowflake tables (default path): - METRICS_BY_TRACK_FEED_DISTRIBUTOR_ROLLUP JOINs: - DIM_RELEASE Snowflake tables (country_code path): - V_METRICS_BY_TRACK_COUNTRY_FEED_DISTRIBUTOR_ROLLUP Snowflake tables (multi_product path): - MULTI_PRIMARY_PRODUCTS - PRIMARY_PRODUCTS Returns: JSON with aggregate stream counts. """ distributors = request.args.get("distributors", DISTRIBUTORS).split(",") query_params = { "product_id": product_id, "distributors": distributors, "multi_product": request.args.get("multi_product", "false") == "true", "country_ids": sorted( { c.strip().upper() for c in request.args.getlist("country_code") if c.strip() } ), "transfer_product_ownership_enabled": is_insights_transfer_product_ownership_enabled(), } permissions = get_permission_values() return flaskify(product.get_aggregate_streams(query_params, permissions)) @app.route(config.PRODUCT_METADATA_PATH) @access.verify_profile(access=ACCESS_ANALYTICS) def get_product_metadata(product_id): """Return product metadata for a given product_id. Route: GET /product//metadata Path params: product_id (str): Product ID. Snowflake tables: None (data from DynamoDB). Returns: JSON with product metadata. """ query_params = {"product_id": product_id} permissions = get_permission_values() return flaskify(product_metadata.get_product_metadata(query_params, permissions)) @app.route(config.TOP_SOUND_RECORDINGS_PATH) @access.verify_profile(access=ACCESS_ANALYTICS) def get_top_sound_recordings(): """Return top sound recordings. Route: GET /top-sound-recordings Query params: country_code (list[str]): Country codes to filter by. distributors (str): Comma-separated distributor names. order_by (str): Sort field (default "streams_7_days"). limit (int): Max results (default 25). offset (int): Pagination offset (default 0). Snowflake tables: - STREAMS_BY_TRACK_FEED_DISTRIBUTOR_ROLLUP - V_STREAMS_BY_TRACK_COUNTRY_FEED_DISTRIBUTOR_ROLLUP JOINs: - DIM_RELEASE Returns: JSON with top sound recordings. """ countries = request.args.getlist("country_code") distributors = request.args.get("distributors", DISTRIBUTORS).split(",") query_params = { "country_ids": countries, "distributors": distributors, "order_by": request.args.get("order_by", "streams_7_days"), "limit": int(request.args.get("limit", 25)), "offset": int(request.args.get("offset", 0)), "transfer_product_ownership_enabled": is_insights_transfer_product_ownership_enabled(), } permissions = get_permission_values() return flaskify( top_sound_recordings.get_top_sound_recordings(query_params, permissions) ) @app.route(config.TOP_METRICS_PATH) @access.verify_profile(access=ACCESS_ANALYTICS) def get_top_metrics(): """Return top metrics for sound recordings. Route: GET /top-metrics Query params: upc (str): UPC to filter by. country_code (list[str]): Country codes to filter by. isrc_country_code (list[str]): ISRC country codes. release_period_start (str): Release period start date. release_period_end (str): Release period end date. only_daily_data (str): "true" for gainers mode. only_weekly_data (str): "true" for gainers mode. tiktok_creations_change_threshold (int): Threshold (default 0). active_streams_change_threshold (int): Threshold (default -1). label_manager_id (str): Label manager ID. store_ids (list[int]): Store IDs to filter by. label_ids (list[str]): Label IDs. subaccount_ids (list[str]): Subaccount IDs. global_participant_ids (list[str]): Participant IDs. parent_company (list[str]): Parent company names. company_brand (list[str]): Company brand names. service_tier (str): Service tier filter. distributors (str): Comma-separated distributor names. order_by (str): Sort field (default "streams_7_days"). order_dir (str): Sort direction (default "DESC"). limit (int): Max results (default 50). offset (int): Pagination offset (default 0). Snowflake tables: Normal mode: - METRICS_BY_TRACK_FEED_DISTRIBUTOR_ROLLUP - V_METRICS_BY_TRACK_COUNTRY_FEED_DISTRIBUTOR_ROLLUP - V_METRICS_BY_TRACK_REGION_FEED_DISTRIBUTOR_ROLLUP - METRICS_BY_TRACK_PARTICIPANT_FEED_DISTRIBUTOR_ROLLUP - V_METRICS_BY_TRACK_PARTICIPANT_COUNTRY_FEED_DISTRIBUTOR_ROLLUP - TIKTOK_BY_ISRC_COUNTRY_ROLLUP - TIKTOK_BY_ISRC_PRODUCT_COUNTRY_ROLLUP Gainers mode (only_daily_data/only_weekly_data): - METRICS_BY_TRACK_CREATIONS_THRESHOLD_ROLLUP - METRICS_BY_TRACK_COUNTRY_CREATIONS_THRESHOLD_ROLLUP JOINs: - PRIMARY_PRODUCTS - MAPPINGS_ISRC_TO_MIN_RELEASE_DATE - VENDOR_COMPANY_BRAND_PARENT_COMPANY_SERVICE_TIER_VIEW (filter subquery) Returns: JSON with top metrics. """ query_params = { "upc": request.args.get("upc"), "country_ids": _get_csv_list("country_code"), "isrc_country_ids": request.args.getlist("isrc_country_code"), "release_period_start": request.args.get("release_period_start"), "release_period_end": request.args.get("release_period_end"), "only_daily_data": True if request.args.get("only_daily_data") == "true" else False, "only_weekly_data": True if request.args.get("only_weekly_data") == "true" else False, "tiktok_creations_change_threshold": request.args.get( "tiktok_creations_change_threshold", 0 ), "active_streams_change_threshold": request.args.get( "active_streams_change_threshold", -1 ), "label_manager_id": request.args.get("label_manager_id"), "store_ids": list(map(int, request.args.getlist("store_ids"))), "label_ids": request.args.getlist("label_ids"), "subaccount_ids": request.args.getlist("subaccount_ids"), "global_participant_ids": request.args.getlist("global_participant_ids"), "parent_companies": filter_non_empty_items( request.args.getlist("parent_company") ) or filter_non_empty_items(request.args.getlist("parent_companies")), "company_brands": filter_non_empty_items(request.args.getlist("company_brand")) or filter_non_empty_items(request.args.getlist("company_brands")), "service_tier": request.args.get("service_tier"), "distributors": request.args.get("distributors", DISTRIBUTORS).split(","), "order_by": request.args.get("order_by", "streams_7_days"), "order_dir": request.args.get("order_dir", "DESC"), "limit": request.args.get("limit", 50), "offset": request.args.get("offset", DEFAULT_OFFSET, type=int), "transfer_product_ownership_enabled": is_insights_transfer_product_ownership_enabled(), } permissions = get_permission_values() if query_params["only_daily_data"] or query_params["only_weekly_data"]: return flaskify(top_metrics.get_top_metrics_gainers(query_params, permissions)) else: return flaskify(top_metrics.get_top_metrics(query_params, permissions)) @app.route(config.TOP_SOUND_RECORDING_FAMILIES_IDS_PATH) @access.verify_profile(access=ACCESS_ANALYTICS) def get_top_sound_recording_families_ids(): """Return sound recording family IDs by filters. Route: GET /top-sound-recording-families-ids Query params: country_code (list[str]): Country codes to filter by. store_ids (list[int]): Store IDs to filter by. distributors (str): Comma-separated distributor names. fin_label_ids (list[str]): Financial label IDs. label_ids (list[str]): Label IDs (mapped to fin_label_ids). upper_profit_center_ids (list[str]): Profit center IDs. parent_company (str): Parent company name. company_brand (str): Company brand name. service_tier (str): Service tier filter. global_participant_ids (list[str]): Participant IDs. order_by (str): Sort field (default "streams_7_days"). order_dir (str): Sort direction (default "DESC"). limit (int): Max results (default 25). offset (int): Pagination offset (default 0). Snowflake tables: - METRICS_BY_TRACK_FEED_DISTRIBUTOR_PRODFAM_ROLLUP_V2 - V_METRICS_BY_TRACK_COUNTRY_FEED_DISTRIBUTOR_PRODFAM_ROLLUP_V2 - V_METRICS_BY_TRACK_PARTICIPANT_COUNTRY_FEED_DISTRIBUTOR_PRODFAM_ROLLUP JOINs: - MAPPINGS_PRODUCT_FAMILY_TO_ISRCS - MAPPINGS_UPPER_PROFIT_CENTER_TO_PRODUCT_IDS - DIM_RELEASE - MAPPINGS_FINANCIAL_LABEL_ID_TO_PRODUCT_IDS - VENDOR_COMPANY_BRAND_PARENT_COMPANY_SERVICE_TIER_VIEW (filter subquery) Returns: JSON with sound recording family IDs. """ fin_label_ids = request.args.getlist("fin_label_ids") or list( filter( None, map( lambda label_id: ACCOUNT_TO_FIN_LABEL_MAPPING.get(int(label_id)), request.args.getlist("label_ids"), ), ) ) query_params = { "country_ids": request.args.getlist("country_code"), "store_ids": list(map(int, request.args.getlist("store_ids"))), "distributors": request.args.get("distributors", DISTRIBUTORS).split(","), "fin_label_ids": fin_label_ids, "upper_profit_center_ids": request.args.getlist("upper_profit_center_ids") or [], "parent_company": request.args.get("parent_company"), "company_brand": request.args.get("company_brand"), "service_tier": request.args.get("service_tier"), "global_participant_ids": request.args.getlist("global_participant_ids"), "order_by": request.args.get("order_by", "streams_7_days"), "order_dir": request.args.get("order_dir", "DESC"), "limit": int(request.args.get("limit", 25)), "offset": int(request.args.get("offset", 0)), } permissions = get_permission_values() return flaskify( top_sound_recording_families.get_top_sound_recording_families_ids( query_params, permissions ) ) @app.route(config.TOP_SOUND_RECORDING_FAMILIES_BY_IDS_PATH, methods=["POST"]) @access.verify_profile(access=ACCESS_ANALYTICS) def get_top_sound_recording_families_by_ids(): """Return top sound recording families by IDs. Route: POST /top-sound-recording-families-by-ids Body params: ids (list[str]): Product family IDs. params.countryCodes (list[str]): Country codes. params.storeIds (list[int]): Store IDs. params.distributors (list[str]): Distributors. params.finLabelIds (list[str]): Financial label IDs. params.labelIds (list[str]): Label IDs. params.upperProfitCenters (list[str]): Profit center IDs. params.parentCompanyId (str): Parent company ID. params.companyBrandId (str): Company brand ID. params.serviceTierId (str): Service tier ID. params.globalParticipantIds (list[str]): Participant IDs. params.orderBy (str): Sort field (default "streams_7_days"). params.orderDir (str): Sort direction (default "DESC"). params.limit (int): Max results. params.offset (int): Pagination offset. Snowflake tables: - METRICS_BY_TRACK_FEED_DISTRIBUTOR_PRODFAM_ROLLUP_V2 - V_METRICS_BY_TRACK_COUNTRY_FEED_DISTRIBUTOR_PRODFAM_ROLLUP_V2 - V_METRICS_BY_TRACK_PARTICIPANT_COUNTRY_FEED_DISTRIBUTOR_PRODFAM_ROLLUP - TIKTOK_TREND_SCORE_BY_ISRC_DBT JOINs: - MAPPINGS_PRODUCT_FAMILY_TO_ISRCS - MAPPINGS_UPPER_PROFIT_CENTER_TO_PRODUCT_IDS - DIM_RELEASE - MAPPINGS_FINANCIAL_LABEL_ID_TO_PRODUCT_IDS - VENDOR_COMPANY_BRAND_PARENT_COMPANY_SERVICE_TIER_VIEW (filter subquery) Returns: JSON with sound recording family metrics. """ data = request.get_json() input_params = data.get("params") fin_label_ids = input_params.get("finLabelIds") or list( filter( None, map( lambda label_id: ACCOUNT_TO_FIN_LABEL_MAPPING.get(int(label_id)), input_params.get("labelIds") or [], ), ) ) query_params = { "prod_fam_ids": data.get("ids"), "country_ids": input_params.get("countryCodes", []), "store_ids": list(map(int, input_params.get("storeIds", []))), "distributors": input_params.get("distributors", DISTRIBUTORS.split(",")), "fin_label_ids": fin_label_ids, "upper_profit_center_ids": input_params.get("upperProfitCenters"), "parent_company": input_params.get("parentCompanyId"), "company_brand": input_params.get("companyBrandId"), "service_tier": input_params.get("serviceTierId"), "global_participant_ids": input_params.get("globalParticipantIds", []), "order_by": input_params.get("orderBy", "streams_7_days"), "order_dir": input_params.get("orderDir", "DESC"), "limit": int(input_params.get("limit")) if input_params.get("limit") else None, "offset": int(input_params.get("offset")) if input_params.get("limit") else None, } permissions = get_permission_values() return flaskify( top_sound_recording_families.get_top_sound_recording_families_by_ids( query_params, permissions ) ) @app.route(config.TOP_ACCOUNTS_METRICS_PATH) @access.verify_profile(access=ACCESS_ANALYTICS) def get_top_accounts_metrics(): """Return top metrics for accounts. Route: GET /top-accounts-metrics Query params: countries (list[str]): Country codes to filter by. store_ids (list[int]): Store IDs to filter by. distributors (str): Comma-separated distributor names. label_ids (list[int]): Label IDs. subaccount_ids (list[int]): Subaccount IDs. parent_company (str): Parent company name. company_brand (str): Company brand name. service_tier (str): Service tier filter. label_manager (str): Label manager filter. include_subaccounts (str): "true" or "false" (default). order_by (str): Sort field (default "streams_28_days"). order_dir (str): Sort direction (default "DESC"). limit (int): Max results (default 25). offset (int): Pagination offset (default 0). Snowflake tables: - METRICS_BY_ACCOUNT_COUNTRY_FEED_DISTRIBUTOR_ROLLUP JOINs: - VENDOR_COMPANY_BRAND_PARENT_COMPANY_SERVICE_TIER_VIEW (filter subquery) Returns: JSON with top account metrics. """ distributors = request.args.get("distributors", DISTRIBUTORS_ORCH_AWAL).split(",") query_params = { "distributors": distributors, "countries": request.args.getlist("countries"), "store_ids": [int(s) for s in request.args.getlist("store_ids")], "label_ids": [int(s) for s in request.args.getlist("label_ids") if int(s) > 0], "subaccount_ids": [ int(s) for s in request.args.getlist("subaccount_ids") if int(s) > 0 ], "parent_company": request.args.get("parent_company"), "company_brand": request.args.get("company_brand"), "service_tier": request.args.get("service_tier"), "label_manager": request.args.get("label_manager"), "include_subaccounts": request.args.get("include_subaccounts", "false") == "true", "order_by": request.args.get("order_by", "streams_28_days"), "order_dir": request.args.get("order_dir", "DESC"), "limit": int(request.args.get("limit", 25)), "offset": int(request.args.get("offset", 0)), "transfer_product_ownership_enabled": is_insights_transfer_product_ownership_enabled(), } permissions = get_permission_values() return flaskify( top_accounts_metrics.get_top_accounts_metrics(query_params, permissions) ) @app.route(config.TOP_VIDEO_METRICS_PATH) @access.verify_profile(access=ACCESS_ANALYTICS) def get_top_video_metrics(): """Return top video metrics. Route: GET /top-video-metrics Query params: country_code (list[str]): Country codes to filter by. store_ids (list[int]): Store IDs to filter by. distributors (str): Comma-separated distributor names. label_ids (list[int]): Label IDs. subaccount_ids (list[int]): Subaccount IDs. global_participant_ids (list[str]): Participant IDs. channel_ids (list[str]): Channel IDs. track_types (list[str]): Track types (default ["video"]). order_by (str): Sort field (default "views_1_month_back"). order_dir (str): Sort direction (default "DESC"). limit (int): Max results (default 25). offset (int): Pagination offset (default 0). Snowflake tables: - METRICS_BY_VIDEO_FEED_DISTRIBUTOR_ROLLUP - METRICS_BY_VIDEO_COUNTRY_FEED_DISTRIBUTOR_ROLLUP JOINs: - YOUTUBE_VIDEO - MAPPINGS_VIDEO_TO_TRACKS_V3 (filter subquery) - DIM_SUBACCOUNT (filter subquery) - MAPPINGS_VIDEO_TO_PARTICIPANTS (filter subquery) - DIM_RELEASE (permissions subquery) Returns: JSON with top video metrics. """ distributors = request.args.get("distributors", DISTRIBUTORS).split(",") query_params = { "distributors": distributors, "countries": request.args.getlist("country_code"), "store_ids": _parse_int_list(request, "store_ids"), "label_ids": _parse_int_list(request, "label_ids"), "subaccount_ids": _parse_int_list(request, "subaccount_ids"), "global_participant_ids": request.args.getlist("global_participant_ids"), "channel_ids": request.args.getlist("channel_ids"), "track_types": request.args.getlist("track_types") or ["video"], "order_by": request.args.get("order_by", "views_1_month_back"), "order_dir": request.args.get("order_dir", "DESC"), "limit": int(request.args.get("limit", 25)), "offset": int(request.args.get("offset", 0)), } permissions = get_permission_values() return flaskify(top_video_metrics.get_top_video_metrics(query_params, permissions)) @app.route(config.TOP_CHANNEL_METRICS_PATH) @access.verify_profile(access=ACCESS_ANALYTICS) def get_top_channel_metrics(): """Return top channel metrics. Route: GET /top-channel-metrics Query params: country_code (list[str]): Country codes to filter by. store_ids (list[int]): Store IDs to filter by. distributors (str): Comma-separated distributor names. label_ids (list[int]): Label IDs. subaccount_ids (list[int]): Subaccount IDs. global_participant_ids (list[str]): Participant IDs. order_by (str): Sort field (default "views_1_month_back"). order_dir (str): Sort direction (default "DESC"). limit (int): Max results (default 25). offset (int): Pagination offset (default 0). Snowflake tables: - METRICS_BY_CHANNEL_FEED_DISTRIBUTOR_ROLLUP - METRICS_BY_CHANNEL_COUNTRY_FEED_DISTRIBUTOR_ROLLUP JOINs: - MAPPINGS_VIDEO_TO_TRACKS_V3 (filter/permissions) - DIM_RELEASE (permissions subquery) - MAPPINGS_VIDEO_TO_PARTICIPANTS (filter subquery) - ORCHARD_APP_REPORTING_V2.ART_RELATIONS_PROD_ART_RELATIONS.YOUTUBE_CHANNEL (filter subquery) Returns: JSON with top channel metrics. """ distributors = request.args.get("distributors", DISTRIBUTORS).split(",") query_params = { "distributors": distributors, "countries": request.args.getlist("country_code"), "store_ids": _parse_int_list(request, "store_ids"), "label_ids": _parse_int_list(request, "label_ids"), "subaccount_ids": _parse_int_list(request, "subaccount_ids"), "global_participant_ids": request.args.getlist("global_participant_ids"), "order_by": request.args.get("order_by", "views_1_month_back"), "order_dir": request.args.get("order_dir", "DESC"), "limit": int(request.args.get("limit", 25)), "offset": int(request.args.get("offset", 0)), } permissions = get_permission_values() return flaskify( top_channel_metrics.get_top_channel_metrics(query_params, permissions) ) @app.route(config.UGC_VIDEO_METRICS_PATH) @access.verify_profile(access=ACCESS_ANALYTICS) def get_ugc_video_metrics(): """Return UGC video metrics. Route: GET /ugc-video-metrics Query params: store_ids (list[int]): Store IDs to filter by. video_ids (list[str]): Video IDs to filter by. track_ids (list[int]): Track IDs to filter by. asset_ids (list[str]): Asset IDs to filter by. track_isrcs (list[str]): Track ISRCs to filter by. countries (list[str]): Country codes to filter by. claim_types (list[str]): Claim types to filter by. claim_statuses (list[str]): Claim statuses to filter by. distributors (str): Comma-separated distributor names. order_by (str): Sort field (default "views"). order_dir (str): Sort direction (default "DESC"). limit (int): Max results (default 25). offset (int): Pagination offset (default 0). Snowflake tables: - V_METRICS_UGC_BY_VIDEO_ASSET_FEED_DISTRIBUTOR_ROLLUP - V_METRICS_UGC_BY_VIDEO_ASSET_COUNTRY_FEED_DISTRIBUTOR_ROLLUP JOINs: - MAPPINGS_VIDEO_TO_ASSETS_BY_TRACK_ID (filter subquery) - MAPPINGS_VIDEO_TO_ASSETS (filter subquery) - MAPPINGS_TRACK_TO_VIDEOS (filter subquery) - DIM_RELEASE (permissions subquery) - LABEL_PARTICIPANT_PARTICIPATED_IN_ORCHARD_PRODUCT (permissions subquery) Returns: JSON with UGC video metrics. """ distributors = request.args.get("distributors") or DISTRIBUTORS query_params = { "distributors": distributors.split(","), "store_ids": _parse_int_list(request, "store_ids"), "video_ids": request.args.getlist("video_ids"), "track_ids": _parse_int_list(request, "track_ids"), "asset_ids": request.args.getlist("asset_ids"), "track_isrcs": request.args.getlist("track_isrcs"), "countries": request.args.getlist("countries"), "claim_types": request.args.getlist("claim_types"), "claim_statuses": request.args.getlist("claim_statuses"), "order_by": request.args.get("order_by", "views"), "order_dir": request.args.get("order_dir", "DESC"), "limit": int(request.args.get("limit", 25)), "offset": int(request.args.get("offset", 0)), "transfer_product_ownership_enabled": is_insights_transfer_product_ownership_enabled(), } permissions = get_permission_values() return flaskify(ugc_video_metrics.get_ugc_video_metrics(query_params, permissions)) @app.route(config.SOUND_RECORDING_METADATA_PATH) def get_sound_recording_metadata(isrc): """Return sound-recording metadata for a given isrc. Route: GET /sound-recording//metadata Path params: isrc (str): ISRC identifier. Query params: include_deleted (str): "true" or "false" (default "false"). Snowflake tables: None (data from DynamoDB). Returns: JSON with sound recording metadata. """ query_params = { "isrc": isrc, "include_deleted": request.args.get("include_deleted", "false") == "true", } permissions = get_permission_values() return flaskify( sound_recording_metadata.get_sound_recording_metadata(query_params, permissions) ) @app.route(config.AGGREGATE_STREAMS_PATH, methods=["POST"]) @access.verify_profile(access=ACCESS_ANALYTICS) def get_aggregate_streams(): """Return sound-recording aggregate streams for a list of isrcs. Route: POST /sound-recording/aggregate-streams Query params: country_code (list[str]): Country codes to filter by. store_ids (list[int]): Store IDs to filter by. distributors (str): Comma-separated distributor names. Body params: isrcs (list[str]): ISRCs to aggregate. Snowflake tables: - STREAMS_BY_TRACK_FEED_DISTRIBUTOR_ROLLUP - V_STREAMS_BY_TRACK_COUNTRY_FEED_DISTRIBUTOR_ROLLUP JOINs: - DIM_RELEASE - LABEL_PARTICIPANT_PARTICIPATED_IN_ORCHARD_PRODUCT (permissions subquery) Returns: JSON with aggregate streams per ISRC. """ data = request.get_json() isrcs = data.get("isrcs", []) if len(isrcs) == 0: return flaskify(response.Response({})) countries, store_ids, _, _, distributors = _get_global_filters() query_params = { "isrcs": isrcs, "country_ids": countries, "store_ids": store_ids, "distributors": distributors, "transfer_product_ownership_enabled": is_insights_transfer_product_ownership_enabled(), } permissions = get_permission_values() return flaskify(aggregate_streams.get_aggregate_streams(query_params, permissions)) @app.route(config.STREAMS_BULK_PATH, methods=["POST"]) @access.verify_profile(access=ACCESS_ANALYTICS) def get_streams_bulk(): """Return sound-recording streams for a list of isrcs. Route: POST /sound-recording/streams Query params: country_code (list[str]): Country codes to filter by. store_ids (list[int]): Store IDs to filter by. start_date (str): Start date YYYY-MM-DD. end_date (str): End date YYYY-MM-DD. distributors (str): Comma-separated distributor names. Body params: isrcs (list[str]): ISRCs to query. Snowflake tables: - V_STREAMS_BY_TRACK_FEED_DISTRIBUTOR_DAILY - V_STREAMS_BY_TRACK_COUNTRY_FEED_DISTRIBUTOR_DAILY - STREAMS_BY_TRACK_FEED_DISTRIBUTOR_ROLLUP - V_STREAMS_BY_TRACK_COUNTRY_FEED_DISTRIBUTOR_ROLLUP JOINs: - DIM_RELEASE - LABEL_PARTICIPANT_PARTICIPATED_IN_ORCHARD_PRODUCT (permissions subquery) Returns: JSON with streams data per ISRC. """ data = request.get_json() isrcs = data.get("isrcs", []) if len(isrcs) == 0: return flaskify(response.Response({})) countries, store_ids, start_date, end_date, distributors = _get_global_filters() query_params = { "isrcs": isrcs, "country_ids": countries, "store_ids": store_ids, "start_date": start_date, "end_date": end_date, "distributors": distributors, "transfer_product_ownership_enabled": is_insights_transfer_product_ownership_enabled(), } permissions = get_permission_values() return flaskify(streams_bulk.get_streams_bulk(query_params, permissions)) @app.route(config.TOP_MARKETS_PATH) @access.verify_profile(access=ACCESS_ANALYTICS) def get_top_markets(isrc): """Return sound-recording top markets for a given isrc. Route: GET /sound-recording//top-markets Path params: isrc (str): ISRC identifier. Query params: country_code (list[str]): Country codes to filter by. store_ids (list[int]): Store IDs to filter by. distributors (str): Comma-separated distributor names. Snowflake tables: - STREAMS_BY_TRACK_COUNTRY_REGION_FEED_DISTRIBUTOR_ROLLUP JOINs: - DIM_RELEASE - LABEL_PARTICIPANT_PARTICIPATED_IN_ORCHARD_PRODUCT (permissions subquery) Returns: JSON with top markets by country/region. """ countries, store_ids, _, _, distributors = _get_global_filters() query_params = { "isrc": isrc, "country_ids": countries, "store_ids": store_ids, "distributors": distributors, "transfer_product_ownership_enabled": is_insights_transfer_product_ownership_enabled(), } permissions = get_permission_values() return flaskify(top_markets.get_top_markets(query_params, permissions)) @app.route(config.DEMOGRAPHICS_PATH) @access.verify_profile(access=ACCESS_ANALYTICS) def get_demographics(isrc): """Return demographics for a given isrc. Route: GET /sound-recording//demographics Path params: isrc (str): ISRC identifier. Query params: country_code (list[str]): Country codes to filter by. store_ids (list[int]): Store IDs to filter by. start_date (str): Start date YYYY-MM-DD. end_date (str): End date YYYY-MM-DD. distributors (str): Comma-separated distributor names. Snowflake tables: - V_STREAMS_DEMOGRAPHICS_BY_TRACK_COUNTRY_FEED_DISTRIBUTOR_DAILY - DATA_AVAILABILITY_BY_STORE_DAILY JOINs: - DIM_RELEASE - DATA_AVAILABILITY_SKIPS_SAVES_BY_FEED_SUMMARY Returns: JSON with demographics breakdown. """ countries, store_ids, start_date, end_date, distributors = _get_global_filters() query_params = { "isrc": isrc, "query_type": "isrc", "countries": countries, "store_ids": store_ids, "start_date": start_date, "end_date": end_date, "distributors": distributors, "transfer_product_ownership_enabled": is_insights_transfer_product_ownership_enabled(), } permissions = get_permission_values() return jsonify(demographics.get_demographics(query_params, permissions)), 200 @app.route(config.DEMOGRAPHICS_2_PATH) @access.verify_profile(access=ACCESS_ANALYTICS) def get_demographics_2(): """Return demographics for a given song, artist or account. Route: GET /demographics Query params: countries (list[str]): Country codes to filter by. store_ids (list[int]): Store IDs to filter by. start_date (str): Start date YYYY-MM-DD. end_date (str): End date YYYY-MM-DD. isrc (str): ISRC identifier (mutually exclusive with global_participant_id and account_id). global_participant_id (str): Participant ID. account_id (str): Account ID. account_type (str): Account type (used with account_id). Snowflake tables: - STREAMS_DEMOGRAPHICS_BY_TRACK_FEED_DISTRIBUTOR_ROLLUP - V_STREAMS_DEMOGRAPHICS_BY_TRACK_FEED_DISTRIBUTOR_DAILY - V_STREAMS_DEMOGRAPHICS_BY_TRACK_COUNTRY_FEED_DISTRIBUTOR_DAILY - STREAMS_DEMOGRAPHICS_BY_PARTICIPANT_FEED_DISTRIBUTOR_ROLLUP - V_STREAMS_DEMOGRAPHICS_BY_PARTICIPANT_FEED_DISTRIBUTOR_DAILY - STREAMS_DEMOGRAPHICS_BY_PARTICIPANT_COUNTRY_FEED_DISTRIBUTOR_ROLLUP - V_STREAMS_DEMOGRAPHICS_BY_PARTICIPANT_COUNTRY_FEED_DISTRIBUTOR_DAILY - STREAMS_DEMOGRAPHICS_BY_PRODUCT_FEED_DISTRIBUTOR_ROLLUP - V_STREAMS_DEMOGRAPHICS_BY_PRODUCT_FEED_DISTRIBUTOR_DAILY - STREAMS_DEMOGRAPHICS_BY_PRODUCT_COUNTRY_FEED_DISTRIBUTOR_ROLLUP - V_STREAMS_DEMOGRAPHICS_BY_PRODUCT_COUNTRY_FEED_DISTRIBUTOR_DAILY - DATA_AVAILABILITY_BY_STORE_DAILY JOINs: - DIM_RELEASE - DIM_RELEASE (permissions subquery) - LABEL_PARTICIPANT_PARTICIPATED_IN_ORCHARD_PRODUCT - DATA_AVAILABILITY_SKIPS_SAVES_BY_FEED_SUMMARY Returns: JSON with demographics breakdown by age/gender. """ query_params = { "countries": request.args.getlist("countries"), "store_ids": request.args.getlist("store_ids"), "start_date": request.args.get("start_date"), "end_date": request.args.get("end_date"), "transfer_product_ownership_enabled": is_insights_transfer_product_ownership_enabled(), } if isrc := request.args.get("isrc"): query_params.update(query_type="isrc", isrc=isrc) elif global_participant_id := request.args.get("global_participant_id"): query_params.update( query_type="global_participant_id", global_participant_id=global_participant_id, ) elif account_id := request.args.get("account_id"): query_params.update( account_id=account_id, query_type="account_id", account_type=request.args.get("account_type"), ) query_params["is_apple_demographics_breakdown"] = query_params[ "query_type" ] == "isrc" and query_params["store_ids"] == ["1"] query_params["is_spotify_demographics_breakdown"] = query_params[ "query_type" ] == "isrc" and query_params["store_ids"] == ["286"] permissions = get_permission_values() return jsonify(demographics.get_demographics_2(query_params, permissions)), 200 @app.route(config.TOP_COUNTRIES_VIDEOS_PATH) @access.verify_profile(access=ACCESS_ANALYTICS) def get_top_countries_videos(video_id): """Return top countries with views for a given video. Route: GET /videos//top-countries Path params: video_id (str): Video ID. Query params: store_ids (list[int]): Store IDs to filter by. start_date (str): Start date YYYY-MM-DD. end_date (str): End date YYYY-MM-DD. distributors (str): Comma-separated distributor names. Snowflake tables: - V_VIEWS_BY_VIDEO_COUNTRY_FEED_DISTRIBUTOR_DAILY JOINs: - MAPPINGS_VIDEO_TO_CHANNELS - MAPPINGS_VIDEO_TO_PRODUCTS (permissions subquery) - DIM_RELEASE (permissions subquery) Returns: JSON with top countries and view counts. """ _, store_ids, start_date, end_date, distributors = _get_global_filters() if start_date and not isinstance(start_date, str): start_date = start_date.strftime("%Y-%m-%d") if end_date and not isinstance(end_date, str): end_date = end_date.strftime("%Y-%m-%d") query_params = { "video_id": video_id, "store_ids": store_ids, "start_date": start_date, "end_date": end_date, "distributors": distributors, } permissions = get_permission_values() return flaskify( top_countries_videos.get_top_countries_videos(query_params, permissions) ) @app.route(config.TOP_COUNTRIES_CHANNELS_PATH) @access.verify_profile(access=ACCESS_ANALYTICS) def get_top_countries_channels(channel_id): """Return top countries with views for a given channel. Route: GET /channels//top-countries Path params: channel_id (str): Channel ID. Query params: store_ids (list[int]): Store IDs to filter by. start_date (str): Start date YYYY-MM-DD. end_date (str): End date YYYY-MM-DD. distributors (str): Comma-separated distributor names. Snowflake tables: - VIEWS_BY_CHANNEL_COUNTRY_FEED_DISTRIBUTOR_DAILY JOINs: - ORCHARD_APP_REPORTING_V2.ART_RELATIONS_PROD_ART_RELATIONS.YOUTUBE_CHANNEL (permissions subquery) Returns: JSON with top countries and view counts. """ _, store_ids, start_date, end_date, distributors = _get_global_filters() if start_date and not isinstance(start_date, str): start_date = start_date.strftime("%Y-%m-%d") if end_date and not isinstance(end_date, str): end_date = end_date.strftime("%Y-%m-%d") query_params = { "channel_id": channel_id, "store_ids": store_ids, "start_date": start_date, "end_date": end_date, "distributors": distributors, } permissions = get_permission_values() return flaskify( top_countries_channels.get_top_countries_channels(query_params, permissions) ) @app.route(config.TOP_VIDEO_TRAFFIC_SOURCES_PATH) @access.verify_profile(access=ACCESS_ANALYTICS) def get_top_video_traffic_sources(video_id): """Return top traffic sources for a given video. Route: GET /videos//top-traffic-sources Path params: video_id (str): Video ID. Query params: country_code (list[str]): Country codes to filter by. store_ids (list[int]): Store IDs to filter by. start_date (str): Start date YYYY-MM-DD. days (int): Number of days (default 28). distributors (list[str]): Distributor names. limit (int): Max results (default 5). Snowflake tables: - V_VIEWS_BY_VIDEO_SOURCE_FEED_DISTRIBUTOR_DAILY - V_VIEWS_BY_VIDEO_SOURCE_COUNTRY_FEED_DISTRIBUTOR_DAILY JOINs: - MAPPINGS_VIDEO_TO_CHANNELS - DIM_TRAFFICSOURCE - MAPPINGS_VIDEO_TO_PRODUCTS (permissions subquery) - DIM_RELEASE (permissions subquery) Returns: JSON with top traffic sources and view counts. """ if request.args.get("start_date") == data_availability.HIGHWATERMARK_DATE: start_date, end_date = data_availability.get_date_range( data_availability.HIGHWATERMARK_DATE, int(request.args.get("days", 28)), downloads=False, videos=True, ) else: start_date, end_date = data_availability.get_date_range( request.args.get( "start_date", data_availability.get_videos_max_available_date().strftime("%Y-%m-%d"), ), int(request.args.get("days", 28)), downloads=False, videos=True, ) query_params = { "video_id": video_id, "country_ids": request.args.getlist("country_code"), "store_ids": list(map(int, request.args.getlist("store_ids"))), "start_date": start_date.strftime("%Y-%m-%d") if start_date else None, "end_date": end_date.strftime("%Y-%m-%d") if end_date else None, "distributors": request.args.getlist("distributors") or DISTRIBUTORS.split(","), "limit": request.args.get("limit", 5), } permissions = get_permission_values() return flaskify( top_video_traffic_sources.get_top_video_traffic_sources( query_params, permissions ) ) @app.route(config.TOP_CHANNEL_TRAFFIC_SOURCES_PATH) @access.verify_profile(access=ACCESS_ANALYTICS) def get_top_channel_traffic_sources(channel_id): """Return top traffic sources for a given channel. Route: GET /channels//top-traffic-sources Path params: channel_id (str): Channel ID. Query params: country_code (list[str]): Country codes to filter by. store_ids (list[int]): Store IDs to filter by. start_date (str): Start date YYYY-MM-DD. end_date (str): End date YYYY-MM-DD. distributors (str): Comma-separated distributor names. limit (int): Max results (default 5). Snowflake tables: - VIEWS_BY_CHANNEL_SOURCE_FEED_DISTRIBUTOR_DAILY - VIEWS_BY_CHANNEL_SOURCE_FEED_DISTRIBUTOR_ROLLUP - VIEWS_BY_CHANNEL_SOURCE_COUNTRY_FEED_DISTRIBUTOR_DAILY - VIEWS_BY_CHANNEL_SOURCE_COUNTRY_FEED_DISTRIBUTOR_ROLLUP JOINs: - DIM_TRAFFICSOURCE - ORCHARD_APP_REPORTING_V2.ART_RELATIONS_PROD_ART_RELATIONS.YOUTUBE_CHANNEL (permissions subquery) Returns: JSON with top traffic sources and view counts. """ country_ids, store_ids, start_date, end_date, distributors = _get_global_filters() if start_date and not isinstance(start_date, str): start_date = start_date.strftime("%Y-%m-%d") if end_date and not isinstance(end_date, str): end_date = end_date.strftime("%Y-%m-%d") query_params = { "channel_id": channel_id, "country_ids": country_ids, "store_ids": store_ids, "start_date": start_date, "end_date": end_date, "distributors": distributors, "limit": int(request.args.get("limit", 5)), } permissions = get_permission_values() return flaskify( channel_traffic_sources.get_top_channel_traffic_sources( query_params, permissions ) ) @app.route(config.TOP_CHANNEL_VIDEOS_PATH) @access.verify_profile(access=ACCESS_ANALYTICS) def get_channel_top_videos(channel_id): """Return top videos for a given channel. Route: GET /channels//top-videos Path params: channel_id (str): Channel ID. Query params: country_code (list[str]): Country codes to filter by. store_ids (list[int]): Store IDs to filter by. start_date (str): Start date YYYY-MM-DD. end_date (str): End date YYYY-MM-DD. distributors (str): Comma-separated distributor names. limit (int): Max results (default 5). Snowflake tables: - VIEWS_BY_VIDEO_FEED_DISTRIBUTOR_DAILY - V_VIEWS_BY_VIDEO_COUNTRY_FEED_DISTRIBUTOR_DAILY - VIEWS_BY_VIDEO_FEED_DISTRIBUTOR_ROLLUP - VIEWS_BY_VIDEO_COUNTRY_FEED_DISTRIBUTOR_ROLLUP JOINs: - YOUTUBE_VIDEO - ORCHARD_APP_REPORTING_V2.ART_RELATIONS_PROD_ART_RELATIONS.YOUTUBE_CHANNEL (permissions subquery) Returns: JSON with top videos and view counts. """ country_ids, store_ids, start_date, end_date, distributors = _get_global_filters() if start_date == ALL_TIME: start_date = None # if start_date and end_date are not str, convert to str in order to avoid conflict with fields.Date() if start_date and not isinstance(start_date, str): start_date = start_date.strftime("%Y-%m-%d") if end_date and not isinstance(end_date, str): end_date = end_date.strftime("%Y-%m-%d") if not (start_date and end_date): all_time = True else: all_time = False query_params = { "channel_id": channel_id, "country_ids": country_ids, "store_ids": list(map(int, store_ids)), "start_date": start_date, "end_date": end_date, "all_time": all_time, "distributors": distributors, "limit": int(request.args.get("limit", 5)), } permissions = get_permission_values() return flaskify( channel_top_videos.get_channel_top_videos(query_params, permissions) ) @app.route(config.VIDEO_METRICS_BULK_PATH, methods=["POST"]) @access.verify_profile(access=ACCESS_ANALYTICS) def get_video_metrics_bulk(): """Return metrics for the given video_ids. Route: POST /video-metrics-bulk Body params: params.video_ids (list[str]): Video IDs. params.country_ids (list[str]): Country codes. params.store_ids (list[int]): Store IDs. params.start_date (str): Start date YYYY-MM-DD. params.days (int): Number of days. Snowflake tables: - VIEWS_BY_VIDEO_FEED_DISTRIBUTOR_DAILY - V_VIEWS_BY_VIDEO_COUNTRY_FEED_DISTRIBUTOR_DAILY JOINs: - MAPPINGS_VIDEO_TO_CHANNELS - MAPPINGS_VIDEO_TO_PRODUCTS (permissions subquery) - DIM_RELEASE (permissions subquery) Returns: JSON with video metrics per video ID. """ input_params = request.get_json().get("params") query_params = { "video_ids": input_params.get("video_ids", []), "country_ids": input_params.get("country_ids", []), "store_ids": list(map(int, input_params.get("store_ids", []))), "start_date": input_params.get("start_date"), "days": input_params.get("days"), } permissions = get_permission_values() return jsonify(video_metrics.get_video_metrics_bulk(query_params, permissions)) @app.route(config.ALL_TIME_VIDEO_METRICS_PATH) @access.verify_profile(access=ACCESS_ANALYTICS) def get_all_time_video_metrics(video_id): """Return all-time metrics for a given video. Route: GET /all-time-video-metrics/ Path params: video_id (str): Video ID. Query params: country_code (list[str]): Country codes to filter by. store_ids (list[int]): Store IDs to filter by. distributors (str): Comma-separated distributor names. Snowflake tables: - VIEWS_BY_VIDEO_FEED_DISTRIBUTOR_ROLLUP - VIEWS_BY_VIDEO_COUNTRY_FEED_DISTRIBUTOR_ROLLUP JOINs: - MAPPINGS_VIDEO_TO_CHANNELS - MAPPINGS_VIDEO_TO_PRODUCTS (permissions subquery) - DIM_RELEASE (permissions subquery) Returns: JSON with all-time video metrics. """ country_ids, store_ids, _, _, distributors = _get_global_filters() query_params = { "video_id": video_id, "country_ids": country_ids, "store_ids": store_ids, "distributors": distributors, } permissions = get_permission_values() return flaskify(video_metrics.get_all_time_video_metrics(query_params, permissions)) @app.route(config.CHANNEL_METRICS_PATH) @access.verify_profile(access=ACCESS_ANALYTICS) def get_channel_metrics(channel_id): """Return metrics for a given channel. Route: GET /channel-metrics/ Path params: channel_id (str): Channel ID. Query params: country_code (list[str]): Country codes to filter by. store_ids (list[int]): Store IDs to filter by. start_date (str): Start date YYYY-MM-DD. end_date (str): End date YYYY-MM-DD. distributors (str): Comma-separated distributor names. Snowflake tables: - VIEWS_BY_CHANNEL_FEED_DISTRIBUTOR_DAILY - VIEWS_BY_CHANNEL_COUNTRY_FEED_DISTRIBUTOR_DAILY JOINs: - ORCHARD_APP_REPORTING_V2.ART_RELATIONS_PROD_ART_RELATIONS.YOUTUBE_CHANNEL (permissions subquery) Returns: JSON with channel metrics and timeseries. """ country_ids, store_ids, start_date, end_date, distributors = _get_global_filters() if not (start_date and end_date): start_date, end_date = data_availability.get_date_range( data_availability.HIGHWATERMARK_DATE, days=28, downloads=False, videos=True, ) if start_date == ALL_TIME: start_date = None # if start_date and end_date are not str, convert to str in order to avoid conflict with fields.Date() if start_date is not None and not isinstance(start_date, str): start_date = start_date.strftime("%Y-%m-%d") if not isinstance(end_date, str): end_date = end_date.strftime("%Y-%m-%d") query_params = { "channel_id": channel_id, "country_ids": country_ids, "store_ids": list(map(int, store_ids)), "start_date": start_date, "end_date": end_date, "all_time": False, "distributors": distributors, } permissions = get_permission_values() return flaskify(channel_metrics.get_channel_metrics(query_params, permissions)) @app.route(config.ALL_TIME_CHANNEL_METRICS_PATH) @access.verify_profile(access=ACCESS_ANALYTICS) def get_all_time_channel_metrics(channel_id): """Return all-time metrics for a given channel. Route: GET /all-time-channel-metrics/ Path params: channel_id (str): Channel ID. Query params: country_code (list[str]): Country codes to filter by. store_ids (list[int]): Store IDs to filter by. distributors (str): Comma-separated distributor names. Snowflake tables: - VIEWS_BY_CHANNEL_FEED_DISTRIBUTOR_ROLLUP - VIEWS_BY_CHANNEL_COUNTRY_FEED_DISTRIBUTOR_ROLLUP JOINs: - ORCHARD_APP_REPORTING_V2.ART_RELATIONS_PROD_ART_RELATIONS.YOUTUBE_CHANNEL (permissions subquery) Returns: JSON with all-time channel metrics. """ country_ids, store_ids, start_date, end_date, distributors = _get_global_filters() query_params = { "channel_id": channel_id, "country_ids": country_ids, "store_ids": list(map(int, store_ids)), "all_time": True, "distributors": distributors, } permissions = get_permission_values() return flaskify(channel_metrics.get_channel_metrics(query_params, permissions)) @app.route(config.PRODUCT_METRICS_BY_TRACK_PATH) @access.verify_profile(access=ACCESS_ANALYTICS) def get_product_metrics_by_track(product_id): """Return product metrics by track. Route: GET /product//metrics-by-track Path params: product_id (str): Product ID. Query params: country_code (list[str]): Country codes to filter by. store_ids (list[int]): Store IDs to filter by. start_date (str): Start date YYYY-MM-DD. end_date (str): End date YYYY-MM-DD. distributors (str): Comma-separated distributor names. limit (int): Max results (default 100). offset (int): Pagination offset (default 0). order_by (str): Sort field (default "streams_7_days"). order_dir (str): Sort direction (default "DESC"). Snowflake tables: - METRICS_BY_TRACK_FEED_DISTRIBUTOR_ROLLUP - V_METRICS_BY_TRACK_COUNTRY_FEED_DISTRIBUTOR_ROLLUP JOINs: - DIM_TRACK Returns: JSON with per-track metrics for the product. """ countries, store_ids, _start_date, _end_date, distributors = _get_global_filters() query_params = { "product_id": product_id, "distributors": distributors, "country_ids": countries, "store_ids": store_ids, "order_by": request.args.get("order_by", "streams_7_days"), "order_dir": request.args.get("order_dir", "DESC").upper(), "limit": int(request.args.get("limit", 100)), "offset": int(request.args.get("offset", 0)), "transfer_product_ownership_enabled": is_insights_transfer_product_ownership_enabled(), } permissions = get_permission_values() return flaskify(product.get_metrics_by_track(query_params, permissions)) @app.route(config.SOUND_RECORDING_AGGREGATED_STREAMS_PATH) @access.verify_profile(access=ACCESS_ANALYTICS) def sound_recording_aggregated_streams(isrc): """Return aggregated streams for a sound recording. Route: GET /sound-recording//aggregated-streams Path params: isrc (str): ISRC identifier. 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, SOS_V2, 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_ISRC_COUNTRY_FEED_DISTRIBUTOR_ROLLUP - V_STREAMS_BY_TRACK_FEED_DISTRIBUTOR_DAILY - V_STREAMS_BY_TRACK_COUNTRY_FEED_DISTRIBUTOR_DAILY - DATA_AVAILABILITY_BY_STORE_DISTRIBUTOR_DAILY dimension=SOS or SOS_V2: - V_STREAMS_BY_TRACK_FEED_DISTRIBUTOR_DAILY - V_STREAMS_BY_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 = {"isrc": isrc} 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 = streams.get_aggregated_streams(query_params, permissions) response_body = {**aggregated_streams} return jsonify(response_body), 200 @app.route(config.HIGHWATERMARK_PATH) @access.verify_profile(access=ACCESS_ANALYTICS) def get_highwatermark(): """Return highwatermark for videos, downloads, or streams. Route: GET /highwatermark Query params: type (str): One of "videos", "streams", "downloads". Snowflake tables: - DATA_AVAILABILITY_BY_STORE_DAILY Returns: JSON with highwatermark date. """ highwatermark_type = request.args.get("type") if highwatermark_type not in ("videos", "streams", "downloads"): return flaskify(response.create_fatal_response("bad object_type")) return flaskify(highwatermark.get_highwatermark(highwatermark_type)) @app.route(config.STORES_PATH, methods=["POST"]) @access.verify_profile(access=ACCESS_ANALYTICS) def get_stores(): """Return store metadata for a list of store IDs. Route: POST /stores Body params: store_ids (list[int]): Store IDs to look up. Snowflake tables: - DATA_AVAILABILITY_BY_STORE_DISTRIBUTOR_DAILY JOINs: - DIM_FEED Returns: JSON with store metadata and outage info. """ data = request.get_json() store_ids = data.get("store_ids", []) response_body = stores.get_stores(store_ids) return jsonify(response_body), 200 @app.route(config.STORE_OUTAGES_PATH) @access.verify_profile(access=ACCESS_ANALYTICS) def get_store_outages(): """Return store outages currently being experienced. Route: GET /store-outages Snowflake tables: - DATA_AVAILABILITY_BY_STORE_DAILY JOINs: - DATA_AVAILABILITY_SKIPS_SAVES_BY_FEED_SUMMARY Returns: JSON with list of current store outages. """ return flaskify( response.Response({"store_outages": stores.add_outage_error_to_stores()}) ) @app.route(config.MARKET_RANKS_PATH) @access.verify_profile(access=ACCESS_ANALYTICS) def get_market_ranks(): """Return market ranks. Route: GET /market-ranks Query params: store_ids (list[int]): Store IDs to filter by. Snowflake tables: - MARKET_SIZE_BY_STORE_COUNTRY Returns: JSON with market size rankings by country. """ query_params = { "store_ids": request.args.getlist("store_ids"), "transfer_product_ownership_enabled": is_insights_transfer_product_ownership_enabled(), } permissions = get_permission_values() return flaskify(market_ranks.get_market_ranks(query_params, permissions)) @app.route(config.TADAS_TRENDS_PATH) @access.verify_profile(access=ACCESS_ANALYTICS) def get_tadas_trends(): """Return TADAS trends (employees only). Route: GET /tadas-trends Query params: markets (list[str]): Market codes to filter by. parent_company_uuids (list[str]): Parent company UUIDs. company_brand_uuids (list[str]): Company brand UUIDs. include_participants (list[str]): Participant IDs to include. exclude_participants (list[str]): Participant IDs to exclude. release_start_date (str): Release start date YYYY-MM-DD. release_end_date (str): Release end date YYYY-MM-DD. rep_owner (list[str]): Rep owner API enum names to filter by. limit (int): Max results (default 200). offset (int): Pagination offset (default 0). order_by (str): Sort field (default "TADAS_DAYS_TRENDING"). order_dir (str): Sort direction (default "asc"). Snowflake tables: - METRICS_TADAS - MAPPING_ISRC_TO_GLOBAL_PARTICIPANT Returns: JSON with TADAS trends and count. """ from datetime import datetime release_start_date = request.args.get("release_start_date") release_end_date = request.args.get("release_end_date") # Validate date format for each provided date start = None end = None if release_start_date: try: start = datetime.strptime(release_start_date, "%Y-%m-%d") except ValueError: return flaskify( response.create_fatal_response( "Invalid release_start_date format. Expected YYYY-MM-DD format." ) ) if release_end_date: try: end = datetime.strptime(release_end_date, "%Y-%m-%d") except ValueError: return flaskify( response.create_fatal_response( "Invalid release_end_date format. Expected YYYY-MM-DD format." ) ) # Validate that end_date is not earlier than start_date when both are provided if start and end and end < start: return flaskify( response.create_fatal_response( "release_end_date cannot be earlier than release_start_date" ) ) query_params = { "markets": request.args.getlist("markets"), "parent_company_uuids": request.args.getlist("parent_company_uuids"), "company_brand_uuids": request.args.getlist("company_brand_uuids"), "include_participants": request.args.getlist("include_participants"), "exclude_participants": request.args.getlist("exclude_participants"), "release_start_date": release_start_date, "release_end_date": release_end_date, "fin_label_parent_cds": request.args.getlist("fin_label_parent_cd"), "limit": request.args.get("limit", 200), "offset": request.args.get("offset", 0), "order_by": request.args.get("order_by", "TADAS_DAYS_TRENDING"), "order_dir": request.args.get("order_dir", "asc"), } permissions = get_permission_values() if user_has_full_access(permissions): # TADAS is for employees only return jsonify(tadas.get_tadas_trends(query_params, permissions)) else: return jsonify({"trends": [], "count": 0}) @app.route(config.TADAS_TREND_GLOBALSOUNDRECORDING_BY_ISRC_PATH) @access.verify_profile(access=ACCESS_ANALYTICS) def get_tadas_trend_by_isrc(isrc): """Return TADAS trend for a specific ISRC (employees only). Route: GET /tadas-trends/ Path params: isrc (str): ISRC identifier. Query params: markets (list[str]): Market codes to filter by. Snowflake tables: - METRICS_TADAS_HYBRID Returns: JSON with TADAS trend data for the ISRC. """ path_params = {"isrc": isrc} query_params = { **path_params, "markets": request.args.getlist("markets"), } permissions = get_permission_values() if user_has_full_access(permissions): # TADAS is for employees only return jsonify( tadas.get_tadas_trend_globalsoundrecording_by_isrc( query_params, permissions ) ) else: return jsonify({"trends": []}) @app.route(config.TADAS_DATA_AVAILABILITY_PATH, methods=["GET"]) def get_tadas_data_availability(): """Return TADAS data availability. Route: GET /tadas/data-availability Snowflake tables: - METRICS_TADAS Returns: JSON with TADAS data availability dates. """ result = tadas.get_tadas_data_availability({}, {}) return jsonify(result) @app.route(config.REP_OWNERS_PATH, methods=["GET"]) @access.verify_profile(access=ACCESS_ANALYTICS) def get_rep_owners(): """Return rep owners and the label ids that roll up to each (employees only). Route: GET /tadas/rep-owners Snowflake tables: - MAPPINGS_FINANCIAL_LABEL_PARENT_NAME_TO_LABEL_IDS Returns: JSON list of rep owners with code, name and label_ids. """ permissions = get_permission_values() if user_has_full_access(permissions): # rep owners are for employees only return jsonify(tadas.get_rep_owners({}, {})) else: return jsonify([])