流量选品月-告警品牌数据源切换

Pyspark_job/dim/dim_asin_detail.py

@@ -27,6 +27,7 @@ from utils.common_util import CommonUtil, DateTypes
 from yswg_utils.common_udf import udf_new_asin_flag
 from utils.hdfs_utils import HdfsUtils
 from utils.db_util import DBUtil
+from utils.DorisHelper import DorisHelper
 from datetime import datetime
 from pyspark.storagelevel import StorageLevel
 from yswg_utils.common_udf import udf_get_package_quantity_with_flag as udf_get_package_quantity, udf_parse_seller_json, udf_parse_amazon_orders
@@ -444,18 +445,12 @@ class DimAsinDetail(object):
             .withColumn("asin_quantity_variation_type", F.when((F.lower(F.col("asin_size").cast("string")) != 'null') &
                                (F.lower(F.col("asin_size").cast("string")) != 'none') &
                                (F.lower(F.col("asin_size").cast("string")).contains('quantity')), 1).otherwise(F.lit(0)))
-        if self.site_name == 'us':
-            pg_sql = f"""
-            select asin_brand_name, 1 as asin_is_alarm from (select lower(trim(brand_name)) as asin_brand_name from brand_alert_erp where brand_name is not null) t group by asin_brand_name"""
-            db_type = "postgresql_cluster"
-            con_info = DBUtil.get_connection_info(db_type=db_type, site_name=self.site_name)
-            if con_info is not None:
-                df_alarm_brand = SparkUtil.read_jdbc_query(
-                    session=self.spark, url=con_info['url'], pwd=con_info['pwd'], username=con_info['username'], query=pg_sql)
-                df_alarm_brand = df_alarm_brand.repartition(100)
-                self.df_asin_detail = self.df_asin_detail.join(df_alarm_brand, on=['asin_brand_name'], how='left')
-        else:
-            self.df_asin_detail = self.df_asin_detail.withColumn("asin_is_alarm", F.lit(0))
+        # 报警品牌：改为从 Doris selection.brand_alert_erp 读取（对齐最近30天 us_flow_asin_30day 的做法），
+        # 该表不区分站点(全站点共用黑名单)，us/uk/de 统一处理，不再像之前一样只处理us站点、uk/de硬编码为0
+        doris_sql = "SELECT DISTINCT LOWER(TRIM(brand_name)) AS asin_brand_name FROM selection.brand_alert_erp WHERE brand_name IS NOT NULL"
+        df_alarm_brand = DorisHelper.spark_import_with_sql(self.spark, doris_sql).withColumn("asin_is_alarm", F.lit(1))
+        df_alarm_brand = df_alarm_brand.repartition(100)
+        self.df_asin_detail = self.df_asin_detail.join(df_alarm_brand, on=['asin_brand_name'], how='left')
         self.df_asin_detail = self.df_asin_detail.na.fill({"asin_is_alarm": 0})
         # 处理是否内部asin信息
         self.df_asin_detail = self.df_asin_detail.join(self.df_self_asin, on=['asin'], how='left')