package com.xzzn.ems.service.impl;

import com.alibaba.fastjson2.JSON;
import com.alibaba.fastjson2.JSONArray;
import com.alibaba.fastjson2.JSONException;
import com.alibaba.fastjson2.JSONObject;
import com.alibaba.fastjson2.TypeReference;
import com.fasterxml.jackson.databind.JsonNode;
import com.fasterxml.jackson.databind.ObjectMapper;
import com.xzzn.common.constant.Constants;
import com.xzzn.common.constant.RedisKeyConstants;
import com.xzzn.common.core.redis.RedisCache;
import com.xzzn.common.enums.AlarmLevelStatus;
import com.xzzn.common.enums.AlarmStatus;
import com.xzzn.common.enums.BranchStatus;
import com.xzzn.common.enums.ChargeStatus;
import com.xzzn.common.enums.CommunicationStatus;
import com.xzzn.common.enums.ControlModeStatus;
import com.xzzn.common.enums.DeviceCategory;
import com.xzzn.common.enums.DeviceRunningStatus;
import com.xzzn.common.enums.DeviceType;
import com.xzzn.common.enums.GridStatus;
import com.xzzn.common.enums.SiteDevice;
import com.xzzn.common.enums.SiteEnum;
import com.xzzn.common.enums.SwitchStatus;
import com.xzzn.common.utils.DateUtils;
import com.xzzn.common.utils.MapUtils;
import com.xzzn.common.utils.StringUtils;
import com.xzzn.ems.domain.*;
import com.xzzn.ems.domain.vo.AmmeterRevenueStatisListVo;
import com.xzzn.ems.domain.vo.EnergyPriceTimeRange;
import com.xzzn.ems.domain.vo.EnergyPriceVo;
import com.xzzn.ems.enums.DeviceMatchTable;
import com.xzzn.ems.mapper.EmsAmmeterDataMapper;
import com.xzzn.ems.mapper.EmsBatteryClusterMapper;
import com.xzzn.ems.mapper.EmsBatteryDataMapper;
import com.xzzn.ems.mapper.EmsBatteryDataMinutesMapper;
import com.xzzn.ems.mapper.EmsBatteryGroupMapper;
import com.xzzn.ems.mapper.EmsBatteryStackMapper;
import com.xzzn.ems.mapper.EmsClusterAlarmDataMapper;
import com.xzzn.ems.mapper.EmsCoolingAlarmDataMapper;
import com.xzzn.ems.mapper.EmsCoolingDataMapper;
import com.xzzn.ems.mapper.EmsDailyChargeDataMapper;
import com.xzzn.ems.mapper.EmsDailyEnergyDataMapper;
import com.xzzn.ems.mapper.EmsDevicesSettingMapper;
import com.xzzn.ems.mapper.EmsDhDataMapper;
import com.xzzn.ems.mapper.EmsEmsDataMapper;
import com.xzzn.ems.mapper.EmsPcsAlarmDataMapper;
import com.xzzn.ems.mapper.EmsPcsBranchDataMapper;
import com.xzzn.ems.mapper.EmsPcsDataMapper;
import com.xzzn.ems.mapper.EmsPointConfigMapper;
import com.xzzn.ems.mapper.EmsSiteMonitorPointMatchMapper;
import com.xzzn.ems.mapper.EmsStackAlarmDataMapper;
import com.xzzn.ems.mapper.EmsXfDataMapper;
import com.xzzn.ems.service.IDeviceDataProcessService;
import com.xzzn.ems.service.IEmsAlarmRecordsService;
import com.xzzn.ems.service.IEmsDeviceSettingService;
import com.xzzn.ems.service.IEmsEnergyPriceConfigService;
import com.xzzn.ems.service.InfluxPointDataWriter;
import com.xzzn.ems.utils.AbstractBatteryDataProcessor;
import com.xzzn.ems.utils.DevicePointMatchDataProcessor;

import java.lang.reflect.Field;
import java.math.BigDecimal;
import java.math.RoundingMode;
import java.time.LocalDate;
import java.time.LocalDateTime;
import java.time.LocalTime;
import java.time.YearMonth;
import java.time.ZoneId;
import java.time.format.DateTimeFormatter;
import java.time.temporal.ChronoUnit;
import java.util.ArrayDeque;
import java.util.ArrayList;
import java.util.Arrays;
import java.util.Date;
import java.util.Deque;
import java.util.HashMap;
import java.util.HashSet;
import java.util.LinkedHashMap;
import java.util.LinkedHashSet;
import java.util.List;
import java.util.Map;
import java.util.Objects;
import java.util.Set;
import java.util.Collections;
import java.util.concurrent.BlockingQueue;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.Executors;
import java.util.concurrent.LinkedBlockingQueue;
import java.util.concurrent.ScheduledExecutorService;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicBoolean;
import java.util.regex.Matcher;
import java.util.regex.Pattern;
import java.util.stream.Collectors;

import javax.annotation.PostConstruct;
import javax.annotation.PreDestroy;

import org.apache.commons.collections4.CollectionUtils;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.BeanUtils;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Service;

@Service
public class DeviceDataProcessServiceImpl extends AbstractBatteryDataProcessor implements IDeviceDataProcessService {
    private static final Logger log = LoggerFactory.getLogger(DeviceDataProcessServiceImpl.class);

    private static final Pattern PATTERN = Pattern.compile("(BMSD\\d{2})(ZT|SOC|SOH|DL|DY|BDSC)");
    // 匹配DTDC+数字格式的正则（提取序号）
    private static final Pattern DTDC_PATTERN = Pattern.compile("DTDC(\\d+)([A-Za-z]*)");
    private static final Pattern DB_NAME_PATTERN = Pattern.compile("^[A-Za-z0-9_]+$");
    private static final Pattern VARIABLE_SAFE_PATTERN = Pattern.compile("[^A-Za-z0-9_]");
    private static final Pattern SIMPLE_CALC_EXPRESSION_PATTERN = Pattern.compile("^[0-9A-Za-z_+\\-*/().,?:<>=!&|\\s]+$");
    private static final Pattern AUTO_DAY_DIFF_FUNCTION_PATTERN = Pattern.compile("(?i)DAY_DIFF\\s*\\(\\s*([A-Za-z_][A-Za-z0-9_]*)\\s*\\)");
    private static final Pattern AUTO_MONTH_DIFF_FUNCTION_PATTERN = Pattern.compile("(?i)MONTH_DIFF\\s*\\(\\s*([A-Za-z_][A-Za-z0-9_]*)\\s*\\)");
    private static final Pattern AUTO_HOUR_DIFF_FUNCTION_PATTERN = Pattern.compile("(?i)HOUR_DIFF\\s*\\(\\s*([A-Za-z_][A-Za-z0-9_]*)\\s*\\)");
    private static final int POINT_QUEUE_CAPACITY = 100000;
    private static final int POINT_FLUSH_BATCH_SIZE = 2000;
    private static final int POINT_FLUSH_MAX_DRAIN_PER_RUN = 20000;
    private static final int POINT_ENQUEUE_RETRY_TIMES = 3;
    private static final long POINT_ENQUEUE_RETRY_WAIT_MS = 10;
    private static final long POINT_FLUSH_INTERVAL_MS = 100;
    private static final String SITE_LEVEL_CALC_DEVICE_ID = "SITE_CALC";
    private static final int MONITOR_POINT_MATCH_REDIS_TTL_SECONDS = 300;
    private static final String DELETED_FIELD_MARK = "__DELETED__";
    private static final String AUTO_DAY_DIFF_CONTEXT_PREFIX = "__AUTO_DAY_DIFF__";
    private static final String AUTO_MONTH_DIFF_CONTEXT_PREFIX = "__AUTO_MONTH_DIFF__";
    private static final String AUTO_HOUR_DIFF_CONTEXT_PREFIX = "__AUTO_HOUR_DIFF__";
    private static final String MONITOR_FIELD_PREFIX_HOME = "home__";
    private static final String MONITOR_FIELD_PREFIX_TJBB_DBBB = "tjbb_dbbb__";
    private static final String FIELD_CODE_TOTAL_CHARGE = "home__totalChargedCap";
    private static final String FIELD_CODE_TOTAL_DISCHARGE = "home__totalDischargedCap";
    private static final String FIELD_CODE_DAY_CHARGE = "home__dayChargedCap";
    private static final String FIELD_CODE_DAY_DISCHARGE = "home__dayDisChargedCap";
    private static final String FIELD_CODE_PEAK_CHARGE = "tjbb_dbbb__activePeakKwh";
    private static final String FIELD_CODE_PEAK_DISCHARGE = "tjbb_dbbb__reActivePeakKwh";
    private static final String FIELD_CODE_HIGH_CHARGE = "tjbb_dbbb__activeHighKwh";
    private static final String FIELD_CODE_HIGH_DISCHARGE = "tjbb_dbbb__reActiveHighKwh";
    private static final String FIELD_CODE_FLAT_CHARGE = "tjbb_dbbb__activeFlatKwh";
    private static final String FIELD_CODE_FLAT_DISCHARGE = "tjbb_dbbb__reActiveFlatKwh";
    private static final String FIELD_CODE_VALLEY_CHARGE = "tjbb_dbbb__activeValleyKwh";
    private static final String FIELD_CODE_VALLEY_DISCHARGE = "tjbb_dbbb__reActiveValleyKwh";
    private static final String[] FIELD_SUFFIX_TOTAL_CHARGE = {
            "activeTotalKwh", "totalChargeData", "totalCharge", "totalChargedCap", "chargedCap"
    };
    private static final String[] FIELD_SUFFIX_TOTAL_DISCHARGE = {
            "reActiveTotalKwh", "totalDischargeData", "totalDischarge", "totalDischargedCap", "disChargedCap"
    };
    private static final String[] FIELD_SUFFIX_DAY_CHARGE = {
            "dayChargedCap", "dayChargeData", "dailyChargeData", "dailyChargedCap"
    };
    private static final String[] FIELD_SUFFIX_DAY_DISCHARGE = {
            "dayDisChargedCap", "dayDischargeData", "dailyDischargeData", "dailyDisChargedCap"
    };
    private static final String[] FIELD_SUFFIX_PEAK_CHARGE = {"activePeakKwh", "peakChargeDiff", "peakCharge"};
    private static final String[] FIELD_SUFFIX_PEAK_DISCHARGE = {"reActivePeakKwh", "peakDischargeDiff", "peakDischarge"};
    private static final String[] FIELD_SUFFIX_HIGH_CHARGE = {"activeHighKwh", "highChargeDiff", "highCharge"};
    private static final String[] FIELD_SUFFIX_HIGH_DISCHARGE = {"reActiveHighKwh", "highDischargeDiff", "highDischarge"};
    private static final String[] FIELD_SUFFIX_FLAT_CHARGE = {"activeFlatKwh", "flatChargeDiff", "flatCharge"};
    private static final String[] FIELD_SUFFIX_FLAT_DISCHARGE = {"reActiveFlatKwh", "flatDischargeDiff", "flatDischarge"};
    private static final String[] FIELD_SUFFIX_VALLEY_CHARGE = {"activeValleyKwh", "valleyChargeDiff", "valleyCharge"};
    private static final String[] FIELD_SUFFIX_VALLEY_DISCHARGE = {"reActiveValleyKwh", "valleyDischargeDiff", "valleyDischarge"};
    private static final long DAILY_CHARGE_INFLUX_QUERY_WINDOW_MS = TimeUnit.DAYS.toMillis(2);
    private static final long CALC_POINT_INFLUX_QUERY_WINDOW_MS = TimeUnit.DAYS.toMillis(2);

    @Autowired
    private EmsBatteryClusterMapper emsBatteryClusterMapper;
    @Autowired
    private EmsBatteryStackMapper emsBatteryStackMapper;
    @Autowired
    private EmsBatteryDataMapper emsBatteryDataMapper;
    @Autowired
    private EmsPcsDataMapper emsPcsDataMapper;
    @Autowired
    private EmsPointConfigMapper emsPointConfigMapper;
    @Autowired
    private EmsPcsBranchDataMapper emsPcsBranchDataMapper;
    @Autowired
    private EmsAmmeterDataMapper emsAmmeterDataMapper;
    @Autowired
    private EmsDevicesSettingMapper emsDevicesSettingMapper;
    @Autowired
    private EmsBatteryDataMinutesMapper emsBatteryDataMinutesMapper;
    @Autowired
    private EmsDailyChargeDataMapper emsDailyChargeDataMapper;
    @Autowired
    private EmsXfDataMapper emsXfDataMapper;
    @Autowired
    private EmsDhDataMapper emsDhDataMapper;
    @Autowired
    private EmsCoolingDataMapper emsCoolingDataMapper;
    @Autowired
    private EmsDailyEnergyDataMapper emsDailyEnergyDataMapper;
    @Autowired
    private EmsBatteryGroupMapper emsBatteryGroupMapper;
    @Autowired
    private EmsEmsDataMapper emsEmsDataMapper;
    @Autowired
    private EmsSiteMonitorPointMatchMapper emsSiteMonitorPointMatchMapper;

    @Autowired
    private EmsCoolingAlarmDataMapper emsCoolingAlarmDataMapper;
    @Autowired
    private EmsStackAlarmDataMapper emsStackAlarmDataMapper;
    @Autowired
    private EmsClusterAlarmDataMapper emsClusterAlarmDataMapper;
    @Autowired
    private EmsPcsAlarmDataMapper emsPcsAlarmDataMapper;

    @Autowired
    private EmsBatteryDailyLatestServiceImpl emsBatteryDailyLatestServiceImpl;
    @Autowired
    private IEmsAlarmRecordsService iEmsAlarmRecordsService;
    @Autowired
    private IEmsEnergyPriceConfigService emsEnergyPriceConfigService;
    @Autowired
    private IEmsDeviceSettingService iEmsDeviceSettingService;
    @Autowired
    private DevicePointMatchDataProcessor devicePointMatchDataProcessor;

    @Autowired
    private RedisCache redisCache;
    @Autowired
    private InfluxPointDataWriter influxPointDataWriter;
    private final BlockingQueue<PointDataRecord> pointDataQueue = new LinkedBlockingQueue<>(POINT_QUEUE_CAPACITY);
    private final Map<String, CompiledExpression> calcExpressionCache = new ConcurrentHashMap<>();
    private final ScheduledExecutorService pointDataWriter = Executors.newSingleThreadScheduledExecutor(r -> {
        Thread thread = new Thread(r);
        thread.setName("point-data-writer");
        thread.setDaemon(true);
        return thread;
    });
    private final AtomicBoolean pointDataFlushing = new AtomicBoolean(false);

    // 构造方法（调用父类构造）
    public DeviceDataProcessServiceImpl(ObjectMapper objectMapper) {
        super(objectMapper);
    }

    @PostConstruct
    public void initPointDataBatchWriter() {
        pointDataWriter.scheduleWithFixedDelay(this::flushPointDataSafely,
                POINT_FLUSH_INTERVAL_MS, POINT_FLUSH_INTERVAL_MS, TimeUnit.MILLISECONDS);
    }

    @PreDestroy
    public void destroyPointDataBatchWriter() {
        flushPointDataSafely();
        pointDataWriter.shutdown();
    }

    @Override
    public void handleDeviceData(String message, String siteId) {
        JSONArray arraylist = parseJsonData(message);
        if (arraylist == null) return;
        long startMs = System.currentTimeMillis();
        log.info("开始处理设备数据, siteId: {}, messageSize: {}", siteId, arraylist.size());
        Set<String> deviceIds = new LinkedHashSet<>();
        Map<String, String> deviceCategoryMap = buildDeviceCategoryMap(siteId, arraylist);
        for (int i = 0; i < arraylist.size(); i++) {
            JSONObject obj = JSONObject.parseObject(arraylist.get(i).toString());

            String deviceId = obj.getString("Device");
            String jsonData = resolveJsonDataWithRedisFallback(siteId, deviceId, obj.getString("Data"));
            Long timestamp = obj.getLong("timestamp");
            Date dataUpdateTime = DateUtils.convertUpdateTime(timestamp);

            if (StringUtils.isNotBlank(deviceId)) {
                deviceIds.add(deviceId);
            }
            if (checkJsonDataEmpty(jsonData)) {
                continue;
            }

            try {
                String mergedJsonData = mergeWithLatestRedisData(siteId, deviceId, jsonData);
                obj.put("Data", mergedJsonData);
                // 保留每个设备最新一条原始报文，供“点位最新值”从 Redis 读取
                redisCache.setCacheObject(RedisKeyConstants.ORIGINAL_MQTT_DATA + siteId + "_" + deviceId, obj);
                redisCache.setCacheObject(RedisKeyConstants.SYNC_DATA + siteId + "_" + deviceId, obj, 1, TimeUnit.MINUTES);

                String deviceCategory = resolveDeviceCategory(siteId, deviceId, deviceCategoryMap);
                // 旧设备表落库链路已下线，MQTT 数据仅走点位映射与站点监控数据同步。
                Map<String, BigDecimal> pointIdValueMap = processPointConfigData(siteId, deviceId, deviceCategory, mergedJsonData, dataUpdateTime);
                List<String> pointIdSamples = pointIdValueMap.keySet().stream().limit(10).collect(Collectors.toList());
                log.info("MQTT点位映射完成，siteId: {}, deviceId: {}, deviceCategory: {}, pointCount: {}, pointIdSamples: {}",
                        siteId, deviceId, deviceCategory, pointIdValueMap.size(), pointIdSamples);
                cacheLatestPointValuesByPointId(siteId, deviceId, pointIdValueMap, dataUpdateTime);
                iEmsDeviceSettingService.syncSiteMonitorDataByMqtt(siteId, deviceId, JSON.toJSONString(pointIdValueMap), dataUpdateTime);
            } catch (Exception e) {
                log.warn("设备数据处理失败，siteId: {}, deviceId: {}, err: {}",
                        siteId, deviceId, e.getMessage(), e);
            }
        }
        log.info("结束处理设备数据, siteId: {}, deviceCount: {}, deviceIds: {}, costMs: {}",
                siteId, deviceIds.size(), String.join(",", deviceIds), System.currentTimeMillis() - startMs);
    }

    private Map<String, BigDecimal> processPointConfigData(String siteId, String deviceId, String deviceCategory, String jsonData, Date dataUpdateTime) {
        Map<String, BigDecimal> pointIdValueMap = new HashMap<>();
        if (StringUtils.isAnyBlank(siteId, deviceId, jsonData)) {
            return pointIdValueMap;
        }
        if (!DB_NAME_PATTERN.matcher(siteId).matches()) {
            log.warn("站点ID不合法，跳过点位映射落库，siteId: {}", siteId);
            return pointIdValueMap;
        }

        Map<String, Object> dataMap = JSON.parseObject(jsonData, new TypeReference<Map<String, Object>>() {
        });
        if (org.apache.commons.collections4.MapUtils.isEmpty(dataMap)) {
            return pointIdValueMap;
        }

        List<EmsPointConfig> pointConfigs = getPointConfigsWithCache(siteId, deviceId, deviceCategory);
        if (CollectionUtils.isEmpty(pointConfigs)) {
            return pointIdValueMap;
        }

        List<EmsPointConfig> dataPointConfigs = new ArrayList<>();
        Set<String> uniquePointKeys = new HashSet<>();
        for (EmsPointConfig pointConfig : pointConfigs) {
            String pointContextKey = resolvePointContextKey(pointConfig);
            if (pointConfig == null || StringUtils.isBlank(pointContextKey)) {
                continue;
            }
            if (!uniquePointKeys.add(pointContextKey)) {
                continue;
            }
            if (isComputedPoint(pointConfig)) {
                continue;
            }
            dataPointConfigs.add(pointConfig);
        }
        Map<String, BigDecimal> rawDataValues = new HashMap<>();
        for (Map.Entry<String, Object> entry : dataMap.entrySet()) {
            if (StringUtils.isBlank(entry.getKey())) {
                continue;
            }
            BigDecimal pointValue = StringUtils.getBigDecimal(entry.getValue());
            if (pointValue == null) {
                continue;
            }
            rawDataValues.put(entry.getKey().trim().toUpperCase(), pointValue);
        }

        for (EmsPointConfig dataPointConfig : dataPointConfigs) {
            BigDecimal sourceValue = getSourceValueFromRawData(dataPointConfig, rawDataValues);
            if (sourceValue == null) {
                continue;
            }
            BigDecimal convertedValue = convertPointValue(sourceValue, dataPointConfig);
            String pointId = resolveInfluxPointKey(dataPointConfig);
            if (StringUtils.isBlank(pointId)) {
                continue;
            }
            enqueuePointData(siteId, deviceId, pointId, convertedValue, dataUpdateTime);
            pointIdValueMap.put(pointId, convertedValue);
        }
        return pointIdValueMap;
    }

    private void updateDailyChargeDataByQuartz(String siteId,
                                               Map<String, BigDecimal> pointIdValueMap,
                                               Date dataUpdateTime) {
        if (StringUtils.isBlank(siteId)
                || org.apache.commons.collections4.MapUtils.isEmpty(pointIdValueMap)) {
            return;
        }

        List<EmsSiteMonitorPointMatch> mappingList = getPointMatchesBySiteId(siteId);
        if (CollectionUtils.isEmpty(mappingList)) {
            return;
        }

        Map<String, BigDecimal> normalizedPointValueMap = new HashMap<>();
        for (Map.Entry<String, BigDecimal> entry : pointIdValueMap.entrySet()) {
            if (StringUtils.isBlank(entry.getKey()) || entry.getValue() == null) {
                continue;
            }
            normalizedPointValueMap.put(entry.getKey().trim().toUpperCase(), entry.getValue());
        }
        if (normalizedPointValueMap.isEmpty()) {
            return;
        }

        EmsSiteMonitorPointMatch totalChargeMatch = firstNonBlankMatchByFieldSuffixIgnoreDevice(
                mappingList, FIELD_SUFFIX_TOTAL_CHARGE
        );
        EmsSiteMonitorPointMatch totalDischargeMatch = firstNonBlankMatchByFieldSuffixIgnoreDevice(
                mappingList, FIELD_SUFFIX_TOTAL_DISCHARGE
        );
        EmsSiteMonitorPointMatch dayChargeMatch = firstNonBlankMatchByFieldSuffixIgnoreDevice(
                mappingList, FIELD_SUFFIX_DAY_CHARGE
        );
        EmsSiteMonitorPointMatch dayDischargeMatch = firstNonBlankMatchByFieldSuffixIgnoreDevice(
                mappingList, FIELD_SUFFIX_DAY_DISCHARGE
        );
        String totalChargePointId = totalChargeMatch == null ? null : StringUtils.trimToNull(totalChargeMatch.getDataPoint());
        String totalDischargePointId = totalDischargeMatch == null ? null : StringUtils.trimToNull(totalDischargeMatch.getDataPoint());
        String dayChargePointId = dayChargeMatch == null ? null : StringUtils.trimToNull(dayChargeMatch.getDataPoint());
        String dayDischargePointId = dayDischargeMatch == null ? null : StringUtils.trimToNull(dayDischargeMatch.getDataPoint());
        log.info("日充放电自检, siteId: {}, totalCharge: {}->{} , totalDischarge: {}->{} , dayCharge: {}->{} , dayDischarge: {}->{}",
                siteId,
                totalChargeMatch == null ? "null" : StringUtils.defaultString(totalChargeMatch.getFieldCode()),
                totalChargePointId,
                totalDischargeMatch == null ? "null" : StringUtils.defaultString(totalDischargeMatch.getFieldCode()),
                totalDischargePointId,
                dayChargeMatch == null ? "null" : StringUtils.defaultString(dayChargeMatch.getFieldCode()),
                dayChargePointId,
                dayDischargeMatch == null ? "null" : StringUtils.defaultString(dayDischargeMatch.getFieldCode()),
                dayDischargePointId);
        BigDecimal nowTotalCharge = getPointValueByIdRaw(normalizedPointValueMap, totalChargePointId);
        BigDecimal nowTotalDischarge = getPointValueByIdRaw(normalizedPointValueMap, totalDischargePointId);
        BigDecimal nowDayCharge = getLatestPointValueFromRedis(dayChargePointId);
        BigDecimal nowDayDischarge = getLatestPointValueFromRedis(dayDischargePointId);
        if (nowTotalCharge == null || nowTotalDischarge == null) {
            log.info("日充放电累计未匹配到有效点位值, siteId: {}, totalChargePointId: {}, totalDischargePointId: {}, pointKeys: {}",
                    siteId, totalChargePointId, totalDischargePointId, normalizedPointValueMap.keySet());
            return;
        }

        EmsDailyChargeData dailyChargeData = new EmsDailyChargeData();
        Date recordDateTime = dataUpdateTime == null ? DateUtils.getNowDate() : dataUpdateTime;
        dailyChargeData.setSiteId(siteId);
        dailyChargeData.setDateTime(recordDateTime);
        dailyChargeData.setTotalChargeData(nowTotalCharge);
        dailyChargeData.setTotalDischargeData(nowTotalDischarge);
        dailyChargeData.setChargeData(nowDayCharge);
        dailyChargeData.setDischargeData(nowDayDischarge);

        EmsDailyChargeData existedDailyData = emsDailyChargeDataMapper.selectBySiteIdAndDateTime(siteId, recordDateTime);
        if (existedDailyData != null && existedDailyData.getId() != null) {
            dailyChargeData.setId(existedDailyData.getId());
            dailyChargeData.setDateTime(existedDailyData.getDateTime());
        }

        Date now = DateUtils.getNowDate();
        dailyChargeData.setCreateBy("system");
        dailyChargeData.setCreateTime(now);
        dailyChargeData.setUpdateBy("system");
        dailyChargeData.setUpdateTime(now);
        emsDailyChargeDataMapper.insertOrUpdateData(dailyChargeData);

        log.info("日充放电累计落库完成, siteId: {}, total[充:{} 放:{}], daily[充:{} 放:{}], points[总充:{} 总放:{} 日充:{} 日放:{}]",
                siteId,
                dailyChargeData.getTotalChargeData(), dailyChargeData.getTotalDischargeData(),
                dailyChargeData.getChargeData(), dailyChargeData.getDischargeData(),
                totalChargePointId, totalDischargePointId, dayChargePointId, dayDischargePointId);
    }

