keepa类型解析增加翻译

Pyspark_job/dim/dim_keepa_asin_info.py

@@ -5,6 +5,7 @@ description: Keepa 数据聚合 — 一站式 Hive → Hive + Doris
                1) 读 Hive ods_keepa_asin_detail 当日分区，解析 last_detail JSON 各字段
                   派生 keepa_launch_time = min(listed_since, tracking_since) 转 yyyy-MM-dd HH:mm:ss
                   weight 字段已废弃置 NULL（Doris 端不再保留）
+                  keepa_type / keepa_item_type 增量翻译为中文存入 *_cn（已翻译过的值复用，不重复调接口）
                2) 与 Hive 历史 dim_keepa_asin_info union 按 asin 去重保留 updated_time 最新
                3) 写入 Hive dim_keepa_asin_info（当日分区）+ 删除所有 date_info < 今日的历史分区
                4) 当日新数据（不含历史）写入 Doris dwd.dwd_keepa_asin_detail
@@ -13,6 +14,8 @@ description: Keepa 数据聚合 — 一站式 Hive → Hive + Doris
 """
 import os
 import sys
+import time
+import requests
 
 sys.path.append(os.path.dirname(sys.path[0]))
 
@@ -23,6 +26,8 @@ from utils.DorisHelper import DorisHelper
 from utils.hdfs_utils import HdfsUtils
 from datetime import datetime, timedelta
 
+# 谷歌翻译中转服务（ct_google_translate.py，部署在可访问外网的容器中）
+TRANSLATE_SERVER = "http://192.168.10.249:30101"
 
 class DimKeepaAsinInfo(object):
 
@@ -40,12 +45,39 @@ class DimKeepaAsinInfo(object):
         self.doris_db = "dwd"
         self.doris_table = "dwd_keepa_asin_detail"
         self.df_to_doris = self.spark.sql(f"select 1+1;")
+        self.translate_cols = ['keepa_type', 'keepa_item_type']
 
     def run(self):
         self.read_data()
         self.handle_data()
         self.save_data()
 
+    @staticmethod
+    def translate_words(words, batch_size=100, retry=3):
+        """调用翻译中转服务批量英译中，返回 {原词: 译文}；服务不可用时跳过，不影响主流程"""
+        try:
+            requests.get(f"{TRANSLATE_SERVER}/health", timeout=60).raise_for_status()
+        except Exception as e:
+            print(f"翻译服务不可用，跳过本次翻译：{e}")
+            return {}
+        result = {}
+        for i in range(0, len(words), batch_size):
+            batch = words[i:i + batch_size]
+            for n in range(retry):
+                try:
+                    resp = requests.post(f"{TRANSLATE_SERVER}/translate", timeout=180,
+                                         json={"words": [w.replace('-', ' ') for w in batch]})
+                    resp.raise_for_status()
+                    trans = resp.json()["result"]
+                    assert len(trans) == len(batch), f"译文数量{len(trans)}与原词数量{len(batch)}不一致"
+                    result.update({w: t for w, t in zip(batch, trans) if t})
+                    break
+                except Exception as e:
+                    print(f"翻译失败第{n + 1}次: {e}")
+                    time.sleep(3)
+            print(f"翻译进度：{min(i + batch_size, len(words))}/{len(words)}")
+        return result
+
     def read_data(self):
         sql = f"""
             select asin, last_detail, update_at as updated_time from ods_keepa_asin_detail 
@@ -97,25 +129,16 @@ class DimKeepaAsinInfo(object):
             F.col('product_type'),
             F.col('title'),
             F.col('keepa_type'),
-            F.col('keepa_item_type')
+            F.col('keepa_item_type'),
+            F.lit(None).cast('string').alias('keepa_type_cn'),
+            F.lit(None).cast('string').alias('keepa_item_type_cn')
         ).filter(
             # 过滤脏数据：productType in (3,4,5) 且 title 为空的异常数据不做保留
             ~(F.col("product_type").isin(3, 4, 5) & F.col("title").isNull())
         ).cache()
