diff --git a/bin/script/send_modbus_and_register.py b/bin/script/send_modbus_and_register.py index cf65ad6b..88c9607b 100644 --- a/bin/script/send_modbus_and_register.py +++ b/bin/script/send_modbus_and_register.py @@ -8,13 +8,11 @@ TARGET_PORT = 8888 # 请根据实际情况修改 # 元数组 modbus 指令 MODBUS_DATA_HEX = ( - "7e80D9A718F9362A7e", - "0103204135317841ab8e39bf27bccd3f1fb19ac2b231bbc0b6f82040ae3207c3f9f80efa23", - "020320413b9fa441acd34dc0024a003fa2a59ac2af34a0c18e17ea413167dbc3f9c7fe6d40", - "0303204139565641ab21ddbff6af333fe26800c2aec626c18685244176ee82c3f9bd7568c8", + "7e80DFEB502006887e", + "01033041011b7b0000000041eaafb742d19bd74513d545450dd9a4450ec53d00000000420ccf64420c74a9420c74a9c2d200009d99" ) -# 重要-四川远德MQTT推送 +# 重要-四川远德MQTT推送 （倾角计） # "7e803042593632517e", # "01033041011b7b0000000041eaafb742d19bd74513d545450dd9a4450ec53d00000000420ccf64420c74a9420c74a9c2d200009d99" # 深部位移计 diff --git a/docs/knowledge-base/06-platform-integration-design.md b/docs/knowledge-base/06-platform-integration-design.md index 78dec70d..a7b2a142 100644 --- a/docs/knowledge-base/06-platform-integration-design.md +++ b/docs/knowledge-base/06-platform-integration-design.md @@ -10,11 +10,11 @@ ## 1. 背景与目标 -FastBee 需要将设备上报数据推送到第三方监测平台，当前已落地 DCP 结构监测 SaaS。模块目标是把“平台、接口、流程、设备绑定、字段映射、推送日志”配置化，降低后续接入新平台或调整 DCP 接口的代码改动。 +FastBee 需要将设备上报数据推送到第三方监测平台，当前已落地 DCP 结构监测 SaaS。模块目标是把"平台、用户认证、接口、流程、设备绑定、字段映射、推送日志"配置化，降低后续接入新平台或调整 DCP 接口的代码改动。 当前实现采用两条主链路： -1. 流程执行链路：用于平台初始化、Token 刷新、设备入网等多步骤调用。 +1. 流程执行链路：用于用户初始化、Token 刷新、设备入网等多步骤调用。 2. 实时推送链路：设备上报后异步命中平台设备绑定，由 DCP 适配器组装请求体并推送。 设计边界： @@ -23,6 +23,7 @@ FastBee 需要将设备上报数据推送到第三方监测平台，当前已落 - 实时数据推送不走流程引擎，按设备绑定上的 `push_interface_id`、`field_mapping`、`extra_mapping` 直接组装请求。 - 当前只有 DCP 适配器实现，平台路由按 `platform_instance.code` 分发，`code=dcp` 时进入 `DcpPlatformAdapter`。 - 网关多子设备场景通过独立子表 `platform_device_sub_mapping` 管理，不再只依赖 `extra_mapping`。 +- **Token/认证已从平台级（PlatformInstance）迁移到用户级（PlatformUserAuth）**，每个用户有独立的 Token、过期时间和项目权限。 --- @@ -34,18 +35,19 @@ FastBee 需要将设备上报数据推送到第三方监测平台，当前已落 | 流程 | 触发方式 | 典型步骤 | 结果落库 | |---|---|---|---| -| 平台初始化 | 平台管理页手动初始化，或手动执行流程 | `dcp_auth` -> `dcp_get_user` -> `dcp_get_projects` | `access_token`、`token_expires_at`、`extra_config.userId`、`extra_config.projectId`、`extra_config.projectName` | -| Token 刷新 | HTTP 调用前自动触发，也可手动执行 | `dcp_auth` | `access_token`、`token_expires_at` | +| 用户初始化 | 用户管理页点击初始化 | `dcp_auth` -> `dcp_get_user` -> `dcp_get_projects` | `platform_user_auth.access_token`、`platform_user_auth.platform_user_id`、`platform_user_auth.platform_user_name`、`platform_user_auth.extra_config` | +| Token 刷新 | HTTP 调用前自动触发 | `dcp_auth` | `platform_user_auth.access_token`、`platform_user_auth.token_expires_at` | | 直连设备入网 | 设备绑定页点击入网 | `dcp_create_point` -> 对应传感器绑定接口 | 主表 `platform_point_id`、`platform_device_id`、`net_status` | | 网关设备入网 | 设备绑定页点击入网 | 对每个 monitor 子设备逐个执行入网流程 | 子表 `platform_point_id`、`platform_device_id`、`net_status`；全部 monitor 成功后主表 `net_status=1` | 入口代码： -- `PlatformInstanceServiceImpl.initialize(Long id)` -- `PlatformFlowServiceImpl.executeManually(Long id, Long mappingId)` -- `PlatformDeviceMappingServiceImpl.onboard(Long mappingId)` +- `PlatformUserAuthServiceImpl.initialize(Long userAuthId)` - 用户初始化 +- `PlatformDeviceMappingServiceImpl.onboard(Long mappingId)` - 设备入网 - `PlatformFlowExecutor.execute(PlatformFlow flow, FlowContext context)` +> **注意：** 流程手动执行端点（`POST /platform/flow/{id}/execute`）已移除，流程执行只能通过业务入口（初始化/入网/Token 刷新）触发。 + ### 2.2 直接调用实时推送 设备上报后由 `DeviceReportTrigger.onDeviceReport(serialNumber, payload)` 异步触发： @@ -53,9 +55,10 @@ FastBee 需要将设备上报数据推送到第三方监测平台，当前已落 1. `PlatformDeviceMappingMapper.selectBySerialNumber` 同时匹配直连主设备和网关子设备。 2. 过滤未启用的平台和绑定关系。 3. 按平台 `code` 分发，目前 `dcp` 进入 `DcpPlatformAdapter`。 -4. 适配器按推送接口的 `request_schema` 与设备绑定字段映射组装请求体。 -5. 检查并刷新 Token。 -6. 发送 HTTP POST，写入 `platform_push_log`。 +4. 加载 `PlatformUserAuth`，传入适配器用于 Token 获取和字段映射。 +5. 适配器按推送接口的 `request_schema` 与设备绑定字段映射组装请求体。 +6. 检查并刷新 Token（从 userAuth 获取）。 +7. 发送 HTTP POST，写入 `platform_push_log`。 实时推送不经过 `PlatformFlowExecutor`，因此不会使用流程步骤的 `param_mapping`、`result_key`、`result_save`。 @@ -71,7 +74,7 @@ FastBee 需要将设备上报数据推送到第三方监测平台，当前已落 调度器 `PlatformPushScheduledJob` 使用 Spring `@Scheduled(fixedDelay = 300_000)` 每 5 分钟检查一次： 1. 查询所有 `enable='1'`、`push_mode=2`、`push_interval > 0` 的绑定。 -2. 逐条判断 `now - last_push_time >= push_interval * 60 * 1000`；从未推送过的立即触发。 +2. 加载绑定关联的 `PlatformUserAuth`，逐条判断 `now - last_push_time >= push_interval * 60 * 1000`；从未推送过的立即触发。 3. 从 Redis 缓存读取最新数据，按直连/网关分别调用 `DcpPlatformAdapter` 推送。 4. 推送成功后更新 `last_push_time`；失败时保留不变，下次继续尝试。 @@ -79,22 +82,25 @@ FastBee 需要将设备上报数据推送到第三方监测平台，当前已落 ### 2.4 Token 自动刷新 -流程执行器和 DCP 实时推送适配器各自调用 `ensureToken`： +流程执行器调用 `ensureToken(FlowContext ctx)`： +- 从 `FlowContext.platformUserAuth` 获取用户认证信息。 - `access_token` 为空，或 `token_expires_at` 距当前时间不足 5 分钟时视为过期。 - 自动查找同平台 `flow_type=3` 且 `enable='1'` 的 Token 刷新流程。 - 刷新流程本身不注入 Authorization，避免递归。 -- 刷新成功后重新读取或更新 `platform_instance.access_token` 和 `token_expires_at`。 +- 刷新成功后更新 `platform_user_auth.access_token` 和 `token_expires_at`（不再是 `platform_instance`）。 + +> **架构变更：** Token 管理已从 `PlatformInstance` 迁移到 `PlatformUserAuth`。`PlatformInstance.access_token` / `PlatformInstance.token_expires_at` 字段不再使用，所有 Token 读写都通过 `platform_user_auth` 表。 --- -## 3. 数据库设计（7 张表） +## 3. 数据库设计（8 张表） DDL 以 `sql/platform-integration.sql` 为准。 ### 3.1 `platform_instance` 平台实体 -保存第三方平台实例、认证配置和初始化结果。 +保存第三方平台实例和认证模板。 关键字段： @@ -104,15 +110,47 @@ DDL 以 `sql/platform-integration.sql` 为准。 | `code` | 平台唯一标识，当前 DCP 为 `dcp`，有唯一索引 | | `base_url` | 平台基础 URL | | `auth_type` | 认证类型：`1=密码Token`、`2=ApiKey`、`3=OAuth2` | -| `auth_config` | 认证配置 JSON，例如 `userName`、`password`、`clientId`、`scope` | -| `extra_config` | 初始化后保存的业务上下文，例如 `userId`、`projectId`、`projectName` | -| `access_token` | 当前 Token | -| `token_expires_at` | Token 过期时间 | -| `init_status` | 初始化状态 | +| `auth_config` | 认证配置模板 JSON，新增用户时可复制；例如 `userName`、`password`、`clientId`、`scope` | +| `extra_config` | 平台级业务参数，如站点编码 | | `enable` | 是否启用 | | `tenant_id` | 租户 ID | -### 3.2 `platform_interface` 接口定义 +**废弃字段：** `access_token`、`token_expires_at`、`init_status` 为历史遗留，实际不再使用。Token 和初始化状态现在存储在 `platform_user_auth` 表。 + +**架构说明：** 一个平台可以对应多个用户认证（`platform_user_auth`），每个用户有独立的 Token、项目权限。`platform_instance.auth_config` 仅作为新增用户时的默认模板。 + +### 3.2 `platform_user_auth` 平台用户认证 + +支持一个平台对应多个用户认证，每个用户有独立的 Token 和项目权限。 + +关键字段： + +| 字段 | 说明 | +|---|---| +| `platform_id` | 平台实例 ID | +| `platform_user_id` | 平台侧用户 ID（如 DCP 的 userId），初始化流程执行后回填，允许为空 | +| `platform_user_name` | 平台侧用户名，初始化流程执行后回填 | +| `auth_config` | 用户认证凭证 JSON，例如 `{userName, password, clientId, scope}` | +| `access_token` | 该用户的 Token | +| `token_expires_at` | Token 过期时间 | +| `init_status` | 初始化状态：`0=未初始化`、`1=已初始化` | +| `extra_config` | 用户级配置，如项目信息等 | +| `enable` | 是否启用 | + +**使用流程：** + +1. 新增用户认证：仅输入认证配置（用户名/密码/clientId/scope） +2. 用户记录处于 `init_status=0`，`platform_user_id` 为空 +3. 执行初始化流程： + - 用认证配置获取 Token + - 用 Token 获取 `userId` 和用户名 + - 自动回填 `platform_user_id` 和 `platform_user_name` + - 标记 `init_status=1` +4. 设备绑定时选择已初始化的用户，再获取项目列表 + +**唯一约束：** `uk_platform_user (platform_id, platform_user_id)`，注意 `platform_user_id` 允许为空（未初始化时）。 + +### 3.3 `platform_interface` 接口定义 保存第三方 HTTP 接口定义。 @@ -135,8 +173,8 @@ DDL 以 `sql/platform-integration.sql` 为准。 |---|---| | `fixed` | 固定值，取 `value` | | `uuid` | 每次请求生成新的 UUID | -| `auth` | 从 `platform_instance.auth_config` 按 `field` 取值 | -| `extra` | 从 `platform_instance.extra_config` 按 `field` 取值 | +| `auth` | 优先从 `platform_user_auth.auth_config` 取值，fallback 到 `platform_instance.auth_config` | +| `extra` | 优先从 `platform_user_auth.extra_config` 取值，fallback 到 `platform_instance.extra_config` | | `context` | 从 `FlowContext` 按 `contextKey` 取值 | | `device` | 从 `PlatformDeviceMapping` / 当前网关子设备上下文取 `platform_device_id`、`platform_point_id`、`serial_number`、`device_name`、`gateway_serial_number`、`monitor_serial_number` | | `thingsModel` | 从流程上下文 payload 按物模型标识取值，通常用于手动流程或扩展流程 | @@ -167,8 +205,8 @@ DDL 以 `sql/platform-integration.sql` 为准。 |---|---| | `fixed` | 固定值 | | `device` | 从绑定关系取平台设备/测点/序列号 | -| `auth` | 从认证 JSON 取值 | -| `extra` | 从扩展 JSON 取值 | +| `auth` | 优先从 `platform_user_auth` 认证 JSON 取值 | +| `extra` | 优先从 `platform_user_auth` 扩展 JSON 取值 | | `thingsModel` / `identifier` | 按 `field_mapping` 或 `itemFieldMappings` 从上报 payload 取物模型值 | | `ts` | 从物模型值的 `__ts` 派生时间，缺省为当前时间 | | `context` | 优先从 payload 的 `contextKey` 取值；`reportTime`/`epoch` 会按时间字段处理 | @@ -176,7 +214,7 @@ DDL 以 `sql/platform-integration.sql` 为准。 `response_schema.path` 支持对象路径、数组下标路径，例如 `id`、`[0].id`、`$.data.id`。 -### 3.3 `platform_flow` 流程定义 +### 3.4 `platform_flow` 流程定义 关键字段： @@ -188,7 +226,7 @@ DDL 以 `sql/platform-integration.sql` 为准。 | `sensor_type` | 入网流程适用传感器类型，如 `inclino`、`deep_inclino`、`stress`、`hyeto`、`static_level` | | `enable` | 是否启用 | -### 3.4 `platform_flow_step` 流程步骤 +### 3.5 `platform_flow_step` 流程步骤 关键字段： @@ -199,17 +237,21 @@ DDL 以 `sql/platform-integration.sql` 为准。 | `interface_id` | 关联接口 | | `param_mapping` | 覆盖接口默认入参来源 | | `result_key` | 步骤结果写入上下文的 key | -| `result_save` | 将响应字段持久化到平台或设备绑定 | +| `result_save` | 将响应字段持久化到用户认证或设备绑定 | `result_save` 支持： -- `extra_config.xxx` -- `platform_instance.access_token` -- `platform_instance.token_expires_at`，值按 `expires_in` 秒数转换为 Date +- `platform_user_auth.access_token` - 用户级 Token +- `platform_user_auth.token_expires_at` - 值按 `expires_in` 秒数转换为 Date +- `platform_user_auth.platform_user_id` - 平台侧用户 ID +- `platform_user_auth.platform_user_name` - 平台侧用户名 +- `extra_config.xxx` - 用户级配置 - `device_mapping.platform_device_id` - `device_mapping.platform_point_id` -### 3.5 `platform_device_mapping` 设备绑定主表 +> **注意：** `result_save` 已从写 `platform_instance.*` 改为写 `platform_user_auth.*`。配置中的 `platform_instance.access_token` 等旧值应更新为 `platform_user_auth.access_token`。 + +### 3.6 `platform_device_mapping` 设备绑定主表 保存 FastBee 设备与第三方平台的绑定关系。 @@ -218,6 +260,9 @@ DDL 以 `sql/platform-integration.sql` 为准。 | 字段 | 说明 | |---|---| | `platform_id` | 所属平台 | +| `platform_user_auth_id` | 关联用户认证 ID，选择已初始化的用户 | +| `platform_project_id` | 平台侧项目 ID，用户选择后存储 | +| `platform_project_name` | 平台侧项目名称 | | `device_id` / `serial_number` | FastBee 设备 ID 与序列号 | | `device_name` | 推送给平台的设备名称，默认取平台设备名称，可手动修改 | | `platform_device_id` | 平台侧 deviceId，直连设备入网后回填 | @@ -233,9 +278,11 @@ DDL 以 `sql/platform-integration.sql` 为准。 | `push_interval` | 定时推送间隔（分钟），`0` 表示不自动推送 | | `last_push_time` | 最近推送时间，由定时调度器更新 | -### 3.6 `platform_device_sub_mapping` 子设备映射表 +**级联选择流程：** 新增设备绑定时，先选择平台 → 加载已初始化用户列表 → 选择用户 → 加载项目列表 → 选择项目。 -用于网关多子设备场景，尤其是静力水准。 +### 3.7 `platform_device_sub_mapping` 子设备映射表 + +用于网关多子设备场景，尤其是静力水准和深部位移。 关键字段： @@ -252,7 +299,7 @@ DDL 以 `sql/platform-integration.sql` 为准。 唯一约束：`parent_mapping_id + sub_serial_number`。 -### 3.7 `platform_push_log` 推送日志 +### 3.8 `platform_push_log` 推送日志 记录流程步骤调用、实时数据推送和失败信息。 @@ -281,6 +328,7 @@ DDL 以 `sql/platform-integration.sql` 为准。 fastbee-service/fastbee-iot-service/src/main/java/com/fastbee/iot/platform/ ├── domain/ │ ├── PlatformInstance +│ ├── PlatformUserAuth # 平台用户认证，Token/认证从这里获取 │ ├── PlatformInterface │ ├── PlatformFlow │ ├── PlatformFlowStep @@ -289,6 +337,7 @@ fastbee-service/fastbee-iot-service/src/main/java/com/fastbee/iot/platform/ │ └── PlatformPushLog ├── mapper/ │ ├── PlatformInstanceMapper +│ ├── PlatformUserAuthMapper # 用户认证 Mapper │ ├── PlatformInterfaceMapper │ ├── PlatformFlowMapper │ ├── PlatformFlowStepMapper @@ -296,30 +345,32 @@ fastbee-service/fastbee-iot-service/src/main/java/com/fastbee/iot/platform/ │ ├── PlatformDeviceSubMappingMapper │ └── PlatformPushLogMapper ├── service/ -│ ├── IPlatformInstanceService -│ ├── IPlatformFlowService +│ ├── IPlatformInstanceService # 仅保留 selectList +│ ├── IPlatformUserAuthService # 用户认证（新增，接管初始化/项目/Token） +│ ├── IPlatformFlowService # 移除 executeManually │ ├── IPlatformDeviceMappingService │ ├── IPlatformDeviceSubMappingService │ ├── IPlatformPushLogService │ └── impl/ ├── flow/ -│ ├── PlatformFlowExecutor -│ └── FlowContext +│ ├── PlatformFlowExecutor # Token 从 userAuth 获取 +│ └── FlowContext # platformUserAuth 为必填字段 ├── adapter/ │ └── dcp/ -│ ├── DcpPlatformAdapter +│ ├── DcpPlatformAdapter # 所有推送方法增加 userAuth 参数 │ └── StaticLevelBufferService ├── trigger/ -│ └── DeviceReportTrigger +│ └── DeviceReportTrigger # 加载 userAuth 传入适配器 └── job/ - └── PlatformPushScheduledJob + └── PlatformPushScheduledJob # 加载 userAuth 传入适配器 ``` ```text fastbee-open-api/src/main/java/com/fastbee/controller/platform/ -├── PlatformInstanceController # /platform/instance +├── PlatformInstanceController # /platform/instance（移除 initialize、getProjects） +├── PlatformUserAuthController # /platform/user-auth（新增，接管初始化/项目/用户管理） ├── PlatformInterfaceController # /platform/interface -├── PlatformFlowController # /platform/flow +├── PlatformFlowController # /platform/flow（移除 execute 端点） ├── PlatformDeviceMappingController # /platform/device-mapping └── PlatformPushLogController # /platform/push-log ``` @@ -338,18 +389,26 @@ execute(flow, context) 对每个 step： 1. 读取关联 platform_interface 2. 合并 request_schema 与 step.param_mapping - 3. 按 source 解析参数 + 3. 按 source 解析参数（优先从 userAuth 获取，fallback 到 platform.authConfig） 4. 若参数中存在 list-of-lists，展开为多批请求 5. 每批请求刷新 uuid 参数 6. 替换 path 中的 {变量} 7. GET 参数放 query；form 接口编码为 x-www-form-urlencoded；其他接口发 JSON body - 8. 非认证步骤自动注入 Authorization: Bearer {token} + 8. 非认证步骤自动注入 Authorization: Bearer {token}（从 userAuth 获取） 9. 执行 HTTP 10. 按 response_schema 抽取响应，写入 FlowContext - 11. 按 result_save 持久化 + 11. 按 result_save 持久化到 platform_user_auth 或 device_mapping 12. 异步写 platform_push_log，log_type=2 ``` +**FlowContext 构造：** `platformUserAuth` 是必填字段，所有构造函数都需要传入。旧的 `FlowContext(PlatformInstance)` 构造函数已移除。 + +**重要变更：** `persistResultSave` 不再向 `platform_instance` 表写 Token/过期时间，而是写入 `platform_user_auth` 表。`result_save` 配置中的 `saveTo` 路径应使用 `platform_user_auth.*` 前缀而非 `platform_instance.*`。 + +`auth` 和 `extra` 参数来源优先级： +- `auth source`：`PlatformUserAuth.auth_config` → `PlatformInstance.auth_config` +- `extra source`：`PlatformUserAuth.extra_config` → `PlatformInstance.extra_config` + 批量展开用于 `referenceSubDeviceSensorIdPairs` 这类参数。例如静力水准入网时，一个网关下多个 monitor 子设备会拆成多次绑定请求。 ### 5.2 设备入网 @@ -358,25 +417,28 @@ execute(flow, context) 直连设备： -1. 查找同平台、`flow_type=2`、`sensor_type` 匹配且启用的流程。 -2. 创建 `FlowContext(platform, mapping, null)`。 -3. 执行流程。 -4. 若主表已有 `platform_device_id`，将主绑定 `net_status` 改为 `1`。 +1. 根据设备绑定的 `platformUserAuthId` 查询用户认证信息。 +2. 查找同平台、`flow_type=2`、`sensor_type` 匹配且启用的流程。 +3. 创建 `FlowContext(platform, mapping, userAuth)`。 +4. 执行流程。 +5. 若主表已有 `platform_device_id`，将主绑定 `net_status` 改为 `1`。 网关设备： 1. 读取 `platform_device_sub_mapping`。 2. 筛选 `device_role=monitor` 且未入网的子设备。 -3. 对每个 monitor 创建独立 `FlowContext`： - - `currentMonitorSerialNumber = monitor.sub_serial_number` - - `currentMonitorDeviceName = monitor.sub_device_name` - - `currentMonitorIndex = monitor 在待入网 monitor 列表中的下标` - - `subMappingOnboardConfig = monitor.onboard_config` +3. 对每个 monitor 创建独立 `FlowContext`，通过构造函数一次性传入所有上下文字段： + - `platformUserAuth` = 设备绑定关联的用户认证 + - `currentMonitorSerialNumber`、`currentMonitorDeviceName` + - `currentMonitorIndex` = monitor 在待入网 monitor 列表中的下标 + - `subMappingOnboardConfig` = monitor.onboard_config 4. 执行同一个入网流程。若绑定接口 schema 使用 `source=onboard_config` 且 `value` 为数组，会按 `currentMonitorIndex` 拆出当前子设备的值。 5. 从上下文 `step1.id`、`step2.id` 取 pointId/deviceId，写回当前 monitor 子表。 6. 所有 monitor 都入网后，主绑定 `net_status=1`。 7. 若部分成功部分失败，已成功子设备会保留结果，接口返回明确失败信息。 +> **注意：** 入网不再校验 `platformProjectId` 是否为空。网关子设备 `FlowContext` 现在通过构造函数一次性设置所有字段，不再使用 `setXxx` 逐个设置。 + ### 5.3 设备上报触发 `DeviceReportTrigger.onDeviceReport` 使用异步线程池 `FastBeeConstant.TASK.PUSH_TO_THIRD_PLATFORM_ASYNC`。 @@ -399,8 +461,9 @@ WHERE s.sub_serial_number = #{serialNumber} AND m.enable = '1' 分发规则： -- `platform.code=dcp` 且 `device_type=0`：调用 `DcpPlatformAdapter.pushData`。 -- `platform.code=dcp` 且 `device_type=1`：先在子表中按上报序列号找 `PlatformDeviceSubMapping`，找到后调用 `pushGatewayData`。 +- 加载 `PlatformUserAuth`，未启用或无用户认证时跳过。 +- `platform.code=dcp` 且 `device_type=0`：调用 `DcpPlatformAdapter.pushData(platform, userAuth, mapping, ...)`。 +- `platform.code=dcp` 且 `device_type=1`：先在子表中按上报序列号找 `PlatformDeviceSubMapping`，找到后调用 `pushGatewayData(platform, userAuth, mapping, subMapping, ...)`。 - 未知平台 code 只记录 warning，不推送。 ### 5.4 定时推送调度器 @@ -428,10 +491,10 @@ public void execute() { `isDue` 判断：`push_interval` 转换为毫秒后，若 `nowMs - lastPushTime.getTime() >= intervalMs` 则触发。`lastPushTime` 为空时立即触发。 -`pushLatestData` 从 Redis 缓存读取设备最新数据，按 `device_type` 分发： +`pushLatestData` 加载 `PlatformUserAuth`，从 Redis 缓存读取设备最新数据，按 `device_type` 分发： -- 直连设备（`device_type=0`）：调用 `DcpPlatformAdapter.pushData`。 -- 网关设备（`device_type=1`）：遍历所有子设备，逐个从 Redis 读取并调用 `DcpPlatformAdapter.pushGatewayData`。 +- 直连设备（`device_type=0`）：调用 `DcpPlatformAdapter.pushData(platform, userAuth, mapping, ...)`。 +- 网关设备（`device_type=1`）：遍历所有子设备，逐个从 Redis 读取并调用 `DcpPlatformAdapter.pushGatewayData(platform, userAuth, mapping, subMapping, ...)`。 推送完成后调用 `IPlatformDeviceMappingService.updateLastPushTime` 更新 `last_push_time`。 @@ -443,8 +506,9 @@ public void execute() { ┌──────────────────────────────────────┐ │ PlatformPushScheduledJob（每5分钟） │ │ push_mode=2 且达到时间间隔 │ +│ → 加载 PlatformUserAuth │ │ → 从 Redis 读最新数据 │ -│ → 推送到平台 │ +│ → 推送到平台（传入 userAuth） │ │ → 更新 last_push_time │ └──────────────────────────────────────┘ ``` @@ -462,12 +526,17 @@ DCP 实时推送必须在设备绑定中配置 `push_interface_id`： - payload 支持 JSON 对象和数组两种格式。 - 数组格式按 `id` 或 `identifier` 转成 `{identifier: value}`，并把 `ts` 保存为 `{identifier}__ts`。 - 有 `request_schema` 时按 schema 组装。 -- 若 schema 组装结果为空，或没有任何 payload 驱动字段，则回退到 legacy 组装： - - `deviceId = platform_device_id` - - `epoch = 当前时间` - - 其他字段来自 `field_mapping` +- 若 schema 组装结果为空，或没有任何 payload 驱动字段，则回退到 legacy 组装。 - 普通推送请求体最终是数组：`[ {...} ]`。 +所有推送方法签名增加了 `PlatformUserAuth userAuth` 参数，用于 Token 获取： + +```java +// 变更后 +dcpAdapter.pushData(platform, userAuth, mapping, reportSerialNumber, payload); +dcpAdapter.pushGatewayData(platform, userAuth, mapping, subMapping, reportSerialNumber, payload); +``` + 时间字段会统一转为带时区的 ISO-8601 毫秒格式，支持输入毫秒时间戳、秒时间戳、ISO 字符串、本地时间字符串。 ### 5.6 网关与数组字段推送 @@ -509,7 +578,7 @@ DCP 实时推送必须在设备绑定中配置 `push_interface_id`： ] ``` -4. 若监测点先到或基准点缓存过期，监测点数据会写入 Redis monitor 缓存。当前代码已有 monitor 缓存读写能力，但推送路径主要依赖“监测点到达时基准点已就绪”的分支。 +4. 若监测点先到或基准点缓存过期，监测点数据会写入 Redis monitor 缓存。 缓存键： @@ -527,11 +596,11 @@ Redis TTL 为 `reportIntervalSec * 2`，读取时超过 `reportIntervalSec * 1.5 1. 查询当前绑定下所有 `device_role=monitor` 的子设备，顺序以 `platform_device_sub_mapping` 返回顺序为准（深部位移场景为浅到深）。 2. 当前上报子设备使用本次 payload，其余子设备从 Redis 读取缓存；读取 Redis 时使用子设备真实 `product_id + sub_serial_number`，避免误用网关主设备产品 ID。 3. 每个 monitor 组装为一个请求体对象，`deviceId` 使用该 monitor 入网后回填的 `platform_device_id`，`xTotalBias`、`yTotalBias` 等主体字段使用该 monitor 自身数据。 -4. 每个对象中的 `sensorData` 不是单点，而是从该 monitor 到末端 monitor 的后缀数组；深部位移中表现为“当前深度 + 更深点位”，顺序保持从浅到深。 +4. 每个对象中的 `sensorData` 不是单点，而是从该 monitor 到末端 monitor 的后缀数组；深部位移中表现为"当前深度 + 更深点位"，顺序保持从浅到深。 5. 最浅节点对象包含全孔从浅到深的点位数据，最深节点对象只包含自己。 6. 最终一次 HTTP 请求推送整个数组。 -批量推送不再从“数据推送接口”的 `request_schema` 读取入网参数 `sensorDepths`。`sensorDepths`、`holeDepth` 等仍属于深部位移入网/绑定接口配置；数据推送接口只需配置实际推送字段，例如 `deviceId`、`sensorData`、累计偏移量等。 +批量推送不再从"数据推送接口"的 `request_schema` 读取入网参数 `sensorDepths`。`sensorDepths`、`holeDepth` 等仍属于深部位移入网/绑定接口配置；数据推送接口只需配置实际推送字段，例如 `deviceId`、`sensorData`、累计偏移量等。 示例请求体结构（4 个串联点，从浅到深为 A/B/C/D）： @@ -570,13 +639,34 @@ Redis TTL 为 `reportIntervalSec * 2`，读取时超过 `reportIntervalSec * 1.5 配置与排查要点： -- 在平台接口管理中，为需要整孔推送的“数据推送接口”开启批量推送，即 `platform_interface.is_batch_push=1`。 +- 在平台接口管理中，为需要整孔推送的"数据推送接口"开启批量推送，即 `platform_interface.is_batch_push=1`。 - 推送接口 `request_schema` 必须包含实际数据推送字段；不要把入网接口的 `sensorDepths` 当作批量推送前置条件。 - 子设备必须已入网，monitor 子表行必须有 `platform_device_id`。 - Redis 中需要有各 monitor 子设备的最新物模型缓存；缺失数据的点位不会进入对应后缀 `sensorData`，若某 monitor 自身无数据则该 monitor 对象会被跳过。 - 推送日志可在 `platform_push_log` 查看；成功日志会出现 `[DCP] 批量推送完成`。 -### 5.8 日志与重试 +### 5.8 Token 管理 + +**架构变更：** Token 管理已从平台级迁移到用户级。`PlatformInstance` 上的 `access_token`、`token_expires_at` 字段不再使用。 + +`PlatformFlowExecutor.ensureToken(FlowContext ctx)`： + +- 从 `ctx.getPlatformUserAuth()` 获取用户认证信息。 +- `access_token` 为空，或 `token_expires_at` 距当前时间不足 5 分钟时视为过期。 +- 自动查找同平台 `flow_type=3` 且 `enable='1'` 的 Token 刷新流程。 +- 创建带有 `platform` 和 `userAuth` 的 `FlowContext` 执行刷新流程。 +- 刷新流程本身不注入 Authorization，避免递归。 +- 刷新成功后更新 `platform_user_auth.access_token` 和 `token_expires_at`。 +- Token 刷新流程最后步骤应配置 `result_key=tokenStep`，以便 `refreshToken` 方法提取 `access_token` 和 `expires_in`。 + +`DcpPlatformAdapter.ensureToken`（适配器层）： + +- 读取 `platformUserAuth.access_token` 和 `token_expires_at` 判断是否过期。 +- 过期时查找 `flow_type=3` 流程并执行 `flowExecutor.execute(tokenFlow, ctx)`。 +- 刷新后重新从数据库读取最新 `platform_user_auth` 记录。 +- 返回 `platformUserAuth.access_token`。 + +### 5.9 日志与重试 - 流程步骤日志：`log_type=2`。 - 实时数据推送日志：`log_type=1`。 @@ -601,7 +691,34 @@ Redis TTL 为 `reportIntervalSec * 2`，读取时超过 `reportIntervalSec * 1.5 - 查询平台列表。 - 新增、编辑、删除平台。 - 平台详情接口会清空 `access_token`，避免前端拿到 Token。 -- 初始化平台：`POST /platform/instance/{id}/initialize`。 +- `initStatus` 通过查询 `platform_user_auth` 表计算：存在有效用户则显示已初始化。 + +**用户管理：** + +点击"用户"按钮打开用户管理弹窗，后端接口：`/platform/user-auth` + +- 查询用户认证列表：`GET /platform/user-auth/list?platformId={id}` +- 新增用户：仅输入认证配置（用户名/密码/clientId/scope），`platformUserId` 和 `platformUserName` 为空 +- 初始化用户：`POST /platform/user-auth/{id}/initialize` + - 执行初始化流程获取 Token + - 从 Token 解析或调用接口获取 `userId` 和用户名 + - 自动回填 `platformUserId` 和 `platformUserName` + - 标记 `initStatus=1` +- 编辑用户：显示 `platformUserId` 和 `platformUserName`（只读） +- 删除用户：`DELETE /platform/user-auth/{ids}` + +用户初始化流程配置（flow_type=1）需要在 `result_save` 中指定： + +```json +[ + {"field": "access_token", "saveTo": "platform_user_auth.access_token"}, + {"field": "expires_in", "saveTo": "platform_user_auth.token_expires_at"}, + {"field": "userId", "saveTo": "platform_user_auth.platform_user_id"}, + {"field": "userName", "saveTo": "platform_user_auth.platform_user_name"} +] +``` + +> **注意：** `saveTo` 应使用 `platform_user_auth.*` 前缀而非旧的 `platform_instance.*`。 ### 6.2 接口定义 @@ -627,9 +744,10 @@ Redis TTL 为 `reportIntervalSec * 2`，读取时超过 `reportIntervalSec * 1.5 - 按平台查询流程。 - 新增、编辑流程及步骤。 - 查看流程详情时返回 `{flow, steps}`。 -- 手动执行：`POST /platform/flow/{id}/execute`，可选 `mappingId`。 - 步骤入参覆盖同样支持 `source=onboard_config`，语义与接口管理一致：配置 `value`，不配置子映射字段名。 +> **注意：** 流程手动执行按钮已移除，流程只能通过业务入口触发。 + ### 6.4 设备绑定 路径：`yack-web/src/views/platform/device-mapping/index.vue` @@ -638,7 +756,11 @@ Redis TTL 为 `reportIntervalSec * 2`，读取时超过 `reportIntervalSec * 1.5 主要能力： -- 管理平台、设备、传感器类型、设备类型、推送接口、字段映射。 +- 管理平台、用户、项目、设备、传感器类型、设备类型、推送接口、字段映射。 +- 级联选择（新增）： + - 选择平台 → 加载已初始化用户列表：`GET /platform/user-auth/initialized/{platformId}` + - 选择用户 → 加载项目列表：`GET /platform/user-auth/{id}/projects` + - 选择项目 → 回填 `platformProjectId` 和 `platformProjectName` - 入网：`POST /platform/device-mapping/{id}/onboard`。 - 子设备映射： - 查询：`GET /platform/device-mapping/{id}/sub-mappings` @@ -646,6 +768,10 @@ Redis TTL 为 `reportIntervalSec * 2`，读取时超过 `reportIntervalSec * 1.5 网关设备必须配置子映射，monitor 子设备才会执行入网并获得平台 deviceId。 +**级联选择接口权限：** 使用 `hasAnyPermi` 支持跨页面调用： +- `/platform/user-auth/initialized/{platformId}` - 允许 `platform:userAuth:list` 或 `platform:deviceMapping:*` +- `/platform/user-auth/{id}/projects` - 同上 + ### 6.5 推送日志 路径：`yack-web/src/views/platform/push-log/index.vue` @@ -690,7 +816,7 @@ SET @dcp_platform_id = (SELECT id FROM platform_instance WHERE code = 'dcp' LIMI ### 7.2 内置流程（6 个） | code | flow_type | sensor_type | 步骤 | -|---|---:|---|---| +|---|---|---|---:|---| | `dcp_init` | 1 | - | `dcp_auth` -> `dcp_get_user` -> `dcp_get_projects` | | `dcp_token_refresh` | 3 | - | `dcp_auth` | | `dcp_onboard_inclino` | 2 | `inclino` | `dcp_create_point` -> `dcp_bind_inclino` | @@ -704,7 +830,7 @@ SET @dcp_platform_id = (SELECT id FROM platform_instance WHERE code = 'dcp' LIMI - 深部位移绑定接口的 `sensorIds` 使用 `source=cascadedSubDeviceSensorIds`，`sensorDepths` 使用 `source=cascadedSubDeviceDepths`。 - 静力水准绑定接口的 `sensorIds` 使用 `source=referenceSubDeviceSensorIdPairs`。 - 静力水准推送接口定义了数组字段 `sensorData`，其 `itemSchema` 包含 `epoch`、`height`、`temperature`、`sensorId`。 -- 轴力计和雨量计 DCP 文档要求累计值递增；当前代码未在适配器中实现“查询上次成功值并跳过非递增值”的通用校验，若目标平台严格校验，需要在推送前补充该逻辑或依赖平台返回失败日志。 +- 轴力计和雨量计 DCP 文档要求累计值递增；当前代码未在适配器中实现"查询上次成功值并跳过非递增值"的通用校验，若目标平台严格校验，需要在推送前补充该逻辑或依赖平台返回失败日志。 ### 7.4 深部位移计扩展配置 @@ -739,7 +865,7 @@ DCP 深部位移绑定接口： ] ``` -网关多子设备绑定深部位移计时，子映射表中的 monitor 顺序表示从浅到深。`sensorDepths.value` 也按浅到深配置完整孔深数组，例如 `[1.7,3.4,5.1,6.8]`。流程执行器会按当前 monitor 截取“当前节点 + 更深节点”，再倒序成 DCP 要求的“最深到当前”： +网关多子设备绑定深部位移计时，子映射表中的 monitor 顺序表示从浅到深。`sensorDepths.value` 也按浅到深配置完整孔深数组，例如 `[1.7,3.4,5.1,6.8]`。流程执行器会按当前 monitor 截取"当前节点 + 更深节点"，再倒序成 DCP 要求的"最深到当前"： - 第 1 个 monitor（1.7m）：`sensorDepths=[6.8,5.1,3.4,1.7]`，`sensorIds=[6.8m sensorId,5.1m sensorId,3.4m sensorId,1.7m sensorId]`。 - 第 2 个 monitor（3.4m）：`sensorDepths=[6.8,5.1,3.4]`，`sensorIds=[6.8m sensorId,5.1m sensorId,3.4m sensorId]`。 @@ -768,14 +894,20 @@ DCP 深部位移绑定接口： ## 9. 当前实现限制与注意事项 +- **Token 迁移：** Token/认证已从 `PlatformInstance`（平台级）迁移到 `PlatformUserAuth`（用户级）。旧的 `result_save` 配置如果使用了 `platform_instance.*` 前缀，需要更新为 `platform_user_auth.*`。`PlatformInstance.access_token`/`token_expires_at` 字段已废弃。 +- **Service 简化：** `PlatformInstanceServiceImpl` 仅保留 `selectList`；初始化、Token 管理、项目查询等功能已迁移到 `PlatformUserAuthServiceImpl`。 +- **Controller 拆分：** `PlatformInstanceController` 移除了 `initialize` 和 `getProjects` 端点，这些功能已迁移到 `PlatformUserAuthController`。 +- **FlowContext 变更：** `platformUserAuth` 现在是必填字段。旧的 `FlowContext(PlatformInstance)` 构造函数已移除。 +- **流程手动执行移除：** `POST /platform/flow/{id}/execute` 端点已移除，流程只能通过业务入口（初始化/入网/Token 刷新）触发。 +- **入网校验变更：** 入网时不再校验 `platformProjectId` 是否为空，改为校验 `platformUserAuth` 是否存在。 - 当前适配器只有 DCP。新增平台需要新增 adapter，并在 `DeviceReportTrigger.dispatchPush` 中按平台 `code` 路由。 -- 文档中的“可视化配置”已覆盖平台、接口、流程、设备绑定、子映射和日志，但平台适配器路由仍需要代码扩展。 +- 文档中的"可视化配置"已覆盖平台、接口、流程、设备绑定、子映射和日志，但平台适配器路由仍需要代码扩展。 - DCP 实时推送必须配置 `push_interface_id`，未配置时不会按传感器类型自动回退路径。 - 实时推送只支持 POST JSON，流程执行支持 GET、form、JSON。 - `idempotency_key` 字段已建表，但实时推送未生成幂等键。 -- `PlatformInstanceController.getInfo` 只清空 `access_token`，注释提到密码脱敏，但当前代码未改写 `auth_config` 内密码。 +- `PlatformUserAuthController.getInfo` 会清空 `access_token`。 - `platform_device_mapping.platform_device_id` 对网关设备不是主要数据来源；网关 monitor 的平台 deviceId 保存在 `platform_device_sub_mapping.platform_device_id`。 -- 静力水准 monitor 先到时会缓存 monitor 数据，但当前主推送分支是在 monitor 到达时读取基准点缓存；若要实现“基准点后到自动补推所有等待 monitor”，需要在基准点上报分支补充扫描 monitor 缓存并推送。 +- 静力水准 monitor 先到时会缓存 monitor 数据，但当前主推送分支是在 monitor 到达时读取基准点缓存；若要实现"基准点后到自动补推所有等待 monitor"，需要在基准点上报分支补充扫描 monitor 缓存并推送。 - 定时推送依赖 Redis 缓存数据，若设备数据未及时写入 Redis，定时推送可能推送过期数据或无数据可推。 - 定时推送网关设备时会遍历所有子设备，大量子设备场景需关注性能和外部接口调用频率。 - 定时推送当前使用 `@Scheduled(fixedDelay)`，不支持 Cron 表达式；如需灵活调度需引入 Quartz。 diff --git a/fastbee-open-api/src/main/java/com/fastbee/controller/platform/PlatformFlowController.java b/fastbee-open-api/src/main/java/com/fastbee/controller/platform/PlatformFlowController.java index 818831e6..e2b8596a 100644 --- a/fastbee-open-api/src/main/java/com/fastbee/controller/platform/PlatformFlowController.java +++ b/fastbee-open-api/src/main/java/com/fastbee/controller/platform/PlatformFlowController.java @@ -71,14 +71,6 @@ public class PlatformFlowController extends BaseController { return success(); } - @PreAuthorize("@ss.hasPermi('platform:flow:edit')") - @PostMapping("/{id}/execute") - @ApiOperation("手动执行流程") - public AjaxResult execute(@PathVariable Long id, @RequestParam(required = false) Long mappingId) { - flowService.executeManually(id, mappingId); - return success("执行完成"); - } - @Data public static class FlowWithSteps { private PlatformFlow flow; diff --git a/fastbee-open-api/src/main/java/com/fastbee/controller/platform/PlatformInstanceController.java b/fastbee-open-api/src/main/java/com/fastbee/controller/platform/PlatformInstanceController.java index 9fdf5bc7..91ff8cbd 100644 --- a/fastbee-open-api/src/main/java/com/fastbee/controller/platform/PlatformInstanceController.java +++ b/fastbee-open-api/src/main/java/com/fastbee/controller/platform/PlatformInstanceController.java @@ -11,6 +11,7 @@ import org.springframework.beans.factory.annotation.Autowired; import org.springframework.security.access.prepost.PreAuthorize; import org.springframework.web.bind.annotation.*; +import javax.websocket.server.PathParam; import java.util.List; @RestController @@ -65,19 +66,4 @@ public class PlatformInstanceController extends BaseController { } return success(); } - - @PreAuthorize("@ss.hasPermi('platform:instance:edit')") - @PostMapping("/{id}/initialize") - @ApiOperation("初始化平台（执行初始化流程，获取userId）") - public AjaxResult initialize(@PathVariable Long id) { - instanceService.initialize(id); - return success("初始化成功"); - } - - @PreAuthorize("@ss.hasAnyPermi('platform:instance:list,platform:deviceMapping:list,platform:deviceMapping:add,platform:deviceMapping:edit')") - @GetMapping("/{id}/projects") - @ApiOperation("查询平台项目列表") - public AjaxResult getProjects(@PathVariable Long id) { - return success(instanceService.getProjectsByPlatformId(id)); - } } diff --git a/fastbee-open-api/src/main/java/com/fastbee/controller/platform/PlatformUserAuthController.java b/fastbee-open-api/src/main/java/com/fastbee/controller/platform/PlatformUserAuthController.java new file mode 100644 index 00000000..b00ba9cd --- /dev/null +++ b/fastbee-open-api/src/main/java/com/fastbee/controller/platform/PlatformUserAuthController.java @@ -0,0 +1,92 @@ +package com.fastbee.controller.platform; + +import com.fastbee.common.core.controller.BaseController; +import com.fastbee.common.core.domain.AjaxResult; +import com.fastbee.common.core.page.TableDataInfo; +import com.fastbee.iot.platform.domain.PlatformUserAuth; +import com.fastbee.iot.platform.service.IPlatformUserAuthService; +import io.swagger.annotations.Api; +import io.swagger.annotations.ApiOperation; +import org.springframework.beans.factory.annotation.Autowired; +import org.springframework.security.access.prepost.PreAuthorize; +import org.springframework.web.bind.annotation.*; + +import java.util.List; + +@RestController +@RequestMapping("/platform/user-auth") +@Api(tags = "平台用户认证") +public class PlatformUserAuthController extends BaseController { + + @Autowired + private IPlatformUserAuthService userAuthService; + + @PreAuthorize("@ss.hasPermi('platform:userAuth:list')") + @GetMapping("/list") + @ApiOperation("查询用户认证列表") + public TableDataInfo list(PlatformUserAuth query) { + startPage(); + List list = userAuthService.selectList(query); + return getDataTable(list); + } + + @PreAuthorize("@ss.hasAnyPermi('platform:userAuth:list,platform:deviceMapping:list,platform:deviceMapping:add,platform:deviceMapping:edit')") + @GetMapping("/initialized/{platformId}") + @ApiOperation("查询平台下已初始化的用户列表") + public AjaxResult getInitializedUsers(@PathVariable Long platformId) { + List list = userAuthService.selectInitializedByPlatformId(platformId); + return success(list); + } + + @PreAuthorize("@ss.hasPermi('platform:userAuth:query')") + @GetMapping("/{id}") + @ApiOperation("获取用户认证详情") + public AjaxResult getInfo(@PathVariable Long id) { + PlatformUserAuth userAuth = userAuthService.getById(id); + // 脱敏：不返回 access_token + if (userAuth != null) { + userAuth.setAccessToken(null); + } + return success(userAuth); + } + + @PreAuthorize("@ss.hasPermi('platform:userAuth:add')") + @PostMapping + @ApiOperation("新增用户认证") + public AjaxResult add(@RequestBody PlatformUserAuth userAuth) { + return toAjax(userAuthService.save(userAuth)); + } + + @PreAuthorize("@ss.hasPermi('platform:userAuth:edit')") + @PutMapping + @ApiOperation("修改用户认证") + public AjaxResult edit(@RequestBody PlatformUserAuth userAuth) { + return toAjax(userAuthService.updateById(userAuth)); + } + + @PreAuthorize("@ss.hasPermi('platform:userAuth:remove')") + @DeleteMapping("/{ids}") + @ApiOperation("删除用户认证") + public AjaxResult remove(@PathVariable Long[] ids) { + for (Long id : ids) { + userAuthService.removeById(id); + } + return success(); + } + + @PreAuthorize("@ss.hasPermi('platform:userAuth:initialize')") + @PostMapping("/{id}/initialize") + @ApiOperation("初始化用户认证（执行初始化流程，获取platformUserId）") + public AjaxResult initialize(@PathVariable Long id) { + userAuthService.initialize(id); + return success("初始化成功"); + } + + @PreAuthorize("@ss.hasAnyPermi('platform:userAuth:list,platform:deviceMapping:list,platform:deviceMapping:add,platform:deviceMapping:edit')") + @GetMapping("/{id}/projects") + @ApiOperation("查询用户的项目列表") + public AjaxResult getProjects(@PathVariable Long id) { + return success(userAuthService.getProjectsByUserAuthId(id)); + } + +} \ No newline at end of file diff --git a/fastbee-plugs/fastbee-mqtt-client/src/main/java/com/fastbee/mqttclient/PubMqttCallBack.java b/fastbee-plugs/fastbee-mqtt-client/src/main/java/com/fastbee/mqttclient/PubMqttCallBack.java index a2a15f0c..4f696d84 100644 --- a/fastbee-plugs/fastbee-mqtt-client/src/main/java/com/fastbee/mqttclient/PubMqttCallBack.java +++ b/fastbee-plugs/fastbee-mqtt-client/src/main/java/com/fastbee/mqttclient/PubMqttCallBack.java @@ -96,13 +96,13 @@ public class PubMqttCallBack implements MqttCallbackExtended { @Override public void connectComplete(boolean reconnect, String serverURI) { log.info("MQTT内部客户端已经连接!"); - System.out.print( + System.out.print("" + "▌▌▌▌ ▌▌▌▌ ▌▌▌▌▌▌ ▍▌▌▌▌▍ ▌▌▌ ▌▌▌▌ ▌▌▌▌▌▍ ▌▌▌▌\n" + " ▌▌▌▌▎▌▌▌▌ ▌▌▌ ▌▌▌ ▍▌▌▌▌▌▌▎ ▌▌▌ ▌▌▌▎▌▌ ▌▌▌▌▌▌▌ ▌▌▌▌\n" + " ▌▌▌▌▌▌ ▌▌▌▌ ▌▌▌ ▍▌▌▌▏▌▌▌▌ ▌▌▌ ▌▌▌ ▌▌▌ ▌▌▌▌▌▌▌▌ ▌▌▌▌\n" + " ▌▌▌▍ ▌▌▌▍▍▍▌▌▌▌ ▌▌▌▌ ▌▌▌▌▌▌▌ ▌▌▌▌▍▍▍▌▌▌ ▌▌▌▌ ▌▌▌▌▌▌▌▌\n" + " ▌▌▌▏ ▌▌▌▌▌▌▌▌▌▌▌▌ ▌▌▌▌ ▌▌▌▌▌▌ ▌▌▌▌▌▌▌▌▌▌▌▌ ▌▌▌▌ ▌▌▌▌▌▌\n" + - " ▌▌▌▏ ▌▌▌▌ ▌▌▌▌ ▌▌▌▌ ▌▌▌▌▌ ▌▌▌▌ ▌▌▌▌ ▌▌▌▌ ▌▌▌▌▌" + " ▌▌▌▏ ▌▌▌▌ ▌▌▌▌ ▌▌▌▌ ▌▌▌▌▌ ▌▌▌▌ ▌▌▌▌ ▌▌▌▌ ▌▌▌▌▌\n" ); 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 bd81756b..2d03498d 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 @@ -13,16 +13,12 @@ import com.fastbee.http.service.HttpClientFactory; import com.fastbee.iot.cache.ITSLCache; import com.fastbee.iot.domain.Device; import com.fastbee.iot.model.ThingsModels.ValueItem; -import com.fastbee.iot.platform.domain.PlatformDeviceMapping; -import com.fastbee.iot.platform.domain.PlatformDeviceSubMapping; -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.domain.*; 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.mapper.PlatformUserAuthMapper; import com.fastbee.iot.platform.service.IPlatformDeviceSubMappingService; import com.fastbee.iot.platform.service.IPlatformInstanceService; import com.fastbee.iot.platform.service.IPlatformPushLogService; @@ -38,7 +34,6 @@ import java.time.OffsetDateTime; import java.time.ZoneId; import java.time.format.DateTimeFormatter; import java.time.format.DateTimeParseException; -import java.util.ArrayList; import java.util.Date; import java.util.HashMap; import java.util.LinkedHashMap; @@ -69,6 +64,11 @@ public class DcpPlatformAdapter { 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 PlatformInterfaceMapper interfaceMapper; + @Autowired + private PlatformUserAuthMapper platformUserAuthMapper; + @Autowired private IPlatformInstanceService instanceService; @Autowired @@ -78,8 +78,6 @@ public class DcpPlatformAdapter { @Autowired private StaticLevelBufferService bufferService; @Autowired - private PlatformInterfaceMapper interfaceMapper; - @Autowired private IPlatformDeviceSubMappingService subMappingService; @Autowired private ITSLCache itslCache; @@ -89,7 +87,7 @@ public class DcpPlatformAdapter { /** * 推送设备上报数据到 DCP（直连设备） */ - public void pushData(PlatformInstance platform, PlatformDeviceMapping mapping, + public void pushData(PlatformInstance platform, PlatformUserAuth userAuth, PlatformDeviceMapping mapping, String reportSerialNumber, String payload) { if (mapping.getNetStatus() == null || mapping.getNetStatus() != 1) { log.warn("[DCP] 设备 {} 未入网，跳过推送", mapping.getSerialNumber()); @@ -102,7 +100,7 @@ public class DcpPlatformAdapter { try { String pushPath = resolvePushPath(mapping, pushInterface); - handleNormalPush(platform, mapping, reportSerialNumber, payload, pushPath, interfaceCode, requestSchema); + handleNormalPush(platform, userAuth, mapping, reportSerialNumber, payload, pushPath, interfaceCode, requestSchema); } catch (Exception e) { String serialNumber = StringUtils.hasText(reportSerialNumber) ? reportSerialNumber : mapping.getSerialNumber(); @@ -121,7 +119,7 @@ public class DcpPlatformAdapter { * 数组字段由 requestSchema.arrayMode 控制组包方式： * single 只推当前子设备，reference_pair 组装 [refItem, monitorItem] 推送。 */ - public void pushGatewayData(PlatformInstance platform, PlatformDeviceMapping mapping, + public void pushGatewayData(PlatformInstance platform, PlatformUserAuth userAuth, PlatformDeviceMapping mapping, PlatformDeviceSubMapping subMapping, String reportSerialNumber, String payload) { if (mapping.getNetStatus() == null || mapping.getNetStatus() != 1) { @@ -142,14 +140,14 @@ public class DcpPlatformAdapter { // 判断是否批量推送 if (pushInterface.getIsBatchPush() != null && pushInterface.getIsBatchPush() == 1) { - pushBatchSubDevicesData(platform, mapping, subMapping, reportSerialNumber, payload, pushPath, interfaceCode, requestSchema); + pushBatchSubDevicesData(platform, userAuth, mapping, subMapping, reportSerialNumber, payload, pushPath, interfaceCode, requestSchema); return; } // 无数组字段 → 简单推送，使用子设备 platformDeviceId 和主表 fieldMapping if (!hasArrayField(requestSchema)) { // 如果只有一个设备完全可以走 - handleSubDeviceSimplePush(platform, mapping, subMapping, reportSerialNumber, payload, pushPath, interfaceCode, requestSchema); + handleSubDeviceSimplePush(platform, userAuth, mapping, subMapping, reportSerialNumber, payload, pushPath, interfaceCode, requestSchema); return; } @@ -159,7 +157,7 @@ public class DcpPlatformAdapter { // 解析 extra_mapping 中的 itemFieldMappings 和 reportIntervalSec int reportIntervalSec = 300; JSONArray itemFieldMappings = null; - String fieldName = resolveArrayFieldName(requestSchema, null); + String fieldName = resolveArrayFieldName(requestSchema); JSONObject extra = null; if (StringUtils.hasText(mapping.getExtraMapping())) { extra = JSON.parseObject(mapping.getExtraMapping()); @@ -190,20 +188,18 @@ public class DcpPlatformAdapter { return; } - JSONObject currentBodyFields = buildObjectBySchema(platform, mapping, requestSchema, reportData, fieldMappings); + JSONObject currentBodyFields = buildObjectBySchema(platform, userAuth, mapping, requestSchema, reportData, fieldMappings); // 单点数组推送：sensorData 中只包含当前上报子设备这一条数据。 if (ARRAY_MODE_SINGLE.equals(arrayMode)) { - pushSingleArrayData(platform, mapping, subMapping, reportSerialNumber, baseUrl, - interfaceCode, fieldName, currentItem, currentBodyFields); + pushSingleArrayData(platform, userAuth, mapping, subMapping, reportSerialNumber, baseUrl, interfaceCode, fieldName, currentItem, currentBodyFields); return; } if (subMapping.isReference()) { // 基准点上报：缓存当前基准点，并补推此前已缓存的监测点 - bufferService.saveReferenceData( - platform.getId(), fieldName, reportSerialNumber, currentItem, reportIntervalSec); - int replayCount = replayPendingMonitorData(platform, mapping, fieldName, currentItem, + bufferService.saveReferenceData(platform.getId(), fieldName, reportSerialNumber, currentItem, reportIntervalSec); + int replayCount = replayPendingMonitorData(platform, userAuth, mapping, fieldName, currentItem, reportIntervalSec, baseUrl, interfaceCode); log.info("[DCP] 基准点 {} 已缓存，补推监测点 {} 条", reportSerialNumber, replayCount); return; @@ -219,8 +215,7 @@ public class DcpPlatformAdapter { // 获取基准点序列号，优先使用子映射中当前标记的 reference 行 String refSerialNumber = resolveRefSerialNumber(subMapping, mapping); - JSONObject refData = bufferService.getReferenceDataIfFresh( - platform.getId(), fieldName, refSerialNumber, reportIntervalSec); + JSONObject refData = bufferService.getReferenceDataIfFresh(platform.getId(), fieldName, refSerialNumber, reportIntervalSec); if (refData != null && refData.getJSONObject("itemData") != null) { JSONArray sensorData = new JSONArray(); sensorData.add(refData.getJSONObject("itemData")); @@ -228,12 +223,11 @@ public class DcpPlatformAdapter { JSONArray pushBody = buildArrayPushBody(mapping, currentBodyFields, fieldName, sensorData, monitorDeviceId); PlatformPushLog logEntry = createLogEntry(platform, mapping, interfaceCode, reportSerialNumber); - doHttpPost(platform, baseUrl, JSON.toJSONString(pushBody), logEntry); + doHttpPost(platform, userAuth, mapping, baseUrl, JSON.toJSONString(pushBody), logEntry); pushLogService.asyncSave(logEntry); log.info("[DCP] 监测点 {} 基准点就绪，已推送 deviceId={}", reportSerialNumber, monitorDeviceId); } else { - bufferService.saveMonitorData( - platform.getId(), fieldName, reportSerialNumber, currentItem, currentBodyFields, reportIntervalSec); + bufferService.saveMonitorData(platform.getId(), fieldName, reportSerialNumber, currentItem, currentBodyFields, reportIntervalSec); log.info("[DCP] 监测点 {} 基准点未就绪，已缓存等待", reportSerialNumber); } } catch (Exception e) { @@ -245,7 +239,7 @@ public class DcpPlatformAdapter { } } - private void pushSingleArrayData(PlatformInstance platform, PlatformDeviceMapping mapping, + private void pushSingleArrayData(PlatformInstance platform, PlatformUserAuth userAuth, PlatformDeviceMapping mapping, PlatformDeviceSubMapping subMapping, String reportSerialNumber, String baseUrl, String interfaceCode, String fieldName, JSONObject currentItem, JSONObject currentBodyFields) { @@ -259,7 +253,7 @@ public class DcpPlatformAdapter { JSONArray pushBody = buildArrayPushBody(mapping, currentBodyFields, fieldName, sensorData, deviceId); PlatformPushLog logEntry = createLogEntry(platform, mapping, interfaceCode, reportSerialNumber); - doHttpPost(platform, baseUrl, JSON.toJSONString(pushBody), logEntry); + doHttpPost(platform, userAuth, mapping, baseUrl, JSON.toJSONString(pushBody), logEntry); pushLogService.asyncSave(logEntry); log.info("[DCP] 子设备 {} 单点数组已推送 deviceId={}", reportSerialNumber, deviceId); } @@ -280,10 +274,7 @@ public class DcpPlatformAdapter { return false; } - private String resolveArrayFieldName(JSONArray requestSchema, String fallback) { - if (StringUtils.hasText(fallback)) { - return fallback; - } + private String resolveArrayFieldName(JSONArray requestSchema) { JSONObject firstArrayField = findFirstArrayFieldSchema(requestSchema); if (firstArrayField != null && StringUtils.hasText(firstArrayField.getString("name"))) { return firstArrayField.getString("name"); @@ -335,7 +326,7 @@ public class DcpPlatformAdapter { /** * 基准点后到时，补推此前缓存的监测点数据。 */ - private int replayPendingMonitorData(PlatformInstance platform, PlatformDeviceMapping mapping, + private int replayPendingMonitorData(PlatformInstance platform, PlatformUserAuth userAuth,PlatformDeviceMapping mapping, String fieldName, JSONObject referenceItem, int reportIntervalSec, String baseUrl, String interfaceCode) { List subMappings = subMappingService.selectByParentId(mapping.getId()); @@ -375,7 +366,7 @@ public class DcpPlatformAdapter { sensorData, monitor.getPlatformDeviceId()); PlatformPushLog logEntry = createLogEntry(platform, mapping, interfaceCode, monitor.getSubSerialNumber()); - doHttpPost(platform, baseUrl, JSON.toJSONString(pushBody), logEntry); + doHttpPost(platform, userAuth, mapping, baseUrl, JSON.toJSONString(pushBody), logEntry); pushLogService.asyncSave(logEntry); if (logEntry.getStatus() != null && logEntry.getStatus() == 1) { bufferService.removeMonitorData(platform.getId(), fieldName, monitor.getSubSerialNumber()); @@ -392,20 +383,20 @@ public class DcpPlatformAdapter { // ==================== 普通推送 ==================== - private void handleNormalPush(PlatformInstance platform, PlatformDeviceMapping mapping, + private void handleNormalPush(PlatformInstance platform, PlatformUserAuth userAuth, PlatformDeviceMapping mapping, String reportSerialNumber, String payload, String pushPath, String interfaceCode, JSONArray requestSchema) { - Object body = buildPushBody(platform, mapping, payload, requestSchema); + Object body = buildPushBody(platform, userAuth, 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); + doHttpPost(platform, userAuth, mapping, platform.getBaseUrl() + pushPath, JSON.toJSONString(body), logEntry); pushLogService.asyncSave(logEntry); } - private Object buildPushBody(PlatformInstance platform, PlatformDeviceMapping mapping, + private Object buildPushBody(PlatformInstance platform, PlatformUserAuth userAuth,PlatformDeviceMapping mapping, String payload, JSONArray requestSchema) { if (!StringUtils.hasText(payload)) { return null; @@ -420,7 +411,7 @@ public class DcpPlatformAdapter { return buildLegacyPushBody(mapping.getPlatformDeviceId(), fieldMappings, reportData); } - JSONObject dcpData = buildObjectBySchema(platform, mapping, requestSchema, reportData, fieldMappings); + JSONObject dcpData = buildObjectBySchema(platform, userAuth, mapping, requestSchema, reportData, fieldMappings); if (dcpData.isEmpty() || hasPayloadDrivenValue(dcpData, requestSchema)) { return buildLegacyPushBody(mapping.getPlatformDeviceId(), fieldMappings, reportData); } @@ -466,18 +457,16 @@ public class DcpPlatformAdapter { *

