datatools/csrSync/job.go

package main

import (
	util "app.yhyue.com/moapp/jybase/common"
	"fmt"
	monitor "github.com/tencentcloud/tencentcloud-sdk-go/tencentcloud/asr/v20190614"
	"github.com/tencentcloud/tencentcloud-sdk-go/tencentcloud/common"
	"github.com/tencentcloud/tencentcloud-sdk-go/tencentcloud/common/errors"
	"github.com/tencentcloud/tencentcloud-sdk-go/tencentcloud/common/profile"
	"log"
	"time"
)

func job() {
	log.Println("语音识别定时任务开始")
	credential := common.NewCredential(
		"keyid",
		"key",
	)
	cpf := profile.NewClientProfile()
	cpf.HttpProfile.Endpoint = "monitor.tencentcloudapi.com"
	client, _ = monitor.NewClient(credential, "ap-shanghai", cpf)
	sql := `SELECT id,MonitorFilename FROM voice_record WHERE %s MonitorFilename <> '' AND  ISNULL(callText) ORDER  BY createTime DESC `
	addSql := ``
	if cfg.VoiceRecordId > 0 {
		addSql = fmt.Sprintf("id > %d  AND ", cfg.VoiceRecordId)
	}
	sql = fmt.Sprintf(sql, addSql)
	TiDb.SelectByBath(cfg.Bath, func(l *[]map[string]interface{}) bool {
		if len(*l) > 0 {
			cfg.VoiceRecordId = util.Int64All((*l)[0]["id"])
			log.Println(fmt.Sprintf("此次处理数据量：%d,最大id：%d", len(*l), cfg.VoiceRecordId))
			for _, v := range *l {
				ch <- true
				go FormatData(v)
			}
		}
		return true
	}, sql) // 需要加where进行增量
	log.Println("语音识别定时任务结束,id:", cfg.VoiceRecordId)
}

// FormatData 获取TaskId 并获取录音识别结果
func FormatData(data map[string]interface{}) {
	defer util.Catch()
	request := monitor.NewCreateRecTaskRequest()
	request.EngineModelType = common.StringPtr("8k_zh")
	request.ChannelNum = common.Uint64Ptr(1)
	request.SourceType = common.Uint64Ptr(0)
	request.ResTextFormat = common.Uint64Ptr(0)
	request.Url = common.StringPtr(util.ObjToString(data["MonitorFilename"]))
	response, err := client.CreateRecTask(request)
	if _, ok := err.(*errors.TencentCloudSDKError); ok {
		<-ch
		fmt.Printf("An API error has returned: %s", err)
		return
	}
	if err != nil {
		<-ch
		panic(err)
		return
	}
	var (
		recordId = util.Int64All(data["id"])
		pollTime = map[int64]int{}
		//获取 response.Response.Data.TaskId
		//查询第二个接口： 查询任务结果  需要轮询
		taskId   = response.Response.Data.TaskId
		requests = monitor.NewDescribeTaskStatusRequest()
	)
	requests.TaskId = taskId
L:
	for {
		if pollTime[recordId] > cfg.PollCount {
			//轮询大于pollCount，则跳出，防止死循环
			break L
		}
		pollTime[recordId] += 1
		r, errs := client.DescribeTaskStatus(requests)
		log.Println(r, errs)
		if errs != nil {
			log.Println("根据taskId 请求语音识别结果异常：", errs.Error())
			break L
		}
		switch *r.Response.Data.StatusStr {
		case "success":
			if updateVoiceRecord(*r.Response.Data.Result, recordId) {
				break L
			}
		case "waiting":
			time.Sleep(time.Duration(cfg.WaitingTime) * time.Second)
			continue L
		case "doing":
			time.Sleep(time.Duration(cfg.DoingTime) * time.Second)
			continue L
		case "failed":
			if updateVoiceRecord(*r.Response.Data.ErrorMsg, recordId) {
				break L
			}
		}
	}
	<-ch
}

func updateVoiceRecord(content string, id int64) bool {
	return TiDb.UpdateOrDeleteBySql(`UPDATE  voice_record  SET callText  = ? WHERE  id  = ?`, content, id) > 0
}