添加收集点和机构的汇总

wangzhibo

2026-07-18 167510c1019f2c72ce9a544daec91aa5fe5409a5

添加收集点和机构的汇总

 src/modules/report/config.ts

New file
@@ -0,0 +1,14 @@
import { ModuleConfig } from '@cool-midway/core';

/**
 * 运维统计与ETL日志模块配置
 */
export default () => {
  return {
    name: '运维汇总模块',
    description: '提供站点与部门每日数据统计汇总及ETL执行日志审计功能',
    middlewares: [],
    globalMiddlewares: [],
    order: 0
  } as ModuleConfig;
};

 src/modules/report/controller/admin/dailydepartment.ts

New file
@@ -0,0 +1,17 @@
import { CoolController, BaseController } from '@cool-midway/core';
import { ReportDailyDepartmentEntity } from '../../entity/dailydepartment';
import { ReportDailyDepartmentService } from '../../service/dailydepartment';

/**
 * 每日部门汇总控制器
 */
@CoolController({
  api: ['add', 'delete', 'update', 'info', 'list', 'page'],
  entity: ReportDailyDepartmentEntity,
  service: ReportDailyDepartmentService,
  pageQueryOp: {
    keyWordLikeFields: ['a.departmentName'],
    fieldEq: ['a.date', 'a.departmentId']
  }
})
export class AdminReportDailydepartmentController extends BaseController {}

 src/modules/report/controller/admin/dailysite.ts

New file
@@ -0,0 +1,17 @@
import { CoolController, BaseController } from '@cool-midway/core';
import { ReportDailysiteEntity } from '../../entity/dailysite';
import { ReportDailysiteService } from '../../service/dailysite';

/**
 * 每日站点汇总控制器
 */
@CoolController({
  api: ['add', 'delete', 'update', 'info', 'list', 'page'],
  entity: ReportDailysiteEntity,
  service: ReportDailysiteService,
  pageQueryOp: {
    keyWordLikeFields: ['a.siteName'],
    fieldEq: ['a.date', 'a.siteId']
  }
})
export class AdminReportDailysiteController extends BaseController {}

 src/modules/report/controller/admin/joblog.ts

New file
@@ -0,0 +1,17 @@
import { CoolController, BaseController } from '@cool-midway/core';
import { ReportJoblogEntity } from '../../entity/joblog';
import { ReportJoblogService } from '../../service/joblog';

/**
 * ETL任务日志控制器
 */
@CoolController({
  api: ['add', 'delete', 'update', 'info', 'list', 'page'],
  entity: ReportJoblogEntity,
  service: ReportJoblogService,
  pageQueryOp: {
    keyWordLikeFields: ['a.jobName'],
    fieldEq: ['a.dataDate', 'a.status']
  }
})
export class AdminReportJoblogController extends BaseController {}

 src/modules/report/entity/dailydepartment.ts

New file
@@ -0,0 +1,163 @@
import { BaseEntity } from '../../base/entity/base';
import { Column, Entity, Index, Unique } from 'typeorm';

/**
 * 每日站点汇总
 */
@Entity('t_report_daily_department')
@Unique('uk_date_department', ['date', 'departmentId']) // 💡 部门表的复合唯一约束
export class ReportDailyDepartmentEntity extends BaseEntity {
  @Index()
  @Column({ comment: '日期', type: 'date' })
  date: string;

  @Index()
  @Column({ comment: '部门ID' })
  departmentId: number;

  @Index()
  @Column({ comment: '部门名称' })
  departmentName: string;

  @Column({ comment: '总重量', type: 'decimal', precision: 15, scale: 2, default: 0.00 })
  weightTotal: number;

  @Column({ comment: 'sw60重量', type: 'decimal', precision: 15, scale: 2, default: 0.00 })
  weightSw60: number;

  @Column({ comment: 'sw61重量', type: 'decimal', precision: 15, scale: 2, default: 0.00 })
  weightSw61: number;

  @Column({ comment: 'sw62重量', type: 'decimal', precision: 15, scale: 2, default: 0.00 })
  weightSw62: number;

