"""YouTube scrapping flow.""" import datetime import logging import os import config from fangare import catchers from helpers import get_youtube_api_credentials from pyyoutube import Api, PyYouTubeException from requests import ReadTimeout from snowflake_executor import YouTubeStatsSFExecutor def get_id_from_link(text): """Return YouTube channel link from config.""" return text.replace(config.YOUTUBE_LINK, '') def today(): """Return today date.""" return datetime.date.today() def if_first_time(stat, first_time): """Return stat depending onn first_time bool value.""" return stat if not first_time else 0 def zero_if_none(value): return value if value is not None else 0 catchers.init_sentry(sentry_dsn=os.getenv('SENTRY_DSN')) class YouTubeTracker: """YouTubeTracker class for scrapping artists data.""" def __init__(self): logging.basicConfig(level=logging.INFO) self.__yt_api__ = None self.__executor__ = YouTubeStatsSFExecutor(sf_config=config.SF_CONFIG) self.__start__ = datetime.datetime.now() self.__profiles__ = dict() self.__last_week_stats__ = dict() self.__first_time__ = False self.__channel__ = None def get_profiles(self): """Return artists' profiles.""" return self.__profiles__ def update_profiles(self): """Delete updated artists from profiles.""" updated_artists = \ [artist for tuple_ in list(artist for artist in self.get_executor(). select_recent_stats()) for artist in tuple_] for profile in list(self.get_profiles()): if profile in updated_artists: self.get_profiles().pop(profile) def get_channel(self): """Return current channel.""" return self.__channel__ 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_yt_api(self): """Return YouTube API instance.""" return self.__yt_api__ def on_failure(self): executor = self.get_executor() artists = [a for a in executor.select_artists_stats() if ( a[-2] < datetime.date.today() and a[-1] > datetime.date.today() - datetime.timedelta(days=14))] for artist in artists: # artist[0] returns artist's name executor.update_artist_stats_last_processing_date( artist[0], datetime.datetime.today()) self.update_staging_raw_table_with_stats() logging.error('Flow has failed, but all the stats kept and updated ' 'preventing the dashboards being off.') def set_last_week_stats(self): self.__last_week_stats__.update( dict((artist_name, { 'id': youtube_link, 'current_subscribers': current_subscribers_count, 'last_week_subscribers': last_week_subscribers_count, 'weekly_change_in_subscribers': weekly_change_in_subscribers, 'percentage_change_in_subscribers': percentage_change_in_subscribers, 'current_videos': videos_count, 'current_likes': current_likes_count, 'last_week_likes': last_week_likes_count, 'weekly_change_in_likes': weekly_change_in_likes, 'percentage_change_in_likes': percentage_change_in_likes, 'current_dislikes': current_dislikes_count, 'last_week_dislikes': last_week_dislikes_count, 'weekly_change_in_dislikes': weekly_change_in_dislikes, 'percentage_change_in_dislikes': percentage_change_in_dislikes, 'current_views': current_views_count, 'last_week_views': last_week_views_count, 'weekly_change_in_views': weekly_change_in_views, 'percentage_change_in_views': percentage_change_in_views, 'current_comments': current_comments_count, 'last_week_comments': last_week_comments_count, 'weekly_change_in_comments': weekly_change_in_comments, 'percentage_change_in_comments': percentage_change_in_comments, 'likes_engagement': likes_engagement, 'comments_engagement': comments_engagement, 'the_latest_release': the_latest_release, 'last_processing_date': last_processing_date, 'last_week_processing_date': last_week_processing_date}) for artist_name, youtube_link, current_subscribers_count, last_week_subscribers_count, weekly_change_in_subscribers, percentage_change_in_subscribers, videos_count, current_likes_count, last_week_likes_count, weekly_change_in_likes, percentage_change_in_likes, current_dislikes_count, last_week_dislikes_count, weekly_change_in_dislikes, percentage_change_in_dislikes, current_views_count, last_week_views_count, weekly_change_in_views, percentage_change_in_views, current_comments_count, last_week_comments_count, weekly_change_in_comments, percentage_change_in_comments, likes_engagement, comments_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() 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_youtube_statistics() if 'already exists' in query_result[0]: logging.info('%s %s', config.snowflake_table_names[ 'youtube_statistics'].upper(), exists) self.update_profiles() self.set_last_week_stats() else: self.__first_time__ = True query_result = executor.create_staging_raw_youtube_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()): 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_subscribers': 0, 'last_week_subscribers': 0, 'weekly_change_in_subscribers': 0, 'percentage_change_in_subscribers': 0, 'current_videos': 0, 'current_likes': 0, 'last_week_likes': 0, 'weekly_change_in_likes': 0, 'percentage_change_in_likes': 0, 'current_dislikes': 0, 'last_week_dislikes': 0, 'weekly_change_in_dislikes': 0, 'percentage_change_in_dislikes': 0, 'current_views': 0, 'last_week_views': 0, 'weekly_change_in_views': 0, 'percentage_change_in_views': 0, 'current_comments': 0, 'last_week_comments': 0, 'weekly_change_in_comments': 0, 'percentage_change_in_comments': 0, 'likes_engagement': 0, 'comments_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 ['likes', 'views', 'comments']: 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 elif stat_name == 'subscribers': percentage_change = ( current_weekly_change / (current - current_weekly_change) if (current - current_weekly_change) != 0 else 0) acceleration = (current_weekly_change - last_weekly_change) / last_weekly_change if \ last_weekly_change != 0 else 0 else: percentage_change, acceleration = 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 get_all_the_content_with_next_page_token( self, id_, page_token, items: list, type_='playlist'): """Recursive method getting all the content using next page token.""" items_ = [] token = None if type_ == 'playlist': try: playlist = self.get_yt_api().get_playlist_items( playlist_id=id_, page_token=page_token) items_, token = playlist.items, playlist.nextPageToken except PyYouTubeException as error: if 'cannot be found' in error.message: logging.error("Playlist doesn't seem to exist.") elif type_ == 'comment_thread': try: comments = self.get_yt_api(). \ get_comment_threads(video_id=id_, page_token=page_token) items_, token = comments.items, comments.nextPageToken except ReadTimeout: logging.error('The comments cannot be read.') items.extend(items_) if token: self.get_all_the_content_with_next_page_token( id_, token, items, type_) return items def get_channel_info(self, artist, log=False): """Return channel info: title, videos, subscribers etc.""" channel_id = self.get_profiles()[artist].get('id') channel = self.get_yt_api().get_channel_info( channel_id=channel_id).items[0] title = channel.snippet.title or '' description = channel.snippet.description or '' videos_count = channel.statistics.videoCount or '0' subscribers = channel.statistics.subscriberCount or '0' total_views = channel.statistics.viewCount or '0' all_videos = channel.contentDetails.relatedPlaylists.uploads or '0' if log: logging.info(artist + ', YouTube Channel ID: ' + channel_id) self.__channel__ = {'title': title, 'videos_count': videos_count, 'description': description, 'subscribers': subscribers, 'total_views': total_views, 'all_videos': all_videos } def get_video_ids(self, all_videos): """Return video IDs using recursive function defined above.""" videos_ = self.get_all_the_content_with_next_page_token( all_videos, None, [], type_='playlist') videos = list(map(lambda v: v.snippet.resourceId.videoId, videos_)) return videos def get_video_info(self, video_id): """Return video info: title, publishing date, views, likes etc.""" video_by_id = self.get_yt_api().get_video_by_id( video_id=video_id).items[0] title = video_by_id.snippet.title description = video_by_id.snippet.description date_published = video_by_id.snippet.publishedAt views = video_by_id.statistics.viewCount or '0' likes = video_by_id.statistics.likeCount or '0' dislikes = video_by_id.statistics.dislikeCount or '0' comments_count = video_by_id.statistics.commentCount or '0' try: comments_ = self.get_all_the_content_with_next_page_token( video_id, None, [], type_='comment_thread') comments = list( map(lambda c: {'comment': c.snippet.topLevelComment.snippet.textDisplay, 'likes': c.snippet.topLevelComment.snippet.likeCount, 'author': c.snippet.topLevelComment.snippet. authorDisplayName, 'authorChannelId': comments_[0].snippet.topLevelComment.snippet. authorChannelUrl.replace( config.HTTP_YOUTUBE_LINK, '')}, comments_)) except PyYouTubeException as error: if 'has disabled comments' in error.message: logging.info('Comments seem to be disabled.') else: logging.error(error.message) comments = [] video_info = {'video_title': title, 'date_published': date_published, 'description': description, 'author': video_by_id.snippet.channelTitle, 'channel_id': video_by_id.snippet.channelId, 'views': views, 'likes': likes, 'dislikes': dislikes, 'comments_count': comments_count, 'comments': comments } return video_info @catchers.catch_all_and_print(sentry_dsn=os.getenv('SENTRY_DSN')) def get_channel_stats(self, artist): """Return count of subscribers and videos count. Args: artist (str): artist's name.""" profiles = self.get_profiles() stats = self.get_last_week_stats() self.get_channel_info(artist) channel = self.get_channel() current = {'subscribers': int(channel.get('subscribers'))} logging.info('Artist: %s', artist) profiles[artist].update(self.compare_current_and_last_week_stats( current, 'subscribers', stats, artist)) profiles[artist].update({'videos_count': channel.get('videos_count')}) @catchers.catch_all_and_print(sentry_dsn=os.getenv('SENTRY_DSN')) def get_likes_dislikes_stats(self, artist, t_likes=0, t_dislikes=0): """Return count of likes, dislikes. Args: artist (str): artist's name. t_likes (int): total likes counter t_dislikes (int): total dislikes counter """ stats = self.get_last_week_stats() profiles = self.get_profiles() channel = self.get_channel() for video in self.get_video_ids(channel.get('all_videos')): info = self.get_video_info(video) t_likes += int(info.get('likes') or 0) t_dislikes += int(info.get('dislikes') or 0) logging.info('"%s" got %s likes and %s dislikes.', info.get('video_title'), info.get('likes'), info.get('dislikes')) current = {'likes': t_likes, 'dislikes': t_dislikes} profiles[artist].update(self.compare_current_and_last_week_stats( current, 'likes', stats, artist)) profiles[artist].update(self.compare_current_and_last_week_stats( current, 'dislikes', stats, artist)) @catchers.catch_all_and_print(sentry_dsn=os.getenv('SENTRY_DSN')) def get_views_comments_stats(self, artist, t_comments=0): """Return count of likes, dislikes. Args: artist (str): artist's name. t_comments (int): total comments counter """ stats = self.get_last_week_stats() last_release_date = datetime.datetime.strptime( 'Jan 1 2000 10:07AM', '%b %d %Y %I:%M%p') profiles = self.get_profiles() channel = self.get_channel() views = int(channel.get('total_views')) for video in self.get_video_ids(channel.get('all_videos')): info = self.get_video_info(video) release_date = datetime.datetime.strptime( info.get('date_published'), '%Y-%m-%dT%H:%M:%SZ') if release_date > last_release_date and release_date is not None: last_release_date = release_date else: last_release_date = last_release_date t_comments += int(info.get('comments_count')) logging.info('"' + info.get('video_title') + '" got ' + str(info.get('comments_count')) + ' comments.\n' + 'Total views on the video: ' + info.get('views')) current = {'views': views, 'comments': t_comments} profiles[artist].update(self.compare_current_and_last_week_stats( current, 'views', stats, artist)) profiles[artist].update(self.compare_current_and_last_week_stats( current, 'comments', stats, artist)) likes_engagement = profiles[artist].get( 'current_likes') / views * 100 \ if views > 0 else 0 # likes / views comments_engagement = profiles[artist].get( 'current_comments') / views * 100 \ if views > 0 else 0 # comments / views logging.info("The latest artist's release has been dropped %s", last_release_date.strftime('%Y-%m-%d')) profiles[artist].update( {'likes_engagement': round(likes_engagement, 2), 'comments_engagement': round(comments_engagement, 2), 'the_latest_release': last_release_date.strftime('%Y-%m-%d')}) @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 updated.", artist) self.get_executor().insert_artist_stats( artist, self.get_profiles()[artist], table_name=self.get_executor().staging_raw_table) logging.info("%s's stats has been inserted.", artist) def update_missing_days_stats(self, artist): 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( id=profiles[artist].get('id'), videos_count=profiles[artist].get('videos_count'), likes_engagement=profiles[artist].get('likes_engagement'), comments_engagement=profiles[artist].get('comments_engagement'), the_latest_release=profiles[artist].get('the_latest_release'), )), days.sort() if missing_days > 0: missing_daily_subscribers = profiles[artist].get( 'daily_change_in_subscribers') / missing_days missing_daily_likes = zero_if_none(profiles[artist].get( 'daily_change_in_likes')) / missing_days missing_daily_dislikes = zero_if_none(profiles[artist].get( 'daily_change_in_dislikes')) / missing_days missing_daily_views = profiles[artist].get( 'daily_change_in_views') / missing_days missing_daily_comments = zero_if_none(profiles[artist].get( 'daily_change_in_comments')) / missing_days for day in days: today_, stats = day, \ self.get_last_week_stats() # calculating actual metrics for subscribers, likes, views # dislikes, and comments missing_days_stats.update( self.compare_current_and_last_week_stats( stats[artist].get('current_subscribers') + missing_daily_subscribers, 'subscribers', stats, artist, today=today_)) missing_days_stats.update( self.compare_current_and_last_week_stats( stats[artist].get('current_likes') + missing_daily_likes, 'likes', stats, artist, today=today_)) missing_days_stats.update( self.compare_current_and_last_week_stats( stats[artist].get('current_dislikes') + missing_daily_dislikes, 'dislikes', stats, artist, today=today_)) missing_days_stats.update( self.compare_current_and_last_week_stats( stats[artist].get('current_views') + missing_daily_views, 'views', stats, artist, today=today_)) missing_days_stats.update( self.compare_current_and_last_week_stats( stats[artist].get('current_comments') + missing_daily_comments, 'comments', stats, artist, today=today_)) # 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 missing day self.set_last_week_stats() def get_and_update_stats(self): """Update stats for every single artist.""" tracking_start = datetime.datetime.now() # browser = login_to_youtube() for artist in self.get_profiles(): self.__first_time__ = False try: start = datetime.datetime.now() self.get_channel_stats(artist) self.get_likes_dislikes_stats(artist) self.get_views_comments_stats(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 PyYouTubeException as error: if 'you have exceeded your' in error.message: logging.warning('The request cannot be completed because ' 'YouTube quota has been exceeded.') logging.info('The flow will continue scrapping artists' 'data tomorrow starting with %s' "'s channel.", artist) return {'stop': True, 'artist': artist} except TypeError as error: if 'object is not subscriptable' in str(error): logging.error("%s's channel seems to be deleted.", artist) tracking_finish = datetime.datetime.now() logging.info('For all the artists it took %s seconds.\n', str((tracking_finish - tracking_start).total_seconds())) return {'stop': False} 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['staging_raw']) self.get_executor().update_stating_raw_youtube_stats() @catchers.catch_all_and_print(sentry_dsn=os.getenv('SENTRY_DSN')) def perform_tracking(self): """Perform tracking with pyyoutube.""" self.create_tables() res = dict(stop=True) counter = 0 while res.get('stop') and len(self.get_profiles()) > counter: try: self.__yt_api__ = Api(api_key=get_youtube_api_credentials() .get('client_id')[counter]) except IndexError: logging.info('YT API Quotas exceeded.') break res = self.get_and_update_stats() counter += 1 self.update_profiles() # if not res.get('stop'): # self.update_staging_raw_table_with_stats() if __name__ == '__main__': tracker = YouTubeTracker() tracker.perform_tracking()