# -*- coding: utf-8 -*-
"""
Created on Tue May 12 00:00:00 2020
@author: Shaji
"""
from . import exceptions
from datetime import datetime
import os
import pandas as pd
def column_level_check(source_df,target_df,primary_keys):
"""
Usage: [arg1]:[Pandas DataFrame - source], [arg2]:[Pandas DataFrame - target], [arg3]:[Primary keys (separated by comma)]
Description: Performs column level testing between two DataFrames by joining using the primary keys.
Returns: [Mismatch Count], [Test Log (list)], [Pandas dataframe - mismatch (if any)]
"""
global execution_status
systime=datetime.now()
start_time=systime.strftime("%Y")+'-'+systime.strftime("%m")+'-'+systime.strftime("%d")+' '+systime.strftime("%H")+':'+systime.strftime("%M")+':'+systime.strftime("%S")
log_list=[]
execution_status='RUNNING'
log_list.append('START TIME: '+start_time)
key_list=primary_keys.split(',')
src=source_df
tgt=target_df
log_list.append(str(datetime.now())+': DIFFERENTIATING SOURCE AND TARGET COLUMNS')
if execution_status!='FAILED':
try:
src_k=[]
src_columns=[]
for i in src.columns:
if str.lower(i) in [str.lower(key) for key in key_list]:
src_columns.append(str.lower(i))
src_k.append(str.lower(i))
else:
src_columns.append(str(i) + '_src')
src.columns = src_columns
tgt_k=[]
tgt_columns=[]
for i in tgt.columns:
if str.lower(i) in [str.lower(key) for key in key_list]:
tgt_columns.append(str.lower(i))
tgt_k.append(str.lower(i))
else:
tgt_columns.append(str(i) + '_tgt')
tgt.columns = tgt_columns
except Exception as e:
print('Failed while DIFFERENTIATING SOURCE AND TARGET COLUMNS: '+str(e))
log_list.append('Failed while DIFFERENTIATING SOURCE AND TARGET COLUMNS: '+str(e))
execution_status='FAILED'
log_list.append(str(datetime.now())+': CHECKING IF THE GROUP BY MAKES THE RECORD LEVEL SAME AS ACTUAL')
if execution_status!='FAILED':
try:
index_unique_flag=[]
if src.groupby(src_k).count().shape[0]==src.shape[0]:
index_unique_flag.append(True)
else:
index_unique_flag.append(False)
if tgt.groupby(tgt_k).count().shape[0]==tgt.shape[0]:
index_unique_flag.append(True)
else:
index_unique_flag.append(False)
except Exception as e:
print('Failed while CHECKING IF THE GROUP BY MAKES THE RECORD LEVEL SAME AS ACTUAL: '+str(e))
log_list.append('Failed while CHECKING IF THE GROUP BY MAKES THE RECORD LEVEL SAME AS ACTUAL: '+str(e))
execution_status='FAILED'
if execution_status!='FAILED':
try:
if all(index_unique_flag)==True:
log_list.append(str(datetime.now())+': JOINING THE TABLES')
try:
df=tgt.set_index(tgt_k).join(src.set_index(src_k),how='left')
except Exception as e:
print('Failed while JOINING THE TABLES: '+str(e))
log_list.append('Failed while JOINING THE TABLES: '+str(e))
execution_status='FAILED'
log_list.append(str(datetime.now())+': FINDING THE TARGET COLUMN AND SOURCE COLUMN TO BE COMPARED')
if execution_status!='FAILED':
try:
ma_list=[]
for i in range(len(df.columns)):
if df.columns[i][-3:]=='tgt':
for j in range(len(df.columns)):
if df.columns[j][-3:]=='src':
if str.lower(df.columns[i][:-4])==str.lower(df.columns[j][:-4]):
ma_list.append([j,i])
match_cols=''
for i in range(len(ma_list)):
match_cols+=str(i+1)+': '+df.columns[ma_list[i][1]]+' = '+df.columns[ma_list[i][0]]+' , '
log_list.append('Matching columns '+match_cols)
except Exception as e:
print('Failed while FINDING THE TARGET COLUMN AND SOURCE COLUMN TO BE COMPARED: '+str(e))
log_list.append('Failed while FINDING THE TARGET COLUMN AND SOURCE COLUMN TO BE COMPARED: '+str(e))
execution_status='FAILED'
log_list.append(str(datetime.now())+': COMPARISION STARTED')
if execution_status!='FAILED':
try:
mis_cols=[]
res=[]
index=[]
for i in range(len(ma_list)):
if all(df[df.columns[ma_list[i][0]]].apply(lambda x:str(x).strip()).astype(str).fillna(str(0))==df[df.columns[ma_list[i][1]]].apply(lambda x:str(x).strip()).astype(str).fillna(str(0)))==True:
res.append(True)
else:
res.append(False)
mis_cols.append(df.columns[ma_list[i][0]])
mis_cols.append(df.columns[ma_list[i][1]])
for j in range(len(df[df.columns[ma_list[i][0]]].apply(lambda x:str(x).strip()).astype(str).fillna(str(0))==df[df.columns[ma_list[i][1]]].apply(lambda x:str(x).strip()).astype(str).fillna(str(0)))):
if list(df[df.columns[ma_list[i][0]]].apply(lambda x:str(x).strip()).astype(str).fillna(str(0))==df[df.columns[ma_list[i][1]]].apply(lambda x:str(x).strip()).astype(str).fillna(str(0)))[j]==False:
index.append(j)
un_df=df[mis_cols].iloc[list(set(index))]
except Exception as e:
print('Failed while COMPARING: '+str(e))
log_list.append('Failed while COMPARING: '+str(e))
execution_status='FAILED'
log_list.append(str(datetime.now())+': TEST RESULT:')
if execution_status!='FAILED':
try:
if all(res)==True:
mismatch_count=0
print('COLUMN LEVEL CHECK PASSED')
execution_status='SUCCESS'
log_list.append('COLUMN LEVEL CHECK PASSED')
else:
log_list.append((str(len(set(index)))+' records unmatched'))
log_list.append('Column level check Failed')
mismatch_count=str(len(set(index)))
execution_status='SUCCESS'
except Exception as e:
print('Failed while getting the TEST RESULT: '+str(e))
log_list.append('Failed while getting the TEST RESULT: '+str(e))
execution_status='FAILED'
else:
log_list.append('The records grouped at the level of key columns are not unique')
except Exception as e:
log_list.append('Failed while CHECKING IF THE GROUP BY MAKES THE RECORD LEVEL SAME AS ACTUAL: '+str(e))
execution_status='FAILED'
if execution_status=='FAILED':
print('Check Logs for the error message')
raise exceptionsExecutionError
return mismatch_count,log_list,un_df
def sort_and_compare(source_df,target_df):
"""
Usage: [arg1]:[Pandas DataFrame - source], [arg2]:[Pandas DataFrame - target]
Description: Sort and Compare two datasets.
Returns: [Mismatch Count], [Test Log (list)], [Pandas dataframe - mismatch (if any)]
"""
log_list=[]
col1=source_df.columns
col2=target_df.columns
cols=list(set(col1.sort_values()).intersection(set(col2.sort_values())))
log_list.append('Common column(s): '+', '.join(cols))
source_df.sort_values(cols, axis=0, ascending=True, inplace=True)
target_df.sort_values(cols, axis=0, ascending=True, inplace=True)
data1=source_df[cols].reset_index(drop=True)
data2=target_df[cols].reset_index(drop=True)
data1.head()
data2.head()
result=data1==data2
bool_list=[]
mis_cols=[]
mis_index=[]
for i in cols:
if all(result[i])==True:
bool_list.append(True)
else:
bool_list.append(False)
mis_cols.append(i)
for j in range(len(result[i])):
if result[i][j]==False:
mis_index.append(j)
un_df=pd.concat([data1.iloc[list(set(mis_index))],data2.iloc[list(set(mis_index))]],axis=1)
mismatch_count=0
if all(bool_list)==True:
log_list.append('Records are matching')
else:
mismatch_count=len(set(mis_index))
log_list.append(str(mismatch_count)+' records unmatched')
log_list.append('Column(s): '+', '.join(mis_cols))
return mismatch_count,log_list,un_df[mis_cols]