add

injection/data_sync/tasks.py

@@ -25,27 +25,27 @@ exec_count = 0
 @shared_task
 def write_to_es(es_type, pk_list, use_batch_query_set=False):
     try:
-        # pk_list = list(frozenset(pk_list))
-        #
-        # if es_type == "register_user_tag":
-        #     RegisterUserTag.get_register_user_tag(pk_list)
-        # elif es_type == "attention_user_tag":
-        #     RegisterUserTag.get_user_attention_tag(pk_list)
-        # else:
-        #     type_info_map = get_type_info_map()
-        #     type_info = type_info_map[es_type]
-        #
-        #     logging.info("consume es_type:%s" % str(es_type))
-        #     type_info.insert_table_by_pk_list(
-        #         sub_index_name=es_type,
-        #         pk_list=pk_list,
-        #         use_batch_query_set=use_batch_query_set,
-        #         es=ESPerform.get_cli()
-        #     )
+        pk_list = list(frozenset(pk_list))
+
+        if es_type == "register_user_tag":
+            RegisterUserTag.get_register_user_tag(pk_list)
+        elif es_type == "attention_user_tag":
+            RegisterUserTag.get_user_attention_tag(pk_list)
+        else:
+            type_info_map = get_type_info_map()
+            type_info = type_info_map[es_type]
+
+            logging.info("consume es_type:%s" % str(es_type))
+            type_info.insert_table_by_pk_list(
+                sub_index_name=es_type,
+                pk_list=pk_list,
+                use_batch_query_set=use_batch_query_set,
+                es=ESPerform.get_cli()
+            )
 
         # if es_type == 'topic':
-        auto_click(pk_list)
-        auto_reply(pk_list)
+            auto_click(pk_list)
+            auto_reply(pk_list)
 
 
     except:
@@ -168,8 +168,6 @@ def auto_click(pk_list):
                         sleep_time = 50
                         timer = threading.Timer(sleep_time, fun_timer)
                         timer.start()
-
-
                 sleep_time = random.randint(300, 540)
                 logging.info("----------1-----------")
                 timer = threading.Timer(10, fun_timer)  # 首次启动