  @Column({ comment: 'sw63重量', type: 'decimal', precision: 15, scale: 2, default: 0.00 })
  weightSw63: number;

  @Column({ comment: 'sw64重量', type: 'decimal', precision: 15, scale: 2, default: 0.00 })
  weightSw64: number;

  @Column({ comment: 'sw62_001重量', type: 'decimal', precision: 15, scale: 2, default: 0.00 })
  weightSw62001: number;

  @Column({ comment: 'sw62_002重量', type: 'decimal', precision: 15, scale: 2, default: 0.00 })
  weightSw62002: number;

  @Column({ comment: 'sw62_003重量', type: 'decimal', precision: 15, scale: 2, default: 0.00 })
  weightSw62003: number;

  @Column({ comment: 'sw62_004重量', type: 'decimal', precision: 15, scale: 2, default: 0.00 })
  weightSw62004: number;

  @Column({ comment: 'sw62_005重量', type: 'decimal', precision: 15, scale: 2, default: 0.00 })
  weightSw62005: number;

  @Column({ comment: 'sw62_006重量', type: 'decimal', precision: 15, scale: 2, default: 0.00 })
  weightSw62006: number;

  @Column({ comment: 'sw62_007重量', type: 'decimal', precision: 15, scale: 2, default: 0.00 })
  weightSw62007: number;

  @Column({ comment: '总称重次数', type: 'decimal', precision: 10, scale: 2, default: 0.00 })
  timeTotal: number;

  @Column({ comment: 'sw60次数', type: 'decimal', precision: 10, scale: 2, default: 0.00 })
  timeSw60: number;

  @Column({ comment: 'sw61次数', type: 'decimal', precision: 10, scale: 2, default: 0.00 })
  timeSw61: number;

  @Column({ comment: 'sw62次数', type: 'decimal', precision: 10, scale: 2, default: 0.00 })
  timeSw62: number;

  @Column({ comment: 'sw63次数', type: 'decimal', precision: 10, scale: 2, default: 0.00 })
  timeSw63: number;

  @Column({ comment: 'sw64次数', type: 'decimal', precision: 10, scale: 2, default: 0.00 })
  timeSw64: number;

  @Column({ comment: 'sw62001次数', type: 'decimal', precision: 10, scale: 2, default: 0.00 })
  timeSw62001: number;

  @Column({ comment: 'sw62002次数', type: 'decimal', precision: 10, scale: 2, default: 0.00 })
  timeSw62002: number;

  @Column({ comment: 'sw62003次数', type: 'decimal', precision: 10, scale: 2, default: 0.00 })
  timeSw62003: number;

  @Column({ comment: 'sw62004次数', type: 'decimal', precision: 10, scale: 2, default: 0.00 })
  timeSw62004: number;

  @Column({ comment: 'sw62005次数', type: 'decimal', precision: 10, scale: 2, default: 0.00 })
  timeSw62005: number;

  @Column({ comment: 'sw62006次数', type: 'decimal', precision: 10, scale: 2, default: 0.00 })
  timeSw62006: number;

  @Column({ comment: 'sw62007次数', type: 'decimal', precision: 10, scale: 2, default: 0.00 })
  timeSw62007: number;

  @Column({ comment: '总碳排量', type: 'decimal', precision: 15, scale: 2, default: 0.0 })
  carbonTotal: number;

  @Column({ comment: 'sw60碳排量', type: 'decimal', precision: 15, scale: 2, default: 0.0 })
  carbonSw60: number;

  @Column({ comment: 'sw61碳排量', type: 'decimal', precision: 15, scale: 2, default: 0.0 })
  carbonSw61: number;

  @Column({ comment: 'sw62碳排量', type: 'decimal', precision: 15, scale: 2, default: 0.0 })
  carbonSw62: number;

  @Column({ comment: 'sw63碳排量', type: 'decimal', precision: 15, scale: 2, default: 0.0 })
  carbonSw63: number;

  @Column({ comment: 'sw64碳排量', type: 'decimal', precision: 15, scale: 2, default: 0.0 })
  carbonSw64: number;

