feat(device): 新增设备同步与盐城/南京实时数据转发功能

iot-module/iot-manager/src/main/java/cc/iotkit/manager/config/DeviceSyncConfig.java

@@ -0,0 +1,92 @@
+package cc.iotkit.manager.config;
+
+import lombok.Data;
+import org.springframework.boot.context.properties.ConfigurationProperties;
+import org.springframework.stereotype.Component;
+
+import java.util.ArrayList;
+import java.util.List;
+
+/**
+ * 设备同步配置（盐城设备台账同步、盐城/南京实时数据转发）
+ * <p>
+ * 配置前缀：device-sync
+ *
+ * @author codex
+ */
+@Component
+@ConfigurationProperties(prefix = "device-sync")
+@Data
+public class DeviceSyncConfig {
+
+    /**
+     * 总开关，联调前保持 false
+     */
+    private boolean enable = false;
+
+    /**
+     * 失败重试次数
+     */
+    private int retry = 3;
+
+    /**
+     * 盐城 token 加密密钥（DES/CBC/PKCS5Padding，密钥需 8 字节，如 88888888）
+     */
+    private String encryptKey = "";
+
+    /**
+     * 对接厂商名称（盐城 addorEdit 的 dockingCompany）
+     */
+    private String dockingCompany = "";
+
+    /**
+     * 盐城设备新增/修改接口
+     */
+    private String addOrEditUrl = "";
+
+    /**
+     * 盐城设备删除接口
+     */
+    private String deleteUrl = "";
+
+    /**
+     * 盐城实时数据接口
+     */
+    private String yanchengRealtimeUrl = "";
+
+    /**
+     * 南京远传数据接口（测试环境：http://36.7.170.39:9880/crgas/cr/alarm/remoteTrans/receive）
+     */
+    private String nanjingReportUrl = "";
+
+    /**
+     * 参与同步的产品白名单（产品 key）；为空表示不限制
+     */
+    private List<String> syncProductKeys = new ArrayList<>();
+
+    /**
+     * 盐城实时数据转发开关
+     */
+    private boolean yanchengRealtimeEnable = false;
+
+    /**
+     * 南京数据转发开关
+     */
+    private boolean nanjingEnable = false;
+
+    /**
+     * 心跳推送周期（秒），默认 10 秒
+     */
+    private int heartbeatIntervalSeconds = 10;
+
+    /**
+     * 事件类消息去重窗口（秒），默认 5 秒
+     */
+    private int dedupSeconds = 5;
+
+    /**
+     * 盐城设备同步 deviceCode 是否使用设备名（IMEI），默认 true；
+     * false 时使用 deviceId
+     */
+    private boolean useDeviceNameAsCode = true;
+}

iot-module/iot-manager/src/main/java/cc/iotkit/manager/event/DeviceSyncEvent.java

@@ -0,0 +1,29 @@
+package cc.iotkit.manager.event;
+
+import cc.iotkit.model.device.DeviceInfo;
+import lombok.Getter;
+import org.springframework.context.ApplicationEvent;
+
+/**
+ * 设备信息同步事件
+ * <p>
+ * 设备创建/修改/删除后发布，由外部对接监听器（如盐城设备台账同步）消费。
+ *
+ * @author codex
+ */
+@Getter
+public class DeviceSyncEvent extends ApplicationEvent {
+
+    /**
+     * 操作类型：create / update / delete
+     */
+    private final String operation;
+
+    private final DeviceInfo device;
+
+    public DeviceSyncEvent(Object source, String operation, DeviceInfo device) {
+        super(source);
+        this.operation = operation;
+        this.device = device;
+    }
+}

iot-module/iot-openapi/src/main/java/cc/iotkit/openapi/sync/RealtimeDataForwarder.java

