import { Provide } from "@midwayjs/core"; import { InjectRepository } from "@midwayjs/sequelize"; import { Repository } from "sequelize-typescript"; import { Op } from "sequelize"; 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"; export interface IccLineSyncResult { /** ICC 拉取到的原始条数 */ pulled: number; /** 解析落库的线路条数(成功 upsert,含新增/更新) */ synced: number; /** 新增的线路条数 */ inserted: number; /** 更新的线路条数 */ updated: number; /** 因缺少合法 line_code 或无法匹配客户/设备而跳过的条数 */ skipped: number; } const SYNC_TYPE = "electric_line"; /** * 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,不抛业务错误 */ @Provide() export class IccLineSyncService { @InjectRepository(ElectricLine) lineModel!: Repository; @InjectRepository(IccSyncLog) syncLogModel!: Repository; @InjectRepository(Customer) customerModel!: Repository; @InjectRepository(CustomerEquipment) equipmentModel!: Repository; @InjectRepository(CustomerBuilding) buildingModel!: Repository; async syncFromIcc(ctx?: Context): Promise { 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 = { Accept: "application/json" }; if (token) headers.Authorization = `Bearer ${token}`; const res = await fetch(url, { method: "GET", headers, signal: AbortSignal.timeout(60000) // 按项目约定 HTTP 超时 60s }); if (!res.ok) { throw new Error(`ICC 线路接口返回异常:HTTP ${res.status}`); } 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); 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?.code === "ERR_MODULE_NOT_FOUND" || err?.message ? String(err?.message || err) : String(err); await this.writeLog(startTime, 0, SYNC_TYPE, result.pulled, message); return result; } } /** 批量按 line_code upsert(幂等),并做客户/设备/建筑外键反查。 */ private async batchUpsert(records: IccLineRawRecord[]): Promise<{ inserted: number; updated: number; skipped: number }> { let inserted = 0; let updated = 0; let skipped = 0; 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))]; 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])); const now = new Date(); for (const raw of records) { const mapped = mapIccLineRecord(raw); if (!mapped.lineCode) { skipped++; continue; } const equip = mapped.equipCode ? equipByCode.get(mapped.equipCode) : undefined; // equip_id 与 customer_id 为 NOT NULL,无法解析时跳过该行 if (!equip) { 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; } 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, status: mapped.status, syncedAt: now } as any; const existing = await this.lineModel.findOne({ where: { lineCode: mapped.lineCode } }); if (existing) { await existing.update(payload); updated++; } else { await this.lineModel.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); } } }