'内部详情have同步到pg15'

Pyspark_job/sqoop_export/export_self_asin_details.py

+#!/usr/bin/env python3
+"""Hive 导出到 PostgreSQL 15 —— 单站点
+
+流程：
+  从 Hive 表 ods_self_asin_detail 用 sqoop 导出到 PG15 归档表 {site}_self_asin_detail_{年}
+
+用法：
+    python export_hive_to_pg15.py <site_name> <date_type> <date_info>
+      site_name : 站点，如 us / uk / it / de / es / fr
+      date_type : week / 4_week / month / quarter / day
+      date_info : 数据日期分区，如 2026-07-08
+"""
+import os
+import sys
+import time
+
+sys.path.append(os.path.dirname(sys.path[0]))
+from utils.secure_db_client import get_remote_engine
+
+# Hive 源表
+HIVE_TABLE = 'ods_self_asin_detail'
+# 导出列（对应 ods_self_asin_detail / PG 目标表的字段）
+COLS = [
+    'asin', 'img_url', 'title', 'title_len', 'price', 'rating', 'total_comments',
+    'buy_box_seller_type', 'page_inventory', 'category', 'volume', 'weight', 'rank',
+    'launch_time', 'video_url', 'add_url', 'material', 'img_num', 'img_type', 'qa_num',
+    'brand', 'ac_name', 'node_id', 'sp_num', 'mpn', 'online_time', 'describe',
+    'one_star', 'two_star', 'three_star', 'four_star', 'five_star', 'low_star',
+    'asin_type', 'is_coupon', 'search_category', 'weight_str', 'account_name',
+    'other_seller_name', 'bsr_date_info', 'account_id', 'package_quantity',
+    'pattern_name', 'together_asin', 'activity_type', 'one_two_val', 'three_four_val',
+    'five_six_val', 'eight_val', 'product_description', 'follow_sellers', 'buy_sales',
+    'image_view', 'product_json', 'productdetail_json', 'review_ai_text',
+    'review_label_json', 'lob_asin_json', 'sp_initial_seen_asins_json',
+    'sp_4stars_initial_seen_asins_json', 'sp_delivery_initial_seen_asins_json',
+    'compare_similar_asin_json', 'customer_reviews_json', 'together_asin_json',
+    'min_match_asin_json', 'seller_json', 'returns', 'created_at', 'updated_at',
+    'result_list_json', 'variat_list', 'bundle_asin_component_json', 'cart_type',
+    'site', 'comment_json', 'date_info',
+]
+
+
+def export_site(site_name, date_type, date_info):
+    """把某站点某天的数据从 Hive(ods_self_asin_detail) 导出到 PG15 归档年表。"""
+    engine = get_remote_engine(
+        site_name=site_name,        # -> database "selection"
+        db_type="postgresql_15",    # -> 服务端 PostgreSQL 15
+    )
+
+    # Hive 分区过滤（按实际分区字段确认：按 site_name + date_type + date_info 分区）
+    partitions = {
+        'site_name': site_name,
+        'date_type': date_type,
+        'date_info': date_info,
+    }
+    year = time.strftime("%Y", time.localtime())
+    import_table = f'{site_name}_self_asin_detail_{year}'
+
+    # 先删该 date_info 的数据，再导入（幂等：重复跑同一天不会产生重复行）
+    del_sql = f"DELETE FROM {import_table} WHERE date_info = '{date_info}'"
+    print(f"[{site_name}] 删除 date_info={date_info} 数据: {del_sql}")
+    engine.execute(del_sql)
+
+    print(f"[{site_name}] hive={HIVE_TABLE} -> pg15={import_table}, partitions={partitions}")
+    engine.sqoop_raw_export(
+        hive_table=HIVE_TABLE,
+        import_table=import_table,
+        partitions=partitions,
+        m=1,
+        cols=','.join(COLS),
+    )
+    print(f"[{site_name}] 导出完成")
+
+
+def main():
+    if len(sys.argv) < 4:
+        print(__doc__)
+        print("参数不足：需要 <site_name> <date_type> <date_info>")
+        sys.exit(1)
+
+    site_name = sys.argv[1]  # 站点，如 us
+    date_type = sys.argv[2]  # week / 4_week / month / quarter / day
+    date_info = sys.argv[3]  # 数据日期分区，如 2026-07-08
+
+    export_site(site_name, date_type, date_info)
+
+
+if __name__ == '__main__':
+    main()