@@ -0,0 +1,323 @@
+package cc.iotkit.openapi.sync;
+
+import cc.iotkit.common.constant.Constants;
+import cc.iotkit.common.thing.ThingModelMessage;
+import cc.iotkit.data.manager.IDeviceInfoData;
+import cc.iotkit.manager.config.DeviceSyncConfig;
+import cc.iotkit.model.device.DeviceInfo;
+import cc.iotkit.mq.ConsumerHandler;
+import cc.iotkit.mq.MqConsumer;
+import cn.hutool.core.util.ObjectUtil;
+import com.fasterxml.jackson.databind.ObjectMapper;
+import lombok.AllArgsConstructor;
+import lombok.Data;
+import lombok.extern.slf4j.Slf4j;
+import okhttp3.MediaType;
+import okhttp3.OkHttpClient;
+import okhttp3.Request;
+import okhttp3.RequestBody;
+import okhttp3.Response;
+import org.apache.commons.lang3.StringUtils;
+import org.springframework.beans.factory.annotation.Autowired;
+import org.springframework.beans.factory.annotation.Qualifier;
+import org.springframework.stereotype.Component;
+
+import javax.annotation.PostConstruct;
+import javax.annotation.PreDestroy;
+import java.util.ArrayList;
+import java.util.List;
+import java.util.Map;
+import java.util.concurrent.ConcurrentHashMap;
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.Executors;
+import java.util.concurrent.LinkedBlockingQueue;
+import java.util.concurrent.ScheduledExecutorService;
+import java.util.concurrent.ThreadPoolExecutor;
+import java.util.concurrent.TimeUnit;
+
+/**
+ * 实时数据转发器
+ * <p>
+ * 订阅物模型消息总线（THING_MODEL_MESSAGE_TOPIC），将设备上报/告警/故障/恢复/上下线
+ * 按协议翻译后推送至盐城 dockingData 与南京 deviceInfo/receive。
+ * 心跳按固定周期（默认 10 秒）定时推送，事件类消息即时推送并做简单去重。
+ * <p>
+ * 默认开关关闭，联调确认细节后再开启。
+ *
+ * @author codex
+ */
+@Slf4j
+@Component
+public class RealtimeDataForwarder implements ConsumerHandler<ThingModelMessage>, Runnable {
+
+    @Autowired
+    @Qualifier("deviceInfoDataCache")
+    private IDeviceInfoData deviceInfoData;
+
+    @Autowired
+    private MqConsumer<ThingModelMessage> thingModelMessageConsumer;
+
+    @Autowired
+    private DeviceSyncConfig config;
+
+    private final ObjectMapper objectMapper = new ObjectMapper();
+
+    private final OkHttpClient httpClient = new OkHttpClient.Builder()
+            .connectTimeout(5, TimeUnit.SECONDS)
+            .readTimeout(10, TimeUnit.SECONDS)
+            .build();
+
+    /**
+     * 设备最近一次上报数据缓存：deviceId -> 数据
+     */
+    private final Map<String, DeviceLatestData> latestDataCache = new ConcurrentHashMap<>();
+
+    /**
+     * 事件去重：deviceName:dedupKey -> 上次推送时间
+     */
+    private final Map<String, Long> dedupCache = new ConcurrentHashMap<>();
+
+    /**
+     * 推送线程池：与消息总线消费解耦，避免阻塞规则引擎
+     */
+    private final ExecutorService pushPool = new ThreadPoolExecutor(2, 8, 60L, TimeUnit.SECONDS,
+            new LinkedBlockingQueue<>(200), new ThreadPoolExecutor.CallerRunsPolicy());
+
+    private ScheduledExecutorService heartbeatScheduler;
+
+    @PostConstruct
+    public void init() {
+        thingModelMessageConsumer.consume(Constants.THING_MODEL_MESSAGE_TOPIC, this);
+        heartbeatScheduler = Executors.newSingleThreadScheduledExecutor(r -> {
+            Thread t = new Thread(r, "realtime-data-heartbeat");
+            t.setDaemon(true);
+            return t;
+        });
+        int interval = Math.max(1, config.getHeartbeatIntervalSeconds());
+        heartbeatScheduler.scheduleAtFixedRate(this, interval, interval, TimeUnit.SECONDS);
+        log.info("实时数据转发器已启动：heartbeatInterval={}s, yanchengRealtime={}, nanjing={}",
+                interval, config.isYanchengRealtimeEnable(), config.isNanjingEnable());
+    }
+
+    @PreDestroy
+    public void destroy() {
+        if (heartbeatScheduler != null) {
+            heartbeatScheduler.shutdown();
+        }
+        pushPool.shutdown();
+    }
+
+    @Override
+    public void handler(ThingModelMessage msg) {
+        try {
+            if (!config.isYanchengRealtimeEnable() && !config.isNanjingEnable()) {
+                return;
+            }
+            if (msg == null || StringUtils.isBlank(msg.getDeviceName())) {
+                return;
+            }
+            DeviceInfo device = deviceInfoData.findByDeviceName(msg.getDeviceName());
+            if (device == null || !inSyncProducts(device.getProductKey())) {
+                return;
+            }
+
+            // 设备上下线
+            if (ThingModelMessage.TYPE_STATE.equals(msg.getType())) {
+                if (ThingModelMessage.ID_OFFLINE.equals(msg.getIdentifier())) {
+                    latestDataCache.remove(device.getDeviceId());
+                    pushOffline(device);
+                }
+                return;
+            }
+
+            // 仅处理属性上报
+            if (!ThingModelMessage.TYPE_PROPERTY.equals(msg.getType())
+                    || !ThingModelMessage.ID_PROPERTY_REPORT.equals(msg.getIdentifier())) {
+                return;
+            }
+            if (!(msg.getData() instanceof Map)) {
+                return;
+            }
+
+            @SuppressWarnings("unchecked")
+            Map<String, Object> data = (Map<String, Object>) msg.getData();
+            long time = msg.getTime() != null ? msg.getTime() : System.currentTimeMillis();
+            latestDataCache.put(device.getDeviceId(), new DeviceLatestData(time, data, device));
+
+            RealtimeProtocolMapper.ClassifiedEvent event = RealtimeProtocolMapper.classify(data);
+            if (event == null) {
+                return;
+            }
+            // 事件类消息去重，心跳由定时任务负责
+            if (!tryAcquireDedup(device.getDeviceName(), event.getDedupKey())) {
+                return;
+            }
+            pushEvent(event, device, data, time);
+        } catch (Throwable e) {
+            log.error("实时数据转发处理异常：deviceName={}",
+                    msg != null ? msg.getDeviceName() : null, e);
+        }
+    }
+
+    /**
+     * 定时心跳：按固定周期扫描最近有上报的设备，推送盐城 102 / 南京 01
+     */
+    @Override
+    public void run() {
+        try {
+            if (!config.isYanchengRealtimeEnable() && !config.isNanjingEnable()) {
+                return;
+            }
+            long now = System.currentTimeMillis();
+            long staleAfter = Math.max(1, config.getHeartbeatIntervalSeconds()) * 3L * 1000L;
+            List<String> staleKeys = new ArrayList<>();
+            for (Map.Entry<String, DeviceLatestData> entry : latestDataCache.entrySet()) {
+                DeviceLatestData latest = entry.getValue();
+                try {
+                    if (now - latest.getTime() > staleAfter) {
+                        staleKeys.add(entry.getKey());
+                        continue;
+                    }
+                    RealtimeProtocolMapper.ClassifiedEvent event = buildHeartbeatEvent(latest.getData());
+                    pushEvent(event, latest.getDevice(), latest.getData(), latest.getTime());
+                } catch (Exception e) {
+                    log.error("定时心跳推送异常：deviceName={}", latest.getDevice().getDeviceName(), e);
+                }
+            }
+            staleKeys.forEach(latestDataCache::remove);
+        } catch (Throwable e) {
+            log.error("定时心跳推送异常", e);
+        }
+    }
+
+    private RealtimeProtocolMapper.ClassifiedEvent buildHeartbeatEvent(Map<String, Object> data) {
+        RealtimeProtocolMapper.ClassifiedEvent event = RealtimeProtocolMapper.classify(data);
+        if (event == null) {
+            event = new RealtimeProtocolMapper.ClassifiedEvent();
+            event.setNodeId(RealtimeProtocolMapper.resolveNodeId(data));
+        }
+        event.setKind(RealtimeProtocolMapper.EVENT_HEARTBEAT);
+        event.setGasSensorState(0);
+        event.setErrorCode(0);
+        event.setContent("正常");
+        event.setDedupKey("heartbeat");
+        return event;
+    }
+
+    private void pushOffline(DeviceInfo device) {
+        RealtimeProtocolMapper.ClassifiedEvent event = new RealtimeProtocolMapper.ClassifiedEvent();
+        event.setKind(RealtimeProtocolMapper.EVENT_OFFLINE);
+        event.setGasSensorState(0);
+        event.setErrorCode(10);
+        event.setContent("离线");
+        event.setDedupKey("offline");
+        if (!tryAcquireDedup(device.getDeviceName(), event.getDedupKey())) {
+            return;
+        }
+        pushEvent(event, device, null, System.currentTimeMillis());
+    }
+
+    private void pushEvent(RealtimeProtocolMapper.ClassifiedEvent event, DeviceInfo device,
+                           Map<String, Object> data, long time) {
+        if (config.isYanchengRealtimeEnable()) {
+            Map<String, Object> ycBody = RealtimeProtocolMapper.buildYanchengBody(
+                    event, device, data, time, config.getDockingCompany());
+            sendYancheng(ycBody, device.getDeviceName());
+        }
+        if (config.isNanjingEnable()) {
+            Map<String, Object> njBody = RealtimeProtocolMapper.buildNanjingBody(
+                    event, device, data, time, config.getDockingCompany());
+            sendNanjing(njBody, device.getDeviceName());
+        }
+    }
+
+    private void sendYancheng(Map<String, Object> body, String deviceName) {
+        if (StringUtils.isBlank(config.getYanchengRealtimeUrl())) {
+            log.warn("盐城实时数据 URL 未配置，跳过：deviceName={}", deviceName);
+            return;
+        }
+        pushPool.submit(() -> {
+            int maxRetry = Math.max(1, config.getRetry());
+            for (int i = 0; i < maxRetry; i++) {
+                try {
+                    Request request = new Request.Builder()
+                            .url(config.getYanchengRealtimeUrl())
+                            .post(RequestBody.create(objectMapper.writeValueAsString(body),
+                                    MediaType.parse("application/json; charset=utf-8")))
+                            .build();
+                    try (Response response = httpClient.newCall(request).execute()) {
+                        if (!response.isSuccessful()) {
+                            throw new RuntimeException("HTTP " + response.code());
+                        }
+                    }
+                    log.info("盐城实时数据推送成功：deviceName={}, serviceType={}, body={}",
+                            deviceName, body.get("serviceType"), objectMapper.writeValueAsString(body));
+                    return;
+                } catch (Exception e) {
+                    log.warn("盐城实时数据推送失败，第{}次：deviceName={}, error={}",
+                            i + 1, deviceName, e.getMessage());
+                }
+            }
+        });
+    }
+
+    private void sendNanjing(Map<String, Object> body, String deviceName) {
+        if (StringUtils.isBlank(config.getNanjingReportUrl())) {
+            log.warn("南京数据上报 URL 未配置，跳过：deviceName={}", deviceName);
+            return;
+        }
+        pushPool.submit(() -> {
+            int maxRetry = Math.max(1, config.getRetry());
+            for (int i = 0; i < maxRetry; i++) {
+                try {
+                    Request request = new Request.Builder()
+                            .url(config.getNanjingReportUrl())
+                            .post(RequestBody.create(objectMapper.writeValueAsString(body),
+                                    MediaType.parse("application/json; charset=utf-8")))
+                            .build();
+                    try (Response response = httpClient.newCall(request).execute()) {
+                        if (!response.isSuccessful()) {
+                            throw new RuntimeException("HTTP " + response.code());
+                        }
+                    }
+                    log.info("南京数据上报推送成功：deviceName={}, messageType={}, body={}",
+                            deviceName, body.get("messageType"), objectMapper.writeValueAsString(body));
+                    return;
+                } catch (Exception e) {
+                    log.warn("南京数据上报推送失败，第{}次：deviceName={}, error={}",
+                            i + 1, deviceName, e.getMessage());
+                }
+            }
+        });
+    }
+
+    private boolean inSyncProducts(String productKey) {
+        if (ObjectUtil.isEmpty(config.getSyncProductKeys())) {
+            return true;
+        }
+        return config.getSyncProductKeys().contains(productKey);
+    }
+
+    private boolean tryAcquireDedup(String deviceName, String dedupKey) {
+        if (StringUtils.isBlank(dedupKey)) {
+            return true;
+        }
+        long now = System.currentTimeMillis();
+        long dedupMs = Math.max(1, config.getDedupSeconds()) * 1000L;
+        String key = deviceName + ":" + dedupKey;
+        Long last = dedupCache.get(key);
+        if (last != null && now - last < dedupMs) {
+            return false;
+        }
+        dedupCache.put(key, now);
+        return true;
+    }
+
+    @Data
+    @AllArgsConstructor
+    private static class DeviceLatestData {
+        private long time;
+        private Map<String, Object> data;
+        private DeviceInfo device;
+    }
+}

