filter refactor

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

@@ -6,6 +6,8 @@ import lombok.Data;
 public class User {
 
     private String id;
+    private String uid;
+    private String mid;
     private UserAttention userAttention;
 
     // user profile 基础画像信息

recommend-server-service/src/main/java/com/tzld/piaoquan/recommend/server/framework/recaller/AbstractFilter.java

@@ -0,0 +1,29 @@
+package com.tzld.piaoquan.recommend.server.framework.recaller;
+
+import com.tzld.piaoquan.recommend.server.framework.candidiate.Candidate;
+import com.tzld.piaoquan.recommend.server.framework.common.User;
+import com.tzld.piaoquan.recommend.server.gen.recommend.RecommendRequest;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+public abstract class AbstractFilter<T> {
+    protected final static Logger LOGGER = LoggerFactory.getLogger(AbstractFilter.class);
+    protected final FilterConfigInfo filterConfigInfo;
+    protected final RecommendRequest requestContext;
+    protected final User user;
+
+    public AbstractFilter(FilterConfigInfo filterConfigInfo,
+                          RecommendRequest requestContext,
+                          User user) {
+
+        this.filterConfigInfo = filterConfigInfo;
+        this.requestContext = requestContext;
+        this.user = user;
+    }
+
+    public FilterConfigInfo getFilterConfigInfo() {
+        return filterConfigInfo;
+    }
+
+    public abstract boolean predicate(Candidate candidate, T t);
+}

recommend-server-service/src/main/java/com/tzld/piaoquan/recommend/server/framework/recaller/BaseRecaller.java

@@ -1,6 +1,7 @@
 package com.tzld.piaoquan.recommend.server.framework.recaller;
 
 
+import com.google.common.base.Predicate;
 import com.google.common.collect.FluentIterable;
 import com.google.common.collect.Lists;
 import com.google.common.collect.Maps;
@@ -11,6 +12,7 @@ import com.tzld.piaoquan.recommend.server.framework.recaller.provider.QueueProvi
 import com.tzld.piaoquan.recommend.server.gen.recommend.RecommendRequest;
 import org.apache.commons.collections4.CollectionUtils;
 import org.apache.commons.lang.exception.ExceptionUtils;
+import org.checkerframework.checker.nullness.qual.Nullable;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
@@ -26,9 +28,14 @@ public class BaseRecaller<Video> {
     private static final Logger LOGGER = LoggerFactory.getLogger(BaseRecaller.class);
     private static final long DEFAULT_QUEUE_LOAD_TIMEOUT = 150; // ms
     private static final long DEFAULT_PARALLEL_FILTER_TIMEOUT = 200; // ms
+
+    private static final String FILTER_CONF = ""; // ms
     private static final ExecutorService filterExecutorService = Executors.newFixedThreadPool(128);
     private static final ExecutorService fetchQueueExecutorService = Executors.newFixedThreadPool(128);
     private final QueueProvider<Video> queueProvider;
+
+    private final FilterConfig filterConfig;
+
     private final long QUEUE_LOAD_TIMEOUT;
 
     public BaseRecaller(QueueProvider<Video> queueProvider) {
@@ -38,9 +45,20 @@ public class BaseRecaller<Video> {
     public BaseRecaller(QueueProvider<Video> queueProvider,
                         long queueLoadTimeout) {
         this.queueProvider = queueProvider;
+        this.filterConfig = new FilterConfig(FILTER_CONF);
         this.QUEUE_LOAD_TIMEOUT = queueLoadTimeout;
     }
 
+
+    public BaseRecaller(QueueProvider<Video> queueProvider,
+                        FilterConfig filterConfig,
+                        long queueLoadTimeout) {
+        this.queueProvider = queueProvider;
+        this.filterConfig = filterConfig;
+        this.QUEUE_LOAD_TIMEOUT = queueLoadTimeout;
+    }
+
+
     public String extractItemId(Entry<Video> entry) {
         return entry.id;
     }
@@ -162,14 +180,20 @@ public class BaseRecaller<Video> {
             }
         }
 
-        List<RankItem> result = convertToRankItem(candidateQueueMap);
+        List<RankItem> result = convertToRankItem(candidateQueueMap, requestData, user);
         return result;
     }
 
 
     // 转成RankItem
     // 同时给Filter预留处理
-    private List<RankItem> convertToRankItem(Map<Candidate, Queue<Video>> candidateQueueMap) {
+    private List<RankItem> convertToRankItem(Map<Candidate, Queue<Video>> candidateQueueMap,
+                                             RecommendRequest requestData,
+                                             User user) {
+
+
+        final FilterPipeline<Video> recallFilter = new FilterPipeline<Video>(this.filterConfig, requestData, user);
+
 
         final List<Callable<List<RankItem>>> callables = new ArrayList<Callable<List<RankItem>>>();
         int expectedRecallSum = 0;
@@ -181,7 +205,12 @@ public class BaseRecaller<Video> {
                     final Candidate candidate = entry.getKey();
                     try {
                         // 1. filter  TODO 待后续增加自定义filter
-                        Iterable<Entry<Video>> entries = FluentIterable.from(entry.getValue()).limit(candidate.getCandidateNum());
+                        Iterable<Entry<Video>> entries = FluentIterable.from(entry.getValue()).filter(new Predicate<Entry<Video>>() {
+                            @Override
+                            public boolean apply(@Nullable Entry<Video> videoEntry) {
+                                return recallFilter.predicate(candidate, videoEntry.item);
+                            }
+                        }).limit(candidate.getCandidateNum());
 
                         // 2. toHits
                         candidateHits.addAll(toHits(entries, candidate));

recommend-server-service/src/main/java/com/tzld/piaoquan/recommend/server/framework/recaller/FilterConfig.java

@@ -0,0 +1,99 @@
+package com.tzld.piaoquan.recommend.server.framework.recaller;
+
+
+import com.typesafe.config.Config;
+import com.typesafe.config.ConfigFactory;
+import com.typesafe.config.ConfigObject;
+import com.typesafe.config.ConfigValue;
+
+import org.apache.commons.lang.exception.ExceptionUtils;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.util.ArrayList;
+import java.util.List;
+
+/**
+
+ */
+
+public class FilterConfig {
+    private static Logger LOGGER = LoggerFactory.getLogger(FilterConfig.class);
+    private List<FilterConfigInfo> filterConfigInfoList = new ArrayList<FilterConfigInfo>();
+
+    public FilterConfig(Config config) {
+        this.load(config);
+    }
+
+
+    public FilterConfig(String configFile) {
+        this.load(configFile);
+    }
+
+    public FilterConfig() {
+    }
+
+    public boolean load(String configFile) {
+        Config filterConfig = ConfigFactory.parseResources(configFile);
+        return load(filterConfig);
+    }
+
+    public boolean load(Config config) {
+        Config recallConfig = config.getConfig("recall-config");
+
+        Config filterConf = recallConfig.getConfig("filter-config");
+        try {
+            loadFilters(filterConf);
+            int pos = 0;
+            for (FilterConfigInfo filterConfigInfo : filterConfigInfoList) {
+                LOGGER.info("filter at position [{}], priority [{}] filter name [{}] ",
+                        new Object[]{pos++, filterConfigInfo.getFilterPriority(), filterConfigInfo.getConfigName()});
+            }
+            LOGGER.debug("Load filter config success");
+        } catch (Exception e) {
+            LOGGER.error("Load filter config failed, [{}]", ExceptionUtils.getFullStackTrace(e));
+            return false;
+        }
+
+        return true;
+    }
+
+    public List<FilterConfigInfo> getFilterConfigInfoList() {
+        return filterConfigInfoList;
+    }
+
+    private void loadFilters(Config config) throws Exception {
+        ConfigObject confObj = config.root();
+        for (ConfigObject.Entry<String, ConfigValue> it : confObj.entrySet()) {
+            Config conf = ((ConfigObject) it.getValue()).toConfig();
+
+            // parse config
+            String configName = it.getKey();
+            String filterName = conf.getString("filter-name");
+            int filterPriority = 0;
+            if (conf.hasPath("filter-priority"))
+                filterPriority = conf.getInt("filter-priority");
+
+            List<Integer> disableExpIds = new ArrayList<Integer>();
+            if (conf.hasPath("disable-expids")) {
+                disableExpIds = conf.getIntList("disable-expids");
+            }
+            FilterConfigInfo filterConfigInfo = new FilterConfigInfo(configName,
+                    filterName, filterPriority, disableExpIds);
+            LOGGER.debug("parse filter config info [{}]", filterConfigInfo);
+
+            addConfigByPriority(filterConfigInfoList, filterConfigInfo);
+        }
+    }
+
+    private void addConfigByPriority(List<FilterConfigInfo> configInfoList, FilterConfigInfo addConfigInfo) {
+
+        int pos = 0;
+        for (; pos < configInfoList.size(); pos++) {
+            if (configInfoList.get(pos).getFilterPriority() <= addConfigInfo.getFilterPriority()) {
+                break;
+            }
+        }
+        configInfoList.add(pos, addConfigInfo);
+    }
+}

recommend-server-service/src/main/java/com/tzld/piaoquan/recommend/server/framework/recaller/FilterConfigInfo.java

@@ -0,0 +1,59 @@
+package com.tzld.piaoquan.recommend.server.framework.recaller;
+
+
+import com.google.common.base.Function;
+import com.google.common.base.Joiner;
+import com.google.common.collect.FluentIterable;
+
+import java.util.HashSet;
+import java.util.List;
+import java.util.Set;
+
+import javax.annotation.Nullable;
+
+
+public class FilterConfigInfo {
+    private Set<Integer> disableExpIds;
+    private String filterName;
+    private Integer filterPriority;
+    private String configName;
+
+    public FilterConfigInfo(String configName,
+                            String filterName,
+                            Integer filterPriority,
+                            List<Integer> expIds) {
+        this.configName = configName;
+        this.filterName = filterName;
+        this.filterPriority = filterPriority;
+        this.disableExpIds = new HashSet<Integer>();
+        this.disableExpIds.addAll(expIds);
+    }
+
+    public Integer getFilterPriority() {
+        return filterPriority;
+    }
+
+    public String getFilterName() {
+        return filterName;
+    }
+
+    public Set<Integer> getDisableExpIds() {
+        return disableExpIds;
+    }
+
+    public String getConfigName() {
+        return configName;
+    }
+
+    @Override
+    public String toString() {
+        return configName + ":" + filterPriority + ":" + filterName + ":" +
+                Joiner.on(",").join(FluentIterable.from(disableExpIds).transform(new Function<Integer, String>() {
+                    @Nullable
+                    @Override
+                    public String apply(Integer integer) {
+                        return integer.toString();
+                    }
+                }));
+    }
+}

recommend-server-service/src/main/java/com/tzld/piaoquan/recommend/server/framework/recaller/FilterPipeline.java

@@ -0,0 +1,65 @@
+package com.tzld.piaoquan.recommend.server.framework.recaller;
+
+import com.tzld.piaoquan.recommend.server.framework.common.User;
+import com.tzld.piaoquan.recommend.server.gen.recommend.RecommendRequest;
+import org.apache.commons.lang.exception.ExceptionUtils;
+import com.tzld.piaoquan.recommend.server.framework.candidiate.Candidate;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.util.ArrayList;
+import java.util.List;
+
+
+public class FilterPipeline<T> {
+
+    private static Logger LOGGER = LoggerFactory.getLogger(FilterPipeline.class);
+    public int filterNum;
+    private FilterConfig config;
+    private List<AbstractFilter<T>> filters;
+    private RecommendRequest requestContext;
+    private User user;
+    private int requestIndex;
+
+    public FilterPipeline(FilterConfig config,
+                        RecommendRequest requestContext,
+                        User user) {
+
+        this.config = config;
+        this.requestContext = requestContext;
+        this.user = user;
+        this.filters = new ArrayList<AbstractFilter<T>>();
+        this.constructFilters(config);
+    }
+
+    public List<AbstractFilter<T>> getFilters() {
+        return filters;
+    }
+
+    public void constructFilters(FilterConfig config) {
+        // add filter
+        if (config != null) {
+            for (FilterConfigInfo filterConfigInfo : config.getFilterConfigInfoList()) {
+                try {
+                    AbstractFilter<T> filter = (AbstractFilter) Class.forName(filterConfigInfo.getFilterName())
+                            .getConstructor(FilterConfigInfo.class, RecommendRequest.class, User.class, Integer.class)
+                            .newInstance(filterConfigInfo, this.requestContext, this.user, this.requestIndex);
+
+                    this.filters.add(filter);
+                } catch (Exception e) {
+                    LOGGER.error("Filter config info construct error, [{}] [{}]",
+                            filterConfigInfo, ExceptionUtils.getFullStackTrace(e));
+                }
+            }
+        }
+    }
+
+    public boolean predicate(Candidate candidate, T t) {
+        for (AbstractFilter filter : filters) {
+            if (!filter.predicate(candidate, t)) {
+                return false;
+            }
+        }
+        return true;
+    }
+}

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

@@ -0,0 +1,100 @@
+package com.tzld.piaoquan.recommend.server.implement.recall;
+
+
+import com.google.common.collect.Lists;
+import com.google.common.hash.Hashing;
+import com.tzld.piaoquan.recommend.server.common.enums.AppTypeEnum;
+import com.tzld.piaoquan.recommend.server.framework.candidiate.Candidate;
+import com.tzld.piaoquan.recommend.server.framework.common.User;
+import com.tzld.piaoquan.recommend.server.framework.recaller.AbstractFilter;
+import com.tzld.piaoquan.recommend.server.framework.recaller.FilterConfigInfo;
+import com.tzld.piaoquan.recommend.server.gen.recommend.RecommendRequest;
+import com.tzld.piaoquan.recommend.server.model.Video;
+import com.tzld.piaoquan.recommend.server.service.PreViewedService;
+import com.tzld.piaoquan.recommend.server.service.filter.FilterParam;
+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.data.redis.core.RedisTemplate;
+
+import java.util.*;
+
+public class AllowListFilter extends AbstractFilter<Video> {
+
+    @Autowired
+    @Qualifier("longVideoRedisTemplate")
+    private RedisTemplate<String, String> redisTemplate;
+
+    private static final String VIDEO_ALLOW_LIST_BITMAP_KEY_SET_PREFIX = "movie:videoid:allowSet:";
+    private static final String RELIGION_VIDEO_ALLOW_LIST_BITMAP_KEY = "mp:religion:allowlist:videoid:bitmap";
+
+    public AllowListFilter(FilterConfigInfo filterConfigInfo,
+                           RecommendRequest recommendRequest,
+                           User user) {
+        super(filterConfigInfo, recommendRequest, user);
+
+    }
+
+    public boolean predicate(Candidate candidate, Video video) {
+        return !isAllowList(user, video);
+    }
+
+
+    public boolean isAllowList(User user, Video video) {
+        //不是新小程序 在白名单则不显示
+        Set<Long> retainVideoIds = new LinkedHashSet<>();
+        int appType = requestContext.getAppType();
+        if ( appType != AppTypeEnum.WAN_NENG_VIDEO.getCode()
+                && appType != AppTypeEnum.LAO_HAO_KAN_VIDEO.getCode()
+                && appType != AppTypeEnum.ZUI_JING_QI.getCode()
+                && appType!= AppTypeEnum.H5.getCode()) {
+            // 如果不在新小程序白名单，则允许
+            return !isMemberOfVideoAllowList(video.getVideoId());
+
+        } else if (appType == AppTypeEnum.WAN_NENG_VIDEO.getCode()
+                || appType == AppTypeEnum.LAO_HAO_KAN_VIDEO.getCode()
+                || appType == AppTypeEnum.ZUI_JING_QI.getCode()
+                || appType == AppTypeEnum.H5.getCode()) {
+            return !isMemberOfReligionVideoAllowList(video.getVideoId());
+
+        }
+        return true;
+    }
+
+    private boolean isMemberOfVideoAllowList(Long videoId) {
+        if (Objects.isNull(videoId)) {
+            return false;
+        }
+        try {
+            int newIdx = Math.abs(Hashing.murmur3_32().hashLong(videoId).asInt()) % 100;
+            String newPrefix = VIDEO_ALLOW_LIST_BITMAP_KEY_SET_PREFIX + newIdx;
+            Boolean result = redisTemplate.opsForSet().isMember(newPrefix, String.valueOf(videoId));
+            return result;
+        } catch (Exception e) {
+            LOGGER.error("isMemberOfVideoAllowList error {}", videoId, e);
+        }
+        return false;
+    }
+
+    public boolean isMemberOfReligionVideoAllowList(Long videoId) {
+        if (Objects.isNull(videoId)) {
+            return false;
+        }
+        try {
+            return redisTemplate.opsForValue().getBit(RELIGION_VIDEO_ALLOW_LIST_BITMAP_KEY, videoId);
+        } catch (Exception e) {
+            LOGGER.error("isMemberOfReligionVideoAllowList error {}", e, videoId);
+        }
+        return false;
+    }
+
+
+
+
+    
+}
+
+
+
+

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

@@ -0,0 +1,49 @@
+package com.tzld.piaoquan.recommend.server.implement.recall;
+
+
+import com.tzld.piaoquan.recommend.server.framework.candidiate.Candidate;
+import com.tzld.piaoquan.recommend.server.framework.common.User;
+import com.tzld.piaoquan.recommend.server.framework.recaller.AbstractFilter;
+import com.tzld.piaoquan.recommend.server.framework.recaller.FilterConfigInfo;
+import com.tzld.piaoquan.recommend.server.gen.recommend.RecommendRequest;
+import com.tzld.piaoquan.recommend.server.model.Video;
+import com.tzld.piaoquan.recommend.server.service.PreViewedService;
+import org.apache.commons.lang3.StringUtils;
+import org.springframework.beans.factory.annotation.Autowired;
+
+import java.util.Set;
+
+public class PreViewedFilter extends AbstractFilter<Video> {
+
+    @Autowired
+    private PreViewedService preViewedService;
+
+    public PreViewedFilter(FilterConfigInfo filterConfigInfo,
+                           RecommendRequest recommendRequest,
+                           User user) {
+        super(filterConfigInfo, recommendRequest, user);
+
+    }
+
+    public boolean predicate(Candidate candidate, Video video) {
+        return !isPreViewed(user, video);
+    }
+
+
+    public boolean isPreViewed(User user, Video video) {
+
+        if (StringUtils.isBlank(user.getMid())
+                || StringUtils.isBlank(String.valueOf(video.getVideoId()))) {
+            return false;
+        }
+        Set<Long> preViewedVideoIds = preViewedService.getVideoIds(this.requestContext.getAppType(), user.getMid());
+
+        return preViewedVideoIds.contains(video.getVideoId());
+    }
+
+
+}
+
+
+
+

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

@@ -0,0 +1,71 @@
+package com.tzld.piaoquan.recommend.server.implement.recall;
+
+
+import com.tzld.piaoquan.recommend.server.common.ThreadPoolFactory;
+import com.tzld.piaoquan.recommend.server.framework.candidiate.Candidate;
+import com.tzld.piaoquan.recommend.server.framework.common.User;
+import com.tzld.piaoquan.recommend.server.framework.recaller.AbstractFilter;
+import com.tzld.piaoquan.recommend.server.framework.recaller.FilterConfigInfo;
+import com.tzld.piaoquan.recommend.server.gen.recommend.RecommendRequest;
+import com.tzld.piaoquan.recommend.server.model.Video;
+import com.tzld.piaoquan.recommend.server.repository.WxVideoStatus;
+import com.tzld.piaoquan.recommend.server.repository.WxVideoStatusRepository;
+import com.tzld.piaoquan.recommend.server.service.PreViewedService;
+import org.apache.commons.collections4.CollectionUtils;
+import org.apache.commons.lang3.RandomUtils;
+import org.apache.commons.lang3.StringUtils;
+import org.apache.commons.lang3.math.NumberUtils;
+import org.springframework.beans.factory.annotation.Autowired;
+import org.springframework.beans.factory.annotation.Qualifier;
+import org.springframework.dao.DataAccessException;
+import org.springframework.data.redis.core.RedisOperations;
+import org.springframework.data.redis.core.RedisTemplate;
+import org.springframework.data.redis.core.SessionCallback;
+import org.springframework.data.redis.core.ValueOperations;
+
+import java.util.*;
+import java.util.concurrent.TimeUnit;
+import java.util.stream.Collectors;
+
+public class RecommendStatusFilter extends AbstractFilter<Video> {
+
+    @Autowired
+    @Qualifier("redisTemplate")
+    private RedisTemplate<String, String> redisTemplate;
+
+    @Autowired
+    private WxVideoStatusRepository wxVideoStatusRepository;
+
+    private String keyFormat = "video:recommend:status:%s";
+
+    private static final int RECOMMEND_STATUS = -6;
+
+
+    @Autowired
+    private PreViewedService preViewedService;
+
+    public RecommendStatusFilter(FilterConfigInfo filterConfigInfo,
+                                 RecommendRequest recommendRequest,
+                                 User user) {
+        super(filterConfigInfo, recommendRequest, user);
+
+    }
+
+    public boolean predicate(Candidate candidate, Video video) {
+        return isStatusAllow(user, video);
+    }
+
+    public boolean isStatusAllow(User user, Video video){
+        //
+        return true;
+
+
+    }
+
+
+
+}
+
+
+
+

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

@@ -0,0 +1,41 @@
+package com.tzld.piaoquan.recommend.server.implement.recall;
+
+
+import com.tzld.piaoquan.recommend.server.framework.candidiate.Candidate;
+import com.tzld.piaoquan.recommend.server.framework.common.User;
+import com.tzld.piaoquan.recommend.server.framework.recaller.AbstractFilter;
+import com.tzld.piaoquan.recommend.server.framework.recaller.FilterConfigInfo;
+import com.tzld.piaoquan.recommend.server.gen.recommend.RecommendRequest;
+import com.tzld.piaoquan.recommend.server.model.Video;
+import org.apache.commons.collections4.CollectionUtils;
+
+import java.util.*;
+
+public class RiskVideoFilter extends AbstractFilter<Video> {
+
+
+    public RiskVideoFilter(FilterConfigInfo filterConfigInfo,
+                           RecommendRequest recommendRequest,
+                           User user) {
+        super(filterConfigInfo, recommendRequest, user);
+    }
+
+    public boolean predicate(Candidate candidate, Video video) {
+
+        return !isRiskVideo(video);
+
+
+    }
+
+
+    public boolean isRiskVideo(Video video) {
+
+     // TODO 风险过滤标签加载video中
+     //
+
+
+
+
+
+
+}

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

@@ -0,0 +1,100 @@
+package com.tzld.piaoquan.recommend.server.implement.recall;
+
+
+import com.google.common.cache.CacheBuilder;
+import com.google.common.cache.CacheLoader;
+import com.google.common.cache.LoadingCache;
+import com.google.common.collect.Lists;
+import com.tzld.piaoquan.recommend.server.common.ThreadPoolFactory;
+import com.tzld.piaoquan.recommend.server.framework.candidiate.Candidate;
+import com.tzld.piaoquan.recommend.server.framework.common.User;
+import com.tzld.piaoquan.recommend.server.framework.recaller.AbstractFilter;
+import com.tzld.piaoquan.recommend.server.framework.recaller.FilterConfigInfo;
+import com.tzld.piaoquan.recommend.server.gen.recommend.RecommendRequest;
+import com.tzld.piaoquan.recommend.server.model.Video;
+import com.tzld.piaoquan.recommend.server.repository.WxVideoTagRel;
+import com.tzld.piaoquan.recommend.server.repository.WxVideoTagRelRepository;
+import com.tzld.piaoquan.recommend.server.util.CommonCollectionUtils;
+import org.apache.commons.collections4.CollectionUtils;
+import org.apache.commons.lang.StringUtils;
+import org.apache.commons.lang.math.NumberUtils;
+import org.springframework.beans.factory.annotation.Autowired;
+import org.springframework.beans.factory.annotation.Value;
+
+import javax.annotation.PostConstruct;
+import java.util.*;
+import java.util.concurrent.TimeUnit;
+
+public class TagFilter extends AbstractFilter<Video> {
+
+    @Autowired
+    private WxVideoTagRelRepository repository;
+
+    @Value("${video.filter.tagids:}")
+    private String videoFilterTagIds;
+
+    private LoadingCache<Long, Set<Long>> videoTagCache = CacheBuilder.newBuilder()
+            .maximumSize(100)
+            .refreshAfterWrite(60, TimeUnit.SECONDS)
+            .expireAfterWrite(60, TimeUnit.SECONDS)
+            .expireAfterAccess(60, TimeUnit.SECONDS)
+            .build(new CacheLoader<Long, Set<Long>>() {
+                @Override
+                public Set<Long> load(Long tagId) {
+                    List<WxVideoTagRel> rels = repository.findAllByTagId(tagId);
+                    return CommonCollectionUtils.toSet(rels, WxVideoTagRel::getVideoId);
+                }
+            });
+
+
+    public TagFilter(FilterConfigInfo filterConfigInfo,
+                     RecommendRequest recommendRequest,
+                     User user) {
+        super(filterConfigInfo, recommendRequest, user);
+
+        if (org.apache.commons.lang.StringUtils.isNotBlank(this.videoFilterTagIds)) {
+            String[] tags = this.videoFilterTagIds.split(",");
+            for (String tag : tags) {
+                if (org.apache.commons.lang.StringUtils.isBlank(tag)) {
+                    continue;
+                }
+                this.videoTagCache.getUnchecked(NumberUtils.toLong(tag));
+            }
+        }
+    }
+
+
+    public boolean predicate(Candidate candidate, Video video) {
+        if (video == null ) {
+            return false;
+        }
+       return !hasVideoRelTagId(video.getVideoId(), this.videoFilterTagIds);
+    }
+
+
+    private boolean hasVideoRelTagId(Long videoId, String videoFilterTagIds) {
+        // TODO 主要是涉政标签
+        List<Long> tagIds = new ArrayList<>();
+
+        String[] tags = videoFilterTagIds.split(",");
+        for (String tag : tags) {
+            if (Objects.isNull(tag) || Objects.equals("", tag)) {
+                continue;
+            }
+            tagIds.add(Long.parseLong(tag));
+        }
+
+        for (Long tagId : tagIds) {
+            if (Objects.isNull(videoId) || videoId <= 0L || Objects.isNull(tagId) || tagId <= 0L) {
+                return false;
+            }
+            Set<Long> videos = this.videoTagCache.getUnchecked(tagId);
+            if (CollectionUtils.isEmpty(videos)) {
+                return false;
+            }
+            return videos.contains(videoId);
+        }
+        return false;
+    }
+
+}

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

@@ -0,0 +1,106 @@
+package com.tzld.piaoquan.recommend.server.implement.recall;
+
+
+import com.tzld.piaoquan.recommend.server.common.ThreadPoolFactory;
+import com.tzld.piaoquan.recommend.server.framework.candidiate.Candidate;
+import com.tzld.piaoquan.recommend.server.framework.common.User;
+import com.tzld.piaoquan.recommend.server.framework.recaller.AbstractFilter;
+import com.tzld.piaoquan.recommend.server.framework.recaller.FilterConfigInfo;
+import com.tzld.piaoquan.recommend.server.gen.recommend.RecommendRequest;
+import com.tzld.piaoquan.recommend.server.model.Video;
+import com.tzld.piaoquan.recommend.server.service.filter.strategy.VideoView;
+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.dao.DataAccessException;
+import org.springframework.data.mongodb.core.MongoTemplate;
+import org.springframework.data.mongodb.core.query.Criteria;
+import org.springframework.data.mongodb.core.query.Query;
+import org.springframework.data.redis.core.RedisOperations;
+import org.springframework.data.redis.core.RedisTemplate;
+import org.springframework.data.redis.core.SessionCallback;
+import org.springframework.data.redis.core.SetOperations;
+
+import java.util.HashSet;
+import java.util.List;
+import java.util.Set;
+import java.util.concurrent.TimeUnit;
+
+public class ViewedHistoryFilter extends AbstractFilter<Video> {
+
+
+    protected Set<String> historySet;
+
+    @Autowired
+    @Qualifier("longVideoRedisTemplate")
+    private RedisTemplate<String, String> redisTemplate;
+
+    @Autowired
+    private MongoTemplate mongoTemplate;
+
+    private String keyFormat = "user:exclude:videoidset:%s";
+
+    public ViewedHistoryFilter(FilterConfigInfo filterConfigInfo,
+                               RecommendRequest recommendRequest,
+                               User user) {
+        super(filterConfigInfo, recommendRequest, user);
+        historySet = this.loadUserHistory(user);
+        if (historySet == null)
+            historySet = new HashSet<>();
+    }
+
+    public boolean predicate(Candidate candidate, Video video) {
+        return !this.historySet.contains(String.valueOf(video.getVideoId()));
+    }
+
+
+    public Set<String> loadUserHistory(User user) {
+        String userid = StringUtils.isNotBlank(user.getUid()) ? user.getUid() : user.getMid();
+        if (StringUtils.isBlank(userid)) {
+            return new HashSet<String>();
+        }
+
+        String key = String.format(keyFormat, user.getMid());
+        Set<String> viewedVideoIds = redisTemplate.opsForSet().members(key);
+        if (CollectionUtils.isEmpty(viewedVideoIds)) {
+            // 从mongo取曝光数据
+            Criteria criteria = new Criteria();
+            criteria.and("uid").is(user);
+            Query query = new Query();
+            query.addCriteria(criteria);
+            List<VideoView> list = mongoTemplate.find(query, VideoView.class);
+
+            //TODO 为什么限制最多10000条？是不是限制近几天更合适？
+            if (CollectionUtils.isNotEmpty(list)) {
+                int limit = 10000;
+                for (int i = list.size() - 1; i >= 0 && limit-- > 0; i--) {
+                    viewedVideoIds.add(String.valueOf(list.get(i).getVideoId()));
+                }
+
+                // 异步写Redis
+                ThreadPoolFactory.defaultPool().execute(() -> {
+                    redisTemplate.executePipelined(new SessionCallback<String>() {
+                        @Override
+                        public <A, B> String execute(RedisOperations<A, B> redisOperations) throws DataAccessException {
+                            SetOperations<String, String> operations =
+                                    (SetOperations<String, String>) redisOperations.opsForSet();
+                            operations.add(key, viewedVideoIds.toArray(new String[viewedVideoIds.size()]));
+                            redisTemplate.expire(key, 360 * 3600, TimeUnit.SECONDS);
+                            return null;
+                        }
+                    });
+                });
+            }
+        }
+
+        return viewedVideoIds;
+    }
+
+
+
+
+
+
+
+}