import { Provide, Inject } from '@midwayjs/core';
|
import { InjectEntityModel } from '@midwayjs/typeorm';
|
import { Repository, In } from 'typeorm';
|
import { ReportJoblogEntity } from '../entity/joblog';
|
import { TaskServiceSite } from './taskservicesite'
|
import { TaskServiceDepartment } from './taskservicedepartment'
|
|
|
|
@Provide()
|
export class ExecutorSiteDeptDaily {
|
|
@InjectEntityModel(ReportJoblogEntity)
|
reportJoblogEntity: Repository<ReportJoblogEntity>;
|
|
@Inject()
|
taskServiceSite: TaskServiceSite;
|
|
@Inject()
|
taskServiceDepartment: TaskServiceDepartment;
|
|
async execute() {
|
// 捞出所有待处理任务
|
const pendingSiteJobs = await this.reportJoblogEntity.find({
|
where: {
|
status: In(['INIT', 'FAILED']),
|
jobName: In(['SITE_DAILY', 'DEPT_DAILY']),
|
},
|
order: {
|
dataDate: 'ASC', // 先算历史，后算今天
|
// 确保在同一天内，SITE_DAILY 先跑，DEPT_DAILY 后跑
|
// 在 JS 数组中我们可以进一步控制排序，或者由业务逻辑严格控制依赖
|
},
|
});
|
|
for (const job of pendingSiteJobs) {
|
// 严格依赖检查：如果是部门日表任务，必须确保同天的站点日表任务已经是 SUCCESS
|
if (job.jobName === 'DEPT_DAILY') {
|
const siteJob = await this.reportJoblogEntity.findOneBy({
|
dataDate: job.dataDate,
|
jobName: 'SITE_DAILY',
|
});
|
if (!siteJob || siteJob.status !== 'SUCCESS') {
|
// 源头明细还没算好/或还在运行，部门表任务必须等待，跳过本次执行
|
continue;
|
}
|
}
|
|
// 锁定任务状态为 RUNNING（轻量级乐观锁锁机制）
|
const lockResult = await this.reportJoblogEntity.update(
|
{ id: job.id, status: job.status }, // 带有原始状态检查，防止被别的线程抢跑
|
{ status: 'RUNNING', startTime: new Date() }
|
);
|
|
if (lockResult.affected === 0) {
|
continue; // 没抢到锁，说明别的定时器在跑它，跳过
|
}
|
|
// 真正开始跑数据
|
try {
|
let recordsProcessed = 0;
|
|
if (job.jobName === 'SITE_DAILY') {
|
// 调用你写的 站点日表 提取逻辑
|
recordsProcessed = await this.taskServiceSite.aggregateDailyData(job.dataDate) || 0;
|
} else if (job.jobName === 'DEPT_DAILY') {
|
// 调用你写的 部门树轰炸 提取逻辑
|
recordsProcessed = await this.taskServiceDepartment.aggregateDailyData(job.dataDate) || 0;
|
}
|
|
// 成功：更新状态
|
await this.reportJoblogEntity.update(job.id, {
|
status: 'SUCCESS',
|
recordsProcessed: recordsProcessed,
|
endTime: new Date(),
|
errorMessage: '',
|
});
|
|
} catch (error) {
|
// 失败：记录堆栈，等待下一小时重试
|
|
var errorMessage = '';
|
if (error instanceof Error) {
|
errorMessage = error.stack ?? error.message;
|
}
|
if (typeof error === 'string') {
|
errorMessage = error;
|
}
|
try {
|
errorMessage = JSON.stringify(error);
|
} catch {
|
errorMessage = String(error);
|
}
|
|
await this.reportJoblogEntity.update(job.id, {
|
status: 'FAILED',
|
endTime: new Date(),
|
errorMessage: errorMessage,
|
failedCount: () => 'failedCount + 1',
|
});
|
}
|
}
|
}
|
|
}
|
