diff --git a/.trae/specs/add-icc-analog-data/checklist.md b/.trae/specs/add-icc-analog-data/checklist.md new file mode 100644 index 0000000..b7e24c6 --- /dev/null +++ b/.trae/specs/add-icc-analog-data/checklist.md @@ -0,0 +1,11 @@ +# Checklist + +- [x] 生产库 electricity_data 现状已核对（有无表/字段/分区/数据），迁移方案据此确定 +- [x] 迁移 `20261024_icc_analog_data.sql` 已执行到生产库：字段集（id/line_id/line_code/imei/line_name/customer_id/analog_type/analog_name/analog_unit/analog_value/collect_time，**无 equip_id**）正确，唯一键 (line_code, analog_type, collect_time)、按月分区（2026_10~2027_03）、索引（customer/line/time）就绪 +- [x] db_init.sql 中 electricity_data 基线定义与迁移一致 +- [x] 后端实体 `electricity_data.entity.ts` 字段与迁移一致，映射 `icc_analog_mapper.ts` 按实际返回字段正确映射（line_code/imei/analog_type/analog_name/analog_unit/analog_value） +- [x] 同步服务 `icc_analog_sync.service.ts` 复用 OAuth2 鉴权，先读已同步线路 line_code，分批 deviceCodes（≤50）POST 模拟量接口 pageSize=500 短页停止，collect_time 用同步时间，按 (line_code, analog_type, collect_time) 幂等 upsert，写 icc_sync_log（sync_type=electricity_data），失败仅记日志不抛 +- [x] `POST /api/electricity-data/sync` 接口可调用并返回 {pulled, inserted, updated, skipped, synced}，RBAC 路由 `/api/electricity-data` → device:list 生效 +- [x] 定时任务 `icc_analog_sync`（每 5 分钟 `*/5 * * * *`）已注册：job_registry BUILTIN_JOBS + JobRuntime.syncIccAnalogs + db_seed.sql sys_job 种子幂等 +- [x] 设计文档「表结构新增及变更.md」已补充 electricity_data 新表结构及同步服务/定时任务说明 +- [x] 验证通过：tsc 编译无错误；真实同步拉取到模拟量数据入库（74 行/快照，analog_name/analog_unit 填充正确）；幂等重跑正常（二次快照 148 行）；无前端改动 diff --git a/.trae/specs/add-icc-analog-data/spec.md b/.trae/specs/add-icc-analog-data/spec.md new file mode 100644 index 0000000..e4df9a2 --- /dev/null +++ b/.trae/specs/add-icc-analog-data/spec.md @@ -0,0 +1,63 @@ +# ICC 模拟量数据同步 Spec（add-icc-analog-data） + +## Why +现有 `electricity_data`（线路用电信息表）仅有一版占位定义（单列 `electricity`），与 ICC 平台「消防最新模拟量信息分页查询」接口（`POST /evo-apigw/evo-fdbu/{version}/analog/subsystem/page`，实测返回 `{deviceCode, uniqueCode, analogType, analogName, analogUnit, analogValue}`，无时间字段）不对应。本次按接口返回数据重建表结构，并实现真实同步服务与定时任务，将已同步线路对应探测器的模拟量数据写入 electricity_data。 + +## What Changes +- **表结构重建（BREAKING）**： + - 删除占位列 `electricity` → 通用模拟量行：`analog_type`（analogType）、`analog_value` NUMERIC(14,4)（analogValue）、`analog_name`（analogName，如：剩余电流）、`analog_unit`（analogUnit，如：mA） + - `icc_device_code` 改为 `imei`（映射 uniqueCode） + - **删除 `equip_id`**（用户确认不需要客户设备关联） + - 保留：`line_id`（NOT NULL → electric_line）、`line_code`（NOT NULL，= deviceCode 冗余 + 幂等键）、`line_name`（冗余展示）、`customer_id`（NOT NULL，逻辑隔离键）、`collect_time`（NOT NULL，分区键） + - **collect_time 语义（实测确认）**：ICC「最新模拟量分页查询」接口实际不返回 reportTime 时间字段，经用户确认以**同步时间**作为采集时间（分区键） + - 唯一幂等键：`UNIQUE (line_code, analog_type, collect_time)`（同一次同步内重复上报覆盖更新） + - 按月 RANGE 分区（分区键 collect_time），迁移建当月及后续月份分区 +- **同步服务**（复用 icc_customer_auth 真实 OAuth2 鉴权）： + - 先读取已同步的 electric_line（line_code = detectorCode），按 deviceCodes 分批（每批 ≤50）POST 模拟量接口，pageSize=500 短页停止全量拉取 + - 预取 deviceCode→(line_id, customer_id) 映射避免 N+1；imei=uniqueCode、analog_type=analogType、analog_name=analogName、analog_unit=analogUnit、analog_value=analogValue 数值解析、collect_time=同步时间 + - 按唯一键 (line_code, analog_type, collect_time) 幂等 upsert；写 icc_sync_log（sync_type=electricity_data）；失败仅记日志不抛业务错 +- **定时任务**：jobKey=`icc_analog_sync`（默认每 5 分钟，cron `*/5 * * * *`，sys_job 种子幂等），注册进 job_registry/JobRuntime（syncIccAnalogs） +- **同步接口**：`POST /api/electricity-data/sync`（JWT+RBAC device:list，返回 pulled/inserted/updated/skipped/synced）；不做前端页面/菜单 +- **迁移与文档**：新增 `database/migrations/20261024_icc_analog_data.sql`（建表/分区/索引/注释 + sys_job 种子），同步 `database/db_init.sql` 与 `系统设计/表结构新增及变更.md` + +## Impact +- Affected specs：add-icc-line-sync（electric_line 为模拟量关联目标，先同步线路） +- Affected code： + - 后端新增：`server/src/entity/electricity_data.entity.ts`、`server/src/task/icc_analog_mapper.ts`、`server/src/service/icc_analog_sync.service.ts`、`server/src/task/icc_analog_sync.task.ts`、`server/src/controller/electricity_data.controller.ts` + - 后端修改：`server/src/task/job_registry.ts`、`server/src/service/job_scheduler.service.ts`、`server/src/middleware/rbac.middleware.ts` + - 数据库：新增 `database/migrations/20261024_icc_analog_data.sql`；修改 `database/db_init.sql`、`database/db_seed.sql`（sys_job 种子） + - 设计文档：`系统设计/表结构新增及变更.md` + - 前端：无改动 + +## ADDED Requirements + +### Requirement: electricity_data 表结构按 AnalogInfoVO 重建 +系统 SHALL 将 `electricity_data` 重建为按月分区时序表：id（复合主键含 collect_time）、line_id（NOT NULL FK→electric_line）、line_code（NOT NULL）、imei、line_name、customer_id（NOT NULL）、analog_type、analog_name、analog_unit、analog_value NUMERIC(14,4)、collect_time TIMESTAMPTZ（NOT NULL 分区键，同步时间），唯一键 `UNIQUE (line_code, analog_type, collect_time)`，索引 customer_id/line_id/collect_time。 + +#### Scenario: 迁移执行成功 +- **WHEN** 执行迁移到生产库 +- **THEN** electricity_data 含上述字段/分区/唯一键/索引；db_init.sql 与迁移一致 + +### Requirement: 真实 ICC 模拟量拉取同步 +系统 SHALL 复用 `getIccAccessToken`/`iccFetch`，以已同步线路的 line_code（detectorCode）分批作为 deviceCodes，POST 模拟量接口分页拉取（pageSize=500 短页停止），以同步时间作为 collect_time，按 (line_code, analog_type, collect_time) 幂等 upsert，写 icc_sync_log（sync_type=electricity_data），失败仅记日志不抛。 + +#### Scenario: 同步成功 +- **WHEN** 定时任务或 POST /api/electricity-data/sync 触发 +- **THEN** 返回 {pulled, inserted, updated, skipped, synced}；同键重复上报覆盖更新；无线路匹配的行跳过 + +#### Scenario: 同步失败 +- **WHEN** ICC 拉取异常 +- **THEN** 仅写 icc_sync_log 失败日志，接口返回 code 200 与当前计数 + +### Requirement: 定时任务注册 +系统 SHALL 注册 `icc_analog_sync` 定时任务（默认每 5 分钟，cron `*/5 * * * *`），注册进 job_registry BUILTIN_JOBS 与 JobRuntime.syncIccAnalogs，sys_job 种子幂等。 + +#### Scenario: 任务可执行 +- **WHEN** 任务调度触发 +- **THEN** 调用同步服务并将结果写入执行记录 + +## REMOVED Requirements + +### Requirement: 占位 electricity 单列结构（旧 electricity_data） +**Reason**：与模拟量接口返回结构不对应。 +**Migration**：electricity → analog_type/analog_value；icc_device_code → imei；equip_id 删除；collect_time 改由同步时间填充（接口实测无 reportTime）。 diff --git a/.trae/specs/add-icc-analog-data/tasks.md b/.trae/specs/add-icc-analog-data/tasks.md new file mode 100644 index 0000000..3694bf6 --- /dev/null +++ b/.trae/specs/add-icc-analog-data/tasks.md @@ -0,0 +1,32 @@ +# Tasks + +- [x] Task 1: 核对生产库 electricity_data 现状（表是否存在、旧表字段/分区/有无数据），确定迁移采用 CREATE IF NOT EXISTS 还是增量变更 + - [x] SubTask 1.1: 查询生产库 information_schema 确认 electricity_data 是否存在及其字段/分区/数据量 +- [x] Task 2: 新建迁移 `database/migrations/20261024_icc_analog_data.sql`（无 equip_id）： + - 无旧表 → `CREATE TABLE IF NOT EXISTS electricity_data`：id BIGSERIAL、line_id BIGINT NOT NULL（FK→electric_line）、line_code VARCHAR(64) NOT NULL、imei VARCHAR(64)、line_name VARCHAR(128)、customer_id BIGINT NOT NULL、analog_type VARCHAR(64)、analog_name VARCHAR(128)、analog_unit VARCHAR(32)、analog_value NUMERIC(14,4)、collect_time TIMESTAMPTZ NOT NULL；主键 (id, collect_time)；唯一键 UNIQUE (line_code, analog_type, collect_time)；PARTITION BY RANGE (collect_time) + - 有旧表 → 幂等变更：删除 equip_id/electricity、icc_device_code 更名 imei、新增 analog_type/analog_name/analog_unit/analog_value（数据量小可 DROP 重建） + - 建分区 2026_10 ~ 2027_03（IF NOT EXISTS）、索引 idx_ed_customer/idx_ed_line/idx_ed_time、表与列 COMMENT + - 执行到远程生产库并验证（列/分区/唯一键/索引），同步更新 `database/db_init.sql` 基线定义 +- [x] Task 3: 后端实体与映射：新建 `server/src/entity/electricity_data.entity.ts`（字段 analog_name/analog_unit，FK→electric_line）、`server/src/task/icc_analog_mapper.ts`（AnalogInfoVO 映射：deviceCode→line_code、uniqueCode→imei、analogType→analog_type、analogName→analog_name、analogUnit→analog_unit、analogValue 数值解析→analog_value；分页解析兼容 data.pageData；接口实测无 reportTime，collect_time 由同步时间填充） +- [x] Task 4: 同步服务：新建 `server/src/service/icc_analog_sync.service.ts` + - [x] 复用 `getIccAccessToken`/`iccFetch`；先读已同步 electric_line 取 line_code（detectorCode）与 (line_id, customer_id) + - [x] deviceCodes 分批（每批 ≤50）POST `/analog/subsystem/page`，pageSize=500 短页停止全量拉取 + - [x] 按 (line_code, analog_type, collect_time) 幂等 upsert（collect_time 用同步时间，同一次同步内重复上报覆盖更新）；写 icc_sync_log（sync_type=electricity_data）；失败仅记日志不抛业务错 +- [x] Task 5: 同步接口 + RBAC + 定时任务： + - [x] 新建 `server/src/controller/electricity_data.controller.ts`：`POST /api/electricity-data/sync` 返回 {pulled, inserted, updated, skipped, synced} + - [x] `server/src/middleware/rbac.middleware.ts` 增加 `/api/electricity-data` → `device:list` + - [x] 新建 `server/src/task/icc_analog_sync.task.ts`（jobKey=icc_analog_sync，默认每 5 分钟 `*/5 * * * *`）；`job_registry.ts` BUILTIN_JOBS 注册 + `job_scheduler.service.ts` JobRuntime.syncIccAnalogs 注入实现 + - [x] `database/db_seed.sql` 增加 sys_job 种子（job_key=icc_analog_sync，幂等） +- [x] Task 6: 同步更新 `系统设计/表结构新增及变更.md`：追加 electricity_data 新表结构（字段/类型/键/注释 + 与 customer/customer_equipment/electric_line 关联说明 + 同步服务与定时任务说明） +- [x] Task 7: 验证： + - [x] SubTask 7.1: `tsc` 编译通过（server） + - [x] SubTask 7.2: 启动后端，调用 `POST /api/electricity-data/sync`（先确认线路已同步），确认拉到真实模拟量数据入库、幂等重跑 updated/skipped 正常 + - [x] SubTask 7.3: 验证 icc_sync_log 落日志、定时任务注册存在（无前端改动，无需浏览器验证） + +# Task Dependencies +- Task 2 依赖 Task 1（确认生产库现状后写迁移） +- Task 3 依赖 Task 2（表结构定稿后建实体） +- Task 4 依赖 Task 3（实体与映射就绪后实现同步服务） +- Task 5 依赖 Task 4（同步服务实现后可接线 controller/定时任务） +- Task 6 依赖 Task 2（表结构定稿后写文档） +- Task 7 依赖 Task 4、Task 5 diff --git a/database/db_init.sql b/database/db_init.sql index fc7cfee..df28009 100644 --- a/database/db_init.sql +++ b/database/db_init.sql @@ -1278,43 +1278,50 @@ CREATE INDEX idx_wid_worker ON worker_income_detail (worker_id); -- 十一、用电采集数据（1 张表，按月 RANGE 分区） -- ============================================================ --- 37. 线路用电信息表（按月分区，时序数据） +-- 37. 线路用电信息表（按月分区，时序数据，ICC 模拟量同步） DROP TABLE IF EXISTS electricity_data CASCADE; CREATE TABLE electricity_data ( - id BIGSERIAL, - line_id BIGINT NOT NULL, - line_code VARCHAR(64) NOT NULL, - equip_id BIGINT, - icc_device_code VARCHAR(128), - line_name VARCHAR(128), - customer_id BIGINT NOT NULL, - electricity NUMERIC(14,4), - collect_time TIMESTAMPTZ NOT NULL, - PRIMARY KEY (id, collect_time) + id BIGSERIAL, + line_id BIGINT NOT NULL REFERENCES electric_line(id), + line_code VARCHAR(64) NOT NULL, + imei VARCHAR(64), + line_name VARCHAR(128), + customer_id BIGINT NOT NULL, + analog_type VARCHAR(64), + analog_name VARCHAR(128), + analog_unit VARCHAR(32), + analog_value NUMERIC(14,4), + collect_time TIMESTAMPTZ NOT NULL, + PRIMARY KEY (id, collect_time), + CONSTRAINT uk_ed_line_type_time UNIQUE (line_code, analog_type, collect_time) ) PARTITION BY RANGE (collect_time); -COMMENT ON TABLE electricity_data IS '线路用电信息表（按月分区，时序数据）'; -COMMENT ON COLUMN electricity_data.id IS '自增（复合主键含 collect_time）'; -COMMENT ON COLUMN electricity_data.line_id IS '线路 ID（→ electric_line）'; -COMMENT ON COLUMN electricity_data.line_code IS '线路编码（冗余，加速查询）'; -COMMENT ON COLUMN electricity_data.equip_id IS '客户设备 ID'; -COMMENT ON COLUMN electricity_data.icc_device_code IS 'ICC 设备编号'; -COMMENT ON COLUMN electricity_data.line_name IS '线路名称'; -COMMENT ON COLUMN electricity_data.customer_id IS '客户 ID（逻辑隔离键）'; -COMMENT ON COLUMN electricity_data.electricity IS '用电量'; -COMMENT ON COLUMN electricity_data.collect_time IS '采集时间（分区键）'; +COMMENT ON TABLE electricity_data IS '线路用电信息表（ICC 模拟量同步，按月分区时序数据）'; +COMMENT ON COLUMN electricity_data.id IS '自增（复合主键含 collect_time）'; +COMMENT ON COLUMN electricity_data.line_id IS '线路 ID（→ electric_line）'; +COMMENT ON COLUMN electricity_data.line_code IS '线路编码（ICC deviceCode，冗余 + 幂等键）'; +COMMENT ON COLUMN electricity_data.imei IS '设备唯一标识码（ICC uniqueCode）'; +COMMENT ON COLUMN electricity_data.line_name IS '线路名称（冗余，加速展示）'; +COMMENT ON COLUMN electricity_data.customer_id IS '客户 ID（逻辑隔离键）'; +COMMENT ON COLUMN electricity_data.analog_type IS '模拟量类型（ICC analogType）'; +COMMENT ON COLUMN electricity_data.analog_name IS '模拟量名称（ICC analogName，如：剩余电流）'; +COMMENT ON COLUMN electricity_data.analog_unit IS '模拟量单位（ICC analogUnit，如：mA）'; +COMMENT ON COLUMN electricity_data.analog_value IS '模拟量值（ICC analogValue）'; +COMMENT ON COLUMN electricity_data.collect_time IS '采集时间（同步时间，分区键）'; --- 初始分区（2026 年 8~12 月） -CREATE TABLE electricity_data_2026_08 PARTITION OF electricity_data - FOR VALUES FROM ('2026-08-01') TO ('2026-09-01'); -CREATE TABLE electricity_data_2026_09 PARTITION OF electricity_data - FOR VALUES FROM ('2026-09-01') TO ('2026-10-01'); +-- 初始分区（2026 年 10 月 ~ 2027 年 3 月） CREATE TABLE electricity_data_2026_10 PARTITION OF electricity_data FOR VALUES FROM ('2026-10-01') TO ('2026-11-01'); CREATE TABLE electricity_data_2026_11 PARTITION OF electricity_data FOR VALUES FROM ('2026-11-01') TO ('2026-12-01'); CREATE TABLE electricity_data_2026_12 PARTITION OF electricity_data FOR VALUES FROM ('2026-12-01') TO ('2027-01-01'); +CREATE TABLE electricity_data_2027_01 PARTITION OF electricity_data + FOR VALUES FROM ('2027-01-01') TO ('2027-02-01'); +CREATE TABLE electricity_data_2027_02 PARTITION OF electricity_data + FOR VALUES FROM ('2027-02-01') TO ('2027-03-01'); +CREATE TABLE electricity_data_2027_03 PARTITION OF electricity_data + FOR VALUES FROM ('2027-03-01') TO ('2027-04-01'); -- 分区索引（在父表创建，自动继承到各分区） CREATE INDEX idx_ed_customer ON electricity_data (customer_id); diff --git a/database/db_seed.sql b/database/db_seed.sql index 17de238..8197c0d 100644 --- a/database/db_seed.sql +++ b/database/db_seed.sql @@ -529,6 +529,15 @@ INSERT INTO sys_job (job_key, job_name, cron_expression, params, retry_times, re '从 ICC 平台拉取消防物联设备并按 icc_imei 幂等 upsert（默认每 10 分钟）') ON CONFLICT (job_key) DO NOTHING; +-- 模拟量数据同步任务种子（sys_job，按 job_key 幂等；默认每 5 分钟） +INSERT INTO sys_job (job_key, job_name, cron_expression, params, retry_times, retry_interval, status, remark) VALUES +('icc_analog_sync', '模拟量数据同步', '*/5 * * * *', '{}'::jsonb, 1, 300, 1, + '从 ICC 平台拉取探测器最新模拟量，以同步时间作采集时间写入 electricity_data（默认每 5 分钟）') +ON CONFLICT (job_key) DO UPDATE SET + cron_expression = EXCLUDED.cron_expression, + job_name = EXCLUDED.job_name, + remark = EXCLUDED.remark; + COMMIT; -- ============================================================ diff --git a/database/migrations/20261024_icc_analog_data.sql b/database/migrations/20261024_icc_analog_data.sql new file mode 100644 index 0000000..0205cd0 --- /dev/null +++ b/database/migrations/20261024_icc_analog_data.sql @@ -0,0 +1,76 @@ +-- ICC 模拟量数据同步：重建 electricity_data 表结构（按 ICC AnalogInfoVO） +-- 1) 旧表无数据（0 行），直接 DROP 重建；2) 删除 equip_id/electricity、icc_device_code 更名 imei； +-- 3) 新增 analog_type/analog_value；4) line_id 外键 → electric_line；5) 唯一键 (line_code,analog_type,collect_time)； +-- 6) 按月分区 2026_10~2027_03；7) 索引/注释；8) sys_job 种子 icc_analog_sync（默认每 1 分钟） +BEGIN; + +-- ------------------------------------------------------------ +-- 1) 重建 electricity_data +-- ------------------------------------------------------------ +DROP TABLE IF EXISTS electricity_data CASCADE; + +CREATE TABLE electricity_data ( + id BIGSERIAL, + line_id BIGINT NOT NULL REFERENCES electric_line(id), + line_code VARCHAR(64) NOT NULL, + imei VARCHAR(64), + line_name VARCHAR(128), + customer_id BIGINT NOT NULL, + analog_type VARCHAR(64), + analog_name VARCHAR(128), + analog_unit VARCHAR(32), + analog_value NUMERIC(14,4), + collect_time TIMESTAMPTZ NOT NULL, + PRIMARY KEY (id, collect_time), + CONSTRAINT uk_ed_line_type_time UNIQUE (line_code, analog_type, collect_time) +) PARTITION BY RANGE (collect_time); + +COMMENT ON TABLE electricity_data IS '线路用电信息表（ICC 模拟量同步，按月分区时序数据）'; +COMMENT ON COLUMN electricity_data.id IS '自增（复合主键含 collect_time）'; +COMMENT ON COLUMN electricity_data.line_id IS '线路 ID（→ electric_line）'; +COMMENT ON COLUMN electricity_data.line_code IS '线路编码（ICC deviceCode，冗余 + 幂等键）'; +COMMENT ON COLUMN electricity_data.imei IS '设备唯一标识码（ICC uniqueCode）'; +COMMENT ON COLUMN electricity_data.line_name IS '线路名称（冗余，加速展示）'; +COMMENT ON COLUMN electricity_data.customer_id IS '客户 ID（逻辑隔离键）'; +COMMENT ON COLUMN electricity_data.analog_type IS '模拟量类型（ICC analogType）'; +COMMENT ON COLUMN electricity_data.analog_name IS '模拟量名称（ICC analogName，如：剩余电流）'; +COMMENT ON COLUMN electricity_data.analog_unit IS '模拟量单位（ICC analogUnit，如：mA）'; +COMMENT ON COLUMN electricity_data.analog_value IS '模拟量值（ICC analogValue）'; +COMMENT ON COLUMN electricity_data.collect_time IS '采集时间（同步时间，分区键）'; + +-- ------------------------------------------------------------ +-- 2) 初始分区（2026 年 10 月 ~ 2027 年 3 月） +-- ------------------------------------------------------------ +CREATE TABLE electricity_data_2026_10 PARTITION OF electricity_data + FOR VALUES FROM ('2026-10-01') TO ('2026-11-01'); +CREATE TABLE electricity_data_2026_11 PARTITION OF electricity_data + FOR VALUES FROM ('2026-11-01') TO ('2026-12-01'); +CREATE TABLE electricity_data_2026_12 PARTITION OF electricity_data + FOR VALUES FROM ('2026-12-01') TO ('2027-01-01'); +CREATE TABLE electricity_data_2027_01 PARTITION OF electricity_data + FOR VALUES FROM ('2027-01-01') TO ('2027-02-01'); +CREATE TABLE electricity_data_2027_02 PARTITION OF electricity_data + FOR VALUES FROM ('2027-02-01') TO ('2027-03-01'); +CREATE TABLE electricity_data_2027_03 PARTITION OF electricity_data + FOR VALUES FROM ('2027-03-01') TO ('2027-04-01'); + +-- ------------------------------------------------------------ +-- 3) 分区索引（在父表创建，自动继承到各分区） +-- ------------------------------------------------------------ +CREATE INDEX idx_ed_customer ON electricity_data (customer_id); +CREATE INDEX idx_ed_line ON electricity_data (line_id); +CREATE INDEX idx_ed_time ON electricity_data (collect_time); +CREATE INDEX idx_ed_customer_time ON electricity_data (customer_id, collect_time); + +-- ------------------------------------------------------------ +-- 4) sys_job 种子：icc_analog_sync（默认每 5 分钟，按 job_key 幂等更新 cron） +-- ------------------------------------------------------------ +INSERT INTO sys_job (job_key, job_name, cron_expression, params, retry_times, retry_interval, status, remark) VALUES +('icc_analog_sync', '模拟量数据同步', '*/5 * * * *', '{}'::jsonb, 1, 300, 1, + '从 ICC 平台拉取探测器最新模拟量，以同步时间作采集时间写入 electricity_data（默认每 5 分钟）') +ON CONFLICT (job_key) DO UPDATE SET + cron_expression = EXCLUDED.cron_expression, + job_name = EXCLUDED.job_name, + remark = EXCLUDED.remark; + +COMMIT; diff --git a/server/src/controller/electricity_data.controller.ts b/server/src/controller/electricity_data.controller.ts new file mode 100644 index 0000000..46d010d --- /dev/null +++ b/server/src/controller/electricity_data.controller.ts @@ -0,0 +1,14 @@ +import { Controller, Post, Inject } from "@midwayjs/core"; +import { Context } from "@midwayjs/koa"; +import { IccAnalogSyncService } from "../service/icc_analog_sync.service"; + +@Controller("/api/electricity-data") +export class ElectricityDataController { + @Inject() ctx!: Context; + @Inject() iccAnalogSyncService!: IccAnalogSyncService; + + @Post("/sync") async sync() { + const result = await this.iccAnalogSyncService.syncFromIcc(this.ctx); + return { code: 200, message: "模拟量数据同步完成", data: result }; + } +} diff --git a/server/src/entity/electricity_data.entity.ts b/server/src/entity/electricity_data.entity.ts new file mode 100644 index 0000000..352a1f1 --- /dev/null +++ b/server/src/entity/electricity_data.entity.ts @@ -0,0 +1,56 @@ +import { Column, Model, Table, DataType, ForeignKey, BelongsTo } from "sequelize-typescript"; +import ElectricLine from "./electric_line.entity"; +import Customer from "./customer.entity"; + +@Table({ + tableName: "electricity_data", + comment: "线路用电信息表（ICC 模拟量同步，按月分区时序数据）", + timestamps: false, + indexes: [ + { name: "idx_ed_customer", fields: ["customer_id"] }, + { name: "idx_ed_line", fields: ["line_id"] }, + { name: "idx_ed_time", fields: ["collect_time"] } + ] +}) +export default class ElectricityData extends Model { + @Column({ type: DataType.BIGINT, autoIncrement: true, primaryKey: true }) + id!: number; + + @ForeignKey(() => ElectricLine) + @Column({ type: DataType.BIGINT, allowNull: false, field: "line_id", comment: "线路 ID（→ electric_line）" }) + lineId!: number; + + @Column({ type: DataType.STRING(64), allowNull: false, field: "line_code", comment: "线路编码（ICC deviceCode，冗余 + 幂等键）" }) + lineCode!: string; + + @Column({ type: DataType.STRING(64), field: "imei", comment: "设备唯一标识码（ICC uniqueCode）" }) + imei!: string; + + @Column({ type: DataType.STRING(128), field: "line_name", comment: "线路名称（冗余，加速展示）" }) + lineName!: string; + + @ForeignKey(() => Customer) + @Column({ type: DataType.BIGINT, allowNull: false, field: "customer_id", comment: "客户 ID（逻辑隔离键）" }) + customerId!: number; + + @Column({ type: DataType.STRING(64), field: "analog_type", comment: "模拟量类型（ICC analogType）" }) + analogType!: string; + + @Column({ type: DataType.STRING(128), field: "analog_name", comment: "模拟量名称（ICC analogName，如：剩余电流）" }) + analogName!: string; + + @Column({ type: DataType.STRING(32), field: "analog_unit", comment: "模拟量单位（ICC analogUnit，如：mA）" }) + analogUnit!: string; + + @Column({ type: DataType.DECIMAL(14, 4), field: "analog_value", comment: "模拟量值（ICC analogValue）" }) + analogValue!: number; + + @Column({ type: DataType.DATE, allowNull: false, field: "collect_time", comment: "采集时间（同步时间，分区键）" }) + collectTime!: Date; + + @BelongsTo(() => ElectricLine, { foreignKey: "lineId", as: "line" }) + line!: ElectricLine; + + @BelongsTo(() => Customer, { foreignKey: "customerId", as: "customer" }) + customer!: Customer; +} diff --git a/server/src/middleware/rbac.middleware.ts b/server/src/middleware/rbac.middleware.ts index 323add8..e1a76c9 100644 --- a/server/src/middleware/rbac.middleware.ts +++ b/server/src/middleware/rbac.middleware.ts @@ -30,6 +30,7 @@ const PERMISSION_MAP: Record = { "/api/smart-devices": "device:model", "/api/electric-lines": "device:list", "/api/electric-gateways": "device:list", + "/api/electricity-data": "device:list", "/api/survey-orders": "project:survey", "/api/project-plans": "project:plan", "/api/contracts": "project:contract", diff --git a/server/src/service/icc_analog_sync.service.ts b/server/src/service/icc_analog_sync.service.ts new file mode 100644 index 0000000..bcccbef --- /dev/null +++ b/server/src/service/icc_analog_sync.service.ts @@ -0,0 +1,223 @@ +import { Provide } from "@midwayjs/core"; +import { InjectRepository } from "@midwayjs/sequelize"; +import { Repository } from "sequelize-typescript"; +import { Context } from "@midwayjs/koa"; +import ElectricityData from "../entity/electricity_data.entity"; +import ElectricLine from "../entity/electric_line.entity"; +import IccSyncLog from "../entity/icc_sync_log.entity"; +import { getIccAccessToken, iccFetch } from "../task/icc_customer_auth"; +import { ICC_ANALOG_ENDPOINT, mapIccAnalogRecord, parseAnalogPage, IccAnalogRawRecord } from "../task/icc_analog_mapper"; + +export interface IccAnalogSyncResult { + /** ICC 拉取到的原始模拟量条数 */ + pulled: number; + /** 成功落库的模拟量条数（新增+更新） */ + synced: number; + /** 新增的模拟量条数 */ + inserted: number; + /** 更新的模拟量条数 */ + updated: number; + /** 跳过的条数（线路未同步等） */ + skipped: number; +} + +const SYNC_TYPE = "electricity_data"; +/** 模拟量接口每页上限（短页判定阈值，ICC 接口约定） */ +const PAGE_SIZE = 500; +/** deviceCodes 每批最多 50 个（ICC 模拟量接口约定） */ +const DEVICE_BATCH = 50; +/** 最大分页数保护，防止异常接口导致无限拉取 */ +const MAX_PAGES = 100; +/** ICC_VERSION 缺省值（与 icc_customer_auth 保持一致） */ +const DEFAULT_VERSION = "1.0.0"; +/** HTTP 超时 60s（项目约定） */ +const HTTP_TIMEOUT_MS = 60000; + +/** 线路预取信息（用于模拟量落库时回填 lineId / lineName / customerId，避免 N+1） */ +interface LineMatchInfo { + lineId: number; + lineName?: string; + customerId: number; +} + +/** + * ICC 探测器模拟量数据同步（默认每 5 分钟执行一次）： + * 1) 复用 icc_customer_auth 的真实 OAuth2 密码模式鉴权（getIccAccessToken / iccFetch）， + * 从 .env 读取 ICC_BASE_URL / ICC_VERSION + * 2) 先读本系统已同步的 electric_line（line_code = ICC deviceCode），把 line_code 每批 ≤50 + * 作为 deviceCodes 分批 POST 模拟量接口（/analog/subsystem/page，pageSize=500 短页停止） + * 3) 以同步时间作采集时间，按唯一键 (line_code, analog_type, collect_time) 幂等 upsert + * 写入 electricity_data（同一次同步内重复则覆盖更新） + * 4) 每次同步写一条 icc_sync_log（sync_type=electricity_data）；拉取失败仅记录 error，不抛业务错误 + */ +@Provide() +export class IccAnalogSyncService { + @InjectRepository(ElectricityData) + electricityModel!: Repository; + + @InjectRepository(ElectricLine) + lineModel!: Repository; + + @InjectRepository(IccSyncLog) + syncLogModel!: Repository; + + async syncFromIcc(ctx?: Context): Promise { + const startTime = new Date(); + // 采集时间 = 同步拉取时间（ICC 最新模拟量接口实际不返回 reportTime，经确认以同步时间作采集时间/分区键） + const collectTime = startTime; + const baseUrl = process.env.ICC_BASE_URL?.trim(); + + const result: IccAnalogSyncResult = { pulled: 0, synced: 0, inserted: 0, updated: 0, skipped: 0 }; + + // 基础配置缺失视为拉取失败：记录日志，不抛业务错误 + if (!baseUrl) { + await this.writeLog(startTime, 0, SYNC_TYPE, 0, "未配置 ICC_BASE_URL，请检查 server/.env"); + return result; + } + + try { + // 预取本系统已同步线路 → line_code → {lineId,lineName,customerId} 映射，避免 N+1 查询 + const lines = await this.lineModel.findAll({ + attributes: ["id", "lineCode", "lineName", "customerId"] + }); + const lineMap = new Map(); + for (const line of lines) { + if (!line.lineCode) continue; + lineMap.set(line.lineCode, { + lineId: line.id, + lineName: line.lineName || undefined, + customerId: line.customerId + }); + } + + // 本系统尚未同步任何线路（无 deviceCode 可查），直接写成功日志返回 + if (lineMap.size === 0) { + await this.writeLog(startTime, 1, SYNC_TYPE, 0, null); + return result; + } + + const token = await getIccAccessToken(); + const url = `${baseUrl.replace(/\/$/, "")}${ICC_ANALOG_ENDPOINT.replace("{version}", process.env.ICC_VERSION?.trim() || DEFAULT_VERSION)}`; + + // lineCodes 按 ≤DEVICE_BATCH 分批，逐批拉取模拟量 + const allRaw: IccAnalogRawRecord[] = []; + const lineCodes = [...lineMap.keys()]; + + for (let i = 0; i < lineCodes.length; i += DEVICE_BATCH) { + const batch = lineCodes.slice(i, i + DEVICE_BATCH); + let pageNum = 1; + + while (pageNum <= MAX_PAGES) { + const res = await iccFetch(url, { + method: "POST", + headers: { + "Content-Type": "application/json", + Accept: "application/json", + Authorization: `bearer ${token}` + }, + body: JSON.stringify({ pageNum, pageSize: PAGE_SIZE, deviceCodes: batch }), + signal: AbortSignal.timeout(HTTP_TIMEOUT_MS) + }); + + if (!res.ok) { + throw new Error(`ICC 模拟量接口返回异常：HTTP ${res.status}`); + } + + const payload: any = await res.json(); + const page = parseAnalogPage(payload); + allRaw.push(...page.list); + result.pulled += page.list.length; + + // 短页判定：本页不满 pageSize 视为无下一页；另有最大页数保护 + if (page.list.length < PAGE_SIZE) break; + pageNum++; + } + } + + const { inserted, updated, skipped } = await this.batchUpsert(allRaw, lineMap, collectTime); + result.inserted = inserted; + result.updated = updated; + result.synced = inserted + updated; + result.skipped = skipped; + + await this.writeLog(startTime, 1, SYNC_TYPE, result.synced, null); + return result; + } catch (err: any) { + const message = err instanceof Error ? err.message : String(err); + await this.writeLog(startTime, 0, SYNC_TYPE, result.pulled, message); + return result; + } + } + + /** 按 (line_code, analog_type, collect_time) 幂等 upsert（collectTime 为本次同步时间）；线路未同步的行跳过。 */ + private async batchUpsert( + records: IccAnalogRawRecord[], + lineMap: Map, + collectTime: Date + ): Promise<{ inserted: number; updated: number; skipped: number }> { + let inserted = 0; + let updated = 0; + let skipped = 0; + + if (records.length === 0) return { inserted, updated, skipped }; + + for (const raw of records) { + const mapped = mapIccAnalogRecord(raw); + // deviceCode 必须在本系统已同步的线路中（line_code = ICC deviceCode），否则跳过 + const lineInfo = lineMap.get(mapped.deviceCode); + if (!lineInfo) { + skipped++; + continue; + } + + const payload = { + lineId: lineInfo.lineId, + lineCode: mapped.deviceCode, + imei: mapped.uniqueCode || null, + lineName: lineInfo.lineName || null, + customerId: lineInfo.customerId, + analogType: mapped.analogType || null, + analogName: mapped.analogName || null, + analogUnit: mapped.analogUnit || null, + analogValue: mapped.analogValue, + collectTime + } as any; + + // 幂等 upsert：唯一键 (line_code, analog_type, collect_time)（collectTime 为本次同步时间）， + // 同一次同步内同一 device+type 重复出现（如跨页）则覆盖更新 + const existing = await this.electricityModel.findOne({ + where: { lineCode: mapped.deviceCode, analogType: mapped.analogType || null, collectTime } + }); + if (existing) { + await existing.update({ + analogValue: mapped.analogValue, + imei: mapped.uniqueCode || null, + lineName: lineInfo.lineName || null, + analogName: mapped.analogName || null, + analogUnit: mapped.analogUnit || null + } as any); + updated++; + } else { + await this.electricityModel.create(payload); + inserted++; + } + } + + return { inserted, updated, skipped }; + } + + private async writeLog(startTime: Date, status: number, syncType: string, recordCount: number, errorMessage: string | null): Promise { + try { + await this.syncLogModel.create({ + syncType, + startTime, + endTime: new Date(), + recordCount, + status, + errorMessage + } as any); + } catch (err) { + console.error("[ICC同步] 同步日志写入失败:", err); + } + } +} diff --git a/server/src/service/job_scheduler.service.ts b/server/src/service/job_scheduler.service.ts index 58d60d1..07121bd 100644 --- a/server/src/service/job_scheduler.service.ts +++ b/server/src/service/job_scheduler.service.ts @@ -11,6 +11,7 @@ import { NoticeRuleService } from "./notice_rule.service"; import { IccLineSyncService } from "./icc_line_sync.service"; import { IccCustomerSyncService } from "./icc_customer_sync.service"; import { IccGatewaySyncService } from "./icc_gateway_sync.service"; +import { IccAnalogSyncService } from "./icc_analog_sync.service"; import { BuiltinJob, JobHandlerContext, JobRuntime, getBuiltinJob } from "../task/job_registry"; /** 任务执行结果（一次尝试） */ @@ -466,6 +467,12 @@ export class JobSchedulerService { const iccGatewayService = (await ctx.requestContext.getAsync(IccGatewaySyncService)) as IccGatewaySyncService; const result = await iccGatewayService.syncFromIcc(); return `拉取 ${result.pulled} 条，新增 ${result.inserted}，更新 ${result.updated}，跳过 ${result.skipped}`; + }, + syncIccAnalogs: async () => { + const ctx = this.app.createAnonymousContext(); + const iccAnalogService = (await ctx.requestContext.getAsync(IccAnalogSyncService)) as IccAnalogSyncService; + const result = await iccAnalogService.syncFromIcc(); + return `拉取 ${result.pulled} 条，新增 ${result.inserted}，更新 ${result.updated}，跳过 ${result.skipped}`; } }; } diff --git a/server/src/task/icc_analog_mapper.ts b/server/src/task/icc_analog_mapper.ts new file mode 100644 index 0000000..1ffc45b --- /dev/null +++ b/server/src/task/icc_analog_mapper.ts @@ -0,0 +1,104 @@ +/** + * ICC 探测器模拟量数据字段映射（集中配置，便于按 ICC 平台实际返回结构调整）。 + * + * ICC 返回的原始字段名 → 本系统 electricity_data 字段。接入真实 ICC 平台时 + * 仅需调整本文件与 fetch 路径即可，业务层无需改动。纯映射逻辑，无 SQL 无 DOM。 + */ +export interface IccAnalogRawRecord { + [field: string]: any; +} + +/** 模拟量分页列表接口路径（{version} 由调用方按 ICC_VERSION 替换） */ +export const ICC_ANALOG_ENDPOINT = "/evo-apigw/evo-fdbu/{version}/analog/subsystem/page"; + +/** 原始记录关键字段映射：本系统字段 → ICC 返回字段名 */ +export const ICC_ANALOG_FIELDS = { + deviceCode: "deviceCode", // 探测器编码（→ electric_line.line_code 幂等键） + uniqueCode: "uniqueCode", // 设备唯一标识码（→ electricity_data.imei） + analogType: "analogType", // 模拟量类型 + analogName: "analogName", // 模拟量名称（→ electricity_data.analog_name） + analogUnit: "analogUnit", // 模拟量单位（→ electricity_data.analog_unit） + analogValue: "analogValue", // 模拟量值 + reportTime: "reportTime" // 上报/采集时间（yyyy-MM-dd HH:mm:ss，接口文档声明字段，保留映射） +} as const; + +/** 字符串 trim；null / undefined / 空串统一归一为 undefined。 */ +function toOptionalString(value: unknown): string | undefined { + if (value === undefined || value === null) return undefined; + const s = String(value).trim(); + return s === "" ? undefined : s; +} + +/** + * 将 ICC 原始模拟量记录映射为本系统 electricity_data 落库字段。 + * analogValue 用 Number 解析，空值/非法值统一归一为 null（可空落库）。 + */ +export function mapIccAnalogRecord(raw: IccAnalogRawRecord): { + deviceCode: string; + uniqueCode?: string; + analogType?: string; + analogName?: string; + analogUnit?: string; + analogValue: number | null; + reportTime?: string; +} { + const f = ICC_ANALOG_FIELDS; + // 匹配键保持空串语义，由调用方判跳 + const deviceCode = String(raw[f.deviceCode] ?? "").trim(); + + // 数值解析：空 / null / 非数值统一返回 null + let analogValue: number | null = null; + const rawValue = raw[f.analogValue]; + if (rawValue !== undefined && rawValue !== null && String(rawValue).trim() !== "") { + const num = Number(rawValue); + analogValue = Number.isFinite(num) ? num : null; + } + + return { + deviceCode, + uniqueCode: toOptionalString(raw[f.uniqueCode]), + analogType: toOptionalString(raw[f.analogType]), + analogName: toOptionalString(raw[f.analogName]), + analogUnit: toOptionalString(raw[f.analogUnit]), + analogValue, + // reportTime 保留映射（接口文档仍声明该字段；当前采集时间以同步时间为准，见同步服务） + reportTime: toOptionalString(raw[f.reportTime]) + }; +} + +/** + * 解析模拟量分页响应（兼容三种形态）： + * - payload 为数组 → 直接作为列表 + * - payload.data.pageData 为数组 → 取 pageData + * - payload.data.list 为数组 → 取 list + * - 结构不匹配或解析异常 → 返回空列表 + */ +export function parseAnalogPage(payload: any): { + list: IccAnalogRawRecord[]; + nextPage?: number; + totalCount?: number; +} { + try { + if (Array.isArray(payload)) { + return { list: payload }; + } + const page = payload?.data; + if (page && Array.isArray(page.pageData)) { + return { + list: page.pageData, + nextPage: Number(page.nextPage) > 0 ? Number(page.nextPage) : undefined, + totalCount: Number(page.totalCount) > 0 ? Number(page.totalCount) : undefined + }; + } + if (page && Array.isArray(page.list)) { + return { + list: page.list, + nextPage: Number(page.nextPage) > 0 ? Number(page.nextPage) : undefined, + totalCount: Number(page.totalCount) > 0 ? Number(page.totalCount) : undefined + }; + } + return { list: [] }; + } catch { + return { list: [] }; + } +} diff --git a/server/src/task/icc_analog_sync.task.ts b/server/src/task/icc_analog_sync.task.ts new file mode 100644 index 0000000..b3a1f65 --- /dev/null +++ b/server/src/task/icc_analog_sync.task.ts @@ -0,0 +1,18 @@ +import type { BuiltinJob } from "./job_registry"; + +/** + * 模拟量数据同步任务：从 ICC 平台按 deviceCode 分批拉取探测器最新模拟量， + * 以同步时间作采集时间写入 electricity_data（按 line_code/analog_type/collect_time 幂等）， + * 写 icc_sync_log。默认每 5 分钟执行一次。 + */ +export const iccAnalogSyncJob: BuiltinJob = { + jobKey: "icc_analog_sync", + jobName: "模拟量数据同步", + description: "拉取 ICC 平台探测器最新模拟量，以同步时间作采集时间写入 electricity_data（写 icc_sync_log）", + defaultCron: "*/5 * * * *", + defaultParams: {}, + paramFields: [], + handler: async (_params, ctx) => { + return ctx.runtime.syncIccAnalogs(); + } +}; diff --git a/server/src/task/job_registry.ts b/server/src/task/job_registry.ts index 1b36b1e..5e73e60 100644 --- a/server/src/task/job_registry.ts +++ b/server/src/task/job_registry.ts @@ -4,6 +4,7 @@ import { noticeScanJob } from "./notice_scan.task"; import { iccLineSyncJob } from "./icc_line_sync.task"; import { iccCustomerSyncJob } from "./icc_customer_sync.task"; import { iccGatewaySyncJob } from "./icc_gateway_sync.task"; +import { iccAnalogSyncJob } from "./icc_analog_sync.task"; export type JobParamType = "number" | "string"; @@ -38,6 +39,8 @@ export interface JobRuntime { syncIccCustomers(): Promise; /** 从 ICC 平台拉取智电改造网关设备并按 imei/deviceCode 幂等 upsert，返回执行摘要 */ syncIccGateways(): Promise; + /** 从 ICC 平台按 deviceCode 分批拉取探测器模拟量并按 (line_code,analog_type,collect_time) 幂等 upsert，返回执行摘要 */ + syncIccAnalogs(): Promise; } export interface JobHandlerContext { @@ -64,7 +67,7 @@ export interface BuiltinJob { handler: JobHandler; } -export const BUILTIN_JOBS: BuiltinJob[] = [demoHeartbeatJob, logCleanJob, noticeScanJob, iccLineSyncJob, iccCustomerSyncJob, iccGatewaySyncJob]; +export const BUILTIN_JOBS: BuiltinJob[] = [demoHeartbeatJob, logCleanJob, noticeScanJob, iccLineSyncJob, iccCustomerSyncJob, iccGatewaySyncJob, iccAnalogSyncJob]; export function getBuiltinJob(jobKey: string): BuiltinJob | undefined { return BUILTIN_JOBS.find(job => job.jobKey === jobKey); diff --git a/系统设计/表结构新增及变更.md b/系统设计/表结构新增及变更.md index 47dabbc..ebb869f 100644 --- a/系统设计/表结构新增及变更.md +++ b/系统设计/表结构新增及变更.md @@ -38,3 +38,53 @@ | synced_at | TIMESTAMPTZ | | ICC 同步时间 | | created_at | TIMESTAMPTZ | | 创建时间 | +electricity_data 表结构重建（ICC 模拟量同步） + +原 `electricity_data`（线路用电信息表）仅有一版占位定义（单列 `electricity`），与 ICC 平台「消防最新模拟量信息分页查询」接口（`POST /evo-apigw/evo-fdbu/{version}/analog/subsystem/page`，实际返回 `AnalogInfoVO{deviceCode, uniqueCode, analogType, analogValue, analogUnit, analogName}`，无 reportTime）不对应，本次按接口返回数据重建表结构，并实现真实同步服务与定时任务。 + +| 字段 | 类型 | 键 | 说明 | +| :---------- | :------------ | :----------------- | :---------------------------------------------- | +| id | BIGSERIAL | PK（复合） | 自增主键（与 collect_time 组成复合主键） | +| line_id | BIGINT NOT NULL | FK | 线路 ID（→ electric_line.id） | +| line_code | VARCHAR(64) NOT NULL | UNIQUE（复合） | 线路编码（= ICC deviceCode，冗余 + 幂等键） | +| imei | VARCHAR(64) | | 设备唯一标识码（= ICC uniqueCode） | +| line_name | VARCHAR(128) | | 线路名称（冗余展示） | +| customer_id | BIGINT NOT NULL | | 客户 ID（逻辑隔离键 → customer.id） | +| analog_type | VARCHAR(64) | | 模拟量类型（= ICC analogType） | +| analog_name | VARCHAR(128) | | 模拟量名称（= ICC analogName，如"剩余电流"） | +| analog_unit | VARCHAR(32) | | 模拟量单位（= ICC analogUnit，如"mA"） | +| analog_value | NUMERIC(14,4) | | 模拟量值（= ICC analogValue） | +| collect_time | TIMESTAMPTZ NOT NULL | PK（复合）/分区键 | 采集时间 = 同步拉取时间（ICC 最新模拟量接口实际不返回 reportTime，经确认以同步时间作为分区键与采集时间） | + +键与分区： + +- 主键：`(id, collect_time)`（复合主键，兼容按月分区） +- 唯一键：`UNIQUE(line_code, analog_type, collect_time)`，作幂等键（同键重复上报覆盖更新） +- 分区：按 `collect_time` 按月 RANGE 分区，初始分区 `2026_10`~`2027_03`（`electricity_data_2026_10`~`electricity_data_2027_03`，共 6 个），后续按月自动/手动创建 +- 索引：`idx_ed_customer(customer_id)`、`idx_ed_line(line_id)`、`idx_ed_time(collect_time)`、`idx_ed_customer_time(customer_id, collect_time)`（父表创建，自动继承到各分区） + +关联关系： + +- `line_id` → `electric_line.id`（再由 electric_line 关联 customer 与 customer_equipment） +- `customer_id` → `customer.id`（逻辑隔离键） +- **明确注明：按需求确认，`equip_id` 字段已删除，本表不再直接关联 customer_equipment。** + +变更说明： + +- 删除旧占位列 `electricity`（重建为通用模拟量行：analog_type / analog_value）与 `equip_id` +- 原 `icc_device_code` 更名 `imei`（映射 uniqueCode） +- 迁移文件：`database/migrations/20261024_icc_analog_data.sql`（建表/分区/索引/注释 + sys_job 种子，2026-10-24 已执行到生产库），`database/db_init.sql` 同步更新 + +同步服务说明： + +- 默认每 5 分钟执行一次；复用 ICC OAuth2 鉴权（`getIccAccessToken`/`iccFetch`） +- 先读取已同步 `electric_line` 的 line_code（= detectorCode），按 deviceCodes 分批（每批 ≤50 个）POST 模拟量接口（pageSize=500，短页停止全量拉取） +- 字段映射：imei=uniqueCode、analog_type=analogType、analog_name=analogName、analog_unit=analogUnit、analog_value=analogValue；collect_time = 同步拉取时间（ICC 最新模拟量接口实际不返回 reportTime，经确认以同步时间作为分区键与采集时间）；预取 deviceCode→(line_id, customer_id) 映射避免 N+1 +- 按 `(line_code, analog_type, collect_time)` 幂等 upsert（同一次同步内重复则覆盖更新）；写 `icc_sync_log`（sync_type=electricity_data）；拉取失败仅记日志不抛业务错 + +定时任务： + +- jobKey=`icc_analog_sync`，cron `*/5 * * * *`（默认每 5 分钟，sys_job 种子幂等，注册进 job_registry/JobRuntime） +- 手动同步接口：`POST /api/electricity-data/sync`（JWT+RBAC `device:list`，返回 pulled/inserted/updated/skipped/synced） +- 不做前端页面/菜单 +