# Import python packages

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

def setup_email_params():

    send_the_email = True
    send_only_to_myself = False

    client = boto3.client('ses',region_name='us-east-1')

    email_params = {
        'client':client,
        'send_the_email':send_the_email,
        'send_only_to_myself':send_only_to_myself
    }

    return email_params

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"]

    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

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 delphi_query(sql_query,sf_params):
    ctx = sf_params['snowflake'].connector.connect(
        user=sf_params['sf_user'],
        private_key=sf_params['pkb'],
        account='delphi',
        warehouse='AWAL_ANALYTICS_LARGE_WAREHOUSE',
        region='us-east-1'
    )
    cs = ctx.cursor(sf_params['snowflake'].connector.DictCursor)
    try:
        cs.execute(sql_query)
        result = cs.fetchall()
    finally:
        cs.close()
    ctx.close()
    return pd.DataFrame(result)

def upload_df_to_snowflake_table(df,table_name,sf_params):
    import streamlit # write_pandas won't run without streamlit package for some reason
    from snowflake.connector.pandas_tools import write_pandas

    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'],
        database='awal',
        schema='awal_ar'
    )
    try:
        write_pandas(ctx, df, table_name, auto_create_table=True)
    finally:
        ctx.close()

def upload_temp_stage_to_delphi(stage_name,stage_data,sf_params):
    delphi_query(f'REMOVE @~/{stage_name}.csv.gz',sf_params)
    with open(f'/tmp/{stage_name}.csv', 'w') as f:
        for each_item in stage_data:
            f.write(f"{each_item}\n")
    file_path = os.path.abspath(f'/tmp/{stage_name}.csv')
    delphi_query(f'PUT file://{file_path} @~ AUTO_COMPRESS=TRUE',sf_params)

def remove_off_limits_artists(df,sf_params):

    # artists
    off_limits_artists = orcd_query('select * from awal.awal_ar.off_limits_artists',sf_params)['ARTIST'].str.lower()
    df = df.loc[~df['NAME'].str.lower().isin(off_limits_artists)]

    # collabs
    off_limits_artists = orcd_query('select * from awal.awal_ar.off_limits_artists_collabs',sf_params)
    off_limits_artists = off_limits_artists['ARTIST'].str.lower().values.tolist()
    df = df.loc[~(df['NAME'].str.lower()).str.contains('|'.join(off_limits_artists))].reset_index(drop=True)

    # shows
    off_limits_shows = orcd_query('select * from awal.awal_ar.off_limits_shows',sf_params)
    off_limits_shows = off_limits_shows['NAME'].str.lower().values.tolist()
    df = df.loc[~(df['NAME'].str.lower()).str.contains('|'.join(off_limits_shows))].reset_index(drop=True)

    return df

def drop_sony_orchard(result):
    tmp = result.drop(result.loc[result['LABEL'].fillna('').str.lower().str.contains('sony')].index)
    tmp.drop(tmp.loc[tmp['LABEL'].fillna('').str.lower().str.contains('orchard')].index,inplace=True)
    tmp.reset_index(inplace=True,drop=True)

    return tmp

def prep_and_send_email(email_params,destination_emails,email_subject,full_email):

    if email_params['send_the_email']:

        if email_params['send_only_to_myself']:
            destination_emails = ['joselyn.ho@awal.com']

        response = email_params['client'].send_email(
            Source='awalresearch@dev.theorchard.io',
            Destination={
                'ToAddresses': destination_emails,
            },
            Message={
                'Subject': {
                    'Data': email_subject,
                    'Charset': 'UTF-8'
                },
                'Body': {
                    'Html': {
                        'Data': full_email,
                        'Charset': 'UTF-8'
                    }
                }
            },
            SourceArn='arn:aws:ses:us-east-1:103233932089:identity/dev.theorchard.io',
            ReplyToAddresses=[
                'joselyn.ho@awal.com'
            ],
        )

        return response['ResponseMetadata']['HTTPStatusCode'] == 200

    else:
        # this is for local testing where no email is sent. Just opens a browser with the contents

        import webbrowser
        import os

        # Save the file
        file_path = os.path.abspath("preview_email.html")
        with open(file_path, "w", encoding="utf-8") as f:
            f.write(full_email)

        # Open it in the default web browser
        webbrowser.open(f"file://{file_path}")

###########################

# Charts

###########################