    public void syncDailyChargeDataFromInfluxByQuartz() {
        syncDailyChargeDataFromInfluxByQuartz(null);
    }

    public void syncDailyChargeDataFromInfluxByQuartz(String siteId) {
        List<String> siteIds;
        if (StringUtils.isNotBlank(siteId)) {
            siteIds = Collections.singletonList(siteId.trim());
        } else {
            siteIds = emsSiteMonitorPointMatchMapper.selectDistinctSiteIds();
        }
        if (CollectionUtils.isEmpty(siteIds)) {
            log.info("Quartz同步日充放电跳过，未找到站点列表, siteIdParam: {}", siteId);
            return;
        }
        for (String currentSiteId : siteIds) {
            if (StringUtils.isBlank(currentSiteId)) {
                continue;
            }
            try {
                syncSingleSiteDailyChargeFromInflux(currentSiteId.trim());
            } catch (Exception ex) {
                log.error("Quartz同步日充放电失败, siteId: {}, err: {}", currentSiteId, ex.getMessage(), ex);
            }
        }
    }

    private void syncSingleSiteDailyChargeFromInflux(String siteId) {
        List<EmsSiteMonitorPointMatch> mappingList = getPointMatchesBySiteId(siteId);
        if (CollectionUtils.isEmpty(mappingList)) {
            log.info("Quartz同步日充放电跳过，未找到点位映射, siteId: {}", siteId);
            return;
        }

        Date queryEndTime = DateUtils.getNowDate();
        Date queryStartTime = new Date(queryEndTime.getTime() - DAILY_CHARGE_INFLUX_QUERY_WINDOW_MS);
        String totalChargePointId = resolvePointIdByFieldCodeIgnoreDevice(mappingList, FIELD_CODE_TOTAL_CHARGE);
        String totalDischargePointId = resolvePointIdByFieldCodeIgnoreDevice(mappingList, FIELD_CODE_TOTAL_DISCHARGE);
        String dayChargePointId = resolvePointIdByFieldCodeIgnoreDevice(mappingList, FIELD_CODE_DAY_CHARGE);
        String dayDischargePointId = resolvePointIdByFieldCodeIgnoreDevice(mappingList, FIELD_CODE_DAY_DISCHARGE);
        if (StringUtils.isBlank(dayChargePointId)) {
            dayChargePointId = firstNonBlankPointByFieldSuffixIgnoreDevice(mappingList, FIELD_SUFFIX_DAY_CHARGE);
        }
        if (StringUtils.isBlank(dayDischargePointId)) {
            dayDischargePointId = firstNonBlankPointByFieldSuffixIgnoreDevice(mappingList, FIELD_SUFFIX_DAY_DISCHARGE);
        }
        log.info("Quartz同步日充放电日累自检, siteId: {}, dayChargePoint: {}, dayDischargePoint: {}",
                siteId,
                dayChargePointId,
                dayDischargePointId);
        String peakChargePointId = resolvePointIdByFieldCodeIgnoreDevice(mappingList, FIELD_CODE_PEAK_CHARGE);
        String peakDischargePointId = resolvePointIdByFieldCodeIgnoreDevice(mappingList, FIELD_CODE_PEAK_DISCHARGE);
        String highChargePointId = resolvePointIdByFieldCodeIgnoreDevice(mappingList, FIELD_CODE_HIGH_CHARGE);
        String highDischargePointId = resolvePointIdByFieldCodeIgnoreDevice(mappingList, FIELD_CODE_HIGH_DISCHARGE);
        String flatChargePointId = resolvePointIdByFieldCodeIgnoreDevice(mappingList, FIELD_CODE_FLAT_CHARGE);
        String flatDischargePointId = resolvePointIdByFieldCodeIgnoreDevice(mappingList, FIELD_CODE_FLAT_DISCHARGE);
        String valleyChargePointId = resolvePointIdByFieldCodeIgnoreDevice(mappingList, FIELD_CODE_VALLEY_CHARGE);
        String valleyDischargePointId = resolvePointIdByFieldCodeIgnoreDevice(mappingList, FIELD_CODE_VALLEY_DISCHARGE);

        log.info("Quartz同步日充放电自检, siteId: {}, totalCharge: {} , totalDischarge: {}, dayCharge: {}, dayDischarge: {}, peak: {} / {}, high: {} / {}, flat: {} / {}, valley: {} / {}",
                siteId,
                totalChargePointId,
                totalDischargePointId,
                dayChargePointId,
                dayDischargePointId,
                peakChargePointId,
                peakDischargePointId,
                highChargePointId,
                highDischargePointId,
                flatChargePointId,
                flatDischargePointId,
                valleyChargePointId,
                valleyDischargePointId);

        Map<String, LatestPointValue> latestPointValues = new HashMap<>();
        Date latestDataTime = null;
        for (String pointId : Arrays.asList(
                totalChargePointId, totalDischargePointId,
                dayChargePointId, dayDischargePointId,
                peakChargePointId, peakDischargePointId,
                highChargePointId, highDischargePointId,
                flatChargePointId, flatDischargePointId,
                valleyChargePointId, valleyDischargePointId)) {
            if (StringUtils.isBlank(pointId)) {
                continue;
            }
            LatestPointValue latestPointValue = getLatestPointValuePreferRedis(siteId, pointId, queryStartTime, queryEndTime);
            if (latestPointValue == null || latestPointValue.getValue() == null) {
                continue;
            }
            latestPointValues.put(pointId.trim().toUpperCase(), latestPointValue);
            if (latestPointValue.getDataTime() != null
                    && (latestDataTime == null || latestPointValue.getDataTime().after(latestDataTime))) {
                latestDataTime = latestPointValue.getDataTime();
            }
        }

        if (latestPointValues.isEmpty()) {
            log.info("Quartz同步日充放电跳过，点位最新值为空, siteId: {}", siteId);
            return;
        }

        Date recordDateTime = latestDataTime == null ? DateUtils.getNowDate() : latestDataTime;
        BigDecimal totalChargeValue = getPointValueById(latestPointValues, totalChargePointId);
        BigDecimal totalDischargeValue = getPointValueById(latestPointValues, totalDischargePointId);
        BigDecimal dayChargeValue = getPointValueById(latestPointValues, dayChargePointId);
        BigDecimal dayDischargeValue = getPointValueById(latestPointValues, dayDischargePointId);
        log.info("单站点位值汇总, siteId: {}, totalCharge: {}({}), totalDischarge: {}({}), dayCharge: {}({}), dayDischarge: {}({}), dataTime: {}",
                siteId,
                totalChargeValue,
                totalDischargeValue,
                dayChargeValue,
                dayDischargeValue,
                pointValueSource(latestPointValues, totalChargePointId),
                pointValueSource(latestPointValues, totalDischargePointId),
                pointValueSource(latestPointValues, dayChargePointId),
                pointValueSource(latestPointValues, dayDischargePointId),
                DateUtils.parseDateToStr(DateUtils.YYYY_MM_DD_HH_MM_SS, recordDateTime));
        if (totalChargeValue == null && totalDischargeValue == null) {
            log.info("Quartz同步日充放电跳过，总充/总放值为空, siteId: {}", siteId);
        } else {
            upsertDailyChargeData(siteId, recordDateTime, totalChargeValue, totalDischargeValue, dayChargeValue, dayDischargeValue);
        }

        updateDailyEnergyHourlyDataByQuartz(
                siteId,
                recordDateTime,
                peakChargePointId,
                peakDischargePointId,
                highChargePointId,
                highDischargePointId,
                flatChargePointId,
                flatDischargePointId,
                valleyChargePointId,
                valleyDischargePointId,
                queryStartTime,
                queryEndTime
        );
        syncDailyChargeRevenueByQuartz(siteId, recordDateTime);
    }

    private void syncDailyChargeRevenueByQuartz(String siteId, Date dataUpdateTime) {
        if (StringUtils.isBlank(siteId)) {
            return;
        }
        Date revenueDate = dataUpdateTime == null ? DateUtils.getNowDate() : dataUpdateTime;
        String day = DateUtils.parseDateToStr(DateUtils.YYYY_MM_DD, revenueDate);
        List<AmmeterRevenueStatisListVo> revenueRows = emsDailyEnergyDataMapper.getRevenueDataBySiteId(siteId, day, day);

        BigDecimal dayRevenue = BigDecimal.ZERO;
        if (CollectionUtils.isNotEmpty(revenueRows)) {
            AmmeterRevenueStatisListVo revenueRow = revenueRows.get(0);
            BigDecimal activeTotalPrice = safeBigDecimal(revenueRow.getActivePeakPrice())
                    .add(safeBigDecimal(revenueRow.getActiveHighPrice()))
                    .add(safeBigDecimal(revenueRow.getActiveFlatPrice()))
                    .add(safeBigDecimal(revenueRow.getActiveValleyPrice()));
            BigDecimal reActiveTotalPrice = safeBigDecimal(revenueRow.getReActivePeakPrice())
                    .add(safeBigDecimal(revenueRow.getReActiveHighPrice()))
                    .add(safeBigDecimal(revenueRow.getReActiveFlatPrice()))
                    .add(safeBigDecimal(revenueRow.getReActiveValleyPrice()));
            dayRevenue = reActiveTotalPrice.subtract(activeTotalPrice);
        }
        BigDecimal totalRevenue = getYestLastData(siteId).add(dayRevenue);

        syncDailyRevenueToChargeData(siteId, revenueDate, totalRevenue, dayRevenue);

        log.info("Quartz同步日收益到充放电表完成, siteId: {}, date: {}, dayRevenue: {}, totalRevenue: {}",
                siteId, day, dayRevenue, totalRevenue);
    }

    private void updateDailyEnergyHourlyDataByQuartz(String siteId,
                                                     Date dataUpdateTime,
                                                     String peakChargePointId,
                                                     String peakDischargePointId,
                                                     String highChargePointId,
                                                     String highDischargePointId,
                                                     String flatChargePointId,
                                                     String flatDischargePointId,
                                                     String valleyChargePointId,
                                                     String valleyDischargePointId,
                                                     Date queryStartTime,
                                                     Date queryEndTime) {
        log.info("Quartz日电量小时落库入口, siteId: {}, dataTime: {}",
                siteId,
                dataUpdateTime == null ? "null" : DateUtils.parseDateToStr(DateUtils.YYYY_MM_DD_HH_MM_SS, dataUpdateTime));
        if (StringUtils.isBlank(siteId)) {
            return;
        }

        Date recordDateTime = dataUpdateTime == null ? DateUtils.getNowDate() : dataUpdateTime;
        Integer dataHour = resolveHourOfDay(recordDateTime);

        LocalDateTime recordLocalTime = LocalDateTime.ofInstant(recordDateTime.toInstant(), ZoneId.systemDefault());
        String priceKey = RedisKeyConstants.ENERGY_PRICE_TIME + siteId + "_" + recordLocalTime.getYear()
                + recordLocalTime.getMonthValue();
        EnergyPriceVo priceVo = redisCache.getCacheObject(priceKey);
        if (priceVo == null) {
            priceVo = emsEnergyPriceConfigService.getCurrentMonthPrice(siteId);
            redisCache.setCacheObject(priceKey, priceVo, 31, TimeUnit.DAYS);
        }
        if (priceVo == null || CollectionUtils.isEmpty(priceVo.getRange())) {
            log.info("Quartz日电量小时落库跳过，未找到有效电价时段配置, siteId: {}", siteId);
            return;
        }

        String costType = null;
        for (EnergyPriceTimeRange timeRange : priceVo.getRange()) {
            if (timeRange == null || StringUtils.isBlank(timeRange.getCostType())) {
                continue;
            }
            if (isInPriceTimeRange(timeRange.getStartTime(), timeRange.getEndTime(), recordDateTime)) {
                costType = timeRange.getCostType();
                break;
            }
        }
        if (StringUtils.isBlank(costType)) {
            log.info("Quartz日电量小时落库跳过，未命中电价时段, siteId: {}, dataTime: {}",
                    siteId, DateUtils.parseDateToStr(DateUtils.YYYY_MM_DD_HH_MM_SS, recordDateTime));
            return;
        }

        log.info("Quartz日电量小时自检, siteId: {}, peak: {} / {} , high: {} / {} , flat: {} / {} , valley: {} / {}",
                siteId,
                peakChargePointId,
                peakDischargePointId,
                highChargePointId,
                highDischargePointId,
                flatChargePointId,
                flatDischargePointId,
                valleyChargePointId,
                valleyDischargePointId);
        String currentChargePointId;
        String currentDischargePointId;
        BigDecimal currentCharge;
        BigDecimal currentDischarge;
        switch (costType) {
            case "peak":
                currentChargePointId = peakChargePointId;
                currentDischargePointId = peakDischargePointId;
                currentCharge = getLatestPointValuePreferRedisValue(siteId, peakChargePointId, queryStartTime, queryEndTime);
                currentDischarge = getLatestPointValuePreferRedisValue(siteId, peakDischargePointId, queryStartTime, queryEndTime);
                break;
            case "high":
                currentChargePointId = highChargePointId;
                currentDischargePointId = highDischargePointId;
                currentCharge = getLatestPointValuePreferRedisValue(siteId, highChargePointId, queryStartTime, queryEndTime);
                currentDischarge = getLatestPointValuePreferRedisValue(siteId, highDischargePointId, queryStartTime, queryEndTime);
                break;
            case "flat":
                currentChargePointId = flatChargePointId;
                currentDischargePointId = flatDischargePointId;
                currentCharge = getLatestPointValuePreferRedisValue(siteId, flatChargePointId, queryStartTime, queryEndTime);
                currentDischarge = getLatestPointValuePreferRedisValue(siteId, flatDischargePointId, queryStartTime, queryEndTime);
                break;
            case "valley":
                currentChargePointId = valleyChargePointId;
                currentDischargePointId = valleyDischargePointId;
                currentCharge = getLatestPointValuePreferRedisValue(siteId, valleyChargePointId, queryStartTime, queryEndTime);
                currentDischarge = getLatestPointValuePreferRedisValue(siteId, valleyDischargePointId, queryStartTime, queryEndTime);
                break;
            default:
                log.info("Quartz日电量小时落库跳过，未知电价时段类型, siteId: {}, costType: {}", siteId, costType);
                return;
        }

        if (currentCharge == null && currentDischarge == null) {
            log.info("Quartz日电量小时值未匹配到有效Redis点位值, siteId: {}, costType: {}, chargePoint: {}, dischargePoint: {}",
                    siteId, costType, currentChargePointId, currentDischargePointId);
            return;
        }

        EmsDailyEnergyData energyData = new EmsDailyEnergyData();
        energyData.setSiteId(siteId);
        energyData.setDataDate(recordDateTime);
        energyData.setDataHour(dataHour);
        if ("peak".equals(costType)) {
            energyData.setPeakChargeDiff(currentCharge);
            energyData.setPeakDischargeDiff(currentDischarge);
        } else if ("high".equals(costType)) {
            energyData.setHighChargeDiff(currentCharge);
            energyData.setHighDischargeDiff(currentDischarge);
        } else if ("flat".equals(costType)) {
            energyData.setFlatChargeDiff(currentCharge);
            energyData.setFlatDischargeDiff(currentDischarge);
        } else if ("valley".equals(costType)) {
            energyData.setValleyChargeDiff(currentCharge);
            energyData.setValleyDischargeDiff(currentDischarge);
        }
        energyData.setCalcTime(recordDateTime);

        EmsDailyEnergyData existedEnergyData = emsDailyEnergyDataMapper.selectBySiteIdAndDateHour(siteId, recordDateTime, dataHour);
        if (existedEnergyData != null && existedEnergyData.getId() != null) {
            energyData.setId(existedEnergyData.getId());
            energyData.setDataDate(existedEnergyData.getDataDate());
            energyData.setDataHour(existedEnergyData.getDataHour());
        }

        Date now = DateUtils.getNowDate();
        energyData.setCreateBy("system");
        energyData.setCreateTime(now);
        energyData.setUpdateBy("system");
        energyData.setUpdateTime(now);
        emsDailyEnergyDataMapper.insertOrUpdateData(energyData);

        log.info("Quartz日电量小时落库完成, siteId: {}, hour: {}, costType: {}, charge: {}, discharge: {}, chargePoint: {}, dischargePoint: {}",
                siteId, dataHour, costType, currentCharge, currentDischarge, currentChargePointId, currentDischargePointId);
    }

    private boolean isInPriceTimeRange(String startTime, String endTime, Date dataUpdateTime) {
        if (StringUtils.isAnyBlank(startTime, endTime) || dataUpdateTime == null) {
            return false;
        }
        try {
            String normalizedStartTime = startTime.trim();
            String normalizedEndTime = endTime.trim();
            LocalTime current = LocalTime.parse(DateUtils.parseDateToStr("HH:mm", dataUpdateTime));
            LocalTime start = LocalTime.parse(normalizedStartTime);

            // 兼容配置中的 24:00，将其视为次日 00:00。
            if ("24:00".equals(normalizedEndTime) || "24:00:00".equals(normalizedEndTime)) {
                return !current.isBefore(start);
            }

            LocalTime end = LocalTime.parse(normalizedEndTime);
            if (start.equals(end)) {
                return true;
            }
            if (start.isBefore(end)) {
                return !current.isBefore(start) && current.isBefore(end);
            }
            return !current.isBefore(start) || current.isBefore(end);
        } catch (Exception ex) {
            log.warn("解析电价时段失败, start: {}, end: {}, err: {}", startTime, endTime, ex.getMessage());
            return false;
        }
    }

    private Integer resolveHourOfDay(Date dateTime) {
        Date targetTime = dateTime == null ? DateUtils.getNowDate() : dateTime;
        String hourText = DateUtils.parseDateToStr("HH", targetTime);
        try {
            return Integer.parseInt(hourText);
        } catch (NumberFormatException ex) {
            return LocalDateTime.now().getHour();
        }
    }

    public void syncCalcPointDataFromInfluxByQuartz() {
        syncCalcPointDataFromInfluxByQuartz(null);
    }

    public void syncCalcPointDataFromInfluxByQuartz(String siteId) {
        List<String> siteIds;
        if (StringUtils.isNotBlank(siteId)) {
            siteIds = Collections.singletonList(siteId.trim());
        } else {
            siteIds = emsSiteMonitorPointMatchMapper.selectDistinctSiteIds();
        }
        if (CollectionUtils.isEmpty(siteIds)) {
            return;
        }
        for (String currentSiteId : siteIds) {
            if (StringUtils.isBlank(currentSiteId)) {
                continue;
            }
            try {
                syncSingleSiteCalcPointFromInflux(currentSiteId.trim());
            } catch (Exception ex) {
                log.error("Quartz同步计算点失败, siteId: {}, err: {}", currentSiteId, ex.getMessage(), ex);
            }
        }
    }

    private void syncSingleSiteCalcPointFromInflux(String siteId) {
        List<EmsPointConfig> calcPointConfigs = getSiteCalcPointConfigs(siteId);
        if (CollectionUtils.isEmpty(calcPointConfigs)) {
            return;
        }
        List<EmsPointConfig> sourcePointConfigs = getSiteSourcePointConfigs(siteId);
        if (CollectionUtils.isEmpty(sourcePointConfigs)) {
            log.info("Quartz同步计算点跳过，未找到源点位配置, siteId: {}", siteId);
            return;
        }

        Date queryEndTime = DateUtils.getNowDate();
        Date queryStartTime = new Date(queryEndTime.getTime() - CALC_POINT_INFLUX_QUERY_WINDOW_MS);
        Map<String, BigDecimal> contextValues = new HashMap<>();
        Set<String> reservedCalcContextKeys = calcPointConfigs.stream()
                .map(this::resolveCalcContextKey)
                .filter(StringUtils::isNotBlank)
                .map(String::trim)
                .map(String::toUpperCase)
                .collect(Collectors.toSet());
        Date latestDataTime = null;
        for (EmsPointConfig sourcePointConfig : sourcePointConfigs) {
            if (sourcePointConfig == null) {
                continue;
            }
            String pointId = resolveInfluxPointKey(sourcePointConfig);
            if (StringUtils.isBlank(pointId)) {
                continue;
            }
            InfluxPointDataWriter.PointValue latestPointValue = influxPointDataWriter.queryLatestPointValueByPointKey(
                    siteId, pointId, queryStartTime, queryEndTime
            );
            if (latestPointValue == null || latestPointValue.getPointValue() == null) {
                continue;
            }
            putSourcePointValueToContext(sourcePointConfig, latestPointValue.getPointValue(), contextValues, reservedCalcContextKeys);
            if (latestDataTime == null
                    || (latestPointValue.getDataTime() != null && latestPointValue.getDataTime().after(latestDataTime))) {
                latestDataTime = latestPointValue.getDataTime();
            }
        }

        if (org.apache.commons.collections4.MapUtils.isEmpty(contextValues)) {
            log.info("Quartz同步计算点跳过，InfluxDB未查询到源点位最新值, siteId: {}", siteId);
            return;
        }
        Date calcTime = latestDataTime == null ? queryEndTime : latestDataTime;
        Map<String, BigDecimal> calcPointIdValueMap = executeCalcPointConfigs(
                siteId, SITE_LEVEL_CALC_DEVICE_ID, calcPointConfigs, contextValues, calcTime
        );
        if (!calcPointIdValueMap.isEmpty()) {
            cacheLatestPointValuesByPointId(siteId, SITE_LEVEL_CALC_DEVICE_ID, calcPointIdValueMap, calcTime);
            iEmsDeviceSettingService.syncSiteMonitorDataByMqtt(
                    siteId, SITE_LEVEL_CALC_DEVICE_ID, JSON.toJSONString(calcPointIdValueMap), calcTime
            );
        }
    }

