Merge branch 'feature_20250526_zhaohaipeng_douhot' of algorithm/recommend-server into master

recommend-server-service/pom.xml

@@ -270,6 +270,10 @@
             <artifactId>recommend-similarity</artifactId>
             <version>1.0.0</version>
         </dependency>
+        <dependency>
+            <groupId>org.springframework.cloud</groupId>
+            <artifactId>spring-cloud-starter-openfeign</artifactId>
+        </dependency>
     </dependencies>
 
 

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

@@ -1,13 +1,12 @@
 package com.tzld.piaoquan.recommend.server;
 
 
-import com.tzld.piaoquan.abtest.client.ABTestClient;
-import com.tzld.piaoquan.recommend.feature.client.FeatureClient;
 import com.tzld.piaoquan.recommend.feature.client.FeatureV2Client;
 import org.springframework.boot.SpringApplication;
 import org.springframework.boot.autoconfigure.SpringBootApplication;
 import org.springframework.boot.autoconfigure.data.redis.RedisReactiveAutoConfiguration;
 import org.springframework.cloud.netflix.eureka.EnableEurekaClient;
+import org.springframework.cloud.openfeign.EnableFeignClients;
 import org.springframework.context.annotation.Bean;
 import org.springframework.context.annotation.ComponentScan;
 import org.springframework.context.annotation.EnableAspectJAutoProxy;
@@ -33,6 +32,7 @@ import org.springframework.scheduling.annotation.EnableScheduling;
 @EnableEurekaClient
 @EnableAspectJAutoProxy
 @EnableScheduling