  @Column({ comment: 'sw62001碳排量', type: 'decimal', precision: 15, scale: 2, default: 0.0 })
  carbonSw62001: number;

  @Column({ comment: 'sw62002碳排量', type: 'decimal', precision: 15, scale: 2, default: 0.0 })
  carbonSw62002: number;

  @Column({ comment: 'sw62003碳排量', type: 'decimal', precision: 15, scale: 2, default: 0.0 })
  carbonSw62003: number;

  @Column({ comment: 'sw62004碳排量', type: 'decimal', precision: 15, scale: 2, default: 0.0 })
  carbonSw62004: number;

  @Column({ comment: 'sw62005碳排量', type: 'decimal', precision: 15, scale: 2, default: 0.0 })
  carbonSw62005: number;

  @Column({ comment: 'sw62006碳排量', type: 'decimal', precision: 15, scale: 2, default: 0.0 })
  carbonSw62006: number;

  @Column({ comment: 'sw62007碳排量', type: 'decimal', precision: 15, scale: 2, default: 0.0 })
  carbonSw62007: number;

  @Column({ comment: 'sw62估算销售额', type: 'decimal', precision: 15, scale: 2, default: 0.00 })
  estimatedSalesSw62: number;

  @Column({ comment: 'sw62_001估算销售额', type: 'decimal', precision: 15, scale: 2, default: 0.00 })
  estimatedSalesSw62001: number;

  @Column({ comment: 'sw62_002估算销售额', type: 'decimal', precision: 15, scale: 2, default: 0.00 })
  estimatedSalesSw62002: number;

  @Column({ comment: 'sw62_003估算销售额', type: 'decimal', precision: 15, scale: 2, default: 0.00 })
  estimatedSalesSw62003: number;

  @Column({ comment: 'sw62_004估算销售额', type: 'decimal', precision: 15, scale: 2, default: 0.00 })
  estimatedSalesSw62004: number;

  @Column({ comment: 'sw62_005估算销售额', type: 'decimal', precision: 15, scale: 2, default: 0.00 })
  estimatedSalesSw62005: number;

  @Column({ comment: 'sw62_006估算销售额', type: 'decimal', precision: 15, scale: 2, default: 0.00 })
  estimatedSalesSw62006: number;

  @Column({ comment: 'sw62_007估算销售额', type: 'decimal', precision: 15, scale: 2, default: 0.00 })
  estimatedSalesSw62007: number;

}

 src/modules/report/entity/dailysite.ts

New file
@@ -0,0 +1,163 @@
import { BaseEntity } from '../../base/entity/base';
import { Column, Entity, Index, Unique } from 'typeorm';

/**
 * 每日站点汇总
 */
@Entity('t_report_daily_site')
@Unique('uk_date_site', ['date', 'siteId']) // 💡 重点：在这里加上复合唯一约束，第一个参数是索引名，第二个是字段数组
export class ReportDailysiteEntity extends BaseEntity {
  @Index()
  @Column({ comment: '日期', type: 'date' })
  date: string;

  @Index()
  @Column({ comment: '站点ID' })
  siteId: number;

  @Index()
  @Column({ comment: '站点名称' })
  siteName: string;

  @Column({ comment: '总重量', type: 'decimal', precision: 15, scale: 2, default: 0.00 })
  weightTotal: number;

  @Column({ comment: 'sw60重量', type: 'decimal', precision: 15, scale: 2, default: 0.00 })
  weightSw60: number;

  @Column({ comment: 'sw61重量', type: 'decimal', precision: 15, scale: 2, default: 0.00 })
  weightSw61: number;

  @Column({ comment: 'sw62重量', type: 'decimal', precision: 15, scale: 2, default: 0.00 })
  weightSw62: number;

  @Column({ comment: 'sw63重量', type: 'decimal', precision: 15, scale: 2, default: 0.00 })
  weightSw63: number;

  @Column({ comment: 'sw64重量', type: 'decimal', precision: 15, scale: 2, default: 0.00 })
  weightSw64: number;

