#! /usr/bin/env python3 from email.mime.multipart import MIMEMultipart from email.mime.text import MIMEText from contextlib import closing from logging import info import snowflake.connector import psycopg2 import smtplib import re import argparse import sys import json import datetime import boto3 import calendar # connection handles sf_conn = None pg_conn = None with open('config.json', 'r') as f: config = json.load(f) sf_user = config['snowflake']["user"] sf_password = config['snowflake']["password"] sf_account = config['snowflake']["account"] sf_warehouse = config['snowflake']['warehouse'] local_db_name = config['failure_counter']["database"] local_db_user = config['failure_counter']["user"] local_db_password = config['failure_counter']["password"] local_db_host = config['failure_counter']["host"] local_db_port = config['failure_counter']["port"] failure_count_increase = "UPDATE failed_count SET failed_count = failed_count + 1, last = 'FAILED', last_updated = NOW() WHERE count_id = 'aegdprw_thursday';" failure_count_to_zero = "UPDATE failed_count SET failed_count = 0, last = 'OK', last_updated = NOW() WHERE count_id = 'aegdprw_thursday';" def postgres_update_failure_count(sql_query): with closing(psycopg2.connect(database=local_db_name, user=local_db_user, password=local_db_password, host=local_db_host, port=local_db_port)) as connection: with connection.cursor() as cursor: cursor.execute(sql_query) connection.commit() def create_arg_parser(): parser = argparse.ArgumentParser() parser.add_argument('-m', '--mail', nargs='+', action='store', dest='recipients', type=str) parser.add_argument('-p', '--print', action="store_const", const=True) return parser def send_mail(recipients: str, message: str, status: str) -> None: # Create a multipart message msg = MIMEMultipart() body_part = MIMEText(message, 'plain') msg['Subject'] = f"[{status}][PROD] Appreciation Engine GDPR weekly monitoring" msg['From'] = "support.sme@dataart.com" msg['To'] = recipients # Add body to email msg.attach(body_part) # Create SMTP object server = smtplib.SMTP('relay1.dataart.com', 25) server.sendmail(msg['From'], msg['To'].split(','), msg.as_string()) server.quit() def snowflake_fetch_data(sql_query) -> dict: # fetching data from snowflake global sf_conn if not sf_conn: try: sf_conn = snowflake.connector.connect( user=sf_user, password=sf_password, account=sf_account, warehouse=sf_warehouse, network_timeout=1200 ) except: return False cs = sf_conn.cursor(snowflake.connector.DictCursor) try: cs.execute(sql_query) result_dict = cs.fetchall() finally: cs.close() return result_dict def check_4(id_list): # Step 4: Weekly data clean-up verification - MEMBER sql_query = 'select count(*) from "DELPHI_EXPLORATION"."MAIN"."AE_MEMBER" where ID in ( ' + id_list +' ) and ID in (select MEMBER_ID from DELPHI_EXPLORATION.MAIN.AE_MEMBERS_GDPR_UNLISTED_MAPPING where MEMBER_ID is not null);' result_raw = snowflake_fetch_data(sql_query) if len(result_raw) > 0: # we have got some rows from snowflake result_count = int(result_raw[0]['COUNT(*)']) print("Check 4: count = " + str(result_count)) return result_count else: return False def check_5(id_list): # Step 5: Weekly data clean-up verification - MEMBER_PERSONAL sql_query = 'select count(*) from "DELPHI_EXPLORATION"."MAIN"."AE_MEMBER_PERSONAL" where MEMBER_ID in ( ' +id_list + ' ) and MEMBER_ID in (select MEMBER_ID from DELPHI_EXPLORATION.MAIN.AE_MEMBERS_GDPR_UNLISTED_MAPPING where MEMBER_ID is not null);' result_raw = snowflake_fetch_data(sql_query) if len(result_raw) > 0: # we have got some rows from snowflake result_count = int(result_raw[0]['COUNT(*)']) print("Check 5: count = " + str(result_count)) return result_count else: return False def check_6(id_list): # Step 6: Weekly data clean-up verification - MEMBERS_PROFILES sql_query = 'select count(*) from "DELPHI_EXPLORATION"."MAIN"."AE_MEMBERS_PROFILES" where MEMBER_ID in ( ' +id_list + ' ) and MEMBER_ID in (select MEMBER_ID from DELPHI_EXPLORATION.MAIN.AE_MEMBERS_GDPR_UNLISTED_MAPPING where MEMBER_ID is not null);' result_raw = snowflake_fetch_data(sql_query) if len(result_raw) > 0: # we have got some rows from snowflake result_count = int(result_raw[0]['COUNT(*)']) print("Check 6: count = " + str(result_count)) return result_count else: return False def check_7(id_list): # Step 7: Weekly data clean-up verification - MEMBERS_SERVICES sql_query = 'select count(*) from "DELPHI_EXPLORATION"."MAIN"."AE_MEMBERS_SERVICES" where MEMBER_ID in ( ' +id_list + ' ) and MEMBER_ID in (select MEMBER_ID from DELPHI_EXPLORATION.MAIN.AE_MEMBERS_GDPR_UNLISTED_MAPPING where MEMBER_ID is not null);' result_raw = snowflake_fetch_data(sql_query) if len(result_raw) > 0: # we have got some rows from snowflake result_count = int(result_raw[0]['COUNT(*)']) print("Check 7: count = " + str(result_count)) return result_count else: return False def check_8(id_list): # Step 8: Weekly data clean-up verification - MEMBERS_DOMAINS sql_query = 'select count(*) from "DELPHI_EXPLORATION"."MAIN"."AE_MEMBERS_DOMAINS" where MEMBER_ID in ( ' +id_list + ' ) and MEMBER_ID in (select MEMBER_ID from DELPHI_EXPLORATION.MAIN.AE_MEMBERS_GDPR_UNLISTED_MAPPING where MEMBER_ID is not null);' result_raw = snowflake_fetch_data(sql_query) if len(result_raw) > 0: # we have got some rows from snowflake result_count = int(result_raw[0]['COUNT(*)']) print("Check 8: count = " + str(result_count)) return result_count else: return False def check_9(id_list): # Step 9: Weekly data clean-up verification - MEMBERS_OPT_INS sql_query = 'select count(*) from "DELPHI_EXPLORATION"."MAIN"."AE_MEMBERS_OPT_INS" where MEMBER_ID in ( ' +id_list + ' ) and MEMBER_ID in (select MEMBER_ID from DELPHI_EXPLORATION.MAIN.AE_MEMBERS_GDPR_UNLISTED_MAPPING where MEMBER_ID is not null);' result_raw = snowflake_fetch_data(sql_query) if len(result_raw) > 0: # we have got some rows from snowflake result_count = int(result_raw[0]['COUNT(*)']) print("Check 9: count = " + str(result_count)) return result_count else: return False def check_10(id_list): # Step 10: Weekly data clean-up verification - MAIN.AE_MEMBERS_LOGINS sql_query = 'select count(*) from "DELPHI_EXPLORATION"."MAIN"."AE_MEMBERS_LOGINS" where ID in ( ' +id_list + ' ) and ID in (select MEMBER_ID from DELPHI_EXPLORATION.MAIN.AE_MEMBERS_GDPR_UNLISTED_MAPPING where MEMBER_ID is not null);' result_raw = snowflake_fetch_data(sql_query) if len(result_raw) > 0: # we have got some rows from snowflake result_count = int(result_raw[0]['COUNT(*)']) print("Check 10: count = " + str(result_count)) return result_count else: return False def check_11(id_list): # Step 11: Weekly data clean-up verification - ACTIVITIES_FEED_PERSONAL sql_query = 'select count(*) from "DELPHI_EXPLORATION"."MAIN"."AE_ACTIVITIES_FEED_PERSONAL" where MEMBER_ID in ( ' +id_list + ' ) and MEMBER_ID in (select MEMBER_ID from DELPHI_EXPLORATION.MAIN.AE_MEMBERS_GDPR_UNLISTED_MAPPING where MEMBER_ID is not null);' result_raw = snowflake_fetch_data(sql_query) if len(result_raw) > 0: # we have got some rows from snowflake result_count = int(result_raw[0]['COUNT(*)']) print("Check 11: count = " + str(result_count)) return result_count else: return False def check_12(id_list): # Step 12: Weekly data clean-up verification - MAIN.ACTIVITIES_FEED sql_query = 'select count(*) from "DELPHI_EXPLORATION"."MAIN"."AE_ACTIVITIES_FEED" where MEMBER_ID in ( ' +id_list + ' ) and MEMBER_ID in (select MEMBER_ID from DELPHI_EXPLORATION.MAIN.AE_MEMBERS_GDPR_UNLISTED_MAPPING where MEMBER_ID is not null);' result_raw = snowflake_fetch_data(sql_query) if len(result_raw) > 0: # we have got some rows from snowflake result_count = int(result_raw[0]['COUNT(*)']) print("Check 12: count = " + str(result_count)) return result_count else: return False def check_13(id_list): # Step 13: Weekly data clean-up verification - RAW.AE_MEMBERS_LOGINS sql_query = 'select count(*) from "DELPHI_EXPLORATION"."RAW"."AE_MEMBERS_LOGINS" where ID in ( ' +id_list + ' ) and ID in (select MEMBER_ID from DELPHI_EXPLORATION.MAIN.AE_MEMBERS_GDPR_UNLISTED_MAPPING where MEMBER_ID is not null);' result_raw = snowflake_fetch_data(sql_query) if len(result_raw) > 0: # we have got some rows from snowflake result_count = int(result_raw[0]['COUNT(*)']) print("Check 13: count = " + str(result_count)) return result_count else: return False def check_14(id_list): # Step 14: Weekly data clean-up verification - MEMBER_EXTENDED sql_query = 'select count(*) from "DELPHI_EXPLORATION"."RAW"."AE_MEMBER_EXTENDED" where ID in ( ' +id_list + ' ) and ID in (select MEMBER_ID from DELPHI_EXPLORATION.MAIN.AE_MEMBERS_GDPR_UNLISTED_MAPPING where MEMBER_ID is not null);' result_raw = snowflake_fetch_data(sql_query) if len(result_raw) > 0: # we have got some rows from snowflake result_count = int(result_raw[0]['COUNT(*)']) print("Check 14: count = " + str(result_count)) return result_count else: return False def check_15(id_list): # Step 15: Weekly data clean-up verification - RAW.ACTIVITIES_FEED sql_query = 'select count(*) from "DELPHI_EXPLORATION"."RAW"."AE_ACTIVITIES_FEED" where MEMBER_ID in ( ' +id_list + ' ) and MEMBER_ID in (select MEMBER_ID from DELPHI_EXPLORATION.MAIN.AE_MEMBERS_GDPR_UNLISTED_MAPPING where MEMBER_ID is not null);' result_raw = snowflake_fetch_data(sql_query) if len(result_raw) > 0: # we have got some rows from snowflake result_count = int(result_raw[0]['COUNT(*)']) print("Check 15: count = " + str(result_count)) return result_count else: return False ################# run steps ###################### def run_steps(id_list) -> dict: message_dict = {'name':"[PROD] Appreciation Engine GDPR (weekly thursday)",'status':"",'date':"", 'info':""} # set date according to sql searching date message_dict["date"] = (datetime.datetime.now() - datetime.timedelta(days=0)).strftime("%Y-%m-%d") failed_4 = False result_4 = check_4(id_list) if result_4 > 0: failed_4 = True failed_5 = False result_5 = check_5(id_list) if result_5 > 0: failed_5 = True failed_6 = False result_6 = check_6(id_list) if result_6 > 0: failed_6 = True failed_7 = False result_7 = check_7(id_list) if result_7 > 0: failed_7 = True failed_8 = False result_8 = check_8(id_list) if result_8 > 0: failed_8 = True failed_9 = False result_9 = check_9(id_list) if result_9 > 0: failed_9 = True failed_10 = False result_10 = check_10(id_list) if result_10 > 0: failed_10 = True failed_11 = False result_11 = check_11(id_list) if result_11 > 0: failed_11 = True failed_12 = False result_12 = check_12(id_list) if result_12 > 0: failed_12 = True failed_13 = False result_13 = check_13(id_list) if result_13 > 0: failed_13 = True failed_14 = False result_14 = check_14(id_list) if result_14 > 0: failed_14 = True failed_15 = False result_15 = check_15(id_list) if result_15 > 0: failed_15 = True full_log = {} if not failed_4 and not failed_5 and not failed_6 and not failed_7 and not failed_8 and not failed_9 and not failed_10 and not failed_11 and not failed_12 and not failed_13 and not failed_14 and not failed_15 : # All steps are successfull postgres_update_failure_count(failure_count_to_zero) message_dict['status'] = "OK" message_dict['info'] = "GDPR monitoring: all step completed successfully, no deviations found." else: # some step has failed postgres_update_failure_count(failure_count_increase) message_dict['status'] = "FAILED" message_dict['info'] = "GDPR monitoring: some steps have failed" if failed_4: message_dict['step_4_details'] = "Step 4 failed: Count = " + str(result_4) if failed_5: message_dict['step_5_details'] = "Step 5 failed: Count = " + str(result_5) if failed_6: message_dict['step_6_details'] = "Step 6 failed: Count = " + str(result_6) if failed_7: message_dict['step_7_details'] = "Step 7 failed: Count = " + str(result_7) if failed_8: message_dict['step_8_details'] = "Step 8 failed: Count = " + str(result_8) if failed_9: message_dict['step_9_details'] = "Step 9 failed: Count = " + str(result_9) if failed_10: message_dict['step_10_details'] = "Step 10 failed: Count = " + str(result_10) if failed_11: message_dict['step_11_details'] = "Step 11 failed: Count = " + str(result_11) if failed_12: message_dict['step_12_details'] = "Step 12 failed: Count = " + str(result_12) if failed_13: message_dict['step_13_details'] = "Step 13 failed: Count = " + str(result_13) if failed_14: message_dict['step_14_details'] = "Step 14 failed: Count = " + str(result_14) if failed_15: message_dict['step_15_details'] = "Step 15 failed: Count = " + str(result_15) full_log["Step 4 count"] = result_4 full_log["Step 5 count"] = result_5 full_log["Step 6 count"] = result_6 full_log["Step 7 count"] = result_7 full_log["Step 8 count"] = result_8 full_log["Step 9 count"] = result_9 full_log["Step 10 count"] = result_10 full_log["Step 11 count"] = result_11 full_log["Step 12 count"] = result_12 full_log["Step 13 count"] = result_13 full_log["Step 14 count"] = result_14 full_log["Step 15 count"] = result_15 message_dict['full_log'] = full_log return message_dict # Close all connections on exit def exit(): global sf_conn global pg_conn if sf_conn: sf_conn.close() if pg_conn: pg_conn.close() if __name__ == '__main__': # # check command line parameters and print or email result parser = create_arg_parser() arg_list = parser.parse_args() # finding the date of last friday day = datetime.date.today() oneday = datetime.timedelta(days=1) day -= oneday while day.weekday() != calendar.FRIDAY: day -= oneday last_friday = day.strftime("%F") last_friday_report = '/root/cron/GDPR/results/' + last_friday + '_step_3.txt' if len(sys.argv[1:]) > 0: try: with open(last_friday_report) as f: id_list = f.read() if len(id_list.strip()) == 0: print("Friday ID list is empty") result = {'status': "OK", 'info': "GDPR monitoring: friday ID list is empty"} else: result = run_steps(id_list) except: print("Could not find last friday results.") result = {'status': "FAILED", 'info': "GDPR monitoring: could not find last friday results"} with open('/tmp/GDPR_weekly_thursday_log.json', 'w') as json_file: json.dump(result, json_file, indent=4) if arg_list.recipients: recipients = arg_list.recipients[0] send_mail(recipients=recipients, message=json.dumps(result, indent=4), status=result['status']) if arg_list.print: print(json.dumps(result, indent=4)) else: parser.print_help() exit()