    private String firstNonBlankPointByFieldSuffixIgnoreDevice(List<EmsSiteMonitorPointMatch> mappingList,
                                                               String[] fieldSuffixes,
                                                               String... fieldPrefixes) {
        EmsSiteMonitorPointMatch match = firstNonBlankMatchByFieldSuffixIgnoreDevice(
                mappingList, fieldSuffixes, fieldPrefixes
        );
        return match == null ? null : StringUtils.trimToNull(match.getDataPoint());
    }

    private EmsSiteMonitorPointMatch firstNonBlankMatchByFieldSuffixIgnoreDevice(List<EmsSiteMonitorPointMatch> mappingList,
                                                                                 String[] fieldSuffixes,
                                                                                 String... fieldPrefixes) {
        if (CollectionUtils.isEmpty(mappingList) || fieldSuffixes == null || fieldSuffixes.length == 0) {
            return null;
        }
        List<String> normalizedPrefixes = new ArrayList<>();
        if (fieldPrefixes != null && fieldPrefixes.length > 0) {
            for (String prefix : fieldPrefixes) {
                String normalizedPrefix = StringUtils.trimToNull(prefix);
                if (normalizedPrefix != null) {
                    normalizedPrefixes.add(normalizedPrefix.toLowerCase());
                }
            }
        }
        for (String suffix : fieldSuffixes) {
            if (StringUtils.isBlank(suffix)) {
                continue;
            }
            String normalizedSuffix = suffix.trim().toLowerCase();
            for (EmsSiteMonitorPointMatch mapping : mappingList) {
                if (mapping == null || StringUtils.isBlank(mapping.getFieldCode())) {
                    continue;
                }
                String dataPoint = StringUtils.defaultString(mapping.getDataPoint()).trim();
                if (StringUtils.isBlank(dataPoint) || DELETED_FIELD_MARK.equalsIgnoreCase(dataPoint)) {
                    continue;
                }
                String fieldCode = mapping.getFieldCode().trim().toLowerCase();
                if (!normalizedPrefixes.isEmpty()) {
                    if (fieldCode.equals(normalizedSuffix)) {
                        return mapping;
                    }
                    boolean matchedPrefix = false;
                    for (String prefix : normalizedPrefixes) {
                        if (fieldCode.startsWith(prefix)) {
                            matchedPrefix = true;
                            String stripped = fieldCode.substring(prefix.length());
                            if (stripped.equals(normalizedSuffix)) {
                                return mapping;
                            }
                        }
                    }
                    if (matchedPrefix) {
                        continue;
                    }
                } else if (fieldCode.equals(normalizedSuffix)) {
                    return mapping;
                }
            }
        }
        return null;
    }

    private String mergeWithLatestRedisData(String siteId, String deviceId, String currentJsonData) {
        Map<String, Object> currentDataMap = parseDataJsonToMap(currentJsonData);
        if (StringUtils.isAnyBlank(siteId, deviceId)) {
            return JSON.toJSONString(currentDataMap);
        }
        String redisKey = RedisKeyConstants.ORIGINAL_MQTT_DATA + siteId + "_" + deviceId;
        Object latestObj = redisCache.getCacheObject(redisKey);
        Map<String, Object> latestDataMap = extractDataMapFromLatest(latestObj);
        if (latestDataMap.isEmpty()) {
            return JSON.toJSONString(currentDataMap);
        }
        Map<String, Object> merged = new LinkedHashMap<>(latestDataMap);
        currentDataMap.forEach((key, value) -> {
            if (StringUtils.isBlank(key) || value == null) {
                return;
            }
            merged.put(key, value);
        });
        return JSON.toJSONString(merged);
    }

    private String resolveJsonDataWithRedisFallback(String siteId, String deviceId, String jsonData) {
        if (!checkJsonDataEmpty(jsonData)) {
            return jsonData;
        }
        if (StringUtils.isAnyBlank(siteId, deviceId)) {
            return jsonData;
        }
        String redisKey = RedisKeyConstants.ORIGINAL_MQTT_DATA + siteId + "_" + deviceId;
        Object latestObj = redisCache.getCacheObject(redisKey);
        Map<String, Object> latestDataMap = extractDataMapFromLatest(latestObj);
        if (latestDataMap.isEmpty()) {
            return jsonData;
        }
        return JSON.toJSONString(latestDataMap);
    }

    private Map<String, Object> extractDataMapFromLatest(Object latestObj) {
        if (latestObj == null) {
            return new LinkedHashMap<>();
        }
        JSONObject latestJson;
        if (latestObj instanceof JSONObject) {
            latestJson = (JSONObject) latestObj;
        } else if (latestObj instanceof String) {
            latestJson = JSON.parseObject((String) latestObj);
        } else if (latestObj instanceof Map) {
            latestJson = new JSONObject((Map<String, Object>) latestObj);
        } else {
            latestJson = JSON.parseObject(JSON.toJSONString(latestObj));
        }
        if (latestJson == null || latestJson.isEmpty()) {
            return new LinkedHashMap<>();
        }
        Object dataObj = latestJson.get("Data");
        if (dataObj == null) {
            return new LinkedHashMap<>();
        }
        if (dataObj instanceof JSONObject) {
            return new LinkedHashMap<>(((JSONObject) dataObj));
        }
        if (dataObj instanceof Map) {
            return new LinkedHashMap<>((Map<String, Object>) dataObj);
        }
        if (dataObj instanceof String) {
            return parseDataJsonToMap((String) dataObj);
        }
        return parseDataJsonToMap(JSON.toJSONString(dataObj));
    }

    private Map<String, Object> parseDataJsonToMap(String jsonData) {
        if (StringUtils.isBlank(jsonData)) {
            return new LinkedHashMap<>();
        }
        Map<String, Object> parsed = JSON.parseObject(jsonData, new TypeReference<Map<String, Object>>() {
        });
        return parsed == null ? new LinkedHashMap<>() : new LinkedHashMap<>(parsed);
    }

    private BigDecimal getPointValueById(Map<String, LatestPointValue> pointIdValueMap, String pointId) {
        if (pointIdValueMap == null || pointIdValueMap.isEmpty() || StringUtils.isBlank(pointId)) {
            return null;
        }
        LatestPointValue latestPointValue = pointIdValueMap.get(pointId.trim().toUpperCase());
        return latestPointValue == null ? null : latestPointValue.getValue();
    }

    private BigDecimal getPointValueByIdRaw(Map<String, BigDecimal> pointIdValueMap, String pointId) {
        if (pointIdValueMap == null || pointIdValueMap.isEmpty() || StringUtils.isBlank(pointId)) {
            return null;
        }
        return pointIdValueMap.get(pointId.trim().toUpperCase());
    }

    private String pointValueSource(Map<String, LatestPointValue> pointIdValueMap, String pointId) {
        if (pointIdValueMap == null || pointIdValueMap.isEmpty() || StringUtils.isBlank(pointId)) {
            return "-";
        }
        LatestPointValue latestPointValue = pointIdValueMap.get(pointId.trim().toUpperCase());
        if (latestPointValue == null) {
            return "-";
        }
        return latestPointValue.isFromRedis() ? "Redis" : "Influx";
    }

    private BigDecimal getLatestPointValueFromRedis(String pointId) {
        String normalizedPointId = StringUtils.trimToNull(pointId);
        if (normalizedPointId == null) {
            return null;
        }
        log.info("点位值自检-Redis读取, pointId: {}", normalizedPointId);
        List<String> candidates = new ArrayList<>();
        candidates.add(normalizedPointId);
        String upperPointId = normalizedPointId.toUpperCase();
        if (!StringUtils.equals(upperPointId, normalizedPointId)) {
            candidates.add(upperPointId);
        }
        String lowerPointId = normalizedPointId.toLowerCase();
        if (!StringUtils.equals(lowerPointId, normalizedPointId)
                && !StringUtils.equals(lowerPointId, upperPointId)) {
            candidates.add(lowerPointId);
        }
        for (String candidate : candidates) {
            Object raw = redisCache.getCacheObject(candidate);
            if (raw == null) {
                continue;
            }
            try {
                JSONObject valueObj = raw instanceof JSONObject
                        ? (JSONObject) raw
                        : JSON.parseObject(JSON.toJSONString(raw));
                if (valueObj == null) {
                    continue;
                }
                BigDecimal pointValue = StringUtils.getBigDecimal(valueObj.get("pointValue"));
                if (pointValue == null) {
                    pointValue = readBigDecimalField(raw, "pointValue");
                }
                if (pointValue != null) {
                    log.info("点位值自检-Redis命中, pointId: {}, redisKey: {}", normalizedPointId, candidate);
                    return pointValue;
                }
            } catch (Exception ex) {
                log.warn("读取点位Redis值失败, pointId: {}, redisKey: {}, err: {}", normalizedPointId, candidate, ex.getMessage());
            }
        }
        return null;
    }

    private LatestPointValue getLatestPointValuePreferRedis(String siteId, String pointId, Date queryStartTime, Date queryEndTime) {
        if (StringUtils.isBlank(siteId) || StringUtils.isBlank(pointId)) {
            return null;
        }
        BigDecimal redisValue = getLatestPointValueFromRedis(pointId);
        if (redisValue != null) {
            return new LatestPointValue(redisValue, null, true);
        }
        log.info("点位值自检-Influx读取, siteId: {}, pointId: {}, range: {} ~ {}",
                siteId, pointId,
                DateUtils.parseDateToStr(DateUtils.YYYY_MM_DD_HH_MM_SS, queryStartTime),
                DateUtils.parseDateToStr(DateUtils.YYYY_MM_DD_HH_MM_SS, queryEndTime));
        InfluxPointDataWriter.PointValue influxPointValue = influxPointDataWriter.queryLatestPointValueByPointKey(
                siteId, pointId, queryStartTime, queryEndTime
        );
        if (influxPointValue == null || influxPointValue.getPointValue() == null) {
            return null;
        }
        return new LatestPointValue(influxPointValue.getPointValue(), influxPointValue.getDataTime(), false);
    }

    private BigDecimal getLatestPointValuePreferRedisValue(String siteId, String pointId, Date queryStartTime, Date queryEndTime) {
        LatestPointValue value = getLatestPointValuePreferRedis(siteId, pointId, queryStartTime, queryEndTime);
        return value == null ? null : value.getValue();
    }

    private String resolvePointIdByFieldCodeIgnoreDevice(List<EmsSiteMonitorPointMatch> mappingList, String fieldCode) {
        if (CollectionUtils.isEmpty(mappingList) || StringUtils.isBlank(fieldCode)) {
            return null;
        }
        String normalized = fieldCode.trim().toLowerCase();
        for (EmsSiteMonitorPointMatch mapping : mappingList) {
            if (mapping == null || StringUtils.isBlank(mapping.getFieldCode())) {
                continue;
            }
            String dataPoint = StringUtils.defaultString(mapping.getDataPoint()).trim();
            if (StringUtils.isBlank(dataPoint) || DELETED_FIELD_MARK.equalsIgnoreCase(dataPoint)) {
                continue;
            }
            if (mapping.getFieldCode().trim().toLowerCase().equals(normalized)) {
                return dataPoint;
            }
        }
        return null;
    }

    private void upsertDailyChargeData(String siteId,
                                       Date recordDateTime,
                                       BigDecimal totalChargeValue,
                                       BigDecimal totalDischargeValue,
                                       BigDecimal dayChargeValue,
                                       BigDecimal dayDischargeValue) {
        if (StringUtils.isBlank(siteId)) {
            return;
        }
        Date persistTime = recordDateTime == null ? DateUtils.getNowDate() : recordDateTime;
        EmsDailyChargeData dailyChargeData = new EmsDailyChargeData();
        dailyChargeData.setSiteId(siteId);
        dailyChargeData.setDateTime(persistTime);
        dailyChargeData.setTotalChargeData(totalChargeValue);
        dailyChargeData.setTotalDischargeData(totalDischargeValue);
        dailyChargeData.setChargeData(dayChargeValue);
        dailyChargeData.setDischargeData(dayDischargeValue);

        EmsDailyChargeData existedDailyData = emsDailyChargeDataMapper.selectBySiteIdAndDateTime(siteId, persistTime);
        if (existedDailyData != null && existedDailyData.getId() != null) {
            dailyChargeData.setId(existedDailyData.getId());
            dailyChargeData.setDateTime(existedDailyData.getDateTime());
        }

        Date now = DateUtils.getNowDate();
        dailyChargeData.setCreateBy("system");
        dailyChargeData.setCreateTime(now);
        dailyChargeData.setUpdateBy("system");
        dailyChargeData.setUpdateTime(now);
        emsDailyChargeDataMapper.insertOrUpdateData(dailyChargeData);
        log.info("日充放电累计落库完成, siteId: {}, totalCharge: {}, totalDischarge: {}, dayCharge: {}, dayDischarge: {}",
                siteId, totalChargeValue, totalDischargeValue, dayChargeValue, dayDischargeValue);
    }

    private static class LatestPointValue {
        private final BigDecimal value;
        private final Date dataTime;
        private final boolean fromRedis;

        LatestPointValue(BigDecimal value, Date dataTime, boolean fromRedis) {
            this.value = value;
            this.dataTime = dataTime;
            this.fromRedis = fromRedis;
        }

        public BigDecimal getValue() {
            return value;
        }

        public Date getDataTime() {
            return dataTime;
        }

        public boolean isFromRedis() {
            return fromRedis;
        }
    }

    private BigDecimal readBigDecimalField(Object source, String fieldName) {
        Object fieldValue = readFieldValue(source, fieldName);
        return StringUtils.getBigDecimal(fieldValue);
    }

    private Object readFieldValue(Object source, String fieldName) {
        if (source == null || StringUtils.isBlank(fieldName)) {
            return null;
        }
        Class<?> current = source.getClass();
        while (current != null) {
            try {
                Field field = current.getDeclaredField(fieldName);
                field.setAccessible(true);
                return field.get(source);
            } catch (NoSuchFieldException ex) {
                current = current.getSuperclass();
            } catch (Exception ex) {
                return null;
            }
        }
        return null;
    }

    private String firstNonBlankPointByFieldSuffix(Map<String, EmsSiteMonitorPointMatch> mappingByFieldAndDevice,
                                                   String deviceId,
                                                   String[] fieldSuffixes) {
        return firstNonBlankPointByFieldSuffix(mappingByFieldAndDevice, deviceId, fieldSuffixes, MONITOR_FIELD_PREFIX_HOME);
    }

    private String firstNonBlankPointByFieldSuffix(Map<String, EmsSiteMonitorPointMatch> mappingByFieldAndDevice,
                                                   String deviceId,
                                                   String[] fieldSuffixes,
                                                   String... fieldPrefixes) {
        if (mappingByFieldAndDevice == null || fieldSuffixes == null || fieldSuffixes.length == 0) {
            return null;
        }
        String normalizedDeviceId = StringUtils.defaultString(deviceId).trim();
        for (String suffix : fieldSuffixes) {
            if (StringUtils.isBlank(suffix)) {
                continue;
            }
            String normalizedSuffix = suffix.trim().toLowerCase();
            EmsSiteMonitorPointMatch exactMatch = resolvePointMatchByFieldSuffix(
                    mappingByFieldAndDevice, normalizedDeviceId, normalizedSuffix, fieldPrefixes
            );
            if (exactMatch != null && StringUtils.isNotBlank(exactMatch.getDataPoint())) {
                return exactMatch.getDataPoint().trim();
            }
            EmsSiteMonitorPointMatch fallbackMatch = resolvePointMatchByFieldSuffix(
                    mappingByFieldAndDevice, "", normalizedSuffix, fieldPrefixes
            );
            if (fallbackMatch != null && StringUtils.isNotBlank(fallbackMatch.getDataPoint())) {
                return fallbackMatch.getDataPoint().trim();
            }
        }
        return null;
    }

    private EmsSiteMonitorPointMatch resolvePointMatchByFieldSuffix(Map<String, EmsSiteMonitorPointMatch> mappingByFieldAndDevice,
                                                                    String deviceId,
                                                                    String suffixLowerCase,
                                                                    String... fieldPrefixes) {
        if (mappingByFieldAndDevice == null || StringUtils.isBlank(suffixLowerCase)) {
            return null;
        }
        List<String> normalizedPrefixes = new ArrayList<>();
        if (fieldPrefixes != null && fieldPrefixes.length > 0) {
            for (String prefix : fieldPrefixes) {
                String normalizedPrefix = StringUtils.trimToNull(prefix);
                if (normalizedPrefix != null) {
                    normalizedPrefixes.add(normalizedPrefix.toLowerCase());
                }
            }
        }
        for (Map.Entry<String, EmsSiteMonitorPointMatch> entry : mappingByFieldAndDevice.entrySet()) {
            EmsSiteMonitorPointMatch mapping = entry.getValue();
            if (mapping == null || StringUtils.isBlank(mapping.getFieldCode())) {
                continue;
            }
            String mappingDeviceId = StringUtils.defaultString(mapping.getDeviceId()).trim();
            if (!StringUtils.equals(mappingDeviceId, StringUtils.defaultString(deviceId).trim())) {
                continue;
            }
            String dataPoint = StringUtils.defaultString(mapping.getDataPoint()).trim();
            if (StringUtils.isBlank(dataPoint) || DELETED_FIELD_MARK.equalsIgnoreCase(dataPoint)) {
                continue;
            }
            String fieldCode = mapping.getFieldCode().trim().toLowerCase();
            if (!normalizedPrefixes.isEmpty() && normalizedPrefixes.stream().noneMatch(fieldCode::startsWith)) {
                continue;
            }
            if (fieldCode.equals(suffixLowerCase)) {
                return mapping;
            }
        }
        return null;
    }

    private Map<String, EmsSiteMonitorPointMatch> getMonitorPointMappingByFieldAndDevice(String siteId) {
        List<EmsSiteMonitorPointMatch> mappingList = getPointMatchesBySiteId(siteId);
        if (CollectionUtils.isEmpty(mappingList)) {
            return Collections.emptyMap();
        }
        return mappingList.stream()
                .filter(item -> item != null && StringUtils.isNotBlank(item.getFieldCode()))
                .collect(Collectors.toMap(
                        item -> buildFieldDeviceKey(item.getFieldCode(), item.getDeviceId()),
                        item -> item,
                        (a, b) -> b
                ));
    }

    private List<EmsSiteMonitorPointMatch> getPointMatchesBySiteId(String siteId) {
        if (StringUtils.isBlank(siteId)) {
            return Collections.emptyList();
        }
        String normalizedSiteId = siteId.trim();
        String redisKey = RedisKeyConstants.SITE_MONITOR_POINT_MATCH + normalizedSiteId;
        log.info("单站映射自检-读取, siteId: {}, redisKey: {}", normalizedSiteId, redisKey);
        Object cacheObj = redisCache.getCacheObject(redisKey);
        List<EmsSiteMonitorPointMatch> cached = parsePointMatchCache(cacheObj);
        if (cached != null) {
            log.info("单站映射自检-命中缓存, siteId: {}, size: {}", normalizedSiteId, cached.size());
            return cached;
        }
        List<EmsSiteMonitorPointMatch> latest = emsSiteMonitorPointMatchMapper.selectBySiteId(normalizedSiteId);
        if (latest == null) {
            latest = Collections.emptyList();
        }
        redisCache.setCacheObject(redisKey, latest, MONITOR_POINT_MATCH_REDIS_TTL_SECONDS, TimeUnit.SECONDS);
        log.info("单站映射自检-落库后写缓存, siteId: {}, size: {}", normalizedSiteId, latest.size());
        return latest;
    }

    private List<EmsSiteMonitorPointMatch> parsePointMatchCache(Object cacheObj) {
        if (cacheObj == null) {
            return null;
        }
        try {
            if (cacheObj instanceof String) {
                return JSON.parseArray((String) cacheObj, EmsSiteMonitorPointMatch.class);
            }
            if (cacheObj instanceof List) {
                List<?> cacheList = (List<?>) cacheObj;
                if (cacheList.isEmpty()) {
                    return Collections.emptyList();
                }
                if (cacheList.get(0) instanceof EmsSiteMonitorPointMatch) {
                    return (List<EmsSiteMonitorPointMatch>) cacheList;
                }
                return JSON.parseArray(JSON.toJSONString(cacheObj), EmsSiteMonitorPointMatch.class);
            }
            return JSON.parseArray(JSON.toJSONString(cacheObj), EmsSiteMonitorPointMatch.class);
        } catch (Exception ex) {
            log.warn("解析单站监控点位映射缓存失败，key类型={}, err={}",
                    cacheObj.getClass().getName(), ex.getMessage());
            return null;
        }
    }

    private String buildFieldDeviceKey(String fieldCode, String deviceId) {
        return StringUtils.defaultString(fieldCode).trim() + "|" + StringUtils.defaultString(deviceId).trim();
    }

    private BigDecimal safeBigDecimal(BigDecimal value) {
        return value == null ? BigDecimal.ZERO : value;
    }

    private Map<String, String> buildDeviceCategoryMap(String siteId, JSONArray arraylist) {
        Map<String, String> deviceCategoryMap = new HashMap<>();
        if (arraylist == null || arraylist.isEmpty()) {
            return deviceCategoryMap;
        }
        for (int i = 0; i < arraylist.size(); i++) {
            JSONObject item = JSONObject.parseObject(arraylist.get(i).toString());
            if (item == null) {
                continue;
            }
            String deviceId = item.getString("Device");
            if (StringUtils.isBlank(deviceId) || deviceCategoryMap.containsKey(deviceId)) {
                continue;
            }
            String category = getDeviceCategory(siteId, deviceId);
            if (StringUtils.isNotBlank(category)) {
                deviceCategoryMap.put(deviceId, category);
            }
        }
        return deviceCategoryMap;
    }

    private String resolveDeviceCategory(String siteId, String deviceId, Map<String, String> deviceCategoryMap) {
        if (StringUtils.isBlank(deviceId)) {
            return "";
        }
        String mappedCategory = deviceCategoryMap == null ? null : deviceCategoryMap.get(deviceId);
        if (StringUtils.isNotBlank(mappedCategory)) {
            return mappedCategory;
        }
        String runtimeCategory = getDeviceCategory(siteId, deviceId);
        if (StringUtils.isNotBlank(runtimeCategory) && deviceCategoryMap != null) {
            deviceCategoryMap.put(deviceId, runtimeCategory);
        }
        return StringUtils.defaultString(runtimeCategory);
    }