  @Column({ comment: 'sw62_001重量', type: 'decimal', precision: 15, scale: 2, default: 0.00 })
  weightSw62001: number;

  @Column({ comment: 'sw62_002重量', type: 'decimal', precision: 15, scale: 2, default: 0.00 })
  weightSw62002: number;

  @Column({ comment: 'sw62_003重量', type: 'decimal', precision: 15, scale: 2, default: 0.00 })
  weightSw62003: number;

  @Column({ comment: 'sw62_004重量', type: 'decimal', precision: 15, scale: 2, default: 0.00 })
  weightSw62004: number;

  @Column({ comment: 'sw62_005重量', type: 'decimal', precision: 15, scale: 2, default: 0.00 })
  weightSw62005: number;

  @Column({ comment: 'sw62_006重量', type: 'decimal', precision: 15, scale: 2, default: 0.00 })
  weightSw62006: number;

  @Column({ comment: 'sw62_007重量', type: 'decimal', precision: 15, scale: 2, default: 0.00 })
  weightSw62007: number;

  @Column({ comment: '总称重次数', type: 'decimal', precision: 10, scale: 2, default: 0.00 })
  timeTotal: number;

  @Column({ comment: 'sw60次数', type: 'decimal', precision: 10, scale: 2, default: 0.00 })
  timeSw60: number;

  @Column({ comment: 'sw61次数', type: 'decimal', precision: 10, scale: 2, default: 0.00 })
  timeSw61: number;

  @Column({ comment: 'sw62次数', type: 'decimal', precision: 10, scale: 2, default: 0.00 })
  timeSw62: number;

  @Column({ comment: 'sw63次数', type: 'decimal', precision: 10, scale: 2, default: 0.00 })
  timeSw63: number;

  @Column({ comment: 'sw64次数', type: 'decimal', precision: 10, scale: 2, default: 0.00 })
  timeSw64: number;

  @Column({ comment: 'sw62001次数', type: 'decimal', precision: 10, scale: 2, default: 0.00 })
  timeSw62001: number;

  @Column({ comment: 'sw62002次数', type: 'decimal', precision: 10, scale: 2, default: 0.00 })
  timeSw62002: number;

  @Column({ comment: 'sw62003次数', type: 'decimal', precision: 10, scale: 2, default: 0.00 })
  timeSw62003: number;

  @Column({ comment: 'sw62004次数', type: 'decimal', precision: 10, scale: 2, default: 0.00 })
  timeSw62004: number;

  @Column({ comment: 'sw62005次数', type: 'decimal', precision: 10, scale: 2, default: 0.00 })
  timeSw62005: number;

  @Column({ comment: 'sw62006次数', type: 'decimal', precision: 10, scale: 2, default: 0.00 })
  timeSw62006: number;

  @Column({ comment: 'sw62007次数', type: 'decimal', precision: 10, scale: 2, default: 0.00 })
  timeSw62007: number;

  @Column({ comment: '总碳排量', type: 'decimal', precision: 15, scale: 2, default: 0.0 })
  carbonTotal: number;

  @Column({ comment: 'sw60碳排量', type: 'decimal', precision: 15, scale: 2, default: 0.0 })
  carbonSw60: number;

  @Column({ comment: 'sw61碳排量', type: 'decimal', precision: 15, scale: 2, default: 0.0 })
  carbonSw61: number;

  @Column({ comment: 'sw62碳排量', type: 'decimal', precision: 15, scale: 2, default: 0.0 })
  carbonSw62: number;

  @Column({ comment: 'sw63碳排量', type: 'decimal', precision: 15, scale: 2, default: 0.0 })
  carbonSw63: number;

  @Column({ comment: 'sw64碳排量', type: 'decimal', precision: 15, scale: 2, default: 0.0 })
  carbonSw64: number;

  @Column({ comment: 'sw62001碳排量', type: 'decimal', precision: 15, scale: 2, default: 0.0 })
  carbonSw62001: number;

  @Column({ comment: 'sw62002碳排量', type: 'decimal', precision: 15, scale: 2, default: 0.0 })
  carbonSw62002: number;

