import random from datetime import timedelta from typing import List from flask import jsonify from google.protobuf.json_format import MessageToDict from delphi_api.utils import DateUtils from delphi_api.v2.constants import AMAZON_MUSIC_COUNTRY_CODES from delphi_api.v2.proto.gen.amazon_v2_pb2 import AmazonTrackStreamStats from delphi_api.v2.views.response_faker import ResponseFaker class AmazonMusicViewFake: # pragma: no cover """Temporary class for consumers that generates fake data before the database is prepared. ToDo: Remove after database implemented """ @staticmethod def get_tracks_streams(isrc: List[str], start_date: str, end_date: str, country_code: str = None, group_by: str = None, limit: int = 10, offset: int = 0): # Don't generate more than 100, use limit param if less than max limit = min(limit, 100) countries = list([country_code]) if country_code else list(AMAZON_MUSIC_COUNTRY_CODES) fake_date = DateUtils.from_str('2019-01-01') items = [] for i in range(limit): parent = AmazonTrackStreamStats() message = ResponseFaker.populate_protobuf_models(parent) item = MessageToDict(message, preserving_proto_field_name=True) item.update({ 'country_code': random.choice(countries), 'isrc': random.choice(isrc), }) if group_by == 'date': date = DateUtils.as_str(fake_date + timedelta(days=i)) item.update({'date': date}) items.append(item) response_dict = { 'items': items, 'count': len(items), } return jsonify(response_dict) @staticmethod def get_playlist_tracks_streams(isrc: List[str], playlist_id: str, start_date: str, end_date: str, country_code: str = None, group_by: str = None, limit: int = 10, offset: int = 0): # Don't generate more than 100, use limit param if less than max limit = min(limit, 100) countries = list([country_code]) if country_code else list(AMAZON_MUSIC_COUNTRY_CODES) fake_date = DateUtils.from_str('2019-01-01') items = [] for i in range(limit): parent = AmazonTrackStreamStats() message = ResponseFaker.populate_protobuf_models(parent) item = MessageToDict(message, preserving_proto_field_name=True) item.update({ 'country_code': random.choice(countries), 'isrc': random.choice(isrc), 'playlist_id': playlist_id, }) if group_by == 'date': date = DateUtils.as_str(fake_date + timedelta(days=i)) item.update({'date': date}) items.append(item) response_dict = { 'items': items, 'count': len(items), } return jsonify(response_dict) @staticmethod def get_playlist_streams(playlist_id: List[str], start_date: str, end_date: str, country_code: str = None, group_by: str = None, limit: int = 10, offset: int = 0): # Don't generate more than 100, use limit param if less than max limit = min(limit, 100) countries = list([country_code]) if country_code else list(AMAZON_MUSIC_COUNTRY_CODES) fake_date = DateUtils.from_str('2019-01-01') items = [] for i in range(limit): parent = AmazonTrackStreamStats() message = ResponseFaker.populate_protobuf_models(parent) item = MessageToDict(message, preserving_proto_field_name=True) item.update({ 'country_code': random.choice(countries), 'playlist_id': random.choice(playlist_id), }) if group_by == 'date': date = DateUtils.as_str(fake_date + timedelta(days=i)) item.update({'date': date}) items.append(item) response_dict = { 'items': items, 'count': len(items), } return jsonify(response_dict)