# HistoryAndForecastListenerFileServiceByPep.java

    package cloud.wconnect.rateshop.service.rateshop.historyandforecast.messagelistener;

    import cloud.wconnect.commonlib.component.BeanMapper;
    import cloud.wconnect.commonlib.cons.BaseConstant;
    import cloud.wconnect.commonlib.cons.BaseErrorCodeEnum;
    import cloud.wconnect.commonlib.cons.SystemPlatformEnum;
    import cloud.wconnect.commonlib.exception.BadRequestException;
    import cloud.wconnect.commonlib.exception.ParseReportException;
    import cloud.wconnect.commonlib.pojo.dto.PropertyDTO;
    import cloud.wconnect.commonlib.pojo.dto.RoomsConfigDTO;
    import cloud.wconnect.commonlib.pojo.dto.calculateoperation.CalculateOOORoomsDTO;
    import cloud.wconnect.commonlib.pojo.dto.calculateoperation.CalculateScOccDTO;
    import cloud.wconnect.commonlib.pojo.dto.calculateoperation.CalculateTransientRoomsDTO;
    import cloud.wconnect.commonlib.pojo.dto.internationalization.InternationalizationNumberDTO;
    import cloud.wconnect.commonlib.service.CalculatePropertyOperationService;
    import cloud.wconnect.commonlib.service.RoomsConfigInternalService;
    import cloud.wconnect.commonlib.util.CsvFileUtils;
    import cloud.wconnect.commonlib.util.DateTimeUtils;
    import cloud.wconnect.commonlib.util.InternationalizationUtils;
    import cloud.wconnect.commonlib.util.JsonUtils;
    import cloud.wconnect.rateshop.cons.revenue.RevenueCons;
    import cloud.wconnect.rateshop.entity.rateshop.PropertyOperation;
    import cloud.wconnect.rateshop.entity.rateshop.RateShopOccForecast;
    import cloud.wconnect.rateshop.entity.revenue.RevenueAdrConfig;
    import cloud.wconnect.rateshop.pojo.bo.rateshop.HistoryAndForecastForRateShopBO;
    import cloud.wconnect.rateshop.pojo.dto.rateshop.HistoryAndForecastOccDTO;
    import cloud.wconnect.rateshop.pojo.dto.rateshop.occforcast.OccForecastDTO;
    import cloud.wconnect.rateshop.service.rateshop.IPropertyOperationService;
    import cloud.wconnect.rateshop.service.rateshop.IRateShopOccForecastService;
    import cloud.wconnect.rateshop.service.revenue.IRevenueAdrConfigService;
    import com.baomidou.dynamic.datasource.annotation.DS;
    import com.fasterxml.jackson.databind.ObjectMapper;
    import com.google.common.collect.Lists;
    import lombok.RequiredArgsConstructor;
    import lombok.extern.slf4j.Slf4j;
    import org.apache.commons.collections.CollectionUtils;
    import org.apache.commons.csv.CSVFormat;
    import org.apache.commons.csv.CSVRecord;
    import org.apache.commons.lang3.StringUtils;
    import org.springframework.stereotype.Component;
    import org.springframework.transaction.annotation.Transactional;

    import java.io.IOException;
    import java.io.InputStream;
    import java.io.InputStreamReader;
    import java.math.BigDecimal;
    import java.math.RoundingMode;
    import java.time.LocalDate;
    import java.util.*;
    import java.util.function.Function;
    import java.util.stream.Collectors;

    /**
     * 解析 History And Forecast 报告
     *
     * @author yyt
     * @date 2026/3/10
     */
    @RequiredArgsConstructor
    @Component
    @DS("rate_shop")
    @Slf4j
    public class HistoryAndForecastListenerFileServiceByPep implements IHistoryAndForecastListenerFileService {

        private final RoomsConfigInternalService roomsConfigInternalService;
        private final IPropertyOperationService propertyOperationService;
        private final IRateShopOccForecastService rateShopOccForecastService;
        private final IRevenueAdrConfigService revenueAdrConfigService;
        private final CalculatePropertyOperationService calculatePropertyOperationService;
        private final BeanMapper beanMapper;
        private final ObjectMapper objectMapper;

        /**
         * 支持的报告来源
         *
         * @param source 报告来源
         * @return boolean
         * @author yyt
         * @date 2026/3/10
         */
        @Override
        public boolean supports(String source) {
            return SystemPlatformEnum.PEP.getDesc().equals(source);
        }

        /**
         * 解析报告
         *
         * @param in          报告文件流
         * @param propertyDTO 资产DTO
         * @param startDate   数据日期，开始时间(整点时刻时间戳)
         * @param endDate     数据日期，结束时间(整点时刻时间戳)
         * @return java.util.List<cloud.wconnect.rateshop.pojo.dto.rateshop.messagelistener.HistoryAndForecastForRateShopDTO>
         * @author yyt
         * @date 2026/3/10
         */
        @Override
        public List<HistoryAndForecastForRateShopBO> parse(InputStream in, PropertyDTO propertyDTO, long startDate, long endDate) throws ParseReportException {
            String timezone = propertyDTO.getTimezone();
            // 获取国际化的配置
            InternationalizationNumberDTO number = propertyDTO.getInternationalization().getNumber();
            RoundingMode roundingMode = InternationalizationUtils.getRoundingMode(number.getCalculationRule());

            // 获取房间数配置
            RoomsConfigDTO roomsConfig = roomsConfigInternalService.getRoomsConfig(propertyDTO.getId());
            Integer configTotalRooms = Objects.isNull(roomsConfig) ? null : roomsConfig.getRooms();

            try (InputStreamReader inputStreamReader = new InputStreamReader(in)) {
                ArrayList<HistoryAndForecastForRateShopBO> resultList = Lists.newArrayList();

                //获取CSV数据迭代器
                Iterable<CSVRecord> iterable = CSVFormat.EXCEL.parse(inputStreamReader);
                Iterator<CSVRecord> iterator = iterable.iterator();

                //跳过第一行标题行
                iterator.next();

                while (iterator.hasNext()) {
                    CSVRecord csvRecord = iterator.next();

                    //日期
                    String dateStr = CsvFileUtils.get(csvRecord, "A");
                    if(StringUtils.isEmpty(dateStr)) continue;
                    LocalDate localDate = DateTimeUtils.strToLocalDate(dateStr, BaseConstant.DATE_FORMAT_MMMDDYYYY_USA);
                    long dataDate = DateTimeUtils.localDateToLong(localDate, timezone);
                    if (dataDate < startDate || dataDate > endDate) continue;

                    String formattedDateStr = DateTimeUtils.timestampToStr(dataDate, BaseConstant.DATE_FORMAT_YYYY_MM_DD, propertyDTO.getTimezone());

                    // 已售房间数
                    String roomsSoldStr = CsvFileUtils.getAndFormatNumber(csvRecord, "D");
                    Integer roomsSold = StringUtils.isEmpty(roomsSoldStr) ? null : Integer.parseInt(roomsSoldStr);

                    // 总房间数
                    String totalRoomsStr = CsvFileUtils.getAndFormatNumber(csvRecord, "C");
                    Integer totalRooms = StringUtils.isEmpty(totalRoomsStr) ? null : Integer.parseInt(totalRoomsStr);

                    // 获取 PC occ
                    BigDecimal pcOccPercent;
                    // 如果配置了房间数，则根据 pc occ 计算规则判断：1.使用配置的房间数计算 pc occ，2.直接使用文件中的 occ 作为 pc occ
                    if (Objects.nonNull(roomsConfig) && Objects.nonNull(roomsSold)
                            && (roomsConfig.getOccRule() == RevenueCons.USE_ROOMS_TO_CALCULATE)) {
                        // pc occ = sold rooms / total rooms
                        pcOccPercent = BigDecimal.valueOf(roomsSold).multiply(BigDecimal.valueOf(100))
                                .divide(new BigDecimal(configTotalRooms), BaseConstant.DATA_PRECISION, roundingMode);
                    } else {
                        // 入住率
                        String occupancyStr = CsvFileUtils.getAndFormatNumber(csvRecord, "M");
                        pcOccPercent = StringUtils.isEmpty(occupancyStr) ?
                                null : new BigDecimal(occupancyStr).setScale(BaseConstant.DATA_PRECISION, roundingMode);
                    }

                    // 平均房价
                    String adrStr = CsvFileUtils.getAndFormatNumber(csvRecord, "L");
                    BigDecimal adr = StringUtils.isEmpty(adrStr) ? null : new BigDecimal(adrStr)
                            .setScale(BaseConstant.DATA_PRECISION, roundingMode);

                    // Rooms Available
                    String availableRoomsStr = CsvFileUtils.getAndFormatNumber(csvRecord, "E");
                    Integer roomsAvailable = StringUtils.isEmpty(availableRoomsStr) ? null : Integer.parseInt(availableRoomsStr);

                    resultList.add(HistoryAndForecastForRateShopBO.builder()
                            .date(formattedDateStr)
                            .roomsSold(roomsSold)
                            .pcOcc(pcOccPercent)
                            .adr(adr)
                            .roomsAvailable(roomsAvailable)
                            .totalRooms(totalRooms)
                            .build());
                }
                return resultList;
            } catch (IOException e) {
                throw new BadRequestException(BaseErrorCodeEnum.SYSTEM_ERROR.getI18nCode(),
                        BaseErrorCodeEnum.SYSTEM_ERROR.getCode(),
                        "Parse pep [History And Forecast] report error", e);
            }
        }

        /**
         * 处理数据
         *
         * @param propertyDTO 资产DTO
         * @param boList      报告数据
         * @param operatorId  操作人 id
         * @author yyt
         * @date 2026/3/10
         */
        @Override
        @Transactional(rollbackFor = Exception.class)
        public void saveReportData(PropertyDTO propertyDTO, List<HistoryAndForecastForRateShopBO> boList, int operatorId) {
            // 保存数据
            saveHistoryAndForecastReportData(propertyDTO, boList, operatorId);
            // 保存occ数据
            saveOccForecast(propertyDTO, operatorId, boList);
        }

        /**
         * 保存 history and forecast 数据
         *
         * @param propertyDTO 资产
         * @param boList      history and forecast 数据
         * @param operatorId  操作人
         * @author yyt
         * @date 2026/3/10
         **/
        private void saveHistoryAndForecastReportData(PropertyDTO propertyDTO, List<HistoryAndForecastForRateShopBO> boList ,int operatorId) {
            if(CollectionUtils.isEmpty(boList)) return;

            // 查询已存在的数据，然后重新计算相关数据
            Integer propertyId = propertyDTO.getId();
            String timezone = propertyDTO.getTimezone();
            List<PropertyOperation> propertyOperations = boList.stream().map(item -> {
                PropertyOperation propertyOperation = beanMapper.map(item, PropertyOperation.class);
                propertyOperation.setPropertyId(propertyId);
                propertyOperation.setTargetDate(DateTimeUtils.dateStrToEpochSecond(item.getDate(), BaseConstant.DATE_FORMAT_YYYY_MM_DD, timezone));
                return propertyOperation;
            }).sorted(Comparator.comparing(PropertyOperation::getTargetDate)).collect(Collectors.toList());
            Long startDate = propertyOperations.get(0).getTargetDate();
            Long endDate = propertyOperations.get(propertyOperations.size() - 1).getTargetDate();
            Map<Long, PropertyOperation> existPropertyOperationByDate = propertyOperationService.listByDateRange(propertyDTO.getId(), startDate, endDate)
                    .stream().collect(Collectors.toMap(PropertyOperation::getTargetDate, v -> v));

            //资产配置的规则
            RevenueAdrConfig revenueAdrConfig = revenueAdrConfigService.get(propertyDTO.getId());
            boolean checkTransientDeduct = Objects.nonNull(revenueAdrConfig) && revenueAdrConfig.getCheckTransientDeduct();

            long now = DateTimeUtils.nowTimestamp();
            RoundingMode roundingMode = InternationalizationUtils.getRoundingMode(propertyDTO.getInternationalization().getNumber().getCalculationRule());

            for (PropertyOperation propertyOperation : propertyOperations) {
                PropertyOperation existPropertyOperation = existPropertyOperationByDate.getOrDefault(propertyOperation.getTargetDate(), new PropertyOperation());
                // 计算故障房
                propertyOperation.setCompRooms(existPropertyOperation.getCompRooms());
                propertyOperation.setOooRooms(calculatePropertyOperationService.calculateOooRooms(beanMapper.map(propertyOperation, CalculateOOORoomsDTO.class)));

                // 计算散客
                propertyOperation.setGroupSold(existPropertyOperation.getGroupSold());
                propertyOperation.setTransientSold(checkTransientDeduct ?
                        calculatePropertyOperationService.calculateTransientRooms(new CalculateTransientRoomsDTO(propertyOperation.getRoomsSold(), propertyOperation.getGroupSold())) : null);

                // 计算 sc occ
                propertyOperation.setScOcc(calculatePropertyOperationService.calculateScOcc(beanMapper.map(propertyOperation, CalculateScOccDTO.class), roundingMode));
                propertyOperation.setPmsUpdatedBy(operatorId);
                propertyOperation.setPmsUpdatedTime(now);
            }

            propertyOperationService.saveHistoryAndForecastReportData(propertyDTO, propertyOperations);
        }

        /**
         * 保存 occ forecast 数据
         *
         * @param propertyDTO 资产
         * @param boList      history and forecast 数据
         * @param operatorId  操作人
         * @author yyt
         * @date 2026/3/10
         **/
        private void saveOccForecast(PropertyDTO propertyDTO, int operatorId, List<HistoryAndForecastForRateShopBO> boList) {
            if (CollectionUtils.isEmpty(boList)) return;

            int propertyId = propertyDTO.getId();
            String timezone = propertyDTO.getTimezone();

            //转化为DTO
            List<HistoryAndForecastOccDTO> historyAndForecastOccDTOS = beanMapper.mapList(boList, HistoryAndForecastOccDTO.class,
                    (bo, dto) -> dto.setTheDate(Long.parseLong(bo.getDate().replace("-", ""))));

            List<Long> theDateList = historyAndForecastOccDTOS.stream().map(HistoryAndForecastOccDTO::getTheDate).toList();
            long startDateQuery = Collections.min(theDateList);
            long endDateQuery = Collections.max(theDateList);

            // 查找指定时间范围内的数据
            List<RateShopOccForecast> occForecastList = rateShopOccForecastService.listByDateRangeForUpdate(propertyId, startDateQuery, endDateQuery);
            Map<Long, RateShopOccForecast> theDate2OccForecast = occForecastList.stream()
                    .collect(Collectors.toMap(RateShopOccForecast::getTheDate, Function.identity()));

            RoundingMode roundingMode = InternationalizationUtils.getRoundingMode(propertyDTO.getInternationalization().getNumber().getCalculationRule());

            // 获取当前时间的 key
            String currentKey = DateTimeUtils.parseToLong(DateTimeUtils.dateTimeNow(timezone),
                    BaseConstant.DATE_FORMAT_YYYYMMDDHH).toString();
            long nowTimestamp = DateTimeUtils.nowTimestamp();

            //循环报告中的Occ数据
            for (HistoryAndForecastOccDTO historyAndForecastOccDTO : historyAndForecastOccDTOS) {
                //获取数据库中指定营业日期的occ数据
                long theDate = historyAndForecastOccDTO.getTheDate();
                RateShopOccForecast oriRateShopOccForecast = theDate2OccForecast.get(theDate);
                //数据库中的数据为空时，新增
                if (Objects.isNull(oriRateShopOccForecast)) {
                    OccForecastDTO currentOccForecastDTO = rateShopOccForecastService.packageOccForecastDTO(currentKey, historyAndForecastOccDTO, roundingMode, operatorId, nowTimestamp);
                    occForecastList.add(new RateShopOccForecast()
                            .setPropertyId(propertyId)
                            .setTheDate(theDate)
                            .setValues(JsonUtils.convertObjListToString(objectMapper, Lists.newArrayList(currentOccForecastDTO))));
                } else {//更新
                    //获取数据库中不同时刻的Occ数据
                    List<OccForecastDTO> oriValuesDTOList = JsonUtils.convertStringToList(objectMapper,
                            oriRateShopOccForecast.getValues(), OccForecastDTO.class);

                    Optional<OccForecastDTO> optional = oriValuesDTOList.stream().filter(dto -> dto.getKey().equals(currentKey)).findFirst();
                    if (optional.isPresent()) {
                        //存在当前时刻的occ数据，更新
                        OccForecastDTO occForecastDTO = optional.get();
                        oriValuesDTOList.remove(occForecastDTO);
                        historyAndForecastOccDTO.setCompRooms(occForecastDTO.getCompRooms());
                        OccForecastDTO currentOccForecastDTO = rateShopOccForecastService.packageOccForecastDTO(currentKey, historyAndForecastOccDTO, roundingMode, operatorId, nowTimestamp);
                        oriValuesDTOList.add(currentOccForecastDTO);
                    } else {
                        //不存在当前时刻的occ数据，新增
                        OccForecastDTO currentOccForecastDTO = rateShopOccForecastService.packageOccForecastDTO(currentKey, historyAndForecastOccDTO, roundingMode, operatorId, nowTimestamp);
                        oriValuesDTOList.add(currentOccForecastDTO);
                    }
                    oriRateShopOccForecast.setValues(JsonUtils.convertObjListToString(objectMapper, oriValuesDTOList));
                }
            }
            // 批量保存
            rateShopOccForecastService.saveOrUpdateBatch(occForecastList);
        }

    }