增加第三方微信公众号调用接口

long-article-recommend-service/pom.xml

@@ -136,6 +136,12 @@
             <artifactId>poi-ooxml</artifactId>
             <version>5.2.3</version>
         </dependency>
+
+        <dependency>
+            <groupId>com.aliyun.openservices</groupId>
+            <artifactId>ons-client</artifactId>
+            <version>1.8.4.Final</version>
+        </dependency>
     </dependencies>
 
 

long-article-recommend-service/src/main/java/com/tzld/longarticle/recommend/server/common/enums/SecretEnum.java

@@ -0,0 +1,27 @@
+package com.tzld.longarticle.recommend.server.common.enums;
+
+
+import java.util.Objects;
+
+public enum SecretEnum {
+
+    SECRET_ENUM_1("fe1a0b632ff1a612b9e37da9c0ba2a25", "魅力"),
+    SECRET_ENUM_2("fe1a0b632ff1a612b9e37da9c0ba2a26", "老来福");
+
+    SecretEnum(String secret, String desc) {
+        this.secret = secret;
+        this.desc = desc;
+    }
+
+    public final String secret;
+    public final String desc;
+
+    public static boolean contains(String secret) {
+        for (SecretEnum secretEnum : SecretEnum.values()) {
+            if (Objects.equals(secretEnum.secret, secret)) {
+                return true;
+            }
+        }
+        return false;
+    }
+}

long-article-recommend-service/src/main/java/com/tzld/longarticle/recommend/server/common/enums/cgi/ReplyStrategyServiceEnum.java

@@ -2,23 +2,23 @@ package com.tzld.longarticle.recommend.server.common.enums.cgi;
 
 public enum ReplyStrategyServiceEnum {
 
-     BUCKET_STRATEGY_V1("BUCKET_STRATEGY_V1", "分桶策略v1"),
+    BUCKET_STRATEGY_V1("BUCKET_STRATEGY_V1", "分桶策略v1"),
 
-     ;
+    PUSH_MESSAGE_STRATEGY_V1("PUSH_MESSAGE_STRATEGY_V1", "第三方微信公众号推送策略V1");
 
-     private final String key;
-     private final String desc;
+    private final String key;
+    private final String desc;
 
     ReplyStrategyServiceEnum(String key, String desc) {
-         this.key = key;
-         this.desc = desc;
-     }
+        this.key = key;
+        this.desc = desc;
+    }
 
-     public String getKey() {
-         return key;
-     }
+    public String getKey() {
+        return key;
+    }
 
-     public String getDesc() {
-         return desc;
-     }
+    public String getDesc() {
+        return desc;
+    }
 }

long-article-recommend-service/src/main/java/com/tzld/longarticle/recommend/server/config/MqConfig.java

@@ -0,0 +1,71 @@
+package com.tzld.longarticle.recommend.server.config;
+
+import com.aliyun.openservices.ons.api.MessageListener;
+import com.aliyun.openservices.ons.api.PropertyKeyConst;
+import com.aliyun.openservices.ons.api.bean.ConsumerBean;
+import com.aliyun.openservices.ons.api.bean.ProducerBean;
+import com.aliyun.openservices.ons.api.bean.Subscription;
+import com.tzld.longarticle.recommend.server.mq.MessageCallbackCustomer;
+import org.springframework.beans.factory.annotation.Autowired;
+import org.springframework.beans.factory.annotation.Value;
+import org.springframework.context.annotation.Bean;
+import org.springframework.context.annotation.Configuration;
+
+import javax.annotation.Resource;
+import java.util.HashMap;
+import java.util.Map;
+import java.util.Properties;
+
+@Configuration
+public class MqConfig {
+
+    @Value("${rocketmq.accessKey}")
+    private String accessKey;
+    @Value("${rocketmq.secretKey}")
+    private String secretKey;
+    @Value("${rocketmq.nameSrvAddr}")
+    private String nameSrvAddr;
+    @Value("${pushMessage.callback.topic}")
+    private String callbackTopic;
+    @Value("${pushMessage.callback.tag}")
+    private String callbackTag;
+
+    @Resource
+    private MessageCallbackCustomer messageCallbackCustomer;
+
+    public Properties getMqProperties() {
+        Properties properties = new Properties();
+        properties.setProperty(PropertyKeyConst.AccessKey, this.accessKey);
+        properties.setProperty(PropertyKeyConst.SecretKey, this.secretKey);
+        properties.setProperty(PropertyKeyConst.NAMESRV_ADDR, this.nameSrvAddr);
+        return properties;
+    }
+
+    @Bean(initMethod = "start", destroyMethod = "shutdown")
+    public ProducerBean buildProducer() {
+        ProducerBean producer = new ProducerBean();
+        producer.setProperties(getMqProperties());
+        return producer;
+    }
+
+
+    @Bean(initMethod = "start", destroyMethod = "shutdown")
+    public ConsumerBean buildConsumer() {
+        ConsumerBean consumerBean = new ConsumerBean();
+        //配置文件
+        Properties properties = getMqProperties();
+        properties.setProperty(PropertyKeyConst.ConsumeThreadNums, "20");
+        consumerBean.setProperties(properties);
+        //订阅关系
+        Map<Subscription, MessageListener> subscriptionTable = new HashMap<>();
+        Subscription subscription = new Subscription();
+        subscription.setTopic(callbackTopic);
+        subscription.setExpression(callbackTag);
+        subscriptionTable.put(subscription, messageCallbackCustomer);
+        //订阅多个topic如上面设置
+        consumerBean.setSubscriptionTable(subscriptionTable);
+        return consumerBean;
+    }
+
+
+}

