import json import os import struct from datetime import datetime import boto3 import botocore import psycopg2 from botocore.exceptions import ClientError from google.cloud import bigtable from sme_logger import get_logger from aws_utils import AwsUtils from const import APP_NAME, ENV, aws, chartmetric, gcp from utility import Utilities # - - - - Variables for Performance Logs - - - - # process_type = "S3 - BIG TABLE WRITE" last_run = Utilities().get_last_run(process_type) parent_id = Utilities().get_last_run("Fan Metrics") start_time = datetime.today().strftime("%Y-%m-%d %H:%M:%S") today = datetime.today().strftime("%Y-%m-%d") # - - - - Variables for Performance Logs - - - - # logger = get_logger(APP_NAME, os.environ.get("ENVIRONMENT", ENV).lower() != ENV) class BigTable(object): def __init__(self): self.column_family_meta = "meta" self.column_family_metrics = "metrics" self.rows = [] self.s3 = boto3.client("s3") self.db = AwsUtils().connect_to_database() self.cur = self.db.cursor() self.default_timestamp = datetime(1970, 1, 1) self.list_files_processed = [] # function to write the data to Big Table by passing PROJECT_ID, INSTANCE, TABLE parameter def write_to_bigtable(self, project_id, instance, table): try: client = bigtable.Client(project=project_id, admin=True) # Create a google client if ENV == "stage": instance = client.instance("stg-delphi-consumer-analytics-hdd") elif ENV == "prod": instance = client.instance("delphi-consumer-analytics-hdd") else: instance = client.instance(instance) table = instance.table(table) except Exception as e: logger.error("Exception: {0}".format(e)) query_res = None # - - - - - - Try Block to fetch file processing data from Postgres - - - - - - # try: self.cur.execute( aws.get("RDS", None).get("queries", None).get("GET_FILE_PROCESSING_INFO", None)) query_res = self.cur.fetchall() logger.info("Filenames fetched from postgres") except psycopg2.OperationalError as e: logger.error(f"Error getting file processing info: {e}") if query_res: logger.debug(query_res) for qry in query_res: file_name = qry[0] report_date = qry[1] file_path = ( f"chartmetric/{chartmetric.get('CHARTMETRIC_VERSION', None)}/report_date={report_date}/" f"report_licensor={chartmetric.get('REPORT_LICENSOR', None)}/{file_name}" ) logger.debug("Report Date is: {0}".format(report_date)) logger.debug("Currently working on: {0}".format(file_name)) # - - - - - - Try Block to read files from AWS S3 bucket - - - - - - # try: obj = self.s3.get_object( Bucket=aws.get("s3", None).get("BUCKET_DECOMPRESSED", None), Key=file_path, ) logger.info( f"{file_name} was found in {aws.get('s3', None).get('BUCKET_DECOMPRESSED', None)}" ) # - - - - - Try block to process if the file was found in Bucket - - - - - - # try: for line in obj["Body"]._raw_stream: body = json.loads(line) # Check if the dsp body is of type dictionary if (isinstance(body, dict) # and body.get("interpolated") != str(True) #For blocking interpolated True data ): try: gras_id = body.get("gras_id") source_dsp = body.get("dsp") metrics_col = body.get("metrics") metrics_value = struct.pack( gcp.get("BIGTABLE_INT_BYTE_FORMAT", None), body.get("value"), ) if source_dsp in chartmetric.get( "DATA_SOURCES_METRICS", None): if (metrics_col in chartmetric.get( "DATA_SOURCES_METRICS", None)[source_dsp]): row_key = f"artist_daily~GRAS_{gras_id}~{report_date}~{source_dsp}" logger.debug( f"Populated RowKey, other columns Data from file_name {file_name} from S3" ) row = table.direct_row(row_key) # Set the column values and Set the Timestamp to 0 row.set_cell( self.column_family_meta, "report_date", str(report_date), timestamp=self. default_timestamp, ) row.set_cell( self.column_family_meta, "artist_id", "GRAS_" + str(gras_id), timestamp=self. default_timestamp, ) row.set_cell( self.column_family_meta, "dsp", source_dsp, timestamp=self. default_timestamp, ) row.set_cell( self.column_family_metrics, metrics_col, metrics_value, timestamp=self. default_timestamp, ) # appends the individual row to rows list self.rows.append(row) else: logger.warning( f"No valid metrics for: {metrics_col}" ) else: logger.warning( f"No valid data source: {source_dsp}" ) except Exception as e: logger.error( f"Error Writing to Big Table: {body}: {e}" ) if self.rows: logger.debug(f"Writing data into BigTable {table}") table.mutate_rows(self.rows) logger.info( f"Data is successfully written to BigTable {table}" ) self.rows.clear() try: self.cur.execute( aws.get("RDS", None).get("queries", None).get( "UPDATE_FILE_PROCESSING", None).format( datetime.today().strftime( "%Y-%m-%d %H:%M:%S"), file_name, )) self.db.commit() logger.info( f"file {file_name} successfully updated in Postgres Table" ) except psycopg2.OperationalError as e: logger.error( f"Error updating file processing table: {e}") except Exception as e: logger.warning( f"No valid data for Date: {report_date}{e}") except botocore.exceptions.ClientError as e: if e.response["Error"]["Code"] == "404": logger.error( f"{file_name} was not found {aws.get('s3', None).get('BUCKET_DECOMPRESSED', None)}" ) else: logger.error("Exception occurred: {0}".format( e.response["Error"])) else: logger.info("No files to Process") def run_s3_to_big_table(self): Utilities().set_google_credentials(gcp.get("GOOGLE_SECRET_NAME", None)) self.write_to_bigtable( gcp.get("PROJECT_ID", None), gcp.get("INSTANCE_ID", None), gcp.get("TABLE", None), ) if __name__ == "__main__": BigTable().run_s3_to_big_table() end_time = datetime.today().strftime("%Y-%m-%d %H:%M:%S") Utilities().insert_performance_logs(process_type, f"{last_run} + 1", start_time, end_time, parent_id) 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}" )