sync_amz_data/sync_amz_data/DataTransform/Data_ETL.py

from sync_amz_data.public.amz_ad_client 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)
from datetime import datetime,timezone,timedelta

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):
        """
        时间戳转换为utc
        """
        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):
        """
        TZ时间格式转换为utc
        """
        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):
        """
        拆分placement与percentage列
        """
        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):
        """
        拆分type，value列
        """
        df[segment] = df[segment].astype("string")
        df[segment+str("_type")] = df[segment].str.extract(r"'type':\s{0,1}'(.+?)',")
        df[segment+str("_value")] = df[segment].str.extract(r"'value':\s{0,1}[',[,{](.+)'")
        df[segment+str("_value")] = df[segment+str("_value")].map(lambda x: x if pd.isna(x) or "," not in x else "["+x+"'}]")
        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_portfolios()
        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 adGroups_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"])
        return self.columnsName_modify(df_budget)

class SB_ETL(SBClient,Common_ETLMethod):
    reportMetrics = [
        'applicableBudgetRuleId',
        'applicableBudgetRuleName',
        'attributedConversions14d',
        'attributedConversions14dSameSKU',
        'attributedDetailPageViewsClicks14d',
        'attributedOrderRateNewToBrand14d',
        'attributedOrdersNewToBrand14d',
        'attributedOrdersNewToBrandPercentage14d',
        'attributedSales14d',
        'attributedSales14dSameSKU',
        'attributedSalesNewToBrand14d',
        'attributedSalesNewToBrandPercentage14d',
        'attributedUnitsOrderedNewToBrand14d',
        'attributedUnitsOrderedNewToBrandPercentage14d',
        'campaignBudget',
        'campaignBudgetType',
        'campaignId',
        'campaignName',
        'campaignRuleBasedBudget',
        'campaignStatus',
        'clicks',
        'cost',
        'dpv14d',
        'impressions',
        'unitsSold14d',
        'attributedBrandedSearches14d',
        'topOfSearchImpressionShare']
    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 adGroups_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 keywords_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"])
        return self.columnsName_modify(df_budget)

    def report_campaignsRecord_ETL(self):
        today = datetime.today()
        date = (datetime(today.year,today.month,today.day,tzinfo=timezone.utc)-timedelta(days=1)).strftime("%Y%m%d")
        print(date)
        need_removedList = []
        if need_removedList is not None:
            [SB_ETL.reportMetrics.remove(i) for i in need_removedList]
        list_campaigns_report = self.get_v3_report(record_type="campaigns",metrics=SB_ETL.reportMetrics,report_date=date)
        # print(list_campaigns_report)
        df_campaign_report = pd.json_normalize(list_campaigns_report)
        return df_campaign_report


class SD_ETL(SDClient,Common_ETLMethod):
    def campaigns_ETL(self):
        list_campaign_SD = self.get_campaigns()
        df_campaign = pd.json_normalize(list_campaign_SD)
        df_campaign['startDate'] = df_campaign['startDate'].map(lambda x: datetime.strptime(x,"%Y%m%d").date()) # 转换为标准时间格式
        df_campaign['portfolioId'] = df_campaign['portfolioId'].fillna(-1).astype("int64") # 将portfolio列为空的填充为-1
        return self.columnsName_modify(df_campaign)

    def adGroups_ETL(self,**param):
        list_adGroups_SD = [row for _ in list(self.iter_adGroups(**param)) for row in _]
        df_adGroups_SD = pd.json_normalize(list_adGroups_SD)
        tactic = {"T00020":"Contextual targeting","T00030":"Audiences targeting"}
        df_adGroups_SD["tactic_type"] = df_adGroups_SD['tactic'].map(tactic) # T00020、T00030解释字段
        return self.columnsName_modify(df_adGroups_SD)

    def ads_ETL(self):
        list_ads_SD = [row for _ in list(self.iter_ads()) for row in _]
        df_ads_SD = pd.json_normalize(list_ads_SD)
        return self.columnsName_modify(df_ads_SD)

    def targets_ETL(self,**param):
        list_targets = [row for _ in list(self.iter_targets(**param)) for row in _]
        df_targets = pd.json_normalize(list_targets)
        df_targets = self.expression_split(df_targets, "resolvedExpression")
        return self.columnsName_modify(df_targets)

    def budget_ETL(self,campaignsIds:list):
        list_budget = self.get_budget(campaignIds=campaignsIds)['success']
        df_budget = pd.json_normalize(list_budget)
        df_budget = self.TZ_Deal(df_budget,["usageUpdatedTimestamp"])
        return self.columnsName_modify(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.report_targetsRecord_ETL())