def group_chart_results(listings,country_list,which_type=[]):

    chart_date = listings['CHART_DATE'].iloc[0]
    
    listings.drop('CHART_DATE',axis=1,inplace=True)

    # sort by region priority
    listings = listings.merge(country_list,how='left',on='COUNTRY')
    listings.sort_values(['KEY_SORT','GENRE'],inplace=True,ignore_index=True)
    listings.drop('KEY_SORT',axis=1,inplace=True)
    
    # combine columns into a listing column
    if listings['GENRE'].isnull().all() or listings['GENRE'].fillna('').iloc[0] in ['regional','viral','All Genres']:
        # don't specify genre if Shazam or spotify top 200 or spotify viral or "All Genres"
        listings['LISTING'] = '<b>' + listings['COUNTRY'] + '</b> #' + listings['RANK'].astype('int64').astype(str)
    else:
        listings['LISTING'] = '<b>' + listings['COUNTRY'] + '/' + listings['GENRE'].fillna('') + '</b> #' + listings['RANK'].astype('int64').astype(str)

    # highlight yellow for new listings but not for re-entry
    listings['LISTING'] = np.where(
        listings['MOVEMENT']=='NEW',
        np.where(
            listings['DAYS_ON_CHART'] > 1,
            listings['LISTING'],
            '<span style="background-color: yellow;">' + listings['LISTING'] + '</span>'
        ),
        listings['LISTING']
    )

    # green arrow to indicate upwards movement
    listings['LISTING'] = np.where(
        listings['MOVEMENT']=='UP',
        listings['LISTING'] + ' <font color="green">▲' + listings['CHG'].fillna(0).astype('int64').astype(str) + '</font>',
        listings['LISTING']
    )

    # Days on chart in blue. If re-entry, specify as such. If peak, specify
    listings['LISTING'] = np.where(
        listings['DAYS_ON_CHART'] > 1,

        # has charted before
        np.where(
            listings['MOVEMENT']=='NEW',

            # is new re-entry
            np.where(
                listings['PEAK'],

                # is peak
                listings['LISTING'] + ' <font color="DodgerBlue">(<b>re-entry</b>, <b>new peak</b>, ' + listings['DAYS_ON_CHART'].fillna(0).astype('int64').astype(str) + 'd)</font>',
                
                # is not peak
                listings['LISTING'] + ' <font color="DodgerBlue">(<b>re-entry</b>, ' + listings['DAYS_ON_CHART'].fillna(0).astype('int64').astype(str) + 'd)</font>'
            ),
            
            # is not a new re-entry
            np.where(
                listings['PEAK'],

                # is peak
                listings['LISTING'] + ' <font color="DodgerBlue">(<b>new peak</b>, ' + listings['DAYS_ON_CHART'].fillna(0).astype('int64').astype(str) + 'd)</font>',

                # is not peak
                listings['LISTING'] + ' <font color="DodgerBlue">(' + listings['DAYS_ON_CHART'].fillna(0).astype('int64').astype(str) + 'd)</font>'
            )
            
        ),     

        # has not charted before, don't need blue notes   
        listings['LISTING']
    )
    
    listings.drop(['COUNTRY','GENRE','RANK'],axis=1,inplace=True)

    # group listings into 1 row per song
    listings['NAME'] = listings['ARTIST'] + ' - ' + listings['SONG_TITLE']
    listings.drop(['ARTIST','SONG_TITLE'],axis=1,inplace=True)
    listings = listings.groupby(['SONG_ID','NAME'], as_index=False).agg(LISTINGS=('LISTING', ' | '.join))
    
    # count num listings
    listings['N_LISTINGS'] = listings['LISTINGS'].str.count(',')+1

    # for the sake of multiple versions of a song - here we will group together names and charts into a list
    grouped_name = listings.groupby(['SONG_ID'])['NAME'].apply(list).reset_index()[['SONG_ID','NAME']]
    grouped_charts = listings.groupby(['SONG_ID'])['LISTINGS'].apply(list).reset_index()[['SONG_ID','LISTINGS']]
    grouped_num = listings.groupby(['SONG_ID'])['N_LISTINGS'].apply(list).reset_index()[['SONG_ID','N_LISTINGS']]

    # merging
    listings.drop(['NAME','LISTINGS','N_LISTINGS'],axis=1,inplace=True)
    listings = listings.merge(grouped_name,how='left',on='SONG_ID')
    listings = listings.merge(grouped_charts,how='left',on='SONG_ID')
    listings = listings.merge(grouped_num,how='left',on='SONG_ID')
    
    listings.drop_duplicates('SONG_ID',inplace=True,ignore_index=True)

    return listings,chart_date

# multi chart listings
def organize_multi_listings(df):

    # separate into groups for further processing of the multi song category
    df['MULTIPLE_SONGS'] = df['N_LISTINGS'].apply(lambda x: True if len(x)>1 else False)

    single_songs = df.loc[df['MULTIPLE_SONGS']==False].reset_index(drop=True).drop(['MULTIPLE_SONGS'],axis=1)
    multiple_songs = df.loc[df['MULTIPLE_SONGS']==True].reset_index(drop=True).drop(['MULTIPLE_SONGS'],axis=1)

    # single songs
    for col in ['NAME','LISTINGS','N_LISTINGS']:
        single_songs[col] = single_songs[col].apply(lambda x: x[0])
    single_songs.drop('NAME',axis=1,inplace=True)

    # multiple songs
    names = multiple_songs.explode('NAME',ignore_index=True)[['SONG_ID','NAME']]
    charts = multiple_songs.explode('LISTINGS',ignore_index=True)[['LISTINGS']]
    n = multiple_songs.explode('N_LISTINGS',ignore_index=True)[['N_LISTINGS']]

    # concat
    multiple_songs = pd.concat([names, charts, n], axis=1)

    ## attach names for multi chart situations.

    # at this point, everything is split except chart listings - split those again per song
    multiple_songs['LISTINGS'] = multiple_songs['LISTINGS'].apply(lambda x: x.split('|'))
    multiple_songs = multiple_songs.explode('LISTINGS',ignore_index=True)

    # identify countries that appear more than once (so multiple versions of the song on the same chart)
    multiple_songs['country'] = multiple_songs['LISTINGS'].str.extract(r'<b>(.*?)</b>')
    multiple_songs['is_duplicate_country'] = multiple_songs.duplicated(subset=['SONG_ID', 'country'],keep=False)
    duplicate_rows = multiple_songs.loc[multiple_songs['is_duplicate_country']]

    multiple_songs = multiple_songs.drop(duplicate_rows.index)
    multiple_songs = multiple_songs.sort_values('country')
    
    # then attach name to each listing only if on the same country chart; drop name col
    duplicate_rows['LISTINGS'] = duplicate_rows['LISTINGS'] + ' (' + duplicate_rows['NAME'] + ')'
    duplicate_rows = duplicate_rows.sort_values('country')

    # return them to multiple_songs
    multiple_songs = pd.concat([multiple_songs,duplicate_rows]).reset_index()
    
    multiple_songs.drop('NAME',axis=1,inplace=True)

    # then combine rows
    multiple_songs = multiple_songs.groupby(['SONG_ID'], as_index=False).agg(LISTINGS=('LISTINGS', ' | '.join))
    multiple_songs['N_LISTINGS'] = multiple_songs['LISTINGS'].str.count('|')+1

    # concat back to all songs
    all_songs = pd.concat([multiple_songs,single_songs])

    # all_songs['LISTINGS'] = all_songs['LISTINGS'].str.split(',')


    return all_songs

