import random from dataclasses import asdict from datetime import datetime, timedelta from typing import List import pytz from delphi_api.errors import Codes, InvalidInputError from delphi_api.utils import DateUtils from delphi_api.v3.constants import ( DUMMY_FOLLOWERS, DUMMY_FOLLOWERS_CHANGE, DUMMY_STREAMS, DUMMY_STREAMS_CHANGE, DUMMY_STREAMS_DSP_CHANGE_FACTOR, ) from delphi_api.v3.data_models.postgres_db import Artist, db from delphi_api.v3.data_models.schemas.demographics import ( AppleAgeBandSchema, GenderSchema, SpotifyAgeBandSchema, YouTubeAgeBandSchema, ) from delphi_api.v3.data_models.schemas.generics import ( FollowerSchema, PopularitySchema, StreamDemographicSchema, StreamSchema, ) from delphi_api.v3.enums import GroupByParam, TracklistType from delphi_api.v3.view_models.chart import ChartsViewModel from delphi_api.v3.view_models.params import Params from delphi_api.v3.view_models.playlist import PlaylistsViewModel from delphi_api.v3.view_models.query_builder import QueryBuilder from tests.conftest import TEST_ARTIST_IDS class FakeUtils: # pragma: no cover """ Temporary class to return simulated data in lieu of BigTable """ SCHEMAS = { 'streams': StreamSchema, 'spotify_popularity': PopularitySchema, } COUNTRY_CODES = ['us', 'gb', 'de', 'jp', 'nz'] @staticmethod def get_randint_noise(x, lo=6000, hi=20 * 1000) -> int: a = 1 b = 10 c = lo d = hi return a * x ** 2 + b * x * random.randint(c, d) @staticmethod def get_fake_temporal_data(params: Params, name='streams', count: int = DUMMY_STREAMS, change_amount: int = DUMMY_STREAMS_CHANGE, count_max: int = 0): """ This function generates most stream objects. Nested objects are mostly abstracted in other methods """ schema = FakeUtils.SCHEMAS.get(name) if params.include == 'demographics': schema = StreamDemographicSchema if not schema: raise InvalidInputError('missing schema definition') date_range = None if params.group_by == GroupByParam.DATE.value: date_range = DateUtils.date_range(params.start_date, params.end_date) if len(date_range): date_range.pop(-1) country_codes = FakeUtils.COUNTRY_CODES if name == 'streams' and params.country_code: country_codes = [code.lower() for code in params.country_code] playlist_id, playlist_ids = None, [] chart_id, chart_ids = None, [] if params.track_id: group, group_name = QueryBuilder.param_as_list(params.track_id), 'track_id' elif params.isrc: group, group_name = QueryBuilder.param_as_list(params.isrc), 'isrc' elif params.artist_id: group, group_name = QueryBuilder.param_as_list(params.artist_id), 'artist_id' elif params.playlist_id: group, group_name = QueryBuilder.param_as_list(params.playlist_id), 'playlist_id' elif params.chart_id: group, group_name = QueryBuilder.param_as_list(params.chart_id), 'chart_id' else: raise InvalidInputError({ 'code': Codes.invalid_input.value, 'description': 'Missing at least one identifiying parameter from set: ' '[track_id, isrc, artist_id, playlist_id, product_id]' }) dsps = QueryBuilder.param_as_list(params.dsp) if params.dsp else [None] if (params.track_id or params.isrc) and params.playlist_id: playlist_ids = QueryBuilder.param_as_list(params.playlist_id) playlist_id = playlist_ids[0] if (params.isrc or params.track_id) and params.chart_id: chart_ids = QueryBuilder.param_as_list(params.chart_id) chart_id = chart_ids[0] n = 1 schemas = [] for idx, dsp in enumerate(dsps): if idx > 0: n *= DUMMY_STREAMS_DSP_CHANGE_FACTOR count = FakeUtils.get_randint_noise(n) for country in country_codes: for id_param in group: schema_data = { group_name: id_param, # ex: 'streams' name: count, 'dsp': dsp } if params.artist_id: schema_data.update({'artist_id': params.artist_id}) if country: schema_data.update({'country_code': country}) if playlist_id: schema_data.update({ 'playlist_id': playlist_ids.pop() if playlist_ids else playlist_id, }) if chart_id: schema_data.update({ 'chart_id': chart_ids.pop() if chart_ids else chart_id, }) if date_range: for date in date_range: schema_data.update({'date': date, name: count}) if params.include == 'demographics': demographics = FakeUtils.get_fake_demographics(n, change_amount) schema_data.update(demographics) schemas.append(schema().load(schema_data)) count = FakeUtils.get_randint_noise(n) n += 1 if count_max and count > count_max: count = count_max else: schema_data.update({name: count}) if params.include == 'demographics': demographics = FakeUtils.get_fake_demographics(count, change_amount) schema_data.update(demographics) schemas.append(schema().load(schema_data)) count = FakeUtils.get_randint_noise(n) n += 1 if count_max and count > count_max: count = count_max return schemas @staticmethod def get_fake_artist_followers(params: Params, count: int = DUMMY_FOLLOWERS, change_amount: int = DUMMY_FOLLOWERS_CHANGE): date_range = DateUtils.date_range(params.start_date, params.end_date) def get_schema_data(base_count): base_count *= random.randrange(1, 20, 1) ints = random.sample(range(1, 20), k=10) return { 'date': None, 'apple_followers': base_count // ints.pop(), 'facebook_likes': base_count // ints.pop(), 'facebook_storytellers': base_count // ints.pop(), 'instagram_followers': base_count // ints.pop(), 'spotify_followers': base_count, 'twitter_followers': base_count // ints.pop(), 'youtube_channel_subscribers': base_count * ints.pop(), 'youtube_channel_views': base_count * ints.pop(), } schemas = [] for date in date_range: schema_data = get_schema_data(count) schema_data.update({'date': date}) schemas.append(FollowerSchema().load(schema_data)) count += change_amount return schemas @staticmethod def get_fake_demographics(count: int = DUMMY_STREAMS, change_amount=DUMMY_STREAMS_CHANGE) -> dict: n = 1 male = FakeUtils.get_randint_noise(n) female = FakeUtils.get_randint_noise(n) unknown = int(male + female * 0.02) neutral = unknown // 2 total = male + female + unknown data = {} dsp_fields = { 'apple': AppleAgeBandSchema._declared_fields.keys(), 'spotify': SpotifyAgeBandSchema._declared_fields.keys(), 'youtube': YouTubeAgeBandSchema._declared_fields.keys(), } for dsp, fields in dsp_fields.items(): data[dsp] = {} for field in fields: n += 1 if field.startswith('male'): all_field = field.replace('male', 'all') male = FakeUtils.get_randint_noise(n) data[dsp][field] = male data[dsp][all_field] = data[dsp].get(all_field, 0) data[dsp][all_field] += male total += male elif field.startswith('female'): all_field = field.replace('female', 'all') female = FakeUtils.get_randint_noise(n) data[dsp][field] = female data[dsp][all_field] = data[dsp].get(all_field, 0) data[dsp][all_field] += female total += female elif field.startswith('unknown'): all_field = field.replace('unknown', 'all') unknown = int(unknown * 1.02) data[dsp][field] = unknown data[dsp][all_field] = data[dsp].get(all_field, 0) data[dsp][all_field] += unknown total += unknown elif field.startswith('neutral'): all_field = field.replace('neutral', 'all') neutral = int(neutral * 1.02) data[dsp][field] = neutral data[dsp][all_field] = data[dsp].get(all_field, 0) data[dsp][all_field] += neutral total += neutral genders_data = { 'male': male, 'female': female, 'unknown': unknown, 'neutral': neutral, } return { 'streams': total, 'genders': GenderSchema().load(genders_data), 'apple_age_bands': AppleAgeBandSchema().load(data['apple']), 'spotify_age_bands': SpotifyAgeBandSchema().load(data['spotify']), 'youtube_age_bands': YouTubeAgeBandSchema().load(data['youtube']), } @staticmethod def get_charts_from_db(dsp: List[str], limit=0, params: Params = None): params_dict = asdict(params) if params else {'limit': limit, 'dsp': dsp} return ChartsViewModel.get_many(params_dict) @staticmethod def get_playlists_from_db(dsp: List[str], limit=0, params: Params = None): params_dict = asdict(params) if params else {'limit': limit, 'dsp': dsp} return PlaylistsViewModel.get_many(params_dict) @staticmethod def get_isrcs_from_db(artist_id, limit=10): artist = db.session.query(Artist).get(artist_id) tracks = artist.tracks_ext[:limit] return [track.isrc for track in tracks] @staticmethod def get_fake_track_positions(params: Params): max_items = 100 params.limit = max_items if params.limit > max_items else params.limit dsps = ['spotify', 'apple'] if not params.dsp else [params.dsp] items, item_id, item_name = [], '', '' if params.tracklist_type == TracklistType.CHARTS.value: item_id = 'chart_id' item_name = 'chart' items = FakeUtils.get_charts_from_db(dsps, limit=params.limit) elif params.tracklist_type == TracklistType.PLAYLISTS.value: item_id = 'playlist_id' item_name = 'playlist' items = FakeUtils.get_playlists_from_db(dsps, limit=params.limit) isrcs = params.isrc if not isrcs: isrcs = FakeUtils.get_isrcs_from_db(TEST_ARTIST_IDS[0]) floor = 5 ceil = 195 num_samples = len(isrcs) + 1 num_samples *= (len(items) + 1) if num_samples > ceil - floor: num_samples = ceil - floor ints = random.sample(range(floor, ceil), k=num_samples) def get_schema_data(pos, isrc, additional_data: dict): pos_change = random.randint(-4, 4) return { 'dsp': dsps[0], 'isrc': isrc, 'date': params.end_date, 'current': pos, 'previous': pos + pos_change, 'top': pos + pos_change, 'num_weeks_on': 16 + pos_change, **additional_data, } data = [] for isrc in isrcs: for item in items: if not len(ints): # short circuit return data addl_data = {item_id: item.get(item_id)} if params.include: addl_data.update({item_name: item}) data.append(get_schema_data(ints.pop(), isrc, addl_data)) return data @staticmethod def get_fake_data_statuses(): dsps = ['amazon', 'apple', 'spotify', 'youtube'] data = [] today = datetime.now(tz=pytz.utc) for dsp in dsps: offset = random.randint(0, 7) date = today - timedelta(days=offset) data.append({ 'dsp_id': dsp, 'updated_date': date, }) return data