import concurrent.futures import json import multiprocessing import os import sys from concurrent.futures.thread import ThreadPoolExecutor from datetime import date, datetime, timedelta from time import sleep sys.path.insert(0, os.path.dirname(os.path.dirname(os.path.abspath(__file__)))) import grequests from gevent import monkey from sme_logger import get_logger from aws_utils import AwsUtils from const import APP_NAME, ENV, aws, chartmetric from endpoints.refreshToken import ChartmetricTokenRefresh from utility import Utilities, measure monkey.patch_all() process_type = "Fan Metrics" last_run = int(Utilities().get_last_run(process_type)) parent_id = last_run start_time = datetime.today().strftime("%Y-%m-%d %H:%M:%S") today = datetime.today().strftime("%Y-%m-%d") Utilities().insert_performance_logs_fan_metrics(process_type, last_run + 1, start_time, parent_id + 1) # - - - - Variables for Performance Logs - - - - # yesterday = date.today() - timedelta(days=chartmetric.get("TIME_DELTA", None)) logger = get_logger(APP_NAME, os.environ.get("ENVIRONMENT", ENV).lower() != ENV) logger.debug(f"Running FanMetrics in Environment :{ENV}") logger.debug("Bucket used : {0}".format( aws.get("s3", None).get("BUCKET_QUARANTINE", None))) class Properties(object): """ is_retrying: flag to check if the fetching from chartmetric is a new day or a retry attempt artists: list of artists based on if its a new or retry attempt data_sources: Depend on the type of fetch retry - get sources from postgres, new - get sources from const """ is_retrying = False # Set the flag to check if its a retry attempt if Utilities().get_last_processed_date() is not None and Utilities( ).get_last_processed_date() == str(date.today()): logger.info(f"Attempting retry for date: {str(date.today())}") is_retrying = True artists = Utilities().get_artists_from_reproc() if not is_retrying: logger.info(f"Attempting new update for date: {str(date.today())}") artists = Utilities().get_artists_from_postgres() data_source = chartmetric.get("DATA_SOURCES", None) # Default data sources from const class StatusLog(object): log = {} class FanMetrics(object): """ status_log: Logging in responses with status codes != 200 refresh_chart_metric_token: Need to refresh token every 1 hour as per Chartmetric artists: List of artists fetched from PostgreSql urls: For every artist ID every URL is generated by appending the ID current_cm_artist: Current Chartmetric ID required to append with Chartmetric response current_sony_artist: Current GRASS ID required to append with Chartmetric response resp_code: Response code for each request to chartmetric """ def __init__(self): self.aws_client = AwsUtils().create_aws_client( service_name="firehose", region_name=aws.get("AWS_DEFAULT_REGION", None)) self.token = ChartmetricTokenRefresh().refresh_chartmetric_token() self.today = str(datetime.today().strftime("%Y-%m-%d")) self.yesterday = str(yesterday.strftime("%Y-%m-%d")) self.headers = {"Authorization": f"Bearer {self.token}"} self.resp_code = "" self.rs = None # - - - - COUNT VARIABLES - - - - # self.TOTAL_PROCESSED_DATA = 0 self.TOTAL_EMPTY_DATA_RECEIVED = 0 self.TOTAL_DATA_NOT_UPDATED_RECEIEVD = 0 self.TOTAL_INVALID_RESPONSES_RECEIVED = 0 self.TOTAL_SUCCESSFUL_UPDATED_DATA_RECEIVED = 0 self.TOTAL_INVALID_REQUESTS_MADE = 0 self.last_token_refresh = datetime.now() self.aws_object = AwsUtils() # - - - - Variables for Performance Logs - - - - # def get_metrics(self, artist_list: list, thread_id: int): artists = artist_list def get_data(): """ urls: Will have all url's for every artist for all the sources we need to fetch data current_cm_artist: Notes the current Chartmetric ID we are fetching data for current_sony_artist: Notes the current Gras ID we are mapping the data with final_data: The Valid data received will be mapped in and sent to Firehose immediately """ urls = [] current_cm_artist = "" current_sony_artist = "" final_data = {} # loop through the artists fetched from Postgres for artist in artists: if not Properties.is_retrying: data_source = Properties.data_source else: data_source = artist[ 3] # Consider source list from Postgres if retrying current_cm_artist = artist[1] current_sony_artist = artist[0] StatusLog.log[f"{current_sony_artist}"] = {} sources = {} # Need to loop through the sources to map each artist data with respective source for source in data_source: if source != 'youtube': self.TOTAL_PROCESSED_DATA += 1 url = (chartmetric.get("urls", None).get( "CM_FAN_METRICS_URL", None).format(artist[1], source, self.yesterday, self.today)) urls.append( url ) # Urls for all the sources for each individual artist sources[f"{source}"] = {} StatusLog.log[f"{current_sony_artist}"] = sources def fetch_data(): processes = [] # """ # Status log has following important keys: # 1. Response Success: 1, Failure: 0 # 2. Request Success: 1, Failure: 0 # 3. Valid Data Yes: 1, No: 0 # """ try: # Send in a list of all the sources for one artist at once self.rs = (grequests.get(u, headers=self.headers) for u in urls) responses = grequests.map(self.rs) for chartmetric_response, chartmetric_source in zip( responses, data_source): def validate_date(): # Edge case to check if the response code was anything other than 200 if (chartmetric_response is None or chartmetric_response.status_code != 200): StatusLog.log[f"{current_sony_artist}"][ chartmetric_source] = { "chartmetric_id": f"{current_cm_artist}", "update_date": f"{self.today}", "valid_data": 0, "response": 0, "request": 1, "response_code": chartmetric_response.status_code, } logger.warning( f"Invalid response received: " f"{{'chartmetric_id':{current_cm_artist}," f"'GRASS_ID': {current_sony_artist}," f"'source: {chartmetric_source}," f"'response_code': {chartmetric_response.status_code}}}" ) self.TOTAL_INVALID_RESPONSES_RECEIVED += 1 # If the response code is 200 else: self.resp_code = str( chartmetric_response.status_code) response_dict = json.loads( chartmetric_response.content) # Object not in response, which means empty data if "obj" not in response_dict: StatusLog.log[ f"{current_sony_artist}"][ chartmetric_source] = { "chartmetric_id": f"{current_cm_artist}", "update_date": f"{self.today}", "valid_data": 0, "response": 1, "request": 1, "response_code": chartmetric_response. status_code, } logger.warning( f"Empty Data Received: " f"{{'chartmetric_id': {current_cm_artist}," f"'GRASS_ID': {current_sony_artist}," f"'source: {chartmetric_source}," f"'response_code': {chartmetric_response.status_code}}}" ) self.TOTAL_EMPTY_DATA_RECEIVED += 1 # Object is present but there is no data, in short empty data if len(response_dict["obj"]) == 0: StatusLog.log[ f"{current_sony_artist}"][ chartmetric_source] = { "chartmetric_id": f"{current_cm_artist}", "update_date": f"{self.today}", "valid_data": 0, "response": 1, "request": 1, "response_code": chartmetric_response. status_code, } logger.warning( f"Empty Data Received: " f"{{'chartmetric_id': {current_cm_artist}," f"'GRASS_ID': {current_sony_artist}," f"'source: {chartmetric_source}," f"'response_code': {chartmetric_response.status_code}}}" ) self.TOTAL_EMPTY_DATA_RECEIVED += 1 # Object is present needs validation checks else: # Eliminating irrelevant urls if "link" in response_dict["obj"].keys( ): response_dict["obj"].pop("link") if "url" in response_dict["obj"].keys( ): response_dict["obj"].pop("url") # Loop through the metrics in responses for data flattening for metric_key in response_dict[ "obj"].keys(): def process_metrics(): # Edge case if the metric is required by us if (metric_key not in chartmetric.get( "DATA_SOURCES_METRICS" )[chartmetric_source]): logger.debug( f"Invalid Metric Found: " f"{{'chartmetric_id': {current_cm_artist}," f"'GRASS_ID': {current_sony_artist}," f"'source: {chartmetric_source}," f"metric: {metric_key}," f"'response_code': {chartmetric_response.status_code}}}" ) else: # Edge case: Nested metric has no data in it if (len(response_dict[ "obj"][metric_key]) == 0 or len(response_dict[ "obj"] [metric_key] [-1]) == 0): StatusLog.log[f"{current_sony_artist}"][ chartmetric_source] = { "chartmetric_id": f"{current_cm_artist}", "update_date": f"{self.today}", "valid_data": 0, "response": 1, "request": 1, "response_code": chartmetric_response .status_code, } logger.warning( f"Empty Data Received: " f"{{'chartmetric_id':{current_cm_artist}," f"'GRASS_ID':{current_sony_artist}," f"'source:{chartmetric_source}," f"'response_code':{chartmetric_response.status_code}}}" ) self.TOTAL_EMPTY_DATA_RECEIVED += ( 1) # Edge Case: Nested metric has data in it, validate if data is correct else: # Timestamps by chartmetric are weird, normalize them if ("Z" in response_dict[ "obj"] [metric_key][-1] ["timestp"]): changed_date_format = response_dict[ "obj"][metric_key][ -1]["timestp"].split( "T")[0] response_dict["obj"][ metric_key][-1][ "timestp"] = changed_date_format else: changed_date_format = datetime.strptime( response_dict[ "obj"] [metric_key] [-1] ["timestp"], "%a %b %d %Y", ).strftime( "%Y-%m-%d") response_dict["obj"][ metric_key][-1][ "timestp"] = changed_date_format # Chartmetric sends the most recent updated data, check if it matches todays date if (changed_date_format == self.today): final_data.update({ "dsp": f"{chartmetric_source}", "gras_id": current_sony_artist, "chartmetric_id": current_cm_artist, "update_date": f"{self.today}", "metrics": f"{metric_key}", "value": response_dict[ "obj"] [metric_key] [-1]["value"], "timestp": f"{response_dict['obj'][metric_key][-1]['timestp']}", }) if ("interpolated" in response_dict[ "obj"] [metric_key] [-1].keys()): final_data.update({ "interpolated": f"{response_dict['obj'][metric_key][-1]['interpolated']}" }) else: logger.debug( f"Interpolation set to default value false for: {current_cm_artist, chartmetric_source, metric_key}" ) final_data.update({ "interpolated": "False" }) resp = self.aws_object.deliver_to_firehose( client=self. aws_client, delivery_stream_name =aws.get( "AWS_FIREHOSE_CHARTMETRIC_DATA_STREAM", None, ), data=final_data, ) StatusLog.log[f"{current_sony_artist}"][ chartmetric_source] = { "chartmetric_id": f"{current_cm_artist}", "update_date": f"{self.today}", "valid_data": 1, "response": 1, "request": 1, "response_code": chartmetric_response . status_code, } logger.debug( f"Valid Data Received: " f"{{'chartmetric_id':{current_cm_artist}," f"'GRASS_ID':{current_sony_artist}," f"'source:{chartmetric_source}," f"'response_code':{chartmetric_response.status_code}," f"'last_updated_by_chartmetric':{changed_date_format}}}" ) self.TOTAL_SUCCESSFUL_UPDATED_DATA_RECEIVED += ( 1) logger.debug( f"Final Data: {final_data}" ) final_data.clear() # Data was received but the timestamp doesnt match todays date else: StatusLog.log[f"{current_sony_artist}"][ chartmetric_source] = { "chartmetric_id": f"{current_cm_artist}", "update_date": f"{self.today}", "valid_data": 0, "response": 1, "request": 1, "response_code": chartmetric_response . status_code, } logger.warning( f"Old Data Received: " f"{{'chartmetric_id':{current_cm_artist}," f"'GRASS_ID':{current_sony_artist}," f"'source:{chartmetric_source}," f"'response_code':{chartmetric_response.status_code}," f"'last_updated_by_chartmetric':{changed_date_format}}}" ) self.TOTAL_DATA_NOT_UPDATED_RECEIEVD += ( 1) with concurrent.futures.ProcessPoolExecutor( max_workers=2) as executor: executor.map(process_metrics()) with concurrent.futures.ProcessPoolExecutor( max_workers=2) as executor: executor.map(validate_date()) except Exception as e: self.TOTAL_INVALID_REQUESTS_MADE += 1 logger.error(f"Fan Metrics Exception: " f"{{'chartmetric_id':{current_cm_artist}," f"'GRASS_ID':{current_sony_artist}," f"{e}}}") fetch_data() urls.clear() final_data.clear() get_data() if (datetime.now() - self.last_token_refresh).seconds > int( chartmetric.get("TOKEN_REFRESH_DELTA", None)): self.last_token_refresh = datetime.now() self.token = ChartmetricTokenRefresh().refresh_chartmetric_token() self.headers = {"Authorization": f"Bearer {self.token}"} sleep(chartmetric.get("REFRESH_SLEEP_TIME", None)) @measure def main(self): num_cores = multiprocessing.cpu_count() logger.info(f"No. of Cores used:{num_cores}") pool = ThreadPoolExecutor(num_cores * 4) try: for i in range(len(Properties.artists)): pool.submit(self.get_metrics, Properties.artists[i:i + 1], i) except Exception as e: logger.error(e) pool.shutdown() def fetch_data(self): try: self.main() except Exception as e: logger.error(f"Exception:{e}") finally: # ------------- WRITE LOGS TO POSTGRES ---------------------- # Utilities().logs_to_postgres(StatusLog.log, Properties.is_retrying) logger.debug("Logs written to postgres") # ------------- WRITE LOGS TO S3---------------------- # AwsUtils().save_json_data_to_s3( bucket_name=aws.get("s3", None).get("BUCKET_LOGS", None), path="chartmetric", file_name_to_save=f"status_report_{datetime.now()}.json", data=StatusLog.log, ) logger.info( f"Status written to : {aws.get('s3', None).get('BUCKET_LOGS', None)}" ) end_time = datetime.today().strftime("%Y-%m-%d %H:%M:%S") Utilities().update_performance_logs_fan_metrics( end_time, start_time, f"Total processed requests: {self.TOTAL_PROCESSED_DATA}, Empty metrics: {self.TOTAL_EMPTY_DATA_RECEIVED}, Not updated metrics: {self.TOTAL_DATA_NOT_UPDATED_RECEIEVD}, " f"Invalid Responses: {self.TOTAL_INVALID_RESPONSES_RECEIVED}, valid metrics received: {self.TOTAL_SUCCESSFUL_UPDATED_DATA_RECEIVED}, " f"Invalid Requests: {self.TOTAL_INVALID_REQUESTS_MADE} ", process_type, f"{last_run + 1}", ) time_delta_format = "%Y-%m-%d %H:%M:%S" time_delta = datetime.strptime( end_time, time_delta_format) - datetime.strptime( start_time, time_delta_format) logger.info( f"{process_type}: Start time: {start_time}, End Time: {end_time}, Total Execution Time: {time_delta}" ) logger.info( f"Total processed requests: {self.TOTAL_PROCESSED_DATA}, Empty metrics: {self.TOTAL_EMPTY_DATA_RECEIVED}, " f"Not updated metrics: {self.TOTAL_DATA_NOT_UPDATED_RECEIEVD}, Invalid Responses: {self.TOTAL_INVALID_RESPONSES_RECEIVED}, " f"valid metrics received: {self.TOTAL_SUCCESSFUL_UPDATED_DATA_RECEIVED}, Invalid Requests: {self.TOTAL_INVALID_REQUESTS_MADE}" ) sys.exit(0) if __name__ == "__main__": FanMetrics().fetch_data()