导出pg集群表待调用keepa数据

Pyspark_job/dwt/dwt_asin_sync.py

@@ -200,10 +200,10 @@ class DwtAsinSync(Templates):
         df_keepa = self.spark.sql(f"""
             select asin from dim_keepa_asin_info
             where site_name = '{self.site_name}'
-                      and package_length >= 0
-                      and package_width  >= 0
-                      and package_height >= 0
-                      and weight         >= 0
+              and package_length > 0
+              and package_width > 0
+              and package_height > 0
+              and (package_weight > 0 or item_weight > 0)
         """).repartition(40, 'asin')
         df = df.join(df_keepa, on='asin', how='left_anti').cache()
         print(f"排除keepa后数据量: {df.count()}")

Pyspark_job/script/export_asin_without_keepa.py

@@ -244,15 +244,15 @@ class ExportAsinWithoutKeepa(object):
             print(f"筛选后数据量: {df.count()}")
 
         # 排除 dim_keepa_asin_info 中已有有效keepa数据的ASIN
-        # 若 package_length/width/height/weight 任意一个 < 0，视为数据异常，不排除（需重新抓取）
+        # 有效定义：长/宽/高 >= 0，且 package_weight 或 item_weight 任意一个 > 0（与 export_need_profit_rate 取数规则一致）
         print("8. 排除已有keepa数据的ASIN (dim_keepa_asin_info)")
         df_keepa = self.spark.sql(f"""
             select asin from dim_keepa_asin_info
             where site_name = '{self.site_name}'
-              and package_length >= 0
-              and package_width  >= 0
-              and package_height >= 0
-              and weight         >= 0
+              and package_length > 0
+              and package_width > 0
+              and package_height > 0
+              and (package_weight > 0 or item_weight > 0)
         """).repartition(40, 'asin')
         df = df.join(df_keepa, on='asin', how='left_anti').cache()
         print(f"排除keepa后数据量: {df.count()}")