    private Map<String, BigDecimal> buildBatchContextBySite(String siteId, JSONArray arraylist, Map<String, String> deviceCategoryMap) {
        Map<String, BigDecimal> contextBySite = new HashMap<>();
        if (arraylist == null || arraylist.isEmpty()) {
            return contextBySite;
        }
        for (int i = 0; i < arraylist.size(); i++) {
            JSONObject item = JSONObject.parseObject(arraylist.get(i).toString());
            if (item == null) {
                continue;
            }
            String deviceId = item.getString("Device");
            String devicePrefix = normalizeVariablePart(deviceId);
            String deviceCategory = resolveDeviceCategory(siteId, deviceId, deviceCategoryMap);
            String jsonData = resolveJsonDataWithRedisFallback(siteId, deviceId, item.getString("Data"));
            if (checkJsonDataEmpty(jsonData)) {
                continue;
            }
            Map<String, Object> dataMap = JSON.parseObject(jsonData, new TypeReference<Map<String, Object>>() {
            });
            if (org.apache.commons.collections4.MapUtils.isEmpty(dataMap)) {
                continue;
            }
            Map<String, EmsPointConfig> pointConfigByDataKey = new HashMap<>();
            List<EmsPointConfig> pointConfigs = getPointConfigsWithCache(siteId, deviceId, deviceCategory);
            if (!CollectionUtils.isEmpty(pointConfigs)) {
                for (EmsPointConfig pointConfig : pointConfigs) {
                    if (isComputedPoint(pointConfig)) {
                        continue;
                    }
                    String dataKey = StringUtils.defaultString(pointConfig.getDataKey()).trim().toUpperCase();
                    if (StringUtils.isBlank(dataKey) || pointConfigByDataKey.containsKey(dataKey)) {
                        continue;
                    }
                    pointConfigByDataKey.put(dataKey, pointConfig);
                }
            }
            for (Map.Entry<String, Object> entry : dataMap.entrySet()) {
                if (StringUtils.isBlank(entry.getKey())) {
                    continue;
                }
                BigDecimal pointValue = StringUtils.getBigDecimal(entry.getValue());
                if (pointValue == null) {
                    continue;
                }
                String normalizedKey = normalizeVariablePart(entry.getKey());
                if (StringUtils.isBlank(normalizedKey)) {
                    continue;
                }
                // 兼容旧表达式：继续支持直接使用变量名（如 WD4）
                contextBySite.put(normalizedKey, pointValue);
                // 新规则：支持设备前缀变量（如 DONGHUAN_WD4）
                if (StringUtils.isNotBlank(devicePrefix)) {
                    contextBySite.put(devicePrefix + "_" + normalizedKey, pointValue);
                }
                // 同时支持按点位ID引用（如 POINT4092）
                EmsPointConfig pointConfig = pointConfigByDataKey.get(entry.getKey().trim().toUpperCase());
                if (pointConfig != null) {
                    String pointIdKey = resolvePointContextKey(pointConfig);
                    if (StringUtils.isNotBlank(pointIdKey)) {
                        BigDecimal convertedValue = convertPointValue(pointValue, pointConfig);
                        contextBySite.put(pointIdKey, convertedValue);
                    }
                }
            }
        }
        return contextBySite;
    }

    private String normalizeVariablePart(String source) {
        if (StringUtils.isBlank(source)) {
            return null;
        }
        String normalized = VARIABLE_SAFE_PATTERN.matcher(source.trim().toUpperCase()).replaceAll("_");
        if (StringUtils.isBlank(normalized)) {
            return null;
        }
        normalized = normalized.replaceAll("_+", "_");
        normalized = StringUtils.strip(normalized, "_");
        return StringUtils.isBlank(normalized) ? null : normalized;
    }

    private boolean isComputedPoint(EmsPointConfig pointConfig) {
        if (pointConfig == null) {
            return false;
        }
        String pointType = StringUtils.defaultString(pointConfig.getPointType()).trim().toLowerCase();
        return "calc".equals(pointType);
    }

    private void enqueuePointData(String siteId, String deviceId, String pointKey, BigDecimal pointValue, Date dataTime) {
        if (StringUtils.isAnyBlank(siteId, deviceId, pointKey) || pointValue == null) {
            return;
        }
        PointDataRecord record = new PointDataRecord(siteId, deviceId, pointKey, pointValue, dataTime);
        if (pointDataQueue.offer(record)) {
            if (pointDataQueue.remainingCapacity() <= POINT_FLUSH_BATCH_SIZE) {
                flushPointDataSafely();
            }
            return;
        }

        // 队列压力较高时，优先触发冲刷并短暂重试，避免 MQTT 处理线程直接阻塞在数据库写入上。
        for (int i = 0; i < POINT_ENQUEUE_RETRY_TIMES; i++) {
            flushPointDataSafely();
            try {
                if (pointDataQueue.offer(record, POINT_ENQUEUE_RETRY_WAIT_MS, TimeUnit.MILLISECONDS)) {
                    return;
                }
            } catch (InterruptedException interruptedException) {
                Thread.currentThread().interrupt();
                break;
            }
        }

        log.warn("点位写入队列持续拥塞，降级同步写入 InfluxDB，siteId: {}, queueSize: {}",
                siteId, pointDataQueue.size());
        writePointDataToInflux(Collections.singletonList(record));
    }

    private BigDecimal convertPointValue(BigDecimal sourceValue, EmsPointConfig pointConfig) {
        if (sourceValue == null || pointConfig == null) {
            return sourceValue;
        }
        BigDecimal a = pointConfig.getDataA() == null ? BigDecimal.ZERO : pointConfig.getDataA();
        BigDecimal k = pointConfig.getDataK() == null ? BigDecimal.ONE : pointConfig.getDataK();
        BigDecimal b = pointConfig.getDataB() == null ? BigDecimal.ZERO : pointConfig.getDataB();
        return a.multiply(sourceValue).multiply(sourceValue)
                .add(k.multiply(sourceValue))
                .add(b);
    }

    private String resolveInfluxPointKey(EmsPointConfig pointConfig) {
        if (pointConfig == null) {
            return null;
        }
        if (StringUtils.isNotBlank(pointConfig.getPointId())) {
            return pointConfig.getPointId().trim();
        }
        return null;
    }

    private Map<String, BigDecimal> executeCalcPointConfigs(String siteId, String deviceId, List<EmsPointConfig> calcPointConfigs,
                                                            Map<String, BigDecimal> contextValues, Date dataUpdateTime) {
        Map<String, BigDecimal> calcPointIdValueMap = new HashMap<>();
        if (CollectionUtils.isEmpty(calcPointConfigs)) {
            return calcPointIdValueMap;
        }
        Map<String, EmsPointConfig> remaining = new HashMap<>();
        for (EmsPointConfig calcPointConfig : calcPointConfigs) {
            String calcContextKey = resolveCalcContextKey(calcPointConfig);
            if (calcPointConfig == null || StringUtils.isBlank(calcContextKey)) {
                continue;
            }
            remaining.putIfAbsent(calcContextKey, calcPointConfig);
        }
        if (remaining.isEmpty()) {
            return calcPointIdValueMap;
        }

        while (!remaining.isEmpty()) {
            boolean progressed = false;
            List<String> finishedKeys = new ArrayList<>();
            List<String> unresolvedKeys = new ArrayList<>();

            for (Map.Entry<String, EmsPointConfig> entry : remaining.entrySet()) {
                String calcDataKey = entry.getKey();
                EmsPointConfig calcPointConfig = entry.getValue();
                try {
                    ExpressionValue calcValue = evaluateCalcExpression(
                            calcPointConfig.getCalcExpression(), contextValues, siteId, dataUpdateTime);
                    if (calcValue.isNumber()) {
                        contextValues.put(calcDataKey, calcValue.asNumber());
                        putPointValueToContext(calcPointConfig, calcValue.asNumber(), contextValues);
                    }
                    // 计算点按站点维度统一落库，不再按配置中的 device_id 分流
                    String pointId = resolveInfluxPointKey(calcPointConfig);
                    if (StringUtils.isNotBlank(pointId)) {
                        BigDecimal storedValue = calcValue.asNumber();
                        if (calcPointConfig.getDecimalScale() != null && calcPointConfig.getDecimalScale() >= 0) {
                            storedValue = storedValue.setScale(calcPointConfig.getDecimalScale(), RoundingMode.HALF_UP);
                        }
                        enqueuePointData(siteId, deviceId, pointId, storedValue, dataUpdateTime);
                        calcPointIdValueMap.put(pointId, storedValue);
                    }
                    finishedKeys.add(calcDataKey);
                    progressed = true;
                } catch (MissingVariableException missingVariableException) {
                    unresolvedKeys.add(calcDataKey + " <- " + missingVariableException.getVariable());
                } catch (IllegalArgumentException expressionException) {
                    log.warn("计算点表达式执行失败，siteId: {}, deviceId: {}, dataKey: {}, expression: {}, err: {}",
                            siteId, deviceId, calcPointConfig.getDataKey(), calcPointConfig.getCalcExpression(),
                            expressionException.getMessage());
                    finishedKeys.add(calcDataKey);
                }
            }
            finishedKeys.forEach(remaining::remove);

            if (!progressed) {
                if (!unresolvedKeys.isEmpty()) {
                    log.warn("计算点依赖未就绪，跳过本轮计算，siteId: {}, deviceId: {}, unresolved: {}",
                            siteId, deviceId, String.join(", ", unresolvedKeys));
                }
                break;
            }
        }
        return calcPointIdValueMap;
    }

    private BigDecimal getSourceValueFromRawData(EmsPointConfig pointConfig, Map<String, BigDecimal> rawDataValues) {
        if (pointConfig == null || rawDataValues == null || rawDataValues.isEmpty()) {
            return null;
        }
        String dataKey = StringUtils.defaultString(pointConfig.getDataKey()).trim().toUpperCase();
        return StringUtils.isBlank(dataKey) ? null : rawDataValues.get(dataKey);
    }

    private void putPointValueToContext(EmsPointConfig pointConfig, BigDecimal pointValue, Map<String, BigDecimal> contextValues) {
        if (pointConfig == null || pointValue == null || contextValues == null) {
            return;
        }
        String pointKey = resolvePointContextKey(pointConfig);
        if (StringUtils.isNotBlank(pointKey)) {
            contextValues.put(pointKey, pointValue);
        }
    }

    private String resolveCalcContextKey(EmsPointConfig pointConfig) {
        return resolvePointContextKey(pointConfig);
    }

    private String resolvePointContextKey(EmsPointConfig pointConfig) {
        if (pointConfig == null) {
            return null;
        }
        String pointId = StringUtils.defaultString(pointConfig.getPointId()).trim().toUpperCase();
        return StringUtils.isNotBlank(pointId) ? pointId : null;
    }

    private ExpressionValue evaluateCalcExpression(String expression, Map<String, BigDecimal> contextValues,
                                                   String siteId, Date dataUpdateTime) {
        if (StringUtils.isBlank(expression)) {
            throw new IllegalArgumentException("计算表达式为空");
        }
        if (!SIMPLE_CALC_EXPRESSION_PATTERN.matcher(expression).matches()) {
            throw new IllegalArgumentException("计算表达式仅支持数字、字母、下划线、空格、运算符和函数语法");
        }
        Map<String, BigDecimal> evaluateContext = contextValues == null
                ? new HashMap<>()
                : new HashMap<>(contextValues);
        prepareAutoPeriodDiffContextValues(expression, evaluateContext, siteId, dataUpdateTime);
        CompiledExpression compiledExpression = calcExpressionCache.computeIfAbsent(expression, this::compileExpression);
        return compiledExpression.evaluate(evaluateContext);
    }

    private void prepareAutoPeriodDiffContextValues(String expression, Map<String, BigDecimal> contextValues,
                                                    String siteId, Date dataUpdateTime) {
        if (StringUtils.isBlank(expression) || contextValues == null || contextValues.isEmpty()) {
            return;
        }
        Date effectiveTime = dataUpdateTime == null ? DateUtils.getNowDate() : dataUpdateTime;
        prepareSingleAutoDiffContext(expression, contextValues, siteId, effectiveTime, "DAY_DIFF",
                AUTO_DAY_DIFF_FUNCTION_PATTERN);
        prepareSingleAutoDiffContext(expression, contextValues, siteId, effectiveTime, "MONTH_DIFF",
                AUTO_MONTH_DIFF_FUNCTION_PATTERN);
        prepareSingleAutoDiffContext(expression, contextValues, siteId, effectiveTime, "HOUR_DIFF",
                AUTO_HOUR_DIFF_FUNCTION_PATTERN);
    }

    private void prepareSingleAutoDiffContext(String expression, Map<String, BigDecimal> contextValues, String siteId,
                                              Date effectiveTime, String functionName, Pattern functionPattern) {
        Set<String> variables = extractAutoDiffVariables(expression, functionPattern);
        if (variables.isEmpty()) {
            return;
        }
        for (String variable : variables) {
            BigDecimal currentValue = contextValues.get(variable);
            if (currentValue == null) {
                continue;
            }
            BigDecimal diffValue = calculateAutoPeriodDiffValue(siteId, functionName, variable, currentValue, effectiveTime);
            contextValues.put(resolveAutoDiffContextKey(functionName, variable), diffValue);
        }
    }

    private Set<String> extractAutoDiffVariables(String expression, Pattern functionPattern) {
        Set<String> variables = new HashSet<>();
        Matcher matcher = functionPattern.matcher(expression);
        while (matcher.find()) {
            String variable = StringUtils.defaultString(matcher.group(1)).trim().toUpperCase();
            if (StringUtils.isNotBlank(variable)) {
                variables.add(variable);
            }
        }
        return variables;
    }

    private BigDecimal calculateAutoPeriodDiffValue(String siteId, String functionName, String variable,
                                                    BigDecimal currentValue, Date effectiveTime) {
        if (currentValue == null || StringUtils.isAnyBlank(siteId, functionName, variable) || effectiveTime == null) {
            return BigDecimal.ZERO;
        }
        if ("DAY_DIFF".equalsIgnoreCase(functionName)) {
            return calculateDayDiffValueFromInflux(siteId, variable, currentValue, effectiveTime);
        }
        if ("MONTH_DIFF".equalsIgnoreCase(functionName)) {
            return calculateMonthDiffValueFromInflux(siteId, variable, currentValue, effectiveTime);
        }
        if ("HOUR_DIFF".equalsIgnoreCase(functionName)) {
            return calculateHourDiffValueFromInflux(siteId, variable, currentValue, effectiveTime);
        }
        return BigDecimal.ZERO;
    }

    private BigDecimal calculateDayDiffValueFromInflux(String siteId, String variable,
                                                       BigDecimal currentValue, Date effectiveTime) {
        LocalDateTime currentDateTime = DateUtils.toLocalDateTime(effectiveTime);
        if (currentDateTime == null) {
            return BigDecimal.ZERO;
        }
        LocalDate currentDate = currentDateTime.toLocalDate();
        Date currentDayStart = DateUtils.toDate(currentDate);
        Date previousDayStart = DateUtils.toDate(currentDate.minusDays(1));
        Date previousDayEnd = new Date(currentDayStart.getTime() - 1);
        return calculateDiffByPreviousPeriodLastValue(siteId, variable, currentValue, previousDayStart, previousDayEnd);
    }

    private BigDecimal calculateMonthDiffValueFromInflux(String siteId, String variable,
                                                         BigDecimal currentValue, Date effectiveTime) {
        LocalDateTime currentDateTime = DateUtils.toLocalDateTime(effectiveTime);
        if (currentDateTime == null) {
            return BigDecimal.ZERO;
        }
        YearMonth currentMonth = YearMonth.from(currentDateTime);
        LocalDate currentMonthStartDate = currentMonth.atDay(1);
        Date currentMonthStart = DateUtils.toDate(currentMonthStartDate);
        YearMonth previousMonth = currentMonth.minusMonths(1);
        Date previousMonthStart = DateUtils.toDate(previousMonth.atDay(1));
        Date previousMonthEnd = new Date(currentMonthStart.getTime() - 1);
        return calculateDiffByPreviousPeriodLastValue(siteId, variable, currentValue, previousMonthStart, previousMonthEnd);
    }

    private BigDecimal calculateHourDiffValueFromInflux(String siteId, String variable,
                                                        BigDecimal currentValue, Date effectiveTime) {
        LocalDateTime currentDateTime = DateUtils.toLocalDateTime(effectiveTime);
        if (currentDateTime == null) {
            return BigDecimal.ZERO;
        }
        LocalDateTime currentHourStartTime = currentDateTime.withMinute(0).withSecond(0).withNano(0);
        Date currentHourStart = DateUtils.toDate(currentHourStartTime);
        Date previousHourStart = DateUtils.toDate(currentHourStartTime.minusHours(1));
        Date previousHourEnd = new Date(currentHourStart.getTime() - 1);
        return calculateDiffByPreviousPeriodLastValue(siteId, variable, currentValue, previousHourStart, previousHourEnd);
    }

    private BigDecimal calculateDiffByPreviousPeriodLastValue(String siteId, String variable,
                                                              BigDecimal currentValue, Date previousPeriodStart,
                                                              Date previousPeriodEnd) {
        if (StringUtils.isAnyBlank(siteId, variable)
                || currentValue == null
                || previousPeriodStart == null
                || previousPeriodEnd == null
                || previousPeriodEnd.before(previousPeriodStart)) {
            return BigDecimal.ZERO;
        }
        String pointKey = resolveInfluxPointKeyByVariable(siteId, variable);
        if (StringUtils.isBlank(pointKey)) {
            return BigDecimal.ZERO;
        }
        InfluxPointDataWriter.PointValue previousPeriodLastPoint = influxPointDataWriter.queryLatestPointValueByPointKey(
                siteId, pointKey, previousPeriodStart, previousPeriodEnd
        );
        if (previousPeriodLastPoint == null || previousPeriodLastPoint.getPointValue() == null) {
            return BigDecimal.ZERO;
        }
        return currentValue.subtract(previousPeriodLastPoint.getPointValue());
    }

    private String resolveInfluxPointKeyByVariable(String siteId, String variable) {
        if (StringUtils.isAnyBlank(siteId, variable)) {
            return null;
        }
        String normalizedVariable = normalizeVariablePart(variable);
        String upperVariable = variable.trim().toUpperCase();
        List<EmsPointConfig> sourcePointConfigs = getSiteSourcePointConfigs(siteId);
        for (EmsPointConfig sourcePointConfig : sourcePointConfigs) {
            if (sourcePointConfig == null) {
                continue;
            }
            String pointKey = resolveInfluxPointKey(sourcePointConfig);
            if (StringUtils.isBlank(pointKey)) {
                continue;
            }

            String pointIdKey = resolvePointContextKey(sourcePointConfig);
            if (StringUtils.isNotBlank(pointIdKey) && upperVariable.equals(pointIdKey)) {
                return pointKey;
            }

            String normalizedDataKey = normalizeVariablePart(sourcePointConfig.getDataKey());
            if (StringUtils.isNotBlank(normalizedVariable)
                    && StringUtils.isNotBlank(normalizedDataKey)
                    && normalizedVariable.equals(normalizedDataKey)) {
                return pointKey;
            }

            String devicePrefix = normalizeVariablePart(sourcePointConfig.getDeviceId());
            if (StringUtils.isNotBlank(devicePrefix)
                    && StringUtils.isNotBlank(normalizedDataKey)
                    && StringUtils.isNotBlank(normalizedVariable)
                    && normalizedVariable.equals(devicePrefix + "_" + normalizedDataKey)) {
                return pointKey;
            }
        }
        return null;
    }

    private static String resolveAutoDayDiffContextKey(String variable) {
        return AUTO_DAY_DIFF_CONTEXT_PREFIX + StringUtils.defaultString(variable).trim().toUpperCase();
    }

    private static String resolveAutoMonthDiffContextKey(String variable) {
        return AUTO_MONTH_DIFF_CONTEXT_PREFIX + StringUtils.defaultString(variable).trim().toUpperCase();
    }

    private static String resolveAutoHourDiffContextKey(String variable) {
        return AUTO_HOUR_DIFF_CONTEXT_PREFIX + StringUtils.defaultString(variable).trim().toUpperCase();
    }

    private static String resolveAutoDiffContextKey(String functionName, String variable) {
        if ("DAY_DIFF".equalsIgnoreCase(functionName)) {
            return resolveAutoDayDiffContextKey(variable);
        }
        if ("MONTH_DIFF".equalsIgnoreCase(functionName)) {
            return resolveAutoMonthDiffContextKey(variable);
        }
        if ("HOUR_DIFF".equalsIgnoreCase(functionName)) {
            return resolveAutoHourDiffContextKey(variable);
        }
        return null;
    }

    private CompiledExpression compileExpression(String expression) {
        return new CompiledExpression(expression);
    }

    private void flushPointDataSafely() {
        if (!pointDataFlushing.compareAndSet(false, true)) {
            return;
        }
        try {
            int drainedCount = 0;
            while (drainedCount < POINT_FLUSH_MAX_DRAIN_PER_RUN) {
                List<PointDataRecord> batch = new ArrayList<>(POINT_FLUSH_BATCH_SIZE);
                pointDataQueue.drainTo(batch, POINT_FLUSH_BATCH_SIZE);
                if (batch.isEmpty()) {
                    break;
                }
                drainedCount += batch.size();
                writePointDataToInflux(batch);
            }

            if (pointDataQueue.remainingCapacity() <= POINT_FLUSH_BATCH_SIZE) {
                log.warn("点位写入队列积压较高，queueSize: {}", pointDataQueue.size());
            }
        } catch (Exception e) {
            log.error("批量写入点位数据失败: {}", e.getMessage(), e);
        } finally {
            pointDataFlushing.set(false);
        }
    }

    private void writePointDataToInflux(List<PointDataRecord> records) {
        if (CollectionUtils.isEmpty(records)) {
            return;
        }
        List<InfluxPointDataWriter.PointWritePayload> payloads = records.stream()
                .map(item -> new InfluxPointDataWriter.PointWritePayload(
                        item.getSiteId(),
                        item.getDeviceId(),
                        item.getPointKey(),
                        item.getPointValue(),
                        item.getDataTime()))
                .collect(Collectors.toList());
        influxPointDataWriter.writeBatch(payloads);
    }

    private List<EmsPointConfig> getPointConfigsWithCache(String siteId, String deviceId, String deviceCategory) {
        String normalizedCategory = StringUtils.defaultString(deviceCategory).trim().toUpperCase();
        String cacheKey = RedisKeyConstants.POINT_CONFIG_DEVICE + siteId + "_" + deviceId + "_" + normalizedCategory;
        List<EmsPointConfig> cached = redisCache.getCacheObject(cacheKey);
        if (cached != null) {
            return cached;
        }

        // 1) 设备级点位（原有逻辑）
        EmsPointConfig query = new EmsPointConfig();
        query.setSiteId(siteId);
        query.setDeviceId(deviceId);
        List<EmsPointConfig> cacheValue = emsPointConfigMapper.selectEmsPointConfigList(query);
        redisCache.setCacheObject(cacheKey, cacheValue);
        return cacheValue;
    }

    private List<EmsPointConfig> getSiteCalcPointConfigs(String siteId) {
        if (StringUtils.isBlank(siteId)) {
            return Collections.emptyList();
        }
        EmsPointConfig query = new EmsPointConfig();
        query.setSiteId(siteId);
        query.setPointType("calc");
        List<EmsPointConfig> calcPoints = emsPointConfigMapper.selectEmsPointConfigList(query);
        if (CollectionUtils.isEmpty(calcPoints)) {
            return Collections.emptyList();
        }
        Map<String, EmsPointConfig> uniqueByPointId = new LinkedHashMap<>();
        for (EmsPointConfig calcPoint : calcPoints) {
            if (!isComputedPoint(calcPoint)) {
                continue;
            }
            String calcKey = resolvePointContextKey(calcPoint);
            if (StringUtils.isBlank(calcKey)) {
                continue;
            }
            uniqueByPointId.putIfAbsent(calcKey, calcPoint);
        }
        return new ArrayList<>(uniqueByPointId.values());
    }

