Merge branch 'feature/20240514/sunxy/addPushFromIndex' of algorithm/recommend-server into master

recommend-server-service/src/main/java/com/tzld/piaoquan/recommend/server/service/RecommendService.java

@@ -5,12 +5,8 @@ import com.google.common.base.Stopwatch;
 import com.google.common.base.Strings;
 import com.google.common.reflect.TypeToken;
 import com.tzld.piaoquan.recommend.server.common.base.Constant;
-import com.tzld.piaoquan.recommend.server.common.enums.AppTypeEnum;
 import com.tzld.piaoquan.recommend.server.gen.common.Result;
-import com.tzld.piaoquan.recommend.server.gen.recommend.MachineInfoProto;
-import com.tzld.piaoquan.recommend.server.gen.recommend.RecommendRequest;
-import com.tzld.piaoquan.recommend.server.gen.recommend.RecommendResponse;
-import com.tzld.piaoquan.recommend.server.gen.recommend.VideoProto;
+import com.tzld.piaoquan.recommend.server.gen.recommend.*;
 import com.tzld.piaoquan.recommend.server.model.MachineInfo;
 import com.tzld.piaoquan.recommend.server.model.RecommendParam;
 import com.tzld.piaoquan.recommend.server.model.Video;
@@ -136,6 +132,15 @@ public class RecommendService {
         // 更新position
         List<VideoProto> vps = new ArrayList<>();
         for (int i = 0; i < videos.size(); i++) {
+            Map<String, List<String>> pushFromIndex = videos.get(i).getPushFromIndex();
+            List<PushFromIndex> pushFromIndexList = new ArrayList<>();
+            if (MapUtils.isNotEmpty(pushFromIndex)) {
+                pushFromIndex.forEach((k, v) -> {
+                    PushFromIndex.Builder builder = PushFromIndex.newBuilder().setPushFrom(k);
+                    v.forEach(builder::addIndex);
+                    pushFromIndexList.add(builder.build());
+                });
+            }
             vps.add(VideoProto.newBuilder()
                     .setPosition(i + 1)
                     .setPushFrom(Strings.nullToEmpty(videos.get(i).getPushFrom()))
@@ -146,6 +151,7 @@ public class RecommendService {
                     .setFlowPool(Strings.nullToEmpty(videos.get(i).getFlowPool()))
                     .setIsInFlowPool(videos.get(i).isInFlowPool() ? 1 : 0)
                     .setRand(videos.get(i).getRand())
+                    .addAllPushFromIndex(pushFromIndexList)
                     .build());
         }
 

recommend-server-service/src/main/java/com/tzld/piaoquan/recommend/server/service/VideoRecommendService.java

@@ -182,6 +182,15 @@ public class VideoRecommendService {
         // 更新position
         List<VideoProto> vps = new ArrayList<>();
         for (int i = 0; i < videos.size(); i++) {
+            Map<String, List<String>> pushFromIndex = videos.get(i).getPushFromIndex();
+            List<PushFromIndex> pushFromIndexList = new ArrayList<>();
+            if (MapUtils.isNotEmpty(pushFromIndex)) {
+                pushFromIndex.forEach((k, v) -> {
+                    PushFromIndex.Builder builder = PushFromIndex.newBuilder().setPushFrom(k);
+                    v.forEach(builder::addIndex);
+                    pushFromIndexList.add(builder.build());
+                });
+            }
             vps.add(VideoProto.newBuilder()
                     .setPosition(i + 1)
                     .setPushFrom(Strings.nullToEmpty(videos.get(i).getPushFrom()))
@@ -192,6 +201,7 @@ public class VideoRecommendService {
                     .setFlowPool(Strings.nullToEmpty(videos.get(i).getFlowPool()))
                     .setIsInFlowPool(videos.get(i).isInFlowPool() ? 1 : 0)
                     .setRand(videos.get(i).getRand())
+                    .addAllPushFromIndex(pushFromIndexList)
                     .build());
         }
         return videos;

recommend-server-service/src/main/java/com/tzld/piaoquan/recommend/server/service/rank/RankService.java

@@ -97,6 +97,9 @@ public class RankService {
             return new RankResult(results);
         }
 
+        // 给所有重复视频打上多个召回源标记
+        tagDuplicateVideos(param);
+
         // 2 正常走分发 排序+冷启动
         List<Video> rovRecallRank = mergeAndRankRovRecall(param);
         List<Video> flowPoolRank = mergeAndRankFlowPoolRecall(param);
@@ -110,6 +113,35 @@ public class RankService {
         return mergeAndSort(param, rovRecallRank, flowPoolRank);
     }
 
+    private void tagDuplicateVideos(RankParam param) {
+        if (CollectionUtils.isEmpty(param.getRecallResult().getData())) {
+            return;
+        }
+        Map<Long, Set<String>> videoIdPushFromMap = new HashMap<>();
+        for (RecallResult.RecallData data : param.getRecallResult().getData()) {
+            if (CollectionUtils.isEmpty(data.getVideos())) {
+                continue;
+            }
+            for (Video video : data.getVideos()) {
+                Set<String> pushFromSet = videoIdPushFromMap.computeIfAbsent(video.getVideoId(), k -> new HashSet<>());
+                pushFromSet.add(data.getPushFrom());
+            }
+        }
+        for (RecallResult.RecallData data : param.getRecallResult().getData()) {
+            if (CollectionUtils.isEmpty(data.getVideos())) {
+                continue;
+            }
+            for (Video video : data.getVideos()) {
+                Set<String> pushFromSet = videoIdPushFromMap.get(video.getVideoId());
+                if (pushFromSet != null && pushFromSet.size() > 0) {
+                    Map<String, List<String>> map = new HashMap<>();
+                    pushFromSet.forEach(p -> map.computeIfAbsent(p, k -> new ArrayList<>()));
+                    video.setPushFromIndex(map);
+                }
+            }
+        }
+    }
+
     public void rankFilter(RankParam param, List<Video> rovRecallRank, List<Video> flowPoolRank) {
     }