"""SoundCloud scrapping flow.""" import datetime import logging import os import time import config from fangare import catchers from helpers import abbreviated_value_to_int from selenium import webdriver from selenium.common.exceptions import NoSuchElementException from selenium.webdriver.remote.errorhandler import \ NoSuchElementException as RemoteNoSuchElementException from snowflake_executor import SoundCloudStatsSFExecutor catchers.init_sentry(sentry_dsn=os.getenv('SENTRY_DSN')) def get_id_from_link(text): """Return SoundCloud channel link from config.""" return text. \ replace(config.SOUNDCLOUD_LINK, '') def today(): """Return today date.""" return datetime.date.today() def zero_if_none(value): """Return zero if the value is None.""" return value if value is not None else 0 def scroll_down(driver_, last_height=0): """Scroll a page down.""" driver_.execute_script('window.scrollTo(0, document.body.scrollHeight);') time.sleep(2) height = driver_.execute_script('return document.body.scrollHeight') if last_height == height: return scroll_down(driver_, height) @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-infobars') options.add_argument('start-maximized') driver = webdriver.Chrome('/soundcloud-statistics/chromedriver', options=options) return driver def if_first_time(stat, first_time): """Return stat depending onn first_time bool value.""" return stat if not first_time else 0 def try_to_get_entry(_dict: dict, key): """Try to get a value using the key.""" try: _dict[key] except KeyError: pass class SoundCloudTracker: """SoundCloudTracker class for scrapping artists data.""" def __init__(self): """Initialise SC tracker.""" logging.basicConfig(level=logging.INFO) self.__executor__ = SoundCloudStatsSFExecutor( sf_config=config.SF_CONFIG) self.__start__ = datetime.datetime.now() self.__profiles__ = dict() self.__last_week_stats__ = dict() self.__first_time__ = False self.__driver__ = set_webdriver() self.__unreachable_accounts__ = list() def get_profiles(self): """Return artists' profiles.""" return self.__profiles__ def get_last_week_stats(self): """Return last week stats from SnowFlake.""" return self.__last_week_stats__ def get_executor(self): """Return SnowFlake executor.""" return self.__executor__ def get_driver(self): """Return Selenium webdriver.""" return self.__driver__ def set_driver(self, driver): """Set current page driver.""" self.__driver__ = driver def get_unreachable_profiles(self): """Return artists' profiles.""" return self.__unreachable_accounts__ def add_to_unreachable_profiles(self, artist): """Return artists' profiles.""" self.__unreachable_accounts__.append(artist) def get_stat_from_(self, attrib_name: str, multiple=False, _type='xpath'): """Get statistics with its attribute from DOM.""" if _type == 'xpath': if multiple: return self.__driver__.find_elements_by_xpath( config.xpaths.get(attrib_name)) return self.__driver__.find_element_by_xpath( config.xpaths.get(attrib_name)) else: if multiple: return self.__driver__.find_elements_by_css_selector( config.css_paths.get(attrib_name)) return self.__driver__.find_element_by_css_selector( config.css_paths.get(attrib_name)) def set_last_week_stats(self): """Set last week statistics to a profile.""" self.__last_week_stats__.update( dict((artist, { 'id': soundcloud_link, 'current_followers_count': current_followers_count, 'last_week_followers_count': last_week_followers_count, 'weekly_change_in_followers_count': weekly_change_in_followers_count, 'percentage_change_in_followers_count': percentage_change_in_followers_count, 'current_followings_count': current_followings_count, 'last_week_followings_count': last_week_followings_count, 'weekly_change_in_followings_count': weekly_change_in_followings_count, 'percentage_change_in_followings_count': percentage_change_in_followings_count, 'current_tracks_count': current_tracks_count, 'last_week_tracks_count': last_week_tracks_count, 'weekly_change_in_tracks_count': weekly_change_in_tracks_count, 'percentage_change_in_tracks_count': percentage_change_in_tracks_count, 'current_plays_count': current_plays_count, 'last_week_plays_count': last_week_plays_count, 'weekly_change_in_plays_count': weekly_change_in_plays_count, 'percentage_change_in_plays_count': percentage_change_in_plays_count, 'current_likes_count': current_likes_count, 'last_week_likes_count': last_week_likes_count, 'weekly_change_in_likes_count': weekly_change_in_likes_count, 'percentage_change_in_likes_count': percentage_change_in_likes_count, 'current_reposts_count': current_reposts_count, 'last_week_reposts_count': last_week_reposts_count, 'weekly_change_in_reposts_count': weekly_change_in_reposts_count, 'percentage_change_in_reposts_count': percentage_change_in_reposts_count, 'plays_engagement': plays_engagement, 'the_latest_release': the_latest_release, 'last_processing_date': last_processing_date, 'last_week_processing_date': last_week_processing_date}) for artist, soundcloud_link, current_followers_count, last_week_followers_count, weekly_change_in_followers_count, percentage_change_in_followers_count, current_followings_count, last_week_followings_count, weekly_change_in_followings_count, percentage_change_in_followings_count, current_tracks_count, last_week_tracks_count, weekly_change_in_tracks_count, percentage_change_in_tracks_count, current_plays_count, last_week_plays_count, weekly_change_in_plays_count, percentage_change_in_plays_count, current_likes_count, last_week_likes_count, weekly_change_in_likes_count, percentage_change_in_likes_count, current_reposts_count, last_week_reposts_count, weekly_change_in_reposts_count, percentage_change_in_reposts_count, plays_engagement, the_latest_release, last_processing_date, last_week_processing_date in self.get_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() query_result = executor.create_soundcloud_statistics() if 'already exists' in query_result[0]: logging.info('%s %s', config.snowflake_table_names[ 'soundcloud_statistics'].upper(), exists) self.set_last_week_stats() else: self.__first_time__ = True if 'already exists' in query_result[0]: logging.info('%s %s', config.snowflake_table_names[ 'soundcloud_artist_to_track'].upper(), exists) self.__profiles__.update( dict((username, dict(soundcloud_link=link)) for username, link in executor.select_artists_links() if link is not None and config.SOUNDCLOUD_LINK in link)) query_result = executor.create_staging_raw_soundcloud_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.', deleted[0][0]) def compare_current_and_last_week_stats( self, current, stat_name: str, lasts: dict, artist: str, today=today()): """Compare today's and yesterday's statistics.""" if isinstance(current, dict): current = current.get(stat_name) try: lasts[artist] = dict(zip( lasts.get(artist).keys(), map(lambda key: 0 if lasts.get(artist).get(key) is None else lasts.get(artist).get(key), lasts.get(artist).keys()))) except AttributeError: lasts[artist] = { 'id': None, 'current_followers_count': 0, 'last_week_followers_count': 0, 'weekly_change_in_followers_count': 0, 'percentage_change_in_followers_count': 0, 'current_followings_count': 0, 'last_week_followings_count': 0, 'weekly_change_in_followings_count': 0, 'percentage_change_in_followings_count': 0, 'current_tracks_count': 0, 'last_week_tracks_count': 0, 'weekly_change_in_tracks_count': 0, 'percentage_change_in_tracks_count': 0, 'current_plays_count': 0, 'last_week_plays_count': 0, 'weekly_change_in_plays_count': 0, 'percentage_change_in_plays_count': 0, 'current_likes_count': 0, 'last_week_likes_count': 0, 'weekly_change_in_likes_count': 0, 'percentage_change_in_likes_count': 0, 'current_reposts_count': 0, 'last_week_reposts_count': 0, 'weekly_change_in_reposts_count': 0, 'percentage_change_in_reposts_count': 0, 'plays_engagement': 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) last_week_stat = \ lasts.get(artist).get('last_week_' + 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 logging.info('Current count of %s is %i', stat_name, current) daily_change = current - previous_stat current_weekly_change = self.get_executor().\ select_weekly_change(artist)[stat_name] + daily_change 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] logging.info('Last week period change in %s is %i', stat_name, last_weekly_change) if stat_name in ['followers', 'followers_count']: percentage_change = (current_weekly_change / ( current - current_weekly_change) if (current - current_weekly_change) != 0 else 0) else: percentage_change = (current_weekly_change - last_weekly_change) /\ last_weekly_change if last_weekly_change != 0 else 0 acceleration = (current_weekly_change - last_weekly_change) /\ last_weekly_change if last_weekly_change != 0 else 0 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 get_driver_of_artist_page(self, artist): """Return driver of artist's SC page.""" driver = self.get_driver() driver.get(self.get_profiles().get(artist).get( 'soundcloud_link') + config.TRACKS_LINK) scroll_down(driver) return driver @catchers.catch_all_and_print(sentry_dsn=os.getenv('SENTRY_DSN')) def get_artist_info(self, artist): """Return artist's common stats: followers, followings, tracks.""" driver = self.get_driver_of_artist_page(artist) self.set_driver(driver) profiles = self.get_profiles() stats = self.get_last_week_stats() followers = abbreviated_value_to_int( self.get_stat_from_('followers', _type='css').text) following = int(''.join(c for c in self.get_stat_from_( 'followings').text if c.isnumeric())) try: tracks_count = int(''.join(c for c in self.get_stat_from_( 'tracks_count').text if c.isnumeric())) except ValueError: tracks_count = 0 try: # the_latest_release_string = self.get_stat_from_( # 'release', _type='css')[0].attrib['datetime'] - prev. v. the_latest_release_string = self.get_stat_from_( 'release', _type='css').get_attribute('datetime') the_latest_release = datetime.datetime.strptime( the_latest_release_string[:int( the_latest_release_string.find('T'))], '%Y-%m-%d') logging.info("%s's latest release has been dropped %s", artist, str(the_latest_release)) except NoSuchElementException: the_latest_release = None logging.info("%s's tracks are hidden or absent.", artist) current = dict(followers_count=followers, followings_count=following, tracks_count=tracks_count) try_to_get_entry(stats, artist) profiles[artist].update(self.compare_current_and_last_week_stats( current, 'followers_count', stats, artist)) profiles[artist].update(self.compare_current_and_last_week_stats( current, 'followings_count', stats, artist)) profiles[artist].update(self.compare_current_and_last_week_stats( current, 'tracks_count', stats, artist)) profiles[artist].update({'the_latest_release': the_latest_release}) @catchers.catch_all_and_print(sentry_dsn=os.getenv('SENTRY_DSN')) def get_tracks_info(self, artist): """Return video info: title, publishing date, views, likes etc.""" profiles = self.get_profiles() stats = self.get_last_week_stats() tracks = len(self.get_stat_from_( 'tracks', multiple=True)) # tracks number plays = [abbreviated_value_to_int(x.text) for x in self.get_stat_from_('plays', multiple=True)[:tracks]] likes = [abbreviated_value_to_int(x.text) for x in self.get_stat_from_('likes', multiple=True)[:tracks]] reposts = [abbreviated_value_to_int(x.text) for x in self.get_stat_from_('reposts', multiple=True)[:tracks]] tracks_stats = dict(zip(['plays_count', 'likes_count', 'reposts_count'], [sum(plays), sum(likes), sum(reposts)])) try_to_get_entry(stats, artist) profiles[artist].update(self.compare_current_and_last_week_stats( tracks_stats, 'plays_count', stats, artist)) profiles[artist].update(self.compare_current_and_last_week_stats( tracks_stats, 'likes_count', stats, artist)) profiles[artist].update(self.compare_current_and_last_week_stats( tracks_stats, 'reposts_count', stats, artist)) plays_engagement = \ tracks_stats.get('likes_count') /\ tracks_stats.get('plays_count') \ if tracks_stats.get('plays_count') != 0 else 0 weekly_plays_engagement = \ profiles[artist].get('current_weekly_change_in_likes_count') /\ profiles[artist].get('current_weekly_change_in_plays_count') \ if profiles[artist].get( 'current_weekly_change_in_plays_count') != 0 else 0 logging.info("%s's engagement is %s", artist, plays_engagement) logging.info("%s's weekly engagement is %s", artist, weekly_plays_engagement) profiles[artist].update(plays_engagement=weekly_plays_engagement) @catchers.catch_all_and_print(sentry_dsn=os.getenv('SENTRY_DSN')) def update_new_artists_stats(self, artist): """Update all stats for artist at SnowFlake.""" if self.__first_time__: self.get_executor().insert_artist_stats( artist, self.get_profiles()[artist]) logging.info("%s's stats has been inserted.", artist) self.get_executor().insert_artist_stats( artist, self.get_profiles()[artist], table_name=self.get_executor().staging_raw_table) def update_missing_days_stats(self, artist): """Update 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, soundcloud_link=profiles[artist].get('soundcloud_link'), plays_engagement=profiles[artist].get('plays_engagement'), the_latest_release=profiles[artist].get('the_latest_release'), )), days.sort() if missing_days > 0: missing_daily_followers = profiles[artist].get( 'daily_change_in_followers_count') / missing_days missing_daily_likes = zero_if_none(profiles[artist].get( 'daily_change_in_likes_count')) / missing_days missing_daily_tracks = zero_if_none(profiles[artist].get( 'daily_change_in_tracks_count')) / missing_days missing_daily_followings = profiles[artist].get( 'daily_change_in_followings_count') / missing_days missing_daily_plays = zero_if_none(profiles[artist].get( 'daily_change_in_plays_count')) / missing_days missing_daily_reposts = zero_if_none(profiles[artist].get( 'daily_change_in_reposts_count')) / missing_days for day in days: today_, stats = day, \ self.get_last_week_stats() # calculating actual metrics for followers, likes, tracks # followings, plays, and reposts missing_days_stats.update( self.compare_current_and_last_week_stats( stats[artist].get('current_followers_count') + missing_daily_followers, 'followers_count', stats, artist, today=today_)) missing_days_stats.update( self.compare_current_and_last_week_stats( stats[artist].get('current_likes_count') + missing_daily_likes, 'likes_count', stats, artist, today=today_)) missing_days_stats.update( self.compare_current_and_last_week_stats( stats[artist].get('current_tracks_count') + missing_daily_tracks, 'tracks_count', stats, artist, today=today_)) missing_days_stats.update( self.compare_current_and_last_week_stats( stats[artist].get('current_followings_count') + missing_daily_followings, 'followings_count', stats, artist, today=today_)) missing_days_stats.update( self.compare_current_and_last_week_stats( stats[artist].get('current_plays_count') + missing_daily_plays, 'plays_count', stats, artist, today=today_)) missing_days_stats.update( self.compare_current_and_last_week_stats( stats[artist].get('current_reposts_count') + missing_daily_reposts, 'reposts_count', stats, artist, today=today_)) # updating stats table with new calculations # to keep the growth gradual self.get_executor().update_artist_stats( missing_days_stats, artist) # 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() def get_and_update_stats(self): """Update stats for every single artist.""" tracking_start = datetime.datetime.now() for artist in self.get_profiles(): self.__first_time__ = False try: start = datetime.datetime.now() self.get_artist_info(artist) self.get_tracks_info(artist) # 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) finish = datetime.datetime.now() logging.info('For %s it took %s seconds.\n', artist, str((finish - start).total_seconds())) except NoSuchElementException or RemoteNoSuchElementException: logging.error("Something went wrong with %s's page.\n", artist) self.add_to_unreachable_profiles(dict( artist=artist, soundcloud_link=self. get_profiles()[artist].get('soundcloud_link'))) tracking_finish = datetime.datetime.now() logging.info('For all the artists it took %s seconds.\n', str((tracking_finish - tracking_start).total_seconds())) self.get_driver().quit() @catchers.catch_all_and_print(sentry_dsn=os.getenv('SENTRY_DSN')) def update_staging_raw_table_with_stats(self): """Update all stats for artists at SnowFlake.""" logging.info('All the stats has been updated at %s.\n', config.snowflake_table_names['soundcloud_statistics']) self.get_executor().update_stating_raw_soundcloud_stats() @catchers.catch_all_and_print(sentry_dsn=os.getenv('SENTRY_DSN')) def perform_tracking(self): """Perform tracking of SoundCloud.""" self.create_tables() self.get_and_update_stats() if len(self.get_unreachable_profiles()) > 0: catchers.sent_a_message_to_slack( title='Unreachable accounts', message='\n'. join(list(map(lambda x: x.get('artist') + ': ' + x.get('soundcloud_link'), self.get_unreachable_profiles()))), username='A&R Support', url=os.getenv('SLACK_WEBHOOK_URL'), icon_emoji=':soundcloud:', color='#EE9133') self.update_staging_raw_table_with_stats() if __name__ == '__main__': tracker = SoundCloudTracker() tracker.perform_tracking()