* 使用子设备 platformDeviceId 和主表 fieldMapping 组装请求体 */ - private void handleSubDeviceSimplePush(PlatformInstance platform, PlatformDeviceMapping mapping, - PlatformDeviceSubMapping subMapping, - String reportSerialNumber, String payload, - String pushPath, String interfaceCode, - JSONArray requestSchema) { - Object body = buildSubDevicePushBody(platform, mapping, subMapping, payload, requestSchema); + private void handleSubDeviceSimplePush(PlatformInstance platform, PlatformUserAuth userAuth, PlatformDeviceMapping mapping, + PlatformDeviceSubMapping subMapping, String reportSerialNumber, String payload, + String pushPath, String interfaceCode, JSONArray requestSchema) { + Object body = buildSubDevicePushBody(platform, userAuth, mapping, subMapping, payload, requestSchema); if (body == null) { log.warn("[DCP] 网关子设备 {} 字段映射结果为空，跳过推送", reportSerialNumber); return; } PlatformPushLog logEntry = createLogEntry(platform, mapping, interfaceCode, reportSerialNumber); - doHttpPost(platform, platform.getBaseUrl() + pushPath, JSON.toJSONString(body), logEntry); + doHttpPost(platform, userAuth,mapping, platform.getBaseUrl() + pushPath, JSON.toJSONString(body), logEntry); pushLogService.asyncSave(logEntry); } @@ -486,7 +475,7 @@ public class DcpPlatformAdapter { *

* 复用主表 fieldMapping 进行物模型字段映射，deviceId 取自子映射行 */ - private Object buildSubDevicePushBody(PlatformInstance platform, PlatformDeviceMapping mapping, + private Object buildSubDevicePushBody(PlatformInstance platform, PlatformUserAuth userAuth, PlatformDeviceMapping mapping, PlatformDeviceSubMapping subMapping, String payload, JSONArray requestSchema) { JSONObject reportData = parsePayloadToMap(payload); @@ -503,7 +492,7 @@ public class DcpPlatformAdapter { return buildLegacyPushBody(platformDeviceId, fieldMappings, reportData); } - JSONObject dcpData = buildObjectBySchema(platform, mapping, requestSchema, reportData, fieldMappings); + JSONObject dcpData = buildObjectBySchema(platform, userAuth, mapping, requestSchema, reportData, fieldMappings); if (dcpData.isEmpty() || hasPayloadDrivenValue(dcpData, requestSchema)) { return buildLegacyPushBody(platformDeviceId, fieldMappings, reportData); @@ -538,7 +527,7 @@ public class DcpPlatformAdapter { return arr; } - private JSONObject buildObjectBySchema(PlatformInstance platform, PlatformDeviceMapping mapping, + private JSONObject buildObjectBySchema(PlatformInstance platform, PlatformUserAuth userAuth, PlatformDeviceMapping mapping, JSONArray requestSchema, JSONObject reportData, JSONArray fieldMappings) { JSONObject body = new JSONObject(); @@ -554,7 +543,7 @@ public class DcpPlatformAdapter { if (!StringUtils.hasText(fieldName)) { continue; } - Object value = resolveSchemaFieldValue(platform, mapping, schemaField, reportData, fieldMappings); + Object value = resolveSchemaFieldValue(platform, userAuth, mapping, schemaField, reportData, fieldMappings); if (value != null) { body.put(fieldName, value); } @@ -562,7 +551,7 @@ public class DcpPlatformAdapter { return body; } - private Object resolveSchemaFieldValue(PlatformInstance platform, PlatformDeviceMapping mapping, + private Object resolveSchemaFieldValue(PlatformInstance platform, PlatformUserAuth userAuth, PlatformDeviceMapping mapping, JSONObject schemaField, JSONObject reportData, JSONArray fieldMappings) { String source = schemaField.getString("source"); @@ -577,10 +566,10 @@ public class DcpPlatformAdapter { return resolveDeviceField(mapping, configField); } if ("auth".equals(source)) { - return resolveJsonField(platform.getAuthConfig(), configField); + return resolveJsonField(userAuth.getAuthConfig(), configField); } if ("extra".equals(source)) { - return resolveJsonField(platform.getExtraConfig(), configField); + return resolveJsonField(userAuth.getExtraConfig(), configField); } if (isThingsModelSource(source)) { return resolveThingsModelValue(reportData, fieldMappings, fieldName, contextKey); @@ -776,8 +765,7 @@ public class DcpPlatformAdapter { } String sensorType = mapping == null ? null : mapping.getSensorType(); String serialNumber = mapping == null ? null : mapping.getSerialNumber(); - throw new RuntimeException("未配置推送接口，请在设备绑定中选择推送接口" - + "，设备=" + serialNumber + "，传感器类型=" + sensorType); + throw new RuntimeException("未配置推送接口，请在设备绑定中选择推送接口" + "，设备=" + serialNumber + "，传感器类型=" + sensorType); } private String resolveInterfaceCode(String sensorType, PlatformInterface pushInterface) { @@ -917,13 +905,13 @@ public class DcpPlatformAdapter { // ==================== HTTP 执行 ==================== - private void doHttpPost(PlatformInstance platform, String url, String bodyStr, PlatformPushLog logEntry) { + private void doHttpPost(PlatformInstance platform, PlatformUserAuth userAuth, PlatformDeviceMapping mapping, String url, String bodyStr, PlatformPushLog logEntry) { long startMs = System.currentTimeMillis(); logEntry.setRequestUrl(url); logEntry.setRequestBody(bodyStr); try { - String token = ensureToken(platform); + String token = ensureToken(platform, userAuth, mapping); Map headers = new HashMap<>(); headers.put("Authorization", "Bearer " + token); headers.put("Content-Type", "application/json"); @@ -1080,7 +1068,7 @@ public class DcpPlatformAdapter { * 当前上报的子设备使用传入的 payload（最新数据），其余子设备从 Redis 缓存读取。 * 推送顺序：从浅到深（与绑定时从深到浅相反）。 */ - private void pushBatchSubDevicesData(PlatformInstance platform, PlatformDeviceMapping mapping, + private void pushBatchSubDevicesData(PlatformInstance platform, PlatformUserAuth userAuth, PlatformDeviceMapping mapping, PlatformDeviceSubMapping currentSubMapping, String reportSerialNumber, String payload, String pushPath, String interfaceCode, JSONArray requestSchema) { @@ -1117,7 +1105,7 @@ public class DcpPlatformAdapter { // 3. 准备公共参数（复用已有 fieldMapping + requestSchema） JSONArray fieldMappings = parseJsonArray(mapping.getFieldMapping()); - String fieldName = resolveArrayFieldName(requestSchema, null); + String fieldName = resolveArrayFieldName(requestSchema); JSONObject extra = null; if (StringUtils.hasText(mapping.getExtraMapping())) { extra = JSON.parseObject(mapping.getExtraMapping()); @@ -1190,7 +1178,7 @@ public class DcpPlatformAdapter { continue; } - JSONObject bodyFields = buildObjectBySchema(platform, mapping, requestSchema, reportData, fieldMappings); + JSONObject bodyFields = buildObjectBySchema(platform, userAuth, mapping, requestSchema, reportData, fieldMappings); JSONArray deviceBodyArr = buildArrayPushBody(mapping, bodyFields, fieldName, sensorDataArr, sub.getPlatformDeviceId()); if (!deviceBodyArr.isEmpty()) { @@ -1207,7 +1195,7 @@ public class DcpPlatformAdapter { // 7. 一次 HTTP 请求推送 String baseUrl = platform.getBaseUrl() + pushPath; PlatformPushLog logEntry = createLogEntry(platform, mapping, interfaceCode, reportSerialNumber); - doHttpPost(platform, baseUrl, JSON.toJSONString(pushBodyArray), logEntry); + doHttpPost(platform, userAuth, mapping, baseUrl, JSON.toJSONString(pushBodyArray), logEntry); pushLogService.asyncSave(logEntry); log.info("[DCP] 批量推送完成 mappingId={} 当前子设备={} 推送设备对象数={}/{}", @@ -1281,33 +1269,30 @@ public class DcpPlatformAdapter { // ==================== Token 管理 ==================== - private String ensureToken(PlatformInstance platform) { + private String ensureToken(PlatformInstance platform, PlatformUserAuth platformUserAuth, PlatformDeviceMapping mapping) { + + // 校验是否过期 Date now = new Date(); - boolean expired = platform.getTokenExpiresAt() == null - || platform.getTokenExpiresAt().getTime() - now.getTime() < 5 * 60 * 1000L; - if (!StringUtils.hasText(platform.getAccessToken()) || expired) { - 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) { - PlatformFlowMapper flowMapper = SpringUtils.getBean(PlatformFlowMapper.class); - PlatformFlow tokenFlow = flowMapper.selectOne( - Wrappers.lambdaQuery() - .eq(PlatformFlow::getPlatformId, fresh.getId()) - .eq(PlatformFlow::getFlowType, 3) - .eq(PlatformFlow::getEnable, "1") - .last("LIMIT 1")); - if (tokenFlow != null) { - FlowContext ctx = new FlowContext(fresh); - flowExecutor.execute(tokenFlow, ctx); - fresh = instanceService.getById(platform.getId()); - } else { - log.warn("[DCP] 平台 {} 未配置 Token 刷新流程(flow_type=3)", fresh.getId()); - } + boolean expired = platformUserAuth.getTokenExpiresAt() == null + || platformUserAuth.getTokenExpiresAt().getTime() - now.getTime() < 5 * 60 * 1000L; + if (!StringUtils.hasText(platformUserAuth.getAccessToken()) || expired) { + // 执行刷新token流程 + PlatformFlowMapper flowMapper = SpringUtils.getBean(PlatformFlowMapper.class); + PlatformFlow tokenFlow = flowMapper.selectOne( + Wrappers.lambdaQuery() + .eq(PlatformFlow::getPlatformId, platform.getId()) + .eq(PlatformFlow::getFlowType, 3) + .eq(PlatformFlow::getEnable, "1") + .last("LIMIT 1")); + if (tokenFlow != null) { + FlowContext ctx = new FlowContext(platform, mapping, platformUserAuth); + flowExecutor.execute(tokenFlow, ctx); + platformUserAuth = platformUserAuthMapper.selectById(mapping.getPlatformUserAuthId()); + } else { + log.warn("[DCP] 平台 {} 未配置 Token 刷新流程(flow_type=3)", platform.getId()); } - platform.setAccessToken(fresh.getAccessToken()); - platform.setTokenExpiresAt(fresh.getTokenExpiresAt()); } - return platform.getAccessToken(); + + return platformUserAuth.getAccessToken(); } } 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 7b592d15..34ba5046 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 @@ -28,6 +28,9 @@ public class PlatformDeviceMapping implements Serializable { @ApiModelProperty("所属平台ID") private Long platformId; + @ApiModelProperty("关联用户认证ID") + private Long platformUserAuthId; + @ApiModelProperty("DCP项目ID,绑定页面用户自己选择") private String platformProjectId; diff --git a/fastbee-service/fastbee-iot-service/src/main/java/com/fastbee/iot/platform/domain/PlatformInstance.java b/fastbee-service/fastbee-iot-service/src/main/java/com/fastbee/iot/platform/domain/PlatformInstance.java index 7feeb526..d8d367be 100644 --- a/fastbee-service/fastbee-iot-service/src/main/java/com/fastbee/iot/platform/domain/PlatformInstance.java +++ b/fastbee-service/fastbee-iot-service/src/main/java/com/fastbee/iot/platform/domain/PlatformInstance.java @@ -36,10 +36,10 @@ public class PlatformInstance implements Serializable { @ApiModelProperty("认证类型(1=密码Token,2=ApiKey,3=OAuth2)") private Integer authType; - @ApiModelProperty("认证凭证JSON {userName,password,clientId,scope}") + @ApiModelProperty("认证配置模板（新增用户时复制）{userName,password,clientId,scope}") private String authConfig; - @ApiModelProperty("初始化后保存的平台数据 {userId,projectId,...}") + @ApiModelProperty("平台级配置（站点编码等）") private String extraConfig; @ApiModelProperty("当前有效Token") diff --git a/fastbee-service/fastbee-iot-service/src/main/java/com/fastbee/iot/platform/domain/PlatformUserAuth.java b/fastbee-service/fastbee-iot-service/src/main/java/com/fastbee/iot/platform/domain/PlatformUserAuth.java new file mode 100644 index 00000000..eeaa0f8c --- /dev/null +++ b/fastbee-service/fastbee-iot-service/src/main/java/com/fastbee/iot/platform/domain/PlatformUserAuth.java @@ -0,0 +1,68 @@ +package com.fastbee.iot.platform.domain; + +import com.baomidou.mybatisplus.annotation.IdType; +import com.baomidou.mybatisplus.annotation.TableId; +import com.baomidou.mybatisplus.annotation.TableName; +import com.fasterxml.jackson.annotation.JsonFormat; +import io.swagger.annotations.ApiModel; +import io.swagger.annotations.ApiModelProperty; +import lombok.Data; + +import java.io.Serializable; +import java.util.Date; + +/** + * 平台用户认证实体 + * 支持一个平台对应多个用户认证 + */ +@ApiModel(value = "PlatformUserAuth", description = "平台用户认证实体") +@Data +@TableName("platform_user_auth") +public class PlatformUserAuth implements Serializable { + private static final long serialVersionUID = 1L; + + @TableId(value = "id", type = IdType.AUTO) + @ApiModelProperty("id") + private Long id; + + @ApiModelProperty("平台实例ID") + private Long platformId; + + @ApiModelProperty("平台侧用户ID（如DCP的userId）") + private String platformUserId; + + @ApiModelProperty("平台侧用户名") + private String platformUserName; + + @ApiModelProperty("用户认证凭证JSON {userName,password,clientId,scope}") + private String authConfig; + + @ApiModelProperty("该用户的Token") + private String accessToken; + + @JsonFormat(pattern = "yyyy-MM-dd HH:mm:ss") + @ApiModelProperty("Token过期时间") + private Date tokenExpiresAt; + + @ApiModelProperty("初始化状态(0=未初始化,1=已初始化)") + private Integer initStatus; + + @ApiModelProperty("用户级配置 {projectId,projectName,...}") + private String extraConfig; + + @ApiModelProperty("是否启用(0=禁用,1=启用)") + private String enable; + + @ApiModelProperty("备注") + private String remark; + + private String createBy; + + @JsonFormat(pattern = "yyyy-MM-dd HH:mm:ss") + private Date createTime; + + private String updateBy; + + @JsonFormat(pattern = "yyyy-MM-dd HH:mm:ss") + private Date updateTime; +} \ No newline at end of file diff --git a/fastbee-service/fastbee-iot-service/src/main/java/com/fastbee/iot/platform/flow/FlowContext.java b/fastbee-service/fastbee-iot-service/src/main/java/com/fastbee/iot/platform/flow/FlowContext.java index 0a992b79..38ae8245 100644 --- a/fastbee-service/fastbee-iot-service/src/main/java/com/fastbee/iot/platform/flow/FlowContext.java +++ b/fastbee-service/fastbee-iot-service/src/main/java/com/fastbee/iot/platform/flow/FlowContext.java @@ -3,6 +3,7 @@ package com.fastbee.iot.platform.flow; import com.fastbee.iot.platform.domain.PlatformDeviceMapping; import com.fastbee.iot.platform.domain.PlatformDeviceSubMapping; import com.fastbee.iot.platform.domain.PlatformInstance; +import com.fastbee.iot.platform.domain.PlatformUserAuth; import java.util.HashMap; import java.util.List; @@ -17,17 +18,24 @@ public class FlowContext { * 平台实体 */ private final PlatformInstance platform; + + /** + * 用户认证（用户级流程时使用） + */ + private final PlatformUserAuth platformUserAuth; + /** * 绑定推送关系 */ - private final PlatformDeviceMapping deviceMapping; + private PlatformDeviceMapping deviceMapping; // 绑定推送子关系 private List monitorSubMappings; /** * 设备上报的原始 payload（实时推送时有值） */ - private final String payload; + private String payload; + /** * 步骤出参存储，key = step.resultKey，value = 解析后的 JSON Map */ @@ -62,20 +70,55 @@ public class FlowContext { */ private String referenceSubMappingOnboardConfig; - public FlowContext(PlatformInstance platform, PlatformDeviceMapping deviceMapping, String payload) { + public FlowContext(PlatformInstance platform, PlatformDeviceMapping deviceMapping, PlatformUserAuth platformUserAuth, String payload) { this.platform = platform; this.deviceMapping = deviceMapping; + this.platformUserAuth = platformUserAuth; this.payload = payload; } - public FlowContext(PlatformInstance platform) { - this(platform, null, null); + public FlowContext(PlatformInstance platform, PlatformDeviceMapping deviceMapping, PlatformUserAuth platformUserAuth) { + this.platform = platform; + this.deviceMapping = deviceMapping; + this.platformUserAuth = platformUserAuth; + } + + public FlowContext(PlatformInstance platform, PlatformUserAuth platformUserAuth) { + this.platform = platform; + this.platformUserAuth = platformUserAuth; + } + + public FlowContext(PlatformInstance platform, PlatformUserAuth platformUserAuth, PlatformDeviceMapping deviceMapping, List monitorSubMappings, + String currentMonitorSerialNumber, String currentMonitorDeviceName, Integer currentMonitorIndex, String subMappingOnboardConfig) { + this(platform, platformUserAuth, deviceMapping, monitorSubMappings, null, currentMonitorSerialNumber, currentMonitorDeviceName, currentMonitorIndex, subMappingOnboardConfig, null, null, null); + } + + public FlowContext(PlatformInstance platform, PlatformUserAuth platformUserAuth, PlatformDeviceMapping deviceMapping, + List monitorSubMappings, String payload, String currentMonitorSerialNumber, + String currentMonitorDeviceName, Integer currentMonitorIndex, String subMappingOnboardConfig, + String currentReferenceSerialNumber, String currentReferenceDeviceName, String referenceSubMappingOnboardConfig) { + this.platform = platform; + this.platformUserAuth = platformUserAuth; + this.deviceMapping = deviceMapping; + this.monitorSubMappings = monitorSubMappings; + this.payload = payload; + this.currentMonitorSerialNumber = currentMonitorSerialNumber; + this.currentMonitorDeviceName = currentMonitorDeviceName; + this.currentMonitorIndex = currentMonitorIndex; + this.subMappingOnboardConfig = subMappingOnboardConfig; + this.currentReferenceSerialNumber = currentReferenceSerialNumber; + this.currentReferenceDeviceName = currentReferenceDeviceName; + this.referenceSubMappingOnboardConfig = referenceSubMappingOnboardConfig; } public PlatformInstance getPlatform() { return platform; } + public PlatformUserAuth getPlatformUserAuth() { + return platformUserAuth; + } + public PlatformDeviceMapping getDeviceMapping() { return deviceMapping; } 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 447a0e86..a854d24a 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 @@ -13,11 +13,7 @@ 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.PlatformDeviceSubMappingMapper; -import com.fastbee.iot.platform.mapper.PlatformFlowStepMapper; -import com.fastbee.iot.platform.mapper.PlatformInstanceMapper; -import com.fastbee.iot.platform.mapper.PlatformInterfaceMapper; +import com.fastbee.iot.platform.mapper.*; import com.fastbee.iot.platform.service.IPlatformPushLogService; import com.dtflys.forest.http.ForestRequestType; import lombok.extern.slf4j.Slf4j; @@ -244,10 +240,21 @@ public class PlatformFlowExecutor { ? JSON.parseObject(step.getParamMapping(), Map.class) : Collections.emptyMap(); - JSONObject authConfig = StringUtils.hasText(ctx.getPlatform().getAuthConfig()) - ? JSON.parseObject(ctx.getPlatform().getAuthConfig()) : new JSONObject(); - JSONObject extraConfig = StringUtils.hasText(ctx.getPlatform().getExtraConfig()) - ? JSON.parseObject(ctx.getPlatform().getExtraConfig()) : new JSONObject(); + // 优先从 userAuth 获取 authConfig，fallback 到 platform.authTemplate + JSONObject authConfig = new JSONObject(); + if (ctx.getPlatformUserAuth() != null && StringUtils.hasText(ctx.getPlatformUserAuth().getAuthConfig())) { + authConfig = JSON.parseObject(ctx.getPlatformUserAuth().getAuthConfig()); + } else if (StringUtils.hasText(ctx.getPlatform().getAuthConfig())) { + authConfig = JSON.parseObject(ctx.getPlatform().getAuthConfig()); + } + + // 优先从 userAuth 获取 extraConfig，fallback 到 platform.extraConfig + JSONObject extraConfig = new JSONObject(); + if (ctx.getPlatformUserAuth() != null && StringUtils.hasText(ctx.getPlatformUserAuth().getExtraConfig())) { + extraConfig = JSON.parseObject(ctx.getPlatformUserAuth().getExtraConfig()); + } else if (StringUtils.hasText(ctx.getPlatform().getExtraConfig())) { + extraConfig = JSON.parseObject(ctx.getPlatform().getExtraConfig()); + } for (int i = 0; i < schema.size(); i++) { JSONObject field = schema.getJSONObject(i); @@ -1086,21 +1093,31 @@ public class PlatformFlowExecutor { return value; } - /** 将 result_save 中指定字段持久化到 platform_instance.extra_config 或 platform_device_mapping */ + /** + * 将 result_save 中指定字段持久化到 platform_user_auth 或 platform_device_mapping + * @param step 流程步骤，第几步 + * @param responseData 请求平台接口返回值 + * @param context 容器上下文 + */ private void persistResultSave(PlatformFlowStep step, Object responseData, FlowContext context) { try { JSONArray saves = JSON.parseArray(step.getResultSave()); JSONObject response = toJsonObject(responseData); if (response == null) return; - PlatformInstance platform = context.getPlatform(); + //PlatformInstance platform = context.getPlatform(); + PlatformUserAuth userAuth = context.getPlatformUserAuth(); - JSONObject extra = StringUtils.hasText(platform.getExtraConfig()) - ? JSON.parseObject(platform.getExtraConfig()) : new JSONObject(); + JSONObject extra = StringUtils.hasText(userAuth.getExtraConfig()) + ? JSON.parseObject(userAuth.getExtraConfig()) : new JSONObject(); boolean extraChanged = false; - boolean platformFieldChanged = false; + boolean userAuthFieldChanged = false; + // 平台用户授权信息(token、userId) + PlatformUserAuth userAuthUpdate = new PlatformUserAuth(); PlatformDeviceMapping mapping = context.getDeviceMapping(); boolean mappingChanged = false; + // 保存平台映射关系(测点ID、设备ID) + PlatformDeviceMapping mappingUpdate = new PlatformDeviceMapping(); for (int i = 0; i < saves.size(); i++) { JSONObject save = saves.getJSONObject(i); @@ -1109,58 +1126,62 @@ public class PlatformFlowExecutor { Object value = response.get(field); if (value == null || saveTo == null) continue; - if (saveTo.startsWith("extra_config.")) { - String key = saveTo.substring("extra_config.".length()); - extra.put(key, value); - extraChanged = true; - } else if (saveTo.startsWith("platform_instance.")) { - String col = saveTo.substring("platform_instance.".length()); - if ("access_token".equals(col)) { - platform.setAccessToken(value.toString()); - platformFieldChanged = true; - } else if ("token_expires_at".equals(col)) { - try { - long seconds = Long.parseLong(value.toString()); - platform.setTokenExpiresAt(new Date(System.currentTimeMillis() + seconds * 1000)); - platformFieldChanged = true; - } catch (Exception ignored) {} + if (saveTo.startsWith("platform_user_auth.")) { + // 新增：保存到用户认证表 + String col = saveTo.substring("platform_user_auth.".length()); + switch (col) { + case "access_token": + userAuthUpdate.setAccessToken(value.toString()); + userAuthFieldChanged = true; + break; + case "token_expires_at": + try { + long seconds = Long.parseLong(value.toString()); + userAuthUpdate.setTokenExpiresAt(new Date(System.currentTimeMillis() + seconds * 1000)); + userAuthFieldChanged = true; + } catch (Exception ignored) { + log.warn("保存token过期时间出错"); + } + break; + case "platform_user_id": + userAuthUpdate.setPlatformUserId(value.toString()); + userAuthFieldChanged = true; + break; + case "platform_user_name": + userAuthUpdate.setPlatformUserName(value.toString()); + userAuthFieldChanged = true; + break; } } else if (saveTo.startsWith("device_mapping.") && mapping != null) { String col = saveTo.substring("device_mapping.".length()); if ("platform_device_id".equals(col)) { - mapping.setPlatformDeviceId(value.toString()); + mappingUpdate.setPlatformDeviceId(value.toString()); mappingChanged = true; } else if ("platform_point_id".equals(col)) { - mapping.setPlatformPointId(value.toString()); + mappingUpdate.setPlatformPointId(value.toString()); mappingChanged = true; } + } else if (saveTo.startsWith("extra_config.")) { + String key = saveTo.substring("extra_config.".length()); + extra.put(key, value); + userAuthUpdate.setExtraConfig(extra.toJSONString()); + extraChanged = true; } } - if (extraChanged) { - // 只更新 extra_config 字段，不覆盖其他字段 - platform.setExtraConfig(extra.toJSONString()); - PlatformInstance update = new PlatformInstance(); - update.setId(platform.getId()); - update.setExtraConfig(extra.toJSONString()); - instanceMapper.updateById(update); - } - if (platformFieldChanged) { - // 只更新 access_token 和 token_expires_at，不覆盖 extra_config - PlatformInstance update = new PlatformInstance(); - update.setId(platform.getId()); - update.setAccessToken(platform.getAccessToken()); - update.setTokenExpiresAt(platform.getTokenExpiresAt()); - instanceMapper.updateById(update); + // 保存用户认证字段 + if (userAuthFieldChanged || extraChanged) { + userAuthUpdate.setId(userAuth.getId()); + PlatformUserAuthMapper userAuthMapper = SpringUtils.getBean(PlatformUserAuthMapper.class); + userAuthMapper.updateById(userAuthUpdate); } + + // 保存设备映射字段 if (mappingChanged) { - PlatformDeviceMapping update = new PlatformDeviceMapping(); - update.setId(mapping.getId()); - update.setPlatformDeviceId(mapping.getPlatformDeviceId()); - update.setPlatformPointId(mapping.getPlatformPointId()); + mappingUpdate.setId(mapping.getId()); // 直接用 SpringUtils 获取 mapper，避免循环依赖注入 PlatformDeviceMappingMapper mappingMapper = SpringUtils.getBean(PlatformDeviceMappingMapper.class); - mappingMapper.updateById(update); + mappingMapper.updateById(mappingUpdate); } } catch (Exception e) { log.warn("[PlatformFlow] persistResultSave 失败: {}", e.getMessage()); @@ -1196,7 +1217,7 @@ public class PlatformFlowExecutor { * 解析当前步骤应使用的 Token： * - flow_type=3（Token刷新）：不注入，避免循环 * - flow_type=1（平台初始化）step1：认证步骤本身不注入 - * - 其他情况：从 platform_instance 取（step1 执行后已通过 persistResultSave 写入），过期则自动刷新 + * - 其他情况：从 userAuth 获取，过期则自动刷新 */ private String resolveToken(PlatformFlow flow, PlatformFlowStep step, FlowContext ctx) { if (flow.getFlowType() == 3) { @@ -1206,42 +1227,50 @@ public class PlatformFlowExecutor { // 初始化流程第1步是认证接口本身，不注入 token return null; } - // 其余情况（含初始化流程 step2+）：从 platform 对象取 - // step1 执行后 persistResultSave 已将 token 写入 platform 内存对象和 DB - return ensureToken(ctx.getPlatform()); + return ensureToken(ctx); } - /** 确保 Token 有效，过期则刷新 */ - public String ensureToken(PlatformInstance platform) { + /** + * 确保 Token 有效，过期则刷新 + */ + public String ensureToken(FlowContext ctx) { + if (ctx == null || ctx.getPlatformUserAuth() == null) { + throw new RuntimeException("入网失败，accessToken为空"); + } + PlatformUserAuth userAuth = ctx.getPlatformUserAuth(); Date now = new Date(); // 提前 5 分钟视为过期 - boolean expired = platform.getTokenExpiresAt() == null - || platform.getTokenExpiresAt().getTime() - now.getTime() < 5 * 60 * 1000L; - if (!StringUtils.hasText(platform.getAccessToken()) || expired) { - refreshToken(platform); + boolean expired = userAuth.getTokenExpiresAt() == null + || userAuth.getTokenExpiresAt().getTime() - now.getTime() < 5 * 60 * 1000L; + if (!StringUtils.hasText(userAuth.getAccessToken()) || expired) { + refreshToken(ctx); } - return platform.getAccessToken(); + return userAuth.getAccessToken(); } - /** 执行 flow_type=3 的 Token 刷新流程 */ - private void refreshToken(PlatformInstance platform) { - com.baomidou.mybatisplus.core.conditions.query.LambdaQueryWrapper wrapper = - Wrappers.lambdaQuery() - .eq(PlatformFlow::getPlatformId, platform.getId()) - .eq(PlatformFlow::getFlowType, 3) - .eq(PlatformFlow::getEnable, "1"); - // 通过 ApplicationContext 获取 PlatformFlowMapper，避免循环依赖 - // 此处直接用 instanceMapper 所在的 Spring 容器获取 - // 实际注入见 PlatformFlowExecutor 字段 - com.fastbee.iot.platform.mapper.PlatformFlowMapper flowMapper = - com.fastbee.common.utils.spring.SpringUtils.getBean( - com.fastbee.iot.platform.mapper.PlatformFlowMapper.class); - PlatformFlow tokenFlow = flowMapper.selectOne(wrapper); + /** + * 执行 flow_type=3 的 Token 刷新流程 + */ + private void refreshToken(FlowContext ctx) { + PlatformUserAuth userAuth = ctx.getPlatformUserAuth(); + PlatformFlowMapper flowMapper = SpringUtils.getBean(PlatformFlowMapper.class); + PlatformFlow tokenFlow = flowMapper.selectOne(Wrappers.lambdaQuery() + .eq(PlatformFlow::getPlatformId, userAuth.getPlatformId()) + .eq(PlatformFlow::getFlowType, 3) + .eq(PlatformFlow::getEnable, "1") + .last("LIMIT 1")); if (tokenFlow == null) { - log.warn("[PlatformFlow] 平台 {} 未配置 Token 刷新流程(flow_type=3)", platform.getId()); + log.warn("[PlatformFlow] 平台 {} 未配置 Token 刷新流程(flow_type=3)", userAuth.getPlatformId()); return; } - FlowContext ctx = new FlowContext(platform); + + // 获取平台实例 + PlatformInstance platform = ctx.getPlatform(); + if (platform == null) { + log.warn("[PlatformFlow] 平台不存在: {}", userAuth.getPlatformId()); + return; + } + execute(tokenFlow, ctx); // 从上下文中提取 token（约定 Token 刷新流程最后一步 resultKey="tokenStep"） @@ -1252,12 +1281,13 @@ public class PlatformFlowExecutor { String accessToken = (String) tokenResp.get("access_token"); Object expiresIn = tokenResp.get("expires_in"); if (StringUtils.hasText(accessToken)) { - platform.setAccessToken(accessToken); + userAuth.setAccessToken(accessToken); if (expiresIn != null) { long seconds = Long.parseLong(expiresIn.toString()); - platform.setTokenExpiresAt(new Date(System.currentTimeMillis() + seconds * 1000)); + userAuth.setTokenExpiresAt(new Date(System.currentTimeMillis() + seconds * 1000)); } - instanceMapper.updateById(platform); + PlatformUserAuthMapper userAuthMapper = SpringUtils.getBean(PlatformUserAuthMapper.class); + userAuthMapper.updateById(userAuth); } } } diff --git a/fastbee-service/fastbee-iot-service/src/main/java/com/fastbee/iot/platform/job/PlatformPushScheduledJob.java b/fastbee-service/fastbee-iot-service/src/main/java/com/fastbee/iot/platform/job/PlatformPushScheduledJob.java index 0dde9d74..14835bf0 100644 --- a/fastbee-service/fastbee-iot-service/src/main/java/com/fastbee/iot/platform/job/PlatformPushScheduledJob.java +++ b/fastbee-service/fastbee-iot-service/src/main/java/com/fastbee/iot/platform/job/PlatformPushScheduledJob.java @@ -8,14 +8,12 @@ import com.fastbee.iot.cache.ITSLCache; import com.fastbee.iot.domain.Device; import com.fastbee.iot.model.ThingsModels.ValueItem; import com.fastbee.iot.platform.adapter.dcp.DcpPlatformAdapter; -import com.fastbee.iot.platform.domain.PlatformDeviceMapping; -import com.fastbee.iot.platform.domain.PlatformDeviceSubMapping; -import com.fastbee.iot.platform.domain.PlatformInstance; -import com.fastbee.iot.platform.domain.PlatformInterface; +import com.fastbee.iot.platform.domain.*; import com.fastbee.iot.platform.mapper.PlatformInterfaceMapper; import com.fastbee.iot.platform.service.IPlatformDeviceMappingService; import com.fastbee.iot.platform.service.IPlatformDeviceSubMappingService; import com.fastbee.iot.platform.service.IPlatformInstanceService; +import com.fastbee.iot.platform.service.IPlatformUserAuthService; import com.fastbee.iot.service.IDeviceService; import lombok.extern.slf4j.Slf4j; import org.apache.commons.collections4.CollectionUtils; @@ -59,6 +57,8 @@ public class PlatformPushScheduledJob { private IDeviceService deviceService; @Autowired private PlatformInterfaceMapper interfaceMapper; + @Autowired + private IPlatformUserAuthService platformUserAuthService; /** * 每5分钟执行一次，检查并触发到期的推送任务 @@ -128,6 +128,12 @@ public class PlatformPushScheduledJob { return; } + PlatformUserAuth platformUserAuth = platformUserAuthService.getById(mapping.getPlatformUserAuthId()); + if (platformUserAuth == null || !"1".equals(platformUserAuth.getEnable())) { + log.warn("[PlatformPushJob] 平台已授权用户不存在或已禁用 platformUserAuthId={}", mapping.getPlatformUserAuthId()); + return; + } + // 数据库 last_push_time 为 DATETIME（秒精度），此处截断毫秒， // 避免 Redis 中带毫秒的时间与库中已截断的时间比较时误判为"数据变化"而重复推送 Date dataTime = truncateToSeconds(resolveLatestDataTime(mapping)); @@ -145,10 +151,10 @@ public class PlatformPushScheduledJob { PlatformInterface pushInterface = loadPushInterface(mapping); if (pushInterface != null && pushInterface.getIsBatchPush() != null && pushInterface.getIsBatchPush() == 1) { // 批量推送 - pushBatchGatewayData(platform, mapping); + pushBatchGatewayData(platform, platformUserAuth, mapping); } else { // 非批量：逐个子设备推送 - pushGatewayLatestData(platform, mapping); + pushGatewayLatestData(platform, platformUserAuth, mapping); } } else { // 直连设备 @@ -159,7 +165,7 @@ public class PlatformPushScheduledJob { return; } String payload = buildPayloadFromCache(cacheMap); - dcpAdapter.pushData(platform, mapping, serialNumber, payload); + dcpAdapter.pushData(platform, platformUserAuth, mapping, serialNumber, payload); } // lastPushTime 记录最近已推送的数据时间，用于幂等判断 @@ -172,7 +178,7 @@ public class PlatformPushScheduledJob { * 取最深层（最后一个）监测点作为触发点，构造模拟 payload， * 调用 pushGatewayData 进入批量推送路径，推送从浅到深的全量子设备数据。 */ - private void pushBatchGatewayData(PlatformInstance platform, PlatformDeviceMapping mapping) { + private void pushBatchGatewayData(PlatformInstance platform, PlatformUserAuth userAuth, PlatformDeviceMapping mapping) { List subMappings = subMappingService.selectByParentId(mapping.getId()); if (CollectionUtils.isEmpty(subMappings)) { log.warn("[PlatformPushJob] 网关设备 {} 无子设备映射", mapping.getSerialNumber()); @@ -202,14 +208,14 @@ public class PlatformPushScheduledJob { } String payload = buildPayloadFromCache(cacheMap); - dcpAdapter.pushGatewayData(platform, mapping, lastMonitor, lastMonitor.getSubSerialNumber(), payload); + dcpAdapter.pushGatewayData(platform, userAuth, mapping, lastMonitor, lastMonitor.getSubSerialNumber(), payload); log.info("[PlatformPushJob] 批量推送完成 mappingId={} 最深层监测点={} 子设备总数={}", mapping.getId(), lastMonitor.getSubSerialNumber(), monitors.size()); } /** * 推送网关设备最新数据（非批量，逐个子设备推送） */ - private void pushGatewayLatestData(PlatformInstance platform, PlatformDeviceMapping mapping) { + private void pushGatewayLatestData(PlatformInstance platform, PlatformUserAuth userAuth, PlatformDeviceMapping mapping) { List subMappings = subMappingService.selectByParentId(mapping.getId()); if (CollectionUtils.isEmpty(subMappings)) { log.warn("[PlatformPushJob] 网关设备 {} 无子设备映射", mapping.getSerialNumber()); @@ -230,7 +236,7 @@ public class PlatformPushScheduledJob { } String payload = buildPayloadFromCache(subCacheMap); - dcpAdapter.pushGatewayData(platform, mapping, subMapping, subSerialNumber, payload); + dcpAdapter.pushGatewayData(platform, userAuth, mapping, subMapping, subSerialNumber, payload); } catch (Exception e) { log.error("[PlatformPushJob] 网关子设备推送失败 subSn={}", subMapping.getSubSerialNumber(), e); } diff --git a/fastbee-service/fastbee-iot-service/src/main/java/com/fastbee/iot/platform/mapper/PlatformUserAuthMapper.java b/fastbee-service/fastbee-iot-service/src/main/java/com/fastbee/iot/platform/mapper/PlatformUserAuthMapper.java new file mode 100644 index 00000000..e934f7d7 --- /dev/null +++ b/fastbee-service/fastbee-iot-service/src/main/java/com/fastbee/iot/platform/mapper/PlatformUserAuthMapper.java @@ -0,0 +1,32 @@ +package com.fastbee.iot.platform.mapper; + +import com.baomidou.mybatisplus.core.mapper.BaseMapper; +import com.fastbee.iot.platform.domain.PlatformUserAuth; +import org.apache.ibatis.annotations.Mapper; +import org.apache.ibatis.annotations.Param; + +import java.util.List; + +/** + * 平台用户认证 Mapper 接口 + */ +@Mapper +public interface PlatformUserAuthMapper extends BaseMapper { + + /** + * 查询列表 + */ + List selectList(PlatformUserAuth query); + + /** + * 根据平台ID查询已初始化的用户列表 + */ + List selectInitializedByPlatformId(@Param("platformId") Long platformId); + + /** + * 根据平台ID和用户ID查询 + */ + PlatformUserAuth selectByPlatformAndUserId(@Param("platformId") Long platformId, @Param("platformUserId") String platformUserId); + + Integer countEffectivePlatformUser(@Param("platformId") Long platformId); +} \ No newline at end of file diff --git a/fastbee-service/fastbee-iot-service/src/main/java/com/fastbee/iot/platform/service/IPlatformFlowService.java b/fastbee-service/fastbee-iot-service/src/main/java/com/fastbee/iot/platform/service/IPlatformFlowService.java index 009f2b88..7d162e99 100644 --- a/fastbee-service/fastbee-iot-service/src/main/java/com/fastbee/iot/platform/service/IPlatformFlowService.java +++ b/fastbee-service/fastbee-iot-service/src/main/java/com/fastbee/iot/platform/service/IPlatformFlowService.java @@ -10,21 +10,9 @@ public interface IPlatformFlowService extends IService { List selectList(PlatformFlow query); - /** 查询某平台指定类型的流程 */ - PlatformFlow selectByPlatformAndType(Long platformId, Integer flowType); - - /** 查询某平台指定类型+传感器类型的流程（设备入网用） */ - PlatformFlow selectByPlatformTypeAndSensor(Long platformId, Integer flowType, String sensorType); - /** 查询流程的有序步骤列表 */ List selectSteps(Long flowId); - /** 手动触发流程执行 */ - void executeManually(Long flowId); - - /** 手动触发流程执行（设备入网流程需传 mappingId） */ - void executeManually(Long flowId, Long mappingId); - /** 保存流程及步骤（含步骤的新增/更新/删除） */ void saveWithSteps(PlatformFlow flow, List steps); } diff --git a/fastbee-service/fastbee-iot-service/src/main/java/com/fastbee/iot/platform/service/IPlatformInstanceService.java b/fastbee-service/fastbee-iot-service/src/main/java/com/fastbee/iot/platform/service/IPlatformInstanceService.java index 4d95fa82..6e492a3b 100644 --- a/fastbee-service/fastbee-iot-service/src/main/java/com/fastbee/iot/platform/service/IPlatformInstanceService.java +++ b/fastbee-service/fastbee-iot-service/src/main/java/com/fastbee/iot/platform/service/IPlatformInstanceService.java @@ -3,25 +3,9 @@ package com.fastbee.iot.platform.service; import com.baomidou.mybatisplus.extension.service.IService; import com.fastbee.iot.platform.domain.PlatformInstance; -import java.util.Date; import java.util.List; -import java.util.Map; public interface IPlatformInstanceService extends IService { List selectList(PlatformInstance query); - - /** 手动触发平台初始化流程（flow_type=1） */ - void initialize(Long platformId); - - /** 更新 Token 和过期时间 */ - void updateToken(Long platformId, String accessToken, Date tokenExpiresAt); - - /** 更新 extra_config 中的某个字段 */ - void updateExtraConfig(Long platformId, String key, String value); - - /** - * 根据平台ID查询项目列表 - */ - List> getProjectsByPlatformId(Long platformId); } diff --git a/fastbee-service/fastbee-iot-service/src/main/java/com/fastbee/iot/platform/service/IPlatformUserAuthService.java b/fastbee-service/fastbee-iot-service/src/main/java/com/fastbee/iot/platform/service/IPlatformUserAuthService.java new file mode 100644 index 00000000..26da0149 --- /dev/null +++ b/fastbee-service/fastbee-iot-service/src/main/java/com/fastbee/iot/platform/service/IPlatformUserAuthService.java @@ -0,0 +1,42 @@ +package com.fastbee.iot.platform.service; + +import com.baomidou.mybatisplus.extension.service.IService; +import com.fastbee.iot.platform.domain.PlatformUserAuth; + +import java.util.List; + +/** + * 平台用户认证 Service 接口 + */ +public interface IPlatformUserAuthService extends IService { + + /** + * 查询列表 + */ + List selectList(PlatformUserAuth query); + + /** + * 根据平台ID查询已初始化的用户列表 + */ + List selectInitializedByPlatformId(Long platformId); + + /** + * 初始化用户认证（执行初始化流程获取 userId） + */ + void initialize(Long userAuthId); + + /** + * 更新 Token + */ + void updateToken(Long userAuthId, String accessToken, java.util.Date tokenExpiresAt); + + /** + * 更新用户级配置 + */ + void updateExtraConfig(Long userAuthId, String key, String value); + + /** + * 获取用户的项目列表 + */ + List> getProjectsByUserAuthId(Long userAuthId); +} \ No newline at end of file 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 41816ccc..d4bb4c03 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 @@ -3,17 +3,16 @@ 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.fastbee.iot.platform.domain.PlatformDeviceMapping; -import com.fastbee.iot.platform.domain.PlatformDeviceSubMapping; -import com.fastbee.iot.platform.domain.PlatformFlow; -import com.fastbee.iot.platform.domain.PlatformInstance; +import com.fastbee.iot.platform.domain.*; import com.fastbee.iot.platform.flow.FlowContext; import com.fastbee.iot.platform.flow.PlatformFlowExecutor; import com.fastbee.iot.platform.mapper.PlatformDeviceMappingMapper; import com.fastbee.iot.platform.mapper.PlatformFlowMapper; import com.fastbee.iot.platform.mapper.PlatformInstanceMapper; +import com.fastbee.iot.platform.mapper.PlatformUserAuthMapper; import com.fastbee.iot.platform.service.IPlatformDeviceMappingService; import com.fastbee.iot.platform.service.IPlatformDeviceSubMappingService; +import com.fastbee.iot.platform.service.IPlatformUserAuthService; import lombok.extern.slf4j.Slf4j; import org.springframework.beans.factory.annotation.Autowired; import org.springframework.stereotype.Service; @@ -40,6 +39,8 @@ public class PlatformDeviceMappingServiceImpl extends ServiceImpl selectList(PlatformDeviceMapping query) { @@ -72,15 +73,16 @@ public class PlatformDeviceMappingServiceImpl extends ServiceImpllambdaQuery() .eq(PlatformFlow::getPlatformId, mapping.getPlatformId()) .eq(PlatformFlow::getFlowType, 2) @@ -92,17 +94,17 @@ public class PlatformDeviceMappingServiceImpl extends ServiceImpl subMappings = subMappingService.selectByParentId(mapping.getId()); PlatformDeviceSubMapping reference = resolveGatewayReference(mapping, subMappings); List monitors = subMappings.stream() @@ -132,12 +134,8 @@ public class PlatformDeviceMappingServiceImpl extends ServiceImpl selectList(PlatformFlow query) { @@ -43,58 +30,11 @@ public class PlatformFlowServiceImpl extends ServiceImpllambdaQuery() - .eq(PlatformFlow::getPlatformId, platformId) - .eq(PlatformFlow::getFlowType, flowType) - .eq(PlatformFlow::getEnable, "1") - .last("LIMIT 1")); - } - - @Override - public PlatformFlow selectByPlatformTypeAndSensor(Long platformId, Integer flowType, String sensorType) { - return this.getOne(Wrappers.lambdaQuery() - .eq(PlatformFlow::getPlatformId, platformId) - .eq(PlatformFlow::getFlowType, flowType) - .eq(PlatformFlow::getSensorType, sensorType) - .eq(PlatformFlow::getEnable, "1") - .last("LIMIT 1")); - } - @Override public List selectSteps(Long flowId) { return stepMapper.selectByFlowId(flowId); } - @Override - public void executeManually(Long flowId) { - executeManually(flowId, null); - } - - @Override - public void executeManually(Long flowId, Long mappingId) { - PlatformFlow flow = this.getById(flowId); - if (flow == null) throw new RuntimeException("流程不存在: " + flowId); - PlatformInstance platform = instanceMapper.selectById(flow.getPlatformId()); - if (platform == null) throw new RuntimeException("平台不存在: " + flow.getPlatformId()); - - // flow_type=2（设备入网）必须传 mappingId - if (flow.getFlowType() != null && flow.getFlowType() == 2) { - if (mappingId == null) { - throw new RuntimeException("设备入网流程需要指定设备绑定关系，请从设备绑定页面点击「入网」操作"); - } - PlatformDeviceMapping mapping = - SpringUtils.getBean(PlatformDeviceMappingMapper.class).selectById(mappingId); - if (mapping == null) throw new RuntimeException("设备绑定关系不存在: " + mappingId); - FlowContext ctx = new FlowContext(platform, mapping, null); - flowExecutor.execute(flow, ctx); - } else { - FlowContext ctx = new FlowContext(platform); - flowExecutor.execute(flow, ctx); - } - } - @Override @Transactional(rollbackFor = Exception.class) public void saveWithSteps(PlatformFlow flow, List steps) { diff --git a/fastbee-service/fastbee-iot-service/src/main/java/com/fastbee/iot/platform/service/impl/PlatformInstanceServiceImpl.java b/fastbee-service/fastbee-iot-service/src/main/java/com/fastbee/iot/platform/service/impl/PlatformInstanceServiceImpl.java index 7e0a985b..03a5a65b 100644 --- a/fastbee-service/fastbee-iot-service/src/main/java/com/fastbee/iot/platform/service/impl/PlatformInstanceServiceImpl.java +++ b/fastbee-service/fastbee-iot-service/src/main/java/com/fastbee/iot/platform/service/impl/PlatformInstanceServiceImpl.java @@ -1,41 +1,25 @@ package com.fastbee.iot.platform.service.impl; -import com.alibaba.fastjson2.JSON; -import com.alibaba.fastjson2.JSONObject; 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.common.core.redis.RedisCache; -import com.fastbee.common.core.redis.RedisKeyBuilder; -import com.fastbee.http.model.HttpClientConfig; -import com.fastbee.http.service.HttpClientFactory; -import com.fastbee.iot.platform.domain.PlatformFlow; +import com.fastbee.common.enums.StatusEnum; import com.fastbee.iot.platform.domain.PlatformInstance; -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.PlatformInstanceMapper; +import com.fastbee.iot.platform.mapper.PlatformUserAuthMapper; import com.fastbee.iot.platform.service.IPlatformInstanceService; import org.springframework.beans.factory.annotation.Autowired; import org.springframework.stereotype.Service; import org.springframework.util.StringUtils; import java.util.*; -import java.util.concurrent.TimeUnit; -import java.util.stream.Collectors; @Service public class PlatformInstanceServiceImpl extends ServiceImpl implements IPlatformInstanceService { @Autowired - private PlatformFlowMapper flowMapper; - @Autowired - private PlatformFlowExecutor flowExecutor; - @Autowired - private RedisCache redisCache; + private PlatformUserAuthMapper platformUserAuthMapper; @Override public List selectList(PlatformInstance query) { @@ -43,114 +27,12 @@ public class PlatformInstanceServiceImpl extends ServiceImpllambdaQuery() - .eq(PlatformFlow::getPlatformId, platformId) - .eq(PlatformFlow::getFlowType, 1) - .eq(PlatformFlow::getEnable, "1") - .last("LIMIT 1")); - if (initFlow == null) throw new RuntimeException("平台未配置初始化流程(flow_type=1)"); - FlowContext ctx = new FlowContext(platform); - flowExecutor.execute(initFlow, ctx); - // 标记初始化完成（extra_config 已在 persistResultSave 中更新） - platform.setInitStatus(1); - this.updateById(platform); - } - - @Override - public void updateToken(Long platformId, String accessToken, Date tokenExpiresAt) { - PlatformInstance update = new PlatformInstance(); - update.setId(platformId); - update.setAccessToken(accessToken); - update.setTokenExpiresAt(tokenExpiresAt); - this.updateById(update); - } - - @Override - public void updateExtraConfig(Long platformId, String key, String value) { - PlatformInstance platform = this.getById(platformId); - if (platform == null) return; - JSONObject extra = StringUtils.hasText(platform.getExtraConfig()) - ? JSON.parseObject(platform.getExtraConfig()) : new JSONObject(); - extra.put(key, value); - PlatformInstance update = new PlatformInstance(); - update.setId(platformId); - update.setExtraConfig(extra.toJSONString()); - this.updateById(update); - } - - @Override - public List> getProjectsByPlatformId(Long platformId) { - PlatformInstance platform = this.getById(platformId); - if (platform == null) { - throw new RuntimeException("平台不存在: " + platformId); + List resultList = this.list(wrapper); + if (resultList != null) { + resultList.forEach(instance -> + instance.setInitStatus(platformUserAuthMapper.countEffectivePlatformUser(instance.getId()) > 0 ? + StatusEnum.SUCCESS.getStatus() : StatusEnum.FAIL.getStatus())); } - - // 从extra_config读取userId - JSONObject extraConfig = JSON.parseObject(platform.getExtraConfig()); - String userId = extraConfig == null ? null : extraConfig.getString("userId"); - if (!StringUtils.hasText(userId)) { - throw new RuntimeException("平台未完成初始化，缺少userId。请先执行平台初始化流程。"); - } - - String key = RedisKeyBuilder.buildDCPProjectInfoKey(userId); - Object cacheObject = redisCache.getCacheObject(key); - if (cacheObject != null) { - return (List>) cacheObject; - } - - // 确保 token 有效（过期自动刷新） - String token = flowExecutor.ensureToken(platform); - String url = platform.getBaseUrl() + "/api/app/project/by-user-id/" + userId; - - HttpClientConfig config = HttpClientConfig.builder() - .url(url) - .method(ForestRequestType.GET) - .headers(Collections.singletonMap("Authorization", "Bearer " + token)) - .build(); - - ForestRequest request = HttpClientFactory.instance(config); - String responseBody = request.executeAsString(); - - if (!StringUtils.hasText(responseBody)) { - return Collections.emptyList(); - } - - List> result = parseProjectsResponse(responseBody); - redisCache.setCacheObject(key, result, 10, TimeUnit.MINUTES); - return result; - } - - @SuppressWarnings("unchecked") - private List> parseProjectsResponse(String responseBody) { - String trimmed = responseBody.trim(); - List> rawList; - - if (trimmed.startsWith("[")) { - rawList = JSON.parseArray(trimmed, Map.class); - } else { - JSONObject json = JSON.parseObject(trimmed); - Object data = json.get("data"); - if (data instanceof List) { - rawList = (List>) data; - } else { - return Collections.emptyList(); - } - } - - return rawList.stream() - .map(m -> { - Map item = new HashMap<>(); - item.put("id", m.get("id")); - item.put("name", m.get("name")); - return item; - }) - .collect(Collectors.toList()); + return resultList; } } diff --git a/fastbee-service/fastbee-iot-service/src/main/java/com/fastbee/iot/platform/service/impl/PlatformUserAuthServiceImpl.java b/fastbee-service/fastbee-iot-service/src/main/java/com/fastbee/iot/platform/service/impl/PlatformUserAuthServiceImpl.java new file mode 100644 index 00000000..b203ff86 --- /dev/null +++ b/fastbee-service/fastbee-iot-service/src/main/java/com/fastbee/iot/platform/service/impl/PlatformUserAuthServiceImpl.java @@ -0,0 +1,177 @@ +package com.fastbee.iot.platform.service.impl; + +import com.alibaba.fastjson2.JSON; +import com.alibaba.fastjson2.JSONObject; +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.common.core.redis.RedisCache; +import com.fastbee.common.core.redis.RedisKeyBuilder; +import com.fastbee.http.model.HttpClientConfig; +import com.fastbee.http.service.HttpClientFactory; +import com.fastbee.iot.platform.domain.PlatformFlow; +import com.fastbee.iot.platform.domain.PlatformInstance; +import com.fastbee.iot.platform.domain.PlatformUserAuth; +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.PlatformInstanceMapper; +import com.fastbee.iot.platform.mapper.PlatformUserAuthMapper; +import com.fastbee.iot.platform.service.IPlatformUserAuthService; +import org.springframework.beans.factory.annotation.Autowired; +import org.springframework.stereotype.Service; +import org.springframework.util.StringUtils; + +import java.util.*; +import java.util.concurrent.TimeUnit; +import java.util.stream.Collectors; + +@Service +public class PlatformUserAuthServiceImpl extends ServiceImpl + implements IPlatformUserAuthService { + + @Autowired + private PlatformInstanceMapper instanceMapper; + @Autowired + private PlatformFlowMapper flowMapper; + @Autowired + private PlatformFlowExecutor flowExecutor; + @Autowired + private RedisCache redisCache; + + @Override + public List selectList(PlatformUserAuth query) { + return baseMapper.selectList(query); + } + + @Override + public List selectInitializedByPlatformId(Long platformId) { + return baseMapper.selectInitializedByPlatformId(platformId); + } + + @Override + public void initialize(Long userAuthId) { + PlatformUserAuth userAuth = this.getById(userAuthId); + if (userAuth == null) { + throw new RuntimeException("用户认证不存在: " + userAuthId); + } + PlatformInstance platform = instanceMapper.selectById(userAuth.getPlatformId()); + if (platform == null) { + throw new RuntimeException("平台不存在: " + userAuth.getPlatformId()); + } + + // 查找初始化流程 + PlatformFlow initFlow = flowMapper.selectOne(Wrappers.lambdaQuery() + .eq(PlatformFlow::getPlatformId, userAuth.getPlatformId()) + .eq(PlatformFlow::getFlowType, 1) + .eq(PlatformFlow::getEnable, "1") + .last("LIMIT 1")); + if (initFlow == null) { + throw new RuntimeException("平台未配置初始化流程(flow_type=1)"); + } + + // 执行流程，传入用户认证上下文 + FlowContext ctx = new FlowContext(platform, userAuth); + flowExecutor.execute(initFlow, ctx); + + // 标记初始化完成 + userAuth.setInitStatus(1); + this.updateById(userAuth); + } + + @Override + public void updateToken(Long userAuthId, String accessToken, Date tokenExpiresAt) { + PlatformUserAuth update = new PlatformUserAuth(); + update.setId(userAuthId); + update.setAccessToken(accessToken); + update.setTokenExpiresAt(tokenExpiresAt); + this.updateById(update); + } + + @Override + public void updateExtraConfig(Long userAuthId, String key, String value) { + PlatformUserAuth userAuth = this.getById(userAuthId); + if (userAuth == null) return; + JSONObject extra = StringUtils.hasText(userAuth.getExtraConfig()) + ? JSON.parseObject(userAuth.getExtraConfig()) : new JSONObject(); + extra.put(key, value); + PlatformUserAuth update = new PlatformUserAuth(); + update.setId(userAuthId); + update.setExtraConfig(extra.toJSONString()); + this.updateById(update); + } + + @Override + public List> getProjectsByUserAuthId(Long userAuthId) { + PlatformUserAuth userAuth = this.getById(userAuthId); + if (userAuth == null) { + throw new RuntimeException("用户认证不存在: " + userAuthId); + } + + PlatformInstance platform = instanceMapper.selectById(userAuth.getPlatformId()); + if (platform == null) { + throw new RuntimeException("平台不存在: " + userAuth.getPlatformId()); + } + + String userId = userAuth.getPlatformUserId(); + if (!StringUtils.hasText(userId)) { + throw new RuntimeException("用户未完成初始化，缺少 platformUserId。请先执行用户初始化流程。"); + } + + // 尝试从缓存获取 + String key = RedisKeyBuilder.buildDCPProjectInfoKey(userId); + Object cacheObject = redisCache.getCacheObject(key); + if (cacheObject != null) { + return (List>) cacheObject; + } + + // 确保 Token 有效 + String token = flowExecutor.ensureToken(new FlowContext(platform, userAuth)); + String url = platform.getBaseUrl() + "/api/app/project/by-user-id/" + userId; + + HttpClientConfig config = HttpClientConfig.builder() + .url(url) + .method(ForestRequestType.GET) + .headers(Collections.singletonMap("Authorization", "Bearer " + token)) + .build(); + + ForestRequest request = HttpClientFactory.instance(config); + String responseBody = request.executeAsString(); + + if (!StringUtils.hasText(responseBody)) { + return Collections.emptyList(); + } + + List> result = parseProjectsResponse(responseBody); + redisCache.setCacheObject(key, result, 10, TimeUnit.MINUTES); + return result; + } + + @SuppressWarnings("unchecked") + private List> parseProjectsResponse(String responseBody) { + String trimmed = responseBody.trim(); + List> rawList; + + if (trimmed.startsWith("[")) { + rawList = JSON.parseArray(trimmed, Map.class); + } else { + JSONObject json = JSON.parseObject(trimmed); + Object data = json.get("data"); + if (data instanceof List) { + rawList = (List>) data; + } else { + return Collections.emptyList(); + } + } + + return rawList.stream() + .map(m -> { + Map item = new HashMap<>(); + item.put("id", m.get("id")); + item.put("name", m.get("name")); + return item; + }) + .collect(Collectors.toList()); + } +} \ No newline at end of file diff --git a/fastbee-service/fastbee-iot-service/src/main/java/com/fastbee/iot/platform/trigger/DeviceReportTrigger.java b/fastbee-service/fastbee-iot-service/src/main/java/com/fastbee/iot/platform/trigger/DeviceReportTrigger.java index f0a94183..01245fdc 100644 --- a/fastbee-service/fastbee-iot-service/src/main/java/com/fastbee/iot/platform/trigger/DeviceReportTrigger.java +++ b/fastbee-service/fastbee-iot-service/src/main/java/com/fastbee/iot/platform/trigger/DeviceReportTrigger.java @@ -5,9 +5,11 @@ import com.fastbee.iot.platform.adapter.dcp.DcpPlatformAdapter; import com.fastbee.iot.platform.domain.PlatformDeviceMapping; import com.fastbee.iot.platform.domain.PlatformDeviceSubMapping; import com.fastbee.iot.platform.domain.PlatformInstance; +import com.fastbee.iot.platform.domain.PlatformUserAuth; import com.fastbee.iot.platform.service.IPlatformDeviceMappingService; import com.fastbee.iot.platform.service.IPlatformDeviceSubMappingService; import com.fastbee.iot.platform.service.IPlatformInstanceService; +import com.fastbee.iot.platform.service.IPlatformUserAuthService; import lombok.extern.slf4j.Slf4j; import org.springframework.beans.factory.annotation.Autowired; import org.springframework.scheduling.annotation.Async; @@ -30,6 +32,8 @@ public class DeviceReportTrigger { @Autowired private IPlatformInstanceService instanceService; @Autowired + private IPlatformUserAuthService platformUserAuthService; + @Autowired private DcpPlatformAdapter dcpAdapter; /** @@ -58,7 +62,9 @@ public class DeviceReportTrigger { PlatformInstance platform = instanceService.getById(mapping.getPlatformId()); if (platform == null || !"1".equals(platform.getEnable())) continue; - dispatchPush(platform, mapping, serialNumber, payload); + PlatformUserAuth platformUserAuth = platformUserAuthService.getById(mapping.getPlatformUserAuthId()); + if (platformUserAuth == null || !"1".equals(platformUserAuth.getEnable())) continue; + dispatchPush(platform, platformUserAuth, mapping, serialNumber, payload); } } catch (Exception e) { log.error("[DeviceReportTrigger] 推送异常 serialNumber={}", serialNumber, e); @@ -66,16 +72,16 @@ public class DeviceReportTrigger { } /** 根据平台 code 路由到对应 Adapter */ - private void dispatchPush(PlatformInstance platform, PlatformDeviceMapping mapping, + private void dispatchPush(PlatformInstance platform, PlatformUserAuth userAuth, PlatformDeviceMapping mapping, String reportSerialNumber, String payload) { switch (platform.getCode()) { case "dcp": if (mapping.getDeviceType() != null && mapping.getDeviceType() == 1) { // 网关设备：查子表确定该 serialNumber 的角色和 platformDeviceId - dispatchGatewayPush(platform, mapping, reportSerialNumber, payload); + dispatchGatewayPush(platform, userAuth, mapping, reportSerialNumber, payload); } else { // 直连设备 - dcpAdapter.pushData(platform, mapping, reportSerialNumber, payload); + dcpAdapter.pushData(platform, userAuth, mapping, reportSerialNumber, payload); } break; default: @@ -89,7 +95,7 @@ public class DeviceReportTrigger { * - 上报者是子表中的 reference → adapter 仅缓存基准数据 * - 上报者是网关本身(serial_number 匹配主表) → 判断是否在子表中标记为 reference */ - private void dispatchGatewayPush(PlatformInstance platform, PlatformDeviceMapping mapping, + private void dispatchGatewayPush(PlatformInstance platform, PlatformUserAuth userAuth, PlatformDeviceMapping mapping, String reportSerialNumber, String payload) { // 查找该序列号在子表中的记录 List subMappings = subMappingService.selectByParentId(mapping.getId()); @@ -104,6 +110,6 @@ public class DeviceReportTrigger { return; } - dcpAdapter.pushGatewayData(platform, mapping, matchedSub, reportSerialNumber, payload); + dcpAdapter.pushGatewayData(platform, userAuth, mapping, matchedSub, reportSerialNumber, payload); } } diff --git a/fastbee-service/fastbee-iot-service/src/main/resources/mapper/iot/platform/PlatformDeviceMappingMapper.xml b/fastbee-service/fastbee-iot-service/src/main/resources/mapper/iot/platform/PlatformDeviceMappingMapper.xml index 50a774a6..0cb9f21c 100644 --- a/fastbee-service/fastbee-iot-service/src/main/resources/mapper/iot/platform/PlatformDeviceMappingMapper.xml +++ b/fastbee-service/fastbee-iot-service/src/main/resources/mapper/iot/platform/PlatformDeviceMappingMapper.xml @@ -5,6 +5,7 @@ + diff --git a/fastbee-service/fastbee-iot-service/src/main/resources/mapper/iot/platform/PlatformUserAuthMapper.xml b/fastbee-service/fastbee-iot-service/src/main/resources/mapper/iot/platform/PlatformUserAuthMapper.xml new file mode 100644 index 00000000..2d582711 --- /dev/null +++ b/fastbee-service/fastbee-iot-service/src/main/resources/mapper/iot/platform/PlatformUserAuthMapper.xml @@ -0,0 +1,73 @@ + + + + + + + + + + + + + + + + + + + + + + + + id, platform_id, platform_user_id, platform_user_name, auth_config, access_token, + token_expires_at, init_status, extra_config, enable, remark, + create_by, create_time, update_by, update_time + + + + + + + + + + + \ No newline at end of file diff --git a/sql/platform-integration.sql b/sql/platform-integration.sql index 2ceff41f..0295e1b0 100644 --- a/sql/platform-integration.sql +++ b/sql/platform-integration.sql @@ -120,6 +120,7 @@ CREATE TABLE IF NOT EXISTS `platform_flow_step` ( CREATE TABLE IF NOT EXISTS `platform_device_mapping` ( `id` bigint(20) NOT NULL AUTO_INCREMENT COMMENT 'id', `platform_id` bigint(20) NOT NULL COMMENT '所属平台ID', + `platform_user_auth_id` bigint(20) DEFAULT NULL COMMENT '关联用户认证ID', `platform_project_id` varchar(100) NOT NULL COMMENT 'DCP项目ID', `platform_project_name` varchar(255) DEFAULT NULL COMMENT 'DCP项目名称', `device_id` bigint(20) NOT NULL COMMENT 'FastBee设备ID', @@ -144,7 +145,8 @@ CREATE TABLE IF NOT EXISTS `platform_device_mapping` ( `update_time` datetime DEFAULT CURRENT_TIMESTAMP ON UPDATE CURRENT_TIMESTAMP COMMENT '更新时间', PRIMARY KEY (`id`), KEY `idx_serial` (`serial_number`), - KEY `idx_platform_device` (`platform_id`, `device_id`) + KEY `idx_platform_device` (`platform_id`, `device_id`), + KEY `idx_user_auth` (`platform_user_auth_id`) ) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COMMENT='平台设备绑定关系'; -- ---------------------------- @@ -238,3 +240,38 @@ CREATE TABLE IF NOT EXISTS `platform_push_log` ( -- 注意：platform_id 需在创建 DCP 平台实例后更新，此处用占位符 0 -- 实际由 DcpPlatformAdapter 在初始化时写入 +-- ---------------------------- +-- 7. 创建平台用户认证表 +-- ---------------------------- +CREATE TABLE `platform_user_auth` ( + `id` bigint(20) NOT NULL AUTO_INCREMENT COMMENT 'id', + `platform_id` bigint(20) NOT NULL COMMENT '平台实例ID', + `platform_user_id` varchar(100) DEFAULT NULL COMMENT '平台侧用户ID（如DCP的userId）', + `platform_user_name` varchar(100) DEFAULT NULL COMMENT '平台侧用户名', + `auth_config` json DEFAULT NULL COMMENT '用户认证凭证JSON {userName,password,clientId,scope}', + `access_token` varchar(2000) DEFAULT NULL COMMENT '该用户的Token', + `token_expires_at` datetime DEFAULT NULL COMMENT 'Token过期时间', + `init_status` tinyint(1) NOT NULL DEFAULT 0 COMMENT '初始化状态(0=未初始化,1=已初始化)', + `extra_config` json DEFAULT NULL COMMENT '用户级配置 {projectId,projectName,...}', + `enable` char(1) NOT NULL DEFAULT '1' COMMENT '是否启用', + `remark` varchar(500) DEFAULT NULL COMMENT '备注', + `create_by` varchar(64) DEFAULT '' COMMENT '创建者', + `create_time` datetime DEFAULT CURRENT_TIMESTAMP COMMENT '创建时间', + `update_by` varchar(64) DEFAULT '' COMMENT '更新者', + `update_time` datetime DEFAULT CURRENT_TIMESTAMP ON UPDATE CURRENT_TIMESTAMP COMMENT '更新时间', + PRIMARY KEY (`id`), + UNIQUE KEY `uk_platform_user` (`platform_id`, `platform_user_id`), + KEY `idx_token_expire` (`token_expires_at`), + KEY `idx_init_status` (`init_status`), + KEY `idx_platform_id` (`platform_id`) +) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COMMENT='平台用户认证'; + +-- ---------------------------- +-- 2. 修改 platform_device_mapping 表 +-- ---------------------------- +-- 增加用户认证关联字段 +ALTER TABLE `platform_device_mapping` + ADD COLUMN `platform_user_auth_id` bigint(20) DEFAULT NULL COMMENT '关联用户认证ID' AFTER `platform_id`; + +-- 增加索引 +CREATE INDEX `idx_user_auth` ON `platform_device_mapping`(`platform_user_auth_id`); \ No newline at end of file diff --git a/yack-web/src/api/platform/flow.js b/yack-web/src/api/platform/flow.js index 0a6056fb..37cd4ab6 100644 --- a/yack-web/src/api/platform/flow.js +++ b/yack-web/src/api/platform/flow.js @@ -15,7 +15,3 @@ export function savePlatformFlow(data) { export function delPlatformFlow(id) { return request({ url: `/platform/flow/${id}`, method: 'delete' }) } - -export function executePlatformFlow(id) { - return request({ url: `/platform/flow/${id}/execute`, method: 'post' }) -} diff --git a/yack-web/src/api/platform/instance.js b/yack-web/src/api/platform/instance.js index 98d11aaf..f6d802e5 100644 --- a/yack-web/src/api/platform/instance.js +++ b/yack-web/src/api/platform/instance.js @@ -19,11 +19,3 @@ export function updatePlatformInstance(data) { export function delPlatformInstance(ids) { return request({ url: `/platform/instance/${ids}`, method: 'delete' }) } - -export function initializePlatform(id) { - return request({ url: `/platform/instance/${id}/initialize`, method: 'post' }) -} - -export function getProjectsByPlatformId(id) { - return request({ url: `/platform/instance/${id}/projects`, method: 'get' }) -} diff --git a/yack-web/src/api/platform/userAuth.js b/yack-web/src/api/platform/userAuth.js new file mode 100644 index 00000000..bd57e214 --- /dev/null +++ b/yack-web/src/api/platform/userAuth.js @@ -0,0 +1,33 @@ +import request from '@/utils/request' + +export function listPlatformUserAuth(params) { + return request({ url: '/platform/user-auth/list', method: 'get', params }) +} + +export function getInitializedUsers(platformId) { + return request({ url: `/platform/user-auth/initialized/${platformId}`, method: 'get' }) +} + +export function getPlatformUserAuth(id) { + return request({ url: `/platform/user-auth/${id}`, method: 'get' }) +} + +export function addPlatformUserAuth(data) { + return request({ url: '/platform/user-auth', method: 'post', data }) +} + +export function updatePlatformUserAuth(data) { + return request({ url: '/platform/user-auth', method: 'put', data }) +} + +export function delPlatformUserAuth(ids) { + return request({ url: `/platform/user-auth/${ids}`, method: 'delete' }) +} + +export function initializeUserAuth(id) { + return request({ url: `/platform/user-auth/${id}/initialize`, method: 'post' }) +} + +export function getProjectsByUserAuthId(id) { + return request({ url: `/platform/user-auth/${id}/projects`, method: 'get' }) +} diff --git a/yack-web/src/views/platform/device-mapping/index.vue b/yack-web/src/views/platform/device-mapping/index.vue index 7601378e..24ba7530 100644 --- a/yack-web/src/views/platform/device-mapping/index.vue +++ b/yack-web/src/views/platform/device-mapping/index.vue @@ -134,8 +134,13 @@ + + + + + - + @@ -331,7 +336,8 @@