来客吧交易数据接收同步

kxs-stat/kxs-stat-api/src/main/java/com/kxs/stat/api/amqp/rabbit/RabbitLkbTradeQueueMQ.java

@@ -37,7 +37,7 @@ public class RabbitLkbTradeQueueMQ extends AbstractMQ {
     /**
      * 死队列名称
      */
-    public static final String DEAD_QUEUE_NAME = "DEAD_QUEUE_LKB_TRADE_DIVISION";
+    public static final String DEAD_QUEUE_NAME = null;
 
     /**
      * 内置msg 消息体定义
@@ -53,55 +53,55 @@ public class RabbitLkbTradeQueueMQ extends AbstractMQ {
         /**
          * 订单状态（0待支付，1已支付）
          */
-        private String  Status;
+        private String  status;
         /**
          * 订单创建时间
          */
-        private String CreateDate;
+        private String createDate;
         /**
          * 支付时间
          */
-        private String PayDate;
+        private String payDate;
         /**
          * 好哒订单号
          */
-        private String HdOrderNo;
+        private String hdOrderNo;
         /**
          * 支付宝微信平台交易号
          */
-        private String TradeNo;
+        private String tradeNo;
         /**
          * 码牌SN
          */
-        private String SnNo;
+        private String snNo;
         /**
          * 支付金额
          */
-        private BigDecimal PayMoney;
+        private BigDecimal payMoney;
         /**
          * 支付方式（1支付宝，2微信）
          */
-        private String PayMode;
+        private String payMode;
         /**
          * 订单号
          */
-        private String OrderNo;
+        private String orderNo;
         /**
          * 商户Id
          */
-        private String MerchantId;
+        private String merchantId;
         /**
          * 活动标识（1活动，0非活动）
          */
-        private String IsAct;
+        private String isAct;
         /**
          * 用户ID
          */
-        private Integer UserId;
+        private Integer userId;
         /**
          * 激活时间
          */
-        private Integer  BrandId;
+        private Integer  brandId;
 
 
     }

kxs-stat/kxs-stat-api/src/main/java/com/kxs/stat/api/model/KxsLkbTrade.java

@@ -108,5 +108,12 @@ public class KxsLkbTrade extends Model<KxsLkbTrade> implements Serializable {
     @Schema(description = "通道品牌ID（0直联，1银联，2好哒）")
     private Integer brandId;
 
+
+    /**
+     * 统计类型 0个人 1团队
+     */
+    @Schema(description = "统计类型 0个人 1团队")
+    private Integer totalType;
+
 }
 

kxs-stat/kxs-stat-biz/src/main/java/com/kxs/stat/biz/mq/RabbitLkbQueueListener.java

@@ -2,11 +2,17 @@ package com.kxs.stat.biz.mq;
 
 import com.alibaba.fastjson.JSON;
 import com.baomidou.mybatisplus.core.toolkit.Wrappers;
-import com.kxs.product.api.amqp.rabbit.RabbitShopTimeoutQueueMQ;
-import com.kxs.product.api.model.KxsShopOrder;
+import com.kxs.common.core.constant.SecurityConstants;
+import com.kxs.common.core.constant.enums.ErrorTypeEnum;
+import com.kxs.common.core.exception.GlobalCustomerException;
+import com.kxs.common.core.util.R;
+import com.kxs.common.core.util.RetOps;
 import com.kxs.stat.api.amqp.rabbit.RabbitLkbTradeQueueMQ;
 import com.kxs.stat.api.model.KxsLkbTrade;
+import com.kxs.stat.biz.constant.enums.TotalTypeEnum;
 import com.kxs.stat.biz.service.KxsLkbTradeService;
+import com.kxs.user.api.feign.RemoteKxsUserService;
+import com.kxs.user.api.model.KxsUser;
 import com.rabbitmq.client.Channel;
 import lombok.RequiredArgsConstructor;
 import lombok.extern.slf4j.Slf4j;
@@ -17,6 +23,8 @@ import org.springframework.beans.BeanUtils;
 import org.springframework.stereotype.Component;
 
 import java.io.IOException;
