import concurrent.futures import csv import os import time import json from concurrent.futures.thread import ThreadPoolExecutor from datetime import datetime, timedelta from io import StringIO from typing import List import boto3 import requests from flask import Blueprint, render_template, request, current_app, jsonify from werkzeug.wrappers import Response from apollo import Apollo from auth0 import Auth0 from db import DB from delphi import Delphi cache = {} # it's works properly with 1 uwsgi worker properly. s3 = boto3.resource('s3') bp = Blueprint('artist', __name__, url_prefix='/artist') IMAGE_SERVICE_HOST = "https://images-api.atlas.stream" def get_artist_list() -> List[str]: # Auth0.get_user(os.environ.get('AUTH0_USER_ID'))['app_metadata']['ama']['artistPermissions'] f = open("mapping.json", "r") data = json.load(f) ids = data.get("spotify_image_fallbacks", {}) return list(ids.keys()) @bp.route('//data_report', methods=('GET',)) def data_report(artist_id): res = requests.get(IMAGE_SERVICE_HOST + "/v2/artists/by_gras_participant_id/" + artist_id) image_exists = res.status_code == 200 # Examples: artist_data = Delphi.request(f"artists/GRAS_{artist_id}")[0] spotify_artist_id = artist_data["spotify_artist_id"].split(":")[-1] stream_data = Apollo.request(f"vendor-api/spotify/v1/artists/{spotify_artist_id}") report = { "GRAS_artist_id": artist_id, "artist_data": artist_data, "stream_data": stream_data, "image_exists": image_exists } return jsonify(report) @bp.route('/artists_names', methods=('GET',)) def get_artists_names(): import html artist_list = get_artist_list() artist_names = {} for artist in artist_list: artist_name, _ = Delphi.request(f"artists/GRAS_{artist}") artist_name = html.unescape(artist_name.get('full_name', '-')) artist_names[artist] = artist_name return jsonify(artist_names) @bp.route('/artists_images_report', methods=('GET',)) def missed_artists_images(): artist_list = get_artist_list() stats = [] step = 100 _from = 0 _to = _from + step while _from <= len(artist_list): scope = set(artist_list[_from:_to]) res = requests.get(IMAGE_SERVICE_HOST + "/v2/artists/by_gras_participant_ids/" + ",".join(scope)) keys = set(list(json.loads(res.text).keys())) for key in scope - keys: res = requests.get(IMAGE_SERVICE_HOST + "/v2/artists/by_gras_participant_id/" + key) stats.append([int(key), res.status_code]) for key in keys: stats.append([int(key), 200]) _from += step _to += step return jsonify(stats) @bp.route('/albums_images_report', methods=('GET',)) def missed_albums_imges(): artist_list = get_artist_list() artist_len = len(artist_list) stats = [] token = Delphi.get_token() i = 1 _from = int(request.args.get('from', default=0)) _to = int(request.args.get('to', default=0)) for data in artist_list[_from:_to]: payload = {'artist_id': "GRAS_" + data, 'group_by': 'isrc', 'sort_by': 'release_date', 'sort_order': 'desc'} tracks, _ = Delphi.request('tracks', payload, token) if not tracks or "items" not in tracks: print(tracks, flush=True) i += 1 continue products_set = set() for product in tracks['items']: products_set.add(product['product']['product_id'].replace('GRAS_', '')) if len(products_set) == 100: print(f"{i}/{artist_len}", f"artist_gras_id: {data}", f"albums_count: {len(products_set)}", flush=True) res = requests.get(IMAGE_SERVICE_HOST + "/v2/albums/by_gras_product_ids/" + ",".join(products_set)) try: keys = set(list(json.loads(res.text).keys())) except json.decoder.JSONDecodeError: keys = set() for key_404 in products_set - keys: stats.append([data, key_404, 404]) for key_200 in keys: stats.append([data, key_200, 200]) products_set = set() print(f"{i}/{artist_len}", f"artist_gras_id: {data}", f"albums_count: {len(products_set)}", flush=True) res = requests.get(IMAGE_SERVICE_HOST + "/v2/albums/by_gras_product_ids/" + ",".join(products_set)) keys_200 = set() if res.status_code == 200: keys_200 = set(list(json.loads(res.text).keys())) for key_404 in products_set - keys_200: stats.append([data, key_404, 404]) for key_200 in keys_200: stats.append([data, key_200, 200]) i += 1 return jsonify(stats) @bp.route('/', methods=('GET',)) def index(): artist_list = Auth0.get_user(os.environ.get('AUTH0_USER_ID'))['app_metadata']['ama']['artistPermissions'] cache_hash = f"artists-{os.environ.get('AUTH0_USER_ID')}" if cache_hash not in cache: token = Delphi.get_token() names = {} with concurrent.futures.ThreadPoolExecutor() as executor: futures = {} for data in artist_list: futures[data.replace('GRAS_', '')] = executor.submit( Delphi.request, current_app.config['DELPHI_API_URL'] + f"artists/{data}", None, token ) for artist_id, future in futures.items(): names[artist_id], _ = future.result() artists = [] for data in artist_list: artist_id = data.replace('GRAS_', '') artists.append({ # 'name': names[artist_id]["full_name"], 'name': 'Artist name', 'id': artist_id, 'img': f"https://images-api.atlas.stream/v2/artists/by_gras_participant_id/{artist_id}" }) cache[cache_hash] = artists return render_template('artist/index.html', artists=cache[cache_hash]) @bp.route('//top', methods=('GET', 'POST')) def top_countries(id): token = Delphi.get_token() selected_time = int(request.args.get('list_time', default='28')) end_date = datetime.today() - timedelta(days=1) start_date = end_date - timedelta(days=(selected_time - 1)) end_date_str = end_date.strftime("%Y-%m-%d") start_date_str = start_date.strftime("%Y-%m-%d") queries = {} # Artist artist, _ = DB.fetchone(""" SELECT * FROM common.dim_artists da WHERE da.artist_id = %s""", id) gras_id = 'GRAS_' + id artist_key = artist[0] # Top 5 top_five, queries['top_five'] = DB.fetchall(""" SELECT country_code, sum(num_streams) AS total_streams FROM spotify.fact_streams fs2 WHERE artist_key = %s AND report_date BETWEEN %s AND %s GROUP BY country_code ORDER BY total_streams DESC LIMIT 5 """, artist_key, start_date_str, end_date_str) # Get RDB for Spotify/Apple/Amazon with ThreadPoolExecutor() as executor: top_five_tuple = tuple([c[0] for c in top_five]) spotify_streams_rdb = executor.submit(DB.fetchall, """ SELECT country_code, sum(num_streams) AS total_streams FROM spotify.fact_streams fs2 WHERE artist_key = %s AND country_code IN %s AND report_date BETWEEN %s AND %s GROUP BY country_code""", artist_key, top_five_tuple, start_date_str, end_date_str) apple_streams_rdb = executor.submit(DB.fetchall, """ SELECT country_code, sum(num_streams) AS total_streams FROM apple.fact_streams fs2 WHERE artist_key = %s AND country_code IN %s AND report_date BETWEEN %s AND %s GROUP BY country_code""", artist_key, top_five_tuple, start_date_str, end_date_str) amazon_streams_rdb = executor.submit(DB.fetchall, """ SELECT country_code, sum(num_streams) AS total_streams FROM amazon.vw_fact_streams_all fs2 WHERE artist_key = %s AND country_code IN %s AND report_date BETWEEN %s AND %s GROUP BY country_code""", artist_key, top_five_tuple, start_date_str, end_date_str) futures = {} for country in top_five: country = country[0] payload = { 'start_date': start_date_str, 'end_date': end_date_str, 'artist_id': gras_id, 'agg_by': 'artist', 'country_code': country.lower(), 'dsp': None } futures[country] = {} for dsp in ['spotify', 'apple', 'amazon']: futures[country][dsp] = executor.submit(Delphi.request, 'streams', dict(payload, **{'dsp': dsp}), token) spotify_streams_rdb, queries['spotify_rdb'] = spotify_streams_rdb.result() apple_streams_rdb, queries['apple_rdb'] = apple_streams_rdb.result() amazon_streams_rdb, queries['amazon_rdb'] = amazon_streams_rdb.result() top_five_countries = [] for country in top_five: country = country[0] spotify_streams_delphi, queries['spotify_delphi_' + country] = futures[country]['spotify'].result() apple_streams_delphi, queries['apple_delphi_' + country] = futures[country]['apple'].result() amazon_streams_delphi, queries['amazon_delphi_' + country] = futures[country]['amazon'].result() top_five_countries.append([ country.lower(), dict(spotify_streams_rdb).get(country, 0), dict(apple_streams_rdb).get(country, 0), dict(amazon_streams_rdb).get(country, 0), spotify_streams_delphi["items"][0]["streams"], apple_streams_delphi["items"][0]["streams"], amazon_streams_delphi["items"][0]["streams"] ]) return render_template( 'artist/top_countries.html', id=id, artist=artist, top_five_countries=top_five_countries, selected_time=selected_time, start_date=start_date_str, end_date=end_date_str, queries=queries ) @bp.route('//country/', methods=('GET', 'POST')) def country(id, country_id): token = Delphi.get_token() selected_time = int(request.args.get('list_time', default='28')) end_date = datetime.today() - timedelta(days=1) start_date = end_date - timedelta(days=(selected_time - 1)) end_date_str = end_date.strftime("%Y-%m-%d") start_date_str = start_date.strftime("%Y-%m-%d") queries = {} # Artist artist, _ = DB.fetchone( "SELECT *" " FROM common.dim_artists da" " WHERE da.artist_id = %s", id ) gras_id = 'GRAS_' + id artist_key = artist[0] # Spotify spotify_streams_rdb, query = DB.fetchone( "SELECT s.report_date," " sum(s.num_streams)" " FROM spotify.fact_streams s" " WHERE s.artist_key=%s" " AND s.report_date BETWEEN %s AND %s" " AND s.country_code=%s" " AND s.content_type = \'A\'" " GROUP BY s.report_date" " ORDER BY s.report_date", artist_key, start_date.strftime("%Y-%m-%d"), end_date.strftime("%Y-%m-%d"), country_id.upper() ) queries['spotify_rdb_' + country_id] = query # Apple apple_streams_rdb, query = DB.fetchone( "SELECT s.report_date," " sum(s.num_streams)" " FROM apple.fact_streams s" " WHERE s.artist_key=%s" " AND s.report_date BETWEEN %s AND %s" " AND s.country_code=%s" " AND s.media_type_key = 1924" " GROUP BY s.report_date" " ORDER BY s.report_date", artist_key, start_date.strftime("%Y-%m-%d"), end_date.strftime("%Y-%m-%d"), country_id.upper() ) queries['apple_rdb_' + country_id] = query # Amazon amazon_streams_rdb, query = DB.fetchone( "SELECT s.report_date," " sum(s.num_streams)" " FROM amazon.vw_fact_streams_all s" " WHERE s.artist_key=%s" " AND s.report_date BETWEEN %s AND %s" " AND s.country_code=%s" " GROUP BY s.report_date" " ORDER BY s.report_date", artist_key, start_date.strftime("%Y-%m-%d"), end_date.strftime("%Y-%m-%d"), country_id.upper() ) queries['amazon_rdb_' + country_id] = query payload = { 'start_date': start_date_str, 'end_date': end_date_str, 'artist_id': gras_id, 'country_code': country_id, 'agg_by': 'artist', 'group_by': 'date', 'dsp': None } payload.update({'dsp': 'spotify'}) spotify_streams_delphi, queries['spotify_delphi_' + country_id] = Delphi.request('streams', payload, token) payload.update({'dsp': 'apple'}) apple_streams_delphi, queries['apple_delphi_' + country_id] = Delphi.request('streams', payload, token) payload.update({'dsp': 'amazon'}) amazon_streams_delphi, queries['amazon_delphi_' + country_id] = Delphi.request('streams', payload, token) # Build the results delta = timedelta(days=1) all_streams = [] current_date = start_date while current_date <= end_date: today = {"date": current_date.strftime("%Y-%m-%d")} for streams in spotify_streams_rdb: if streams[0].strftime("%Y-%m-%d") == current_date.strftime("%Y-%m-%d"): today["spotify_streams_rdb"] = streams[1] for streams in apple_streams_rdb: if streams[0].strftime("%Y-%m-%d") == current_date.strftime("%Y-%m-%d"): today["apple_streams_rdb"] = streams[1] for streams in amazon_streams_rdb: if streams[0].strftime("%Y-%m-%d") == current_date.strftime("%Y-%m-%d"): today["amazon_streams_rdb"] = streams[1] for streams in spotify_streams_delphi["items"]: if streams["date"] == current_date.strftime("%Y-%m-%d"): today["spotify_streams_delphi"] = streams["streams"] for streams in apple_streams_delphi["items"]: if streams["date"] == current_date.strftime("%Y-%m-%d"): today["apple_streams_delphi"] = streams["streams"] for streams in amazon_streams_delphi["items"]: if streams["date"] == current_date.strftime("%Y-%m-%d"): today["amazon_streams_delphi"] = streams["streams"] current_date += delta all_streams.append(today) return render_template( 'artist/country.html', id=id, country_id=country_id.upper(), artist=artist, all_streams=all_streams, selected_time=selected_time, start_date=start_date_str, end_date=end_date_str, queries=queries ) @bp.route('/track_performance', methods=('GET',)) def track_performance(): start = time.time() delphi_token = Delphi.get_token() isrc = str(request.args['isrc']) # track = Delphi.request(f'tracks', {'isrc': isrc}, delphi_token) artist = Delphi.request(f'artists', {'isrc': isrc}, delphi_token) tracks = Delphi.request(f'tracks', { 'artist_id': artist[0]["items"][0]["artist_id"], 'sort_by': 'release_date', }, delphi_token) isrcs = [] data = {} for track in tracks[0]["items"]: if track['release_date'] is None: continue isrcs.append(track) track_names = {} for track in isrcs[:30]: isrc = track['isrc'] track_names[isrc] = track['name'] payload = { 'start_date': ( datetime.strptime(track['release_date'], '%Y-%m-%d') - timedelta(days=7)).strftime( "%Y-%m-%d"), 'end_date': '2030-01-01', 'isrc': [isrc, ], 'group_by': 'date', 'sort_by': 'date', 'sort_order': 'asc', } _data = Delphi.request(f'streams', payload, delphi_token) if not len(_data[0]['items']): continue data[isrc] = {} results = {} for item in _data[0]['items']: if item['country_code'] not in results: results[item['country_code']] = {} results[item['country_code']][item['date']] = item['streams'] invalid_markets = [] for market in results: data[isrc][market] = {} i = 0 total = 0 _date = track['release_date'] for date in results[market]: i += 1 total += results[market][date] print(market, date, results[market][date]) if (datetime.strptime(date, '%Y-%m-%d') - datetime.strptime(_date, '%Y-%m-%d')).days > 1: invalid_markets.append([isrc, market]) break if i in (4, 12, 21): data[isrc][market][i] = total elif datetime.strptime(date, '%Y-%m-%d').isoweekday() == 5 and i <= 7: data[isrc][market]['F'] = total _date = date for isrc, market in invalid_markets: if isrc in data and market in data[isrc]: del data[isrc][market] for isrc in data: if market in data[isrc]: if len(data[isrc][market]) < 4: del data[isrc][market] return render_template('track_performance.html', report={ 'data': data, 'meta': { 'track_names': track_names, 'time': time.time() - start, } }) @bp.route('/data_analysis', methods=('GET',)) def data_analysis(): def get_sony_tracks(gras_artist_id, delphi_token): payload = {'artist_id': gras_artist_id, 'group_by': 'isrc'} tracks, _ = Delphi.request('tracks', payload, delphi_token) return [track['isrc'] for track in tracks.get('items', [])] start = time.time() delphi_token = Delphi.get_token() artist_list = cache.get('artist_list', []) if not artist_list: artist_list = Auth0.get_user(os.environ.get('AUTH0_USER_ID'))['app_metadata']['ama']['artistPermissions'] cache['artist_list'] = artist_list sony_tracks = cache.get('sony_tracks', []) if not sony_tracks: futures = {} with concurrent.futures.ThreadPoolExecutor(20) as executor: for gras_artist_id in artist_list: futures[gras_artist_id] = executor.submit( get_sony_tracks, gras_artist_id, delphi_token, ) for gras_artist_id, future in futures.items(): sony_tracks += future.result() cache['sony_tracks'] = sony_tracks end_date = datetime.today() - timedelta(days=1) start_date = end_date - timedelta(days=7) # Query params end_date = request.args.get('end_date', default=end_date.strftime("%Y-%m-%d")) start_date = request.args.get('start_date', default=start_date.strftime("%Y-%m-%d")) cache_key = f"data_{end_date}_{start_date}", data = cache.get(cache_key, {}) if not data and sony_tracks: tracks, _ = DB.fetchall( "SELECT product_key" " FROM common.dim_products dp" " WHERE isrc_cd IN %s", tuple(sony_tracks) ) tracks = tuple([track[0] for track in tracks]) _data, _ = DB.fetchall( " SELECT sum(fs2.num_streams), count(distinct fs2.product_key), fs2.country_code, 'SPOTIFY'" " FROM spotify.fact_streams fs2" " WHERE fs2.product_key IN %s" " AND fs2.report_date BETWEEN %s AND %s" " GROUP BY fs2.country_code" " UNION ALL" " SELECT sum(fs2.num_streams), count(distinct fs2.product_key), fs2.country_code, 'APPLE'" " FROM apple.fact_streams fs2" " WHERE fs2.product_key IN %s" " AND fs2.report_date BETWEEN %s AND %s" " GROUP BY fs2.country_code", tracks, start_date, end_date, tracks, start_date, end_date, ) _data = sorted(_data, key=lambda d: d[0] / d[1], reverse=True) for d in _data: if d[3] not in data: data[d[3]] = [] data[d[3]].append({ "streams": d[0], "tracks_amount": d[1], "market": d[2], "index": round(d[0] / d[1]), }) cache[cache_key] = data return render_template('data_analysis.html', report={ 'data': data, 'meta': { 'artist_count': len(artist_list), 'sony_tracks_count': len(sony_tracks), 'time': time.time() - start, 'start_date': start_date, 'end_date': end_date, } }) @bp.route('/report', methods=('GET',)) def report(): def calculate_position(country_code, date, position): _country_code = country_code if _country_code == "worldwide": _country_code = "global" url = f"spotifycharts/daily/regional/{date}/{_country_code}.csv" if url not in cache: from app import application with application.app_context(): bucket = s3.Bucket(current_app.config['CHARTS_BUCKET']) obj = bucket.Object(url) try: decoded_content = obj.get().get("Body").read().decode("utf-8") cr = csv.reader(decoded_content.splitlines(), delimiter=",") cr = list(cr) cr.pop(0) cr.pop(0) cache[url] = {int(i[0]): i for i in cr} except: pass track = cache.get(url, {}).get(position, (None, None, None, None, None,)) chart_position, track_name, _, streams, _ = track return track_name, streams, country_code, date def update_tracks_data(data, source, isrc, country_code, date, dsp, streams, tracks_meta): dsp = dsp.upper() source = source.upper() country_code = country_code.upper() if isrc not in data: data[isrc] = { 'details': tracks_meta[isrc], 'country_codes': {}, } if country_code not in data[isrc]['country_codes']: data[isrc]['country_codes'][country_code] = {} if date not in data[isrc]['country_codes'][country_code]: data[isrc]['country_codes'][country_code][date] = { 'AMAZONMUSICUNLIMITED': { 'DELPHI': None, 'RDB': None, }, 'AMAZONPRIME': { 'DELPHI': None, 'RDB': None, }, 'AMAZONADSUPPORTED': { 'DELPHI': None, 'RDB': None, }, 'APPLE': { 'DELPHI': None, 'RDB': None, }, 'SPOTIFY': { 'DELPHI': None, 'RDB': None, }, } data[isrc]['country_codes'][country_code][date][dsp][source] = streams delphi = data[isrc]['country_codes'][country_code][date][dsp].get('DELPHI', 0) rdb = data[isrc]['country_codes'][country_code][date][dsp].get('RDB', 0) if delphi and rdb: data[isrc]['country_codes'][country_code][date][dsp]['DELTA'] = round((rdb / delphi - 1) * 100, 2) return data def calculate_artist(gras_artist_id, days, top_tracks, countries, delphi_token): artists_result = {} artist_id = gras_artist_id.replace('GRAS_', '') # img =============================================================== img = f"https://images-api.atlas.stream/v2/artists/by_gras_participant_id/{artist_id}" r = requests.get(img) artists_result['image'] = { 'exists': r.status_code == 200, 'url': img } # social ============================================================= end_date = datetime.today() - timedelta(days=1) start_date = end_date - timedelta(days=(days - 1)) end_date = end_date.strftime("%Y-%m-%d") start_date = start_date.strftime("%Y-%m-%d") payload = { 'start_date': start_date, 'end_date': end_date, 'group_by': 'date', 'sort_by': 'date', 'sort_order': 'desc', } media_data = Delphi.request(f'artists/{gras_artist_id}/followers', payload, delphi_token)[0]['items'] artists_result['social'] = [{ 'date': item['date'], 'facebook_likes': item['facebook_likes'], 'twitter_followers': item['twitter_followers'], 'instagram_followers': item['instagram_followers'], } for item in media_data] # tracks ================================================================ payload = {'artist_id': gras_artist_id, 'group_by': 'isrc', 'sort_by': 'release_date', 'sort_order': 'desc'} tracks, _ = Delphi.request('tracks', payload, delphi_token) tracks = sorted(tracks['items'], key=lambda track: track["release_date"] if track["release_date"] else '', reverse=True) if not len(tracks): return artists_result tracks_meta = {track['isrc']: { 'name': track['name'], 'release_date': track['release_date'], } for track in tracks[:top_tracks]} tracks, _ = DB.fetchall( "SELECT product_key, product_name, suppl_title, isrc_cd" " FROM common.dim_products dp" " WHERE isrc_cd IN %s", tuple([track['isrc'] for track in tracks[:top_tracks]]) ) tracks_id = tuple([track[0] for track in tracks]) tracks_isrc = [track[3] for track in tracks] end_date = datetime.today() - timedelta(days=1) start_date = end_date - timedelta(days=(days - 1)) end_date = end_date.strftime("%Y-%m-%d") start_date = start_date.strftime("%Y-%m-%d") selected_countries = tuple([country_code.upper() for country_code in countries]) streams_rdb, _ = DB.fetchall( " SELECT fs2.report_date, count(fs2.num_streams ), fs2.product_key, fs2.country_code, 'SPOTIFY'" " FROM spotify.fact_streams fs2" " WHERE fs2.product_key IN %s" " AND fs2.country_code IN %s" " AND fs2.report_date BETWEEN %s AND %s" " GROUP BY fs2.report_date, fs2.product_key, fs2.country_code" " UNION ALL" " SELECT fs2.report_date, count(fs2.num_streams ), fs2.product_key, fs2.country_code, 'APPLE'" " FROM apple.fact_streams fs2" " WHERE fs2.product_key IN %s" " AND fs2.country_code IN %s" " AND fs2.report_date BETWEEN %s AND %s" " GROUP BY fs2.report_date, fs2.product_key, fs2.country_code" " UNION ALL" " SELECT fs2.report_date, count(fs2.num_streams ), fs2.product_key, fs2.country_code, 'AMAZONMUSICUNLIMITED'" " FROM amazon.fact_streams fs2" " WHERE fs2.product_key IN %s" " AND fs2.country_code IN %s" " AND fs2.report_date BETWEEN %s AND %s" " AND fs2.consumer_key != -1" " AND fs2.royalty_bearing_key = 170" " GROUP BY fs2.report_date, fs2.product_key, fs2.country_code" , tracks_id, selected_countries, start_date, end_date, tracks_id, selected_countries, start_date, end_date, tracks_id, selected_countries, start_date, end_date, ) payload = { 'start_date': start_date, 'end_date': end_date, 'isrc': tracks_isrc, 'country_code': [country_code.lower() for country_code in countries], 'agg_by': 'isrc', 'group_by': ['date', 'sub_dsp'], 'dsp': ['spotify', 'apple', 'amazon'], 'sort_by': 'date', 'sort_order': 'desc', } streams_delphi, _ = Delphi.request('streams', payload, delphi_token) tracks_data = {} tracks_mapping = {track[0]: track[3] for track in tracks} for sd in streams_delphi['items']: source, isrc, country_code, date, dsp, streams = 'DELPHI', sd['isrc'], sd['country_code'], sd['date'], \ sd['dsp'], sd['streams'] tracks_data = update_tracks_data(tracks_data, source, isrc, country_code, date, dsp, streams, tracks_meta) for sd in streams_rdb: source, isrc, country_code, date, dsp, streams = 'RDB', sd[2], sd[3], sd[0], sd[4], sd[1] date = date.strftime("%Y-%m-%d") isrc = tracks_mapping[isrc] tracks_data = update_tracks_data(tracks_data, source, isrc, country_code, date, dsp, streams, tracks_meta) artists_result['tracks'] = tracks_data # charts ============================================================================================ payload = { 'start_date': start_date, 'end_date': end_date, 'isrc': list(tracks_mapping.values()), 'include': 'charts', 'sort_by': 'date', 'sort_order': 'desc', 'dsp': 'spotify' } spotify_charts_delphi, _ = Delphi.request('track-positions/charts', payload, delphi_token) spotify_charts_delphi = spotify_charts_delphi['items'] charts = {} if len(spotify_charts_delphi) > 0: with concurrent.futures.ThreadPoolExecutor() as executor: for item in spotify_charts_delphi: if item['isrc'] not in charts: charts[item['isrc']] = {} if item['chart']['country_code'] not in charts[item['isrc']]: charts[item['isrc']][item['chart']['country_code']] = {} if item['date'] not in charts[item['isrc']][item['chart']['country_code']]: charts[item['isrc']][item['chart']['country_code']][item['date']] = { 'delphi_position': item['current'], 'delphi_track_name': None, 'streams': None, } futures = {executor.submit( calculate_position, item['chart']['country_code'], item['date'], item['current'] ): item['isrc'] for item in spotify_charts_delphi} for future in concurrent.futures.as_completed(futures): track_name, streams, country_code, date = future.result() track_isrc = futures[future] charts[track_isrc][country_code][date]['spotiify_track_name'] = track_name charts[track_isrc][country_code][date]['streams'] = streams artists_result['charts'] = charts return artists_result start = time.time() if 'artist_list' in cache: artist_list = cache['artist_list'] else: artist_list = Auth0.get_user(os.environ.get('AUTH0_USER_ID'))['app_metadata']['ama']['artistPermissions'] cache['artist_list'] = artist_list total_result = {} futures = {} delphi_token = Delphi.get_token() limit = int(request.args.get('limit', default='10')) offset = int(request.args.get('offset', default='0')) days = int(request.args.get('days', default='28')) top_tracks = int(request.args.get('top_tracks', default='15')) countries = request.args.get('countries', default='us,de,fr,gb,ca,au').split(',') is_simple_report = request.args.get('report_type', default='simple') == 'simple' # simple|full with concurrent.futures.ThreadPoolExecutor(20) as executor: for gras_artist_id in artist_list[offset:offset + limit]: futures[gras_artist_id] = executor.submit( calculate_artist, gras_artist_id, days, top_tracks, countries, delphi_token, ) for gras_artist_id, future in futures.items(): total_result[gras_artist_id] = future.result() return render_template('artist/report.html', report={ 'time': time.time() - start, 'artists': total_result, 'is_simple_report': is_simple_report, }) @bp.route('//tracks', methods=('GET', 'POST')) def tracks(id): token = Delphi.get_token() queries = {} # Artist artist, _ = DB.fetchone( "SELECT *" " FROM common.dim_artists da" " WHERE da.artist_id = %s", id ) gras_id = 'GRAS_' + id payload = {'artist_id': gras_id, 'group_by': 'isrc', 'sort_by': 'release_date', 'sort_order': 'desc'} tracks, queries['tracks'] = Delphi.request('tracks', payload, token) tracks = sorted(tracks['items'], key=lambda track: track["release_date"] if track["release_date"] else '', reverse=True) return render_template('artist/tracks.html', artist=artist, tracks=tracks, queries=queries) @bp.route('//tracks/', methods=('GET', 'POST')) def track(id, isrc): token = Delphi.get_token() selected_time = int(request.args.get('list_time', default='28')) selected_country = request.args.get('list_country', default='worldwide') end_date = datetime.today() - timedelta(days=1) start_date = end_date - timedelta(days=(selected_time - 1)) end_date_str = end_date.strftime("%Y-%m-%d") start_date_str = start_date.strftime("%Y-%m-%d") queries = {} all_streams = [] # Artist artist, _ = DB.fetchone( "SELECT *" " FROM common.dim_artists da" " WHERE da.artist_id = %s", id ) # Track track, _ = DB.fetchone( "SELECT product_key, product_name, suppl_title, isrc_cd" " FROM common.dim_products dp" " WHERE isrc_cd = %s", isrc ) # Spotify if selected_country == 'worldwide': spotify_streams_rdb, query = DB.fetchall( "SELECT fs2.report_date, count(fs2.num_streams )" " FROM spotify.fact_streams fs2" " WHERE product_key = %s" " AND fs2.report_date BETWEEN %s AND %s" " GROUP BY fs2.report_date" " ORDER BY fs2.report_date", track[0], start_date_str, end_date_str ) else: spotify_streams_rdb, query = DB.fetchall( "SELECT fs2.report_date, count(fs2.num_streams )" " FROM spotify.fact_streams fs2" " WHERE product_key = %s" " AND country_code = %s" " AND fs2.report_date BETWEEN %s AND %s" " GROUP BY fs2.report_date" " ORDER BY fs2.report_date", track[0], selected_country.upper(), start_date_str, end_date_str ) queries['spotify_streams_rdb'] = query # Apple if selected_country == 'worldwide': apple_streams_rdb, query = DB.fetchall( "SELECT fs2.report_date, count(fs2.num_streams )" " FROM apple.fact_streams fs2" " WHERE product_key = %s" " AND fs2.report_date BETWEEN %s AND %s" " GROUP BY fs2.report_date" " ORDER BY fs2.report_date", track[0], start_date_str, end_date_str ) else: apple_streams_rdb, query = DB.fetchall( "SELECT fs2.report_date, count(fs2.num_streams )" " FROM apple.fact_streams fs2" " WHERE product_key = %s" " AND country_code = %s" " AND fs2.report_date BETWEEN %s AND %s" " GROUP BY fs2.report_date" " ORDER BY fs2.report_date", track[0], selected_country.upper(), start_date_str, end_date_str ) queries['apple_streams_rdb'] = query # Amazon if selected_country == 'worldwide': amazon_streams_rdb, query = DB.fetchall( # '''select # report_date, # sum(streams) total # from # ( # select # report_date, count(num_streams) streams # from # amazon.fact_streams # WHERE # product_key = %s # AND report_date BETWEEN %s AND %s # group by # report_date # union all # select # report_date, sum(sub_streams) streams # from # ( # select # report_date, count(num_streams) sub_streams # from # amazon.fact_streams_free # WHERE # product_key = %s # AND report_date BETWEEN %s AND %s # group by # report_date # union all # select # report_date, count(num_streams) sub_streams # from # amazon.fact_streams_prime # WHERE # product_key = %s # AND report_date BETWEEN %s AND %s # group by # report_date ) s # group by # report_date ) t # group by # report_date''', (track[0], start_date_str, end_date_str, track[0], start_date_str, end_date_str, track[0], start_date_str, end_date_str) '''SELECT fs2.report_date, count(fs2.num_streams ) FROM amazon.vw_fact_streams_all fs2 WHERE product_key = %s AND fs2.report_date BETWEEN %s AND %s AND consumer_key != -1 AND royalty_bearing_key = 170 GROUP BY fs2.report_date ORDER BY fs2.report_date''', track[0], start_date_str, end_date_str ) else: amazon_streams_rdb, query = DB.fetchall( "SELECT fs2.report_date, count(fs2.num_streams )" " FROM amazon.vw_fact_streams_all fs2" " WHERE product_key = %s" " AND country_code = %s" " AND fs2.report_date BETWEEN %s AND %s" " AND consumer_key != -1" " AND royalty_bearing_key = 170" " GROUP BY fs2.report_date" " ORDER BY fs2.report_date", track[0], selected_country.upper(), start_date_str, end_date_str ) queries['amazon_streams_rdb'] = query payload = { 'start_date': start_date_str, 'end_date': end_date_str, 'isrc': isrc, 'country_code': selected_country.lower(), 'agg_by': 'isrc', 'group_by': 'date', 'dsp': None, } payload.update({'dsp': 'spotify'}) spotify_streams_delphi, queries['spotify_streams_delphi'] = Delphi.request('streams', payload, token) payload.update({'dsp': 'apple'}) apple_streams_delphi, queries['apple_streams_delphi'] = Delphi.request('streams', payload, token) payload.update({'dsp': 'amazon'}) amazon_streams_delphi, queries['amazon_streams_delphi'] = Delphi.request('streams', payload, token) # Build the results delta = timedelta(days=1) current_date = start_date while current_date <= end_date: today = {} today["date"] = current_date.strftime("%Y-%m-%d") for streams in spotify_streams_rdb: if streams[0].strftime("%Y-%m-%d") == current_date.strftime("%Y-%m-%d"): today["spotify_streams_rdb"] = streams[1] for streams in apple_streams_rdb: if streams[0].strftime("%Y-%m-%d") == current_date.strftime("%Y-%m-%d"): today["apple_streams_rdb"] = streams[1] for streams in amazon_streams_rdb: if streams[0].strftime("%Y-%m-%d") == current_date.strftime("%Y-%m-%d"): today["amazon_streams_rdb"] = streams[1] for streams in spotify_streams_delphi["items"]: if streams["date"] == current_date.strftime("%Y-%m-%d"): today["spotify_streams_delphi"] = streams["streams"] for streams in apple_streams_delphi["items"]: if streams["date"] == current_date.strftime("%Y-%m-%d"): today["apple_streams_delphi"] = streams["streams"] for streams in amazon_streams_delphi["items"]: if streams["date"] == current_date.strftime("%Y-%m-%d"): today["amazon_streams_delphi"] = streams["streams"] current_date += delta all_streams.append(today) def generate_csv(): data = StringIO() w = csv.writer(data) # write header w.writerow((artist[2], track[1], track[3])) w.writerow( ('date', 'Spotify RDB', 'Spotify Delphi', 'Apple RDB', 'Apple Delphi', 'Amazon RDB', 'Amazon Delphi')) yield data.getvalue() data.seek(0) data.truncate(0) # write each stream item for streams in all_streams: w.writerow(( streams["date"] if ("date" in streams) else "", streams["spotify_streams_rdb"] if ("spotify_streams_rdb" in streams) else "", streams["spotify_streams_delphi"] if ("spotify_streams_delphi" in streams) else "", round((((streams["spotify_streams_rdb"] / streams["spotify_streams_delphi"]) - 1) * 100), 2) if ( "spotify_streams_rdb" in streams and "spotify_streams_delphi" in streams) else "", streams["apple_streams_rdb"] if ("apple_streams_rdb" in streams) else "", streams["apple_streams_delphi"] if ("apple_streams_delphi" in streams) else "", round((((streams["apple_streams_rdb"] / streams["apple_streams_delphi"]) - 1) * 100), 2) if ( "apple_streams_rdb" in streams and "apple_streams_delphi" in streams) else "", streams["amazon_streams_rdb"] if ("amazon_streams_rdb" in streams) else "", streams["amazon_streams_delphi"] if ("amazon_streams_delphi" in streams) else "", round((((streams["amazon_streams_rdb"] / streams["amazon_streams_delphi"]) - 1) * 100), 2) if ( "amazon_streams_rdb" in streams and "amazon_streams_delphi" in streams) else "", )) yield data.getvalue() data.seek(0) data.truncate(0) w.writerow(('')) # write each stream item for query in queries.items(): w.writerow(( query[0], query[1], )) yield data.getvalue() data.seek(0) data.truncate(0) if request.method == 'GET': _countries, _ = Delphi.request('regions') countries = [] for country in _countries['items']: if country['country_code'] in ['us', 'de', 'fr', 'ca', 'au', 'gb']: countries.append(country) return render_template( 'artist/track.html', artist=artist, track=track, countries=countries, all_streams=all_streams, selected_time=selected_time, selected_country=selected_country, start_date=start_date_str, end_date=end_date_str, queries=queries ) if request.method == 'POST': response = Response(generate_csv(), mimetype='text/csv') # add a filename response.headers.set("Content-Disposition", "attachment", filename="log.csv") return response @bp.route('//tracks//charts', methods=('GET', 'POST')) def charts(id, isrc): token = Delphi.get_token() selected_time = int(request.args.get('list_time', default='28')) end_date = datetime.today() - timedelta(days=1) start_date = end_date - timedelta(days=(selected_time - 1)) end_date_str = end_date.strftime("%Y-%m-%d") start_date_str = start_date.strftime("%Y-%m-%d") queries = {} # Get the artist artist, _ = DB.fetchone( "SELECT *" " FROM common.dim_artists da" " WHERE da.artist_id = %s", id ) # Track track, _ = DB.fetchone( "SELECT product_key, product_name, suppl_title, isrc_cd" " FROM common.dim_products dp" " WHERE isrc_cd = %s", isrc ) payload = { 'start_date': start_date_str, 'end_date': end_date_str, 'isrc': isrc, 'include': 'charts', 'sort_by': 'date', 'sort_order': 'desc', 'dsp': None, } payload.update({'dsp': 'spotify'}) spotify_charts_delphi, queries['spotify_charts_delphi'] = Delphi.request('track-positions/charts', payload, token) spotify_charts_delphi = spotify_charts_delphi['items'] def calculate_position(country_code, date, position): _country_code = country_code if _country_code == "worldwide": _country_code = "global" url = f"https://spotifycharts.com/regional/{_country_code}/daily/{date}/download" if url not in cache: download = requests.get(url) decoded_content = download.content.decode('utf-8') cr = csv.reader(decoded_content.splitlines(), delimiter=',') cache[url] = list(cr) track = cache[url][position + 1] chart_position, track_name, _, streams, _ = track return track_name, streams, country_code, date, spotify_charts_original = {} if len(spotify_charts_delphi) > 0: with concurrent.futures.ThreadPoolExecutor() as executor: futures = {executor.submit( calculate_position, item['chart']['country_code'], item['date'], item['current'] ): item for item in spotify_charts_delphi} for future in concurrent.futures.as_completed(futures): track_name, streams, country_code, date = future.result() if date not in spotify_charts_original: spotify_charts_original[date] = {} spotify_charts_original[date][country_code] = { 'track_name': track_name, 'streams': streams, } payload.update({'dsp': 'apple'}) apple_charts_delphi, queries['apple_charts_delphi'] = Delphi.request('track-positions/charts', payload, token) apple_charts_delphi = apple_charts_delphi['items'] return render_template( 'artist/charts.html', artist=artist, track=track, spotify_charts_delphi=spotify_charts_delphi, spotify_charts_original=spotify_charts_original, apple_charts_delphi=apple_charts_delphi, start_date=start_date_str, end_date=end_date_str, queries=queries ) @bp.route('//tracks//amazon', methods=('GET', 'POST')) def amazon(id, isrc): token = Delphi.get_token() selected_time = int(request.args.get('list_time', default='28')) selected_country = request.args.get('list_country', default='worldwide') end_date = datetime.today() - timedelta(days=1) start_date = end_date - timedelta(days=(selected_time - 1)) end_date_str = end_date.strftime("%Y-%m-%d") start_date_str = start_date.strftime("%Y-%m-%d") queries = {} all_streams = [] # Artist artist, _ = DB.fetchone( "SELECT *" " FROM common.dim_artists da" " WHERE da.artist_id = %s", id ) # Track track, _ = DB.fetchone( "SELECT product_key, product_name, suppl_title, isrc_cd" " FROM common.dim_products dp" " WHERE isrc_cd = %s", isrc ) # Amazon Total if selected_country == 'worldwide': amazon_all_rdb, query = DB.fetchall( "SELECT fs2.report_date, count(fs2.num_streams )" " FROM amazon.vw_fact_streams_all fs2" " WHERE product_key = %s" " AND fs2.report_date BETWEEN %s AND %s" " GROUP BY fs2.report_date" " ORDER BY fs2.report_date", track[0], start_date_str, end_date_str ) else: amazon_all_rdb, query = DB.fetchall( "SELECT fs2.report_date, count(fs2.num_streams )" " FROM amazon.vw_fact_streams_all fs2" " WHERE product_key = %s" " AND country_code = %s" " AND fs2.report_date BETWEEN %s AND %s" " GROUP BY fs2.report_date" " ORDER BY fs2.report_date", track[0], selected_country.upper(), start_date_str, end_date_str ) queries['amazon_all_rdb'] = query # Amazon Unlimited if selected_country == 'worldwide': amazon_unlimited_rdb, query = DB.fetchall( "SELECT fs2.report_date, count(fs2.num_streams )" " FROM amazon.fact_streams fs2" " WHERE product_key = %s" " AND fs2.report_date BETWEEN %s AND %s" " GROUP BY fs2.report_date" " ORDER BY fs2.report_date", track[0], start_date_str, end_date_str ) else: amazon_unlimited_rdb, query = DB.fetchall( "SELECT fs2.report_date, count(fs2.num_streams )" " FROM amazon.fact_streams fs2" " WHERE product_key = %s" " AND country_code = %s" " AND fs2.report_date BETWEEN %s AND %s" " GROUP BY fs2.report_date" " ORDER BY fs2.report_date", track[0], selected_country.upper(), start_date_str, end_date_str ) queries['amazon_unlimited_rdb'] = query # Amazon Free if selected_country == 'worldwide': amazon_free_rdb, query = DB.fetchall( "SELECT fs2.report_date, count(fs2.num_streams )" " FROM amazon.fact_streams_free fs2" " WHERE product_key = %s" " AND fs2.report_date BETWEEN %s AND %s" " GROUP BY fs2.report_date" " ORDER BY fs2.report_date", track[0], start_date_str, end_date_str ) else: amazon_free_rdb, query = DB.fetchall( "SELECT fs2.report_date, count(fs2.num_streams )" " FROM amazon.fact_streams_free fs2" " WHERE product_key = %s" " AND country_code = %s" " AND fs2.report_date BETWEEN %s AND %s" " GROUP BY fs2.report_date" " ORDER BY fs2.report_date", track[0], selected_country.upper(), start_date_str, end_date_str ) queries['amazon_free_rdb'] = query # Amazon Prime if selected_country == 'worldwide': amazon_prime_rdb, query = DB.fetchall( "SELECT fs2.report_date, count(fs2.num_streams )" " FROM amazon.fact_streams_prime fs2" " WHERE product_key = %s" " AND fs2.report_date BETWEEN %s AND %s" " GROUP BY fs2.report_date" " ORDER BY fs2.report_date", track[0], start_date_str, end_date_str ) else: amazon_prime_rdb, query = DB.fetchall( "SELECT fs2.report_date, count(fs2.num_streams )" " FROM amazon.fact_streams_prime fs2" " WHERE product_key = %s" " AND country_code = %s" " AND fs2.report_date BETWEEN %s AND %s" " GROUP BY fs2.report_date" " ORDER BY fs2.report_date", track[0], selected_country.upper(), start_date_str, end_date_str ) queries['amazon_prime_rdb'] = query payload = {'start_date': start_date_str, 'end_date': end_date_str, 'isrc': isrc, 'dsp': 'amazon', 'country_code': selected_country.lower(), 'agg_by': 'isrc', 'group_by': 'date'} amazon_streams_delphi, queries['amazon_streams_delphi'] = Delphi.request('streams', payload, token) # Build the results delta = timedelta(days=1) current_date = start_date while current_date <= end_date: today = {"date": current_date.strftime("%Y-%m-%d")} for streams in amazon_all_rdb: if streams[0].strftime("%Y-%m-%d") == current_date.strftime("%Y-%m-%d"): today["amazon_all_rdb"] = streams[1] for streams in amazon_unlimited_rdb: if streams[0].strftime("%Y-%m-%d") == current_date.strftime("%Y-%m-%d"): today["amazon_unlimited_rdb"] = streams[1] for streams in amazon_free_rdb: if streams[0].strftime("%Y-%m-%d") == current_date.strftime("%Y-%m-%d"): today["amazon_free_rdb"] = streams[1] for streams in amazon_prime_rdb: if streams[0].strftime("%Y-%m-%d") == current_date.strftime("%Y-%m-%d"): today["amazon_prime_rdb"] = streams[1] for streams in amazon_streams_delphi["items"]: if streams["date"] == current_date.strftime("%Y-%m-%d"): today["amazon_streams_delphi"] = streams["streams"] current_date += delta all_streams.append(today) countries, _ = Delphi.request('regions') return render_template( 'artist/track_amazon.html', artist=artist, track=track, countries=countries['items'], all_streams=all_streams, selected_time=selected_time, selected_country=selected_country, start_date=start_date_str, end_date=end_date_str, queries=queries ) @bp.route('//tracks//', methods=('GET', 'POST')) def date(id, isrc, date): token = Delphi.get_token() queries = {} # Artist artist, _ = DB.fetchone( "SELECT *" " FROM common.dim_artists da" " WHERE da.artist_id = %s", id ) # Track track, _ = DB.fetchone( "SELECT product_key, product_name, suppl_title, isrc_cd" " FROM common.dim_products dp" " WHERE isrc_cd = %s", isrc ) countries = Delphi.request('regions') all_streams = [] for country in countries['items']: country_result = {'country': country['country_code']} spotify_streams_rdb, query = DB.fetchall( "SELECT fs2.country_code , count(fs2.num_streams )" " FROM spotify.fact_streams fs2" " WHERE product_key = %s" " AND fs2.report_date BETWEEN %s AND %s" " AND fs2.country_code = %s" " GROUP BY fs2.country_code" " ORDER BY fs2.country_code", track[0], date, date, country['country_code'].upper() ) if spotify_streams_rdb: country_result['spotify_streams_rdb'] = spotify_streams_rdb[1] queries['spotify_streams_rdb'] = query apple_streams_rdb, query = DB.fetchall( "SELECT fs2.country_code , count(fs2.num_streams )" " FROM apple.fact_streams fs2" " WHERE product_key = %s" " AND fs2.report_date BETWEEN %s AND %s" " AND fs2.country_code = %s" " GROUP BY fs2.country_code" " ORDER BY fs2.country_code", track[0], date, date, country['country_code'].upper() ) if apple_streams_rdb: country_result['apple_streams_rdb'] = apple_streams_rdb[1] queries['apple_streams_rdb'] = query amazon_streams_rdb, query = DB.fetchall( "SELECT fs2.country_code , count(fs2.num_streams )" " FROM amazon.vw_fact_streams_all fs2" " WHERE product_key = %s" " AND fs2.report_date BETWEEN %s AND %s" " AND fs2.country_code = %s" " GROUP BY fs2.country_code" " ORDER BY fs2.country_code", track[0], date, date, country['country_code'].upper() ) if amazon_streams_rdb: country_result['amazon_streams_rdb'] = amazon_streams_rdb[1] queries['amazon_streams_rdb'] = query payload = {'start_date': date, 'end_date': date, 'isrc': isrc, 'country_code': country['country_code'].lower(), 'agg_by': 'isrc', 'group_by': 'date', 'dsp': None, } payload.update({'dsp': 'spotify'}) items, queries['spotify_streams_delphi_' + country['country_code'].lower()] = Delphi.request('streams', payload, token) items = items['items'] if items: country_result['spotify_streams_delphi'] = items[0]['streams'] payload.update({'dsp': 'apple'}) items, queries['apple_streams_delphi_' + country['country_code'].lower()] = Delphi.request('streams', payload, token) items = items['items'] if items: country_result['apple_streams_delphi'] = items[0]['streams'] payload.update({'dsp': 'amazon'}) items, queries['amazon_streams_delphi_' + country['country_code'].lower()] = Delphi.request('streams', payload, token) items = items['items'] if items: country_result['amazon_streams_delphi'] = items[0]['streams'] all_streams.append(country_result) # if i > 15: # break # i+=1 return render_template( 'artist/track_day.html', artist=artist, track=track, all_streams=all_streams, start_date=date, queries=queries ) @bp.route('//social_media', methods=('GET', 'POST')) def social(id): token = Delphi.get_token() selected_time = 10 end_date = datetime.today() - timedelta(days=1) start_date = end_date - timedelta(days=(selected_time - 1)) end_date_str = end_date.strftime("%Y-%m-%d") start_date_str = start_date.strftime("%Y-%m-%d") payload = { 'start_date': start_date_str, 'end_date': end_date_str, 'group_by': 'date' } media_data = Delphi.request('artists/GRAS_{}/followers'.format(id), payload, token)[0]['items'] dates = [] fb_likes = [] tw_followers = [] insta_followers = [] for item in media_data: dates.append(item['date']) fb_likes.append(item['facebook_likes']) tw_followers.append(item['twitter_followers']) insta_followers.append(str(item['instagram_followers'])) return render_template('artist/social.html', date=dates, facebook_likes=fb_likes, instagram_followers=insta_followers, twitter_followers=tw_followers)