iot-module/iot-openapi/src/main/java/cc/iotkit/openapi/sync/RealtimeProtocolMapper.java

@@ -0,0 +1,263 @@
+package cc.iotkit.openapi.sync;
+
+import cc.iotkit.model.device.DeviceInfo;
+import lombok.Data;
+import org.apache.commons.lang3.StringUtils;
+
+import java.text.SimpleDateFormat;
+import java.util.LinkedHashMap;
+import java.util.Map;
+import java.util.Date;
+
+/**
+ * 实时数据协议映射器
+ * <p>
+ * 将平台内部物模型消息（ThingModelMessage.data）分类并翻译成
+ * 盐城 dockingData / 南京 deviceInfo/receive 两家平台的报文。
+ *
+ * @author codex
+ */
+public class RealtimeProtocolMapper {
+
+    /**
+     * 事件分类
+     */
+    public static final String EVENT_HEARTBEAT = "heartbeat";
+    public static final String EVENT_ALARM = "alarm";
+    public static final String EVENT_FAULT = "fault";
+    public static final String EVENT_RECOVER = "recover";
+    public static final String EVENT_OFFLINE = "offline";
+
+    /**
+     * 盐城 serviceType
+     */
+    public static final int YANCHENG_TYPE_HEARTBEAT = 102;
+    public static final int YANCHENG_TYPE_ALARM = 1001;
+    public static final int YANCHENG_TYPE_FAULT = 1002;
+    public static final int YANCHENG_TYPE_RECOVER = 1003;
+
+    /**
+     * 南京 messageType
+     */
+    public static final String NANJING_TYPE_MONITOR = "01";
+    public static final String NANJING_TYPE_FAULT = "02";
+    public static final String NANJING_TYPE_ALARM = "03";
+
+    private RealtimeProtocolMapper() {
+    }
+
+    /**
+     * 根据解析器输出 data 分类事件。
+     * 正常上报(128/129/130) → 心跳；报警 → 告警；故障/断网 → 故障；恢复 → 恢复。
+     *
+     * @return 分类结果，无法识别返回 null
+     */
+    public static ClassifiedEvent classify(Map<String, Object> data) {
+        if (data == null || data.isEmpty()) {
+            return null;
+        }
+
+        Object eventTypeObj = data.get("eventType");
+        Object eventTypeValueObj = data.get("eventTypeValue");
+        String value = eventTypeValueObj != null ? eventTypeValueObj.toString() : "";
+        int eventType = eventTypeObj instanceof Number ? ((Number) eventTypeObj).intValue() : -1;
+
+        ClassifiedEvent event = new ClassifiedEvent();
+        event.setNodeId(resolveNodeId(data));
+        event.setContent(value);
+        event.setDedupKey(value);
+
+        // 正常数据上报（心跳/监测）
+        if (eventType == 128 || eventType == 129 || eventType == 130
+                || value.contains("正常数据上报")) {
+            event.setKind(EVENT_HEARTBEAT);
+            event.setGasSensorState(0);
+            event.setErrorCode(0);
+            return event;
+        }
+
+        // 浓度变化上报按监测数据处理
+        if (value.contains("浓度")) {
+            event.setKind(EVENT_HEARTBEAT);
+            event.setGasSensorState(0);
+            event.setErrorCode(0);
+            return event;
+        }
+
+        // 告警恢复 / 故障恢复
+        if (value.contains("恢复")) {
+            event.setKind(EVENT_RECOVER);
+            event.setGasSensorState(0);
+            event.setErrorCode(0);
+            return event;
+        }
+
+        // 故障（传感器故障 / 通讯断网 / 设备故障）
+        if (value.contains("故障") || value.contains("断网")) {
+            event.setKind(EVENT_FAULT);
+            event.setGasSensorState(0);
+            event.setErrorCode(5);
+            return event;
+        }
+
+        // 告警（首警/报警/动火离人报警等）
+        if (value.contains("报警") || value.contains("告警") || value.contains("首警")) {
+            event.setKind(EVENT_ALARM);
+            // 高限 → 2，低限/其他 → 1
+            event.setGasSensorState(value.contains("高") ? 2 : 1);
+            event.setErrorCode(0);
+            return event;
+        }
+
+        return null;
+    }
+
+    /**
+     * 组装盐城 dockingData 请求体。
+     */
+    public static Map<String, Object> buildYanchengBody(ClassifiedEvent event, DeviceInfo device,
+                                                        Map<String, Object> data, long timestamp,
+                                                        String manufacturer) {
+        Map<String, Object> payload = new LinkedHashMap<>();
+        payload.put("gas_sensor_state", event.getGasSensorState());
+        payload.put("error_code", event.getErrorCode());
+        String potency = resolvePotency(data);
+        if (potency != null) {
+            payload.put("gas_sample_value", potency);
+        }
+        if (StringUtils.isNotBlank(device.getSignalStrength())) {
+            payload.put("signal_power", device.getSignalStrength());
+        }
+        if (StringUtils.isNotBlank(manufacturer)) {
+            payload.put("manufacturer_name", manufacturer);
+        }
+        payload.put("IMEI", device.getDeviceName());
+
+        Map<String, Object> body = new LinkedHashMap<>();
+        body.put("IMEI", device.getDeviceName());
+        body.put("serviceType", resolveYanchengServiceType(event.getKind()));
+        body.put("timestamp", timestamp);
+        body.put("payload", payload);
+        return body;
+    }
+
+    /**
+     * 组装南京 deviceInfo/receive 请求体。
+     */
+    public static Map<String, Object> buildNanjingBody(ClassifiedEvent event, DeviceInfo device,
+                                                       Map<String, Object> data, long timestamp,
+                                                       String manufacturer) {
+        Map<String, Object> body = new LinkedHashMap<>();
+        body.put("manufacturer", manufacturer);
+        body.put("sendTime", formatTime(timestamp));
+
+        // hostNo/deviceNo：单点设备取 IMEI；多点设备 hostNo 取 IMEI、deviceNo 取节点编号
+        String imei = device.getDeviceName();
+        body.put("hostNo", imei);
+        body.put("deviceNo", StringUtils.isNotBlank(event.getNodeId()) ? event.getNodeId() : imei);
+
+        body.put("messageType", resolveNanjingMessageType(event.getKind()));
+
+        String potency = resolvePotency(data);
+        if (potency != null) {
+            body.put("potency", potency);
+        }
+        if (StringUtils.isNotBlank(event.getContent())) {
+            body.put("content", event.getContent());
+        }
+        if (isMonitorLike(event.getKind())) {
+            body.put("onlineStatus", "00");
+        }
+        if (EVENT_OFFLINE.equals(event.getKind())) {
+            body.put("onlineStatus", "01");
+        }
+        body.put("longitude", device.getLongitude());
+        body.put("latitude", device.getLatitude());
+        body.put("remark", event.getContent());
+        return body;
+    }
+
+    private static boolean isMonitorLike(String kind) {
+        return EVENT_HEARTBEAT.equals(kind) || EVENT_RECOVER.equals(kind);
+    }
+
+    private static int resolveYanchengServiceType(String kind) {
+        switch (kind) {
+            case EVENT_ALARM:
+                return YANCHENG_TYPE_ALARM;
+            case EVENT_FAULT:
+            case EVENT_OFFLINE:
+                return YANCHENG_TYPE_FAULT;
+            case EVENT_RECOVER:
+                return YANCHENG_TYPE_RECOVER;
+            default:
+                return YANCHENG_TYPE_HEARTBEAT;
+        }
+    }
+
+    private static String resolveNanjingMessageType(String kind) {
+        switch (kind) {
+            case EVENT_ALARM:
+                return NANJING_TYPE_ALARM;
+            case EVENT_FAULT:
+            case EVENT_OFFLINE:
+                return NANJING_TYPE_FAULT;
+            default:
+                return NANJING_TYPE_MONITOR;
+        }
+    }
+
+    /**
+     * 浓度值：优先取带节点后缀的 dataDetectorValue0，其次取 dataDetectorValue。
+     * 仅返回可解析为数值的浓度，避免把联动箱反馈状态等字符串误当浓度上报。
+     */
+    private static String resolvePotency(Map<String, Object> data) {
+        if (data == null) {
+            return null;
+        }
+        Object v = data.get("dataDetectorValue0");
+        if (v == null) {
+            v = data.get("dataDetectorValue");
+        }
+        if (v == null) {
+            return null;
+        }
+        try {
+            Double.parseDouble(v.toString());
+            return v.toString();
+        } catch (NumberFormatException e) {
+            return null;
+        }
+    }
+
+    /**
+     * 探头编号：优先 nodeId0，其次 nodeId。
+     */
+    public static String resolveNodeId(Map<String, Object> data) {
+        if (data == null) {
+            return null;
+        }
+        Object v = data.get("nodeId0");
+        if (v == null) {
+            v = data.get("nodeId");
+        }
+        return v != null ? v.toString() : null;
+    }
+
+    private static String formatTime(long timestamp) {
+        return new SimpleDateFormat("yyyy-MM-dd HH:mm:ss").format(new Date(timestamp));
+    }
+
+    /**
+     * 事件分类结果
+     */
+    @Data
+    public static class ClassifiedEvent {
+        private String kind;
+        private int gasSensorState;
+        private int errorCode;
+        private String content;
+        private String nodeId;
+        private String dedupKey;
+    }
+}

