asin详情同步-兼容历史字段缺失

Pyspark_job/sqoop_import/ods_asin_detail.py

@@ -71,85 +71,109 @@ if __name__ == '__main__':
 
     check_syn(engine=engine)
 
+    # 源表(PG爬虫表)这7个字段是分阶段才有的，早于对应月份的源表没有这几列，
+    # 直接select会报"列不存在"，历史回溯同步时按date_info降级为对应类型的NULL占位
+    has_2025_cols = date_info >= '2025-01'  # bundles_this_asins_json/video_m3u8_url/result_list_json/bundle_asin_component_json/review_json_list/fbm_delivery_price
+    has_2026_cols = date_info >= '2026-01'  # amazon_label/landing_asin/sub_title
+
+    bundles_this_asins_json_expr = ("REPLACE(REPLACE(REPLACE(bundles_this_asins_json, E'\n',' '), E'\r',' '), E'\t',' ') AS bundles_this_asins_json"
+                                     if has_2025_cols else "CAST(NULL AS VARCHAR) AS bundles_this_asins_json")
+    video_m3u8_url_expr = ("REPLACE(REPLACE(REPLACE(video_m3u8_url, E'\n',' '), E'\r',' '), E'\t',' ') AS video_m3u8_url"
+                            if has_2025_cols else "CAST(NULL AS VARCHAR) AS video_m3u8_url")
+    result_list_json_expr = ("REPLACE(REPLACE(REPLACE(result_list_json, E'\n',' '), E'\r',' '), E'\t',' ') AS result_list_json"
+                              if has_2025_cols else "CAST(NULL AS VARCHAR) AS result_list_json")
+    bundle_asin_component_json_expr = ("REPLACE(REPLACE(REPLACE(bundle_asin_component_json, E'\n',' '), E'\r',' '), E'\t',' ') AS bundle_asin_component_json"
+                                       if has_2025_cols else "CAST(NULL AS VARCHAR) AS bundle_asin_component_json")
+    review_json_list_expr = ("REPLACE(REPLACE(REPLACE(review_json_list, E'\n',' '), E'\r',' '), E'\t',' ') AS review_json_list"
+                              if has_2025_cols else "CAST(NULL AS VARCHAR) AS review_json_list")
+    fbm_delivery_price_expr = "fbm_delivery_price" if has_2025_cols else "CAST(NULL AS NUMERIC) AS fbm_delivery_price"
+
+    amazon_label_expr = ("REPLACE(REPLACE(REPLACE(amazon_label, E'\n',' '), E'\r',' '), E'\t',' ') AS amazon_label"
+                         if has_2026_cols else "CAST(NULL AS VARCHAR) AS amazon_label")
+    landing_asin_expr = ("REPLACE(REPLACE(REPLACE(landing_asin, E'\n',' '), E'\r',' '), E'\t',' ') AS landing_asin"
+                         if has_2026_cols else "CAST(NULL AS VARCHAR) AS landing_asin")
+    sub_title_expr = ("REPLACE(REPLACE(REPLACE(sub_title, E'\n',' '), E'\r',' '), E'\t',' ') AS sub_title"
+                      if has_2026_cols else "CAST(NULL AS VARCHAR) AS sub_title")
+
     query = f"""
-    SELECT
-  id,
-  asin,
-  REPLACE(REPLACE(REPLACE(img_url, E'\n',' '), E'\r',' '), E'\t',' ') AS img_url,
-  REPLACE(REPLACE(REPLACE(title, E'\n',' '), E'\r',' '), E'\t',' ') AS title,
-  title_len,
-  price,
-  rating,
-  total_comments,
-  buy_box_seller_type,
-  page_inventory,
-  REPLACE(REPLACE(REPLACE(category, E'\n',' '), E'\r',' '), E'\t',' ') AS category,
-  REPLACE(REPLACE(REPLACE(volume, E'\n',' '), E'\r',' '), E'\t',' ') AS volume,
-  weight,
-  rank,
-  launch_time,
-  created_time AS created_at, updated_time AS updated_at,
-  category_state,
-  img_num,
-  REPLACE(REPLACE(REPLACE(img_type, E'\n',' '), E'\r',' '), E'\t',' ') AS img_type,
-  REPLACE(REPLACE(REPLACE(activity_type, E'\n',' '), E'\r',' '), E'\t',' ') AS activity_type,
-  one_two_val,
-  three_four_val,
-  five_six_val,
-  eight_val,
-  qa_num,
-  one_star,
-  two_star,
-  three_star,
-  four_star,
-  five_star,
-  low_star,
-  REPLACE(REPLACE(REPLACE(together_asin, E'\n',' '), E'\r',' '), E'\t',' ') AS together_asin,
-  REPLACE(REPLACE(REPLACE(brand, E'\n',' '), E'\r',' '), E'\t',' ') AS brand,
-  REPLACE(REPLACE(REPLACE(ac_name, E'\n',' '), E'\r',' '), E'\t',' ') AS ac_name,
-  REPLACE(REPLACE(REPLACE(material, E'\n',' '), E'\r',' '), E'\t',' ') AS material,
-  REPLACE(REPLACE(REPLACE(node_id, E'\n',' '), E'\r',' '), E'\t',' ') AS node_id,
-  data_type,
-  REPLACE(REPLACE(REPLACE(sp_num, E'\n',' '), E'\r',' '), E'\t',' ') AS sp_num,
-  REPLACE(REPLACE(REPLACE(describe, E'\n',' '), E'\r',' '), E'\t',' ') AS describe,
-  REPLACE(REPLACE(REPLACE(weight_str, E'\n',' '), E'\r',' '), E'\t',' ') AS weight_str,
-  REPLACE(REPLACE(REPLACE(package_quantity, E'\n',' '), E'\r',' '), E'\t',' ') AS package_quantity,
-  REPLACE(REPLACE(REPLACE(pattern_name, E'\n',' '), E'\r',' '), E'\t',' ') AS pattern_name,
-  follow_sellers,
-  REPLACE(REPLACE(REPLACE(product_description, E'\n',' '), E'\r',' '), E'\t',' ') AS product_description,
-  REPLACE(REPLACE(REPLACE(buy_sales, E'\n',' '), E'\r',' '), E'\t',' ') AS buy_sales,
-  image_view,
-  spider_int,
-  REPLACE(REPLACE(REPLACE(lob_asin_json, E'\n',' '), E'\r',' '), E'\t',' ') AS lob_asin_json,
-  REPLACE(REPLACE(REPLACE(seller_json, E'\n',' '), E'\r',' '), E'\t',' ') AS seller_json,
-  REPLACE(REPLACE(REPLACE(customer_reviews_json, E'\n',' '), E'\r',' '), E'\t',' ') AS customer_reviews_json,
-  REPLACE(REPLACE(REPLACE(product_json, E'\n',' '), E'\r',' '), E'\t',' ') AS product_json,
-  REPLACE(REPLACE(REPLACE(product_detail_json, E'\n',' '), E'\r',' '), E'\t',' ') AS product_detail_json,
-  REPLACE(REPLACE(REPLACE(review_ai_text, E'\n',' '), E'\r',' '), E'\t',' ') AS review_ai_text,
-  REPLACE(REPLACE(REPLACE(review_label_json, E'\n',' '), E'\r',' '), E'\t',' ') AS review_label_json,
-  REPLACE(REPLACE(REPLACE(sp_initial_seen_asins_json, E'\n',' '), E'\r',' '), E'\t',' ') AS sp_initial_seen_asins_json,
-  REPLACE(REPLACE(REPLACE(sp_4stars_initial_seen_asins_json, E'\n',' '), E'\r',' '), E'\t',' ') AS sp_4stars_initial_seen_asins_json,
-  REPLACE(REPLACE(REPLACE(sp_delivery_initial_seen_asins_json, E'\n',' '), E'\r',' '), E'\t',' ') AS sp_delivery_initial_seen_asins_json,
-  REPLACE(REPLACE(REPLACE(compare_similar_asin_json, E'\n',' '), E'\r',' '), E'\t',' ') AS compare_similar_asin_json,
-  REPLACE(REPLACE(REPLACE(together_asin_json, E'\n',' '), E'\r',' '), E'\t',' ') AS together_asin_json,
-  REPLACE(REPLACE(REPLACE(min_match_asin_json, E'\n',' '), E'\r',' '), E'\t',' ') AS min_match_asin_json,
-  variat_num,
-  REPLACE(REPLACE(REPLACE(current_asin, E'\n',' '), E'\r',' '), E'\t',' ') AS current_asin,
-  REPLACE(REPLACE(REPLACE(img_list, E'\n',' '), E'\r',' '), E'\t',' ') AS img_list,
-  REPLACE(REPLACE(REPLACE(variat_list, E'\n',' '), E'\r',' '), E'\t',' ') AS variat_list,
-  REPLACE(REPLACE(REPLACE(parent_asin, E'\n',' '), E'\r',' '), E'\t',' ') AS parent_asin,
-  REPLACE(REPLACE(REPLACE(bundles_this_asins_json, E'\n',' '), E'\r',' '), E'\t',' ') AS bundles_this_asins_json,
-  REPLACE(REPLACE(REPLACE(video_m3u8_url, E'\n',' '), E'\r',' '), E'\t',' ') AS video_m3u8_url,
-  REPLACE(REPLACE(REPLACE(result_list_json, E'\n',' '), E'\r',' '), E'\t',' ') AS result_list_json,
-  REPLACE(REPLACE(REPLACE(bundle_asin_component_json, E'\n',' '), E'\r',' '), E'\t',' ') AS bundle_asin_component_json,
-  REPLACE(REPLACE(REPLACE(review_json_list, E'\n',' '), E'\r',' '), E'\t',' ') AS review_json_list,
-  fbm_delivery_price,
-  REPLACE(REPLACE(REPLACE(amazon_label, E'\n',' '), E'\r',' '), E'\t',' ') AS amazon_label,
-  REPLACE(REPLACE(REPLACE(landing_asin, E'\n',' '), E'\r',' '), E'\t',' ') AS landing_asin,
-  REPLACE(REPLACE(REPLACE(sub_title, E'\n',' '), E'\r',' '), E'\t',' ') AS sub_title
-FROM {import_table}
-WHERE  1=1 AND \$CONDITIONS
-"""
+        SELECT
+        id,
+        asin,
+        REPLACE(REPLACE(REPLACE(img_url, E'\n',' '), E'\r',' '), E'\t',' ') AS img_url,
+        REPLACE(REPLACE(REPLACE(title, E'\n',' '), E'\r',' '), E'\t',' ') AS title,
+        title_len,
+        price,
+        rating,
+        total_comments,
+        buy_box_seller_type,
+        page_inventory,
+        REPLACE(REPLACE(REPLACE(category, E'\n',' '), E'\r',' '), E'\t',' ') AS category,
+        REPLACE(REPLACE(REPLACE(volume, E'\n',' '), E'\r',' '), E'\t',' ') AS volume,
+        weight,
+        rank,
+        launch_time,
+        created_time AS created_at, updated_time AS updated_at,
+        category_state,
+        img_num,
+        REPLACE(REPLACE(REPLACE(img_type, E'\n',' '), E'\r',' '), E'\t',' ') AS img_type,
+        REPLACE(REPLACE(REPLACE(activity_type, E'\n',' '), E'\r',' '), E'\t',' ') AS activity_type,
+        one_two_val,
+        three_four_val,
+        five_six_val,
+        eight_val,
+        qa_num,
+        one_star,
+        two_star,
+        three_star,
+        four_star,
+        five_star,
+        low_star,
+        REPLACE(REPLACE(REPLACE(together_asin, E'\n',' '), E'\r',' '), E'\t',' ') AS together_asin,
+        REPLACE(REPLACE(REPLACE(brand, E'\n',' '), E'\r',' '), E'\t',' ') AS brand,
+        REPLACE(REPLACE(REPLACE(ac_name, E'\n',' '), E'\r',' '), E'\t',' ') AS ac_name,
+        REPLACE(REPLACE(REPLACE(material, E'\n',' '), E'\r',' '), E'\t',' ') AS material,
+        REPLACE(REPLACE(REPLACE(node_id, E'\n',' '), E'\r',' '), E'\t',' ') AS node_id,
+        data_type,
+        REPLACE(REPLACE(REPLACE(sp_num, E'\n',' '), E'\r',' '), E'\t',' ') AS sp_num,
+        REPLACE(REPLACE(REPLACE(describe, E'\n',' '), E'\r',' '), E'\t',' ') AS describe,
+        REPLACE(REPLACE(REPLACE(weight_str, E'\n',' '), E'\r',' '), E'\t',' ') AS weight_str,
+        REPLACE(REPLACE(REPLACE(package_quantity, E'\n',' '), E'\r',' '), E'\t',' ') AS package_quantity,
+        REPLACE(REPLACE(REPLACE(pattern_name, E'\n',' '), E'\r',' '), E'\t',' ') AS pattern_name,
+        follow_sellers,
+        REPLACE(REPLACE(REPLACE(product_description, E'\n',' '), E'\r',' '), E'\t',' ') AS product_description,
+        REPLACE(REPLACE(REPLACE(buy_sales, E'\n',' '), E'\r',' '), E'\t',' ') AS buy_sales,
+        image_view,
+        spider_int,
+        REPLACE(REPLACE(REPLACE(lob_asin_json, E'\n',' '), E'\r',' '), E'\t',' ') AS lob_asin_json,
+        REPLACE(REPLACE(REPLACE(seller_json, E'\n',' '), E'\r',' '), E'\t',' ') AS seller_json,
+        REPLACE(REPLACE(REPLACE(customer_reviews_json, E'\n',' '), E'\r',' '), E'\t',' ') AS customer_reviews_json,
+        REPLACE(REPLACE(REPLACE(product_json, E'\n',' '), E'\r',' '), E'\t',' ') AS product_json,
+        REPLACE(REPLACE(REPLACE(product_detail_json, E'\n',' '), E'\r',' '), E'\t',' ') AS product_detail_json,
+        REPLACE(REPLACE(REPLACE(review_ai_text, E'\n',' '), E'\r',' '), E'\t',' ') AS review_ai_text,
+        REPLACE(REPLACE(REPLACE(review_label_json, E'\n',' '), E'\r',' '), E'\t',' ') AS review_label_json,
+        REPLACE(REPLACE(REPLACE(sp_initial_seen_asins_json, E'\n',' '), E'\r',' '), E'\t',' ') AS sp_initial_seen_asins_json,
+        REPLACE(REPLACE(REPLACE(sp_4stars_initial_seen_asins_json, E'\n',' '), E'\r',' '), E'\t',' ') AS sp_4stars_initial_seen_asins_json,
+        REPLACE(REPLACE(REPLACE(sp_delivery_initial_seen_asins_json, E'\n',' '), E'\r',' '), E'\t',' ') AS sp_delivery_initial_seen_asins_json,
+        REPLACE(REPLACE(REPLACE(compare_similar_asin_json, E'\n',' '), E'\r',' '), E'\t',' ') AS compare_similar_asin_json,
+        REPLACE(REPLACE(REPLACE(together_asin_json, E'\n',' '), E'\r',' '), E'\t',' ') AS together_asin_json,
+        REPLACE(REPLACE(REPLACE(min_match_asin_json, E'\n',' '), E'\r',' '), E'\t',' ') AS min_match_asin_json,
+        variat_num,
+        REPLACE(REPLACE(REPLACE(current_asin, E'\n',' '), E'\r',' '), E'\t',' ') AS current_asin,
+        REPLACE(REPLACE(REPLACE(img_list, E'\n',' '), E'\r',' '), E'\t',' ') AS img_list,
+        REPLACE(REPLACE(REPLACE(variat_list, E'\n',' '), E'\r',' '), E'\t',' ') AS variat_list,
+        REPLACE(REPLACE(REPLACE(parent_asin, E'\n',' '), E'\r',' '), E'\t',' ') AS parent_asin,
+        {bundles_this_asins_json_expr},
+        {video_m3u8_url_expr},
+        {result_list_json_expr},
+        {bundle_asin_component_json_expr},
+        {review_json_list_expr},
+        {fbm_delivery_price_expr},
+        {amazon_label_expr},
+        {landing_asin_expr},
+        {sub_title_expr}
+        FROM {import_table}
+        WHERE  1=1 AND \$CONDITIONS
+    """
 
     engine.sqoop_raw_import(
         # query=f"SELECT {cols} FROM {import_table} WHERE 1=1 and \$CONDITIONS",