diff --git a/intc-admin/pom.xml b/intc-admin/pom.xml
index 54c3ee2..3f5f7ac 100644
--- a/intc-admin/pom.xml
+++ b/intc-admin/pom.xml
@@ -108,6 +108,12 @@
com.intc
intc-workflow
+
+
+ com.intc
+ intc-iot
+ ${revision}
+
de.codecentric
diff --git a/intc-admin/src/main/resources/alarm-config-example.yml b/intc-admin/src/main/resources/alarm-config-example.yml
new file mode 100644
index 0000000..534b3c8
--- /dev/null
+++ b/intc-admin/src/main/resources/alarm-config-example.yml
@@ -0,0 +1,32 @@
+# ==================== 告警配置示例 ====================
+# 将以下配置添加到 application.yml 或 application-dev.yml 中
+
+# 告警阈值配置
+alarm:
+ threshold:
+ # 溶解氧阈值（mg/L）
+ dissolved-oxygen:
+ min: 4.0 # 最低阈值
+ max: 15.0 # 最高阈值
+
+ # 温度阈值（℃）
+ temperature:
+ min: 10.0 # 最低阈值
+ max: 35.0 # 最高阈值
+
+ # pH值阈值
+ ph:
+ min: 6.5 # 最低阈值
+ max: 8.5 # 最高阈值
+
+ # 盐度阈值
+ salinity:
+ max: 35.0 # 最高阈值
+
+ # 电池电量阈值（%）
+ battery:
+ min: 20.0 # 最低阈值
+
+ # 告警通知配置
+ notification:
+ interval: 30 # 告警通知间隔时间（分钟），同一设备在此时间内不会重复发送通知
diff --git a/intc-admin/src/main/resources/application-dev.yml b/intc-admin/src/main/resources/application-dev.yml
index 031d4c5..9c42941 100644
--- a/intc-admin/src/main/resources/application-dev.yml
+++ b/intc-admin/src/main/resources/application-dev.yml
@@ -13,7 +13,7 @@ spring.boot.admin.client:
--- # snail-job 配置
snail-job:
- enabled: true
+ enabled: false
# 需要在 SnailJob 后台组管理创建对应名称的组,然后创建任务的时候选择对应的组,才能正确分派任务
group: "ruoyi_group"
# SnailJob 接入验证令牌 详见 script/sql/ry_job.sql `sj_group_config` 表
@@ -35,7 +35,7 @@ spring:
# 动态数据源文档 https://www.kancloud.cn/tracy5546/dynamic-datasource/content
dynamic:
# 性能分析插件(有性能损耗 不建议生产环境使用)
- p6spy: true
+ p6spy: false
# 设置默认的数据源或者数据源组,默认值即为 master
primary: master
# 严格模式 匹配不到数据源则报错
@@ -204,6 +204,44 @@ sms:
code-template-id: 'SMS_465720430' # TODO: 请填写您的阿里云短信模板CODE
+--- # 阿里云生活物联网平台（飞燕平台）配置
+aliyun:
+ living-iot:
+ # 阿里云 AccessKey ID（必填）
+ access-key-id: LTAI5tRnPowmTLjH181nSbsR
+ # 阿里云 AccessKey Secret（必填）
+ access-key-secret: Vh2LoAM1t3XuMUVy2wTWSACJ97kOUW
+ # 地域节点（必填）
+ region-id: cn-shanghai
+ # 飞燕平台项目ID（Project ID）
+ project-id: a123nMibvh0q4UnU
+ # App Key（用于 API 调用）
+ app-key: 334224397
+ # App Secret
+ app-secret: 70de3018ec39423e9ca1e1b6a6a84ad6
+ # AMQP 服务端订阅配置（使用数据同步的 AppKey + AppSecret）
+ amqp:
+ # 是否启用
+ enabled: true
+ # 数据同步 AppKey（与上面的 app-key 不同！）
+ data-sync-app-key: 334224409
+ # 数据同步 AppSecret（请在阿里云控制台查看）
+ data-sync-app-secret: 17fdd58f9a4c4c90be236897b1f8e8f7
+ # AMQP 接入点地址（使用生活物联网平台官方地址）
+ endpoint: amqps://ilop.iot-amqp.cn-shanghai.aliyuncs.com:5671
+ # 消费组 ID（按照官方文档，应该与数据同步 AppKey 相同）
+ consumer-group-id: 334224409
+ # 客户端 ID（建议使用机器 UUID、MAC 地址等唯一标识）
+ client-id: fishery-backend-001
+ # 连接超时时间（毫秒）
+ connection-timeout: 80000
+ # 是否自动重连
+ auto-reconnect: true
+ # 最大重连次数
+ max-reconnect-attempts: 10
+ # 重连延迟（毫秒）
+ reconnect-delay: 30000
+
--- # 三方授权
justauth:
# 前端外网访问地址
diff --git a/intc-admin/src/main/resources/application.yml b/intc-admin/src/main/resources/application.yml
index aed01f2..9f29768 100644
--- a/intc-admin/src/main/resources/application.yml
+++ b/intc-admin/src/main/resources/application.yml
@@ -35,7 +35,7 @@ captcha:
# 日志配置
logging:
level:
- org.dromara: @logging.level@
+ org.dromara: info
org.springframework: warn
org.mybatis.spring.mapper: error
org.apache.fury: warn
@@ -69,7 +69,7 @@ spring:
# 国际化资源文件路径
basename: i18n/messages
profiles:
- active: @profiles.active@
+ active: dev
# 文件上传
servlet:
multipart:
@@ -118,6 +118,9 @@ security:
- /*/api-docs
- /*/api-docs/**
- /warm-flow-ui/config
+ - /iot/test
+ - /iot/device/**
+ - /iot/amqp/**
# 多租户配置
tenant:
diff --git a/intc-common/intc-common-mybatis/src/main/resources/spy.properties b/intc-common/intc-common-mybatis/src/main/resources/spy.properties
index f3ed7d8..dc6e699 100644
--- a/intc-common/intc-common-mybatis/src/main/resources/spy.properties
+++ b/intc-common/intc-common-mybatis/src/main/resources/spy.properties
@@ -17,4 +17,4 @@ databaseDialectTimestampFormat=yyyy-MM-dd HH:mm:ss
# 是否过滤 Log
filter=true
# 过滤 Log 时所排除的 sql 关键字，以逗号分隔
-exclude=
+exclude=information_schema.ins_tables
diff --git a/intc-modules/intc-iot/pom.xml b/intc-modules/intc-iot/pom.xml
index 5f33d62..b37a80a 100644
--- a/intc-modules/intc-iot/pom.xml
+++ b/intc-modules/intc-iot/pom.xml
@@ -31,11 +31,40 @@
7.46.0
-
+
org.eclipse.paho
org.eclipse.paho.client.mqttv3
1.2.5
+ true
+
+
+
+
+ org.apache.qpid
+ qpid-jms-client
+ 0.53.0
+
+
+
+
+ commons-codec
+ commons-codec
+ 1.15
+
+
+
+
+ com.aliyun.mns
+ aliyun-sdk-mns
+ 1.1.9.2
+
+
+
+
+ com.aliyun
+ aliyun-java-sdk-dyvmsapi
+ 1.1.0
@@ -81,11 +110,30 @@
intc-common-web
+
+ com.intc
+ intc-common-excel
+
+
com.intc
intc-common-tenant
+
+
+ com.intc
+ intc-tdengine
+ 5.5.0
+ compile
+
+
+ com.intc
+ intc-fishery
+ 5.5.0
+ compile
+
+
diff --git a/intc-modules/intc-iot/src/main/java/com/intc/iot/config/AliyunIotConfiguration.java b/intc-modules/intc-iot/src/main/java/com/intc/iot/config/AliyunIotConfiguration.java
index a8af4d3..f40b6ef 100644
--- a/intc-modules/intc-iot/src/main/java/com/intc/iot/config/AliyunIotConfiguration.java
+++ b/intc-modules/intc-iot/src/main/java/com/intc/iot/config/AliyunIotConfiguration.java
@@ -6,6 +6,7 @@ import com.aliyuncs.profile.DefaultProfile;
import lombok.RequiredArgsConstructor;
import lombok.extern.slf4j.Slf4j;
import org.springframework.boot.autoconfigure.condition.ConditionalOnProperty;
+import org.springframework.boot.context.properties.EnableConfigurationProperties;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
@@ -17,6 +18,7 @@ import org.springframework.context.annotation.Configuration;
@Slf4j
@Configuration
@RequiredArgsConstructor
+@EnableConfigurationProperties(AliyunIotProperties.class)
@ConditionalOnProperty(prefix = "aliyun.living-iot", name = "access-key-id")
public class AliyunIotConfiguration {
diff --git a/intc-modules/intc-iot/src/main/java/com/intc/iot/config/AliyunIotProperties.java b/intc-modules/intc-iot/src/main/java/com/intc/iot/config/AliyunIotProperties.java
index 475a573..a1b4888 100644
--- a/intc-modules/intc-iot/src/main/java/com/intc/iot/config/AliyunIotProperties.java
+++ b/intc-modules/intc-iot/src/main/java/com/intc/iot/config/AliyunIotProperties.java
@@ -2,7 +2,6 @@ package com.intc.iot.config;
import lombok.Data;
import org.springframework.boot.context.properties.ConfigurationProperties;
-import org.springframework.stereotype.Component;
/**
* 阿里云生活物联网平台（飞燕平台）配置属性
@@ -10,7 +9,6 @@ import org.springframework.stereotype.Component;
* @author intc
*/
@Data
-@Component
@ConfigurationProperties(prefix = "aliyun.living-iot")
public class AliyunIotProperties {
@@ -50,10 +48,15 @@ public class AliyunIotProperties {
private String categoryKey;
/**
- * MQTT 配置
+ * MQTT 配置（可选，用于设备直连）
*/
private MqttConfig mqtt = new MqttConfig();
+ /**
+ * AMQP 服务端订阅配置（推荐）
+ */
+ private AmqpConfig amqp = new AmqpConfig();
+
@Data
public static class MqttConfig {
/**
@@ -97,4 +100,62 @@ public class AliyunIotProperties {
private Boolean cleanSession = true;
}
+ /**
+ * AMQP 服务端订阅配置（使用数据同步的 AppKey + AppSecret）
+ */
+ @Data
+ public static class AmqpConfig {
+ /**
+ * 是否启用 AMQP 订阅
+ */
+ private Boolean enabled = false;
+
+ /**
+ * 数据同步 AppKey（与外层的 appKey 不同！）
+ */
+ private String dataSyncAppKey;
+
+ /**
+ * 数据同步 AppSecret
+ */
+ private String dataSyncAppSecret;
+
+ /**
+ * AMQP 接入点地址（使用 amqps 协议）
+ * 中国内地：amqps://ilop.iot-amqp.cn-shanghai.aliyuncs.com:5671
+ * 新加坡：amqps://ilop.iot-amqp.ap-southeast-1.aliyuncs.com:5671
+ */
+ private String endpoint;
+
+ /**
+ * 消费组 ID
+ */
+ private String consumerGroupId;
+
+ /**
+ * 客户端 ID（建议使用机器 UUID、MAC 地址等唯一标识）
+ */
+ private String clientId;
+
+ /**
+ * 连接超时时间（毫秒）
+ */
+ private Integer connectionTimeout = 80000;
+
+ /**
+ * 自动重连
+ */
+ private Boolean autoReconnect = true;
+
+ /**
+ * 最大重连次数
+ */
+ private Integer maxReconnectAttempts = 10;
+
+ /**
+ * 重连延迟（毫秒）
+ */
+ private Integer reconnectDelay = 30000;
+ }
+
}
diff --git a/intc-modules/intc-iot/src/main/java/com/intc/iot/config/AmqpConfiguration.java b/intc-modules/intc-iot/src/main/java/com/intc/iot/config/AmqpConfiguration.java
new file mode 100644
index 0000000..32cde01
--- /dev/null
+++ b/intc-modules/intc-iot/src/main/java/com/intc/iot/config/AmqpConfiguration.java
@@ -0,0 +1,254 @@
+package com.intc.iot.config;
+
+import com.intc.iot.service.AmqpMessageHandler;
+import lombok.RequiredArgsConstructor;
+import lombok.extern.slf4j.Slf4j;
+import org.apache.qpid.jms.JmsConnection;
+import org.apache.qpid.jms.JmsConnectionListener;
+import org.apache.qpid.jms.message.JmsInboundMessageDispatch;
+import org.springframework.boot.autoconfigure.condition.ConditionalOnProperty;
+import org.springframework.boot.context.properties.EnableConfigurationProperties;
+import org.springframework.context.annotation.Bean;
+import org.springframework.context.annotation.Configuration;
+
+import javax.crypto.Mac;
+import javax.crypto.spec.SecretKeySpec;
+import javax.jms.*;
+import javax.jms.Connection;
+import javax.naming.Context;
+import javax.naming.InitialContext;
+import org.apache.commons.codec.binary.Hex;
+import java.net.URI;
+import java.util.Hashtable;
+import java.util.concurrent.ThreadLocalRandom;
+
+/**
+ * 阿里云 IoT AMQP 服务端订阅配置
+ *
+ * 使用说明：
+ * 1. 在阿里云IoT控制台开启设备数据同步
+ * 2. 配置 AMQP 连接参数（endpoint、consumerGroupId 等）
+ * 3. 启用配置：aliyun.living-iot.amqp.enabled=true
+ *
+ * 注意：
+ * - 阿里云生活物联网平台使用 AMQP 1.0 协议
+ * - 使用 AppKey + AppSecret 认证（不是 AccessKey）
+ * - 连接地址：amqps://ilop.iot-amqp.cn-shanghai.aliyuncs.com:5671
+ *
+ * @author intc
+ */
+@Slf4j
+@Configuration
+@RequiredArgsConstructor
+@EnableConfigurationProperties(AliyunIotProperties.class)
+@ConditionalOnProperty(prefix = "aliyun.living-iot.amqp", name = "enabled", havingValue = "true")
+public class AmqpConfiguration {
+
+ private final AliyunIotProperties iotProperties;
+ private final AmqpMessageHandler messageHandler;
+
+ /**
+ * 创建 AMQP 1.0 连接（使用 JMS）
+ */
+ @Bean
+ public Connection amqpConnection() throws Exception {
+ AliyunIotProperties.AmqpConfig amqp = iotProperties.getAmqp();
+ // 使用数据同步的 AppKey/AppSecret 进行 AMQP 认证
+ String dataSyncAppKey = amqp.getDataSyncAppKey();
+ String dataSyncAppSecret = amqp.getDataSyncAppSecret();
+
+ log.info("========== AMQP 连接配置 ==========");
+ log.info("Endpoint: {}", amqp.getEndpoint());
+ log.info("数据同步 AppKey: {}", dataSyncAppKey);
+ log.info("数据同步 AppSecret: {}...", dataSyncAppSecret != null && dataSyncAppSecret.length() > 4
+ ? dataSyncAppSecret.substring(0, 4) + "****" : "NULL");
+ log.info("ConsumerGroupId: {}", amqp.getConsumerGroupId());
+ log.info("ClientId: {}", amqp.getClientId());
+ log.info("ConnectionTimeout: {}ms", amqp.getConnectionTimeout());
+ log.info("====================================");
+
+ try {
+ // 生成随机数（按照官方示例）
+ long random = ThreadLocalRandom.current().nextLong(Long.MAX_VALUE);
+
+ // 构建用户名（完全按照官方示例）
+ String userName = amqp.getClientId() + "|authMode=appkey"
+ + ",signMethod=SHA256"
+ + ",random=" + random
+ + ",appKey=" + dataSyncAppKey
+ + ",groupId=" + amqp.getConsumerGroupId() + "|";
+
+ // 计算密码（完全按照官方示例）
+ String signContent = "random=" + random;
+ String password = doSign(signContent, dataSyncAppSecret, "HmacSHA256");
+
+ log.info("用户名: {}", userName);
+ log.info("签名内容: {}", signContent);
+ log.info("密码（完整）: {}", password);
+ log.info("密码长度: {}", password.length());
+ log.info("ConsumerGroupId（用于groupId）: {}", amqp.getConsumerGroupId());
+
+ // 构建连接 URL（带故障转移和 SSL 配置）
+ String connectionUrl = String.format(
+ "failover:(%s?amqp.idleTimeout=%d&transport.verifyHost=false)?failover.maxReconnectAttempts=%d&failover.reconnectDelay=%d",
+ amqp.getEndpoint(),
+ amqp.getConnectionTimeout(),
+ amqp.getMaxReconnectAttempts(),
+ amqp.getReconnectDelay()
+ );
+
+ log.info("连接 URL: {}", connectionUrl);
+
+ // 使用 JNDI 初始化连接
+ Hashtable hashtable = new Hashtable<>();
+ hashtable.put("connectionfactory.SBCF", connectionUrl);
+ hashtable.put("queue.QUEUE", "default");
+ hashtable.put(Context.INITIAL_CONTEXT_FACTORY, "org.apache.qpid.jms.jndi.JmsInitialContextFactory");
+
+ Context context = new InitialContext(hashtable);
+ ConnectionFactory cf = (ConnectionFactory) context.lookup("SBCF");
+ Destination queue = (Destination) context.lookup("QUEUE");
+
+ log.info("正在连接到 AMQP 服务器...");
+ log.info("请耐必等待，连接可能需要几秒...");
+
+ // 创建连接
+ Connection connection = cf.createConnection(userName, password);
+
+ // 添加连接监听器
+ if (connection instanceof JmsConnection) {
+ ((JmsConnection) connection).addConnectionListener(new JmsConnectionListener() {
+ @Override
+ public void onConnectionEstablished(URI remoteURI) {
+ log.info("✅ AMQP 连接建立成功: {}", remoteURI);
+ }
+
+ @Override
+ public void onConnectionFailure(Throwable error) {
+ log.error("❌ AMQP 连接失败回调");
+ log.error("失败原因: {}", error.getMessage());
+ log.error("异常类型: {}", error.getClass().getName());
+ if (error.getCause() != null) {
+ log.error("根本原因: {}", error.getCause().getMessage());
+ log.error("根本原因类型: {}", error.getCause().getClass().getName());
+ }
+ // 打印完整堆栈
+ log.error("完整异常信息:", error);
+ }
+
+ @Override
+ public void onConnectionInterrupted(URI remoteURI) {
+ log.warn("⚠️ AMQP 连接中断: {}", remoteURI);
+ }
+
+ @Override
+ public void onConnectionRestored(URI remoteURI) {
+ log.info("✅ AMQP 连接恢复: {}", remoteURI);
+ }
+
+ @Override
+ public void onInboundMessage(JmsInboundMessageDispatch envelope) {}
+
+ @Override
+ public void onSessionClosed(Session session, Throwable cause) {}
+
+ @Override
+ public void onConsumerClosed(MessageConsumer consumer, Throwable cause) {}
+
+ @Override
+ public void onProducerClosed(MessageProducer producer, Throwable cause) {}
+ });
+ }
+
+ // 启动连接
+ connection.start();
+ log.info("✅ AMQP 连接创建成功！");
+
+ // 启动消息消费
+ startConsuming(connection, queue);
+
+ return connection;
+ } catch (Exception e) {
+ log.error("❌ AMQP 连接失败", e);
+ log.error("===== 详细错误信息 =====");
+ log.error("异常类型: {}", e.getClass().getName());
+ log.error("异常信息: {}", e.getMessage());
+
+ // 打印完整异常链
+ Throwable cause = e.getCause();
+ int level = 1;
+ while (cause != null) {
+ log.error("原因 {} - 类型: {}", level, cause.getClass().getName());
+ log.error("原因 {} - 信息: {}", level, cause.getMessage());
+ cause = cause.getCause();
+ level++;
+ }
+
+ log.error("===== 配置信息 =====");
+ log.error(" - Endpoint: {}", amqp.getEndpoint());
+ log.error(" - 数据同步 AppKey: {}", dataSyncAppKey);
+ log.error(" - 数据同步 AppSecret: {}...", dataSyncAppSecret != null && dataSyncAppSecret.length() > 4 ? dataSyncAppSecret.substring(0, 4) + "****" : "NULL");
+ log.error(" - ConsumerGroupId: {}", amqp.getConsumerGroupId());
+ log.error(" - ClientId: {}", amqp.getClientId());
+
+ log.error("===== 可能原因 =====");
+ log.error("1. 网络无法连接到服务器: {}", amqp.getEndpoint());
+ log.error("2. 数据同步 AppKey 或 AppSecret 错误");
+ log.error("3. 消费组 ID 不存在或未授权");
+ log.error("4. 未开启设备数据同步");
+ log.error("5. 防火墙阻止了端口 5671");
+ log.error("6. 签名算法或格式错误");
+
+ throw new RuntimeException("创建 AMQP 连接失败: " + e.getMessage(), e);
+ }
+ }
+
+ /**
+ * 启动消息消费
+ */
+ private void startConsuming(Connection connection, Destination queue) {
+ new Thread(() -> {
+ try {
+ // 创建会话（自动 ACK）
+ Session session = connection.createSession(false, Session.AUTO_ACKNOWLEDGE);
+
+ // 创建消费者
+ MessageConsumer consumer = session.createConsumer(queue);
+
+ log.info("开始消费 AMQP 消息...");
+
+ // 设置消息监听器
+ consumer.setMessageListener(message -> {
+ try {
+ // 阿里云 IoT 发送的是 BytesMessage
+ byte[] body = message.getBody(byte[].class);
+ String content = new String(body, "UTF-8");
+
+ // 处理消息（异步）
+ messageHandler.handleMessage(content);
+
+ } catch (Exception e) {
+ log.error("处理 AMQP 消息失败", e);
+ }
+ });
+
+ } catch (Exception e) {
+ log.error("启动 AMQP 消费失败", e);
+ }
+ }, "AMQP-Consumer").start();
+ }
+
+ /**
+ * 计算签名（完全按照官方示例）
+ */
+ private String doSign(String toSignString, String secret, String signMethod) throws Exception {
+ SecretKeySpec signingKey = new SecretKeySpec(secret.getBytes(), signMethod);
+ Mac mac = Mac.getInstance(signMethod);
+ mac.init(signingKey);
+ byte[] rawHmac = mac.doFinal(toSignString.getBytes());
+
+ // 使用 Hex 编码（按照官方示例）
+ return Hex.encodeHexString(rawHmac);
+ }
+
+}
diff --git a/intc-modules/intc-iot/src/main/java/com/intc/iot/config/MqttConfiguration.java b/intc-modules/intc-iot/src/main/java/com/intc/iot/config/MqttConfiguration.java
index e768c07..81ada4a 100644
--- a/intc-modules/intc-iot/src/main/java/com/intc/iot/config/MqttConfiguration.java
+++ b/intc-modules/intc-iot/src/main/java/com/intc/iot/config/MqttConfiguration.java
@@ -5,6 +5,7 @@ import lombok.extern.slf4j.Slf4j;
import org.eclipse.paho.client.mqttv3.MqttClient;
import org.eclipse.paho.client.mqttv3.MqttConnectOptions;
import org.eclipse.paho.client.mqttv3.persist.MemoryPersistence;
+import org.springframework.boot.autoconfigure.condition.ConditionalOnClass;
import org.springframework.boot.autoconfigure.condition.ConditionalOnProperty;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
@@ -17,6 +18,7 @@ import org.springframework.context.annotation.Configuration;
@Slf4j
@Configuration
@RequiredArgsConstructor
+@ConditionalOnClass(MqttClient.class)
@ConditionalOnProperty(prefix = "aliyun.living-iot.mqtt", name = "broker-url")
public class MqttConfiguration {
@@ -24,26 +26,40 @@ public class MqttConfiguration {
/**
* 创建 MQTT 客户端
+ * 添加容错机制，连接失败时不影响应用启动
*/
@Bean
- public MqttClient mqttClient() throws Exception {
+ public MqttClient mqttClient() {
AliyunIotProperties.MqttConfig mqtt = iotProperties.getMqtt();
- MemoryPersistence persistence = new MemoryPersistence();
- MqttClient client = new MqttClient(mqtt.getBrokerUrl(), mqtt.getClientId(), persistence);
+ try {
+ MemoryPersistence persistence = new MemoryPersistence();
+ MqttClient client = new MqttClient(mqtt.getBrokerUrl(), mqtt.getClientId(), persistence);
- MqttConnectOptions options = new MqttConnectOptions();
- options.setUserName(mqtt.getUsername());
- options.setPassword(mqtt.getPassword().toCharArray());
- options.setConnectionTimeout(mqtt.getConnectionTimeout());
- options.setKeepAliveInterval(mqtt.getKeepAliveInterval());
- options.setAutomaticReconnect(mqtt.getAutoReconnect());
- options.setCleanSession(mqtt.getCleanSession());
+ MqttConnectOptions options = new MqttConnectOptions();
+ options.setUserName(mqtt.getUsername());
+ options.setPassword(mqtt.getPassword().toCharArray());
+ options.setConnectionTimeout(mqtt.getConnectionTimeout());
+ options.setKeepAliveInterval(mqtt.getKeepAliveInterval());
+ options.setAutomaticReconnect(mqtt.getAutoReconnect());
+ options.setCleanSession(mqtt.getCleanSession());
- client.connect(options);
- log.info("MQTT 客户端连接成功，Broker: {}", mqtt.getBrokerUrl());
-
- return client;
+ client.connect(options);
+ log.info("MQTT 客户端连接成功，Broker: {}", mqtt.getBrokerUrl());
+
+ return client;
+ } catch (Exception e) {
+ log.error("MQTT 客户端连接失败，将返回未连接的客户端: {}", e.getMessage());
+
+ try {
+ // 返回未连接的客户端，由 MqttServiceImpl 中的自动重连机制处理
+ MemoryPersistence persistence = new MemoryPersistence();
+ return new MqttClient(mqtt.getBrokerUrl(), mqtt.getClientId(), persistence);
+ } catch (Exception ex) {
+ log.error("MQTT 客户端创建失败，请检查配置: {}", ex.getMessage());
+ throw new RuntimeException("创建 MQTT 客户端失败", ex);
+ }
+ }
}
}
diff --git a/intc-modules/intc-iot/src/main/java/com/intc/iot/config/VmsMnsAutoStarter.java b/intc-modules/intc-iot/src/main/java/com/intc/iot/config/VmsMnsAutoStarter.java
new file mode 100644
index 0000000..79a532e
--- /dev/null
+++ b/intc-modules/intc-iot/src/main/java/com/intc/iot/config/VmsMnsAutoStarter.java
@@ -0,0 +1,41 @@
+package com.intc.iot.config;
+
+import com.intc.iot.service.VmsMnsConsumerService;
+import lombok.RequiredArgsConstructor;
+import lombok.extern.slf4j.Slf4j;
+import org.springframework.boot.ApplicationArguments;
+import org.springframework.boot.ApplicationRunner;
+import org.springframework.boot.autoconfigure.condition.ConditionalOnBean;
+import org.springframework.stereotype.Component;
+
+import jakarta.annotation.PreDestroy;
+
+/**
+ * VMS MNS 消费服务自动启动器
+ * 在 Spring 容器启动后自动启动 MNS 消费
+ *
+ * @author intc-iot
+ */
+@Component
+@RequiredArgsConstructor
+@ConditionalOnBean(VmsMnsConsumerService.class)
+@Slf4j
+public class VmsMnsAutoStarter implements ApplicationRunner {
+
+ private final VmsMnsConsumerService vmsMnsConsumerService;
+
+ @Override
+ public void run(ApplicationArguments args) throws Exception {
+ log.info("自动启动 VMS MNS 回执消费服务");
+ vmsMnsConsumerService.start();
+ }
+
+ /**
+ * 容器销毁时停止服务
+ */
+ @PreDestroy
+ public void onDestroy() {
+ log.info("容器销毁，停止 VMS MNS 回执消费服务");
+ vmsMnsConsumerService.stop();
+ }
+}
diff --git a/intc-modules/intc-iot/src/main/java/com/intc/iot/controller/IotController.java b/intc-modules/intc-iot/src/main/java/com/intc/iot/controller/IotController.java
index ef79aac..a328ebb 100644
--- a/intc-modules/intc-iot/src/main/java/com/intc/iot/controller/IotController.java
+++ b/intc-modules/intc-iot/src/main/java/com/intc/iot/controller/IotController.java
@@ -1,10 +1,20 @@
package com.intc.iot.controller;
import com.intc.common.core.domain.R;
+import com.intc.common.mybatis.core.page.PageQuery;
+import com.intc.common.mybatis.core.page.TableDataInfo;
import com.intc.common.web.core.BaseController;
+import com.intc.iot.domain.bo.DeviceRealtimeDataBo;
+import com.intc.iot.domain.vo.DeviceRealtimeDataVo;
import com.intc.iot.service.DeviceDataService;
+import com.intc.iot.service.DeviceRealtimeDataService;
+import com.intc.iot.service.DeviceStatusService;
import com.intc.iot.service.IotDeviceService;
import com.intc.iot.service.MqttService;
+import com.intc.iot.service.VmsMnsConsumerService;
+import com.intc.iot.service.VmsNoticeService;
+import com.intc.iot.service.WarnCallNoticeService;
+import com.intc.iot.utils.AliyunAmqpSignUtil;
import io.swagger.v3.oas.annotations.Operation;
import io.swagger.v3.oas.annotations.Parameter;
import io.swagger.v3.oas.annotations.tags.Tag;
@@ -36,22 +46,115 @@ public class IotController extends BaseController {
@Autowired(required = false)
private DeviceDataService deviceDataService;
+ @Autowired(required = false)
+ private DeviceRealtimeDataService deviceRealtimeDataService;
+
+ @Autowired(required = false)
+ private DeviceStatusService deviceStatusService;
+
+ @Autowired(required = false)
+ private VmsNoticeService vmsNoticeService;
+
+ @Autowired(required = false)
+ private VmsMnsConsumerService vmsMnsConsumerService;
+
+ @Autowired(required = false)
+ private WarnCallNoticeService warnCallNoticeService;
+
+ @Autowired(required = false)
+ private javax.jms.Connection amqpConnection;
+
@Operation(summary = "测试接口")
@GetMapping("/test")
public R test() {
return R.ok("飞燕平台模块测试成功！");
}
+ @Operation(summary = "查询 AMQP 连接状态")
+ @GetMapping("/amqp/status")
+ public R> getAmqpStatus() {
+ Map status = new java.util.HashMap<>();
+
+ if (amqpConnection == null) {
+ status.put("configured", false);
+ status.put("connected", false);
+ status.put("message", "AMQP 未启用或未配置");
+ } else {
+ try {
+ status.put("configured", true);
+ // JMS Connection 没有直接的 isOpen 方法，通过尝试获取元数据来检查连接
+ javax.jms.ConnectionMetaData metaData = amqpConnection.getMetaData();
+ status.put("connected", true);
+ status.put("provider", metaData.getJMSProviderName());
+ status.put("version", metaData.getProviderVersion());
+ status.put("message", "AMQP 连接正常");
+ } catch (Exception e) {
+ status.put("configured", true);
+ status.put("connected", false);
+ status.put("message", "AMQP 连接已关闭或异常: " + e.getMessage());
+ }
+ }
+
+ return R.ok(status);
+ }
+
+ @Operation(summary = "生成 AMQP 配置信息")
+ @GetMapping("/amqp/generate-config")
+ public R> generateAmqpConfig(
+ @Parameter(description = "AccessKey ID") @RequestParam String accessKeyId,
+ @Parameter(description = "AccessKey Secret") @RequestParam String accessKeySecret,
+ @Parameter(description = "消费组 ID") @RequestParam String consumerGroupId,
+ @Parameter(description = "阿里云账号 UID") @RequestParam String uid,
+ @Parameter(description = "地域 ID（如：cn-shanghai）") @RequestParam String regionId) {
+ try {
+ String host = AliyunAmqpSignUtil.generateHost(uid, regionId);
+ String virtualHost = AliyunAmqpSignUtil.generateVirtualHost(accessKeyId);
+ String username = AliyunAmqpSignUtil.generateUsername(accessKeyId);
+ String password = AliyunAmqpSignUtil.generatePassword(accessKeySecret, consumerGroupId);
+
+ Map config = new java.util.HashMap<>();
+ config.put("host", host);
+ config.put("port", 5672);
+ config.put("virtualHost", virtualHost);
+ config.put("username", username);
+ config.put("password", password);
+ config.put("consumerGroupId", consumerGroupId);
+
+ return R.ok(config);
+ } catch (Exception e) {
+ log.error("生成 AMQP 配置失败", e);
+ return R.fail("生成配置失败: " + e.getMessage());
+ }
+ }
+
+ @Operation(summary = "根据 ProductKey 和 DeviceName 查询设备信息")
+ @GetMapping("/device/find")
+ public R> findDeviceByProductKeyAndName(
+ @Parameter(description = "产品Key") @RequestParam String productKey,
+ @Parameter(description = "设备名称") @RequestParam String deviceName) {
+ try {
+ if (iotDeviceService == null) {
+ return R.fail("飞燕平台配置未启用");
+ }
+ Map response = iotDeviceService.findDeviceByProductKeyAndName(productKey, deviceName);
+ return R.ok(response);
+ } catch (Exception e) {
+ log.error("根据 ProductKey 和 DeviceName 查询设备信息失败", e);
+ return R.fail("查询设备信息失败: " + e.getMessage());
+ }
+ }
+
@Operation(summary = "查询设备列表")
@GetMapping("/device/list")
public R> queryDeviceList(
+ @Parameter(description = "产品Key（必填）") @RequestParam(value = "productKey", required = true) String productKey,
@Parameter(description = "页码") @RequestParam(defaultValue = "1") Integer pageNo,
@Parameter(description = "每页大小") @RequestParam(defaultValue = "20") Integer pageSize) {
try {
if (iotDeviceService == null) {
return R.fail("飞燕平台配置未启用");
}
- Map response = iotDeviceService.queryDeviceList(pageNo, pageSize);
+ Map response = iotDeviceService.queryDeviceList(productKey, pageNo, pageSize);
return R.ok(response);
} catch (Exception e) {
log.error("查询设备列表失败", e);
@@ -103,7 +206,7 @@ public class IotController extends BaseController {
Map response = iotDeviceService.setDeviceProperty(iotId, properties);
return R.ok(response);
} catch (Exception e) {
- log.error("设置设备属性失败", e);
+ log.error("设置设备属性失败: {}", e.getMessage());
return R.fail("设置设备属性失败: " + e.getMessage());
}
}
@@ -121,7 +224,7 @@ public class IotController extends BaseController {
Map response = iotDeviceService.invokeService(iotId, identifier, args);
return R.ok(response);
} catch (Exception e) {
- log.error("调用设备服务失败", e);
+ log.error("调用设备服务失败: {}", e.getMessage());
return R.fail("调用设备服务失败: " + e.getMessage());
}
}
@@ -137,11 +240,65 @@ public class IotController extends BaseController {
Map response = iotDeviceService.unbindDevice(iotId);
return R.ok(response);
} catch (Exception e) {
- log.error("解绑设备失败", e);
+ log.error("解绑设备失败: {}", e.getMessage());
return R.fail("解绑设备失败: " + e.getMessage());
}
}
+ @Operation(summary = "查询设备物模型（模板）")
+ @GetMapping("/device/thing-model")
+ public R> queryThingModel(
+ @Parameter(description = "产品Key") @RequestParam String productKey) {
+ try {
+ if (iotDeviceService == null) {
+ return R.fail("飞燕平台配置未启用");
+ }
+ Map response = iotDeviceService.queryThingModel(productKey);
+ return R.ok(response);
+ } catch (Exception e) {
+ log.error("查询设备物模型失败: {}", e.getMessage());
+ return R.fail("查询设备物模型失败: " + e.getMessage());
+ }
+ }
+
+ @Operation(summary = "查询设备当前状态")
+ @GetMapping("/device/status")
+ public R queryDeviceStatus(
+ @Parameter(description = "产品Key") @RequestParam String productKey,
+ @Parameter(description = "设备名称") @RequestParam String deviceName) {
+ try {
+ if (deviceStatusService == null) {
+ return R.fail("设备状态服务未启用");
+ }
+ com.intc.iot.domain.IotDeviceStatus status = deviceStatusService.queryDeviceStatus(productKey, deviceName);
+ return R.ok(status);
+ } catch (Exception e) {
+ log.error("查询设备状态失败: {}", e.getMessage());
+ return R.fail("查询设备状态失败: " + e.getMessage());
+ }
+ }
+
+ @Operation(summary = "判断设备是否在线")
+ @GetMapping("/device/online")
+ public R> isDeviceOnline(
+ @Parameter(description = "产品Key") @RequestParam String productKey,
+ @Parameter(description = "设备名称") @RequestParam String deviceName) {
+ try {
+ if (deviceStatusService == null) {
+ return R.fail("设备状态服务未启用");
+ }
+ boolean online = deviceStatusService.isDeviceOnline(productKey, deviceName);
+ Map result = new java.util.HashMap<>();
+ result.put("online", online);
+ result.put("productKey", productKey);
+ result.put("deviceName", deviceName);
+ return R.ok(result);
+ } catch (Exception e) {
+ log.error("查询设备在线状态失败: {}", e.getMessage());
+ return R.fail("查询设备在线状态失败: " + e.getMessage());
+ }
+ }
+
@Operation(summary = "发布MQTT消息")
@PostMapping("/mqtt/publish")
public R publishMqtt(
@@ -155,7 +312,7 @@ public class IotController extends BaseController {
mqttService.publish(topic, payload, qos);
return R.ok("消息发布成功");
} catch (Exception e) {
- log.error("发布MQTT消息失败", e);
+ log.error("发布MQTT消息失败: {}", e.getMessage());
return R.fail("发布MQTT消息失败: " + e.getMessage());
}
}
@@ -172,7 +329,7 @@ public class IotController extends BaseController {
mqttService.subscribe(topic, qos);
return R.ok("订阅成功");
} catch (Exception e) {
- log.error("订阅MQTT主题失败", e);
+ log.error("订阅MQTT主题失败: {}", e.getMessage());
return R.fail("订阅MQTT主题失败: " + e.getMessage());
}
}
@@ -188,11 +345,28 @@ public class IotController extends BaseController {
mqttService.unsubscribe(topic);
return R.ok("取消订阅成功");
} catch (Exception e) {
- log.error("取消订阅MQTT主题失败", e);
+ log.error("取消订阅失败: {}", e.getMessage());
return R.fail("取消订阅MQTT主题失败: " + e.getMessage());
}
}
+ @Operation(summary = "订阅设备状态 Topic")
+ @PostMapping("/mqtt/subscribe/status")
+ public R subscribeDeviceStatus(
+ @Parameter(description = "产品Key") @RequestParam String productKey,
+ @Parameter(description = "设备名称") @RequestParam String deviceName) {
+ try {
+ if (mqttService == null) {
+ return R.fail("MQTT配置未启用");
+ }
+ mqttService.subscribeDeviceStatus(productKey, deviceName);
+ return R.ok("订阅设备状态成功");
+ } catch (Exception e) {
+ log.error("订阅设备状态失败: {}", e.getMessage());
+ return R.fail("订阅设备状态失败: " + e.getMessage());
+ }
+ }
+
@Operation(summary = "订阅设备实时数据(按产品)")
@PostMapping("/device/data/subscribe")
public R subscribeDeviceData(
@@ -204,9 +378,208 @@ public class IotController extends BaseController {
deviceDataService.subscribeAllDevices(productKey);
return R.ok("订阅成功，设备数据将实时推送");
} catch (Exception e) {
- log.error("订阅设备数据失败", e);
+ log.error("订阅设备数据失败: {}", e.getMessage());
return R.fail("订阅设备数据失败: " + e.getMessage());
}
}
+ @Operation(summary = "查询设备实时数据列表（分页）")
+ @GetMapping("/device/realtime/list")
+ public R> queryRealtimeDataList(
+ DeviceRealtimeDataBo bo, PageQuery pageQuery) {
+ try {
+ if (deviceRealtimeDataService == null) {
+ return R.fail("设备数据服务未启用");
+ }
+ TableDataInfo dataInfo = deviceRealtimeDataService.queryPageList(bo, pageQuery);
+ return R.ok(dataInfo);
+ } catch (Exception e) {
+ log.error("查询实时数据失败: {}", e.getMessage());
+ return R.fail("查询设备实时数据列表失败: " + e.getMessage());
+ }
+ }
+
+ @Operation(summary = "查询设备最新数据")
+ @GetMapping("/device/realtime/latest")
+ public R queryLatestData(
+ @Parameter(description = "产品Key") @RequestParam String productKey,
+ @Parameter(description = "设备名称") @RequestParam String deviceName) {
+ try {
+ if (deviceRealtimeDataService == null) {
+ return R.fail("设备数据服务未启用");
+ }
+ DeviceRealtimeDataVo data = deviceRealtimeDataService.queryLatestData(productKey, deviceName);
+ return R.ok(data);
+ } catch (Exception e) {
+ log.error("查询最新数据失败: {}", e.getMessage());
+ return R.fail("查询设备最新数据失败: " + e.getMessage());
+ }
+ }
+
+ @Operation(summary = "查询设备最新属性数据")
+ @GetMapping("/device/realtime/latest-property")
+ public R queryLatestPropertyData(
+ @Parameter(description = "产品Key") @RequestParam String productKey,
+ @Parameter(description = "设备名称") @RequestParam String deviceName) {
+ try {
+ if (deviceRealtimeDataService == null) {
+ return R.fail("设备数据服务未启用");
+ }
+ DeviceRealtimeDataVo data = deviceRealtimeDataService.queryLatestPropertyData(productKey, deviceName);
+ return R.ok(data);
+ } catch (Exception e) {
+ log.error("查询最新属性失败: {}", e.getMessage());
+ return R.fail("查询设备最新属性数据失败: " + e.getMessage());
+ }
+ }
+
+ @Operation(summary = "删除设备实时数据")
+ @DeleteMapping("/device/realtime/{ids}")
+ public R deleteRealtimeData(
+ @Parameter(description = "主键ID数组") @PathVariable Long[] ids) {
+ try {
+ if (deviceRealtimeDataService == null) {
+ return R.fail("设备数据服务未启用");
+ }
+ boolean success = deviceRealtimeDataService.deleteByIds(ids);
+ return success ? R.ok() : R.fail("删除失败");
+ } catch (Exception e) {
+ log.error("删除实时数据失败: {}", e.getMessage());
+ return R.fail("删除设备实时数据失败: " + e.getMessage());
+ }
+ }
+
+ // ======================== VMS 语音通知相关接口 ========================
+
+ @Operation(summary = "发送语音通知（测试接口）")
+ @PostMapping("/vms/call")
+ public R> sendVoiceCall(
+ @Parameter(description = "手机号") @RequestParam String phoneNumber,
+ @Parameter(description = "模板参数JSON") @RequestParam String params,
+ @Parameter(description = "业务ID") @RequestParam(required = false) String outId) {
+ try {
+ if (vmsNoticeService == null) {
+ return R.fail("VMS语音服务未启用");
+ }
+ Map paramsMap = cn.hutool.json.JSONUtil.toBean(params, Map.class);
+ com.intc.iot.domain.VmsNoticeResponse response = vmsNoticeService.sendTtsCall(
+ phoneNumber, paramsMap, outId != null ? outId : "TEST"
+ );
+
+ Map result = new java.util.HashMap<>();
+ result.put("success", response.isSuccess());
+ result.put("code", response.getCode());
+ result.put("message", response.getMessage());
+ result.put("callId", response.getCallId());
+
+ return R.ok(result);
+ } catch (Exception e) {
+ log.error("发送语音通知失败: {}", e.getMessage());
+ return R.fail("发送语音通知失败: " + e.getMessage());
+ }
+ }
+
+ @Operation(summary = "启动 VMS MNS 回执消费服务")
+ @PostMapping("/vms/mns/start")
+ public R startVmsMnsConsumer() {
+ try {
+ if (vmsMnsConsumerService == null) {
+ return R.fail("VMS MNS消费服务未配置");
+ }
+ if (vmsMnsConsumerService.isRunning()) {
+ return R.fail("VMS MNS消费服务已在运行中");
+ }
+ vmsMnsConsumerService.start();
+ return R.ok("VMS MNS消费服务启动成功");
+ } catch (Exception e) {
+ log.error("启动VMS失败: {}", e.getMessage());
+ return R.fail("启动失败: " + e.getMessage());
+ }
+ }
+
+ @Operation(summary = "停止 VMS MNS 回执消费服务")
+ @PostMapping("/vms/mns/stop")
+ public R stopVmsMnsConsumer() {
+ try {
+ if (vmsMnsConsumerService == null) {
+ return R.fail("VMS MNS消费服务未配置");
+ }
+ if (!vmsMnsConsumerService.isRunning()) {
+ return R.fail("VMS MNS消费服务未在运行");
+ }
+ vmsMnsConsumerService.stop();
+ return R.ok("VMS MNS消费服务已停止");
+ } catch (Exception e) {
+ log.error("停止VMS失败: {}", e.getMessage());
+ return R.fail("停止失败: " + e.getMessage());
+ }
+ }
+
+ @Operation(summary = "查询 VMS MNS 消费服务运行状态")
+ @GetMapping("/vms/mns/status")
+ public R> getVmsMnsStatus() {
+ try {
+ if (vmsMnsConsumerService == null) {
+ return R.fail("VMS MNS消费服务未配置");
+ }
+ Map status = new java.util.HashMap<>();
+ status.put("running", vmsMnsConsumerService.isRunning());
+ return R.ok(status);
+ } catch (Exception e) {
+ log.error("查询VMS状态失败: {}", e.getMessage());
+ return R.fail("查询失败: " + e.getMessage());
+ }
+ }
+
+ @Operation(summary = "手动触发 VMS 回执处理")
+ @PostMapping("/vms/callback/process")
+ public R> processVmsCallbacks() {
+ try {
+ if (warnCallNoticeService == null) {
+ return R.fail("告警通知服务未启用");
+ }
+ int count = warnCallNoticeService.processUnhandledCallbacks();
+ Map result = new java.util.HashMap<>();
+ result.put("processedCount", count);
+ return R.ok(result);
+ } catch (Exception e) {
+ log.error("处理回执失败: {}", e.getMessage());
+ return R.fail("处理失败: " + e.getMessage());
+ }
+ }
+
+ @Operation(summary = "移除指定设备的待通知记录")
+ @DeleteMapping("/vms/pending/{deviceId}")
+ public R> removePendingNotifications(
+ @Parameter(description = "设备ID") @PathVariable Long deviceId) {
+ try {
+ if (warnCallNoticeService == null) {
+ return R.fail("告警通知服务未启用");
+ }
+ int count = warnCallNoticeService.removePendingNotificationsByDevice(deviceId);
+ Map result = new java.util.HashMap<>();
+ result.put("removedCount", count);
+ return R.ok(result);
+ } catch (Exception e) {
+ log.error("移除待通知失败: {}", e.getMessage());
+ return R.fail("移除失败: " + e.getMessage());
+ }
+ }
+
+ @Operation(summary = "手动清理超时的通知记录")
+ @PostMapping("/vms/cleanup/expired")
+ public R> cleanupExpiredNotifications() {
+ try {
+ if (warnCallNoticeService == null) {
+ return R.fail("告警通知服务未启用");
+ }
+ int count = warnCallNoticeService.cleanupExpiredNotifications();
+ Map result = new java.util.HashMap<>();
+ result.put("cleanedCount", count);
+ return R.ok(result);
+ } catch (Exception e) {
+ log.error("清理超时失败: {}", e.getMessage());
+ return R.fail("清理失败: " + e.getMessage());
+ }
+ }
}
diff --git a/intc-modules/intc-iot/src/main/java/com/intc/iot/domain/AquAlarmHistory.java b/intc-modules/intc-iot/src/main/java/com/intc/iot/domain/AquAlarmHistory.java
new file mode 100644
index 0000000..302d846
--- /dev/null
+++ b/intc-modules/intc-iot/src/main/java/com/intc/iot/domain/AquAlarmHistory.java
@@ -0,0 +1,123 @@
+package com.intc.iot.domain;
+
+import com.baomidou.mybatisplus.annotation.IdType;
+import com.baomidou.mybatisplus.annotation.TableId;
+import com.baomidou.mybatisplus.annotation.TableName;
+import lombok.Data;
+
+import java.io.Serializable;
+import java.time.LocalDateTime;
+
+/**
+ * 水质告警历史记录
+ *
+ * @author intc
+ */
+@Data
+@TableName("aqu_alarm_history")
+public class AquAlarmHistory implements Serializable {
+
+ private static final long serialVersionUID = 1L;
+
+ /**
+ * 主键ID
+ */
+ @TableId(type = IdType.AUTO)
+ private Long id;
+
+ /**
+ * 设备ID
+ */
+ private Long deviceId;
+
+ /**
+ * 设备名称
+ */
+ private String deviceName;
+
+ /**
+ * 用户ID
+ */
+ private Long userId;
+
+ /**
+ * 塘口名称
+ */
+ private String pondName;
+
+ /**
+ * 告警类型（溶解氧、温度、pH、盐度、电池等）
+ */
+ private String alarmType;
+
+ /**
+ * 告警级别：1-一般 2-重要 3-紧急
+ */
+ private Integer alarmLevel;
+
+ /**
+ * 告警内容
+ */
+ private String alarmContent;
+
+ /**
+ * 告警值
+ */
+ private Double alarmValue;
+
+ /**
+ * 阈值（最小或最大）
+ */
+ private Double thresholdValue;
+
+ /**
+ * 告警状态：0-未处理 1-已处理 2-已恢复
+ */
+ private Integer alarmStatus;
+
+ /**
+ * 告警时间
+ */
+ private LocalDateTime alarmTime;
+
+ /**
+ * 恢复时间
+ */
+ private LocalDateTime recoveryTime;
+
+ /**
+ * 处理时间
+ */
+ private LocalDateTime handleTime;
+
+ /**
+ * 处理人
+ */
+ private Long handleBy;
+
+ /**
+ * 处理备注
+ */
+ private String handleRemark;
+
+ /**
+ * 是否已通知：0-未通知 1-已通知
+ */
+ private Integer notified;
+
+ /**
+ * 通知记录ID（关联 aqu_call_notice）
+ */
+ private Long noticeId;
+
+ /**
+ * 创建时间
+ */
+ private LocalDateTime createdTime;
+
+ /**
+ * 更新时间
+ */
+ private LocalDateTime updatedTime;
+
+}
diff --git a/intc-modules/intc-iot/src/main/java/com/intc/iot/domain/AquWarnCallNotice.java b/intc-modules/intc-iot/src/main/java/com/intc/iot/domain/AquWarnCallNotice.java
new file mode 100644
index 0000000..8acdf65
--- /dev/null
+++ b/intc-modules/intc-iot/src/main/java/com/intc/iot/domain/AquWarnCallNotice.java
@@ -0,0 +1,144 @@
+package com.intc.iot.domain;
+
+import com.baomidou.mybatisplus.annotation.IdType;
+import com.baomidou.mybatisplus.annotation.TableId;
+import com.baomidou.mybatisplus.annotation.TableName;
+import lombok.Data;
+
+import java.io.Serializable;
+import java.time.LocalDateTime;
+
+/**
+ * 告警电话通知记录
+ * 对应 C# DbAquWarnCallNotice
+ *
+ * @author intc-iot
+ */
+@Data
+@TableName("aqu_call_notice")
+public class AquWarnCallNotice implements Serializable {
+
+ private static final long serialVersionUID = 1L;
+
+ /**
+ * 主键ID
+ */
+ @TableId(type = IdType.AUTO)
+ private Long id;
+
+ /**
+ * 用户ID
+ */
+ private Long userId;
+
+ /**
+ * 通知手机号
+ */
+ private String mobilePhone;
+
+ /**
+ * 设备ID
+ */
+ private Long deviceId;
+
+ /**
+ * 塘口名称
+ */
+ private String pondName;
+
+ /**
+ * 设定的呼叫时间
+ */
+ private LocalDateTime callTime;
+
+ /**
+ * 呼叫状态
+ * 0-未呼叫(NoCall)
+ * 1-呼叫成功，等待结果反馈(CallSuccess)
+ * 2-呼叫失败(CallFail)
+ * 3-呼叫反馈成功(CallBackSuccess)
+ * 4-呼叫反馈失败(CallBackFail)
+ */
+ private Integer callStatus;
+
+ /**
+ * 呼叫的CallId，CallStatus为1/3/4时有值
+ */
+ private String callId;
+
+ // ==================== 以下为呼叫回执字段 ====================
+
+ /**
+ * 呼叫结果状态码
+ * 200000: 用户听完语音
+ * 200001: 用户未听完挂断
+ * 其他: 失败
+ */
+ private String statusCode;
+
+ /**
+ * 挂断方向（user: 用户挂断, system: 系统挂断）
+ */
+ private String hangupDirection;
+
+ /**
+ * 主叫号码
+ */
+ private String caller;
+
+ /**
+ * 被叫响铃时间
+ */
+ private String ringTime;
+
+ /**
+ * 通话时长（秒），未接通时为0
+ */
+ private String duration;
+
+ /**
+ * 话单类型（voice-普通话单，asr-asr话单，smart_transfer-智能外呼转接话单）
+ */
+ private String voiceType;
+
+ /**
+ * 呼叫发起时间
+ */
+ private String originateTime;
+
+ /**
+ * 通话接通时间
+ */
+ private String startTime;
+
+ /**
+ * 通话结束时间
+ */
+ private String endTime;
+
+ /**
+ * 结果描述
+ */
+ private String statusMsg;
+
+ /**
+ * 扩展字段回传
+ */
+ private String outId;
+
+ /**
+ * 通话类型（LOCAL-市话，PROVINCE-省内长途，DOMESTIC-国内长途，INTERNATIONAL-国际长途，UNKNOWN-未知）
+ */
+ private String tollType;
+
+ /**
+ * 创建时间
+ */
+ private LocalDateTime createdTime;
+
+ /**
+ * 更新时间
+ */
+ private LocalDateTime updatedTime;
+
+}
diff --git a/intc-modules/intc-iot/src/main/java/com/intc/iot/domain/DeviceRealtimeData.java b/intc-modules/intc-iot/src/main/java/com/intc/iot/domain/DeviceRealtimeData.java
new file mode 100644
index 0000000..39b73e3
--- /dev/null
+++ b/intc-modules/intc-iot/src/main/java/com/intc/iot/domain/DeviceRealtimeData.java
@@ -0,0 +1,82 @@
+package com.intc.iot.domain;
+
+import com.baomidou.mybatisplus.annotation.IdType;
+import com.baomidou.mybatisplus.annotation.TableId;
+import com.baomidou.mybatisplus.annotation.TableName;
+import com.intc.common.mybatis.core.domain.BaseEntity;
+import lombok.Data;
+import lombok.EqualsAndHashCode;
+
+import java.io.Serial;
+import java.time.LocalDateTime;
+
+/**
+ * 设备实时数据
+ *
+ * @author intc
+ */
+@Data
+@EqualsAndHashCode(callSuper = true)
+@TableName("iot_device_realtime_data")
+public class DeviceRealtimeData extends BaseEntity {
+
+ @Serial
+ private static final long serialVersionUID = 1L;
+
+ /**
+ * 主键ID
+ */
+ @TableId(type = IdType.AUTO)
+ private Long id;
+
+ /**
+ * 产品Key
+ */
+ private String productKey;
+
+ /**
+ * 设备名称
+ */
+ private String deviceName;
+
+ /**
+ * IoT平台设备ID
+ */
+ private String iotId;
+
+ /**
+ * 消息ID
+ */
+ private String messageId;
+
+ /**
+ * 数据类型：property-属性，event-事件
+ */
+ private String dataType;
+
+ /**
+ * 属性/事件标识符
+ */
+ private String identifier;
+
+ /**
+ * 数据内容（JSON格式）
+ */
+ private String dataContent;
+
+ /**
+ * 数据上报时间
+ */
+ private LocalDateTime reportTime;
+
+ /**
+ * 原始Topic
+ */
+ private String topic;
+
+ /**
+ * 备注
+ */
+ private String remark;
+
+}
diff --git a/intc-modules/intc-iot/src/main/java/com/intc/iot/domain/IotDeviceStatus.java b/intc-modules/intc-iot/src/main/java/com/intc/iot/domain/IotDeviceStatus.java
new file mode 100644
index 0000000..b0e6f34
--- /dev/null
+++ b/intc-modules/intc-iot/src/main/java/com/intc/iot/domain/IotDeviceStatus.java
@@ -0,0 +1,79 @@
+package com.intc.iot.domain;
+
+import com.baomidou.mybatisplus.annotation.IdType;
+import com.baomidou.mybatisplus.annotation.TableId;
+import com.baomidou.mybatisplus.annotation.TableName;
+import lombok.Data;
+
+import java.io.Serializable;
+import java.time.LocalDateTime;
+
+/**
+ * 设备上下线状态记录
+ * 对应 C# IOTTopicDeviceStatus
+ *
+ * @author intc-iot
+ */
+@Data
+@TableName("iot_device_status")
+public class IotDeviceStatus implements Serializable {
+
+ private static final long serialVersionUID = 1L;
+
+ /**
+ * 主键ID
+ */
+ @TableId(type = IdType.AUTO)
+ private Long id;
+
+ /**
+ * 产品Key
+ */
+ private String productKey;
+
+ /**
+ * 设备名称
+ */
+ private String deviceName;
+
+ /**
+ * 设备IotId
+ */
+ private String iotId;
+
+ /**
+ * 状态（online-在线，offline-离线）
+ */
+ private String status;
+
+ /**
+ * 状态变更时间（秒时间戳）
+ */
+ private Long statusTime;
+
+ /**
+ * 上次更新时间（秒时间戳）
+ */
+ private Long lastTime;
+
+ /**
+ * 客户端IP
+ */
+ private String clientIp;
+
+ /**
+ * 用户标识（可选）
+ */
+ private Long userId;
+
+ /**
+ * 创建时间
+ */
+ private LocalDateTime createdTime;
+
+ /**
+ * 更新时间
+ */
+ private LocalDateTime updatedTime;
+
+}
diff --git a/intc-modules/intc-iot/src/main/java/com/intc/iot/domain/VmsCallback.java b/intc-modules/intc-iot/src/main/java/com/intc/iot/domain/VmsCallback.java
new file mode 100644
index 0000000..bcc43fe
--- /dev/null
+++ b/intc-modules/intc-iot/src/main/java/com/intc/iot/domain/VmsCallback.java
@@ -0,0 +1,110 @@
+package com.intc.iot.domain;
+
+import com.baomidou.mybatisplus.annotation.IdType;
+import com.baomidou.mybatisplus.annotation.TableId;
+import com.baomidou.mybatisplus.annotation.TableName;
+import lombok.Data;
+
+import java.io.Serializable;
+import java.time.LocalDateTime;
+
+/**
+ * VMS 语音回执实体
+ * 对应 C# AliVmsCallBack
+ *
+ * @author intc-iot
+ */
+@Data
+@TableName("vms_callback")
+public class VmsCallback implements Serializable {
+
+ private static final long serialVersionUID = 1L;
+
+ /**
+ * 主键ID
+ */
+ @TableId(type = IdType.AUTO)
+ private Long id;
+
+ /**
+ * 呼叫唯一标识
+ */
+ private String callId;
+
+ /**
+ * 业务扩展字段（用于区分系统）
+ */
+ private String outId;
+
+ /**
+ * 状态码
+ * 200000: 用户接听并听完
+ * 200001: 用户未听完挂断
+ * 其他: 失败
+ */
+ private String statusCode;
+
+ /**
+ * 状态描述
+ */
+ private String statusMsg;
+
+ /**
+ * 挂断方向
+ * user: 用户挂断
+ * system: 系统挂断
+ */
+ private String hangupDirection;
+
+ /**
+ * 被叫号码
+ */
+ private String caller;
+
+ /**
+ * 呼叫类型
+ * tts: 文本转语音
+ */
+ private String voiceType;
+
+ /**
+ * 通话发起时间（秒时间戳）
+ */
+ private Long originateTime;
+
+ /**
+ * 通话开始时间（秒时间戳）
+ */
+ private Long startTime;
+
+ /**
+ * 通话结束时间（秒时间戳）
+ */
+ private Long endTime;
+
+ /**
+ * 振铃时长（秒）
+ */
+ private Integer ringTime;
+
+ /**
+ * 通话时长（秒）
+ */
+ private Integer duration;
+
+ /**
+ * 计费类型
+ */
+ private String tollType;
+
+ /**
+ * 是否已处理（0-未处理，1-已处理）
+ */
+ private Integer processed;
+
+ /**
+ * 创建时间
+ */
+ private LocalDateTime createdTime;
+
+}
diff --git a/intc-modules/intc-iot/src/main/java/com/intc/iot/domain/VmsNoticeResponse.java b/intc-modules/intc-iot/src/main/java/com/intc/iot/domain/VmsNoticeResponse.java
new file mode 100644
index 0000000..96ec8ff
--- /dev/null
+++ b/intc-modules/intc-iot/src/main/java/com/intc/iot/domain/VmsNoticeResponse.java
@@ -0,0 +1,22 @@
+package com.intc.iot.domain;
+
+import lombok.Data;
+
+/**
+ * 语音通知返回结果（Java 版）。
+ */
+@Data
+public class VmsNoticeResponse {
+
+ /** 是否调用成功 */
+ private boolean success;
+
+ /** 阿里云返回的状态码 */
+ private String code;
+
+ /** 状态码描述信息 */
+ private String message;
+
+ /** 呼叫ID */
+ private String callId;
+}
diff --git a/intc-modules/intc-iot/src/main/java/com/intc/iot/domain/bo/DeviceRealtimeDataBo.java b/intc-modules/intc-iot/src/main/java/com/intc/iot/domain/bo/DeviceRealtimeDataBo.java
new file mode 100644
index 0000000..e778228
--- /dev/null
+++ b/intc-modules/intc-iot/src/main/java/com/intc/iot/domain/bo/DeviceRealtimeDataBo.java
@@ -0,0 +1,66 @@
+package com.intc.iot.domain.bo;
+
+import com.intc.common.mybatis.core.domain.BaseEntity;
+import com.intc.iot.domain.DeviceRealtimeData;
+import io.github.linpeilie.annotations.AutoMapper;
+import jakarta.validation.constraints.NotBlank;
+import lombok.Data;
+import lombok.EqualsAndHashCode;
+
+import java.io.Serial;
+import java.time.LocalDateTime;
+
+/**
+ * 设备实时数据业务对象
+ *
+ * @author intc
+ */
+@Data
+@EqualsAndHashCode(callSuper = true)
+@AutoMapper(target = DeviceRealtimeData.class, reverseConvertGenerate = false)
+public class DeviceRealtimeDataBo extends BaseEntity {
+
+ @Serial
+ private static final long serialVersionUID = 1L;
+
+ /**
+ * 主键ID
+ */
+ private Long id;
+
+ /**
+ * 产品Key
+ */
+ private String productKey;
+
+ /**
+ * 设备名称
+ */
+ private String deviceName;
+
+ /**
+ * IoT平台设备ID
+ */
+ private String iotId;
+
+ /**
+ * 数据类型：property-属性，event-事件
+ */
+ private String dataType;
+
+ /**
+ * 属性/事件标识符
+ */
+ private String identifier;
+
+ /**
+ * 开始时间
+ */
+ private LocalDateTime startTime;
+
+ /**
+ * 结束时间
+ */
+ private LocalDateTime endTime;
+
+}
diff --git a/intc-modules/intc-iot/src/main/java/com/intc/iot/domain/vo/DeviceRealtimeDataVo.java b/intc-modules/intc-iot/src/main/java/com/intc/iot/domain/vo/DeviceRealtimeDataVo.java
new file mode 100644
index 0000000..8f5a409
--- /dev/null
+++ b/intc-modules/intc-iot/src/main/java/com/intc/iot/domain/vo/DeviceRealtimeDataVo.java
@@ -0,0 +1,99 @@
+package com.intc.iot.domain.vo;
+
+
+import cn.idev.excel.annotation.ExcelIgnoreUnannotated;
+import cn.idev.excel.annotation.ExcelProperty;
+import com.intc.iot.domain.DeviceRealtimeData;
+import io.github.linpeilie.annotations.AutoMapper;
+import lombok.Data;
+
+import java.io.Serial;
+import java.io.Serializable;
+import java.time.LocalDateTime;
+
+/**
+ * 设备实时数据视图对象
+ *
+ * @author intc
+ */
+@Data
+@ExcelIgnoreUnannotated
+@AutoMapper(target = DeviceRealtimeData.class)
+public class DeviceRealtimeDataVo implements Serializable {
+
+ @Serial
+ private static final long serialVersionUID = 1L;
+
+ /**
+ * 主键ID
+ */
+ @ExcelProperty(value = "主键ID")
+ private Long id;
+
+ /**
+ * 产品Key
+ */
+ @ExcelProperty(value = "产品Key")
+ private String productKey;
+
+ /**
+ * 设备名称
+ */
+ @ExcelProperty(value = "设备名称")
+ private String deviceName;
+
+ /**
+ * IoT平台设备ID
+ */
+ @ExcelProperty(value = "设备ID")
+ private String iotId;
+
+ /**
+ * 消息ID
+ */
+ @ExcelProperty(value = "消息ID")
+ private String messageId;
+
+ /**
+ * 数据类型：property-属性，event-事件
+ */
+ @ExcelProperty(value = "数据类型")
+ private String dataType;
+
+ /**
+ * 属性/事件标识符
+ */
+ @ExcelProperty(value = "标识符")
+ private String identifier;
+
+ /**
+ * 数据内容（JSON格式）
+ */
+ @ExcelProperty(value = "数据内容")
+ private String dataContent;
+
+ /**
+ * 数据上报时间
+ */
+ @ExcelProperty(value = "上报时间")
+ private LocalDateTime reportTime;
+
+ /**
+ * 原始Topic
+ */
+ @ExcelProperty(value = "Topic")
+ private String topic;
+
+ /**
+ * 备注
+ */
+ @ExcelProperty(value = "备注")
+ private String remark;
+
+ /**
+ * 创建时间
+ */
+ @ExcelProperty(value = "创建时间")
+ private LocalDateTime createTime;
+
+}
diff --git a/intc-modules/intc-iot/src/main/java/com/intc/iot/handler/DeviceDataHandler.java b/intc-modules/intc-iot/src/main/java/com/intc/iot/handler/DeviceDataHandler.java
index f70d590..53904ba 100644
--- a/intc-modules/intc-iot/src/main/java/com/intc/iot/handler/DeviceDataHandler.java
+++ b/intc-modules/intc-iot/src/main/java/com/intc/iot/handler/DeviceDataHandler.java
@@ -1,10 +1,29 @@
package com.intc.iot.handler;
+import cn.hutool.core.util.StrUtil;
import cn.hutool.json.JSONObject;
import cn.hutool.json.JSONUtil;
+import com.intc.iot.domain.AquAlarmHistory;
+import com.intc.iot.domain.AquWarnCallNotice;
+import com.intc.iot.domain.VmsNoticeResponse;
+import com.intc.iot.mapper.AquAlarmHistoryMapper;
+import com.intc.iot.mapper.AquWarnCallNoticeMapper;
+import com.intc.iot.mapper.IotDeviceMapper;
+import com.intc.iot.service.VmsNoticeService;
+import com.intc.fishery.domain.Device;
+import com.intc.fishery.mapper.DeviceMapper;
+import com.intc.tdengine.domain.DeviceSensorData;
+import com.intc.tdengine.service.IDeviceSensorDataService;
+import lombok.RequiredArgsConstructor;
import lombok.extern.slf4j.Slf4j;
+import org.springframework.boot.autoconfigure.condition.ConditionalOnBean;
import org.springframework.stereotype.Component;
+import java.time.LocalDateTime;
+import java.time.format.DateTimeFormatter;
+import java.util.HashMap;
+import java.util.Map;
+
/**
* 设备数据处理器
*
@@ -12,8 +31,76 @@ import org.springframework.stereotype.Component;
*/
@Slf4j
@Component
+@RequiredArgsConstructor
+@ConditionalOnBean(IDeviceSensorDataService.class)
public class DeviceDataHandler {
+ /**
+ * TDengine 服务，用于存储设备时序数据
+ */
+ private final IDeviceSensorDataService deviceSensorDataService;
+
+ /**
+ * 告警通知 Mapper
+ */
+ private final AquWarnCallNoticeMapper warnCallNoticeMapper;
+
+ /**
+ * 告警历史记录 Mapper
+ */
+ private final AquAlarmHistoryMapper alarmHistoryMapper;
+
+ /**
+ * IoT设备 Mapper
+ */
+ private final IotDeviceMapper iotDeviceMapper;
+
+ /**
+ * 设备 Mapper
+ */
+ private final DeviceMapper deviceMapper;
+
+ /**
+ * VMS 语音通知服务
+ */
+ private final VmsNoticeService vmsNoticeService;
+
+ private static final DateTimeFormatter DATETIME_FORMATTER = DateTimeFormatter.ofPattern("yyyy-MM-dd HH:mm:ss");
+
+ /**
+ * 呼叫状态：未呼叫
+ */
+ private static final int CALL_STATUS_NO_CALL = 0;
+
+ /**
+ * 告警状态：未处理
+ */
+ private static final int ALARM_STATUS_PENDING = 0;
+
+ /**
+ * 告警状态：已恢复
+ */
+ private static final int ALARM_STATUS_RECOVERED = 2;
+
+ /**
+ * 告警间隔时间（分钟）- 同一设备同一类型告警在此时间内不重复发送
+ */
+ private static final int ALARM_NOTIFICATION_INTERVAL = 30;
+
+ /**
+ * 告警类型常量
+ */
+ private static final String ALARM_TYPE_DISSOLVED_OXYGEN = "溶解氧";
+ private static final String ALARM_TYPE_TEMPERATURE = "温度";
+ private static final String ALARM_TYPE_BATTERY = "电池电量";
+
+ /**
+ * 告警级别
+ */
+ private static final int ALARM_LEVEL_NORMAL = 1; // 一般
+ private static final int ALARM_LEVEL_IMPORTANT = 2; // 重要
+ private static final int ALARM_LEVEL_URGENT = 3; // 紧急
+
/**
* 处理设备属性上报数据
*
@@ -21,32 +108,60 @@ public class DeviceDataHandler {
* @param payload 消息内容
*/
public void handlePropertyPost(String topic, String payload) {
- log.info("收到设备属性上报，Topic: {}", topic);
-
try {
JSONObject data = JSONUtil.parseObj(payload);
-
+
// 解析飞燕平台消息格式
- String method = data.getStr("method");
- String id = data.getStr("id");
JSONObject params = data.getJSONObject("params");
-
- log.info("设备属性数据 - Method: {}, ID: {}, Params: {}", method, id, params);
-
- // TODO: 这里添加您的业务逻辑
- // 1. 存储到数据库
- // 2. 触发告警
- // 3. 推送到前端（通过 WebSocket/SSE）
- // 4. 数据分析处理
-
- if (params != null) {
- params.forEach((key, value) -> {
- log.info("属性: {} = {}", key, value);
- // 根据不同属性进行不同处理
- handleProperty(key, value);
- });
+
+ // 从Topic中解析产品Key和设备名称
+ // Topic格式: /sys/{ProductKey}/{DeviceName}/thing/event/property/post
+ String[] topicParts = topic.split("/");
+ if (topicParts.length < 4) {
+ log.warn("设备属性数据 Topic 格式不正确: {}", topic);
+ return;
}
-
+
+ String productKey = topicParts[2];
+ String deviceName = topicParts[3];
+
+ if (productKey == null || productKey.isEmpty() || deviceName == null || deviceName.isEmpty()) {
+ log.warn("设备属性数据 ProductKey 或 DeviceName 为空: {}", topic);
+ return;
+ }
+
+ // 保存到 TDengine 时序数据库
+ DeviceSensorData sensorData = null;
+ if (params != null && params.size() > 3) {
+// // 检查 sensorerrorcode 是否为 6
+// Object errorCode = params.get("sensorerrorcode");
+// if (errorCode == null || !"6".equals(String.valueOf(errorCode))) {
+// log.debug("数据误码不为6，已过滤: {} (errorCode: {})", deviceName, errorCode);
+// return;
+// }
+
+ try {
+ sensorData = convertToSensorData(productKey, deviceName, params);
+ if (sensorData != null) {
+ // 1. 插入到 TDengine
+ deviceSensorDataService.batchInsertDeviceSensorData(java.util.Collections.singletonList(sensorData));
+ log.debug("数据已保存: {}", deviceName);
+
+ // 2. 更新设备表的实时数据
+ updateDeviceRealTimeData(deviceName, sensorData);
+ }
+ } catch (Exception e) {
+ log.error("保存数据失败[{}]: {}", deviceName, e.getMessage());
+ }
+ } else if (params != null && params.size() < 3) {
+ log.debug("数据字段不足3个，已过滤: {} (字段数: {})", deviceName, params.size());
+ }
+//
+// // 检查是否触发报警
+// if (params != null && sensorData != null) {
+// checkAndTriggerAlarm(productKey, deviceName, sensorData);
+// }
+
} catch (Exception e) {
log.error("处理设备属性数据失败", e);
}
@@ -59,76 +174,602 @@ public class DeviceDataHandler {
* @param payload 消息内容
*/
public void handleEventPost(String topic, String payload) {
- log.info("收到设备事件上报，Topic: {}", topic);
-
try {
JSONObject data = JSONUtil.parseObj(payload);
-
- String method = data.getStr("method");
- String id = data.getStr("id");
+
JSONObject params = data.getJSONObject("params");
-
- log.info("设备事件数据 - Method: {}, ID: {}, Params: {}", method, id, params);
-
+
+ // 从Topic中解析产品Key、设备名称和事件标识符
+ // Topic格式: /sys/{ProductKey}/{DeviceName}/thing/event/{EventIdentifier}/post
+ String[] topicParts = topic.split("/");
+ if (topicParts.length < 7) {
+ log.warn("设备事件数据 Topic 格式不正确: {}", topic);
+ return;
+ }
+
+ String productKey = topicParts[2];
+ String deviceName = topicParts[3];
+ String eventIdentifier = topicParts[6];
+
+ if (productKey == null || productKey.isEmpty() ||
+ deviceName == null || deviceName.isEmpty() ||
+ eventIdentifier == null || eventIdentifier.isEmpty()) {
+ log.warn("设备事件数据 ProductKey、DeviceName 或 EventIdentifier 为空: {}", topic);
+ return;
+ }
+
+ // 保存到 TDengine 时序数据库
+ DeviceSensorData sensorData = new DeviceSensorData();
+
+ // 时间字段
+ String now = LocalDateTime.now().format(DATETIME_FORMATTER);
+ sensorData.setTime(now);
+ sensorData.setCreateTime(now);
+
+ // 设备标识
+ sensorData.setSerialNum(deviceName);
+ sensorData.setDeviceName(deviceName);
+
+ // 存储完整的事件数据（可以根据需要扩展）
+ // 目前暂不处理事件数据的详细字段映射
+
+ try {
+ deviceSensorDataService.batchInsertDeviceSensorData(java.util.Collections.singletonList(sensorData));
+ log.debug("事件已保存: {} - {}", deviceName, eventIdentifier);
+ } catch (Exception e) {
+ log.error("保存事件失败[{}]: {}", deviceName, e.getMessage());
+ }
+
// TODO: 处理设备事件
// 例如：故障告警、状态变化等
-
+ handleDeviceEvent(productKey, deviceName, eventIdentifier, params);
+
} catch (Exception e) {
log.error("处理设备事件数据失败", e);
}
}
+
/**
- * 处理单个属性
+ * 处理设备事件
*
- * @param propertyName 属性名称
- * @param propertyValue 属性值
+ * @param productKey 产品Key
+ * @param deviceName 设备名称
+ * @param eventIdentifier 事件标识符
+ * @param params 事件参数
*/
- private void handleProperty(String propertyName, Object propertyValue) {
- // 根据属性名称进行不同的业务处理
- switch (propertyName) {
- case "temperature":
- handleTemperature(propertyValue);
+ private void handleDeviceEvent(String productKey, String deviceName, String eventIdentifier, JSONObject params) {
+ // 根据事件类型进行不同处理
+ switch (eventIdentifier) {
+ case "error":
+ // 故障事件
+ log.warn("设备故障告警: {}", params);
+ // 直接触发电话告警
+ triggerDeviceErrorAlarm(deviceName, params);
break;
- case "humidity":
- handleHumidity(propertyValue);
- break;
- case "status":
- handleStatus(propertyValue);
+ case "alert":
+ // 告警事件
+ log.warn("设备告警: {}", params);
break;
default:
- log.debug("未处理的属性: {} = {}", propertyName, propertyValue);
+ log.debug("未处理的事件类型: {}", eventIdentifier);
+ }
+ }
+
+ // ==================== 报警逻辑 ====================
+
+ /**
+ * 检查并触发报警
+ * 根据传感器数据判断是否超过阈值，并触发电话通知
+ *
+ * @param productKey 产品Key
+ * @param deviceName 设备名称
+ * @param sensorData 传感器数据
+ */
+ private void checkAndTriggerAlarm(String productKey, String deviceName, DeviceSensorData sensorData) {
+ try {
+ // 查询设备信息，获取设备配置和告警阈值
+ Device device = deviceMapper.selectOne(
+ new com.baomidou.mybatisplus.core.conditions.query.LambdaQueryWrapper()
+ .eq(Device::getSerialNum, deviceName)
+ .last("LIMIT 1")
+ );
+
+ if (device == null) {
+ return;
+ }
+
+ Long deviceId = device.getId();
+ StringBuilder alarmMessage = new StringBuilder();
+ boolean hasAlarm = false;
+ java.util.List alarmList = new java.util.ArrayList<>();
+
+ // 1. 检查溶解氧（使用设备配置的阈值）
+ if (sensorData.getDissolvedOxygen() != null && device.getOxyWarnCallOpen() != null && device.getOxyWarnCallOpen() == 1) {
+ Double dissolvedOxygen = sensorData.getDissolvedOxygen();
+ Double oxyWarnLower = device.getOxyWarnLower();
+
+ if (oxyWarnLower != null && dissolvedOxygen < oxyWarnLower) {
+ String content = String.format("溶解氧过低: %.2f mg/L (最低: %.2f)",
+ dissolvedOxygen, oxyWarnLower);
+ alarmMessage.append(content).append("; ");
+
+ AquAlarmHistory alarm = createAlarmHistory(deviceId, deviceName, ALARM_TYPE_DISSOLVED_OXYGEN,
+ ALARM_LEVEL_URGENT, content, dissolvedOxygen, oxyWarnLower);
+ alarmList.add(alarm);
+ hasAlarm = true;
+ }
+ }
+
+ // 2. 检查水温（使用设备配置的阈值）
+ if (sensorData.getTemperature() != null && device.getTempWarnCallOpen() != null && device.getTempWarnCallOpen() == 1) {
+ Double temperature = sensorData.getTemperature();
+ Double tempWarnLower = device.getTempWarnLower();
+ Double tempWarnUpper = device.getTempWarnUpper();
+
+ if (tempWarnLower != null && temperature < tempWarnLower) {
+ String content = String.format("水温过低: %.2f °C (最低: %.2f)",
+ temperature, tempWarnLower);
+ alarmMessage.append(content).append("; ");
+
+ AquAlarmHistory alarm = createAlarmHistory(deviceId, deviceName, ALARM_TYPE_TEMPERATURE,
+ ALARM_LEVEL_IMPORTANT, content, temperature, tempWarnLower);
+ alarmList.add(alarm);
+ hasAlarm = true;
+ } else if (tempWarnUpper != null && temperature > tempWarnUpper) {
+ String content = String.format("水温过高: %.2f °C (最高: %.2f)",
+ temperature, tempWarnUpper);
+ alarmMessage.append(content).append("; ");
+
+ AquAlarmHistory alarm = createAlarmHistory(deviceId, deviceName, ALARM_TYPE_TEMPERATURE,
+ ALARM_LEVEL_IMPORTANT, content, temperature, tempWarnUpper);
+ alarmList.add(alarm);
+ hasAlarm = true;
+ }
+ }
+
+ // 3. 检查电池电量（使用设备配置的阈值）
+ if (sensorData.getBattery() != null && device.getBatteryWarnCallOpen() != null && device.getBatteryWarnCallOpen() == 1) {
+ Double battery = sensorData.getBattery();
+ Double batteryWarnLower = device.getBatteryWarnLower() != null ? device.getBatteryWarnLower().doubleValue() : null;
+
+ if (batteryWarnLower != null && battery < batteryWarnLower) {
+ String content = String.format("电池电量低: %.2f%% (最低: %.2f%%)",
+ battery, batteryWarnLower);
+ alarmMessage.append(content).append("; ");
+
+ AquAlarmHistory alarm = createAlarmHistory(deviceId, deviceName, ALARM_TYPE_BATTERY,
+ ALARM_LEVEL_NORMAL, content, battery, batteryWarnLower);
+ alarmList.add(alarm);
+ hasAlarm = true;
+ }
+ }
+
+ // 如果有告警，保存告警历史并触发通知
+ if (hasAlarm) {
+ log.warn("[告警] {} - {}", deviceName, alarmMessage);
+
+ // 保存告警历史
+ for (AquAlarmHistory alarm : alarmList) {
+ alarmHistoryMapper.insert(alarm);
+ }
+
+ // 触发通知（只对紧急和重要告警发送通知）
+ boolean shouldNotify = alarmList.stream()
+ .anyMatch(alarm -> alarm.getAlarmLevel() >= ALARM_LEVEL_IMPORTANT);
+
+ if (shouldNotify) {
+ triggerAlarmNotification(device, alarmMessage.toString(), sensorData, alarmList);
+ }
+ } else {
+ // 没有告警，检查是否有未恢复的告警需要标记为已恢复
+ checkAndRecoverAlarms(deviceId, device, sensorData);
+ }
+
+ } catch (Exception e) {
+ log.error("检查报警失败: {}", e.getMessage(), e);
}
}
/**
- * 处理温度数据
+ * 触发告警通知
+ * 创建电话通知记录，并发送语音通知
+ *
+ * @param device 设备信息
+ * @param alarmMessage 告警信息
+ * @param sensorData 传感器数据
+ * @param alarmList 告警列表
*/
- private void handleTemperature(Object value) {
- // 示例：温度告警
- if (value instanceof Number) {
- double temp = ((Number) value).doubleValue();
- if (temp > 80) {
- log.warn("温度过高告警: {}°C", temp);
- // TODO: 发送告警通知
+ private void triggerAlarmNotification(Device device, String alarmMessage,
+ DeviceSensorData sensorData, java.util.List alarmList) {
+ try {
+ Long deviceId = device.getId();
+ String deviceName = device.getDeviceName();
+
+ // 检查设备的免打扰设置
+ boolean shouldSkip = false;
+ for (AquAlarmHistory alarm : alarmList) {
+ if (ALARM_TYPE_DISSOLVED_OXYGEN.equals(alarm.getAlarmType()) &&
+ device.getOxyWarnCallNoDis() != null && device.getOxyWarnCallNoDis() == 1) {
+ shouldSkip = true;
+ break;
+ }
+ if (ALARM_TYPE_TEMPERATURE.equals(alarm.getAlarmType()) &&
+ device.getTempWarnCallNoDis() != null && device.getTempWarnCallNoDis() == 1) {
+ shouldSkip = true;
+ break;
+ }
+ if (ALARM_TYPE_BATTERY.equals(alarm.getAlarmType()) &&
+ device.getBatteryWarnCallNoDis() != null && device.getBatteryWarnCallNoDis() == 1) {
+ shouldSkip = true;
+ break;
+ }
+ }
+
+ if (shouldSkip) {
+ return;
+ }
+
+ // 检查是否在告警间隔时间内已经发送过通知（防止频繁通知）
+ LocalDateTime intervalTime = LocalDateTime.now().minusMinutes(ALARM_NOTIFICATION_INTERVAL);
+
+ // 查询最近的通知记录
+ long recentNoticeCount = warnCallNoticeMapper.selectCount(
+ new com.baomidou.mybatisplus.core.conditions.query.LambdaQueryWrapper()
+ .eq(AquWarnCallNotice::getDeviceId, deviceId)
+ .ge(AquWarnCallNotice::getCallTime, intervalTime)
+ );
+
+ if (recentNoticeCount > 0) {
+ log.debug("[告警] {} - {}分钟内已通知", deviceName, ALARM_NOTIFICATION_INTERVAL);
+
+ // 更新告警历史为已通知，但不发送实际通知
+ for (AquAlarmHistory alarm : alarmList) {
+ alarm.setNotified(1);
+ alarm.setUpdatedTime(LocalDateTime.now());
+ alarmHistoryMapper.updateById(alarm);
+ }
+ return;
+ }
+
+ // 从设备信息获取用户信息
+ Long userId = device.getUserId();
+ // TODO: 从用户表查询手机号
+ String mobilePhone = sensorData.getMobilePhone(); // 临时使用传感器数据中的手机号
+ // TODO: 从塘口表查询塘口名称
+ String pondName = deviceName; // 临时使用设备名
+
+ // 验证必要信息
+ if (userId == null || StrUtil.isBlank(mobilePhone)) {
+ log.warn("[告警] {} - 信息不完整", deviceName);
+ return;
+ }
+
+ // 创建告警通知记录
+ AquWarnCallNotice callNotice = new AquWarnCallNotice();
+ callNotice.setUserId(userId);
+ callNotice.setMobilePhone(mobilePhone);
+ callNotice.setDeviceId(deviceId);
+ callNotice.setPondName(pondName);
+ callNotice.setCallTime(LocalDateTime.now());
+ callNotice.setCallStatus(CALL_STATUS_NO_CALL);
+ callNotice.setCreatedTime(LocalDateTime.now());
+ callNotice.setUpdatedTime(LocalDateTime.now());
+
+ // 保存到数据库
+ warnCallNoticeMapper.insert(callNotice);
+
+ // 更新告警历史中的通知ID
+ for (AquAlarmHistory alarm : alarmList) {
+ alarm.setNoticeId(callNotice.getId());
+ alarm.setNotified(1);
+ alarm.setUpdatedTime(LocalDateTime.now());
+ alarmHistoryMapper.updateById(alarm);
+ }
+
+ // 发送语音通知
+ sendVoiceNotification(callNotice, alarmMessage);
+
+ } catch (Exception e) {
+ log.error("[告警通知] 触发告警通知失败: {}", e.getMessage(), e);
+ }
+ }
+
+ /**
+ // 发送语音通知
+ * @param callNotice 通知记录
+ * @param alarmMessage 告警信息
+ */
+ private void sendVoiceNotification(AquWarnCallNotice callNotice, String alarmMessage) {
+ try {
+ // 构建语音通知参数
+ Map ttsParams = new HashMap<>();
+ ttsParams.put("pond_name", callNotice.getPondName() != null ? callNotice.getPondName() : "未知塘口");
+ ttsParams.put("alarm_info", alarmMessage);
+
+ // 生成业务标识（用于回执关联）
+ String outId = "ALARM_" + callNotice.getId() + "_" + System.currentTimeMillis();
+
+ // 调用 VMS 发送语音通知
+ VmsNoticeResponse response = vmsNoticeService.sendTtsCall(
+ callNotice.getMobilePhone(),
+ ttsParams,
+ outId
+ );
+
+ if (response != null && response.isSuccess()) {
+ // 更新通知记录状态
+ callNotice.setCallStatus(1); // 呼叫成功，等待结果反馈
+ callNotice.setCallId(response.getCallId());
+ callNotice.setOutId(outId);
+ callNotice.setUpdatedTime(LocalDateTime.now());
+ warnCallNoticeMapper.updateById(callNotice);
+
+ log.info("[告警] 呼叫成功: {}", callNotice.getMobilePhone());
+ } else {
+ // 呼叫失败
+ callNotice.setCallStatus(2); // 呼叫失败
+ callNotice.setStatusMsg(response != null ? response.getMessage() : "呼叫失败");
+ callNotice.setUpdatedTime(LocalDateTime.now());
+ warnCallNoticeMapper.updateById(callNotice);
+
+ log.error("[告警] 呼叫失败: {} - {}", callNotice.getMobilePhone(), callNotice.getStatusMsg());
+ }
+
+ } catch (Exception e) {
+ log.error("[告警] 语音通知异常: {}", e.getMessage());
+
+ // 更新为呼叫失败
+ try {
+ callNotice.setCallStatus(2);
+ callNotice.setStatusMsg("发送异常: " + e.getMessage());
+ callNotice.setUpdatedTime(LocalDateTime.now());
+ warnCallNoticeMapper.updateById(callNotice);
+ } catch (Exception ex) {
+ log.error("[告警] 更新失败: {}", ex.getMessage());
}
}
}
/**
- * 处理湿度数据
+ * 触发设备故障告警
+ * 针对设备事件（error）的紧急告警
+ *
+ * @param deviceName 设备名称
+ * @param params 事件参数
*/
- private void handleHumidity(Object value) {
- // 示例：湿度处理逻辑
- log.debug("湿度数据: {}", value);
+ private void triggerDeviceErrorAlarm(String deviceName, JSONObject params) {
+ try {
+ String errorMsg = params != null ? params.toString() : "设备故障";
+ log.error("[设备故障] 设备: {}, 故障信息: {}", deviceName, errorMsg);
+
+ // TODO: 实现紧急告警逻辑
+ // 1. 查询设备信息和用户
+ // 2. 创建紧急告警记录
+ // 3. 立即发送语音通知
+ // 4. 可选：发送短信、推送消息等多种通知方式
+
+ } catch (Exception e) {
+ log.error("[设备故障] 触发故障告警失败: {}", e.getMessage(), e);
+ }
}
/**
- * 处理状态数据
+ * 创建告警历史记录
+ *
+ * @param deviceId 设备ID
+ * @param deviceName 设备名称
+ * @param alarmType 告警类型
+ * @param alarmLevel 告警级别
+ * @param alarmContent 告警内容
+ * @param alarmValue 告警值
+ * @param thresholdValue 阈值
+ * @return 告警历史对象
*/
- private void handleStatus(Object value) {
- // 示例：设备状态变化
- log.info("设备状态变化: {}", value);
+ private AquAlarmHistory createAlarmHistory(Long deviceId, String deviceName, String alarmType,
+ int alarmLevel, String alarmContent,
+ Double alarmValue, Double thresholdValue) {
+ AquAlarmHistory alarm = new AquAlarmHistory();
+ alarm.setDeviceId(deviceId);
+ alarm.setDeviceName(deviceName);
+ alarm.setAlarmType(alarmType);
+ alarm.setAlarmLevel(alarmLevel);
+ alarm.setAlarmContent(alarmContent);
+ alarm.setAlarmValue(alarmValue);
+ alarm.setThresholdValue(thresholdValue);
+ alarm.setAlarmStatus(ALARM_STATUS_PENDING);
+ alarm.setAlarmTime(LocalDateTime.now());
+ alarm.setNotified(0);
+ alarm.setCreatedTime(LocalDateTime.now());
+ alarm.setUpdatedTime(LocalDateTime.now());
+ return alarm;
+ }
+
+ /**
+ * 检查并标记已恢复的告警
+ * 当传感器数据恢复正常时，将未恢复的告警标记为已恢复
+ *
+ * @param deviceId 设备ID
+ * @param device 设备信息
+ * @param sensorData 传感器数据
+ */
+ private void checkAndRecoverAlarms(Long deviceId, Device device, DeviceSensorData sensorData) {
+ try {
+ // 查询该设备所有未恢复的告警
+ java.util.List pendingAlarms = alarmHistoryMapper.selectList(
+ new com.baomidou.mybatisplus.core.conditions.query.LambdaQueryWrapper()
+ .eq(AquAlarmHistory::getDeviceId, deviceId)
+ .ne(AquAlarmHistory::getAlarmStatus, ALARM_STATUS_RECOVERED)
+ );
+
+ if (pendingAlarms.isEmpty()) {
+ return;
+ }
+
+ LocalDateTime now = LocalDateTime.now();
+
+ for (AquAlarmHistory alarm : pendingAlarms) {
+ boolean recovered = false;
+
+ // 根据告警类型检查是否已恢复（使用设备配置的阈值）
+ switch (alarm.getAlarmType()) {
+ case ALARM_TYPE_DISSOLVED_OXYGEN:
+ if (sensorData.getDissolvedOxygen() != null) {
+ Double value = sensorData.getDissolvedOxygen();
+ Double oxyWarnLower = device.getOxyWarnLower();
+ recovered = oxyWarnLower != null && value >= oxyWarnLower;
+ }
+ break;
+
+ case ALARM_TYPE_TEMPERATURE:
+ if (sensorData.getTemperature() != null) {
+ Double value = sensorData.getTemperature();
+ Double tempWarnLower = device.getTempWarnLower();
+ Double tempWarnUpper = device.getTempWarnUpper();
+ recovered = (tempWarnLower == null || value >= tempWarnLower) &&
+ (tempWarnUpper == null || value <= tempWarnUpper);
+ }
+ break;
+
+ case ALARM_TYPE_BATTERY:
+ if (sensorData.getBattery() != null) {
+ Double batteryWarnLower = device.getBatteryWarnLower() != null ? device.getBatteryWarnLower().doubleValue() : null;
+ recovered = batteryWarnLower != null && sensorData.getBattery() >= batteryWarnLower;
+ }
+ break;
+ }
+
+ // 如果告警已恢复，更新状态
+ if (recovered) {
+ alarm.setAlarmStatus(ALARM_STATUS_RECOVERED);
+ alarm.setRecoveryTime(now);
+ alarm.setUpdatedTime(now);
+ alarmHistoryMapper.updateById(alarm);
+ log.info("[恢复] {} - {}", alarm.getDeviceName(), alarm.getAlarmType());
+ }
+ }
+
+ } catch (Exception e) {
+ log.error("检查告警恢复失败: {}", e.getMessage(), e);
+ }
+ }
+
+ /**
+ * 转换为 TDengine 传感器数据格式
+ *
+ * @param productKey 产品Key
+ * @param deviceName 设备名称
+ * @param params 设备参数
+ * @return DeviceSensorData
+ */
+ private DeviceSensorData convertToSensorData(String productKey, String deviceName, JSONObject params) {
+ try {
+ DeviceSensorData sensorData = new DeviceSensorData();
+
+ // 时间字段
+ String now = LocalDateTime.now().format(DATETIME_FORMATTER);
+ sensorData.setTime(now);
+ sensorData.setCreateTime(now);
+
+ // 设备标识（使用 deviceName 作为 serialNum）
+ sensorData.setSerialNum(deviceName);
+ sensorData.setDeviceName(deviceName);
+
+ // 解析传感器数据（根据实际字段映射）
+ if (params.containsKey("dissolvedoxygen")) {
+ sensorData.setDissolvedOxygen(params.getDouble("dissolvedoxygen"));
+ }
+ if (params.containsKey("temperature")) {
+ sensorData.setTemperature(params.getDouble("temperature"));
+ }
+ if (params.containsKey("saturability")) {
+ sensorData.setSaturability(params.getDouble("saturability"));
+ }
+ if (params.containsKey("ph")) {
+ sensorData.setPh(params.getDouble("ph"));
+ }
+ if (params.containsKey("salinity")) {
+ sensorData.setSalinity(params.getDouble("salinity"));
+ }
+ if (params.containsKey("treference")) {
+ sensorData.setTreference(params.getDouble("treference"));
+ }
+ if (params.containsKey("tfluorescence")) {
+ sensorData.setTfluorescence(params.getDouble("tfluorescence"));
+ }
+ if (params.containsKey("phasedifference")) {
+ sensorData.setPhaseDifference(params.getDouble("phasedifference"));
+ }
+ if (params.containsKey("battery")) {
+ sensorData.setBattery(params.getDouble("battery"));
+ }
+
+ // TODO: 如果需要设置 deviceId、userId 等标签字段，需要从设备管理系统查询
+ // 可以通过 deviceName 或 productKey 查询设备信息
+ // DeviceInfo deviceInfo = deviceService.getByDeviceName(deviceName);
+ // if (deviceInfo != null) {
+ // sensorData.setDeviceId(deviceInfo.getId());
+ // sensorData.setUserId(deviceInfo.getUserId());
+ // sensorData.setUserName(deviceInfo.getUserName());
+ // sensorData.setMobilePhone(deviceInfo.getMobilePhone());
+ // sensorData.setDeviceType(deviceInfo.getDeviceType());
+ // }
+
+ return sensorData;
+ } catch (Exception e) {
+ log.error("转换传感器数据失败: {}", e.getMessage(), e);
+ return null;
+ }
+ }
+
+ /**
+ * 更新设备表的实时数据
+ *
+ * @param deviceName 设备名称
+ * @param sensorData 传感器数据
+ */
+ private void updateDeviceRealTimeData(String deviceName, DeviceSensorData sensorData) {
+ try {
+ // 查询设备
+ Device device = deviceMapper.selectOne(
+ new com.baomidou.mybatisplus.core.conditions.query.LambdaQueryWrapper()
+ .eq(Device::getSerialNum, deviceName)
+ .last("LIMIT 1")
+ );
+
+ if (device == null) {
+ log.debug("设备不存在，无法更新实时数据: {}", deviceName);
+ return;
+ }
+
+ // 更新实时数据
+ Device updateDevice = new Device();
+ updateDevice.setId(device.getId());
+
+ if (sensorData.getDissolvedOxygen() != null) {
+ updateDevice.setValueDissolvedOxygen(sensorData.getDissolvedOxygen());
+ }
+ if (sensorData.getTemperature() != null) {
+ updateDevice.setValueTemperature(sensorData.getTemperature());
+ }
+ if (sensorData.getSaturability() != null) {
+ updateDevice.setValueSaturability(sensorData.getSaturability());
+ }
+ if (sensorData.getPh() != null) {
+ updateDevice.setValuePh(sensorData.getPh());
+ }
+ if (sensorData.getSalinity() != null) {
+ updateDevice.setValueSalinity(sensorData.getSalinity());
+ }
+
+ // 执行更新
+ int updated = deviceMapper.updateById(updateDevice);
+ if (updated > 0) {
+ log.debug("设备实时数据已更新: {}", deviceName);
+ }
+ } catch (Exception e) {
+ log.error("更新设备实时数据失败[{}]: {}", deviceName, e.getMessage());
+ }
}
}
diff --git a/intc-modules/intc-iot/src/main/java/com/intc/iot/mapper/AquAlarmHistoryMapper.java b/intc-modules/intc-iot/src/main/java/com/intc/iot/mapper/AquAlarmHistoryMapper.java
new file mode 100644
index 0000000..e237e35
--- /dev/null
+++ b/intc-modules/intc-iot/src/main/java/com/intc/iot/mapper/AquAlarmHistoryMapper.java
@@ -0,0 +1,15 @@
+package com.intc.iot.mapper;
+
+import com.baomidou.mybatisplus.core.mapper.BaseMapper;
+import com.intc.iot.domain.AquAlarmHistory;
+import org.apache.ibatis.annotations.Mapper;
+
+/**
+ * 告警历史记录 Mapper
+ *
+ * @author intc
+ */
+@Mapper
+public interface AquAlarmHistoryMapper extends BaseMapper {
+
+}
diff --git a/intc-modules/intc-iot/src/main/java/com/intc/iot/mapper/AquWarnCallNoticeMapper.java b/intc-modules/intc-iot/src/main/java/com/intc/iot/mapper/AquWarnCallNoticeMapper.java
new file mode 100644
index 0000000..42edeeb
--- /dev/null
+++ b/intc-modules/intc-iot/src/main/java/com/intc/iot/mapper/AquWarnCallNoticeMapper.java
@@ -0,0 +1,14 @@
+package com.intc.iot.mapper;
+
+import com.baomidou.mybatisplus.core.mapper.BaseMapper;
+import com.intc.iot.domain.AquWarnCallNotice;
+import org.apache.ibatis.annotations.Mapper;
+
+/**
+ * 告警电话通知记录 Mapper
+ *
+ * @author intc-iot
+ */
+@Mapper
+public interface AquWarnCallNoticeMapper extends BaseMapper {
+}
diff --git a/intc-modules/intc-iot/src/main/java/com/intc/iot/mapper/DeviceRealtimeDataMapper.java b/intc-modules/intc-iot/src/main/java/com/intc/iot/mapper/DeviceRealtimeDataMapper.java
new file mode 100644
index 0000000..5fca9af
--- /dev/null
+++ b/intc-modules/intc-iot/src/main/java/com/intc/iot/mapper/DeviceRealtimeDataMapper.java
@@ -0,0 +1,14 @@
+package com.intc.iot.mapper;
+
+import com.intc.common.mybatis.core.mapper.BaseMapperPlus;
+import com.intc.iot.domain.DeviceRealtimeData;
+import com.intc.iot.domain.vo.DeviceRealtimeDataVo;
+
+/**
+ * 设备实时数据Mapper接口
+ *
+ * @author intc
+ */
+public interface DeviceRealtimeDataMapper extends BaseMapperPlus {
+
+}
diff --git a/intc-modules/intc-iot/src/main/java/com/intc/iot/mapper/IotDeviceMapper.java b/intc-modules/intc-iot/src/main/java/com/intc/iot/mapper/IotDeviceMapper.java
new file mode 100644
index 0000000..18eb86f
--- /dev/null
+++ b/intc-modules/intc-iot/src/main/java/com/intc/iot/mapper/IotDeviceMapper.java
@@ -0,0 +1,15 @@
+package com.intc.iot.mapper;
+
+import com.baomidou.mybatisplus.core.mapper.BaseMapper;
+import com.intc.iot.domain.IotDevice;
+import org.apache.ibatis.annotations.Mapper;
+
+/**
+ * IoT设备 Mapper
+ *
+ * @author intc
+ */
+@Mapper
+public interface IotDeviceMapper extends BaseMapper {
+
+}
diff --git a/intc-modules/intc-iot/src/main/java/com/intc/iot/mapper/IotDeviceStatusMapper.java b/intc-modules/intc-iot/src/main/java/com/intc/iot/mapper/IotDeviceStatusMapper.java
new file mode 100644
index 0000000..fdaf174
--- /dev/null
+++ b/intc-modules/intc-iot/src/main/java/com/intc/iot/mapper/IotDeviceStatusMapper.java
@@ -0,0 +1,14 @@
+package com.intc.iot.mapper;
+
+import com.baomidou.mybatisplus.core.mapper.BaseMapper;
+import com.intc.iot.domain.IotDeviceStatus;
+import org.apache.ibatis.annotations.Mapper;
+
+/**
+ * 设备上下线状态 Mapper
+ *
+ * @author intc-iot
+ */
+@Mapper
+public interface IotDeviceStatusMapper extends BaseMapper {
+}
diff --git a/intc-modules/intc-iot/src/main/java/com/intc/iot/mapper/VmsCallbackMapper.java b/intc-modules/intc-iot/src/main/java/com/intc/iot/mapper/VmsCallbackMapper.java
new file mode 100644
index 0000000..5adca7b
--- /dev/null
+++ b/intc-modules/intc-iot/src/main/java/com/intc/iot/mapper/VmsCallbackMapper.java
@@ -0,0 +1,14 @@
+package com.intc.iot.mapper;
+
+import com.baomidou.mybatisplus.core.mapper.BaseMapper;
+import com.intc.iot.domain.VmsCallback;
+import org.apache.ibatis.annotations.Mapper;
+
+/**
+ * VMS 语音回执 Mapper
+ *
+ * @author intc-iot
+ */
+@Mapper
+public interface VmsCallbackMapper extends BaseMapper {
+}
diff --git a/intc-modules/intc-iot/src/main/java/com/intc/iot/service/AmqpMessageHandler.java b/intc-modules/intc-iot/src/main/java/com/intc/iot/service/AmqpMessageHandler.java
new file mode 100644
index 0000000..0cf40c7
--- /dev/null
+++ b/intc-modules/intc-iot/src/main/java/com/intc/iot/service/AmqpMessageHandler.java
@@ -0,0 +1,19 @@
+package com.intc.iot.service;
+
+/**
+ * AMQP 消息处理器接口
+ *
+ * 用于处理从阿里云 IoT 平台通过 AMQP 推送的设备消息
+ *
+ * @author intc
+ */
+public interface AmqpMessageHandler {
+
+ /**
+ * 处理接收到的 AMQP 消息
+ *
+ * @param message JSON 格式的消息内容
+ */
+ void handleMessage(String message);
+
+}
diff --git a/intc-modules/intc-iot/src/main/java/com/intc/iot/service/DeviceDataService.java b/intc-modules/intc-iot/src/main/java/com/intc/iot/service/DeviceDataService.java
index ec975c5..2cca6b9 100644
--- a/intc-modules/intc-iot/src/main/java/com/intc/iot/service/DeviceDataService.java
+++ b/intc-modules/intc-iot/src/main/java/com/intc/iot/service/DeviceDataService.java
@@ -39,4 +39,13 @@ public interface DeviceDataService {
*/
void unsubscribeDevice(String iotId) throws Exception;
+ /**
+ * 保存设备数据（AMQP 推送的数据）
+ *
+ * @param deviceName 设备名称
+ * @param productKey 产品Key
+ * @param data 设备数据
+ */
+ void saveDeviceData(String deviceName, String productKey, Object data);
+
}
diff --git a/intc-modules/intc-iot/src/main/java/com/intc/iot/service/DeviceRealtimeDataService.java b/intc-modules/intc-iot/src/main/java/com/intc/iot/service/DeviceRealtimeDataService.java
new file mode 100644
index 0000000..42a60b3
--- /dev/null
+++ b/intc-modules/intc-iot/src/main/java/com/intc/iot/service/DeviceRealtimeDataService.java
@@ -0,0 +1,68 @@
+package com.intc.iot.service;
+
+import com.intc.common.mybatis.core.page.PageQuery;
+import com.intc.common.mybatis.core.page.TableDataInfo;
+import com.intc.iot.domain.bo.DeviceRealtimeDataBo;
+import com.intc.iot.domain.vo.DeviceRealtimeDataVo;
+
+import java.util.List;
+
+/**
+ * 设备实时数据服务接口
+ *
+ * @author intc
+ */
+public interface DeviceRealtimeDataService {
+
+ /**
+ * 查询设备实时数据列表（分页）
+ *
+ * @param bo 查询条件
+ * @param pageQuery 分页参数
+ * @return 设备实时数据列表
+ */
+ TableDataInfo queryPageList(DeviceRealtimeDataBo bo, PageQuery pageQuery);
+
+ /**
+ * 查询设备实时数据列表
+ *
+ * @param bo 查询条件
+ * @return 设备实时数据列表
+ */
+ List queryList(DeviceRealtimeDataBo bo);
+
+ /**
+ * 根据ID查询设备实时数据
+ *
+ * @param id 主键ID
+ * @return 设备实时数据
+ */
+ DeviceRealtimeDataVo queryById(Long id);
+
+ /**
+ * 查询设备最新数据
+ *
+ * @param productKey 产品Key
+ * @param deviceName 设备名称
+ * @return 最新数据
+ */
+ DeviceRealtimeDataVo queryLatestData(String productKey, String deviceName);
+
+ /**
+ * 查询设备最新属性数据
+ *
+ * @param productKey 产品Key
+ * @param deviceName 设备名称
+ * @return 最新属性数据
+ */
+ DeviceRealtimeDataVo queryLatestPropertyData(String productKey, String deviceName);
+
+ /**
+ * 删除设备实时数据
+ *
+ * @param ids 主键ID数组
+ * @return 是否成功
+ */
+ Boolean deleteByIds(Long[] ids);
+
+}
diff --git a/intc-modules/intc-iot/src/main/java/com/intc/iot/service/DeviceStatusService.java b/intc-modules/intc-iot/src/main/java/com/intc/iot/service/DeviceStatusService.java
new file mode 100644
index 0000000..dcd01e0
--- /dev/null
+++ b/intc-modules/intc-iot/src/main/java/com/intc/iot/service/DeviceStatusService.java
@@ -0,0 +1,40 @@
+package com.intc.iot.service;
+
+import com.intc.iot.domain.IotDeviceStatus;
+
+import java.util.Map;
+
+/**
+ * 设备状态服务接口
+ * 对应 C# 的设备上下线状态处理
+ *
+ * @author intc-iot
+ */
+public interface DeviceStatusService {
+
+ /**
+ * 处理设备状态变更事件
+ * 对应 C# IOTTopicDeviceStatus 的处理
+ *
+ * @param statusData 状态数据
+ */
+ void handleStatusChange(Map statusData);
+
+ /**
+ * 查询设备当前状态
+ *
+ * @param productKey 产品Key
+ * @param deviceName 设备名称
+ * @return 设备状态
+ */
+ IotDeviceStatus queryDeviceStatus(String productKey, String deviceName);
+
+ /**
+ * 查询设备是否在线
+ *
+ * @param productKey 产品Key
+ * @param deviceName 设备名称
+ * @return true-在线，false-离线
+ */
+ boolean isDeviceOnline(String productKey, String deviceName);
+}
diff --git a/intc-modules/intc-iot/src/main/java/com/intc/iot/service/IotCloudService.java b/intc-modules/intc-iot/src/main/java/com/intc/iot/service/IotCloudService.java
new file mode 100644
index 0000000..2c65781
--- /dev/null
+++ b/intc-modules/intc-iot/src/main/java/com/intc/iot/service/IotCloudService.java
@@ -0,0 +1,56 @@
+package com.intc.iot.service;
+
+import java.util.Map;
+
+/**
+ * IoT 云端操作服务（Java 版等价封装）。
+ *
+ * 封装常用的设备查询、属性查询与属性下发能力，
+ * 底层基于 {@link com.intc.iot.service.IotDeviceService} 和阿里云 IoT Java SDK。
+ */
+public interface IotCloudService {
+
+ // ================= 按 productKey + deviceName 操作 =================
+
+ /**
+ * 根据 productKey + deviceName 查询设备详情。
+ */
+ Map getDeviceInfo(String productKey, String deviceName) throws Exception;
+
+ /**
+ * 根据 productKey + deviceName 查询设备属性状态。
+ */
+ Map getDeviceProperties(String productKey, String deviceName) throws Exception;
+
+ /**
+ * 根据 productKey + deviceName 设置设备属性。
+ */
+ boolean setProperty(String productKey, String deviceName, Map properties, boolean checkSuccess, int retryCount) throws Exception;
+
+ /**
+ * 根据 productKey + deviceName 调用设备服务。
+ */
+ Map invokeService(String productKey, String deviceName, String identifier, String args) throws Exception;
+
+ // ================= 按 iotId 操作 =================
+
+ /**
+ * 根据 iotId 查询设备详情。
+ */
+ Map getDeviceInfo(String iotId) throws Exception;
+
+ /**
+ * 根据 iotId 查询设备属性状态。
+ */
+ Map getDeviceProperties(String iotId) throws Exception;
+
+ /**
+ * 根据 iotId 设置设备属性。
+ */
+ boolean setProperty(String iotId, Map properties, boolean checkSuccess, int retryCount) throws Exception;
+
+ /**
+ * 调用设备服务（如自定义服务能力）。
+ */
+ Map invokeService(String iotId, String identifier, String args) throws Exception;
+}
diff --git a/intc-modules/intc-iot/src/main/java/com/intc/iot/service/IotDeviceService.java b/intc-modules/intc-iot/src/main/java/com/intc/iot/service/IotDeviceService.java
index 3dd8aea..7a93d4a 100644
--- a/intc-modules/intc-iot/src/main/java/com/intc/iot/service/IotDeviceService.java
+++ b/intc-modules/intc-iot/src/main/java/com/intc/iot/service/IotDeviceService.java
@@ -12,12 +12,23 @@ public interface IotDeviceService {
/**
* 查询设备列表
*
+ * @param productKey 产品Key（可选）
* @param pageNo 页码
* @param pageSize 每页大小
* @return 设备列表
* @throws Exception 异常
*/
- Map queryDeviceList(Integer pageNo, Integer pageSize) throws Exception;
+ Map queryDeviceList(String productKey, Integer pageNo, Integer pageSize) throws Exception;
+
+ /**
+ * 根据 ProductKey 和 DeviceName 查询设备信息
+ *
+ * @param productKey 产品Key
+ * @param deviceName 设备名称
+ * @return 设备详情（包含 iotId 等）
+ * @throws Exception 异常
+ */
+ Map findDeviceByProductKeyAndName(String productKey, String deviceName) throws Exception;
/**
* 查询设备详情
@@ -67,4 +78,14 @@ public interface IotDeviceService {
*/
Map unbindDevice(String iotId) throws Exception;
+ /**
+ * 查询设备物模型（模板）
+ * 对应 C# IIOTCloudService.GetTemplate
+ *
+ * @param productKey 产品Key
+ * @return 物模型数据
+ * @throws Exception 异常
+ */
+ Map queryThingModel(String productKey) throws Exception;
+
}
diff --git a/intc-modules/intc-iot/src/main/java/com/intc/iot/service/MqttService.java b/intc-modules/intc-iot/src/main/java/com/intc/iot/service/MqttService.java
index 8b48a9b..e8de9c6 100644
--- a/intc-modules/intc-iot/src/main/java/com/intc/iot/service/MqttService.java
+++ b/intc-modules/intc-iot/src/main/java/com/intc/iot/service/MqttService.java
@@ -34,4 +34,14 @@ public interface MqttService {
*/
void unsubscribe(String topic) throws Exception;
+ /**
+ * 订阅设备状态 Topic
+ * 对应 C# 的 /as/mqtt/status/ Topic 订阅
+ *
+ * @param productKey 产品Key
+ * @param deviceName 设备名称
+ * @throws Exception 异常
+ */
+ void subscribeDeviceStatus(String productKey, String deviceName) throws Exception;
+
}
diff --git a/intc-modules/intc-iot/src/main/java/com/intc/iot/service/VmsMnsConsumerService.java b/intc-modules/intc-iot/src/main/java/com/intc/iot/service/VmsMnsConsumerService.java
new file mode 100644
index 0000000..d97f5a5
--- /dev/null
+++ b/intc-modules/intc-iot/src/main/java/com/intc/iot/service/VmsMnsConsumerService.java
@@ -0,0 +1,28 @@
+package com.intc.iot.service;
+
+/**
+ * VMS MNS 回执消费服务接口
+ *
+ * @author intc-iot
+ */
+public interface VmsMnsConsumerService {
+
+ /**
+ * 启动 MNS 消费者
+ *
+ * @throws Exception 启动异常
+ */
+ void start() throws Exception;
+
+ /**
+ * 停止 MNS 消费者
+ */
+ void stop();
+
+ /**
+ * 判断是否正在运行
+ *
+ * @return true: 运行中, false: 已停止
+ */
+ boolean isRunning();
+}
diff --git a/intc-modules/intc-iot/src/main/java/com/intc/iot/service/VmsNoticeService.java b/intc-modules/intc-iot/src/main/java/com/intc/iot/service/VmsNoticeService.java
new file mode 100644
index 0000000..9dc1232
--- /dev/null
+++ b/intc-modules/intc-iot/src/main/java/com/intc/iot/service/VmsNoticeService.java
@@ -0,0 +1,22 @@
+package com.intc.iot.service;
+
+import com.intc.iot.domain.VmsNoticeResponse;
+
+import java.util.Map;
+
+/**
+ * 语音电话通知服务（VMS）。
+ */
+public interface VmsNoticeService {
+
+ /**
+ * 发送语音 TTS 呼叫（验证码或通知）。
+ *
+ * @param phoneNum 被叫手机号
+ * @param params 模板参数键值对
+ * @param outId 业务唯一标识（会在回执中返回）
+ * @return 调用结果
+ * @throws Exception 调用阿里云接口失败
+ */
+ VmsNoticeResponse sendTtsCall(String phoneNum, Map params, String outId) throws Exception;
+}
diff --git a/intc-modules/intc-iot/src/main/java/com/intc/iot/service/WarnCallNoticeService.java b/intc-modules/intc-iot/src/main/java/com/intc/iot/service/WarnCallNoticeService.java
new file mode 100644
index 0000000..853a4cf
--- /dev/null
+++ b/intc-modules/intc-iot/src/main/java/com/intc/iot/service/WarnCallNoticeService.java
@@ -0,0 +1,51 @@
+package com.intc.iot.service;
+
+import com.intc.iot.domain.VmsCallback;
+
+/**
+ * 告警电话通知服务接口
+ * 对应 C# AlarmProcessor 中的 UpdateCallNoticeFromCallback 逻辑
+ *
+ * @author intc-iot
+ */
+public interface WarnCallNoticeService {
+
+ /**
+ * 处理 VMS 回执，更新告警通知记录
+ * 对应 C# ProcessVmsCallbacks + UpdateCallNoticeFromCallback
+ *
+ * @param callback VMS 回执
+ * @return true: 处理成功, false: 未找到对应记录
+ */
+ boolean processVmsCallback(VmsCallback callback);
+
+ /**
+ * 批量处理未处理的 VMS 回执
+ * 由定时任务调用
+ *
+ * @return 处理的回执数量
+ */
+ int processUnhandledCallbacks();
+
+ /**
+ * 移除指定设备的待通知记录
+ * 当某个设备的语音通知成功后，移除该设备的其他待通知记录
+ * 对应 C# RemovePendingNotificationsForDevice
+ *
+ * @param deviceId 设备ID
+ * @return 移除的记录数量
+ */
+ int removePendingNotificationsByDevice(Long deviceId);
+
+ /**
+ * 清理超时的通知记录
+ * 对应 C# CleanupExpiredNotifications
+ *
+ * 清理规则：
+ * - 已呼叫成功（callStatus=1）但超过超时时间未收到回执
+ * - 超时时间：3 分钟（与 C# 保持一致）
+ *
+ * @return 清理的记录数量
+ */
+ int cleanupExpiredNotifications();
+}
diff --git a/intc-modules/intc-iot/src/main/java/com/intc/iot/service/impl/AmqpMessageHandlerImpl.java b/intc-modules/intc-iot/src/main/java/com/intc/iot/service/impl/AmqpMessageHandlerImpl.java
new file mode 100644
index 0000000..cf9d2ab
--- /dev/null
+++ b/intc-modules/intc-iot/src/main/java/com/intc/iot/service/impl/AmqpMessageHandlerImpl.java
@@ -0,0 +1,252 @@
+package com.intc.iot.service.impl;
+
+import cn.hutool.json.JSONObject;
+import cn.hutool.json.JSONUtil;
+import com.intc.iot.handler.DeviceDataHandler;
+import com.intc.iot.service.AmqpMessageHandler;
+import lombok.RequiredArgsConstructor;
+import lombok.extern.slf4j.Slf4j;
+import org.springframework.beans.factory.annotation.Autowired;
+import org.springframework.stereotype.Service;
+
+/**
+ * AMQP 消息处理器实现类
+ *
+ * 处理从阿里云 IoT 平台推送的设备消息，包括：
+ * - 设备上报数据
+ * - 设备上线/下线
+ * - 设备状态变化
+ * - 设备属性变化
+ *
+ * @author intc
+ */
+@Slf4j
+@Service
+@RequiredArgsConstructor
+public class AmqpMessageHandlerImpl implements AmqpMessageHandler {
+
+ @Autowired(required = false)
+ private DeviceDataHandler deviceDataHandler;
+
+ @Override
+ public void handleMessage(String message) {
+ try {
+ JSONObject json = JSONUtil.parseObj(message);
+
+ // 检查是否包含 items 字段（设备属性数据）
+ if (json.containsKey("items")) {
+ // 直接处理设备属性数据
+ handlePropertyMessage(json);
+ return;
+ }
+
+ // 获取消息类型
+ String messageType = json.getStr("messageType");
+
+ if (messageType == null) {
+ // 如果没有 messageType，可能是飞燕平台的原始消息格式
+ // 尝试从 topic 或 method 判断消息类型
+ String topic = json.getStr("topic");
+ String method = json.getStr("method");
+
+ if (topic != null) {
+ // 根据 topic 判断消息类型
+ if (topic.contains("/property/post")) {
+ handlePropertyMessageFromTopic(topic, message);
+ } else if (topic.contains("/event/")) {
+ handleEventMessageFromTopic(topic, message);
+ } else {
+ log.warn("未知的 Topic 格式: {}", topic);
+ }
+ return;
+ }
+ }
+
+ switch (messageType) {
+ case "property":
+ // 设备属性上报
+ handlePropertyMessage(json);
+ break;
+ case "event":
+ // 设备事件上报
+ handleEventMessage(json);
+ break;
+ case "status":
+ // 设备状态变化
+ handleStatusMessage(json);
+ break;
+ case "deviceLifeCycle":
+ // 设备生命周期（上线/下线）
+ handleLifeCycleMessage(json);
+ break;
+ default:
+ log.warn("未知的消息类型: {}", messageType);
+ }
+
+ } catch (Exception e) {
+ log.error("处理 AMQP 消息失败: {}", message, e);
+ throw new RuntimeException("消息处理失败", e);
+ }
+ }
+
+ /**
+ * 处理设备属性上报消息
+ */
+ private void handlePropertyMessage(JSONObject json) {
+ String deviceName = json.getStr("deviceName");
+ String productKey = json.getStr("productKey");
+ JSONObject items = json.getJSONObject("items");
+
+ // 如果没有 DeviceDataHandler，只记录日志
+ if (deviceDataHandler == null) {
+ log.warn("设备数据处理器未配置，跳过数据处理");
+ return;
+ }
+
+ // 使用 DeviceDataHandler 处理数据（包括存储和报警）
+ try {
+ // 转换 items 格式：{"dissolvedOxygen": {"value": 7.82, "time": xxx}} -> {"dissolvedOxygen": 7.82}
+ JSONObject params = new JSONObject();
+ if (items != null) {
+ for (String key : items.keySet()) {
+ Object itemValue = items.get(key);
+ if (itemValue instanceof JSONObject) {
+ JSONObject itemObj = (JSONObject) itemValue;
+ // 提取 value 字段
+ if (itemObj.containsKey("value")) {
+ // 字段名映射：currentTemperature -> temperature, dosat -> saturability
+ String mappedKey = mapFieldName(key);
+ params.set(mappedKey, itemObj.get("value"));
+ }
+ } else {
+ // 如果不是嵌套格式，直接使用
+ String mappedKey = mapFieldName(key);
+ params.set(mappedKey, itemValue);
+ }
+ }
+ }
+
+ // 构造 topic 格式：/sys/{ProductKey}/{DeviceName}/thing/event/property/post
+ String topic = String.format("/sys/%s/%s/thing/event/property/post", productKey, deviceName);
+ // 构造飞燕平台消息格式
+ JSONObject message = new JSONObject();
+ message.set("method", "thing.event.property.post");
+ message.set("params", params);
+
+ deviceDataHandler.handlePropertyPost(topic, message.toString());
+ } catch (Exception e) {
+ log.error("使用 DeviceDataHandler 处理属性数据失败", e);
+ }
+ }
+
+ /**
+ * 字段名映射：将阿里云字段名映射为应用字段名
+ */
+ private String mapFieldName(String fieldName) {
+ switch (fieldName) {
+ case "currentTemperature":
+ return "temperature";
+ case "dosat":
+ return "saturability";
+ case "Treference":
+ return "treference";
+ case "Tfluorescence":
+ return "tfluorescence";
+ case "Tcorrect":
+ return "tcorrect";
+ case "Tsignal":
+ return "tsignal";
+ case "salinitySet":
+ return "salinity";
+ default:
+ // 其他字段名保持不变（转为小写）
+ return fieldName.toLowerCase();
+ }
+ }
+
+ /**
+ * 处理从 Topic 判断的属性消息（飞燕平台原始格式）
+ */
+ private void handlePropertyMessageFromTopic(String topic, String payload) {
+ if (deviceDataHandler == null) {
+ log.warn("设备数据处理器未配置，跳过数据处理");
+ return;
+ }
+
+ try {
+ deviceDataHandler.handlePropertyPost(topic, payload);
+ } catch (Exception e) {
+ log.error("处理属性消息失败", e);
+ }
+ }
+
+ /**
+ * 处理设备事件消息
+ */
+ private void handleEventMessage(JSONObject json) {
+ String deviceName = json.getStr("deviceName");
+ String productKey = json.getStr("productKey");
+ String identifier = json.getStr("identifier");
+ JSONObject value = json.getJSONObject("value");
+
+ if (deviceDataHandler == null) {
+ log.warn("设备数据处理器未配置，跳过事件处理");
+ return;
+ }
+
+ try {
+ // 构造 topic 格式：/sys/{ProductKey}/{DeviceName}/thing/event/{EventIdentifier}/post
+ String topic = String.format("/sys/%s/%s/thing/event/%s/post", productKey, deviceName, identifier);
+ // 构造飞燕平台消息格式
+ JSONObject message = new JSONObject();
+ message.set("method", "thing.event." + identifier + ".post");
+ message.set("params", value);
+
+ deviceDataHandler.handleEventPost(topic, message.toString());
+ } catch (Exception e) {
+ log.error("处理设备事件失败", e);
+ }
+ }
+
+ /**
+ * 处理从 Topic 判断的事件消息（飞燕平台原始格式）
+ */
+ private void handleEventMessageFromTopic(String topic, String payload) {
+ if (deviceDataHandler == null) {
+ log.warn("设备数据处理器未配置，跳过事件处理");
+ return;
+ }
+
+ try {
+ deviceDataHandler.handleEventPost(topic, payload);
+ } catch (Exception e) {
+ log.error("处理事件消息失败", e);
+ }
+ }
+
+ /**
+ * 处理设备状态变化消息
+ */
+ private void handleStatusMessage(JSONObject json) {
+ String deviceName = json.getStr("deviceName");
+ String status = json.getStr("status");
+
+ log.debug("设备状态: {} - {}", deviceName, status);
+
+ // TODO: 更新设备状态
+ }
+
+ /**
+ * 处理设备生命周期消息（上线/下线）
+ */
+ private void handleLifeCycleMessage(JSONObject json) {
+ String deviceName = json.getStr("deviceName");
+ String action = json.getStr("action");
+
+ log.debug("设备{}: {}", "online".equals(action) ? "上线" : "下线", deviceName);
+
+ // action: "online" 或 "offline"
+ // TODO: 更新设备在线状态
+ }
+
+}
diff --git a/intc-modules/intc-iot/src/main/java/com/intc/iot/service/impl/DeviceDataServiceImpl.java b/intc-modules/intc-iot/src/main/java/com/intc/iot/service/impl/DeviceDataServiceImpl.java
index 3deaaeb..9d77d06 100644
--- a/intc-modules/intc-iot/src/main/java/com/intc/iot/service/impl/DeviceDataServiceImpl.java
+++ b/intc-modules/intc-iot/src/main/java/com/intc/iot/service/impl/DeviceDataServiceImpl.java
@@ -7,6 +7,7 @@ import lombok.RequiredArgsConstructor;
import lombok.extern.slf4j.Slf4j;
import org.eclipse.paho.client.mqttv3.MqttClient;
import org.springframework.boot.autoconfigure.condition.ConditionalOnBean;
+import org.springframework.boot.autoconfigure.condition.ConditionalOnClass;
import org.springframework.stereotype.Service;
/**
@@ -17,6 +18,7 @@ import org.springframework.stereotype.Service;
@Slf4j
@Service
@RequiredArgsConstructor
+@ConditionalOnClass(MqttClient.class)
@ConditionalOnBean(MqttClient.class)
public class DeviceDataServiceImpl implements DeviceDataService {
@@ -44,24 +46,14 @@ public class DeviceDataServiceImpl implements DeviceDataService {
public void subscribeDeviceProperties(String iotId) throws Exception {
// 注意：需要根据 iotId 获取 productKey 和 deviceName
// 这里简化处理，实际应该从数据库或缓存中查询
- log.info("订阅设备属性上报，IotId: {}", iotId);
-
- // 示例：假设从设备信息中获取
- // String productKey = getProductKeyByIotId(iotId);
- // String deviceName = getDeviceNameByIotId(iotId);
- // String topic = String.format(PROPERTY_POST_TOPIC, productKey, deviceName);
-
- // mqttService.subscribe(topic, 1);
-
- log.warn("请先实现 iotId 到 productKey/deviceName 的映射逻辑");
+ log.warn("订阅设备属性上报功能未实现，请先实现 iotId 到 productKey/deviceName 的映射逻辑");
+ throw new UnsupportedOperationException("订阅设备属性功能未实现，请使用 subscribeAllDevices 方法");
}
@Override
public void subscribeDeviceEvents(String iotId) throws Exception {
- log.info("订阅设备事件上报，IotId: {}", iotId);
-
- // 类似属性订阅，需要映射关系
- log.warn("请先实现 iotId 到 productKey/deviceName 的映射逻辑");
+ log.warn("订阅设备事件上报功能未实现，请先实现 iotId 到 productKey/deviceName 的映射逻辑");
+ throw new UnsupportedOperationException("订阅设备事件功能未实现，请使用 subscribeAllDevices 方法");
}
@Override
@@ -76,10 +68,22 @@ public class DeviceDataServiceImpl implements DeviceDataService {
@Override
public void unsubscribeDevice(String iotId) throws Exception {
- log.info("取消订阅设备数据，IotId: {}", iotId);
+ log.warn("取消订阅设备数据功能未实现，请先实现 iotId 到 productKey/deviceName 的映射逻辑");
+ throw new UnsupportedOperationException("取消订阅设备功能未实现");
+ }
+
+ @Override
+ public void saveDeviceData(String deviceName, String productKey, Object data) {
+ log.info("保存设备数据 - 设备: {}, 产品: {}, 数据: {}", deviceName, productKey, data);
- // 需要取消对应的 Topic 订阅
- log.warn("请先实现 iotId 到 productKey/deviceName 的映射逻辑");
+ // TODO: 实现数据保存逻辑
+ // 1. 保存到 MySQL/PostgreSQL（结构化数据）
+ // 2. 保存到 TDengine（时序数据）
+ // 3. 触发告警逻辑
+ // 4. 发送通知
+
+ // 示例：保存到实时数据表
+ // deviceRealtimeDataService.save(deviceName, productKey, data);
}
}
diff --git a/intc-modules/intc-iot/src/main/java/com/intc/iot/service/impl/DeviceRealtimeDataServiceImpl.java b/intc-modules/intc-iot/src/main/java/com/intc/iot/service/impl/DeviceRealtimeDataServiceImpl.java
new file mode 100644
index 0000000..b4a9562
--- /dev/null
+++ b/intc-modules/intc-iot/src/main/java/com/intc/iot/service/impl/DeviceRealtimeDataServiceImpl.java
@@ -0,0 +1,95 @@
+package com.intc.iot.service.impl;
+
+import cn.hutool.core.util.ObjectUtil;
+import com.baomidou.mybatisplus.core.conditions.query.LambdaQueryWrapper;
+import com.baomidou.mybatisplus.core.toolkit.Wrappers;
+import com.baomidou.mybatisplus.extension.plugins.pagination.Page;
+import com.intc.common.core.utils.MapstructUtils;
+import com.intc.common.mybatis.core.page.PageQuery;
+import com.intc.common.mybatis.core.page.TableDataInfo;
+import com.intc.iot.domain.DeviceRealtimeData;
+import com.intc.iot.domain.bo.DeviceRealtimeDataBo;
+import com.intc.iot.domain.vo.DeviceRealtimeDataVo;
+import com.intc.iot.mapper.DeviceRealtimeDataMapper;
+import com.intc.iot.service.DeviceRealtimeDataService;
+import lombok.RequiredArgsConstructor;
+import lombok.extern.slf4j.Slf4j;
+import org.springframework.stereotype.Service;
+
+import java.util.Arrays;
+import java.util.List;
+
+/**
+ * 设备实时数据服务实现
+ *
+ * @author intc
+ */
+@Slf4j
+@Service
+@RequiredArgsConstructor
+public class DeviceRealtimeDataServiceImpl implements DeviceRealtimeDataService {
+
+ private final DeviceRealtimeDataMapper deviceRealtimeDataMapper;
+
+ /**
+ * 构建查询条件
+ */
+ private LambdaQueryWrapper buildQueryWrapper(DeviceRealtimeDataBo bo) {
+ LambdaQueryWrapper wrapper = Wrappers.lambdaQuery();
+ wrapper.eq(ObjectUtil.isNotEmpty(bo.getProductKey()), DeviceRealtimeData::getProductKey, bo.getProductKey())
+ .eq(ObjectUtil.isNotEmpty(bo.getDeviceName()), DeviceRealtimeData::getDeviceName, bo.getDeviceName())
+ .eq(ObjectUtil.isNotEmpty(bo.getIotId()), DeviceRealtimeData::getIotId, bo.getIotId())
+ .eq(ObjectUtil.isNotEmpty(bo.getDataType()), DeviceRealtimeData::getDataType, bo.getDataType())
+ .eq(ObjectUtil.isNotEmpty(bo.getIdentifier()), DeviceRealtimeData::getIdentifier, bo.getIdentifier())
+ .ge(ObjectUtil.isNotEmpty(bo.getStartTime()), DeviceRealtimeData::getReportTime, bo.getStartTime())
+ .le(ObjectUtil.isNotEmpty(bo.getEndTime()), DeviceRealtimeData::getReportTime, bo.getEndTime())
+ .orderByDesc(DeviceRealtimeData::getReportTime);
+ return wrapper;
+ }
+
+ @Override
+ public TableDataInfo queryPageList(DeviceRealtimeDataBo bo, PageQuery pageQuery) {
+ LambdaQueryWrapper wrapper = buildQueryWrapper(bo);
+ Page page = deviceRealtimeDataMapper.selectVoPage(pageQuery.build(), wrapper);
+ return TableDataInfo.build(page);
+ }
+
+ @Override
+ public List queryList(DeviceRealtimeDataBo bo) {
+ LambdaQueryWrapper wrapper = buildQueryWrapper(bo);
+ return deviceRealtimeDataMapper.selectVoList(wrapper);
+ }
+
+ @Override
+ public DeviceRealtimeDataVo queryById(Long id) {
+ return deviceRealtimeDataMapper.selectVoById(id);
+ }
+
+ @Override
+ public DeviceRealtimeDataVo queryLatestData(String productKey, String deviceName) {
+ LambdaQueryWrapper wrapper = Wrappers.lambdaQuery();
+ wrapper.eq(DeviceRealtimeData::getProductKey, productKey)
+ .eq(DeviceRealtimeData::getDeviceName, deviceName)
+ .orderByDesc(DeviceRealtimeData::getReportTime)
+ .last("LIMIT 1");
+ return deviceRealtimeDataMapper.selectVoOne(wrapper);
+ }
+
+ @Override
+ public DeviceRealtimeDataVo queryLatestPropertyData(String productKey, String deviceName) {
+ LambdaQueryWrapper wrapper = Wrappers.lambdaQuery();
+ wrapper.eq(DeviceRealtimeData::getProductKey, productKey)
+ .eq(DeviceRealtimeData::getDeviceName, deviceName)
+ .eq(DeviceRealtimeData::getDataType, "property")
+ .orderByDesc(DeviceRealtimeData::getReportTime)
+ .last("LIMIT 1");
+ return deviceRealtimeDataMapper.selectVoOne(wrapper);
+ }
+
+ @Override
+ public Boolean deleteByIds(Long[] ids) {
+ int rows = deviceRealtimeDataMapper.deleteBatchIds(Arrays.asList(ids));
+ return rows > 0;
+ }
+
+}
diff --git a/intc-modules/intc-iot/src/main/java/com/intc/iot/service/impl/DeviceStatusServiceImpl.java b/intc-modules/intc-iot/src/main/java/com/intc/iot/service/impl/DeviceStatusServiceImpl.java
new file mode 100644
index 0000000..6a1d9ae
--- /dev/null
+++ b/intc-modules/intc-iot/src/main/java/com/intc/iot/service/impl/DeviceStatusServiceImpl.java
@@ -0,0 +1,124 @@
+package com.intc.iot.service.impl;
+
+import cn.hutool.json.JSONUtil;
+import com.baomidou.mybatisplus.core.conditions.query.LambdaQueryWrapper;
+import com.intc.iot.domain.IotDeviceStatus;
+import com.intc.iot.mapper.IotDeviceStatusMapper;
+import com.intc.iot.service.DeviceStatusService;
+import lombok.RequiredArgsConstructor;
+import lombok.extern.slf4j.Slf4j;
+import org.springframework.boot.autoconfigure.condition.ConditionalOnBean;
+import org.springframework.stereotype.Service;
+import org.springframework.transaction.annotation.Transactional;
+
+import java.time.LocalDateTime;
+import java.util.Map;
+
+/**
+ * 设备状态服务实现
+ * 对应 C# IOTTopicDeviceStatus 的处理逻辑
+ *
+ * @author intc-iot
+ */
+@Service
+@RequiredArgsConstructor
+@ConditionalOnBean(IotDeviceStatusMapper.class)
+@Slf4j
+public class DeviceStatusServiceImpl implements DeviceStatusService {
+
+ private final IotDeviceStatusMapper deviceStatusMapper;
+
+ @Override
+ @Transactional(rollbackFor = Exception.class)
+ public void handleStatusChange(Map statusData) {
+ try {
+ String productKey = (String) statusData.get("productKey");
+ String deviceName = (String) statusData.get("deviceName");
+ String status = (String) statusData.get("status");
+ Long statusTime = getLongValue(statusData, "statusTime");
+ Long lastTime = getLongValue(statusData, "lastTime");
+ String clientIp = (String) statusData.get("clientIp");
+
+ if (productKey == null || deviceName == null || status == null) {
+ log.warn("[设备状态] 状态数据不完整，跳过处理: {}", JSONUtil.toJsonStr(statusData));
+ return;
+ }
+
+ log.info("[设备状态] 设备状态变更 - ProductKey: {}, DeviceName: {}, Status: {}",
+ productKey, deviceName, status);
+
+ // 查询是否已存在记录
+ IotDeviceStatus existingStatus = deviceStatusMapper.selectOne(
+ new LambdaQueryWrapper()
+ .eq(IotDeviceStatus::getProductKey, productKey)
+ .eq(IotDeviceStatus::getDeviceName, deviceName)
+ );
+
+ LocalDateTime now = LocalDateTime.now();
+
+ if (existingStatus != null) {
+ // 更新现有记录
+ existingStatus.setStatus(status);
+ existingStatus.setStatusTime(statusTime);
+ existingStatus.setLastTime(lastTime);
+ existingStatus.setClientIp(clientIp);
+ existingStatus.setUpdatedTime(now);
+ deviceStatusMapper.updateById(existingStatus);
+
+ log.debug("[设备状态] 更新设备状态记录 - ID: {}, Status: {}", existingStatus.getId(), status);
+ } else {
+ // 新增记录
+ IotDeviceStatus newStatus = new IotDeviceStatus();
+ newStatus.setProductKey(productKey);
+ newStatus.setDeviceName(deviceName);
+ newStatus.setIotId((String) statusData.get("iotId"));
+ newStatus.setStatus(status);
+ newStatus.setStatusTime(statusTime);
+ newStatus.setLastTime(lastTime);
+ newStatus.setClientIp(clientIp);
+ newStatus.setCreatedTime(now);
+ newStatus.setUpdatedTime(now);
+ deviceStatusMapper.insert(newStatus);
+
+ log.info("[设备状态] 新增设备状态记录 - ProductKey: {}, DeviceName: {}, Status: {}",
+ productKey, deviceName, status);
+ }
+ } catch (Exception e) {
+ log.error("[设备状态] 处理设备状态变更异常: {}", e.getMessage(), e);
+ throw e;
+ }
+ }
+
+ @Override
+ public IotDeviceStatus queryDeviceStatus(String productKey, String deviceName) {
+ return deviceStatusMapper.selectOne(
+ new LambdaQueryWrapper()
+ .eq(IotDeviceStatus::getProductKey, productKey)
+ .eq(IotDeviceStatus::getDeviceName, deviceName)
+ );
+ }
+
+ @Override
+ public boolean isDeviceOnline(String productKey, String deviceName) {
+ IotDeviceStatus status = queryDeviceStatus(productKey, deviceName);
+ return status != null && "online".equalsIgnoreCase(status.getStatus());
+ }
+
+ /**
+ * 从 Map 中获取 Long 值
+ */
+ private Long getLongValue(Map map, String key) {
+ Object value = map.get(key);
+ if (value == null) {
+ return null;
+ }
+ if (value instanceof Number) {
+ return ((Number) value).longValue();
+ }
+ try {
+ return Long.parseLong(value.toString());
+ } catch (NumberFormatException e) {
+ return null;
+ }
+ }
+}
diff --git a/intc-modules/intc-iot/src/main/java/com/intc/iot/service/impl/IotCloudServiceImpl.java b/intc-modules/intc-iot/src/main/java/com/intc/iot/service/impl/IotCloudServiceImpl.java
new file mode 100644
index 0000000..0951ebb
--- /dev/null
+++ b/intc-modules/intc-iot/src/main/java/com/intc/iot/service/impl/IotCloudServiceImpl.java
@@ -0,0 +1,181 @@
+package com.intc.iot.service.impl;
+
+import cn.hutool.json.JSONUtil;
+import com.aliyuncs.IAcsClient;
+import com.intc.iot.service.IotCloudService;
+import com.intc.iot.service.IotDeviceService;
+import lombok.RequiredArgsConstructor;
+import lombok.extern.slf4j.Slf4j;
+import org.springframework.boot.autoconfigure.condition.ConditionalOnBean;
+import org.springframework.stereotype.Service;
+
+import java.util.List;
+import java.util.Map;
+
+/**
+ * IoT 云端操作服务实现。
+ *
+ * 基于 {@link IotDeviceService} 封装常用的设备查询、属性查询与属性下发能力，
+ * 方便业务侧直接按 iotId 进行操作。
+ */
+@Slf4j
+@Service
+@RequiredArgsConstructor
+@ConditionalOnBean({IAcsClient.class, IotDeviceService.class})
+public class IotCloudServiceImpl implements IotCloudService {
+
+ private final IotDeviceService iotDeviceService;
+
+ /**
+ * 根据 productKey + deviceName 解析 iotId。
+ */
+ @SuppressWarnings("unchecked")
+ private String resolveIotId(String productKey, String deviceName) throws Exception {
+ Map resp = iotDeviceService.findDeviceByProductKeyAndName(productKey, deviceName);
+ Object successObj = resp.get("success");
+ boolean success = successObj instanceof Boolean && (Boolean) successObj;
+ if (!success) {
+ log.warn("[IotCloudService] 根据 productKey={} deviceName={} 查询设备失败: {}", productKey, deviceName, resp);
+ return null;
+ }
+
+ Object dataObj = resp.get("data");
+ if (dataObj == null) {
+ log.warn("[IotCloudService] 查询设备返回数据为空");
+ return null;
+ }
+
+ // 处理阿里云 SDK 返回的数据结构
+ try {
+ // 尝试获取 deviceList
+ List deviceList = null;
+ if (dataObj instanceof Map) {
+ Map dataMap = (Map) dataObj;
+ Object listObj = dataMap.get("deviceList");
+ if (listObj instanceof List) {
+ deviceList = (List) listObj;
+ }
+ } else {
+ // 如果 dataObj 本身有 getDeviceList 方法（通过反射调用）
+ java.lang.reflect.Method method = dataObj.getClass().getMethod("getDeviceList");
+ Object result = method.invoke(dataObj);
+ if (result instanceof List) {
+ deviceList = (List) result;
+ }
+ }
+
+ if (deviceList == null || deviceList.isEmpty()) {
+ log.warn("[IotCloudService] 未找到设备, productKey={}, deviceName={}", productKey, deviceName);
+ return null;
+ }
+
+ // 获取第一个设备的 iotId
+ Object deviceObj = deviceList.get(0);
+ if (deviceObj instanceof Map) {
+ Map deviceMap = (Map) deviceObj;
+ Object iotIdObj = deviceMap.get("iotId");
+ return iotIdObj != null ? iotIdObj.toString() : null;
+ } else {
+ // 通过反射获取 iotId
+ java.lang.reflect.Method method = deviceObj.getClass().getMethod("getIotId");
+ Object iotIdObj = method.invoke(deviceObj);
+ return iotIdObj != null ? iotIdObj.toString() : null;
+ }
+ } catch (Exception e) {
+ log.error("[IotCloudService] 解析设备信息失败, productKey={}, deviceName={}", productKey, deviceName, e);
+ return null;
+ }
+ }
+
+ @Override
+ public Map getDeviceInfo(String productKey, String deviceName) throws Exception {
+ String iotId = resolveIotId(productKey, deviceName);
+ if (iotId == null) {
+ return null;
+ }
+ return getDeviceInfo(iotId);
+ }
+
+ @Override
+ public Map getDeviceProperties(String productKey, String deviceName) throws Exception {
+ String iotId = resolveIotId(productKey, deviceName);
+ if (iotId == null) {
+ return null;
+ }
+ return getDeviceProperties(iotId);
+ }
+
+ @Override
+ public boolean setProperty(String productKey, String deviceName, Map properties, boolean checkSuccess, int retryCount) throws Exception {
+ String iotId = resolveIotId(productKey, deviceName);
+ if (iotId == null) {
+ return false;
+ }
+ return setProperty(iotId, properties, checkSuccess, retryCount);
+ }
+
+ @Override
+ public Map invokeService(String productKey, String deviceName, String identifier, String args) throws Exception {
+ String iotId = resolveIotId(productKey, deviceName);
+ if (iotId == null) {
+ return null;
+ }
+ return invokeService(iotId, identifier, args);
+ }
+
+ @Override
+ public Map getDeviceInfo(String iotId) throws Exception {
+ log.info("[IotCloudService] 查询设备详情, iotId={}", iotId);
+ return iotDeviceService.queryDeviceInfo(iotId);
+ }
+
+ @Override
+ public Map getDeviceProperties(String iotId) throws Exception {
+ log.info("[IotCloudService] 查询设备属性, iotId={}", iotId);
+ return iotDeviceService.queryDeviceProperties(iotId);
+ }
+
+ @Override
+ public boolean setProperty(String iotId, Map properties, boolean checkSuccess, int retryCount) throws Exception {
+ log.info("[IotCloudService] 设置设备属性, iotId={}, properties={} ", iotId, properties);
+
+ // 将属性 Map 转为阿里云 SDK 需要的 JSON 字符串
+ String itemsJson = JSONUtil.toJsonStr(properties);
+
+ int attempt = 0;
+ while (true) {
+ Map resp = iotDeviceService.setDeviceProperty(iotId, itemsJson);
+ Object successObj = resp.get("success");
+ boolean success = successObj instanceof Boolean && (Boolean) successObj;
+
+ if (!checkSuccess) {
+ // 不强制校验 success，直接按照返回值判断即可
+ return success;
+ }
+
+ if (success) {
+ return true;
+ }
+
+ attempt++;
+ if (attempt > retryCount) {
+ log.warn("[IotCloudService] 设置设备属性失败且重试次数耗尽, iotId={}, properties={}, resp={}", iotId, properties, resp);
+ return false;
+ }
+
+ log.warn("[IotCloudService] 设置设备属性失败, 准备重试 {}/{} 次, iotId={}, properties={}, resp={}", attempt, retryCount, iotId, properties, resp);
+ try {
+ Thread.sleep(500L);
+ } catch (InterruptedException e) {
+ Thread.currentThread().interrupt();
+ return false;
+ }
+ }
+ }
+
+ @Override
+ public Map invokeService(String iotId, String identifier, String args) throws Exception {
+ log.info("[IotCloudService] 调用设备服务, iotId={}, identifier={} ", iotId, identifier);
+ return iotDeviceService.invokeService(iotId, identifier, args);
+ }
+}
diff --git a/intc-modules/intc-iot/src/main/java/com/intc/iot/service/impl/IotDeviceServiceImpl.java b/intc-modules/intc-iot/src/main/java/com/intc/iot/service/impl/IotDeviceServiceImpl.java
index 7e194d7..ae76d86 100644
--- a/intc-modules/intc-iot/src/main/java/com/intc/iot/service/impl/IotDeviceServiceImpl.java
+++ b/intc-modules/intc-iot/src/main/java/com/intc/iot/service/impl/IotDeviceServiceImpl.java
@@ -7,6 +7,7 @@ import com.intc.iot.service.IotDeviceService;
import lombok.RequiredArgsConstructor;
import lombok.extern.slf4j.Slf4j;
import org.springframework.boot.autoconfigure.condition.ConditionalOnBean;
+import org.springframework.boot.autoconfigure.condition.ConditionalOnClass;
import org.springframework.stereotype.Service;
import java.util.HashMap;
@@ -20,6 +21,7 @@ import java.util.Map;
@Slf4j
@Service
@RequiredArgsConstructor
+@ConditionalOnClass(IAcsClient.class)
@ConditionalOnBean(IAcsClient.class)
public class IotDeviceServiceImpl implements IotDeviceService {
@@ -27,15 +29,27 @@ public class IotDeviceServiceImpl implements IotDeviceService {
private final AliyunIotProperties iotProperties;
@Override
- public Map queryDeviceList(Integer pageNo, Integer pageSize) throws Exception {
- log.info("查询设备列表，页码: {}, 每页大小: {}", pageNo, pageSize);
+ public Map queryDeviceList(String productKey, Integer pageNo, Integer pageSize) throws Exception {
+ log.info("查询设备列表，ProductKey: {}, 页码: {}, 每页大小: {}", productKey, pageNo, pageSize);
QueryDeviceRequest request = new QueryDeviceRequest();
+ // ProductKey 是必填参数，如果为空则报错
+ if (productKey == null || productKey.isEmpty()) {
+ log.error("查询设备列表失败：ProductKey 不能为空");
+ throw new IllegalArgumentException("ProductKey is mandatory for this action");
+ }
+ request.setProductKey(productKey);
request.setCurrentPage(pageNo);
request.setPageSize(pageSize);
+ log.info("调用阿里云 API，请求参数: ProductKey={}, CurrentPage={}, PageSize={}",
+ request.getProductKey(), request.getCurrentPage(), request.getPageSize());
+
QueryDeviceResponse response = acsClient.getAcsResponse(request);
+ log.info("阿里云 API 返回: Success={}, ErrorMessage={}",
+ response.getSuccess(), response.getErrorMessage());
+
Map result = new HashMap<>();
result.put("success", response.getSuccess());
result.put("data", response.getData());
@@ -43,6 +57,38 @@ public class IotDeviceServiceImpl implements IotDeviceService {
return result;
}
+ @Override
+ public Map findDeviceByProductKeyAndName(String productKey, String deviceName) throws Exception {
+ log.info("根据 ProductKey 和 DeviceName 查询设备信息，ProductKey: {}, DeviceName: {}", productKey, deviceName);
+
+ // 使用 QueryDeviceDetail API 查询特定设备
+ QueryDeviceDetailRequest request = new QueryDeviceDetailRequest();
+ request.setProductKey(productKey);
+ request.setDeviceName(deviceName);
+
+ QueryDeviceDetailResponse response = acsClient.getAcsResponse(request);
+
+ Map result = new HashMap<>();
+ result.put("success", response.getSuccess());
+
+ // 将设备详情数据包装成类似 QueryDevice 的格式，以保持接口兼容性
+ if (response.getSuccess() && response.getData() != null) {
+ // 创建一个设备列表，包含查询到的单个设备
+ Map dataWrapper = new HashMap<>();
+ java.util.List