Merge branch 'yifan' of ASJ_ADS/sync_amz_data into master

sync_amz_data/public/sp_api_client.py

@@ -210,19 +210,35 @@ class SpApiRequest:
         print("duration:",end-start)
         return decom_df
 
-    def get_mainImage_url(self, sku):
+    def get_listing_info(self, sku,seller_id):
         listingClient = ListingsItems(credentials=self.credentials, marketplace=self.marketplace)
         try:
-            r1 = listingClient.get_listings_item(sellerId=self.shopInfo['advertiser_id'], sku=sku)
-            img = r1.payload.get("summaries")[0].get("mainImage")
-            img_url = None if img is None else img.get("link")
+            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:
-            print("获取图片url过程错误重试, 错误message: ",e)
-            time.sleep(3)
-            r1 = listingClient.get_listings_item(sellerId=self.shopInfo['advertiser_id'], sku=sku)
-            img = r1.payload.get("summaries")[0].get("mainImage")
-            img_url = None if img is None else img.get("link")
-        return img_url
+            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",
@@ -275,19 +291,25 @@ class SpApiRequest:
                     return datetime(1999,12,31,0,0,0)
 
     def update_data(self,df,seller_id,country_code,conn):
+        conn = SpApiRequest.mysql_connect_auth()
         cursor = conn.cursor()
-        columns = ['listing-id','seller_id',
-                              'asin1','seller-sku','country_code',
-                              'marketplace_id','quantity','fulfillment_channel',
-                              'price','opendate','status','update_datetime','product-id','product-id-type'
-                              ]
+        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
-                                ansjer_dvadmin.seller_listings where seller_id='{seller_id}' and country_code='{country_code}'""")
+                                ansjer_dvadmin.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)
@@ -303,6 +325,7 @@ class SpApiRequest:
             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']
@@ -310,13 +333,18 @@ class SpApiRequest:
                 fulfillment_channel = temp_df['fulfillment_channel']
                 price = temp_df['price']
                 product_id = temp_df['product-id']
-                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""")
+                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,country_code,sku,listing_id,product_id))
+                    delete_list.append((seller_id,marketplace_id,sku,listing_id,product_id))
                 row += 1
             print("判断不同数据条数",len(delete_list))
             print("预计更新数据条数",len(df_data))
@@ -324,7 +352,7 @@ class SpApiRequest:
                 # print(tuple(delete_list))
                 if len(delete_list)>0:
                     query = f"""delete from ansjer_dvadmin.seller_listings 
-                           where (seller_id,country_code,sku,listing_id,product_id) in %s""" #where (seller_id,country_code) in %s"""
+                           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()
@@ -374,10 +402,11 @@ class SpApiRequest:
             df['update_datetime'] = datetime.now(pytz.UTC).date()
 
             origin_columns = ['listing-id','seller_id',
-                              'asin1','seller-sku','country_code',
+                              '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()
             cursor = conn.cursor()
             cursor.execute("""select product_id,asin from (select * from ansjer_dvadmin.seller_listings where asin is not null 
                                 and asin<>'' and product_id is not null and product_id <>'') t1 group by product_id,asin""")
@@ -386,33 +415,61 @@ class SpApiRequest:
             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):
-                if 'B0' in str(product_id):
+
+            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=='':
+                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
-                else:
-                    return asin1
+                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['asin1'] = merged_df.apply(lambda x:func_(x['asin'],x['asin1'],x['product-id']),axis=1) #x['asin'] if pd.isna(x['asin1']) or x['asin1']=='' else x['asin1']
             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)
-            # update_df.to_csv("fr.csv")
             if len(update_df)==0:
                 return '无更新数据插入'
-
+            # update_df['country_code'] = update_df['country_code'].map({"GB":"UK"})
+            conn = SpApiRequest.mysql_connect_auth()
             cursor = conn.cursor()
 
             try:
                 insertsql = """insert into
-                ansjer_dvadmin.seller_listings(listing_id,seller_id,asin,sku,country_code,marketplace_id,quantity,
+                ansjer_dvadmin.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)"""
+                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)
@@ -427,8 +484,9 @@ class SpApiRequest:
         columns_name = [i[0] for i in cursor.description]
         rel = cursor.fetchall()
         df = pd.DataFrame(rel, columns=columns_name)
-
-        for refresh_token in df['refresh_token'].to_numpy().tolist():
+        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凭证
@@ -444,9 +502,10 @@ class SpApiRequest:
             account_name = single_info['account_name'].values[0]
             if region_circle == 'NA':
                 pass
-                for marketplace in [Marketplaces.US, Marketplaces.BR, Marketplaces.CA, Marketplaces.MX]:
+                for marketplace in [Marketplaces.US, Marketplaces.BR, Marketplaces.CA,Marketplaces.MX]:
                     sp_api = SpApiRequest(aws_credentials, marketplace)
                     try:
+                        auth_conn = SpApiRequest.mysql_connect_auth()
                         sp_api.GET_FLAT_FILE_OPEN_LISTINGS_DATA(auth_conn, seller_id)
                     except Exception as e:
                         print(e)
@@ -455,15 +514,16 @@ class SpApiRequest:
                 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()
                         sp_api.GET_FLAT_FILE_OPEN_LISTINGS_DATA(auth_conn, seller_id)
                     except Exception as e:
                         print(e)
 
             else:
+                auth_conn = SpApiRequest.mysql_connect_auth()
                 marketplace = eval(f'Marketplaces.{region_circle}')
                 sp_api = SpApiRequest(aws_credentials, marketplace)
                 sp_api.GET_FLAT_FILE_OPEN_LISTINGS_DATA(auth_conn, seller_id)