fastbee/docs/scripts/production_to_test_db_bridge.groovy

// 生产 -> 测试库 数据库桥接脚本（数据流脚本，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;
    }
}