-        # 写入 Doris 需带 site_name 分区字段，并把 keepa_launch_time / updated_time 转为 DATETIME（DDL 类型已改）
-        # weight 字段在 Doris dwd_keepa_asin_detail 已废弃，select 不带
-        self.df_to_doris = self.df_keepa_asin.select(
-            F.lit(self.site_name).alias('site_name'),
-            'asin', 'package_length', 'package_width', 'package_height', 'package_weight', 'item_weight',
-            'listed_since', 'release_date', 'tracking_since',
-            F.to_timestamp(F.col('keepa_launch_time')).alias('keepa_launch_time'),
-            F.to_timestamp(F.col('updated_time')).alias('updated_time'),
-            'keepa_type', 'keepa_item_type'
-        )
-
         # 读取历史数据
         sql = f"""
-            select asin, package_length, package_width, package_height, package_weight, item_weight, weight, keepa_launch_time, updated_time, listed_since, release_date, tracking_since, product_type, title, keepa_type, keepa_item_type
+            select asin, package_length, package_width, package_height, package_weight, item_weight, weight, keepa_launch_time, updated_time, listed_since, release_date, tracking_since, product_type, title, keepa_type, keepa_item_type, keepa_type_cn, keepa_item_type_cn
             from dim_keepa_asin_info where site_name = '{self.site_name}';
         """
         self.df_keepa_asin_history = self.spark.sql(sqlQuery=sql).repartition(40, 'asin').cache()
@@ -128,7 +151,39 @@ class DimKeepaAsinInfo(object):
         )
         self.df_save = self.df_save.withColumn(
             'rank', F.row_number().over(window=window)
-        ).filter('rank = 1').drop('rank')
+        ).filter('rank = 1').drop('rank').cache()
+
+        # 增量翻译：已有译文的值复用，仅对新出现的非空值调接口
+        cols_order = self.df_save.columns
+        df_today = self.df_keepa_asin
+        for col in self.translate_cols:
+            col_cn = f'{col}_cn'
+            df_valid = self.df_save.filter(F.col(col).isNotNull() & (F.trim(F.col(col)) != ''))
+            # 字典收集到 driver 再重建，切断与 Hive 历史分区的血缘（写 Doris 时历史分区已删除）
+            dict_rows = {r[0]: r[1] for r in df_valid.filter(F.col(col_cn).isNotNull() & (F.col(col_cn) != '')).select(
+                col, col_cn
+            ).dropDuplicates([col]).collect()}
+            words = [r[0] for r in df_valid.select(col).distinct().collect() if r[0] not in dict_rows]
+            print(f"{col} 已有译文数量：{len(dict_rows)}，待翻译数量：{len(words)}")
+            if words:
+                trans_map = self.translate_words(words)
+                print(f"{col} 翻译成功数量：{len(trans_map)}")
+                dict_rows.update(trans_map)
+            df_dict = F.broadcast(self.spark.createDataFrame(list(dict_rows.items()), f"{col} string, {col_cn} string"))
+            self.df_save = self.df_save.drop(col_cn).join(df_dict, on=col, how='left')
+            df_today = df_today.drop(col_cn).join(df_dict, on=col, how='left')
+        self.df_save = self.df_save.select(*cols_order)
+
+        # 写入 Doris 需带 site_name 分区字段，并把 keepa_launch_time / updated_time 转为 DATETIME（DDL 类型已改）
+        # weight 字段在 Doris dwd_keepa_asin_detail 已废弃，select 不带
+        self.df_to_doris = df_today.select(
+            F.lit(self.site_name).alias('site_name'),
+            'asin', 'package_length', 'package_width', 'package_height', 'package_weight', 'item_weight',
+            'listed_since', 'release_date', 'tracking_since',
+            F.to_timestamp(F.col('keepa_launch_time')).alias('keepa_launch_time'),
+            F.to_timestamp(F.col('updated_time')).alias('updated_time'),
+            'keepa_type', 'keepa_item_type', 'keepa_type_cn', 'keepa_item_type_cn'
+        )
 
     def save_data(self):
         self.df_save = self.df_save.withColumn(
@@ -172,7 +227,8 @@ class DimKeepaAsinInfo(object):
         print("往doris存储最新keepa详情信息：")
         doris_table_columns = """
         site_name, asin, package_length, package_width, package_height, package_weight, item_weight,
-        listed_since, release_date, tracking_since, keepa_launch_time, updated_time, keepa_type, keepa_item_type
+        listed_since, release_date, tracking_since, keepa_launch_time, updated_time, keepa_type, keepa_item_type,
+        keepa_type_cn, keepa_item_type_cn
         """
         DorisHelper.spark_export_with_columns(
             df_save=self.df_to_doris,