sync_amz_data/sync_amz_data/public/sp_api_client.py

import clickhouse_connect
import time
import warnings
warnings.filterwarnings('ignore')
import numpy as np
from pymysql import Timestamp
from sp_api.util import throttle_retry, load_all_pages
from sp_api.api import Orders,ListingsItems,Inventories,Reports,CatalogItems
from sp_api.base import Marketplaces,ReportType,ProcessingStatus
import pandas as pd
import gzip
from io import BytesIO,StringIO
from datetime import datetime, timedelta,timezone
import pytz
import time
from dateutil.parser import parse
import pymysql
from typing import List, Literal


class SpApiRequest:

    def __init__(self, credentials,marketplace):
        self.credentials = credentials
        self.marketplace = marketplace
        # self.shopInfo = shop_infos('3006125408623189')
        # self.timezone = self.shopInfo['time_zone']
        # self.profileid = '3006125408623189'

    @classmethod
    def mysql_connect_auth(cls):
        conn = pymysql.connect(user="admin",
        password="NSYbBSPbkGQUbOSNOeyy",
        host="retail-data.cnrgrbcygoap.us-east-1.rds.amazonaws.com",
        database="ansjer_dvadmin",
        port=3306)

        return conn

    @classmethod
    def mysql_connect_auth_lst(cls):
        conn = pymysql.connect(user="huangyifan",
                               password="123456",
                               host="127.0.0.1",
                               database="amz_sp_api",
                               port=3306)
        return conn

    @classmethod
    def mysql_connect(cls):
        conn = pymysql.connect(user="huangyifan",
                               password="123456",
                               host="127.0.0.1",
                               database="amz_sp_api",
                               port=3306)
        return conn

    @classmethod
    def mysql_adTest_connect(cls):
        conn = pymysql.connect(user="root",
                               password="sandbox",
                               host="192.168.1.225",
                               database="asj_ads",
                               port=3306)
        return conn

    @classmethod
    def get_catelog(cls,account_name,country=Marketplaces.US,asin=None):
        if country in [Marketplaces.US, Marketplaces.BR, Marketplaces.CA,Marketplaces.MX]:
            region = 'NA'
        elif country in [Marketplaces.DE,Marketplaces.AE, Marketplaces.BE,  Marketplaces.PL,
                        Marketplaces.EG,Marketplaces.ES,  Marketplaces.GB, Marketplaces.IN, Marketplaces.IT,
                        Marketplaces.NL, Marketplaces.SA, Marketplaces.SE, Marketplaces.TR,Marketplaces.UK,Marketplaces.FR,
                        ]:
            region = 'EU'
        else:
            region = str(country)[-2:]
        df = cls.auth_info()
        try:
            refresh_token = df.query("account_name==@account_name and region==@region")['refresh_token'].values[0]
        except:
            print("请输入正确的account name与Marketplace")
            return '获取失败'
        cred = {
                'refresh_token': refresh_token,
                'lwa_app_id': 'amzn1.application-oa2-client.1f9d3d4747e14b22b4b598e54e6b922e',  # 卖家中心里面开发者资料LWA凭证
                'lwa_client_secret': 'amzn1.oa2-cs.v1.3af0f5649f5b8e151cd5bd25c10f2bf3113172485cd6ffc52ccc6a5e8512b490',
                'aws_access_key': 'AKIARBAGHTGOZC7544GN',
                'aws_secret_key': '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',
                'role_arn': 'arn:aws:iam::070880041373:role/Amazon_SP_API_ROLE'
            }
        cate_item = CatalogItems(credentials=cred, marketplace=country)
        images_info = cate_item.get_catalog_item(asin=asin,**{"includedData":['images']})
        images = images_info.images[0].get('images')[0]['link']
        title_info = cate_item.get_catalog_item(asin=asin)
        title = title_info.payload['summaries'][0]['itemName']
        return {'images':images,'title':title}

    def create_report(self,**kwargs):
        reportType = kwargs['reportType']
        reportOptions =kwargs.get("reportOptions")

        dataStartTime = datetime.now().strftime("%Y-%m-%dT%H:%M:%S") if kwargs.get("dataStartTime") is None else kwargs.get("dataStartTime")+"T00:00:00"
        dataEndTime = datetime.now().strftime("%Y-%m-%dT%H:%M:%S") if kwargs.get("dataEndTime") is None else kwargs.get("dataEndTime")+"T23:59:59"

        report = Reports(credentials=self.credentials, marketplace=self.marketplace)
        rel = report.create_report(
            reportType=reportType,marketplaceIds=[kwargs['marketpalceids'] if kwargs.get('marketpalceids') is not None else self.marketplace.marketplace_id],
            reportOptions=reportOptions,dataStartTime=dataStartTime,dataEndTime=dataEndTime
                        )
        reportId = rel.payload.get("reportId")
        # print(reportId)
        return reportId

    def decompression(self,reportId):
        report = Reports(credentials=self.credentials, marketplace=self.marketplace)
        while True:
            reportId_info = report.get_report(reportId=reportId)
            # print(reportId_info.payload)
            print("please wait...")
            if reportId_info.payload.get("processingStatus")==ProcessingStatus.DONE:
                reportDocumentId = reportId_info.payload.get("reportDocumentId")
                rp_table = report.get_report_document(reportDocumentId=reportDocumentId,download=False)
                print(rp_table)
                if rp_table.payload.get('compressionAlgorithm') is not None and self.marketplace.marketplace_id not in ['A1VC38T7YXB528']:#
                    df = pd.read_table(filepath_or_buffer=rp_table.payload['url'],compression={"method":'gzip'},encoding='iso-8859-1')
                    return df
                elif rp_table.payload.get('compressionAlgorithm') is not None and self.marketplace.marketplace_id in ['A1VC38T7YXB528']:
                    df = pd.read_table(filepath_or_buffer=rp_table.payload['url'], compression={"method": 'gzip'},
                                       encoding='Shift-JIS')
                    # df.columns =
                    return df
                elif rp_table.payload.get('compressionAlgorithm') is None and self.marketplace.marketplace_id not in ['A1VC38T7YXB528']:
                    df = pd.read_table(rp_table.payload.get("url"),encoding='iso-8859-1')
                    return df
                elif rp_table.payload.get('compressionAlgorithm') is None and self.marketplace.marketplace_id in ['A1VC38T7YXB528']:
                    df = pd.read_table(rp_table.payload.get("url"),encoding='Shift-JIS')
                    return df
            elif reportId_info.payload.get("processingStatus") in [ProcessingStatus.CANCELLED,ProcessingStatus.FATAL]:
                print(reportId_info)
                print("取消或失败")
                break
            time.sleep(15)
            print("please wait...")


    def data_deal(self,decom_df,seller_id):
        decom_df['mainImageUrl'] = decom_df['seller-sku'].map(lambda x: self.get_mainImage_url(x))
        url_columns = [i for i in decom_df.columns if "url" in i.lower()]
        if len(url_columns) > 0:
            decom_df[url_columns] = decom_df[url_columns].astype("string")
        asin_columns = [i for i in decom_df.columns if 'asin' in i.lower()]
        if len(asin_columns) > 0:
            decom_df[asin_columns] = decom_df[asin_columns].astype("string")
        if 'pending-quantity' in decom_df.columns:
            decom_df['pending-quantity'] = decom_df['pending-quantity'].map(
                lambda x: 0 if pd.isna(x) or np.isinf(x) else x).astype("int32")
        deletecolumns = [i for i in decom_df.columns if 'zshop' in i.lower()]
        decom_df.drop(columns=deletecolumns, inplace=True)
        if 'quantity' in decom_df.columns:
            decom_df['quantity'] = decom_df['quantity'].map(lambda x: 0 if pd.isna(x) or np.isinf(x) else x).astype(
                "int32")
        decom_df['opendate_date'] = decom_df['open-date'].map(lambda x: self.datetime_deal(x))
        if 'add-delete' in decom_df.columns:
            decom_df['add-delete'] = decom_df['add-delete'].astype('string', errors='ignore')
        if 'will-ship-internationally' in decom_df.columns:
            decom_df['will-ship-internationally'] = decom_df['will-ship-internationally'].astype('string',errors='ignore')
        if 'expedited-shipping' in decom_df.columns:
            decom_df['expedited-shipping'] = decom_df['expedited-shipping'].astype('string',errors='ignore')
        decom_df['updateTime'] = datetime.now()
        decom_df['timezone'] = "UTC"
        decom_df['seller_id'] = seller_id
        #
        decom_df['item-description'] = decom_df['item-description'].str.slice(0,500)
        decom_df[decom_df.select_dtypes(float).columns] = decom_df[decom_df.select_dtypes(float).columns].fillna(0.0)
        decom_df[decom_df.select_dtypes(int).columns] = decom_df[decom_df.select_dtypes(int).columns].fillna(0)
        decom_df[decom_df.select_dtypes(datetime).columns] = decom_df[decom_df.select_dtypes(datetime).columns].astype(
            'string')
        decom_df.fillna('', inplace=True)
        # print(decom_df.info())
        return decom_df

    def GET_MERCHANT_LISTINGS_ALL_DATA(self,limit=None):
        start = time.time()
        para = {"reportType":ReportType.GET_MERCHANT_LISTINGS_ALL_DATA}
        reportid = self.create_report(**para)
        decom_df = self.decompression(reportid)
        print("连接数据库")
        conn = self.mysql_connect()
        print("连接成功")
        cursor = conn.cursor()
        timezone = "UTC" #pytz.timezone(self.timezone)
        bondary_date = (datetime.now()).strftime("%Y-%m-%d") #+ timedelta(days=-28)
        cursor.execute(f"""select * from amz_sp_api.productInfo where (mainImageUrl is not null and mainImageUrl not in ('', ' ')) and 
                        (`seller-sku` not in ('',' ') and `seller-sku` is not null) and 
                        `updateTime`>='{bondary_date}'""") #`seller-sku`,`updateTime`,`mainImageUrl`
        col = [i[0] for i in cursor.description]
        query_rel = cursor.fetchall()

        if len(query_rel)!=0:
            print(query_rel[0])
            df = pd.DataFrame(query_rel,columns=col)
            listingid = df['listing-id'].to_numpy().tolist()
            decom_df = decom_df.query("`listing-id` not in @listingid")
            print("数据条数: ",len(decom_df))
            # print(f"delete * from amz_sp_api.productInfo where `listing-id` not in {tuple(listingid)}")

            # conn.commit()

        if len(decom_df)==0:
            return "Done"

        if limit != None:
            decom_df = decom_df.iloc[:limit,:]
        print("getting mainImageInfo...")
        rowcount = 0
        while rowcount < len(decom_df):
            df_insert = decom_df.copy()
            df_insert = df_insert.iloc[rowcount:rowcount + 200, :]

            df_insert = self.data_deal(df_insert)
            list_df = df_insert.to_numpy().tolist()

            # print(list(conn.query("select * from amz_sp_api.orderReport")))
            sql = f"""
                        insert into amz_sp_api.productInfo
                        values (%s,%s,%s,%s,%s,%s,%s, %s,%s,%s,%s,%s,%s,%s, %s,%s,%s,%s,%s,%s,%s, %s,%s,%s,%s,%s,%s,%s,%s)
                    """ #ok
            # print(sql)
            conn = self.mysql_connect()
            cursor = conn.cursor()
            try:
                conn.begin()
                cursor.executemany(sql, list_df)
                conn.commit()
                print("插入中...")
                insert_listingid = df_insert['listing-id'].to_numpy().tolist()
                cursor.execute(f"delete from amz_sp_api.productInfo where `listing-id` not in {tuple(insert_listingid)} and `updateTime`<'{bondary_date}'")
                conn.commit()
                rowcount += 200
            except Exception as e:
                conn.rollback()
                print(e)
                try:
                    conn = self.mysql_connect()
                    cursor = conn.cursor()
                    conn.begin()
                    cursor.executemany(sql, list_df)
                    conn.commit()
                    insert_listingid = df_insert['listing-id'].to_numpy().tolist()
                    cursor.execute(f"delete from amz_sp_api.productInfo where `listing-id` not in {tuple(insert_listingid)} and `updateTime`<'{bondary_date}'")
                    conn.commit()
                except Exception as e:
                    conn.rollback()
                    print(e)
                    break
                # break
        conn.close()
        print("全部完成")
        end =time.time()
        print("duration:",end-start)
        return decom_df

    def get_listing_info(self, sku,seller_id):
        listingClient = ListingsItems(credentials=self.credentials, marketplace=self.marketplace)
        try:
            r1 = listingClient.get_listings_item(sellerId=seller_id, sku=sku)
            # print(r1.payload)
            json_content = r1.payload.get("summaries")[0]
            item_name = json_content.get("itemName")
            item_name ='###' if item_name==None else item_name
            img = json_content.get("mainImage")
            img_url = '###' if img is None else img.get("link")
            # print(str(img_url)+"-----"+ str(item_name))
            return str(img_url)+"-----"+ str(item_name)
        except Exception as e:
            try:
                print("获取图片url过程错误重试, 错误message: ",e)
                time.sleep(3)
                r1 = listingClient.get_listings_item(sellerId=seller_id, sku=sku)
                print(r1.payload)
                json_content = r1.payload.get("summaries")[0]

                item_name = json_content.get("itemName")
                item_name = '###' if item_name == None else item_name
                img = json_content.get("mainImage")
                img_url = '###' if img is None else img.get("link")
                return str(img_url)+"-----"+ str(item_name)
            except Exception as e:
                print(e)
                return "###-----###"


    def datetime_deal(self,timestring):
        timezone_ = {"AEST":"Australia/Sydney",
                    "AEDT":"Australia/Sydney",
                    "PST":"America/Los_Angeles",
                    "PDT":"America/Los_Angeles",
                    "CST":"America/Chicago",
                    "CDT":"America/Chicago",
                    "MET":"MET",
                    "MEST":"MET",
                    "BST":"Europe/London",
                    "GMT":"GMT",
                    "CET":"CET",
                    "CEST":"CET",
                    "JST":"Asia/Tokyo",
                    "BRT":"America/Sao_Paulo"}

        date_list = str.split(timestring,sep = ' ')
        if len(date_list)<3:
            try:
                return datetime.strptime(date_list[0],"%Y-%m-%d")
            except:
                try:
                    return datetime.strptime(date_list[0], "%Y/%m/%d")
                except:
                    try:
                        return datetime.strptime(date_list[0], "%d/%m/%Y")
                    except Exception as e:
                        print(e)
                        return datetime(1999, 12, 31, 0, 0, 0)
        try:
            time_date = datetime.strptime(date_list[0]+date_list[1],"%Y-%m-%d%H:%M:%S")
            timezone = pytz.timezone(timezone_[date_list[2]])
            time_ = timezone.localize(time_date)
            return time_.astimezone(pytz.UTC)
        except:
            try:
                time_date = datetime.strptime(date_list[0] + date_list[1], "%d/%m/%Y%H:%M:%S")
                timezone = pytz.timezone(timezone_[date_list[2]])
                time_ = timezone.localize(time_date)
                return time_.astimezone(pytz.UTC)
            except :
                try:
                    time_date = datetime.strptime(date_list[0] + date_list[1], "%Y/%m/%d%H:%M:%S")
                    timezone = pytz.timezone(timezone_[date_list[2]])
                    time_ = timezone.localize(time_date)
                    return time_.astimezone(pytz.UTC)
                except Exception as e1:
                    print(e1)
                    return datetime(1999,12,31,0,0,0)

    def update_data(self,df,seller_id,country_code,conn):
        conn = SpApiRequest.mysql_connect_auth_lst()
        cursor = conn.cursor()
        columns = ['listing-id', 'seller_id',
         'asin1', 'seller-sku', 'title', 'image_link', 'country_code',
         'marketplace_id', 'quantity', 'fulfillment_channel',
         'price', 'opendate', 'status', 'update_datetime', 'product-id', 'product-id-type'
         ]
        if country_code=='GB':
            country_code="UK"
            df['country_code'] = "UK"


        df_data = pd.DataFrame(columns=columns)
        delete_list = []

        marketplace_id = self.marketplace.marketplace_id
        try:
            cursor.execute(f"""select * from
                                amz_sp_api.seller_listings where seller_id='{seller_id}' and marketplace_id='{marketplace_id}'""")
            col = [i[0] for i in cursor.description]
            query_rel = cursor.fetchall()
            df_rel = pd.DataFrame(query_rel, columns=col)
            df_rel['quantity'] = df_rel['quantity'].fillna(0).astype('int64')
            df_rel['price'] = df_rel['price'].fillna(0.0).astype('float64')
            df_rel['product_id_type'] = df_rel['product_id_type'].astype('int64')
            df['update_datetime'] =df['update_datetime'].astype('datetime64[ns]')

            df['quantity'] = df['quantity'].fillna(0).astype('int64')
            df['price']= df['price'].fillna(0.0).astype('float64')
            # print(df_rel.dtypes)
            # print(df[columns].dtypes)
            row = 0
            while row < len(df):
                temp_df = df.iloc[row, :]

                listing_id = temp_df['listing-id']
                asin = temp_df['asin1']
                sku = temp_df['seller-sku']
                quantity = temp_df['quantity']
                fulfillment_channel = temp_df['fulfillment_channel']
                price = temp_df['price']
                product_id = temp_df['product-id']
                title = temp_df['title']
                imageurl = temp_df['image_link']
                temp = df_rel.query("""listing_id==@listing_id and asin==@asin and sku==@sku and quantity==@quantity and fulfillment_channel==@fulfillment_channel and price==@price and product_id==@product_id and country_code==@country_code and seller_id==@seller_id and title==@title and image_link==@imageurl""")
                print("需要关注数据（是否异常）：",len(temp),temp.to_numpy().tolist()) if len(temp)>1 else 1
                if len(temp)>1:
                    temp = temp.head(1).to_numpy().tolist()
                    df_data = df_data.append(temp_df, ignore_index=True)
                    delete_list.append((seller_id, marketplace_id, sku, listing_id, product_id))
                # print(len(temp))
                if len(temp)==0:
                    df_data = df_data.append(temp_df,ignore_index=True)
                    delete_list.append((seller_id,marketplace_id,sku,listing_id,product_id))
                row += 1
            print("判断不同数据条数",len(delete_list))
            print("预计更新数据条数",len(df_data))
            try:
                # print(tuple(delete_list))
                if len(delete_list)>0:
                    query = f"""delete from amz_sp_api.seller_listings 
                           where (seller_id,marketplace_id,sku,listing_id,product_id) in %s""" #where (seller_id,country_code) in %s"""
                    cursor.execute(query,(delete_list,))

                    conn.commit()
                    print(delete_list)
                    print("进行中...")
            except Exception as e:
                print(e)
                conn.rollback()
            return df_data
        except Exception as e:
            print("错误：", e)
            return df


    def GET_FLAT_FILE_OPEN_LISTINGS_DATA(self,conn=None,seller_id=None):
        para = {"reportType": ReportType.GET_MERCHANT_LISTINGS_ALL_DATA}
        reportid = self.create_report(**para)
        df = self.decompression(reportid)
        if len(df)>0:
            if self.marketplace.marketplace_id =='A1VC38T7YXB528':
                df.columns = ['item-name','listing-id','seller-sku','price','quantity','open-date','product-id-type','item-description',
                              'item-condition','overseas shipping','fast shipping','asin1','stock_number','fulfillment-channel','merchant-shipping-group','status']

            df['seller_id'] = seller_id
            df['marketplace_id'] = self.marketplace.marketplace_id
            df['country_code'] = str(self.marketplace)[-2:]
            if 'fulfilment-channel' in df.columns:
                print("changed fulfilment-channel:")
                print(seller_id,self.marketplace)
                df['fulfillment-channel'] = df['fulfilment-channel'].copy()
            df['fulfillment_channel'] = df['fulfillment-channel'].map(lambda x:"FBA" if not pd.isna(x) and len(x)>0 and str(x)[1:4] in "AMAZON" else x)
            df['fulfillment_channel'] = df['fulfillment_channel'].map(lambda x: "FBM" if not pd.isna(x) and len(x)>0 and str(x)[1:4] in "DEFAULT" else x)

            if 'asin1' not in df.columns:
                df['asin1'] = ''
            if 'product-id' not in df.columns:
                df['product-id'] = ''

            # 空值处理
            df['quantity'] = df['quantity'].fillna(0).astype('int64',errors='ignore')
            df[['listing-id','seller_id','asin1','seller-sku','country_code','marketplace_id','fulfillment_channel','status','product-id']] = df[['listing-id','seller_id','asin1','seller-sku','country_code','marketplace_id','fulfillment_channel','status','product-id']].fillna('').astype('string',errors='ignore')
            df['price'] = df['price'].fillna(0.0).astype('float64',errors='ignore')
            df.fillna('',inplace=True)

            # 时间处理
            df['opendate'] = df['open-date'].map(lambda x: self.datetime_deal(x))
            df['update_datetime'] = datetime.now(pytz.UTC).date()

            origin_columns = ['listing-id','seller_id',
                              'asin1','seller-sku','title','image_link','country_code',
                              'marketplace_id','quantity','fulfillment_channel',
                              'price','opendate','status','update_datetime','product-id','product-id-type'
                              ]
            conn = SpApiRequest.mysql_connect_auth_lst()
            cursor = conn.cursor()
            cursor.execute("""select product_id,asin from (select * from amz_sp_api.seller_listings where asin is not null 
                                and asin<>'' and product_id is not null and product_id <>'') t1 group by product_id,asin""")
            query_ = cursor.fetchall()
            col_name = [i[0] for i in cursor.description]
            df_datatable = pd.DataFrame(query_, columns=col_name)
            merged_df = df.merge(df_datatable[['product_id','asin']],how='left',left_on='product-id',right_on='product_id')
            print(merged_df.head())

            def func_(asin,asin1,product_id,cred,market_p,seller_id,sku):
                if 'B0' in str(product_id)[:3]:
                    return str(product_id)
                if (pd.isna(asin1) or asin1=='') and (pd.isna(asin)==False and asin !=''):
                    if 'B0' in asin[:3]:
                        return asin
                elif (pd.isna(asin1)==False and asin1!=''):
                    if 'B0' in asin1[:3]:
                        return asin1

                listingClient = ListingsItems(credentials=cred, marketplace=market_p)
                try:
                    r1 = listingClient.get_listings_item(sellerId=seller_id, sku=sku)
                    print(r1.payload)
                    asin = r1.payload.get("summaries")[0].get("asin")
                    return asin
                except Exception as e:
                    print("获取图片url过程错误重试, 错误message: ", e)
                    time.sleep(3)
                    r1 = listingClient.get_listings_item(sellerId=seller_id, sku=sku)
                    print(r1.payload)
                    asin = r1.payload.get("summaries")[0].get("asin")
                    return asin

            merged_df['asin1'] = merged_df.apply(lambda x:func_(x['asin'],x['asin1'],x['product-id'],self.credentials,self.marketplace,seller_id,x['seller-sku']),axis=1) #x['asin'] if pd.isna(x['asin1']) or x['asin1']=='' else x['asin1']
            # merged_df.to_csv("tmp.csv")
            # merged_df = merged_df.loc[:10,:].copy()

            print("获取listing Info...")
            merged_df['temp_columns'] = merged_df.apply(lambda x: self.get_listing_info(x['seller-sku'],seller_id),axis=1)
            merged_df[['image_link','title']] = merged_df['temp_columns'].str.split("-----",expand=True)
            # merged_df['image_link'] = ''
            # merged_df['title'] = ''


            merged_df.fillna('',inplace=True)
            df1 = merged_df.copy()
            print(df1[origin_columns].head(1))
            update_df = self.update_data(df1,seller_id,str(self.marketplace)[-2:],conn)
            if len(update_df)==0:
                return '无更新数据插入'
            # update_df['country_code'] = update_df['country_code'].map({"GB":"UK"})
            conn = SpApiRequest.mysql_connect_auth_lst()
            cursor = conn.cursor()

            try:
                insertsql = """insert into
                amz_sp_api.seller_listings(listing_id,seller_id,asin,sku,title,image_link,country_code,marketplace_id,quantity,
                        fulfillment_channel,price,launch_datetime,status,update_datetime,product_id,product_id_type)
                values(%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s)"""
                conn.begin()
                cursor.executemany(insertsql,tuple(update_df[origin_columns].to_numpy().tolist()))
                conn.commit()
                print("插入完成")
                return '插入完成'
            except Exception as e:
                print("插入错误：",e)
                conn.rollback()
                return '出错回滚'

    @staticmethod
    def auth_info():
        auth_conn = SpApiRequest.mysql_connect_auth()
        cursor = auth_conn.cursor()
        cursor.execute("select * from amazon_sp_report.amazon_sp_auth_info;")
        columns_name = [i[0] for i in cursor.description]
        rel = cursor.fetchall()
        df = pd.DataFrame(rel, columns=columns_name)
        return df

    @classmethod
    def get_orders_allShops(cls):
        pass

    @staticmethod
    def data_judge_secondTry(sp_api,data_type,seller_id,auth_conn):
        try:
            SpApiRequest.data_judge(sp_api, data_type, seller_id, auth_conn)
        except:
            time.sleep(3)
            SpApiRequest.data_judge(sp_api, data_type, seller_id, auth_conn)


    @staticmethod
    def data_judge(sp_api,data_type,seller_id,auth_conn):
        if data_type == "GET_FLAT_FILE_OPEN_LISTINGS_DATA":
            return sp_api.GET_FLAT_FILE_OPEN_LISTINGS_DATA(auth_conn,seller_id)
        elif data_type =="GET_FLAT_FILE_ALL_ORDERS_DATA_BY_ORDER_DATE_GENERAL":
            return sp_api.GET_FLAT_FILE_ALL_ORDERS_DATA_BY_ORDER_DATE_GENERAL(seller_id)
        elif data_type =="GET_FLAT_FILE_RETURNS_DATA_BY_RETURN_DATE":
            return sp_api.GET_FLAT_FILE_RETURNS_DATA_BY_RETURN_DATE(seller_id)
        else:
            return ""

    @classmethod
    def get_refreshtoken(cls):
        df = cls.auth_info()
        refreshtoken_list = (df['refresh_token'].to_numpy().tolist())
        return refreshtoken_list

    @classmethod
    def get_allShops(cls,data_type=Literal["GET_FLAT_FILE_OPEN_LISTINGS_DATA","GET_FLAT_FILE_ALL_ORDERS_DATA_BY_ORDER_DATE_GENERAL"]):
        df = cls.auth_info()
        refreshtoken_list = (df['refresh_token'].to_numpy().tolist())
        refreshtoken_list.reverse()
        for refresh_token in refreshtoken_list:
            aws_credentials = {
                'refresh_token': refresh_token,
                'lwa_app_id': 'amzn1.application-oa2-client.1f9d3d4747e14b22b4b598e54e6b922e',  # 卖家中心里面开发者资料LWA凭证
                'lwa_client_secret': 'amzn1.oa2-cs.v1.3af0f5649f5b8e151cd5bd25c10f2bf3113172485cd6ffc52ccc6a5e8512b490',
                'aws_access_key': 'AKIARBAGHTGOZC7544GN',
                'aws_secret_key': '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',
                'role_arn': 'arn:aws:iam::070880041373:role/Amazon_SP_API_ROLE'
            }

            single_info = df.query("refresh_token==@refresh_token")
            region_circle = single_info['region'].values[0]
            seller_id = single_info['selling_partner_id'].values[0]
            account_name = single_info['account_name'].values[0]
            if region_circle == 'NA':
                pass
                for marketplace in [Marketplaces.US, Marketplaces.BR, Marketplaces.CA,Marketplaces.MX]:
                    sp_api = SpApiRequest(aws_credentials, marketplace)
                    try:
                        auth_conn = SpApiRequest.mysql_connect_auth()
                        cls.data_judge_secondTry(sp_api, data_type, seller_id, auth_conn)
                        ## sp_api.GET_FLAT_FILE_OPEN_LISTINGS_DATA(auth_conn, seller_id)
                    except Exception as e:
                        print(e)
            elif region_circle == 'EU':
                pass
                for marketplace in [Marketplaces.DE,Marketplaces.AE, Marketplaces.BE,  Marketplaces.PL,
                                    Marketplaces.EG,Marketplaces.ES,  Marketplaces.GB, Marketplaces.IN, Marketplaces.IT,
                                    Marketplaces.NL, Marketplaces.SA, Marketplaces.SE, Marketplaces.TR,Marketplaces.UK,Marketplaces.FR,
                                    ]:
                    sp_api = SpApiRequest(aws_credentials, marketplace)
                    try:
                        auth_conn = SpApiRequest.mysql_connect_auth()
                        cls.data_judge_secondTry(sp_api, data_type, seller_id, auth_conn)
                        ## sp_api.GET_FLAT_FILE_OPEN_LISTINGS_DATA(auth_conn, seller_id)
                    except Exception as e:
                        print(e)

            else:
            # if region_circle not in ['NA','EU']:
                auth_conn = SpApiRequest.mysql_connect_auth()
                print(region_circle)
                marketplace = eval(f'Marketplaces.{region_circle}')
                sp_api = SpApiRequest(aws_credentials, marketplace)
                cls.data_judge_secondTry(sp_api, data_type, seller_id, auth_conn)
            ## sp_api.GET_FLAT_FILE_OPEN_LISTINGS_DATA(auth_conn, seller_id)

    def timeDeal(self, orgTime):
        orgTime = parse(orgTime)
        timezone = pytz.timezone("UTC")
        shopTime = orgTime.astimezone(timezone)
        shopTime_datetime = datetime(shopTime.year, shopTime.month, shopTime.day, shopTime.hour, shopTime.minute,
                                     shopTime.second)
        return shopTime_datetime

    def GET_FLAT_FILE_RETURNS_DATA_BY_RETURN_DATE(self,seller_id):
        shopReportday = (datetime.now() + timedelta(days=-2)).strftime("%Y-%m-%d")
        # print(shopReportday)
        para = {"reportType": ReportType.GET_SELLER_FEEDBACK_DATA,
                "dataStartTime": shopReportday, "dataEndTime": shopReportday,
                }
        reportid = self.create_report(**para)  # {"ShowSalesChannel":"true"}
        decom_df = self.decompression(reportid)
        print(decom_df)
        # print(decom_df.columns)

    def GET_FLAT_FILE_ALL_ORDERS_DATA_BY_ORDER_DATE_GENERAL(self,seller_id):
        # timezone_ = pytz.timezone(self.timezone)
        shopReportday = (datetime.now() + timedelta(days=-1)).strftime("%Y-%m-%d")
        # print(shopReportday)
        para = {"reportType":ReportType.GET_FLAT_FILE_ALL_ORDERS_DATA_BY_ORDER_DATE_GENERAL,"dataStartTime":shopReportday,"dataEndTime":shopReportday,"reportOptions":{"ShowSalesChannel":"true"}}
        reportid = self.create_report(**para) #{"ShowSalesChannel":"true"}
        decom_df = self.decompression(reportid)
        decom_df[decom_df.select_dtypes(float).columns] = decom_df[decom_df.select_dtypes(float).columns].fillna(0.0)
        decom_df[decom_df.select_dtypes(int).columns] = decom_df[decom_df.select_dtypes(int).columns].fillna(0)
        decom_df[decom_df.select_dtypes(datetime).columns] = decom_df[decom_df.select_dtypes(datetime).columns].astype('string')
        if "purchase-order-number" in decom_df.columns:
            decom_df['purchase-order-number'] = decom_df['purchase-order-number'].astype("string")

        decom_df.fillna('',inplace=True)
        # decom_df.to_csv('order.csv')
        decom_df["ReportDate"] = parse(shopReportday)
        # decom_df['timezone'] =  decom_df["purchase-date"].map(lambda x: parse(x).tzname()).fillna(method='bfill')
        decom_df['timezone'] = "UTC"
        print("==========================================================")
        decom_df[["purchase-date", "last-updated-date"]] = decom_df[["purchase-date", "last-updated-date"]].applymap(
            lambda x: self.timeDeal(x) if pd.isna(x) == False or x != None else x)
        if 'is-business-order' not in decom_df.columns:
            decom_df['is-business-order'] = None
        if 'purchase-order-number' not in decom_df.columns:
            decom_df['purchase-order-number'] = '-'
        if 'price-designation' not in decom_df.columns:
            decom_df['price-designation'] = '-'

        decom_df['seller_id'] = seller_id
        country_code = str(self.marketplace)[-2:]
        if country_code=='GB':
            country_code="UK"
            # decom_df['country_code'] = "UK"
        decom_df['country_code'] = country_code
        # print(decom_df[])
        reserve_columns = ["amazon-order-id","merchant-order-id","purchase-date","last-updated-date","order-status",
                        "fulfillment-channel","sales-channel","order-channel","ship-service-level","product-name",
                        "sku","asin","item-status","quantity","currency","item-price","item-tax","shipping-price",
                        "shipping-tax","gift-wrap-price","gift-wrap-tax","item-promotion-discount",
                        "ship-promotion-discount","ship-city","ship-state","ship-postal-code","ship-country",
                        "promotion-ids","is-business-order","purchase-order-number","price-designation","ReportDate",
                        "timezone","seller_id","country_code"
                        ]
        list_df = decom_df[reserve_columns].to_numpy().tolist()
        # print(list_df)
        # print(list_df[0])
        # tuple_data = [tuple(i) for i in list_df]
        conn = self.mysql_connect()
        cursor = conn.cursor()
        # print(list(conn.query("select * from amz_sp_api.orderReport")))
        sql = f"""
            insert into amz_sp_api.orderreport_renew1
            values (%s,%s,%s,%s,%s,%s,%s, %s,%s,%s,%s,%s,%s,%s, %s,%s,%s,%s,%s,%s,%s, %s,%s,%s,%s,%s,%s,%s, %s,%s,%s,%s,%s,%s,%s)
        """ #ok
        # print(sql)
        try:
            conn.begin()
            cursor.executemany(sql,list_df)
            conn.commit()
            print("插入完成")
        except Exception as e:
            conn.rollback()
            print(e)


