import gspread from gspread_formatting import ( ConditionalFormatRule, GridRange, BooleanRule, BooleanCondition, CellFormat, Color, format_cell_range, get_conditional_format_rules, TextFormat, numberFormat, batch_updater ) from gspread.utils import rowcol_to_a1 import pandas as pd import numpy as np import json import re import time import datetime as dt from dateutil.relativedelta import relativedelta, FR import pytz import requests import html import boto3 import os import sys 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"] 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 } gspread_creds = gspread_creds.encode('utf-8').decode('unicode_escape').encode("utf-8") gspread_creds = json.loads(gspread_creds) 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 # Define tiers def get_multiplier(value): if value < 25000: return 1.5 elif value < 50000: return 1.4 elif value < 100000: return 1.3 elif value < 250000: return 1.2 elif value < 500000: return 1.1 elif value < 1000000: return 1.05 else: return 1.01 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 # 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 build_rich_text_cell(cell_text, base_url): """ Returns a tuple: (full text, textFormatRuns) Each part (single or multi-ID) gets its own clickable link. """ parts = [p.strip() for p in cell_text.split(",")] text = ", ".join(parts) runs = [] idx = 0 for p in parts: runs.append({ "startIndex": idx, "format": {"link": {"uri": f"{base_url}{p}"}} }) idx += len(p) + 2 # account for comma + space return text, runs def upload_organize_google_sheet(gc,df,is_test,logger,most_recent_thursday): logger.info('Connecting to google sheets...') doc_name = 'AWAL Artists - Long Term Growth' sh = gc.open(doc_name) logger.info('Create sheet / initial upload...') # create new sheet only if does not exist if is_test: sheet_name = f'[TEST] W/E {most_recent_thursday.strftime("%m/%d/%y")}' else: sheet_name = f'W/E {most_recent_thursday.strftime("%m/%d/%y")}' sheet_exists = True sheet_counter = 1 # create a sheet name that doesn't yet exist while sheet_exists: try: worksheet = sh.worksheet(sheet_name) # if this succeeds then need a new sheet name sheet_name = sheet_name + f' ({sheet_counter})' sheet_counter = sheet_counter+1 except gspread.exceptions.WorksheetNotFound: sheet_exists = False time.sleep(3) # create new worksheet worksheet = sh.add_worksheet(title=sheet_name, rows="100", cols="26") data = [df.columns.tolist()] + df.values.tolist() # Upload worksheet.update('A1', data, value_input_option="USER_ENTERED") BATCH_LIMIT = 50000 rules = get_conditional_format_rules(worksheet) headers = worksheet.row_values(1) requests_sh = [] # add hyperlinks to label id accounting for multiple label id's base_url = "https://insights.awal.com/account/" for row_idx, value in enumerate(df["LABELID"], start=2): text, runs = build_rich_text_cell(value, base_url) if not text or not runs: continue requests_sh.append({ "repeatCell": { "range": { "sheetId": worksheet.id, "startRowIndex": row_idx - 1, "endRowIndex": row_idx, "startColumnIndex": 15, # column P "endColumnIndex": 16 }, "cell": { "userEnteredValue": {"stringValue": text}, "textFormatRuns": runs }, "fields": "userEnteredValue.stringValue,textFormatRuns" } }) if len(requests_sh) > 0: sh.batch_update({"requests": requests_sh}) # Modify headers updated_headers = [ "ARTIST" if header.endswith("ARTIST_NAME") else "LAST REL" if header.endswith("RELEASE_DATE") else "STREAMING TIER" if header.endswith("STREAMING_TIER") else "LABEL TIER" if header.endswith("LABEL_TIER") else "GROWTH TIER" if header.endswith("GROWTH_TIER") else "ACCT" if header.endswith("LABELID") else "REG" if header.endswith("COUNTRY") else "8W\nGAIN" if header.endswith("GROWTH") else "8W\nCHG" if header.endswith("CHG") else "ACTIVE\nDRIVEN" if header.endswith("ACTIVE") else header for header in headers ] worksheet.update('1:1', [updated_headers]) logger.info('Formatting columns...') headers = worksheet.row_values(1) requests = [] requests_sh = [] indices = [i for i, col in enumerate(headers)] pct_cols = [i for i, col in enumerate(headers) if col.endswith(('GAIN','DRIVEN'))] pct_chg_cols = [i for i, col in enumerate(headers) if col.endswith(('GAIN'))] num_comma_cols = [i for i, col in enumerate(headers) if col.startswith(('LW','8W\nCHG'))] artist_name_col = [i for i, col in enumerate(headers) if col.startswith(('ARTIST'))] tier_cols = [i for i, col in enumerate(headers) if col.endswith(('TIER'))] chg_col = [i for i, col in enumerate(headers) if col.endswith(('CHG'))] date_cols = [i for i, col in enumerate(headers) if col.endswith(('REL'))] reg_col = [i for i, col in enumerate(headers) if col.startswith(('REG'))] for col_index in indices: col_letter = col_num_to_letter(col_index) col_range = f"{col_letter}1:{col_letter}" # bold and center align - HEADER requests_sh.append({ "repeatCell": { "range": { "sheetId": worksheet._properties['sheetId'], "startRowIndex": 0, "endRowIndex": 1, "startColumnIndex": col_index, "endColumnIndex": col_index + 1 }, "cell": { "userEnteredFormat": { "horizontalAlignment": "CENTER", "textFormat": { "bold": True }, "wrapStrategy": "WRAP" } }, "fields": "userEnteredFormat.wrapStrategy,userEnteredFormat.textFormat.bold,userEnteredFormat.horizontalAlignment" } }) # add color to header requests_sh.append({ "updateCells": { "range": { "sheetId": worksheet._properties['sheetId'], "startRowIndex": 0, "endRowIndex": 1, "startColumnIndex": col_index, "endColumnIndex": col_index + 1 }, "rows": [ { "values": [ { "userEnteredFormat": { "backgroundColor": { "red": 0.6431, "green": 0.7608, "blue": 0.9569 } } } ] } ], "fields": "userEnteredFormat.backgroundColor" } }) # percent formatting if col_index in pct_cols: requests.append((col_range, CellFormat(numberFormat=numberFormat(type='PERCENT', pattern='0%')))) if col_index in pct_chg_cols: # color coding thresholds = [ (3, Color(0.35, 0.60, 0.35)), # dark green (1, Color(0.50, 0.75, 0.50)), (.5, Color(0.70, 0.88, 0.70)), (0, Color(0.88, 0.96, 0.88)) # light green ] for threshold, color in thresholds: rule = ConditionalFormatRule( ranges=[GridRange.from_a1_range(col_range, worksheet)], booleanRule=BooleanRule( condition=BooleanCondition('CUSTOM_FORMULA', [f'=AND(${col_letter}1<>"", ROUND(${col_letter}1,2)>={threshold})']), format=CellFormat(backgroundColor=color) ) ) rules.append(rule) # num comma formatting if col_index in num_comma_cols: requests.append((col_range, CellFormat(numberFormat=numberFormat(type='NUMBER', pattern='#,##0')))) if col_index in chg_col: # color coding thresholds = [ (750000, Color(0.35, 0.63, 0.75)), # dark blue (100000, Color(0.55, 0.78, 0.85)), (0, Color(0.80, 0.93, 0.95)) # light blue ] for threshold, color in thresholds: rule = ConditionalFormatRule( ranges=[GridRange.from_a1_range(col_range, worksheet)], booleanRule=BooleanRule( condition=BooleanCondition('CUSTOM_FORMULA', [f'=AND(${col_letter}1<>"", ROUND(${col_letter}1,2)>={threshold})']), format=CellFormat(backgroundColor=color) ) ) rules.append(rule) if len(requests) >= BATCH_LIMIT: with batch_updater(sh) as batch: for addr, fmt in requests: batch.format_cell_range(worksheet, addr, fmt) requests = [] # col width if col_index in artist_name_col: pixelSize = 175 elif col_index in tier_cols: pixelSize = 100 elif col_index in num_comma_cols: pixelSize = 80 elif col_index in date_cols: pixelSize = 80 elif col_index in pct_chg_cols: pixelSize = 75 elif col_index in pct_cols: pixelSize = 80 elif col_index in reg_col: pixelSize = 55 else: pixelSize = 100 requests_sh.append({ "updateDimensionProperties": { "range": { "sheetId": worksheet._properties['sheetId'], "dimension": "COLUMNS", "startIndex": col_index, "endIndex": col_index+1 }, "properties": { "pixelSize": pixelSize }, "fields": "pixelSize" } }) # remove underline in whole sheet rows = len(df) + 5 end_col_letter = col_num_to_letter(reg_col[0]) col_range = f"A1:{end_col_letter}{rows}" requests.append((col_range, CellFormat(textFormat=TextFormat(underline=False)))) if len(requests_sh) > 0: sh.batch_update({"requests": requests_sh}) # Flush last batch if requests: with batch_updater(sh) as batch: for addr, fmt in requests: batch.format_cell_range(worksheet, addr, fmt) rules.save() logger.info('Final formatting...') # Freeze rows/columns n_cols_to_freeze = 1 freeze_request = { "requests": [{ "updateSheetProperties": { "properties": { "sheetId": worksheet._properties['sheetId'], "gridProperties": { "frozenRowCount": 1, "frozenColumnCount": n_cols_to_freeze } }, "fields": "gridProperties.frozenRowCount,gridProperties.frozenColumnCount" } }] } sh.batch_update(freeze_request) # Add a filter to all columns in header n_columns = len(df.columns) filter_request = { "setBasicFilter": { "filter": { "range": { "sheetId": worksheet._properties['sheetId'], "startRowIndex": 0, "startColumnIndex": 0, "endColumnIndex": n_columns } } } } sh.batch_update({"requests": [filter_request]}) # Move new sheet to the first sheet my_worksheet = sh.worksheet(sheet_name) # Get a list of all worksheets worksheets = sh.worksheets() # Create a new list with your target worksheet first, # followed by all other worksheets except that one. ordered = [my_worksheet] + [ws for ws in worksheets if ws.id != my_worksheet.id] # Apply the new order sh.reorder_worksheets(ordered) logger.info('Sheet complete') ################################################### ################################################### def handler(event,context): is_test = False logger = setup_logger() logger.info('Starting handler function...') import snowflake.connector sf_params,gspread_creds = init_secrets() sf_params['snowflake'] = snowflake gc = gspread.service_account_from_dict(gspread_creds) et_timezone = pytz.timezone('US/Eastern') today = dt.datetime.now(et_timezone).date() most_recent_thursday = today - relativedelta(weekday=FR(-1)) - relativedelta(days=1) ################################################### ################################################### # READ/ORGANIZE DATA ################################################### ################################################### df = orcd_query(""" with weeks AS ( SELECT 'LW1' AS which_week, 1 AS wk UNION ALL SELECT 'LW2', 2 UNION ALL SELECT 'LW3', 3 UNION ALL SELECT 'LW4', 4 UNION ALL SELECT 'LW5', 5 UNION ALL SELECT 'LW6', 6 UNION ALL SELECT 'LW7', 7 UNION ALL SELECT 'LW8', 8 ), streams_base as ( SELECT gplp.GLOBAL_PARTICIPANT_ID as PARTICIPANT_ID, gp.name as artist_name, CASE when s.download_activity_date >= DATEADD(DAY, -7*1, PREVIOUS_DAY(CURRENT_DATE(), 'Friday')) then 'LW1' when s.download_activity_date >= DATEADD(DAY, -7*2, PREVIOUS_DAY(CURRENT_DATE(), 'Friday')) then 'LW2' when s.download_activity_date >= DATEADD(DAY, -7*3, PREVIOUS_DAY(CURRENT_DATE(), 'Friday')) then 'LW3' when s.download_activity_date >= DATEADD(DAY, -7*4, PREVIOUS_DAY(CURRENT_DATE(), 'Friday')) then 'LW4' when s.download_activity_date >= DATEADD(DAY, -7*5, PREVIOUS_DAY(CURRENT_DATE(), 'Friday')) then 'LW5' when s.download_activity_date >= DATEADD(DAY, -7*6, PREVIOUS_DAY(CURRENT_DATE(), 'Friday')) then 'LW6' when s.download_activity_date >= DATEADD(DAY, -7*7, PREVIOUS_DAY(CURRENT_DATE(), 'Friday')) then 'LW7' when s.download_activity_date >= DATEADD(DAY, -7*8, PREVIOUS_DAY(CURRENT_DATE(), 'Friday')) then 'LW8' END AS which_week, dl.labelid, st.sort_order as tier_key, co.artist_country as country, s.isrc, r.product_id, s.download_activity_date, s.transaction_country_code, SUM(s.streams) as streams FROM INTELLIGENCE.DBT_PROD.SUMMARY_STREAMS s JOIN intelligence.dbt_prod.awal_tier_groups g ON g.labelid=s.label_id JOIN facts.prod.dim_release r ON r.releaseid=s.upc JOIN facts.prod.dim_artist da ON da.artistid=r.artistid JOIN facts.prod.dim_label dl ON dl.labelid=s.label_id JOIN FACTS.prod.SERVICE_TIER st ON st.uuid=dl.servicetierid JOIN FACTS.PROD.MAPPING_ISRC_TO_GLOBAL_PARTICIPANT_DBT m ON m.isrc=s.isrc JOIN FACTS.PROD.GLOBAL_PARTICIPANT_REPRESENTS_LABEL_PARTICIPANT gplp ON gplp.GLOBAL_PARTICIPANT_ID = m.GLOBAL_PARTICIPANT_ID JOIN FACTS.PROD.LABEL_PARTICIPANT lp ON lp.UUID = gplp.LABEL_PARTICIPANT_UUID AND lp.COMPANY_BRAND_UUID=dl.brandid AND lp.VENDOR_ID = s.label_id JOIN FACTS.PROD.GLOBAL_PARTICIPANT gp ON gp.ID = gplp.GLOBAL_PARTICIPANT_ID AND lp.name=gp.name LEFT JOIN INTELLIGENCE.DBT_PROD.cm_track_artists co ON co.artist_name=da.artistname AND co.isrc=s.isrc WHERE 1=1 AND s.download_activity_date BETWEEN DATEADD(DAY, -7*8, PREVIOUS_DAY(CURRENT_DATE(), 'Friday')) AND PREVIOUS_DAY(CURRENT_DATE(), 'Thursday') AND g.brand_name='awal' AND s.label_id != 32551 AND contains(da.artistname, gp.name) GROUP BY all ), sap_agg AS ( SELECT sap.global_participant_id, sap.isrc, sap.product_id, sap.download_activity_date, sap.country_code, SUM(sap.streams_active) + SUM(sap.streams_collection) as active_streams FROM FACTS.PROD.V_STREAMS_BY_PARTICIPANT_TRACK_COUNTRY_FEED_DISTRIBUTOR_DAILY sap INNER JOIN (SELECT DISTINCT PARTICIPANT_ID, isrc, product_id FROM streams_base) sb ON sb.PARTICIPANT_ID = sap.global_participant_id AND sb.isrc = sap.isrc AND sb.product_id = sap.product_id WHERE 1=1 AND sap.download_activity_date BETWEEN DATEADD(DAY, -7*8, PREVIOUS_DAY(CURRENT_DATE(), 'Friday')) AND PREVIOUS_DAY(CURRENT_DATE(), 'Thursday') AND sap.store_id != 453 AND sap.distributor in ('theorchard','awal') GROUP BY all ), pre_table AS ( SELECT sb.PARTICIPANT_ID, sb.artist_name, sb.which_week, sb.labelid, sb.tier_key, sb.country, SUM(sb.streams) as streams, SUM(sap.active_streams) as active_streams FROM streams_base sb LEFT JOIN sap_agg sap ON sap.global_participant_id = sb.PARTICIPANT_ID AND sap.isrc = sb.isrc AND sap.product_id = sb.product_id AND sap.download_activity_date = sb.download_activity_date AND sap.country_code = sb.transaction_country_code GROUP BY all ), -- clean up where isrcs are not matched, resulting in a mix of null countries for an artist pre_table_cleaned_country as ( select t1.* exclude(country), FIRST_VALUE(country) IGNORE NULLS OVER ( PARTITION BY participant_id ORDER BY streams desc ) as country from pre_table t1 ), -- add in tax country if country result is still null full_table as ( select t1.* exclude(country), coalesce(t1.country,dc.country_code) as country from pre_table_cleaned_country t1 left join ORCHARD_APP_REPORTING_V2.PROD_ROYALTY_ACCOUNTING_ROYALTY_ACCOUNTING.ACCOUNT_TAX_INFO tx on tx.account_id=t1.labelid left join FACTS.PROD.DIM_COUNTRY dc on dc.ISO3166A3=tx.COUNTRY_OF_TAX_RESIDENCE ), -- PM info product_manager as ( select v.vendor_id as labelid, concat(u.f_name,' ',u.l_name) PM from ORCHARD_APP_REPORTING_V2.ART_RELATIONS_PROD_ART_RELATIONS.ORCHADMIN_USERS u join ORCHARD_APP_REPORTING_V2.ART_RELATIONS_PROD_ART_RELATIONS.PRODUCT_MANAGER_MAPPING_VENDOR v on v.product_manager_id=u.id join full_table f on f.labelid=v.vendor_id where u.active = 'Y' ), product_manager_cleaned as ( select distinct labelid, LISTAGG(distinct PM, ', ') WITHIN GROUP (ORDER BY PM) AS PM from product_manager group by 1 ), closer as ( select v.vendor_id as labelid, concat(u.f_name,' ',u.l_name) PM from ORCHARD_APP_REPORTING_V2.ART_RELATIONS_PROD_ART_RELATIONS.ORCHADMIN_USERS u join ORCHARD_APP_REPORTING_V2.ART_RELATIONS_PROD_ART_RELATIONS.VENDOR_CLOSERS v on v.orchadmin_user_id=u.id join full_table f on f.labelid=v.vendor_id where u.active = 'Y' ), closer_cleaned as ( select distinct labelid, LISTAGG(distinct PM, ', ') WITHIN GROUP (ORDER BY PM) AS PM from closer group by 1 ), pm_final as ( select distinct coalesce(p.labelid,c.labelid) labelid, coalesce(p.PM,c.PM) PM from product_manager_cleaned p full outer join closer_cleaned c ON c.labelid = p.labelid ), artists_tier AS ( SELECT PARTICIPANT_ID,min(tier_key) as tier_key FROM full_table group by 1 ), artists_labels as ( select distinct PARTICIPANT_ID,labelid from full_table ), artists_pms as ( select distinct a.PARTICIPANT_ID,a.labelid,p.pm from artists_labels a join pm_final p on p.labelid=a.labelid ), artists_labels_agg as ( SELECT distinct PARTICIPANT_ID, LISTAGG(distinct labelid, ', ') WITHIN GROUP (ORDER BY labelid) AS labelid FROM artists_labels group by 1 ), artists_pms_agg as ( SELECT PARTICIPANT_ID, LISTAGG(pm, ', ') WITHIN GROUP (ORDER BY min_labelid) AS pm FROM ( SELECT PARTICIPANT_ID, pm, MIN(labelid) AS min_labelid FROM artists_pms GROUP BY PARTICIPANT_ID, pm ) t GROUP BY PARTICIPANT_ID ), artists_countries as ( SELECT distinct PARTICIPANT_ID, LISTAGG(distinct country, ', ') WITHIN GROUP (ORDER BY country) AS country FROM full_table group by 1 ), artist_names as ( select DISTINCT PARTICIPANT_ID, artist_name from full_table ), artists as ( select distinct arn.PARTICIPANT_ID, arn.artist_name, at.tier_key, al.labelid, ap.pm, ac.country from artist_names arn join artists_tier at on arn.PARTICIPANT_ID=at.PARTICIPANT_ID join artists_labels_agg al on al.PARTICIPANT_ID=arn.PARTICIPANT_ID join artists_countries ac on ac.PARTICIPANT_ID=arn.PARTICIPANT_ID left join artists_pms_agg ap on ap.PARTICIPANT_ID=arn.PARTICIPANT_ID ), cleaned_table as ( SELECT a.participant_id, a.artist_name, a.tier_key, a.labelid, a.pm, a.country, w.which_week, COALESCE(SUM(ft.streams), 0) AS streams, COALESCE(SUM(ft.active_streams), 0) AS active_streams FROM artists a CROSS JOIN weeks w LEFT JOIN full_table ft ON ft.PARTICIPANT_ID = a.PARTICIPANT_ID AND ft.which_week = w.which_week GROUP BY all ORDER BY 1,2 ), artists_below_threshold as ( select distinct PARTICIPANT_ID from cleaned_table where which_week='LW1' and streams < 10000 ), final_table as ( select c.* from cleaned_table c full join artists_below_threshold a on a.PARTICIPANT_ID=c.PARTICIPANT_ID where a.PARTICIPANT_ID is null ), release_dates as ( select ap.participant_id, max(cast(dr.releasedate as date)) release_date from FACTS.PROD.DIM_TRACK dt join FACTS.PROD.DIM_RELEASE dr on dr.releaseid=dt.upc join FACTS.PROD.DIM_ARTIST da on da.artistid=dr.artistid join FACTS.PROD.DIM_LABEL dl on dl.labelid=dt.labelid join FACTS.PROD.MAPPING_ISRC_TO_GLOBAL_PARTICIPANT_DBT m on m.isrc=dt.isrc JOIN FACTS.PROD.GLOBAL_PARTICIPANT_REPRESENTS_LABEL_PARTICIPANT gplp ON gplp.GLOBAL_PARTICIPANT_ID = m.GLOBAL_PARTICIPANT_ID JOIN FACTS.PROD.LABEL_PARTICIPANT lp ON lp.UUID = gplp.LABEL_PARTICIPANT_UUID and lp.COMPANY_BRAND_UUID=dl.brandid AND lp.VENDOR_ID = dl.labelid JOIN FACTS.PROD.GLOBAL_PARTICIPANT gp ON gp.ID = gplp.GLOBAL_PARTICIPANT_ID and lp.name=gp.name join artists ap on ap.participant_id=gplp.GLOBAL_PARTICIPANT_ID full join INTELLIGENCE.DBT_PROD.SUMMARY_STREAMS ss on ss.isrc=dt.isrc and ss.upc=dr.releaseid where 1=1 and contains (da.artistname,ap.artist_name) and dl.brandid='31f4f0f0-cbb4-4a2c-9eb0-d7288c5a2588' and dr.product_type='digital' and dr.physical_product_type not like 'Music Video' and dr.releasedate <= CURRENT_DATE() and ss.download_activity_date > DATEADD(DAY, -7*8, PREVIOUS_DAY(CURRENT_DATE(), 'Friday')) and ss.store='Spotify' and ss.isrc is not null and dt.isrc not like '%ISRC' group by all ) select distinct ft.*, rd.release_date, CASE when rd.release_date < DATEADD(MONTH, -18, CURRENT_DATE()) then 0 when rd.release_date is null then 0 else 1 END as REL_18_MO, CASE when rd.release_date >= DATEADD(DAY, -7*8, PREVIOUS_DAY(CURRENT_DATE(), 'Friday')) then 1 else 0 END as NEW_RELEASE_LAST_8_WEEKS from final_table ft left join release_dates rd on lower(rd.PARTICIPANT_ID)=lower(ft.PARTICIPANT_ID) join artists a on a.PARTICIPANT_ID=ft.PARTICIPANT_ID """,sf_params) df = df.loc[~df['PM'].str.contains('Trust & Safety Termination',na=False)] tiers = pd.DataFrame({ 'TIER_KEY':[1,2,3,4,5,6,7], 'LABEL_TIER':[ 'Premium Services', 'Tier 1', 'Tier 2', 'Tier 3', 'Managed Basic', 'Basic', 'Untiered' ] }) df = df.merge(tiers,how='inner',on='TIER_KEY') df = df.drop(columns='TIER_KEY') # pivot table adj_result = df.pivot(index=['ARTIST_NAME','PARTICIPANT_ID','LABELID','PM','COUNTRY','RELEASE_DATE','REL_18_MO','NEW_RELEASE_LAST_8_WEEKS','LABEL_TIER'], columns='WHICH_WEEK', values='STREAMS').reset_index() active_streams = df.pivot(index=['ARTIST_NAME','PARTICIPANT_ID','LABELID','PM','COUNTRY','RELEASE_DATE','REL_18_MO','NEW_RELEASE_LAST_8_WEEKS','LABEL_TIER'], columns='WHICH_WEEK', values='ACTIVE_STREAMS').reset_index() active_streams['CHG_ACTIVE'] = active_streams['LW1'] - active_streams['LW8'] # # set limit of 0% # active_streams['CHG_ACTIVE'] = np.where( # active_streams['CHG_ACTIVE'] < 0, # 0, # active_streams['CHG_ACTIVE'] # ) # Last week is at least 20k more streams than 8 weeks ago adj_result = adj_result.loc[adj_result['LW1']>adj_result['LW8']+20000] # if LW8 is above 5M, LW1 should be at least 250k more streams to_remove = adj_result.loc[(adj_result['LW8']>=5000000) & (adj_result['LW1']=1000000) & (adj_result['LW1']=500000) & (adj_result['LW1']=300000) & (adj_result['LW1']=100000) & (adj_result['LW1'] adj_result['LW8'] step2 = adj_result['LW6'] > adj_result['LW7'] step3 = adj_result['LW5'] > adj_result['LW6'] step4 = adj_result['LW4'] > adj_result['LW5'] step5 = adj_result['LW3'] > adj_result['LW4'] step6 = adj_result['LW2'] > adj_result['LW3'] step7 = adj_result['LW1'] > adj_result['LW2'] growth_steps = step1.astype(int) + step2.astype(int) + step3.astype(int) + step4.astype(int) + step5.astype(int) + step6.astype(int) + step7.astype(int) # average of recent weeks is above the multiplier threshold recent_avg = adj_result[['LW4', 'LW3', 'LW2', 'LW1']].mean(axis=1) early_avg = adj_result[['LW8', 'LW7','LW6', 'LW5']].mean(axis=1) # Check that recent average is significantly higher than early average meaningful_growth = recent_avg > (adj_result['multiplier'] * early_avg) # 4. Final filter final_result = adj_result[ (growth_steps >= 2) & (adj_result['LW1'] > early_avg) # LW1 greater than first 2 weeks & (adj_result['LW2'] > early_avg) # LW2 greater than first 2 weeks & meaningful_growth # avg of most recent 2 weeks at least x amount greater than avg of first 2 weeks & ( (adj_result['LW1'] > adj_result['LW2']) | (adj_result['LW2'] > adj_result['LW3']) ) # at least either LW2 or LW1 is up & ((~(adj_result['multiplier'] == 1.01)) | (adj_result['RATIO'] >= 1.12)) & ((~(adj_result['multiplier'] == 1.05)) | (adj_result['RATIO'] >= 1.15)) & ((~(adj_result['multiplier'] == 1.1)) | (adj_result['RATIO'] >= 1.27)) & ((~(adj_result['multiplier'] == 1.2)) | (adj_result['RATIO'] >= 1.28)) & ((~(adj_result['multiplier'] == 1.3)) | (adj_result['RATIO'] >= 1.4)) & ((~(adj_result['multiplier'] == 1.4)) | (adj_result['RATIO'] >= 1.7)) & ((~(adj_result['multiplier'] == 1.5)) | (adj_result['RATIO'] >= 2)) ] final_result['GROWTH'] = np.where( final_result['LW8']>0, (final_result['LW1'] - final_result['LW8']) / final_result['LW8'], (final_result['LW1'] - final_result['LW8']) / .001 ) # cap at 1000% final_result['GROWTH'] = np.where( final_result['GROWTH'] > 10, 10, final_result['GROWTH'] ) final_result['CHG'] = final_result['LW1'] - final_result['LW8'] final_result = final_result.merge(active_streams[['PARTICIPANT_ID','CHG_ACTIVE']],how='left',on='PARTICIPANT_ID') final_result['GROWTH_RANK'] = 0 final_result['GROWTH_RANK'] = np.where( final_result['GROWTH'].round(2)<.5, 1, final_result['GROWTH_RANK'] ) final_result['GROWTH_RANK'] = np.where( final_result['GROWTH'].round(2)>=.5, 2, final_result['GROWTH_RANK'] ) final_result['GROWTH_RANK'] = np.where( final_result['GROWTH'].round(2)>=1, 3, final_result['GROWTH_RANK'] ) final_result['GROWTH_RANK'] = np.where( final_result['GROWTH'].round(2)>=3, 4, final_result['GROWTH_RANK'] ) growth_tiers = pd.DataFrame({ 'GROWTH_RANK':[1,2,3,4], 'GROWTH_TIER':[ 'MODERATE', 'SUBSTANTIAL', 'SIGNIFICANT', 'GREATEST GAINERS' ] }) final_result = final_result.merge(growth_tiers,how='left',on='GROWTH_RANK') final_result['STREAMING_TIER'] = np.where( final_result['LW1']<100000, 'SMALL', np.where( final_result['LW1'].between(100000,750000), 'MODERATE', 'LARGE' ) ) final_result = final_result.sort_values(['GROWTH_RANK','GROWTH','LW1'],ascending=False) final_result['STATUS'] = np.where( final_result['REL_18_MO']==0, 'CATALOG', np.where( final_result['NEW_RELEASE_LAST_8_WEEKS']==1, 'ACTIVE FRONTLINE', 'INACTIVE FRONTLINE' ) ) final_result['PCT_ACTIVE'] = final_result['CHG_ACTIVE']/final_result['CHG'] final_result = final_result[[ 'ARTIST_NAME','GROWTH','CHG','PCT_ACTIVE', 'LW1','LW2','LW3','LW4','LW5','LW6','LW7','LW8', 'RELEASE_DATE','STATUS', 'LABEL_TIER','LABELID','PARTICIPANT_ID', 'PM','COUNTRY']] # remove artists whose peak week was before LW1 and is >1.8x LW1 selected_columns = ['LW1', 'LW2', 'LW3', 'LW4', 'LW5', 'LW6', 'LW7', 'LW8'] final_result['PEAK_WEEK'] = final_result[selected_columns].idxmax(axis=1) final_result['PEAK_VALUE'] = final_result[selected_columns].max(axis=1) to_drop = final_result.loc[(final_result['PEAK_WEEK'].str[-1].astype(int)>1) & (final_result['PEAK_VALUE']>1.8*final_result['LW1'])].index final_result = final_result.drop(to_drop) final_result['RELEASE_DATE'] = pd.to_datetime(final_result['RELEASE_DATE']).astype(str).replace('NaT', '') final_result['PM'] = final_result['PM'].fillna('') final_result['PARTICIPANT_ID'] = final_result['PARTICIPANT_ID'].fillna('') final_result['COUNTRY'] = final_result['COUNTRY'].fillna('') # add hyperlinks to artist participant id final_result["ARTIST_NAME"] = '=HYPERLINK("' + 'https://insights.awal.com/artist/' + final_result["PARTICIPANT_ID"] + '", "' + final_result['ARTIST_NAME'] + '")' final_result = final_result.drop(['PEAK_WEEK','PEAK_VALUE','PARTICIPANT_ID'],axis=1) ################################################ upload_organize_google_sheet(gc,final_result,is_test,logger,most_recent_thursday) ################################################ logger.info('Finished') return { "statusCode": 200, "message": "Handler function ran successfully" }