feat(ems/record): 优化告警处理与实时告警逻辑

ruoyi-ems/src/main/java/org/dromara/ems/record/constant/EmsAlarmPushStatusConstants.java

@@ -0,0 +1,38 @@
+package org.dromara.ems.record.constant;
+
+/**
+ * 告警推送状态常量。
+ *
+ * <p>迁移阶段统一使用字符串语义值，而不是旧库里的单字符占位，
+ * 这样统计 SQL、推送日志和前端展示才能共用同一套口径。</p>
+ */
+public final class EmsAlarmPushStatusConstants {
+
+    private EmsAlarmPushStatusConstants() {
+    }
+
+    /**
+     * 无推送配置，不纳入推送统计。
+     */
+    public static final String NONE = "NONE";
+
+    /**
+     * 已生成待执行推送任务。
+     */
+    public static final String PENDING = "PENDING";
+
+    /**
+     * 推送执行中。
+     */
+    public static final String PROCESSING = "PROCESSING";
+
+    /**
+     * 推送成功。
+     */
+    public static final String SUCCESS = "SUCCESS";
+
+    /**
+     * 推送失败。
+     */
+    public static final String FAIL = "FAIL";
+}

ruoyi-ems/src/main/java/org/dromara/ems/record/constant/EmsAlarmStatusConstants.java

@@ -0,0 +1,23 @@
+package org.dromara.ems.record.constant;
+
+/**
+ * 告警状态常量。
+ *
+ * <p>这里先只统一当前仓库已经真实落库的主状态，
+ * 避免“已处理/未处理”在前后端之间出现魔法值漂移。</p>
+ */
+public final class EmsAlarmStatusConstants {
+
+    private EmsAlarmStatusConstants() {
+    }
+
+    /**
+     * 已处理
+     */
+    public static final long HANDLED = 0L;
+
+    /**
+     * 未处理
+     */
+    public static final long UNHANDLED = 1L;
+}

ruoyi-ems/src/main/java/org/dromara/ems/record/domain/vo/AlarmHandleResultVo.java

@@ -0,0 +1,32 @@
+package org.dromara.ems.record.domain.vo;
+
+import lombok.Data;
+
+import java.io.Serial;
+import java.io.Serializable;
+import java.util.ArrayList;
+import java.util.List;
+
+/**
+ * 告警处理结果。
+ */
+@Data
+public class AlarmHandleResultVo implements Serializable {
+
+    @Serial
+    private static final long serialVersionUID = 1L;
+
+    private Integer requestCount = 0;
+
+    private Integer updatedCount = 0;
+
+    private Integer alreadyHandledCount = 0;
+
+    private Integer missingCount = 0;
+
+    private List<Long> successIds = new ArrayList<>();
+
+    private List<Long> alreadyHandledIds = new ArrayList<>();
+
+    private List<Long> missingIds = new ArrayList<>();
+}

ruoyi-ems/src/main/java/org/dromara/ems/record/domain/vo/EmsAlarmHandleResultVo.java

@@ -0,0 +1,34 @@
+package org.dromara.ems.record.domain.vo;
+
+import lombok.Data;
+
+import java.io.Serial;
+import java.io.Serializable;
+import java.util.ArrayList;
+import java.util.List;
+
+/**
+ * 告警处置结果。
+ *
+ * <p>前端需要知道“真的更新了几条”，不能再把 HTTP 200 当成业务成功。</p>
+ */
+@Data
+public class EmsAlarmHandleResultVo implements Serializable {
+
+    @Serial
+    private static final long serialVersionUID = 1L;
+
+    private Integer requestCount = 0;
+
+    private Integer updatedCount = 0;
+
+    private Integer alreadyHandledCount = 0;
+
+    private Integer missingCount = 0;
+
+    private List<Long> successIds = new ArrayList<>();
+
+    private List<Long> alreadyHandledIds = new ArrayList<>();
+
+    private List<Long> missingIds = new ArrayList<>();
+}

ruoyi-ems/src/main/java/org/dromara/ems/record/domain/vo/EmsRealtimeAlarmBatchResultVo.java

@@ -0,0 +1,44 @@
+package org.dromara.ems.record.domain.vo;
+
+import lombok.Data;
+import org.dromara.ems.record.domain.EmsRecordAlarmData;
+
+import java.io.Serial;
+import java.io.Serializable;
+import java.util.ArrayList;
+import java.util.List;
+
+/**
+ * 实时告警批量入库结果。
+ */
+@Data
+public class EmsRealtimeAlarmBatchResultVo implements Serializable {
+
+    @Serial
+    private static final long serialVersionUID = 1L;
+
+    /**
+     * 接收到的总条数。
+     */
+    private Integer totalCount = 0;
+
+    /**
+     * 实际新增条数。
+     */
+    private Integer insertedCount = 0;
+
+    /**
+     * 去重命中条数。
+     */
+    private Integer duplicateCount = 0;
+
+    /**
+     * 校验失败或跳过条数。
+     */
+    private Integer failedCount = 0;
+
+    /**
+     * 实际入库后的告警记录，供前端实时弹窗和后续联调直接复用。
+     */
+    private List<EmsRecordAlarmData> records = new ArrayList<>();
+}

ruoyi-ems/src/main/java/org/dromara/ems/record/domain/vo/EmsRealtimeAlarmEventVo.java