long-article-recommend-service/src/main/java/com/tzld/longarticle/recommend/server/model/bo/MiniData.java

@@ -0,0 +1,20 @@
+package com.tzld.longarticle.recommend.server.model.bo;
+
+import lombok.Data;
+
+@Data
+public class MiniData {
+
+    private Integer msgType;
+
+    private String title;
+
+    private String coverUrl;
+
+    private String miniAppId;
+
+    private String miniPagePath;
+
+    private Long miniVideoId;
+
+}

long-article-recommend-service/src/main/java/com/tzld/longarticle/recommend/server/model/bo/ReplyInfo.java

@@ -0,0 +1,11 @@
+package com.tzld.longarticle.recommend.server.model.bo;
+
+import lombok.Data;
+
+@Data
+public class ReplyInfo {
+
+    private Integer msgType;
+
+    private Long videoId;
+}

long-article-recommend-service/src/main/java/com/tzld/longarticle/recommend/server/model/param/CallbackParam.java

@@ -0,0 +1,21 @@
+package com.tzld.longarticle.recommend.server.model.param;
+
+import com.tzld.longarticle.recommend.server.model.bo.ReplyInfo;
+import lombok.Data;
+
+import java.util.List;
+
+@Data
+public class CallbackParam {
+
+    private String ghId;
+
+    private String accessToken;
+
+    private String openId;
+
+    private Long timestamp;
+
+    private List<ReplyInfo> replyInfo;
+
+}

long-article-recommend-service/src/main/java/com/tzld/longarticle/recommend/server/model/param/PushMessageParam.java

@@ -0,0 +1,11 @@
+package com.tzld.longarticle.recommend.server.model.param;
+
+import lombok.Data;
+
+@Data
+public class PushMessageParam {
+
+    private String accessToken;
+
+    private String ghId;
+}

long-article-recommend-service/src/main/java/com/tzld/longarticle/recommend/server/model/vo/AccessTokenVo.java

@@ -0,0 +1,11 @@
+package com.tzld.longarticle.recommend.server.model.vo;
+
+import lombok.Data;
+
+@Data
+public class AccessTokenVo {
+
+    private String accessToken;
+
+    private Long expires;
+}

long-article-recommend-service/src/main/java/com/tzld/longarticle/recommend/server/model/vo/PushMessageVo.java

@@ -0,0 +1,14 @@
+package com.tzld.longarticle.recommend.server.model.vo;
+
+import com.tzld.longarticle.recommend.server.model.bo.MiniData;
+import lombok.Data;
+
+import java.util.List;
+
+@Data
+public class PushMessageVo {
+
+    private Integer groupIdx;
+
+    private List<MiniData> components;
+}

long-article-recommend-service/src/main/java/com/tzld/longarticle/recommend/server/mq/MessageCallbackCustomer.java

@@ -0,0 +1,24 @@
+package com.tzld.longarticle.recommend.server.mq;
+
+
+import com.alibaba.fastjson.JSONObject;
+import com.aliyun.openservices.ons.api.Action;
+import com.aliyun.openservices.ons.api.ConsumeContext;
+import com.aliyun.openservices.ons.api.Message;
+import com.aliyun.openservices.ons.api.MessageListener;
+import com.tzld.longarticle.recommend.server.model.param.CallbackParam;
+import lombok.extern.slf4j.Slf4j;
+import org.springframework.stereotype.Component;
+
+@Slf4j
+@Component
+public class MessageCallbackCustomer implements MessageListener {
+
+    @Override
+    public Action consume(Message message, ConsumeContext consumeContext) {
+        System.out.println("Receive: " + message);
+        CallbackParam param = JSONObject.parseObject(new String(message.getBody()), CallbackParam.class);
+        log.info("param = {}", param);
+        return Action.CommitMessage;
+    }
+}

long-article-recommend-service/src/main/java/com/tzld/longarticle/recommend/server/mq/MessageCallbackProducer.java

