#! /usr/bin/env python3 from email.mime.multipart import MIMEMultipart from email.mime.text import MIMEText from contextlib import closing from sql import sql_queries from datetime import date, timedelta import snowflake.connector import psycopg2 import smtplib import argparse import sys import json import datetime 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'] db_name_slz = config['postgres_slz']["database"] db_user_slz = config['postgres_slz']["user"] db_password_slz = config['postgres_slz']["password"] db_host_slz = config['postgres_slz']["host"] db_port_slz = config['postgres_slz']["port"] 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 = 'bw_hourly';" failure_count_to_zero = "UPDATE failed_count SET failed_count = 0, last = 'OK', last_updated = NOW() WHERE count_id = 'bw_hourly';" 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] Brandwatch (hourly)" 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: ctx = snowflake.connector.connect( user=sf_user, password=sf_password, account=sf_account, warehouse=sf_warehouse ) cs = ctx.cursor(snowflake.connector.DictCursor) try: cs.execute(sql_query) result_dict = cs.fetchall() finally: cs.close() ctx.close() return result_dict def postgres_fetch_data_slz(sql_query) -> list: result_list = [] with closing(psycopg2.connect(database=db_name_slz, user=db_user_slz, password=db_password_slz, host=db_host_slz, port=db_port_slz)) as connection: with connection.cursor() as cursor: cursor.execute(sql_query) columns = cursor.description rows = cursor.fetchall() for row in rows: tmp = {} for i in range(len(columns)): tmp[columns[i][0]] = row[i] result_list.append(tmp) return result_list def check_6_to_8() -> dict: result_dict = {} full_log = {} ### Check 6 SLZ status: DAILY artist_ingestion is currently running successfully (daily ingestion for today is currently in progress) (expected result: “LOGS_CREATED” in every row >0) ################################################ tmp_pg_check_6 = postgres_fetch_data_slz(sql_queries.pg_check_6) print(f"CHECK_6: - {tmp_pg_check_6}") #Check if tmp_pg_check_6: full_log["CHECK_6"] = {} result_dict_tmp = {} result_dict_tmp["CHECK_6"] = {} for item in tmp_pg_check_6: full_log["CHECK_6"].update({item['REPORT_DATE'].strftime("%Y-%m-%d"): item['LOGS_CREATED']}) if item['LOGS_CREATED'] == 0: result_dict["CHECK_6"] = {} result_dict_tmp["CHECK_6"].update({item['REPORT_DATE'].strftime("%Y-%m-%d"): item['LOGS_CREATED']}) print(f"{item['REPORT_DATE']} - {item['LOGS_CREATED']}") result_dict["CHECK_6"] = result_dict_tmp["CHECK_6"] else: full_log["CHECK_6"] = {'OK': 'Query returned 0 records'} ### Check 6.1 SLZ status: query log failed tmp_pg_check_6_1 = postgres_fetch_data_slz(sql_queries.pg_check_6_1) print(f"CHECK_6.1: - {tmp_pg_check_6_1}") #Check if tmp_pg_check_6_1: full_log["CHECK_6.1"] = {} result_dict_tmp = {} result_dict_tmp["CHECK_6.1"] = {} for item in tmp_pg_check_6_1: full_log["CHECK_6.1"].update({item['REPORT_DATE'].strftime("%Y-%m-%d"): item['LOGS_CREATED']}) if item['LOGS_CREATED'] != 0: result_dict["CHECK_6.1"] = {} result_dict_tmp["CHECK_6.1"].update({item['REPORT_DATE'].strftime("%Y-%m-%d"): item['LOGS_CREATED']}) print(f"{item['REPORT_DATE']} - {item['LOGS_CREATED']}") result_dict["CHECK_6.1"] = result_dict_tmp["CHECK_6.1"] else: full_log["CHECK_6.1"] = {'OK': 'Query returned 0 records'} ### Check 7 SLZ status: BACKFILL artist_ingestion is successful (expected result: “LOGS_CREATED” in every row >0) ################################################ tmp_pg_check_7 = postgres_fetch_data_slz(sql_queries.pg_check_7) print(f"CHECK_7: - {tmp_pg_check_7}") #Check if tmp_pg_check_7: full_log["CHECK_7"] = {} result_dict_tmp = {} result_dict_tmp["CHECK_7"] = {} for item in tmp_pg_check_7: full_log["CHECK_7"].update({item['REPORT_DATE'].strftime("%Y-%m-%d"): item['LOGS_CREATED']}) if item['LOGS_CREATED'] == 0: result_dict["CHECK_7"] = {} result_dict_tmp["CHECK_7"].update({item['REPORT_DATE'].strftime("%Y-%m-%d"): item['LOGS_CREATED']}) print(f"{item['REPORT_DATE']} - {item['LOGS_CREATED']}") result_dict["CHECK_7"] = result_dict_tmp["CHECK_7"] else: full_log["CHECK_7"] = {'OK': 'Query returned 0 records'} ### Check 8 SNOWFLAKE status - S3 → RAW (expected result: 0 results) ############################################# tmp_sf_check_8 = snowflake_fetch_data(sql_queries.sf_check_8) print(f"CHECK_8: {tmp_sf_check_8}") if tmp_sf_check_8: result_dict["CHECK_8"] = {'FAILED': 'Query does not returned 0 results'} full_log["CHECK_8"] = result_dict["CHECK_8"] else: full_log["CHECK_8"] = {'OK': 'Query returned 0 results'} return result_dict, full_log ################# run the all steps ###################### def run_steps() -> dict: message_dict = {'name':"[PROD] Brandwatch (hourly)",'status':"",'date':"", 'info':"", 'details':"", 'full_log':""} # set date according to sql searching date, currently it's `today` message_dict["date"] = datetime.datetime.now().strftime("%Y-%m-%d") result_dict_6_to_8, monitoring_6_to_8_log = check_6_to_8() message_dict['full_log'] = {**monitoring_6_to_8_log} message_dict['details'] = {**result_dict_6_to_8} if result_dict_6_to_8: postgres_update_failure_count(failure_count_increase) message_dict['status'] = "FAILED" message_dict['info'] = "Ingestion issues" else: postgres_update_failure_count(failure_count_to_zero) message_dict['status'] = "OK" message_dict['info'] = "Ingestion in progress" return message_dict if __name__ == '__main__': # # check command line parameters and print or email result parser = create_arg_parser() arg_list = parser.parse_args() if len(sys.argv[1:]) > 0: result = run_steps() with open('/tmp/brandwatch_hourly_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()