    private List<EmsPointConfig> getSiteSourcePointConfigs(String siteId) {
        if (StringUtils.isBlank(siteId)) {
            return Collections.emptyList();
        }
        EmsPointConfig query = new EmsPointConfig();
        query.setSiteId(siteId);
        List<EmsPointConfig> pointConfigs = emsPointConfigMapper.selectEmsPointConfigList(query);
        if (CollectionUtils.isEmpty(pointConfigs)) {
            return Collections.emptyList();
        }
        Map<String, EmsPointConfig> uniqueByPointId = new LinkedHashMap<>();
        for (EmsPointConfig pointConfig : pointConfigs) {
            if (pointConfig == null || isComputedPoint(pointConfig)) {
                continue;
            }
            String pointId = resolveInfluxPointKey(pointConfig);
            if (StringUtils.isBlank(pointId)) {
                continue;
            }
            uniqueByPointId.putIfAbsent(pointId.trim().toUpperCase(), pointConfig);
        }
        return new ArrayList<>(uniqueByPointId.values());
    }

    private void putSourcePointValueToContext(EmsPointConfig pointConfig,
                                              BigDecimal pointValue,
                                              Map<String, BigDecimal> contextValues,
                                              Set<String> reservedCalcContextKeys) {
        if (pointConfig == null || pointValue == null || contextValues == null) {
            return;
        }
        String pointContextKey = resolvePointContextKey(pointConfig);
        if (StringUtils.isNotBlank(pointContextKey)) {
            contextValues.put(pointContextKey, pointValue);
        }
        String normalizedDataKey = normalizeVariablePart(pointConfig.getDataKey());
        if (StringUtils.isNotBlank(normalizedDataKey)) {
            String reservedKey = normalizedDataKey.trim().toUpperCase();
            if (CollectionUtils.isEmpty(reservedCalcContextKeys) || !reservedCalcContextKeys.contains(reservedKey)) {
                contextValues.put(normalizedDataKey, pointValue);
            } else {
                log.debug("计算点上下文跳过源点位 dataKey 裸变量，避免覆盖计算点，siteId/deviceId/pointId={}/{}/{}, dataKey={}",
                        pointConfig.getSiteId(), pointConfig.getDeviceId(), pointConfig.getPointId(), pointConfig.getDataKey());
            }
            String devicePrefix = normalizeVariablePart(pointConfig.getDeviceId());
            if (StringUtils.isNotBlank(devicePrefix)) {
                contextValues.put(devicePrefix + "_" + normalizedDataKey, pointValue);
            }
        }
    }

    private void cacheLatestPointValuesByPointId(String siteId, String deviceId,
                                                 Map<String, BigDecimal> pointIdValueMap,
                                                 Date dataTime) {
        if (StringUtils.isBlank(siteId) || org.apache.commons.collections4.MapUtils.isEmpty(pointIdValueMap)) {
            log.warn("MQTT点位写Redis跳过，siteId: {}, deviceId: {}, pointMapEmpty: {}",
                    siteId, deviceId, org.apache.commons.collections4.MapUtils.isEmpty(pointIdValueMap));
            return;
        }
        Date latestTime = dataTime == null ? DateUtils.getNowDate() : dataTime;
        int writeCount = 0;
        List<String> writeSamples = new ArrayList<>();
        for (Map.Entry<String, BigDecimal> entry : pointIdValueMap.entrySet()) {
            String pointId = StringUtils.trimToNull(entry.getKey());
            BigDecimal pointValue = entry.getValue();
            if (StringUtils.isBlank(pointId) || pointValue == null) {
                log.warn("MQTT点位写Redis忽略无效值，siteId: {}, deviceId: {}, rawPointId: {}, pointValue: {}",
                        siteId, deviceId, entry.getKey(), pointValue);
                continue;
            }
            JSONObject cacheValue = new JSONObject();
            cacheValue.put("siteId", siteId);
            cacheValue.put("deviceId", deviceId);
            cacheValue.put("pointId", pointId);
            cacheValue.put("pointValue", pointValue);
            cacheValue.put("dataTime", latestTime);
            cacheValue.put("timestamp", latestTime == null ? null : latestTime.getTime());
            redisCache.setCacheObject(pointId, cacheValue);
            writeCount++;
            if (writeSamples.size() < 10) {
                writeSamples.add(pointId);
            }
        }
        log.info("MQTT点位写Redis完成，siteId: {}, deviceId: {}, writeCount: {}, writeSamples: {}, dataTime: {}",
                siteId, deviceId, writeCount, writeSamples, latestTime);
    }

    private static class PointDataRecord {
        private final String siteId;
        private final String deviceId;
        private final String pointKey;
        private final BigDecimal pointValue;
        private final Date dataTime;

        private PointDataRecord(String siteId, String deviceId, String pointKey, BigDecimal pointValue, Date dataTime) {
            this.siteId = siteId;
            this.deviceId = deviceId;
            this.pointKey = pointKey;
            this.pointValue = pointValue;
            this.dataTime = dataTime;
        }

        private String getSiteId() {
            return siteId;
        }

        private String getDeviceId() {
            return deviceId;
        }

        private String getPointKey() {
            return pointKey;
        }

        private BigDecimal getPointValue() {
            return pointValue;
        }

        private Date getDataTime() {
            return dataTime;
        }
    }

    private static class PointLatestCacheValue {
        private String siteId;
        private String deviceId;
        private String pointId;
        private BigDecimal pointValue;
        private Date dataTime;

        private String getSiteId() {
            return siteId;
        }

        private void setSiteId(String siteId) {
            this.siteId = siteId;
        }

        private String getDeviceId() {
            return deviceId;
        }

        private void setDeviceId(String deviceId) {
            this.deviceId = deviceId;
        }

        private String getPointId() {
            return pointId;
        }

        private void setPointId(String pointId) {
            this.pointId = pointId;
        }

        private BigDecimal getPointValue() {
            return pointValue;
        }

        private void setPointValue(BigDecimal pointValue) {
            this.pointValue = pointValue;
        }

        private Date getDataTime() {
            return dataTime;
        }

        private void setDataTime(Date dataTime) {
            this.dataTime = dataTime;
        }
    }

    private static class MissingVariableException extends IllegalArgumentException {
        private final String variable;

        private MissingVariableException(String variable) {
            super("缺少变量: " + variable);
            this.variable = variable;
        }

        private String getVariable() {
            return variable;
        }
    }

    private enum ExprTokenType {
        NUMBER,
        STRING,
        IDENTIFIER,
        OPERATOR,
        LEFT_PAREN,
        RIGHT_PAREN,
        COMMA,
        QUESTION,
        COLON,
        EOF
    }

    private static class ExprToken {
        private final ExprTokenType type;
        private final String text;
        private final BigDecimal number;
        private final String stringValue;

        private ExprToken(ExprTokenType type, String text, BigDecimal number, String stringValue) {
            this.type = type;
            this.text = text;
            this.number = number;
            this.stringValue = stringValue;
        }

        private static ExprToken number(BigDecimal value) {
            return new ExprToken(ExprTokenType.NUMBER, null, value, null);
        }

        private static ExprToken string(String value) {
            return new ExprToken(ExprTokenType.STRING, null, null, value);
        }

        private static ExprToken identifier(String value) {
            return new ExprToken(ExprTokenType.IDENTIFIER, value, null, null);
        }

        private static ExprToken operator(String value) {
            return new ExprToken(ExprTokenType.OPERATOR, value, null, null);
        }

        private static ExprToken symbol(ExprTokenType type, String text) {
            return new ExprToken(type, text, null, null);
        }
    }

    private static class ExpressionValue {
        private final BigDecimal numberValue;
        private final String textValue;

        private ExpressionValue(BigDecimal numberValue, String textValue) {
            this.numberValue = numberValue;
            this.textValue = textValue;
        }

        private static ExpressionValue ofNumber(BigDecimal numberValue) {
            return new ExpressionValue(numberValue, null);
        }

        private static ExpressionValue ofText(String textValue) {
            return new ExpressionValue(null, textValue == null ? "" : textValue);
        }

        private boolean isNumber() {
            return numberValue != null;
        }

        private BigDecimal asNumber() {
            if (numberValue == null) {
                throw new IllegalArgumentException("表达式值不是数值类型: " + textValue);
            }
            return numberValue;
        }

        private String asText() {
            return numberValue != null ? numberValue.stripTrailingZeros().toPlainString() : textValue;
        }

        private boolean asBoolean() {
            return isNumber() ? BigDecimal.ZERO.compareTo(numberValue) != 0 : StringUtils.isNotBlank(textValue);
        }

    }

    private interface ExpressionNode {
        ExpressionValue evaluate(Map<String, BigDecimal> contextValues);
    }

    private static class NumberNode implements ExpressionNode {
        private final BigDecimal value;

        private NumberNode(BigDecimal value) {
            this.value = value;
        }

        @Override
        public ExpressionValue evaluate(Map<String, BigDecimal> contextValues) {
            return ExpressionValue.ofNumber(value);
        }
    }

    private static class StringNode implements ExpressionNode {
        private final String value;

        private StringNode(String value) {
            this.value = value;
        }

        @Override
        public ExpressionValue evaluate(Map<String, BigDecimal> contextValues) {
            return ExpressionValue.ofText(value);
        }
    }

    private static class VariableNode implements ExpressionNode {
        private final String name;

        private VariableNode(String name) {
            this.name = name;
        }

        @Override
        public ExpressionValue evaluate(Map<String, BigDecimal> contextValues) {
            BigDecimal value = contextValues.get(name);
            if (value == null) {
                throw new MissingVariableException(name);
            }
            return ExpressionValue.ofNumber(value);
        }

        private String getName() {
            return name;
        }
    }

    private static class UnaryNode implements ExpressionNode {
        private final String operator;
        private final ExpressionNode node;

        private UnaryNode(String operator, ExpressionNode node) {
            this.operator = operator;
            this.node = node;
        }

        @Override
        public ExpressionValue evaluate(Map<String, BigDecimal> contextValues) {
            ExpressionValue value = node.evaluate(contextValues);
            switch (operator) {
                case "+":
                    return value;
                case "-":
                    return ExpressionValue.ofNumber(value.asNumber().negate());
                case "!":
                    return value.asBoolean() ? ExpressionValue.ofNumber(BigDecimal.ZERO) : ExpressionValue.ofNumber(BigDecimal.ONE);
                default:
                    throw new IllegalArgumentException("不支持的一元操作符: " + operator);
            }
        }
    }

    private static class BinaryNode implements ExpressionNode {
        private final String operator;
        private final ExpressionNode left;
        private final ExpressionNode right;

        private BinaryNode(String operator, ExpressionNode left, ExpressionNode right) {
            this.operator = operator;
            this.left = left;
            this.right = right;
        }

        @Override
        public ExpressionValue evaluate(Map<String, BigDecimal> contextValues) {
            if ("&&".equals(operator)) {
                ExpressionValue leftValue = left.evaluate(contextValues);
                if (!leftValue.asBoolean()) {
                    return ExpressionValue.ofNumber(BigDecimal.ZERO);
                }
                return right.evaluate(contextValues).asBoolean()
                        ? ExpressionValue.ofNumber(BigDecimal.ONE)
                        : ExpressionValue.ofNumber(BigDecimal.ZERO);
            }
            if ("||".equals(operator)) {
                ExpressionValue leftValue = left.evaluate(contextValues);
                if (leftValue.asBoolean()) {
                    return ExpressionValue.ofNumber(BigDecimal.ONE);
                }
                return right.evaluate(contextValues).asBoolean()
                        ? ExpressionValue.ofNumber(BigDecimal.ONE)
                        : ExpressionValue.ofNumber(BigDecimal.ZERO);
            }

            ExpressionValue leftValue = left.evaluate(contextValues);
            ExpressionValue rightValue = right.evaluate(contextValues);
            switch (operator) {
                case "+":
                    if (!leftValue.isNumber() || !rightValue.isNumber()) {
                        return ExpressionValue.ofText(leftValue.asText() + rightValue.asText());
                    }
                    return ExpressionValue.ofNumber(leftValue.asNumber().add(rightValue.asNumber()));
                case "-":
                    return ExpressionValue.ofNumber(leftValue.asNumber().subtract(rightValue.asNumber()));
                case "*":
                    return ExpressionValue.ofNumber(leftValue.asNumber().multiply(rightValue.asNumber()));
                case "/":
                    if (BigDecimal.ZERO.compareTo(rightValue.asNumber()) == 0) {
                        throw new IllegalArgumentException("除数不能为0");
                    }
                    return ExpressionValue.ofNumber(leftValue.asNumber().divide(rightValue.asNumber(), 10, RoundingMode.HALF_UP));
                case ">":
                    return leftValue.asNumber().compareTo(rightValue.asNumber()) > 0
                            ? ExpressionValue.ofNumber(BigDecimal.ONE) : ExpressionValue.ofNumber(BigDecimal.ZERO);
                case ">=":
                    return leftValue.asNumber().compareTo(rightValue.asNumber()) >= 0
                            ? ExpressionValue.ofNumber(BigDecimal.ONE) : ExpressionValue.ofNumber(BigDecimal.ZERO);
                case "<":
                    return leftValue.asNumber().compareTo(rightValue.asNumber()) < 0
                            ? ExpressionValue.ofNumber(BigDecimal.ONE) : ExpressionValue.ofNumber(BigDecimal.ZERO);
                case "<=":
                    return leftValue.asNumber().compareTo(rightValue.asNumber()) <= 0
                            ? ExpressionValue.ofNumber(BigDecimal.ONE) : ExpressionValue.ofNumber(BigDecimal.ZERO);
                case "==":
                    if (leftValue.isNumber() && rightValue.isNumber()) {
                        return leftValue.asNumber().compareTo(rightValue.asNumber()) == 0
                                ? ExpressionValue.ofNumber(BigDecimal.ONE) : ExpressionValue.ofNumber(BigDecimal.ZERO);
                    }
                    return Objects.equals(leftValue.asText(), rightValue.asText())
                            ? ExpressionValue.ofNumber(BigDecimal.ONE) : ExpressionValue.ofNumber(BigDecimal.ZERO);
                case "!=":
                    if (leftValue.isNumber() && rightValue.isNumber()) {
                        return leftValue.asNumber().compareTo(rightValue.asNumber()) != 0
                                ? ExpressionValue.ofNumber(BigDecimal.ONE) : ExpressionValue.ofNumber(BigDecimal.ZERO);
                    }
                    return !Objects.equals(leftValue.asText(), rightValue.asText())
                            ? ExpressionValue.ofNumber(BigDecimal.ONE) : ExpressionValue.ofNumber(BigDecimal.ZERO);
                default:
                    throw new IllegalArgumentException("不支持的操作符: " + operator);
            }
        }
    }

    private static class TernaryNode implements ExpressionNode {
        private final ExpressionNode condition;
        private final ExpressionNode trueNode;
        private final ExpressionNode falseNode;

        private TernaryNode(ExpressionNode condition, ExpressionNode trueNode, ExpressionNode falseNode) {
            this.condition = condition;
            this.trueNode = trueNode;
            this.falseNode = falseNode;
        }

        @Override
        public ExpressionValue evaluate(Map<String, BigDecimal> contextValues) {
            return condition.evaluate(contextValues).asBoolean()
                    ? trueNode.evaluate(contextValues)
                    : falseNode.evaluate(contextValues);
        }
    }

    private static class FunctionNode implements ExpressionNode {
        private final String functionName;
        private final List<ExpressionNode> arguments;

        private FunctionNode(String functionName, List<ExpressionNode> arguments) {
            this.functionName = StringUtils.defaultString(functionName).trim().toUpperCase();
            this.arguments = arguments == null ? Collections.emptyList() : arguments;
        }

        @Override
        public ExpressionValue evaluate(Map<String, BigDecimal> contextValues) {
            if (("DAY_DIFF".equals(functionName)
                    || "MONTH_DIFF".equals(functionName)
                    || "HOUR_DIFF".equals(functionName)) && arguments.size() == 1) {
                ExpressionNode argument = arguments.get(0);
                if (!(argument instanceof VariableNode)) {
                    throw new IllegalArgumentException(functionName + "单参数模式仅支持变量参数");
                }
                String variable = ((VariableNode) argument).getName();
                BigDecimal autoDiffValue = contextValues.get(resolveAutoDiffContextKey(functionName, variable));
                if (autoDiffValue == null) {
                    throw new MissingVariableException(variable);
                }
                return ExpressionValue.ofNumber(autoDiffValue);
            }
            if ("DAY_DIFF".equals(functionName)
                    || "MONTH_DIFF".equals(functionName)
                    || "HOUR_DIFF".equals(functionName)) {
                throw new IllegalArgumentException(functionName + "函数参数数量错误，需1个参数");
            }
            throw new IllegalArgumentException("不支持的函数: " + functionName);
        }
    }

    private static class CompiledExpression {
        private final ExpressionNode root;

        private CompiledExpression(String expression) {
            List<ExprToken> tokens = tokenizeExpression(expression);
            ExpressionParser parser = new ExpressionParser(tokens);
            this.root = parser.parseExpression();
            if (!parser.isEnd()) {
                ExprToken token = parser.peek();
                throw new IllegalArgumentException("表达式尾部有多余内容: " + token.text);
            }
        }

        private ExpressionValue evaluate(Map<String, BigDecimal> contextValues) {
            return root.evaluate(contextValues);
        }
    }

    private static class ExpressionParser {
        private final List<ExprToken> tokens;
        private int index;

        private ExpressionParser(List<ExprToken> tokens) {
            this.tokens = tokens;
            this.index = 0;
        }

        private ExpressionNode parseExpression() {
            return parseTernary();
        }

        private ExpressionNode parseTernary() {
            ExpressionNode condition = parseOr();
            if (match(ExprTokenType.QUESTION)) {
                ExpressionNode trueNode = parseTernary();
                expect(ExprTokenType.COLON, "三元表达式缺少 ':'");
                ExpressionNode falseNode = parseTernary();
                return new TernaryNode(condition, trueNode, falseNode);
            }
            return condition;
        }

        private ExpressionNode parseOr() {
            ExpressionNode left = parseAnd();
            while (matchOperator("||")) {
                left = new BinaryNode("||", left, parseAnd());
            }
            return left;
        }

        private ExpressionNode parseAnd() {
            ExpressionNode left = parseEquality();
            while (matchOperator("&&")) {
                left = new BinaryNode("&&", left, parseEquality());
            }
            return left;
        }

        private ExpressionNode parseEquality() {
            ExpressionNode left = parseComparison();
            while (true) {
                if (matchOperator("==")) {
                    left = new BinaryNode("==", left, parseComparison());
                    continue;
                }
                if (matchOperator("!=")) {
                    left = new BinaryNode("!=", left, parseComparison());
                    continue;
                }
                break;
            }
            return left;
        }

        private ExpressionNode parseComparison() {
            ExpressionNode left = parseAddSub();
            while (true) {
                if (matchOperator(">=")) {
                    left = new BinaryNode(">=", left, parseAddSub());
                    continue;
                }
                if (matchOperator("<=")) {
                    left = new BinaryNode("<=", left, parseAddSub());
                    continue;
                }
                if (matchOperator(">")) {
                    left = new BinaryNode(">", left, parseAddSub());
                    continue;
                }
                if (matchOperator("<")) {
                    left = new BinaryNode("<", left, parseAddSub());
                    continue;
                }
                break;
            }
            return left;
        }

        private ExpressionNode parseAddSub() {
            ExpressionNode left = parseMulDiv();
            while (true) {
                if (matchOperator("+")) {
                    left = new BinaryNode("+", left, parseMulDiv());
                    continue;
                }
                if (matchOperator("-")) {
                    left = new BinaryNode("-", left, parseMulDiv());
                    continue;
                }
                break;
            }
            return left;
        }

        private ExpressionNode parseMulDiv() {
            ExpressionNode left = parseUnary();
            while (true) {
                if (matchOperator("*")) {
                    left = new BinaryNode("*", left, parseUnary());
                    continue;
                }
                if (matchOperator("/")) {
                    left = new BinaryNode("/", left, parseUnary());
                    continue;
                }
                break;
            }
            return left;
        }

        private ExpressionNode parseUnary() {
            if (matchOperator("+")) {
                return new UnaryNode("+", parseUnary());
            }
            if (matchOperator("-")) {
                return new UnaryNode("-", parseUnary());
            }
            if (matchOperator("!")) {
                return new UnaryNode("!", parseUnary());
            }
            return parsePrimary();
        }

        private ExpressionNode parsePrimary() {
            ExprToken token = peek();
            if (match(ExprTokenType.NUMBER)) {
                return new NumberNode(token.number);
            }
            if (match(ExprTokenType.STRING)) {
                return new StringNode(token.stringValue);
            }
            if (match(ExprTokenType.IDENTIFIER)) {
                String identifier = token.text;
                if (match(ExprTokenType.LEFT_PAREN)) {
                    return parseFunction(identifier);
                }
                return new VariableNode(identifier.toUpperCase());
            }
            if (match(ExprTokenType.LEFT_PAREN)) {
                ExpressionNode nested = parseExpression();
                expect(ExprTokenType.RIGHT_PAREN, "括号不匹配，缺少右括号");
                return nested;
            }
            throw new IllegalArgumentException("表达式语法错误，当前位置: " + token.text);
        }

        private ExpressionNode parseFunction(String identifier) {
            if ("IF".equalsIgnoreCase(identifier)) {
                ExpressionNode condition = parseExpression();
                expect(ExprTokenType.COMMA, "IF函数参数格式错误，缺少第1个逗号");
                ExpressionNode trueNode = parseExpression();
                expect(ExprTokenType.COMMA, "IF函数参数格式错误，缺少第2个逗号");
                ExpressionNode falseNode = parseExpression();
                expect(ExprTokenType.RIGHT_PAREN, "IF函数缺少右括号");
                return new TernaryNode(condition, trueNode, falseNode);
            }
            if ("DAY_DIFF".equalsIgnoreCase(identifier)) {
                List<ExpressionNode> arguments = parseFunctionArguments(identifier);
                if (arguments.size() != 1) {
                    throw new IllegalArgumentException("DAY_DIFF函数参数数量错误，需1个参数");
                }
                return new FunctionNode(identifier, arguments);
            }
            if ("MONTH_DIFF".equalsIgnoreCase(identifier)) {
                List<ExpressionNode> arguments = parseFunctionArguments(identifier);
                if (arguments.size() != 1) {
                    throw new IllegalArgumentException("MONTH_DIFF函数参数数量错误，需1个参数");
                }
                return new FunctionNode(identifier, arguments);
            }
            if ("HOUR_DIFF".equalsIgnoreCase(identifier)) {
                List<ExpressionNode> arguments = parseFunctionArguments(identifier);
                if (arguments.size() != 1) {
                    throw new IllegalArgumentException("HOUR_DIFF函数参数数量错误，需1个参数");
                }
                return new FunctionNode(identifier, arguments);
            }
            throw new IllegalArgumentException("不支持的函数: " + identifier);
        }

        private List<ExpressionNode> parseFunctionArguments(String functionName) {
            List<ExpressionNode> arguments = new ArrayList<>();
            if (match(ExprTokenType.RIGHT_PAREN)) {
                return arguments;
            }
            while (true) {
                arguments.add(parseExpression());
                if (match(ExprTokenType.COMMA)) {
                    continue;
                }
                expect(ExprTokenType.RIGHT_PAREN, functionName + "函数缺少右括号");
                return arguments;
            }
        }

        private ExprToken peek() {
            if (index >= tokens.size()) {
                return ExprToken.symbol(ExprTokenType.EOF, "<eof>");
            }
            return tokens.get(index);
        }

        private boolean isEnd() {
            return peek().type == ExprTokenType.EOF;
        }

        private boolean match(ExprTokenType type) {
            if (peek().type != type) {
                return false;
            }
            index++;
            return true;
        }

