# PriceMessagePostProcessorImpl.java

    package cloud.wconnect.rateshop.service.rateshop.impl;

    import cloud.wconnect.commonlib.component.BeanMapper;
    import cloud.wconnect.commonlib.cons.BaseConstant;
    import cloud.wconnect.commonlib.util.DateTimeUtils;
    import cloud.wconnect.rateshop.entity.rateshop.Setting;
    import cloud.wconnect.rateshop.pojo.dto.rateshop.PriceCacheDTO;
    import cloud.wconnect.rateshop.pojo.dto.rateshop.PriceCacheKeyDTO;
    import cloud.wconnect.rateshop.pojo.dto.rateshop.consumer.hotelprice.PlatformDTO;
    import cloud.wconnect.rateshop.pojo.dto.rateshop.consumer.hotelprice.PriceDTO;
    import cloud.wconnect.rateshop.pojo.dto.rateshop.consumer.hotelprice.PriceMessage;
    import cloud.wconnect.rateshop.pojo.dto.rateshop.consumer.hotelprice.RoomDTO;
    import cloud.wconnect.rateshop.pojo.dto.rateshop.sender.MsgMinPrice;
    import cloud.wconnect.rateshop.service.rateshop.IAlertMessageService;
    import cloud.wconnect.rateshop.service.rateshop.IIwcSqsService;
    import cloud.wconnect.rateshop.service.rateshop.IPriceMessagePostProcessor;
    import com.baomidou.dynamic.datasource.annotation.DS;
    import com.google.common.collect.Lists;
    import org.apache.commons.collections.CollectionUtils;
    import org.springframework.beans.factory.annotation.Autowired;
    import org.springframework.data.redis.core.RedisCallback;
    import org.springframework.data.redis.core.RedisTemplate;
    import org.springframework.stereotype.Service;

    import java.math.BigDecimal;
    import java.time.LocalDate;
    import java.time.temporal.ChronoUnit;
    import java.util.*;
    import java.util.stream.Collectors;

    /**
     * 房价消息后置处理器
     *
     * @author lichen
     * @date 2025/4/4
     */
    @Service
    @DS("rate_shop")
    public class PriceMessagePostProcessorImpl implements IPriceMessagePostProcessor {

        @Autowired
        private IAlertMessageService alertMessageService;
        @Autowired
        private RedisTemplate<String, String> redisTemplate;
        @Autowired
        private IIwcSqsService iwcSqsService;
        @Autowired
        private BeanMapper beanMapper;

        /**
         * 房价消息的后置处理
         *
         * @param priceMessage 房价消息
         * @return void
         * @author lichen
         * @date 2025/4/4
         */
        @Override
        public void process(PriceMessage priceMessage) {
            //过滤出未来32天以内的数据
            List<PriceDTO> priceDTOList = priceMessage.getData().stream().filter(priceDto -> {
                Long checkInDate = priceDto.getCheckInDate();

                //把checkInDate转化为localDate
                LocalDate checkInDateLocalDate = DateTimeUtils.strToLocalDate(checkInDate.toString(), BaseConstant.DATE_FORMAT_YYYYMMDD);
                //获取当前时间
                LocalDate nowLocalDate = LocalDate.now();
                // 比较两个日期之间的天数差
                long diffDays = ChronoUnit.DAYS.between(checkInDateLocalDate, nowLocalDate);
                //判断checkInDate是否在未来32天之内
                return Math.abs(diffDays) <= 32;
            }).filter(item -> CollectionUtils.isNotEmpty(item.getPlatform())).toList();

            if (CollectionUtils.isEmpty(priceDTOList)) return;

            //获取消息中的最低价格
            Map<Integer, List<MsgMinPrice>> containTaxFlag2LowestPrice = getMessageLowestPrice(priceMessage, priceDTOList);

            //根据含税标志生成警报消息
            for (Integer containTaxFlag : Lists.newArrayList(Setting.BEFORE_TAX_RATE, Setting.AFTER_TAX_RATE)) {
                //指定含税标志的最低价格集合
                List<MsgMinPrice> msgMinPrices = containTaxFlag2LowestPrice.get(containTaxFlag);
                //发送价格变化的警报消息
                alertMessageService.sendPriceChangeAlert(msgMinPrices);
            }

            //发送计算IWC的消息
            iwcSqsService.sendCalculateIWCMessage(priceMessage, containTaxFlag2LowestPrice);

            //当前日期
            LocalDate now = LocalDate.now();

            //刷新当日房价的缓存
            for (Map.Entry<Integer, List<MsgMinPrice>> entry : containTaxFlag2LowestPrice.entrySet()) {
                List<PriceCacheDTO> priceCacheDTOList = entry.getValue().stream().map(msgMinPrice -> {
                    //缓存key
                    String cacheKey = beanMapper.map(msgMinPrice, PriceCacheKeyDTO.class).generateLowestRateCacheKey();

                    //缓存value
                    BigDecimal minPrice = msgMinPrice.getMinPrice();
                    String cacheValue = Objects.nonNull(minPrice) ? minPrice.toString() : "";

                    //计算入住日期和当前时间的差值
                    LocalDate checkInDateLocalDate = DateTimeUtils.strToLocalDate(msgMinPrice.getCheckInDate().toString(), BaseConstant.DATE_FORMAT_YYYYMMDD);
                    // 比较两个日期之间的天数差
                    long diffDays = ChronoUnit.DAYS.between(checkInDateLocalDate, now);
                    //最低房价存储到缓存中的过期时间 = 时间差值 + 1天
                    long expireTime = Math.abs(diffDays) + 1;

                    return new PriceCacheDTO(cacheKey, cacheValue, expireTime);
                }).toList();
                List<String> cacheKeyList = priceCacheDTOList.stream().map(PriceCacheDTO::getKey).toList();

                //刷新缓存
                Map<String, String> cacheKey2Value = priceCacheDTOList.stream().collect(Collectors.toMap(PriceCacheDTO::getKey, PriceCacheDTO::getValue));
                redisTemplate.opsForValue().multiSet(cacheKey2Value);

                Map<String, Long> cacheKey2ExpireTime = priceCacheDTOList.stream().collect(Collectors.toMap(PriceCacheDTO::getKey, PriceCacheDTO::getExpireTime));
                //用一次连接，设置过期时间
                redisTemplate.executePipelined((RedisCallback<Void>) connection -> {
                    for (String key : cacheKeyList) {
                        connection.expire(key.getBytes(), 3600L * 24L * cacheKey2ExpireTime.get(key));
                    }
                    return null;
                });
            }
        }

        /**
         * 获取消息中的最低价格
         *
         * @param priceMessage 房价消息
         * @param priceDTOList 房价数据
         * @return java.util.Map<java.lang.String, cloud.wconnect.rateshop.pojo.dto.rateshop.sender.MsgMinPrice>
         * @author lichen
         * @date 2025/4/4
         */
        private Map<Integer, List<MsgMinPrice>> getMessageLowestPrice(PriceMessage priceMessage, List<PriceDTO> priceDTOList) {
            //不同含税标志下的最低房价
            Map<Integer, List<MsgMinPrice>> containTaxFlag2MsgMinPrice = new HashMap<>();

            for (Integer containTaxFlag : Lists.newArrayList(Setting.BEFORE_TAX_RATE, Setting.AFTER_TAX_RATE)) {
                List<MsgMinPrice> msgMinPrices = new ArrayList<>();

                //获取消息中的最低价格
                for (PriceDTO priceDTO : priceDTOList) {
                    Long checkInDate = priceDTO.getCheckInDate();
                    for (PlatformDTO platformDTO : priceDTO.getPlatform()) {
                        //平台id
                        Integer platformId = platformDTO.getId();

                        /**
                         * 获取当前消息中的最低价格
                         * roomDTOS是empty时表示sold out，最低价格记录为0
                         */
                        List<RoomDTO> roomDTOS = platformDTO.getRoom();
                        BigDecimal minPrice;
                        if (CollectionUtils.isEmpty(roomDTOS)) {
                            minPrice = BigDecimal.ZERO;
                        } else {
                            minPrice = roomDTOS.stream()
                                    .map(roomDTO -> containTaxFlag == Setting.BEFORE_TAX_RATE ? roomDTO.getRate() : roomDTO.getAfterTaxRate())
                                    .filter(Objects::nonNull).min(BigDecimal::compareTo)
                                    .orElse(null);
                        }

                        //构建最低价格对象
                        MsgMinPrice msgMinPrice = MsgMinPrice.builder()
                                .hotelId(priceMessage.getHotelId())
                                .platformId(platformId)
                                .checkInDate(checkInDate)
                                .minPrice(minPrice)
                                .currencyId(priceMessage.getCurrencyId())
                                .rateType(priceMessage.getRateType())
                                .datasourceId(priceMessage.getDatasourceId())
                                .containTaxFlag(containTaxFlag)
                                .pos(priceMessage.getPos())
                                .build();
                        msgMinPrices.add(msgMinPrice);
                    }
                }

                //将最低价格对象放入map中
                containTaxFlag2MsgMinPrice.put(containTaxFlag, msgMinPrices);
            }

            return containTaxFlag2MsgMinPrice;
        }

    }