fix: 太阳能控制下发指令，自测bug修复。

intc-admin/src/main/resources/application.yml

@@ -138,6 +138,7 @@ tenant:
     - sys_user_role
     - sys_client
     - sys_oss_config
+    - iot_device_status
     - flow_spel
 
 # MyBatisPlus配置

intc-modules/intc-fishery/src/main/java/com/intc/fishery/domain/vo/DeviceVo.java

@@ -336,12 +336,17 @@ public class DeviceVo implements Serializable {
     @ExcelProperty(value = "用户名")
     private String userName;
 
-        /**
+    /**
      * 塘口名称
      */
     @ExcelProperty(value = "塘口名称")
     private String pondName;
 
+    /**
+     * 设备在线状态（online/offline）
+     */
+    private String onlineStatus;
+
     /**
      * 是否过期（0-未过期，1-已过期）
      */

intc-modules/intc-fishery/src/main/java/com/intc/fishery/service/impl/DeviceServiceImpl.java

@@ -5,6 +5,7 @@ import java.util.stream.Collectors;
 
 import com.baomidou.mybatisplus.core.conditions.query.LambdaQueryWrapper;
 import com.baomidou.mybatisplus.core.conditions.update.LambdaUpdateWrapper;
+import com.intc.common.core.config.properties.DeviceTypeProperties;
 import com.intc.common.core.utils.MapstructUtils;
 import com.intc.common.core.utils.StringUtils;
 import com.intc.common.mybatis.core.page.PageQuery;
@@ -60,11 +61,22 @@ public class DeviceServiceImpl implements IDeviceService {
     private final DeviceSwitchMapper deviceSwitchMapper;
     private final LinkedCtrlMapper linkedCtrlMapper;
     private final TimingCtrlMapper timingCtrlMapper;
+    private final DeviceTypeProperties deviceTypeProperties;
 
     /** Redis 离线等待标记前缀（与 DeviceDataHandler 保持一致） */
     private static final String DEVICE_OFFLINE_WAIT_KEY_PREFIX = "device:offline:wait:";
     /** Redis 溶解氧探头离线等待标记前缀 */
     private static final String DETECTOR_OXY_OFFLINE_WAIT_KEY_PREFIX = "device:detector:oxy:offline:wait:";
+    /** 太阳能网控设备类型 */
+    private static final Integer DEVICE_TYPE_SOLAR_CONTROLLER = 3;
+    /** 设备在线状态表别名 */
+    private static final String DEVICE_STATUS_ALIAS = "ids";
+    /** 设备在线状态表状态字段 */
+    private static final String DEVICE_STATUS_COLUMN = DEVICE_STATUS_ALIAS + ".status";
+    /** 设备在线状态展示字段 */
+    private static final String DEVICE_STATUS_DISPLAY_COLUMN = "LOWER(COALESCE(" + DEVICE_STATUS_COLUMN + ", 'offline'))";
+    /** 设备在线状态表状态字段（小写） */
+    private static final String DEVICE_STATUS_LOWER_COLUMN = "LOWER(" + DEVICE_STATUS_COLUMN + ")";
 
     /**
      * 查询设备管理
@@ -115,6 +127,7 @@ public class DeviceServiceImpl implements IDeviceService {
     private MPJLambdaWrapper<Device> buildQueryWrapper(DeviceBo bo) {
         Map<String, Object> params = bo.getParams();
         MPJLambdaWrapper<Device> wrapper = new MPJLambdaWrapper<>();
+        boolean solarQuery = DEVICE_TYPE_SOLAR_CONTROLLER.equals(bo.getDeviceType());
 
         // Select main table fields
         wrapper.selectAll(Device.class);
@@ -127,6 +140,12 @@ public class DeviceServiceImpl implements IDeviceService {
         wrapper.leftJoin(AquUser.class, AquUser::getId, Device::getUserId);
         // Left join with pond table
         wrapper.leftJoin(Pond.class, Pond::getId, Device::getPondId);
+        if (solarQuery) {
+            wrapper.selectAs(DEVICE_STATUS_DISPLAY_COLUMN, DeviceVo::getOnlineStatus)
+                .leftJoin(buildSolarStatusJoinSql());
+            applySolarOnlineStatusFilter(wrapper, params);
+            wrapper.orderByAsc("CASE WHEN " + DEVICE_STATUS_LOWER_COLUMN + " = 'online' THEN 0 ELSE 1 END");
+        }
 
         // 处理排序逼辑：如果expiredFlag为1，按过期时间倒序，否则按创建时间倒序
         String expiredFlag = params != null ? (String) params.get("expiredFlag") : null;
@@ -154,6 +173,33 @@ public class DeviceServiceImpl implements IDeviceService {
         return wrapper;
     }
 
+    private String buildSolarStatusJoinSql() {
+        String sql = "iot_device_status " + DEVICE_STATUS_ALIAS + " ON "
+            + DEVICE_STATUS_ALIAS + ".device_name = t.serial_num";
+        String productKey = deviceTypeProperties.getSolarController();
+        if (StringUtils.isNotBlank(productKey)) {
+            sql += " AND " + DEVICE_STATUS_ALIAS + ".product_key = '" + productKey.replace("'", "''") + "'";
+        }
+        return sql;
+    }
+
+    /**
+     * 太阳能网控在线状态查询。
+     */
+    private void applySolarOnlineStatusFilter(MPJLambdaWrapper<Device> wrapper, Map<String, Object> params) {
+        if (params == null || params.isEmpty()) {
+            return;
+        }
+        String onlineStatus = params.get("onlineStatus") != null ? params.get("onlineStatus").toString() : null;
+        if ("online".equalsIgnoreCase(onlineStatus)) {
+            wrapper.apply(DEVICE_STATUS_LOWER_COLUMN + " = {0}", "online");
+        } else if ("offline".equalsIgnoreCase(onlineStatus)) {
+            wrapper.and(w -> w.isNull(DEVICE_STATUS_COLUMN)
+                .or()
+                .apply(DEVICE_STATUS_LOWER_COLUMN + " <> {0}", "online"));
+        }
+    }
+
     /**
      * 处理额外的查询参数
      *

intc-modules/intc-iot/src/main/java/com/intc/iot/handler/DeviceDataHandler.java

@@ -63,6 +63,7 @@ import java.util.concurrent.TimeUnit;
 public class DeviceDataHandler {
 
     public static final String PENDING_SOLAR_PROPERTY_KEY_PREFIX = "device:pending:solar:property:";
+    private static final String PENDING_SOLAR_PROPERTY_FLUSH_LOCK_PREFIX = "device:pending:solar:property:flush:";
     private static final int DEVICE_TYPE_SOLAR_CONTROLLER = 3;
 
     /**
@@ -1011,8 +1012,23 @@ public class DeviceDataHandler {
                 deviceName, device.getId(), device.getDeviceType());
             return;
         }
+        if (StrUtil.isBlank(device.getSerialNum())) {
+            log.debug("[太阳能属性] 上线设备编号为空，跳过补发: deviceName={}, deviceId={}, iotId={}",
+                deviceName, device.getId(), device.getIotId());
+            return;
+        }
+
+        String flushLockKey = PENDING_SOLAR_PROPERTY_FLUSH_LOCK_PREFIX + device.getSerialNum();
+        boolean locked = false;
 
         try {
+            locked = RedisUtils.setObjectIfAbsent(flushLockKey, "1", Duration.ofSeconds(30));
+            if (!locked) {
+                log.debug("[太阳能属性] 已有补发任务执行中，跳过重复补发: deviceName={}, deviceId={}",
+                    deviceName, device.getId());
+                return;
+            }
+
             PendingSolarProperty pendingProperty = findPendingSolarProperty(device);
             if (pendingProperty == null) {
                 logNoPendingSolarProperty(trigger, deviceName, device);
@@ -1047,6 +1063,15 @@ public class DeviceDataHandler {
                 trigger, deviceName, pendingProperty.redisKey, pendingProperty.deviceId, pendingProperty.properties);
         } catch (Exception e) {
             log.error("[太阳能属性] {}补发异常: deviceName={}, err={}", trigger, deviceName, e.getMessage(), e);
+        } finally {
+            if (locked) {
+                try {
+                    RedisUtils.deleteObject(flushLockKey);
+                } catch (Exception e) {
+                    log.warn("[太阳能属性] 释放补发锁失败: deviceName={}, lockKey={}, err={}",
+                        deviceName, flushLockKey, e.getMessage());
+                }
+            }
         }
     }
 

intc-modules/intc-iot/src/main/java/com/intc/iot/service/impl/AmqpMessageHandlerImpl.java

@@ -5,11 +5,15 @@ import cn.hutool.json.JSONObject;
 import cn.hutool.json.JSONUtil;
 import com.intc.iot.handler.DeviceDataHandler;
 import com.intc.iot.service.AmqpMessageHandler;
+import com.intc.iot.service.DeviceStatusService;
 import lombok.RequiredArgsConstructor;
 import lombok.extern.slf4j.Slf4j;
 import org.springframework.beans.factory.annotation.Autowired;
 import org.springframework.stereotype.Service;
 
+import java.util.HashMap;
+import java.util.Map;
+
 /**
  * AMQP 消息处理器实现类
  * 
@@ -29,6 +33,9 @@ public class AmqpMessageHandlerImpl implements AmqpMessageHandler {
     @Autowired(required = false)
     private DeviceDataHandler deviceDataHandler;
 
+    @Autowired(required = false)
+    private DeviceStatusService deviceStatusService;
+
     @Override
     public void handleMessage(String message) {
         try {
@@ -218,17 +225,11 @@ public class AmqpMessageHandlerImpl implements AmqpMessageHandler {
     private void handleStatusMessage(JSONObject json) {
         String deviceName = json.getStr("deviceName");
         String status = json.getStr("status");
-        if (status != null) {
+        status = normalizeStatus(status);
-            status = status.trim().toLowerCase();
-        }
 
         log.debug("设备状态: {} - {}", deviceName, status);
 
-        if ("offline".equals(status) && deviceDataHandler != null && deviceName != null) {
+        handleDeviceStatusChange(json, deviceName, status);
-            deviceDataHandler.handleDeviceOffline(deviceName);
-        } else if ("online".equals(status) && deviceDataHandler != null && deviceName != null) {
-            deviceDataHandler.handleDeviceOnline(deviceName);
-        }
     }
 
     /**
@@ -241,17 +242,52 @@ public class AmqpMessageHandlerImpl implements AmqpMessageHandler {
         if (action == null) {
             action = json.getStr("status");
         }
-        if (action != null) {
+        action = normalizeStatus(action);
-            action = action.trim().toLowerCase();
-        }
 
         log.debug("设备{}: {}", "online".equals(action) ? "上线" : "下线", deviceName);
 
-        if ("offline".equals(action) && deviceDataHandler != null && deviceName != null) {
+        handleDeviceStatusChange(json, deviceName, action);
+    }
+
+    private void handleDeviceStatusChange(JSONObject json, String deviceName, String status) {
+        if (deviceName == null || status == null) {
+            log.warn("设备状态消息缺少设备名称或状态: {}", json);
+            return;
+        }
+
+        String productKey = json.getStr("productKey");
+        if (deviceStatusService != null && productKey != null && !productKey.trim().isEmpty()) {
+            try {
+                Map<String, Object> statusData = new HashMap<>();
+                statusData.put("productKey", productKey);
+                statusData.put("deviceName", deviceName);
+                statusData.put("status", status);
+                statusData.put("statusTime", json.get("statusTime"));
+                statusData.put("lastTime", json.get("lastTime"));
+                statusData.put("clientIp", json.getStr("clientIp"));
+                statusData.put("iotId", json.getStr("iotId"));
+                statusData.put("userId", json.get("userId"));
+                deviceStatusService.handleStatusChange(statusData);
+                return;
+            } catch (Exception e) {
+                log.error("写入设备状态失败，降级执行上下线处理: deviceName={}, status={}, err={}",
+                    deviceName, status, e.getMessage(), e);
+            }
+        }
+
+        if ("offline".equals(status) && deviceDataHandler != null) {
             deviceDataHandler.handleDeviceOffline(deviceName);
-        } else if ("online".equals(action) && deviceDataHandler != null && deviceName != null) {
+        } else if ("online".equals(status) && deviceDataHandler != null) {
             deviceDataHandler.handleDeviceOnline(deviceName);
         }
     }
 
+    private String normalizeStatus(String status) {
+        if (status == null) {
+            return null;
+        }
+        status = status.trim().toLowerCase();
+        return ("online".equals(status) || "offline".equals(status)) ? status : null;
+    }
+
 }

intc-modules/intc-iot/src/main/java/com/intc/iot/service/impl/DeviceStatusServiceImpl.java

@@ -2,6 +2,8 @@ package com.intc.iot.service.impl;
 
 import cn.hutool.json.JSONUtil;
 import com.baomidou.mybatisplus.core.conditions.query.LambdaQueryWrapper;
+import com.baomidou.mybatisplus.core.conditions.update.LambdaUpdateWrapper;
+import com.intc.common.tenant.helper.TenantHelper;
 import com.intc.iot.handler.DeviceDataHandler;
 import com.intc.iot.domain.IotDeviceStatus;
 import com.intc.iot.mapper.IotDeviceStatusMapper;
@@ -10,10 +12,11 @@ import lombok.RequiredArgsConstructor;
 import lombok.extern.slf4j.Slf4j;
 import org.springframework.beans.factory.annotation.Autowired;
 import org.springframework.boot.autoconfigure.condition.ConditionalOnBean;
+import org.springframework.dao.DuplicateKeyException;
 import org.springframework.stereotype.Service;
-import org.springframework.transaction.annotation.Transactional;
 
 import java.time.LocalDateTime;
+import java.util.List;
 import java.util.Map;
 
 /**
@@ -34,30 +37,36 @@ public class DeviceStatusServiceImpl implements DeviceStatusService {
     private DeviceDataHandler deviceDataHandler;
 
     @Override
-    @Transactional(rollbackFor = Exception.class)
     public void handleStatusChange(Map<String, Object> statusData) {
         try {
-            String productKey = (String) statusData.get("productKey");
+            if (statusData == null || statusData.isEmpty()) {
-            String deviceName = (String) statusData.get("deviceName");
+                log.warn("[设备状态] 状态数据为空，跳过处理");
-            String status = (String) statusData.get("status");
+                return;
+            }
+            String productKey = trimToNull(statusData.get("productKey"));
+            String deviceName = trimToNull(statusData.get("deviceName"));
+            String status = normalizeStatus(statusData.get("status"));
             Long statusTime = getLongValue(statusData, "statusTime");
             Long lastTime = getLongValue(statusData, "lastTime");
-            String clientIp = (String) statusData.get("clientIp");
+            String clientIp = trimToNull(statusData.get("clientIp"));
+            String iotId = trimToNull(statusData.get("iotId"));
+            Long userId = getLongValue(statusData, "userId");
 
             if (productKey == null || deviceName == null || status == null) {
                 log.warn("[设备状态] 状态数据不完整，跳过处理: {}", JSONUtil.toJsonStr(statusData));
                 return;
             }
+            if (statusTime == null) {
+                statusTime = System.currentTimeMillis() / 1000;
+            }
+            if (lastTime == null) {
+                lastTime = statusTime;
+            }
 
             log.info("[设备状态] 设备状态变更 - ProductKey: {}, DeviceName: {}, Status: {}", 
                     productKey, deviceName, status);
 
-            // 查询是否已存在记录
+            IotDeviceStatus existingStatus = findExistingStatus(productKey, deviceName);
-            IotDeviceStatus existingStatus = deviceStatusMapper.selectOne(
-                    new LambdaQueryWrapper<IotDeviceStatus>()
-                            .eq(IotDeviceStatus::getProductKey, productKey)
-                            .eq(IotDeviceStatus::getDeviceName, deviceName)
-            );
 
             LocalDateTime now = LocalDateTime.now();
 
@@ -67,8 +76,12 @@ public class DeviceStatusServiceImpl implements DeviceStatusService {
                 existingStatus.setStatusTime(statusTime);
                 existingStatus.setLastTime(lastTime);
                 existingStatus.setClientIp(clientIp);
+                existingStatus.setIotId(iotId);
+                existingStatus.setUserId(userId);
                 existingStatus.setUpdatedTime(now);
-                deviceStatusMapper.updateById(existingStatus);
+                TenantHelper.ignore(() -> {
+                    deviceStatusMapper.updateById(existingStatus);
+                });
 
                 log.debug("[设备状态] 更新设备状态记录 - ID: {}, Status: {}", existingStatus.getId(), status);
             } else {
@@ -76,21 +89,33 @@ public class DeviceStatusServiceImpl implements DeviceStatusService {
                 IotDeviceStatus newStatus = new IotDeviceStatus();
                 newStatus.setProductKey(productKey);
                 newStatus.setDeviceName(deviceName);
-                newStatus.setIotId((String) statusData.get("iotId"));
+                newStatus.setIotId(iotId);
                 newStatus.setStatus(status);
                 newStatus.setStatusTime(statusTime);
                 newStatus.setLastTime(lastTime);
                 newStatus.setClientIp(clientIp);
+                newStatus.setUserId(userId);
                 newStatus.setCreatedTime(now);
                 newStatus.setUpdatedTime(now);
-                deviceStatusMapper.insert(newStatus);
+                try {
+                    TenantHelper.ignore(() -> {
+                        deviceStatusMapper.insert(newStatus);
+                    });
+                } catch (DuplicateKeyException e) {
+                    log.warn("[设备状态] 状态记录并发插入冲突，改为更新: productKey={}, deviceName={}", productKey, deviceName);
+                    updateStatusByDevice(productKey, deviceName, newStatus);
+                }
 
                 log.info("[设备状态] 新增设备状态记录 - ProductKey: {}, DeviceName: {}, Status: {}", 
                         productKey, deviceName, status);
             }
 
-            if ("online".equalsIgnoreCase(status) && deviceDataHandler != null) {
+            if (deviceDataHandler != null) {
-                deviceDataHandler.handleDeviceOnline(deviceName);
+                if ("online".equals(status)) {
+                    deviceDataHandler.handleDeviceOnline(deviceName);
+                } else if ("offline".equals(status)) {
+                    deviceDataHandler.handleDeviceOffline(deviceName);
+                }
             }
         } catch (Exception e) {
             log.error("[设备状态] 处理设备状态变更异常: {}", e.getMessage(), e);
@@ -100,11 +125,11 @@ public class DeviceStatusServiceImpl implements DeviceStatusService {
 
     @Override
     public IotDeviceStatus queryDeviceStatus(String productKey, String deviceName) {
-        return deviceStatusMapper.selectOne(
+        return TenantHelper.ignore(() -> deviceStatusMapper.selectOne(
                 new LambdaQueryWrapper<IotDeviceStatus>()
                         .eq(IotDeviceStatus::getProductKey, productKey)
                         .eq(IotDeviceStatus::getDeviceName, deviceName)
-        );
+        ));
     }
 
     @Override
@@ -130,4 +155,53 @@ public class DeviceStatusServiceImpl implements DeviceStatusService {
             return null;
         }
     }
+
+    private IotDeviceStatus findExistingStatus(String productKey, String deviceName) {
+        return TenantHelper.ignore(() -> {
+            List<IotDeviceStatus> list = deviceStatusMapper.selectList(
+                new LambdaQueryWrapper<IotDeviceStatus>()
+                    .eq(IotDeviceStatus::getProductKey, productKey)
+                    .eq(IotDeviceStatus::getDeviceName, deviceName)
+            );
+            if (list == null || list.isEmpty()) {
+                return null;
+            }
+            if (list.size() > 1) {
+                log.warn("[设备状态] 存在重复状态记录: productKey={}, deviceName={}, count={}",
+                    productKey, deviceName, list.size());
+            }
+            return list.get(0);
+        });
+    }
+
+    private void updateStatusByDevice(String productKey, String deviceName, IotDeviceStatus status) {
+        TenantHelper.ignore(() -> {
+            deviceStatusMapper.update(status,
+                new LambdaUpdateWrapper<IotDeviceStatus>()
+                    .eq(IotDeviceStatus::getProductKey, productKey)
+                    .eq(IotDeviceStatus::getDeviceName, deviceName)
+            );
+        });
+    }
+
+    private String normalizeStatus(Object status) {
+        String text = trimToNull(status);
+        if (text == null) {
+            return null;
+        }
+        text = text.toLowerCase();
+        if ("online".equals(text) || "offline".equals(text)) {
+            return text;
+        }
+        log.warn("[设备状态] 未知状态值，跳过处理: {}", text);
+        return null;
+    }
+
+    private String trimToNull(Object value) {
+        if (value == null) {
+            return null;
+        }
+        String text = value.toString().trim();
+        return text.isEmpty() ? null : text;
+    }
 }