适配信息箱

.config/.dev.yaml

@@ -1,8 +1,9 @@
 TEST_VALUE:
   name: cookie
 # 数据库相关
-DATABASE: drive
+DRIVE_DB: drive
+SSH_DB: ssh
 USER: root
-PORT: 3307
-HOST: 127.0.0.1
-PASSWORD: '123456'
+PORT: 3306
+HOST: 192.168.10.62
+PASSWORD: 'zhonglian405'

.config/.prod.yaml

@@ -1,8 +1,9 @@
 TEST_VALUE:
   name: cookie
 # 数据库相关
-DATABASE: drive
+DRIVE_DB: drive
+SSH_DB: ssh
 USER: root
-PORT: 3306
+PORT: 3307
 HOST: 127.0.0.1
-PASSWORD: zhonglian405
+PASSWORD: '123456'

prod.yaml

@@ -1,8 +1,9 @@
 TEST_VALUE:
   name: cookie
 # 数据库相关
-DATABASE: drive
+DRIVE_DB: drive
+SSH_DB: ssh
 USER: root
 PORT: 3306
 HOST: 127.0.0.1
-PASSWORD: '123456'
+PASSWORD: zhonglian405

src/app.module.ts

@@ -4,7 +4,7 @@ import { AppService } from './app.service';
 import { ClientModule } from './client/client.module';
 import { GatewayModule } from './gateway/gateway.module';
 import { TypeOrmModule, TypeOrmModuleAsyncOptions } from '@nestjs/typeorm';
