# MarketDemandTrendConsumer.java

    package cloud.wconnect.rateshop.consumer.marketdemand;

    import cloud.wconnect.commonlib.util.JsonUtils;
    import cloud.wconnect.commonlib.util.TransactionUtils;
    import cloud.wconnect.rateshop.cons.rateshop.RateShopCons;
    import cloud.wconnect.rateshop.entity.rateshop.Trend;
    import cloud.wconnect.rateshop.entity.rateshop.TrendSourceLocation;
    import cloud.wconnect.rateshop.pojo.dto.rateshop.consumer.marketdemand.OpenHotel;
    import cloud.wconnect.rateshop.pojo.dto.rateshop.consumer.marketdemand.TrendMessage;
    import cloud.wconnect.rateshop.service.rateshop.IAlertMessageService;
    import cloud.wconnect.rateshop.service.rateshop.IIwcSqsService;
    import cloud.wconnect.rateshop.service.rateshop.ITrendService;
    import cloud.wconnect.rateshop.service.rateshop.ITrendSourceLocationService;
    import com.baomidou.dynamic.datasource.annotation.DS;
    import com.baomidou.mybatisplus.core.conditions.query.LambdaQueryWrapper;
    import com.fasterxml.jackson.databind.ObjectMapper;
    import com.google.common.collect.Lists;
    import lombok.extern.slf4j.Slf4j;
    import org.apache.commons.collections.CollectionUtils;
    import org.springframework.beans.factory.annotation.Autowired;
    import org.springframework.stereotype.Component;
    import org.springframework.transaction.annotation.Transactional;
    import software.amazon.awssdk.services.sqs.model.Message;

    import java.math.BigDecimal;
    import java.math.RoundingMode;
    import java.util.ArrayList;
    import java.util.List;
    import java.util.Map;
    import java.util.Objects;
    import java.util.concurrent.Executor;
    import java.util.function.Function;
    import java.util.stream.Collectors;

    /**
     * 市场需求 消息消费方
     *
     * @author lichen
     * @date 2023/6/8
     */
    @Component
    @DS("rate_shop")
    @Slf4j
    public class MarketDemandTrendConsumer {

        @Autowired
        private ObjectMapper objectMapper;
        @Autowired
        private IAlertMessageService alertMessageService;
        @Autowired
        private ITrendSourceLocationService locationService;
        @Autowired
        private IIwcSqsService iwcSqsService;
        @Autowired
        private ITrendService trendService;
        @Autowired
        private Executor executor;

        /**
         * 处理市场需求消息
         *
         * @param message 消息对象
         * @return void
         * @author lichen
         * @date 2023/6/8
         */
        @Transactional(rollbackFor = Exception.class)
        public void handleMessage(Message message) {
            //把消息内容转化成TrendMessage
            TrendMessage trendMessage = JsonUtils.convertStringToObject(objectMapper, message.body(), TrendMessage.class);
            //查询locationId
            TrendSourceLocation trendSourceLocation = locationService.getById(trendMessage.getLocationId());

            if (Objects.isNull(trendSourceLocation)) return;

            if (trendMessage.getType() == RateShopCons.TYPE_TOTAL_HOTEL_NUM) {//地区注册总酒店数
                Integer totalHotel = trendMessage.getTotalHotel();
                if (Objects.isNull(totalHotel) || totalHotel == 0) return;

                //更新地区的总酒店数
                TrendSourceLocation updateEntity = new TrendSourceLocation();
                updateEntity.setId(trendSourceLocation.getId());
                updateEntity.setTotalHotelNum(totalHotel);
                updateEntity.setQueryDate(trendMessage.getQueryDate());
                locationService.updateById(updateEntity);
            } else {//地区营业酒店数

                //地区的总酒店数为空时，无法计算级别
                Integer totalHotelNum = trendSourceLocation.getTotalHotelNum();
                if (Objects.isNull(totalHotelNum)) return;

                //营业酒店数为空时，不做处理
                List<OpenHotel> openHotelList = trendMessage.getOpenHotel();
                if (CollectionUtils.isEmpty(openHotelList)) return;

                //查询更新前的数据
                List<Long> dateList = openHotelList.stream().map(OpenHotel::getDate).toList();
                List<Trend> trendListByUpdateBefore = trendService.list(new LambdaQueryWrapper<Trend>().eq(Trend::getTrendLocationId, trendMessage.getLocationId())
                        .in(Trend::getCheckInDate, dateList));
                Map<Long, Trend> date2TrendByUpdateBefore = trendListByUpdateBefore.stream().collect(Collectors.toMap(Trend::getCheckInDate, Function.identity()));

                ArrayList<Trend> insertList = Lists.newArrayListWithCapacity(openHotelList.size());
                ArrayList<Trend> updateList = Lists.newArrayListWithCapacity(openHotelList.size());
                //新增或修改的数据集合
                ArrayList<Trend> allList = Lists.newArrayListWithCapacity(openHotelList.size());

                for (OpenHotel openHotel : openHotelList) {
                    //计算趋势比率:（该地区酒店总数 - 该地区营业的酒店数）/ 该地区酒店总数
                    int diffHotels = totalHotelNum - openHotel.getNumber();
                    BigDecimal rate = new BigDecimal(diffHotels).divide(new BigDecimal(totalHotelNum), 6, RoundingMode.HALF_UP);

                    //计算趋势级别
                    int level;
                    if (rate.compareTo(BigDecimal.ZERO) > 0 && rate.compareTo(new BigDecimal("0.3")) <= 0) {
                        level = 1;
                    } else if (rate.compareTo(new BigDecimal("0.3")) > 0 && rate.compareTo(new BigDecimal("0.7")) <= 0) {
                        level = 2;
                    } else {//大于0.7
                        level = 3;
                    }

                    Trend trend = Trend.builder().checkInDate(openHotel.getDate())
                            .trendLocationId(trendMessage.getLocationId())
                            .level(level)
                            .rate(rate)
                            .queryDate(trendMessage.getQueryDate())
                            .build();

                    //更新前的数据
                    Trend oriTrend = date2TrendByUpdateBefore.get(openHotel.getDate());
                    if(Objects.isNull(oriTrend)){
                        insertList.add(trend);

                        allList.add(trend);
                    }else{
                        if(!oriTrend.getRate().equals(rate)){
                            trend.setId(oriTrend.getId());
                            updateList.add(trend);

                            allList.add(trend);
                        }
                    }
                }

                trendService.saveBatch(insertList);
                trendService.updateBatchById(updateList);

                if(CollectionUtils.isNotEmpty(allList)){
                    //事务提交后执行
                    TransactionUtils.afterCommitAsyncExecute(executor, () -> {
                        //生成趋势级别变化的消息并发送
                        alertMessageService.generateMarketDemandMessage(allList, trendMessage.getLocationId());
                        //发送计算iwc消息
                        iwcSqsService.sendCalculateIWCMessage(trendMessage.getLocationId(),
                                allList.stream().map(Trend::getCheckInDate).toList());
                    });
                }
            }
        }
    }