import sys
import os
from sf_conn import snowflake_connection as sf
import pandas as pd
import re
from pandas import json_normalize
import json
import ast

def sf_to_pd(sql):
    conn = sf.get_snowflake_connection()
    cur = conn.cursor()

    cur.execute(sql)
    df = cur.fetch_pandas_all()

    conn.close()
    return df

#Below dfs are the queries sent to Snowflake
SaxTable = sf_to_pd("select distinct * from royalty_accounting.prod.sax_agreement_denorm WHERE agreement_type NOT IN  ('Synch Only') and (is_terminated = 'N' or novation_flag not like '%Novated to%') and agreement_id not in ('AGR663514', 'AGR663521', 'AGR663560', 'AGR663568', 'AGR686290', 'AGR622724', 'AGR663311', 'AGR663516', 'AGR663519', 'AGR506599', 'AGR663213',     'AGR663291', 'AGR663524')  and (right_description not like '%Pro Rata Share of One-Off Income%' or right_description is null) ")
AbacusContract = sf_to_pd('select distinct contract_id, sax_agreement_id as agreement_id from royalty_accounting.prod.vw_dim_abacus_contract')
RefExtSource = sf_to_pd('select distinct * from orchard_app_reporting_v2.PROD_royalty_accounting_royalty_accounting.reference_external_source')
AbacusUpcs = sf_to_pd("select distinct contract_id, UPC as list_upcs from royalty_accounting.prod.contract_denormalized_distro where upc is not null")
AbacusIsrcs = sf_to_pd("select distinct contract_id, ISRC as abacus_isrcs from royalty_accounting.prod.contract_denormalized_distro where isrc is not null")
AbacusTermType = sf_to_pd("select distinct contract_id, contract_term_id, term_type, attachments, IS_BASE_TERM from orchard_app_reporting_v2.prod_royalty_accounting_royalty_accounting.contract_term")
AbacusTermCondition = sf_to_pd("select distinct contract_term_condition_id, term_rate, ARRAY_TO_STRING(PARSE_JSON(max(countries)), ',') as list_countries, ARRAY_TO_STRING(PARSE_JSON(max(transaction_types)), ', ') as list_transaction_types from ( select distinct contract_term_condition_id, conditions, term_rate, iff(key='countries', value::string,null) as countries, iff(key='transaction_types', value::string,null) as transaction_types from orchard_app_reporting_v2.prod_royalty_accounting_royalty_accounting.contract_term_condition, lateral flatten (input => CONDITIONS) at ) group by 1,2")
SaxIso = sf_to_pd("select ter.description as sax_territory_desc, c.iso as sax_iso from awal_sax.sax.territories ter left join royalty_accounting.prod.sax_territory_to_orchard_iso c on c.id = ter.id")
# AbacusRemapUpcs = art_relations- > releases -> Manufacture upc where upc != upc
# Remove all the ones from Abacus and then the remaps 


#Renaming the column to have a clean join
AbacusContract.rename(columns={'SAX_AGREEMENT_ID':'AGREEMENT_ID'}, inplace=True)
SaxTable.rename(columns={'ABACUS_RATE':'SAX_ABACUS_RATE'}, inplace=True)

#Inner Join of SAX agreement with Abacus Contract 
SaxContract = SaxTable.merge(AbacusContract, on='AGREEMENT_ID', how='inner')

#Applying filters on Reference External Source DF to get the rights ID and Contract TErm Condition 
RightsId_df = RefExtSource[
        (RefExtSource['PARENT_TABLE_NAME']=='contract_term_condition')
        & (RefExtSource['EXTERNAL_SOURCE'].isin(['sax_rights_id', 'sax_right_id']))
        & (RefExtSource['_FIVETRAN_DELETED']==False)
]

#Logic to get the Rights Id starting with RGT followed by the number
RightsId_df['ABACUS_RIGHTS_ID'] = RightsId_df['EXTERNAL_SOURCE_ID'].str.extract(r'(RGT\d+)')

# print(list(RightsId_df.columns.values))

#Replacing NULL values with 'no_value' as it will be an issue in left join
SaxContract['RIGHTS_ID'].fillna('no_value', inplace=True)

