import pytest from app.data import ISRCS from app.src.tests.insights.tests_common import markets_to_test, \ start_date, end_date, date_range from app.utils import get_insights_market, generate_excel_file QUERY = 'tiktokTimeseries' apollo_endpoint = "/dsp-api/delphi/tiktok/tracks/analytics?breakdowns=content_type_country%2Ctotals&metrics=creations%2Cvideo_views&isrc={0}&start_date={1}&end_date={2}&country_code={3}" @pytest.mark.parametrize("market", markets_to_test) @pytest.mark.parametrize("isrc", ISRCS) def test_tiktokTimeseries(isrc, market, graphql, api_apollo): variables = {"isrc": isrc, "type": "ALL", "startDate": start_date, "endDate": end_date, "countries": get_insights_market(market), "ids": []} response_insights = graphql.fetch_data(QUERY, variables) insights_items = response_insights['globalSoundRecordingByIsrc']['tiktok']['timeseries']['items'] insights_views = [item['views'] for item in insights_items] insights_creations = [item['creations'] for item in insights_items] response_apollo = api_apollo.get(apollo_endpoint.format(isrc, start_date, end_date, market)) apollo_breakdowns = response_apollo['breakdowns']['content_type_country'] apollo_views = [a + b for a, b in zip(apollo_breakdowns['pgc'][market]['video_views'], apollo_breakdowns['ugc'][market]['video_views'])] apollo_creations = [a + b for a, b in zip(apollo_breakdowns['pgc'][market]['creations'], apollo_breakdowns['ugc'][market]['creations'])] diff_views = [abs(int((b - a) / a * 100)) if a != 0 else None for a, b in zip(apollo_views, insights_views)] diff_creations = [abs(int((b - a) / a * 100)) if a != 0 else None for a, b in zip(apollo_creations, insights_creations)] data_dict = { "date": date_range, "apollo_views": apollo_views, "insights_views": insights_views, "change_views": diff_views, "apollo_creations": apollo_creations, "insights_creations": insights_creations, "change_creations": diff_creations, } generate_excel_file(data_dict, QUERY, [isrc, start_date, end_date, market])