+import java.util.ArrayList;
+import java.util.List;
 
 /**
  * rabbit 队列侦听器 商品下单过期队列
@@ -31,6 +39,8 @@ public class RabbitLkbQueueListener {
 
     private final KxsLkbTradeService kxsLkbTradeService;
 
+    private final RemoteKxsUserService remoteKxsUserService;
+
     /**
      * 监听 商品下单过期 队列的处理器
      *
@@ -42,9 +52,22 @@ public class RabbitLkbQueueListener {
         try {
             RabbitLkbTradeQueueMQ.MsgEntity parse = RabbitLkbTradeQueueMQ.parse(msg);
             log.info("来客吧交易消费端Payload: " + parse);
+
             KxsLkbTrade kxsLkbTrade = new KxsLkbTrade();
-			BeanUtils.copyProperties(parse, kxsLkbTrade);
-			KxsLkbTrade lkbTrade = kxsLkbTradeService.getOne(Wrappers.<KxsLkbTrade>lambdaQuery()
+            kxsLkbTrade.setUserId(parse.getUserId());
+            kxsLkbTrade.setBrandId(parse.getBrandId());
+
+            //活动标识（1活动，0非活动）
+            if("1".equals(parse.getIsAct())){
+                kxsLkbTrade.setActTradeAmt(parse.getPayMoney());
+            }else{
+                kxsLkbTrade.setTradeAmt(parse.getPayMoney());
+            }
+            kxsLkbTrade.setTotalType(TotalTypeEnum.SELF.getType());
+
+
+            //个人业绩
+            KxsLkbTrade lkbTrade = kxsLkbTradeService.getOne(Wrappers.<KxsLkbTrade>lambdaQuery()
 					.eq(KxsLkbTrade::getUserId, kxsLkbTrade.getUserId())
 					.eq(KxsLkbTrade::getBrandId, kxsLkbTrade.getBrandId())
 					.eq(KxsLkbTrade::getTradeMonth, kxsLkbTrade.getTradeMonth())
@@ -56,33 +79,54 @@ public class RabbitLkbQueueListener {
 				lkbTrade.setTradeAmt(lkbTrade.getTradeAmt().add(kxsLkbTrade.getTradeAmt()));
 				kxsLkbTradeService.updateById(lkbTrade);
 			}
+            //团队业绩
+            R<KxsUser> kxsUserR = remoteKxsUserService.loadUserById(parse.getUserId(), SecurityConstants.FROM_IN);
+            KxsUser user = RetOps.of(kxsUserR)
+                    .getData()
+                    .orElseThrow(() -> new GlobalCustomerException(ErrorTypeEnum.USER_NOT_FOUND.getDescription()));
+
+            String pidPath = user.getPidPath();
+            String[] split = pidPath.split(",");
+            List<KxsLkbTrade> batch = new ArrayList<>();
+            for (String userId : split) {
+                int pid = Integer.parseInt(userId);
+                if(pid > 0){
+                    KxsLkbTrade kxsLkbTradeBatch = new KxsLkbTrade();
+                    BeanUtils.copyProperties(kxsLkbTrade, kxsLkbTradeBatch);
+                    kxsLkbTradeBatch.setUserId(pid);
+                    kxsLkbTradeBatch.setTotalType(TotalTypeEnum.TEAM.getType());
+                    batch.add(kxsLkbTradeBatch);
+                }
+            }
+            kxsLkbTradeService.saveBatch(batch);
+            log.info("来客吧交易数据消费成功:{}", JSON.toJSONString(parse));
             channel.basicAck(message.getMessageProperties().getDeliveryTag(), false);
         } catch (Exception e) {
-            log.error(e.getMessage(), e);
+            log.error("来客吧交易数据消费失败:{}", JSON.toJSONString(msg), e);
             try {
-                channel.basicReject(message.getMessageProperties().getDeliveryTag(), false);
+                channel.basicAck(message.getMessageProperties().getDeliveryTag(), false);
             } catch (IOException ex) {
-                log.error("来客吧进入死信队列失败" + ex.getMessage(), ex);
+                log.error("来客吧交易数据消费失败" + ex.getMessage(), ex);
             }
         }
     }
 
-    /**
-     * 监听死信 队列的处理器
-     *
-     * @param message 消息
-     */
-    @RabbitListener(queues = RabbitLkbTradeQueueMQ.DEAD_QUEUE_NAME)
-    @RabbitHandler
-    public void onDeadMessage(String msg, Message message, Channel channel) {
-        log.info("来客吧死信队列Payload: " + JSON.toJSONString(msg));
-        try {
-            RabbitShopTimeoutQueueMQ.MsgEntity parse = RabbitShopTimeoutQueueMQ.parse(msg);
-        } catch (Exception e) {
-            log.error(e.getMessage(), e);
-
-        }
-    }
+//    /**
+//     * 监听死信 队列的处理器
+//     *
+//     * @param message 消息
+//     */
+//    @RabbitListener(queues = RabbitLkbTradeQueueMQ.DEAD_QUEUE_NAME)
+//    @RabbitHandler
+//    public void onDeadMessage(String msg, Message message, Channel channel) {
+//        log.info("来客吧死信队列Payload: " + JSON.toJSONString(msg));
+//        try {
+//            RabbitShopTimeoutQueueMQ.MsgEntity parse = RabbitShopTimeoutQueueMQ.parse(msg);
+//        } catch (Exception e) {
+//            log.error(e.getMessage(), e);
+//
+//        }
+//    }
 
 
 }

kxs-stat/kxs-stat-biz/src/main/resources/mapper/KxsLkbTradeMapper.xml

@@ -14,6 +14,7 @@
         <result column="act_trade_amt" property="actTradeAmt"/>
         <result column="user_id" property="userId"/>
         <result column="brand_id" property="brandId"/>
+        <result column="total_type" property="totalType"/>
 
     </resultMap>
     <select id="getUserTradeMonthList" resultType="com.kxs.stat.api.model.KxsLkbTrade" fetchSize="1000">