def organize_spotify(full_listing,charts_country_key):

    song_ids = []
   
    priority_country_list = [
        'GLOBAL',
        'US',
        'GB',
        'CA',
        'AU',
        'IE'
    ]

    secondary_country_list = [
        'NG',
        'FR',
        'MX',
        'SE',
        'ES',
        'NL',
        'ZA',
        'PH',
        'EG',
        'IN'
    ]

    ## NEW entries ##

    listings = full_listing.copy()
    listings = listings.loc[listings['MOVEMENT']=='NEW']
    listings['N_LISTINGS'] = listings['SONG_ID'].map(listings['SONG_ID'].value_counts())

    song_ids_to_keep = listings.loc[
        (listings['COUNTRY'].isin(priority_country_list))
        | ((listings['COUNTRY'].isin(secondary_country_list)) & (listings['RANK']<=50))
        | (listings['N_LISTINGS']>1)
    ]['SONG_ID'].drop_duplicates()
    song_ids.append(song_ids_to_keep)

    ## Moved UP ##

    listings = full_listing.copy()
    listings = listings.loc[listings['MOVEMENT']=='UP']
    listings['N_LISTINGS'] = listings['SONG_ID'].map(listings['SONG_ID'].value_counts())

    listings = listings.loc[
        (
            ((listings['CHG']>=5) & (listings['RANK']<=50))
            | ((listings['CHG']>=10) & (listings['RANK']>50))
        )
    ]

    # Filter: non-priority countries with rank < 50
    non_priority_filtered = listings[
        (~listings['COUNTRY'].isin(priority_country_list))
    ]
    
    # Identify countries with at least 2 such listings
    valid_non_priority_countries = non_priority_filtered['SONG_ID'].value_counts()
    valid_non_priority_countries = valid_non_priority_countries[valid_non_priority_countries > 1].index

    song_ids_to_keep = listings.loc[
        (listings['COUNTRY'].isin(priority_country_list)) 
        | (listings['SONG_ID'].isin(valid_non_priority_countries))
    ]['SONG_ID'].drop_duplicates()
    song_ids.append(song_ids_to_keep)

    ####### #######
    
    song_ids = pd.concat(song_ids).reset_index(drop=True)

    listings = full_listing.copy()
    listings = listings.loc[listings['SONG_ID'].isin(song_ids)]

    if len(listings) > 0:
    
        updated_listings,charts_dates = group_chart_results(listings,charts_country_key)
        
        updated_listings = organize_multi_listings(updated_listings)
        
        updated_listings.sort_values('N_LISTINGS',ascending=False,inplace=True)

        updated_listings['CHART_DATE'] = charts_dates

    else:
        updated_listings = pd.DataFrame()

    return updated_listings

def organize_apple_genres(full_listing,charts_country_key):
    
    priority_country_list = [
        'GLOBAL',
        'US',
        'GB',
        'CA',
        'AU',
        'IE'
    ]
    
    ## NEW entries ##
    
    listings = full_listing.copy()
    listings = listings.loc[listings['MOVEMENT']=='NEW']
    
    # Filter: non-priority countries with rank < 50
    non_priority_filtered = listings[
        (~listings['COUNTRY'].isin(priority_country_list)) & (listings['RANK'] <= 50)
    ][['SONG_ID']]
    non_priority_filtered['N_LISTINGS'] = non_priority_filtered['SONG_ID'].map(non_priority_filtered['SONG_ID'].value_counts())
    non_priority_filtered = non_priority_filtered.drop_duplicates()
    
    # Filter: priority countries with rank < 100
    priority_filtered = listings[
        (listings['COUNTRY'].isin(priority_country_list)) & (listings['RANK'] <= 100)
    ][['SONG_ID']]
    priority_filtered['N_LISTINGS'] = priority_filtered['SONG_ID'].map(priority_filtered['SONG_ID'].value_counts())
    priority_filtered = priority_filtered.drop_duplicates()
    
    ## Moved UP ##
    
    listings = full_listing.copy()
    listings = listings.loc[listings['MOVEMENT']=='UP']
    
    ups = listings.loc[(listings['CHG']>=10) & (listings['RANK']<=30)][['SONG_ID']]
    ups['N_LISTINGS'] = ups['SONG_ID'].map(ups['SONG_ID'].value_counts())
    ups = ups.drop_duplicates()
    
    ######
    
    combined = pd.concat([non_priority_filtered,priority_filtered,ups])
    
    combined['N_LISTINGS'] = combined.groupby('SONG_ID')['N_LISTINGS'].transform('sum')
    song_ids = combined.loc[combined['N_LISTINGS']>1]['SONG_ID'].drop_duplicates()

    listings = full_listing.copy()
    listings = listings.loc[listings['SONG_ID'].isin(song_ids)]

    if len(listings) > 0:
    
        updated_listings,charts_dates = group_chart_results(listings,charts_country_key)
        
        updated_listings = organize_multi_listings(updated_listings)
        
        updated_listings.sort_values('N_LISTINGS',ascending=False,inplace=True)
    
        updated_listings['CHART_DATE'] = charts_dates

    else:
        updated_listings = pd.DataFrame()   

    return updated_listings