        private boolean matchOperator(String operator) {
            ExprToken token = peek();
            if (token.type != ExprTokenType.OPERATOR || !operator.equals(token.text)) {
                return false;
            }
            index++;
            return true;
        }

        private void expect(ExprTokenType type, String message) {
            if (!match(type)) {
                throw new IllegalArgumentException(message);
            }
        }
    }

    private static List<ExprToken> tokenizeExpression(String expression) {
        if (StringUtils.isBlank(expression)) {
            throw new IllegalArgumentException("计算表达式为空");
        }
        List<ExprToken> tokens = new ArrayList<>();
        int index = 0;
        while (index < expression.length()) {
            char ch = expression.charAt(index);
            if (Character.isWhitespace(ch)) {
                index++;
                continue;
            }
            if (Character.isDigit(ch) || ch == '.') {
                int start = index;
                boolean hasDot = ch == '.';
                index++;
                while (index < expression.length()) {
                    char next = expression.charAt(index);
                    if (Character.isDigit(next)) {
                        index++;
                        continue;
                    }
                    if (next == '.' && !hasDot) {
                        hasDot = true;
                        index++;
                        continue;
                    }
                    break;
                }
                String text = expression.substring(start, index);
                try {
                    tokens.add(ExprToken.number(new BigDecimal(text)));
                } catch (NumberFormatException ex) {
                    throw new IllegalArgumentException("数值格式错误: " + text);
                }
                continue;
            }
            if (Character.isLetter(ch) || ch == '_') {
                int start = index;
                index++;
                while (index < expression.length()) {
                    char next = expression.charAt(index);
                    if (Character.isLetterOrDigit(next) || next == '_') {
                        index++;
                        continue;
                    }
                    break;
                }
                String text = expression.substring(start, index).trim().toUpperCase();
                tokens.add(ExprToken.identifier(text));
                continue;
            }
            if (ch == '\'' || ch == '"') {
                char quote = ch;
                index++;
                StringBuilder sb = new StringBuilder();
                boolean escaped = false;
                while (index < expression.length()) {
                    char current = expression.charAt(index++);
                    if (escaped) {
                        switch (current) {
                            case 'n':
                                sb.append('\n');
                                break;
                            case 't':
                                sb.append('\t');
                                break;
                            case 'r':
                                sb.append('\r');
                                break;
                            case '\\':
                                sb.append('\\');
                                break;
                            case '\'':
                                sb.append('\'');
                                break;
                            case '"':
                                sb.append('"');
                                break;
                            default:
                                sb.append(current);
                                break;
                        }
                        escaped = false;
                        continue;
                    }
                    if (current == '\\') {
                        escaped = true;
                        continue;
                    }
                    if (current == quote) {
                        break;
                    }
                    sb.append(current);
                }
                if (escaped || index > expression.length() || expression.charAt(index - 1) != quote) {
                    throw new IllegalArgumentException("字符串字面量未闭合");
                }
                tokens.add(ExprToken.string(sb.toString()));
                continue;
            }
            if (ch == '(') {
                tokens.add(ExprToken.symbol(ExprTokenType.LEFT_PAREN, "("));
                index++;
                continue;
            }
            if (ch == ')') {
                tokens.add(ExprToken.symbol(ExprTokenType.RIGHT_PAREN, ")"));
                index++;
                continue;
            }
            if (ch == ',') {
                tokens.add(ExprToken.symbol(ExprTokenType.COMMA, ","));
                index++;
                continue;
            }
            if (ch == '?') {
                tokens.add(ExprToken.symbol(ExprTokenType.QUESTION, "?"));
                index++;
                continue;
            }
            if (ch == ':') {
                tokens.add(ExprToken.symbol(ExprTokenType.COLON, ":"));
                index++;
                continue;
            }
            if (index + 1 < expression.length()) {
                String twoChars = expression.substring(index, index + 2);
                if ("&&".equals(twoChars)
                        || "||".equals(twoChars)
                        || ">=".equals(twoChars)
                        || "<=".equals(twoChars)
                        || "==".equals(twoChars)
                        || "!=".equals(twoChars)) {
                    tokens.add(ExprToken.operator(twoChars));
                    index += 2;
                    continue;
                }
            }
            if (ch == '+' || ch == '-' || ch == '*' || ch == '/'
                    || ch == '>' || ch == '<' || ch == '!') {
                tokens.add(ExprToken.operator(String.valueOf(ch)));
                index++;
                continue;
            }
            throw new IllegalArgumentException("表达式包含非法字符: " + ch);
        }
        tokens.add(ExprToken.symbol(ExprTokenType.EOF, "<eof>"));
        return tokens;
    }

    private JSONArray parseJsonData(String message) {
        try {
            JSONObject object = JSONObject.parseObject(message);
            return object.getJSONArray("payload");
        } catch (JSONException e) {
            log.info("mqtt message is not a json object: " + e.getMessage());
            try {
                return JSONArray.parseArray(message);
            } catch (Exception arrayException) {
                log.info("mqtt message is not a json array: " + e.getMessage());
            }
        }
        return null;
    }

    private String getDeviceCategory(String siteId, String deviceId) {
        EmsDevicesSetting emsDevicesSetting = getDeviceSettingWithCache(siteId, deviceId);
        if (emsDevicesSetting != null) {
            return emsDevicesSetting.getDeviceCategory();
        }
        return "";
    }

    private void emsDataProcess(String siteId, String deviceId, String jsonData, Date dataUpdateTime) {
        //EMS
        Map<String, Object> obj = JSON.parseObject(jsonData, new TypeReference<Map<String, Object>>() {
        });
        // 点位匹配数据
        List<EmsPointMatch> pointMatchList = devicePointMatchDataProcessor.getDevicePointMatch(siteId, deviceId, DeviceMatchTable.EMS.getCode());
        if (CollectionUtils.isEmpty(pointMatchList)) {
            log.info("未找到匹配的点位数据，无法处理EMS数据，siteId: " + siteId + "，deviceId: " + deviceId);
            return;
        }

        EmsEmsData emsData = new EmsEmsData();

        saveDeviceData(pointMatchList, obj, emsData);

        emsData.setDataUpdateTime(dataUpdateTime);
        emsData.setCreateBy("system");
        emsData.setCreateTime(DateUtils.getNowDate());
        emsData.setUpdateBy("system");
        emsData.setUpdateTime(DateUtils.getNowDate());
        emsData.setSiteId(siteId);
        emsData.setDeviceId(deviceId);

        emsEmsDataMapper.insertEmsEmsData(emsData);

        redisCache.setCacheObject(RedisKeyConstants.EMS + siteId + "_" + deviceId, emsData);
    }

    private void coolingDataProcess(String siteId, String deviceId, String jsonData, Date dataUpdateTime) {
        //中水冷却
        Map<String, Object> obj = JSON.parseObject(jsonData, new TypeReference<Map<String, Object>>() {
        });
        // 点位匹配数据
        List<EmsPointMatch> pointMatchList = devicePointMatchDataProcessor.getDevicePointMatch(siteId, deviceId, DeviceMatchTable.COOLING.getCode());
        if (CollectionUtils.isEmpty(pointMatchList)) {
            log.info("未找到匹配的点位数据，无法处理中水冷却数据，siteId: " + siteId + "，deviceId: " + deviceId);
            return;
        }

        EmsCoolingData coolingData = new EmsCoolingData();

        saveDeviceData(pointMatchList, obj, coolingData);

        coolingData.setDataUpdateTime(dataUpdateTime);
        coolingData.setCreateBy("system");
        coolingData.setCreateTime(DateUtils.getNowDate());
        coolingData.setUpdateBy("system");
        coolingData.setUpdateTime(DateUtils.getNowDate());
        coolingData.setSiteId(siteId);
        coolingData.setDeviceId(deviceId);

        emsCoolingDataMapper.insertEmsCoolingData(coolingData);

        redisCache.setCacheObject(RedisKeyConstants.COOLING + siteId + "_" + deviceId, coolingData);

    }

    public void meteXFProcess(String siteId, String deviceId, String dataJson, Date dataUpdateTime) {
        //消防
        Map<String, Object> obj = JSON.parseObject(dataJson, new TypeReference<Map<String, Object>>() {
        });
        // 点位匹配数据
        List<EmsPointMatch> pointMatchList = devicePointMatchDataProcessor.getDevicePointMatch(siteId, deviceId, DeviceMatchTable.XF.getCode());
        if (CollectionUtils.isEmpty(pointMatchList)) {
            log.info("未找到匹配的点位数据，无法处理消防数据，siteId: " + siteId + "，deviceId: " + deviceId);
            return;
        }

        // 暂时只更新设备表的设备状态 ZDYBYDCZT-主电源备用电池状态
        // 数据存表
        EmsXfData xfData = new EmsXfData();
        xfData.setDataTimestamp(dataUpdateTime);

        saveDeviceData(pointMatchList, obj, xfData);

        xfData.setCreateBy("system");
        xfData.setCreateTime(DateUtils.getNowDate());
        xfData.setUpdateBy("system");
        xfData.setUpdateTime(DateUtils.getNowDate());
        xfData.setSiteId(siteId);
        xfData.setDeviceId(deviceId);
        emsXfDataMapper.insertEmsXfData(xfData);

        redisCache.setCacheObject(RedisKeyConstants.XF + siteId + "_" + deviceId, xfData);
        // 状态枚举还没有提供
        EmsDevicesSetting emsDevicesSetting = getDeviceSettingWithCache(siteId, deviceId);
        if (emsDevicesSetting != null) {
            emsDevicesSetting.setCommunicationStatus(StringUtils.getString(xfData.getDczt()));
            emsDevicesSetting.setUpdatedAt(DateUtils.getNowDate());
            emsDevicesSettingMapper.updateEmsDevicesSetting(emsDevicesSetting);
            cacheDeviceSetting(siteId, deviceId, emsDevicesSetting);
        }
    }

    private void dhDataProcess(String siteId, String deviceId, String dataJson, Date dateUpdateTime) {
        //动环
        Map<String, Object> obj = JSON.parseObject(dataJson, new TypeReference<Map<String, Object>>() {
        });
        // 点位匹配数据
        List<EmsPointMatch> pointMatchList = devicePointMatchDataProcessor.getDevicePointMatch(siteId, deviceId, DeviceMatchTable.DH.getCode());
        if (CollectionUtils.isEmpty(pointMatchList)) {
            log.info("未找到匹配的点位数据，无法处理动环数据，siteId: " + siteId + "，deviceId: " + deviceId);
            return;
        }

        //DH
        EmsDhData dhData = new EmsDhData();

        saveDeviceData(pointMatchList, obj, dhData);

        dhData.setDataUpdateTime(dateUpdateTime);
        dhData.setCreateBy("system");
        dhData.setCreateTime(DateUtils.getNowDate());
        dhData.setUpdateBy("system");
        dhData.setUpdateTime(DateUtils.getNowDate());
        dhData.setSiteId(siteId);
        dhData.setDeviceId(deviceId);
        emsDhDataMapper.insertEmsDhData(dhData);

        redisCache.setCacheObject(RedisKeyConstants.DH + siteId + "_" + deviceId, dhData);

    }

    private void batteryStackDataProcess(String siteId, String deviceId, String dataJson, Date dataUpdateTime) {

        //电池堆
        Map<String, Object> obj = JSON.parseObject(dataJson, new TypeReference<Map<String, Object>>() {
        });

        // 将原始数据存一下方便后面簇数据使用
        redisCache.setCacheObject(RedisKeyConstants.ORIGINAL_BMSD + siteId + "_" + deviceId, obj);

        // 点位匹配数据
        List<EmsPointMatch> pointMatchList = devicePointMatchDataProcessor.getDevicePointMatch(siteId, deviceId, DeviceMatchTable.STACK.getCode());
        if (CollectionUtils.isEmpty(pointMatchList)) {
            log.info("未找到匹配的点位数据，无法处理电池堆数据，siteId: " + siteId + "，deviceId: " + deviceId);
            return;
        }

        //BMS 电池簇
        EmsBatteryStack dataStack = new EmsBatteryStack();

        // 其他非 BigDecimal 字段
        dataStack.setDataUpdateTime(dataUpdateTime);
//        dataStack.setWorkStatus(WorkStatus.NORMAL.getCode()); // 或其他默认值
        dataStack.setPcsCommunicationStatus(CommunicationStatus.OK.getCode());
        dataStack.setEmsCommunicationStatus(CommunicationStatus.OK.getCode());

        saveDeviceData(pointMatchList, obj, dataStack);

        dataStack.setCreateBy("system");
        dataStack.setCreateTime(DateUtils.getNowDate());
        dataStack.setUpdateBy("system");
        dataStack.setUpdateTime(DateUtils.getNowDate());
        dataStack.setSiteId(siteId);
        dataStack.setDeviceId(deviceId);

        //TODO 临时解决上送数据为浮点的问题
        String status = dataStack.getWorkStatus();
        int result = 5;

        if (status != null && !status.trim().isEmpty() && status.contains("null")) {
            // 1.0 -> 1, 2.0 -> 2
            result = (int) Double.parseDouble(status.trim());
        }

        dataStack.setWorkStatus(String.valueOf(result));

        emsBatteryStackMapper.insertEmsBatteryStack(dataStack);

        redisCache.setCacheObject(RedisKeyConstants.STACK + siteId + "_" + deviceId, dataStack);

        // 同步更新设备工作状态
        saveDeviceWorkStatus(siteId, deviceId, DeviceMatchTable.STACK.getCode(), dataStack.getWorkStatus());
    }

    private void batteryGroupDataProcess(String siteId, String deviceId, String jsonData) {
//        if (!SiteEnum.DDS.getCode().equals(siteId)) {
//            return;
//        }
        //电池组
        Map<String, Object> obj = JSON.parseObject(jsonData, new TypeReference<Map<String, Object>>() {
        });

        Map<String, EmsBatteryGroup> groupMap = new HashMap<>();
        for (Map.Entry<String, Object> entry : obj.entrySet()) {
            String key = entry.getKey();
            // 跳过空键
            if (key == null || key.trim().isEmpty()) {
                continue;
            }
            Matcher matcher = PATTERN.matcher(key);
            if (matcher.matches()) {
                String groupDeviceId = matcher.group(1);
                String property = matcher.group(2);
                EmsBatteryGroup dataGroup = groupMap.getOrDefault(groupDeviceId, new EmsBatteryGroup());
                dataGroup.setDeviceId(groupDeviceId);
                dataGroup.setSiteId(siteId);
                dataGroup.setCreateBy("system");
                dataGroup.setCreateTime(DateUtils.getNowDate());
                dataGroup.setUpdateBy("system");
                dataGroup.setUpdateTime(DateUtils.getNowDate());

                setPropertyValue(dataGroup, property, entry.getValue());
                groupMap.put(groupDeviceId, dataGroup);
            }
        }

        // 批量插入数据库
        if (org.apache.commons.collections4.MapUtils.isNotEmpty(groupMap)) {
            List<EmsBatteryGroup> batteryGroupList = new ArrayList<>(groupMap.values());
            emsBatteryGroupMapper.batchInsertGroupData(batteryGroupList);
            redisCache.setCacheObject(RedisKeyConstants.GROUP + siteId + "_" + deviceId, batteryGroupList);
        }

    }

    private void batteryDataProcessFromBmsd(String siteId, String deviceId, String dataJson, Date dataUpdateTime) {
//        if (!SiteEnum.DDS.getCode().equals(siteId)) {
//            return;
//        }
        //电池组
        Map<String, Object> obj = JSON.parseObject(dataJson, new TypeReference<Map<String, Object>>() {
        });
        List<EmsBatteryData> dataList = new ArrayList<>();

        // 前一个小时
        LocalDateTime oneHourAgo = LocalDateTime.now().minus(1, ChronoUnit.HOURS);
        DateTimeFormatter formatter = DateTimeFormatter.ofPattern("yyyy-MM-dd HH:mm:ss");
        String oneHourAgoStr = oneHourAgo.format(formatter);

        Map<String, EmsBatteryData> dataMap = new HashMap<>();
        Map<String, EmsBatteryDataDailyLatest> dailyMap = new HashMap<>();
        Map<String, EmsBatteryDataMinutes> minutesMap = new HashMap<>();

        String clusterId = getClusterDeviceIdByParentDeviceId(siteId, deviceId);
        for (Map.Entry<String, Object> entry : obj.entrySet()) {
            String key = entry.getKey();

            if (key.startsWith("DTDC")) {
                Matcher matcher = DTDC_PATTERN.matcher(key);
                if (matcher.matches()) {
                    String batteryCellId = matcher.group(1);
                    String property = matcher.group(2);

                    EmsBatteryData data = dataMap.getOrDefault(batteryCellId, new EmsBatteryData());
                    if (StringUtils.isNotEmpty(batteryCellId)) {
                        data.setDataTimestamp(dataUpdateTime);
                        data.setBatteryPack(deviceId);
                        data.setBatteryCluster(clusterId);
                        data.setClusterDeviceId(clusterId);
                        data.setBatteryCellId(batteryCellId);
                        data.setSiteId(siteId);
                        data.setDeviceId(batteryCellId);
                        data.setCreateBy("system");
                        data.setCreateTime(DateUtils.getNowDate());
                        data.setUpdateBy("system");
                        data.setUpdateTime(DateUtils.getNowDate());
                    }
                    // 根据后缀设置对应属性
                    setDTDCPropertyValue(data, property, entry.getValue());
                    dataMap.put(batteryCellId, data);

                    // 每日最新数据：按batteryCellId去重
                    EmsBatteryDataDailyLatest daily = dailyMap.getOrDefault(batteryCellId, new EmsBatteryDataDailyLatest());
                    BeanUtils.copyProperties(data, daily);
                    daily.setDateDay(DateUtils.getNowDate());
                    dailyMap.put(batteryCellId, daily);

                    // 分钟级的表，上报的数据直接存入，数据上限是 1 个小时
                    EmsBatteryDataMinutes minutes = minutesMap.getOrDefault(batteryCellId, new EmsBatteryDataMinutes());
                    BeanUtils.copyProperties(data, minutes);
                    minutesMap.put(batteryCellId, minutes);
                }
            }
        }
        if (!CollectionUtils.sizeIsEmpty(dataMap)) {
            dataList = new ArrayList<>(dataMap.values());
            emsBatteryDataMapper.insertEmsBatteryDataList(new ArrayList<>(dataList));

            redisCache.deleteList(RedisKeyConstants.BATTERY + siteId + "_" + clusterId);
            redisCache.setCacheList(RedisKeyConstants.BATTERY + siteId + "_" + clusterId, dataList);
        }
        // 批量处理每日最新数据
        List<EmsBatteryDataDailyLatest> dailyList = new ArrayList<>(dailyMap.values());
        if (!dailyList.isEmpty()) {
            dailyList = new ArrayList<>(dailyMap.values());
            emsBatteryDailyLatestServiceImpl.batchProcessBatteryData(dailyList);
        }

        // 实时插入每分钟数据
        List<EmsBatteryDataMinutes> minutesList = new ArrayList<>(minutesMap.values());
        if (!minutesList.isEmpty()) {
            emsBatteryDataMinutesMapper.insertMinutesBatteryDataList(minutesList);
        }
        // 清理分钟级表里一小时前数据
        emsBatteryDataMinutesMapper.deleteByTimeBeforeOneHour(oneHourAgoStr);

        // 分片处理时级，天级，月级数据
        if (CollectionUtils.isNotEmpty(dataList)) {
            super.processBatch(dataList);
        }
    }

    private String getClusterDeviceIdByParentDeviceId(String siteId, String deviceId) {
        String clusterId = "BMSC01";
        Map<String, List<EmsDevicesSetting>> map = redisCache.getCacheObject(RedisKeyConstants.INIT_DEVICE_INFO);
        if (map == null || map.isEmpty()) {
            map = iEmsDeviceSettingService.initDeviceInfo();
        }
        // 不为空，则查找子类的簇id
        if (map != null && !map.isEmpty()) {
            List<EmsDevicesSetting> list = map.get(siteId);
            if (CollectionUtils.isNotEmpty(list)) {
                for (EmsDevicesSetting emsDevicesSetting : list) {
                    if (deviceId.equals(emsDevicesSetting.getParentId())) {
                        clusterId = emsDevicesSetting.getDeviceId();
                    }
                }
            }
        }
        return clusterId;
    }

    private void setDTDCPropertyValue(EmsBatteryData data, String property, Object value) {
        BigDecimal numberValue = null;
        if (value instanceof Number) {
            numberValue = new BigDecimal(value.toString());
        }
        switch (property) {
            case "DY":
                data.setVoltage(numberValue);
                break;
            case "WD":
                data.setTemperature(numberValue);
                break;
            case "NZ":
                data.setInterResistance(numberValue);
                break;
        }
    }

    //根据属性名设置对应的值
    private static void setPropertyValue(EmsBatteryGroup groupData, String property, Object value) {
        BigDecimal numberValue = null;
        if (value instanceof Number) {
            numberValue = new BigDecimal(value.toString());
        }

        switch (property) {
            case "ZT":
                if (numberValue != null) {
                    groupData.setStatus(numberValue.toString());
                }
                break;
            case "SOC":
                groupData.setSoc(numberValue);
                break;
            case "SOH":
                groupData.setSoh(numberValue);
                break;
            case "DL":
                groupData.setCurrent(numberValue);
                break;
            case "DY":
                groupData.setVoltage(numberValue);
                break;
            case "BDSC":
                groupData.setEstimatedBackupDuration(numberValue);
                break;
        }
    }

    /**
     *
     * @param pointMatchList 字段匹配规则列表
     * @param obj 原始数据Map
     * @param entity 目标Java对象
     */
    private void saveDeviceData(List<EmsPointMatch> pointMatchList, Map<String, Object> obj, Object entity) {
        if (null == obj || obj.isEmpty()) {
            return;
        }
        Map<String, String> pointMatchMap = pointMatchList.stream()
                .filter(data -> StringUtils.isNotEmpty(data.getDataPoint()))
                .collect(Collectors.toMap(
                        data -> StringUtils.toCamelCase(data.getMatchField()), // 源字段名转为驼峰
                        EmsPointMatch::getDataPoint, // 获取目标点位名
                        (existing, replacement) -> replacement)); // 处理重复键
        Field[] fields = entity.getClass().getDeclaredFields();
        for (Field field : fields) {
            String fieldName = field.getName();
            if (pointMatchMap.containsKey(fieldName)) {
                // 处理匹配的字段名，并设置值
                field.setAccessible(true); // 允许访问私有字段
                try {
                    // 1. 从原始数据中获取匹配值
                    Object matchValue = obj.get(pointMatchMap.get(fieldName));

                    // 2. 类型转换
                    Class<?> fieldType = field.getType();
                    if (String.class.equals(fieldType)) {
                        matchValue = StringUtils.getString(matchValue);
                    } else if (Long.class.equals(fieldType)) {
                        matchValue = StringUtils.getLong(matchValue);
                    } else if (BigDecimal.class.equals(fieldType)) {
                        matchValue = StringUtils.getBigDecimal(matchValue);
                    } else if (Integer.class.equals(fieldType)) {
                        matchValue = MapUtils.getInteger(obj, pointMatchMap.get(fieldName));
                    }

                    // 4. 设置字段值
                    field.set(entity, matchValue);
                } catch (IllegalAccessException e) {
                    log.warn("deviceDataProcess 设置字段值时出错", e);
                }
            }
        }
    }

