fastbee/docs/knowledge-base/06-platform-integration-design.md
2026-05-31 08:42:02 +08:00

367 lines
16 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 字段)
### 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',
`field_mapping` json DEFAULT NULL COMMENT '字段映射 [{identifier,platformField,transform}]',
`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)
for each mapping where enable=1 and net_status=1:
adapter = getAdapter(mapping.platformId) // 按 platform.code 路由
adapter.pushData(mapping, payload)
// pushData 内部:字段映射转换 → 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[{height,sensorId,epoch}] | 需两个 sensorId |
递增约束处理:推送前查询 `platform_push_log` 中上次成功推送的值,若当前值 ≤ 上次值则跳过并记录警告。
---
## 6. 前端页面(yack-web)
菜单:**系统管理 → 平台对接**
### 平台管理
- 列表:名称、标识、baseURL、Token状态(有效/过期/未认证)、初始化状态、启用
- 操作:新增/编辑、**初始化**(执行 flow_type=1 流程)、启用/禁用
### 流程编排
- 顶部选择平台,列表展示该平台所有流程
- 编辑页:步骤列表(可排序),每步选择接口 + 配置字段映射
- 字段来源下拉:固定值 / 认证配置 / 上下文(下拉展示前序步骤出参)/ 平台已保存数据
- 操作:保存、**手动执行**(展示实时执行日志)
### 设备绑定
- 顶部选择平台,列表展示绑定关系
- 列表关联展示:设备名称(非ID)、入网状态标签
- 新增弹窗:
- 设备搜索选择器(输入名称/序列号模糊搜索)
- 传感器类型下拉
- 字段映射:物模型字段下拉(自动加载该设备产品的物模型)+ DCP字段(根据传感器类型预填建议值)
- 操作:**入网**(执行 flow_type=2 流程,成功后自动回填 DCP deviceId)、启用/禁用
### 推送日志
- 筛选:平台、设备(搜索选择)、状态、时间范围
- 列表:时间、平台、设备名称、接口、状态标签、耗时
- 详情弹窗:请求URL/Body、响应码/Body(JSON格式化)、错误信息
---
## 7. DCP 内置初始化数据
系统内置 DCP 平台的接口定义和流程配置(SQL 初始化数据),用户只需填入账号密码即可使用。
**内置接口(6个):**
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
**内置流程(5个):**
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
---
## 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 基类 |