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 前面
|
pendingSiteJobs.sort((a, b) => {
|
const d = String(a.dataDate).localeCompare(String(b.dataDate));
|
if (d !== 0) return d;
|
return a.jobName === 'SITE_DAILY' ? -1 : 1;
|
});
|
|
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) {
|
// 失败：记录堆栈，等待下一小时重试
|
let errorMessage = '';
|
if (error instanceof Error) {
|
errorMessage = error.stack ?? error.message;
|
console.error('[JOB-FAILED]', job.jobName, job.dataDate, error?.message ?? error);
|
console.error(error.stack);
|
} else if (typeof error === 'string') {
|
errorMessage = error;
|
} else {
|
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',
|
});
|
}
|
}
|
}
|
|
}
|