    private void batteryClusterDataProcess(String siteId, String deviceId, String dataJson, Date dataUpdateTime) {

        Map<String, Object> obj = JSON.parseObject(dataJson, new TypeReference<Map<String, Object>>() {
        });
        // 点位匹配数据
        List<EmsPointMatch> pointMatchList = devicePointMatchDataProcessor.getDevicePointMatch(siteId, deviceId, DeviceMatchTable.CLUSTER.getCode());
        if (CollectionUtils.isEmpty(pointMatchList)) {
            log.info("未找到匹配的点位数据，无法处理电池簇数据，siteId: " + siteId + "，deviceId: " + deviceId);
            return;
        }
       String stackDeviceId = getStackDeviceId(deviceId);
        //BMS 电池簇
        EmsBatteryCluster data = new EmsBatteryCluster();

        saveDeviceData(pointMatchList, obj, data);

        data.setDataUpdateTime(dataUpdateTime);
//        data.setWorkStatus(WorkStatus.NORMAL.getCode()); // 或其他默认值
        data.setPcsCommunicationStatus(CommunicationStatus.OK.getCode());
        data.setEmsCommunicationStatus(CommunicationStatus.OK.getCode());
        data.setCreateBy("system");
        data.setCreateTime(DateUtils.getNowDate());
        data.setUpdateBy("system");
        data.setUpdateTime(DateUtils.getNowDate());
        data.setSiteId(siteId);
        data.setDeviceId(deviceId);
        if (StringUtils.isNotBlank(stackDeviceId)) {
            data.setStackDeviceId(stackDeviceId);
        } else {
            data.setStackDeviceId("1");
        }
        emsBatteryClusterMapper.insertEmsBatteryCluster(data);

        redisCache.setCacheObject(RedisKeyConstants.CLUSTER + siteId + "_" + deviceId, data);

        // 同步更新设备工作状态
        saveDeviceWorkStatus(siteId, deviceId, DeviceMatchTable.CLUSTER.getCode(), data.getWorkStatus());
    }

    private String getStackDeviceId(String deviceId) {
        EmsDevicesSetting joken = new EmsDevicesSetting();
        joken.setDeviceId(deviceId);
        List<EmsDevicesSetting> up = emsDevicesSettingMapper.selectEmsDevicesSettingList(joken);
        String stackDeviceId = "";
        if (CollectionUtils.isNotEmpty(up)) {
            stackDeviceId = up.get(0).getParentId();
            if (StringUtils.isNotEmpty(stackDeviceId)) {
                stackDeviceId = "1";
            }
        }

        return stackDeviceId;
    }

    private void batteryDataProcessFromBmsc(String siteId, String deviceId, String dataJson, Date dataUpdateTime) {

        //单体电池
        Map<String, Map<String, Object>> records = processData(JSON.parseObject(dataJson, new TypeReference<Map<String, Object>>() {
        }));

        log.info("站点：{}，单体电池数据：{}", siteId, records);
        List<EmsBatteryData> list = new ArrayList<>();
        List<EmsBatteryDataDailyLatest> dailyList = new ArrayList<>();
        List<EmsBatteryDataMinutes> minutesList = new ArrayList<>();
        // 前一个小时
        LocalDateTime oneHourAgo = LocalDateTime.now().minus(1, ChronoUnit.HOURS);
        DateTimeFormatter formatter = DateTimeFormatter.ofPattern("yyyy-MM-dd HH:mm:ss");
        String oneHourAgoStr = oneHourAgo.format(formatter);
        //单体电池
        for (Map.Entry<String, Map<String, Object>> record : records.entrySet()) {
            String recordId = record.getKey();
            Map<String, Object> fields = record.getValue();
            String batteryDeviceId = recordId + deviceId;
            // 点位匹配数据
            List<EmsPointMatch> pointMatchList = devicePointMatchDataProcessor.getDevicePointMatch(siteId, batteryDeviceId, DeviceMatchTable.BATTERY.getCode());

            if (CollectionUtils.isEmpty(pointMatchList)) {
                pointMatchList = devicePointMatchDataProcessor.getDeviceDefaultPointMatch(siteId, DeviceMatchTable.BATTERY.getCode());
            } else {
                Map<String, Object> newFields = new HashMap<>();
                for (Map.Entry<String, Object> entry : fields.entrySet()) {
                    newFields.put(entry.getKey() + recordId, entry.getValue());
                }
                fields = newFields;
            }

            if (CollectionUtils.isEmpty(pointMatchList)) {
                log.info("未找到匹配的点位数据，无法处理单体电池数据，siteId: " + siteId + "，deviceId: " + batteryDeviceId);
                return;
            }

            EmsBatteryData batteryData = new EmsBatteryData();
            //TODO 单体电池 deviceid 拼接簇的 deviceid
            batteryData.setDeviceId(recordId+deviceId);
            batteryData.setBatteryCellId(recordId);

            saveDeviceData(pointMatchList, fields, batteryData);

            batteryData.setBatteryCluster(deviceId);
            batteryData.setBatteryPack(getStackDeviceId(deviceId));

            // 时间戳
            batteryData.setDataTimestamp(dataUpdateTime);
            // 系统管理字段
            batteryData.setCreateBy("system");
            batteryData.setCreateTime(DateUtils.getNowDate());
            batteryData.setUpdateBy("system");
            batteryData.setUpdateTime(DateUtils.getNowDate());
            // ID字段
            batteryData.setSiteId(siteId);
            batteryData.setClusterDeviceId(deviceId);
            list.add(batteryData);

            // 每日最新数据
            EmsBatteryDataDailyLatest daily = new EmsBatteryDataDailyLatest();
            BeanUtils.copyProperties(batteryData, daily);
            daily.setDateDay(DateUtils.getNowDate());
            dailyList.add(daily);

            // 分钟级的表，上报的数据直接存入，数据上限是 1 个小时
            EmsBatteryDataMinutes minutes = new EmsBatteryDataMinutes();
            BeanUtils.copyProperties(batteryData, minutes);
            minutesList.add(minutes);
        }

        if (CollectionUtils.isNotEmpty(list)) {
            emsBatteryDataMapper.insertEmsBatteryDataList(list);

            redisCache.deleteList(RedisKeyConstants.BATTERY + siteId + "_" + deviceId);
            redisCache.setCacheList(RedisKeyConstants.BATTERY + siteId + "_" + deviceId, list);

        }

        // 批量处理每日最新数据
        if (CollectionUtils.isNotEmpty(dailyList)) {
            emsBatteryDailyLatestServiceImpl.batchProcessBatteryData(dailyList);
        }
        // 实时插入每分钟数据
        if (CollectionUtils.isNotEmpty(minutesList)) {
            emsBatteryDataMinutesMapper.insertMinutesBatteryDataList(minutesList);
        }
        // 清理分钟级表里一小时前数据
        emsBatteryDataMinutesMapper.deleteByTimeBeforeOneHour(oneHourAgoStr);

        // 分片处理时级，天级，月级数据
        if (CollectionUtils.isNotEmpty(list)) {
            super.processBatch(list);
        }
    }

    private void pcsDataProcess(String siteId, String deviceId, String dataJson, Date dataUpdateTime) {
        Map<String, Object> obj = JSON.parseObject(dataJson, new TypeReference<Map<String, Object>>() {
        });

        // 点位匹配数据
        List<EmsPointMatch> pointMatchList = devicePointMatchDataProcessor.getDevicePointMatch(siteId, deviceId, DeviceMatchTable.PCS.getCode());
        if (CollectionUtils.isEmpty(pointMatchList)) {
            log.info("未找到匹配的点位数据，无法处理PCS数据，siteId: " + siteId + "，deviceId: " + deviceId);
            return;
        }

        //pcs
        EmsPcsData pcsData = new EmsPcsData();

        saveDeviceData(pointMatchList, obj, pcsData);

        // 状态指示类
        pcsData.setBranchStatus(BranchStatus.NORMAL.getCode());
        pcsData.setDischargeStatus(ChargeStatus.CHARGING.getCode());
        pcsData.setAcSwitchStatus(SwitchStatus.CLOSED.getCode());
        pcsData.setDcSwitchStatus(SwitchStatus.CLOSED.getCode());
        pcsData.setRemoteControlStatus(ControlModeStatus.REMOTE.getCode());

        // 时间与状态类字段
        pcsData.setDataUpdateTime(dataUpdateTime);
//        pcsData.setWorkStatus(WorkStatus.NORMAL.getCode());
        pcsData.setGridStatus(GridStatus.GRID.getCode());
        pcsData.setDeviceStatus(DeviceRunningStatus.ONLINE.getCode());
        pcsData.setControlMode(ControlModeStatus.REMOTE.getCode());

        // 系统管理字段
        pcsData.setCreateBy("system");
        pcsData.setCreateTime(DateUtils.getNowDate());
        pcsData.setUpdateBy("system");
        pcsData.setUpdateTime(DateUtils.getNowDate());
        pcsData.setSiteId(siteId);
        pcsData.setDeviceId(deviceId);
        pcsData.setDateMonth(DateUtils.getNowMonthLong());
        pcsData.setDateDay(DateUtils.getNowDayLong());

        //TODO 临时解决设备状态不展示数据问题
        pcsData.setDeviceStatus(DeviceRunningStatus.ONLINE.getCode());
        //TODO end

        //TODO 临时解决上送数据为浮点的问题 start
        String status = pcsData.getWorkStatus();
        int result = 0;
        if (status != null && !status.trim().isEmpty() && status.contains("null")) {
            // 1.0 -> 1, 2.0 -> 2
            result = (int) Double.parseDouble(status.trim());
        }
        pcsData.setWorkStatus(String.valueOf(result));
        //TODO end


        emsPcsDataMapper.insertEmsPcsData(pcsData);
        redisCache.setCacheObject(RedisKeyConstants.PCS + siteId + "_" + deviceId, pcsData);

        // 同步更新PCS设备工作状态
        saveDeviceWorkStatus(siteId, deviceId, DeviceMatchTable.PCS.getCode(), pcsData.getWorkStatus());

//        if (SiteEnum.FX.getCode().equals(siteId)) {
//            //更新每日充放电数据
//            dealFXXDailyChargeDate(siteId, deviceId, pcsData);
//        }
    }

    private void pcsBranchDataProcess(String siteId, String deviceId, String dataJson) {

        Map<String, Map<String, Object>> records = processDataPrefix(JSON.parseObject(dataJson, new TypeReference<Map<String, Object>>() {
        }));
        List<EmsPcsBranchData> list = new ArrayList<>();
        // 点位匹配数据
        List<EmsPointMatch> pointMatchList = devicePointMatchDataProcessor.getDevicePointMatch(siteId, deviceId, DeviceMatchTable.BRANCH.getCode());
        if (CollectionUtils.isEmpty(pointMatchList)) {
            log.info("未找到匹配的点位数据，无法处理PCS支路数据，siteId: " + siteId + "，deviceId: " + deviceId);
            return;
        }

        //PCS支路
        for (Map.Entry<String, Map<String, Object>> record : records.entrySet()) {
            String recordId = record.getKey();
            Map<String, Object> fields = record.getValue();

            EmsPcsBranchData data = new EmsPcsBranchData();
            data.setDeviceId(deviceId);
            data.setSiteId(siteId);
            data.setGridStatus(GridStatus.GRID.getCode());

            saveDeviceData(pointMatchList, fields, data);

            data.setBranchId(recordId);
            data.setCreateBy("system");
            data.setCreateTime(DateUtils.getNowDate());
            data.setUpdateBy("system");
            data.setUpdateTime(DateUtils.getNowDate());
            list.add(data);
        }
        if (CollectionUtils.isNotEmpty(list)) {
            emsPcsBranchDataMapper.insertPcsBranchDataList(list);

            redisCache.setCacheObject(RedisKeyConstants.BRANCH + siteId + "_" + deviceId, list);

        }

    }

    private void loadDataProcess(String siteId, String deviceId, String dataJson, Date dataUpdateTime) {

        //总表
        Map<String, Object> obj = JSON.parseObject(dataJson, new TypeReference<Map<String, Object>>() {
        });
        // 点位匹配数据
        List<EmsPointMatch> pointMatchList = devicePointMatchDataProcessor.getDevicePointMatch(siteId, deviceId, DeviceMatchTable.AMMETER.getCode());
        if (CollectionUtils.isEmpty(pointMatchList)) {
            pointMatchList = devicePointMatchDataProcessor.getDeviceDefaultPointMatch(siteId, DeviceMatchTable.AMMETER.getCode());
        }
        if (CollectionUtils.isEmpty(pointMatchList)) {
            log.info("未找到匹配的点位数据，无法处理LOAD总表数据，siteId: " + siteId + "，deviceId: " + deviceId);
            return;
        }

        EmsAmmeterData dataLoad = new EmsAmmeterData();
        // 更新时间
        dataLoad.setDataUpdateTime(dataUpdateTime);

        saveDeviceData(pointMatchList, obj, dataLoad);

        dataLoad.setCreateBy("system");
        dataLoad.setCreateTime(DateUtils.getNowDate());
        dataLoad.setUpdateBy("system");
        dataLoad.setUpdateTime(DateUtils.getNowDate());
        dataLoad.setSiteId(siteId);
        dataLoad.setDeviceId(deviceId);

        emsAmmeterDataMapper.insertEmsAmmeterData(dataLoad);

        redisCache.setCacheObject(RedisKeyConstants.AMMETER + siteId + "_" + deviceId, dataLoad);

    }

    private void batteryClusterDataProcess(String siteId, String dataJson, Date dataUpdateTime) {
//        if (!SiteEnum.DDS.getCode().equals(siteId)) {
//            return;
//        }

        Map<String, Object> obj = JSON.parseObject(dataJson, new TypeReference<Map<String, Object>>() {
        });

        List<String> deviceIds = Arrays.asList("BMSC02", "BMSC01");
        for (String deviceId : deviceIds) {
            // 点位匹配数据
            List<EmsPointMatch> pointMatchList = devicePointMatchDataProcessor.getDevicePointMatch(siteId, deviceId, DeviceMatchTable.CLUSTER.getCode());
            if (CollectionUtils.isEmpty(pointMatchList)) {
                log.info("未找到匹配的点位数据，无法处理电池簇数据，siteId: " + siteId + "，deviceId: " + deviceId);
                return;
            }
            //BMSC 电池簇
            EmsBatteryCluster data = new EmsBatteryCluster();
            // 其他非 BigDecimal 字段
            data.setDataUpdateTime(dataUpdateTime);
//            data.setWorkStatus(WorkStatus.NORMAL.getCode()); // 或其他默认值
            data.setPcsCommunicationStatus(CommunicationStatus.OK.getCode());
            data.setEmsCommunicationStatus(CommunicationStatus.OK.getCode());
            data.setCreateBy("system");
            data.setCreateTime(DateUtils.getNowDate());
            data.setUpdateBy("system");
            data.setUpdateTime(DateUtils.getNowDate());
            data.setSiteId(siteId);
            String stackDeviceId = getDeviceParent(siteId, deviceId);
            if (StringUtils.isNotBlank(stackDeviceId)) {
                data.setStackDeviceId(stackDeviceId);
            } else {
                data.setStackDeviceId("1");
            }


            // 获取redis获取最新的BMSD数据
            Map<String, Object> stackObj = redisCache.getCacheObject(RedisKeyConstants.ORIGINAL_BMSD + siteId + "_" + stackDeviceId);
            data.setDeviceId(deviceId);
            saveDeviceData(pointMatchList, obj, data);

            // 取堆里面数据
            saveDeviceData(pointMatchList, stackObj, data);


            //TODO 临时解决上送数据为浮点的问题 start
            log.info("临时解决上送数据为浮点的问题");
            String status = data.getWorkStatus();
            int result = 5;
            if (status != null && !status.trim().isEmpty() && status.contains("null")) {
                // 1.0 -> 1, 2.0 -> 2
                result = (int) Double.parseDouble(status.trim());
            }
            data.setWorkStatus(String.valueOf(result));
            //TODO end


            emsBatteryClusterMapper.insertEmsBatteryCluster(data);
            redisCache.setCacheObject(RedisKeyConstants.CLUSTER + siteId + "_" + deviceId, data);

            // 同步更新设备工作状态
            saveDeviceWorkStatus(siteId, deviceId, DeviceMatchTable.CLUSTER.getCode(), data.getWorkStatus());

        }

    }

    private void saveDeviceWorkStatus(String siteId, String deviceId, String deviceCategory, String workStatus) {
        if (StringUtils.isEmpty(workStatus)) {
            return;
        }
//        Map<String, List<EmsPointEnumMatch>> pointEnumMatchMap = devicePointMatchDataProcessor.getPointEnumMatchMap(siteId, deviceCategory);
//        // 枚举值转换
//        List<EmsPointEnumMatch> pointEnumMatchList = pointEnumMatchMap.get("workStatus");
//        if (CollectionUtils.isNotEmpty(pointEnumMatchList) && workStatus != null) {
//            String finalMatchValue = workStatus.replace(".0", "");
//            Optional<EmsPointEnumMatch> enumMatch = pointEnumMatchList.stream()
//                    .filter(data -> data.getDataEnumCode().equals(finalMatchValue)).findFirst();
//            if (enumMatch.isPresent()) {
//                workStatus = enumMatch.get().getEnumCode();
//            }
//        }
        EmsDevicesSetting emsDevicesSetting = getDeviceSettingWithCache(siteId, deviceId);
        if (emsDevicesSetting == null) {
            return;
        }
        emsDevicesSetting.setWorkStatus(workStatus);
        emsDevicesSetting.setUpdatedAt(DateUtils.getNowDate());
        emsDevicesSettingMapper.updateEmsDevicesSetting(emsDevicesSetting);
        cacheDeviceSetting(siteId, deviceId, emsDevicesSetting);
    }

    private EmsDevicesSetting getDeviceSettingWithCache(String siteId, String deviceId) {
        if (StringUtils.isAnyBlank(siteId, deviceId)) {
            return null;
        }
        String cacheKey = RedisKeyConstants.DEVICE_SETTING + siteId + "_" + deviceId;
        EmsDevicesSetting cached = redisCache.getCacheObject(cacheKey);
        if (cached != null) {
            return cached;
        }
        EmsDevicesSetting fromInitCache = getDeviceSettingFromInitCache(siteId, deviceId);
        if (fromInitCache != null) {
            cacheDeviceSetting(siteId, deviceId, fromInitCache);
            return fromInitCache;
        }
        EmsDevicesSetting fromDb = emsDevicesSettingMapper.getDeviceBySiteAndDeviceId(deviceId, siteId);
        if (fromDb != null) {
            cacheDeviceSetting(siteId, deviceId, fromDb);
        }
        return fromDb;
    }

    private EmsDevicesSetting getDeviceSettingFromInitCache(String siteId, String deviceId) {
        Map<String, List<EmsDevicesSetting>> map = redisCache.getCacheObject(RedisKeyConstants.INIT_DEVICE_INFO);
        if (map == null || map.isEmpty()) {
            return null;
        }
        List<EmsDevicesSetting> list = map.get(siteId);
        if (CollectionUtils.isEmpty(list)) {
            return null;
        }
        for (EmsDevicesSetting item : list) {
            if (item != null && deviceId.equals(item.getDeviceId())) {
                return item;
            }
        }
        return null;
    }

    private void cacheDeviceSetting(String siteId, String deviceId, EmsDevicesSetting setting) {
        if (StringUtils.isAnyBlank(siteId, deviceId) || setting == null) {
            return;
        }
        redisCache.setCacheObject(RedisKeyConstants.DEVICE_SETTING + siteId + "_" + deviceId, setting, 10, TimeUnit.MINUTES);
    }

    private String getDeviceParent(String siteId, String deviceId) {
        Map<String, List<EmsDevicesSetting>> map = redisCache.getCacheObject(RedisKeyConstants.INIT_DEVICE_INFO);
        if (map == null || map.isEmpty()) {
            map = iEmsDeviceSettingService.initDeviceInfo();
        }
        // 不为空，则查找父类
        String stackDeviceId = "1";
        if (map != null && !map.isEmpty()) {
            List<EmsDevicesSetting> list = map.get(siteId);
            if (list == null || list.isEmpty()) {
                EmsDevicesSetting deviceInfo = new EmsDevicesSetting();
                deviceInfo.setDeviceId(deviceId);
                deviceInfo.setSiteId(siteId);
                list = emsDevicesSettingMapper.selectEmsDevicesSettingList(deviceInfo);
                if (list == null || list.isEmpty()) {
                    return stackDeviceId;
                }
            }
            for (EmsDevicesSetting emsDevicesSetting : list) {
                if (deviceId.equals(emsDevicesSetting.getDeviceId())) {
                    stackDeviceId = emsDevicesSetting.getParentId();
                }
            }
        }
        return stackDeviceId;
    }

    private void dealFXXDailyChargeDate(String siteId, String deviceId, EmsPcsData pcsData) {
        log.info("start dealFXXDailyChargeDate");
        //日充放电数据
        if (pcsData == null) {
            log.info("日充放电数据为空, deviceId: " + deviceId);
            return;
        }

        // 初始化当日数据
        EmsDailyChargeData emsDailyChargeData = new EmsDailyChargeData();
        emsDailyChargeData.setSiteId(siteId);
        emsDailyChargeData.setDateTime(DateUtils.getNowDate());
        emsDailyChargeData.setTotalChargeData(pcsData.getTotalAcChargeEnergy());
        emsDailyChargeData.setTotalDischargeData(pcsData.getTotalAcDischargeEnergy());
        emsDailyChargeData.setChargeData(pcsData.getDailyAcChargeEnergy());
        emsDailyChargeData.setDischargeData(pcsData.getDailyAcDischargeEnergy());
        emsDailyChargeData.setCreateBy("system");
        emsDailyChargeData.setCreateTime(DateUtils.getNowDate());
        emsDailyChargeData.setUpdateBy("system");
        emsDailyChargeData.setUpdateTime(DateUtils.getNowDate());
        // 插入或更新每日充放电数据表
        emsDailyChargeDataMapper.insertOrUpdateData(emsDailyChargeData);
        log.info("end dealFXXDailyChargeDate");
    }

    private void meteDataProcess(String siteId, String deviceId, String dataJson, Date dataUpdateTime) {

        //总表
        Map<String, Object> obj = JSON.parseObject(dataJson, new TypeReference<Map<String, Object>>() {
        });
        // 点位匹配数据
        List<EmsPointMatch> pointMatchList = devicePointMatchDataProcessor.getDevicePointMatch(siteId, deviceId, DeviceMatchTable.AMMETER.getCode());
        if (CollectionUtils.isEmpty(pointMatchList)) {
            pointMatchList = devicePointMatchDataProcessor.getDeviceDefaultPointMatch(siteId, DeviceMatchTable.AMMETER.getCode());
        }
        if (CollectionUtils.isEmpty(pointMatchList)) {
            log.info("未找到匹配的点位数据，无法处理电表数据，siteId: " + siteId + "，deviceId: " + deviceId);
            return;
        }

        EmsAmmeterData dataMete = new EmsAmmeterData();
        // 更新时间
        dataMete.setDataUpdateTime(dataUpdateTime);

        saveDeviceData(pointMatchList, obj, dataMete);

        dataMete.setCreateBy("system");
        dataMete.setCreateTime(DateUtils.getNowDate());
        dataMete.setUpdateBy("system");
        dataMete.setUpdateTime(DateUtils.getNowDate());
        dataMete.setSiteId(siteId);
        dataMete.setDeviceId(deviceId);

        emsAmmeterDataMapper.insertEmsAmmeterData(dataMete);

        redisCache.setCacheObject(RedisKeyConstants.AMMETER + siteId + "_" + deviceId, dataMete);

        if (deviceId.equals(SiteDevice.METE.name())) {
            // 处理储能电表-METE每日充放电数据
            dealDailyChargeDate(siteId, deviceId, dataMete);

            // ems_daily_energy_data 仅保留页面修正与Quartz写入
        }
    }

