Merge branch 'master' into gyz_fix

libs/es.py

@@ -188,19 +188,22 @@ class ESPerform(object):
     @classmethod
     def get_search_results(cls, es_cli, sub_index_name, query_body, offset=0, size=10,
                            auto_create_index=False, doc_type="_doc", aggregations_query=False, is_suggest_request=False,
-                           batch_search=False, routing=None):
+                           batch_search=False, routing=None,if_official_index_name=False):
         try:
             assert (es_cli is not None)
 
-            official_index_name = cls.get_official_index_name(sub_index_name, "read")
-            index_exists = es_cli.indices.exists(official_index_name)
-            if not index_exists:
-                if not auto_create_index:
-                    logging.error("index:%s is not existing,get_search_results error!" % official_index_name)
-                    return None
-                else:
-                    cls.create_index(es_cli, sub_index_name)
-                    cls.put_index_mapping(es_cli, sub_index_name)
+            if if_official_index_name:
+                official_index_name = sub_index_name
+            else:
+                official_index_name = cls.get_official_index_name(sub_index_name, "read")
+                index_exists = es_cli.indices.exists(official_index_name)
+                if not index_exists:
+                    if not auto_create_index:
+                        logging.error("index:%s is not existing,get_search_results error!" % official_index_name)
+                        return None
+                    else:
+                        cls.create_index(es_cli, sub_index_name)
+                        cls.put_index_mapping(es_cli, sub_index_name)
 
             logging.info("duan add,query_body:%s" % str(query_body).encode("utf-8"))
 
@@ -401,6 +404,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}}
                                 ]
                             }
@@ -411,9 +415,18 @@ class ESPerform(object):
                     }
                 },
                 "_source": {
-                    "include": ["id", "user_id", "latest_reply_time"]
+                    "include": ["id", "user_id", "latest_reply_time", "topic_ctr_30", "like_rate_30"]
                 },
                 "sort": [
+                    {
+                        "_script": {
+                            "order": "desc",
+                            "script": {
+                                "inline": "5*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"}},
@@ -446,10 +459,69 @@ class ESPerform(object):
 
             logging.info("topic_id_list:%s" % str(topic_id_dict))
 
-            logging.info("linucb_tag_id_list_2_same_tagset_ids:" + str(tag_id))
+            logging.info("get_tag_topic_list_dict:gyz" + str(q) + str(result_dict))
             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()

linucb/views/collect_data.py

@@ -263,6 +263,7 @@ class CollectData(object):
                             if "type" in raw_val_dict and \
                                     (raw_val_dict["type"] in ("on_click_feed_topic_card","on_click_button")):
                                 click_topic_tag_list = list()
+                                device_id = ""
                                 if "on_click_feed_topic_card" == raw_val_dict["type"]:
                                     topic_id =  raw_val_dict["params"]["topic_id"]
                                     device_id = raw_val_dict["device"]["device_id"]

search/views/group.py

@@ -303,7 +303,7 @@ def pictorial_topic(topic_id=-1, offset=0, size=10):
 
 
 @bind("physical/search/pictorial_topic_sort")
-def pictorial_topic_sort(pictorial_id=-1, offset=0, size=10, sort_type=PICTORIAL_TOPIC_SORT.HOT):
+def pictorial_topic_sort(pictorial_id=-1, offset=0, size=10, sort_type=PICTORIAL_TOPIC_SORT.HOT, user_id=-1):
     """
     :remark 画报排序 人气 部分
     人气按照票数从大到小排序，相同票数按照图片票数更新时间由旧到新排序
@@ -420,7 +420,7 @@ def pictorial_topic_sort(pictorial_id=-1, offset=0, size=10, sort_type=PICTORIAL
         pict_pictorial_ids_list = []
         # 获取es链接对象
         es_cli_obj = ESPerform.get_cli()
-        result_dict = ESPerform.get_search_results(es_cli_obj, "topic", q, offset, size)
+        result_dict = ESPerform.get_search_results(es_cli=es_cli_obj, sub_index_name="mv-alpha-topic-prod-190905001", query_body=q, offset=offset, size=size,if_official_index_name=True)
 
         # logging.info("get pictorial_topic_sort res:%s" % result_dict)
 

trans2es/management/commands/redis_new_topic_sub.py

@@ -4,7 +4,7 @@ import datetime
 from libs.cache import redis_client
 from libs.error import logging_exception
 from django.conf import settings
-from trans2es.models.portrait_stat import LikeDeviceTagStat
+from trans2es.models.portrait_stat import LikeTopicStat
 
 try:
     ps = redis_client.pubsub()
@@ -28,9 +28,9 @@ try:
                 insert_list = []
                 for topic in insert_topic_ids:
                     insert_list.append(
-                        LikeDeviceTagStat(create_time=datetime.datetime.today(), update_time=datetime.datetime.today(),
+                        LikeTopicStat(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)
+                LikeTopicStat.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))

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 LikeTopicStat
 
 
 
@@ -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(
+                                LikeTopicStat(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))
+                        LikeTopicStat.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())