// 生产 -> 测试库 数据库桥接脚本(数据流脚本,scriptAction = 5 数据库存储) // 配套桥接配置(数据库存储类型)的 SQL 模板请填写:${bridgeSql} // // 注意:processRuleScript 传入的 msgContext.getSerialNumber() 是网关SN, // 而 mapping 的键是子设备SN,所以必须用上报体里的 subSerialNumber 做映射。 // // 上报格式(子设备包装): // [{ // "message": [ // {"calculated":false,"id":"WaterH","name":"水位","ts":"2026-08-22 12:24:03.738","value":"27.249924"}, // ... // ], // "sources":"...", // "subSerialNumber":"D14OIE7K7FAL", // "subDeviceName":"485静力水准仪_2", // "subSlaveId":2 // }] import cn.hutool.json.JSONArray; import cn.hutool.json.JSONObject; import cn.hutool.json.JSONUtil; import cn.hutool.core.util.NumberUtil; // 数据库桥接ID(对应“数据库存储”类型桥接配置的 id,加 L 保证是 Long) msgContext.setData("databaseBridgeID", 10L); // 设备编号映射 生产子设备SN -> 测试子设备SN(映射本身即白名单) def mapping = [ "D1G79IFDNZDX": "D1KB7I625466", "D1J178V592N2": "D1K4NLZ65M0V", "D1Q14L400AGG": "D1151C27FW6C", "D170H3R44Q9S": "D147S65648BV" ]; // 网关SN(msgContext 里带的,仅作为缺少 subSerialNumber 时的兜底) String gwSn = msgContext.getSerialNumber(); String sql = "SELECT 1;"; try { String payload = msgContext.getPayload(); // 兼容三种形态:子设备包装数组、直接物模型数组、单个物模型对象 JSONArray wrappers = null; try { wrappers = JSONUtil.parseArray(payload); } catch (Exception e1) { wrappers = null; } if (wrappers == null && payload != null) { try { JSONObject single = JSONUtil.parseObj(payload); wrappers = new JSONArray(); wrappers.add(single); } catch (Exception e2) { wrappers = null; } } if (wrappers == null || wrappers.isEmpty()) { msgContext.setData("bridgeSql", "SELECT 1;"); return; } // 一个物模型一行,整条消息拼成一条多行 INSERT(单条语句,兼容 Statement.execute) StringBuilder values = new StringBuilder(); int rowCount = 0; for (int w = 0; w < wrappers.size(); w++) { JSONObject wrapper = wrappers.getJSONObject(w); // 取该包装里的物模型数组 JSONArray msgs = wrapper.getJSONArray("message"); if (msgs == null && wrapper.getStr("id") != null) { // 直接物模型数组:wrapper 本身就是一个物模型 msgs = new JSONArray(); msgs.add(wrapper); } if (msgs == null || msgs.isEmpty()) { continue; } // 目标SN:优先子设备SN,缺省回退网关SN String srcSN = wrapper.getStr("subSerialNumber"); if (srcSN == null || srcSN.trim().isEmpty()) { srcSN = gwSn; } String targetSN = lookupTarget(srcSN); if (targetSN == null) { // 该子设备不在映射中,跳过,不中断整条消息 continue; } for (int i = 0; i < msgs.size(); i++) { JSONObject item = msgs.getJSONObject(i); String identity = item.getStr("id"); String modelName = item.getStr("name"); if (modelName == null || modelName.trim().isEmpty()) { modelName = identity; } String logValue = item.getStr("value"); String logTime = normalizeTime(item.get("ts")); if (rowCount > 0) { values.append(","); } values.append("(") .append("'").append(escapeSql(identity)).append("', ") .append("'").append(escapeSql(modelName)).append("', ") .append("1, ") // log_type = 1 属性上报 .append("'").append(escapeSql(logValue)).append("', ") .append("'").append(escapeSql(targetSN)).append("', ") .append("0, 2, ") // is_monitor = 0, mode = 2 在线模式 .append(logTime.isEmpty() ? "NOW()" : "'" + logTime + "'") .append(")"); rowCount++; } } if (rowCount == 0) { // 本次上报没有命中映射的设备,给无害语句 msgContext.setData("bridgeSql", "SELECT 1;"); return; } sql = "INSERT INTO fastbee.iot_device_log " + "(identity, model_name, log_type, log_value, serial_number, is_monitor, mode, create_time) " + "VALUES " + values.toString() + ";"; msgContext.setData("bridgeSql", sql); } catch (Exception e) { // 解析或拼装失败时降级为无害语句,不让桥接节点反复报 SQL 错误 msgContext.setData("bridgeSql", "SELECT 1;"); try { msgContext.logger.error("数据库桥接脚本执行失败: gwSn={}, err={}", gwSn, e.getMessage(), e); } catch (Exception ignore) { } } try { msgContext.logger.info("数据库桥接 SQL: {}", sql); } catch (Exception ignore) { } // ---- 辅助方法 ---- // 映射查询:先按原值,再按大写匹配(FastBee 设备编号通常大写存储) String lookupTarget(String srcSN) { if (srcSN == null) { return null; } String key = srcSN.trim(); String target = mapping[key]; if (target == null && !key.equals(key.toUpperCase())) { target = mapping[key.toUpperCase()]; } return target; } // 转义 SQL 字符串:单引号翻倍,反斜杠转义(MySQL 默认模式) String escapeSql(String v) { if (v == null) { return ""; } return v.replace("\\", "\\\\").replace("'", "''"); } // ts 归一化: // - "2026-08-22 12:24:03.738" 截断到秒(create_time 为 datetime,不带毫秒) // - 毫秒/秒时间戳转 "yyyy-MM-dd HH:mm:ss" // - 缺省返回空串(调用处用 NOW()) String normalizeTime(Object ts) { if (ts == null || ts.toString().trim().isEmpty()) { return ""; } String s = ts.toString().trim(); if (s.length() > 19 && s.charAt(4) == '-') { return s.substring(0, 19); } try { long t = Long.parseLong(s); java.util.Date d = t > 100000000000L ? new java.util.Date(t) : new java.util.Date(t * 1000L); return new java.text.SimpleDateFormat("yyyy-MM-dd HH:mm:ss").format(d); } catch (Exception e) { return s; } }