"""Spotify scrapping flow.""" import datetime import json import logging import os import re from bs4 import BeautifulSoup import config from fangare import catchers from lxml import etree import requests from requests.exceptions import ConnectionError from requests.exceptions import ReadTimeout from selenium import webdriver from snowflake_executor import SpotifyStatsSFExecutor import spotipy from spotipy import SpotifyException from spotipy.oauth2 import SpotifyClientCredentials catchers.init_sentry(sentry_dsn=os.getenv('SENTRY_DSN')) def today(): """Return today's date.""" return datetime.date.today() def zero_if_none(value): """Return zero if None.""" return value if value is not None else 0 def get_id_from_link(text): # text = config.artists_to_track.get(artist) """Get ID from Spotify profile link.""" return text. \ replace('/artist/', ''). \ replace(config.SPOTIFY_LINK, ''). \ rsplit('?', 1)[0] def get_playlists_ids(uri): """Get playlist IDs using Access Token.""" client = requests.session() response = client.get(config.SPOTIFY_LINK, headers=config.request_headers) token = re.search('"accessToken":\s*"([^"]+)', response.text) bearer = token.group(1) config.request_headers['Authorization'] = 'Bearer {}'. \ format(bearer) variables_params = json.dumps({'uri': 'spotify:artist:{}'. format(uri)}) extensions_params = json.dumps({ 'persistedQuery': {'version': 1, 'sha256Hash': config.SHA256HASH}}) params = { 'operationName': 'queryArtistDiscoveredOn', 'variables': variables_params, 'extensions': extensions_params, } discovered_on = client.get( 'https://api-partner.spotify.com/pathfinder/v1/query', params=params, headers=config.request_headers) discovered_on_url = discovered_on.url.replace('+', ''). \ replace('%5D', '').replace('%2Fartist%2F', '') # correct url response = client.get(discovered_on_url, headers=config.request_headers) pl_ids = [] try: playlists = json.loads(response.text).get('data').get('artist'). \ get('relatedContent').get('discoveredOn').get('items') for playlist in playlists: pl_ids.append(playlist.get('id')) except AttributeError: logging.warning('Some errors on API side have occurred' ' while trying to reach playlists.') return pl_ids def if_first_time(stat, first_time): """Return stat depending onn first_time bool value.""" return stat if not first_time else 0 def abbreviated_value_to_int(value: str): """Convert values like 4M, 35.6K to int.""" try: if 'M' in value: followers = int(float( value.replace(',', '.').rstrip('M')) * 1000000) elif 'K' in value: followers = int(float( value.replace(',', '.').rstrip('K')) * 1000) else: followers = int(value) return followers except: # noqa logging.warning('Failed to parse and convert ' 'the listeners number: %s', value) return None @catchers.catch_all_and_print(sentry_dsn=os.getenv('SENTRY_DSN')) def set_webdriver(): """Set class' webdriver argument.""" options = webdriver.ChromeOptions() options.add_argument('headless') options.add_argument('--disable-dev-shm-usage') options.add_argument('--no-sandbox') options.add_argument('--single-process') options.add_argument('--disable-extensions') options.add_argument('--disable-gpu') options.add_argument('disable-infobars') options.add_argument('start-maximized') driver = webdriver.Chrome( '/spotify-statistics/chromedriver', options=options) return driver class SpotifyTracker: """Spotify Tracker class for scrapping data.""" def __init__(self): """Spotify tracker attributes.""" logging.basicConfig(level=logging.INFO) self.__client_credentials_manager__ = \ SpotifyClientCredentials( client_id=config.spotify.get('client_id'), client_secret=config.spotify.get('client_secret')) self.__spotify__ = spotipy. \ Spotify(client_credentials_manager=self. __client_credentials_manager__) self.__executor__ = SpotifyStatsSFExecutor(sf_config=config.SF_CONFIG) self.__driver__ = set_webdriver() self.__dom__ = None self.__profiles__ = dict() self.__last_week_stats__ = dict() self.__first_time__ = False def get_profiles(self): """Get profiles.""" return self.__profiles__ def get_last_week_stats(self): """Get last week stats.""" return self.__last_week_stats__ def get_executor(self): """Get SnowFlake executor instance.""" return self.__executor__ def get_driver(self): """Return Selenium webdriver.""" return self.__driver__ def get_dom(self): """Return current page dom.""" return self.__dom__ def set_dom(self, dom): """Set current page dom.""" self.__dom__ = dom def get_spotify(self): """Get Spotify API instance.""" return self.__spotify__ def set_last_week_stats(self): """Set last week stats.""" self.__last_week_stats__.update( dict((artist_name, {'id': spotify_link, 'current_monthly_listeners': abbreviated_value_to_int(monthly_listeners), 'last_week_monthly_listeners': abbreviated_value_to_int( last_week_monthly_listeners), 'weekly_change_in_monthly_listeners': abbreviated_value_to_int( weekly_change_in_monthly_listeners), 'percentage_change_in_monthly_listeners': percentage_change_in_monthly_listeners, 'current_followers': current_followers_count, 'last_week_followers': last_week_follower_count, 'weekly_change_in_followers': weekly_change_in_followers, 'percentage_change_in_followers': percentage_change_in_followers, 'current_popularity': spotify_popularity, 'last_week_popularity': last_week_spotify_popularity, 'editorial_playlists_adds': editorial_playlists_adds, 'editorial_playlists_adds_count': editorial_playlists_adds_count, 'last_week_editorial_playlists_adds': last_week_editorial_playlists_adds, 'last_week_editorial_playlists_adds_count': last_week_editorial_playlists_adds_count, 'notable_playlists_adds': notable_playlists_adds, 'notable_playlists_adds_count': notable_playlists_adds_count, 'last_week_notable_playlists_adds': last_week_notable_playlists_adds, 'last_week_notable_playlists_adds_count': last_week_notable_playlists_adds_count, 'the_latest_release': the_latest_release, 'last_processing_date': last_processing_date, 'last_week_processing_date': last_week_processing_date}) for artist_name, spotify_link, monthly_listeners, last_week_monthly_listeners, weekly_change_in_monthly_listeners, percentage_change_in_monthly_listeners, current_followers_count, last_week_follower_count, weekly_change_in_followers, percentage_change_in_followers, spotify_popularity, last_week_spotify_popularity, editorial_playlists_adds, editorial_playlists_adds_count, last_week_editorial_playlists_adds, last_week_editorial_playlists_adds_count, notable_playlists_adds, notable_playlists_adds_count, last_week_notable_playlists_adds, last_week_notable_playlists_adds_count, the_latest_release, last_processing_date, last_week_processing_date in self.__executor__.select_artists_stats())) def create_tables(self): """Create tables if not exist and read last week stats.""" exists = ' table already exists.' executor = self.get_executor() self.__profiles__.update(dict((artist, dict(id=get_id_from_link(link))) for artist, link in executor.select_artists_links() if link is not None)) query_result = executor.create_spotify_statistics() if 'already exists' in query_result[0]: logging.info('%s%s', config.snowflake_table_names[ 'spotify_statistics'].upper(), exists) self.set_last_week_stats() else: self.__first_time__ = True query_result = executor.create_staging_raw_spotify_statistics() if 'already exists' in query_result[0]: logging.info('%s%s\n', config.snowflake_table_names[ 'staging_raw'].upper(), exists) deleted = executor.delete_from_staging_raw() logging.info('%i rows have been deleted.\n', deleted[0][0]) def compare_current_and_last_week_stats( self, current, stat_name: str, lasts: dict, artist: str, today=today()): """Compare artist's statistics.""" if isinstance(current, dict): current = current.get(stat_name) try: lasts[artist] = {k: v or 0 for k, v in lasts.get(artist).items()} except AttributeError: lasts[artist] = { 'id': None, 'current_monthly_listeners': 0, 'last_week_monthly_listeners': 0, 'weekly_change_in_monthly_listeners': 0, 'percentage_change_in_monthly_listeners': 0, 'current_followers': 0, 'last_week_followers': 0, 'weekly_change_in_followers': 0, 'percentage_change_in_followers': 0, 'current_popularity': 0, 'last_week_popularity': 0, 'editorial_playlists_adds': '', 'editorial_playlists_adds_count': 0, 'last_week_editorial_playlists_adds': '', 'last_week_editorial_playlists_adds_count': 0, 'notable_playlists_adds': '', 'notable_playlists_adds_count': 0, 'last_week_notable_playlists_adds': '', 'last_week_notable_playlists_adds_count': 0, 'the_latest_release': None, 'last_processing_date': today, 'last_week_processing_date': today} lasts[artist]['current_' + stat_name] = current lasts[artist]['last_week_' + stat_name] = current self.__first_time__ = True previous_stat = \ lasts.get(artist).get( 'current_' + stat_name) if stat_name not in [ 'editorial_playlists_adds', 'notable_playlists_adds_count', 'editorial_playlists_adds_count', 'notable_playlists_adds'] else lasts.get(artist).get(stat_name) last_week_stat = \ lasts.get(artist).get('last_week_' + stat_name) last_week_percentage_change = \ lasts.get(artist).get('percentage_change_in_' + stat_name) last_week_processing_date = \ lasts.get(artist).get('last_week_processing_date') current = 0 if current is None else current last_week_stat = previous_stat \ if today >= last_week_processing_date + datetime. \ timedelta(days=7) else last_week_stat if stat_name not in ['editorial_playlists_adds', 'notable_playlists_adds', 'popularity']: logging.info('Current count of %s is %i', stat_name, current) daily_change = current - previous_stat if \ previous_stat is not None else current current_weekly_change = self.get_executor( ).select_weekly_change( artist)[stat_name] + daily_change if stat_name in [ 'followers', 'monthly_listeners'] else 0 logging.info('Current weekly change in %s is %i', stat_name, current_weekly_change) last_weekly_change = self.get_executor().select_weekly_change( artist, first_day_of_period=-6, last_day_of_period=-14)[stat_name] if stat_name in [ 'followers', 'monthly_listeners'] else 0 logging.info('Last week period change in %s is %i', stat_name, last_weekly_change) percentage_change = ((current - last_week_stat) / last_week_stat if last_week_stat != 0 else 0) if \ today >= last_week_processing_date + datetime.timedelta( days=7) else last_week_percentage_change logging.info('Percentage change in %s is %s', stat_name, str( percentage_change)) acceleration = \ (current_weekly_change - last_weekly_change) / \ last_weekly_change if stat_name in [ 'followers', 'monthly_listeners'] and last_weekly_change != 0 else 0 else: current_weekly_change, last_weekly_change = None, None percentage_change, daily_change, acceleration = \ None, None, None return {'current_' + stat_name: current, 'last_week_' + stat_name: last_week_stat, 'daily_change_in_' + stat_name: daily_change, 'current_weekly_change_in_' + stat_name: current_weekly_change, 'weekly_change_in_' + stat_name: last_weekly_change, 'percentage_change_in_' + stat_name: percentage_change, 'acceleration_' + stat_name: acceleration, 'last_processing_date': today, 'last_week_processing_date': today if today >= last_week_processing_date + datetime.timedelta( days=7) else last_week_processing_date } def load_profile(self, artist_name, log=False): """Load artist's Spotify profile. Args: artist_name (str): artist's name. log (bool): log loading info or not. """ artist_id = self.get_profiles()[artist_name].get('id') profile = self.get_spotify().artist(artist_id) if log: logging.info(artist_name + ', Spotify ID: ' + artist_id) return profile, artist_id @catchers.catch_all_and_print(sentry_dsn=os.getenv('SENTRY_DSN')) def get_common_stats(self, artist): """Return count of followers and popularity. Args: artist (str): artist's name. """ profiles = self.get_profiles() stats = self.get_last_week_stats() profile = self.load_profile(artist, log=True) followers = profile[0].get('followers').get('total') popularity = profile[0].get('popularity') profiles[artist].update(self.compare_current_and_last_week_stats( followers, 'followers', stats, artist)) profiles[artist].update(self.compare_current_and_last_week_stats( popularity, 'popularity', stats, artist)) @catchers.catch_all_and_print(sentry_dsn=os.getenv('SENTRY_DSN')) def get_playlists_stats(self, artist): """Return count of posts, posts likes and engagement. Args: artist (str): artist's name. """ profiles = self.get_profiles() stats = self.get_last_week_stats() playlists_ids = get_playlists_ids(profiles.get(artist).get('id')) editorial_playlists = dict() notable_playlists = dict() def get_playlist_tracks(_items: list, _playlist): try: _items.extend(_playlist.get('items')) except AttributeError: return if _playlist.get('next') is not None: try: results = self.get_spotify().next(_playlist) get_playlist_tracks(_items, results) except ReadTimeout: logging.warning('The playlist is unreachable.') return except ConnectionError: logging.warning('The playlist cannot be read ' 'due to connection issues.') return else: return def search_for_artists_track(details, _tracks: list, _playlist=None): try: for track in _tracks: for _ in [ x.get('name') for x in track.get( 'track').get('artists') if x.get( 'name') == artist]: track_name_ = track.get('track').get('name') _playlist.update( {p: p + " - '" + name + "': " + track_name_}) except AttributeError as error: logging.warning('%s\nFor %s playlist, the song' 'seems to be broken on API side...', str(error), details[1]) logging.warning("Playlist's ID is %s", details[0]) editorial_playlists_adds_count, notable_playlists_adds_count = 0, 0 for p in playlists_ids: try: current_playlist = self.get_spotify().playlist(p) owner = current_playlist.get('owner').get('display_name') followers = current_playlist.get('followers').get('total') tracks_items = [] get_playlist_tracks( tracks_items, current_playlist.get('tracks')) if owner == 'Spotify': name = current_playlist.get('name') search_for_artists_track( [p, name], tracks_items, editorial_playlists) editorial_playlists_adds_count += 1 elif followers and followers >= 3000: name = current_playlist.get('name') search_for_artists_track( [p, name], tracks_items, notable_playlists) notable_playlists_adds_count += 1 except SpotifyException: logging.warning('The playlist %s is unreachable.', str(p)) editorial_playlists_string = ';\n'.join(editorial_playlists.values()) notable_playlists_string = ';\n'.join(notable_playlists.values()) current = { 'editorial_playlists_adds_count': editorial_playlists_adds_count, 'notable_playlists_adds_count': notable_playlists_adds_count, 'editorial_playlists_adds': editorial_playlists_string, 'notable_playlists_adds': notable_playlists_string} profiles[artist].update(self.compare_current_and_last_week_stats( current, 'editorial_playlists_adds_count', stats, artist)) profiles[artist].update(self.compare_current_and_last_week_stats( current, 'editorial_playlists_adds', stats, artist)) profiles[artist].update(self.compare_current_and_last_week_stats( current, 'notable_playlists_adds_count', stats, artist)) profiles[artist].update(self.compare_current_and_last_week_stats( current, 'notable_playlists_adds', stats, artist)) last_release_date = datetime.datetime.strptime( 'Jan 1 2000 10:07AM', '%b %d %Y %I:%M%p') spotify_latest_release = self.get_spotify(). \ artist_albums(profiles.get(artist).get('id')). \ get('items')[0].get('release_date') spotify_latest_release = spotify_latest_release if len( spotify_latest_release) > 4 \ else datetime.datetime.strptime( spotify_latest_release, '%Y').strftime('%Y-%m-%d') the_latest_release = last_release_date.strftime('%Y-%m-%d') if datetime.datetime.strptime( spotify_latest_release, '%Y-%m-%d') > last_release_date: if spotify_latest_release is not None: the_latest_release = spotify_latest_release profiles[artist].update({'the_latest_release': the_latest_release}) logging.info("The latest artist's release has been dropped %s", str(the_latest_release)) def get_dom_of_artist_page(self, artist): """Return DOM of artist's SC page.""" driver = self.get_driver() driver.get(config.SPOTIFY_LINK + '/artist/' + self.get_profiles().get(artist).get('id')) soup = BeautifulSoup(driver.page_source, 'html.parser') html = soup.prettify() dom = etree.HTML(html) return dom @catchers.catch_all_and_print(sentry_dsn=os.getenv('SENTRY_DSN')) def get_monthly_listeners(self, artist): """Return artist's common stats: followers, followings, tracks.""" dom = self.get_dom_of_artist_page(artist) self.set_dom(dom) profiles = self.get_profiles() stats = self.get_last_week_stats() try: contains = "//*[contains(text(), 'monthly listeners')]" monthly_listeners = dom.xpath(contains)[0].text[ :dom.xpath(contains)[0].text.index( 'monthly')].replace( '. Artist', '').split()[-1] except IndexError: monthly_listeners = '' profiles[artist].update(self.compare_current_and_last_week_stats( dict(monthly_listeners=abbreviated_value_to_int( monthly_listeners)), 'monthly_listeners', stats, artist)) def update_missing_days_stats(self, artist): """Update artists' missing days statistics.""" missing_days_stats, profiles, stats = \ {}, self.get_profiles(), self.get_last_week_stats() missing_days = (datetime.date.today() - stats[artist]. get('last_processing_date')).days days = [datetime.date.today() - datetime.timedelta(days=i) for i in range(missing_days)] missing_days_stats.update(dict( artist=artist, id=profiles[artist].get('id'), current_popularity=profiles[artist].get('current_popularity'), last_week_popularity=profiles[artist].get( 'last_week_popularity'), current_editorial_playlists_adds=profiles[artist].get( 'current_editorial_playlists_adds'), last_week_editorial_playlists_adds=profiles[artist].get( 'last_week_editorial_playlists_adds'), current_notable_playlists_adds=profiles[artist].get( 'current_notable_playlists_adds'), last_week_notable_playlists_adds=profiles[artist].get( 'last_week_notable_playlists_adds'), the_latest_release=profiles[artist].get( 'the_latest_release'))), days.sort() if missing_days > 0: missing_daily_monthly_listeners = profiles[artist].get( 'daily_change_in_monthly_listeners') / missing_days missing_daily_followers = profiles[artist].get( 'daily_change_in_followers') / missing_days missing_daily_editorial_playlists_adds_count = zero_if_none( profiles[artist].get('daily_change_in_editorial_' 'playlists_adds_count')) / missing_days missing_daily_notable_playlists_adds_count = zero_if_none( profiles[artist].get('daily_change_in_notable_' 'playlists_adds_count')) / missing_days for day in days: today_, stats = day, \ self.get_last_week_stats() # calculating actual metrics for # followers, ML, and playlists adds missing_days_stats.update( self.compare_current_and_last_week_stats( stats[artist].get('current_monthly_listeners') + int(missing_daily_monthly_listeners), 'monthly_listeners', stats, artist, today=today_)) missing_days_stats.update( self.compare_current_and_last_week_stats( stats[artist].get('current_followers') + int( missing_daily_followers), 'followers', stats, artist, today=today_)) missing_days_stats.update( self.compare_current_and_last_week_stats( stats[artist].get('editorial_playlists_adds_count') + int(missing_daily_editorial_playlists_adds_count), 'editorial_playlists_adds_count', stats, artist)) missing_days_stats.update( self.compare_current_and_last_week_stats( stats[artist].get('notable_playlists_adds_count') + int(missing_daily_notable_playlists_adds_count), 'notable_playlists_adds_count', stats, artist)) # updating stats table with new calculations # to keep the growth gradual self.get_executor().update_artist_stats(missing_days_stats) # updating staging raw table with new # calculations to get correct weekly change self.get_executor().insert_artist_stats( artist, missing_days_stats, table_name=config.snowflake_table_names['staging_raw']) # setting last week stats with the # newest data for the missing day self.set_last_week_stats() @catchers.catch_all_and_print(sentry_dsn=os.getenv('SENTRY_DSN')) def update_new_artists_stats(self, profile): """Update stats for new artists at SnowFlake.""" if self.__first_time__: self.get_executor().insert_artist_stats( profile, self.get_profiles()[profile]) logging.info("%s's stats has been inserted.\n", profile) self.get_executor().insert_artist_stats( profile, self.get_profiles()[profile], table_name=self.get_executor().staging_raw_table) def get_all_stats_and_update(self): """Update stats for every single artist.""" tracking_start = datetime.datetime.now() # browser = login_to_spotify() for artist in self.get_profiles(): self.__first_time__ = False try: start = datetime.datetime.now() self.get_monthly_listeners(artist) self.get_common_stats(artist) try: self.get_playlists_stats(artist) except ReadTimeout: logging.warning('The playlist cannot be read.') except AttributeError: logging.warning("The playlist's currently " 'broken on the API side..') except IndexError: logging.warning("Something's happened to %s's " 'Spotify profile.', artist) finish = datetime.datetime.now() logging.info('For %s it took %i seconds.', artist, (finish - start).total_seconds()) # updates staging raw with missing days stats if artist in self.get_last_week_stats(): self.update_missing_days_stats(artist) self.update_new_artists_stats(artist) except ConnectionError: logging.warning('An error occurred on requests side.') tracking_finish = datetime.datetime.now() logging.info('For all the artists it took %i seconds.\n', (tracking_finish - tracking_start).total_seconds()) @catchers.catch_all_and_print(sentry_dsn=os.getenv('SENTRY_DSN')) def update_staging_raw_table(self): """Update appropriate staging raw table.""" logging.info( 'All the stats has been updated at %s.\n', config.snowflake_table_names['spotify_statistics']) self.get_executor().update_stating_raw_spotify_stats() @catchers.catch_all_and_print(sentry_dsn=os.getenv('SENTRY_DSN')) def perform_tracking(self): """Perform tracking flow.""" self.create_tables() self.get_all_stats_and_update() # self.update_staging_raw_table() if __name__ == '__main__': tracker = SpotifyTracker() tracker.perform_tracking()