-import { getMysql } from './utils/config';
+import { getDriveDb, getSshDb } from './utils/config';
 // import 'winston-daily-rotate-file';
 
 @Module({
@@ -13,7 +13,18 @@ import { getMysql } from './utils/config';
     TypeOrmModule.forRootAsync({
       useFactory: async () => {
         return {
-          ...getMysql(), entities: [__dirname + '/**/*.entity{.ts,.js}'], // 扫描本项目中.entity.ts或者.entity.js的文件
+          ...getDriveDb(),
+          entities: [__dirname + '/**/*.entity{.ts,.js}'], // 扫描本项目中.entity.ts或者.entity.js的文件
+          synchronize: false,
+          // logging: true,
+        } as TypeOrmModuleAsyncOptions;
+      },
+    }),
+    TypeOrmModule.forRootAsync({
+      useFactory: async () => {
+        return {
+          ...getSshDb(),
+          entities: [__dirname + '/**/*.entity{.ts,.js}'], // 扫描本项目中.entity.ts或者.entity.js的文件
           synchronize: false,
           // logging: true,
         } as TypeOrmModuleAsyncOptions;

src/client/client.controller.ts

@@ -9,7 +9,7 @@ import {
 } from '@nestjs/websockets';
 import { Server, Socket } from 'socket.io';
 import { EventGateway } from 'src/gateway/gateway';
-import {BusinessException} from "../common/exceptions/business.exception";
+import { BusinessException } from "../common/exceptions/business.exception";
 import AppLogger from "../utils/logger";
 
 @Controller()
@@ -28,10 +28,10 @@ export class ClientController {
    */
   @Post('/api/v1/szjt/hcc/queryNum')
   queryHccNum(@Body() { areaCode }) {
-    if (areaCode && (areaCode.length!=12)){
+    if (areaCode && (areaCode.length != 12)) {
       this.logger.error(null, '输入参数不符合标准12位码');
       throw BusinessException.throwValidateError('输入参数不符合标准12位码')
-    }else {
+    } else {
       return this.clientService.queryHccNumV2({ areaCode });
     }
   }
@@ -46,7 +46,7 @@ export class ClientController {
    */
   @Post('/api/v1/szjt/user/queryNum')
   queryUserNum(@Body() { areaCode }) {
-    if (areaCode && (areaCode.length!=12)){
+    if (areaCode && (areaCode.length != 12)) {
       this.logger.error(null, '输入参数不符合标准12位码');
       throw BusinessException.throwValidateError('输入参数不符合标准12位码')
     }
@@ -62,7 +62,7 @@ export class ClientController {
    */
   @Post('/api/v1/szjt/mp/queryNum')
   queryMpNum(@Body() { topLimit, areaCode, type }) {
-    if (areaCode && (areaCode.length!=12)){
+    if (areaCode && (areaCode.length != 12)) {
       this.logger.error(null, '输入参数不符合标准12位码');
       throw BusinessException.throwValidateError('输入参数不符合标准12位码')
     }
@@ -108,7 +108,8 @@ export class ClientController {
    */
   @Post('/api/v1/szjt/user/queryWarnList')
   queryWarnList(@Body() { code, type, time, page }) {
-    return this.clientService.queryWarnList({ code, type, time, page });
+    // return this.clientService.queryWarnList({ code, type, time, page });
+    return this.clientService.queryWarnListV2({ code, type, time, page });
   }
 
   /**
@@ -125,7 +126,9 @@ export class ClientController {
 
   @Post('/mp/api/v1/callback')
   @SubscribeMessage('new')
-  async callBack(@Body() { payload }, @ConnectedSocket() client: Socket) {
+  async callBack(@Body() { payload },
+    // @ConnectedSocket() client: Socket
+  ) {
     const data = await this.clientService.callBack({
       payload,
     });

src/client/client.mapper.ts

@@ -3,15 +3,20 @@ import { UserNum } from '../entities/user-num.entity';
 import { UserInfo } from 'src/entities/user-info.entity';
 import { MpInfo } from 'src/entities/mp-info.entity';
 import { UserArchieve } from 'src/entities/user-archieve.entity';
-import { Between, Like } from 'typeorm';
+import { Between, getConnection, Like, Repository } from 'typeorm';
 import { UserHealth } from 'src/entities/user-health.entity';
 import { HomeInfo } from 'src/entities/home-linfo.entity';
 import { AlarmInfo } from 'src/entities/alarm_info.entity';
+import { HccAlarmInfo } from 'src/entities/hcc_alarm_info.entity';
 const moment = require('moment');
 moment.locale('zh_CN');
 
 @Injectable()
 class ClientMapper {
+  // constructor(
+  //   @InjectConnection('drive') private readonly drive: Connection,
+  //   @InjectConnection('ssh') private readonly ssh: Connection,
+  // ) { }
   /**
    * 获取家庭统计信息
    */
@@ -433,6 +438,30 @@ class ClientMapper {
     });
   }
 
+  async queryWarnListV2({ code, startTime, endTime, page }) {
+    const obj: any = {}
+    const { pageSize, pageIndex } = page
+    obj.createTime = Between(startTime, endTime);
+    // const r = getConnection('ssh').manager.find(HccAlarmInfo, {
+    //   where: obj,
+    //   select: ['content', 'createTime', 'msgType'],
+    //   take: pageSize,
+    //   skip: pageIndex * pageSize,
+    // })
+    // console.log(r);
+
+    const [data, count] = await HccAlarmInfo.findAndCount({
+      where: obj,
+      select: ['content', 'createTime', 'msgType'],
+      take: Number(pageSize),
+      skip: Number(pageIndex - 1) * Number(pageSize),
+    })
+    return {
+      data,
+      count
+    }
+  }
+
   async queryDataUseNum({ areaCode, startTime, endTime, limit, dateRange }) {
     let sql;
     if (areaCode) {

src/client/client.service.ts

@@ -475,6 +475,39 @@ export class ClientService {
     return info;
   }
 
+  async queryWarnListV2({ code, type, time, page }) {
+    let startTime, endTime
+    if (type == 2) {
+      // 当日
+      startTime = moment().format('YYYY-MM-DD 00:00:00')
+      endTime = moment().format('YYYY-MM-DD 23:59:59')
+    } else {
+      // 当月
+      startTime = moment().format('YYYY-MM-01')
+      endTime = moment().day(31).format('YYYY-MM-DD 23:59:59')
+    }
+    const { data, count } = await clientMapper.queryWarnListV2({ code, startTime, endTime, page });
+    data.forEach((e: any) => {
+      e.homeCode = e?.addressCode || ''
+      e.address = e?.address || ''
+      e.typeName = e?.typeName || ''
+      e.alarmDate = moment(e.createTime).format('YYYY-MM-DD hh:mm:ss');
+      delete e.createTime
+    })
+    let result: any = {}
+    const { pageSize, pageIndex } = page
+    result = {
+      alarmList: data,
+      page: {
+        pageIndex,
+        pageSize,
+        totalNum: count,
+        totalPage: Math.ceil(count / pageSize)
+      }
+    }
+    return result;
+  }
+
   /**
    * 4.9获取平台家庭数据使用记录
    */

src/entities/alarm_info.entity.ts

@@ -3,7 +3,7 @@ import { Entity, PrimaryColumn, Column, BaseEntity } from 'typeorm';
 /**
  * 用户档案列表
  */
-@Entity('alarm_info')
+@Entity({ name: 'alarm_info', database: 'drive' })
 export class AlarmInfo extends BaseEntity {
   //id
   @PrimaryColumn({ type: 'bigint' })

src/entities/hcc_alarm_info.entity.ts

@@ -0,0 +1,28 @@
+import { Entity, PrimaryColumn, Column, BaseEntity } from 'typeorm';
+
+/**
+ * 用户档案列表
+ */
+@Entity({ name: 'ssh_msg_push_record', database: 'ssh' })
+export class HccAlarmInfo extends BaseEntity {
+  //id
+  @PrimaryColumn({ type: 'bigint' })
+  id: number;
+
+  //标题
+  @Column({ type: 'varchar' })
+  title: string;
+
+  //内容
+  @Column({ type: 'varchar' })
+  content: string;
+
+  //时间
+  @Column({ name: 'create_time', type: 'varchar' })
+  createTime: string;
+
+  //类型
+  @Column({ name: 'msg_type', type: 'int' })
+  msgType: number;
+
+}

src/entities/hcc_info.ts

@@ -0,0 +1,16 @@
+import { Entity, PrimaryColumn, Column, BaseEntity } from 'typeorm';
+
+/**
+ * 用户档案列表
+ */
+@Entity({ name: 'ssh_hcc_info', database: 'ssh' })
+export class HccInfo extends BaseEntity {
+  //id
+  @PrimaryColumn({ type: 'varchar' })
+  id: string;
+
+  //标题
+  @Column({ type: 'varchar', name: 'address_detail' })
+  addressDetail: string;
+
+}

src/entities/home-linfo.entity.ts

@@ -3,7 +3,7 @@ import { Entity, PrimaryColumn, Column, BaseEntity } from 'typeorm';
 /**
  * 家庭数据使用记录表
  */
-@Entity('home_info')
+@Entity({ name: 'home_info', database: 'drive' })
 export class HomeInfo extends BaseEntity {
   //id
   @PrimaryColumn({ type: 'bigint' })

src/entities/mp-info.entity.ts

@@ -3,7 +3,7 @@ import { Entity, PrimaryColumn, Column, BaseEntity } from 'typeorm';
 /**
  * 应用服务统计信息表
  */
-@Entity('mp_info')
+@Entity({ name: 'mp_info', database: 'drive' })
 export class MpInfo extends BaseEntity {
   //地区编码
   @PrimaryColumn({ type: 'varchar', name: 'areaCode' })

src/entities/user-archieve.entity.ts

@@ -3,7 +3,7 @@ import { Entity, PrimaryColumn, Column, BaseEntity } from 'typeorm';
 /**
  * 用户档案列表
  */
-@Entity('user_archieve')
+@Entity({ name: 'user_archieve', database: 'drive' })
 export class UserArchieve extends BaseEntity {
   //id
   @PrimaryColumn({ type: 'bigint' })

src/entities/user-health.entity.ts

@@ -3,7 +3,7 @@ import { Entity, PrimaryColumn, Column, BaseEntity } from 'typeorm';
 /**
  * 用户健康档案列表
  */
-@Entity('user_health')
+@Entity({ name: 'user_health', database: 'drive' })
 export class UserHealth extends BaseEntity {
   //id
   @PrimaryColumn({ type: 'bigint' })

src/entities/user-info.entity.ts

@@ -3,7 +3,7 @@ import { Entity, PrimaryColumn, Column, BaseEntity } from 'typeorm';
 /**
  * 用户统计信息表
  */
-@Entity('user-info')
+@Entity({ name: 'user_info', database: 'drive' })
 export class UserInfo extends BaseEntity {
   @PrimaryColumn({ type: 'varchar', comment: '地区编码' })
   areaCode: string;

src/entities/user-num.entity.ts

@@ -3,7 +3,7 @@ import { Entity, PrimaryColumn, Column, BaseEntity } from 'typeorm';
 /**
  * 家庭信息在线表
  */
-@Entity('user_num')
+@Entity({ name: 'user_num', database: 'drive' })
 export class UserNum extends BaseEntity {
   @PrimaryColumn({ type: 'varchar', comment: '地区编码' })
   areaCode: string;

src/main.ts

@@ -12,6 +12,7 @@ import { AllExceptionsFilter } from './common/exceptions/base.exception.filter';
 import { HttpExceptionFilter } from './common/exceptions/http.exception.filter';
 import { NestExpressApplication } from '@nestjs/platform-express';
 import { join } from 'path';
+import { Mqtt } from './mqtt/mqtt';
 
 async function bootstrap() {
   if (module.hot) {
@@ -36,6 +37,7 @@ async function bootstrap() {
   app.useGlobalPipes(new ValidationPipe());
 
   app.useStaticAssets(join(__dirname, '..', 'public'));
+  Mqtt.init()
   await app.listen(7666, '0.0.0.0');
 }
 bootstrap();

src/mqtt/mqtt.ts

@@ -0,0 +1,71 @@
+import * as mqtt from 'mqtt';
+import MqttTool from './mqttTool'
+import AppLogger from '../utils/logger';
+import { EventGateway } from 'src/gateway/gateway';
+import { HccInfo } from 'src/entities/hcc_info';
+const logger = new AppLogger();
+
+export class Mqtt {
+  protected static client: mqtt.MqttClient;
+
+  static async init() {
+    const {
+      clientId,
+      userName,
+      password,
+      securityKey,
+      mqttUrl
+    } = await new MqttTool().getMqttInfo()
+    console.log(`clientId:${clientId},mqttUrl:${mqttUrl}`);
+    if (!clientId) {
+      return;
+    }
+
+    const options = {
+      clientId,
+      username: userName,
+      password,
+      clean: true, // true: 清除会话, false: 保留会话
+      connectTimeout: 60 * 60, //超时时间
+    };
+
+    this.client = mqtt.connect(mqttUrl, options);
+
+    // 订阅主题
+    this.client.subscribe(['rep/hcc/#/msgPush/report']);
+
+    this.client.on('connect', async () => {
+      //初始化日志数据
+      // logger.info(null, `成功连接到服务器[${getConfig().MQTT_URL}]`);
+      console.log(`成功连接到服务器[${mqttUrl}]`);
+    });
+
+    this.client.on('reconnect', (err) => {
+      // logger.info(null, 'mqtt 正在重连:', err);
+      console.log('mqtt 正在重连');
+    });
+
+    this.client.on('message', async (topic, mes) => {
+      const arr = topic.split('/');
+      logger.info(null, `订阅到topic: ${topic}`);
+      const data = JSON.parse(mes.toString())
+      logger.info(null, `data:${JSON.stringify(data)}`);
+      // 获取监听到的数据
+      const { pushFlag, pushId, msgType, title, content } = data
+      // 查询当前信息箱所在地址
+      const list = await HccInfo.find()
+      const address = list[0].addressDetail
+      new EventGateway().message({ message: `${address}${content}` })
+    });
+
+    this.client.on('error', (err) => {
+      logger.info(null, `mqtt连接失败,${err}`);
+      console.log('mqtt连接失败');
+    });
+
+    this.client.on('close', async () => {
+      logger.info(null, '已断开');
+      console.log('已断开');
+    });
+  }
+}

src/mqtt/mqttTool.ts

@@ -0,0 +1,16 @@
+import axios from 'axios';
+
+export default class MqttTool {
+  // 先请求接口获取mqtt连接地址
+  async getMqttInfo() {
+    try {
+      const result = await axios.post('http:localhost:8000/iot/hcc/getConnInfo')
+      const connInfo = result.data.connInfo;
+      return connInfo;
+    } catch (error) {
+      console.log('请求失败');
+      return {}
+    }
+  }
+
+}

src/utils/config.ts

@@ -1,9 +1,9 @@
 // /* eslint-disable @typescript-eslint/no-var-requires */
 import { parse } from 'yaml';
-// import path from 'path';
-// import fs from 'fs';
 const path = require('path');
 const fs = require('fs');
+import { TypeOrmModuleOptions, TypeOrmOptionsFactory } from '@nestjs/typeorm';
+import { Injectable } from '@nestjs/common';
 
 // 获取项目运行环境
 export const getEnv = () => {
@@ -21,29 +21,45 @@ export const getConfig = () => {
   const yamlPath = path.join(process.cwd(), `./${environment}.yaml`);
   const file = fs.readFileSync(yamlPath, 'utf8');
   let config = parse(file);
+  console.log('读取配置', JSON.stringify(config))
   // 融合注入参数
   config = Object.assign(config, process.env);
   configAll = { ...config, isExist: true }
   return config;
 };
 
-export const getMysql = () => {
+export const getDriveDb = () => {
   const mysql = {
+    name: 'drive',
     type: 'mysql',
     host: getConfig().HOST,
     port: getConfig().PORT,
     username: getConfig().USER,
     password: getConfig().PASSWORD,
-    database: getConfig().DATABASE,
-    // type: 'mysql',
-    // host: '192.168.10.13',
-    // port: 3306,
+    database: getConfig().DRIVE_DB,
+    // host: '127.0.0.1',
+    // port: '3307',
     // username: 'root',
     // password: '123456',
     // database: 'drive',
-    // entities: [__dirname + '/**/*.entity{.ts,.js}'], // 扫描本项目中.entity.ts或者.entity.js的文件
-    // synchronize: false,
-    // logging: true,
+  }
+  return mysql
+}
+
+export const getSshDb = () => {
+  const mysql = {
+    name: 'ssh',
+    type: 'mysql',
+    host: getConfig().HOST,
+    port: getConfig().PORT,
+    username: getConfig().USER,
+    password: getConfig().PASSWORD,
+    database: getConfig().SSH_DB,
+    // host: '127.0.0.1',
+    // port: '3307',
+    // username: 'root',
+    // password: '123456',
+    // database: 'ssh',
   }
   return mysql
 }