add

wangzhibo

2026-07-17 6e210cdaada04907dd127ce6db5a67f2e6e11f58

 src/modules/push/service/open.ts

@@ -7,6 +7,7 @@
import { PushAuditlogEntity } from '../entity/auditlog';
import { PushCameraGpsEntity } from '../entity/cameragps';
import { PushVehicleGpsEntity } from '../entity/vehiclegps';
import { PushDeviceOnoffEntity } from '../entity/deviceonoff';
import { CachingFactory, MidwayCache } from '@midwayjs/cache-manager';

/**
@@ -25,6 +26,10 @@

  @InjectEntityModel(PushVehicleGpsEntity)
  pushVehicleGpsEntity: Repository<PushVehicleGpsEntity>;

  @InjectEntityModel(PushDeviceOnoffEntity)
  pushDeviceOnoffEntity: Repository<PushDeviceOnoffEntity>;


  @InjectEntityModel(PushAuditlogEntity)
  pushAuditlogEntity: Repository<PushAuditlogEntity>;
@@ -76,34 +81,105 @@
      } else {
        siteWeight.businessId = 'UNKNOWN';
        siteWeight.departmentId = -1;
        await this.writePushAuditLog("weight",{ "deviceNo": data.deviceNo, "requestContent": data, "errorMessage": "电子秤还未分配到垃圾收集点." });
        await this.writePushAuditLog("weight", { "deviceNo": data.deviceNo, "requestContent": data, "errorMessage": "电子秤还未分配到垃圾收集点." });
      }

    } catch (err) {
      await this.writePushAuditLog("weight",{ "deviceNo": data.deviceNo || "UNKNOW", "requestContent": data, "errorMessage": err });
      await this.writePushAuditLog("weight", { "deviceNo": data.deviceNo || "UNKNOW", "requestContent": data, "errorMessage": err });
    }

  }

  /**
   * 接收GPS消息并缓存
   * 接收萤石云 消息并缓存
   */
  async receiveCameraGpsMessage(body: any) {
    const messageId = body.msgSeq || Math.random().toString(36).substring(2);
    await this.midwayCache.set(`push:cameragps:queue:${messageId}`, JSON.stringify(body), 24 * 3600);
    this.processCameraGpsAsync(messageId).catch(err => {
  async saveYingshiyunMessage(body: any) {
    const messageId = body?.header?.messageId || Math.random().toString(36).substring(2);
    await this.midwayCache.set(`push:yingshiyun:queue:${messageId}`, JSON.stringify(body), 24 * 3600);
    this.processYingshiyunAsync(messageId).catch(err => {
      console.error('异步处理GPS消息失败:', err);
    });
    return  messageId ;
    return messageId;
  }

  /**
   * 异步解析车载监控视频的位置数据并入库
   * 异步解析萤石云消息，按照消息类型 入库
   */
  async processCameraGpsAsync(messageId: string) {
    const dataStr = await this.midwayCache.get(`push:cameragps:queue:${messageId}`) as string | null;
  async processYingshiyunAsync(messageId: string) {
    const dataStr = await this.midwayCache.get(`push:yingshiyun:queue:${messageId}`) as string | null;
    if (!dataStr) return;
    const body = JSON.parse(dataStr);

    const type = body?.header?.type;

    try {
      switch (type) {

        // 设备上下线
        case 'ys.onoffline': {
          await this.pushDeviceOnoffEntity.save({
            deviceId:body.header.deviceId,
            channelNo: body.header.channelNo,
            messageId: body.header.messageId,
            messageTime: new Date(body.header.messageTime),
            msgType: body.body.msgType,
            devType: body.body.devType,
            deviceName: body.body.deviceName,
            subSerial: body.body.subSerial,
            dasId: body.body.dasId,
            natIp: body.body.natIp,
            occurTime: body.body.occurTime,
            regTime: body.body.regTime
          });
          break;
        }

        // 萤石告警
        /*
        case 'ys.alarm': {
          await this.pushAlarmRepo.save({
            messageId,
            raw: JSON.stringify(body),
            createTime: new Date(),
          });
          break;
        }
          */

        // 海康 ISAPI
        //case 'ys.open.isapi': {
          //processCameraGpsAsync(body)
          //await this.pushIsapiRepo.save({
          //  messageId,
          //  raw: JSON.stringify(body),
          //  createTime: new Date(),
          //});
          //break;
        //}

        default:
          console.warn('Unknown yingshi webhook type:', type);
      }
    } catch (err) {
      await this.writePushAuditLog(type || 'unknown', {
        deviceNo: body?.sn || 'UNKNOWN',
        requestContent: body,
        errorMessage: err,
      });
    }

    // 不管成功与否，删除缓存中的任务队列
    await this.midwayCache.del(`push:yingshiyun:queue:${messageId}`);

}



  /**
   * 异步解析车载监控的GPS数据，存储并添加业务表
   */
  async processCameraGpsAsync(body: any) {


    try {
      let vehicleNo = '';
@@ -155,21 +231,20 @@
          });

        } else {
          await this.writePushAuditLog("camera_gps",{ "deviceNo": body.sn, "requestContent": body, "errorMessage": "车载监控还未分配到车辆" });
          await this.writePushAuditLog("camera_gps", { "deviceNo": body.sn, "requestContent": body, "errorMessage": "车载监控还未分配到车辆" });
        }
      } else {
        await this.writePushAuditLog("camera_gps",{ "deviceNo": "UNKOWN", "requestContent": body, "errorMessage": "未包含设备号SN" });
        await this.writePushAuditLog("camera_gps", { "deviceNo": "UNKOWN", "requestContent": body, "errorMessage": "未包含设备号SN" });
      }

      await this.midwayCache.del(`push:gps:queue:${messageId}`);

    } catch (err: any) {
      await this.writePushAuditLog("camera_gps",{ "deviceNo": body.sn || "UNKNOW", "requestContent": body, "errorMessage": err });

      await this.writePushAuditLog("camera_gps", { "deviceNo": body.sn || "UNKNOW", "requestContent": body, "errorMessage": err });

    }
  }

  async writePushAuditLog(pushType:string,data: any) {
  async writePushAuditLog(pushType: string, data: any) {
    let isSuccess = true;
    let errorMsg = '';
    try {