224 lines
7.9 KiB
TypeScript
224 lines
7.9 KiB
TypeScript
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);
|
||
}
|
||
}
|
||
}
|