  @Column({ comment: 'sw62003碳排量', type: 'decimal', precision: 15, scale: 2, default: 0.0 })
  carbonSw62003: number;

  @Column({ comment: 'sw62004碳排量', type: 'decimal', precision: 15, scale: 2, default: 0.0 })
  carbonSw62004: number;

  @Column({ comment: 'sw62005碳排量', type: 'decimal', precision: 15, scale: 2, default: 0.0 })
  carbonSw62005: number;

  @Column({ comment: 'sw62006碳排量', type: 'decimal', precision: 15, scale: 2, default: 0.0 })
  carbonSw62006: number;

  @Column({ comment: 'sw62007碳排量', type: 'decimal', precision: 15, scale: 2, default: 0.0 })
  carbonSw62007: number;

  @Column({ comment: 'sw62估算销售额', type: 'decimal', precision: 15, scale: 2, default: 0.00 })
  estimatedSalesSw62: number;

  @Column({ comment: 'sw62_001估算销售额', type: 'decimal', precision: 15, scale: 2, default: 0.00 })
  estimatedSalesSw62001: number;

  @Column({ comment: 'sw62_002估算销售额', type: 'decimal', precision: 15, scale: 2, default: 0.00 })
  estimatedSalesSw62002: number;

  @Column({ comment: 'sw62_003估算销售额', type: 'decimal', precision: 15, scale: 2, default: 0.00 })
  estimatedSalesSw62003: number;

  @Column({ comment: 'sw62_004估算销售额', type: 'decimal', precision: 15, scale: 2, default: 0.00 })
  estimatedSalesSw62004: number;

  @Column({ comment: 'sw62_005估算销售额', type: 'decimal', precision: 15, scale: 2, default: 0.00 })
  estimatedSalesSw62005: number;

  @Column({ comment: 'sw62_006估算销售额', type: 'decimal', precision: 15, scale: 2, default: 0.00 })
  estimatedSalesSw62006: number;

  @Column({ comment: 'sw62_007估算销售额', type: 'decimal', precision: 15, scale: 2, default: 0.00 })
  estimatedSalesSw62007: number;

}

 src/modules/report/entity/joblog.ts

New file
@@ -0,0 +1,40 @@
import { BaseEntity } from '../../base/entity/base';
import { Column, Entity, Index } from 'typeorm';

/**
 * ETL任务日志
 */
@Entity('t_report_joblog')
@Index(['jobName', 'dataDate'], { unique: true })
export class ReportJoblogEntity extends BaseEntity {
  @Index()
  @Column({ comment: '任务类型: SITE_DAILY(站点日汇总), DEPT_DAILY(部门日汇总)' })
  jobName: string;

  @Index()
  @Column({ comment: '数据目标日期', type: 'date' })
  dataDate: string;

  @Column({ type: 'datetime',comment: '开始时间', nullable: true })
  startTime: Date;

  @Column({ type: 'datetime',comment: '结束时间', nullable: true })
  endTime: Date;

  @Column({
    type: 'enum',
    enum: ['INIT', 'RUNNING', 'SUCCESS', 'FAILED'],
    default: 'INIT',
    comment: '状态: 未开始, 运行中, 成功, 失败',
  })
  status: 'INIT' | 'RUNNING' | 'SUCCESS' | 'FAILED';

  @Column({ comment: '处理记录数', default: 0 })
  recordsProcessed: number;

  @Column({ comment: '失败次数', default: 0 })
  failedCount: number;

  @Column({ comment: '错误堆栈信息', type: 'text', nullable: true })
  errorMessage: string;
}

 src/modules/report/service/dailydepartment.ts

New file
@@ -0,0 +1,14 @@
import { Provide } from '@midwayjs/core';
import { BaseService } from '@cool-midway/core';
import { InjectEntityModel } from '@midwayjs/typeorm';
import { Repository } from 'typeorm';
import { ReportDailyDepartmentEntity } from '../entity/dailydepartment';

/**
 * 每日部门汇总服务
 */
