import { Provide } from '@midwayjs/core';
|
import { BaseService } from '@cool-midway/core';
|
import { InjectEntityModel } from '@midwayjs/typeorm';
|
import { Repository } from 'typeorm';
|
import { ReportDailysiteEntity } from '../entity/dailysite';
|
import { PushSiteweightEntity } from '../../push/entity/siteweight';
|
import { BasicdataSiteEntity } from '../../basicdata/entity/site'
|
|
@Provide()
|
export class TaskSiteService extends BaseService {
|
@InjectEntityModel(ReportDailysiteEntity)
|
reportDailysiteEntity: Repository<ReportDailysiteEntity>;
|
|
@InjectEntityModel(PushSiteweightEntity)
|
pushSiteweightEntity: Repository<PushSiteweightEntity>;
|
|
@InjectEntityModel(BasicdataSiteEntity)
|
basicdataSiteEntity: Repository<BasicdataSiteEntity>;
|
|
async aggregateDailyData(targetDate: string) {
|
const [year, month, day] = targetDate.split('-').map(Number);
|
const startDateTime = new Date(year, month - 1, day);
|
const end = new Date(startDateTime);
|
end.setHours(24, 0, 0, 0);
|
const endDateTime = end;
|
|
// 1. 内存优化核心：提前把所有的垃圾类型单价、碳排因数加载到内存字典中，避免循环内查库
|
const wasteTypes = await this.nativeQuery('SELECT code, price, carbonFactor, carbonDirection FROM t_basicdata_wastetype');
|
const wasteMap = new Map<string, { price: number; carbonFactor: number; carbonDirection: number }>();
|
for (const w of wasteTypes) {
|
wasteMap.set(w.code, {
|
price: Number(w.price) || 0,
|
carbonFactor: Number(w.carbonFactor) || 0,
|
carbonDirection: Number(w.carbonDirection) || 0
|
});
|
}
|
|
// 2. 数据库只做一件事情：过滤并捞出当天“收集点”的所有明细流数据（"过滤条件为"）
|
// 数据库只负责：精准过滤出当天属于“收集点”的、且真正产生称重的所有流水
|
const rawDetails = await this.pushSiteweightEntity.createQueryBuilder('weight')
|
.innerJoin(BasicdataSiteEntity, 'site', 'weight.businessId = site.businessId')
|
.select([
|
'site.id AS siteId', // 拿最新的自增ID作为每日表的 siteId
|
'site.businessName AS siteName', // 永远拿当前后台最新的站点名称，防止设备传旧名称
|
'site.departmentId AS departmentId',
|
'weight.weight AS weight',
|
'weight.garbageType AS garbageType'
|
])
|
.where('site.siteType = :siteType', { siteType: '收集点' })
|
.andWhere('weight.uploadTime BETWEEN :start AND :end', { start: startDateTime, end: endDateTime })
|
.getRawMany();
|
|
if (!rawDetails || rawDetails.length === 0) return;
|
|
// 3. 内存聚合核心：定义支持动态键名累加的对象字典
|
const siteAggMap = new Map<string, any>();
|
|
// 映射表：将明细里的 garbageType 映射到你 Entity 里的字段后缀
|
const typeKeyMap: Record<string, string> = {
|
'SW60': 'Sw60', 'SW61': 'Sw61', 'SW62': 'Sw62', 'SW63': 'Sw63', 'SW64': 'Sw64',
|
'900-001-S62': 'Sw62001', '900-002-S62': 'Sw62002', '900-003-S62': 'Sw62003',
|
'900-004-S62': 'Sw62004', '900-005-S62': 'Sw62005', '900-006-S62': 'Sw62006', '900-007-S62': 'Sw62007'
|
};
|
|
|
for (const row of rawDetails) {
|
const siteId = row.siteId;
|
const weight = Number(row.weight) || 0;
|
const gType = row.garbageType || 'SW64';
|
const fieldSuffix = typeKeyMap[gType];
|
|
// 初始化该站点的聚合结构
|
if (!siteAggMap.has(siteId)) {
|
const initData: any = {
|
date: targetDate,
|
siteId: siteId,
|
siteName: row.siteName,
|
departmentId:row.departmentId,
|
weightTotal: 0, timeTotal: 0, carbonTotal: 0
|
};
|
// 动态初始化宽表里各子类型的 0 值
|
Object.values(typeKeyMap).forEach(suffix => {
|
initData[`weight${suffix}`] = 0;
|
initData[`time${suffix}`] = 0;
|
initData[`carbon${suffix}`] = 0;
|
// 仅 sw62 系列有估算销售额
|
if (suffix.startsWith('Sw62')) {
|
initData[`estimatedSales${suffix}`] = 0;
|
}
|
});
|
siteAggMap.set(siteId, initData);
|
}
|
|
const agg = siteAggMap.get(siteId);
|
|
// 通用总计累加
|
agg.weightTotal += weight;
|
agg.timeTotal += 1; // 称重次数 + 1
|
|
// 提取该垃圾类型的碳排与销售额计算规则
|
const wasteCfg = wasteMap.get(row.garbageType) || { price: 0, carbonFactor: 0, carbonDirection: 0 };
|
|
// 计算单笔碳排
|
let currentCarbon = weight * wasteCfg.carbonFactor;
|
if (wasteCfg.carbonDirection === 1) currentCarbon = -currentCarbon; // 负向碳排
|
agg.carbonTotal += currentCarbon;
|
|
// 如果属于我们统计的行转列子类型，进行精确归类
|
if (fieldSuffix) {
|
agg[`weight${fieldSuffix}`] += weight;
|
agg[`time${fieldSuffix}`] += 1;
|
agg[`carbon${fieldSuffix}`] += currentCarbon;
|
|
// 如果是 sw62 及其子类，额外计算估算销售额
|
if (fieldSuffix.startsWith('Sw62')) {
|
const currentSales = weight * wasteCfg.price;
|
agg[`estimatedSales${fieldSuffix}`] += currentSales;
|
}
|
}
|
}
|
|
// 4. 高性能写入：转换为数组，利用批量插入 + ON DUPLICATE KEY UPDATE 提交给 MySQL
|
const finalRecords = Array.from(siteAggMap.values()).map(item => {
|
// 浮点数防精度丢失处理，保留两位小数
|
Object.keys(item).forEach(key => {
|
if (typeof item[key] === 'number') {
|
item[key] = Number(item[key].toFixed(2));
|
}
|
});
|
return item;
|
});
|
|
if (finalRecords.length > 0) {
|
// 1. 动态抓取所有的键，将主键、联合唯一键彻底从“待更新字段”中排除
|
// 这样能确保 ON DUPLICATE KEY UPDATE 后面的赋值语句绝对干净
|
const allUpdateFields = Object.keys(finalRecords[0]).filter(
|
k => k !== 'date' && k !== 'siteId' && k !== 'id'
|
);
|
|
// 2. 依然推荐采用 Chunk 分批，这是应对海量数据/多子字段最安全、执行效率最高的做法
|
const chunkSize = 50;
|
for (let i = 0; i < finalRecords.length; i += chunkSize) {
|
const chunk = finalRecords.slice(i, i + chunkSize);
|
|
await this.reportDailysiteEntity.createQueryBuilder()
|
.insert()
|
.values(chunk)
|
// 💡 显式声明：当 ['date', 'siteId'] 发生冲突时，强行把 allUpdateFields 里的字段全部更新一遍
|
.orUpdate(allUpdateFields, ['date', 'siteId'])
|
.execute();
|
}
|
}
|
|
return finalRecords.length;
|
}
|
}
|
