fastbee/docs/knowledge-base/06-platform-integration-design.md
2026-06-06 16:35:49 +08:00

411 lines
19 KiB
Markdown
Raw Blame History

This file contains ambiguous Unicode characters

This file contains Unicode characters that might be confused with other characters. If you think that this is intentional, you can safely ignore this warning. Use the Escape button to reveal them.

# 平台对接设计文档
## 1. 背景与目标
FastBee 需要将设备数据推送至第三方监测平台(当前为 DCP 结构监测 SaaS),并支持未来对接多个平台。不同平台的认证方式、接口路径、字段格式各不相同,需要一套可配置、可扩展的对接框架。
**设计原则:**
- 平台、接口、流程、设备绑定全部可视化配置,无需改代码即可对接新平台
- 严格区分三类执行模式,避免过度设计
- 字段映射通过 UI 关联选择,不手填字符串
- 推送全程有日志,可追溯
---
## 2. 三种执行模式
### 模式一:流程编排(多步串联)
**适用场景:** 多个接口必须按顺序调用,且上一步的返回值是下一步的入参。
| 流程 | 触发方式 | 步骤 | 频率 |
|---|---|---|---|
| 平台初始化 | 手动点击一次 | Token → userId → projectId | 一次性 |
| 设备入网 | 手动点击(每设备) | 创建测点 → 绑定传感器 → 获得 DCP deviceId | 每设备一次 |
执行结果自动持久化:userId/projectId 写入 `platform_instance.extra_config`,DCP deviceId 回填到 `platform_device_mapping`。
### 模式二:直接调用(实时推送)
**适用场景:** 设备上报数据时,按字段映射转换后调用单个推送接口。
流程:设备上报 → 字段映射转换 → 直接 HTTP 调用推送接口 → 写推送日志
**不走流程引擎**,最短路径,不引入额外开销。
### 模式三:Token 自动刷新(横切关注点)
每次 HTTP 调用前自动检查 `token_expires_at`,提前 5 分钟视为过期,过期则先调用认证接口刷新 Token,对业务逻辑完全透明。
---
## 3. 数据库设计(6 张表)
### 3.1 `platform_instance` — 平台实体
```sql
CREATE TABLE `platform_instance` (
`id` bigint NOT NULL AUTO_INCREMENT,
`name` varchar(100) NOT NULL COMMENT '平台名称,如 DCP结构监测',
`code` varchar(50) NOT NULL UNIQUE COMMENT '平台唯一标识,如 dcp',
`base_url` varchar(255) NOT NULL COMMENT '基础URL',
`auth_type` tinyint NOT NULL DEFAULT 1 COMMENT '认证类型(1=密码Token,2=ApiKey,3=OAuth2)',
`auth_config` json DEFAULT NULL COMMENT '认证凭证JSON {userName,password,clientId,scope}',
`extra_config` json DEFAULT NULL COMMENT '初始化后保存的平台数据 {userId,projectId,...}',
`access_token` varchar(2000) DEFAULT NULL COMMENT '当前有效Token',
`token_expires_at` datetime DEFAULT NULL COMMENT 'Token过期时间',
`init_status` tinyint NOT NULL DEFAULT 0 COMMENT '初始化状态(0=未初始化,1=已初始化)',
`enable` char(1) NOT NULL DEFAULT '1',
`tenant_id` bigint DEFAULT NULL,
`remark` varchar(500) DEFAULT NULL,
`create_by` varchar(64) DEFAULT '',
`create_time` datetime DEFAULT CURRENT_TIMESTAMP,
`update_by` varchar(64) DEFAULT '',
`update_time` datetime DEFAULT CURRENT_TIMESTAMP ON UPDATE CURRENT_TIMESTAMP,
PRIMARY KEY (`id`)
) COMMENT='第三方平台实体';
```
### 3.2 `platform_interface` — 接口定义
```sql
CREATE TABLE `platform_interface` (
`id` bigint NOT NULL AUTO_INCREMENT,
`platform_id` bigint NOT NULL COMMENT '所属平台',
`name` varchar(100) NOT NULL COMMENT '接口名称',
`code` varchar(100) NOT NULL COMMENT '接口标识',
`method` varchar(10) NOT NULL COMMENT 'HTTP方法',
`path` varchar(255) NOT NULL COMMENT '接口路径,支持路径变量 {userId}',
`content_type` varchar(100) DEFAULT 'application/json',
`request_schema` json DEFAULT NULL COMMENT '入参字段定义',
`response_schema` json DEFAULT NULL COMMENT '出参字段定义',
`sort_order` int DEFAULT 0,
`remark` varchar(500) DEFAULT NULL,
`create_by` varchar(64) DEFAULT '',
`create_time` datetime DEFAULT CURRENT_TIMESTAMP,
`update_by` varchar(64) DEFAULT '',
`update_time` datetime DEFAULT CURRENT_TIMESTAMP ON UPDATE CURRENT_TIMESTAMP,
PRIMARY KEY (`id`),
KEY `idx_platform_id` (`platform_id`)
) COMMENT='平台接口定义';
```
**`request_schema` 字段 source 枚举:**
- `fixed` — 固定值(value 字段)
- `auth` — 从 `auth_config` 取(field 字段指定 key)
- `extra` — 从 `extra_config` 取(初始化后保存的 userId/projectId 等)
- `context` — 从流程上下文或运行时上下文取(contextKey 指定,如 `step1.access_token`)
- `device` — 从设备绑定关系取(如 `platform_device_id`)
- `thingsModel` — 从设备物模型实时值取(设备绑定页选择 identifier)
- `ts` — 自动取设备上报时间
**数组字段补充:**
- `type=array` 的字段通过 `itemSchema` 描述数组元素结构
- `itemSchema.source` 额外支持 `subDeviceId`,用于数组元素内绑定子设备平台 ID(如静力水准 `sensorId`)
### 3.3 `platform_flow` — 流程定义
```sql
CREATE TABLE `platform_flow` (
`id` bigint NOT NULL AUTO_INCREMENT,
`platform_id` bigint NOT NULL,
`name` varchar(100) NOT NULL,
`code` varchar(100) NOT NULL,
`flow_type` tinyint NOT NULL COMMENT '流程类型(1=平台初始化,2=设备入网,3=Token刷新)',
`sensor_type` varchar(50) DEFAULT NULL COMMENT '传感器类型,flow_type=2时有效',
`enable` char(1) NOT NULL DEFAULT '1',
`remark` varchar(500) DEFAULT NULL,
`create_by` varchar(64) DEFAULT '',
`create_time` datetime DEFAULT CURRENT_TIMESTAMP,
`update_by` varchar(64) DEFAULT '',
`update_time` datetime DEFAULT CURRENT_TIMESTAMP ON UPDATE CURRENT_TIMESTAMP,
PRIMARY KEY (`id`),
KEY `idx_platform_type` (`platform_id`, `flow_type`)
) COMMENT='平台对接流程';
```
**flow_type 说明:**
- `1` 平台初始化:手动触发,获取 userId/projectId 等基础数据
- `2` 设备入网:手动触发,创建测点+绑定传感器,获得 DCP deviceId
- `3` Token 刷新:系统自动触发,不在 UI 中手动执行
### 3.4 `platform_flow_step` — 流程步骤
```sql
CREATE TABLE `platform_flow_step` (
`id` bigint NOT NULL AUTO_INCREMENT,
`flow_id` bigint NOT NULL,
`step_order` int NOT NULL DEFAULT 1 COMMENT '执行顺序,从1开始',
`interface_id` bigint NOT NULL COMMENT '关联接口',
`param_mapping` json DEFAULT NULL COMMENT '入参来源覆盖配置',
`result_key` varchar(100) DEFAULT NULL COMMENT '出参写入上下文的key,如 step1',
`result_save` json DEFAULT NULL COMMENT '需持久化的出参字段 [{field,saveTo}]',
`remark` varchar(255) DEFAULT NULL,
PRIMARY KEY (`id`),
KEY `idx_flow_id` (`flow_id`)
) COMMENT='流程步骤';
```
**`result_save` 示例:** 初始化流程 Step2 获取 userId 后自动保存
```json
[
{"field": "id", "saveTo": "extra_config.userId"}
]
```
### 3.5 `platform_device_mapping` — 设备绑定
```sql
CREATE TABLE `platform_device_mapping` (
`id` bigint NOT NULL AUTO_INCREMENT,
`platform_id` bigint NOT NULL,
`device_id` bigint NOT NULL COMMENT 'FastBee设备ID',
`serial_number` varchar(64) NOT NULL COMMENT 'FastBee设备序列号',
`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,
`create_by` varchar(64) DEFAULT '',
`create_time` datetime DEFAULT CURRENT_TIMESTAMP,
`update_by` varchar(64) DEFAULT '',
`update_time` datetime DEFAULT CURRENT_TIMESTAMP ON UPDATE CURRENT_TIMESTAMP,
PRIMARY KEY (`id`),
KEY `idx_serial` (`serial_number`),
KEY `idx_platform_device` (`platform_id`, `device_id`)
) COMMENT='平台设备绑定关系';
```
### 3.6 `platform_push_log` — 推送日志
```sql
CREATE TABLE `platform_push_log` (
`id` bigint NOT NULL AUTO_INCREMENT,
`platform_id` bigint NOT NULL,
`flow_id` bigint DEFAULT NULL,
`step_order` int DEFAULT NULL,
`device_id` bigint DEFAULT NULL,
`serial_number` varchar(64) DEFAULT NULL,
`interface_code` varchar(100) DEFAULT NULL,
`request_url` varchar(500) DEFAULT NULL,
`request_body` mediumtext DEFAULT NULL,
`response_code` int DEFAULT NULL,
`response_body` mediumtext DEFAULT NULL,
`status` tinyint NOT NULL DEFAULT 0 COMMENT '0=进行中,1=成功,2=失败',
`error_msg` varchar(1000) DEFAULT NULL,
`idempotency_key` varchar(100) DEFAULT NULL COMMENT '幂等键',
`retry_count` int NOT NULL DEFAULT 0,
`cost_ms` int DEFAULT NULL,
`create_time` datetime DEFAULT CURRENT_TIMESTAMP,
PRIMARY KEY (`id`),
KEY `idx_platform_status` (`platform_id`, `status`),
KEY `idx_device_time` (`device_id`, `create_time`)
) COMMENT='平台推送日志';
```
---
## 4. 后端代码结构
```
fastbee-service/fastbee-iot-service/
└── com.fastbee.iot.platform/
├── domain/ # 6个实体类
├── mapper/ # MyBatis-Plus Mapper
├── service/
│ ├── IPlatformInstanceService
│ ├── IPlatformFlowService # 含手动触发、流程执行
│ ├── IPlatformDeviceMappingService
│ ├── IPlatformPushLogService
│ └── impl/
├── flow/
│ ├── PlatformFlowExecutor # 核心流程执行引擎
│ └── FlowContext # 步骤间传递的上下文
├── adapter/
│ ├── AbstractPlatformAdapter # Token管理、HTTP调用、日志
│ └── dcp/
│ ├── DcpPlatformAdapter # DCP实现
│ └── strategy/
│ ├── DcpSensorStrategy # 接口
│ ├── InclinoStrategy # 倾角
│ ├── StressStrategy # 轴力计(递增约束)
│ ├── HyetoStrategy # 雨量计(递增约束)
│ └── StaticLevelStrategy # 静力水准
└── trigger/
├── DeviceReportTrigger # 挂接 RuleProcess event=1
└── ScheduledPushJob # Quartz 定时批量推送
fastbee-open-api/
└── com.fastbee.controller.platform/
├── PlatformInstanceController # /platform/instance
├── PlatformInterfaceController # /platform/interface
├── PlatformFlowController # /platform/flow(含手动触发)
├── PlatformDeviceMappingController # /platform/device-mapping(含入网)
└── PlatformPushLogController # /platform/push-log
```
---
## 5. 核心逻辑
### 5.1 流程执行引擎(PlatformFlowExecutor)
```
FlowContext ctx = new FlowContext(platformInstance, deviceMapping?)
for each step (ordered by step_order):
1. 按 request_schema + param_mapping 解析入参
- fixed: 直接取 value
- auth: 从 auth_config JSON 取
- extra: 从 extra_config JSON 取(userId/projectId)
- context: 从 ctx.get(contextKey) 取(上一步出参)
- device: 从 deviceMapping 取(platform_device_id 等)
2. 替换 path 变量,如 /project/by-user-id/{userId}
3. 检查 Token,过期则先刷新
4. HttpClientFactory.instance(config).executeAsString()
5. 将出参写入 ctx(key = step.resultKey)
6. 按 result_save 将指定字段持久化到 platform_instance 或 platform_device_mapping
7. 异步写 platform_push_log
```
### 5.2 Token 自动刷新(AbstractPlatformAdapter)
```
before any HTTP call:
if token is null OR now > token_expires_at - 5min:
find flow where flow_type=3 for this platform
execute flow → get new access_token + expires_in
update platform_instance.access_token + token_expires_at
inject "Authorization: Bearer {token}" into request headers
```
### 5.3 实时推送(DeviceReportTrigger)
```
// 挂接在 RuleProcess.processRuleScript() 调用方之后,异步执行
List<PlatformDeviceMapping> 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(platform, mapping, reportSerialNumber, payload)
// pushData 内部:按 request_schema / itemSchema 动态组装 → Token检查 → HTTP调用 → 写日志
```
### 5.4 DCP 传感器推送策略
| sensor_type | 推送接口 | 关键字段 | 特殊约束 |
|---|---|---|---|
| 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[{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)
菜单:**系统管理 → 平台对接**
### 平台管理
- 列表:名称、标识、baseURL、Token状态(有效/过期/未认证)、初始化状态、启用
- 操作:新增/编辑、**初始化**(执行 flow_type=1 流程)、启用/禁用
### 流程编排
- 顶部选择平台,列表展示该平台所有流程
- 编辑页:步骤列表(可排序),每步选择接口 + 配置字段映射
- 字段来源下拉:固定值 / 认证配置 / 上下文(下拉展示前序步骤出参)/ 平台已保存数据
- 操作:保存、**手动执行**(展示实时执行日志)
### 设备绑定
- 顶部选择平台,列表展示绑定关系
- 列表关联展示:设备名称(非ID)、入网状态标签
- 新增弹窗:
- 设备搜索选择器(输入名称/序列号模糊搜索)
- 传感器类型下拉
- 推送接口选择器:从页面选择平台下已定义的推送接口
- 字段映射:根据所选推送接口 `request_schema` 动态渲染,不再按传感器类型写死
- 若 `request_schema` 含 `type=array` 字段,则自动出现“数组字段配置”区域
- 数组字段配置包含:
- 子设备序列号 ↔ 平台子设备字段映射(如 `sensorId`)
- 数组元素中 `source=thingsModel` 字段的物模型映射(如 `height`、`temperature`)
- 基准点选择与缓存有效间隔
- 操作:**入网**(执行 flow_type=2 流程,成功后自动回填 DCP deviceId)、启用/禁用
### 推送日志
- 筛选:平台、设备(搜索选择)、状态、时间范围
- 列表:时间、平台、设备名称、接口、状态标签、耗时
- 详情弹窗:请求URL/Body、响应码/Body(JSON格式化)、错误信息
---
## 7. DCP 内置初始化数据
系统内置 DCP 平台的接口定义和流程配置(SQL 初始化数据),用户只需填入账号密码即可使用。
**内置接口(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}
4. `dcp_create_point` — POST /api/app/point
5. `dcp_bind_inclino` — POST /api/app/inclino-sensor
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
**内置流程(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` 来源:按设备绑定页配置的基准点,将子设备拆成 `[基准点, 监测点]` 多个请求批次,适合静力水准等成组设备绑定场景
---
## 8. 关键复用点
| 现有资产 | 位置 | 用途 |
|---|---|---|
| `HttpClientFactory` | `fastbee-plugs/fastbee-http` | 发起 HTTP 请求 |
| `HttpClientConfig` | `fastbee-plugs/fastbee-http` | 请求配置 DTO |
| `RuleProcess` | `fastbee-iot-service` | 设备上报触发挂接点(event=1) |
| `sys_job` / Quartz | `fastbee-plugs/quartz` | 定时批量推送 Job |
| `iot_device_log` | DB | 批量推送数据来源 |
| `iot_things_model` | DB | 设备绑定页面物模型字段下拉 |
| MyBatis-Plus ServiceImpl | 已有 | 所有 Service 基类 |