# OperatingStatisticsServiceImpl.java

    package cloud.wconnect.workperformance.service.sva.operatingstatistics.impl;

    import cloud.wconnect.commonlib.component.BeanMapper;
    import cloud.wconnect.commonlib.cons.BaseConstant;
    import cloud.wconnect.commonlib.cons.m3.DailyOperatingStatisticsIndexEnum;
    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.workperformance.cons.operatingstatisticsdaily.OperatingStatisticsDatasourceEnum;
    import cloud.wconnect.workperformance.cons.sva.SvaDailyOperatingStatisticsIndexEnum;
    import cloud.wconnect.workperformance.entity.sva.operatingstatistics.OperatingStatistics;
    import cloud.wconnect.workperformance.entity.sva.operatingstatistics.SvaOperatingStatisticsOriginal;
    import cloud.wconnect.workperformance.mapper.sva.operatingstatistics.OperatingStatisticsMapper;
    import cloud.wconnect.workperformance.pojo.bo.sva.operatingstatistics.OperatingStatisticsListBO;
    import cloud.wconnect.workperformance.pojo.dto.sva.operatingstatistics.OperatingStatisticsDatasourceDTO;
    import cloud.wconnect.workperformance.service.sva.operatingstatistics.IOperatingStatisticsService;
    import cloud.wconnect.workperformance.service.sva.operatingstatistics.ISvaOperatingStatisticsOriginalService;
    import cloud.wconnect.workperformance.service.sva.operatingstatistics.operatingstatistics.IOperatingStatisticsParseService;
    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 org.apache.commons.collections.CollectionUtils;
    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.util.*;
    import java.util.function.Function;
    import java.util.stream.Collectors;

    /**
     * <p>
     * 服务实现类：运营收入分析
     * </p>
     *
     * @author zc
     * @since 2023-03-17
     */
    @Service
    @DS("work_performance")
    public class OperatingStatisticsServiceImpl extends ServiceImpl<OperatingStatisticsMapper, OperatingStatistics> implements IOperatingStatisticsService {
        @Resource
        private ApplicationContext applicationContext;
        @Resource
        private BeanMapper beanMapper;
        @Resource
        private ObjectMapper objectMapper;
        @Resource
        private ISvaOperatingStatisticsOriginalService svaOperatingStatisticsOriginalService;

        /**
         * 获取 Operating Statistics 集合
         *
         * @param propertyId property id
         * @param fromDate   开始日期
         * @param toDate     结束日期
         * @return Operating Statistics 集合
         * @author zc
         * @date 2023/3/17
         */
        @Override
        public List<OperatingStatistics> list(int propertyId, long fromDate, long toDate) {
            return list(new QueryWrapper<OperatingStatistics>().lambda()
                    .eq(OperatingStatistics::getPropertyId, propertyId)
                    .between(OperatingStatistics::getTheDate, fromDate, toDate));
        }

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

        /**
         * 获取资产指定时间段经营数据
         *
         * @param propertyIds 资产id集合
         * @param startDate   开始时间
         * @param endDate     结束时间
         * @return java.util.List<cloud.wconnect.workperformance.entity.sva.operatingstatistics.OperatingStatistics>
         * @author lichen
         * @date 2025/10/15
         */
        @Override
        public List<OperatingStatistics> listByDateRange(List<Integer> propertyIds, long startDate, long endDate) {
            if (CollectionUtils.isEmpty(propertyIds)) return Collections.emptyList();
            return list(new LambdaQueryWrapper<OperatingStatistics>()
                    .in(OperatingStatistics::getPropertyId, propertyIds)
                    .between(OperatingStatistics::getTheDate, startDate, endDate));
        }

        /**
         * 按优先级更新Operating statistics数据
         *
         * @param propertyDTO 资产
         * @param dates       日期集合
         * @author lichen
         * @date 2025/10/15
         */
        @Override
        @Transactional(rollbackFor = Exception.class)
        public void updateOperatingStatisticsByPriority(PropertyDTO propertyDTO, List<Integer> dates) {
            // 查询指定日期的Operating Statistics Original数据集合
            List<SvaOperatingStatisticsOriginal> operatingStatisticsOriginals = svaOperatingStatisticsOriginalService.listByDates(propertyDTO.getId(), dates);
            if (CollectionUtils.isEmpty(operatingStatisticsOriginals)) return;
            // 按照日期(yyyyMMdd)进行分组
            Map<Integer, List<SvaOperatingStatisticsOriginal>> date2OriginalList = operatingStatisticsOriginals.stream().collect(Collectors.groupingBy(SvaOperatingStatisticsOriginal::getDate));

            // 查询指定日期的Operating Statistics数据集合
            List<Long> theDates = dates.stream().map(date ->
                    DateTimeUtils.dateStrToEpochSecond(date.toString(), BaseConstant.DATE_FORMAT_YYYYMMDD, propertyDTO.getTimezone())).collect(Collectors.toList());
            Map<Long, OperatingStatistics> theDate2FinalEntity = list(new LambdaQueryWrapper<OperatingStatistics>()
                    .eq(OperatingStatistics::getPropertyId, propertyDTO.getId())
                    .in(OperatingStatistics::getTheDate, theDates))
                    .stream().collect(Collectors.toMap(OperatingStatistics::getTheDate, Function.identity()));

            ArrayList<OperatingStatistics> insertBatchList = Lists.newArrayListWithCapacity(dates.size());
            ArrayList<OperatingStatistics> updateBatchList = Lists.newArrayListWithCapacity(dates.size());
            for (Integer date : dates) {
                long theDate = DateTimeUtils.dateStrToEpochSecond(date.toString(), BaseConstant.DATE_FORMAT_YYYYMMDD, propertyDTO.getTimezone());

                // 获取指定日期优先级最高的数据
                OperatingStatistics highestPriorityEntity = getHighestPriority(date2OriginalList.getOrDefault(date, Collections.emptyList()));
                if (Objects.isNull(highestPriorityEntity)) continue;

                // 已经存在的按优先级筛选后的最终数据
                OperatingStatistics existFinalEntity = theDate2FinalEntity.get(theDate);

                if (Objects.isNull(existFinalEntity)) {
                    // 不存在则新增
                    highestPriorityEntity.setPropertyId(propertyDTO.getId());
                    highestPriorityEntity.setId(null);
                    highestPriorityEntity.setTheDate(theDate);
                    // 最终数据为空时，使用优先级最高的数据
                    insertBatchList.add(highestPriorityEntity);
                } else {
                    // 筛选非M3 API的数据
                    List<OperatingStatisticsDatasourceDTO> datasourceDTOS = JsonUtils.convertStringToList(objectMapper,
                            existFinalEntity.getDatasource(), OperatingStatisticsDatasourceDTO.class);
                    // 筛选出M3 API数据源的指标
                    List<OperatingStatisticsDatasourceDTO> m3ApiDatasourceDTOS = datasourceDTOS.stream()
                            .filter(item -> OperatingStatisticsDatasourceEnum.M3_API.getValue() == item.getDatasource())
                            .toList();
                    List<String> m3ApiIndexNameList = m3ApiDatasourceDTOS.stream().map(OperatingStatisticsDatasourceDTO::getIndexName).toList();
                    // 筛选出非M3 API的指标集合
                    List<String> nonM3ApiIndexNameList = Arrays.stream(DailyOperatingStatisticsIndexEnum.values())
                            .map(DailyOperatingStatisticsIndexEnum::getIndexName)
                            .filter(indexName -> !m3ApiIndexNameList.contains(indexName)).toList();
                    if (CollectionUtils.isNotEmpty(nonM3ApiIndexNameList)) {
                        // 存在非M3 API的指标，则更新
                        existFinalEntity.setLastUpdatedBy(highestPriorityEntity.getLastUpdatedBy());
                        existFinalEntity.setLastUpdatedTime(highestPriorityEntity.getLastUpdatedTime());

                        // 优先级最高的数据，数据源集合
                        Map<String, Integer> indexName2HighestPriorityDatasource = JsonUtils.convertStringToList(objectMapper,
                                        highestPriorityEntity.getDatasource(), OperatingStatisticsDatasourceDTO.class)
                                .stream().collect(Collectors.toMap(OperatingStatisticsDatasourceDTO::getIndexName, OperatingStatisticsDatasourceDTO::getDatasource));

                        ArrayList<OperatingStatisticsDatasourceDTO> newDatasourceDTOS = Lists.newArrayList(m3ApiDatasourceDTOS);
                        // 非M3 Api的指标设置新的值
                        for (String indexName : nonM3ApiIndexNameList) {
                            // 指定指标优先级最高的来源
                            Integer datasourceByHighestPriority = indexName2HighestPriorityDatasource.get(indexName);
                            if(Objects.isNull(datasourceByHighestPriority)){
                                continue;
                            }
                            // 获取 getter 方法
                            Function<OperatingStatistics, Object> getter = SvaDailyOperatingStatisticsIndexEnum.getOperatingStatisticsGetterFunction(indexName);
                            // 使用 getter 获取值
                            Object value = getter.apply(highestPriorityEntity);
                            // 指标对应的字段setter方法
                            SvaDailyOperatingStatisticsIndexEnum.OperatingStatisticsSetterBiConsumer setterFunction =
                                    SvaDailyOperatingStatisticsIndexEnum.getOperatingStatisticsSetterFunction(indexName);
                            // 给指标对应的字段赋值
                            setterFunction.accept(existFinalEntity, value);

                            newDatasourceDTOS.add(new OperatingStatisticsDatasourceDTO(indexName, datasourceByHighestPriority));
                        }

                        // 更新数据源
                        existFinalEntity.setDatasource(JsonUtils.convertObjListToString(objectMapper, newDatasourceDTOS));

                        updateBatchList.add(existFinalEntity);
                    }
                }
            }

            saveBatch(insertBatchList);

            // 批量更新,支持指标值设置为null
            if (CollectionUtils.isNotEmpty(updateBatchList)) {
                baseMapper.batchUpdateWithNull(updateBatchList);
            }
        }

        /**
         * 获取指定日期优先级最高数据
         * <p>
         * 优先级规则：
         * 1. M3 API > PDF/CSV
         * 2. 如果不存在M3 API，则取lastUpdateTime最大的数据
         * 3. 如果存在M3 API，则逐字段判断：
         * - M3 API字段非空：使用M3 API的字段值
         * - M3 API字段为空：使用PDF/CSV中lastUpdateTime最大的数据的字段值
         *
         * @param originalList 原数据集合（最多3条：PDF/CSV/M3 API）
         * @return 优先级最高的 OperatingStatistics 对象，如果列表为空则返回 null
         * @author lichen
         * @date 2026/4/14
         */
        private OperatingStatistics getHighestPriority(List<SvaOperatingStatisticsOriginal> originalList) {
            if (CollectionUtils.isEmpty(originalList)) {
                return null;
            }

            // 分离 M3 API 数据和非 M3 API 数据（PDF/CSV）
            List<SvaOperatingStatisticsOriginal> m3ApiList = originalList.stream()
                    .filter(item -> OperatingStatisticsDatasourceEnum.M3_API.getValue() == item.getDataType())
                    .toList();

            List<SvaOperatingStatisticsOriginal> nonM3ApiList = originalList.stream()
                    .filter(item -> OperatingStatisticsDatasourceEnum.M3_API.getValue() != item.getDataType())
                    .toList();

            // 获取非 M3 API 中 lastUpdatedTime 最大的数据
            SvaOperatingStatisticsOriginal latestNonM3Api = nonM3ApiList.stream()
                    .max(Comparator.comparing(SvaOperatingStatisticsOriginal::getLastUpdatedTime,
                            Comparator.nullsLast(Comparator.naturalOrder())))
                    .orElse(null);

            if (CollectionUtils.isEmpty(m3ApiList)) {
                // 情况1：不存在 M3 API 数据，直接使用 lastUpdatedTime 最大的非 M3 API 数据
                if (latestNonM3Api == null) {
                    return null;
                }
                OperatingStatistics highestPriorityEntity = beanMapper.map(latestNonM3Api, OperatingStatistics.class);
                // 设置数据源信息
                List<OperatingStatisticsDatasourceDTO> datasourceDTOS = Arrays.stream(DailyOperatingStatisticsIndexEnum.values())
                        .map(indexEnum -> new OperatingStatisticsDatasourceDTO(indexEnum.getIndexName(), latestNonM3Api.getDataType()))
                        .collect(Collectors.toList());
                highestPriorityEntity.setDatasource(JsonUtils.convertObjListToString(objectMapper, datasourceDTOS));
                return highestPriorityEntity;
            } else {
                // 情况2：存在 M3 API 数据
                SvaOperatingStatisticsOriginal m3ApiData = m3ApiList.get(0);

                OperatingStatistics highestPriorityEntity = new OperatingStatistics();
                highestPriorityEntity.setCreateBy(m3ApiData.getCreateBy());
                highestPriorityEntity.setCreatedTime(m3ApiData.getCreatedTime());
                highestPriorityEntity.setLastUpdatedBy(m3ApiData.getLastUpdatedBy());
                highestPriorityEntity.setLastUpdatedTime(m3ApiData.getLastUpdatedTime());

                ArrayList<OperatingStatisticsDatasourceDTO> datasourceDTOS = Lists.newArrayList();

                Arrays.stream(DailyOperatingStatisticsIndexEnum.values()).forEach(indexEnum -> {
                    // 获取原数据对象 getter 方法
                    Function<SvaOperatingStatisticsOriginal, Object> originalGetterFunction = SvaDailyOperatingStatisticsIndexEnum.getOperatingStatisticsOriginalGetterFunction(indexEnum.getIndexName());
                    // 获取M3 API的字段值
                    Object valueByM3Api = originalGetterFunction.apply(m3ApiData);

                    // 获取 Setter 方法
                    SvaDailyOperatingStatisticsIndexEnum.OperatingStatisticsSetterBiConsumer setterFunction = SvaDailyOperatingStatisticsIndexEnum.getOperatingStatisticsSetterFunction(indexEnum.getIndexName());

                    Integer dataType = null;
                    if(Objects.nonNull(valueByM3Api)){
                        // M3 API字段非空：使用M3 API的字段值
                        setterFunction.accept(highestPriorityEntity, valueByM3Api);
                        dataType = OperatingStatisticsDatasourceEnum.M3_API.getValue();
                    }else{
                        if(Objects.nonNull(latestNonM3Api)){
                            // M3 API字段为空：使用PDF/CSV中lastUpdateTime最大的数据的字段值
                            Object valueByNonM3Api = originalGetterFunction.apply(latestNonM3Api);
                            // 设置指标值
                            setterFunction.accept(highestPriorityEntity, valueByNonM3Api);
                            dataType = latestNonM3Api.getDataType();
                        }
                        // 当M3 API此字段值为null，且PDF/CSV没有指定日期的数据，则不设置此指标值
                    }

                    if(Objects.nonNull(dataType)){
                        OperatingStatisticsDatasourceDTO datasourceDTO = new OperatingStatisticsDatasourceDTO(indexEnum.getIndexName(), dataType);
                        datasourceDTOS.add(datasourceDTO);
                    }
                });

                highestPriorityEntity.setDatasource(JsonUtils.convertObjListToString(objectMapper, datasourceDTOS));
                return highestPriorityEntity;
            }
        }

    }