    private EmsDailyChargeData initDailyChargeData(String siteId, String deviceId, BigDecimal nowTotalChargeDate,
                                                   BigDecimal nowTotalDisChargeDate) {
        EmsDailyChargeData emsDailyChargeData = new EmsDailyChargeData();
        emsDailyChargeData.setSiteId(siteId);
        emsDailyChargeData.setDateTime(DateUtils.getNowDate());
        emsDailyChargeData.setTotalChargeData(nowTotalChargeDate);
        emsDailyChargeData.setTotalDischargeData(nowTotalDisChargeDate);
        emsDailyChargeData.setCreateBy("system");
        emsDailyChargeData.setCreateTime(DateUtils.getNowDate());
        emsDailyChargeData.setUpdateBy("system");
        emsDailyChargeData.setUpdateTime(DateUtils.getNowDate());
        return emsDailyChargeData;
    }


    private EmsAmmeterData dealDailyChargeDate(String siteId, String deviceId, EmsAmmeterData currentData) {
        // 初始化今日充放电
        BigDecimal dailyDisChargeDate = new BigDecimal(0);
        BigDecimal dailyChargeDate = new BigDecimal(0);

        BigDecimal nowTotalDisChargeDate = currentData.getCurrentReverseActiveTotal();
        BigDecimal nowTotalChargeDate = currentData.getCurrentForwardActiveTotal();
        // 初始化当日数据-总的
        EmsDailyChargeData emsDailyChargeData = initDailyChargeData(siteId, deviceId, nowTotalChargeDate, nowTotalDisChargeDate);
        // 获取redis存放昨日最晚数据
        String yestDate = DateUtils.getYesterdayDate();
        String yestDateRedisKey = RedisKeyConstants.AMMETER + siteId + "_" + deviceId + "_" + yestDate;
        EmsAmmeterData yestData = redisCache.getCacheObject(yestDateRedisKey);
        if (yestData == null) {
            // redis没有这查电表总数据表取截止到昨日最新第一条数据
            yestData = emsAmmeterDataMapper.getYestLatestDate(siteId, deviceId, yestDate);
            // 数据存redis-有效期1天
            redisCache.setCacheObject(yestDateRedisKey, yestData, Constants.DATE_VALID_TIME, TimeUnit.DAYS);
        }
        if (yestData != null) {
            // 今日总数据-昨日总数据=今日充放电
            BigDecimal yestTotalDisChargeDate = yestData.getCurrentReverseActiveTotal();
            BigDecimal yestTotalChargeDate = yestData.getCurrentForwardActiveTotal();

            dailyChargeDate = nowTotalChargeDate.subtract(yestTotalChargeDate);
            dailyDisChargeDate = nowTotalDisChargeDate.subtract(yestTotalDisChargeDate);
            emsDailyChargeData.setChargeData(dailyChargeDate);
            emsDailyChargeData.setDischargeData(dailyDisChargeDate);
        }

        // 插入或更新每日充放电数据表
        emsDailyChargeDataMapper.insertOrUpdateData(emsDailyChargeData);

        return yestData;
    }

    private BigDecimal getYestLastData(String siteId) {
        // dds存的是累计到昨日总收益
        String yestDate = DateUtils.getYesterdayDayString();
        String redisKey = RedisKeyConstants.DDS_TOTAL_REVENUE + siteId + "_" + yestDate;
        Object cacheObject = redisCache.getCacheObject(redisKey);
        BigDecimal yestLastTotalRevenue;
        if (cacheObject == null) {
            yestLastTotalRevenue = emsDailyEnergyDataMapper.getLastTotalRevenue(siteId);
            if (yestLastTotalRevenue == null) {
                yestLastTotalRevenue = BigDecimal.ZERO;
            }
            redisCache.setCacheObject(redisKey, yestLastTotalRevenue, 1, TimeUnit.DAYS);
        } else {
            log.info("从redis获取昨日总收益cacheObject:{}", cacheObject);
            yestLastTotalRevenue = new BigDecimal(cacheObject.toString());
        }
        return yestLastTotalRevenue;
    }

    private void syncDailyRevenueToChargeData(String siteId, Date revenueDate, BigDecimal totalRevenue, BigDecimal dayRevenue) {
        if (StringUtils.isBlank(siteId)) {
            return;
        }
        Date persistDate = revenueDate == null ? DateUtils.getNowDate() : revenueDate;
        BigDecimal persistTotalRevenue = safeBigDecimal(totalRevenue);
        BigDecimal persistDayRevenue = safeBigDecimal(dayRevenue);

        int updated = emsDailyChargeDataMapper.updateRevenueBySiteAndDate(siteId, persistDate, persistTotalRevenue, persistDayRevenue, "system");
        if (updated > 0) {
            return;
        }

        EmsDailyChargeData existedData = emsDailyChargeDataMapper.selectBySiteIdAndDateTime(siteId, persistDate);
        if (existedData != null) {
            existedData.setTotalRevenue(persistTotalRevenue);
            existedData.setDayRevenue(persistDayRevenue);
            existedData.setUpdateBy("system");
            existedData.setUpdateTime(DateUtils.getNowDate());
            emsDailyChargeDataMapper.updateEmsDailyChargeData(existedData);
            return;
        }

        EmsDailyChargeData chargeData = new EmsDailyChargeData();
        chargeData.setSiteId(siteId);
        chargeData.setDateTime(persistDate);
        chargeData.setTotalRevenue(persistTotalRevenue);
        chargeData.setDayRevenue(persistDayRevenue);
        chargeData.setCreateBy("system");
        chargeData.setCreateTime(DateUtils.getNowDate());
        chargeData.setUpdateBy("system");
        chargeData.setUpdateTime(DateUtils.getNowDate());
        emsDailyChargeDataMapper.insertEmsDailyChargeData(chargeData);
    }

    // 数据分组处理
    private static Map<String, Map<String, Object>> processData(Map<String, Object> rawData) {
        Map<String, Map<String, Object>> records = new HashMap<>();

        for (Map.Entry<String, Object> entry : rawData.entrySet()) {
            String key = entry.getKey();
            // 提取记录ID（最后3位）
            String recordId = key.substring(key.length() - 3);
            try {
                Long.parseLong(recordId);
            } catch (Exception e) {
                continue;
            }

            // 提取字段类型（前缀）
            String fieldType = key.substring(0, key.length() - 3);

            // 初始化记录
            records.putIfAbsent(recordId, new HashMap<>());
            // 存入字段值
            records.get(recordId).put(fieldType, entry.getValue());
        }
        return records;
    }

    private static Map<String, Map<String, Object>> processDataPrefix(Map<String, Object> rawData) {
        Map<String, Map<String, Object>> records = new HashMap<>();

        for (Map.Entry<String, Object> entry : rawData.entrySet()) {
            String key = entry.getKey();
            // 确保键长度足够
            if (key.length() < 3) {
                continue;
            }

            // 提取记录ID（前3位）
            String recordId = key.substring(0, 3);
            if (!recordId.startsWith("DY")) {
                continue;
            }

            // 提取字段类型（剩余部分）
            String fieldType = key.substring(3);

            // 初始化记录
            records.putIfAbsent(recordId, new HashMap<>());
            // 存入字段值
            records.get(recordId).put(fieldType, entry.getValue());
        }
        if (records.isEmpty() && rawData.size() > 0) {
            records.put("DY1", rawData);
        }
        return records;
    }

    @Override
    public void handleAlarmData(String message, String siteId) {
        JSONArray arraylist = JSONArray.parseArray(message);

        // 获取redis缓存-告警信息
        Map<String, EmsAlarmMatchData> alarmMatchInfo = redisCache.getCacheObject(RedisKeyConstants.ALARM_MATCH_INFO);

        for (int i = 0; i < arraylist.size(); i++) {
            JSONObject obj = JSONObject.parseObject(arraylist.get(i).toString());

            String deviceId = obj.getString("Device");
            String jsonData = obj.getString("Data");
            Long timestamp = obj.getLong("timestamp");
            Date dataUpdateTime = DateUtils.convertUpdateTime(timestamp);

            log.info("deviceId:" + deviceId);

            // 存放mqtt原始每个设备最晚一次数据，便于后面点位获取数据
            redisCache.setCacheObject(RedisKeyConstants.ORIGINAL_MQTT_DATA_ALARM + siteId + "_" + deviceId, obj);
            // 存放每次同步数据，失效时间（同同步时间）-用于判断是否正常同步数据
            redisCache.setCacheObject(RedisKeyConstants.SYNC_DATA_ALARM + siteId + "_" + deviceId, obj, 1, TimeUnit.MINUTES);
            // 处理设备数据，根据不同设备类型进行不同的数据处理
            String deviceCategory = processingDeviceAlarmData(siteId, deviceId, jsonData, dataUpdateTime);
            if (StringUtils.isEmpty(deviceCategory)) {
                // 处理告警信息
                alarmDataProcess(siteId, deviceId, jsonData, alarmMatchInfo, deviceCategory);
            }
        }
    }

    public String processingDeviceAlarmData(String siteId, String deviceId, String jsonData, Date dataUpdateTime) {
        // 判断设备类型，并调用对应的方法处理数据
        String deviceCategory = getDeviceCategory(siteId, deviceId);
        log.info("alarm siteId: {}, deviceId: {}siteId, category: {} ", siteId, deviceId, deviceCategory);
        if (deviceId.contains(SiteDevice.ZSLQ.name())
                || DeviceCategory.COOLING.getCode().equals(deviceCategory)) {
            coolingAlarmDataProcess(siteId, deviceId, jsonData, dataUpdateTime);
        } else if (deviceId.contains(SiteDevice.BMSD.name())
                || DeviceCategory.STACK.getCode().equals(deviceCategory)) {
            stackAlarmDataProcess(siteId, deviceId, jsonData, dataUpdateTime);
        } else if (deviceId.contains(SiteDevice.BMSC.name())
                || DeviceCategory.CLUSTER.getCode().equals(deviceCategory)) {
            clusterAlarmDataProcess(siteId, deviceId, jsonData, dataUpdateTime);
        } else if (deviceId.contains(SiteDevice.PCS.name())
                || DeviceCategory.PCS.getCode().equals(deviceCategory) ) {
            pcsAlarmDataProcess(siteId, deviceId, jsonData, dataUpdateTime);
        } else {
            log.info("未找到匹配的点位数据，无法处理设备告警数据，siteId: " + siteId + "，deviceId: " + deviceId);
            return null;
        }

        return deviceCategory;
    }

    private void pcsAlarmDataProcess(String siteId, String deviceId, String jsonData, Date dataUpdateTime) {
        //pcs
        Map<String, Object> obj = JSON.parseObject(jsonData, new TypeReference<Map<String, Object>>() {
        });
        // 点位匹配数据
        List<EmsPointMatch> pointMatchList = devicePointMatchDataProcessor.getDevicePointMatch(siteId, deviceId, DeviceMatchTable.PCS.getCode());
        if (CollectionUtils.isEmpty(pointMatchList)) {
            log.info("未找到匹配的点位数据，无法处理中水冷却告警数据，siteId: " + siteId + "，deviceId: " + deviceId);
            return;
        }
        EmsPcsAlarmData pcsAlarmData = new EmsPcsAlarmData();
        // 更新时间
        pcsAlarmData.setDataTimestamp(dataUpdateTime);

        saveDeviceData(pointMatchList, obj, pcsAlarmData);

        pcsAlarmData.setCreateBy("system");
        pcsAlarmData.setCreateTime(DateUtils.getNowDate());
        pcsAlarmData.setUpdateBy("system");
        pcsAlarmData.setUpdateTime(DateUtils.getNowDate());
        pcsAlarmData.setSiteId(siteId);
        pcsAlarmData.setDeviceId(deviceId);
        emsPcsAlarmDataMapper.insertEmsPcsAlarmData(pcsAlarmData);
    }

    private void stackAlarmDataProcess(String siteId, String deviceId, String jsonData, Date dataUpdateTime) {
        // bmsd
        Map<String, Object> obj = JSON.parseObject(jsonData, new TypeReference<Map<String, Object>>() {
        });

        // 点位匹配数据
        List<EmsPointMatch> pointMatchList = devicePointMatchDataProcessor.getDevicePointMatch(siteId, deviceId, DeviceMatchTable.STACK.getCode());
        if (CollectionUtils.isEmpty(pointMatchList)) {
            log.info("未找到匹配的点位数据，无法处理电池堆告警数据，siteId: " + siteId + "，deviceId: " + deviceId);
            return;
        }

        EmsStackAlarmData stackAlarmData = new EmsStackAlarmData();
        // 更新时间
        stackAlarmData.setDataTimestamp(dataUpdateTime);

        saveDeviceData(pointMatchList, obj, stackAlarmData);

        stackAlarmData.setCreateBy("system");
        stackAlarmData.setCreateTime(DateUtils.getNowDate());
        stackAlarmData.setUpdateBy("system");
        stackAlarmData.setUpdateTime(DateUtils.getNowDate());
        stackAlarmData.setSiteId(siteId);
        stackAlarmData.setDeviceId(deviceId);
        emsStackAlarmDataMapper.insertEmsStackAlarmData(stackAlarmData);
    }

    private void clusterAlarmDataProcess(String siteId, String deviceId, String jsonData, Date dataUpdateTime) {
        // bmsc
        Map<String, Object> obj = JSON.parseObject(jsonData, new TypeReference<Map<String, Object>>() {
        });
        // 点位匹配数据
        List<EmsPointMatch> pointMatchList = devicePointMatchDataProcessor.getDevicePointMatch(siteId, deviceId, DeviceMatchTable.CLUSTER.getCode());
        if (CollectionUtils.isEmpty(pointMatchList)) {
            log.info("未找到匹配的点位数据，无法处理电池簇告警数据，siteId: " + siteId + "，deviceId: " + deviceId);
            return;
        }
        EmsClusterAlarmData clusterAlarmData = new EmsClusterAlarmData();
        // 更新时间
        clusterAlarmData.setDataTimestamp(dataUpdateTime);

        saveDeviceData(pointMatchList, obj, clusterAlarmData);

        clusterAlarmData.setCreateBy("system");
        clusterAlarmData.setCreateTime(DateUtils.getNowDate());
        clusterAlarmData.setUpdateBy("system");
        clusterAlarmData.setUpdateTime(DateUtils.getNowDate());
        clusterAlarmData.setSiteId(siteId);
        clusterAlarmData.setDeviceId(deviceId);
        emsClusterAlarmDataMapper.insertEmsClusterAlarmData(clusterAlarmData);
    }

    private void coolingAlarmDataProcess(String siteId, String deviceId, String jsonData, Date dataUpdateTime) {
        //中水冷却
        Map<String, Object> obj = JSON.parseObject(jsonData, new TypeReference<Map<String, Object>>() {
        });

        // 点位匹配数据
        List<EmsPointMatch> pointMatchList = devicePointMatchDataProcessor.getDeviceAlarmPointMatch(siteId, deviceId, DeviceMatchTable.COOLING.getCode());
        if (CollectionUtils.isEmpty(pointMatchList)) {
            log.info("未找到匹配的点位数据，无法处理中水冷却告警数据，siteId: " + siteId + "，deviceId: " + deviceId);
            return;
        }

        EmsCoolingAlarmData coolingAlarmData = new EmsCoolingAlarmData();
        // 更新时间
        coolingAlarmData.setDataTimestamp(dataUpdateTime);

        saveDeviceData(pointMatchList, obj, coolingAlarmData);

        coolingAlarmData.setCreateBy("system");
        coolingAlarmData.setCreateTime(DateUtils.getNowDate());
        coolingAlarmData.setUpdateBy("system");
        coolingAlarmData.setUpdateTime(DateUtils.getNowDate());
        coolingAlarmData.setSiteId(siteId);
        coolingAlarmData.setDeviceId(deviceId);
        emsCoolingAlarmDataMapper.insertEmsCoolingAlarmData(coolingAlarmData);
    }

    private void alarmDataProcess(String siteId, String deviceId, String jsonData,
                                  Map<String, EmsAlarmMatchData> alarmInfoData, String category) {
        Map<String, Object> obj = JSON.parseObject(jsonData, new TypeReference<Map<String, Object>>() {
        });

        String redisKey = RedisKeyConstants.LATEST_ALARM_RECORD + "_" + siteId + "_" + deviceId;
        // 获取redis里面的当前有效告警遍历添加到已存在告警key里面
        Map<String, Object> currentAlarm = redisCache.getCacheMap(redisKey);
        final Set<String> currentAlarmKeys = new HashSet<>();
        if (currentAlarm != null && !currentAlarm.isEmpty()) {
            currentAlarm.keySet().stream()
                    .filter(Objects::nonNull)
                    .map(Object::toString)
                    .forEach(currentAlarmKeys::add);
        }

        // 结合同步数据，筛选簇需要更新的告警信息
        List<String> needUpdateKeys = obj.entrySet().stream()
                .filter(entry -> {
                    Object valueObj = entry.getValue();
                    if (valueObj == null) {
                        return false;
                    }
                    int value = Integer.parseInt(valueObj.toString());
                    return value == 0 && currentAlarmKeys.contains(entry.getKey());
                })
                .map(Map.Entry::getKey)
                .collect(Collectors.toList());

        // 批量查询数据库-需要更新的数据
        Map<String, EmsAlarmRecords> needUpdateMap = new HashMap<>();
        if (!needUpdateKeys.isEmpty()) {
            List<EmsAlarmRecords> records = iEmsAlarmRecordsService.getAllUnfinishedRecords(needUpdateKeys, siteId, deviceId);
            // 转为Map便于快速获取
            needUpdateMap = records.stream()
                    .collect(Collectors.toMap(
                            EmsAlarmRecords::getAlarmPoint,
                            record -> record
                    ));
        }


        List<EmsAlarmRecords> saveOrUpdateList = new ArrayList<>();
        List<EmsAlarmRecords> newAddRecordList = new ArrayList<>();
        List<String> toRemoveFromRedis = new ArrayList<>();

        // 遍历数据map
        for (Map.Entry<String, Object> entry : obj.entrySet()) {
            if (entry.getValue() == null) {
                continue;
            }
            String key = entry.getKey();
            Long value = (Long) entry.getValue();
            Boolean isCurrentAlarm = currentAlarmKeys.contains(key);
            String matchRedisKey = siteId + "_" + category + "_" + key;
            // 默认告警值0是正常，1是异常
            Long alarmData = 1L;
            Object cacheObj = alarmInfoData.get(matchRedisKey);
            if (cacheObj != null) {
                EmsAlarmMatchData matchInfo = JSON.toJavaObject(cacheObj, EmsAlarmMatchData.class);
                alarmData = matchInfo.getAlarmData();
            }
            // 处理告警
            if (value.equals(alarmData) && !isCurrentAlarm) {
                // 上送值和匹配值相同，新增告警
                EmsAlarmMatchData matchInfo = JSON.toJavaObject(cacheObj, EmsAlarmMatchData.class);
                EmsAlarmRecords emsAlarmRecord = convertAlarmRecord(siteId, deviceId, matchInfo);
                saveOrUpdateList.add(emsAlarmRecord);
                newAddRecordList.add(emsAlarmRecord);
            } else {
                if (isCurrentAlarm) {
                    // 在当前告警里面 - 更新告警已处理
                    EmsAlarmRecords existingAlarm = needUpdateMap.get(key);
                    if (existingAlarm != null) {
                        existingAlarm.setStatus(AlarmStatus.DONE.getCode());
                        existingAlarm.setUpdateTime(new Date());
                        existingAlarm.setAlarmEndTime(new Date());
                        saveOrUpdateList.add(existingAlarm);
                        toRemoveFromRedis.add(key);
                    }
                }
            }
//            // 值为 1且不在当前告警里面 - 新增告警
//            if (value == 1 && !isCurrentAlarm) {
//                String matchRedisKey = category + "_" + key;
//                Object cacheObj = alarmInfoData.get(matchRedisKey);
//                if (cacheObj == null) {
//                    // 处理空数据逻辑
//                    return;
//                }
//                EmsAlarmMatchData matchInfo = JSON.toJavaObject(cacheObj, EmsAlarmMatchData.class);
//                EmsAlarmRecords emsAlarmRecord = convertAlarmRecord(siteId, deviceId, matchInfo);
//                saveOrUpdateList.add(emsAlarmRecord);
//                newAddRecordList.add(emsAlarmRecord);
//            } else if (value == 0 && isCurrentAlarm) {// 值为 0且在当前告警里面 - 更新告警已处理
//                EmsAlarmRecords existingAlarm = needUpdateMap.get(key);
//                if (existingAlarm != null) {
//                    existingAlarm.setStatus(AlarmStatus.DONE.getCode());
//                    existingAlarm.setUpdateTime(new Date());
//                    existingAlarm.setAlarmEndTime(new Date());
//                    saveOrUpdateList.add(existingAlarm);
//                    toRemoveFromRedis.add(key);
//                }
//            }
        }

        // 批量处理插入和更新操作
        if (CollectionUtils.isNotEmpty(saveOrUpdateList)) {
            iEmsAlarmRecordsService.batchProcessAlarmRecords(saveOrUpdateList);
        }
        // 已处理的从redis里面删除
        if (!toRemoveFromRedis.isEmpty()) {
            redisCache.deleteAllCacheMapValue(RedisKeyConstants.LATEST_ALARM_RECORD + siteId, toRemoveFromRedis.toArray());
        }
        // 批量添加新增的告警到Redis
        Map<String, EmsAlarmRecords> newAlarms = newAddRecordList.stream()
                .filter(a -> !AlarmStatus.DONE.getCode().equals(a.getStatus()))
                .collect(Collectors.toMap(EmsAlarmRecords::getAlarmPoint, a -> a));
        if (!newAlarms.isEmpty()) {
            // 本次新增的放入redis
            redisCache.setAllCacheMapValue(redisKey, newAlarms);
        }
    }

    private EmsAlarmRecords convertAlarmRecord(String siteId, String deviceId, EmsAlarmMatchData matchInfo) {
        EmsAlarmRecords emsAlarmRecords = new EmsAlarmRecords();
        emsAlarmRecords.setSiteId(siteId);
        emsAlarmRecords.setDeviceId(deviceId);
        emsAlarmRecords.setDeviceType(DeviceType.TCP.toString());
        emsAlarmRecords.setAlarmLevel(AlarmLevelStatus.GENERAL.getCode());
        emsAlarmRecords.setStatus(AlarmStatus.WAITING.getCode());
        emsAlarmRecords.setAlarmStartTime(DateUtils.getNowDate());
        emsAlarmRecords.setCreateTime(DateUtils.getNowDate());
        emsAlarmRecords.setCreateBy("system");
        emsAlarmRecords.setUpdateTime(DateUtils.getNowDate());
        emsAlarmRecords.setUpdateBy("system");

        if (matchInfo != null) {
            emsAlarmRecords.setAlarmPoint(matchInfo.getPoint());
            emsAlarmRecords.setAlarmContent(matchInfo.getAlarmDescription());
        }
        return emsAlarmRecords;
    }

    @Override
    public void processBatch(List<EmsBatteryData> batchData) {
        super.processBatch(batchData);
    }

    // 空数据不处理
    private boolean checkJsonDataEmpty(String jsonData) {
        if (StringUtils.isEmpty(jsonData)) {
            return true;
        }
        try {
            JsonNode jsonNode = objectMapper.readTree(jsonData);
            // 判断是否为空对象（{}）
            return jsonNode.isObject() && jsonNode.isEmpty();
        } catch (Exception e) {
            log.warn("设备数据Data字段解析失败，按空数据处理，data: {}, err: {}", jsonData, e.getMessage());
            return true;
        }
    }
}