if __name__ == '__main__':
    SpApiRequest.get_allShops("GET_FLAT_FILE_ALL_ORDERS_DATA_BY_ORDER_DATE_GENERAL")

    # rel = SpApiRequest.get_catelog(account_name='ANLAPUS_US',country=Marketplaces.US,asin='B0BVXB4KT9')
    # print(rel)

"""
create database amz_sp_api;
"""
"""

        create table amz_sp_api.productInfo
        (
        `item-name`	VARCHAR(300),
        `item-description`	VARCHAR(1000),
        `listing-id`	VARCHAR(50),
        `seller-sku`	VARCHAR(50),
        `price`	FLOAT,
        `quantity`	INT,
        `open-date`	VARCHAR(70),
        `image-url`	VARCHAR(300),
        `item-is-marketplace`	VARCHAR(50),
        `product-id-type`	INT,
        `item-note`	VARCHAR(300),
        `item-condition`	INT,
        `asin1`	VARCHAR(50),
        `asin2`	VARCHAR(50),
        `asin3`	VARCHAR(50),
        `will-ship-internationally`	VARCHAR(50),
        `expedited-shipping`	VARCHAR(50),
        `product-id`	VARCHAR(50),
        `bid-for-featured-placement`	FLOAT,
        `add-delete`	VARCHAR(50),
        `pending-quantity`	INT,
        `fulfillment-channel`	VARCHAR(50),
        `merchant-shipping-group`	VARCHAR(50),
        `status`	VARCHAR(50),
        `mainImageUrl`	VARCHAR(300),
        `opendate_date`	Date,
        `updateTime`	Date,
        `timezone`	VARCHAR(30)
        )
"""
"""
create table amz_sp_api.orderReport
(`amazon-order-id` VARCHAR(40),
`merchant-order-id` VARCHAR(40),
`purchase-date` DATETIME,
`last-updated-date` DATETIME,
`order-status` VARCHAR(40),
`fulfillment-channel` VARCHAR(40),
`sales-channel` VARCHAR(40),
`order-channel` VARCHAR(40),
`ship-service-level` VARCHAR(40),
`product-name` VARCHAR(250),
`sku` VARCHAR(50),
`asin` VARCHAR(40),
`item-status` VARCHAR(40),
`quantity` INT,
`currency` VARCHAR(40),
`item-price` FLOAT,
`item-tax` FLOAT,
`shipping-price` FLOAT,
`shipping-tax` FLOAT,
`gift-wrap-price` FLOAT,
`gift-wrap-tax` FLOAT,
`item-promotion-discount` FLOAT,
`ship-promotion-discount` FLOAT,
`ship-city` VARCHAR(40),
`ship-state` VARCHAR(40),
`ship-postal-code` VARCHAR(40),
`ship-country` VARCHAR(40),
`promotion-ids` VARCHAR(50),
`cpf` VARCHAR(40),
`is-business-order` BOOL,
`purchase-order-number` VARCHAR(50),
`price-designation` VARCHAR(40),
`signature-confirmation-recommended` BOOL,
`ReportDate` DATE not null,
`timezone` VARCHAR(20) not null
);
"""