20260531:平台对接第2版实现

fastbee-common/src/main/java/com/fastbee/common/config/DeviceTask.java

@@ -97,6 +97,12 @@ public class DeviceTask {
         return builder(FastBeeConstant.TASK.TRIGGER_POWER_ON_TASK_ASYNC);
     }
 
+    /*第三方平台推送*/
+    @Bean(FastBeeConstant.TASK.PUSH_TO_THIRD_PLATFORM_ASYNC)
+    public Executor pushToThirdPlatformExecutor() {
+        return builder(FastBeeConstant.TASK.PUSH_TO_THIRD_PLATFORM_ASYNC);
+    }
+
 
     /*组装线程池*/
     private ThreadPoolTaskExecutor builder(String threadNamePrefix) {

fastbee-common/src/main/java/com/fastbee/common/constant/FastBeeConstant.java

@@ -99,6 +99,9 @@ public interface FastBeeConstant {
 
         // 设备上电即触发异步任务 TriggerPowerOnTaskAsync
         String TRIGGER_POWER_ON_TASK_ASYNC = "triggerPowerOnTaskAsync";
+
+        // 设备上报数据-推送到第三方平台
+        String PUSH_TO_THIRD_PLATFORM_ASYNC = "pushToThirdPlatformAsync";
     }
 
     interface MQTT {

fastbee-iot-data/src/main/java/com/fastbee/iot/data/service/impl/DataHandlerImpl.java

@@ -20,6 +20,7 @@ import com.fastbee.common.utils.gateway.mq.TopicsUtils;
 import com.fastbee.iot.cache.SceneModelTagCache;
 import com.fastbee.iot.data.executor.ConditionTriggerTaskExecutor;
 import com.fastbee.iot.data.service.IDataHandler;
+import com.fastbee.iot.platform.trigger.DeviceReportTrigger;
 import com.fastbee.iot.data.service.IMqttMessagePublish;
 import com.fastbee.iot.data.service.IRuleEngine;
 import com.fastbee.iot.domain.*;
@@ -84,6 +85,9 @@ public class DataHandlerImpl implements IDataHandler {
     @Resource
     private IFunctionLogService functionLogService;
 
+    @Resource
+    private DeviceReportTrigger deviceReportTrigger;
+
     /**
      * 上报属性或功能处理
      *
@@ -116,6 +120,8 @@ public class DataHandlerImpl implements IDataHandler {
                         // payload 与 MQTT 上报格式一致：ThingsModelSimpleItem JSON 数组（含计算物模型）
                         String bridgePayload = JSON.toJSONString(result);
                         ruleProcess.processRuleScript(serialNumber, 1, "tcp/property/post", bridgePayload);
+                        // 平台对接推送（异步，不阻塞主链路）
+                        deviceReportTrigger.onDeviceReport(serialNumber, bridgePayload);
                     } catch (Exception e) {
                         log.error("数据桥接规则引擎执行异常 sn={}, msg={}", serialNumber, e.getMessage(), e);
                     }

fastbee-open-api/src/main/java/com/fastbee/controller/platform/PlatformPushLogController.java

@@ -33,4 +33,12 @@ public class PlatformPushLogController extends BaseController {
     public AjaxResult getInfo(@PathVariable Long id) {
         return success(pushLogService.getById(id));
     }
+
+    @PreAuthorize("@ss.hasPermi('platform:pushLog:edit')")
+    @PostMapping("/{id}/retry")
+    @ApiOperation("手动重推")
+    public AjaxResult retry(@PathVariable Long id) {
+        pushLogService.retry(id);
+        return success("重推成功");
+    }
 }

fastbee-service/fastbee-iot-service/src/main/java/com/fastbee/iot/platform/adapter/dcp/DcpPlatformAdapter.java

@@ -71,6 +71,7 @@ public class DcpPlatformAdapter {
         logEntry.setInterfaceCode("dcp_push_" + sensorType);
         logEntry.setStatus(0);
         logEntry.setRetryCount(0);
+        logEntry.setLogType(1); // 数据推送日志
 
         try {
             // 1. 字段映射转换
@@ -126,7 +127,12 @@ public class DcpPlatformAdapter {
             return null;
         }
         JSONArray fieldMappings = JSON.parseArray(mapping.getFieldMapping());
-        JSONObject reportData = JSON.parseObject(payload);
+
+        // 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());
@@ -158,6 +164,35 @@ public class DcpPlatformAdapter {
         return arr;
     }
 
+    /**
+     * 将设备上报 payload 解析为 {identifier: value} 的扁平 map
+     * 支持两种格式：
+     * 1. 数组: [{"id":"AngX","value":"-1.55","ts":"..."},...]
+     * 2. 对象: {"AngX":"-1.55","T":"21.6",...}
+     */
+    private JSONObject parsePayloadToMap(String payload) {
+        String trimmed = payload.trim();
+        if (trimmed.startsWith("[")) {
+            JSONArray arr = JSON.parseArray(trimmed);
+            JSONObject map = new JSONObject();
+            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 (item.containsKey("ts")) {
+                    map.put(id + "__ts", item.get("ts"));
+                }
+            }
+            return map;
+        } else {
+            return JSON.parseObject(trimmed);
+        }
+    }
+
     /** 确保 Token 有效，过期则执行 flow_type=3 的 Token 刷新流程 */
     private String ensureToken(PlatformInstance platform) {
         Date now = new Date();

fastbee-service/fastbee-iot-service/src/main/java/com/fastbee/iot/platform/domain/PlatformPushLog.java

@@ -57,6 +57,9 @@ public class PlatformPushLog implements Serializable {
     @ApiModelProperty("状态(0=进行中,1=成功,2=失败)")
     private Integer status;
 
+    @ApiModelProperty("日志类型(1=数据推送,2=操作日志)")
+    private Integer logType;
+
     @ApiModelProperty("错误信息")
     private String errorMsg;
 

fastbee-service/fastbee-iot-service/src/main/java/com/fastbee/iot/platform/flow/PlatformFlowExecutor.java

@@ -69,6 +69,7 @@ public class PlatformFlowExecutor {
         logEntry.setInterfaceCode(iface.getCode());
         logEntry.setStatus(0);
         logEntry.setRetryCount(0);
+        logEntry.setLogType(2); // 操作日志（认证/入网等流程步骤）
 
         try {
             // 1. 解析入参

fastbee-service/fastbee-iot-service/src/main/java/com/fastbee/iot/platform/service/IPlatformPushLogService.java

@@ -14,4 +14,7 @@ public interface IPlatformPushLogService extends IService<PlatformPushLog> {
 
     /** 查询某设备某接口最后一次成功推送的响应体（用于递增约束校验） */
     String selectLastSuccessResponseBody(Long deviceId, String interfaceCode);
+
+    /** 手动重推：重新发送已记录的请求，更新日志状态 */
+    void retry(Long id);
 }

fastbee-service/fastbee-iot-service/src/main/java/com/fastbee/iot/platform/service/impl/PlatformPushLogServiceImpl.java

@@ -3,7 +3,13 @@ package com.fastbee.iot.platform.service.impl;
 import com.baomidou.mybatisplus.core.conditions.query.LambdaQueryWrapper;
 import com.baomidou.mybatisplus.core.toolkit.Wrappers;
 import com.baomidou.mybatisplus.extension.service.impl.ServiceImpl;
+import com.dtflys.forest.http.ForestRequest;
+import com.dtflys.forest.http.ForestRequestType;
+import com.fastbee.http.model.HttpClientConfig;
+import com.fastbee.http.service.HttpClientFactory;
+import com.fastbee.iot.platform.domain.PlatformInstance;
 import com.fastbee.iot.platform.domain.PlatformPushLog;
+import com.fastbee.iot.platform.mapper.PlatformInstanceMapper;
 import com.fastbee.iot.platform.mapper.PlatformPushLogMapper;
 import com.fastbee.iot.platform.service.IPlatformPushLogService;
 import org.springframework.beans.factory.annotation.Autowired;
@@ -11,7 +17,9 @@ import org.springframework.scheduling.annotation.Async;
 import org.springframework.stereotype.Service;
 import org.springframework.util.StringUtils;
 
+import java.util.HashMap;
 import java.util.List;
+import java.util.Map;
 
 @Service
 public class PlatformPushLogServiceImpl extends ServiceImpl<PlatformPushLogMapper, PlatformPushLog>
@@ -19,6 +27,8 @@ public class PlatformPushLogServiceImpl extends ServiceImpl<PlatformPushLogMappe
 
     @Autowired
     private PlatformPushLogMapper pushLogMapper;
+    @Autowired
+    private PlatformInstanceMapper instanceMapper;
 
     @Override
     public List<PlatformPushLog> selectList(PlatformPushLog query) {
@@ -26,6 +36,7 @@ public class PlatformPushLogServiceImpl extends ServiceImpl<PlatformPushLogMappe
                 .eq(query.getPlatformId() != null, PlatformPushLog::getPlatformId, query.getPlatformId())
                 .eq(query.getDeviceId() != null, PlatformPushLog::getDeviceId, query.getDeviceId())
                 .eq(query.getStatus() != null, PlatformPushLog::getStatus, query.getStatus())
+                .eq(query.getLogType() != null, PlatformPushLog::getLogType, query.getLogType())
                 .eq(StringUtils.hasText(query.getSerialNumber()), PlatformPushLog::getSerialNumber, query.getSerialNumber())
                 .orderByDesc(PlatformPushLog::getCreateTime);
         return this.list(wrapper);
@@ -41,4 +52,66 @@ public class PlatformPushLogServiceImpl extends ServiceImpl<PlatformPushLogMappe
     public String selectLastSuccessResponseBody(Long deviceId, String interfaceCode) {
         return pushLogMapper.selectLastSuccessResponseBody(deviceId, interfaceCode);
     }
+
+    @Override
+    public void retry(Long id) {
+        PlatformPushLog log = this.getById(id);
+        if (log == null) throw new RuntimeException("日志记录不存在: " + id);
+        if (!StringUtils.hasText(log.getRequestUrl()) || !StringUtils.hasText(log.getRequestBody())) {
+            throw new RuntimeException("日志缺少请求URL或请求体，无法重推");
+        }
+
+        long startMs = System.currentTimeMillis();
+        // 标记为进行中
+        PlatformPushLog update = new PlatformPushLog();
+        update.setId(id);
+        update.setStatus(0);
+        update.setRetryCount(log.getRetryCount() == null ? 1 : log.getRetryCount() + 1);
+        update.setErrorMsg(null);
+        this.updateById(update);
+
+        try {
+            // 获取平台 token
+            String token = null;
+            if (log.getPlatformId() != null) {
+                PlatformInstance platform = instanceMapper.selectById(log.getPlatformId());
+                if (platform != null) {
+                    token = platform.getAccessToken();
+                }
+            }
+
+            Map<String, Object> headers = new HashMap<>();
+            headers.put("Content-Type", "application/json");
+            if (StringUtils.hasText(token)) {
+                headers.put("Authorization", "Bearer " + token);
+            }
+
+            HttpClientConfig config = HttpClientConfig.builder()
+                    .url(log.getRequestUrl())
+                    .method(ForestRequestType.POST)
+                    .headers(headers)
+                    .body(log.getRequestBody())
+                    .build();
+
+            ForestRequest<?> request = HttpClientFactory.instance(config);
+            String responseBody = request.executeAsString();
+
+            PlatformPushLog success = new PlatformPushLog();
+            success.setId(id);
+            success.setStatus(1);
+            success.setResponseBody(responseBody);
+            success.setResponseCode(200);
+            success.setCostMs((int) (System.currentTimeMillis() - startMs));
+            this.updateById(success);
+
+        } catch (Exception e) {
+            PlatformPushLog fail = new PlatformPushLog();
+            fail.setId(id);
+            fail.setStatus(2);
+            fail.setErrorMsg(e.getMessage());
+            fail.setCostMs((int) (System.currentTimeMillis() - startMs));
+            this.updateById(fail);
+            throw new RuntimeException("重推失败: " + e.getMessage(), e);
+        }
+    }
 }

fastbee-service/fastbee-iot-service/src/main/java/com/fastbee/iot/platform/trigger/DeviceReportTrigger.java

@@ -1,5 +1,6 @@
 package com.fastbee.iot.platform.trigger;
 
+import com.fastbee.common.constant.FastBeeConstant;
 import com.fastbee.iot.platform.adapter.dcp.DcpPlatformAdapter;
 import com.fastbee.iot.platform.domain.PlatformDeviceMapping;
 import com.fastbee.iot.platform.domain.PlatformInstance;
@@ -36,7 +37,7 @@ public class DeviceReportTrigger {
      * @param serialNumber 设备序列号
      * @param payload      上报数据 JSON
      */
-    @Async
+    @Async(FastBeeConstant.TASK.PUSH_TO_THIRD_PLATFORM_ASYNC)
     public void onDeviceReport(String serialNumber, String payload) {
         try {
             List<PlatformDeviceMapping> mappings = mappingService.selectBySerialNumber(serialNumber);

sql/platform-dcp-init-data.sql

@@ -236,7 +236,9 @@ SET @flow_inclino_id = LAST_INSERT_ID();
 INSERT INTO `platform_flow_step` (`flow_id`,`step_order`,`interface_id`,`result_key`,`result_save`,`remark`)
 VALUES (@flow_inclino_id, 1,
   (SELECT id FROM platform_interface WHERE platform_id=@dcp_platform_id AND code='dcp_create_point' LIMIT 1),
-  'step1', NULL, '创建测点，获取pointId');
+  'step1',
+  '[{"field":"id","saveTo":"device_mapping.platform_point_id"}]',
+  '创建测点，获取pointId并回填');
 
 INSERT INTO `platform_flow_step` (`flow_id`,`step_order`,`interface_id`,`result_key`,`result_save`,`remark`)
 VALUES (@flow_inclino_id, 2,
@@ -252,10 +254,12 @@ VALUES (@dcp_platform_id, 'DCP轴力计入网', 'dcp_onboard_stress', 2, 'stress
 
 SET @flow_stress_id = LAST_INSERT_ID();
 
-INSERT INTO `platform_flow_step` (`flow_id`,`step_order`,`interface_id`,`result_key`,`remark`)
+INSERT INTO `platform_flow_step` (`flow_id`,`step_order`,`interface_id`,`result_key`,`result_save`,`remark`)
 VALUES (@flow_stress_id, 1,
   (SELECT id FROM platform_interface WHERE platform_id=@dcp_platform_id AND code='dcp_create_point' LIMIT 1),
-  'step1', '创建测点');
+  '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_stress_id, 2,
@@ -271,10 +275,12 @@ VALUES (@dcp_platform_id, 'DCP雨量计入网', 'dcp_onboard_hyeto', 2, 'hyeto',
 
 SET @flow_hyeto_id = LAST_INSERT_ID();
 
-INSERT INTO `platform_flow_step` (`flow_id`,`step_order`,`interface_id`,`result_key`,`remark`)
+INSERT INTO `platform_flow_step` (`flow_id`,`step_order`,`interface_id`,`result_key`,`result_save`,`remark`)
 VALUES (@flow_hyeto_id, 1,
   (SELECT id FROM platform_interface WHERE platform_id=@dcp_platform_id AND code='dcp_create_point' LIMIT 1),
-  'step1', '创建测点');
+  '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_hyeto_id, 2,

sql/platform-integration.sql

@@ -128,6 +128,7 @@ CREATE TABLE IF NOT EXISTS `platform_push_log` (
   `status`           tinyint(1)    NOT NULL DEFAULT 0 COMMENT '状态(0=进行中,1=成功,2=失败)',
   `error_msg`        varchar(1000) DEFAULT NULL COMMENT '错误信息',
   `idempotency_key`  varchar(100)  DEFAULT NULL COMMENT '幂等键，防重复推送',
+  `log_type`         tinyint(1)    NOT NULL DEFAULT 1 COMMENT '日志类型(1=数据推送,2=操作日志)',
   `retry_count`      int(11)       NOT NULL DEFAULT 0 COMMENT '重试次数',
   `cost_ms`          int(11)       DEFAULT NULL COMMENT '耗时毫秒',
   `create_time`      datetime      DEFAULT CURRENT_TIMESTAMP COMMENT '创建时间',

yack-web/src/api/platform/pushLog.js

@@ -7,3 +7,7 @@ export function listPlatformPushLog(params) {
 export function getPlatformPushLog(id) {
   return request({ url: `/platform/push-log/${id}`, method: 'get' })
 }
+
+export function retryPlatformPushLog(id) {
+  return request({ url: `/platform/push-log/${id}/retry`, method: 'post' })
+}

yack-web/src/views/platform/push-log/index.vue

@@ -28,13 +28,18 @@
     <el-card>
       <right-toolbar :showSearch.sync="showSearch" @queryTable="getList" />
 
+      <el-tabs v-model="activeTab" @tab-click="onTabChange">
+        <el-tab-pane label="数据推送" name="1" />
+        <el-tab-pane label="操作日志" name="2" />
+      </el-tabs>
+
       <el-table v-loading="loading" :data="list">
         <el-table-column label="时间" prop="createTime" width="160" />
         <el-table-column label="平台" width="120">
           <template slot-scope="scope">{{ platformName(scope.row.platformId) }}</template>
         </el-table-column>
         <el-table-column label="设备序列号" prop="serialNumber" width="140" />
-        <el-table-column label="接口" prop="interfaceCode" width="160" show-overflow-tooltip />
+        <el-table-column label="接口" prop="interfaceCode" width="180" show-overflow-tooltip />
         <el-table-column label="状态" width="80" align="center">
           <template slot-scope="scope">
             <el-tag size="small" type="info"    v-if="scope.row.status===0">进行中</el-tag>
@@ -44,9 +49,14 @@
         </el-table-column>
         <el-table-column label="耗时(ms)" prop="costMs" width="90" align="right" />
         <el-table-column label="错误信息" prop="errorMsg" min-width="160" show-overflow-tooltip />
-        <el-table-column label="操作" fixed="right" width="80" align="center">
+        <el-table-column label="操作" fixed="right" width="140" align="center">
           <template slot-scope="scope">
             <el-button size="mini" type="text" icon="el-icon-view" @click="handleDetail(scope.row)">详情</el-button>
+            <el-button
+              v-if="scope.row.status===2 && activeTab==='1'"
+              size="mini" type="text" icon="el-icon-refresh"
+              :loading="retryingId===scope.row.id"
+              @click="handleRetry(scope.row)">重推</el-button>
           </template>
         </el-table-column>
       </el-table>
@@ -55,12 +65,13 @@
     </el-card>
 
     <!-- 详情弹窗 -->
-    <el-dialog title="推送详情" :visible.sync="detailOpen" width="800px" append-to-body>
+    <el-dialog title="日志详情" :visible.sync="detailOpen" width="800px" append-to-body>
       <el-descriptions :column="2" border size="small" style="margin-bottom:16px">
         <el-descriptions-item label="平台">{{ platformName(detail.platformId) }}</el-descriptions-item>
         <el-descriptions-item label="接口">{{ detail.interfaceCode }}</el-descriptions-item>
         <el-descriptions-item label="设备序列号">{{ detail.serialNumber }}</el-descriptions-item>
         <el-descriptions-item label="状态">
+          <el-tag size="small" type="info"    v-if="detail.status===0">进行中</el-tag>
           <el-tag size="small" type="success" v-if="detail.status===1">成功</el-tag>
           <el-tag size="small" type="danger"  v-if="detail.status===2">失败</el-tag>
         </el-descriptions-item>
@@ -88,7 +99,7 @@
 </template>
 
 <script>
-import { listPlatformPushLog, getPlatformPushLog } from '@/api/platform/pushLog'
+import { listPlatformPushLog, getPlatformPushLog, retryPlatformPushLog } from '@/api/platform/pushLog'
 import { listPlatformInstance } from '@/api/platform/instance'
 
 export default {
@@ -100,10 +111,12 @@ export default {
       total: 0,
       list: [],
       platformList: [],
+      activeTab: '1',
+      retryingId: null,
       detailOpen: false,
       detailTab: 'req',
       detail: {},
-      queryParams: { pageNum: 1, pageSize: 20, platformId: null, serialNumber: '', status: null }
+      queryParams: { pageNum: 1, pageSize: 20, platformId: null, serialNumber: '', status: null, logType: 1 }
     }
   },
   created() {
@@ -121,6 +134,11 @@ export default {
       const p = this.platformList.find(x => x.id === id)
       return p ? p.name : id
     },
+    onTabChange(tab) {
+      this.queryParams.logType = parseInt(tab.name)
+      this.queryParams.pageNum = 1
+      this.getList()
+    },
     handleQuery() { this.queryParams.pageNum = 1; this.getList() },
     resetQuery() { this.resetForm('queryForm'); this.handleQuery() },
     handleDetail(row) {
@@ -130,6 +148,17 @@ export default {
         this.detailOpen = true
       })
     },
+    handleRetry(row) {
+      this.retryingId = row.id
+      retryPlatformPushLog(row.id).then(() => {
+        this.$message.success('重推成功')
+        this.getList()
+      }).catch(err => {
+        this.$message.error(err.message || '重推失败')
+      }).finally(() => {
+        this.retryingId = null
+      })
+    },
     formatJson(str) {
       if (!str) return ''
       try { return JSON.stringify(JSON.parse(str), null, 2) } catch { return str }