+@EnableFeignClients
 public class Application {
     public static void main(String[] args) {
         SpringApplication.run(Application.class, args);

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

@@ -5,8 +5,17 @@ package com.tzld.piaoquan.recommend.server.common;
  */
 public class RedisKeyConstants {
 
-    public static class Recommend{
+    public static class Recommend {
         public static String riskUserMid = "risk:user:mid";
         public static String riskUserUid = "risk:user:uid";
     }
+
+    public static class DouHot {
+
+        // dou:hot:flow:pool:level:item:{省份}:{level}
+        public static String ITEM_REDIS_KEY_FORMAT = "dou:hot:flow:pool:level:item:%s:%s";
+
+        // dou:hot:flow:pool:local:distribute:count:{video}:{flowPool}
+        public static String LOCAL_DISTRIBUTE_KEY_FORMAT = "dou:hot:flow:pool:local:distribute:count:%s:%s";
+    }
 }

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

@@ -0,0 +1,20 @@
+package com.tzld.piaoquan.recommend.server.feign;
+
+import com.alibaba.fastjson.JSONObject;
+import com.tzld.piaoquan.recommend.server.feign.model.FlowPoolResponse;
+import com.tzld.piaoquan.recommend.server.feign.model.FlowPoolVideoInfo;
+import org.springframework.cloud.openfeign.FeignClient;
+import org.springframework.web.bind.annotation.PostMapping;
+import org.springframework.web.bind.annotation.RequestBody;
+
+import java.util.List;
+
+@FeignClient(value = "flowpool")
+public interface FlowPoolFeign {
+
+    @PostMapping("/flowpool/video/getFlowPoolVideo")
+    FlowPoolResponse<List<FlowPoolVideoInfo>> getFlowPoolVideo(@RequestBody JSONObject param);
+
+    @PostMapping("/flowpool/video/remainViewCount")
+    FlowPoolResponse<List<FlowPoolVideoInfo>> remainViewCount(@RequestBody JSONObject param);
+}

recommend-server-service/src/main/java/com/tzld/piaoquan/recommend/server/feign/model/FlowPoolResponse.java

@@ -0,0 +1,18 @@
+package com.tzld.piaoquan.recommend.server.feign.model;
+
+import lombok.Data;
+import lombok.ToString;
+
+@Data
+@ToString
+public class FlowPoolResponse<T> {
+    private int code;
+
+    private String msg;
+
+    private T data;
+
+    private String redirect;
+
+    private String success;
+}

recommend-server-service/src/main/java/com/tzld/piaoquan/recommend/server/feign/model/FlowPoolVideoInfo.java

@@ -0,0 +1,23 @@
+package com.tzld.piaoquan.recommend.server.feign.model;
+
+import lombok.Data;
+import lombok.ToString;
+
+@Data
+@ToString
+public class FlowPoolVideoInfo {
+
+    private Long videoId;
+
+    private String flowPool;
+
+    private Integer distributeCount;
+
+    private Double score;
+
+    private Integer flowPoolId;
+
+    private Integer flowPoolLevelId;
+
+    private Integer level;
+}

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

@@ -0,0 +1,27 @@
+package com.tzld.piaoquan.recommend.server.repository;
+
+import lombok.Data;
+
+import javax.persistence.Entity;
+import javax.persistence.Id;
+import javax.persistence.Table;
+import java.util.Date;
+
+@Data
+@Entity
+@Table(name = "douhot_video_mapping")
+public class DouHotVideoMapping {
+
+    @Id
+    private Long id;
+
+    private Long videoId;
+
+    private String vid;
+
+    private Integer type;
+
+    private Date createTime;
+
+    private Date updateTime;
+}

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

@@ -0,0 +1,13 @@
+package com.tzld.piaoquan.recommend.server.repository;
+
+import org.springframework.data.jpa.repository.JpaRepository;
+import org.springframework.stereotype.Repository;
+
+import java.util.List;
+
+@Repository
+public interface DouHotVideoMappingRepository extends JpaRepository<DouHotVideoMapping, Long> {
+
+    List<DouHotVideoMapping> findAllByVideoIdIn(List<Long> videoIdList);
+
+}

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

@@ -0,0 +1,33 @@
+package com.tzld.piaoquan.recommend.server.repository;
+
+
+import lombok.Data;
+
+import javax.persistence.Entity;
+import javax.persistence.Id;
+import javax.persistence.Table;
+import java.util.Date;
+
+@Data
+@Entity
+@Table(name = "douhot_video_portrait_data")
+public class DouHotVideoPortraitData {
+
+    @Id
+    private Long id;
+
+    private String vid;
+
+    private String name;
+
+    private Double value;
+
+    private String option;
+
+    private Integer type;
+
+    private Date createTime;
+
+    private Date updateTime;
+
+}

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

@@ -0,0 +1,24 @@
+package com.tzld.piaoquan.recommend.server.repository;
+
+import org.springframework.data.jpa.repository.JpaRepository;
+import org.springframework.data.jpa.repository.Query;
+import org.springframework.data.repository.query.Param;
+import org.springframework.stereotype.Repository;
+
+import java.util.List;
+
+@Repository
+public interface DouHotVideoPortraitDataRepository extends JpaRepository<DouHotVideoPortraitData, Long> {
+
+    List<DouHotVideoPortraitData> findAllByVidInAndType(List<String> videoIds, Integer type);
+
+    @Query(value = "SELECT NEW com.tzld.piaoquan.recommend.server.repository.DouHotVideoProvince(" +
+            "d.vid, d.name, d.option, " +
+            "SUM(CASE WHEN d.type = 1 THEN d.value ELSE 0 END), " +
+            "SUM(CASE WHEN d.type = 2 THEN d.value ELSE 0 END)" +
+            ") FROM DouHotVideoPortraitData d " +
+            "WHERE d.option = 4 AND d.vid IN (:videos) " +
+            "GROUP BY d.vid, d.name, d.option " +
+            "HAVING SUM(CASE WHEN d.type = 2 THEN d.value ELSE 0 END) >= 100")
+    List<DouHotVideoProvince> findRecordTGIGe100AndRateDesc(@Param("videos") List<String> videos);
+}

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

@@ -0,0 +1,14 @@
+package com.tzld.piaoquan.recommend.server.repository;
+
+import lombok.AllArgsConstructor;
+import lombok.Data;
+
+@Data
+@AllArgsConstructor
+public class DouHotVideoProvince {
+    private String vid;
+    private String name;
+    private String option;
+    private Double rate;
+    private Double tgi;
+}

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

@@ -493,6 +493,7 @@ public class RecommendService {
 
         updateFlowPoolCache(request, param, videos);
 
+        updateDouHotVideoCache(request, param, videos);
     }
 
     private void updateFlowPoolCache(RecommendRequest request, RecommendParam param,
@@ -514,6 +515,19 @@ public class RecommendService {
         flowPoolService.updateDistributeCountWithLevel(flowPoolVideos);
     }
 
+    private void updateDouHotVideoCache(RecommendRequest request, RecommendParam param, List<Video> videos) {
+        if (CollectionUtils.isEmpty(videos)) {
+            return;
+        }
+        List<Video> douHotFlowPoolVideo = videos.stream()
+                .filter(v -> v.getPushFrom().equals(DouHotFlowPoolRecallStrategy.PUSH_FROM))
+                .collect(Collectors.toList());
+        if (CollectionUtils.isEmpty(douHotFlowPoolVideo)) {
+            return;
+        }
+        flowPoolService.asyncHandleDouHotCache(douHotFlowPoolVideo, param.getProvince());
+    }
+
     private void updateLastVideoCache(List<Video> videos) {
 
         if (CollectionUtils.isEmpty(videos)) {

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

@@ -1,22 +1,39 @@
 package com.tzld.piaoquan.recommend.server.service.flowpool;
 
+import com.alibaba.fastjson.JSON;
+import com.alibaba.fastjson.JSONObject;
+import com.google.common.collect.Lists;
+import com.google.common.collect.Maps;
+import com.tzld.piaoquan.recommend.server.common.RedisKeyConstants;
+import com.tzld.piaoquan.recommend.server.common.ThreadPoolFactory;
+import com.tzld.piaoquan.recommend.server.feign.FlowPoolFeign;
+import com.tzld.piaoquan.recommend.server.feign.model.FlowPoolResponse;
+import com.tzld.piaoquan.recommend.server.feign.model.FlowPoolVideoInfo;
 import com.tzld.piaoquan.recommend.server.model.TripleConsumer;
 import com.tzld.piaoquan.recommend.server.model.Video;
-import com.tzld.piaoquan.recommend.server.common.ThreadPoolFactory;
+import com.tzld.piaoquan.recommend.server.repository.DouHotVideoMapping;
+import com.tzld.piaoquan.recommend.server.repository.DouHotVideoMappingRepository;
+import com.tzld.piaoquan.recommend.server.repository.DouHotVideoPortraitDataRepository;
+import com.tzld.piaoquan.recommend.server.repository.DouHotVideoProvince;
+import com.tzld.piaoquan.recommend.server.util.RecallUtils;
 import lombok.extern.slf4j.Slf4j;
 import org.apache.commons.collections4.CollectionUtils;
 import org.apache.commons.collections4.MapUtils;
+import org.apache.commons.lang3.StringUtils;
 import org.springframework.beans.factory.annotation.Autowired;
 import org.springframework.beans.factory.annotation.Qualifier;
+import org.springframework.beans.factory.annotation.Value;
 import org.springframework.data.redis.core.RedisTemplate;
 import org.springframework.stereotype.Service;
 
 import java.util.*;
 import java.util.concurrent.ExecutorService;
+import java.util.concurrent.TimeUnit;
 import java.util.stream.Collectors;
 
 import static com.tzld.piaoquan.recommend.server.common.enums.AppTypeEnum.*;
-import static com.tzld.piaoquan.recommend.server.service.flowpool.FlowPoolConstants.*;
+import static com.tzld.piaoquan.recommend.server.service.flowpool.FlowPoolConstants.KEY_WITH_LEVEL_FORMAT;
+import static com.tzld.piaoquan.recommend.server.service.flowpool.FlowPoolConstants.KEY_WITH_LEVEL_FORMAT_V2;
 
 /**
  * @author dyp
@@ -30,6 +47,17 @@ public class FlowPoolService {
     @Autowired
     private FlowPoolConfigService flowPoolConfigService;
 
+    @Autowired
+    private DouHotVideoMappingRepository douHotVideoMappingRepository;
+    @Autowired
+    private DouHotVideoPortraitDataRepository douHotVideoPortraitDataRepository;
+
+    @Value("${dou.hot.flow.pool.id:18}")
+    private Integer douHotFlowPoolId;
+
+    @Autowired
+    private FlowPoolFeign flowPoolFeign;
+
     private final String localDistributeCountFormat = "flow:pool:local:distribute:count:%s:%s";
 
     public final String valueFormat = "%s-%s";
@@ -124,5 +152,195 @@ public class FlowPoolService {
             }
         });
     }
+
+    public void asyncHandleDouHotCache(List<Video> videos, String province) {
+        if (StringUtils.isBlank(province) || CollectionUtils.isEmpty(videos)) {
+            return;
+        }
+        pool.execute(() -> {
+            List<String> needRemoveVideoIds = new ArrayList<>();
+            for (Video video : videos) {
+                String distributeKey = String.format(RedisKeyConstants.DouHot.LOCAL_DISTRIBUTE_KEY_FORMAT, video.getVideoId(), video.getFlowPool());
+                Long count = redisTemplate.opsForValue().decrement(distributeKey);
+                if (Objects.isNull(count) || count <= 0) {
+                    redisTemplate.delete(distributeKey);
+                    needRemoveVideoIds.add(String.format("%d-%s", video.getVideoId(), video.getFlowPool()));
+                }
+            }
+            if (CollectionUtils.isEmpty(needRemoveVideoIds)) {
+                return;
+            }
+
+            // 从流量池缓存中移除达到分发次数限制的视频ID
+            String itemKey = String.format(RedisKeyConstants.DouHot.ITEM_REDIS_KEY_FORMAT, province, "1");
+            log.info("[DouHot view distribute count remove cache] key: {}, item: {}", itemKey, needRemoveVideoIds);
+            redisTemplate.opsForSet().remove(itemKey, needRemoveVideoIds.toArray());
+        });
+    }
+
+    public void syncDouHotFlowPoolVideo() {
+
+        // 获取流量池中 全部热点宝视频
+        List<FlowPoolVideoInfo> allDouHotVideo = this.findAllDouHotVideoFromFlowPool();
+        log.info("[DouHot video size]: {}", allDouHotVideo.size());
+
+        // 获取视频对应的可分发数量
+        Map<String, Integer> videoAndDistributeCountMap = this.findDouHotVideoDistributeCount(allDouHotVideo);
+        log.info("[DouHot view distribute count size]: {}", videoAndDistributeCountMap.size());
+
+        List<Long> allVideoId = allDouHotVideo.stream()
+                .map(FlowPoolVideoInfo::getVideoId)
+                .collect(Collectors.toList());
+
+        // 获取视频对应的省份信息, video -> {省份: tgi}
+        Map<Long, Map<String, Double>> allVideoAndProvinceMap = this.findAllVideoAndProvinceMap(allVideoId);
+        log.info("[DouHot video and province mapping size]: {}", allVideoAndProvinceMap.size());
+
+        Map<String, List<FlowPoolVideoInfo>> provinceAndVideoListMap = new HashMap<>(allVideoAndProvinceMap.size());
+
+        // 获取省份与对应视频的映射
+        Map<Long, List<FlowPoolVideoInfo>> videoIdList = allDouHotVideo.stream()
+                .collect(Collectors.groupingBy(FlowPoolVideoInfo::getVideoId));
+        for (Map.Entry<Long, Map<String, Double>> entry : allVideoAndProvinceMap.entrySet()) {
+            Long videoId = entry.getKey();
+            Set<String> provinceList = entry.getValue().keySet();
+            for (String province : provinceList) {
+                List<FlowPoolVideoInfo> flowPoolVideoInfos = provinceAndVideoListMap.computeIfAbsent(province, s -> new ArrayList<>());
+                flowPoolVideoInfos.addAll(videoIdList.get(videoId));
+            }
+        }
+        provinceAndVideoListMap.forEach((key, value) -> log.info("[DouHot province video size]: province: {}, video size: {}", key, value.size()));
+
+        // 将每个省份的数据写入Redis，并同步写入每个视频在对应流量池中的可分发数量
+        for (Map.Entry<String, List<FlowPoolVideoInfo>> entry : provinceAndVideoListMap.entrySet()) {
+            String province = entry.getKey();
+            List<FlowPoolVideoInfo> flowPoolVideoInfos = entry.getValue();
+
+            String redisKey = String.format(RedisKeyConstants.DouHot.ITEM_REDIS_KEY_FORMAT, province, "1");
+            log.info("[DouHot item redis key]: redisKey: {}, video size: {}", redisKey, flowPoolVideoInfos.size());
+            redisTemplate.delete(redisKey);
+
+            // 将视频添加到Redis缓存中，并设置可分发数量
+            for (FlowPoolVideoInfo flowPoolVideoInfo : flowPoolVideoInfos) {
+                String item = String.format("%d-%s", flowPoolVideoInfo.getVideoId(), flowPoolVideoInfo.getFlowPool());
+
+                // 如果剩余的可分发数量不存在或者小于为 则不添加到Redis中
+                if (!videoAndDistributeCountMap.containsKey(item) || videoAndDistributeCountMap.get(item) <= 0) {
+                    continue;
+                }
+
+                // 获取视频在某个省份的tgi，作为视频的分数
+                double score = 0;
+                if (allVideoAndProvinceMap.containsKey(flowPoolVideoInfo.getVideoId())) {
+                    score = allVideoAndProvinceMap.get(flowPoolVideoInfo.getVideoId()).getOrDefault(province, 0d);
+                }
+
+                redisTemplate.opsForZSet().add(redisKey, item, score);
+
+                String distributeKey = String.format(RedisKeyConstants.DouHot.LOCAL_DISTRIBUTE_KEY_FORMAT, flowPoolVideoInfo.getVideoId(), flowPoolVideoInfo.getFlowPool());
+                redisTemplate.opsForValue().set(distributeKey, String.valueOf(videoAndDistributeCountMap.get(item)));
+                redisTemplate.expire(distributeKey, 24 * 60 * 60, TimeUnit.SECONDS);
+            }
+
+            redisTemplate.expire(redisKey, 24 * 60 * 60, TimeUnit.SECONDS);
+        }
+
+
+    }
+
+    private Map<Long, Map<String, Double>> findAllVideoAndProvinceMap(List<Long> videoIds) {
+        // 获取票圈视频ID与热点宝vid的映射
+        List<List<Long>> videoIdPartition = Lists.partition(videoIds, 500);
+        Map<Long, String> videoIdAndVidMap = videoIdPartition.stream().map(douHotVideoMappingRepository::findAllByVideoIdIn)
+                .flatMap(List::stream)
+                .collect(Collectors.toMap(DouHotVideoMapping::getVideoId, DouHotVideoMapping::getVid, (o1, o2) -> o1));
+
+        // 获取热点宝vid与地域的映射
+        List<List<String>> vidPartition = Lists.partition(new ArrayList<>(videoIdAndVidMap.values()), 500);
+        List<DouHotVideoProvince> douHotVideoProvince = vidPartition.stream()
+                .map(douHotVideoPortraitDataRepository::findRecordTGIGe100AndRateDesc)
+                .flatMap(List::stream)
+                .collect(Collectors.toList());
+
+        // 按vid和name分组，并取每组TGI倒排前10条
+        Map<String, Map<String, Double>> vidAndProvinceListMap = douHotVideoProvince.stream()
+                .collect(Collectors.groupingBy(
+                        DouHotVideoProvince::getVid,  // 按 vid 分组
+                        Collectors.collectingAndThen(
+                                Collectors.toList(),
+                                list -> list.stream()
+                                        .sorted(Comparator.comparingDouble(DouHotVideoProvince::getRate).reversed()) // 按 占比 降序
+                                        .limit(10) // 取前10个
+                                        .collect(Collectors.toMap(i -> RecallUtils.douHotProvinceConvert(i.getName()), DouHotVideoProvince::getTgi))
+                        )
+                ));
+        vidAndProvinceListMap.forEach((key, value) -> log.info("[DouHot vid and province mapping]: vid: {}, province: {}", key, JSON.toJSONString(value)));
+        Map<Long, Map<String, Double>> resultMap = new HashMap<>(videoIdAndVidMap.size());
+        for (Map.Entry<Long, String> entry : videoIdAndVidMap.entrySet()) {
+            Long videoId = entry.getKey();
+            String vid = entry.getValue();
+            if (vidAndProvinceListMap.containsKey(vid)) {
+                resultMap.put(videoId, vidAndProvinceListMap.get(vid));
+            }
+        }
+
+        return resultMap;
+    }
+
+    private List<FlowPoolVideoInfo> findAllDouHotVideoFromFlowPool() {
+        List<FlowPoolVideoInfo> result = new ArrayList<>();
+        int pageNum = 0;
+        while (true) {
+            JSONObject paramJson = new JSONObject();
+            paramJson.put("flowPoolId", douHotFlowPoolId);
+            paramJson.put("appType", 0);
+            paramJson.put("pageSize", 1000);
+            paramJson.put("pageNum", pageNum++);
+
+            log.info("[get DouHot flow pool video] paramJson:{}", paramJson);
+            FlowPoolResponse<List<FlowPoolVideoInfo>> response = flowPoolFeign.getFlowPoolVideo(paramJson);
+            if (0 != response.getCode()) {
+                log.error("[get DouHot flow pool video request error] responseJson: {}", response);
+                break;
+            }
+
+            if (CollectionUtils.isEmpty(response.getData())) {
+                log.error("[get DouHot flow pool video data is empty] responseJson: {}", response);
+                break;
+            }
+            result.addAll(response.getData());
+        }
+        return result;
+    }
+
+    private Map<String, Integer> findDouHotVideoDistributeCount(List<FlowPoolVideoInfo> flowPoolVideoInfos) {
+        List<JSONObject> paramJsonList = Lists.newArrayList();
+        for (FlowPoolVideoInfo flowPoolVideoInfo : flowPoolVideoInfos) {
+            JSONObject paramJson = new JSONObject();
+            paramJson.put("videoId", flowPoolVideoInfo.getVideoId());
+            paramJson.put("flowPool", flowPoolVideoInfo.getFlowPool());
+            paramJsonList.add(paramJson);
+        }
+
+        List<FlowPoolVideoInfo> remainFlowPoolVideoInfos = new ArrayList<>(flowPoolVideoInfos.size());
+        List<List<JSONObject>> partition = Lists.partition(paramJsonList, 10);
+        for (List<JSONObject> videos : partition) {
+            JSONObject paramJson = new JSONObject();
+            paramJson.put("videos", videos);
+            FlowPoolResponse<List<FlowPoolVideoInfo>> response = flowPoolFeign.remainViewCount(paramJson);
+            if (0 != response.getCode()) {
+                log.error("[remain view count error] responseJson: {}", response);
+                continue;
+            }
+            remainFlowPoolVideoInfos.addAll(response.getData());
+        }
+
+        Map<String, Integer> distributeCountMap = Maps.newHashMap();
+        for (FlowPoolVideoInfo videoInfo : remainFlowPoolVideoInfos) {
+            String key = String.format("%s-%s", videoInfo.getVideoId(), videoInfo.getFlowPool());
+            distributeCountMap.put(key, videoInfo.getDistributeCount());
+        }
+        return distributeCountMap;
+    }
 }
 

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

@@ -85,11 +85,13 @@ public abstract class RankService {
         List<Video> rovRecallRank = mergeAndRankRovRecall(param);
         List<Video> flowPoolRank = mergeAndRankFlowPoolRecall(param);
 
-        removeDuplicate(param, rovRecallRank, flowPoolRank);
+        List<Video> douHotFlowPoolRank = extractAndSort(param, DouHotFlowPoolRecallStrategy.PUSH_FROM);
+
+        removeDuplicate(param, rovRecallRank, flowPoolRank, douHotFlowPoolRank);
 
 
         // 融合排序
-        return mergeAndSort(param, rovRecallRank, flowPoolRank);
+        return mergeAndSort(param, rovRecallRank, flowPoolRank, douHotFlowPoolRank);
     }
 
     private void tagDuplicateVideos(RankParam param) {
@@ -200,7 +202,7 @@ public abstract class RankService {
         }
     }
 
-    public void removeDuplicate(RankParam param, List<Video> rovRecallRank, List<Video> flowPoolRank) {
+    public void removeDuplicate(RankParam param, List<Video> rovRecallRank, List<Video> flowPoolRank, List<Video> douHotFlowPoolRank) {
         // TODO 重构 rov和流量池 融合排序
         //    去重原则：
         //        如果视频在ROV召回池topK，则保留ROV召回池，否则保留流量池
@@ -222,6 +224,17 @@ public abstract class RankService {
             }
         }
 
+        // dou hot flow pool 移除topK视频
+        Iterator<Video> douHotFlowPoolIte = douHotFlowPoolRank.iterator();
+        while (douHotFlowPoolIte.hasNext()) {
+            Video data = douHotFlowPoolIte.next();
+            if (rovTopKVideoIds.contains(data.getVideoId())) {
+                douHotFlowPoolIte.remove();
+            } else {
+                flowPoolVideoIds.add(data.getVideoId());
+            }
+        }
+
         // rov pool 移除flow中的视频
         Iterator<Video> rovRecallRankIte = rovRecallRank.iterator();
         while (rovRecallRankIte.hasNext()) {
@@ -232,7 +245,7 @@ public abstract class RankService {
         }
     }
 
-    public abstract RankResult mergeAndSort(RankParam param, List<Video> rovRecallRank, List<Video> flowPoolRank);
+    public abstract RankResult mergeAndSort(RankParam param, List<Video> rovRecallRank, List<Video> flowPoolRank,List<Video> douHotFlowPoolRank);
 
     private boolean matchSpecialApp(int appId) {
         Set<Integer> notSpecialApp = new HashSet<>(Arrays.asList(0, 4, 5));

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

@@ -10,7 +10,7 @@ public class RankExtractorItemTags {
         this.redisTemplate = redisTemplate;
     }
 
-    public void processor(List<Video> rovVideos, List<Video> flowVideos){
+    public void processor(List<Video> rovVideos, List<Video> flowVideos, List<Video> douHotFlowPoolVideos) {
         List<Long> videoIds = new ArrayList<>();
         for (Video v : rovVideos) {
             videoIds.add(v.getVideoId());
@@ -18,6 +18,10 @@ public class RankExtractorItemTags {
         for (Video v : flowVideos) {
             videoIds.add(v.getVideoId());
         }
+        for (Video v : douHotFlowPoolVideos) {
+            videoIds.add(v.getVideoId());
+        }
+
         Map<Long, List<String>> videoTagDict = getVideoTags(redisTemplate, videoIds);
         for (Video v : rovVideos) {
             v.setTags(videoTagDict.getOrDefault(v.getVideoId(), new ArrayList<>()));
@@ -25,6 +29,9 @@ public class RankExtractorItemTags {
         for (Video v : flowVideos) {
             v.setTags(videoTagDict.getOrDefault(v.getVideoId(), new ArrayList<>()));
         }
+        for (Video v : douHotFlowPoolVideos) {
+            v.setTags(videoTagDict.getOrDefault(v.getVideoId(), new ArrayList<>()));
+        }
     }
 
     public static Map<Long, List<String>> getVideoTags(RedisTemplate<String, String> redisHelper, List<Long> videoIds) {

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

@@ -6,7 +6,7 @@ import com.tzld.piaoquan.recommend.server.util.ProbabilityCalculator;
 import java.util.*;
 public class RankProcessorTagFilter {
 
-    public static void processor(List<Video> rov, List<Video> flow, Map<String, Map<String, String>> rules) {
+    public static void processor(List<Video> rov, List<Video> flow,List<Video> douHot, Map<String, Map<String, String>> rules) {
 
         Set<String> filterTags = new HashSet<>();
         Random random = new Random();
@@ -56,6 +56,17 @@ public class RankProcessorTagFilter {
                 }
             }
         }
+
+        iterator = douHot.iterator();
+        while (iterator.hasNext()) {
+            Video video = iterator.next();
+            List<String> tags = video.getTags();
+            for (String tag : tags) {
+                if (filterTags.contains(tag)) {
+                    iterator.remove();
+                }
+            }
+        }
     }
 
 }

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

@@ -16,6 +16,7 @@ import lombok.extern.slf4j.Slf4j;
 import org.apache.commons.collections4.CollectionUtils;
 import org.apache.commons.collections4.MapUtils;
 import org.apache.commons.lang3.RandomUtils;
+import org.springframework.beans.factory.annotation.Value;
 import org.springframework.stereotype.Service;
 
 import java.io.BufferedReader;
@@ -38,6 +39,10 @@ public abstract class RankStrategy4RegionMergeModelBasic extends RankService {
     @ApolloJsonValue("${RankReduceByMergeCateConfig:{}}")
     private Map<String, Map<String, List<Map<String, String>>>> rankReduceConfig = new HashMap<>();
 
+
+    @Value("${new.flow.pool.select.rate:1}")
+    private double newFlowPoolSelectRate;
+
     String CLASS_NAME = this.getClass().getSimpleName();
 
     public void duplicate(Set<Long> setVideo, List<Video> videos) {
@@ -53,7 +58,7 @@ public abstract class RankStrategy4RegionMergeModelBasic extends RankService {
     }
 
     @Override
-    public RankResult mergeAndSort(RankParam param, List<Video> rovVideos, List<Video> flowVideos) {
+    public RankResult mergeAndSort(RankParam param, List<Video> rovVideos, List<Video> flowVideos, List<Video> douHotFlowPoolVideos) {
 
         // 1 兜底策略，rov池子不足时，用冷启池填补。直接返回。
         if (CollectionUtils.isEmpty(rovVideos)) {
@@ -90,11 +95,11 @@ public abstract class RankStrategy4RegionMergeModelBasic extends RankService {
         // 3 标签读取
         if (rulesMap != null && !rulesMap.isEmpty()) {
             RankExtractorItemTags extractorItemTags = new RankExtractorItemTags(this.redisTemplate);
-            extractorItemTags.processor(rovVideos, flowVideos);
+            extractorItemTags.processor(rovVideos, flowVideos, douHotFlowPoolVideos);
         }
         // 6 合并结果时间卡控
         if (rulesMap != null && !rulesMap.isEmpty()) {
-            RankProcessorTagFilter.processor(rovVideos, flowVideos, rulesMap);
+            RankProcessorTagFilter.processor(rovVideos, flowVideos, douHotFlowPoolVideos, rulesMap);
         }
 
 
@@ -127,6 +132,12 @@ public abstract class RankStrategy4RegionMergeModelBasic extends RankService {
                 } else {
                     break;
                 }
+            } else if (this.isInsertDouHotFlowPoolVideo()) {
+                if (flowPoolIndex < douHotFlowPoolVideos.size()) {
+                    result.add(douHotFlowPoolVideos.get(flowPoolIndex++));
+                } else {
+                    break;
+                }
             } else {
                 if (rovPoolIndex < rovVideos.size()) {
                     result.add(rovVideos.get(rovPoolIndex++));
@@ -353,4 +364,8 @@ public abstract class RankStrategy4RegionMergeModelBasic extends RankService {
         return vor;
     }
 
+    private boolean isInsertDouHotFlowPoolVideo() {
+        double rand = RandomUtils.nextDouble(0, 1);
+        return rand <= newFlowPoolSelectRate;
+    }
 }

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

@@ -125,6 +125,7 @@ public class RecallService implements ApplicationContextAware {
         if (!param.isRiskUser()) {
             strategies.add(strategyMap.get(QuickFlowPoolWithLevelRecallStrategy.class.getSimpleName()));
             strategies.add(strategyMap.get(FlowPoolWithLevelRecallStrategyTomson.class.getSimpleName()));
+            strategies.add(strategyMap.get(DouHotFlowPoolRecallStrategy.class.getSimpleName()));
         }
         return strategies;
     }

recommend-server-service/src/main/java/com/tzld/piaoquan/recommend/server/service/recall/strategy/DouHotFlowPoolRecallStrategy.java

@@ -0,0 +1,97 @@
+package com.tzld.piaoquan.recommend.server.service.recall.strategy;
+
+import com.tzld.piaoquan.recommend.server.common.RedisKeyConstants;
+import com.tzld.piaoquan.recommend.server.model.Video;
+import com.tzld.piaoquan.recommend.server.service.filter.FilterParam;
+import com.tzld.piaoquan.recommend.server.service.filter.FilterResult;
+import com.tzld.piaoquan.recommend.server.service.filter.FilterService;
+import com.tzld.piaoquan.recommend.server.service.recall.FilterParamFactory;
+import com.tzld.piaoquan.recommend.server.service.recall.RecallParam;
+import com.tzld.piaoquan.recommend.server.service.recall.RecallStrategy;
+import com.tzld.piaoquan.recommend.server.util.RecallUtils;
+import lombok.extern.slf4j.Slf4j;
+import org.apache.commons.collections4.CollectionUtils;
+import org.apache.commons.lang3.StringUtils;
+import org.springframework.beans.factory.annotation.Autowired;
+import org.springframework.beans.factory.annotation.Qualifier;
+import org.springframework.beans.factory.annotation.Value;
+import org.springframework.data.redis.core.RedisTemplate;
+import org.springframework.data.redis.core.ZSetOperations;
+import org.springframework.stereotype.Component;
+
+import java.util.*;
+
+@Slf4j
+@Component
+public class DouHotFlowPoolRecallStrategy implements RecallStrategy {
+
+    @Autowired
+    @Qualifier("redisTemplate")
+    private RedisTemplate<String, String> redisTemplate;
+
+    @Autowired
+    private FilterService filterService;
+
+    @Value("${dou.hot.recall.video.count:20}")
+    private Integer douHotRecallVideoCnt;
+
+    public static final String PUSH_FROM = "recall_strategy_hotspot";
+
+    @Override
+    public List<Video> recall(RecallParam param) {
+        if (StringUtils.isBlank(param.getProvince())) {
+            return Collections.emptyList();
+        }
+
+        String redisKey = String.format(RedisKeyConstants.DouHot.ITEM_REDIS_KEY_FORMAT, RecallUtils.douHotProvinceConvert(param.getProvince()), "1");
+        log.info("[DouHot recall redisKey] {}", redisKey);
+        Set<ZSetOperations.TypedTuple<String>> redisValues = redisTemplate.opsForZSet().rangeWithScores(redisKey, 0, 10000000);
+
+
+        if (CollectionUtils.isEmpty(redisValues)) {
+            log.error("[DouHot recall is empty]");
+            return Collections.emptyList();
+        }
+
+        Map<Long, String> videoIdAndFlowPoolMap = new HashMap<>(redisValues.size());
+
+        for (ZSetOperations.TypedTuple<String> redisValue : redisValues) {
+            String value = redisValue.getValue();
+            if (StringUtils.isBlank(value)) {
+                continue;
+            }
+            String[] split = value.split("-");
+            if (split.length != 2) {
+                continue;
+            }
+            videoIdAndFlowPoolMap.put(Long.valueOf(split[0]), split[1]);
+        }
+
+        FilterParam filterParam = FilterParamFactory.create(param, new ArrayList<>(videoIdAndFlowPoolMap.keySet()));
+        FilterResult filterResult = filterService.filter(filterParam);
+
+        // 对视频随机打断
+        Collections.shuffle(filterResult.getVideoIds());
+
+        // 截取前20个
+        int size = Math.min(filterResult.getVideoIds().size(), douHotRecallVideoCnt);
+        List<Long> subList = filterResult.getVideoIds().subList(0, size);
+        List<Video> videos = new ArrayList<>(subList.size());
+        for (Long videoId : subList) {
+            Video video = new Video();
+            video.setVideoId(videoId);
+            video.setFlowPool(videoIdAndFlowPoolMap.get(videoId));
+            video.setPushFrom(pushFrom());
+            // 热点宝供给目前只有一层，所以写死
+            video.setLevel("1");
+            videos.add(video);
+        }
+
+        return videos;
+    }
+
+    @Override
+    public String pushFrom() {
+        return PUSH_FROM;
+    }
+}

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

@@ -5,6 +5,7 @@ import com.tzld.piaoquan.recommend.server.service.rank.RankParam;
 import com.tzld.piaoquan.recommend.server.service.recall.RecallResult;
 import com.tzld.piaoquan.recommend.server.service.recall.strategy.*;
 import org.apache.commons.collections4.CollectionUtils;
+import org.apache.commons.lang3.StringUtils;
 
 import java.util.*;
 import java.util.stream.Collectors;
@@ -70,4 +71,25 @@ public class RecallUtils {
             setVideo.addAll(list.stream().map(Video::getVideoId).collect(Collectors.toSet()));
         }
     }
+
+    public static String douHotProvinceConvert(String province) {
+        if (StringUtils.isBlank(province)) {
+            return "";
+        }
+        if (StringUtils.startsWith(province, "西藏")) {
+            return "西藏";
+        }
+        if (StringUtils.startsWith(province, "新疆")) {
+            return "新疆";
+        }
+
+        if (StringUtils.startsWith(province, "内蒙古")) {
+            return "内蒙古";
+        }
+
+        if (StringUtils.endsWith(province, "市")) {
+            return province.replaceAll("市$", "");
+        }
+        return province.replaceAll("省$", "");
+    }
 }

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

@@ -0,0 +1,27 @@
+package com.tzld.piaoquan.recommend.server.xxl;
+
+import com.tzld.piaoquan.recommend.server.service.flowpool.FlowPoolService;
+import com.xxl.job.core.biz.model.ReturnT;
+import com.xxl.job.core.handler.annotation.XxlJob;
+import lombok.extern.slf4j.Slf4j;
+import org.springframework.beans.factory.annotation.Autowired;
+import org.springframework.stereotype.Component;
+
+@Slf4j
+@Component
+public class FlowPoolJob {
+
+    @Autowired
+    private FlowPoolService flowPoolService;
+
+
+    @XxlJob("syncDouHotFlowPoolVideo")
+    public ReturnT<String> syncDouHotFlowPoolVideo(String params) {
+        try {
+            flowPoolService.syncDouHotFlowPoolVideo();
+        }catch (Exception e){
+            log.error("[syncDouHotFlowPoolVideo error]", e);
+        }
+        return ReturnT.SUCCESS;
+    }
+}

recommend-server-service/src/main/resources/logback.xml

@@ -316,6 +316,10 @@
 
     <root level="info">
         <appender-ref ref="CONSOLE"/>
+        <!-- <appender-ref ref="DEBUG_FILE"/> -->
+        <!-- <appender-ref ref="INFO_FILE"/> -->
+        <!-- <appender-ref ref="WARN_FILE"/> -->
+        <!-- <appender-ref ref="ERROR_FILE"/> -->
         <appender-ref ref="loghubAppenderInfo"/>
         <appender-ref ref="loghubAppenderWarn"/>
         <appender-ref ref="loghubAppenderError"/>