@@ -0,0 +1,30 @@
+package org.dromara.ems.record.domain.vo;
+
+import lombok.Data;
+import org.dromara.ems.record.domain.EmsRecordAlarmData;
+
+import java.io.Serial;
+import java.io.Serializable;
+import java.util.ArrayList;
+import java.util.Date;
+import java.util.List;
+
+/**
+ * 推送给前端的实时告警事件包。
+ */
+@Data
+public class EmsRealtimeAlarmEventVo implements Serializable {
+
+    @Serial
+    private static final long serialVersionUID = 1L;
+
+    private String eventType = "ems_alarm_realtime";
+
+    private String channel = "SSE";
+
+    private String source = "saveWebSocketAlarmData";
+
+    private Date generatedAt = new Date();
+
+    private List<EmsRecordAlarmData> alarms = new ArrayList<>();
+}

ruoyi-ems/src/main/java/org/dromara/ems/record/domain/vo/EmsRealtimeAlarmSaveResultVo.java

@@ -0,0 +1,35 @@
+package org.dromara.ems.record.domain.vo;
+
+import lombok.Data;
+import org.dromara.ems.record.domain.EmsRecordAlarmData;
+
+import java.io.Serial;
+import java.io.Serializable;
+import java.util.ArrayList;
+import java.util.List;
+
+/**
+ * 实时告警批量入库结果。
+ *
+ * <p>保留插入/去重/失败明细，是为了让前端和联调脚本能准确判断本次主链到底落到了哪一步。</p>
+ */
+@Data
+public class EmsRealtimeAlarmSaveResultVo implements Serializable {
+
+    @Serial
+    private static final long serialVersionUID = 1L;
+
+    private Integer requestCount = 0;
+
+    private Integer insertedCount = 0;
+
+    private Integer duplicateCount = 0;
+
+    private Integer failedCount = 0;
+
+    private List<EmsRecordAlarmData> insertedAlarms = new ArrayList<>();
+
+    private List<String> duplicateKeys = new ArrayList<>();
+
+    private List<String> failureMessages = new ArrayList<>();
+}

ruoyi-ems/src/main/java/org/dromara/ems/record/enums/EmsAlarmPushStatus.java

@@ -0,0 +1,32 @@
+package org.dromara.ems.record.enums;
+
+import cn.hutool.core.util.StrUtil;
+
+import java.util.Locale;
+
+/**
+ * 告警推送状态统一口径。
+ *
+ * <p>这里把历史遗留的数字值和旧字符串都收敛到标准状态，避免统计口径、日志页和实时链路各说各话。</p>
+ */
+public enum EmsAlarmPushStatus {
+
+    PENDING,
+    PROCESSING,
+    SUCCESS,
+    FAIL;
+
+    public static String normalize(String rawValue) {
+        if (StrUtil.isBlank(rawValue)) {
+            return PENDING.name();
+        }
+        String normalized = rawValue.trim().toUpperCase(Locale.ROOT);
+        return switch (normalized) {
+            case "0", "WAIT", "WAITING", "PENDING" -> PENDING.name();
+            case "1", "PROCESSING" -> PROCESSING.name();
+            case "2", "SUCCESS", "SUCCEED" -> SUCCESS.name();
+            case "3", "FAIL", "FAILED", "ERROR" -> FAIL.name();
+            default -> normalized;
+        };
+    }
+}

ruoyi-ems/src/main/java/org/dromara/ems/record/mapper/EmsRecordAlarmDataMapper.java

@@ -58,6 +58,17 @@ public interface EmsRecordAlarmDataMapper
      */
     public int updateEmsRecordAlarmData(EmsRecordAlarmData emsRecordAlarmData);
 
+    /**
+     * 将告警记录标记为已处理。
+     *
+     * <p>这里单独拆接口，是为了把“仅处理未处理记录”的条件收进 SQL，
+     * 避免并发确认时因为先查后改产生双重成功的假象。</p>
+     *
+     * @param emsRecordAlarmData 告警处理信息
+     * @return 实际更新条数
+     */
+    int markAlarmHandled(EmsRecordAlarmData emsRecordAlarmData);
+
     /**
      * 删除异常数据记录
      *

ruoyi-ems/src/main/java/org/dromara/ems/record/mapper/EmsRecordAlarmRuleMapper.java

@@ -1,5 +1,6 @@
 package org.dromara.ems.record.mapper;
 
+import org.apache.ibatis.annotations.Param;
 import org.dromara.ems.record.domain.EmsRecordAlarmRule;
 
 import java.util.List;
@@ -67,4 +68,13 @@ public interface EmsRecordAlarmRuleMapper
      * @return 告警规则总数
      */
     public int getEmsRecordAlarmRuleTotalCount();
+
+    /**
+     * 根据实时告警上下文匹配启用中的规则。
+     *
+     * <p>这里复用 SOP 命中同一套 `monitorId + cause` 语义，
+     * 避免“实时弹窗命中一套规则、落库补字段又命中另一套规则”的口径漂移。</p>
+     */
+    List<EmsRecordAlarmRule> selectMatchedAlarmRulesByAlarmInfo(@Param("monitorId") String monitorId,
+                                                                @Param("cause") String cause);
 }

ruoyi-ems/src/main/java/org/dromara/ems/record/service/IEmsRecordAlarmDataService.java

@@ -1,6 +1,8 @@
 package org.dromara.ems.record.service;
 
 import org.dromara.ems.record.domain.EmsRecordAlarmData;
+import org.dromara.ems.record.domain.vo.AlarmHandleResultVo;
+import org.dromara.ems.record.domain.vo.EmsRealtimeAlarmBatchResultVo;
 import org.dromara.ems.record.domain.vo.EmsRecordAlarmDataSummaryVo;
 
 import java.util.List;
@@ -74,7 +76,7 @@ public interface IEmsRecordAlarmDataService
      * @param objIds
      * @return
      */
