feat(analog): 新增ICC模拟量数据同步 - 重建electricity_data表（迁移20261024_icc_analog_data.sql：删除equip_id与electricity占位列、icc_device_code更名imei、新增analog_type/analog_name/analog_unit/analog_value、collect_time以同步时间填充、按月分区2026_10~2027_03、唯一键(line_code,analog_type,collect_time)）；同步服务IccAnalogSyncService复用OAuth2鉴权，按已同步线路line_code分批（≤50）拉取最新模拟量（POST /analog/subsystem/page，pageSize=500短页停止）幂等upsert并写icc_sync_log（sync_type=electricity_data）；新增POST /api/electricity-data/sync手动同步（RBAC device:list）；注册icc_analog_sync每5分钟定时任务与sys_job种子；同步db_init/db_seed与设计文档；无前端改动

2026-10-24 14:52:22 +08:00
1 parent 160ad85051
commit c92fc7f298
@@ -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<ElectricityData>;
+
+	@InjectRepository(ElectricLine)
+	lineModel!: Repository<ElectricLine>;
+
+	@InjectRepository(IccSyncLog)
+	syncLogModel!: Repository<IccSyncLog>;
+
+	async syncFromIcc(ctx?: Context): Promise<IccAnalogSyncResult> {
+		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<string, LineMatchInfo>();
+			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<string, LineMatchInfo>,
+		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<void> {
+		try {
+			await this.syncLogModel.create({
+				syncType,
+				startTime,
+				endTime: new Date(),
+				recordCount,
+				status,
+				errorMessage
+			} as any);
+		} catch (err) {
+			console.error("[ICC同步] 同步日志写入失败:", err);
+		}
+	}
+}