流量选品-新增uk、de站点异常月销处理

Pyspark_job/es_handle/es_update_with_asin_bought_month.py

@@ -17,10 +17,12 @@ __es_passwd__ = "Selection20251#+"
 
 class EsUpdate(object):
 
-    def __init__(self):
+    def __init__(self, site_name, date_info):
+        self.site_name = site_name
+        self.date_info = date_info
         self.spark = SparkUtil.get_spark_session(f"{self.__class__.__name__}")
-
-        self.index_name = "us_st_detail_month_2025_09"
+        year, month = self.date_info.split('-')
+        self.index_name = f"{site_name}_st_detail_month_{year}_{month}"
         self.es_options = {
             "es.nodes": __es_ip__,
             "es.port": __es_port__,
@@ -45,10 +47,10 @@ class EsUpdate(object):
         self.update_es_filed()
 
     def get_update_asin(self):
-        sql = """
-            select asin from us_asin_detail_2025_not_buysales
+        sql = f"""
+            select asin from {self.site_name}_asin_detail_2025_buysales_err where date_info = '{self.date_info}'
         """
-        pg_con_info = DBUtil.get_connection_info("postgresql_14", "us")
+        pg_con_info = DBUtil.get_connection_info("postgresql_14", self.site_name)
         self.df_asin = SparkUtil.read_jdbc_query(
             session=self.spark,
             url=pg_con_info['url'],
@@ -84,5 +86,7 @@ class EsUpdate(object):
 
 
 if __name__ == "__main__":
-    handle_obj = EsUpdate()
+    site_name = sys.argv[1]
+    date_info = sys.argv[2]
+    handle_obj = EsUpdate(site_name, date_info)
     handle_obj.run()