-    int handleExceptionsAlarmData(Long[] objIds);
+    AlarmHandleResultVo handleExceptionsAlarmData(Long[] objIds);
 
     /**
      * 物联网数据阈值检查定时任务
@@ -93,5 +95,5 @@ public interface IEmsRecordAlarmDataService
      * @param alarmDataList 告警数据列表
      * @return 实际插入的记录数量
      */
-    public int saveWebSocketAlarmDataBatch(List<EmsRecordAlarmData> alarmDataList);
+    EmsRealtimeAlarmBatchResultVo saveWebSocketAlarmDataBatch(List<EmsRecordAlarmData> alarmDataList);
 }

ruoyi-ems/src/main/java/org/dromara/ems/record/service/impl/EmsRecordAlarmDataServiceImpl.java

@@ -2,12 +2,27 @@ package org.dromara.ems.record.service.impl;
 
 import cn.hutool.core.util.ObjectUtil;
 import lombok.RequiredArgsConstructor;
+import lombok.extern.slf4j.Slf4j;
+import org.dromara.common.core.exception.ServiceException;
 import org.dromara.common.core.utils.DateUtils;
+import org.dromara.common.json.utils.JsonUtils;
+import org.dromara.common.sse.utils.SseMessageUtils;
+import org.dromara.ems.base.domain.EmsAlarmPushLog;
+import org.dromara.ems.base.domain.EmsBaseMonitorInfo;
+import org.dromara.ems.base.domain.bo.EmsAlarmNotifyGroupUserBo;
+import org.dromara.ems.base.domain.vo.EmsAlarmNotifyGroupUserVo;
+import org.dromara.ems.base.mapper.EmsAlarmPushLogMapper;
 import org.dromara.ems.base.mapper.EmsBaseEnergyTypeMapper;
 import org.dromara.ems.base.mapper.EmsBaseMonitorInfoMapper;
+import org.dromara.ems.base.service.IEmsAlarmNotifyGroupUserService;
+import org.dromara.ems.record.constant.EmsAlarmPushStatusConstants;
+import org.dromara.ems.record.constant.EmsAlarmStatusConstants;
 import org.dromara.ems.record.domain.EmsRecordAlarmData;
 import org.dromara.ems.record.domain.EmsRecordAlarmRule;
 import org.dromara.ems.record.domain.RecordIotenvInstant;
+import org.dromara.ems.record.domain.vo.AlarmHandleResultVo;
+import org.dromara.ems.record.domain.vo.EmsRealtimeAlarmBatchResultVo;
+import org.dromara.ems.record.domain.vo.EmsRealtimeAlarmEventVo;
 import org.dromara.ems.record.domain.vo.EmsRecordAlarmDataSummaryVo;
 import org.dromara.ems.record.mapper.EmsRecordAlarmDataMapper;
 import org.dromara.ems.record.mapper.EmsRecordAlarmRuleMapper;
@@ -15,10 +30,13 @@ import org.dromara.ems.record.mapper.RecordIotenvInstantMapper;
 import org.dromara.ems.record.service.IEmsRecordAlarmDataService;
 import org.springframework.stereotype.Service;
 import org.springframework.transaction.annotation.Transactional;
+import org.springframework.transaction.support.TransactionSynchronization;
+import org.springframework.transaction.support.TransactionSynchronizationManager;
 
 import java.math.BigDecimal;
 import java.util.*;
 
+import static org.dromara.common.satoken.utils.LoginHelper.getUserId;
 import static org.dromara.common.satoken.utils.LoginHelper.getUsername;
 
 /**
@@ -28,6 +46,7 @@ import static org.dromara.common.satoken.utils.LoginHelper.getUsername;
  * @date 2024-05-15
  */
 @Service
