"""TikTok scrapping flow.""" import os import requests from bs4 import BeautifulSoup import datetime from datetime import date import logging from fangare import catchers from requests import HTTPError from snowflake_executor import TikTokStatsSFExecutor import config today = date.today() catchers.init_sentry(os.getenv('SENTRY_DSN')) def if_first_time(stat: int, first_time: bool) -> int: """Return stat depending onn first_time bool value.""" return stat if not first_time else 0 def _abbreviated_value_to_int(value: str) -> int: """Convert values like 4M, 35.6K to int.""" 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 def _if_week_period(value, s_value, processing_date): return value if today >= processing_date + datetime.timedelta(days=7) \ else s_value @catchers.catch_all_and_print(os.getenv('SENTRY_DSN')) def get_tiktok_stats(tiktok_url: str, stat_name='all') -> dict: """Get Tiktok stat. Args: tiktok_url (str): URL of the account to parse. stat_name (str): stat to scrape. Returns: int: stat value. Raises: HTTPError: Tiktok returned an error (usually 404). ParsingError: Failed to parse the Tiktok artist page.""" stat, stat_name = {}, 'following' if \ stat_name == 'followings' else stat_name payload = { 'api_key': config.SCRAPER_API_KEY, 'url': tiktok_url, 'premium': 'true' } response = requests.get('http://api.scraperapi.com', params=payload) response.raise_for_status() soup = BeautifulSoup(response.text) if stat_name not in [*config.TIKTOK_METRICS, 'all']: return {} else: stats = [] for stat in config.TIKTOK_METRICS: stat_span = soup.find('strong', {'title': stat.capitalize()}) if not stat_span: stat_span = soup.find('strong', {'data-e2e': 'followers-count'}) if not stat_span: count_infos_h2 = soup.find('h2', {'class': 'count-infos'}) if count_infos_h2: stat_span = count_infos_h2.select_one('div.number:nth-child(2) > strong') if not stat_span: logging.error('Failed to find the %s tag.', stat) return {} stat_ = _abbreviated_value_to_int(stat_span.text) stats.append(stat_) return dict(zip(config.TIKTOK_METRICS, stats)) class TikTokTracker: """TikTokTracker class for scrapping artists data.""" def __init__(self): logging.basicConfig(level=logging.INFO) self.__executor__ = TikTokStatsSFExecutor(sf_config=config.SF_CONFIG) self.__start__ = datetime.datetime.now() self.__profiles__ = dict() self.__last_week_stats__ = dict() self.__first_time__ = False @property def get_profiles(self) -> dict: """Return artists' profiles.""" return self.__profiles__ @property def get_last_week_stats(self): """Return last week stats from SnowFlake.""" return self.__last_week_stats__ @property def get_executor(self): """Return SnowFlake executor.""" return self.__executor__ def on_failure(self): executor = self.get_executor artists = list(filter(lambda a: a[-2] < datetime.date.today() and a[-1] > datetime.date.today() - datetime.timedelta(days=14), executor.select_artists_stats())) for artist in artists: 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.') @catchers.catch_all_and_print(os.getenv('SENTRY_DSN')) 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(link=link)) for artist, link in executor.select_artists_links() if link is not None)) query_result = executor.create_tiktok_statistics() if 'already exists' in query_result[0]: logging.info('%s %s', config.snowflake_table_names['tiktok_statistics'].upper(), exists) self.__last_week_stats__.update( dict((artist, {'tiktok_link': link, 'current_followers': current_followers, 'weekly_change_in_followers': weekly_change_in_followers, 'daily_change_in_followers': daily_change_in_followers, 'current_weekly_change_in_followers': current_weekly_change_in_followers, 'last_week_followers': last_week_followers, 'percentage_change_in_followers': percentage_change_in_followers, 'current_followings': current_followings, 'weekly_change_in_followings': weekly_change_in_followings, 'daily_change_in_followings': daily_change_in_followings, 'current_weekly_change_in_followings': current_weekly_change_in_followings, 'last_week_followings': last_week_followings, 'percentage_change_in_followings': percentage_change_in_followings, 'current_likes': current_likes, 'weekly_change_in_likes': weekly_change_in_likes, 'daily_change_in_likes': daily_change_in_likes, 'current_weekly_change_in_likes': current_weekly_change_in_likes, 'last_week_likes': last_week_likes, 'percentage_change_in_likes': percentage_change_in_likes, 'last_processing_date': last_processing_date, 'last_week_processing_date': last_week_processing_date, }) for artist, link, current_followers, weekly_change_in_followers, daily_change_in_followers, current_weekly_change_in_followers, last_week_followers, percentage_change_in_followers, current_followings, weekly_change_in_followings, daily_change_in_followings, current_weekly_change_in_followings, last_week_followings, percentage_change_in_followings, current_likes, weekly_change_in_likes, daily_change_in_likes, current_weekly_change_in_likes, last_week_likes, percentage_change_in_likes, last_processing_date, last_week_processing_date, in executor.select_artists_stats())) else: self.__first_time__ = True query_result = executor.create_staging_raw_tiktok_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 process_tiktok_stats(self, artist): """Return count of likes, dislikes. Args: artist (str): artist's name. """ profiles = self.get_profiles link = profiles[artist].get('link') lasts = self.get_last_week_stats stats = get_tiktok_stats(link) current = { 'followers': stats.get('followers'), 'followings': stats.get('following'), 'likes': stats.get('likes'), } profiles[artist].update(self.compare_current_and_last_week_stats( current, 'followers', lasts, artist)) profiles[artist].update(self.compare_current_and_last_week_stats( current, 'followings', lasts, artist)) profiles[artist].update(self.compare_current_and_last_week_stats( current, 'likes', lasts, artist)) 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.process_tiktok_stats(artist) self.update_stats(artist) finish = datetime.datetime.now() logging.info('For %s it took %s seconds.\n', artist, str((finish - start).total_seconds())) except HTTPError: logging.error(artist + 'has the wrong TikTok link. Please, reset the link.') tracking_finish = datetime.datetime.now() logging.info('For all the artists it took %s seconds.\n', str((tracking_finish - tracking_start).total_seconds())) @catchers.catch_all_and_print(os.getenv('SENTRY_DSN')) def update_stats(self, artist): """Update all stats for artist at SnowFlake.""" if not self.__first_time__: self.get_executor.update_artist_stats(artist, self.get_profiles[artist]) logging.info("%s's stats has been updated.", artist) else: self.get_executor.insert_artist_stats(artist, self.get_profiles[artist]) logging.info("%s's stats has been inserted.", artist) @catchers.catch_all_and_print(os.getenv('SENTRY_DSN')) def compare_current_and_last_week_stats( self, current, stat_name: str, lasts: dict, artist: str): if isinstance(current, dict): current = current.get(stat_name) if current.get(stat_name) is not None else 0 lasts_ = dict() 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] = {'tiktok_link': None, 'current_followers': 0, 'weekly_change_in_followers': 0, 'daily_change_in_followers': 0, 'current_weekly_change_in_followers': 0, 'last_week_followers': 0, 'percentage_change_in_followers': 0, 'current_followings': 0, 'weekly_change_in_followings': 0, 'daily_change_in_followings': 0, 'current_weekly_change_in_followings': 0, 'last_week_followings': 0, 'percentage_change_in_followings': 0, 'current_likes': 0, 'weekly_change_in_likes': 0, 'daily_change_in_likes': 0, 'current_weekly_change_in_likes': 0, 'last_week_likes': 0, 'percentage_change_in_likes': 0, '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_[artist]. \ get('current_' + stat_name) last_week_stat = lasts_[artist]. \ get('last_week_' + stat_name) last_week_percentage_change = lasts_[artist]. \ get('percentage_change_in_' + stat_name) last_week_processing_date = lasts_[artist]. \ get('last_week_processing_date') 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) 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 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: _if_week_period(previous_stat, last_week_stat, last_week_processing_date), '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': _if_week_period(today, last_week_processing_date, last_week_processing_date) } 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_tiktok_stats() @catchers.catch_flow_failure(on_failure, os.getenv('SENTRY_DSN')) def perform_tracking(self): """Perform tracking of TikTok stats.""" self.create_tables() self.get_and_update_stats() self.update_staging_raw_table_with_stats() if __name__ == '__main__': tracker = TikTokTracker() tracker.perform_tracking()