@Provide()
export class ReportDailyDepartmentService extends BaseService {
  @InjectEntityModel(ReportDailyDepartmentEntity)
  reportDailyDepartmentEntity: Repository<ReportDailyDepartmentEntity>;
}

 src/modules/report/service/dailysite.ts

New file
@@ -0,0 +1,14 @@
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';

/**
 * 每日站点汇总服务
 */
@Provide()
export class ReportDailysiteService extends BaseService {
  @InjectEntityModel(ReportDailysiteEntity)
  reportDailysiteEntity: Repository<ReportDailysiteEntity>;
}

 src/modules/report/service/joblog.ts

New file
@@ -0,0 +1,14 @@
import { Provide } from '@midwayjs/core';
import { BaseService } from '@cool-midway/core';
import { InjectEntityModel } from '@midwayjs/typeorm';
import { Repository } from 'typeorm';
import { ReportJoblogEntity } from '../entity/joblog';

/**
 * ETL任务日志服务
 */
@Provide()
export class ReportJoblogService extends BaseService {
  @InjectEntityModel(ReportJoblogEntity)
  reportJoblogEntity: Repository<ReportJoblogEntity>;
}

 src/modules/report/service/taskdepartmentservice.ts

New file
@@ -0,0 +1,160 @@
import { Provide, Inject } from '@midwayjs/core';
import { BaseService } from '@cool-midway/core';
import { InjectEntityModel } from '@midwayjs/typeorm';
import { Repository } from 'typeorm';
import { ReportDailysiteEntity } from '../entity/dailysite';
import { ReportDailyDepartmentEntity } from '../entity/dailydepartment';
import { BaseSysDepartmentEntity } from '../../base/entity/sys/department'; 
import { BasicdataSiteEntity } from '../../basicdata/entity/site';


/**
 *  定时任务-子任务 汇总部门的各类垃圾数据
 */
@Provide()
export class TaskDepartmentService extends BaseService {
  @InjectEntityModel(ReportDailysiteEntity)
  reportDailysiteEntity: Repository<ReportDailysiteEntity>;

  @InjectEntityModel(ReportDailyDepartmentEntity)
  reportDailyDepartmentEntity: Repository<ReportDailyDepartmentEntity>;

  @InjectEntityModel(BaseSysDepartmentEntity)
  baseSysDepartmentEntity: Repository<BaseSysDepartmentEntity>;

  @InjectEntityModel(BasicdataSiteEntity)
  basicdataSiteEntity: Repository<BasicdataSiteEntity>;

