add pdfjs and traderecord

wangzhibo

2026-07-28 920cec41cedfde27f89d7c19b78e03147b926ab5

 src/modules/report/service/taskservice.ts

@@ -3,8 +3,8 @@
import { InjectEntityModel } from '@midwayjs/typeorm';
import { Repository, In } from 'typeorm';
import { ReportJoblogEntity } from '../entity/joblog';
import { TaskSiteService } from './tasksiteservice'
import { TaskDepartmentService } from './taskdepartmentservice'
import { ExecutorSiteDeptDaily } from './ExecutorSiteDeptDaily'
import { ExecutorVehicleDaily } from './ExecutorVehicleDaily'

/**
 *  定时任务-汇总站点/各个单位的各类垃圾汇总数据
@@ -17,12 +17,13 @@
  @InjectEntityModel(ReportJoblogEntity)
  reportJoblogEntity: Repository<ReportJoblogEntity>;

  // 注入你具体的业务 Service，用于跑真正的汇总 SQL
  @Inject()
  taskSiteService: TaskSiteService;
  executorSiteDeptDaily: ExecutorSiteDeptDaily;

  @Inject()
  departmentCubeService: TaskDepartmentService;
  executorVehicleDaily: ExecutorVehicleDaily;



  /**
   * 外部定时任务（Crontab）每小时调用的总入口
@@ -32,10 +33,17 @@
    // 1. 自动看门狗：确保昨天和今天的任务实例已经在表里初始化了
    await this.ensureJobInstancesCreated();

    // 2. 执行引擎：捞出需要执行或重试的任务
    await this.executePendingJobs();
    // 2. 两类任务同时执行
    // 后续增加新的 task service 比如 更新 萤石云 token， 统计分拣员行为等等。

    return '任务执行成功:ReportTaskService';
    await Promise.all([
      this.executorSiteDeptDaily.execute(),
      this.executorVehicleDaily.execute()

    ]);


    return '任务执行:ReportTaskService';
  }

  /**
@@ -56,10 +64,14 @@

    // 我们需要守护的目标日期和任务类型组合
    const targets = [
      { date: yesterdayStr, type: 'SITE_DAILY',isToday: false},
      { date: yesterdayStr, type: 'DEPT_DAILY',isToday: false},
      { date: todayStr, type: 'SITE_DAILY' ,isToday: true},
      { date: todayStr, type: 'DEPT_DAILY' ,isToday: true},
      { date: yesterdayStr, type: 'SITE_DAILY', isToday: false },
      { date: yesterdayStr, type: 'DEPT_DAILY', isToday: false },
      { date: yesterdayStr, type: 'VEHICLE_DAILY', isToday: false },   // 新增

      { date: todayStr, type: 'SITE_DAILY', isToday: true },
      { date: todayStr, type: 'DEPT_DAILY', isToday: true },
      { date: todayStr, type: 'VEHICLE_DAILY', isToday: true },         // 新增

    ];

    for (const target of targets) {
@@ -75,7 +87,7 @@
          job.dataDate = target.date;
          job.status = 'INIT';
          await this.reportJoblogEntity.save(job);
        }else if (target.isToday && exist.status === 'SUCCESS') {
        } else if (target.isToday && exist.status === 'SUCCESS') {
          // 2. 【核心修复】如果是今天的任务，且已经是 SUCCESS 状态，强行洗回 INIT 参与本小时的重算
          await this.reportJoblogEntity.update(exist.id, {
            status: 'INIT'
@@ -87,88 +99,5 @@
    }
  }

  /**
   * 捞出 INIT 和 FAILED 的任务，并串行严格按顺序执行
   */
  private async executePendingJobs() {
    // 捞出所有待处理任务
    const pendingJobs = await this.reportJoblogEntity.find({
      where: {
        status: In(['INIT', 'FAILED']),
      },
      order: {
        dataDate: 'ASC', // 先算历史，后算今天
        // 确保在同一天内，SITE_DAILY 先跑，DEPT_DAILY 后跑
        // 在 JS 数组中我们可以进一步控制排序，或者由业务逻辑严格控制依赖
      },
    });

    for (const job of pendingJobs) {
      // 严格依赖检查：如果是部门日表任务，必须确保同天的站点日表任务已经是 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.taskSiteService.aggregateDailyData(job.dataDate) || 0;
        } else if (job.jobName === 'DEPT_DAILY') {
          // 调用你写的 部门树轰炸 提取逻辑
          recordsProcessed = await this.departmentCubeService.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',
        });
      }
    }
  }
}