召回改造

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

@@ -157,6 +157,7 @@ public class RecommendService {
         recallParam.setMid(request.getMid());
         recallParam.setSpecialRecommend(true);
         recallParam.setSize(request.getSize());
+        recallParam.setProvince(request.getProvince());
 
         RecallResult recallResult = recallService.recall(recallParam);
         log.info("recallParam {}, recallResult {}", JSONUtils.toJson(recallParam), JSONUtils.toJson(recallResult));
@@ -411,6 +412,8 @@ public class RecommendService {
         recallParam.setAppRegionFiltered(param.getAppRegionFiltered());
         recallParam.setAbExpCodes(param.getAbExpCodes());
 
+        recallParam.setProvince(param.getProvince());
+
 
         return recallParam;
     }

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

@@ -22,6 +22,6 @@ public class FilterParam {
     private Map<Integer, List<String>> appRegionFiltered;
     private List<Long> videosWithRisk;
     private String regionCode;
-    private int forceTruncation;
+    private int forceTruncation = 1000;
     private Set<String> abExpCodes;
 }

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

@@ -19,6 +19,8 @@ public class RankRouter {
 
     @Autowired
     private RankStrategyFlowThompsonModel rankStrategyFlowThompsonModel;
+    @Autowired
+    private RankStrategy4RegionMerge rankStrategy4RegionMerge;
 
 
     public RankResult rank(RankParam param) {
@@ -44,6 +46,8 @@ public class RankRouter {
                 return rankStrategy4Density.rank(param);
             case "60107":
                 return rankStrategyFlowThompsonModel.rank(param);
+            case "60120": // 576
+                return rankStrategy4RegionMerge.rank(param);
             default:
                 break;
         }

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

@@ -0,0 +1,86 @@
+package com.tzld.piaoquan.recommend.server.service.rank.strategy;
+
+
+import com.ctrip.framework.apollo.spring.annotation.ApolloJsonValue;
+import com.tzld.piaoquan.recommend.server.model.Video;
+import com.tzld.piaoquan.recommend.server.service.rank.RankParam;
+import com.tzld.piaoquan.recommend.server.service.rank.RankResult;
+import com.tzld.piaoquan.recommend.server.service.rank.RankService;
+import com.tzld.piaoquan.recommend.server.service.rank.extractor.RankExtractorFeature;
+import com.tzld.piaoquan.recommend.server.service.rank.processor.RankProcessorDensity;
+import com.tzld.piaoquan.recommend.server.service.rank.processor.RankProcessorTagFilter;
+import com.tzld.piaoquan.recommend.server.service.recall.strategy.*;
+import com.tzld.piaoquan.recommend.server.util.JSONUtils;
+import lombok.extern.slf4j.Slf4j;
+import org.apache.commons.collections4.CollectionUtils;
+import org.apache.commons.lang3.RandomUtils;
+import org.apache.commons.lang3.math.NumberUtils;
+import org.springframework.stereotype.Service;
+
+import java.util.*;
+import java.util.stream.Collectors;
+
+/**
+ * @author zhangbo
+ * @desc 地域召回融合
+ */
+@Service
+@Slf4j
+public class RankStrategy4RegionMerge extends RankService {
+
+    public void duplicate(Set<Long> setVideo, List<Video> videos){
+        Iterator<Video> iterator = videos.iterator();
+        while(iterator.hasNext()){
+            Video v = iterator.next();
+            if (setVideo.contains(v.getVideoId())){
+                iterator.remove();
+            }else{
+                setVideo.add(v.getVideoId());
+            }
+        }
+    }
+    @Override
+    public List<Video> mergeAndRankRovRecall(RankParam param) {
+        //-------------------地域内部融合+去重复-------------------
+        List<Video> rovRecallRank = new ArrayList<>();
+        List<Video> v1 = extractAndSort(param, RegionRealtimeRecallStrategyV1.PUSH_FORM);
+        List<Video> v2 = extractAndSort(param, RegionRealtimeRecallStrategyV2.PUSH_FORM);
+        List<Video> v3 = extractAndSort(param, RegionRealtimeRecallStrategyV3.PUSH_FORM);
+        List<Video> v4 = extractAndSort(param, RegionRealtimeRecallStrategyV4.PUSH_FORM);
+        Set<Long> setVideo = new HashSet<>();
+        this.duplicate(setVideo, v1);
+        this.duplicate(setVideo, v2);
+        this.duplicate(setVideo, v3);
+        this.duplicate(setVideo, v4);
+        rovRecallRank.addAll(v1.subList(0, Math.min(10, v1.size())));
+        rovRecallRank.addAll(v2.subList(0, Math.min(5, v2.size())));
+        rovRecallRank.addAll(v3.subList(0, Math.min(10, v3.size())));
+        rovRecallRank.addAll(v4.subList(0, Math.min(5, v4.size())));
+        //-------------------地域 sim returnv2 融合+去重复-------------------
+        List<Video> v5 = extractAndSort(param, SimHotVideoRecallStrategy.PUSH_FORM);
+        List<Video> v6 = extractAndSort(param, ReturnVideoRecallStrategy.PUSH_FORM);
+        this.duplicate(setVideo, v5);
+        this.duplicate(setVideo, v6);
+        rovRecallRank.addAll(v5);
+        rovRecallRank.addAll(v6);
+        //-------------------排-------------------
+        //-------------------序-------------------
+        //-------------------逻-------------------
+        //-------------------辑-------------------
+        List<String> videoIdKeys = rovRecallRank.stream()
+                .map(t -> param.getRankKeyPrefix() + t.getVideoId())
+                .collect(Collectors.toList());
+        List<String> videoScores = this.redisTemplate.opsForValue().multiGet(videoIdKeys);
+        log.info("rank mergeAndRankRovRecall videoIdKeys={}, videoScores={}", JSONUtils.toJson(videoIdKeys),
+                JSONUtils.toJson(videoScores));
+        if (CollectionUtils.isNotEmpty(videoScores)
+                && videoScores.size() == rovRecallRank.size()) {
+            for (int i = 0; i < videoScores.size(); i++) {
+                rovRecallRank.get(i).setSortScore(NumberUtils.toDouble(videoScores.get(i), 0.0));
+            }
+            Collections.sort(rovRecallRank, Comparator.comparingDouble(o -> -o.getSortScore()));
+        }
+        return rovRecallRank;
+    }
+
+}

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

@@ -16,6 +16,7 @@ public class RecallParam {
     private int appType;
     private String dataKey;
     private String ruleKey;
+    private String province;
 
     // hardcode   兼容旧服务的518实验
     private String hDataKey;

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

@@ -82,29 +82,38 @@ public class RecallService implements ApplicationContextAware {
             return strategies;
         }
 
+        String abCode = param.getAbCode();
         //1：通过“产品”控制“召回子策略”
         if (param.getAppType() == AppTypeEnum.LAO_HAO_KAN_VIDEO.getCode()
                 || param.getAppType() == AppTypeEnum.ZUI_JING_QI.getCode()) {
             strategies.addAll(getRegionRecallStrategy(param));
-        } else {
-            //2:通过“流量池标记”控制“流量池召回子策略”
-            if (param.getFlowPoolAbtestGroup().equals(FlowPoolConstants.EXPERIMENTAL_FLOW_SET_LEVEL)) {
-                strategies.addAll(getRegionRecallStrategy(param));
-                strategies.add(strategyMap.get(QuickFlowPoolWithLevelRecallStrategy.class.getSimpleName()));
-                strategies.add(strategyMap.get(FlowPoolWithLevelRecallStrategy.class.getSimpleName()));
-            } else if (param.getFlowPoolAbtestGroup().equals(FlowPoolConstants.EXPERIMENTAL_FLOW_SET_LEVEL_SCORE)) {
-                strategies.addAll(getRegionRecallStrategy(param));
-                strategies.add(strategyMap.get(QuickFlowPoolWithLevelScoreRecallStrategy.class.getSimpleName()));
-                strategies.add(strategyMap.get(FlowPoolWithLevelScoreRecallStrategy.class.getSimpleName()));
-            } else {
+            return strategies;
+        }
+        //2:通过“abcode”控制“地域召回子策略”
+        switch (abCode) {
+            case "60120": // 576
+                strategies.add(strategyMap.get(RegionRealtimeRecallStrategyV1.class.getSimpleName()));
+                strategies.add(strategyMap.get(RegionRealtimeRecallStrategyV2.class.getSimpleName()));
+                strategies.add(strategyMap.get(RegionRealtimeRecallStrategyV3.class.getSimpleName()));
+                strategies.add(strategyMap.get(RegionRealtimeRecallStrategyV4.class.getSimpleName()));
+                break;
+            default:
                 strategies.addAll(getRegionRecallStrategy(param));
-                strategies.add(strategyMap.get(QuickFlowPoolWithScoreRecallStrategy.class.getSimpleName()));
-                strategies.add(strategyMap.get(FlowPoolWithScoreRecallStrategy.class.getSimpleName()));
-            }
+        }
+        //3:通过“流量池标记”控制“流量池召回子策略”
+        if (param.getFlowPoolAbtestGroup().equals(FlowPoolConstants.EXPERIMENTAL_FLOW_SET_LEVEL)) {
+            strategies.add(strategyMap.get(QuickFlowPoolWithLevelRecallStrategy.class.getSimpleName()));
+            strategies.add(strategyMap.get(FlowPoolWithLevelRecallStrategy.class.getSimpleName()));
+        } else if (param.getFlowPoolAbtestGroup().equals(FlowPoolConstants.EXPERIMENTAL_FLOW_SET_LEVEL_SCORE)) {
+            strategies.add(strategyMap.get(QuickFlowPoolWithLevelScoreRecallStrategy.class.getSimpleName()));
+            strategies.add(strategyMap.get(FlowPoolWithLevelScoreRecallStrategy.class.getSimpleName()));
+        } else {
+            strategies.add(strategyMap.get(QuickFlowPoolWithScoreRecallStrategy.class.getSimpleName()));
+            strategies.add(strategyMap.get(FlowPoolWithScoreRecallStrategy.class.getSimpleName()));
         }
 
-        //3:通过“abcode”控制“召回子策略”
-        String abCode = param.getAbCode();
+
+        //4:通过“abcode”控制“新增召回子策略”
         if (abCode == null) {
             // todo 做兜底吗？
             ;
@@ -135,6 +144,7 @@ public class RecallService implements ApplicationContextAware {
                 case "60102": // 通过更改param中的配置实现使用不同数据源 data66 rule68 + 无排序模块
                 case "60103": // 增加地域1小时扩量，通过配置实现
                 case "60105": // 通过更改param中的配置实现使用不同数据源 data66 rule68 + 有排序模块
+                case "60120": // 576
                     strategies.add(strategyMap.get(SimHotVideoRecallStrategy.class.getSimpleName()));
                     strategies.add(strategyMap.get(ReturnVideoRecallStrategy.class.getSimpleName()));
                     break;

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

@@ -0,0 +1,83 @@
+package com.tzld.piaoquan.recommend.server.service.recall.strategy;
+
+import com.alibaba.fastjson.JSONObject;
+import com.google.common.collect.Lists;
+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.RegionFilterService;
+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.service.score.ScorerUtils;
+import lombok.extern.slf4j.Slf4j;
+import org.apache.commons.collections4.CollectionUtils;
+import org.apache.commons.lang3.tuple.Pair;
+import org.springframework.beans.factory.annotation.Autowired;
+
+import java.util.*;
+
+import com.tzld.piaoquan.recommend.server.service.score4recall.ScorerPipeline4Recall;
+import org.springframework.stereotype.Component;
+
+/**
+ * @author zhangbo
+ */
+@Slf4j
+@Component
+public class RegionRealtimeRecallStrategyV1 implements RecallStrategy {
+    @Autowired
+    private RegionFilterService filterService;
+    @Override
+    public List<Video> recall(RecallParam param) {
+        // 1 获取省份key 放入参数map中
+        String provinceCn = param.getProvince();
+        if (provinceCn == null){
+            provinceCn = "中国";
+        }else{
+            provinceCn = provinceCn.replaceAll("省$", "");
+        }
+        Map<String, String> param4Model = new HashMap<>(1);
+        param4Model.put("region_province", provinceCn);
+        // 2 通过model拿到召回list
+        ScorerPipeline4Recall pipeline = ScorerUtils.getScorerPipeline4Recall("feeds_recall_config_region_v1.conf");
+        List<List<Pair<Long, Double>>> results = pipeline.recall(param4Model);
+        List<Pair<Long, Double>> result = results.get(0);
+        for (int i=1; i<results.size(); ++i){
+            result.addAll(results.get(i));
+        }
+        Map<Long, Double> videoMap = new LinkedHashMap<>();
+        for (Pair<Long, Double> v: result){
+            videoMap.put(v.getLeft(), v.getRight());
+        }
+
+        FilterParam filterParam = FilterParamFactory.create(param, Lists.newArrayList(videoMap.keySet()));
+        filterParam.setForceTruncation(10000);
+        long t1 = new Long(System.currentTimeMillis());
+        FilterResult filterResult = filterService.filter(filterParam);
+        long t2 = new Long(System.currentTimeMillis());
+        JSONObject obj = new JSONObject();
+        obj.put("name", "RegionRealtimeRecallStrategyV1");
+        obj.put("filter_time", t2-t1);
+        log.info(obj.toString());
+        List<Video> videosResult = new ArrayList<>();
+        if (filterResult != null && CollectionUtils.isNotEmpty(filterResult.getVideoIds())) {
+            filterResult.getVideoIds().stream().forEach(vid -> {
+                Video video = new Video();
+                video.setVideoId(vid);
+                video.setAbCode(param.getAbCode());
+                video.setRovScore(videoMap.get(vid));
+                video.setPushFrom(pushFrom());
+                videosResult.add(video);
+            });
+        }
+        Collections.sort(videosResult, Comparator.comparingDouble(o -> -o.getRovScore()));
+        return videosResult;
+    }
+    public static final String PUSH_FORM = "recall_strategy_region_1h";
+    @Override
+    public String pushFrom(){
+        return PUSH_FORM;
+    }
+
+}

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

@@ -0,0 +1,68 @@
+package com.tzld.piaoquan.recommend.server.service.recall.strategy;
+
+import com.google.common.collect.Lists;
+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.RegionFilterService;
+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.service.score.ScorerUtils;
+import com.tzld.piaoquan.recommend.server.service.score4recall.ScorerPipeline4Recall;
+import lombok.extern.slf4j.Slf4j;
+import org.apache.commons.collections4.CollectionUtils;
+import org.apache.commons.lang3.tuple.Pair;
+import org.springframework.beans.factory.annotation.Autowired;
+import org.springframework.stereotype.Component;
+
+import java.util.*;
+
+/**
+ * @author zhangbo
+ */
+@Slf4j
+@Component
+public class RegionRealtimeRecallStrategyV2 implements RecallStrategy {
+    @Autowired
+    protected RegionFilterService filterService;
+    @Override
+    public List<Video> recall(RecallParam param) {
+
+        // 2 通过model拿到召回list
+        Map<String, String> param4Model = new HashMap<>(0);
+        ScorerPipeline4Recall pipeline = ScorerUtils.getScorerPipeline4Recall("feeds_recall_config_region_v2.conf");
+        List<List<Pair<Long, Double>>> results = pipeline.recall(param4Model);
+        List<Pair<Long, Double>> result = results.get(0);
+        for (int i=1; i<results.size(); ++i){
+            result.addAll(results.get(i));
+        }
+        Map<Long, Double> videoMap = new LinkedHashMap<>();
+        for (Pair<Long, Double> v: result){
+            videoMap.put(v.getLeft(), v.getRight());
+        }
+
+        FilterParam filterParam = FilterParamFactory.create(param, Lists.newArrayList(videoMap.keySet()));
+        filterParam.setForceTruncation(10000);
+        FilterResult filterResult = filterService.filter(filterParam);
+        List<Video> videosResult = new ArrayList<>();
+        if (filterResult != null && CollectionUtils.isNotEmpty(filterResult.getVideoIds())) {
+            filterResult.getVideoIds().stream().forEach(vid -> {
+                Video video = new Video();
+                video.setVideoId(vid);
+                video.setAbCode(param.getAbCode());
+                video.setRovScore(videoMap.get(vid));
+                video.setPushFrom(pushFrom());
+                videosResult.add(video);
+            });
+        }
+        Collections.sort(videosResult, Comparator.comparingDouble(o -> -o.getRovScore()));
+        return videosResult;
+    }
+    public static final String PUSH_FORM = "recall_strategy_noregion_24h";
+    @Override
+    public String pushFrom(){
+        return PUSH_FORM;
+    }
+
+}

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

@@ -0,0 +1,75 @@
+package com.tzld.piaoquan.recommend.server.service.recall.strategy;
+
+import com.google.common.collect.Lists;
+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.RegionFilterService;
+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.service.score.ScorerUtils;
+import com.tzld.piaoquan.recommend.server.service.score4recall.ScorerPipeline4Recall;
+import lombok.extern.slf4j.Slf4j;
+import org.apache.commons.collections4.CollectionUtils;
+import org.apache.commons.lang3.tuple.Pair;
+import org.springframework.beans.factory.annotation.Autowired;
+import org.springframework.stereotype.Component;
+
+import java.util.*;
+
+/**
+ * @author zhangbo
+ */
+@Slf4j
+@Component
+public class RegionRealtimeRecallStrategyV3 implements RecallStrategy {
+    @Autowired
+    protected RegionFilterService filterService;
+    @Override
+    public List<Video> recall(RecallParam param) {
+        // 1 获取省份key 放入参数map中
+        String provinceCn = param.getProvince();
+        if (provinceCn == null){
+            provinceCn = "中国";
+        }else{
+            provinceCn = provinceCn.replaceAll("省$", "");
+        }
+        Map<String, String> param4Model = new HashMap<>(1);
+        param4Model.put("region_province", provinceCn);
+        // 2 通过model拿到召回list
+        ScorerPipeline4Recall pipeline = ScorerUtils.getScorerPipeline4Recall("feeds_recall_config_region_v3.conf");
+        List<List<Pair<Long, Double>>> results = pipeline.recall(param4Model);
+        List<Pair<Long, Double>> result = results.get(0);
+        for (int i=1; i<results.size(); ++i){
+            result.addAll(results.get(i));
+        }
+        Map<Long, Double> videoMap = new LinkedHashMap<>();
+        for (Pair<Long, Double> v: result){
+            videoMap.put(v.getLeft(), v.getRight());
+        }
+
+        FilterParam filterParam = FilterParamFactory.create(param, Lists.newArrayList(videoMap.keySet()));
+        filterParam.setForceTruncation(10000);
+        FilterResult filterResult = filterService.filter(filterParam);
+        List<Video> videosResult = new ArrayList<>();
+        if (filterResult != null && CollectionUtils.isNotEmpty(filterResult.getVideoIds())) {
+            filterResult.getVideoIds().stream().forEach(vid -> {
+                Video video = new Video();
+                video.setVideoId(vid);
+                video.setAbCode(param.getAbCode());
+                video.setRovScore(videoMap.get(vid));
+                video.setPushFrom(pushFrom());
+                videosResult.add(video);
+            });
+        }
+        Collections.sort(videosResult, Comparator.comparingDouble(o -> -o.getRovScore()));
+        return videosResult;
+    }
+    public static final String PUSH_FORM = "recall_strategy_region_24h";
+    @Override
+    public String pushFrom(){
+        return PUSH_FORM;
+    }
+
+}

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

@@ -0,0 +1,67 @@
+package com.tzld.piaoquan.recommend.server.service.recall.strategy;
+
+import com.google.common.collect.Lists;
+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.RegionFilterService;
+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.service.score.ScorerUtils;
+import com.tzld.piaoquan.recommend.server.service.score4recall.ScorerPipeline4Recall;
+import lombok.extern.slf4j.Slf4j;
+import org.apache.commons.collections4.CollectionUtils;
+import org.apache.commons.lang3.tuple.Pair;
+import org.springframework.beans.factory.annotation.Autowired;
+import org.springframework.stereotype.Component;
+
+import java.util.*;
+
+/**
+ * @author zhangbo
+ */
+@Slf4j
+@Component
+public class RegionRealtimeRecallStrategyV4 implements RecallStrategy {
+    @Autowired
+    protected RegionFilterService filterService;
+    @Override
+    public List<Video> recall(RecallParam param) {
+        // 2 通过model拿到召回list
+        Map<String, String> param4Model = new HashMap<>(0);
+        ScorerPipeline4Recall pipeline = ScorerUtils.getScorerPipeline4Recall("feeds_recall_config_region_v4.conf");
+        List<List<Pair<Long, Double>>> results = pipeline.recall(param4Model);
+        List<Pair<Long, Double>> result = results.get(0);
+        for (int i=1; i<results.size(); ++i){
+            result.addAll(results.get(i));
+        }
+        Map<Long, Double> videoMap = new LinkedHashMap<>();
+        for (Pair<Long, Double> v: result){
+            videoMap.put(v.getLeft(), v.getRight());
+        }
+
+        FilterParam filterParam = FilterParamFactory.create(param, Lists.newArrayList(videoMap.keySet()));
+        filterParam.setForceTruncation(10000);
+        FilterResult filterResult = filterService.filter(filterParam);
+        List<Video> videosResult = new ArrayList<>();
+        if (filterResult != null && CollectionUtils.isNotEmpty(filterResult.getVideoIds())) {
+            filterResult.getVideoIds().stream().forEach(vid -> {
+                Video video = new Video();
+                video.setVideoId(vid);
+                video.setAbCode(param.getAbCode());
+                video.setRovScore(videoMap.get(vid));
+                video.setPushFrom(pushFrom());
+                videosResult.add(video);
+            });
+        }
+        Collections.sort(videosResult, Comparator.comparingDouble(o -> -o.getRovScore()));
+        return videosResult;
+    }
+    public static final String PUSH_FORM = "recall_strategy_noregion_1h";
+    @Override
+    public String pushFrom(){
+        return PUSH_FORM;
+    }
+
+}

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

@@ -2,6 +2,8 @@ package com.tzld.piaoquan.recommend.server.service.score;
 
 
 import com.typesafe.config.Config;
+import com.tzld.piaoquan.recommend.server.service.score4recall.AbstractScorer4Recall;
+import com.tzld.piaoquan.recommend.server.service.score4recall.ScorerPipeline4Recall;
 import lombok.extern.slf4j.Slf4j;
 import org.apache.commons.lang.exception.ExceptionUtils;
 import org.slf4j.Logger;
@@ -18,6 +20,7 @@ public final class ScorerUtils {
     private static Logger LOGGER = LoggerFactory.getLogger(ScorerUtils.class);
 
     private static Map<String, ScorerPipeline> scorerPipelineCache = new ConcurrentHashMap<>();
+    private static Map<String, ScorerPipeline4Recall> scorerPipelineCache4Recall = new ConcurrentHashMap<>();
 
     public static String BASE_CONF = "feeds_score_config_baseline.conf";
     public static String VIDEO_SCORE_CONF_FOR_AD = "video_score_config_for_ad.conf";
@@ -30,6 +33,10 @@ public final class ScorerUtils {
         ScorerUtils.init(BASE_CONF);
         ScorerUtils.init(FLOWPOOL_CONF);
         ScorerUtils.init(VIDEO_SCORE_CONF_FOR_AD);
+        ScorerUtils.init4Recall("feeds_recall_config_region_v1.conf");
+        ScorerUtils.init4Recall("feeds_recall_config_region_v2.conf");
+        ScorerUtils.init4Recall("feeds_recall_config_region_v3.conf");
+        ScorerUtils.init4Recall("feeds_recall_config_region_v4.conf");
     }
 
     private ScorerUtils() {
@@ -44,6 +51,14 @@ public final class ScorerUtils {
         initLoadModel(scorers);
         scorerPipelineCache.put(configFile, new ScorerPipeline(scorers));
     }
+    public static void init4Recall(String configFile) {
+        ScorerConfig scorerConfig = new ScorerConfig();
+        scorerConfig.load(configFile);
+        List<AbstractScorer4Recall> scorers = ScorerUtils.constructScorers4Recall(scorerConfig);
+
+        initLoadModel4Recall(scorers);
+        scorerPipelineCache4Recall.put(configFile, new ScorerPipeline4Recall(scorers));
+    }
 
     /**
      * init load model
@@ -57,6 +72,13 @@ public final class ScorerUtils {
             }
         }
     }
+    public static void initLoadModel4Recall(List<AbstractScorer4Recall> scorers) {
+        for (AbstractScorer4Recall scorer : scorers) {
+            if (scorer.isEnable()) {
+                scorer.loadModel();
+            }
+        }
+    }
 
     public static void initLoadModel(String configFile) {
         ScorerConfig scorerConfig = new ScorerConfig();
@@ -86,6 +108,19 @@ public final class ScorerUtils {
         return pipeline;
     }
 
+    public static ScorerPipeline4Recall getScorerPipeline4Recall(String configFile) {
+        // 不需要保证严格意义的单例
+        if (scorerPipelineCache4Recall.containsKey(configFile)) {
+            return scorerPipelineCache4Recall.get(configFile);
+        }
+        ScorerConfig scorerConfig = new ScorerConfig();
+        scorerConfig.load(configFile);
+        List<AbstractScorer4Recall> scorers = ScorerUtils.constructScorers4Recall(scorerConfig);
+        ScorerPipeline4Recall pipeline = new ScorerPipeline4Recall(scorers);
+        scorerPipelineCache4Recall.put(configFile, pipeline);
+        return pipeline;
+    }
+
     public static ScorerPipeline getScorerPipeline(Config mergeConfig) {
         ScorerConfig scorerConfig = new ScorerConfig();
         scorerConfig.load(mergeConfig);
@@ -126,6 +161,32 @@ public final class ScorerUtils {
         return scorers;
     }
 
+    /**
+     * construct scorers
+     *
+     * @param scorerConfig
+     * @return
+     */
+    public static List<AbstractScorer4Recall> constructScorers4Recall(ScorerConfig scorerConfig) {
+
+        List<AbstractScorer4Recall> scorers = new ArrayList<AbstractScorer4Recall>();
+        for (ScorerConfigInfo configInfo : scorerConfig.getConfigInfoList()) {
+            if (!configInfo.getDisableSwitch()) {
+                try {
+                    AbstractScorer4Recall scorer = (AbstractScorer4Recall) Class.forName(configInfo.getScorerName())
+                            .getConstructor(ScorerConfigInfo.class)
+                            .newInstance(configInfo);
+                    scorers.add(scorer);
+                    LOGGER.debug("construct score [{}]", configInfo.getScorerName());
+                } catch (Exception e) {
+                    LOGGER.error("instance scorer {} failed {}", configInfo.getScorerName(), ExceptionUtils.getFullStackTrace(e));
+                }
+
+            }
+        }
+        return scorers;
+    }
+
     /**
      * construct scorers
      *

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

@@ -0,0 +1,63 @@
+package com.tzld.piaoquan.recommend.server.service.score4recall;
+
+
+import com.tzld.piaoquan.recommend.server.service.score.ScorerConfigInfo;
+import com.tzld.piaoquan.recommend.server.service.score.model.Model;
+import com.tzld.piaoquan.recommend.server.service.score.model.ModelManager;
+import org.apache.commons.lang3.StringUtils;
+import org.apache.commons.lang3.tuple.Pair;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.io.IOException;
+import java.util.List;
+import java.util.Map;
+
+
+public abstract class AbstractScorer4Recall {
+    public static Logger LOGGER = LoggerFactory.getLogger(AbstractScorer4Recall.class);
+    protected ScorerConfigInfo scorerConfigInfo;
+    protected ModelManager modelManager = ModelManager.getInstance();
+
+    public AbstractScorer4Recall(ScorerConfigInfo scorerConfigInfo) {
+        this.scorerConfigInfo = scorerConfigInfo;
+    }
+
+    public abstract void loadModel();
+
+    public boolean isEnable() {
+        return !getScorerConfigInfo().getDisableSwitch();
+    }
+
+
+    public void doLoadModel(Class<? extends Model> modelClass) {
+
+        String modelPath = scorerConfigInfo.getModelPath();
+        if (StringUtils.isNotBlank(modelPath)) {
+            try {
+                // 使用 modelPath 作为 modelName 注册
+                modelManager.registerModel(modelPath, modelPath, modelClass);
+                LOGGER.info("register model success, model path [{}], model class [{}]", modelPath, modelClass);
+            } catch (ModelManager.ModelRegisterException e) {
+                LOGGER.error("register model fail [{}]:[{}]", modelPath, e);
+            } catch (IOException e) {
+                LOGGER.error("register model fail [{}]:[{}]", modelPath, e);
+            }
+        } else {
+            LOGGER.error("modelpath is null, for model class [{}]", modelClass);
+        }
+    }
+    public Model getModel() {
+        if (StringUtils.isBlank(scorerConfigInfo.getModelPath())) {
+            return null;
+        }
+        return modelManager.getModel(scorerConfigInfo.getModelPath());
+    }
+
+    public ScorerConfigInfo getScorerConfigInfo() {
+        return scorerConfigInfo;
+    }
+
+    public abstract List<Pair<Long, Double>> recall(Map<String, String> params);
+
+}

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

@@ -0,0 +1,43 @@
+package com.tzld.piaoquan.recommend.server.service.score4recall;
+
+import com.tzld.piaoquan.recommend.feature.domain.video.base.UserFeature;
+import com.tzld.piaoquan.recommend.server.common.base.RankItem;
+import com.tzld.piaoquan.recommend.server.service.score.AbstractScorer;
+import com.tzld.piaoquan.recommend.server.service.score.ScoreParam;
+import lombok.extern.slf4j.Slf4j;
+import com.tzld.piaoquan.recommend.server.service.score.ScorerPipeline;
+import org.apache.commons.collections4.CollectionUtils;
+import org.apache.commons.lang.exception.ExceptionUtils;
+import org.apache.commons.lang3.tuple.Pair;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.util.ArrayList;
+import java.util.Arrays;
+import java.util.List;
+import java.util.Map;
+import java.util.concurrent.*;
+
+
+@Slf4j
+public class ScorerPipeline4Recall {
+    public static final int corePoolSize = 128;
+    public static final int SCORE_TIME_OUT = 400;
+    public static final Logger LOGGER = LoggerFactory.getLogger(ScorerPipeline4Recall.class);
+    public static final ExecutorService executorService = Executors.newFixedThreadPool(corePoolSize);
+
+    public List<AbstractScorer4Recall> scorers;
+
+    public ScorerPipeline4Recall(List<AbstractScorer4Recall> scorers) {
+        this.scorers = scorers;
+    }
+
+    //todo zhangbo 定义一个接口，对scores做循环，执行recall的功能，返回一个多个list。
+    public List<List<Pair<Long, Double>>> recall(Map<String, String> params) {
+        List<List<Pair<Long, Double>>> result = new ArrayList<>();
+        for (AbstractScorer4Recall scorer : scorers){
+            result.add(scorer.recall(params));
+        }
+        return result;
+    }
+}

recommend-server-service/src/main/java/com/tzld/piaoquan/recommend/server/service/score4recall/model4recall/AbstractModel.java

@@ -0,0 +1,14 @@
+package com.tzld.piaoquan.recommend.server.service.score4recall.model4recall;
+
+
+import java.util.List;
+import com.tzld.piaoquan.recommend.server.service.score.model.Model;
+
+abstract public class AbstractModel extends Model {
+    @Override
+    public int getModelSize(){
+        return -1;
+    }
+
+}
+

recommend-server-service/src/main/java/com/tzld/piaoquan/recommend/server/service/score4recall/model4recall/Model4RecallKeyValue.java

@@ -0,0 +1,58 @@
+package com.tzld.piaoquan.recommend.server.service.score4recall.model4recall;
+
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.io.BufferedReader;
+import java.io.IOException;
+import java.io.InputStreamReader;
+import java.util.ArrayList;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+import org.apache.commons.lang3.tuple.Pair;
+
+public class Model4RecallKeyValue extends AbstractModel {
+    private static final Logger LOGGER = LoggerFactory.getLogger(Model4RecallKeyValue.class);
+    public Map<String, List<Pair<Long, Double>>> kv;
+    public Model4RecallKeyValue() {
+        //配置不同环境的hdfs conf
+        this.kv = new HashMap<>();
+    }
+
+
+    @Override
+    public boolean loadFromStream(InputStreamReader in) throws IOException {
+        BufferedReader input = new BufferedReader(in);
+        String line = null;
+        while ((line = input.readLine()) != null) {
+            String[] items = line.split("\t");
+            if (items.length < 3) {
+                continue;
+            }
+            String key = items[0].trim();
+            String [] l1 = items[1].trim().split(",");
+            String [] l2 = items[2].trim().split(",");
+            if (l1.length != l2.length){
+                continue;
+            }
+            List<Pair<Long, Double>> lists = new ArrayList<>(l1.length);
+            for (int i=0; i<l1.length; ++i){
+                String id = l1[i];
+                String score = l2[i];
+                try{
+                    Pair<Long, Double> pair = Pair.of(Long.valueOf(id), Double.valueOf(score));
+                    lists.add(pair);
+                }catch (Exception e){
+                    LOGGER.error(String.format("something is wrong with parse pair %s %s: ", id, score), e);
+                }
+            }
+            this.kv.put(key, lists);
+        }
+        input.close();
+        in.close();
+        return true;
+    }
+
+
+}

recommend-server-service/src/main/java/com/tzld/piaoquan/recommend/server/service/score4recall/model4recall/Model4RecallList.java

@@ -0,0 +1,46 @@
+package com.tzld.piaoquan.recommend.server.service.score4recall.model4recall;
+
+import org.apache.commons.lang3.tuple.Pair;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.io.BufferedReader;
+import java.io.IOException;
+import java.io.InputStreamReader;
+import java.util.ArrayList;
+import java.util.HashMap;
+import java.util.List;
+
+
+public class Model4RecallList extends AbstractModel {
+    private static final Logger LOGGER = LoggerFactory.getLogger(Model4RecallList.class);
+    public List<Pair<Long, Double>> recallList;
+    public Model4RecallList() {
+        //配置不同环境的hdfs conf
+        this.recallList = new ArrayList<>();
+    }
+
+
+    @Override
+    public boolean loadFromStream(InputStreamReader in) throws IOException {
+        BufferedReader input = new BufferedReader(in);
+        String line = null;
+        while ((line = input.readLine()) != null) {
+            String[] items = line.split("\t");
+            if (items.length < 2) {
+                continue;
+            }
+            String id = items[0].trim();
+            String score = items[1].trim();
+            try{
+                Pair<Long, Double> pair = Pair.of(Long.valueOf(id), Double.valueOf(score));
+                this.recallList.add(pair);
+            }catch (Exception e){
+                LOGGER.error(String.format("something is wrong with parse pair %s %s: ", id, score), e);
+            }
+        }
+        input.close();
+        in.close();
+        return true;
+    }
+}

recommend-server-service/src/main/java/com/tzld/piaoquan/recommend/server/service/score4recall/strategy/RegionRecallScorerV1.java

@@ -0,0 +1,37 @@
+package com.tzld.piaoquan.recommend.server.service.score4recall.strategy;
+
+import com.tzld.piaoquan.recommend.server.service.score4recall.AbstractScorer4Recall;
+import com.tzld.piaoquan.recommend.server.service.score.ScorerConfigInfo;
+import com.tzld.piaoquan.recommend.server.service.score4recall.model4recall.Model4RecallKeyValue;
+import org.apache.commons.lang3.tuple.Pair;
+
+import java.util.ArrayList;
+import java.util.List;
+import java.util.Map;
+
+
+public class RegionRecallScorerV1 extends AbstractScorer4Recall {
+
+    public RegionRecallScorerV1(ScorerConfigInfo configInfo) {
+        super(configInfo);
+    }
+    @Override
+    public void loadModel() {
+        doLoadModel(Model4RecallKeyValue.class);
+    }
+
+    @Override
+    public List<Pair<Long, Double>> recall(Map<String, String> params){
+        // todo zhangbo 这里要写实现功能
+        Model4RecallKeyValue model = (Model4RecallKeyValue) this.getModel();
+        String key = params.getOrDefault("region_province", "中国");
+        List<Pair<Long, Double>> lists = model.kv.getOrDefault(key, new ArrayList<>());
+        if (lists.isEmpty()){
+            lists = model.kv.getOrDefault("中国", new ArrayList<>());
+        }
+
+        return lists.subList(0, Math.min(100, lists.size()));
+    }
+
+
+}

recommend-server-service/src/main/java/com/tzld/piaoquan/recommend/server/service/score4recall/strategy/RegionRecallScorerV2.java

@@ -0,0 +1,31 @@
+package com.tzld.piaoquan.recommend.server.service.score4recall.strategy;
+
+import com.tzld.piaoquan.recommend.server.service.score.ScorerConfigInfo;
+import com.tzld.piaoquan.recommend.server.service.score4recall.AbstractScorer4Recall;
+import com.tzld.piaoquan.recommend.server.service.score4recall.model4recall.Model4RecallList;
+import org.apache.commons.lang3.tuple.Pair;
+
+import java.util.List;
+import java.util.Map;
+
+
+public class RegionRecallScorerV2 extends AbstractScorer4Recall {
+
+    public RegionRecallScorerV2(ScorerConfigInfo configInfo) {
+        super(configInfo);
+    }
+    @Override
+    public void loadModel() {
+        doLoadModel(Model4RecallList.class);
+    }
+
+    @Override
+    public List<Pair<Long, Double>> recall(Map<String, String> params){
+        // todo zhangbo 这里要写实现功能
+        Model4RecallList model = (Model4RecallList) this.getModel();
+        List<Pair<Long, Double>> lists = model.recallList;
+        return lists.subList(0, Math.min(50, lists.size()));
+    }
+
+
+}

recommend-server-service/src/main/java/com/tzld/piaoquan/recommend/server/service/score4recall/strategy/RegionRecallScorerV3.java

@@ -0,0 +1,37 @@
+package com.tzld.piaoquan.recommend.server.service.score4recall.strategy;
+
+import com.tzld.piaoquan.recommend.server.service.score.ScorerConfigInfo;
+import com.tzld.piaoquan.recommend.server.service.score4recall.AbstractScorer4Recall;
+import com.tzld.piaoquan.recommend.server.service.score4recall.model4recall.Model4RecallKeyValue;
+import org.apache.commons.lang3.tuple.Pair;
+
+import java.util.ArrayList;
+import java.util.List;
+import java.util.Map;
+
+
+public class RegionRecallScorerV3 extends AbstractScorer4Recall {
+
+    public RegionRecallScorerV3(ScorerConfigInfo configInfo) {
+        super(configInfo);
+    }
+    @Override
+    public void loadModel() {
+        doLoadModel(Model4RecallKeyValue.class);
+    }
+
+    @Override
+    public List<Pair<Long, Double>> recall(Map<String, String> params){
+        // todo zhangbo 这里要写实现功能
+        Model4RecallKeyValue model = (Model4RecallKeyValue) this.getModel();
+        String key = params.getOrDefault("region_province", "中国");
+        List<Pair<Long, Double>> lists = model.kv.getOrDefault(key, new ArrayList<>());
+        if (lists.isEmpty()){
+            lists = model.kv.getOrDefault("中国", new ArrayList<>());
+        }
+
+        return lists.subList(0, Math.min(100, lists.size()));
+    }
+
+
+}

recommend-server-service/src/main/java/com/tzld/piaoquan/recommend/server/service/score4recall/strategy/RegionRecallScorerV4.java

@@ -0,0 +1,31 @@
+package com.tzld.piaoquan.recommend.server.service.score4recall.strategy;
+
+import com.tzld.piaoquan.recommend.server.service.score.ScorerConfigInfo;
+import com.tzld.piaoquan.recommend.server.service.score4recall.AbstractScorer4Recall;
+import com.tzld.piaoquan.recommend.server.service.score4recall.model4recall.Model4RecallList;
+import org.apache.commons.lang3.tuple.Pair;
+
+import java.util.List;
+import java.util.Map;
+
+
+public class RegionRecallScorerV4 extends AbstractScorer4Recall {
+
+    public RegionRecallScorerV4(ScorerConfigInfo configInfo) {
+        super(configInfo);
+    }
+    @Override
+    public void loadModel() {
+        doLoadModel(Model4RecallList.class);
+    }
+
+    @Override
+    public List<Pair<Long, Double>> recall(Map<String, String> params){
+        // todo zhangbo 这里要写实现功能
+        Model4RecallList model = (Model4RecallList) this.getModel();
+        List<Pair<Long, Double>> lists = model.recallList;
+        return lists.subList(0, Math.min(50, lists.size()));
+    }
+
+
+}

recommend-server-service/src/main/resources/feeds_recall_config_region_v1.conf

@@ -0,0 +1,8 @@
+scorer-config = {
+    score1-config = {
+        scorer-name = "com.tzld.piaoquan.recommend.server.service.score4recall.strategy.RegionRecallScorerV1"
+        scorer-priority = 99
+        model-path = "alg_recall_file/01_recall_region1hour.txt"
+    }
+
+}

recommend-server-service/src/main/resources/feeds_recall_config_region_v2.conf

@@ -0,0 +1,9 @@
+scorer-config = {
+
+    score2-config = {
+        scorer-name = "com.tzld.piaoquan.recommend.server.service.score4recall.strategy.RegionRecallScorerV2"
+        scorer-priority = 98
+        model-path = "alg_recall_file/02_recall_noregion24hour.txt"
+    }
+
+}

recommend-server-service/src/main/resources/feeds_recall_config_region_v3.conf

@@ -0,0 +1,9 @@
+scorer-config = {
+
+    score3-config = {
+        scorer-name = "com.tzld.piaoquan.recommend.server.service.score4recall.strategy.RegionRecallScorerV3"
+        scorer-priority = 97
+        model-path = "alg_recall_file/03_recall_region24hour.txt"
+    }
+
+}

recommend-server-service/src/main/resources/feeds_recall_config_region_v4.conf

@@ -0,0 +1,8 @@
+scorer-config = {
+
+    score4-config = {
+        scorer-name = "com.tzld.piaoquan.recommend.server.service.score4recall.strategy.RegionRecallScorerV4"
+        scorer-priority = 96
+        model-path = "alg_recall_file/04_recall_noregion1h.txt"
+    }
+}