feat(line): 智电改造线路台账接入 ICC 消防探测器真实同步
- 重建 electric_line 表结构(迁移20261024_icc_line_rebuild.sql并同步db_init/db_seed):删building_id/floor_no/location/icc_device_code占位字段,equip_id改可空,新增imei/gateway_id/type_code/type_name/fire_system_code/fire_system_name/part_unit_loop_code/channel_no/channel_sn/produce_date,gateway_id外键→electric_gateway - 同步服务IccLineSyncService复用OAuth2真实鉴权分页拉取探测器(POST /detector/subsystem/page,pageSize=50短页停止),按line_code(detectorCode)幂等upsert写icc_sync_log;customer_id由orgCode→customer.icc_code解析、gateway_id由parentCode→electric_gateway.device_code匹配、equip_id同步不填充 - 列表接口返回gateway/type/imei等新字段,keyword检索含imei;前端台账页更新为新列(IMEI/所属网关/探测器类型/回路通道/消防系统/生产日期),保留手动同步按钮与筛选 - 更新系统设计文档electric_line表定义与spec(add-icc-line-sync)
This commit is contained in:
1 parent
cf0a0d4d31
commit
160ad85051
13 files changed
+510
-200
No files matched your search
@@ -4,9 +4,9 @@ import { Repository } from "sequelize-typescript";
|
||||
import { Op } from "sequelize";
|
||||
import { Context } from "@midwayjs/koa";
|
||||
import ElectricLine from "../entity/electric_line.entity";
|
||||
import ElectricGateway from "../entity/electric_gateway.entity";
|
||||
import Customer from "../entity/customer.entity";
|
||||
import CustomerEquipment from "../entity/customer_equipment.entity";
|
||||
import CustomerBuilding from "../entity/customer_building.entity";
|
||||
import SysUserRole from "../entity/sysUserRole.entity";
|
||||
import SysRole from "../entity/sysRole.entity";
|
||||
|
||||
@@ -21,8 +21,8 @@ export class ElectricLineService {
|
||||
@InjectRepository(CustomerEquipment)
|
||||
equipmentModel!: Repository<CustomerEquipment>;
|
||||
|
||||
@InjectRepository(CustomerBuilding)
|
||||
buildingModel!: Repository<CustomerBuilding>;
|
||||
@InjectRepository(ElectricGateway)
|
||||
gatewayModel!: Repository<ElectricGateway>;
|
||||
|
||||
@InjectRepository(SysUserRole)
|
||||
userRoleModel!: Repository<SysUserRole>;
|
||||
@@ -44,12 +44,12 @@ export class ElectricLineService {
|
||||
if (query.keyword) {
|
||||
where[Op.or] = [
|
||||
{ lineName: { [Op.like]: `%${query.keyword}%` } },
|
||||
{ lineCode: { [Op.like]: `%${query.keyword}%` } }
|
||||
{ lineCode: { [Op.like]: `%${query.keyword}%` } },
|
||||
{ imei: { [Op.like]: `%${query.keyword}%` } }
|
||||
];
|
||||
}
|
||||
if (query.customerId !== undefined && query.customerId !== "") where.customerId = Number(query.customerId);
|
||||
if (query.equipId !== undefined && query.equipId !== "") where.equipId = Number(query.equipId);
|
||||
if (query.buildingId !== undefined && query.buildingId !== "") where.buildingId = Number(query.buildingId);
|
||||
if (query.status !== undefined && query.status !== "" && query.status !== null) where.status = Number(query.status);
|
||||
|
||||
// 非管理员按客户归属自动限定数据范围
|
||||
@@ -95,15 +95,15 @@ export class ElectricLineService {
|
||||
return { code: 200, message: "获取线路台账详情成功", data: decorated[0] };
|
||||
}
|
||||
|
||||
/** 补全客户/设备/建筑展示信息,避免 N+1。 */
|
||||
/** 补全客户/设备/网关展示信息,避免 N+1。 */
|
||||
private async decorate(rows: ElectricLine[]): Promise<any[]> {
|
||||
if (rows.length === 0) return rows as any;
|
||||
|
||||
const customerIds = [...new Set(rows.map((r) => r.customerId).filter(Boolean))];
|
||||
const equipIds = [...new Set(rows.map((r) => r.equipId).filter(Boolean))];
|
||||
const buildingIds = [...new Set(rows.map((r) => r.buildingId).filter(Boolean))];
|
||||
const gatewayIds = [...new Set(rows.map((r) => r.gatewayId).filter(Boolean))];
|
||||
|
||||
const [customers, equips, buildings] = await Promise.all([
|
||||
const [customers, equips, gateways] = await Promise.all([
|
||||
customerIds.length
|
||||
? this.customerModel.findAll({
|
||||
where: { id: { [Op.in]: customerIds } },
|
||||
@@ -116,28 +116,29 @@ export class ElectricLineService {
|
||||
attributes: ["id", "equipCode", "equipName"]
|
||||
})
|
||||
: Promise.resolve([]),
|
||||
buildingIds.length
|
||||
? this.buildingModel.findAll({
|
||||
where: { id: { [Op.in]: buildingIds } },
|
||||
attributes: ["id", "buildingName"]
|
||||
gatewayIds.length
|
||||
? this.gatewayModel.findAll({
|
||||
where: { id: { [Op.in]: gatewayIds } },
|
||||
attributes: ["id", "deviceCode", "deviceName"]
|
||||
})
|
||||
: Promise.resolve([])
|
||||
]);
|
||||
|
||||
const customerMap = new Map(customers.map((c) => [c.id, c]));
|
||||
const equipMap = new Map(equips.map((e) => [e.id, e]));
|
||||
const buildingMap = new Map(buildings.map((b) => [b.id, b]));
|
||||
const gatewayMap = new Map(gateways.map((g) => [g.id, g]));
|
||||
|
||||
return rows.map((row) => {
|
||||
const json: any = row.toJSON();
|
||||
const customer: any = customerMap.get(row.customerId);
|
||||
const equip: any = equipMap.get(row.equipId);
|
||||
const building: any = row.buildingId ? buildingMap.get(row.buildingId) : undefined;
|
||||
const equip: any = row.equipId ? equipMap.get(row.equipId) : undefined;
|
||||
const gateway: any = row.gatewayId ? gatewayMap.get(row.gatewayId) : undefined;
|
||||
json.customerNo = customer?.customerNo || "";
|
||||
json.customerName = customer?.customerName || "";
|
||||
json.equipCode = equip?.equipCode || "";
|
||||
json.equipName = equip?.equipName || "";
|
||||
json.buildingName = building?.buildingName || "";
|
||||
json.gatewayCode = gateway?.deviceCode || "";
|
||||
json.gatewayName = gateway?.deviceName || "";
|
||||
return json;
|
||||
});
|
||||
}
|
||||
|
||||
@@ -6,31 +6,41 @@ import { Context } from "@midwayjs/koa";
|
||||
import ElectricLine from "../entity/electric_line.entity";
|
||||
import IccSyncLog from "../entity/icc_sync_log.entity";
|
||||
import Customer from "../entity/customer.entity";
|
||||
import CustomerEquipment from "../entity/customer_equipment.entity";
|
||||
import CustomerBuilding from "../entity/customer_building.entity";
|
||||
import { ICC_LINE_ENDPOINT, mapIccLineRecord, IccLineRawRecord } from "../task/icc_line_mapper";
|
||||
import ElectricGateway from "../entity/electric_gateway.entity";
|
||||
import { getIccAccessToken, iccFetch } from "../task/icc_customer_auth";
|
||||
import { ICC_LINE_ENDPOINT, mapIccLineRecord, parseLinePage, IccLineRawRecord } from "../task/icc_line_mapper";
|
||||
|
||||
export interface IccLineSyncResult {
|
||||
/** ICC 拉取到的原始条数 */
|
||||
/** ICC 拉取到的原始探测器条数 */
|
||||
pulled: number;
|
||||
/** 解析落库的线路条数(成功 upsert,含新增/更新) */
|
||||
/** 成功落库的线路条数(新增+更新) */
|
||||
synced: number;
|
||||
/** 新增的线路条数 */
|
||||
inserted: number;
|
||||
/** 更新的线路条数 */
|
||||
updated: number;
|
||||
/** 因缺少合法 line_code 或无法匹配客户/设备而跳过的条数 */
|
||||
/** 跳过的条数(缺 lineCode / orgCode 匹配不到客户) */
|
||||
skipped: number;
|
||||
}
|
||||
|
||||
const SYNC_TYPE = "electric_line";
|
||||
/** 分页拉取参数(ICC 探测器分页接口约定,短页判定阈值) */
|
||||
const PAGE_SIZE = 50;
|
||||
/** 最大分页数保护,防止异常接口导致无限拉取 */
|
||||
const MAX_PAGES = 100;
|
||||
/** ICC_VERSION 缺省值(与 icc_customer_auth 保持一致) */
|
||||
const DEFAULT_VERSION = "1.0.0";
|
||||
/** HTTP 超时 60s(项目约定) */
|
||||
const HTTP_TIMEOUT_MS = 60000;
|
||||
|
||||
/**
|
||||
* ICC 智电改造线路拉取同步:
|
||||
* 1) 从 .env 读取 ICC_BASE_URL / ICC_TOKEN 占位配置
|
||||
* 2) 请求 ICC 线路列表(路径与字段映射集中在 icc_line_mapper)
|
||||
* 3) 按 line_code 幂等 upsert 写入 electric_line,更新 synced_at
|
||||
* 4) 每次同步写一条 icc_sync_log;ICC 拉取失败仅记录 error,不抛业务错误
|
||||
* ICC 智电改造线路(消防探测器)拉取同步:
|
||||
* 1) 复用 icc_customer_auth 的真实 OAuth2 密码模式鉴权(getIccAccessToken / iccFetch),
|
||||
* 从 .env 读取 ICC_BASE_URL / ICC_VERSION
|
||||
* 2) POST 分页请求 ICC 探测器列表(/detector/subsystem/page,pageNum/pageSize=50,全量拉取)
|
||||
* 3) 预取 orgCode→customer.id 映射(customer.icc_code)与 parentCode→gateway.id 映射(electric_gateway.device_code),避免 N+1 查询
|
||||
* 4) 按 line_code(detectorCode)幂等 upsert 写入 electric_line,更新 synced_at
|
||||
* 5) 每次同步写一条 icc_sync_log;拉取失败仅记录 error,不抛业务错误
|
||||
*/
|
||||
@Provide()
|
||||
export class IccLineSyncService {
|
||||
@@ -43,52 +53,55 @@ export class IccLineSyncService {
|
||||
@InjectRepository(Customer)
|
||||
customerModel!: Repository<Customer>;
|
||||
|
||||
@InjectRepository(CustomerEquipment)
|
||||
equipmentModel!: Repository<CustomerEquipment>;
|
||||
|
||||
@InjectRepository(CustomerBuilding)
|
||||
buildingModel!: Repository<CustomerBuilding>;
|
||||
@InjectRepository(ElectricGateway)
|
||||
gatewayModel!: Repository<ElectricGateway>;
|
||||
|
||||
async syncFromIcc(ctx?: Context): Promise<IccLineSyncResult> {
|
||||
const startTime = new Date();
|
||||
const baseUrl = process.env.ICC_BASE_URL?.trim();
|
||||
const token = process.env.ICC_TOKEN?.trim();
|
||||
|
||||
const result: IccLineSyncResult = { 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 {
|
||||
const url = `${baseUrl.replace(/\/$/, "")}${ICC_LINE_ENDPOINT}`;
|
||||
const headers: Record<string, string> = { Accept: "application/json" };
|
||||
if (token) headers.Authorization = `Bearer ${token}`;
|
||||
const token = await getIccAccessToken();
|
||||
const url = `${baseUrl.replace(/\/$/, "")}${ICC_LINE_ENDPOINT.replace("{version}", process.env.ICC_VERSION?.trim() || DEFAULT_VERSION)}`;
|
||||
|
||||
const res = await fetch(url, {
|
||||
method: "GET",
|
||||
headers,
|
||||
signal: AbortSignal.timeout(60000) // 按项目约定 HTTP 超时 60s
|
||||
});
|
||||
const allRaw: IccLineRawRecord[] = [];
|
||||
let pageNum = 1;
|
||||
|
||||
if (!res.ok) {
|
||||
throw new Error(`ICC 线路接口返回异常:HTTP ${res.status}`);
|
||||
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 }), // 不传额外筛选条件,全量拉取
|
||||
signal: AbortSignal.timeout(HTTP_TIMEOUT_MS)
|
||||
});
|
||||
|
||||
if (!res.ok) {
|
||||
throw new Error(`ICC 探测器接口返回异常:HTTP ${res.status}`);
|
||||
}
|
||||
|
||||
const payload: any = await res.json();
|
||||
const page = parseLinePage(payload);
|
||||
allRaw.push(...page.list);
|
||||
result.pulled += page.list.length;
|
||||
|
||||
// 短页判定:本页不满 pageSize 视为无下一页;另有最大页数保护
|
||||
if (page.list.length < PAGE_SIZE) break;
|
||||
pageNum++;
|
||||
}
|
||||
|
||||
const payload: any = await res.json();
|
||||
// 兼容缓存工具包裹(data/list 字段)返回格式
|
||||
const rawList: IccLineRawRecord[] = Array.isArray(payload)
|
||||
? payload
|
||||
: Array.isArray(payload.data)
|
||||
? payload.data
|
||||
: Array.isArray(payload.list)
|
||||
? payload.list
|
||||
: [];
|
||||
result.pulled = rawList.length;
|
||||
|
||||
const { inserted, updated, skipped } = await this.batchUpsert(rawList);
|
||||
const { inserted, updated, skipped } = await this.batchUpsert(allRaw);
|
||||
result.inserted = inserted;
|
||||
result.updated = updated;
|
||||
result.synced = inserted + updated;
|
||||
@@ -97,15 +110,13 @@ export class IccLineSyncService {
|
||||
await this.writeLog(startTime, 1, SYNC_TYPE, result.synced, null);
|
||||
return result;
|
||||
} catch (err: any) {
|
||||
const message = err?.code === "ERR_MODULE_NOT_FOUND" || err?.message
|
||||
? String(err?.message || err)
|
||||
: String(err);
|
||||
const message = err instanceof Error ? err.message : String(err);
|
||||
await this.writeLog(startTime, 0, SYNC_TYPE, result.pulled, message);
|
||||
return result;
|
||||
}
|
||||
}
|
||||
|
||||
/** 批量按 line_code upsert(幂等),并做客户/设备/建筑外键反查。 */
|
||||
/** 按 line_code(detectorCode)幂等 upsert,customer_id 解析失败的行跳过。 */
|
||||
private async batchUpsert(records: IccLineRawRecord[]): Promise<{ inserted: number; updated: number; skipped: number }> {
|
||||
let inserted = 0;
|
||||
let updated = 0;
|
||||
@@ -113,71 +124,66 @@ export class IccLineSyncService {
|
||||
|
||||
if (records.length === 0) return { inserted, updated, skipped };
|
||||
|
||||
// 预取外键映射,避免 N+1
|
||||
const equipCodes = [...new Set(records.map((r) => mapIccLineRecord(r).equipCode).filter(Boolean))];
|
||||
const customerNos = [...new Set(records.map((r) => mapIccLineRecord(r).customerNo).filter(Boolean))];
|
||||
const buildingNames = [...new Set(records.map((r) => mapIccLineRecord(r).buildingName).filter(Boolean))];
|
||||
// 预取 orgCode → customer.id 映射,避免 N+1 查询客户
|
||||
const orgCodes = [...new Set(records.map((r) => mapIccLineRecord(r).orgCode).filter(Boolean))];
|
||||
const customers = orgCodes.length
|
||||
? await this.customerModel.findAll({ where: { iccCode: { [Op.in]: orgCodes } }, attributes: ["id", "iccCode"] })
|
||||
: [];
|
||||
const customerIdByOrgCode = new Map(customers.map((c) => [c.iccCode, c.id]));
|
||||
|
||||
const [equips, customers, buildings] = await Promise.all([
|
||||
equipCodes.length
|
||||
? this.equipmentModel.findAll({ where: { equipCode: { [Op.in]: equipCodes } }, attributes: ["id", "equipCode", "customerId", "buildingId"] })
|
||||
: Promise.resolve([]),
|
||||
customerNos.length
|
||||
? this.customerModel.findAll({ where: { customerNo: { [Op.in]: customerNos } }, attributes: ["id", "customerNo"] })
|
||||
: Promise.resolve([]),
|
||||
buildingNames.length
|
||||
? this.buildingModel.findAll({ where: { buildingName: { [Op.in]: buildingNames } }, attributes: ["id", "buildingName", "customerId"] })
|
||||
: Promise.resolve([])
|
||||
]);
|
||||
|
||||
const equipByCode = new Map(equips.map((e) => [e.equipCode, e]));
|
||||
const customerByNo = new Map(customers.map((c) => [c.customerNo, c]));
|
||||
const buildingByName = new Map(buildings.map((b) => [b.buildingName, b]));
|
||||
// 预取 parentCode → gateway.id 映射(electric_gateway.device_code),避免 N+1 查询网关
|
||||
const parentCodes = [...new Set(records.map((r) => mapIccLineRecord(r).parentCode).filter(Boolean))];
|
||||
const gateways = parentCodes.length
|
||||
? await this.gatewayModel.findAll({ where: { deviceCode: { [Op.in]: parentCodes } }, attributes: ["id", "deviceCode"] })
|
||||
: [];
|
||||
const gatewayIdByParentCode = new Map(gateways.map((g) => [g.deviceCode, g.id]));
|
||||
|
||||
const now = new Date();
|
||||
|
||||
for (const raw of records) {
|
||||
const mapped = mapIccLineRecord(raw);
|
||||
// line_code(detectorCode)为空则无同步匹配键,跳过
|
||||
if (!mapped.lineCode) {
|
||||
skipped++;
|
||||
continue;
|
||||
}
|
||||
|
||||
const equip = mapped.equipCode ? equipByCode.get(mapped.equipCode) : undefined;
|
||||
// equip_id 与 customer_id 为 NOT NULL,无法解析时跳过该行
|
||||
if (!equip) {
|
||||
// customer_id NOT NULL:orgCode 必须能解析到客户,否则跳过
|
||||
if (!mapped.orgCode) {
|
||||
skipped++;
|
||||
continue;
|
||||
}
|
||||
const customerId = customerIdByOrgCode.get(mapped.orgCode);
|
||||
if (!customerId) {
|
||||
skipped++;
|
||||
continue;
|
||||
}
|
||||
|
||||
let customerId: number = equip.customerId;
|
||||
if (mapped.customerNo) {
|
||||
const byNo = customerByNo.get(mapped.customerNo);
|
||||
if (byNo) customerId = byNo.id;
|
||||
}
|
||||
|
||||
let buildingId: number | null = equip.buildingId || null;
|
||||
if (mapped.buildingName) {
|
||||
const building = buildingByName.get(mapped.buildingName);
|
||||
if (building) buildingId = building.id;
|
||||
}
|
||||
// gateway_id 可空:parentCode 匹配不到 electric_gateway.device_code 时置空
|
||||
const gatewayId = mapped.parentCode ? (gatewayIdByParentCode.get(mapped.parentCode) ?? null) : null;
|
||||
|
||||
const payload = {
|
||||
lineCode: mapped.lineCode,
|
||||
equipId: equip.id,
|
||||
customerId,
|
||||
buildingId,
|
||||
iccDeviceCode: mapped.iccDeviceCode || null,
|
||||
lineName: mapped.lineName || null,
|
||||
floorNo: mapped.floorNo || null,
|
||||
location: mapped.location || null,
|
||||
imei: mapped.imei || null,
|
||||
// equipId 同步不填充
|
||||
gatewayId,
|
||||
customerId,
|
||||
typeCode: mapped.typeCode || null,
|
||||
typeName: mapped.typeName || null,
|
||||
fireSystemCode: mapped.fireSystemCode || null,
|
||||
fireSystemName: mapped.fireSystemName || null,
|
||||
partUnitLoopCode: mapped.partUnitLoopCode || null,
|
||||
channelNo: mapped.channelNo || null,
|
||||
channelSn: mapped.channelSn || null,
|
||||
produceDate: mapped.produceDate || null,
|
||||
status: mapped.status,
|
||||
syncedAt: now
|
||||
} as any;
|
||||
|
||||
// 幂等 upsert:按 line_code 唯一键匹配
|
||||
const existing = await this.lineModel.findOne({ where: { lineCode: mapped.lineCode } });
|
||||
if (existing) {
|
||||
await existing.update(payload);
|
||||
await existing.update({ ...payload, syncedAt: now });
|
||||
updated++;
|
||||
} else {
|
||||
await this.lineModel.create(payload);
|
||||
@@ -202,4 +208,4 @@ export class IccLineSyncService {
|
||||
console.error("[ICC同步] 同步日志写入失败:", err);
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
Reference in new issue
Block a user