#Left Join of the Rights DF with the SAX Contract to create a new DF named SaxRights
SaxRights = SaxContract.merge(RightsId_df[['PARENT_TABLE_ID', 'ABACUS_RIGHTS_ID','EXTERNAL_SOURCE_ID']], left_on='RIGHTS_ID', right_on='ABACUS_RIGHTS_ID', how='left')

#Replacing the NULL & Blank values with 'no_value'
SaxRights['SAX_RIGHT_TYPE_IDS'].fillna('no_value', inplace=True)
SaxRights['SAX_RIGHT_TYPE_IDS'].replace('', 'no_value', inplace=True)

#Below logic is to check ABACUS RIGHTS ID matching with Abacus and counter checking if Right Type IDs are available or not
SaxRights['SAX_RIGHTS_BLOCK_MATCH_ABACUS_INTERIM'] = SaxRights.apply(lambda row: 'Yes' if row['RIGHTS_ID'] == row['ABACUS_RIGHTS_ID'] else 'No', axis =1)
SaxRights['SAX_RIGHTS_BLOCK_MATCH_ABACUS'] = SaxRights.apply(lambda row: 'No Right available' if (row['SAX_RIGHTS_BLOCK_MATCH_ABACUS_INTERIM'] == 'No' and  row['SAX_RIGHT_TYPE_IDS']=='no_value') else row['SAX_RIGHTS_BLOCK_MATCH_ABACUS_INTERIM'] , axis =1)

#Renaming the column Parent Table ID with Contract Term Condition 
SaxRights.rename(columns={'PARENT_TABLE_ID': 'CONTRACT_TERM_CONDITION_ID'}, inplace=True)
SaxRights.rename(columns={'EXTERNAL_SOURCE_ID': 'ABACUS_RIGHTS_LABEL_ID'}, inplace=True)

#Applying filters on Reference External Source DF to get the Schedule ID and Contract Term 
ScheduleId_df = RefExtSource[
        (RefExtSource['PARENT_TABLE_NAME']=='contract_term')
        & (RefExtSource['EXTERNAL_SOURCE'].isin(['sax_schedule_id', 'sax_recording_schedule_id']))
        & (RefExtSource['_FIVETRAN_DELETED']==False)
]
#Logic to get the Schedule Id starting with RCS followed by the number
ScheduleId_df['ABACUS_SCHEDULE_ID'] = ScheduleId_df['EXTERNAL_SOURCE_ID'].str.extract(r'(RCS\d+)')

SaxRights['RECORDING_SCHEDULE_ID'].fillna('no_value', inplace=True)
SaxRights['SAX_UPCS'].fillna('no_value', inplace=True)
SaxRights['SAX_ISRCS'].fillna('no_value', inplace=True)
SaxRights['SAX_UPCS'].replace('', 'no_value', inplace=True)
SaxRights['SAX_ISRCS'].replace('', 'no_value', inplace=True)


SaxSchedule = SaxRights.merge(ScheduleId_df[['PARENT_TABLE_ID', 'ABACUS_SCHEDULE_ID']], left_on='RECORDING_SCHEDULE_ID', right_on='ABACUS_SCHEDULE_ID', how='left')

SaxSchedule['SAX_SCHEDULE_MATCH_ABACUS_INTERIM'] = SaxSchedule.apply(lambda row: 'Yes' if row['RECORDING_SCHEDULE_ID'] == row['ABACUS_SCHEDULE_ID']  else 'No', axis =1)
SaxSchedule['SAX_SCHEDULE_MATCH_ABACUS'] = SaxSchedule.apply(lambda row: 'No UPC or ISRC available' if (row['SAX_SCHEDULE_MATCH_ABACUS_INTERIM'] == 'No' and (row['SAX_UPCS']=='no_value' or row['SAX_ISRCS']=='no_value')) else row['SAX_SCHEDULE_MATCH_ABACUS_INTERIM'] , axis =1)


SaxSchedule['SAX_SCHEDULE_MATCH_ABACUS'] = SaxSchedule.apply(lambda row: 'No Schedule - Label Base Term' if (row['SAX_SCHEDULE_MATCH_ABACUS'] == 'No' and '-label' in str(row['ABACUS_RIGHTS_LABEL_ID']) ) else row['SAX_SCHEDULE_MATCH_ABACUS'], axis=1)