+@Slf4j
 @RequiredArgsConstructor
 public class EmsRecordAlarmDataServiceImpl implements IEmsRecordAlarmDataService {
 
@@ -45,6 +64,12 @@ public class EmsRecordAlarmDataServiceImpl implements IEmsRecordAlarmDataService
 
     private final EmsBaseEnergyTypeMapper emsBaseEnergyTypeMapper;
 
+    private final EmsAlarmPushLogMapper emsAlarmPushLogMapper;
+
+    private final IEmsAlarmNotifyGroupUserService emsAlarmNotifyGroupUserService;
+
+    private static final long REALTIME_DUPLICATE_WINDOW_MS = 5 * 60 * 1000L;
+
     /**
      * 监测字段字典映射
      * 将前端字典值映射到数据库字段名和中文描述
@@ -171,16 +196,40 @@ public class EmsRecordAlarmDataServiceImpl implements IEmsRecordAlarmDataService
      */
     @Override
     @Transactional(rollbackFor = Exception.class)
-    public int handleExceptionsAlarmData(Long[] objIds) {
-        int result = 0;
+    public AlarmHandleResultVo handleExceptionsAlarmData(Long[] objIds) {
+        AlarmHandleResultVo result = new AlarmHandleResultVo();
+        if (objIds == null || objIds.length == 0) {
+            return result;
+        }
+        result.setRequestCount(objIds.length);
         for (Long objId : objIds) {
+            EmsRecordAlarmData existing = emsRecordAlarmDataMapper.selectEmsRecordAlarmDataByObjId(objId);
+            if (existing == null) {
+                result.setMissingCount(result.getMissingCount() + 1);
+                result.getMissingIds().add(objId);
+                continue;
+            }
+            if (Objects.equals(existing.getAlarmStatus(), EmsAlarmStatusConstants.HANDLED)) {
+                result.setAlreadyHandledCount(result.getAlreadyHandledCount() + 1);
+                result.getAlreadyHandledIds().add(objId);
+                continue;
+            }
             EmsRecordAlarmData alarmData = new EmsRecordAlarmData();
             alarmData.setObjId(objId);
-            alarmData.setAlarmStatus(0L);
+            alarmData.setAlarmStatus(EmsAlarmStatusConstants.HANDLED);
             alarmData.setOperationName(getUsername());
             alarmData.setOperationTime(DateUtils.getNowDate());
-            int i = emsRecordAlarmDataMapper.updateEmsRecordAlarmData(alarmData);
-            result += 1;
+            alarmData.setConfirmUserId(getUserId());
+            alarmData.setUpdateTime(DateUtils.getNowDate());
+            // 只允许从“未处理”切到“已处理”，这样并发确认时第二次提交会拿到 0 行更新。
+            int updated = emsRecordAlarmDataMapper.markAlarmHandled(alarmData);
+            if (updated > 0) {
+                result.setUpdatedCount(result.getUpdatedCount() + updated);
+                result.getSuccessIds().add(objId);
+            } else {
+                result.setAlreadyHandledCount(result.getAlreadyHandledCount() + 1);
+                result.getAlreadyHandledIds().add(objId);
+            }
         }
         return result;
     }
@@ -792,73 +841,71 @@ public class EmsRecordAlarmDataServiceImpl implements IEmsRecordAlarmDataService
 
     /**
      * 批量保存WebSocket告警数据
-     * 处理前端传来的告警数据列表，直接插入数据库
+     * 兼容旧入口，但补回结构化结果，便于前端实时链路和联调页直接复用。
      *
      * @param alarmDataList 告警数据列表
-     * @return 实际插入的记录数量
+     * @return 实时告警处理结果
      */
     @Override
-    public int saveWebSocketAlarmDataBatch(List<EmsRecordAlarmData> alarmDataList) {
+    @Transactional(rollbackFor = Exception.class)
+    public EmsRealtimeAlarmBatchResultVo saveWebSocketAlarmDataBatch(List<EmsRecordAlarmData> alarmDataList) {
+        EmsRealtimeAlarmBatchResultVo result = new EmsRealtimeAlarmBatchResultVo();
         if (alarmDataList == null || alarmDataList.isEmpty()) {
-            return 0;
+            return result;
         }
 
-        System.out.println("==== 开始保存WebSocket告警数据 ====");
-        System.out.println("接收到告警数据数量: " + alarmDataList.size());
-
-        // 设置默认值和创建时间
+        result.setTotalCount(alarmDataList.size());
         Date currentTime = DateUtils.getNowDate();
-        int insertedCount = 0;
-
-        for (EmsRecordAlarmData alarmData : alarmDataList) {
-            try {
-                // 设置创建时间
-                if (alarmData.getCreateTime() == null) {
-                    alarmData.setCreateTime(currentTime);
-                }
-
-                // 设置默认告警状态为未处理
-                if (alarmData.getAlarmStatus() == null) {
-                    alarmData.setAlarmStatus(1L); // 1表示未处理
-                }
-
-                // 数据完整性检查
-                if (ObjectUtil.isEmpty(alarmData.getMonitorId())) {
-                    System.out.println("警告：跳过monitorId为空的告警数据");
-                    continue;
-                }
-
-                if (alarmData.getCollectTime() == null) {
-                    System.out.println("警告：collectTime为空，使用当前时间");
-                    alarmData.setCollectTime(currentTime);
-                }
-
-                // 推送状态、标题和内容属于新增字段，入库前补齐默认值，避免后续列表/详情出现空链路。
-                fillAlarmEnhanceFields(alarmData);
-
-                System.out.println("保存告警数据 - 设备:" + alarmData.getMonitorId() +
-                                 ", 原因:" + alarmData.getCause() +
-                                 ", 数值:" + alarmData.getAlarmData() +
-                                 ", 告警类型:" + alarmData.getAlarmType());
-
-                // 直接插入数据库，不进行去重检查
-                int result = insertEmsRecordAlarmData(alarmData);
-                if (result > 0) {
-                    insertedCount++;
-                    System.out.println("成功保存告警数据 - 设备:" + alarmData.getMonitorId() +
-                                     ", 原因:" + alarmData.getCause());
-                }
-
-            } catch (Exception e) {
-                System.err.println("保存告警数据时发生错误: " + e.getMessage());
-                e.printStackTrace();
+        Set<String> requestDeduplicationKeys = new HashSet<>();
+        for (int index = 0; index < alarmDataList.size(); index++) {
+            EmsRecordAlarmData alarmData = alarmDataList.get(index);
+            Optional<String> invalidReason = validateRealtimeAlarm(alarmData, currentTime);
+            if (invalidReason.isPresent()) {
+                // 这里允许坏消息单条失败，但数据库写入错误仍走事务整体回滚，避免产生半闭环脏数据。
+                result.setFailedCount(result.getFailedCount() + 1);
+                log.warn("实时告警第{}条校验失败，原因={}", index + 1, invalidReason.get());
+                continue;
             }
+
+            String requestKey = buildUniqueKey(alarmData);
+            if (requestKey != null && !requestDeduplicationKeys.add(requestKey)) {
+                result.setDuplicateCount(result.getDuplicateCount() + 1);
+                log.info("实时告警命中批次内去重，businessKey={}", requestKey);
+                continue;
+            }
+
+            List<EmsRecordAlarmRule> matchedRules = emsRecordAlarmRuleMapper.selectMatchedAlarmRulesByAlarmInfo(alarmData.getMonitorId(), alarmData.getCause());
+            enrichAlarmByMonitor(alarmData);
+            enrichAlarmByRule(alarmData, matchedRules);
+            fillAlarmEnhanceFields(alarmData);
+
+            if (isDuplicateRealtimeAlarm(alarmData)) {
+                result.setDuplicateCount(result.getDuplicateCount() + 1);
+                continue;
+            }
+
+            int inserted = insertEmsRecordAlarmData(alarmData);
+            if (inserted != 1 || alarmData.getObjId() == null) {
+                throw new ServiceException("实时告警落库失败，monitorId=" + alarmData.getMonitorId() + ", cause=" + alarmData.getCause());
+            }
+
+            int pushCount = createPendingPushLogs(alarmData, matchedRules);
+            if (pushCount > 0 || !EmsAlarmPushStatusConstants.NONE.equals(alarmData.getPushStatus())) {
+                EmsRecordAlarmData patch = new EmsRecordAlarmData();
+                patch.setObjId(alarmData.getObjId());
+                patch.setPushCount(pushCount);
+                patch.setPushStatus(alarmData.getPushStatus());
+                patch.setUpdateTime(DateUtils.getNowDate());
+                emsRecordAlarmDataMapper.updateEmsRecordAlarmData(patch);
+                alarmData.setPushCount(pushCount);
+            }
+
+            result.getRecords().add(alarmData);
         }
 
-        System.out.println("WebSocket告警数据保存完成，实际插入: " + insertedCount + " 条记录");
-        System.out.println("==== WebSocket告警数据保存结束 ====");
-
-        return insertedCount;
+        result.setInsertedCount(result.getRecords().size());
+        publishRealtimeAlarmEventAfterCommit(result);
+        return result;
     }
 
     private void fillAlarmEnhanceFields(EmsRecordAlarmData alarmData) {
@@ -868,10 +915,7 @@ public class EmsRecordAlarmDataServiceImpl implements IEmsRecordAlarmDataService
         if (alarmData.getPushCount() == null) {
             alarmData.setPushCount(0);
         }
-        if (ObjectUtil.isEmpty(alarmData.getPushStatus())) {
-            // 约定 0 为未推送，先补默认值，避免新增字段因为空值影响筛选和详情展示。
-            alarmData.setPushStatus("0");
-        }
+        alarmData.setPushStatus(normalizePushStatus(alarmData.getPushStatus()));
         if (alarmData.getActualValue() == null && ObjectUtil.isNotEmpty(alarmData.getAlarmData())) {
             try {
                 alarmData.setActualValue(new BigDecimal(alarmData.getAlarmData()));
@@ -906,4 +950,214 @@ public class EmsRecordAlarmDataServiceImpl implements IEmsRecordAlarmDataService
             }
         }
     }
+
+    private Optional<String> validateRealtimeAlarm(EmsRecordAlarmData alarmData, Date fallbackTime) {
+        if (alarmData == null) {
+            return Optional.of("告警对象为空");
+        }
+        if (ObjectUtil.isEmpty(alarmData.getMonitorId())) {
+            return Optional.of("monitorId不能为空");
+        }
+        if (ObjectUtil.isEmpty(alarmData.getCause())) {
+            return Optional.of("cause不能为空，SOP 命中链无法建立");
+        }
+        if (alarmData.getCollectTime() == null) {
+            alarmData.setCollectTime(fallbackTime);
+        }
+        if (alarmData.getCreateTime() == null) {
+            alarmData.setCreateTime(fallbackTime);
+        }
+        if (alarmData.getAlarmStatus() == null) {
+            alarmData.setAlarmStatus(EmsAlarmStatusConstants.UNHANDLED);
+        }
+        return Optional.empty();
+    }
+
+    private boolean isDuplicateRealtimeAlarm(EmsRecordAlarmData alarmData) {
+        Date collectTime = alarmData.getCollectTime() != null ? alarmData.getCollectTime() : DateUtils.getNowDate();
+        Date startTime = new Date(collectTime.getTime() - REALTIME_DUPLICATE_WINDOW_MS);
+        Date endTime = new Date(collectTime.getTime() + REALTIME_DUPLICATE_WINDOW_MS);
+        Integer duplicateCount = emsRecordAlarmDataMapper.checkDuplicateAlarmData(alarmData.getMonitorId(), alarmData.getCause(), startTime, endTime);
+        return duplicateCount != null && duplicateCount > 0;
+    }
+
+    private void enrichAlarmByMonitor(EmsRecordAlarmData alarmData) {
+        if (ObjectUtil.isEmpty(alarmData.getMonitorId())) {
+            return;
+        }
+        EmsBaseMonitorInfo query = new EmsBaseMonitorInfo();
+        query.setMonitorCode(alarmData.getMonitorId());
+        List<EmsBaseMonitorInfo> monitorInfos = emsBaseMonitorInfoMapper.selectEmsBaseMonitorInfoList(query);
+        if (monitorInfos == null || monitorInfos.isEmpty()) {
+            return;
+        }
+        EmsBaseMonitorInfo monitorInfo = monitorInfos.get(0);
+        if (ObjectUtil.isEmpty(alarmData.getMonitorName())) {
+            alarmData.setMonitorName(monitorInfo.getMonitorName());
+        }
+        if (ObjectUtil.isEmpty(alarmData.getMetricCode())) {
+            alarmData.setMetricCode(monitorInfo.getMetricCode());
+        }
+        if (ObjectUtil.isEmpty(alarmData.getTenantId())) {
+            alarmData.setTenantId(monitorInfo.getTenantId());
+        }
+    }
+
+    private void enrichAlarmByRule(EmsRecordAlarmData alarmData, List<EmsRecordAlarmRule> matchedRules) {
+        if (matchedRules == null || matchedRules.isEmpty()) {
+            return;
+        }
+        EmsRecordAlarmRule matchedRule = matchedRules.get(0);
+        if (ObjectUtil.isEmpty(alarmData.getTenantId())) {
+            alarmData.setTenantId(matchedRule.getTenantId());
+        }
+        if (ObjectUtil.isEmpty(alarmData.getMetricCode())) {
+            alarmData.setMetricCode(matchedRule.getMetricCode());
+        }
+        if (ObjectUtil.isEmpty(alarmData.getAlarmLevel())) {
+            alarmData.setAlarmLevel(matchedRule.getAlarmLevel());
+        }
+            if (ObjectUtil.isEmpty(alarmData.getNotifyUser())) {
+                alarmData.setNotifyUser(matchedRule.getNotifyUser());
+            }
+            if (alarmData.getThresholdValue() == null) {
+            if (matchedRule.getTriggerValue() != null) {
+                alarmData.setThresholdValue(matchedRule.getTriggerValue());
+            } else if (Objects.equals(1L, matchedRule.getTriggerRule())) {
+                alarmData.setThresholdValue(matchedRule.getAlarmLower());
+            } else {
+                alarmData.setThresholdValue(matchedRule.getAlarmUpper());
+            }
+        }
+    }
+
+    private int createPendingPushLogs(EmsRecordAlarmData alarmData, List<EmsRecordAlarmRule> matchedRules) {
+        Set<String> pushTargets = new LinkedHashSet<>();
+        collectExplicitNotifyUsers(pushTargets, alarmData.getNotifyUser());
+        if (matchedRules != null) {
+            for (EmsRecordAlarmRule matchedRule : matchedRules) {
+                collectExplicitNotifyUsers(pushTargets, matchedRule.getNotifyUser());
+                if (matchedRule.getNotifyGroupId() != null) {
+                    EmsAlarmNotifyGroupUserBo query = new EmsAlarmNotifyGroupUserBo();
+                    query.setGroupId(matchedRule.getNotifyGroupId());
+                    query.setIsEnable("0");
+                    List<EmsAlarmNotifyGroupUserVo> groupUsers = emsAlarmNotifyGroupUserService.queryList(query);
+                    if (groupUsers == null || groupUsers.isEmpty()) {
+                        pushTargets.add("GROUP:" + matchedRule.getNotifyGroupId());
+                        continue;
+                    }
+                    for (EmsAlarmNotifyGroupUserVo groupUser : groupUsers) {
+                        if (ObjectUtil.isNotEmpty(groupUser.getPhone())) {
+                            pushTargets.add(groupUser.getPhone());
+                        } else if (ObjectUtil.isNotEmpty(groupUser.getEmail())) {
+                            pushTargets.add(groupUser.getEmail());
+                        } else if (ObjectUtil.isNotEmpty(groupUser.getNickName())) {
+                            pushTargets.add(groupUser.getNickName());
+                        } else if (ObjectUtil.isNotEmpty(groupUser.getUserName())) {
+                            pushTargets.add(groupUser.getUserName());
+                        }
+                    }
+                }
+            }
+        }
+
+        for (String target : pushTargets) {
+            EmsAlarmPushLog pushLog = new EmsAlarmPushLog();
+            pushLog.setAlarmObjId(alarmData.getObjId());
+            pushLog.setChannelType(resolveChannelType(target));
+            pushLog.setTargetValue(target);
+            pushLog.setAlarmLevel(alarmData.getAlarmLevel());
+            pushLog.setPushContent(alarmData.getAlarmContent());
+            pushLog.setPushStatus(EmsAlarmPushStatusConstants.PENDING);
+            pushLog.setResponseMsg("已生成待推送日志，等待真实推送执行链接管");
+            pushLog.setPushTime(DateUtils.getNowDate());
+            emsAlarmPushLogMapper.insert(pushLog);
+        }
+
+        // 这里明确区分“已生成推送任务”和“本条告警没有通知配置”，避免统计页把未配置也算成待推送。
+        alarmData.setPushStatus(pushTargets.isEmpty() ? EmsAlarmPushStatusConstants.NONE : EmsAlarmPushStatusConstants.PENDING);
+        return pushTargets.size();
+    }
+
+    private void collectExplicitNotifyUsers(Set<String> pushTargets, String notifyUser) {
+        if (ObjectUtil.isEmpty(notifyUser)) {
+            return;
+        }
+        String[] parts = notifyUser.split("[,;，；\\s]+");
+        for (String part : parts) {
+            if (ObjectUtil.isNotEmpty(part)) {
+                pushTargets.add(part.trim());
+            }
+        }
+    }
+
+    private String resolveChannelType(String targetValue) {
+        if (ObjectUtil.isEmpty(targetValue)) {
+            return "SITE";
+        }
+        if (targetValue.contains("@")) {
+            return "EMAIL";
+        }
+        if (targetValue.matches("^1\\d{10}$")) {
+            return "SMS";
+        }
+        if (targetValue.startsWith("http://") || targetValue.startsWith("https://")) {
+            return "WEBHOOK";
+        }
+        return "SITE";
+    }
+
+    private String normalizePushStatus(String pushStatus) {
+        if (ObjectUtil.isEmpty(pushStatus)) {
+            return EmsAlarmPushStatusConstants.NONE;
+        }
+        String normalized = pushStatus.trim().toUpperCase(Locale.ROOT);
+        if ("0".equals(normalized) || "WAIT".equals(normalized)) {
+            return EmsAlarmPushStatusConstants.PENDING;
+        }
+        if ("1".equals(normalized)) {
+            return EmsAlarmPushStatusConstants.PROCESSING;
+        }
+        if ("2".equals(normalized) || "FAILED".equals(normalized)) {
+            return EmsAlarmPushStatusConstants.FAIL;
+        }
+        if (EmsAlarmPushStatusConstants.NONE.equals(normalized)
+            || EmsAlarmPushStatusConstants.PENDING.equals(normalized)
+            || EmsAlarmPushStatusConstants.PROCESSING.equals(normalized)
+            || EmsAlarmPushStatusConstants.SUCCESS.equals(normalized)
+            || EmsAlarmPushStatusConstants.FAIL.equals(normalized)) {
+            return normalized;
+        }
+        return EmsAlarmPushStatusConstants.NONE;
+    }
+
+    private void publishRealtimeAlarmEventAfterCommit(EmsRealtimeAlarmBatchResultVo result) {
+        if (result == null || result.getInsertedCount() <= 0 || result.getRecords().isEmpty()) {
+            return;
+        }
+        Runnable publishTask = () -> {
+            try {
+                EmsRealtimeAlarmEventVo event = new EmsRealtimeAlarmEventVo();
+                // 这里使用前端约定的 eventType，避免迁移阶段再引入额外的协议转换层。
+                event.setEventType("ems_alarm_realtime");
+                event.setGeneratedAt(DateUtils.getNowDate());
+                event.setSource("saveWebSocketAlarmData");
+                event.setAlarms(result.getRecords());
+                SseMessageUtils.publishAll(JsonUtils.toJsonString(event));
+            } catch (Exception ex) {
+                // 这里不能反向影响主事务，避免“落库成功但广播失败”把记录整体回滚。
+                log.warn("实时告警 SSE 广播失败，insertedCount={}", result.getInsertedCount(), ex);
+            }
+        };
+        if (TransactionSynchronizationManager.isSynchronizationActive()) {
+            TransactionSynchronizationManager.registerSynchronization(new TransactionSynchronization() {
+                @Override
+                public void afterCommit() {
+                    publishTask.run();
+                }
+            });
+            return;
+        }
+        publishTask.run();
+    }
 }

ruoyi-ems/src/main/java/org/dromara/ems/record/service/impl/EmsRecordAlarmRuleServiceImpl.java

@@ -3,6 +3,8 @@ package org.dromara.ems.record.service.impl;
 import lombok.RequiredArgsConstructor;
 import org.dromara.common.core.utils.DateUtils;
 import org.dromara.ems.base.domain.EmsAlarmActionStep;
+import org.dromara.ems.base.domain.EmsAlarmNotifyGroup;
+import org.dromara.ems.base.mapper.EmsAlarmNotifyGroupMapper;
 import org.dromara.ems.base.service.IEmsAlarmActionStepService;
 import org.dromara.ems.record.domain.EmsRecordAlarmRule;
 import org.dromara.ems.record.mapper.EmsRecordAlarmRuleMapper;
@@ -28,6 +30,8 @@ public class EmsRecordAlarmRuleServiceImpl implements IEmsRecordAlarmRuleService
 
     private final IEmsAlarmActionStepService emsAlarmActionStepService;
 
+    private final EmsAlarmNotifyGroupMapper emsAlarmNotifyGroupMapper;
+
     /**
      * 查询异常告警规则
      *
@@ -61,6 +65,7 @@ public class EmsRecordAlarmRuleServiceImpl implements IEmsRecordAlarmRuleService
     @Override
     public int insertEmsRecordAlarmRule(EmsRecordAlarmRule emsRecordAlarmRule)
     {
+        validateNotifyConfig(emsRecordAlarmRule);
         emsRecordAlarmRule.setCreateTime(DateUtils.getNowDate());
         return emsRecordAlarmRuleMapper.insertEmsRecordAlarmRule(emsRecordAlarmRule);
     }
@@ -74,6 +79,7 @@ public class EmsRecordAlarmRuleServiceImpl implements IEmsRecordAlarmRuleService
     @Override
     public int updateEmsRecordAlarmRule(EmsRecordAlarmRule emsRecordAlarmRule)
     {
+        validateNotifyConfig(emsRecordAlarmRule);
         emsRecordAlarmRule.setUpdateTime(DateUtils.getNowDate());
         return emsRecordAlarmRuleMapper.updateEmsRecordAlarmRule(emsRecordAlarmRule);
     }
@@ -152,4 +158,17 @@ public class EmsRecordAlarmRuleServiceImpl implements IEmsRecordAlarmRuleService
     {
     	return emsRecordAlarmRuleMapper.getEmsRecordAlarmRuleTotalCount();
     }
+
+    private void validateNotifyConfig(EmsRecordAlarmRule emsRecordAlarmRule) {
+        if (emsRecordAlarmRule == null || emsRecordAlarmRule.getNotifyGroupId() == null) {
+            return;
+        }
+        EmsAlarmNotifyGroup notifyGroup = emsAlarmNotifyGroupMapper.selectById(emsRecordAlarmRule.getNotifyGroupId());
+        if (notifyGroup == null) {
+            throw new IllegalArgumentException("通知组不存在，无法保存告警规则");
+        }
+        if ("1".equals(notifyGroup.getIsEnable())) {
+            throw new IllegalArgumentException("通知组已停用，请先启用后再绑定到告警规则");
+        }
+    }
 }

ruoyi-ems/src/main/resources/mapper/ems/record/EmsRecordAlarmDataMapper.xml

@@ -82,6 +82,8 @@ PUBLIC "-//mybatis.org//DTD Mapper 3.0//EN"
             <if test="collectDeviceId != null  and collectDeviceId != ''"> and rad.collect_device_id = #{collectDeviceId}</if>
             <if test="collectDeviceName != null  and collectDeviceName != ''"> and ebcdi.collect_device_name like concat('%', #{collectDeviceName}, '%')</if>
             <if test="collectTime != null "> and rad.collect_time = #{collectTime}</if>
+            <if test="beginCollectTime != null and beginCollectTime != ''"> and rad.collect_time &gt;= #{beginCollectTime}</if>
+            <if test="endCollectTime != null and endCollectTime != ''"> and rad.collect_time &lt;= #{endCollectTime}</if>
             <if test="alarmType != null "> and rad.alarm_type = #{alarmType}</if>
             <if test="alarmStatus != null "> and rad.alarm_status = #{alarmStatus}</if>
             <if test="alarmData != null  and alarmData != ''"> and rad.alarm_data = #{alarmData}</if>
@@ -102,6 +104,12 @@ PUBLIC "-//mybatis.org//DTD Mapper 3.0//EN"
             <if test="confirmUserId != null "> and rad.confirm_user_id = #{confirmUserId}</if>
             <if test="confirmRemark != null  and confirmRemark != ''"> and rad.confirm_remark like concat('%', #{confirmRemark}, '%')</if>
             <if test="notifyUser != null  and notifyUser != ''"> and rad.notify_user = #{notifyUser}</if>
+            <if test="monitorIds != null and monitorIds.length > 0">
+                and rad.monitor_id in
+                <foreach collection="monitorIds" item="monitorId" open="(" separator="," close=")">
+                    #{monitorId}
+                </foreach>
+            </if>
         </where>
     </sql>
 
@@ -115,7 +123,7 @@ PUBLIC "-//mybatis.org//DTD Mapper 3.0//EN"
         SELECT COUNT(1) AS totalCount,
                COALESCE(SUM(CASE WHEN rad.alarm_status = 1 THEN 1 ELSE 0 END), 0) AS unhandledCount,
                COALESCE(SUM(CASE WHEN rad.alarm_status = 0 THEN 1 ELSE 0 END), 0) AS handledCount,
-               COALESCE(SUM(CASE WHEN UPPER(COALESCE(rad.push_status, '')) IN ('PENDING', 'WAIT') THEN 1 ELSE 0 END), 0) AS pushPendingCount,
+               COALESCE(SUM(CASE WHEN UPPER(COALESCE(rad.push_status, '')) IN ('PENDING', 'WAIT', '0') THEN 1 ELSE 0 END), 0) AS pushPendingCount,
                COALESCE(SUM(CASE WHEN UPPER(COALESCE(rad.push_status, '')) = 'PROCESSING' THEN 1 ELSE 0 END), 0) AS pushProcessingCount,
                COALESCE(SUM(CASE WHEN UPPER(COALESCE(rad.push_status, '')) = 'SUCCESS' THEN 1 ELSE 0 END), 0) AS pushSuccessCount,
                COALESCE(SUM(CASE WHEN UPPER(COALESCE(rad.push_status, '')) IN ('FAIL', 'FAILED') THEN 1 ELSE 0 END), 0) AS pushFailCount
@@ -223,6 +231,17 @@ PUBLIC "-//mybatis.org//DTD Mapper 3.0//EN"
         where obj_id = #{objId}
     </update>
 
+    <update id="markAlarmHandled" parameterType="EmsRecordAlarmData">
+        update ems_record_alarm_data
+        set alarm_status = #{alarmStatus},
+            operation_name = #{operationName},
+            operation_time = #{operationTime},
+            confirm_user_id = #{confirmUserId},
+            update_time = #{updateTime}
+        where obj_id = #{objId}
+          and alarm_status &lt;&gt; #{alarmStatus}
+    </update>
+
     <delete id="deleteEmsRecordAlarmDataByObjId" parameterType="Long">
         delete from ems_record_alarm_data where obj_id = #{objId}
     </delete>

ruoyi-ems/src/main/resources/mapper/ems/record/EmsRecordAlarmRuleMapper.xml

@@ -106,6 +106,32 @@ PUBLIC "-//mybatis.org//DTD Mapper 3.0//EN"
         where RAL.obj_id = #{objId}
     </select>
 
+    <select id="selectMatchedAlarmRulesByAlarmInfo" resultMap="EmsRecordAlarmRuleResult">
+        <include refid="selectEmsRecordAlarmRuleVo"/>
+        <where>
+            RAL.monitor_id = #{monitorId}
+            and (RAL.is_enable is null or RAL.is_enable = '0')
+            and (
+                (RAL.monitor_field IS NOT NULL AND
+                 CASE RAL.monitor_field
+                     WHEN 0 THEN #{cause} = '温度'
+                     WHEN 1 THEN #{cause} = '湿度'
+                     WHEN 2 THEN #{cause} = '振动-速度(mm/s)'
+                     WHEN 3 THEN #{cause} = '振动-位移(um)'
+                     WHEN 4 THEN #{cause} = '振动-加速度(g)'
+                     WHEN 5 THEN #{cause} = '振动-温度(℃)'
+                     WHEN 6 THEN #{cause} = '噪音'
+                     WHEN 7 THEN #{cause} = '照度'
+                     WHEN 8 THEN #{cause} = '气体浓度'
+                     ELSE 0
+                 END = 1
+                )
+                or (RAL.monitor_field IS NULL)
+            )
+        </where>
+        order by RAL.obj_id desc
+    </select>
+
     <insert id="insertEmsRecordAlarmRule" parameterType="EmsRecordAlarmRule" useGeneratedKeys="true" keyProperty="objId">
         insert into ems_record_alarm_rule
         <trim prefix="(" suffix=")" suffixOverrides=",">