@@ -0,0 +1,55 @@
+package com.tzld.longarticle.recommend.server.mq;
+
+import com.alibaba.fastjson.JSON;
+import com.aliyun.openservices.ons.api.Message;
+import com.aliyun.openservices.ons.api.SendResult;
+import com.aliyun.openservices.ons.api.bean.ProducerBean;
+import com.tzld.longarticle.recommend.server.model.param.CallbackParam;
+import lombok.extern.slf4j.Slf4j;
+import org.springframework.beans.factory.annotation.Autowired;
+import org.springframework.beans.factory.annotation.Value;
+import org.springframework.stereotype.Component;
+
+import java.nio.charset.StandardCharsets;
+
+@Slf4j
+@Component
+public class MessageCallbackProducer {
+
+    @Value("${pushMessage.callback.topic}")
+    private String TOPIC;
+
+    @Value("${pushMessage.callback.tag}")
+    private String TAG;
+
+    @Autowired
+    private ProducerBean producer;
+
+    public void sendMessage(CallbackParam param) {
+        Message message = new Message();
+        message.setTopic(TOPIC);
+        message.setTag(TAG);
+        message.setBody(JSON.toJSONString(param).getBytes(StandardCharsets.UTF_8));
+        try {
+            SendResult sendResult = producer.send(message);
+            log.info("sendResult = {}", sendResult);
+        } catch (Exception e) {
+            log.error("MessageCallbackProducer send param = {} error", param, e);
+            //重试
+            retry(message);
+        }
+
+    }
+
+    private void retry(Message message) {
+        for (int i = 0; i < 3; i++) {
+            try {
+                SendResult sendResult = producer.send(message);
+                log.info("sendResult = {}", sendResult);
+                return;
+            } catch (Exception e) {
+                log.error("MessageCallbackProducer send error {}", i, e);
+            }
+        }
+    }
+}

long-article-recommend-service/src/main/java/com/tzld/longarticle/recommend/server/service/cgi/CgiReplyService.java

@@ -2,6 +2,7 @@ package com.tzld.longarticle.recommend.server.service.cgi;
 
 import com.tzld.longarticle.recommend.server.model.cgi.BucketDataParam;
 import com.tzld.longarticle.recommend.server.model.cgi.ReplyBucketData;