SaxSchedule.rename(columns={'PARENT_TABLE_ID': 'CONTRACT_TERM_ID'}, inplace=True)

#SAX Term Type Addition to SAX Schedule
SaxTermConditions = SaxSchedule.merge(AbacusTermCondition[['CONTRACT_TERM_CONDITION_ID', 'TERM_RATE', 'LIST_COUNTRIES', 'LIST_TRANSACTION_TYPES']], how='left', on='CONTRACT_TERM_CONDITION_ID')

SaxTermConditions['SAX_RATE_MATCHED_ABACUS_RATE'] = SaxTermConditions.apply(lambda row: 'Yes' if row['SAX_ABACUS_RATE'] == row['TERM_RATE'] and not pd.isna(row['TERM_RATE']) and not pd.isna(row['SAX_ABACUS_RATE']) else 'No', axis=1)

SaxTerms = SaxTermConditions.merge(AbacusTermType[['CONTRACT_TERM_ID', 'TERM_TYPE']], how='left', on='CONTRACT_TERM_ID')

#Transaction Types Comparison

SaxTxnTypes_df = SaxSchedule.groupby(['CONTRACT_ID','CONTRACT_TERM_CONDITION_ID'])['ABACUS_TRANSACTION_TYPE_IDS'].unique().agg(list).reset_index(name='SAX_TRANSACTION_TYPES')
AbacusTermCondition_df = AbacusTermCondition.groupby(['CONTRACT_TERM_CONDITION_ID'])['LIST_TRANSACTION_TYPES'].unique().agg(list).reset_index(name='ABACUS_TRANSACTION_TYPES')

AbacusTermCondition_df['CONTRACT_TERM_CONDITION_ID'] = AbacusTermCondition_df['CONTRACT_TERM_CONDITION_ID'].astype(int)

SaxAbacusTxnTypes = SaxTxnTypes_df.merge(AbacusTermCondition_df[['CONTRACT_TERM_CONDITION_ID','ABACUS_TRANSACTION_TYPES']], how='left', on='CONTRACT_TERM_CONDITION_ID')

SaxAbacusTxnTypes['ABACUS_TRANSACTION_TYPES'] = SaxAbacusTxnTypes['ABACUS_TRANSACTION_TYPES'].apply(lambda x: list(set(x)) if isinstance(x, list) else [])
SaxAbacusTxnTypes['SAX_TRANSACTION_TYPES'] = SaxAbacusTxnTypes['SAX_TRANSACTION_TYPES'].apply(lambda x: list(set((map(int, ''.join(x).split(','))))))
SaxAbacusTxnTypes['ABACUS_TRANSACTION_TYPES'] = SaxAbacusTxnTypes['ABACUS_TRANSACTION_TYPES'].apply(lambda x: list(set((map(int, ''.join(x).split(','))))) if x != [] else x)

SaxAbacusTxnTypes['SAX_Txn_Type_not_present_in_ABACUS'] = SaxAbacusTxnTypes.apply(lambda row: [val for val in row['SAX_TRANSACTION_TYPES'] if val not in row['ABACUS_TRANSACTION_TYPES']] , axis=1)
SaxAbacusTxnTypes['SAX_Txn_Type_in_ABACUS[Yes/No]'] = SaxAbacusTxnTypes.apply(lambda row: 'Yes' if row['SAX_Txn_Type_not_present_in_ABACUS'] == [] else 'No', axis=1)


#List of Countries comparison
def split_and_process(value_list):
    result = []
    for value in value_list:
        # Split the string by commas
        items = re.split(r',\s*', value)

        # Use regular expressions to identify and merge specific phrases
    return items

