diff --git a/docs/knowledge-base/06-platform-integration-design.md b/docs/knowledge-base/06-platform-integration-design.md index 45e7ae22..9ae93277 100644 --- a/docs/knowledge-base/06-platform-integration-design.md +++ b/docs/knowledge-base/06-platform-integration-design.md @@ -94,9 +94,14 @@ CREATE TABLE `platform_interface` ( - `fixed` — 固定值（value 字段） - `auth` — 从 `auth_config` 取（field 字段指定 key） - `extra` — 从 `extra_config` 取（初始化后保存的 userId/projectId 等） -- `context` — 从流程上下文取（contextKey 指定，如 `step1.access_token`） +- `context` — 从流程上下文或运行时上下文取（contextKey 指定，如 `step1.access_token`） - `device` — 从设备绑定关系取（如 `platform_device_id`） -- `thingsModel` — 从设备物模型实时值取（identifier 字段） +- `thingsModel` — 从设备物模型实时值取（设备绑定页选择 identifier） +- `ts` — 自动取设备上报时间 + +**数组字段补充：** +- `type=array` 的字段通过 `itemSchema` 描述数组元素结构 +- `itemSchema.source` 额外支持 `subDeviceId`，用于数组元素内绑定子设备平台 ID（如静力水准 `sensorId`） ### 3.3 `platform_flow` — 流程定义 @@ -159,7 +164,9 @@ CREATE TABLE `platform_device_mapping` ( `platform_device_id` varchar(100) DEFAULT NULL COMMENT 'DCP侧deviceId，入网后自动回填', `platform_point_id` varchar(100) DEFAULT NULL COMMENT 'DCP侧pointId', `sensor_type` varchar(50) NOT NULL COMMENT '传感器类型: inclino/stress/hyeto/static_level', + `push_interface_id` bigint DEFAULT NULL COMMENT '推送接口ID，决定设备绑定页动态字段', `field_mapping` json DEFAULT NULL COMMENT '字段映射 [{identifier,platformField,transform}]', + `extra_mapping` json DEFAULT NULL COMMENT '扩展映射，如数组字段子设备映射/基准点配置', `net_status` tinyint NOT NULL DEFAULT 0 COMMENT '入网状态(0=未入网,1=已入网)', `enable` char(1) NOT NULL DEFAULT '1', `remark` varchar(255) DEFAULT NULL, @@ -281,10 +288,11 @@ before any HTTP call: ``` // 挂接在 RuleProcess.processRuleScript() 调用方之后，异步执行 List mappings = findBySerialNumber(serialNumber) +// 既匹配主设备 serial_number，也匹配 extra_mapping 中登记的子设备 serial_number for each mapping where enable=1 and net_status=1: adapter = getAdapter(mapping.platformId) // 按 platform.code 路由 - adapter.pushData(mapping, payload) - // pushData 内部：字段映射转换 → Token检查 → HTTP调用 → 写日志 + adapter.pushData(platform, mapping, reportSerialNumber, payload) + // pushData 内部：按 request_schema / itemSchema 动态组装 → Token检查 → HTTP调用 → 写日志 ``` ### 5.4 DCP 传感器推送策略 @@ -294,10 +302,29 @@ for each mapping where enable=1 and net_status=1: | inclino | POST /api/app/inclino-data/many | xDeg, yDeg, temperature, epoch | 无 | | stress | POST /api/app/stress-data/many | e, epoch | e 必须严格递增 | | hyeto | POST /api/app/hyeto-data/many | value, epoch | value 必须严格递增 | -| static_level | POST /api/app/static-level-data/many | sensorData[{height,sensorId,epoch}] | 需两个 sensorId | +| static_level | POST /api/app/static-level-data/many | sensorData[{epoch,height,temperature,sensorId}] | 网关下子设备成对推送 | 递增约束处理：推送前查询 `platform_push_log` 中上次成功推送的值，若当前值 ≤ 上次值则跳过并记录警告。 +### 5.5 数组入参通用方案 + +当前实现不再把静力水准写成硬编码特例，而是抽象成“数组字段推送”： + +1. 推送接口在 `platform_interface.request_schema` 中定义 `type=array` 字段及 `itemSchema` +2. 设备绑定页读取所选推送接口的 `request_schema` +3. 标量字段中 `source=thingsModel` 的字段，渲染为普通字段映射 +4. 数组字段中 `itemSchema.source=thingsModel` 的字段，渲染为数组项字段映射 +5. 数组字段中 `itemSchema.source=subDeviceId` 的字段，渲染为子设备与平台字段映射 +6. 运行时由 `DeviceReportTrigger` 按实际上报的子设备序列号命中绑定关系，再由适配器组装完整数组请求体 + +静力水准是第一类落地场景： +- 绑定的是网关级设备关系，`platform_device_id` 也是网关级 +- `extra_mapping.arrayFields[].subDeviceMappings[]` 保存子设备序列号与 DCP `sensorId` 的对应关系 +- 其中一个子设备标记为基准点 +- 基准点上报时，推送 `[基准点, 基准点]` +- 监测点上报时，若基准点缓存仍在有效期内，则推送 `[基准点, 当前监测点]` +- 若监测点先到，则先暂存，待下一次基准点上报后补推 + --- ## 6. 前端页面（yack-web） @@ -320,7 +347,13 @@ for each mapping where enable=1 and net_status=1: - 新增弹窗： - 设备搜索选择器（输入名称/序列号模糊搜索） - 传感器类型下拉 - - 字段映射：物模型字段下拉（自动加载该设备产品的物模型）+ DCP字段（根据传感器类型预填建议值） + - 推送接口选择器：从页面选择平台下已定义的推送接口 + - 字段映射：根据所选推送接口 `request_schema` 动态渲染，不再按传感器类型写死 + - 若 `request_schema` 含 `type=array` 字段，则自动出现“数组字段配置”区域 + - 数组字段配置包含： + - 子设备序列号 ↔ 平台子设备字段映射（如 `sensorId`） + - 数组元素中 `source=thingsModel` 字段的物模型映射（如 `height`、`temperature`） + - 基准点选择与缓存有效间隔 - 操作：**入网**（执行 flow_type=2 流程，成功后自动回填 DCP deviceId）、启用/禁用 ### 推送日志 @@ -334,7 +367,7 @@ for each mapping where enable=1 and net_status=1: 系统内置 DCP 平台的接口定义和流程配置（SQL 初始化数据），用户只需填入账号密码即可使用。 -**内置接口（6个）：** +**内置接口（12个）：** 1. `dcp_auth` — POST /connect/token 2. `dcp_get_user` — GET /api/app/user/user-id 3. `dcp_get_projects` — GET /api/app/project/by-user-id/{userId} @@ -343,13 +376,24 @@ for each mapping where enable=1 and net_status=1: 6. `dcp_bind_stress` — POST /api/app/stress-sensor 7. `dcp_bind_hyeto` — POST /api/app/hyeto-sensor 8. `dcp_bind_static_level` — POST /api/app/static-level-device +9. `dcp_push_inclino` — POST /api/app/inclino-data/many +10. `dcp_push_stress` — POST /api/app/stress-data/many +11. `dcp_push_hyeto` — POST /api/app/hyeto-data/many +12. `dcp_push_static_level` — POST /api/app/static-level-data/many -**内置流程（5个）：** +**内置流程（6个）：** 1. `dcp_init`（flow_type=1）：Step1=dcp_auth → Step2=dcp_get_user → Step3=dcp_get_projects 2. `dcp_token_refresh`（flow_type=3）：Step1=dcp_auth 3. `dcp_onboard_inclino`（flow_type=2, sensor_type=inclino）：Step1=dcp_create_point → Step2=dcp_bind_inclino 4. `dcp_onboard_stress`（flow_type=2, sensor_type=stress）：Step1=dcp_create_point → Step2=dcp_bind_stress 5. `dcp_onboard_hyeto`（flow_type=2, sensor_type=hyeto）：Step1=dcp_create_point → Step2=dcp_bind_hyeto +6. `dcp_onboard_static_level`（flow_type=2, sensor_type=static_level）：Step1=dcp_create_point → Step2=dcp_bind_static_level + +补充： +- 上述接口和流程既可通过 SQL 初始化，也可直接在页面维护 +- 当前已具备接口管理页、流程编排页、设备绑定页，不需要再靠改 SQL 才能新增运行系统 +- 接口入参支持 `uuid` 来源：运行时每次请求生成新的 UUID，避免把 `auto` 字符串直接发送给目标平台 +- 接口入参支持 `referenceSubDeviceSensorIdPairs` 来源：按设备绑定页配置的基准点，将子设备拆成 `[基准点, 监测点]` 多个请求批次，适合静力水准等成组设备绑定场景 --- diff --git a/fastbee-service/fastbee-iot-service/src/main/java/com/fastbee/iot/platform/adapter/dcp/DcpPlatformAdapter.java b/fastbee-service/fastbee-iot-service/src/main/java/com/fastbee/iot/platform/adapter/dcp/DcpPlatformAdapter.java index 9506d8df..bc45fb61 100644 --- a/fastbee-service/fastbee-iot-service/src/main/java/com/fastbee/iot/platform/adapter/dcp/DcpPlatformAdapter.java +++ b/fastbee-service/fastbee-iot-service/src/main/java/com/fastbee/iot/platform/adapter/dcp/DcpPlatformAdapter.java @@ -11,11 +11,13 @@ import com.fastbee.http.model.HttpClientConfig; import com.fastbee.http.service.HttpClientFactory; import com.fastbee.iot.platform.domain.PlatformDeviceMapping; import com.fastbee.iot.platform.domain.PlatformFlow; +import com.fastbee.iot.platform.domain.PlatformInterface; import com.fastbee.iot.platform.domain.PlatformInstance; import com.fastbee.iot.platform.domain.PlatformPushLog; import com.fastbee.iot.platform.flow.FlowContext; import com.fastbee.iot.platform.flow.PlatformFlowExecutor; import com.fastbee.iot.platform.mapper.PlatformFlowMapper; +import com.fastbee.iot.platform.mapper.PlatformInterfaceMapper; import com.fastbee.iot.platform.service.IPlatformInstanceService; import com.fastbee.iot.platform.service.IPlatformPushLogService; import lombok.extern.slf4j.Slf4j; @@ -23,80 +25,773 @@ import org.springframework.beans.factory.annotation.Autowired; import org.springframework.stereotype.Component; import org.springframework.util.StringUtils; -import java.text.SimpleDateFormat; import java.time.Instant; -import java.util.*; +import java.time.LocalDateTime; +import java.time.OffsetDateTime; +import java.time.ZoneId; +import java.time.format.DateTimeFormatter; +import java.time.format.DateTimeParseException; +import java.util.Date; +import java.util.HashMap; +import java.util.List; +import java.util.Map; /** * DCP 平台实时数据推送适配器 - * 仅负责实时推送（单步直接调用），不走流程引擎 + *

+ * 支持两种推送模式： + *

    + *
  1. 普通传感器：按 pushInterface.requestSchema + fieldMapping 动态组装请求体
    2. + *
  2. 数组字段传感器：按 array itemSchema + extraMapping 组装，并做基准点/监测点对齐
    3. + *
*/ @Slf4j @Component public class DcpPlatformAdapter { + private static final DateTimeFormatter ISO_OFFSET_MILLIS = + DateTimeFormatter.ofPattern("yyyy-MM-dd'T'HH:mm:ss.SSSXXX"); + private static final DateTimeFormatter DATE_TIME_MILLIS = + DateTimeFormatter.ofPattern("yyyy-MM-dd HH:mm:ss.SSS"); + private static final DateTimeFormatter DATE_TIME_SECONDS = + DateTimeFormatter.ofPattern("yyyy-MM-dd HH:mm:ss"); + @Autowired private IPlatformInstanceService instanceService; @Autowired private IPlatformPushLogService pushLogService; @Autowired private PlatformFlowExecutor flowExecutor; + @Autowired + private StaticLevelBufferService bufferService; + @Autowired + private PlatformInterfaceMapper interfaceMapper; /** 传感器类型 → 推送接口路径 */ private static final Map PUSH_PATH = new HashMap<>(); static { - PUSH_PATH.put("inclino", "/api/app/inclino-data/many"); // 倾角 - PUSH_PATH.put("stress", "/api/app/stress-data/many"); // 轴力 - PUSH_PATH.put("hyeto", "/api/app/hyeto-data/many"); // 雨量 - PUSH_PATH.put("static_level", "/api/app/static-level-data/many"); // 静力水准 + PUSH_PATH.put("inclino", "/api/app/inclino-data/many"); + PUSH_PATH.put("stress", "/api/app/stress-data/many"); + PUSH_PATH.put("hyeto", "/api/app/hyeto-data/many"); + PUSH_PATH.put("static_level", "/api/app/static-level-data/many"); } /** * 推送设备上报数据到 DCP - * - * @param platform 平台实例 - * @param mapping 设备绑定关系 - * @param payload 设备上报的原始 JSON payload */ - public void pushData(PlatformInstance platform, PlatformDeviceMapping mapping, String payload) { + public void pushData(PlatformInstance platform, PlatformDeviceMapping mapping, + String reportSerialNumber, String payload) { if (mapping.getNetStatus() == null || mapping.getNetStatus() != 1) { log.warn("[DCP] 设备 {} 未入网，跳过推送", mapping.getSerialNumber()); return; } String sensorType = mapping.getSensorType(); - String pushPath = PUSH_PATH.get(sensorType); + PlatformInterface pushInterface = loadPushInterface(mapping); + JSONArray requestSchema = loadRequestSchema(pushInterface); + String interfaceCode = resolveInterfaceCode(sensorType, pushInterface); + String pushPath = resolvePushPath(sensorType, pushInterface); if (pushPath == null) { log.warn("[DCP] 未知传感器类型: {}", sensorType); return; } - long startMs = System.currentTimeMillis(); - PlatformPushLog logEntry = new PlatformPushLog(); - logEntry.setPlatformId(platform.getId()); - logEntry.setDeviceId(mapping.getDeviceId()); - logEntry.setSerialNumber(mapping.getSerialNumber()); - logEntry.setInterfaceCode("dcp_push_" + sensorType); - logEntry.setStatus(0); - logEntry.setRetryCount(0); - logEntry.setLogType(1); // 数据推送日志 - try { - // 1. 字段映射转换 - Object body = buildPushBody(mapping, payload, sensorType); - if (body == null) { - log.warn("[DCP] 设备 {} 字段映射结果为空，跳过推送", mapping.getSerialNumber()); - return; + JSONArray arrayFieldsDef = extractArrayFieldsDef(mapping.getExtraMapping()); + if (arrayFieldsDef != null && !arrayFieldsDef.isEmpty()) { + handleArrayFieldPush( + platform, mapping, reportSerialNumber, payload, pushPath, + interfaceCode, requestSchema, arrayFieldsDef); + } else { + handleNormalPush( + platform, mapping, reportSerialNumber, payload, pushPath, + interfaceCode, requestSchema); + } + } catch (Exception e) { + String serialNumber = StringUtils.hasText(reportSerialNumber) + ? reportSerialNumber : mapping.getSerialNumber(); + log.error("[DCP] 推送失败 device={} sensor={}", serialNumber, sensorType, e); + PlatformPushLog logEntry = createLogEntry(platform, mapping, interfaceCode, serialNumber); + logEntry.setStatus(2); + logEntry.setErrorMsg(e.getMessage()); + pushLogService.asyncSave(logEntry); + } + } + + // ==================== 普通推送 ==================== + + private void handleNormalPush(PlatformInstance platform, PlatformDeviceMapping mapping, + String reportSerialNumber, String payload, String pushPath, + String interfaceCode, JSONArray requestSchema) { + Object body = buildPushBody(platform, mapping, payload, requestSchema); + if (body == null) { + log.warn("[DCP] 设备 {} 字段映射结果为空，跳过推送", mapping.getSerialNumber()); + return; + } + PlatformPushLog logEntry = createLogEntry(platform, mapping, interfaceCode, reportSerialNumber); + doHttpPost(platform, platform.getBaseUrl() + pushPath, JSON.toJSONString(body), logEntry); + pushLogService.asyncSave(logEntry); + } + + private Object buildPushBody(PlatformInstance platform, PlatformDeviceMapping mapping, + String payload, JSONArray requestSchema) { + if (!StringUtils.hasText(payload)) { + return null; + } + JSONObject reportData = parsePayloadToMap(payload); + if (reportData == null || reportData.isEmpty()) { + return null; + } + + JSONArray fieldMappings = parseJsonArray(mapping.getFieldMapping()); + if (requestSchema == null || requestSchema.isEmpty()) { + return buildLegacyPushBody(mapping, fieldMappings, reportData); + } + + JSONObject dcpData = buildObjectBySchema(platform, mapping, requestSchema, reportData, fieldMappings); + if (dcpData.isEmpty() || !hasPayloadDrivenValue(dcpData, requestSchema)) { + return buildLegacyPushBody(mapping, fieldMappings, reportData); + } + + JSONArray arr = new JSONArray(); + arr.add(dcpData); + return arr; + } + + private Object buildLegacyPushBody(PlatformDeviceMapping mapping, JSONArray fieldMappings, JSONObject reportData) { + if (fieldMappings == null || fieldMappings.isEmpty()) { + return null; + } + + JSONObject dcpData = new JSONObject(); + dcpData.put("deviceId", mapping.getPlatformDeviceId()); + dcpData.put("epoch", nowIsoTimestamp()); + + for (int i = 0; i < fieldMappings.size(); i++) { + JSONObject fm = fieldMappings.getJSONObject(i); + String identifier = fm.getString("identifier"); + String platformField = fm.getString("platformField"); + if (!StringUtils.hasText(identifier) || !StringUtils.hasText(platformField)) { + continue; + } + if ("deviceId".equals(platformField) || "epoch".equals(platformField)) { + continue; + } + Object value = reportData.get(identifier); + if (value == null) { + continue; + } + dcpData.put(platformField, value); + } + + JSONArray arr = new JSONArray(); + arr.add(dcpData); + return arr; + } + + // ==================== 数组字段推送 ==================== + + /** + * 从 extraMapping 中提取 arrayFields 定义 + */ + private JSONArray extractArrayFieldsDef(String extraMapping) { + if (!StringUtils.hasText(extraMapping)) return null; + try { + JSONObject extra = JSON.parseObject(extraMapping); + JSONArray arrayFields = extra.getJSONArray("arrayFields"); + if (arrayFields != null && !arrayFields.isEmpty()) return arrayFields; + JSONObject staticLevel = extra.getJSONObject("staticLevel"); + if (staticLevel != null) { + return convertLegacyStaticLevel(staticLevel); + } + } catch (Exception e) { + log.warn("[DCP] 解析 extraMapping 失败: {}", e.getMessage()); + } + return null; + } + + /** 将旧格式 {staticLevel:{...}} 转换为新 arrayFields 格式 */ + private JSONArray convertLegacyStaticLevel(JSONObject staticLevel) { + JSONArray arrayFields = new JSONArray(); + JSONObject af = new JSONObject(); + af.put("fieldName", "sensorData"); + af.put("reportIntervalSec", 300); + + JSONArray mappings = staticLevel.getJSONArray("sensorIdMappings"); + JSONArray subDeviceMappings = new JSONArray(); + if (mappings != null) { + String refSN = staticLevel.getString("referenceSerialNumber"); + for (int i = 0; i < mappings.size(); i++) { + JSONObject m = mappings.getJSONObject(i); + JSONObject sdm = new JSONObject(); + sdm.put("serialNumber", m.getString("serialNumber")); + sdm.put("sensorId", m.getString("sensorId")); + sdm.put("isReference", m.getString("serialNumber").equals(refSN)); + subDeviceMappings.add(sdm); + } + } + af.put("subDeviceMappings", subDeviceMappings); + + JSONArray itemMappings = new JSONArray(); + if (StringUtils.hasText(staticLevel.getString("heightIdentifier"))) { + JSONObject h = new JSONObject(); + h.put("platformField", "height"); + h.put("identifier", staticLevel.getString("heightIdentifier")); + itemMappings.add(h); + } + if (StringUtils.hasText(staticLevel.getString("temperatureIdentifier"))) { + JSONObject t = new JSONObject(); + t.put("platformField", "temperature"); + t.put("identifier", staticLevel.getString("temperatureIdentifier")); + itemMappings.add(t); + } + af.put("itemFieldMappings", itemMappings); + + arrayFields.add(af); + return arrayFields; + } + + private void handleArrayFieldPush(PlatformInstance platform, PlatformDeviceMapping mapping, + String reportSerialNumber, String payload, + String pushPath, String interfaceCode, + JSONArray requestSchema, JSONArray arrayFieldsDef) { + JSONObject reportData = parsePayloadToMap(payload); + if (reportData == null || reportData.isEmpty()) { + return; + } + + JSONArray fieldMappings = parseJsonArray(mapping.getFieldMapping()); + String currentSerialNumber = StringUtils.hasText(reportSerialNumber) + ? reportSerialNumber : mapping.getSerialNumber(); + String baseUrl = platform.getBaseUrl() + pushPath; + + for (int i = 0; i < arrayFieldsDef.size(); i++) { + JSONObject af = arrayFieldsDef.getJSONObject(i); + String fieldName = af.getString("fieldName"); + int reportIntervalSec = af.getIntValue("reportIntervalSec", 300); + JSONArray subDeviceMappings = af.getJSONArray("subDeviceMappings"); + JSONArray itemFieldMappings = af.getJSONArray("itemFieldMappings"); + + if (subDeviceMappings == null || subDeviceMappings.isEmpty()) { + log.warn("[DCP] arrayField {} 子设备映射为空", fieldName); + continue; } - // 2. 确保 Token 有效 + JSONObject currentSubDevice = findSubDevice(subDeviceMappings, currentSerialNumber); + if (currentSubDevice == null) { + log.debug("[DCP] 设备 {} 不属于 arrayField {} 的子设备列表，跳过", currentSerialNumber, fieldName); + continue; + } + + JSONObject refSubDevice = findReferenceDevice(subDeviceMappings); + if (refSubDevice == null) { + log.warn("[DCP] arrayField {} 未指定基准点", fieldName); + continue; + } + + String refSerialNumber = refSubDevice.getString("serialNumber"); + // 子设备 sensorId 优先取页面配置；未显式配置时回退到子设备编号，减少手工维护。 + String currentSensorId = resolveSubDeviceSensorId(currentSubDevice, currentSerialNumber); + boolean isCurrentReference = Boolean.TRUE.equals(currentSubDevice.getBoolean("isReference")); + + JSONObject currentItem = buildArrayItem( + platform, mapping, requestSchema, reportData, fieldName, itemFieldMappings, currentSensorId); + if (currentItem == null || currentItem.isEmpty()) { + log.warn("[DCP] 设备 {} arrayField {} 映射结果为空，跳过", currentSerialNumber, fieldName); + continue; + } + + JSONObject currentBodyFields = buildObjectBySchema(platform, mapping, requestSchema, reportData, fieldMappings); + + if (isCurrentReference) { + bufferService.saveReferenceData( + platform.getId(), fieldName, currentSerialNumber, currentItem, reportIntervalSec); + pushReferenceSelfPair(platform, mapping, baseUrl, interfaceCode, currentSerialNumber, + currentBodyFields, fieldName, currentItem); + flushCachedMonitorData(platform, mapping, baseUrl, interfaceCode, fieldName, + subDeviceMappings, currentSerialNumber, currentItem, reportIntervalSec); + continue; + } + + JSONObject refData = bufferService.getReferenceDataIfFresh( + platform.getId(), fieldName, refSerialNumber, reportIntervalSec); + if (refData != null && refData.getJSONObject("itemData") != null) { + // 监测点上报时，只有在缓存中拿到有效基准点数据，才组装成 [基准点, 监测点]。 + JSONArray monitorSensorData = new JSONArray(); + monitorSensorData.add(refData.getJSONObject("itemData")); + monitorSensorData.add(currentItem); + JSONArray monitorBody = buildArrayPushBody(mapping, currentBodyFields, fieldName, monitorSensorData); + + PlatformPushLog monitorLog = createLogEntry(platform, mapping, interfaceCode, currentSerialNumber); + doHttpPost(platform, baseUrl, JSON.toJSONString(monitorBody), monitorLog); + pushLogService.asyncSave(monitorLog); + log.info("[DCP] 监测点 {} 基准点就绪，已推送", currentSerialNumber); + } else { + bufferService.saveMonitorData( + platform.getId(), fieldName, currentSerialNumber, currentItem, currentBodyFields, reportIntervalSec); + log.info("[DCP] 监测点 {} 基准点未就绪，已缓存等待基准点上报", currentSerialNumber); + } + } + } + + private void pushReferenceSelfPair(PlatformInstance platform, PlatformDeviceMapping mapping, + String baseUrl, String interfaceCode, String refSerialNumber, + JSONObject bodyFields, String fieldName, JSONObject refItem) { + JSONArray sensorData = new JSONArray(); + sensorData.add(refItem); + sensorData.add(refItem); + JSONArray body = buildArrayPushBody(mapping, bodyFields, fieldName, sensorData); + + PlatformPushLog logEntry = createLogEntry(platform, mapping, interfaceCode, refSerialNumber); + doHttpPost(platform, baseUrl, JSON.toJSONString(body), logEntry); + pushLogService.asyncSave(logEntry); + log.info("[DCP] 基准点 {} 已推送自配对数据", refSerialNumber); + } + + private void flushCachedMonitorData(PlatformInstance platform, PlatformDeviceMapping mapping, + String baseUrl, String interfaceCode, String fieldName, + JSONArray subDeviceMappings, String refSerialNumber, + JSONObject refItem, int reportIntervalSec) { + for (int i = 0; i < subDeviceMappings.size(); i++) { + JSONObject subDevice = subDeviceMappings.getJSONObject(i); + String monitorSerialNumber = subDevice.getString("serialNumber"); + if (!StringUtils.hasText(monitorSerialNumber) || refSerialNumber.equals(monitorSerialNumber)) { + continue; + } + JSONObject monitorData = bufferService.getMonitorDataIfFresh( + platform.getId(), fieldName, monitorSerialNumber, reportIntervalSec); + if (monitorData == null || monitorData.getJSONObject("itemData") == null) { + continue; + } + + JSONArray sensorData = new JSONArray(); + sensorData.add(refItem); + sensorData.add(monitorData.getJSONObject("itemData")); + JSONArray body = buildArrayPushBody( + mapping, monitorData.getJSONObject("bodyFields"), fieldName, sensorData); + + PlatformPushLog logEntry = createLogEntry(platform, mapping, interfaceCode, monitorSerialNumber); + doHttpPost(platform, baseUrl, JSON.toJSONString(body), logEntry); + pushLogService.asyncSave(logEntry); + if (Integer.valueOf(1).equals(logEntry.getStatus())) { + bufferService.removeMonitorData(platform.getId(), fieldName, monitorSerialNumber); + } + log.info("[DCP] 基准点 {} 上报后补推监测点 {}", refSerialNumber, monitorSerialNumber); + } + } + + // ==================== 构建辅助 ==================== + + private JSONArray buildArrayPushBody(PlatformDeviceMapping mapping, JSONObject bodyFields, + String fieldName, JSONArray sensorData) { + JSONObject body = copyJsonObject(bodyFields); + body.put(fieldName, sensorData); + if (!body.containsKey("deviceId")) { + body.put("deviceId", mapping.getPlatformDeviceId()); + } + if (!body.containsKey("ct")) { + body.put("ct", 1); + } + JSONArray arr = new JSONArray(); + arr.add(body); + return arr; + } + + private JSONObject buildObjectBySchema(PlatformInstance platform, PlatformDeviceMapping mapping, + JSONArray requestSchema, JSONObject reportData, + JSONArray fieldMappings) { + JSONObject body = new JSONObject(); + if (requestSchema == null || requestSchema.isEmpty()) { + return body; + } + for (int i = 0; i < requestSchema.size(); i++) { + JSONObject schemaField = requestSchema.getJSONObject(i); + if ("array".equalsIgnoreCase(schemaField.getString("type"))) { + continue; + } + String fieldName = schemaField.getString("name"); + if (!StringUtils.hasText(fieldName)) { + continue; + } + Object value = resolveSchemaFieldValue(platform, mapping, schemaField, reportData, fieldMappings); + if (value != null) { + body.put(fieldName, value); + } + } + return body; + } + + private Object resolveSchemaFieldValue(PlatformInstance platform, PlatformDeviceMapping mapping, + JSONObject schemaField, JSONObject reportData, + JSONArray fieldMappings) { + String source = schemaField.getString("source"); + String fieldName = schemaField.getString("name"); + String configField = schemaField.getString("field"); + String contextKey = schemaField.getString("contextKey"); + + if ("fixed".equals(source)) { + return schemaField.get("value"); + } + if ("device".equals(source)) { + return resolveDeviceField(mapping, configField); + } + if ("auth".equals(source)) { + return resolveJsonField(platform.getAuthConfig(), configField); + } + if ("extra".equals(source)) { + return resolveJsonField(platform.getExtraConfig(), configField); + } + if (isThingsModelSource(source)) { + return resolveThingsModelValue(reportData, fieldMappings, fieldName, contextKey); + } + if ("ts".equals(source)) { + return resolveTimestampValue(reportData, fieldMappings, fieldName, contextKey); + } + if ("context".equals(source)) { + if ("reportTime".equals(contextKey) || "epoch".equals(fieldName)) { + return resolveTimestampValue(reportData, fieldMappings, fieldName, contextKey); + } + Object contextValue = reportData.get(contextKey); + if (contextValue != null) { + return contextValue; + } + return resolveThingsModelValue(reportData, fieldMappings, fieldName, contextKey); + } + return null; + } + + private JSONObject buildArrayItem(PlatformInstance platform, PlatformDeviceMapping mapping, + JSONArray requestSchema, JSONObject reportData, String fieldName, + JSONArray itemFieldMappings, String subDeviceId) { + JSONObject arrayFieldSchema = findArrayFieldSchema(requestSchema, fieldName); + JSONArray itemSchema = arrayFieldSchema == null ? null : arrayFieldSchema.getJSONArray("itemSchema"); + if (itemSchema == null || itemSchema.isEmpty()) { + return buildLegacyArrayItem(reportData, itemFieldMappings, subDeviceId); + } + + JSONObject item = new JSONObject(); + boolean hasThingsModelField = false; + boolean hasThingsModelValue = false; + for (int i = 0; i < itemSchema.size(); i++) { + JSONObject schemaField = itemSchema.getJSONObject(i); + String itemFieldName = schemaField.getString("name"); + if (!StringUtils.hasText(itemFieldName)) { + continue; + } + if (isThingsModelSource(schemaField.getString("source"))) { + hasThingsModelField = true; + } + Object value = resolveArrayItemFieldValue( + platform, mapping, schemaField, reportData, itemFieldMappings, subDeviceId); + if (value != null) { + item.put(itemFieldName, value); + if (isThingsModelSource(schemaField.getString("source"))) { + hasThingsModelValue = true; + } + } + } + if (hasThingsModelField && !hasThingsModelValue) { + return null; + } + return item; + } + + private JSONObject buildLegacyArrayItem(JSONObject reportData, JSONArray itemFieldMappings, String subDeviceId) { + Map itemValues = extractItemValues(reportData, itemFieldMappings); + Object height = itemValues.get("height"); + Object temperature = itemValues.get("temperature"); + if (height == null && temperature == null) { + return null; + } + String heightIdentifier = findIdentifier(itemFieldMappings, "height"); + String epochIdentifier = StringUtils.hasText(heightIdentifier) + ? heightIdentifier : findFirstIdentifier(itemFieldMappings); + Object epoch = resolveEpoch(reportData, epochIdentifier); + return buildSensorDataItem(epoch, height, temperature, subDeviceId); + } + + private Object resolveArrayItemFieldValue(PlatformInstance platform, PlatformDeviceMapping mapping, + JSONObject schemaField, JSONObject reportData, + JSONArray itemFieldMappings, String subDeviceId) { + String source = schemaField.getString("source"); + String fieldName = schemaField.getString("name"); + if ("fixed".equals(source)) { + return schemaField.get("value"); + } + if ("subDeviceId".equals(source)) { + return subDeviceId; + } + if ("ts".equals(source)) { + return resolveTimestampValue(reportData, itemFieldMappings, fieldName, schemaField.getString("contextKey")); + } + if ("device".equals(source)) { + return resolveDeviceField(mapping, schemaField.getString("field")); + } + if ("auth".equals(source)) { + return resolveJsonField(platform.getAuthConfig(), schemaField.getString("field")); + } + if ("extra".equals(source)) { + return resolveJsonField(platform.getExtraConfig(), schemaField.getString("field")); + } + if (isThingsModelSource(source)) { + String identifier = findIdentifier(itemFieldMappings, fieldName); + return StringUtils.hasText(identifier) ? reportData.get(identifier) : null; + } + if ("context".equals(source)) { + String contextKey = schemaField.getString("contextKey"); + if ("reportTime".equals(contextKey) || "epoch".equals(fieldName)) { + return resolveTimestampValue(reportData, itemFieldMappings, fieldName, contextKey); + } + Object contextValue = reportData.get(contextKey); + if (contextValue != null) { + return contextValue; + } + } + return null; + } + + private JSONObject findArrayFieldSchema(JSONArray requestSchema, String fieldName) { + if (requestSchema == null || requestSchema.isEmpty() || !StringUtils.hasText(fieldName)) { + return null; + } + for (int i = 0; i < requestSchema.size(); i++) { + JSONObject schemaField = requestSchema.getJSONObject(i); + if ("array".equalsIgnoreCase(schemaField.getString("type")) + && fieldName.equals(schemaField.getString("name"))) { + return schemaField; + } + } + return null; + } + + private JSONObject copyJsonObject(JSONObject source) { + if (source == null || source.isEmpty()) { + return new JSONObject(); + } + return JSON.parseObject(source.toJSONString()); + } + + private boolean hasPayloadDrivenValue(JSONObject body, JSONArray requestSchema) { + if (body == null || body.isEmpty() || requestSchema == null || requestSchema.isEmpty()) { + return false; + } + for (int i = 0; i < requestSchema.size(); i++) { + JSONObject schemaField = requestSchema.getJSONObject(i); + if ("array".equalsIgnoreCase(schemaField.getString("type"))) { + continue; + } + if (usesPayload(schemaField.getString("source")) + && body.containsKey(schemaField.getString("name"))) { + return true; + } + } + return false; + } + + private boolean usesPayload(String source) { + return isThingsModelSource(source) || "context".equals(source) || "ts".equals(source); + } + + private boolean isThingsModelSource(String source) { + return "thingsModel".equals(source) || "identifier".equals(source); + } + + private PlatformInterface loadPushInterface(PlatformDeviceMapping mapping) { + if (mapping.getPushInterfaceId() == null) { + return null; + } + return interfaceMapper.selectById(mapping.getPushInterfaceId()); + } + + private JSONArray loadRequestSchema(PlatformInterface pushInterface) { + if (pushInterface == null || !StringUtils.hasText(pushInterface.getRequestSchema())) { + return null; + } + try { + return JSON.parseArray(pushInterface.getRequestSchema()); + } catch (Exception e) { + log.warn("[DCP] 解析 requestSchema 失败 interfaceId={} err={}", + pushInterface.getId(), e.getMessage()); + return null; + } + } + + private String resolvePushPath(String sensorType, PlatformInterface pushInterface) { + if (pushInterface != null && StringUtils.hasText(pushInterface.getPath())) { + return pushInterface.getPath(); + } + return PUSH_PATH.get(sensorType); + } + + private String resolveInterfaceCode(String sensorType, PlatformInterface pushInterface) { + if (pushInterface != null && StringUtils.hasText(pushInterface.getCode())) { + return pushInterface.getCode(); + } + return "dcp_push_" + sensorType; + } + + private JSONArray parseJsonArray(String json) { + if (!StringUtils.hasText(json)) { + return null; + } + try { + return JSON.parseArray(json); + } catch (Exception e) { + log.warn("[DCP] 解析 JSON 数组失败: {}", e.getMessage()); + return null; + } + } + + private Object resolveThingsModelValue(JSONObject reportData, JSONArray fieldMappings, + String platformField, String fallbackKey) { + String identifier = findIdentifier(fieldMappings, platformField); + if (StringUtils.hasText(identifier)) { + return reportData.get(identifier); + } + if (StringUtils.hasText(fallbackKey)) { + return reportData.get(fallbackKey); + } + return null; + } + + private Object resolveTimestampValue(JSONObject reportData, JSONArray fieldMappings, + String platformField, String fallbackKey) { + String identifier = findIdentifier(fieldMappings, platformField); + if (StringUtils.hasText(identifier)) { + return resolveEpoch(reportData, identifier); + } + if (StringUtils.hasText(fallbackKey) && reportData.containsKey(fallbackKey + "__ts")) { + return resolveEpoch(reportData, fallbackKey); + } + String firstIdentifier = findFirstIdentifier(fieldMappings); + if (StringUtils.hasText(firstIdentifier)) { + return resolveEpoch(reportData, firstIdentifier); + } + return nowIsoTimestamp(); + } + + private Object resolveDeviceField(PlatformDeviceMapping mapping, String field) { + if (!StringUtils.hasText(field)) { + return null; + } + switch (field) { + case "platform_device_id": + return mapping.getPlatformDeviceId(); + case "platform_point_id": + return mapping.getPlatformPointId(); + case "serial_number": + return mapping.getSerialNumber(); + default: + return null; + } + } + + private Object resolveJsonField(String json, String field) { + if (!StringUtils.hasText(json) || !StringUtils.hasText(field)) { + return null; + } + try { + return JSON.parseObject(json).get(field); + } catch (Exception e) { + return null; + } + } + + /** 构建静力水准兼容 item */ + private JSONObject buildSensorDataItem(Object epoch, Object height, Object temperature, String sensorId) { + JSONObject item = new JSONObject(); + item.put("epoch", epoch); + item.put("height", height); + item.put("temperature", temperature); + item.put("sensorId", sensorId); + return item; + } + + private Map extractItemValues(JSONObject reportData, JSONArray itemFieldMappings) { + Map values = new HashMap<>(); + if (itemFieldMappings == null) return values; + for (int i = 0; i < itemFieldMappings.size(); i++) { + JSONObject im = itemFieldMappings.getJSONObject(i); + String platformField = im.getString("platformField"); + String identifier = im.getString("identifier"); + if (StringUtils.hasText(identifier) && StringUtils.hasText(platformField)) { + values.put(platformField, reportData.get(identifier)); + } + } + return values; + } + + private String findIdentifier(JSONArray itemFieldMappings, String platformField) { + if (itemFieldMappings == null || !StringUtils.hasText(platformField)) { + return null; + } + for (int i = 0; i < itemFieldMappings.size(); i++) { + JSONObject im = itemFieldMappings.getJSONObject(i); + if (platformField.equals(im.getString("platformField"))) { + return im.getString("identifier"); + } + } + return null; + } + + private String findFirstIdentifier(JSONArray itemFieldMappings) { + if (itemFieldMappings == null || itemFieldMappings.isEmpty()) { + return null; + } + for (int i = 0; i < itemFieldMappings.size(); i++) { + String identifier = itemFieldMappings.getJSONObject(i).getString("identifier"); + if (StringUtils.hasText(identifier)) { + return identifier; + } + } + return null; + } + + private JSONObject findSubDevice(JSONArray mappings, String serialNumber) { + if (mappings == null || !StringUtils.hasText(serialNumber)) { + return null; + } + for (int i = 0; i < mappings.size(); i++) { + JSONObject mapping = mappings.getJSONObject(i); + if (serialNumber.equals(mapping.getString("serialNumber"))) { + return mapping; + } + } + return null; + } + + private JSONObject findReferenceDevice(JSONArray mappings) { + if (mappings == null || mappings.isEmpty()) { + return null; + } + for (int i = 0; i < mappings.size(); i++) { + JSONObject mapping = mappings.getJSONObject(i); + if (Boolean.TRUE.equals(mapping.getBoolean("isReference"))) { + return mapping; + } + } + return mappings.getJSONObject(0); + } + + private String resolveSubDeviceSensorId(JSONObject subDevice, String fallbackSerialNumber) { + if (subDevice == null) { + return normalizeText(fallbackSerialNumber); + } + String sensorId = normalizeText(subDevice.getString("sensorId")); + if (StringUtils.hasText(sensorId)) { + return sensorId; + } + return normalizeText(subDevice.getString("serialNumber")); + } + + // ==================== HTTP 执行 ==================== + + private void doHttpPost(PlatformInstance platform, String url, String bodyStr, PlatformPushLog logEntry) { + long startMs = System.currentTimeMillis(); + logEntry.setRequestUrl(url); + logEntry.setRequestBody(bodyStr); + + try { String token = ensureToken(platform); - - // 3. 构建请求 - String url = platform.getBaseUrl() + pushPath; - String bodyStr = JSON.toJSONString(body); - logEntry.setRequestUrl(url); - logEntry.setRequestBody(bodyStr); - Map headers = new HashMap<>(); headers.put("Authorization", "Bearer " + token); headers.put("Content-Type", "application/json"); @@ -114,63 +809,35 @@ public class DcpPlatformAdapter { logEntry.setResponseBody(responseBody); logEntry.setResponseCode(200); logEntry.setStatus(1); - logEntry.setCostMs((int) (System.currentTimeMillis() - startMs)); - + log.info("[DCP] 推送成功 url={} cost={}ms", url, System.currentTimeMillis() - startMs); } catch (Exception e) { - log.error("[DCP] 推送失败 device={} sensor={}", mapping.getSerialNumber(), sensorType, e); + log.error("[DCP] HTTP 执行失败 url={}", url, e); logEntry.setStatus(2); logEntry.setErrorMsg(e.getMessage()); - logEntry.setCostMs((int) (System.currentTimeMillis() - startMs)); } finally { - pushLogService.asyncSave(logEntry); + logEntry.setCostMs((int) (System.currentTimeMillis() - startMs)); } } - /** - * 根据 field_mapping 将设备上报数据转换为 DCP 推送格式 - */ - private Object buildPushBody(PlatformDeviceMapping mapping, String payload, String sensorType) { - if (!StringUtils.hasText(mapping.getFieldMapping()) || !StringUtils.hasText(payload)) { - return null; - } - JSONArray fieldMappings = JSON.parseArray(mapping.getFieldMapping()); - - // payload 可能是数组 [{id,value,ts},...] 或对象 {key:value,...} - JSONObject reportData = parsePayloadToMap(payload); - if (reportData == null || reportData.isEmpty()) { - return null; - } - - JSONObject dcpData = new JSONObject(); - dcpData.put("deviceId", mapping.getPlatformDeviceId()); - // 时间处理 时间戳转 ISO8601 - Instant now = Instant.now(); - dcpData.put("epoch", now); - - // 按字段映射转换 - for (int i = 0; i < fieldMappings.size(); i++) { - JSONObject fm = fieldMappings.getJSONObject(i); - String identifier = fm.getString("identifier"); - String platformField = fm.getString("platformField"); - //String transform = fm.getString("transform"); - - Object value = reportData.get(identifier); - if (value == null) continue; - dcpData.put(platformField, value); - } - - // 包装为数组（DCP 推送接口接受数组） - JSONArray arr = new JSONArray(); - arr.add(dcpData); - return arr; + private PlatformPushLog createLogEntry(PlatformInstance platform, PlatformDeviceMapping mapping, String ifaceCode) { + return createLogEntry(platform, mapping, ifaceCode, mapping.getSerialNumber()); } - /** - * 将设备上报 payload 解析为 {identifier: value} 的扁平 map - * 支持两种格式： - * 1. 数组: [{"id":"AngX","value":"-1.55","ts":"..."},...] - * 2. 对象: {"AngX":"-1.55","T":"21.6",...} - */ + private PlatformPushLog createLogEntry(PlatformInstance platform, PlatformDeviceMapping mapping, + String ifaceCode, String serialNumber) { + PlatformPushLog logEntry = new PlatformPushLog(); + logEntry.setPlatformId(platform.getId()); + logEntry.setDeviceId(mapping.getDeviceId()); + logEntry.setSerialNumber(StringUtils.hasText(serialNumber) ? serialNumber : mapping.getSerialNumber()); + logEntry.setInterfaceCode(ifaceCode); + logEntry.setStatus(0); + logEntry.setRetryCount(0); + logEntry.setLogType(1); + return logEntry; + } + + // ==================== 数据解析 ==================== + private JSONObject parsePayloadToMap(String payload) { String trimmed = payload.trim(); if (trimmed.startsWith("[")) { @@ -179,35 +846,121 @@ public class DcpPlatformAdapter { for (int i = 0; i < arr.size(); i++) { JSONObject item = arr.getJSONObject(i); String id = item.getString("id"); - if (id == null) id = item.getString("identifier"); - if (id == null) continue; - Object value = item.get("value"); - map.put(id, value); - // 同时保存 ts 供时间转换使用 + if (id == null) { + id = item.getString("identifier"); + } + if (id == null) { + continue; + } + map.put(id, item.get("value")); if (item.containsKey("ts")) { map.put(id + "__ts", item.get("ts")); } } return map; - } else { - return JSON.parseObject(trimmed); } + return JSON.parseObject(trimmed); } - /** 确保 Token 有效，过期则执行 flow_type=3 的 Token 刷新流程 */ + private Object resolveEpoch(JSONObject reportData, String identifier) { + if (!StringUtils.hasText(identifier)) { + return nowIsoTimestamp(); + } + Object ts = reportData.get(identifier + "__ts"); + return normalizeTimestamp(ts); + } + + /** + * DCP 文档要求时间字段为 ISO-8601 字符串；缺省值也保持同样格式，避免序列化歧义。 + */ + private String nowIsoTimestamp() { + return formatInstant(Instant.now()); + } + + /** + * 设备上报时间可能是毫秒时间戳、秒时间戳、ISO 字符串或本地时间字符串，统一转成带时区的 ISO-8601。 + */ + private String normalizeTimestamp(Object rawTs) { + if (rawTs == null) { + return nowIsoTimestamp(); + } + if (rawTs instanceof Number) { + return formatEpochNumber(((Number) rawTs).longValue()); + } + + String text = normalizeText(String.valueOf(rawTs)); + if (!StringUtils.hasText(text)) { + return nowIsoTimestamp(); + } + if (text.matches("^-?\\d+$")) { + try { + return formatEpochNumber(Long.parseLong(text)); + } catch (NumberFormatException ignore) { + return nowIsoTimestamp(); + } + } + try { + return formatInstant(Instant.parse(text)); + } catch (DateTimeParseException ignore) { + // ignore + } + try { + OffsetDateTime dateTime = OffsetDateTime.parse(text); + return dateTime.format(ISO_OFFSET_MILLIS); + } catch (DateTimeParseException ignore) { + // ignore + } + try { + LocalDateTime localDateTime = LocalDateTime.parse(text, DATE_TIME_MILLIS); + return localDateTime.atZone(ZoneId.systemDefault()).toOffsetDateTime().format(ISO_OFFSET_MILLIS); + } catch (DateTimeParseException ignore) { + // ignore + } + try { + LocalDateTime localDateTime = LocalDateTime.parse(text, DATE_TIME_SECONDS); + return localDateTime.atZone(ZoneId.systemDefault()).toOffsetDateTime().format(ISO_OFFSET_MILLIS); + } catch (DateTimeParseException ignore) { + // ignore + } + return nowIsoTimestamp(); + } + + private String formatEpochNumber(long epochValue) { + long epochMillis = Math.abs(epochValue) < 100000000000L ? epochValue * 1000L : epochValue; + return formatInstant(Instant.ofEpochMilli(epochMillis)); + } + + private String formatInstant(Instant instant) { + return instant.atZone(ZoneId.systemDefault()).toOffsetDateTime().format(ISO_OFFSET_MILLIS); + } + + private String normalizeText(String value) { + return value == null ? null : value.trim(); + } + + private boolean sameSerialNumber(String left, String right) { + String normalizedLeft = normalizeSerialNumber(left); + String normalizedRight = normalizeSerialNumber(right); + return StringUtils.hasText(normalizedLeft) && normalizedLeft.equals(normalizedRight); + } + + private String normalizeSerialNumber(String value) { + String text = normalizeText(value); + return StringUtils.hasText(text) ? text.toUpperCase() : null; + } + + // ==================== Token 管理 ==================== + private String ensureToken(PlatformInstance platform) { Date now = new Date(); boolean expired = platform.getTokenExpiresAt() == null || platform.getTokenExpiresAt().getTime() - now.getTime() < 5 * 60 * 1000L; if (!StringUtils.hasText(platform.getAccessToken()) || expired) { - // 重新加载最新 platform（可能其他线程已刷新） PlatformInstance fresh = instanceService.getById(platform.getId()); boolean stillExpired = fresh.getTokenExpiresAt() == null || fresh.getTokenExpiresAt().getTime() - now.getTime() < 5 * 60 * 1000L; if (!StringUtils.hasText(fresh.getAccessToken()) || stillExpired) { - // 执行 flow_type=3 的 Token 刷新流程（不是初始化流程） - PlatformFlowMapper flowMapper = - SpringUtils.getBean(PlatformFlowMapper.class); + PlatformFlowMapper flowMapper = SpringUtils.getBean(PlatformFlowMapper.class); PlatformFlow tokenFlow = flowMapper.selectOne( Wrappers.lambdaQuery() .eq(PlatformFlow::getPlatformId, fresh.getId()) diff --git a/fastbee-service/fastbee-iot-service/src/main/java/com/fastbee/iot/platform/adapter/dcp/StaticLevelBufferService.java b/fastbee-service/fastbee-iot-service/src/main/java/com/fastbee/iot/platform/adapter/dcp/StaticLevelBufferService.java new file mode 100644 index 00000000..4a236278 --- /dev/null +++ b/fastbee-service/fastbee-iot-service/src/main/java/com/fastbee/iot/platform/adapter/dcp/StaticLevelBufferService.java @@ -0,0 +1,95 @@ +package com.fastbee.iot.platform.adapter.dcp; + +import com.alibaba.fastjson2.JSON; +import com.alibaba.fastjson2.JSONObject; +import lombok.extern.slf4j.Slf4j; +import org.springframework.beans.factory.annotation.Autowired; +import org.springframework.data.redis.core.StringRedisTemplate; +import org.springframework.stereotype.Component; + +import java.time.Instant; +import java.util.concurrent.TimeUnit; + +/** + * 数组字段推送缓冲服务 + */ +@Slf4j +@Component +public class StaticLevelBufferService { + + private static final String REF_PREFIX = "static_level:ref:"; + private static final String MONITOR_PREFIX = "static_level:monitor:"; + + @Autowired + private StringRedisTemplate redisTemplate; + + /** + * 保存基准点最新 item 数据 + */ + public void saveReferenceData(long platformId, String fieldName, String refSerialNumber, + JSONObject itemData, int reportIntervalSec) { + String key = refKey(platformId, fieldName, refSerialNumber); + JSONObject data = new JSONObject(); + data.put("itemData", itemData); + data.put("ts", Instant.now().toEpochMilli()); + redisTemplate.opsForValue().set(key, data.toJSONString(), reportIntervalSec * 2L, TimeUnit.SECONDS); + log.debug("[StaticLevelBuffer] 保存基准点缓冲 key={}", key); + } + + /** + * 保存监测点最新 item 数据。用于监测点先到、基准点后到的补推场景。 + */ + public void saveMonitorData(long platformId, String fieldName, String serialNumber, + JSONObject itemData, JSONObject bodyFields, int reportIntervalSec) { + String key = monitorKey(platformId, fieldName, serialNumber); + JSONObject data = new JSONObject(); + data.put("itemData", itemData); + data.put("bodyFields", bodyFields); + data.put("ts", Instant.now().toEpochMilli()); + redisTemplate.opsForValue().set(key, data.toJSONString(), reportIntervalSec * 2L, TimeUnit.SECONDS); + log.debug("[StaticLevelBuffer] 保存监测点缓冲 key={}", key); + } + + /** + * 获取基准点最新 item 数据，如果超过阈值则返回 null + */ + public JSONObject getReferenceDataIfFresh(long platformId, String fieldName, String refSerialNumber, + int reportIntervalSec) { + return getDataIfFresh(refKey(platformId, fieldName, refSerialNumber), reportIntervalSec, "基准点"); + } + + /** + * 获取监测点最新 item 数据，如果超过阈值则返回 null。 + */ + public JSONObject getMonitorDataIfFresh(long platformId, String fieldName, String serialNumber, + int reportIntervalSec) { + return getDataIfFresh(monitorKey(platformId, fieldName, serialNumber), reportIntervalSec, "监测点"); + } + + public void removeMonitorData(long platformId, String fieldName, String serialNumber) { + redisTemplate.delete(monitorKey(platformId, fieldName, serialNumber)); + } + + private JSONObject getDataIfFresh(String key, int reportIntervalSec, String dataName) { + String val = redisTemplate.opsForValue().get(key); + if (val == null) { + return null; + } + JSONObject data = JSON.parseObject(val); + long ts = data.getLongValue("ts", 0L); + long maxStaleMs = (long) (reportIntervalSec * 1.5 * 1000); + if (Instant.now().toEpochMilli() - ts > maxStaleMs) { + log.debug("[StaticLevelBuffer] {}数据已过期 key={}", dataName, key); + return null; + } + return data; + } + + private String refKey(long platformId, String fieldName, String serialNumber) { + return REF_PREFIX + platformId + ":" + fieldName + ":" + serialNumber; + } + + private String monitorKey(long platformId, String fieldName, String serialNumber) { + return MONITOR_PREFIX + platformId + ":" + fieldName + ":" + serialNumber; + } +} diff --git a/fastbee-service/fastbee-iot-service/src/main/java/com/fastbee/iot/platform/domain/PlatformDeviceMapping.java b/fastbee-service/fastbee-iot-service/src/main/java/com/fastbee/iot/platform/domain/PlatformDeviceMapping.java index f57f5b5c..2b60f6d8 100644 --- a/fastbee-service/fastbee-iot-service/src/main/java/com/fastbee/iot/platform/domain/PlatformDeviceMapping.java +++ b/fastbee-service/fastbee-iot-service/src/main/java/com/fastbee/iot/platform/domain/PlatformDeviceMapping.java @@ -48,6 +48,9 @@ public class PlatformDeviceMapping implements Serializable { @ApiModelProperty("字段映射 [{identifier,platformField,transform}]") private String fieldMapping; + @ApiModelProperty("扩展映射配置，如静力水准子设备编号与DCP sensorId映射") + private String extraMapping; + @ApiModelProperty("入网状态(0=未入网,1=已入网)") private Integer netStatus; diff --git a/fastbee-service/fastbee-iot-service/src/main/java/com/fastbee/iot/platform/flow/PlatformFlowExecutor.java b/fastbee-service/fastbee-iot-service/src/main/java/com/fastbee/iot/platform/flow/PlatformFlowExecutor.java index d5fa48a1..e50f377c 100644 --- a/fastbee-service/fastbee-iot-service/src/main/java/com/fastbee/iot/platform/flow/PlatformFlowExecutor.java +++ b/fastbee-service/fastbee-iot-service/src/main/java/com/fastbee/iot/platform/flow/PlatformFlowExecutor.java @@ -5,9 +5,13 @@ import com.alibaba.fastjson2.JSONArray; import com.alibaba.fastjson2.JSONObject; import com.baomidou.mybatisplus.core.toolkit.Wrappers; import com.dtflys.forest.http.ForestRequest; +import com.fastbee.common.core.mq.SubDeviceBo; import com.fastbee.common.utils.spring.SpringUtils; import com.fastbee.http.model.HttpClientConfig; import com.fastbee.http.service.HttpClientFactory; +import com.fastbee.iot.domain.Device; +import com.fastbee.iot.mapper.DeviceMapper; +import com.fastbee.iot.mapper.SubGatewayMapper; import com.fastbee.iot.platform.domain.*; import com.fastbee.iot.platform.mapper.PlatformDeviceMappingMapper; import com.fastbee.iot.platform.mapper.PlatformFlowStepMapper; @@ -40,6 +44,10 @@ public class PlatformFlowExecutor { private PlatformInstanceMapper instanceMapper; @Autowired private IPlatformPushLogService pushLogService; + @Autowired + private SubGatewayMapper subGatewayMapper; + @Autowired + private DeviceMapper deviceMapper; /** * 执行流程 @@ -62,6 +70,39 @@ public class PlatformFlowExecutor { private void executeStep(PlatformFlow flow, PlatformFlowStep step, PlatformInterface iface, FlowContext context) { long startMs = System.currentTimeMillis(); + PlatformPushLog logEntry = createStepLog(flow, step, iface); + + try { + Map baseParams = resolveParams(iface, step, context); + List> paramSets = expandBatchParams(iface, step, baseParams); + Object responseData = executeParamSets(flow, step, iface, context, paramSets, logEntry, startMs); + + // 7. 将出参写入上下文： + // 优先按 response_schema.path 提取，保证前端新增接口定义后运行链路可以直接消费这些配置。 + if (StringUtils.hasText(step.getResultKey()) && responseData != null) { + context.putStepResult(step.getResultKey(), responseData); + } + + // 8. 持久化指定出参字段 + if (StringUtils.hasText(step.getResultSave())) { + persistResultSave(step, responseData, context); + } + + } catch (Exception e) { + if (logEntry.getStatus() == null || logEntry.getStatus() == 0) { + log.error("[PlatformFlow] 步骤执行失败 flowId={} step={} interface={}", + flow.getId(), step.getStepOrder(), iface.getCode(), e); + logEntry.setStatus(2); + logEntry.setErrorMsg(e.getMessage()); + logEntry.setCostMs((int) (System.currentTimeMillis() - startMs)); + } + throw new RuntimeException("流程步骤执行失败: " + iface.getName() + " - " + e.getMessage(), e); + } finally { + pushLogService.asyncSave(logEntry); + } + } + + private PlatformPushLog createStepLog(PlatformFlow flow, PlatformFlowStep step, PlatformInterface iface) { PlatformPushLog logEntry = new PlatformPushLog(); logEntry.setPlatformId(flow.getPlatformId()); logEntry.setFlowId(flow.getId()); @@ -70,112 +111,116 @@ public class PlatformFlowExecutor { logEntry.setStatus(0); logEntry.setRetryCount(0); logEntry.setLogType(2); // 操作日志（认证/入网等流程步骤） + return logEntry; + } - try { - // 1. 解析入参 - Map params = resolveParams(iface, step, context); - - // 2. 解析 URL（替换路径变量） - String url = resolvePath(context.getPlatform().getBaseUrl() + iface.getPath(), params); - logEntry.setRequestUrl(url); - - // 3. 构建请求 - String method = iface.getMethod().toUpperCase(); - String contentType = StringUtils.hasText(iface.getContentType()) - ? iface.getContentType() : "application/json"; - boolean isFormUrlEncoded = contentType.contains("x-www-form-urlencoded"); - - HttpClientConfig.HttpClientConfigBuilder builder = HttpClientConfig.builder() - .url(url) - .method(ForestRequestType.valueOf(method)); - - Map headers = new HashMap<>(); - headers.put("Content-Type", contentType); - - if ("GET".equals(method)) { - // GET：移除路径变量后剩余参数作为 query string - removePathVars(params, iface.getPath()); - if (!params.isEmpty()) { - builder.querys(params); - } - logEntry.setRequestBody(JSON.toJSONString(params)); - } else if (isFormUrlEncoded) { - // form-urlencoded：移除路径变量后，拼接为 key=value&key=value - removePathVars(params, iface.getPath()); - String formBody = params.entrySet().stream() - .filter(e -> e.getValue() != null) - .map(e -> { - try { - return URLEncoder.encode(e.getKey(), "UTF-8") + "=" - + URLEncoder.encode(e.getValue().toString(), "UTF-8"); - } catch (Exception ex) { - return e.getKey() + "=" + e.getValue(); - } - }) - .collect(Collectors.joining("&")); - builder.body(formBody); - logEntry.setRequestBody(formBody); - } else { - // application/json - removePathVars(params, iface.getPath()); - String body = JSON.toJSONString(params); - builder.body(body); - logEntry.setRequestBody(body); - } - - // 4. 注入 Token - String token = resolveToken(flow, step, context); - if (StringUtils.hasText(token)) { - headers.put("Authorization", "Bearer " + token); - } - builder.headers(headers); - - // 5. 校验路径变量是否全部替换（Forest 会把 {xxx} 当模板变量解析，必须提前替换完） - if (url.contains("{") || url.contains("}")) { - throw new RuntimeException("URL路径变量未解析: " + url + "，请检查步骤参数配置"); - } - - // 6. 执行请求 - ForestRequest request = HttpClientFactory.instance(builder.build()); - String responseBody = request.executeAsString(); - int responseCode = request.getResponseEncode() != null ? Integer.parseInt(request.getResponseEncode()) : 200; - - logEntry.setResponseBody(responseBody); - logEntry.setResponseCode(responseCode); - logEntry.setStatus(1); - logEntry.setCostMs((int) (System.currentTimeMillis() - startMs)); - - // 7. 将出参写入上下文（数组响应取第一个元素） - if (StringUtils.hasText(step.getResultKey()) && StringUtils.hasText(responseBody)) { - String trimmed = responseBody.trim(); - if (trimmed.startsWith("[")) { - JSONArray arr = JSON.parseArray(trimmed); - context.putStepResult(step.getResultKey(), - (arr != null && !arr.isEmpty()) ? arr.getJSONObject(0) : null); - } else { - try { - context.putStepResult(step.getResultKey(), JSON.parseObject(trimmed)); - } catch (Exception e) { - context.putStepResult(step.getResultKey(), trimmed); - } - } - } - - // 8. 持久化指定出参字段 - if (StringUtils.hasText(step.getResultSave())) { - persistResultSave(step, responseBody, context); - } - - } catch (Exception e) { - log.error("[PlatformFlow] 步骤执行失败 flowId={} step={} interface={}", - flow.getId(), step.getStepOrder(), iface.getCode(), e); - logEntry.setStatus(2); - logEntry.setErrorMsg(e.getMessage()); - logEntry.setCostMs((int) (System.currentTimeMillis() - startMs)); - throw new RuntimeException("流程步骤执行失败: " + iface.getName() + " - " + e.getMessage(), e); - } finally { - pushLogService.asyncSave(logEntry); + private Object executeParamSets(PlatformFlow flow, PlatformFlowStep step, PlatformInterface iface, + FlowContext context, List> paramSets, + PlatformPushLog firstLog, long stepStartMs) { + if (paramSets == null || paramSets.isEmpty()) { + return null; } + JSONArray batchResponses = new JSONArray(); + Object singleResponse = null; + boolean batchMode = paramSets.size() > 1; + + for (int i = 0; i < paramSets.size(); i++) { + PlatformPushLog currentLog = i == 0 ? firstLog : createStepLog(flow, step, iface); + long requestStartMs = System.currentTimeMillis(); + try { + Object responseData = executeHttpOnce(flow, step, iface, context, + new LinkedHashMap<>(paramSets.get(i)), currentLog, requestStartMs); + if (batchMode) { + batchResponses.add(responseData); + } else { + singleResponse = responseData; + } + } catch (Exception e) { + log.error("[PlatformFlow] 步骤执行失败 flowId={} step={} interface={} batchIndex={}", + flow.getId(), step.getStepOrder(), iface.getCode(), i + 1, e); + currentLog.setStatus(2); + currentLog.setErrorMsg(e.getMessage()); + currentLog.setCostMs((int) (System.currentTimeMillis() - requestStartMs)); + throw new RuntimeException(e.getMessage(), e); + } finally { + if (i > 0) { + pushLogService.asyncSave(currentLog); + } + } + } + if (batchMode) { + firstLog.setCostMs((int) (System.currentTimeMillis() - stepStartMs)); + return batchResponses; + } + return singleResponse; + } + + private Object executeHttpOnce(PlatformFlow flow, PlatformFlowStep step, PlatformInterface iface, + FlowContext context, Map params, + PlatformPushLog logEntry, long startMs) { + String url = resolvePath(context.getPlatform().getBaseUrl() + iface.getPath(), params); + logEntry.setRequestUrl(url); + + String method = iface.getMethod().toUpperCase(); + String contentType = StringUtils.hasText(iface.getContentType()) + ? iface.getContentType() : "application/json"; + boolean isFormUrlEncoded = contentType.contains("x-www-form-urlencoded"); + + HttpClientConfig.HttpClientConfigBuilder builder = HttpClientConfig.builder() + .url(url) + .method(ForestRequestType.valueOf(method)); + + Map headers = new HashMap<>(); + headers.put("Content-Type", contentType); + + if ("GET".equals(method)) { + removePathVars(params, iface.getPath()); + if (!params.isEmpty()) { + builder.querys(params); + } + logEntry.setRequestBody(JSON.toJSONString(params)); + } else if (isFormUrlEncoded) { + removePathVars(params, iface.getPath()); + String formBody = params.entrySet().stream() + .filter(e -> e.getValue() != null) + .map(e -> { + try { + return URLEncoder.encode(e.getKey(), "UTF-8") + "=" + + URLEncoder.encode(e.getValue().toString(), "UTF-8"); + } catch (Exception ex) { + return e.getKey() + "=" + e.getValue(); + } + }) + .collect(Collectors.joining("&")); + builder.body(formBody); + logEntry.setRequestBody(formBody); + } else { + removePathVars(params, iface.getPath()); + String body = JSON.toJSONString(params); + builder.body(body); + logEntry.setRequestBody(body); + } + + String token = resolveToken(flow, step, context); + if (StringUtils.hasText(token)) { + headers.put("Authorization", "Bearer " + token); + } + builder.headers(headers); + + if (url.contains("{") || url.contains("}")) { + throw new RuntimeException("URL路径变量未解析: " + url + "，请检查步骤参数配置"); + } + + ForestRequest request = HttpClientFactory.instance(builder.build()); + String responseBody = request.executeAsString(); + int responseCode = request.getResponseEncode() != null ? Integer.parseInt(request.getResponseEncode()) : 200; + Object responseData = buildResponseData(iface, responseBody); + + logEntry.setResponseBody(responseBody); + logEntry.setResponseCode(responseCode); + logEntry.setStatus(1); + logEntry.setCostMs((int) (System.currentTimeMillis() - startMs)); + return responseData; } /** @@ -231,6 +276,8 @@ public class PlatformFlowExecutor { switch (source) { case "fixed": return fd.get("value"); + case "uuid": + return newGeneratedId(); case "auth": return authConfig.get(fd.getString("field")); case "extra": @@ -244,11 +291,309 @@ public class PlatformFlowExecutor { if ("platform_point_id".equals(df)) return ctx.getDeviceMapping().getPlatformPointId(); if ("serial_number".equals(df)) return ctx.getDeviceMapping().getSerialNumber(); return null; + case "subDeviceSensorIds": + return collectSubDeviceSensorIds(ctx.getDeviceMapping()); + case "referenceSubDeviceSensorIdPairs": + return collectReferenceSubDeviceSensorIdPairs(ctx.getDeviceMapping()); default: return null; } } + private String newGeneratedId() { + return UUID.randomUUID().toString(); + } + + private List> expandBatchParams(PlatformInterface iface, PlatformFlowStep step, + Map baseParams) { + Map.Entry batchEntry = findBatchParam(baseParams); + List> paramSets = new ArrayList<>(); + if (batchEntry == null) { + Map params = new LinkedHashMap<>(baseParams); + refreshGeneratedParams(iface, step, params); + paramSets.add(params); + return paramSets; + } + + List batches = toList(batchEntry.getValue()); + for (Object batchItem : batches) { + Map params = new LinkedHashMap<>(baseParams); + params.put(batchEntry.getKey(), normalizeBatchValue(batchItem)); + refreshGeneratedParams(iface, step, params); + paramSets.add(params); + } + return paramSets; + } + + private Map.Entry findBatchParam(Map params) { + for (Map.Entry entry : params.entrySet()) { + if (isListOfLists(entry.getValue())) { + return entry; + } + } + return null; + } + + private boolean isListOfLists(Object value) { + List list = toList(value); + if (list == null || list.isEmpty()) { + return false; + } + Object first = list.get(0); + return toList(first) != null; + } + + private List toList(Object value) { + if (value instanceof List) { + return (List) value; + } + if (value instanceof JSONArray) { + return (JSONArray) value; + } + return null; + } + + private Object normalizeBatchValue(Object value) { + List list = toList(value); + if (list == null) { + return value; + } + return new ArrayList<>(list); + } + + private void refreshGeneratedParams(PlatformInterface iface, PlatformFlowStep step, Map params) { + for (String fieldName : resolveGeneratedFieldNames(iface, step)) { + params.put(fieldName, newGeneratedId()); + } + } + + private Set resolveGeneratedFieldNames(PlatformInterface iface, PlatformFlowStep step) { + LinkedHashMap sources = new LinkedHashMap<>(); + try { + if (iface != null && StringUtils.hasText(iface.getRequestSchema())) { + JSONArray schema = JSON.parseArray(iface.getRequestSchema()); + for (int i = 0; i < schema.size(); i++) { + JSONObject field = schema.getJSONObject(i); + if (field != null && StringUtils.hasText(field.getString("name"))) { + sources.put(field.getString("name"), field.getString("source")); + } + } + } + if (step != null && StringUtils.hasText(step.getParamMapping())) { + JSONObject overrides = JSON.parseObject(step.getParamMapping()); + for (String fieldName : overrides.keySet()) { + JSONObject override = overrides.getJSONObject(fieldName); + if (override != null && StringUtils.hasText(override.getString("source"))) { + sources.put(fieldName, override.getString("source")); + } + } + } + } catch (Exception e) { + log.warn("[PlatformFlow] 解析自动生成参数失败 interfaceId={}, error={}", + iface == null ? null : iface.getId(), e.getMessage()); + } + return sources.entrySet().stream() + .filter(entry -> "uuid".equals(entry.getValue())) + .map(Map.Entry::getKey) + .collect(Collectors.toCollection(LinkedHashSet::new)); + } + + private List collectSubDeviceSensorIds(PlatformDeviceMapping mapping) { + return collectSubDeviceSensorBindings(mapping).stream() + .map(item -> item.sensorId) + .filter(StringUtils::hasText) + .distinct() + .collect(Collectors.toList()); + } + + /** + * 静力水准类成组设备按“基准点 + 监测点”执行绑定。 + * 返回二维数组后由通用批量参数展开逻辑拆成多次 HTTP 请求。 + */ + private List> collectReferenceSubDeviceSensorIdPairs(PlatformDeviceMapping mapping) { + List bindings = collectSubDeviceSensorBindings(mapping); + if (bindings.size() < 2) { + return Collections.emptyList(); + } + SubDeviceSensorBinding reference = bindings.stream() + .filter(item -> item.reference) + .findFirst() + .orElse(bindings.get(0)); + List> pairs = new ArrayList<>(); + for (SubDeviceSensorBinding binding : bindings) { + if (sameText(reference.serialNumber, binding.serialNumber) + || !StringUtils.hasText(reference.sensorId) + || !StringUtils.hasText(binding.sensorId)) { + continue; + } + pairs.add(Arrays.asList(reference.sensorId, binding.sensorId)); + } + return pairs; + } + + private List collectSubDeviceSensorBindings(PlatformDeviceMapping mapping) { + if (mapping == null) { + return Collections.emptyList(); + } + LinkedHashMap bindings = new LinkedHashMap<>(); + String referenceSerialNumber = appendConfiguredSubDeviceBindings(bindings, mapping); + appendGatewaySubDeviceBindings(bindings, mapping); + + if (!StringUtils.hasText(referenceSerialNumber) && !bindings.isEmpty()) { + referenceSerialNumber = bindings.keySet().iterator().next(); + } + if (StringUtils.hasText(referenceSerialNumber) && bindings.containsKey(referenceSerialNumber)) { + bindings.get(referenceSerialNumber).reference = true; + } + return new ArrayList<>(bindings.values()); + } + + private String appendConfiguredSubDeviceBindings(LinkedHashMap bindings, + PlatformDeviceMapping mapping) { + if (!StringUtils.hasText(mapping.getExtraMapping())) { + return null; + } + try { + JSONObject extra = JSON.parseObject(mapping.getExtraMapping()); + String referenceSerialNumber = null; + + JSONArray arrayFields = extra.getJSONArray("arrayFields"); + if (arrayFields != null) { + for (int i = 0; i < arrayFields.size(); i++) { + JSONObject arrayField = arrayFields.getJSONObject(i); + referenceSerialNumber = appendConfiguredMappingRows(bindings, + arrayField == null ? null : arrayField.getJSONArray("subDeviceMappings"), + referenceSerialNumber); + } + } + + JSONObject staticLevel = extra.getJSONObject("staticLevel"); + if (staticLevel != null) { + referenceSerialNumber = normalizeText(staticLevel.getString("referenceSerialNumber")); + referenceSerialNumber = appendConfiguredMappingRows( + bindings, staticLevel.getJSONArray("sensorIdMappings"), referenceSerialNumber); + } + return referenceSerialNumber; + } catch (Exception e) { + log.warn("[PlatformFlow] 解析子设备传感器列表失败 mappingId={}, error={}", + mapping.getId(), e.getMessage()); + return null; + } + } + + private String appendConfiguredMappingRows(LinkedHashMap bindings, + JSONArray mappings, String referenceSerialNumber) { + if (mappings == null || mappings.isEmpty()) { + return referenceSerialNumber; + } + String ref = referenceSerialNumber; + for (int i = 0; i < mappings.size(); i++) { + JSONObject item = mappings.getJSONObject(i); + if (item == null) { + continue; + } + String serialNumber = normalizeText(item.getString("serialNumber")); + if (!StringUtils.hasText(serialNumber)) { + continue; + } + String sensorId = normalizeText(item.getString("sensorId")); + upsertSubDeviceBinding(bindings, serialNumber, + StringUtils.hasText(sensorId) ? sensorId : serialNumber); + if (Boolean.TRUE.equals(item.getBoolean("isReference"))) { + ref = serialNumber; + } + } + return ref; + } + + private void appendGatewaySubDeviceBindings(LinkedHashMap bindings, + PlatformDeviceMapping mapping) { + String gatewaySerialNumber = resolveGatewaySerialNumber(mapping); + if (!StringUtils.hasText(gatewaySerialNumber)) { + return; + } + + try { + List relationRows = subGatewayMapper.getSubDeviceListByGw(gatewaySerialNumber); + if (relationRows != null) { + for (SubDeviceBo subDevice : relationRows) { + appendGatewaySubDeviceBinding(bindings, subDevice == null ? null : subDevice.getSubDeviceNo()); + } + } + } catch (Exception e) { + log.warn("[PlatformFlow] 查询网关子设备关系失败 gateway={}, error={}", + gatewaySerialNumber, e.getMessage()); + } + + try { + Device query = new Device(); + query.setGwDevCode(gatewaySerialNumber); + List deviceRows = deviceMapper.selectDeviceList(query); + if (deviceRows != null) { + for (Device device : deviceRows) { + appendGatewaySubDeviceBinding(bindings, device == null ? null : device.getSerialNumber()); + } + } + } catch (Exception e) { + log.warn("[PlatformFlow] 查询网关子设备列表失败 gateway={}, error={}", + gatewaySerialNumber, e.getMessage()); + } + } + + private void appendGatewaySubDeviceBinding(LinkedHashMap bindings, + String serialNumber) { + String normalizedSerialNumber = normalizeText(serialNumber); + if (!StringUtils.hasText(normalizedSerialNumber)) { + return; + } + upsertSubDeviceBinding(bindings, normalizedSerialNumber, normalizedSerialNumber); + } + + private void upsertSubDeviceBinding(LinkedHashMap bindings, + String serialNumber, String sensorId) { + SubDeviceSensorBinding binding = bindings.computeIfAbsent(serialNumber, SubDeviceSensorBinding::new); + if (StringUtils.hasText(sensorId)) { + binding.sensorId = sensorId; + } + } + + private String resolveGatewaySerialNumber(PlatformDeviceMapping mapping) { + String serialNumber = normalizeText(mapping.getSerialNumber()); + if (StringUtils.hasText(serialNumber)) { + return serialNumber; + } + if (mapping.getDeviceId() == null) { + return null; + } + try { + Device device = deviceMapper.selectDeviceByDeviceId(mapping.getDeviceId()); + return device == null ? null : normalizeText(device.getSerialNumber()); + } catch (Exception e) { + log.warn("[PlatformFlow] 查询网关设备编号失败 deviceId={}, error={}", + mapping.getDeviceId(), e.getMessage()); + return null; + } + } + + private boolean sameText(String left, String right) { + return Objects.equals(normalizeText(left), normalizeText(right)); + } + + private String normalizeText(String value) { + return value == null ? null : value.trim(); + } + + private static class SubDeviceSensorBinding { + private final String serialNumber; + private String sensorId; + private boolean reference; + + private SubDeviceSensorBinding(String serialNumber) { + this.serialNumber = serialNumber; + this.sensorId = serialNumber; + } + } + /** 替换 URL 路径变量，如 /project/by-user-id/{userId} */ private String resolvePath(String url, Map params) { for (Map.Entry entry : params.entrySet()) { @@ -270,20 +615,170 @@ public class PlatformFlowExecutor { } } + private Object buildResponseData(PlatformInterface iface, String responseBody) { + if (!StringUtils.hasText(responseBody)) { + return null; + } + Object parsed = parseResponseBody(responseBody); + if (parsed == null) { + return null; + } + JSONObject extracted = extractResponseBySchema(iface, parsed); + if (!extracted.isEmpty()) { + return extracted; + } + if (parsed instanceof JSONArray) { + JSONArray arr = (JSONArray) parsed; + return (arr != null && !arr.isEmpty()) ? arr.get(0) : null; + } + return parsed; + } + + private Object parseResponseBody(String responseBody) { + String trimmed = responseBody == null ? null : responseBody.trim(); + if (!StringUtils.hasText(trimmed)) { + return null; + } + try { + if (trimmed.startsWith("[")) { + return JSON.parseArray(trimmed); + } + if (trimmed.startsWith("{")) { + return JSON.parseObject(trimmed); + } + } catch (Exception e) { + log.warn("[PlatformFlow] 解析响应JSON失败: {}", e.getMessage()); + } + return trimmed; + } + + private JSONObject extractResponseBySchema(PlatformInterface iface, Object parsedResponse) { + JSONObject extracted = new JSONObject(); + if (iface == null || !StringUtils.hasText(iface.getResponseSchema()) || parsedResponse == null) { + return extracted; + } + try { + JSONArray schema = JSON.parseArray(iface.getResponseSchema()); + for (int i = 0; i < schema.size(); i++) { + JSONObject field = schema.getJSONObject(i); + if (field == null || !StringUtils.hasText(field.getString("name"))) { + continue; + } + Object value = readPathValue(parsedResponse, field.getString("path")); + if (value != null) { + extracted.put(field.getString("name"), value); + } + } + } catch (Exception e) { + log.warn("[PlatformFlow] 解析 response_schema 失败, interfaceId={}, error={}", + iface.getId(), e.getMessage()); + } + return extracted; + } + + private Object readPathValue(Object root, String path) { + if (root == null || !StringUtils.hasText(path)) { + return null; + } + Object current = root; + for (String segment : splitPath(path)) { + if (current == null) { + return null; + } + current = readSegmentValue(current, segment); + } + return current; + } + + private List splitPath(String path) { + String normalized = path.trim(); + if (normalized.startsWith("$.")) { + normalized = normalized.substring(2); + } else if ("$".equals(normalized)) { + normalized = ""; + } else if (normalized.startsWith("$")) { + normalized = normalized.substring(1); + } + List segments = new ArrayList<>(); + StringBuilder current = new StringBuilder(); + int bracketDepth = 0; + for (int i = 0; i < normalized.length(); i++) { + char ch = normalized.charAt(i); + if (ch == '.' && bracketDepth == 0) { + if (current.length() > 0) { + segments.add(current.toString()); + current.setLength(0); + } + continue; + } + if (ch == '[') { + bracketDepth++; + } else if (ch == ']') { + bracketDepth = Math.max(0, bracketDepth - 1); + } + current.append(ch); + } + if (current.length() > 0) { + segments.add(current.toString()); + } + return segments; + } + + private Object readSegmentValue(Object current, String segment) { + if (!StringUtils.hasText(segment)) { + return current; + } + int bracketIndex = segment.indexOf('['); + String fieldName = bracketIndex >= 0 ? segment.substring(0, bracketIndex) : segment; + Object value = current; + if (StringUtils.hasText(fieldName)) { + if (value instanceof JSONObject) { + value = ((JSONObject) value).get(fieldName); + } else if (value instanceof Map) { + value = ((Map) value).get(fieldName); + } else { + return null; + } + } + int cursor = bracketIndex >= 0 ? bracketIndex : segment.length(); + while (cursor >= 0 && cursor < segment.length()) { + int left = segment.indexOf('[', cursor); + if (left < 0) { + break; + } + int right = segment.indexOf(']', left); + if (right < 0) { + return null; + } + String indexText = segment.substring(left + 1, right).trim(); + if (!StringUtils.hasText(indexText)) { + return null; + } + int index; + try { + index = Integer.parseInt(indexText); + } catch (NumberFormatException e) { + return null; + } + if (value instanceof JSONArray) { + JSONArray arr = (JSONArray) value; + value = (index >= 0 && index < arr.size()) ? arr.get(index) : null; + } else if (value instanceof List) { + List list = (List) value; + value = (index >= 0 && index < list.size()) ? list.get(index) : null; + } else { + return null; + } + cursor = right + 1; + } + return value; + } + /** 将 result_save 中指定字段持久化到 platform_instance.extra_config 或 platform_device_mapping */ - private void persistResultSave(PlatformFlowStep step, String responseBody, FlowContext context) { + private void persistResultSave(PlatformFlowStep step, Object responseData, FlowContext context) { try { JSONArray saves = JSON.parseArray(step.getResultSave()); - // 响应可能是数组（如项目列表），取第一个元素 - JSONObject response; - String trimmed = responseBody.trim(); - if (trimmed.startsWith("[")) { - JSONArray arr = JSON.parseArray(trimmed); - if (arr == null || arr.isEmpty()) return; - response = arr.getJSONObject(0); - } else { - response = JSON.parseObject(trimmed); - } + JSONObject response = toJsonObject(responseData); if (response == null) return; PlatformInstance platform = context.getPlatform(); @@ -360,6 +855,31 @@ public class PlatformFlowExecutor { } } + private JSONObject toJsonObject(Object data) { + if (data == null) { + return null; + } + if (data instanceof JSONObject) { + return (JSONObject) data; + } + if (data instanceof JSONArray) { + JSONArray arr = (JSONArray) data; + if (arr.isEmpty()) { + return null; + } + Object first = arr.get(0); + return toJsonObject(first); + } + if (data instanceof Map) { + return new JSONObject((Map) data); + } + try { + return JSON.parseObject(JSON.toJSONString(data)); + } catch (Exception e) { + return null; + } + } + /** * 解析当前步骤应使用的 Token： * - flow_type=3（Token刷新）：不注入，避免循环 diff --git a/fastbee-service/fastbee-iot-service/src/main/java/com/fastbee/iot/platform/service/impl/PlatformDeviceMappingServiceImpl.java b/fastbee-service/fastbee-iot-service/src/main/java/com/fastbee/iot/platform/service/impl/PlatformDeviceMappingServiceImpl.java index 78013722..24c990be 100644 --- a/fastbee-service/fastbee-iot-service/src/main/java/com/fastbee/iot/platform/service/impl/PlatformDeviceMappingServiceImpl.java +++ b/fastbee-service/fastbee-iot-service/src/main/java/com/fastbee/iot/platform/service/impl/PlatformDeviceMappingServiceImpl.java @@ -39,6 +39,7 @@ public class PlatformDeviceMappingServiceImpl extends ServiceImpl + @@ -23,8 +24,14 @@ diff --git a/sql/platform-dcp-init-data.sql b/sql/platform-dcp-init-data.sql index 2660958b..b26a7db2 100644 --- a/sql/platform-dcp-init-data.sql +++ b/sql/platform-dcp-init-data.sql @@ -73,8 +73,8 @@ VALUES (@dcp_platform_id, '绑定倾角传感器', 'dcp_bind_inclino', 'POST', ' {"name":"pointId", "type":"string","source":"context","contextKey":"step1.id","required":true,"desc":"测点ID，来自创建测点步骤"}, {"name":"deviceType", "type":"number","source":"fixed", "value":5, "required":true,"desc":"设备类型，倾角=5"}, {"name":"name", "type":"string","source":"device", "field":"serial_number","required":true,"desc":"传感器名称"}, - {"name":"moduleId", "type":"string","source":"fixed", "value":"auto", "required":true,"desc":"模块ID，auto表示自动生成UUID"}, - {"name":"virtualId", "type":"string","source":"fixed", "value":"auto", "required":true,"desc":"虚拟ID，auto表示自动生成UUID"}, + {"name":"moduleId", "type":"string","source":"uuid", "required":true,"desc":"模块ID，系统自动生成UUID"}, + {"name":"virtualId", "type":"string","source":"uuid", "required":true,"desc":"虚拟ID，系统自动生成UUID"}, {"name":"dH", "type":"number","source":"fixed", "value":0, "required":false,"desc":"高程"} ]', '[ @@ -91,8 +91,8 @@ VALUES (@dcp_platform_id, '绑定轴力计', 'dcp_bind_stress', 'POST', '/api/ap {"name":"deviceType", "type":"number","source":"fixed", "value":10, "required":true,"desc":"设备类型，轴力计=10"}, {"name":"name", "type":"string","source":"device", "field":"serial_number","required":true,"desc":"传感器名称"}, {"name":"factor", "type":"number","source":"fixed", "value":1, "required":false,"desc":"系数"}, - {"name":"moduleId", "type":"string","source":"fixed", "value":"auto", "required":true}, - {"name":"virtualId", "type":"string","source":"fixed", "value":"auto", "required":true} + {"name":"moduleId", "type":"string","source":"uuid", "required":true}, + {"name":"virtualId", "type":"string","source":"uuid", "required":true} ]', '[{"name":"id","type":"string","path":"id","desc":"DCP设备ID"}]', 6, '将轴力计绑定到测点', 'system', NOW()); @@ -105,8 +105,8 @@ VALUES (@dcp_platform_id, '绑定雨量计', 'dcp_bind_hyeto', 'POST', '/api/app {"name":"pointId", "type":"string","source":"context","contextKey":"step1.id","required":true}, {"name":"deviceType", "type":"number","source":"fixed", "value":3, "required":true,"desc":"雨量计=3"}, {"name":"name", "type":"string","source":"device", "field":"serial_number","required":true}, - {"name":"moduleId", "type":"string","source":"fixed", "value":"auto", "required":true}, - {"name":"virtualId", "type":"string","source":"fixed", "value":"auto", "required":true} + {"name":"moduleId", "type":"string","source":"uuid", "required":true}, + {"name":"virtualId", "type":"string","source":"uuid", "required":true} ]', '[{"name":"id","type":"string","path":"id","desc":"DCP设备ID"}]', 7, '将雨量计绑定到测点', 'system', NOW()); @@ -119,8 +119,9 @@ VALUES (@dcp_platform_id, '绑定静力水准', 'dcp_bind_static_level', 'POST', {"name":"pointId", "type":"string","source":"context","contextKey":"step1.id","required":true}, {"name":"deviceType", "type":"number","source":"fixed", "value":1002, "required":true,"desc":"静力水准=1002"}, {"name":"name", "type":"string","source":"device", "field":"serial_number","required":true}, - {"name":"moduleId", "type":"string","source":"fixed", "value":"auto", "required":true}, - {"name":"virtualId", "type":"string","source":"fixed", "value":"auto", "required":true} + {"name":"moduleId", "type":"string","source":"uuid", "required":true}, + {"name":"virtualId", "type":"string","source":"uuid", "required":true}, + {"name":"sensorIds", "type":"array", "source":"referenceSubDeviceSensorIdPairs","required":true,"desc":"基准点与每个监测点成对绑定"} ]', '[{"name":"id","type":"string","path":"id","desc":"DCP设备ID"}]', 8, '将静力水准传感器绑定到测点', 'system', NOW()); @@ -135,10 +136,10 @@ INSERT INTO `platform_interface` (`platform_id`,`name`,`code`,`method`,`path`,`c VALUES (@dcp_platform_id, '推送倾角数据', 'dcp_push_inclino', 'POST', '/api/app/inclino-data/many', 'application/json', '[ {"name":"deviceId", "type":"string","source":"device","field":"platform_device_id","required":true,"desc":"DCP设备ID"}, - {"name":"epoch", "type":"string","source":"context","contextKey":"reportTime","required":true,"desc":"采集时间，ISO8601格式"}, - {"name":"xDeg", "type":"number","source":"context","contextKey":"xDeg","required":true,"desc":"X轴倾角(度)"}, - {"name":"yDeg", "type":"number","source":"context","contextKey":"yDeg","required":true,"desc":"Y轴倾角(度)"}, - {"name":"temperature","type":"number","source":"context","contextKey":"temperature","required":false,"desc":"温度"} + {"name":"epoch", "type":"string","source":"ts","required":true,"desc":"采集时间，ISO8601格式"}, + {"name":"xDeg", "type":"number","source":"thingsModel","required":true,"desc":"X轴倾角(度)"}, + {"name":"yDeg", "type":"number","source":"thingsModel","required":true,"desc":"Y轴倾角(度)"}, + {"name":"temperature","type":"number","source":"thingsModel","required":false,"desc":"温度"} ]', '[]', 9, '推送倾角传感器实时数据到DCP', 'system', NOW()); @@ -148,8 +149,8 @@ INSERT INTO `platform_interface` (`platform_id`,`name`,`code`,`method`,`path`,`c VALUES (@dcp_platform_id, '推送轴力计数据', 'dcp_push_stress', 'POST', '/api/app/stress-data/many', 'application/json', '[ {"name":"deviceId","type":"string","source":"device","field":"platform_device_id","required":true,"desc":"DCP设备ID"}, - {"name":"epoch", "type":"string","source":"context","contextKey":"reportTime","required":true,"desc":"采集时间，ISO8601格式"}, - {"name":"e", "type":"number","source":"context","contextKey":"e","required":true,"desc":"累计值（必须递增）"} + {"name":"epoch", "type":"string","source":"ts","required":true,"desc":"采集时间，ISO8601格式"}, + {"name":"e", "type":"number","source":"thingsModel","required":true,"desc":"累计值（必须递增）"} ]', '[]', 10, '推送轴力计实时数据到DCP', 'system', NOW()); @@ -159,8 +160,8 @@ INSERT INTO `platform_interface` (`platform_id`,`name`,`code`,`method`,`path`,`c VALUES (@dcp_platform_id, '推送雨量计数据', 'dcp_push_hyeto', 'POST', '/api/app/hyeto-data/many', 'application/json', '[ {"name":"deviceId","type":"string","source":"device","field":"platform_device_id","required":true,"desc":"DCP设备ID"}, - {"name":"epoch", "type":"string","source":"context","contextKey":"reportTime","required":true,"desc":"采集时间，ISO8601格式"}, - {"name":"value", "type":"number","source":"context","contextKey":"value","required":true,"desc":"累计雨量（必须递增）"} + {"name":"epoch", "type":"string","source":"ts","required":true,"desc":"采集时间，ISO8601格式"}, + {"name":"value", "type":"number","source":"thingsModel","required":true,"desc":"累计雨量（必须递增）"} ]', '[]', 11, '推送雨量计实时数据到DCP', 'system', NOW()); @@ -170,17 +171,20 @@ INSERT INTO `platform_interface` (`platform_id`,`name`,`code`,`method`,`path`,`c VALUES (@dcp_platform_id, '推送静力水准数据', 'dcp_push_static_level', 'POST', '/api/app/static-level-data/many', 'application/json', '[ {"name":"deviceId", "type":"string","source":"device","field":"platform_device_id","required":true,"desc":"DCP设备ID"}, - {"name":"epoch", "type":"string","source":"context","contextKey":"reportTime","required":true,"desc":"采集时间，ISO8601格式"}, - {"name":"height", "type":"number","source":"context","contextKey":"height","required":true,"desc":"高程"}, - {"name":"referHeight","type":"number","source":"context","contextKey":"referHeight","required":false,"desc":"参考高程"}, - {"name":"sensorData", "type":"number","source":"context","contextKey":"sensorData","required":false,"desc":"传感器原始数据"} + {"name":"sensorData","type":"array","source":"context","contextKey":"sensorData","required":true,"desc":"静力水准数组","itemSchema":[ + {"name":"epoch","type":"string","source":"ts","desc":"采集时间"}, + {"name":"height","type":"number","source":"thingsModel","desc":"高程"}, + {"name":"temperature","type":"number","source":"thingsModel","desc":"温度"}, + {"name":"sensorId","type":"string","source":"subDeviceId","desc":"DCP侧sensorId"} + ]}, + {"name":"ct","type":"number","source":"fixed","value":1,"required":true,"desc":"采集类型"} ]', '[]', 12, '推送静力水准实时数据到DCP', 'system', NOW()); -- ============================================= --- 二、流程定义（5个） +-- 二、流程定义（6个） -- ============================================= -- 流程1：平台初始化（flow_type=1） @@ -288,3 +292,96 @@ VALUES (@flow_hyeto_id, 2, 'step2', '[{"field":"id","saveTo":"device_mapping.platform_device_id"}]', '绑定雨量计'); + + +-- 流程6：静力水准设备入网（flow_type=2） +INSERT INTO `platform_flow` (`platform_id`,`name`,`code`,`flow_type`,`sensor_type`,`enable`,`remark`,`create_by`,`create_time`) +VALUES (@dcp_platform_id, 'DCP静力水准入网', 'dcp_onboard_static_level', 2, 'static_level', '1', '创建测点→绑定静力水准', 'system', NOW()); + +SET @flow_static_level_id = LAST_INSERT_ID(); + +INSERT INTO `platform_flow_step` (`flow_id`,`step_order`,`interface_id`,`result_key`,`result_save`,`remark`) +VALUES (@flow_static_level_id, 1, + (SELECT id FROM platform_interface WHERE platform_id=@dcp_platform_id AND code='dcp_create_point' LIMIT 1), + 'step1', + '[{"field":"id","saveTo":"device_mapping.platform_point_id"}]', + '创建测点'); + +INSERT INTO `platform_flow_step` (`flow_id`,`step_order`,`interface_id`,`result_key`,`result_save`,`remark`) +VALUES (@flow_static_level_id, 2, + (SELECT id FROM platform_interface WHERE platform_id=@dcp_platform_id AND code='dcp_bind_static_level' LIMIT 1), + 'step2', + '[{"field":"id","saveTo":"device_mapping.platform_device_id"}]', + '绑定静力水准'); + + + + +-- 更新已落库的平台接口配置，去掉 fixed:auto 和静力水准扁平 sensorIds 配置。 +-- 执行前确认平台实例 code，如不是 dcp，请手动替换 @platform_id。 + +SET @platform_id = (SELECT id FROM platform_instance WHERE code = 'dcp' LIMIT 1); + +UPDATE platform_interface +SET request_schema = '[ + {"name":"projectId", "type":"string","source":"extra", "field":"projectId", "required":true,"desc":"项目ID"}, + {"name":"pointId", "type":"string","source":"context","contextKey":"step1.id","required":true,"desc":"测点ID，来自创建测点步骤"}, + {"name":"deviceType", "type":"number","source":"fixed", "value":5, "required":true,"desc":"设备类型，倾角=5"}, + {"name":"name", "type":"string","source":"device", "field":"serial_number","required":true,"desc":"传感器名称"}, + {"name":"moduleId", "type":"string","source":"uuid", "required":true,"desc":"模块ID，系统自动生成UUID"}, + {"name":"virtualId", "type":"string","source":"uuid", "required":true,"desc":"虚拟ID，系统自动生成UUID"}, + {"name":"dH", "type":"number","source":"fixed", "value":0, "required":false,"desc":"高程"} +]', + update_by = 'system', + update_time = NOW() +WHERE platform_id = @platform_id + AND code = 'dcp_bind_inclino'; + +UPDATE platform_interface +SET request_schema = '[ + {"name":"projectId", "type":"string","source":"extra", "field":"projectId", "required":true,"desc":"项目ID"}, + {"name":"pointId", "type":"string","source":"context","contextKey":"step1.id","required":true,"desc":"测点ID"}, + {"name":"deviceType", "type":"number","source":"fixed", "value":10, "required":true,"desc":"设备类型，轴力计=10"}, + {"name":"name", "type":"string","source":"device", "field":"serial_number","required":true,"desc":"传感器名称"}, + {"name":"factor", "type":"number","source":"fixed", "value":1, "required":false,"desc":"系数"}, + {"name":"moduleId", "type":"string","source":"uuid", "required":true}, + {"name":"virtualId", "type":"string","source":"uuid", "required":true} +]', + update_by = 'system', + update_time = NOW() +WHERE platform_id = @platform_id + AND code = 'dcp_bind_stress'; + +UPDATE platform_interface +SET request_schema = '[ + {"name":"projectId", "type":"string","source":"extra", "field":"projectId", "required":true}, + {"name":"pointId", "type":"string","source":"context","contextKey":"step1.id","required":true}, + {"name":"deviceType", "type":"number","source":"fixed", "value":3, "required":true,"desc":"雨量计=3"}, + {"name":"name", "type":"string","source":"device", "field":"serial_number","required":true}, + {"name":"moduleId", "type":"string","source":"uuid", "required":true}, + {"name":"virtualId", "type":"string","source":"uuid", "required":true} +]', + update_by = 'system', + update_time = NOW() +WHERE platform_id = @platform_id + AND code = 'dcp_bind_hyeto'; + +UPDATE platform_interface +SET request_schema = '[ + {"name":"projectId", "type":"string","source":"extra", "field":"projectId", "required":true}, + {"name":"pointId", "type":"string","source":"context","contextKey":"step1.id","required":true}, + {"name":"deviceType", "type":"number","source":"fixed", "value":1002, "required":true,"desc":"静力水准=1002"}, + {"name":"name", "type":"string","source":"device", "field":"serial_number","required":true}, + {"name":"moduleId", "type":"string","source":"uuid", "required":true}, + {"name":"virtualId", "type":"string","source":"uuid", "required":true}, + {"name":"sensorIds", "type":"array", "source":"referenceSubDeviceSensorIdPairs","required":true,"desc":"基准点与每个监测点成对绑定"} +]', + update_by = 'system', + update_time = NOW() +WHERE platform_id = @platform_id + AND code = 'dcp_bind_static_level'; + +SELECT code, request_schema +FROM platform_interface +WHERE platform_id = @platform_id + AND code IN ('dcp_bind_inclino', 'dcp_bind_stress', 'dcp_bind_hyeto', 'dcp_bind_static_level'); diff --git a/sql/platform-integration.sql b/sql/platform-integration.sql index 68a84dca..03879d81 100644 --- a/sql/platform-integration.sql +++ b/sql/platform-integration.sql @@ -98,6 +98,7 @@ CREATE TABLE IF NOT EXISTS `platform_device_mapping` ( `sensor_type` varchar(50) NOT NULL COMMENT '传感器类型: inclino/stress/hyeto/static_level', `push_interface_id` bigint(20) DEFAULT NULL COMMENT '推送接口ID，用于字段映射参照', `field_mapping` json DEFAULT NULL COMMENT '字段映射 [{identifier,platformField,transform}]', + `extra_mapping` json DEFAULT NULL COMMENT '扩展映射配置，如静力水准子设备编号与DCP sensorId映射', `net_status` tinyint(1) NOT NULL DEFAULT 0 COMMENT '入网状态(0=未入网,1=已入网)', `enable` char(1) NOT NULL DEFAULT '1' COMMENT '是否启用', `remark` varchar(255) DEFAULT NULL COMMENT '备注', diff --git a/yack-web/src/views/platform/components/InterfaceDrawer.vue b/yack-web/src/views/platform/components/InterfaceDrawer.vue index 212ed737..efde5678 100644 --- a/yack-web/src/views/platform/components/InterfaceDrawer.vue +++ b/yack-web/src/views/platform/components/InterfaceDrawer.vue @@ -98,8 +98,8 @@ 入参定义（request_schema）
- 定义接口需要的入参字段。source 决定参数值来源： - fixed=固定值、auth=认证配置、extra=平台扩展配置、context=流程上下文（上一步出参）、device=设备信息 + 定义接口需要的入参字段。参数来源使用业务化名称： + 固定值、认证参数、平台参数、上一步结果、当前设备、自动UUID、子设备传感器列表、基准点成对传感器列表、设备物模型、上报时间。
@@ -118,33 +118,49 @@ - + - + + + + + + + + + + + + + @@ -158,7 +174,12 @@ - + + + + @@ -166,6 +187,59 @@ 添加入参字段 + + +
+ 定义数组中每个元素的字段。可选来源：设备物模型、子设备映射、上报时间、固定值。 +
+ + + + + + + + + + + + + + + + + + + + + 添加字段 +
+ 关闭 + 确定 +
+ + 出参定义（response_schema）
@@ -218,6 +292,7 @@ + + diff --git a/yack-web/src/views/platform/flow/index.vue b/yack-web/src/views/platform/flow/index.vue index d9282d5f..4fb92282 100644 --- a/yack-web/src/views/platform/flow/index.vue +++ b/yack-web/src/views/platform/flow/index.vue @@ -1,6 +1,5 @@