sync_amz_data/sync_amz_data/DataTransform/Data_ETL.py

from sync_amz_data.public.amz_ad_client_02 import SPClient,Account,SBClient,SDClient
import pandas as pd
import numpy as np
from dateutil.parser import parse
pd.set_option('display.max_columns', None)
import warnings
warnings.filterwarnings('ignore')
pd.set_option('expand_frame_repr', False)

class Common_ETLMethod:
    def columnsName_modify(self,df):
        df.columns = [i.replace(".","_").lower() for i in df.columns]
        return df

    def time_stamp_convert(self,df,time_columns:list):
        for time_column in time_columns:
            df[time_column] = pd.to_datetime(df[time_column]*1000000).map(lambda x: x.strftime("%Y-%m-%d %H:%M:%S"))
        df[time_columns] = df[time_columns].astype("datetime64")
        return df

    def TZ_Deal(self,df, time_columns):
        for time_column in time_columns:
            df[time_column] = df[time_column].map(lambda x: parse(x).strftime("%Y-%m-%d %H:%M:%S"))
        df[time_columns] = df[time_columns].astype("datetime64")
        return df

    def placement_segmentsplit(self,df,segment):
        df[segment] = df[segment].astype("string")
        df[segment+str("_percentage")] = df[segment].str.extract("'percentage':.+([\d\.]{1,}),").astype('float32')
        df[segment+str("_placement")] = df[segment].str.extract("'placement':.+'(.+)'")
        df.replace(['nan','Nan','NaN'],np.nan,inplace=True)
        df.drop(columns=[segment],inplace=True)
        return df

    def expression_split(self,df,segment):
        df[segment] = df[segment].astype("string")
        df[segment+str("type")] = df[segment].str.extract("'type':.+'(.+)',")
        df[segment+str("value")] = df[segment].str.extract("'value':.+'(.+)'")
        df.replace(['nan','Nan','NaN'],np.nan,inplace=True)
        df.drop(columns=[segment],inplace=True)
        return df

class Acount_ETL(Account,Common_ETLMethod):
    def portfolio_ETL(self):
        list_portfolio = self.get_portfolio()
        df_portfolio = pd.json_normalize(list_portfolio)
        print(self.columnsName_modify(df_portfolio))
        return self.columnsName_modify(df_portfolio)

class SP_ETL(SPClient,Common_ETLMethod):
    def campaigns_ETL(self):
        list_campaign_SP = list(self.iter_campaigns(**{"includeExtendedDataFields":True}))
        df_campaign = pd.json_normalize(list_campaign_SP)
        df_campaign = self.placement_segmentsplit(df_campaign, "dynamicBidding.placementBidding")
        df_campaign = self.TZ_Deal(df_campaign,["extendedData.creationDateTime","extendedData.lastUpdateDateTime"])
        # print(df_campaign)
        return self.columnsName_modify(df_campaign)

    def adGroup_ETL(self):
        list_adGroup_SP = list(self.iter_adGroups(**{"includeExtendedDataFields":True}))
        df_adGroup_SP = pd.json_normalize(list_adGroup_SP)
        df_adGroup_SP = self.TZ_Deal(df_adGroup_SP,["extendedData.creationDateTime","extendedData.lastUpdateDateTime"])
        return self.columnsName_modify(df_adGroup_SP)

    def ads_ETL(self):
        list_adId_SP = list(self.iter_ads(**{"includeExtendedDataFields":True}))
        df_adId_SP = pd.json_normalize(list_adId_SP)
        df_adId_SP = self.TZ_Deal(df_adId_SP,["extendedData.creationDateTime", "extendedData.lastUpdateDateTime"])
        return self.columnsName_modify(df_adId_SP)

    def keywords_ETL(self):
        list_keywords_SP = list(self.iter_keywords(**{"includeExtendedDataFields":True}))
        df_keywords_SP = pd.json_normalize(list_keywords_SP)
        df_keywords_SP = self.TZ_Deal(df_keywords_SP, ["extendedData.creationDateTime", "extendedData.lastUpdateDateTime"])
        return self.columnsName_modify(df_keywords_SP)

    def targets_ETL(self):
        list_targets = list(self.iter_targets())
        df_targets = pd.json_normalize(list_targets)
        df_targets = self.TZ_Deal(df_targets, ["extendedData.creationDateTime", "extendedData.lastUpdateDateTime"])
        return self.columnsName_modify(df_targets)

    def budget_ETL(self,campaign_ids:list):
        list_budget = self.get_budget(campaign_ids = campaign_ids)['success']
        df_budget = pd.json_normalize(list_budget)
        df_budget = self.TZ_Deal(df_budget,["usageUpdatedTimestamp"])
        print(df_budget)