SaxListCountries_df = SaxSchedule.groupby(['CONTRACT_ID','CONTRACT_TERM_CONDITION_ID'])['SAX_TERRITORIES'].unique().agg(list).reset_index(name='LIST_SAX_TERRITORIES')
# SaxListCountries_df = SaxListCountries_df.head(20)
SaxListCountries_df['LIST_SAX_TERRITORIES'] = SaxListCountries_df['LIST_SAX_TERRITORIES'].apply(split_and_process)
# SaxListCountries_df['LIST_SAX_TERRITORIES'] = SaxListCountries_df['LIST_SAX_TERRITORIES'].apply(lambda x: [item.strip() for item in x[0].split(',')])
# SaxListCountriesExt_df = SaxListCountries_df['LIST_SAX_TERRITORIES'].explode().tolist()
# SaxListCountriesExt_df.tolist()

output_excel_filename = 'sax_abacus_country_comparison.xlsx'
with pd.ExcelWriter(output_excel_filename) as excel_writer:
    SaxListCountries_df.to_excel(excel_writer, sheet_name='Overview', index=False)
    # SaxListCountriesExt_df.to_excel(excel_writer, sheet_name='SaxListCountriesExt_df', index=False)

ssss
# UPCs and ISRCs
SaxUpcs_df = SaxSchedule.groupby('CONTRACT_ID')['SAX_UPCS'].unique().agg(list).reset_index(name='SAX_LIST_UPCS')
SaxIsrc_df = SaxSchedule.groupby('CONTRACT_ID')['SAX_ISRCS'].unique().agg(list).reset_index(name='SAX_LIST_ISRCS')

# print(SaxUpcs_df.loc[SaxUpcs_df['CONTRACT_ID']==532301])

AbacusUpcs['LIST_UPCS'] = AbacusUpcs['LIST_UPCS'].astype(int)

AbacusUpcs_df = AbacusUpcs.groupby('CONTRACT_ID')['LIST_UPCS'].unique().agg(list).reset_index(name='LIST_UPCS')
AbacusIsrcs_df = AbacusIsrcs.groupby('CONTRACT_ID')['ABACUS_ISRCS'].unique().agg(list).reset_index(name='ABACUS_ISRCS')

SaxAbacusUpcs = SaxUpcs_df.merge(AbacusUpcs_df[['CONTRACT_ID','LIST_UPCS']], how='left', on='CONTRACT_ID')
SaxAbacusIsrcs = SaxIsrc_df.merge(AbacusIsrcs_df[['CONTRACT_ID','ABACUS_ISRCS']], how='left', on='CONTRACT_ID')


SaxAbacusUpcs['LIST_UPCS'].fillna('no_value', inplace=True)
SaxAbacusUpcs['SAX_LIST_UPCS'].fillna('no_value', inplace=True)

SaxAbacusIsrcs['ABACUS_ISRCS'].fillna('no_value', inplace=True)
SaxAbacusIsrcs['SAX_LIST_ISRCS'].fillna('no_value', inplace=True)


# Function to calculate the difference between two lists
def calculate_difference_isrcs(first_values,second_values):
    
    if second_values == 'no_value':
        second_values=[]
    difference =[]
    for value in first_values:
        if value == 'no_value':
            continue
        if isinstance(value, str):
            first_list = list(value.split(','))
        else:
            continue
        difference.extend(list(set(first_list) - set(second_values)))
    return list(set(difference))

# Function to calculate the difference between two lists
def calculate_difference_upcs(first_values,second_values):

    if second_values == 'no_value':
        second_values=[]
    difference =[]

    for value in first_values:
        # Handle 'no_value'
        if value == 'no_value':
            continue
        if isinstance(value, str):
            first_list = list(map(int, value.split(',')))
        elif isinstance(value, int):
            first_list = [value]
        else:
            continue
        difference.extend(list(set(first_list) - set(second_values)))
    return list(set(difference))
    
#55739
SaxAbacusUpcs['SAX_UPC_not_present_in_ABACUS'] = SaxAbacusUpcs.apply(lambda row: calculate_difference_upcs(row['SAX_LIST_UPCS'], row['LIST_UPCS']) if (row['SAX_LIST_UPCS']!='no_value') else row['LIST_UPCS'], axis=1)
SaxAbacusIsrcs['SAX_ISRC_not_present_in_ABACUS'] = SaxAbacusIsrcs.apply(lambda row: calculate_difference_isrcs(row['SAX_LIST_ISRCS'], row['ABACUS_ISRCS']) if (row['SAX_LIST_ISRCS']!='no_value') else row['ABACUS_ISRCS'], axis=1)


