#! /usr/bin/env python3 import logging from workers import postgres_worker as pg_worker from workers import snowflake_worker as sf_worker from workers import slack_worker from workers import smtp_worker from sql import sql_queries import os import sys import datetime import argparse import json log_folder = "log" files_folder = "files" etl_sources_list = { 'spotify': {'days_offset': 2}, 'apple': {'days_offset': 2}, 'amazonprime': {'days_offset': 2}, 'amazonmusicunlimited': {'days_offset': 2}, 'amazonadsupported': {'days_offset': 2}, 'youtubereporting': {'days_offset': 2}, 'tiktokreporting': {'days_offset': 2} } ## working with script arguments def create_arg_parser(): parser = argparse.ArgumentParser() parser.add_argument('-d', '--date-shifter', nargs=1, type=int, default=0, dest='date_shifter', action="store", choices=[1, 2, 3, 4, 5]) parser.add_argument('-m', '--mail', action="store_const", const=True) parser.add_argument('-p', '--print', action="store_const", const=True) parser.add_argument('-s', '--slack', action="store_const", const=True) return parser def main(): def get_data(date_shifter: int = 0, path: str = files_folder) -> tuple: date_now = (datetime.datetime.now() - datetime.timedelta(days=date_shifter)).strftime("%Y-%m-%d") csv_folder = f"{path}/{date_now}" os.makedirs(csv_folder, exist_ok=True) message_dict = {} message_dict['name'] = "[PROD] Daily Data Alert Email" message_dict['status'] = "OK" message_dict['date'] = datetime.datetime.now().strftime("%Y-%m-%d") ############## 1. get the information about ETL processing for the day before yesterday. etl_current_date_delta = (datetime.datetime.now() - datetime.timedelta(days=date_shifter)).strftime("%Y-%m-%d") etl_report_log_issue_list = pg_worker.select("select distinct source,issue_date,status from audit_log where status != 'OK';") etl_report_log_source_list = {x['source'] for x in etl_report_log_issue_list} etl_date_set_tmp = set([etl_current_date_delta] + list({x['issue_date'] for x in etl_report_log_issue_list})) etl_date_set=sorted(etl_date_set_tmp) # create etl result template etl_data = {x: [] for x in etl_sources_list.keys()} # check etl for the dates from report_log db for date in etl_date_set: pg_worker.fetch_csv(date=date, query=sql_queries.etl_processing, to_file=f'{csv_folder}/etl_{date}.csv') etl_for_date = pg_worker.etl_processing(file=f'{csv_folder}/etl_{date}.csv', date=date, sources_list=etl_sources_list) for dsp in etl_data.keys(): if date == etl_current_date_delta: etl_data[dsp].append(etl_for_date[dsp]) if etl_for_date[dsp]['status'] == "OK": pg_worker.update(source=dsp, issue_date=etl_for_date[dsp]['date'], status=etl_for_date[dsp]['status']) else: pg_worker.insert(source=dsp, issue_date=etl_for_date[dsp]['date'], status=etl_for_date[dsp]['status']) else: if dsp in etl_report_log_source_list: etl_data[dsp].append(etl_for_date[dsp]) if etl_for_date[dsp]['status'] == "OK": pg_worker.update(source=dsp, issue_date=etl_for_date[dsp]['date'], status=etl_for_date[dsp]['status']) message_dict['ETL'] = etl_data ### Youtube Trending Charts message_dict['Youtube_Trending_charts'] = {} try: with open('/tmp/Youtube_Trending_charts_daily_log.json', 'r') as f: message_dict['Youtube_Trending_charts'] = json.load(f) except Exception as e: logging.error(e) ### Shazam Charts message_dict['Shazam_Charts'] = {} try: with open('/tmp/Shazam_Charts_daily_log.json', 'r') as f: message_dict['Shazam_Charts'] = json.load(f) except Exception as e: logging.error(e) ### Fivetran Facebook/Google Ads message_dict['Fivetran_Facebook_Google_Ads'] = {} try: with open('/tmp/Fivetran_Facebook_Google_Ads_daily_log.json', 'r') as f: message_dict['Fivetran_Facebook_Google_Ads'] = json.load(f) except Exception as e: logging.error(e) ### Appreciation Engine message_dict['Appreciation_Engine'] = {} try: with open('/tmp/Appreciation_Engine_daily_log.json', 'r') as f: message_dict['Appreciation_Engine'] = json.load(f) except Exception as e: logging.error(e) ### Linkfire message_dict['Linkfire'] = {} try: with open('/tmp/Linkfire_daily_log.json', 'r') as f: message_dict['Linkfire'] = json.load(f) except Exception as e: logging.error(e) ### Apple charts message_dict['Apple'] = {} try: with open('/tmp/Apple_daily_log.json', 'r') as f: message_dict['Apple'] = json.load(f) except Exception as e: logging.error(e) ### Spotify charts message_dict['Spotify'] = {} try: with open('/tmp/Spotify_daily_log.json', 'r') as f: message_dict['Spotify'] = json.load(f) except Exception as e: logging.error(e) ### Snowflake RTI check steps sf_result_dict = sf_worker.snowflake_fetch_data(sql_queries.snowflake_sql) sf_worker.dict_to_csv(f'{csv_folder}/Chartmetric.csv', sf_result_dict) artists_lags = sf_worker.check_days_lag(sf_result_dict) message_dict['Chartmetric'] = {'status': '', 'details': {}} if artists_lags: message_dict['Chartmetric']['status'] = "ISSUE" message_dict['Chartmetric']['message'] = "We observe lag for artist(s) available through RTI:" message_dict['Chartmetric']['details'] = artists_lags else: message_dict['Chartmetric']['status'] = "OK" message_dict['Chartmetric']['message'] = "No major lags." return message_dict, csv_folder, date_now # # check command line parameters and print or email result parser = create_arg_parser() arg_list = parser.parse_args() if len(sys.argv[1:]) > 0: day_shifter = 0 if arg_list.date_shifter: day_shifter = arg_list.date_shifter[0] message, csv_folder, date_now = get_data(date_shifter=day_shifter) # write down the json log file, just in case os.makedirs(log_folder, exist_ok=True) with open(f'{log_folder}/log_{date_now}.json', 'w') as json_file: json.dump(message, json_file, indent=4) with open('/tmp/daily_data_alert.json', 'w') as json_file: json.dump(message, json_file, indent=4) if arg_list.mail: html = smtp_worker.html_generate(message) smtp_worker.send_mail(message=html, path=csv_folder) if arg_list.slack: pretty_text = slack_worker.prettify_text(message) slack_worker.post_message(pretty_text) slack_worker.files_upload(path=csv_folder) if arg_list.print: print(json.dumps(message, indent=4)) else: parser.print_help() if __name__ == '__main__': main()