  /**
   * 部门日表无级递归汇总引擎入口
   * @param targetDate 目标计算日期格式 'YYYY-MM-DD'
   */
  async aggregateDailyData(targetDate: string) {
    // 1. 异步并行加载基础字典和当天已算好的站点日表数据，最大化I/O效率
    const [allDepts, allSites, dailySites] = await Promise.all([
      this.baseSysDepartmentEntity.find({ select: ['id', 'parentId', 'name'] }),
      this.basicdataSiteEntity.find({ select: ['id', 'departmentId'], where: { siteType: '收集点' } }),
      this.reportDailysiteEntity.findBy({ date: targetDate })
    ]);

    if (dailySites.length === 0) {
      // 如果当天没有任何站点产生流水，顺应“没数不存”原则，部门表也直接不存任何数据，完美收工
      return;
    }

    // 2. 建立 站点ID -> 部门ID 的映射快查表
    const siteToDeptMap = new Map<number, number>();
    allSites.forEach(s => {
      if (s.departmentId) siteToDeptMap.set(s.id, s.departmentId);
    });

    // 3. 构建部门树的关系链（用于无级向上追溯）
    const deptParentMap = new Map<number, number | null>();
    const deptNameMap = new Map<number, string>();
    allDepts.forEach(d => {
      deptParentMap.set(d.id, d.parentId || null);
      deptNameMap.set(d.id, d.name);
    });

    // 4. 定义需要累加的动态分类字段清单（必须和站点表算出来的50多个字段完全对齐）
    const typeSuffixes = [
      'Sw60', 'Sw61', 'Sw62', 'Sw63', 'Sw64',
      'Sw62001', 'Sw62002', 'Sw62003', 'Sw62004', 'Sw62005', 'Sw62006', 'Sw62007'
    ];
   

    // 5. 声明部门数据的内存聚合容器
    const deptAggMap = new Map<number, any>();

    // 初始化一个干净部门对象的辅助函数
    const initDeptData = (deptId: number) => ({
      date: targetDate,
      departmentId: deptId,
      departmentName: deptNameMap.get(deptId) || `未知部门(${deptId})`,
      weightTotal: 0,
      timeTotal: 0,
      carbonTotal: 0,
      estimatedSalesSw62: 0,
      estimatedSalesSw62001: 0,
      estimatedSalesSw62002: 0,
      estimatedSalesSw62003: 0,
      estimatedSalesSw62004: 0,
      estimatedSalesSw62005: 0,
      estimatedSalesSw62006: 0,
      estimatedSalesSw62007: 0,
      ...typeSuffixes.reduce((acc, suffix) => {
        acc[`weight${suffix}`] = 0;
        acc[`time${suffix}`] = 0;
        acc[`carbon${suffix}`] = 0;
        return acc;
      }, {})
    });

    // 6. 核心无级轰炸算法：遍历每一个有数据的站点，向它的所有上级父部门“冒泡”贡献数据
    for (const siteRow of dailySites) {
      const directDeptId = siteToDeptMap.get(siteRow.siteId);
      if (!directDeptId) continue; // 如果站点脱离了部门组织架构，跳过

      // 从当前直属部门开始，沿着 parentId 一路往上爬，直到爬到祖先根节点（parentId 为 null）
      let currentDeptId: number | null | undefined = directDeptId;
      
      while (currentDeptId !== null && currentDeptId !== undefined) {
        // 如果树中没有这个部门ID（异常数据防崩），中断退出
        if (!deptParentMap.has(currentDeptId)) break;

        // 如果内存中还没有这个部门的聚合空间，立即为其独立初始化
        if (!deptAggMap.has(currentDeptId)) {
          deptAggMap.set(currentDeptId, initDeptData(currentDeptId));
        }

        const deptAgg = deptAggMap.get(currentDeptId);

        // 核心轰炸：将该站点的各项数值，叠加到当前部门的所有对应指标上
        deptAgg.weightTotal += Number(siteRow.weightTotal) || 0;
        deptAgg.timeTotal += Number(siteRow.timeTotal) || 0;
        deptAgg.carbonTotal += Number(siteRow.carbonTotal) || 0;

        // 动态轰炸 50 多个子类字段
        typeSuffixes.forEach(suffix => {
          deptAgg[`weight${suffix}`] += Number(siteRow[`weight${suffix}`]) || 0;
          deptAgg[`time${suffix}`] += Number(siteRow[`time${suffix}`]) || 0;
          deptAgg[`carbon${suffix}`] += Number(siteRow[`carbon${suffix}`]) || 0;

          // 回收物子类的估算金额同步叠加
          if (suffix.startsWith('Sw62')) {
            deptAgg[`estimatedSales${suffix}`] += Number(siteRow[`estimatedSales${suffix}`]) || 0;
          }
        });

        // 【关键跃迁】：将指针移向它的亲生父节点，实现无级递归向上冒泡！
        currentDeptId = deptParentMap.get(currentDeptId);
      }
    }

    // 7. 将内存中所有被“轰炸”过、产生了有效数据的部门记录提取出来
    const finalDeptRows = Array.from(deptAggMap.values());

    if (finalDeptRows.length > 0) {
            // 1. 动态抓取所有的键，将主键、联合唯一键彻底从“待更新字段”中排除
            // 这样能确保 ON DUPLICATE KEY UPDATE 后面的赋值语句绝对干净
            const allUpdateFields = Object.keys(finalDeptRows[0]).filter(
                k => k !== 'date' && k !== 'departmentId' && k !== 'id'
            );

            // 2. 依然推荐采用 Chunk 分批，这是应对海量数据/多子字段最安全、执行效率最高的做法
            const chunkSize = 50;
            for (let i = 0; i < finalDeptRows.length; i += chunkSize) {
                const chunk = finalDeptRows.slice(i, i + chunkSize);

                await this.reportDailyDepartmentEntity.createQueryBuilder()
                    .insert()
                    .values(chunk)
                    // 💡 显式声明：当 ['date', 'departmentId'] 发生冲突时，强行把 allUpdateFields 里的字段全部更新一遍
                    .orUpdate(allUpdateFields, ['date', 'departmentId'])
                    .execute();
            }
        }

    return finalDeptRows.length;
  }
}

 src/modules/report/service/taskservice.ts

