Skip to content

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

数据集成：数据清洗

parent 4114c762
Showing with 169 additions and 38 deletions
src/biz/dataIntegration.ts
......@@ -6,21 +6,36 @@ import {
getMaterStateList,
} from "./feiyiClient";
import { mysqlModelMap } from "../model/sqlModelBind";
import moment from "moment";
// ==================== 工具函数 ====================
/**
* 获取今天的日期字符串，格式 YYYYMMDD（使用本地时区，避免时区问题）
*/
function getTodayDateStr(): string {
return moment().format('YYYYMMDD');
}
// ==================== 分页拉取辅助函数 ====================
/**
* 分页拉取所有数据（通用）
* @param fetchFn 分页查询函数，入参 { page, limit }，返回 { page, limit, total, rows }
* @param fetchFn 分页查询函数，入参 { page, limit, ...extraParams }，返回 { page, limit, total, rows }
* @param extraParams 额外的查询参数，会透传每次调用
* @param limit 每页条数，默认 100
*/
async function fetchAllPages(fetchFn: (page: number, limit: number) => Promise<any>, limit: number = 100): Promise<any[]> {
async function fetchAllPages(
fetchFn: (page: number, limit: number, extraParams: Record<string, any>) => Promise<any>,
extraParams: Record<string, any> = {},
limit: number = 100,
): Promise<any[]> {
const allRows: any[] = [];
let page = 1;
let hasMore = true;
while (hasMore) {
const result = await fetchFn(page, limit);
const result = await fetchFn(page, limit, extraParams);
if (!result || !result.rows || result.rows.length === 0) {
hasMore = false;
break;
......@@ -330,11 +345,46 @@ async function integrateAcDeviceData() {
return;
}
// 分页拉取所有内机开关机记录
const rows = await fetchAllPages((page, limit) => getIndoorUnitStateList({ page, limit }));
// 1. 分页拉取当天内机开关机记录（使用 occurrenceTime 过滤当天数据）
const todayDate = getTodayDateStr();
const rows = await fetchAllPages(
(page, limit, extraParams) => getIndoorUnitStateList({ page, limit, occurrenceTime: extraParams.occurrenceTime }),
{ occurrenceTime: todayDate },
);
console.log(`[空调设备数据集成] 共获取 ${rows.length} 条开关机记录`);
if (rows.length === 0) {
console.log('[空调设备数据集成] 当天无开关机记录，跳过');
return;
}
let insertCount = 0;
// 2. 提取所有涉及的 device_id（去重），并过滤出已在 device 表中注册的设备
const deviceIdSet = new Set<string>();
for (const item of rows) {
if (item.indoorUnitAddressFull) {
deviceIdSet.add(item.indoorUnitAddressFull);
}
}
const deviceIds = Array.from(deviceIdSet);
// 3. 一次 SQL GROUP BY 查询这些设备在 device_data 表中各自的最大 device_time
const maxTimeMap = new Map<string, Date | null>();
if (deviceIds.length > 0) {
const maxTimeResults = await deviceDataModel.findAll({
attributes: [
'device_id',
[deviceDataModel.sequelize.fn('MAX', deviceDataModel.sequelize.col('device_time')), 'max_time'],
],
where: { device_id: deviceIds },
group: ['device_id'],
raw: true,
});
for (const r of maxTimeResults as any[]) {
maxTimeMap.set(r.device_id, r.max_time ? new Date(r.max_time) : null);
}
}
// 4. 内存过滤：保留 device_time > 该设备最大时间的记录
const toInsert: any[] = [];
let skipCount = 0;
for (const item of rows) {
const deviceId = item.indoorUnitAddressFull;
......@@ -343,13 +393,26 @@ async function integrateAcDeviceData() {
continue;
}
// 查找对应设备
const device = await deviceModel.findOne({ where: { device_id: deviceId } });
if (!device) {
// 校验设备是否已注册
if (!deviceIdSet.has(deviceId)) {
skipCount++;
continue;
}
const itemTime = item.createTime ? new Date(item.createTime) : null;
const maxTime = maxTimeMap.get(deviceId);
// 新设备（无历史数据）全部保留；已有数据的设备只保留增量
// 使用秒级时间戳比较，避免 MySQL DATETIME（秒精度）与接口毫秒时间戳精度不一致导致重复插入
if (maxTime !== undefined && maxTime !== null && itemTime) {
const itemSec = Math.floor(itemTime.getTime() / 1000);
const maxSec = Math.floor(maxTime.getTime() / 1000);
if (itemSec <= maxSec) {
skipCount++;
continue;
}
}
// 构造数据（按设备数据结构.md 空调设备数据格式）
const data = {
power: item.onOff === 1 ? 'on' : 'off',
......@@ -361,17 +424,23 @@ async function integrateAcDeviceData() {
minTemp: item.tempSetLo || 0,
autoControl: '生效',
};
await deviceDataModel.create({
toInsert.push({
device_id: deviceId,
data: data,
received_time: new Date(),
device_time: item.createTime ? new Date(item.createTime) : null,
device_time: itemTime,
});
insertCount++;
}
console.log(`[空调设备数据集成] 完成，新增 ${insertCount} 条，跳过 ${skipCount} 条`);
// TODO: 考虑是否需要在插入前对数据按 device_time 排序，避免数据库存储顺序混乱
// 5. 批量插入
if (toInsert.length > 0) {
await deviceDataModel.bulkCreate(toInsert);
}
console.log(`[空调设备数据集成] 完成，新增 ${toInsert.length} 条，跳过 ${skipCount} 条`);
}
/**
......@@ -387,11 +456,46 @@ async function integrateMeterDeviceData() {
return;
}
// 分页拉取所有抄表记录
const rows = await fetchAllPages((page, limit) => getMaterStateList({ page, limit }));
// 1. 分页拉取当天抄表记录（使用 occurrenceTime 过滤当天数据）
const todayDate = getTodayDateStr();
const rows = await fetchAllPages(
(page, limit, extraParams) => getMaterStateList({ page, limit, occurrenceTime: extraParams.occurrenceTime }),
{ occurrenceTime: todayDate },
);
console.log(`[电表设备数据集成] 共获取 ${rows.length} 条抄表记录`);
if (rows.length === 0) {
console.log('[电表设备数据集成] 当天无抄表记录，跳过');
return;
}
let insertCount = 0;
// 2. 提取所有涉及的 device_id（去重）
const deviceIdSet = new Set<string>();
for (const item of rows) {
if (item.gatewayCode) {
deviceIdSet.add(item.gatewayCode);
}
}
const deviceIds = Array.from(deviceIdSet);
// 3. 一次 SQL GROUP BY 查询这些设备在 device_data 表中各自的最大 device_time
const maxTimeMap = new Map<string, Date | null>();
if (deviceIds.length > 0) {
const maxTimeResults = await deviceDataModel.findAll({
attributes: [
'device_id',
[deviceDataModel.sequelize.fn('MAX', deviceDataModel.sequelize.col('device_time')), 'max_time'],
],
where: { device_id: deviceIds },
group: ['device_id'],
raw: true,
});
for (const r of maxTimeResults as any[]) {
maxTimeMap.set(r.device_id, r.max_time ? new Date(r.max_time) : null);
}
}
// 4. 内存过滤：保留 device_time > 该设备最大时间的记录
const toInsert: any[] = [];
let skipCount = 0;
for (const item of rows) {
const deviceId = item.gatewayCode;
......@@ -400,11 +504,18 @@ async function integrateMeterDeviceData() {
continue;
}
// 查找对应设备
const device = await deviceModel.findOne({ where: { device_id: deviceId } });
if (!device) {
skipCount++;
continue;
const itemTime = item.deviceTime ? new Date(item.deviceTime) : null;
const maxTime = maxTimeMap.get(deviceId);
// 新设备（无历史数据）全部保留；已有数据的设备只保留增量
// 使用秒级时间戳比较，避免 MySQL DATETIME（秒精度）与接口毫秒时间戳精度不一致导致重复插入
if (maxTime !== undefined && maxTime !== null && itemTime) {
const itemSec = Math.floor(itemTime.getTime() / 1000);
const maxSec = Math.floor(maxTime.getTime() / 1000);
if (itemSec <= maxSec) {
skipCount++;
continue;
}
}
// 构造数据（按设备数据结构.md 电表设备数据格式）
......@@ -416,16 +527,22 @@ async function integrateMeterDeviceData() {
switch: 'open',
};
await deviceDataModel.create({
toInsert.push({
device_id: deviceId,
data: data,
received_time: new Date(),
device_time: item.deviceTime ? new Date(item.deviceTime) : null,
device_time: itemTime,
});
insertCount++;
}
console.log(`[电表设备数据集成] 完成，新增 ${insertCount} 条，跳过 ${skipCount} 条`);
// TODO: 考虑是否需要在插入前对数据按 device_time 排序，避免数据库存储顺序混乱
// 5. 批量插入
if (toInsert.length > 0) {
await deviceDataModel.bulkCreate(toInsert);
}
console.log(`[电表设备数据集成] 完成，新增 ${toInsert.length} 条，跳过 ${skipCount} 条`);
}
// ==================== 暂无对应接口的集成函数（保留空实现） ====================
......
src/biz/running.ts
......@@ -101,9 +101,13 @@ export async function getRunAnalysis(regionGroup: string, regionType: string) {
monthlyYearOnYear
};
resultAny.electricityTrend = {
last24HoursMax: 0,
last24Hours: hourlyEnergy,
last7DaysMax: 0,
last7Days: dailyEnergy7,
last30DaysMax: 0,
last30Days: dailyEnergy30,
lastYearMax: 0,
lastYear: monthlyEnergyYear
};
......@@ -116,12 +120,12 @@ export async function getRunAnalysis(regionGroup: string, regionType: string) {
let airDeviceIds = airQualityDevices.data.map(d => d.device_id);
let qualityIndex = 300;
let co2Trend = [];
let currentCo2 = '21ppm';
let currentTemperature = '21℃';
let currentCo2 = '0ppm';
let currentTemperature = '0℃';
let temperatureTrend = [];
let currentHumidity = '47%';
let currentHumidity = '0%';
let humidityTrend = [];
let currentPm25 = '21μg/m³';
let currentPm25 = '0μg/m³';
let pm25Trend = [];
let co2TrendDetail = [];
......@@ -134,10 +138,10 @@ export async function getRunAnalysis(regionGroup: string, regionType: string) {
}, ["data"]);
if (latestAir.data.length) {
let airData = latestAir.data[0].data;
currentTemperature = `${airData.temperature ?? 21}℃`;
currentHumidity = `${airData.humidity ?? 47}%`;
currentPm25 = `${airData.pm25 ?? 21}μg/m³`;
currentCo2 = `${airData.co2 ?? 400}ppm`;
currentTemperature = `${airData.temperature ?? 0}℃`;
currentHumidity = `${airData.humidity ?? 0}%`;
currentPm25 = `${airData.pm25 ?? 0}μg/m³`;
currentCo2 = `${airData.co2 ?? 0}ppm`;
qualityIndex = 500 - (airData.pm25 ?? 0);
}
// 过去24小时趋势
......@@ -158,10 +162,10 @@ export async function getRunAnalysis(regionGroup: string, regionType: string) {
for (let i = 0; i < 24; i++) {
let key = `${i}时`;
let val = hourlyData.get(key);
temperatureTrend.push({ time: `${i}:00`, value: val?.temp?.toString() ?? '21' });
humidityTrend.push({ time: `${i}:00`, value: val?.hum?.toString() ?? '47' });
pm25Trend.push({ time: `${i}:00`, value: val?.pm?.toString() ?? '21' });
let co2Val = val?.co2 ?? 400;
temperatureTrend.push({ time: `${i}:00`, value: val?.temp?.toString() ?? '0' });
humidityTrend.push({ time: `${i}:00`, value: val?.hum?.toString() ?? '0' });
pm25Trend.push({ time: `${i}:00`, value: val?.pm?.toString() ?? '0' });
let co2Val = val?.co2 ?? 0;
co2TrendDetail.push({ time: `${i}:00`, value: (co2Val / 100).toFixed(1) });
co2Trend.push({ time: `${i}时`, value: co2Val.toString() });
}
......@@ -191,6 +195,11 @@ export async function getRunAnalysis(regionGroup: string, regionType: string) {
{ riskContent: "温度传感器离线", alertTime: "26/02/01", status: "已响应" },
{ riskContent: "湿度超标", alertTime: "26/02/02", status: "未处理" }
];
let alertWorkThreeOrders = [
{ riskContent: "空调压缩机异常", alertTime: "26/02/01", alertAddress: "A馆1F辉煌厅", device_id: "3-1-0-0", faultCode: "LU", status: "已解决" },
{ riskContent: "温度传感器离线", alertTime: "26/02/01", alertAddress: "A馆1F海外厅", device_id: "3-1-0-1", faultCode: "018", status: "已响应" },
{ riskContent: "湿度超标", alertTime: "26/02/02", alertAddress: "A馆1F走廊", device_id: "2-4-0-0", faultCode: "13", status: "未处理" }
];
let alertWorkLevel = {
"一级预警": 10,
"二级预警": 20,
......@@ -206,6 +215,11 @@ export async function getRunAnalysis(regionGroup: string, regionType: string) {
alertWorkType: alertWorkType,
alertWorkOrders: alertWorkOrders
};
// 三级页面，预警信息展示不同
if (regionType) {
resultAny.alertManagement.alertWorkOrders = alertWorkThreeOrders;
}
// 6.1 人流监控及客流量趋势分析
let regionInfos: any = [];
......
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