"""Instagram scrapping flow.""" import datetime import json import logging import os import sqlite3 import time from argparse import ArgumentParser from glob import glob from os.path import expanduser from Crypto.Cipher import AES from Crypto.Protocol.KDF import PBKDF2 import apiclient import config from fangare import catchers import httplib2 import instaloader from instaloader import ConnectionException, ProfileNotExistsException from instaloader import Instaloader from oauth2client.service_account import ServiceAccountCredentials from snowflake_executor import InstagramStatsSFExecutor 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 value is None.""" return value if value is not None else 0 def get_cookies(path=None): """Return cookie file.""" # valid path for MacOs is Users//Library/ # Application Support/Google/Chrome//Cookies cookies = [path] if path else glob(expanduser(os.getcwd() + '/Cookies')) if not cookies: raise SystemExit('No cookies file found. Use -c COOKIEFILE.') return cookies[0] def import_session(cookies_file, session): """Return saved Instagram session.""" def chrome_decrypt(encrypted_value_, key_=None): encrypted_value_ = encrypted_value_[3:] # Strip padding by taking off number indicated by padding # eg if last is '\x0e' then ord('\x0e') == 14, so take off 14. # You'll need to change this function to use ord() for python2. def clean(value_): """Return decoded and cleaned session info.""" return value_[:-value_[-1]].decode('utf8') cipher = AES.new(key_, AES.MODE_CBC, IV=b' ' * 16) decrypted = cipher.decrypt(encrypted_value_) return clean(decrypted) cookies = {} cookies_list = [] logging.info('Using cookies from %s.', cookies_file) key = PBKDF2(password=config.chrome_safe_storage, salt=b'saltysalt', dkLen=16, count=config.ITERATIONS) conn = sqlite3.connect(cookies_file) sql = 'SELECT name, value, encrypted_value FROM ' \ "cookies WHERE host_key LIKE '%instagram.com%'" with conn: for name, value, encrypted_value in conn.execute(sql): if value or (encrypted_value[:3] != b'v10'): cookies_list.append((name, value)) else: decrypted_tuple = (name, chrome_decrypt(encrypted_value, key)) cookies_list.append(decrypted_tuple) cookies.update(cookies_list) instaloader_ = Instaloader(max_connection_attempts=1) instaloader_.context._session.cookies.update(cookies) username = instaloader_.test_login() if not username: raise SystemExit('Not logged in. Are you logged' ' in successfully in Chrome?') logging.info('Imported session cookie for %s.', username) instaloader_.context.username = username instaloader_.save_session_to_file(session) return instaloader_ def if_first_time(stat, first_time): """Return stat depending onn first_time bool value.""" return stat if not first_time else 0 def get_google_sheet(sheet_id): """Get Google Sheet.""" data_set = dict( type='service_account', project_id='sheets-connector-321907', private_key_id=config.GS_CONNECTOR_PRIVATE_KEY_ID, private_key=config.GS_CONNECTOR_PRIVATE_KEY, client_email=config.GS_CONNECTOR_CLIENT_EMAIL, client_id=config.GS_CONNECTOR_CLIENT_ID, auth_uri='https://accounts.google.com/o/oauth2/auth', token_uri='https://oauth2.googleapis.com/token', auth_provider_x509_cert_url='https://www.googleapis' '.com/oauth2/v1/certs', client_x509_cert_url=config.GS_CONNECTOR_CLIENT_X504_CERT_URL) json_dump = json.dumps(data_set) with open(config.GS_CONNECTOR_JSON_FILENAME, 'w') as outfile: outfile.write(json_dump) credentials = ServiceAccountCredentials.from_json_keyfile_name( config.GS_CONNECTOR_JSON_FILENAME, config.GS_CONNECTOR_SCOPES) http_auth = credentials.authorize(httplib2.Http()) service = apiclient.discovery.build('sheets', 'v4', http=http_auth) sheet = service.spreadsheets().get(spreadsheetId=sheet_id, includeGridData=True).execute() return sheet def download_artists_links_from_google_sheets(sheet_id): """Get data from Google Sheet having Sheet ID.""" sheet = get_google_sheet(sheet_id) raw_data = sheet.get('sheets')[0].get('data')[0].get('rowData') artists_links = list(map(lambda x: dict(zip( ['artist', 'instagram', 'spotify', 'youtube', 'soundcloud', 'shazam', 'tiktok', 'id', 'date'], [*x, *[None] * (9 - len(x))])), list(filter( lambda x: x[0] is not None, [[x.get('userEnteredValue').get('stringValue') if x.get( 'userEnteredValue') is not None else None for x in y.get('values')] for y in raw_data])) )) return artists_links def get_username_from_link(link: str, reverse=False): """Get username from link to profile.""" if reverse: return r'https://www.instagram.com/' + link + r'/' return link. \ replace(r'https://', ''). \ replace(r'www.', ''). \ replace(r'instagram.com/', ''). \ replace(r'/?hl=en', '').replace(r'/', '') class InstagramTracker: """Instagram Tracker class for scrapping data.""" def __init__(self, args_, path=None): """Initialize the tracker.""" logging.basicConfig(level=logging.INFO) self.__username__ = config.instagram['username'] self.__password__ = config.instagram['password'] self.__loader__ = import_session(args_.cookiefile or get_cookies(path), args_.sessionfile) self.__executor__ = InstagramStatsSFExecutor( sf_config=config.SF_CONFIG) self.__profiles__ = dict() self.__last_week_stats__ = dict() self.__first_time__ = False self.__unreachable_accounts__ = list() # Often we could just view other profiles as a guest, # but in case we need to read a lot of data, Instagram # will ask us about logging in. So, we could just put # our basic credentials and get 'unlimited access' to # public data and stats. time.sleep(0.02) self.__loader__.login(self.__username__, self.__password__) def get_profiles(self): """Return artists' profiles.""" return self.__profiles__ 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_last_week_stats(self): """Return artists' last week stats from SnowFlake.""" return self.__last_week_stats__ def get_executor(self): """Return SnowFlake executor instance.""" return self.__executor__ def update_artists_table(self): """Update artists' table.""" existing_artists = dict((username, dict( username=get_username_from_link(link), id=instagram_id)) for username, link, instagram_id, spotify, youtube, soundcloud, shazam, tiktok in self.__executor__.select_artists_links() if link is not None) updated_sheet = download_artists_links_from_google_sheets( sheet_id=config.SPREADSHEET_ID) updated_artists = [a.get('artist') for a in updated_sheet] artists_to_add = list(set(updated_artists) - set(existing_artists)) artist_to_delete = list(set(existing_artists) - set(updated_artists)) new_artists = [] for artist in updated_sheet: inst_id = None if artist.get('instagram') is not None: try: try: user = instaloader.Profile.from_username( self.__loader__.context, get_username_from_link(artist.get('instagram'))) except ConnectionException: artist.update({'id': str(inst_id)}) logging.error( 'Instaloader cannot reach %s page.', artist.get('instagram')) artist.update(reason='unreachable'), self.add_to_unreachable_profiles(artist) continue except ProfileNotExistsException: if artist.get('artist') not in existing_artists: logging.warning( "Update %s's Instagram profile link at GS.", artist.get('artist')) inst_id, username = '', get_username_from_link( artist.get('instagram')) artist.update(reason='wrong link'), self.add_to_unreachable_profiles(artist) else: id_ = existing_artists.get( artist.get('artist')).get('id') if id_ not in (None, 'None', ''): try: user = instaloader.Profile.from_id( self.__loader__.context, int(id_)) inst_id, username = user.userid, user.username logging.info( 'New username has been inserted for %s.', artist.get('instagram')) except ProfileNotExistsException: logging.warning( 'The account seems deleted for %s.', artist.get('instagram')) username = get_username_from_link( artist.get('instagram')) artist.update(reason='deleted'), self.add_to_unreachable_profiles(artist) else: logging.error( "Update %s's Instagram profile link at GS.", artist.get('artist')) inst_id, username = '', get_username_from_link( artist.get('instagram')) artist.update(reason='expired link'), self.add_to_unreachable_profiles(artist) else: inst_id, username = user.userid, user.username logging.info('ID has been inserted for %s.', artist.get('instagram')) artist['instagram'] = get_username_from_link( username, reverse=True) artist.update({'id': str(inst_id)}) for artist_to_add in artists_to_add: if artist_to_add == artist.get('artist'): artist.update({ 'date': datetime.datetime.today( ).strftime('%d-%m-%Y')}) artist.update({'id': str(inst_id)}) new_artists.append(artist) self.get_executor().update_artist_link_table(updated_sheet) for artist in artist_to_delete: self.get_executor().delete_artist_link_table(artist) self.get_executor().insert_artist_link_table(new_artists) def set_last_week_stats(self): """Set last week stats.""" self.__last_week_stats__.update( dict((artist_name, { 'link': instagram_link, 'verified': is_verified, 'current_followers': current_followers_count, 'weekly_change_in_followers': weekly_change_in_followers, 'last_week_followers': last_week_follower_count, 'percentage_change_in_followers': percentage_change_in_followers, 'current_posts': current_posts_count, 'weekly_change_in_posts': weekly_change_in_posts, 'last_week_posts': last_week_posts_count, 'current_likes': 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_comments': comments_count, 'weekly_change_in_comments': weekly_change_in_comments, 'current_engagement': engagement, 'last_week_engagement': last_week_engagement, 'last_processing_date': last_processing_date, 'last_week_processing_date': last_week_processing_date}) for artist_name, instagram_link, is_verified, current_followers_count, weekly_change_in_followers, last_week_follower_count, percentage_change_in_followers, current_posts_count, weekly_change_in_posts, last_week_posts_count, likes_count, last_week_likes_count, weekly_change_in_likes, percentage_change_in_likes, comments_count, weekly_change_in_comments, engagement, last_week_engagement, 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.""" executor = self.get_executor() query_result = executor.create_artist_link_table() if 'already exists' in query_result[0]: logging.info( '%s table already exists. Need to update it.', config.snowflake_table_names['artist_to_track'].upper()) self.update_artists_table() else: google_sheet_data = download_artists_links_from_google_sheets( sheet_id=config.SPREADSHEET_ID) for entry in google_sheet_data: if entry.get('date') is None: entry['date'] = today().strftime('%d-%m-%Y') executor.insert_artist_link_table(google_sheet_data) self.__profiles__.update(dict((username, dict( username=get_username_from_link(link))) for username, link, instagram_id, spotify, youtube, soundcloud, shazam, tiktok in self.__executor__.select_artists_links() if link is not None)) query_result = executor.create_instagram_statistics() if 'already exists' in query_result[0]: logging.info( '%s table already exists.', config.snowflake_table_names['instagram_statistics'].upper()) self.set_last_week_stats() else: self.__first_time__ = True query_result = executor.create_staging_raw_instagram_statistics() if 'already exists' in query_result[0]: logging.info('%s table already exists.', config.snowflake_table_names['staging_raw'].upper()) 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 current and last week statistics.""" 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] = { 'link': None, 'verified': None, 'current_followers': 0, 'weekly_change_in_followers': 0, 'last_week_followers': 0, 'percentage_change_in_followers': 0, 'current_posts': 0, 'weekly_change_in_posts': 0, 'last_week_posts': 0, 'current_likes': 0, 'weekly_change_in_likes': 0, 'percentage_change_in_likes': 0, 'current_comments': 0, 'weekly_change_in_comments': 0, 'current_engagement': 0, 'last_week_engagement': 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 = \ zero_if_none(lasts.get(artist).get('current_' + stat_name)) last_week_stat = \ zero_if_none(lasts.get(artist).get('last_week_' + stat_name) if stat_name != 'comments' else lasts.get(artist).get('current_' + stat_name)) last_week_processing_date = \ zero_if_none(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) percentage_change = ( current_weekly_change / (current - current_weekly_change) if (current - current_weekly_change) != 0 else 0) 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', 'likes'] and 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 load_profile(self, artist, log=False): """Load artist's Instagram profile. Args: artist (str): artist's name. log (bool): log loading info or not. """ username = self.get_profiles().get(artist).get('username') if log: logging.info('\n%s, instagram username: %s\n', artist, username) profile = instaloader.Profile.from_username( self.__loader__.context, username) return profile @catchers.catch_all_and_print(sentry_dsn=os.getenv('SENTRY_DSN')) def get_common_stats(self, artist): """Return count of followers, followings, posts. Also returns 'VERIFIED' if artist's page is verified and '' if it's not, and change in these stats. """ profiles = self.get_profiles() stats = self.get_last_week_stats() profile = self.load_profile(artist, log=True) followers = profile.followers followings = profile.followees is_verified = profile.is_verified profiles[artist].update(self.compare_current_and_last_week_stats( followers, 'followers', stats, artist)) logging.info('Followings: %i', followings) profiles[artist].update({'verified': is_verified, 'followees': followings}) @catchers.catch_all_and_print(sentry_dsn=os.getenv('SENTRY_DSN')) def get_posts_stats(self, artist): """Return count of posts, posts likes and engagement.""" profile = self.load_profile(artist) profiles = self.get_profiles() stats = self.get_last_week_stats() total_posts, last_year_posts = profile.get_posts(), 0 posts_likes = 0 posts_comments = 0 for post in total_posts: if today() + datetime.timedelta(days=365) \ > post.date_utc.date(): posts_likes += post.likes posts_comments += post.comments last_year_posts += 1 if total_posts.count == 0: logging.info('Seems like %s has deleted all the' ' posts on their account.', artist) profiles[artist].update(self.compare_current_and_last_week_stats( total_posts.count, 'posts', stats, artist)) profiles[artist].update(self.compare_current_and_last_week_stats( posts_likes, 'likes', stats, artist)) profiles[artist].update(self.compare_current_and_last_week_stats( posts_comments, 'comments', stats, artist)) engagement = posts_likes / (profile.followers * last_year_posts) if \ profile.followers * last_year_posts != 0 else 0 profiles[artist].update(self.compare_current_and_last_week_stats( engagement, 'engagement', stats, artist)) 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, username=profiles[artist].get('username'), verified=profiles[artist].get('verified'), followings=profiles[artist].get('followings'), )), days.sort() if missing_days > 0: missing_daily_followers = profiles[artist].get( 'daily_change_in_followers') / missing_days missing_daily_likes = zero_if_none(profiles[artist].get( 'daily_change_in_likes')) / missing_days missing_daily_comments = zero_if_none(profiles[artist].get( 'daily_change_in_comments')) / missing_days missing_daily_posts = \ (profiles[artist].get('current_posts') - stats[artist].get('current_posts')) / missing_days missing_daily_engagement = \ (profiles[artist].get('current_engagement') - stats[artist].get('current_engagement')) / missing_days for day in days: today_, stats = day, \ self.get_last_week_stats() # calculating actual metrics for followers, likes, and comments missing_days_stats.update( self.compare_current_and_last_week_stats( stats[artist].get('current_followers') + missing_daily_followers, 'followers', 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_comments') + missing_daily_comments, 'comments', stats, artist, today=today_)) missing_days_stats.update( self.compare_current_and_last_week_stats( stats[artist].get('current_posts') + missing_daily_posts, 'posts', stats, artist, today=today_)) missing_days_stats.update( self.compare_current_and_last_week_stats( stats[artist].get('current_engagement') + missing_daily_engagement, 'engagement', stats, artist, today=today_)) # updating stats table to keep the growth gradual self.get_executor().update_artist_stats( artist, missing_days_stats) # updating staging raw table 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 w. stats with the newest data for 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 all stats for 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) logging.info("%s's stats has been inserted.\n", profile) def get_all_stats_and_update(self): """Update stats for every single artist.""" tracking_start = datetime.datetime.now() for artist in list(self.get_profiles()): self.__first_time__ = False try: start_ = datetime.datetime.now() self.get_common_stats(artist) self.get_posts_stats(artist) finish = datetime.datetime.now() logging.info('For %s it took %i seconds.\n', artist, (finish - start_). total_seconds()) except ProfileNotExistsException as error: if ' does not exist.' in error.args[0]: logging.error( "%s's instagram username seems" ' to be changed.\n', artist) except ConnectionException: time.sleep(140) logging.info( "%s's statistics is not available " 'now due to some technical limitations.\n', artist) else: # 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) 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['instagram_statistics']) self.get_executor().update_stating_raw_instagram_stats() @catchers.catch_all_and_print(sentry_dsn=os.getenv('SENTRY_DSN')) def perform_tracking(self): """Perform tracking with instaloader.""" self.create_tables() 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('instagram') + ' - ' + x.get('reason'), self.get_unreachable_profiles()))), username='A&R Support', url=os.getenv('SLACK_WEBHOOK_URL'), icon_emoji=':instagram:') self.get_all_stats_and_update() # self.update_staging_raw_table() if __name__ == '__main__': p = ArgumentParser() p.add_argument('-c', '--cookiefile') p.add_argument('-f', '--sessionfile') args = p.parse_args() tracker = InstagramTracker(args) tracker.perform_tracking()