Skip to content

Q
qingdaoMuseumPlatform
Commit 42295692 by PC-20251223ZVQQ\Administrator
Options

预警工单数据接入

parent 7a8453e5
Showing with 557 additions and 48 deletions
src/biz/dataIntegration.ts
......@@ -4,12 +4,18 @@ import {
getElectricMaterList,
getIndoorUnitStateList,
getMaterStateList,
getIndoorUnitAlarmErrorHis,
getFaultCodeInfo,
getFaultTypeName,
} from "./feiyiClient";
import { mysqlModelMap } from "../model/sqlModelBind";
import moment from "moment";
import XLSX from "xlsx";
import path from "path";
/** 飞奕品牌编码，默认 "1"=日立 */
const FEIYI_BRAND_CODE = "1";
// ==================== 工具函数 ====================
/**
......@@ -733,9 +739,234 @@ export async function customerDeviceData() {
// 暂无对应接口 -> MQTT
}
// ==================== 预警工单集成（飞奕故障记录） ====================
/**
* 数据集成-预警工单数据集成
* 从飞奕云平台拉取当天空调内机故障记录，按 indoorUnitAlarmId 去重后写入 device_fault 表
*/
export async function alertWorkData() {
// TODO: 暂无对应接口
console.log('[预警工单集成] 开始...');
const deviceModel = mysqlModelMap['device'];
const faultModel = mysqlModelMap['device_fault'];
if (!deviceModel || !faultModel) {
console.error('[预警工单集成] device 或 device_fault 模型未初始化，跳过');
return;
}
try {
const todayDate = getTodayDateStr();
// 1. 分页拉取当天所有故障记录
const rows = await fetchAllPages(
(page, limit, extraParams) => getIndoorUnitAlarmErrorHis({
page,
limit,
occurrenceTime: extraParams.occurrenceTime,
}),
{ occurrenceTime: todayDate },
200,
);
console.log(`[预警工单集成] 获取 ${rows.length} 条故障记录`);
if (rows.length === 0) {
console.log('[预警工单集成] 当天无故障记录，跳过');
return;
}
// 2. 收集所有 indoorUnitAlarmId，批量查已有故障记录（去重）
const alarmIds = rows.map((r: any) => r.indoorUnitAlarmId).filter(Boolean);
if (alarmIds.length === 0) {
console.log('[预警工单集成] 所有记录缺少 indoorUnitAlarmId，跳过');
return;
}
const existingFaults = await faultModel.findAll({
where: { fault_origin_id: alarmIds },
attributes: ['fault_origin_id'],
raw: true,
});
const existingIds = new Set((existingFaults as any[]).map((f: any) => f.fault_origin_id));
// 3. 收集 indoorUnitAddressFull，批量查设备确认已注册
const addrList = rows.map((r: any) => r.indoorUnitAddressFull).filter(Boolean);
const devices = await deviceModel.findAll({
where: { device_id: addrList },
attributes: ['device_id'],
raw: true,
});
const registeredDeviceIds = new Set((devices as any[]).map((d: any) => d.device_id));
// 4. 收集所有故障码，批量查询故障详情（缓存去重）
const alarmCodes = [...new Set(rows.map((r: any) => r.alarmCode).filter(Boolean))] as string[];
const faultInfoMap = new Map<string, { errorInfo: string; scheme: string | null }>();
if (alarmCodes.length > 0) {
console.log(`[预警工单集成] 查询 ${alarmCodes.length} 个不同故障码的详情...`);
const infoResults = await Promise.all(
alarmCodes.map(async (code) => {
const info = await getFaultCodeInfo(FEIYI_BRAND_CODE, code);
return { code, info };
})
);
for (const { code, info } of infoResults) {
faultInfoMap.set(code, info);
}
}
// 5. 过滤 + 组装待插入数据
const toInsert: any[] = [];
let skipCount = 0;
for (const r of rows as any[]) {
// 去重：已存在的记录跳过
if (!r.indoorUnitAlarmId || existingIds.has(r.indoorUnitAlarmId)) {
skipCount++;
continue;
}
// 设备未注册跳过
const deviceId = r.indoorUnitAddressFull;
if (!deviceId || !registeredDeviceIds.has(deviceId)) {
skipCount++;
continue;
}
// 获取故障详情
const faultInfo = faultInfoMap.get(r.alarmCode);
const faultFound = faultInfo
&& faultInfo.errorInfo !== r.alarmCode
&& faultInfo.errorInfo !== '查询失败';
const errorMsg = faultFound ? faultInfo.errorInfo : '未找到主内机';
toInsert.push({
device_id: deviceId,
fault_type: getFaultTypeName(r.alarmType),
fault_code: r.alarmCode || '',
fault_description: errorMsg,
fault_origin_id: r.indoorUnitAlarmId,
occurred_time: r.deviceTime ? new Date(r.deviceTime) : new Date(),
level: 2,
status: 0,
});
}
// 6. 批量写入
if (toInsert.length > 0) {
await faultModel.bulkCreate(toInsert);
}
console.log(`[预警工单集成] 完成，新增 ${toInsert.length} 条，跳过 ${skipCount} 条（去重/未注册）`);
} catch (err) {
console.error('[预警工单集成] 执行异常:', err);
}
}
// ==================== 故障状态自动更新 ====================
/**
* 故障状态自动更新
* 故障驱动模式：拉取所有未处理/处理中的故障，其设备一次拉回故障时间之后的全部开机记录，
* 在内存中比对，状态匹配的批量更新为"已解决"。
* 始终发 2 条 SQL，与故障数无关；不设时间兜底，服务宕机重启后也能补全。
*/
export async function processFaultStatus() {
console.log('[故障状态更新] 开始...');
const faultModel = mysqlModelMap['device_fault'];
const deviceDataModel = mysqlModelMap['device_data'];
if (!faultModel || !deviceDataModel) {
console.error('[故障状态更新] device_fault 或 device_data 模型未初始化，跳过');
return;
}
try {
const Sequelize = require('sequelize');
const Op = Sequelize.Op;
// ① 一次查出所有未处理/处理中的故障
const activeFaults = await faultModel.findAll({
where: { status: { [Op.in]: [0, 1] } },
attributes: ['id', 'device_id', 'occurred_time'],
raw: true,
});
console.log(`[故障状态更新] 共 ${activeFaults.length} 条活跃故障`);
if (activeFaults.length === 0) {
console.log('[故障状态更新] 无活跃故障，跳过');
return;
}
// ② 去重 device_id + 取最早故障时间（扫描范围起点）
const deviceSet = new Set<string>();
let minOccurredTime: Date | null = null;
for (const f of activeFaults as any[]) {
deviceSet.add(f.device_id);
const t = f.occurred_time ? new Date(f.occurred_time) : null;
if (t && (!minOccurredTime || t < minOccurredTime)) {
minOccurredTime = t;
}
}
const deviceIds = [...deviceSet];
// ③ 一次拉回所有相关设备的开机记录（故障最早时间之后）
// 不做时间兜底，宕机重启后也能覆盖全部历史
const deviceDataRecords = await deviceDataModel.findAll({
where: {
device_id: { [Op.in]: deviceIds },
...(minOccurredTime ? { device_time: { [Op.gte]: minOccurredTime } } : {}),
},
attributes: ['device_id', 'data', 'device_time'],
order: [['device_id', 'ASC'], ['device_time', 'ASC']],
raw: true,
});
console.log(`[故障状态更新] 共 ${deviceDataRecords.length} 条相关设备数据记录`);
// ④ 内存过滤：只保留开机记录（power='on'），按 device_id 分组取最早开机时间
const powerOnMap = new Map<string, Date>();
for (const r of deviceDataRecords as any[]) {
let power: string | undefined;
const rawData = r.data;
if (typeof rawData === 'string') {
try { power = JSON.parse(rawData).power; } catch { continue; }
} else if (rawData && typeof rawData === 'object') {
power = rawData.power;
}
if (power !== 'on') continue;
const dt = r.device_time ? new Date(r.device_time) : null;
if (!dt) continue;
if (!powerOnMap.has(r.device_id) || dt < powerOnMap.get(r.device_id)!) {
powerOnMap.set(r.device_id, dt);
}
}
// ⑤ 内存比对：开机时间 > 故障时间 → 标记已解决
const toUpdate: { id: number; resolved_time: Date }[] = [];
for (const fault of activeFaults as any[]) {
const powerOnTime = powerOnMap.get(fault.device_id);
if (!powerOnTime) continue;
const occurredTime = fault.occurred_time ? moment(fault.occurred_time) : null;
if (!occurredTime) continue;
if (moment(powerOnTime).isAfter(occurredTime)) {
toUpdate.push({ id: fault.id, resolved_time: powerOnTime });
}
}
if (toUpdate.length === 0) {
console.log('[故障状态更新] 无符合条件的故障需更新');
return;
}
// ⑥ 逐条更新 status=2
for (const item of toUpdate) {
await faultModel.update(
{ status: 2, resolved_time: item.resolved_time, updated_at: new Date() },
{ where: { id: item.id } },
);
}
console.log(`[故障状态更新] 完成，共处理 ${toUpdate.length} 条故障`);
} catch (err) {
console.error('[故障状态更新] 执行异常:', err);
}
}
src/biz/feiyiClient.ts
......@@ -4,7 +4,7 @@
*/
import * as crypto from 'crypto';
import { post } from '../util/request';
import { post, get } from '../util/request';
import { BizError } from '../util/bizError';
import { ERRORENUM } from '../config/errorEnum';
......@@ -379,3 +379,79 @@ export async function getIndoorUnitStateList(params: any): Promise<any> {
const result = await feiYiPost(path, body);
return result.data;
}
/**
* 空调内机故障记录分页查询
* @param params { page, limit, roomIds?, indoorUnitAddressFull?, alarmType?, occurrenceTime? }
* @returns data 部分，包含 page, limit, total, rows
*/
export async function getIndoorUnitAlarmErrorHis(params: any): Promise<any> {
const path = '/openApi/indoorUnit/alarm/errorHis';
const body = { ...params };
const result = await feiYiPost(path, body);
return result.data;
}
/**
* 故障码查询故障详情
* 根据品牌编码和错误编码查询故障描述和解决方案
* 接口文档：空调故障码对接服务与解析流程.pdf
* @param brandCode 品牌编码，如 "1"=日立
* @param errorCode 错误编码，如 "LOST"
* @returns { errorInfo: string, scheme: string | null } 故障信息描述、解决方案
*/
export async function getFaultCodeInfo(brandCode: string, errorCode: string): Promise<{ errorInfo: string; scheme: string | null }> {
const url = 'https://errorcode.feiyikj.cn/prod-api/feiyi/app/errorCode/info';
const query = { brandCode, errorCode };
let result: any;
try {
result = await get(url, query);
} catch (err) {
console.error(`[故障码查询] 请求失败 brandCode=${brandCode} errorCode=${errorCode}:`, err);
return { errorInfo: '查询失败', scheme: null };
}
if (!result || result.code !== 200) {
console.error(`[故障码查询] 接口返回异常 brandCode=${brandCode} errorCode=${errorCode}:`, result);
return { errorInfo: '查询失败', scheme: null };
}
const data = result.data || [];
if (data.length === 0) {
console.warn(`[故障码查询] 未查到故障码信息 brandCode=${brandCode} errorCode=${errorCode}`);
return { errorInfo: errorCode, scheme: null };
}
return {
errorInfo: data[0].errorInfo || errorCode,
scheme: data[0].scheme || null,
};
}
/**
* 故障类型映射（中文 -> 数字）
*/
let faultTypeMap: { [key: string]: number } = {
"内机故障": 5,
"未绑定房间": 15
};
/** 数字 -> 中文的反向映射 */
let faultTypeReverseMap: Map<number, string> | null = null;
/** 构建数字->中文的反向映射 */
function getFaultTypeReverseMap(): Map<number, string> {
if (faultTypeReverseMap) return faultTypeReverseMap;
faultTypeReverseMap = new Map();
for (const [cn, num] of Object.entries(faultTypeMap)) {
faultTypeReverseMap.set(num, cn);
}
return faultTypeReverseMap;
}
/**
* 根据 alarmType 数字获取中文故障类型名称
* @param alarmType 故障类型数字
* @returns 中文名称，未匹配时返回原数字字符串
*/
export function getFaultTypeName(alarmType: number | string): string {
const num = typeof alarmType === 'string' ? parseInt(alarmType, 10) : alarmType;
return getFaultTypeReverseMap().get(num) || String(alarmType);
}
\ No newline at end of file
src/biz/report.ts
......@@ -96,6 +96,7 @@ function dayRange(day: Date): { from: string; to: string } {
/** 某馆的所有设备分组（用于批量查询） */
interface HallDeviceGroup {
acDevices: { device_id: string; device_name: string; region_name: string; region_key: number }[];
envDevices: { device_id: string; device_name: string; region_key: number }[];
customerDevices: { device_id: string; device_name: string; region_key: number }[];
meterDevices: { device_id: string; device_name: string; region_key: number }[];
allDeviceIds: string[];
......@@ -107,7 +108,7 @@ interface HallDeviceGroup {
*/
async function queryHallDeviceGroup(regionIds: number[]): Promise<HallDeviceGroup> {
if (regionIds.length === 0) {
return { acDevices: [], customerDevices: [], meterDevices: [], allDeviceIds: [], allDevices: [] };
return { acDevices: [], envDevices: [], customerDevices: [], meterDevices: [], allDeviceIds: [], allDevices: [] };
}
// 并行查询该馆设备和区域名称
......@@ -134,6 +135,7 @@ async function queryHallDeviceGroup(regionIds: number[]): Promise<HallDeviceGrou
return {
acDevices,
envDevices: allDevices.filter(d => d.device_type === "环境监测"),
customerDevices: allDevices.filter(d => d.device_type === "客流传感器"),
meterDevices: allDevices.filter(d => d.device_type === "电能监测"),
allDeviceIds: allDevices.map(d => d.device_id),
......@@ -142,20 +144,15 @@ async function queryHallDeviceGroup(regionIds: number[]): Promise<HallDeviceGrou
}
/**
* 从空调/新风设备数据中计算室温均值 + 运行时长
* 从空调/新风设备数据中计算运行时长
*/
function calcAcDeviceDaily(records: any[], dayEndTime: Date): { avgTemp: number; runHours: number; tempCount: number } {
function calcAcRunHours(records: any[], dayEndTime: Date): number {
const sorted = records.slice().sort(
(a, b) => new Date(a.device_time).getTime() - new Date(b.device_time).getTime(),
);
let tempSum = 0, tempCount = 0, runHours = 0;
let runHours = 0;
for (let i = 0; i < sorted.length; i++) {
const d = sorted[i].data || {};
if (d.roomTemp != null && Number(d.roomTemp) !== 0) {
tempSum += Number(d.roomTemp);
tempCount++;
}
if (d.power === "on") {
const t1 = new Date(sorted[i].device_time).getTime();
const t2 = i + 1 < sorted.length
......@@ -164,44 +161,52 @@ function calcAcDeviceDaily(records: any[], dayEndTime: Date): { avgTemp: number;
if (t2 > t1) runHours += (t2 - t1) / (1000 * 60 * 60);
}
}
return runHours;
}
const avgTemp = tempCount > 0 ? Math.round(tempSum / tempCount * 10) / 10 : 0;
return { avgTemp, runHours, tempCount };
/**
* 从环境监测设备数据中计算室温均值（使用 data.temperature 字段）
*/
function calcEnvDeviceDaily(records: any[]): { tempSum: number; tempCount: number } {
let tempSum = 0, tempCount = 0;
for (const rec of records) {
const d = rec.data || {};
if (d.temperature != null && Number(d.temperature) !== 0) {
tempSum += Number(d.temperature);
tempCount++;
}
}
return { tempSum, tempCount };
}
/**
* 从客流传感器数据中计算人流总值
* 从客流传感器数据中计算当日平均在馆人数
* current_total 为当时在场人数（瞬时值），取当日所有采集值的算术平均
*/
function calcCustomerDeviceDaily(records: any[]): number {
let visitorSum = 0;
function calcCustomerDeviceDaily(records: any[]): { visitorSum: number; visitorCount: number } {
let sum = 0, count = 0;
for (const rec of records) {
const d = rec.data || {};
if (d.current_total != null) visitorSum += Number(d.current_total);
if (d.current_total != null) {
sum += Number(d.current_total);
count++;
}
}
return visitorSum;
return { visitorSum: sum, visitorCount: count };
}
/**
* 从电表数据中计算24时能耗（末次累计 - 首次累计）
* 从电表数据中计算24时总用电量（累加每次上报的 current 值）
*/
function calcMeterDeviceDaily(records: any[]): number {
let firstEnergy: number | null = null;
let lastEnergy: number | null = null;
const sorted = records.slice().sort(
(a, b) => new Date(a.device_time).getTime() - new Date(b.device_time).getTime(),
);
for (const rec of sorted) {
let total = 0;
for (const rec of records) {
const d = rec.data || {};
if (d.energy != null) {
const val = Number(d.energy);
if (firstEnergy === null) firstEnergy = val;
lastEnergy = val;
if (d.current != null) {
total += Number(d.current);
}
}
return (lastEnergy !== null && firstEnergy !== null && lastEnergy > firstEnergy)
? lastEnergy - firstEnergy
: 0;
return total;
}
/**
......@@ -222,7 +227,7 @@ export async function getWeeklyReportData(weekOffset: number = 0): Promise<any>
const hallDeviceGroups = new Map<string, HallDeviceGroup>();
for (let i = 0; i < HALL_NAMES.length; i++) {
hallDeviceGroups.set(HALL_NAMES[i], groups[i]);
console.log(`[周报告JSON] ${HALL_NAMES[i]}: AC${groups[i].acDevices.length} 客流${groups[i].customerDevices.length} 电表${groups[i].meterDevices.length}`);
console.log(`[周报告JSON] ${HALL_NAMES[i]}: AC${groups[i].acDevices.length} 环境${groups[i].envDevices.length} 客流${groups[i].customerDevices.length} 电表${groups[i].meterDevices.length}`);
}
// ===== 预查询：一次性拉取所有设备未解决故障 =====
......@@ -273,21 +278,30 @@ export async function getWeeklyReportData(weekOffset: number = 0): Promise<any>
}
let dayRunHours = 0, dayTempSum = 0, dayTempCount = 0;
let dayEnergy = 0, dayVisitor = 0;
let dayEnergy = 0, dayVisitorSum = 0, dayVisitorCount = 0;
for (const dev of group.acDevices) {
const records = deviceDataMap.get(dev.device_id) || [];
const { avgTemp, runHours, tempCount } = calcAcDeviceDaily(records, dayEnd);
dayRunHours += runHours;
if (tempCount > 0) { dayTempSum += avgTemp * tempCount; dayTempCount += tempCount; }
dayRunHours += calcAcRunHours(records, dayEnd);
}
for (const dev of group.envDevices) {
const records = deviceDataMap.get(dev.device_id) || [];
const { tempSum, tempCount } = calcEnvDeviceDaily(records);
dayTempSum += tempSum;
dayTempCount += tempCount;
}
for (const dev of group.customerDevices) {
dayVisitor += calcCustomerDeviceDaily(deviceDataMap.get(dev.device_id) || []);
const { visitorSum, visitorCount } = calcCustomerDeviceDaily(deviceDataMap.get(dev.device_id) || []);
dayVisitorSum += visitorSum;
dayVisitorCount += visitorCount;
}
for (const dev of group.meterDevices) {
dayEnergy += calcMeterDeviceDaily(deviceDataMap.get(dev.device_id) || []);
}
// 当日在馆人数均值 = 所有客流传感器当天采集值的平均
const dayVisitor = dayVisitorCount > 0 ? dayVisitorSum : 0; // Math.round(dayVisitorSum / dayVisitorCount)
// 累加到周汇总
hallTotalRunHours += dayRunHours;
if (dayTempCount > 0) { hallTempSum += dayTempSum; hallTempCount += dayTempCount; }
......@@ -299,15 +313,15 @@ export async function getWeeklyReportData(weekOffset: number = 0): Promise<any>
dayRunHours > 0 ? fmtNum(dayRunHours) + "h" : "——",
dayTempCount > 0 ? fmtNum(dayTempSum / dayTempCount) : "——",
dayVisitor > 0 ? String(dayVisitor) : "——",
dayEnergy > 0 ? String(Math.round(dayEnergy * 100) / 100) : "——",
dayEnergy > 0 ? String(Math.round(dayEnergy / 24 * 100) / 100) : "——",
]);
}
// ===== 第四步：汇总各馆 =====
const hallAvgTemp = hallTempCount > 0 ? hallTempSum / hallTempCount : null;
const hallTempStr = hallAvgTemp != null ? fmtNum(hallAvgTemp) + "°C" : "——";
const hallEnergyStr = fmtEnergy(hallTotalEnergy);
const kllStr = String(hallTotalVisitor || "——");
const hallEnergyStr = fmtEnergy(hallTotalEnergy / 7);
const kllStr = hallTotalVisitor > 0 ? String(Math.round(hallTotalVisitor / 7)) : "——";
dataTotal[hallKey] = { nh: hallEnergyStr, kll: kllStr, jw: hallTempStr };
runDatas[hallName] = {
......@@ -336,8 +350,8 @@ export async function getWeeklyReportData(weekOffset: number = 0): Promise<any>
// ===== 第五步：整体汇总 + 评价 =====
dataTotal.ztyx = {
nh: fmtEnergy(overallEnergy),
kll: String(overallVisitor || "——"),
nh: fmtEnergy(overallEnergy / 7),
kll: overallVisitor > 0 ? String(Math.round(overallVisitor / 7)) : "——",
jw: overallTempCount > 0
? fmtNum(overallTempSum / overallTempCount) + "°C"
: "——",
......
src/config/mysqlTableConfig.ts
......@@ -65,11 +65,22 @@ export const TablesConfig = [
allowNull: false,
comment: '故障类型，如“离线”“数据异常”“硬件故障”'
},
fault_code: {
type: DataTypes.STRING(50),
allowNull: true,
comment: '故障代码，如 LOST、E1 等'
},
fault_description: {
type: DataTypes.TEXT,
allowNull: true,
comment: '故障详细描述'
},
fault_origin_id: {
type: DataTypes.STRING(64),
allowNull: true,
unique: true,
comment: '第三方故障记录ID，用于去重'
},
occurred_time: {
type: DataTypes.DATE,
allowNull: false,
......
src/config/schedule.ts
/**
* 定时任务调度器
* - 每分钟执行一次任务
* - 每分钟执行一次数据同步任务
* - 每 10 分钟执行一次预警工单同步任务
* - 内置锁机制，防止任务重叠执行
* - 任务锁死时（超时未释放），会主动强制释放锁，确保后续任务能正常执行
*/
import { dataintegration } from "tencentcloud-sdk-nodejs";
import { region, device, deviceData } from "../biz/dataIntegration";
import { region, device, deviceData, alertWorkData, processFaultStatus } from "../biz/dataIntegration";
// 锁超时时间（毫秒），任务执行超过此时间视为锁死
const LOCK_TIMEOUT_MS = 55 * 1000; // 55秒，留5秒余量给下一次执行
......@@ -111,6 +112,43 @@ async function scheduledTask(): Promise<void> {
// 定时器句柄
let timerHandle: NodeJS.Timeout | null = null;
let alertTimerHandle: NodeJS.Timeout | null = null;
let faultStatusTimerHandle: NodeJS.Timeout | null = null;
// 预警工单定时任务专用锁
const alertTaskLock: TaskLock = {
locked: false,
lockTime: 0,
lockKey: '',
};
/** 预警工单锁超时（毫秒） */
const ALERT_LOCK_TIMEOUT_MS = 9 * 60 * 1000; // 9 分钟
/** 预警工单执行间隔（毫秒） */
const ALERT_TASK_INTERVAL_MS = 10 * 60 * 1000; // 10 分钟
/**
* 尝试获取预警工单任务锁
*/
function tryAcquireAlertLock(): boolean {
if (alertTaskLock.locked) {
const elapsed = Date.now() - alertTaskLock.lockTime;
if (elapsed > ALERT_LOCK_TIMEOUT_MS) {
console.warn(`[预警工单定时任务] 检测到锁死，强制释放锁。已锁定 ${elapsed}ms`);
alertTaskLock.locked = false;
alertTaskLock.lockTime = 0;
alertTaskLock.lockKey = '';
return tryAcquireAlertLock();
}
console.log(`[预警工单定时任务] 上一次任务尚未完成，跳过本次执行`);
return false;
}
alertTaskLock.locked = true;
alertTaskLock.lockTime = Date.now();
alertTaskLock.lockKey = `${Date.now()}_${Math.random().toString(36).substring(2, 8)}`;
return true;
}
/**
* 启动定时任务
......@@ -122,7 +160,7 @@ export function startSchedule(): void {
return;
}
console.log(`[定时任务] 定时器已启动，间隔 ${TASK_INTERVAL_MS / 1000} 秒`);
console.log(`[定时任务] 数据同步定时器已启动，间隔 ${TASK_INTERVAL_MS / 1000} 秒`);
// 使用 setInterval 实现每分钟执行一次
timerHandle = setInterval(() => {
......@@ -131,6 +169,129 @@ export function startSchedule(): void {
}
/**
* 启动预警工单定时任务
* 独立于数据同步定时任务，每 10 分钟执行一次
*/
export function startAlertSchedule(): void {
if (alertTimerHandle) {
console.warn('[预警工单定时任务] 定时器已在运行中，无需重复启动');
return;
}
console.log(`[预警工单定时任务] 定时器已启动，间隔 ${ALERT_TASK_INTERVAL_MS / 1000} 秒`);
alertTimerHandle = setInterval(() => {
scheduledAlertTask();
}, ALERT_TASK_INTERVAL_MS);
}
/**
* 预警工单定时任务的具体业务逻辑
*/
async function scheduledAlertTask(): Promise<void> {
if (!tryAcquireAlertLock()) {
return;
}
try {
console.log(`[预警工单定时任务] 开始执行 - ${new Date().toLocaleString()}`);
await alertWorkData();
console.log(`[预警工单定时任务] 执行完成 - ${new Date().toLocaleString()}`);
} catch (err) {
console.error('[预警工单定时任务] 执行异常:', err);
} finally {
alertTaskLock.locked = false;
alertTaskLock.lockTime = 0;
alertTaskLock.lockKey = '';
}
}
// ==================== 故障状态更新定时任务 ====================
/** 故障状态更新锁超时（毫秒） */
const FAULT_STATUS_LOCK_TIMEOUT_MS = 4 * 60 * 1000; // 4 分钟
/** 故障状态更新执行间隔（毫秒） */
const FAULT_STATUS_TASK_INTERVAL_MS = 5 * 60 * 1000; // 5 分钟
/** 故障状态更新专用锁 */
const faultStatusTaskLock: TaskLock = {
locked: false,
lockTime: 0,
lockKey: '',
};
/**
* 尝试获取故障状态更新任务锁
*/
function tryAcquireFaultStatusLock(): boolean {
if (faultStatusTaskLock.locked) {
const elapsed = Date.now() - faultStatusTaskLock.lockTime;
if (elapsed > FAULT_STATUS_LOCK_TIMEOUT_MS) {
console.warn(`[故障状态更新任务] 检测到锁死，强制释放锁。已锁定 ${elapsed}ms`);
faultStatusTaskLock.locked = false;
faultStatusTaskLock.lockTime = 0;
faultStatusTaskLock.lockKey = '';
return tryAcquireFaultStatusLock();
}
console.log(`[故障状态更新任务] 上一次任务尚未完成，跳过本次执行`);
return false;
}
faultStatusTaskLock.locked = true;
faultStatusTaskLock.lockTime = Date.now();
faultStatusTaskLock.lockKey = `${Date.now()}_${Math.random().toString(36).substring(2, 8)}`;
return true;
}
/**
* 故障状态更新定时任务
*/
async function scheduledFaultStatusTask(): Promise<void> {
if (!tryAcquireFaultStatusLock()) {
return;
}
try {
console.log(`[故障状态更新任务] 开始执行 - ${new Date().toLocaleString()}`);
await processFaultStatus();
console.log(`[故障状态更新任务] 执行完成 - ${new Date().toLocaleString()}`);
} catch (err) {
console.error('[故障状态更新任务] 执行异常:', err);
} finally {
faultStatusTaskLock.locked = false;
faultStatusTaskLock.lockTime = 0;
faultStatusTaskLock.lockKey = '';
}
}
/**
* 启动故障状态更新定时任务
*/
export function startFaultStatusSchedule(): void {
if (faultStatusTimerHandle) {
console.warn('[故障状态更新任务] 定时器已在运行中，无需重复启动');
return;
}
console.log(`[故障状态更新任务] 定时器已启动，间隔 ${FAULT_STATUS_TASK_INTERVAL_MS / 60000} 分钟`);
faultStatusTimerHandle = setInterval(() => {
scheduledFaultStatusTask();
}, FAULT_STATUS_TASK_INTERVAL_MS);
}
/**
* 停止故障状态更新定时任务
*/
export function stopFaultStatusSchedule(): void {
if (faultStatusTimerHandle) {
clearInterval(faultStatusTimerHandle);
faultStatusTimerHandle = null;
console.log('[故障状态更新任务] 定时器已停止');
}
}
/**
* 停止定时任务
*/
export function stopSchedule(): void {
......@@ -140,3 +301,14 @@ export function stopSchedule(): void {
console.log('[定时任务] 定时器已停止');
}
}
/**
* 停止预警工单定时任务
*/
export function stopAlertSchedule(): void {
if (alertTimerHandle) {
clearInterval(alertTimerHandle);
alertTimerHandle = null;
console.log('[预警工单定时任务] 定时器已停止');
}
}
src/main.ts
......@@ -3,7 +3,7 @@ import * as mysqlDB from "./db/mysqlInit";
import * as mqttSrv from "./service/mqttInit";
import { initMysqlModel } from "./model/sqlModelBind";
import { httpServer } from "./net/http_server";
import { startSchedule } from "./config/schedule";
import { startSchedule, startAlertSchedule, startFaultStatusSchedule } from "./config/schedule";
async function lanuch() {
......@@ -18,6 +18,10 @@ async function lanuch() {
await mqttSrv.startMqttClient();
/**启动定时任务 */
startSchedule();
/**启动预警工单定时任务（10分钟） */
startAlertSchedule();
/**启动故障状态更新定时任务（5分钟） */
startFaultStatusSchedule();
console.log('This indicates that the server is started successfully.');
......
Markdown is supported
0% or
You are about to add 0 people to the discussion. Proceed with caution.
Finish editing this message first!
Please register or to comment