# SalesServiceImpl.java

    package cloud.wconnect.workperformance.service.mor.impl;

    import cloud.wconnect.commonlib.component.BeanMapper;
    import cloud.wconnect.commonlib.cons.BaseConstant;
    import cloud.wconnect.commonlib.cons.MoudleEventSubjectEnum;
    import cloud.wconnect.commonlib.exception.ListenerException;
    import cloud.wconnect.commonlib.pojo.dto.PropertyDTO;
    import cloud.wconnect.commonlib.util.DateTimeUtils;
    import cloud.wconnect.commonlib.util.JsonUtils;
    import cloud.wconnect.commonlib.util.TransactionUtils;
    import cloud.wconnect.commonlib.util.YearMonthUtils;
    import cloud.wconnect.sns.http.processor.Sender;
    import cloud.wconnect.workperformance.cons.operatingstatisticsdaily.DataUpdateResult;
    import cloud.wconnect.workperformance.cons.operatingstatisticsdaily.OperatingStatisticsDatasourceEnum;
    import cloud.wconnect.workperformance.entity.mor.Sales;
    import cloud.wconnect.workperformance.entity.mor.SalesOperatingStatistics;
    import cloud.wconnect.workperformance.mapper.mor.SalesMapper;
    import cloud.wconnect.workperformance.pojo.bo.modelevent.salessource.MorSalesSourceEventBO;
    import cloud.wconnect.workperformance.pojo.bo.mor.sales.SalesSourceOperatingStatisticsListBO;
    import cloud.wconnect.workperformance.pojo.dto.mor.sales.*;
    import cloud.wconnect.workperformance.pojo.query.mor.sales.SalesQuery;
    import cloud.wconnect.workperformance.service.mor.ISalesOperatingStatisticsService;
    import cloud.wconnect.workperformance.service.mor.ISalesService;
    import cloud.wconnect.workperformance.service.mor.operatingstatistics.ISalesOperatingStatisticsParseService;
    import com.baomidou.dynamic.datasource.annotation.DS;
    import com.baomidou.mybatisplus.core.conditions.query.LambdaQueryWrapper;
    import com.baomidou.mybatisplus.core.conditions.query.QueryWrapper;
    import com.baomidou.mybatisplus.extension.service.impl.ServiceImpl;
    import com.fasterxml.jackson.databind.ObjectMapper;
    import com.google.common.collect.Lists;
    import com.google.common.collect.Maps;
    import lombok.extern.slf4j.Slf4j;
    import org.apache.commons.collections.CollectionUtils;
    import org.apache.commons.lang3.ObjectUtils;
    import org.springframework.beans.factory.annotation.Qualifier;
    import org.springframework.context.ApplicationContext;
    import org.springframework.stereotype.Service;
    import org.springframework.transaction.annotation.Transactional;

    import jakarta.annotation.Resource;

    import java.io.InputStream;
    import java.math.BigDecimal;
    import java.time.LocalDate;
    import java.time.YearMonth;
    import java.util.*;
    import java.util.concurrent.Executor;
    import java.util.function.Function;
    import java.util.stream.Collectors;

    /**
     * <p>
     * 销售信息 服务实现类
     * </p>
     *
     * @author zc
     * @since 2023-04-04
     */
    @Slf4j
    @Service
    @DS("work_performance")
    public class SalesServiceImpl extends ServiceImpl<SalesMapper, Sales> implements ISalesService {
        @Resource
        private ISalesOperatingStatisticsService salesOperatingStatisticsService;
        @Resource
        @Qualifier("moduleEventSender")
        private Sender moduleEventSender;
        @Resource
        private ObjectMapper objectMapper;
        @Resource
        private BeanMapper beanMapper;
        @Resource
        private Executor executor;
        @Resource
        private ApplicationContext applicationContext;

        /**
         * 获取 sales sources 列表数据
         *
         * @param salesQuery 查询条件
         * @return sales sources
         * @author zc
         * @date 2024/5/31
         */
        @Override
        public List<SalesPropertyDTO> list(SalesQuery salesQuery) {
            Map<Integer, List<Sales>> propertyId2Sales = listSales(salesQuery.getPropertyIds(), salesQuery.getYears()).stream()
                    .collect(Collectors.groupingBy(Sales::getPropertyId));

            return salesQuery.getPropertyIds().stream().map(propertyId -> {
                Map<Integer, Map<Integer, Sales>> month2SalesByYear = propertyId2Sales.getOrDefault(propertyId, Collections.emptyList()).stream()
                        .collect(Collectors.groupingBy(item -> YearMonthUtils.getYearMonth(item.getTheMonth().toString()).getYear(),
                                Collectors.toMap(Sales::getTheMonth, Function.identity())));

                List<SalesYearDTO> salesYearDTOList = salesQuery.getYears().stream().map(year -> {
                    List<SalesMonthDTO> salesMonthDTOList = Lists.newArrayListWithExpectedSize(12);
                    Map<Integer, Sales> month2Sales = month2SalesByYear.getOrDefault(year, Collections.emptyMap());
                    for (int i = 1; i <= 12; i++) {
                        Sales sales = month2Sales.getOrDefault((int) YearMonthUtils.parseYearMonthToLong(YearMonth.of(year, i)), new Sales());
                        salesMonthDTOList.add(SalesMonthDTO.builder()
                                .month(i)
                                .manualSalesTotal(sales.getManualSalesTotal())
                                .reportSalesTotal(sales.getReportSalesTotal()).build());
                    }

                    return SalesYearDTO.builder().year(year).monthList(salesMonthDTOList).build();
                }).collect(Collectors.toList());

                return SalesPropertyDTO.builder()
                        .propertyId(propertyId)
                        .sales(salesYearDTOList).build();
            }).collect(Collectors.toList());
        }

        /**
         * 保存 sales sources 金额
         *
         * @param propertyId   资产 id
         * @param saveSalesDTO 保存内容
         * @param operatorId   操作人 id
         * @author zc
         * @date 2024/5/31
         */
        @Override
        @Transactional(rollbackFor = Exception.class)
        public void save(int propertyId, SaveSalesDTO saveSalesDTO, int operatorId) {
            Sales sales = Sales.builder()
                    .propertyId(propertyId)
                    .theYear(saveSalesDTO.getYear())
                    .theMonth((int) YearMonthUtils.parseYearMonthToLong(YearMonth.of(saveSalesDTO.getYear(), saveSalesDTO.getMonth())))
                    .manualSalesTotal(saveSalesDTO.getTotal())
                    .updatedBy(operatorId)
                    .updatedTime(DateTimeUtils.nowTimestamp())
                    .build();
            baseMapper.saveOrUpdateManualSalesTotalByUnique(sales);
        }

        /**
         * 获取年对应金额
         *
         * @param propertyId property id
         * @return 年对应金额
         * @author zc
         * @date 2023/4/4
         */
        @Override
        public List<SalesYearTotalDTO> listYearTotal(int propertyId) {
            List<Sales> salesList = list(new LambdaQueryWrapper<Sales>()
                    .eq(Sales::getPropertyId, propertyId));
            if (CollectionUtils.isEmpty(salesList)) {
                return Collections.emptyList();
            }
            return salesList.stream().collect(Collectors.groupingBy(Sales::getTheYear, Collectors.reducing(BigDecimal.ZERO, item ->
                            ObjectUtils.firstNonNull(item.getManualSalesTotal(), item.getReportSalesTotal(), BigDecimal.ZERO), BigDecimal::add)))
                    .entrySet().stream().map(entry ->
                            SalesYearTotalDTO.builder()
                                    .year(entry.getKey())
                                    .yearTotalDollars(entry.getValue())
                                    .build()).collect(Collectors.toList());
        }

        /**
         * 获取 sales 数据
         *
         * @param propertyIds property id
         * @param years       year
         * @return sales
         * @author zc
         * @date 2023/4/4
         */
        private List<Sales> listSales(List<Integer> propertyIds, List<Integer> years) {
            if (CollectionUtils.isEmpty(propertyIds) || CollectionUtils.isEmpty(years)) {
                return Collections.emptyList();
            }
            return list(new QueryWrapper<Sales>().lambda()
                    .in(Sales::getPropertyId, propertyIds)
                    .in(Sales::getTheYear, years));
        }

        /**
         * 解析报告
         *
         * @param propertyList         资产集合
         * @param in                   报告文件流
         * @param source               数据源
         * @param propertyId2DateRange 报告数据日期范围(yyyyMMdd)
         * @return SalesSourceOperatingStatisticsListBO 集合
         * @author zc
         * @date 2024/6/3
         */
        @Override
        public List<SalesSourceOperatingStatisticsListBO> parseReport(List<PropertyDTO> propertyList, InputStream in,
                                                                      String source, Map<Integer, List<Long>> propertyId2DateRange) {
            Collection<ISalesOperatingStatisticsParseService> parserList = applicationContext.getBeansOfType(ISalesOperatingStatisticsParseService.class).values();
            for (ISalesOperatingStatisticsParseService item : parserList) {
                if (item.supports(source)) {
                    return item.parse(propertyList, in, propertyId2DateRange);
                }
            }
            throw new ListenerException("Not found report parser");
        }

        /**
         * 处理报告数据
         *
         * @param propertyList         资产
         * @param boList               报告数据 BO
         * @param operatorId           操作人 id
         * @param propertyId2DateRange 数据日期范围，日期格式：yyyyMMdd
         * @param datasource           数据源
         * @return java.util.Map<java.lang.Integer, java.lang.Integer>
         * @author lichen
         * @date 2025/11/22
         */
        @Override
        @Transactional(rollbackFor = Exception.class)
        public Map<Integer, Integer> handleOperatingStatistics(List<PropertyDTO> propertyList, List<SalesSourceOperatingStatisticsListBO> boList,
                                                               int operatorId, Map<Integer, List<Long>> propertyId2DateRange, int datasource) {
            // 资产集合
            Map<Integer, PropertyDTO> propertyId2DTO = propertyList.stream().collect(Collectors.toMap(PropertyDTO::getId, Function.identity()));

            Set<LocalDate> localDates = propertyId2DateRange.values().stream().flatMap(Collection::stream)
                    .map(item -> DateTimeUtils.strToLocalDate(item.toString(), BaseConstant.DATE_FORMAT_YYYYMMDD))
                    .collect(Collectors.toSet());
            // 为防止少处理数据，将最小和最大日期减去1天和加1天
            LocalDate minLocalDate = Collections.min(localDates).minusDays(1);
            LocalDate maxLocalDate = Collections.max(localDates).plusDays(1);
            Integer minDate = Integer.parseInt(DateTimeUtils.localDateToStr(minLocalDate, BaseConstant.DATE_FORMAT_YYYYMMDD));
            Integer maxDate = Integer.parseInt(DateTimeUtils.localDateToStr(maxLocalDate, BaseConstant.DATE_FORMAT_YYYYMMDD));
            // 获取已存在的数据
            List<SalesOperatingStatistics> originList = salesOperatingStatisticsService.listByDateRange(propertyList.stream().map(PropertyDTO::getId).toList(),
                    minDate, maxDate);
            // 已存在的数据按照资产id和日期进行分组
            Map<Integer, Map<Integer, SalesOperatingStatistics>> propertyId2TheDate2OriginData = originList.stream().collect(
                    Collectors.groupingBy(SalesOperatingStatistics::getPropertyId, Collectors.toMap(SalesOperatingStatistics::getTheDate, Function.identity())));

            // 待处理的数据按照资产标识和日期进行分组
            Map<String, Map<String, SalesSourceOperatingStatisticsListBO>> abbr2TheDay2BO = boList.stream().collect(
                    Collectors.groupingBy(SalesSourceOperatingStatisticsListBO::getProperty,
                            Collectors.toMap(SalesSourceOperatingStatisticsListBO::getTheDay, Function.identity())));

            ArrayList<SalesOperatingStatistics> batchInsertList = Lists.newArrayListWithCapacity(boList.size());
            ArrayList<SalesOperatingStatistics> batchUpdateList = Lists.newArrayListWithCapacity(boList.size());
            ArrayList<Long> batchRemoveIdList = Lists.newArrayListWithCapacity(boList.size());
            // 资产对应的数据更新结果
            HashMap<Integer, Integer> propertyId2DataUpdateResult = Maps.newHashMapWithExpectedSize(propertyList.size());
            long nowTimestamp = DateTimeUtils.nowTimestamp();

            propertyId2DateRange.forEach((propertyId, dateRange) -> {
                // 当前资产DTO
                PropertyDTO propertyDTO = propertyId2DTO.get(propertyId);

                // 原来的数据
                Map<Integer, SalesOperatingStatistics> theDate2OriginData = propertyId2TheDate2OriginData.getOrDefault(propertyId, Collections.emptyMap());
                // 待处理的数据
                Map<String, SalesSourceOperatingStatisticsListBO> theDay2BO = abbr2TheDay2BO.getOrDefault(propertyDTO.getAbbr(), Collections.emptyMap());

                LocalDate startLocalDate = DateTimeUtils.strToLocalDate(dateRange.get(0).toString(), BaseConstant.DATE_FORMAT_YYYYMMDD);
                LocalDate endLocalDate = DateTimeUtils.strToLocalDate(dateRange.get(1).toString(), BaseConstant.DATE_FORMAT_YYYYMMDD);

                // 资产未更新的数量
                int nonUpdateCount = 0;
                // 无需处理的日期数量
                int noProcessDateCount = 0;
                // 总的日期数
                int dateCount = 0;

                // 循环开始日期和结束日期的每一天
                for (LocalDate localDate = startLocalDate; !localDate.isAfter(endLocalDate); localDate = localDate.plusDays(1)) {
                    dateCount ++;

                    // 待处理的数据
                    String theDay = DateTimeUtils.localDateToStr(localDate, BaseConstant.DATE_FORMAT_YYYY_MM_DD);
                    SalesSourceOperatingStatisticsListBO bo = theDay2BO.get(theDay);

                    // 原来的数据
                    Integer theDate = Integer.valueOf(DateTimeUtils.localDateToStr(localDate, BaseConstant.DATE_FORMAT_YYYYMMDD));
                    SalesOperatingStatistics originData = theDate2OriginData.get(theDate);

                    if(Objects.isNull(bo)){
                        if(Objects.nonNull(originData)){
                            if (originData.getDatasource() != OperatingStatisticsDatasourceEnum.M3_API.getValue()) {
                                // 待处理数据不存在，原来数据存在且数据源非M3 API时，则删除原来数据
                                batchRemoveIdList.add(originData.getId());
                            } else {
                                // 待处理数据不存在，原来数据存在且数据源为M3 API时，则不能更新
                                nonUpdateCount++;
                            }
                        }else{
                            // 待处理数据不存在，原来数据不存在时，则无需处理
                            noProcessDateCount++;
                        }
                    }else{
                        SalesOperatingStatistics entity = beanMapper.map(bo, SalesOperatingStatistics.class);
                        entity.setPropertyId(propertyDTO.getId());
                        entity.setTheDate(theDate);
                        entity.setUpdatedBy(operatorId);
                        entity.setUpdatedTime(nowTimestamp);
                        entity.setDatasource(datasource);

                        if (Objects.isNull(originData)) {
                            // 待处理数据存在，原来数据不存在时，则插入数据
                            batchInsertList.add(entity);
                        }else{
                            if (originData.getDatasource() == OperatingStatisticsDatasourceEnum.M3_API.getValue()) {
                                // 存在数据源等于M3 API的数据，则不更新
                                nonUpdateCount++;
                            } else {
                                // 存在数据源不等于M3 API的数据，则更新
                                entity.setId(originData.getId());
                                batchUpdateList.add(entity);
                            }
                        }
                    }
                }

                // 需要处理的日期数
                int needProcessDateCount = dateCount - noProcessDateCount;

                int dataUpdateResult;
                if (nonUpdateCount == needProcessDateCount) {
                    // 未更新的数量等于需要处理的数量，表示全部未更新
                    dataUpdateResult = DataUpdateResult.NOT_UPDATED.getValue();
                } else if (nonUpdateCount == 0) {
                    // 未更新的数量等于0，表示全部更新
                    dataUpdateResult = DataUpdateResult.ALL_UPDATED.getValue();
                } else {
                    // 部分更新
                    dataUpdateResult = DataUpdateResult.PARTIAL_UPDATED.getValue();
                }
                propertyId2DataUpdateResult.put(propertyDTO.getId(), dataUpdateResult);
            });

            // 批量新增
            salesOperatingStatisticsService.saveBatch(batchInsertList);

            // 批量更新，支持更新为null
            salesOperatingStatisticsService.batchUpdateWithNull(batchUpdateList);

            // 批量删除
            salesOperatingStatisticsService.removeByIds(batchRemoveIdList);

            // 收集变更的数据
            ArrayList<SalesOperatingStatistics> operatingStatistics = Lists.newArrayListWithCapacity(boList.size());
            if (CollectionUtils.isNotEmpty(batchInsertList)) {
                operatingStatistics.addAll(batchInsertList);
            }
            if (CollectionUtils.isNotEmpty(batchUpdateList)) {
                operatingStatistics.addAll(batchUpdateList);
            }

            // 生成需要新增或更新的 Sales 数据
            List<Sales> salesList = generateUpdateSalesList(propertyList, operatingStatistics);
            // 更新 Sales reportSalesTotal 字段
            saveOrUpdateBatch(salesList);

            // Payroll analysis 指标数据（Sales）发生变化时，发送 SNS 事件
            TransactionUtils.afterCommitAsyncExecute(executor, () -> {
                // 数据发生变更发送事件
                sendModuleEvent(operatingStatistics);
            });

            return propertyId2DataUpdateResult;
        }

        /**
         * 通过M3 API处理报告数据
         *
         * @param propertyDTO 资产
         * @param boList      报告数据 BO
         * @param operatorId  操作人 id
         * @author lichen
         * @date 2025/11/22
         */
        @Override
        @Transactional(rollbackFor = Exception.class)
        public void handleOperatingStatisticsByM3Api(PropertyDTO propertyDTO, List<SalesSourceOperatingStatisticsListBO> boList, int operatorId) {
            if (CollectionUtils.isEmpty(boList)) return;

            List<Integer> theDateList = boList.stream().map(item -> {
                LocalDate localDate = DateTimeUtils.strToLocalDate(item.getTheDay(), BaseConstant.DATE_FORMAT_YYYY_MM_DD);
                return Integer.parseInt(DateTimeUtils.localDateToStr(localDate, BaseConstant.DATE_FORMAT_YYYYMMDD));
            }).toList();
            // 获取该资产在指定时间范围内的数据
            Integer minTheDate = Collections.min(theDateList);
            Integer maxTheDate = Collections.max(theDateList);
            List<SalesOperatingStatistics> originOperatingStatistics = salesOperatingStatisticsService.listByDateRange(Collections.singletonList(propertyDTO.getId()),
                    minTheDate, maxTheDate);
            Map<Integer, SalesOperatingStatistics> theDate2OriginEntity = originOperatingStatistics.stream()
                    .collect(Collectors.toMap(SalesOperatingStatistics::getTheDate, Function.identity()));

            long nowTimestamp = DateTimeUtils.nowTimestamp();
            ArrayList<SalesOperatingStatistics> insertBatch = Lists.newArrayListWithCapacity(boList.size());
            ArrayList<SalesOperatingStatistics> updateBatch = Lists.newArrayListWithCapacity(boList.size());
            for (SalesSourceOperatingStatisticsListBO bo : boList) {
                LocalDate localDate = DateTimeUtils.strToLocalDate(bo.getTheDay(), BaseConstant.DATE_FORMAT_YYYY_MM_DD);
                int theDate = Integer.parseInt(DateTimeUtils.localDateToStr(localDate, BaseConstant.DATE_FORMAT_YYYYMMDD));
                // 指定日期原来的数据
                SalesOperatingStatistics originEntity = theDate2OriginEntity.get(theDate);
                if (Objects.isNull(originEntity)) {
                    SalesOperatingStatistics insertEntity = beanMapper.map(bo, SalesOperatingStatistics.class);
                    insertEntity.setPropertyId(propertyDTO.getId());
                    insertEntity.setTheDate(theDate);
                    insertEntity.setDatasource(OperatingStatisticsDatasourceEnum.M3_API.getValue());
                    insertEntity.setUpdatedBy(operatorId);
                    insertEntity.setUpdatedTime(nowTimestamp);
                    // 原来的数据为null，则新增
                    insertBatch.add(insertEntity);
                } else {
                    originEntity.setTotalRev(bo.getTotalRev());
                    originEntity.setUpdatedBy(operatorId);
                    originEntity.setUpdatedTime(nowTimestamp);
                    originEntity.setDatasource(OperatingStatisticsDatasourceEnum.M3_API.getValue());
                    // 原来的数据不为null，则更新
                    updateBatch.add(originEntity);
                }
            }

            // 批量新增
            salesOperatingStatisticsService.saveBatch(insertBatch);

            // 批量更新，支持更新为null
            salesOperatingStatisticsService.batchUpdateWithNull(updateBatch);

            ArrayList<SalesOperatingStatistics> operatingStatistics = Lists.newArrayListWithCapacity(boList.size());
            if(CollectionUtils.isNotEmpty(insertBatch)) {
                operatingStatistics.addAll(insertBatch);
            }
            if(CollectionUtils.isNotEmpty(updateBatch)) {
                operatingStatistics.addAll(updateBatch);
            }

            // 生成需要新增或更新的 Sales 数据
            List<Sales> salesList = generateUpdateSalesList(Collections.singletonList(propertyDTO), operatingStatistics);
            // 更新 Sales reportSalesTotal 字段
            saveOrUpdateBatch(salesList);

            // Payroll analysis 指标数据（Sales）发生变化时，发送 SNS 事件
            TransactionUtils.afterCommitAsyncExecute(executor, () -> {
                // 数据发生变更发送事件
                sendModuleEvent(operatingStatistics);
            });

        }

        /**
         * 数据发生变更发送事件
         *
         * @param operatingStatistics 数据
         * @author zc
         * @date 2024/8/11
         */
        private void sendModuleEvent(List<SalesOperatingStatistics> operatingStatistics) {
            List<MorSalesSourceEventBO> mappedList = beanMapper.mapList(operatingStatistics, MorSalesSourceEventBO.class, (bo1, bo2) -> {
                bo2.setDate(Long.valueOf(bo1.getTheDate()));
                bo2.setTotalRevenue(bo1.getTotalRev());
            });
            String messages = JsonUtils.convertObjListToString(objectMapper, mappedList);
            moduleEventSender.publish(MoudleEventSubjectEnum.MOR_SALES_SOURCE.getSubject(), messages);
        }

        /**
         * 批量更新 Sales reportSalesTotal 字段
         *
         * @param propertyList        资产
         * @param operatingStatistics Operating 报告数据
         * @author zc
         * @date 2024/6/3
         */
        private List<Sales> generateUpdateSalesList(List<PropertyDTO> propertyList, List<SalesOperatingStatistics> operatingStatistics) {
            if (CollectionUtils.isEmpty(operatingStatistics)) {
                return Collections.emptyList();
            }

            List<Integer> propertyIds = propertyList.stream().map(PropertyDTO::getId).toList();
            // 提取受影响的 theMonth
            List<Integer> theMonthList = operatingStatistics.stream().map(item -> {
                LocalDate localDate = DateTimeUtils.strToLocalDate(item.getTheDate().toString(), BaseConstant.DATE_FORMAT_YYYYMMDD);
                return (int) YearMonthUtils.parseYearMonthToLong(YearMonth.of(localDate.getYear(), localDate.getMonthValue()));
            }).distinct().toList();

            // 提取 theMonth 的最小值、最大值
            Integer minMonth = Collections.min(theMonthList);
            Integer maxMonth = Collections.max(theMonthList);

            // 获取日期区间的 Operating 数据
            int startDate = Integer.parseInt(DateTimeUtils.localDateToStr(YearMonthUtils.getYearMonth(minMonth.toString()).atDay(1),
                    BaseConstant.DATE_FORMAT_YYYYMMDD));
            int endDate = Integer.parseInt(DateTimeUtils.localDateToStr(YearMonthUtils.getYearMonth(maxMonth.toString()).atEndOfMonth(),
                    BaseConstant.DATE_FORMAT_YYYYMMDD));
            List<SalesOperatingStatistics> salesOperatingStatisticsList = salesOperatingStatisticsService.list(new LambdaQueryWrapper<SalesOperatingStatistics>()
                    .between(SalesOperatingStatistics::getTheDate, startDate, endDate)
                    .in(SalesOperatingStatistics::getPropertyId, propertyIds));
            Map<Integer, Map<Integer, BigDecimal>> month2TotalRevByPropertyId = salesOperatingStatisticsList.stream()
                    .filter(item -> Objects.nonNull(item.getTotalRev()))
                    .collect(Collectors.groupingBy(SalesOperatingStatistics::getPropertyId, Collectors.groupingBy(item -> {
                        LocalDate localDate = DateTimeUtils.strToLocalDate(item.getTheDate().toString(), BaseConstant.DATE_FORMAT_YYYYMMDD);
                        return (int) YearMonthUtils.parseYearMonthToLong(YearMonth.of(localDate.getYear(), localDate.getMonthValue()));
                    }, Collectors.reducing(BigDecimal.ZERO, SalesOperatingStatistics::getTotalRev, BigDecimal::add))));

            // 获取日期区间的 Sales 数据
            List<Sales> salesList = list(new LambdaQueryWrapper<Sales>()
                    .in(Sales::getTheMonth, theMonthList)
                    .in(Sales::getPropertyId, propertyIds));
            Map<Integer, Map<Integer, Sales>> month2SalesByPropertyId = salesList.stream()
                    .collect(Collectors.groupingBy(Sales::getPropertyId, Collectors.toMap(Sales::getTheMonth, Function.identity())));

            List<Sales> saveSalesList = Lists.newArrayList();

            for (Map.Entry<Integer, Map<Integer, BigDecimal>> entry : month2TotalRevByPropertyId.entrySet()) {
                Integer propertyId = entry.getKey();
                Map<Integer, Sales> month2Sales = month2SalesByPropertyId.get(propertyId);
                for (Map.Entry<Integer, BigDecimal> month2TotalRev : entry.getValue().entrySet()) {
                    Integer theMonth = month2TotalRev.getKey();
                    BigDecimal totalRevValue = month2TotalRev.getValue();
                    int year = YearMonthUtils.getYearMonth(theMonth.toString()).getYear();
                    if (Objects.isNull(month2Sales)) {
                        saveSalesList.add(Sales.builder()
                                .propertyId(propertyId)
                                .theYear(year)
                                .theMonth(theMonth)
                                .manualSalesTotal(totalRevValue)
                                .reportSalesTotal(totalRevValue)
                                .build());
                    } else {
                        Sales originSales = month2Sales.get(theMonth);
                        if (Objects.isNull(originSales)) {
                            saveSalesList.add(Sales.builder()
                                    .propertyId(propertyId)
                                    .theYear(year)
                                    .theMonth(theMonth)
                                    .manualSalesTotal(totalRevValue)
                                    .reportSalesTotal(totalRevValue)
                                    .build());
                        } else {
                            saveSalesList.add(originSales.setManualSalesTotal(originSales.getUpdatedBy() == 0 ?
                                            totalRevValue : originSales.getManualSalesTotal())
                                    .setReportSalesTotal(totalRevValue));
                        }
                    }
                }
            }
            return saveSalesList;
        }
    }