import pandas as pd import numpy as np import re import time import datetime as dt from dateutil.relativedelta import relativedelta, FR import pytz import requests import boto3 import sys import os import gspread from gspread_formatting import ( ConditionalFormatRule, GridRange, BooleanRule, BooleanCondition, CellFormat, Color, format_cell_range, get_conditional_format_rules, TextFormat ) import json from google.oauth2.service_account import Credentials from googleapiclient.discovery import build def setup_logger(): import logging logger = logging.getLogger() logger.setLevel(logging.INFO) # Add a logger handler if none exists if not logger.hasHandlers(): handler = logging.StreamHandler() formatter = logging.Formatter('%(asctime)s - %(levelname)s - %(message)s') handler.setFormatter(formatter) logger.addHandler(handler) return logger def init_secrets(): session = boto3.session.Session() client = session.client( service_name='secretsmanager', region_name='us-east-1' ) sf_user = client.get_secret_value(SecretId="dev/awal-ar/SNOWFLAKE_USER")["SecretString"] sf_account = client.get_secret_value(SecretId="dev/awal-ar/SNOWFLAKE_ACCOUNT")["SecretString"] sf_warehouse = client.get_secret_value(SecretId="dev/awal-ar/SNOWFLAKE_WAREHOUSE")["SecretString"] sf_password = client.get_secret_value(SecretId="dev/awal-ar/PEM_KEY_PASSWORD")["SecretString"] sf_pem_key = client.get_secret_value(SecretId="dev/awal-ar/PEM_KEY")["SecretString"] gspread_creds = client.get_secret_value(SecretId="dev/awal-weekly-report/GSPREAD_CREDS")["SecretString"] gspread_creds = gspread_creds.encode('utf-8').decode('unicode_escape').encode("utf-8") gspread_creds = json.loads(gspread_creds) from cryptography.hazmat.backends import default_backend from cryptography.hazmat.primitives import serialization p_key = serialization.load_pem_private_key( sf_pem_key.encode('utf-8').decode('unicode_escape').encode("utf-8"), password=sf_password.encode('utf-8'), backend=default_backend() ) pkb = p_key.private_bytes( encoding=serialization.Encoding.DER, format=serialization.PrivateFormat.PKCS8, encryption_algorithm=serialization.NoEncryption()) sf_secrets = { 'sf_user':sf_user, 'sf_account':sf_account, 'sf_warehouse':sf_warehouse, 'sf_password':sf_password, 'pkb':pkb } return sf_secrets,gspread_creds def orcd_query(sql_query,sf_params): ctx = sf_params['snowflake'].connector.connect( user=sf_params['sf_user'], private_key=sf_params['pkb'], account=sf_params['sf_account'], warehouse=sf_params['sf_warehouse'] ) cs = ctx.cursor(sf_params['snowflake'].connector.DictCursor) try: cs.execute(sql_query) result = cs.fetchall() finally: cs.close() ctx.close() result = pd.DataFrame(result) return result def find_sheet_in_folder(name, folder_id, drive_service): query = ( f"'{folder_id}' in parents and " f"name = '{name}' and " f"mimeType = 'application/vnd.google-apps.spreadsheet'" ) results = drive_service.files().list( q=query, fields="files(id, name)" ).execute() return results.get("files", []) # Helper: Convert 0-based index to column letter (A, B, C, ..., AA, AB, etc.) def col_num_to_letter(n): result = '' while n >= 0: result = chr(n % 26 + ord('A')) + result n = n // 26 - 1 return result def read_and_compare_listings(sh,worksheet,new_df,sheet_exists): # read existing data so we can compare for new listings list_of_lists = worksheet.get_all_values() if not list_of_lists[0]: # if empty, meaning the sheet exists but is blank header_exists = False headers = None header_buffer = 0 sheet_exists = False # delete the sheet sh.del_worksheet(worksheet) return worksheet,new_df,headers,header_exists,header_buffer,sheet_exists elif list_of_lists[2]!=['DSP','ARTIST_NAME','TRACK_NAME','PLAYLIST_NAME','PLAYLIST_FOLLOWERS','TRACK_POSITION','NUMBER_OF_TRACKS','TRACK_ADD_DATE','PLAYLIST_URL','CURATOR_COUNTRY','STATUS','TIME_PULLED','ISRC']: # the df column headers do not match / corrupted header_exists = False headers = None header_buffer = 0 sheet_exists = False # delete the sheet sh.del_worksheet(worksheet) return worksheet,new_df,headers,header_exists,header_buffer,sheet_exists # skip first 2 rows (those are expected to be blank bc of the logo etc) if all(item == '' for item in list_of_lists[0]) or list_of_lists[0][0]=='#REF!': header_exists = True header_buffer = 2 list_of_lists = list_of_lists[2:] else: header_exists = False header_buffer = 0 headers = list_of_lists[0] sheet_df = pd.DataFrame(list_of_lists[1:], columns=headers) if len(sheet_df)>0: if 'STATUS' not in sheet_df.columns: sheet_df['STATUS'] = None sheet_df['TRACK_POSITION'] = sheet_df['TRACK_POSITION'].astype('int64') # identify new rows in my df and insert for dsp in ['AMAZON','APPLE','SPOTIFY','DEEZER']: df_0 = sheet_df.loc[sheet_df['DSP']==dsp].reset_index() # reset index to get the index number df_1 = new_df.loc[new_df['DSP']==dsp].reset_index() if len(df_0)>0 or len(df_1)>0: # new playlists new_playlists_idx = df_1.loc[~df_1['PLAYLIST_NAME'].isin(df_0['PLAYLIST_NAME'])]['index'] if len(new_playlists_idx)>0: df_1 = df_1.drop(df_1.loc[df_1['index'].isin(new_playlists_idx)].index) # check for update in track position merge_cols = ['DSP','ISRC','ARTIST_NAME','TRACK_NAME','PLAYLIST_NAME','TRACK_POSITION'] merged_df = pd.merge(df_0, df_1, on=merge_cols, how='outer', indicator=True) update_position_idx = merged_df.loc[merged_df['_merge']=='right_only']['index_y'] # check for update in territories if dsp in ['APPLE','AMAZON']: merge_cols = ['DSP','ISRC','ARTIST_NAME','TRACK_NAME','PLAYLIST_NAME','PLAYLIST_FOLLOWERS'] merged_df = pd.merge(df_0, df_1, on=merge_cols, how='outer', indicator=True) update_territory_idx = merged_df.loc[merged_df['_merge']=='right_only']['index_y'] else: update_territory_idx = [] if (len(update_position_idx)>0) or (len(update_territory_idx)>0) or (len(new_playlists_idx)>0): new_df['STATUS'] = np.where( new_df.reset_index()['index'].isin(new_playlists_idx), 'NEW PLAYLIST - ' + new_df['TIME_PULLED'], np.where( new_df.reset_index()['index'].isin(update_territory_idx), 'UPDATE: TERRITORIES - ' + new_df['TIME_PULLED'], np.where( new_df.reset_index()['index'].isin(update_position_idx), 'UPDATE: POSITION - ' + new_df['TIME_PULLED'], new_df['STATUS'] ) ) ) # attach previous status to all "stayed the same" listings # this has to go here so that apple/amazon includes playlist_followers column keep_status = merged_df.loc[merged_df['_merge']=='both'][['index_y','STATUS_x']] if len(keep_status)>0: # grabbing previous status of listings already marked as new keep_status.rename(columns={'index_y':'index'},inplace=True) new_df = new_df.reset_index().merge(keep_status,how='left',on='index') new_df['STATUS'] = np.where( new_df.reset_index()['index'].isin(keep_status['index']), new_df['STATUS_x'], new_df['STATUS'] ) new_df = new_df.drop(['STATUS_x','index'],axis=1) return worksheet,new_df,headers,header_exists,header_buffer,sheet_exists def upload_sheet_contents(artist_df,sheet_exists,worksheet,headers,header_buffer): # Upload to_upload = artist_df.copy() to_upload = to_upload[['DSP','ARTIST_NAME','TRACK_NAME','PLAYLIST_NAME','PLAYLIST_FOLLOWERS','TRACK_POSITION','NUMBER_OF_TRACKS','TRACK_ADD_DATE','PLAYLIST_URL','CURATOR_COUNTRY','STATUS','TIME_PULLED','ISRC']] to_upload['TRACK_NAME'] = "\'" + to_upload['TRACK_NAME'] to_upload['PLAYLIST_NAME'] = "\'" + to_upload['PLAYLIST_NAME'] if not sheet_exists: to_upload['STATUS'] = 'NEW PLAYLIST - ' + to_upload['TIME_PULLED'] headers = to_upload.columns.tolist() else: worksheet.clear() # headers = headers data = [to_upload.columns.tolist()] + to_upload.values.tolist() worksheet.update(f'A{str(1+header_buffer)}', data, value_input_option="USER_ENTERED") return to_upload,headers def format_sheet(sh,worksheet,metadata,logger,sheet_exists,to_upload,headers,header_exists,header_buffer): rules = get_conditional_format_rules(worksheet) rules.clear() requests_sh = [] cols = [i for i, col in enumerate(headers)] name_cols = [i for i, col in enumerate(headers) if col.endswith(('TRACK_NAME','PLAYLIST_NAME','CURATOR_COUNTRY'))] territory_col = [i for i, col in enumerate(headers) if col.endswith(('PLAYLIST_FOLLOWERS'))] dsp_col = [i for i, col in enumerate(headers) if col.endswith(('DSP'))] artist_col = [i for i, col in enumerate(headers) if col.endswith(('ARTIST_NAME'))] cols_to_hide = [i for i, col in enumerate(headers) if col.endswith(('ISRC','TIME_PULLED'))] status_col = [i for i, col in enumerate(headers) if col.endswith(('STATUS'))] for col_index in cols: col_letter = col_num_to_letter(col_index) col_range = f"{col_letter}{str(2+header_buffer)}:{col_letter}" if col_index in name_cols: pixel_size = 175 elif col_index in artist_col: pixel_size = 125 elif col_index in territory_col: pixel_size = 200 if (col_index in name_cols) or (col_index in artist_col) or (col_index in territory_col): # col width requests_sh.append({ "updateDimensionProperties": { "range": { "sheetId": worksheet._properties['sheetId'], "dimension": "COLUMNS", "startIndex": col_index, "endIndex": col_index+1 }, "properties": { "pixelSize": pixel_size }, "fields": "pixelSize" } }) if col_index in dsp_col: thresholds = [ ('AMAZON', Color(1.0, 0.65, 0.0)), # orange ('APPLE', Color(0.9, 0.9, 0.9)), # light gray ('DEEZER', Color(0.7, 0.85, 1.0)), # light blue ('SPOTIFY', Color(0.8, 1.0, 0.8)), # light green ] for which_dsp, color in thresholds: rule = ConditionalFormatRule( ranges=[GridRange.from_a1_range(col_range, worksheet)], booleanRule=BooleanRule( condition=BooleanCondition('CUSTOM_FORMULA', [f'=${col_letter}{str(2+header_buffer)}="{which_dsp}"']), format=CellFormat(backgroundColor=color) ) ) rules.append(rule) if col_index in cols_to_hide: # hide columns requests_sh.append({ "updateDimensionProperties": { "range": { "sheetId": worksheet._properties['sheetId'], "dimension": "COLUMNS", "startIndex": col_index, "endIndex": col_index+1 }, "properties": { "hiddenByUser": True }, "fields": "hiddenByUser" } }) ## create header if not exists logger.info('Header...') if not header_exists: # insert rows / merge cells. this will already be done if there was a header to begin with, but still need to re-insert the img worksheet.insert_rows([[], []], row=1) # insert 2 rows at the top requests_sh.append({ "mergeCells": { "range": { "sheetId": worksheet._properties['sheetId'], "startRowIndex": 0, "endRowIndex": 1, "startColumnIndex": 0, "endColumnIndex": 3 }, "mergeType": "MERGE_ALL" } }) requests_sh.append({ "updateDimensionProperties": { "range": { "sheetId": worksheet._properties['sheetId'], "dimension": "ROWS", "startIndex": 0, "endIndex": 1, }, "properties": { "pixelSize": 100 }, "fields": "pixelSize", } }) # insert logo regardless worksheet.update_acell("A1", f'=IMAGE("https://lh3.googleusercontent.com/d/1Mpzp-XjdBdTsE6i1VjT1qW4LC0vW4CZR=w400-h400")') # font fmt = CellFormat( textFormat = TextFormat( fontFamily='Karla', fontSize=10 ) ) format_cell_range(worksheet, f"A1:{col_num_to_letter(len(headers))}{len(to_upload)+3}", fmt) ## Protect the sheet except for status column logger.info('Set sheet protection rules...') # check for existing rules; add only if none exist for the current sheet if sheet_exists: sheet_metadata = metadata["sheets"][worksheet.index] existing_rules = sheet_metadata.get("protectedRanges", []) if len(existing_rules) > 0: # protection already exists add_protection_rules = False else: add_protection_rules = True if (not sheet_exists) or add_protection_rules: # Protect rows ABOVE editable status block requests_sh.append({ "addProtectedRange": { "protectedRange": { "range": { "sheetId": worksheet._properties['sheetId'], "startRowIndex": 0, "endRowIndex": 3 }, "warningOnly": True } } }) # Protect rows BELOW editable status block until the end of the sheet requests_sh.append({ "addProtectedRange": { "protectedRange": { "range": { "sheetId": worksheet._properties['sheetId'], "startRowIndex": len(to_upload)+3, "endRowIndex": worksheet.row_count }, "warningOnly": True } } }) # Protect columns LEFT of editable status block requests_sh.append({ "addProtectedRange": { "protectedRange": { "range": { "sheetId": worksheet._properties['sheetId'], "startRowIndex": 3, "endRowIndex": len(to_upload)+3, "startColumnIndex": 0, "endColumnIndex": status_col[0] }, "warningOnly": True } } }) # Protect columns RIGHT of editable status block requests_sh.append({ "addProtectedRange": { "protectedRange": { "range": { "sheetId": worksheet._properties['sheetId'], "startRowIndex": 3, "endRowIndex": len(to_upload)+3, "startColumnIndex": status_col[0]+1, "endColumnIndex": 100 }, "warningOnly": True } } }) ## save any conditional formatting rules ## rules.save() ## sh.batch_update({"requests": requests_sh}) def handler(event,context): logger = setup_logger() logger.info('Starting handler function...') import snowflake.connector sf_params,gspread_creds = init_secrets() sf_params['snowflake'] = snowflake logger.info('Connecting to google...') SCOPES = [ "https://www.googleapis.com/auth/spreadsheets", "https://www.googleapis.com/auth/drive" ] creds = Credentials.from_service_account_info( gspread_creds, scopes=SCOPES ) gc = gspread.authorize(creds) drive_service = build("drive", "v3", credentials=creds) FOLDER_ID = "1TjYIfCLxLN7eSSPf1mp4b3omFDZk7Z59" et_timezone = pytz.timezone('Europe/London') today = dt.datetime.now(et_timezone).date() friday_date = today - relativedelta(weekday=FR(-1)) f_name = f'{friday_date.strftime("%m/%d/%y")} - playlist tracker' # check if spreadsheet for this week exists. if yes, open it; if not, create it files = find_sheet_in_folder(f_name, FOLDER_ID, drive_service) if files: sheet_exists = True else: sheet_exists = False if sheet_exists: sheet_id = files[0]["id"] logger.info(f'Found spreadsheet for {friday_date.strftime("%m/%d/%y")}') else: logger.info(f'Spreadsheet for {friday_date.strftime("%m/%d/%y")} not found; creating...') spreadsheet = gc.create(f_name) sheet_id = spreadsheet.id # Move file into folder # Retrieve current parents file = drive_service.files().get( fileId=sheet_id, fields="parents" ).execute() previous_parents = ",".join(file.get("parents")) drive_service.files().update( fileId=sheet_id, addParents=FOLDER_ID, removeParents=previous_parents, fields="id, parents" ).execute() sh = gc.open_by_key(sheet_id) # read any existing protection rules metadata = sh.fetch_sheet_metadata() # get listings for this week df = orcd_query(f""" with country_list as (select countryname,country_code from FACTS.PROD.DIM_COUNTRY order by country_code), -- SPOTIFY -- latest_time_spotify as ( select max(time_pulled) latest_time from AWAL.AWAL_AR.PLAYLIST_TRACKER_SPOTIFY ), pre_result_spotify as ( select p.* , ROW_NUMBER() OVER (PARTITION BY p.isrc,p.playlist_url ORDER BY p.track_add_date DESC) AS rn from AWAL.AWAL_AR.PLAYLIST_TRACKER_SPOTIFY p join latest_time_spotify on latest_time_spotify.latest_time = p.time_pulled ), result_spotify as ( select * exclude rn , case when playlist_type='NEW_MUSIC_FRIDAY' then 0 when playlist_type='CHART' and playlist_name RLIKE '^Top [0-9]+( .*)?$' then 1 when playlist_type='CHART' and playlist_name like 'Viral %' then 2 when curator_country like 'Spotify Chart' then 3 when curator_country like 'Spotify' then 4 when curator_country like 'Spotify US' then 5 when curator_country like 'Spotify%' then 6 when curator_country like 'Filtr%' then 7 else 8 end as order_key, ROW_NUMBER() OVER (ORDER BY artist_name asc,track_name asc,order_key asc,curator_country asc,playlist_followers desc) AS row_num from pre_result_spotify where rn=1 ), result_spotify_final as ( select 'SPOTIFY' as dsp, isrc, artist_name, track_name, playlist_name, playlist_followers, track_position, number_of_tracks, track_add_date, playlist_url, playlist_type, curator_country, time_pulled, order_key+row_num as order_key from result_spotify ), -- APPLE -- latest_time_apple as ( select max(time_pulled) latest_time from AWAL.AWAL_AR.PLAYLIST_TRACKER_APPLE ), pre_result_apple as ( select p.* , ROW_NUMBER() OVER (PARTITION BY p.isrc,p.playlist_url ORDER BY p.track_add_date DESC) AS rn from AWAL.AWAL_AR.PLAYLIST_TRACKER_APPLE p join latest_time_apple on latest_time_apple.latest_time = p.time_pulled ), result_apple as ( select pre_result_apple.* exclude rn from pre_result_apple where rn=1 order by isrc,playlist_name ), result_apple_agg as ( select r.isrc, r.artist_name, r.track_name, r.playlist_name, r.track_position, r.number_of_tracks, r.track_add_date, r.playlist_url, r.curator_name, r.n_territories, LISTAGG(c.countryname, ', ') WITHIN GROUP (ORDER BY f.index) AS territories, r.time_pulled, case when r.playlist_name='New Music Daily' then 0 when r.curator_name like 'Apple%' then 1 when r.curator_name ilike 'Filtr%' then 2 else 3 end as order_key from result_apple r, LATERAL FLATTEN(input => SPLIT(r.territories, ',')) f JOIN country_list c ON TRIM(f.value) = c.country_code group by all order by artist_name,track_name,order_key,curator_name,playlist_name ), result_apple_final as ( select 'APPLE' as dsp, isrc, artist_name, track_name, playlist_name, concat(n_territories,' countries incl ',territories) as territories, track_position, number_of_tracks, track_add_date, playlist_url, curator_name, time_pulled, order_key from result_apple_agg ), -- AMAZON -- latest_time_amazon as ( select max(time_pulled) latest_time from AWAL.AWAL_AR.PLAYLIST_TRACKER_amazon ), pre_result_amazon as ( select p.* , ROW_NUMBER() OVER (PARTITION BY p.isrc,p.playlist_url ORDER BY p.track_add_date DESC) AS rn from AWAL.AWAL_AR.PLAYLIST_TRACKER_amazon p join latest_time_amazon on latest_time_amazon.latest_time = p.time_pulled ), result_amazon as ( select pre_result_amazon.* exclude rn from pre_result_amazon where rn=1 order by isrc,playlist_name ), result_amazon_agg as ( select r.isrc, r.artist_name, r.track_name, r.playlist_name, r.track_position, r.number_of_tracks, r.track_add_date, r.playlist_url, r.n_territories, LISTAGG(c.countryname, ', ') WITHIN GROUP (ORDER BY f.index) AS territories, r.time_pulled, 1 as order_key from result_amazon r, LATERAL FLATTEN(input => SPLIT(r.territories, ',')) f JOIN country_list c ON TRIM(f.value) = c.country_code group by all order by artist_name,track_name,order_key,playlist_name ), result_amazon_final as ( select 'AMAZON' as dsp, isrc, artist_name, track_name, playlist_name, concat(n_territories,' countries incl ',territories) as territories, track_position, number_of_tracks, track_add_date, playlist_url, time_pulled, order_key from result_amazon_agg ), -- DEEZER -- latest_time_deezer as ( select max(time_pulled) latest_time from AWAL.AWAL_AR.PLAYLIST_TRACKER_deezer ), pre_result_deezer as ( select p.* , ROW_NUMBER() OVER (PARTITION BY p.isrc,p.playlist_url ORDER BY p.track_add_date DESC) AS rn from AWAL.AWAL_AR.PLAYLIST_TRACKER_deezer p join latest_time_deezer on latest_time_deezer.latest_time = p.time_pulled ), result_deezer as ( select * exclude rn , CASE when curator_country='Deezer Charts' then 0 when curator_country ilike '%filtr%' then 1 when curator_country like 'Top Playlists' then 2 when ((curator_country like '%Deezer%') and (curator_country like '%Editor')) then 3 when curator_country like 'Topsify%' then 4 else 5 END as order_key, ROW_NUMBER() OVER (ORDER BY artist_name asc,track_name asc,order_key asc,curator_country asc,playlist_followers desc) AS row_num from pre_result_deezer where rn=1 ), result_deezer_final as ( select 'DEEZER' as dsp, isrc, artist_name, track_name, playlist_name, playlist_followers, track_position, number_of_tracks, track_add_date, playlist_url, curator_country, time_pulled, order_key+row_num order_key from result_deezer ), -- RESULT -- combined as ( select a.dsp, a.isrc, a.artist_name, a.track_name, a.playlist_name, a.territories as playlist_followers, a.track_position, a.number_of_tracks, a.track_add_date, a.playlist_url, a.time_pulled, a.order_key, a.curator_name as curator_country from result_apple_final a UNION select s.dsp, s.isrc, s.artist_name, s.track_name, s.playlist_name, to_char(s.playlist_followers, 'FM999,999,999,999') as playlist_followers, s.track_position, s.number_of_tracks, s.track_add_date, s.playlist_url, s.time_pulled, s.order_key, s.curator_country from result_spotify_final s UNION select a.dsp, a.isrc, a.artist_name, a.track_name, a.playlist_name, a.territories as playlist_followers, a.track_position, a.number_of_tracks, a.track_add_date, a.playlist_url, a.time_pulled, a.order_key, 'N/A' as curator_country from result_amazon_final a UNION select a.dsp, a.isrc, a.artist_name, a.track_name, a.playlist_name, to_char(a.playlist_followers, 'FM999,999,999,999') as playlist_followers, a.track_position, a.number_of_tracks, a.track_add_date, a.playlist_url, a.time_pulled, a.order_key, a.curator_country from result_deezer_final a ), latest_friday as ( SELECT CASE WHEN DAYOFWEEK(CURRENT_DATE()) = 5 THEN CURRENT_DATE() ELSE PREVIOUS_DAY(CURRENT_DATE(), 'FRIDAY') END AS latest_friday_date ), final_table as ( select distinct CASE when da.artistname like 'Juan Luis Guerra%' then 'Juan Luis Guerra 4.40' else da.artistname end as artist_name, c.* exclude(artist_name) from combined c join FACTS.PROD.DIM_TRACK dt on dt.isrc=c.isrc join FACTS.PROD.DIM_RELEASE dr on dt.upc=dr.releaseid join facts.prod.dim_artist da on da.artistid=dr.artistid join awal.awal_ar.playlist_roster_tracks rt on rt.isrc=c.isrc join latest_friday lf on rt.release_week=lf.latest_friday_date join facts.prod.dim_label dl on dr.labelid=dl.labelid where dl.brandid in ('31f4f0f0-cbb4-4a2c-9eb0-d7288c5a2588','d25a4cd1-e820-45f2-be5c-56edcfeb8298') order by artist_name,c.dsp,c.isrc,c.order_key,c.curator_country,c.playlist_name ) select * from final_table ; """,sf_params) for_artist_report = orcd_query('select lower(artist) as artist from awal.awal_ar.playlist_roster_artists',sf_params) artist_reports = [] for_catalog_report = orcd_query('select isrc,catalog_name from awal.awal_ar.playlist_roster_catalogs',sf_params) catalog_reports = {} if len(for_catalog_report) > 0: catalog_reports = {k: [] for k in for_catalog_report['CATALOG_NAME'].unique()} df['TRACK_ADD_DATE'] = pd.to_datetime(df['TRACK_ADD_DATE']) df['TIME_PULLED'] = pd.to_datetime(df['TIME_PULLED']) # remove plus sign's because won't print correctly in spreadsheet df = df.replace(r'\+', '', regex=True) # convert dates to strings df["TRACK_ADD_DATE"] = df["TRACK_ADD_DATE"].dt.strftime("%Y-%m-%d").fillna("") df["TIME_PULLED"] = df["TIME_PULLED"].dt.strftime("%m/%d %H:%M") df['TRACK_POSITION'] = df['TRACK_POSITION'].astype('int64') # normalize capitalization in case of artist variants (uppercase, lowercase etc) artist_map = ( df.groupby(df['ARTIST_NAME'].str.casefold())['ARTIST_NAME'] .first() .to_dict() ) df['ARTIST_NAME'] = ( df['ARTIST_NAME'] .str.casefold() .map(artist_map) ) artists = df['ARTIST_NAME'].drop_duplicates() i = 1 j = 1 for artist in artists: if i%10==0: logger.info('Pause...') time.sleep(15) artist_df = df.loc[df['ARTIST_NAME']==artist] artist_df['STATUS'] = None # prep data for artist-level reports if applicable if "ARTIST" in for_artist_report.columns and artist.lower() in for_artist_report['ARTIST'].tolist(): artist_reports.append(artist_df) # skip current sheet if putting into artist report logger.info(f'Skipping current artist {artist} - moving to artist level report') i=i+1 continue # prep data for catalog-level reports if applicable elif "ISRC" in for_catalog_report.columns: catalog_df = for_catalog_report.merge(artist_df,how='inner',on='ISRC') if len(catalog_df) > 0: catalog_name = catalog_df['CATALOG_NAME'].drop_duplicates().iloc[0] catalog_reports[catalog_name].append(artist_df) # skip current sheet if putting into catalog report logger.info(f'Skipping current artist {artist} - moving to catalog level report') i=i+1 continue sheet_exists = False # check if sheet for that artist exists. if not, create it. try: logger.info(f'Loading sheet for artist {j} of {len(artists)}: {artist}') worksheet = next( ws for ws in sh.worksheets() if ws.title.lower() == artist.lower() ) sheet_exists = True except gspread.exceptions.WorksheetNotFound: logger.info('Sheet not found; creating new sheet') except StopIteration: logger.info('Sheet not found; creating new sheet 1') i = i+1 j = j+1 if sheet_exists: worksheet,artist_df,headers,header_exists,header_buffer,sheet_exists = read_and_compare_listings(sh,worksheet,artist_df,sheet_exists) if not sheet_exists: # IF statement instead of ELSE statement because the previous line can still return sheet_exists=False # sheet does not exist; create logger.info('Creating new sheet...') worksheet = sh.add_worksheet(title=artist, rows=f"{len(artist_df)+3}", cols="100") header_exists = False headers = None header_buffer = 0 to_upload, headers = upload_sheet_contents(artist_df,sheet_exists,worksheet,headers,header_buffer) ## SHEET FORMATTING ## logger.info('Formatting...') format_sheet(sh,worksheet,metadata,logger,sheet_exists,to_upload,headers,header_exists,header_buffer) logger.info('Complete') ###################################################### ###################################################### # ARTIST LEVEL REPORTS ###################################################### ###################################################### logger.info('ARTIST LEVEL REPORTS...') if len(artist_reports) >= 1: FOLDER_ID = "1jYQjcNTyVtVDS5HKNGSOX1VpH26HMpe-" # spreadsheet per artist for artist_report in artist_reports: artist = artist_report['ARTIST_NAME'].drop_duplicates().tolist()[0] logger.info(artist) f_name = f'{artist} - playlist tracker' # check if spreadsheet for this artist exists. if yes, open it; if not, create it files = find_sheet_in_folder(f_name, FOLDER_ID, drive_service) if files: sheet_exists = True else: sheet_exists = False if sheet_exists: sheet_id = files[0]["id"] logger.info(f'Found spreadsheet for {artist}') else: logger.info(f'Spreadsheet for {artist} not found; creating...') spreadsheet = gc.create(f_name) sheet_id = spreadsheet.id # Move file into folder # Retrieve current parents file = drive_service.files().get( fileId=sheet_id, fields="parents" ).execute() previous_parents = ",".join(file.get("parents")) drive_service.files().update( fileId=sheet_id, addParents=FOLDER_ID, removeParents=previous_parents, fields="id, parents" ).execute() sh = gc.open_by_key(sheet_id) # read any existing protection rules metadata = sh.fetch_sheet_metadata() tracks = artist_report['TRACK_NAME'].drop_duplicates() j = 1 # sheet per track for track in tracks: if i%10==0: logger.info('Pause...') time.sleep(15) track_df = artist_report.loc[artist_report['TRACK_NAME'] == track] track_df['STATUS'] = None sheet_exists = False # check if sheet for that track exists. if not, create it. try: logger.info(f'Loading sheet for track {j} of {len(tracks)}: {track}') worksheet = next( ws for ws in sh.worksheets() if ws.title.lower() == track.lower() ) sheet_exists = True except gspread.exceptions.WorksheetNotFound: logger.info('Sheet not found; creating new sheet') except StopIteration: logger.info('Sheet not found; creating new sheet 1') j = j+1 i = i+1 if sheet_exists: worksheet,track_df,headers,header_exists,header_buffer,sheet_exists = read_and_compare_listings(sh,worksheet,track_df,sheet_exists) if not sheet_exists: # IF statement instead of ELSE statement because the previous line can still return sheet_exists=False # sheet does not exist; create logger.info('Creating new sheet...') worksheet = sh.add_worksheet(title=track, rows=f"{len(track_df)+3}", cols="100") header_exists = False headers = None header_buffer = 0 to_upload, headers = upload_sheet_contents(track_df,sheet_exists,worksheet,headers,header_buffer) ## SHEET FORMATTING ## logger.info('Formatting...') format_sheet(sh,worksheet,metadata,logger,sheet_exists,to_upload,headers,header_exists,header_buffer) logger.info('Complete') else: logger.info('No artists for artist level reports') ###################################################### ###################################################### # CATALOG LEVEL REPORTS ###################################################### ###################################################### logger.info('CATALOG LEVEL REPORTS...') if len(catalog_reports) >= 1: FOLDER_ID = "1j25WrygS-QtFjvcaRQr5AxpwgIxoIGGm" for which_catalog in catalog_reports: logger.info(which_catalog) f_name = f'{which_catalog} - playlist tracker' # check if spreadsheet for this catalog exists. if yes, open it; if not, create it files = find_sheet_in_folder(f_name, FOLDER_ID, drive_service) if files: sheet_exists = True else: sheet_exists = False if sheet_exists: sheet_id = files[0]["id"] logger.info(f'Found spreadsheet for {which_catalog}') else: logger.info(f'Spreadsheet for {which_catalog} not found; creating...') spreadsheet = gc.create(f_name) sheet_id = spreadsheet.id # Move file into folder # Retrieve current parents file = drive_service.files().get( fileId=sheet_id, fields="parents" ).execute() previous_parents = ",".join(file.get("parents")) drive_service.files().update( fileId=sheet_id, addParents=FOLDER_ID, removeParents=previous_parents, fields="id, parents" ).execute() sh = gc.open_by_key(sheet_id) # read any existing protection rules metadata = sh.fetch_sheet_metadata() j = 1 # sheet per artist for artist_df in catalog_reports[which_catalog]: if i%10==0: logger.info('Pause...') time.sleep(15) artist_df['STATUS'] = None artist = artist_df['ARTIST_NAME'].drop_duplicates().tolist()[0] sheet_exists = False # check if sheet for that artist exists. if not, create it. try: logger.info(f'Loading sheet for artist {j} of {len(catalog_reports[which_catalog])}: {artist}') worksheet = next( ws for ws in sh.worksheets() if ws.title.lower() == artist.lower() ) sheet_exists = True except gspread.exceptions.WorksheetNotFound: logger.info('Sheet not found; creating new sheet') except StopIteration: logger.info('Sheet not found; creating new sheet 1') j = j+1 i = i+1 if sheet_exists: worksheet,artist_df,headers,header_exists,header_buffer,sheet_exists = read_and_compare_listings(sh,worksheet,artist_df,sheet_exists) if not sheet_exists: # IF statement instead of ELSE statement because the previous line can still return sheet_exists=False # sheet does not exist; create logger.info('Creating new sheet...') worksheet = sh.add_worksheet(title=artist, rows=f"{len(artist_df)+3}", cols="100") header_exists = False headers = None header_buffer = 0 to_upload, headers = upload_sheet_contents(artist_df,sheet_exists,worksheet,headers,header_buffer) ## SHEET FORMATTING ## logger.info('Formatting...') format_sheet(sh,worksheet,metadata,logger,sheet_exists,to_upload,headers,header_exists,header_buffer) logger.info('Complete') else: logger.info('No catalogs for catalog level reports')