def organize_itunes_main(listings,charts_country_key,which_type):

    listings['N_LISTINGS'] = listings['SONG_ID'].map(listings['SONG_ID'].value_counts())
    
    priority_country_list = [
        'GLOBAL',
        'US',
        'GB',
        'CA',
        'AU',
        'IE'
    ]
    
    if which_type=='NEW':
        # Filter: non-priority countries with rank < 50
        non_priority_filtered = listings[
            (~listings['COUNTRY'].isin(priority_country_list)) & (listings['RANK'] <= 50)
        ]
        
        # Identify countries with at least 2 such listings
        valid_non_priority_countries = non_priority_filtered['SONG_ID'].value_counts()
        valid_non_priority_countries = valid_non_priority_countries[valid_non_priority_countries > 1].index

        
        song_ids_to_keep = listings.loc[
            (listings['COUNTRY'].isin(priority_country_list) & (listings['RANK']<=100))
            | (listings['SONG_ID'].isin(valid_non_priority_countries))
        ]['SONG_ID'].drop_duplicates()
    elif which_type=='UP':
        listings = listings.loc[
            (
                ((listings['CHG']>=5) & (listings['RANK']<=50))
                | ((listings['CHG']>=10) & (listings['RANK']>50))
            )
        ]

        # Filter: non-priority countries with rank < 50
        non_priority_filtered = listings[
            (~listings['COUNTRY'].isin(priority_country_list))
        ]
        
        # Identify countries with at least 2 such listings
        valid_non_priority_countries = non_priority_filtered['SONG_ID'].value_counts()
        valid_non_priority_countries = valid_non_priority_countries[valid_non_priority_countries > 1].index

        song_ids_to_keep = listings.loc[
            (listings['COUNTRY'].isin(priority_country_list)) 
            | (listings['SONG_ID'].isin(valid_non_priority_countries))
        ]['SONG_ID'].drop_duplicates()
    
    
    listings = listings.loc[listings['SONG_ID'].isin(song_ids_to_keep)]

    if len(listings) > 0:
    
        updated_listings,charts_dates = group_chart_results(listings,charts_country_key,which_type)
        
        updated_listings = organize_multi_listings(updated_listings)
                
        updated_listings.sort_values('N_LISTINGS',ascending=False,inplace=True)
        
        updated_listings['CHART_DATE'] = charts_dates

    else:
        updated_listings = pd.DataFrame()

    return updated_listings

def organize_itunes_genres(listings,charts_country_key,which_type):

    listings['N_LISTINGS'] = listings['SONG_ID'].map(listings['SONG_ID'].value_counts())
    
    priority_country_list = [
        'GLOBAL',
        'US',
        'GB',
        'CA',
        'AU'
    ]
    
    if which_type=='NEW':
        # Filter: non-priority countries with rank
        non_priority_filtered = listings[
            (~listings['COUNTRY'].isin(priority_country_list)) & (listings['RANK'] <= 20)
        ]
        
        # Identify countries with at least 2 such listings
        valid_non_priority_countries = non_priority_filtered['SONG_ID'].value_counts()
        valid_non_priority_countries = valid_non_priority_countries[valid_non_priority_countries > 1].index

        
        listings = listings.loc[
            (
                (listings['COUNTRY'].isin(priority_country_list) & (listings['RANK']<=30))
                | (listings['SONG_ID'].isin(valid_non_priority_countries))
            )
        ]

        listings['N_LISTINGS'] = listings['SONG_ID'].map(listings['SONG_ID'].value_counts())
        
        song_ids_to_keep = listings.loc[listings['N_LISTINGS'] > 1]['SONG_ID'].drop_duplicates()
        
    elif which_type=='UP':
        listings = listings.loc[
            (
                ((listings['CHG']>=5) & (listings['RANK']<=20))
                # | ((listings['CHG']>=10) & (listings['RANK']<=30))
            )
        ]
        listings['N_LISTINGS'] = listings['SONG_ID'].map(listings['SONG_ID'].value_counts())

        # Filter: non-priority countries 
        non_priority_filtered = listings[
            (~listings['COUNTRY'].isin(priority_country_list))
        ]
        
        # Identify countries with at least 2 such listings
        valid_non_priority_countries = non_priority_filtered['SONG_ID'].value_counts()
        valid_non_priority_countries = valid_non_priority_countries[valid_non_priority_countries > 1].index

        song_ids_to_keep = listings.loc[
            (
                (listings['COUNTRY'].isin(priority_country_list)) 
                | (listings['SONG_ID'].isin(valid_non_priority_countries))
            )
            & (listings['N_LISTINGS'] > 1)
        ]['SONG_ID'].drop_duplicates()
    
    
    listings = listings.loc[listings['SONG_ID'].isin(song_ids_to_keep)]

    if len(listings) > 0:
    
        updated_listings,charts_dates = group_chart_results(listings,charts_country_key,which_type)
        
        updated_listings = organize_multi_listings(updated_listings)
                
        updated_listings.sort_values('N_LISTINGS',ascending=False,inplace=True)
        
        updated_listings['CHART_DATE'] = charts_dates

    else:
        updated_listings = pd.DataFrame()

    return updated_listings