+import com.tzld.longarticle.recommend.server.model.param.PushMessageParam;
 
 public interface CgiReplyService {
     ReplyBucketData getRgiReplyData(BucketDataParam bucketDataParam);

long-article-recommend-service/src/main/java/com/tzld/longarticle/recommend/server/service/exterior/AccessTokenService.java

@@ -0,0 +1,11 @@
+package com.tzld.longarticle.recommend.server.service.exterior;
+
+
+import com.tzld.longarticle.recommend.server.model.vo.AccessTokenVo;
+
+public interface AccessTokenService {
+
+    AccessTokenVo getAccessToken(String secret);
+
+    boolean validateAccessToken(String accessToken);
+}

long-article-recommend-service/src/main/java/com/tzld/longarticle/recommend/server/service/exterior/ThirdPartyService.java

@@ -0,0 +1,15 @@
+package com.tzld.longarticle.recommend.server.service.exterior;
+
+import com.tzld.longarticle.recommend.server.model.param.CallbackParam;
+import com.tzld.longarticle.recommend.server.model.param.PushMessageParam;
+import com.tzld.longarticle.recommend.server.model.vo.PushMessageVo;
+import org.springframework.stereotype.Service;
+
+import java.util.List;
+
+@Service
+public interface ThirdPartyService {
+    List<PushMessageVo> getPushMessage(PushMessageParam param);
+
+    void PushMessageCallback (CallbackParam param);
+}

long-article-recommend-service/src/main/java/com/tzld/longarticle/recommend/server/service/exterior/impl/AccessTokenServiceImpl.java

@@ -0,0 +1,68 @@
+package com.tzld.longarticle.recommend.server.service.exterior.impl;
+
+import cn.hutool.core.lang.UUID;
+import com.tzld.longarticle.recommend.server.common.enums.SecretEnum;
+import com.tzld.longarticle.recommend.server.model.vo.AccessTokenVo;
+import com.tzld.longarticle.recommend.server.service.exterior.AccessTokenService;
+import org.apache.commons.lang3.StringUtils;
+import org.springframework.beans.factory.annotation.Autowired;
+import org.springframework.data.redis.core.RedisTemplate;
+import org.springframework.stereotype.Service;
+
+import java.util.Map;
+import java.util.concurrent.TimeUnit;
+
+@Service
+public class AccessTokenServiceImpl implements AccessTokenService {
+
+    private static final String ACCESS_TOKEN_LIST = "access_token_list";
+
+    @Autowired
+    private RedisTemplate<String, String> redisTemplate;
+
+    @Override
+    public AccessTokenVo getAccessToken(String secret) {
+        if (!SecretEnum.contains(secret)) {
+            throw new RuntimeException("secret 不存在");
+        }
+        AccessTokenVo accessTokenVo = new AccessTokenVo();
+        String accessToken = redisTemplate.opsForValue().get(secret);
+        if (StringUtils.isNotEmpty(accessToken)) {
+            Long expire = redisTemplate.getExpire(secret, TimeUnit.SECONDS);
+            accessTokenVo.setAccessToken(accessToken);
+            accessTokenVo.setExpires(expire);
+            if (redisTemplate.opsForHash().size(ACCESS_TOKEN_LIST) > 20) {
+                Map<Object, Object> entries = redisTemplate.opsForHash().entries(ACCESS_TOKEN_LIST);
+                entries.forEach((k, v) -> {
+                    Long timestamp = (Long) v;
+                    if (System.currentTimeMillis() / 1000 > timestamp) {
+                        redisTemplate.opsForHash().delete(ACCESS_TOKEN_LIST, k);
+                    }
+                });
+            }
+        } else {
+            String newAccessToken = UUID.randomUUID().toString().replace("-", "");
+            redisTemplate.opsForValue().set(secret, newAccessToken, 2L, TimeUnit.HOURS);
+            long expire = 7200L;
+            long timestamp = System.currentTimeMillis() / 1000 + expire + 600;
+            redisTemplate.opsForHash().put(ACCESS_TOKEN_LIST, newAccessToken, timestamp);
+            accessTokenVo.setAccessToken(newAccessToken);
+            accessTokenVo.setExpires(expire);
+        }
+        return accessTokenVo;
+    }
+
+    @Override
+    public boolean validateAccessToken(String accessToken) {
+        Object o = redisTemplate.opsForHash().get(ACCESS_TOKEN_LIST, accessToken);
+        if (o == null) {
+            return false;
+        }
+        Long timestamp = (Long) o;
+        if (System.currentTimeMillis() / 1000 > timestamp) {
+            redisTemplate.opsForHash().delete(ACCESS_TOKEN_LIST, accessToken);
+            return false;
+        }
+        return true;
+    }
+}

long-article-recommend-service/src/main/java/com/tzld/longarticle/recommend/server/service/exterior/impl/ThirdPartyServiceImpl.java

@@ -0,0 +1,97 @@
+package com.tzld.longarticle.recommend.server.service.exterior.impl;
+
+import com.tzld.longarticle.recommend.server.common.enums.cgi.ReplyStrategyServiceEnum;
+import com.tzld.longarticle.recommend.server.model.bo.MiniData;
+import com.tzld.longarticle.recommend.server.model.cgi.BucketDataParam;
+import com.tzld.longarticle.recommend.server.model.cgi.GroupData;
+import com.tzld.longarticle.recommend.server.model.cgi.MsgData;
+import com.tzld.longarticle.recommend.server.model.cgi.ReplyBucketData;
+import com.tzld.longarticle.recommend.server.model.param.CallbackParam;
+import com.tzld.longarticle.recommend.server.model.param.PushMessageParam;
+import com.tzld.longarticle.recommend.server.model.vo.PushMessageVo;
+import com.tzld.longarticle.recommend.server.service.exterior.AccessTokenService;
+import com.tzld.longarticle.recommend.server.service.exterior.ThirdPartyService;
+import com.tzld.longarticle.recommend.server.service.strategy.reply.ReplyStrategyService;
+import lombok.extern.slf4j.Slf4j;
+import org.apache.commons.lang3.StringUtils;
+import org.springframework.beans.BeanUtils;
+import org.springframework.beans.factory.annotation.Autowired;
+import org.springframework.context.ApplicationContext;
+import org.springframework.stereotype.Service;
+import org.springframework.util.CollectionUtils;
+
+import javax.annotation.PostConstruct;
+import java.util.ArrayList;
+import java.util.List;
+import java.util.Map;
+
+@Slf4j
+@Service
+public class ThirdPartyServiceImpl implements ThirdPartyService {
+
+    @Autowired
+    private ApplicationContext applicationContext;
+
+    @Autowired
+    private AccessTokenService accessTokenService;
+
+    private Map<String, ReplyStrategyService> strategyServiceMap;
+
+    @PostConstruct
+    public void init() {
+        strategyServiceMap = applicationContext.getBeansOfType(ReplyStrategyService.class);
+    }
+
+
+    @Override
+    public List<PushMessageVo> getPushMessage(PushMessageParam param) {
+        if (param == null || StringUtils.isEmpty(param.getGhId()) || StringUtils.isEmpty(param.getAccessToken())) {
+            throw new RuntimeException("参数错误");
+        }
+        if (!accessTokenService.validateAccessToken(param.getAccessToken())) {
+            throw new RuntimeException("accessToken错误");
+        }
+        List<PushMessageVo> pushMessageVoList = new ArrayList<>();
+        ReplyBucketData replyBucketData = getPushMessageData(param);
+        if (replyBucketData == null) {
+            log.error("获取推送策略数据失败");
+            return pushMessageVoList;
+        }
+        List<GroupData> groupList = replyBucketData.getGroupList();
+        for (GroupData groupData : groupList) {
+            if(CollectionUtils.isEmpty(groupData.getMsgDataList())){
+                continue;
+            }
+            PushMessageVo pushMessageVo = new PushMessageVo();
+            List<MiniData> components = new ArrayList<>();
+            for (MsgData msgData : groupData.getMsgDataList()){
+                MiniData miniData = new MiniData();
+                BeanUtils.copyProperties(miniData, msgData);
+                components.add(miniData);
+            }
+            pushMessageVo.setGroupIdx(groupData.getGroupIndex());
+            pushMessageVo.setComponents(components);
+            pushMessageVoList.add(pushMessageVo);
+        }
+        return pushMessageVoList;
+    }
+
+    @Override
+    public void PushMessageCallback(CallbackParam param) {
+
+    }
+
+    private ReplyBucketData getPushMessageData(PushMessageParam param) {
+        for (Map.Entry<String, ReplyStrategyService> stringReplyStrategyServiceEntry : strategyServiceMap.entrySet()) {
+            ReplyStrategyService replyStrategyService = stringReplyStrategyServiceEntry.getValue();
+            // 使用策略层
+            if (replyStrategyService.support(ReplyStrategyServiceEnum.PUSH_MESSAGE_STRATEGY_V1)) {
+                BucketDataParam bucketDataParam = new BucketDataParam();
+                bucketDataParam.setGhId(param.getGhId());
+                return replyStrategyService.getResult(bucketDataParam);
+            }
+        }
+        // 无执行策略 不会走到这里
+        return null;
+    }
+}

long-article-recommend-service/src/main/java/com/tzld/longarticle/recommend/server/service/strategy/reply/ReplyStrategyService.java

@@ -4,7 +4,7 @@ import com.tzld.longarticle.recommend.server.common.enums.cgi.ReplyStrategyServi
 import com.tzld.longarticle.recommend.server.model.cgi.BucketDataParam;
 import com.tzld.longarticle.recommend.server.model.cgi.ReplyBucketData;
 
-public interface ReplyStrategyService {
+public interface ReplyStrategyService<T, P> {
 
     ReplyBucketData getResult(BucketDataParam bucketDataParam);
 

long-article-recommend-service/src/main/java/com/tzld/longarticle/recommend/server/service/strategy/reply/impl/PushMessageStrategyV1.java

@@ -0,0 +1,241 @@
+package com.tzld.longarticle.recommend.server.service.strategy.reply.impl;
+
+import com.alibaba.fastjson.JSON;
+import com.alibaba.fastjson.JSONObject;
+import com.tzld.longarticle.recommend.server.common.enums.cgi.ReplyStrategyServiceEnum;
+import com.tzld.longarticle.recommend.server.mapper.crawler.AlgGhAutoreplyVideoRankDataMapper;
+import com.tzld.longarticle.recommend.server.mapper.crawler.CgiReplyBucketDataMapper;
+import com.tzld.longarticle.recommend.server.model.cgi.*;
+import com.tzld.longarticle.recommend.server.repository.model.AlgGhAutoreplyVideoRankData;
+import com.tzld.longarticle.recommend.server.repository.model.AlgGhAutoreplyVideoRankDataExample;
+import com.tzld.longarticle.recommend.server.repository.model.CgiReplyBucketData;
+import com.tzld.longarticle.recommend.server.repository.model.CgiReplyBucketDataExample;
+import com.tzld.longarticle.recommend.server.service.cgi.TouLiuHttpClientService;
+import com.tzld.longarticle.recommend.server.service.strategy.reply.ReplyStrategyService;
+import lombok.extern.slf4j.Slf4j;
+import org.springframework.beans.BeanUtils;
+import org.springframework.beans.factory.annotation.Autowired;
+import org.springframework.stereotype.Service;
+import org.springframework.util.CollectionUtils;
+
+import java.util.*;
+import java.util.stream.Collectors;
+
+@Slf4j
+@Service
+public class PushMessageStrategyV1 implements ReplyStrategyService {
+
+    /**
+     * 实验分桶数量
+     */
+//    private static final Integer bucketNum = 10;
+
+    /**
+     * 分桶实验策略，key为策略，arr为对应桶
+     * {"base":[0,1,2,3],"stg0909-base":[4,5],"stg0909-explore1":[6,7,8],"stg0909-explore2":[9]}
+     * {"stg0909-base":[5,6],"stg0909-explore1":[7],"stg0909-explore2":[8,9]}
+     */
+    private static final String bucketStrategyConfig = "{\"stg0909-base\":[5,6],\"stg0909-explore1\":[7],\"stg0909-explore2\":[8,9]}";
+
+    /**
+     * 自动回复使用小程序Id
+     */
+    private static final String SMALL_APP_Id = "wxbdd2a2e93d9a6e25";
+
+    private static final String CDN_URL = "https://rescdn.piaoquantv.com/";
+
+    private static final String GET_SMALL_PAGE_URL = "https://api.piaoquantv.com";
+
+    @Autowired
+    private AlgGhAutoreplyVideoRankDataMapper algGhAutoreplyVideoRankDataMapper;
+    @Autowired
+    private CgiReplyBucketDataMapper cgiReplyBucketDataMapper;
+    @Autowired
+    private TouLiuHttpClientService httpClientService;
+
+    @Override
+    public ReplyBucketData getResult(BucketDataParam bucketDataParam) {
+        // 0 获取策略key
+        JSONObject bucketStrategyConfigJsonObject = JSON.parseObject(bucketStrategyConfig);
+        Set<String> keyedSet = bucketStrategyConfigJsonObject.keySet();
+        // 1 处理文章--算法引擎--排序文章数据
+//        getWenzhangData();
+        // 2 处理小程序--读取离线数据表--获取策略排序小程序数据
+        List<CgiReplyBucketData> smallDataCgiReplyList = readStrategyOrderSmallData(keyedSet);
+        // 2.1 获取小程序落地页地址 http调用
+        smallDataCgiReplyList = setSmallPageUrl(smallDataCgiReplyList);
+        log.info(JSON.toJSONString(smallDataCgiReplyList));
+        // 3 入库读表
+        insertSmallData(smallDataCgiReplyList, keyedSet);
+        // 4 组装分桶数据
+        return getReplyBucketData(bucketStrategyConfigJsonObject, keyedSet, bucketDataParam.getGhId());
+    }
+
+    private ReplyBucketData getReplyBucketData(JSONObject bucketStrategyConfigJsonObject, Set<String> keyedSet, String ghId) {
+        // 策略小程序数据
+        ReplyBucketData replyBucketData = new ReplyBucketData();
+        List<GroupData> groupDataList = new ArrayList<>();
+        for (String key : keyedSet) {
+            CgiReplyBucketDataExample cgiReplyBucketDataExample = new CgiReplyBucketDataExample();
+            cgiReplyBucketDataExample.createCriteria().andIsDeleteEqualTo(0).andStrategyEqualTo(key).andGhIdEqualTo(ghId);
+            cgiReplyBucketDataExample.setOrderByClause("sort");
+            List<CgiReplyBucketData> cgiReplyBucketData = cgiReplyBucketDataMapper.selectByExample(cgiReplyBucketDataExample);
+            if (CollectionUtils.isEmpty(cgiReplyBucketData)) {
+                CgiReplyBucketDataExample cgiReplyBucketDataExampleNull = new CgiReplyBucketDataExample();
+                cgiReplyBucketDataExampleNull.createCriteria().andIsDeleteEqualTo(0).andStrategyEqualTo(key).andGhIdEqualTo("default");
+                cgiReplyBucketDataExampleNull.setOrderByClause("sort");
+                cgiReplyBucketData = cgiReplyBucketDataMapper.selectByExample(cgiReplyBucketDataExampleNull);
+            }
+            if (CollectionUtils.isEmpty(cgiReplyBucketData)) {
+                log.error("getReplyBucketData get data is null,key:" + key);
+                return null;
+            }
+
+            List<Integer> groupList = bucketStrategyConfigJsonObject.getJSONArray(key).toJavaList(Integer.class);
+            for (Integer group : groupList) {
+                GroupData groupData = new GroupData();
+                groupData.setGroupIndex(group);
+                List<MsgData> msgDataList = new ArrayList<>();
+                for (CgiReplyBucketData cgiReplyBucketDatum : cgiReplyBucketData) {
+                    MsgData msgData = new MsgData();
+                    BeanUtils.copyProperties(cgiReplyBucketDatum, msgData);
+                    if (cgiReplyBucketDatum.getMsgType().equals(1)) {
+                        msgData.setMiniAppId(SMALL_APP_Id);
+                    }
+                    msgDataList.add(msgData);
+                }
+                groupData.setMsgDataList(msgDataList);
+                groupDataList.add(groupData);
+            }
+        }
+        // groupDataList排序
+        replyBucketData.setGroupList(groupDataList.stream().sorted(Comparator.comparingInt(GroupData::getGroupIndex)).collect(Collectors.toList()));
+        return replyBucketData;
+    }
+
+
+    private void insertSmallData(List<CgiReplyBucketData> smallDataCgiReplyList, Set<String> keyedSet) {
+        if (CollectionUtils.isEmpty(smallDataCgiReplyList)) {
+            return;
+        }
+        for (String key : keyedSet) {
+            if ("base".equals(key)) {
+                continue;
+            }
+            List<CgiReplyBucketData> collect = smallDataCgiReplyList.stream().filter(x -> x.getStrategy().equals(key)).collect(Collectors.toList());
+            if (CollectionUtils.isEmpty(collect)) {
+                log.error("insertSmallData 算法排序数据异常，data：" + JSON.toJSONString(smallDataCgiReplyList));
+                continue;
+            }
+            // 清上个版本的策略数据
+            CgiReplyBucketDataExample cgiReplyBucketDataExample = new CgiReplyBucketDataExample();
+            cgiReplyBucketDataExample.createCriteria().andIsDeleteEqualTo(0).andMsgTypeEqualTo(1).andStrategyEqualTo(key);
+            List<CgiReplyBucketData> cgiReplyBucketData1 = cgiReplyBucketDataMapper.selectByExample(cgiReplyBucketDataExample);
+            for (CgiReplyBucketData cgiReplyBucketData : cgiReplyBucketData1) {
+                cgiReplyBucketData.setIsDelete(1);
+                cgiReplyBucketDataMapper.updateByPrimaryKeySelective(cgiReplyBucketData);
+            }
+            // 入库
+            for (CgiReplyBucketData cgiReplyBucketData : collect) {
+                cgiReplyBucketDataMapper.insertSelective(cgiReplyBucketData);
+            }
+        }
+    }
+
+    private List<CgiReplyBucketData> setSmallPageUrl(List<CgiReplyBucketData> smallDataCgiReplyList) {
+        if (CollectionUtils.isEmpty(smallDataCgiReplyList)) {
+            return smallDataCgiReplyList;
+        }
+        Set<String> keys = smallDataCgiReplyList.stream().map(x -> x.getGhId() + "&" + x.getMiniVideoId()).collect(Collectors.toSet());
+        Map<String, SmallPageUrlDetail> keyPageUrl = new HashMap<>();
+        // gh-id + videoId 复用同一page_url及落地页id
+        for (String key : keys) {
+            String[] keyArr = key.split("&");
+            String ghId = keyArr[0];
+            String videoId = keyArr[1];
+            // 查询库里是否存在，如果存在即复用
+            CgiReplyBucketDataExample cgiReplyBucketDataExample = new CgiReplyBucketDataExample();
+            cgiReplyBucketDataExample.createCriteria().andIsDeleteEqualTo(0).andMiniVideoIdEqualTo(Long.valueOf(videoId)).andGhIdEqualTo(ghId);
+            List<CgiReplyBucketData> cgiReplyBucketData = cgiReplyBucketDataMapper.selectByExample(cgiReplyBucketDataExample);
+            SmallPageUrlDetail smallPageUrlDetail = new SmallPageUrlDetail();
+            if (CollectionUtils.isEmpty(cgiReplyBucketData)) {
+                // 库里不存在，调用新生成
+                String response = httpClientService.sendAdFlowAddRequest(GET_SMALL_PAGE_URL, videoId, "公众号", "自动回复小程序", "位置X", ghId);
+                JSONObject jsonObject = JSON.parseObject(response);
+                if (jsonObject.getInteger("code").equals(0)) {
+                    smallPageUrlDetail = jsonObject.getObject("data", SmallPageUrlDetail.class);
+                    keyPageUrl.put(key, smallPageUrlDetail);
+                } else {
+                    log.error("httpClientService get page url error,response:" + response);
+                    throw new RuntimeException("httpClientService get page url error");
+                }
+            } else {
+                // 复用
+                CgiReplyBucketData cgiReplyBucketData1 = cgiReplyBucketData.get(0);
+                smallPageUrlDetail.setId(cgiReplyBucketData1.getPagePathUrlId());
+                smallPageUrlDetail.setUrl(cgiReplyBucketData1.getMiniPagePath());
+            }
+            keyPageUrl.put(key, smallPageUrlDetail);
+        }
+        // 处理数据
+        for (CgiReplyBucketData cgiReplyBucketData : smallDataCgiReplyList) {
+            String key = cgiReplyBucketData.getGhId() + "&" + cgiReplyBucketData.getMiniVideoId();
+            SmallPageUrlDetail smallPageUrlDetail = keyPageUrl.get(key);
+            if (Objects.isNull(smallPageUrlDetail)) {
+                log.error("setSmallPageUrl get map url is null" + JSON.toJSONString(keyPageUrl));
+                throw new RuntimeException("setSmallPageUrl get map url is null");
+            }
+            cgiReplyBucketData.setPagePathUrlId(smallPageUrlDetail.getId());
+            cgiReplyBucketData.setMiniPagePath(smallPageUrlDetail.getUrl());
+        }
+        return smallDataCgiReplyList;
+    }
+
+    private List<CgiReplyBucketData> readStrategyOrderSmallData(Set<String> keyedSet) {
+        List<CgiReplyBucketData> result = new ArrayList<>();
+        for (String key : keyedSet) {
+            if ("base".equals(key)) {
+                // base作为人工控制
+                continue;
+            }
+            // 获取最新dt的策略
+            String dtVersion = algGhAutoreplyVideoRankDataMapper.selectLatestDtVersionByStrategyKey(key);
+            // 判断当前的dtVersion是否已经处理过了
+            CgiReplyBucketDataExample cgiReplyBucketDataExample = new CgiReplyBucketDataExample();
+            cgiReplyBucketDataExample.createCriteria().andIsDeleteEqualTo(0).andStrategyDtEqualTo(dtVersion).andStrategyEqualTo(key);
+            long count = cgiReplyBucketDataMapper.countByExample(cgiReplyBucketDataExample);
+            if (count != 0) {
+                // 说明已处理过该dtVersion数据
+                continue;
+            }
+            // 获取最新dt数据
+            List<AlgGhAutoreplyVideoRankData> dtVserSionStrategyData = getDtVersionStrategyData(key, dtVersion);
+            result.addAll(dtVserSionStrategyData.stream().map(x -> {
+                CgiReplyBucketData cgiReplyBucketData = new CgiReplyBucketData();
+                cgiReplyBucketData.setStrategy(key);
+                cgiReplyBucketData.setSort(x.getSort());
+                cgiReplyBucketData.setStrategyDt(x.getDtVersion());
+                cgiReplyBucketData.setGhId(x.getGhId());
+                cgiReplyBucketData.setMsgType(1);
+                cgiReplyBucketData.setTitle(x.getTitle());
+                cgiReplyBucketData.setCoverUrl(CDN_URL + x.getCoverUrl());
+                cgiReplyBucketData.setMiniAppId(SMALL_APP_Id);
+                cgiReplyBucketData.setMiniVideoId(x.getVideoId());
+                return cgiReplyBucketData;
+            }).collect(Collectors.toList()));
+        }
+        // 获取最新数据版本
+        return CollectionUtils.isEmpty(result) ? null : result;
+    }
+
+    private List<AlgGhAutoreplyVideoRankData> getDtVersionStrategyData(String key, String dtVersion) {
+        AlgGhAutoreplyVideoRankDataExample algGhAutoreplyVideoRankDataExample = new AlgGhAutoreplyVideoRankDataExample();
+        algGhAutoreplyVideoRankDataExample.createCriteria().andIsDeleteEqualTo(0).andDtVersionEqualTo(dtVersion).andStrategyKeyEqualTo(key);
+        return algGhAutoreplyVideoRankDataMapper.selectByExample(algGhAutoreplyVideoRankDataExample);
+    }
+
+    @Override
+    public Boolean support(ReplyStrategyServiceEnum key) {
+        return ReplyStrategyServiceEnum.PUSH_MESSAGE_STRATEGY_V1.equals(key);
+    }
+}

long-article-recommend-service/src/main/java/com/tzld/longarticle/recommend/server/web/ThirdPartyController.java

@@ -0,0 +1,48 @@
+package com.tzld.longarticle.recommend.server.web;
+
+import com.tzld.longarticle.recommend.server.common.response.CommonResponse;
+import com.tzld.longarticle.recommend.server.model.param.CallbackParam;
+import com.tzld.longarticle.recommend.server.model.vo.AccessTokenVo;
+import com.tzld.longarticle.recommend.server.model.param.PushMessageParam;
+import com.tzld.longarticle.recommend.server.model.vo.PushMessageVo;
+import com.tzld.longarticle.recommend.server.mq.MessageCallbackProducer;
+import com.tzld.longarticle.recommend.server.service.exterior.AccessTokenService;
+import com.tzld.longarticle.recommend.server.service.exterior.ThirdPartyService;
+import org.springframework.beans.factory.annotation.Autowired;
+import org.springframework.web.bind.annotation.PostMapping;
+import org.springframework.web.bind.annotation.RequestBody;
+import org.springframework.web.bind.annotation.RequestMapping;
+import org.springframework.web.bind.annotation.RestController;
+
+import javax.annotation.Resource;
+import java.util.List;
+
+@RestController
+@RequestMapping("/3rdParty")
+public class ThirdPartyController {
+
+    @Autowired
+    private AccessTokenService accessTokenService;
+
+    @Autowired
+    private ThirdPartyService thirdPartyService;
+
+    @Resource
+    private MessageCallbackProducer messageCallbackProducer;
+
+    @PostMapping("/accessToken/get")
+    public CommonResponse<AccessTokenVo> getAccessToken(@RequestBody String secret) {
+        return CommonResponse.success(accessTokenService.getAccessToken(secret));
+    }
+
+    @PostMapping("/pushMessage/get")
+    public CommonResponse<List<PushMessageVo>> getPushMessage(@RequestBody PushMessageParam param) {
+        return CommonResponse.success(thirdPartyService.getPushMessage(param));
+    }
+
+    @PostMapping("/pushMessage/callback")
+    public CommonResponse<Void> pushMessageCallback(@RequestBody CallbackParam param) {
+        messageCallbackProducer.sendMessage(param);
+        return CommonResponse.success();
+    }
+}

long-article-recommend-service/src/main/resources/application-dev.yml

@@ -111,4 +111,9 @@ aliyun:
 
 logging:
   file:
-    path: ./${spring.application.name}/logs/
+    path: ./${spring.application.name}/logs/
+
+pushMessage:
+  callback:
+    topic: topic_3rd_party_push_message_callback_dev
+    tag: mini

long-article-recommend-service/src/main/resources/application.yml

@@ -55,3 +55,8 @@ apollo:
 
 mybatis:
   mapper-locations: classpath:/mapper/*.xml
+
+rocketmq:
+  accessKey: LTAI4G7puhXtLyHzHQpD6H7A
+  secretKey: nEbq3xWNQd1qLpdy2u71qFweHkZjSG
+  nameSrvAddr: http://MQ_INST_1894469520484605_BXhXuzkZ.mq-internet-access.mq-internet.aliyuncs.com:80