New file
@@ -0,0 +1,174 @@
import { Provide, Inject } from '@midwayjs/core';
import { BaseService } from '@cool-midway/core';
import { InjectEntityModel } from '@midwayjs/typeorm';
import { Repository, In } from 'typeorm';
import { ReportJoblogEntity } from '../entity/joblog';
import { TaskSiteService } from './tasksiteservice'
import { TaskDepartmentService } from './taskdepartmentservice'

/**
 *  定时任务-汇总站点/各个单位的各类垃圾汇总数据
 *  总入口，创建和调度任务，具体执行调用TaskSiteService 和  TaskDepartmentService
 */


@Provide()
export class ReportTaskService extends BaseService {
  @InjectEntityModel(ReportJoblogEntity)
  reportJoblogEntity: Repository<ReportJoblogEntity>;

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

  @Inject()
  departmentCubeService: TaskDepartmentService;

  /**
   * 外部定时任务（Crontab）每小时调用的总入口
   * 建议表达式: 0 5 * * * * (每小时的第5分钟执行)
   */
  async clockIn() {
    // 1. 自动看门狗：确保昨天和今天的任务实例已经在表里初始化了
    await this.ensureJobInstancesCreated();

    // 2. 执行引擎：捞出需要执行或重试的任务
    await this.executePendingJobs();

    return '任务执行成功:ReportTaskService';
  }

  /**
   * 检查并初始化任务实例（守护动作）
   */
  private async ensureJobInstancesCreated() {
    const now = new Date();
    const year = now.getFullYear();
    const month = String(now.getMonth() + 1).padStart(2, '0');
    const day = String(now.getDate()).padStart(2, '0');
    const yesterdayDate = new Date(now.getTime() - 24 * 60 * 60 * 1000);
    const yYear = yesterdayDate.getFullYear();
    const yMonth = String(yesterdayDate.getMonth() + 1).padStart(2, '0');
    const yDay = String(yesterdayDate.getDate()).padStart(2, '0');

    const todayStr = `${year}-${month}-${day}`;
    const yesterdayStr = `${yYear}-${yMonth}-${yDay}`;

    // 我们需要守护的目标日期和任务类型组合
    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},
    ];

    for (const target of targets) {
      try {
        // 利用 try-catch 和数据库唯一索引，防并发同时也能实现“不存在则创建”
        const exist = await this.reportJoblogEntity.findOneBy({
          jobName: target.type,
          dataDate: target.date,
        });
        if (!exist) {
          const job = new ReportJoblogEntity();
          job.jobName = target.type;
          job.dataDate = target.date;
          job.status = 'INIT';
          await this.reportJoblogEntity.save(job);
        }else if (target.isToday && exist.status === 'SUCCESS') {
          // 2. 【核心修复】如果是今天的任务，且已经是 SUCCESS 状态，强行洗回 INIT 参与本小时的重算
          await this.reportJoblogEntity.update(exist.id, {
            status: 'INIT'
          });
        }
      } catch (err) {
        // 并发写入时可能触发唯一索引冲突，直接忽略即可
      }
    }
  }

  /**
   * 捞出 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',
        });
      }
    }
  }
}

 src/modules/report/service/tasksiteservice.ts

New file
@@ -0,0 +1,154 @@
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', // 永远拿当前后台最新的站点名称，防止设备传旧名称
                '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,
                    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;
    }
}