SaxOutputColumns = ['AGREEMENT_ID', 'CONTRACT_ID', 'SAX_RIGHTS_BLOCK_MATCH_ABACUS', 'RIGHTS_ID', 'CONTRACT_TERM_CONDITION_ID', 'SAX_SCHEDULE_MATCH_ABACUS', 'RECORDING_SCHEDULE_ID', 'CONTRACT_TERM_ID', 'TERM_TYPE', 'SAX_RATE_MATCHED_ABACUS_RATE', 'TERM_RATE', 'SAX_ABACUS_RATE']

SaxOutput = SaxTerms[SaxOutputColumns].drop_duplicates()

SaxTxnTypes = SaxOutput.merge(SaxAbacusTxnTypes[['CONTRACT_ID', 'CONTRACT_TERM_CONDITION_ID','SAX_TRANSACTION_TYPES', 'ABACUS_TRANSACTION_TYPES','SAX_Txn_Type_not_present_in_ABACUS','SAX_Txn_Type_in_ABACUS[Yes/No]']], on=['CONTRACT_ID', 'CONTRACT_TERM_CONDITION_ID'], how='left')

SaxTxnTypes['SAX_TRANSACTION_TYPES'].fillna('no_value', inplace=True)
SaxTxnTypes['ABACUS_TRANSACTION_TYPES'].fillna('no_value', inplace=True)
SaxTxnTypes['SAX_Txn_Type_not_present_in_ABACUS'].fillna('no_value', inplace=True)
SaxTxnTypes['SAX_Txn_Type_in_ABACUS[Yes/No]'].fillna('no_value', inplace=True)

print(SaxTxnTypes.head(9))

OutputSummaryColumns = ['AGREEMENT_ID', 'CONTRACT_ID', 'SAX_RIGHTS_BLOCK_MATCH_ABACUS', 'SAX_SCHEDULE_MATCH_ABACUS', 'SAX_RATE_MATCHED_ABACUS_RATE', 'SAX_Txn_Type_in_ABACUS[Yes/No]']
DetailedOutputColumns = ['AGREEMENT_ID', 'CONTRACT_ID', 'SAX_RIGHTS_BLOCK_MATCH_ABACUS', 'RIGHTS_ID', 'CONTRACT_TERM_CONDITION_ID', 'SAX_SCHEDULE_MATCH_ABACUS', 'RECORDING_SCHEDULE_ID', 'CONTRACT_TERM_ID', 'TERM_TYPE', 'SAX_RATE_MATCHED_ABACUS_RATE', 'TERM_RATE', 'SAX_ABACUS_RATE', 'SAX_Txn_Type_in_ABACUS[Yes/No]', 'SAX_TRANSACTION_TYPES', 'ABACUS_TRANSACTION_TYPES','SAX_Txn_Type_not_present_in_ABACUS']

OutputSummaryColumns = SaxTxnTypes[OutputSummaryColumns].drop_duplicates()
DetailedOutputColumns = SaxTxnTypes[DetailedOutputColumns]

SaxUPC = DetailedOutputColumns.merge(SaxAbacusUpcs, on='CONTRACT_ID', how='left')
SaxISRC = SaxUPC.merge(SaxAbacusIsrcs, on='CONTRACT_ID', how='left')
# SaxTxnTypes = SaxUPC.merge(SaxAbacusTxnTypes, on=['CONTRACT_ID', 'CONTRACT_TERM_CONDITION_ID'], how='left')


# output_excel_filename = 'sax_abacus_upc_comparison.xlsx'
# with pd.ExcelWriter(output_excel_filename) as excel_writer:
#     SaxUPC.to_excel(excel_writer, sheet_name='Overview', index=False)


output_excel_filename = 'sax_abacus_comparison.xlsx'
with pd.ExcelWriter(output_excel_filename, engine='xlsxwriter') as excel_writer:
    OutputSummaryColumns.to_excel(excel_writer, sheet_name='Overview', index=False)
    DetailedOutputColumns.to_excel(excel_writer, sheet_name='Detailed Diff 1', index=False)
    SaxISRC.to_excel(excel_writer, sheet_name='Detailed Diff 2', index=False)