实时任务流程控制表

Pyspark_job/utils/templates.py

@@ -700,17 +700,22 @@ class Templates(object):
     def modify_kafka_state(self):
         # 正式的实时消费才修改状态
         script_name = sys.argv[0].split("/")[-1].split(".")[0]
-        if self.consumer_type == 'latest' and self.test_flag == 'normal' and script_name in ['kafka_flow_asin_detail', 'kafka_asin_detail', 'kafka_rank_asin_detail']:
-            if script_name == 'kafka_flow_asin_detail':
-                # kafka_field = 'kafka_flow_state'
+        if self.consumer_type == 'latest' and self.test_flag == 'normal' and script_name in ['kafka_flow_asin_detail_to_doris', 'kafka_asin_detail']:
+            # 已注释：kafka_flow_asin_detail / kafka_rank_asin_detail 旧脚本统一由 kafka_flow_asin_detail_to_doris 接管（天/月实时任务）
+            # if script_name == 'kafka_flow_asin_detail':
+            #     # kafka_field = 'kafka_flow_state'
+            #     kafka_field = 'kafka_state'
+            #     wx_users = ['chenyuanjie', 'pengyanbing']
+            #     wx_msg = f"站点： {self.site_name} 日期类型： {self.date_type} {self.date_info} asin详情实时消费数据到es准备工作已完成，可以开启详情爬取！"
+            # elif script_name == 'kafka_rank_asin_detail':
+            #     # kafka_field = 'kafka_flow_state'
+            #     kafka_field = 'kafka_state'
+            #     wx_users = ['chenyuanjie', 'pengyanbing']
+            #     wx_msg = f"站点： {self.site_name} 日期类型：{self.date_type}, {self.date_info} 榜单asin详情实时消费数据到es准备工作已完成，可以开启详情爬取！"
+            if script_name == 'kafka_flow_asin_detail_to_doris':
                 kafka_field = 'kafka_state'
                 wx_users = ['chenyuanjie', 'pengyanbing']
-                wx_msg = f"站点： {self.site_name} 日期类型： {self.date_type} {self.date_info} asin详情实时消费数据到es准备工作已完成，可以开启详情爬取！"
-            elif script_name == 'kafka_rank_asin_detail':
-                # kafka_field = 'kafka_flow_state'
-                kafka_field = 'kafka_state'
-                wx_users = ['chenyuanjie', 'pengyanbing']
-                wx_msg = f"站点： {self.site_name} 日期类型：{self.date_type}, {self.date_info} 榜单asin详情实时消费数据到es准备工作已完成，可以开启详情爬取！"
+                wx_msg = f"站点： {self.site_name} 日期类型： {self.date_type} {self.date_info} asin详情实时消费数据到doris准备工作已完成，可以开启详情爬取！"
             elif script_name == 'kafka_asin_detail':
                 kafka_field = 'kafka_state'
                 wx_users = ['fangxingjun', 'pengyanbing']