#! /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 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'] 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 = 'tktkexp';" failure_count_to_zero = "UPDATE failed_count SET failed_count = 0, last = 'OK', last_updated = NOW() WHERE count_id = 'tktkexp';" 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] TikTok Exploration" 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 check_3_to_4() -> dict: result_dict = {} full_log = {} ### Check 3 RAW -> MAIN DIM_TIKTOK_TOP_SONG (expected result: Counts match inside every step) ################################################ tmp_sf_check_3 = snowflake_fetch_data(sql_queries.sf_check_3) sf_check_3_src = int(tmp_sf_check_3[0]['count']) sf_check_3_dest = int(tmp_sf_check_3[1]['count']) print(f"SRC_CHECK_3: - {sf_check_3_src}") print(f"DEST_CHECK_3: - {sf_check_3_dest}") full_log["CHECK_3"] = {'SRC_CHECK_3': sf_check_3_src, 'DEST_CHECK_3': sf_check_3_dest} if sf_check_3_src != sf_check_3_dest: result_dict["CHECK_3"] = full_log["CHECK_3"] ### Check 4 Check Inbound file status table (For every records in the results: there are no NULL values in src, destination_raw or destination_main; src = destination_raw, destination_main)################################################ tmp_sf_check_4 = snowflake_fetch_data(sql_queries.sf_check_4) print(f"CHECK_4: - {tmp_sf_check_4}") #Check if tmp_sf_check_4: full_log["CHECK_4"] = {} temp_result_dict = {} temp_result_dict["CHECK_4"] = {} # result_dict["CHECK_4"] = {} for item in tmp_sf_check_4: date = item['date'].strftime("%Y-%m-%d") cs_id = item['cs_id'] src = item['src'] dest_raw = item['dest_raw'] dest_main = item['dest_main'] full_log["CHECK_4"][cs_id] = {'date': date, 'cs_id': cs_id, 'src': src, 'dest_raw': dest_raw, 'dest_main': dest_main} print(f"'date': {date}, 'cs_id': {cs_id}, 'src': {src}, 'dest_raw': {dest_raw}, 'dest_main': {dest_main}") if src != dest_raw or src != dest_main: temp_result_dict["CHECK_4"][cs_id] = {'date': date, 'cs_id': cs_id, 'src': src, 'dest_raw': dest_raw, 'dest_main': dest_main} if temp_result_dict["CHECK_4"]: result_dict["CHECK_4"] = temp_result_dict["CHECK_4"] return result_dict, full_log ################# run the all steps ###################### def run_steps() -> dict: message_dict = {'name':"[PROD] TikTok Exploration",'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_3_to_4, monitoring_3_to_4_log = check_3_to_4() message_dict['full_log'] = {**monitoring_3_to_4_log} message_dict['details'] = {**result_dict_3_to_4} if result_dict_3_to_4: postgres_update_failure_count(failure_count_increase) message_dict['status'] = "FAILED" message_dict['info'] = "Some steps failed" else: postgres_update_failure_count(failure_count_to_zero) message_dict['status'] = "OK" message_dict['info'] = "All steps executed without errors" 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/TikTok_Exploration_daily_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()