Merge branch 'gyz' into 'master'

lin新回答保量，lin旧回答排序(30天内的ctr+点赞率)

See merge request alpha/physical!498

libs/es.py

@@ -401,6 +401,7 @@ class ESPerform(object):
                                     {"term": {"content_level": 6}},
                                     {"term": {"is_online": True}},
                                     {"term": {"is_deleted": False}},
+                                    {"term": {"is_new_topic": False}},
                                     {"terms": {"tag_list": tag_id}}
                                 ]
                             }
@@ -414,6 +415,15 @@ class ESPerform(object):
                     "include": ["id", "user_id", "latest_reply_time"]
                 },
                 "sort": [
+                    {
+                        "_script": {
+                            "order": "desc",
+                            "script": {
+                                "inline": "doc['topic_ctr_30'].value+doc['like_rate_30'].value"
+                            },
+                            "type": "number"
+                        }
+                    },
                     {"latest_reply_time": {"order": "desc"}},
                     {"create_time_val": {"order": "desc"}},
                     {"language_type": {"order": "asc"}},
@@ -450,6 +460,65 @@ class ESPerform(object):
             logging.info("linucb_tag_id_list_2_same_tagset_ids_2_topics_detail:" + str(topic_id_dict_latest_reply_time))
 
             return topic_id_list, topic_id_dict
+        except:
+            logging.error("catch exception,err_msg:%s" % traceback.format_exc())
+            return list(), dict()
+
+    @classmethod
+    def get_tag_new_topic_list(cls, tag_id, have_read_topic_id_list, size=10):
+        try:
+            functions_list = list()
+            for id in tag_id:
+                functions_list.append(
+                    {
+                        "filter": {"term": {"tag_list": id}},
+                        "weight": 1
+                    }
+                )
+            q = {
+                "query": {
+                    "function_score": {
+                        "query": {
+                            "bool": {
+                                "must": [
+                                    {"term": {"content_level": 6}},
+                                    {"term": {"is_online": True}},
+                                    {"term": {"is_deleted": False}},
+                                    {"term": {"is_new_topic": True}},
+                                    {"terms": {"tag_list": tag_id}}
+                                ]
+                            }
+                        },
+                        "boost_mode": "sum",
+                        "score_mode": "sum",
+                        "functions": functions_list
+                    }
+                },
+                "_source": {
+                    "include": ["id", "user_id"]
+                },
+                "sort": [
+                    {"latest_reply_time": {"order": "desc"}},
+                    {"create_time_val": {"order": "desc"}},
+                    {"language_type": {"order": "asc"}},
+                ],
+                "collapse": {
+                    "field": "user_id"
+                }
+            }
+
+            if len(have_read_topic_id_list) > 0:
+                q["query"]["function_score"]["query"]["bool"]["must_not"] = {
+                    "terms": {
+                        "id": have_read_topic_id_list
+                    }
+                }
+            result_dict = ESPerform.get_search_results(ESPerform.get_cli(), sub_index_name="topic-high-star",
+                                                       query_body=q,
+                                                       offset=0, size=size, routing="6")
+
+            topic_id_list = [item["_source"]["id"] for item in result_dict["hits"]]
+            return topic_id_list
         except:
             logging.error("catch exception,err_msg:%s" % traceback.format_exc())
             return list()

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())