control flow

recommend.py

@@ -315,16 +315,21 @@ def video_recommend(request_id, mid, uid, size, top_K, flow_pool_P, app_type, al
     #print("data['hot_recall']", data['hot_recall'])
     if ab_code == 60058 or ab_code == 60059 or ab_code == 60060 or ab_code == 60061 :
         test_config = None
+        flow_config = None
         if ab_code == 60058:
-            test_config =pool_recall.get_hotrecall_config()
+            test_config = pool_recall.get_hotrecall_config()
+            flow_config = pool_recall.get_hotrecall_flow_config()
         elif ab_code == 60059:
             test_config = pool_recall.get_w2v_config()
+            flow_config = pool_recall.get_w2v_flow_config()
         elif ab_code == 60060:
             test_config = pool_recall.get_test_config()
+            flow_config = pool_recall.get_flow_config()
         elif ab_code == 60061:
             test_config = pool_recall.get_simrecall_config()
+            flow_config = pool_recall.get_simrecall_flow_config()
         #print("test config:",test_config)
-        rank_result = video_sanke_rank(data=data, size=size, top_K=top_K, flow_pool_P=float(flow_pool_P), ab_Code=ab_code, exp_config=test_config)
+        rank_result = video_sanke_rank(data=data, size=size, top_K=top_K, flow_pool_P=float(flow_pool_P), ab_Code=ab_code, exp_config=test_config, flowConfig=flow_config)
     else:
         rank_result = video_rank(data=data, size=size, top_K=top_K, flow_pool_P=float(flow_pool_P))
 
@@ -564,10 +569,19 @@ def video_old_recommend(request_id, mid, uid, size, top_K, flow_pool_P, app_type
             }
     #if ab_code=="ab_new_test":
     #print("before data:", data)
-    rank_result = video_new_rank2(data=data, size=size, top_K=top_K, flow_pool_P=float(flow_pool_P), ab_code=ab_code)
+
+    flow_config = None
+    if ab_code == 60052:
+        flow_config = pool_recall.get_flow_exp_7_config()
+    elif ab_code == 60053:
+        flow_config = pool_recall.get_flow_exp_8_config()
+    elif ab_code == 60057:
+        flow_config = pool_recall.get_flow_exp_6_config()
+    rank_result, flow_result = video_new_rank2(data=data, size=size, top_K=top_K, flow_pool_P=float(flow_pool_P), ab_code=ab_code, flowConfig=flow_config)
     #print(rank_result)
 
     result['rankResult'] = rank_result
+    result['flow_recall_num'] = flow_result
     result['rankTime'] = (time.time() - start_rank) * 1000
 
 

video_rank.py

@@ -592,7 +592,7 @@ def video_rank_with_old_video(rank_result, old_video_recall, size, top_K, old_vi
     return new_rank_result[:size]
 
 
-def video_new_rank2(data, size, top_K, flow_pool_P, ab_code):
+def video_new_rank2(data, size, top_K, flow_pool_P, ab_code, flowConfig=None):
     """
         视频分发排序
         :param data: 各路召回的视频 type-dict {'rov_pool_recall': [], 'flow_pool_recall': []}
@@ -653,30 +653,42 @@ def video_new_rank2(data, size, top_K, flow_pool_P, ab_code):
     else:
         rank_result.extend(flow_recall_rank[:top_K])
         flow_recall_rank = flow_recall_rank[top_K:]
-        # 按概率 p 及score排序获取 size - k 个视频
-    i = 0
-    while i < size - top_K:
-        # 随机生成[0, 1)浮点数
-        rand = random.random()
-        # log_.info('rand: {}'.format(rand))
-        if rand < flow_pool_P:
-            if flow_recall_rank:
-                rank_result.append(flow_recall_rank[0])
-                flow_recall_rank.remove(flow_recall_rank[0])
-            else:
-                rank_result.extend(rov_recall_rank[:size - top_K - i])
-                return rank_result[:size]
-        else:
-            if rov_recall_rank:
+    flow_result = 0
+    if flowConfig == "1" and len(rov_recall_rank)>0:
+        for recall_item in rank_result:
+            flow_recall = recall_item.get("flowPool", '')
+            if flow_recall.split("#") > -1:
+                flow_result = flow_result + 1
+        if flow_result > 0:
+            if len(rov_recall_rank) > 0:
                 rank_result.append(rov_recall_rank[0])
-                rov_recall_rank.remove(rov_recall_rank[0])
+            elif len(flow_recall_rank) > 0:
+                rank_result.append(flow_recall_rank[0])
+    else:
+        # 按概率 p 及score排序获取 size - k 个视频
+        i = 0
+        while i < size - top_K:
+            # 随机生成[0, 1)浮点数
+            rand = random.random()
+            # log_.info('rand: {}'.format(rand))
+            if rand < flow_pool_P:
+                if flow_recall_rank:
+                    rank_result.append(flow_recall_rank[0])
+                    flow_recall_rank.remove(flow_recall_rank[0])
+                else:
+                    rank_result.extend(rov_recall_rank[:size - top_K - i])
+                    return rank_result[:size]
             else:
-                rank_result.extend(flow_recall_rank[:size - top_K - i])
-                return rank_result[:size]
-        i += 1
+                if rov_recall_rank:
+                    rank_result.append(rov_recall_rank[0])
+                    rov_recall_rank.remove(rov_recall_rank[0])
+                else:
+                    rank_result.extend(flow_recall_rank[:size - top_K - i])
+                    return rank_result[:size]
+            i += 1
     return rank_result[:size]
 
-def video_sanke_rank(data, size, top_K, flow_pool_P, ab_Code='', exp_config=None):
+def video_sanke_rank(data, size, top_K, flow_pool_P, ab_Code='', exp_config=None, flowConfig=None):
     """
     视频分发排序
     :param data: 各路召回的视频 type-dict {'rov_pool_recall': [], 'flow_pool_recall': []}
@@ -735,8 +747,8 @@ def video_sanke_rank(data, size, top_K, flow_pool_P, ab_Code='', exp_config=None
     #print("recall_config:", recall_list)
     rov_recall_rank = []
     select_ids = set('')
-    for i in  range(3):
-        if len(rov_recall_rank)>8:
+    for i in  range(4):
+        if len(rov_recall_rank)>16:
             break
         for per_recall_item in recall_list:
             per_recall_name =  per_recall_item[0]
@@ -768,10 +780,10 @@ def video_sanke_rank(data, size, top_K, flow_pool_P, ab_Code='', exp_config=None
                                                          top_K=top_K)
     # log_.info('remove_duplicate finished! rov_recall_rank = {}, flow_recall_rank = {}'.format(
     #     rov_recall_rank, flow_recall_rank))
-
+    # add flow result
     # rank_result = relevant_recall_rank
     rank_result = []
-
+    flow_result = 0
     # 从ROV召回池中获取top k
     if len(rov_recall_rank) > 0:
         rank_result.extend(rov_recall_rank[:top_K])
@@ -779,29 +791,39 @@ def video_sanke_rank(data, size, top_K, flow_pool_P, ab_Code='', exp_config=None
     else:
         rank_result.extend(flow_recall_rank[:top_K])
         flow_recall_rank = flow_recall_rank[top_K:]
-
-    # 按概率 p 及score排序获取 size - k 个视频
-    i = 0
-    while i < size - top_K:
-        # 随机生成[0, 1)浮点数
-        rand = random.random()
-        # log_.info('rand: {}'.format(rand))
-        if rand < flow_pool_P:
-            if flow_recall_rank:
-                rank_result.append(flow_recall_rank[0])
-                flow_recall_rank.remove(flow_recall_rank[0])
-            else:
-                rank_result.extend(rov_recall_rank[:size - top_K - i])
-                return rank_result[:size]
-        else:
-            if rov_recall_rank:
+    if flowConfig=="1" and len(rov_recall_rank)>0:
+        for recall_item in rank_result:
+            flow_recall = recall_item.get("flowPool", '')
+            if flow_recall.split("#")>-1:
+                flow_result = flow_result+1
+        if flow_result>0:
+            if len(rov_recall_rank)>0:
                 rank_result.append(rov_recall_rank[0])
-                rov_recall_rank.remove(rov_recall_rank[0])
+            elif len(flow_recall_rank)>0:
+                rank_result.append(flow_recall_rank[0])
+    else:
+        # 按概率 p 及score排序获取 size - k 个视频
+        i = 0
+        while i < size - top_K:
+            # 随机生成[0, 1)浮点数
+            rand = random.random()
+            # log_.info('rand: {}'.format(rand))
+            if rand < flow_pool_P:
+                if flow_recall_rank:
+                    rank_result.append(flow_recall_rank[0])
+                    flow_recall_rank.remove(flow_recall_rank[0])
+                else:
+                    rank_result.extend(rov_recall_rank[:size - top_K - i])
+                    return rank_result[:size]
             else:
-                rank_result.extend(flow_recall_rank[:size - top_K - i])
-                return rank_result[:size]
-        i += 1
-    return rank_result[:size]
+                if rov_recall_rank:
+                    rank_result.append(rov_recall_rank[0])
+                    rov_recall_rank.remove(rov_recall_rank[0])
+                else:
+                    rank_result.extend(flow_recall_rank[:size - top_K - i])
+                    return rank_result[:size]
+            i += 1
+    return rank_result[:size], flow_result
 
 
 

video_recall.py

@@ -2547,8 +2547,8 @@ class PoolRecall(object):
         else:
             return None
 
-    def get_hotrecall_config(self):
-        recall_key = "ht_config"
+    def get_simrecall_config(self):
+        recall_key = "simrecall_config"
         data = self.redis_helper.get_data_from_redis(key_name=recall_key)
         if data is not None:
             try:
@@ -2561,8 +2561,8 @@ class PoolRecall(object):
         else:
             return None
 
-    def get_simrecall_config(self):
-        recall_key = "simrecall_config"
+    def get_hotrecall_config(self):
+        recall_key = "ht_config:"
         data = self.redis_helper.get_data_from_redis(key_name=recall_key)
         if data is not None:
             try:
@@ -2573,4 +2573,37 @@ class PoolRecall(object):
             except Exception as e:
                 return None
         else:
-            return None
+            return None
+
+    def get_hotrecall_flow_config(self):
+        recall_key = "ht_flow_config:"
+        data = self.redis_helper.get_data_from_redis(key_name=recall_key)
+        return data
+
+    def get_w2v_flow_config(self):
+        recall_key = "w2v_flow_config:"
+        data = self.redis_helper.get_data_from_redis(key_name=recall_key)
+        return data
+
+    def get_flow_config(self):
+        recall_key = "flow_config:"
+        data = self.redis_helper.get_data_from_redis(key_name=recall_key)
+        return data
+    def get_simrecall_flow_config(self):
+        recall_key = "flow_config:"
+        data = self.redis_helper.get_data_from_redis(key_name=recall_key)
+        return data
+
+    def get_flow_exp_7_config(self):
+        recall_key = "exp7_flow_config:"
+        data = self.redis_helper.get_data_from_redis(key_name=recall_key)
+        return data
+
+    def get_flow_exp_8_config(self):
+        recall_key = "exp8_flow_config:"
+        data = self.redis_helper.get_data_from_redis(key_name=recall_key)
+        return data
+    def get_flow_exp_6_config(self):
+        recall_key = "exp6_flow_config:"
+        data = self.redis_helper.get_data_from_redis(key_name=recall_key)
+        return data