redis sub pub py

trans2es/management/commands/trans2es_data2es_parallel.py

@@ -28,10 +28,13 @@ from trans2es.models.topic import Topic,ActionSumAboutTopic
 from search.utils.common import *
 from linucb.views.collect_data import CollectData
 from injection.data_sync.tasks import sync_user_similar_score
+import datetime
 
 from trans2es.models.tag import Tag
 from libs.cache import redis_client
 from trans2es.models.tag import TopicTag
+from libs.error import logging_exception
+from trans2es.models.portrait_stat import LikeDeviceTagStat
 
 
 
@@ -215,6 +218,42 @@ class Command(BaseCommand):
             logging.error("catch exception,err_msg:%s" % traceback.format_exc())
 
 
+    def sub_redis_new_topic_ids(self):
+        try:
+            ps = redis_client.pubsub()
+            ps.subscribe("new_topic_impression")
+            all_new_topic_impression_count_key = "all_new_topic_impression_count_key"
+            for item in ps.listen():
+                if item['type'] == 'message':
+                    new_topic_ids = json.loads(item["data"])
+                    all_new_topic_impression_count = json.loads(redis_client.get(all_new_topic_impression_count_key))
+                    insert_topic_ids = []
+                    for topic in new_topic_ids:
+                        topic = str(topic)
+                        if topic in all_new_topic_impression_count:
+                            all_new_topic_impression_count[topic] = all_new_topic_impression_count[topic] + 1
+                            if all_new_topic_impression_count[topic] > 100:
+                                insert_topic_ids.append(int(topic))
+                                all_new_topic_impression_count.pop(topic)
+                        else:
+                            all_new_topic_impression_count[topic] = 1
+                    if insert_topic_ids:
+                        insert_list = []
+                        for topic in insert_topic_ids:
+                            insert_list.append(
+                                LikeDeviceTagStat(create_time=datetime.datetime.today(),
+                                                  update_time=datetime.datetime.today(),
+                                                  topic_id=topic, is_new_topic=0, topic_ctr_30=0.0, like_rate_30=0.0))
+                        LikeDeviceTagStat.objects.using(settings.MASTER_DB_NAME).bulk_create(insert_list)
+                        logging.info("impressions count gt 100 topic ids" + str(insert_topic_ids))
+                    json_all_new_topic_impression_count = json.dumps(all_new_topic_impression_count)
+                    logging.info("all_new_topic_impression_count" + str(all_new_topic_impression_count))
+                    redis_client.set(all_new_topic_impression_count_key, json_all_new_topic_impression_count)
+        except:
+            logging_exception()
+            logging.error("redis new topic sub error!")
+
+
     def handle(self, *args, **options):
         try:
             type_name_list = get_type_info_map().keys()
@@ -241,5 +280,8 @@ class Command(BaseCommand):
             if len(options["sync_type"]) and options["sync_type"]=="tagname":
                 self.sync_tag_collecction_name_set()
 
+            if len(options["sync_type"]) and options["sync_type"] == "new_topic_sub":
+                self.sub_redis_new_topic_ids()
+
         except:
             logging.error("catch exception,err_msg:%s" % traceback.format_exc())