iot-starter/src/main/resources/application-dev.yml

@@ -276,3 +276,35 @@ plugin:
   mainPackage: cc.iotkit
   pluginPath:
     - ./data/plugins
+
+# 设备同步配置（盐城设备台账同步、盐城/南京实时数据转发）
+# 联调前保持 enable=false，避免误推数据
+device-sync:
+  enable: false
+  retry: 3
+  # DES/CBC/PKCS5Padding 加密密钥（需 8 字节，与甲方 DesUtil.java 一致）
+  encrypt-key: ""
+  # 对接厂商名称
+  docking-company: 天津费加罗电子
+  # 盐城-设备新增/修改
+  add-or-edit-url: http://gas.safeyancheng.cn:8888/ycxf/ycDevice/addorEdit
+  # 盐城-设备删除
+  delete-url: http://gas.safeyancheng.cn:8888/ycxf/ycDevice/deleteByCode
+  # 盐城-实时数据
+  yancheng-realtime-url: http://jiekou.safeyancheng.cn:8889/gas/dockingData
+  # 南京-远传数据（测试环境）
+  nanjing-report-url: http://36.7.170.39:9880/crgas/cr/alarm/remoteTrans/receive
+  # 参与同步的产品白名单（FIGARO 燃气报警器相关产品 key）
+  sync-product-keys:
+    - CEMpmANABN7Tt6Jh   # 家用报警器
+    - XmXYxjzihseT76As   # 独立式报警器
+    - bAASX8tBjYQjBGFP   # 工商业控制器
+    - WfpZZFkMxxbGfRca   # 动火离人
+    - MQFejp7cyDMH3enG   # 联动控制箱
+  yancheng-realtime-enable: false
+  nanjing-enable: false
+  heartbeat-interval-seconds: 10
+  # 事件类消息去重窗口（秒）
+  dedup-seconds: 5
+  # 盐城 deviceCode 使用设备名（IMEI）
+  use-device-name-as-code: true