def organize_youtube(listings,charts_country_key,which_type):
    
    priority_country_list = [
        'US',
        'GB',
        'CA',
        'AU',
        'IE'
    ]

    secondary_country_list = [
        'NG',
        'FR',
        'MX',
        'SE',
        'ES',
        'NL',
        'ZA',
        'PH',
        'EG',
        'IN'
    ]
    
    if which_type=='NEW':
        listings = listings.loc[
            (listings['COUNTRY']=='GLOBAL')
            | (listings['COUNTRY'].isin(priority_country_list) & (listings['RANK']<=40))
            | (listings['COUNTRY'].isin(secondary_country_list) & (listings['RANK']<=25))
        ]

        listings['N_LISTINGS'] = listings['SONG_ID'].map(listings['SONG_ID'].value_counts())

        song_ids_to_keep = listings.loc[listings['N_LISTINGS']>1]['SONG_ID'].drop_duplicates()
        
    elif which_type=='UP':
        listings = listings.loc[
            (
                ((listings['CHG']>=5) & (listings['RANK']<=50))
                | ((listings['CHG']>=10) & (listings['RANK']>50))
            )
        ]

        # Filter: non-priority countries with rank < 50
        non_priority_filtered = listings[
            (~listings['COUNTRY'].isin(priority_country_list))
        ]
        
        # Identify countries with at least 2 such listings
        valid_non_priority_countries = non_priority_filtered['SONG_ID'].value_counts()
        valid_non_priority_countries = valid_non_priority_countries[valid_non_priority_countries > 1].index

        song_ids_to_keep = listings.loc[
            (listings['COUNTRY'].isin(priority_country_list)) 
            | (listings['SONG_ID'].isin(valid_non_priority_countries))
        ]['SONG_ID'].drop_duplicates()
    
    
    listings = listings.loc[listings['SONG_ID'].isin(song_ids_to_keep)]

    if len(listings) > 0:
    
        updated_listings,charts_dates = group_chart_results(listings,charts_country_key,which_type)
        
        updated_listings = organize_multi_listings(updated_listings)
                
        updated_listings.sort_values('N_LISTINGS',ascending=False,inplace=True)
        
        updated_listings['CHART_DATE'] = charts_dates

    else:
        updated_listings = pd.DataFrame()

    return updated_listings


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 handler(event,context):

    logger = setup_logger()
    logger.info('Starting handler function...')

    import snowflake.connector    

    sf_params = init_secrets()
    sf_params['snowflake'] = snowflake

    email_params = setup_email_params()

    ###################################################
    ###################################################

    # READ/ORGANIZE DATA

    ###################################################
    ###################################################

    destination_email_list = orcd_query('select * from awal.awal_ar.email_alerts',sf_params)

    run_charts = True
    
    if run_charts:

        ###################################################
        ###################################################

        # CHARTS

        ###################################################
        ###################################################   

        logger.info('Pulling charts...')

        charts_country_key = orcd_query(f"""
            select * from awal.awal_ar.CHART_KEY_COUNTRIES
        """,sf_params)

        charts_df = orcd_query("""
        select * from awal.awal_ar.CHARTS
        """,sf_params)
        logger.info(f'{len(charts_df)} entries')

        logger.info('Getting corresponding song_ids...')

        isrcs = charts_df['ISRC'].drop_duplicates()
        upload_temp_stage_to_delphi('charts_isrcs',isrcs,sf_params)

        song_ids = delphi_query(f"""
        WITH isrcs as (
            SELECT $1 AS isrc 
            FROM @~/charts_isrcs.csv.gz
        )
        select distinct isrcs.isrc,ref.song_id
            from isrcs
            left join LUMINATE_DB_LISTING_DETAIL.EXTRACT_S.VW_MUSICAL_RECORDING_DS mr on mr.isrc=isrcs.isrc
            left join LUMINATE_DB_LISTING_DETAIL.EXTRACT_S.VW_MR_SONG_MAP_DS ref on ref.mr_id=mr.mr_id    
        where mr.isrc is not null and mr.type like 'Audio'
        """,sf_params)
        song_ids = song_ids.loc[~song_ids['SONG_ID'].isna()]

        charts_df = charts_df.merge(song_ids,how='left',on='ISRC')

        logger.info('Filtering...')

        md_key = orcd_query(f"""
        select 
            distinct r.song_id,
            r.ARTIST,r.title as SONG_TITLE,r.TRACK_URL,r.label,r.release_date,
            coalesce(c.country_of_origin,spf.region) as artist_region

        from awal.awal_ar.discovery_main r

        left join awal.awal_ar.DISCOVERY_ARTIST_MAP_SONG mp on mp.song_id=r.song_id
        left join awal.awal_ar.discovery_artist_table_cleaned c on c.artist_name=mp.final_artist_name
        left join awal.awal_ar.discovery_spotify_followers spf on spf.spotify_artist_id=c.spotify_artist_id

        where r.song_id in {tuple(charts_df['SONG_ID'].dropna().drop_duplicates())}
        """,sf_params)
        md_key = md_key.drop_duplicates('SONG_ID')
        song_ids_to_keep = md_key[['SONG_ID']].copy()
        song_ids_to_keep = song_ids_to_keep.drop_duplicates()

        # criteria to keep

        is_in_dm = charts_df.loc[charts_df['SONG_ID'].isin(song_ids_to_keep['SONG_ID'])]['SONG_ID'].drop_duplicates()

        et_timezone = pytz.timezone('US/Eastern')
        today = dt.datetime.now(et_timezone).date()
        charts_df['RELEASE_DATE'] = pd.to_datetime(charts_df['RELEASE_DATE']).dt.date

        is_new_release = charts_df.loc[charts_df['RELEASE_DATE'].between(today - relativedelta(days=2),today)]['SONG_ID'].drop_duplicates()

        charts_df = charts_df.loc[
            (charts_df['SONG_ID'].isin(is_new_release))
            | (charts_df['SONG_ID'].isin(is_in_dm))
        ]

        logger.info(f'{len(charts_df)} entries')


        logger.info('Organizing...')

        charts_df = charts_df[['SONG_ID','ARTIST','SONG_TITLE','DSP','COUNTRY','GENRE','RANK','CHG','MOVEMENT','CHART_DATE','DATE_UPDATED','LABEL','TRACK_URL','RELEASE_DATE','DAYS_ON_CHART','PEAK']]

        charts_df['SONG_KEY'] = charts_df['ARTIST'] + ' - ' + charts_df['SONG_TITLE']
        charts_df['SONG_ID'] = charts_df['SONG_ID'].fillna(charts_df['SONG_KEY'])

        charts_for_email = []

        ##########################

        ### Spotify Top200 ###
            
        ##########################

        listings = charts_df.loc[
                        (charts_df['DSP']=='SPOTIFY')
                        & (charts_df['GENRE']=='regional')
                    ]
        tmp = organize_spotify(listings,charts_country_key)

        if len(tmp) > 0:
            # final aggregate if needed
            tmp = tmp.groupby(['SONG_ID','CHART_DATE'], as_index=False).agg(LISTINGS=('LISTINGS', ' | '.join))
            
            tmp['TYPE'] = 'Spotify Top 200'
            charts_for_email.append(tmp)

        ##########################

        ### Spotify viral ###

        ##########################
            
        listings = charts_df.loc[
                    (charts_df['DSP']=='SPOTIFY')
                    & (charts_df['GENRE']=='viral')
                ]
        tmp = organize_spotify(listings,charts_country_key)

        if len(tmp) > 0:
            # final aggregate if needed
            tmp = tmp.groupby(['SONG_ID','CHART_DATE'], as_index=False).agg(LISTINGS=('LISTINGS', ' | '.join))
            
            tmp['TYPE'] = 'Spotify Viral Charts'
            charts_for_email.append(tmp)

        ##########################

        ### Apple Music All-Genre ###
            
        ##########################

        ### NEW ###

        listings = charts_df.loc[
                        (charts_df['DSP']=='APPLE')
                        & (charts_df['GENRE']=='All Genres')
                    ]
        tmp = organize_spotify(listings,charts_country_key) # using spotify function on purpose here

        if len(tmp) > 0:
            # final aggregate if needed
            tmp = tmp.groupby(['SONG_ID','CHART_DATE'], as_index=False).agg(LISTINGS=('LISTINGS', ' | '.join))
            
            tmp['TYPE'] = 'Apple Music'
            charts_for_email.append(tmp)

        ##########################

        ### Apple Music Genre-Specific ###
            
        ##########################

        ### NEW ###

        listings = charts_df.loc[
                        (charts_df['DSP']=='APPLE')
                        & (charts_df['GENRE']!='All Genres')
                    ]
        tmp = organize_apple_genres(listings,charts_country_key)

        if len(tmp) > 0:
            # final aggregate if needed
            tmp = tmp.groupby(['SONG_ID','CHART_DATE'], as_index=False).agg(LISTINGS=('LISTINGS', ' | '.join))
            
            tmp['TYPE'] = 'Apple Music (Genre charts)'
            charts_for_email.append(tmp)

        ##########################

        ### iTunes All-Genre ###
            
        ##########################

        ### NEW ###

        listings = charts_df.loc[
                        (charts_df['DSP']=='ITUNES')
                        & (charts_df['GENRE']=='All Genres')
                        & (charts_df['MOVEMENT']=='NEW')
                    ]
        tmp_new = organize_itunes_main(listings,charts_country_key,'NEW') 

        ### UP ###

        listings = charts_df.loc[
                        (charts_df['DSP']=='ITUNES')
                        & (charts_df['GENRE']=='All Genres')
                        & (charts_df['MOVEMENT']=='UP')
                    ]
        tmp_up = organize_itunes_main(listings,charts_country_key,'UP')  

        ### COMBINE ###

        tmp = pd.concat([tmp_new,tmp_up]).reset_index(drop=True)

        if len(tmp) > 0:
            # final aggregate if needed
            tmp = tmp.groupby(['SONG_ID','CHART_DATE'], as_index=False).agg(LISTINGS=('LISTINGS', ' | '.join))
            
            tmp['TYPE'] = 'iTunes'
            charts_for_email.append(tmp)

        ##########################

        ### iTunes Genre-Specific ###
            
        ##########################

        ### NEW ###

        listings = charts_df.loc[
                        (charts_df['DSP']=='ITUNES')
                        & (charts_df['GENRE']!='All Genres')
                        & (charts_df['MOVEMENT']=='NEW')
                    ]
        tmp_new = organize_itunes_genres(listings,charts_country_key,'NEW')

        ### UP ###

        listings = charts_df.loc[
                        (charts_df['DSP']=='ITUNES')
                        & (charts_df['GENRE']!='All Genres')
                        & (charts_df['MOVEMENT']=='UP')
                    ]
        tmp_up = organize_itunes_genres(listings,charts_country_key,'UP')

        ### COMBINE ###

        tmp = pd.concat([tmp_new,tmp_up]).reset_index(drop=True)

        if len(tmp) > 0:
            # final aggregate if needed
            tmp = tmp.groupby(['SONG_ID','CHART_DATE'], as_index=False).agg(LISTINGS=('LISTINGS', ' | '.join))
            
            tmp['TYPE'] = 'iTunes (Genre charts)'
            charts_for_email.append(tmp)

        ##########################

        ### Youtube ###
            
        ##########################

        ### NEW ###

        listings = charts_df.loc[
                        (charts_df['DSP']=='YOUTUBE')
                        & (charts_df['MOVEMENT']=='NEW')
                    ]
        tmp_new = organize_youtube(listings,charts_country_key,'NEW')

        ### UP ###

        listings = charts_df.loc[
                        (charts_df['DSP']=='YOUTUBE')
                        & (charts_df['MOVEMENT']=='UP')
                    ]
        tmp_up = organize_youtube(listings,charts_country_key,'UP')

        ### COMBINE ###

        tmp = pd.concat([tmp_new,tmp_up]).reset_index(drop=True)

        if len(tmp) > 0:
            # final aggregate if needed
            tmp = tmp.groupby(['SONG_ID','CHART_DATE'], as_index=False).agg(LISTINGS=('LISTINGS', ' | '.join))
            
            tmp['TYPE'] = 'Youtube'
            charts_for_email.append(tmp)


        ##########################
        ##########################
        ##########################

        charts_for_email = pd.concat(charts_for_email)

        logger.info('Merging metadata...')

        # md = orcd_query(f"""
        # WITH ranked_songs AS (
        #         SELECT *,
        #             ROW_NUMBER() OVER (
        #                 PARTITION BY SONG_ID 
        #                 ORDER BY gl_wtd_tp DESC, ARTIST_ORDER ASC
        #             ) AS rn
        #         FROM awal.awal_ar.DISCOVERY_MAIN
        #         WHERE song_id in {tuple(charts_for_email.loc[~charts_for_email['SONG_ID'].str.contains('-')]['SONG_ID'])}
        #     ),
        #     result as (
        #         SELECT *
        #         FROM ranked_songs
        #         WHERE rn = 1
        #     )
        #     select 
        #     r.song_id,r.TRACK_URL,r.label,r.release_date,
        #     CASE 
        #         WHEN f.song_id IS NULL THEN 1
        #         ELSE 0
        #     END AS exclude_song
        #     from result r
        #     left join awal.awal_ar.discovery_main f on f.song_id=r.song_id
        # """,sf_params)
        # md.drop_duplicates('SONG_ID',inplace=True)

        # charts_for_email = charts_for_email.merge(md,how='left',on='SONG_ID')
        # charts_for_email['LABEL'] = charts_for_email['LABEL_y'].fillna(charts_for_email['LABEL_x'].fillna('N/A'))

        # merge with metadata from discovery_main; ID which songs aren't in discovery_main
        merged = charts_for_email.merge(md_key.drop_duplicates(),how='left',on='SONG_ID',indicator=True)
        unmatched = merged[merged['_merge'] == 'left_only'].copy()
        matched = merged[merged['_merge'] == 'both'].copy()

        # for the unmatched songs, merge with original metadata from the charts pull
        unmatched.drop(columns=['ARTIST','SONG_TITLE','LABEL','TRACK_URL','RELEASE_DATE'], inplace=True)
        md_key_from_charts = charts_df[['SONG_ID','ARTIST','SONG_TITLE','LABEL','TRACK_URL','RELEASE_DATE']]
        md_key_from_charts = md_key_from_charts.dropna(subset=['ARTIST', 'SONG_TITLE'])
        md_key_from_charts = md_key_from_charts.drop_duplicates('SONG_ID')
        unmatched_merged = unmatched.merge(
            md_key_from_charts,
            how='left',
            on='SONG_ID'
        )
        charts_for_email = pd.concat([matched, unmatched_merged], ignore_index=True)
        

        charts_for_email = drop_sony_orchard(charts_for_email)

        # charts_for_email['EXCLUDE_SONG'] = charts_for_email['EXCLUDE_SONG'].fillna(0)
        # charts_for_email.drop(charts_for_email.loc[charts_for_email['EXCLUDE_SONG']==1].index,inplace=True)

        # charts_for_email['TRACK_URL'] = charts_for_email['TRACK_URL_y'].fillna(charts_for_email['TRACK_URL_x'].fillna('N/A'))
        
        charts_for_email['LUMINATE_STR'] = np.where(
            charts_for_email['SONG_ID']==charts_for_email['ARTIST'] + ' - ' + charts_for_email['SONG_TITLE'],
            "",
            f"https://app.luminatedata.com/song/" + charts_for_email['SONG_ID'].astype('str') + f"?g=AA&d=YTD&stf=Q018U0UsSU5ULFBDfENOLFZJfC8%3D&ssf=Lw%3D%3D&psf=Lw%3D%3D&a=ST&b=CM&sd=2025-01-03&ed={str(today-relativedelta(days=2))}&stgl=R0xCLVBWL1NFfEJTLENOfEJTLENNfEJT&psgl=R0xCLVNULw%3D%3D&ga=&m=VVNNLy9OQVRJT05BTA%3D%3D"
        )
        charts_for_email['LUMINATE_STR'] = np.where(
            charts_for_email['SONG_ID']==charts_for_email['ARTIST'] + ' - ' + charts_for_email['SONG_TITLE'],
            '',
            ' | <a href="' + charts_for_email['LUMINATE_STR'] + '">STREAMS</a>'
        )

        charts_for_email['SONG_TITLE'] = np.where(
            charts_for_email['TRACK_URL'].isna(),
            charts_for_email['SONG_TITLE'],
            '<a href="' + charts_for_email['TRACK_URL'] + '">' + charts_for_email['SONG_TITLE'] + '</a>'
        )

        most_recent_friday = today + relativedelta(days=-1, weekday=FR(-1))
        charts_for_email['NEW_RELEASE_TAG'] = np.where(
            charts_for_email['RELEASE_DATE'] >= most_recent_friday,
            " <font color='red'>(new release)</font>",
            ""
        )

        charts_for_email['TEXT'] = '<b>' + charts_for_email['ARTIST'] + ' - ' + charts_for_email['SONG_TITLE'] + '</b>' + charts_for_email['NEW_RELEASE_TAG'] + ' | <b>Artist Region</b> ' + charts_for_email['ARTIST_REGION'].fillna('N/A') + '<br><b>Label</b> ' + charts_for_email['LABEL'] + ' | <b>Released</b> ' + pd.to_datetime(charts_for_email['RELEASE_DATE']).dt.strftime('%Y-%m-%d') + charts_for_email['LUMINATE_STR'] + '<br><span style="padding-left: 25px;">' + charts_for_email['LISTINGS'] + '</span>'

        
        df_to_print = ''
        for each_type in charts_for_email['TYPE'].drop_duplicates():
            tmp = charts_for_email.loc[charts_for_email['TYPE']==each_type]

            # sort by artist (per DSP)
            tmp = tmp.sort_values('TEXT')

            chart_date = tmp['CHART_DATE'].iloc[0]

            tmp = tmp['TEXT'].str.cat(sep='<br><br>')
            tmp = '<p style="margin-left: 25px;">' + tmp + '</p>'

            df_to_print = df_to_print + '<h3>' + each_type + ' | ' + str(chart_date) + '</h3>' + tmp

        if len(df_to_print) > 0:

            email_subject = f'A&R Research: Charts {today}'

            email_head = f"""\
                    <html>
                        <head>
                            <b>Chart Update</b>
                            <br>
                            <br>
                            KEY:
                            <br>
                            <span style="background-color: yellow;">New entries</span>
                            <br>
                            <font color="green">▲ Moved up</font>
                            <br>
                            <font color='red'>(new release)</font> Released on/after Fri {most_recent_friday}
                            <br>
                            <font color='DodgerBlue'>(# of total days on a chart)</font> + whether this is a re-entry and/or new peak position
                            <br>
                            <br>
                            CHARTS REGION LIST:
                            <br>
                            Global, US, GB, Canada (CA), Australia (AU), Ireland (IE),
                            <br>
                            Nigeria (NG), France (FR), Mexico (MX), Sweden (SE), Spain (ES), Netherlands (NL), South Africa (ZA), Philippines (PH), Egypt (EG), India (IN),
                            <br>
                            Germany (DE), New Zealand (NZ), Denmark (DK), Norway (NO), Korea (KR), Indonesia (ID), Italy (IT), Argentina (AR), Colombia (CO)
                            <br>
                            <br>
                            <hr>
                            <br>
                    """

            full_email = email_head + df_to_print

            destination_emails = destination_email_list.loc[destination_email_list['CATEGORY']=='CHARTS']
            destination_emails = destination_emails['EMAILS'].tolist()

            email_outcome = prep_and_send_email(email_params,destination_emails,email_subject,full_email)
            if email_outcome:
                logger.info('Email sent.')
            else:
                logger.info('Error: Email did not send')

        else:
            logger.info('No chart positions - no need to send email')