class SB_ETL(SBClient,Common_ETLMethod):
    def campaigns_ETL(self):
        list_campaign_SB = list(self.iter_campaigns(**{"includeExtendedDataFields":True}))
        df_campaign = pd.json_normalize(list_campaign_SB)
        df_campaign = self.placement_segmentsplit(df_campaign, "bidding.bidAdjustmentsByPlacement")
        df_campaign = self.time_stamp_convert(df_campaign,["extendedData.creationDate","extendedData.lastUpdateDate"])
        # print(df_campaign)
        return self.columnsName_modify(df_campaign)

    def adGroup_ETL(self):
        list_adGroup_SB = list(self.iter_adGroups(**{"includeExtendedDataFields":True}))
        df_adGroup_SP = pd.json_normalize(list_adGroup_SB)
        df_adGroup_SP = self.time_stamp_convert(df_adGroup_SP,["extendedData.creationDate","extendedData.lastUpdateDate"])
        return self.columnsName_modify(df_adGroup_SP)

    def ads_ETL(self):
        list_adId_SB = list(self.iter_ads(**{"includeExtendedDataFields":True}))
        df_adId_SP = pd.json_normalize(list_adId_SB)
        df_adId_SP = self.time_stamp_convert(df_adId_SP,["extendedData.creationDate","extendedData.lastUpdateDate"])
        return self.columnsName_modify(df_adId_SP)

    def keyword_ETL(self):
        list_keywords_SB = [row for _ in list(self.iter_keywords()) for row in _]
        df_keywords_SP = pd.json_normalize(list_keywords_SB)
        return self.columnsName_modify(df_keywords_SP)

    def targets_ETL(self):
        list_targets = list(self.iter_targets())
        df_targets = pd.json_normalize(list_targets)
        # df_targets = self.TZ_Deal(df_targets, ["extendedData.creationDateTime", "extendedData.lastUpdateDateTime"])
        df_targets = self.expression_split(df_targets,"resolvedExpressions")

        return self.columnsName_modify(df_targets)

    def budget_ETL(self,campaign_ids:list):
        list_budget = self.get_budget(campaignIds = campaign_ids)['success']
        df_budget = pd.json_normalize(list_budget)
        df_budget = self.TZ_Deal(df_budget,["usageUpdatedTimestamp"])
        print(df_budget)

if __name__ == '__main__':
    AWS_CREDENTIALS = {
        'lwa_client_id': 'amzn1.application-oa2-client.ebd701cd07854fb38c37ee49ec4ba109',
        'refresh_token': "Atzr|IwEBIL4ur8kbcwRyxVu_srprAAoTYzujnBvA6jU-0SMxkRgOhGjYJSUNGKvw24EQwJa1jG5RM76mQD2P22AKSq8qSD94LddoXGdKDO74eQVYl0RhuqOMFqdrEZpp1p4bIR6_N8VeSJDHr7UCuo8FiabkSHrkq7tsNvRP-yI-bnpQv4EayPBh7YwHVX3hYdRbhxaBvgJENgCuiEPb35Q2-Z6w6ujjiKUAK2VSbCFpENlEfcHNsjDeY7RCvFlwlCoHj1IeiNIaFTE9yXFu3aEWlExe3LzHv6PZyunEi88QJSXKSh56Um0e0eEg05rMv-VBM83cAqc5POmZnTP1vUdZO8fQv3NFLZ-xU6e1WQVxVPi5Cyqk4jYhGf1Y9t98N654y0tVvw74qNIsTrB-8bGS0Uhfe24oBEWmzObvBY3zhtT1d42myGUJv4pMTU6yPoS83zhPKm3LbUDEpBA1hvvc_09jHk7vUEAuFB-UAZzlht2C1yklzQ",
        'lwa_client_secret': 'cbf0514186db4df91e04a8905f0a91b605eae4201254ced879d8bb90df4b474d',
        'profile_id': "3006125408623189"
    }
    ac_etl = SB_ETL(**AWS_CREDENTIALS)
    # print(ac_etl.budget_ETL(campaign_ids=["126327624499318"]))
    print(ac_etl.get_budget(["144123082741012379"]))