speechAnalysis/service/process.go

package service

import (
	"bytes"
	"context"
	"encoding/json"
	"errors"
	"fmt"
	"github.com/fsnotify/fsnotify"
	"gorm.io/gorm"
	"io"
	"log"
	"mime/multipart"
	"net/http"
	"os"
	"path/filepath"
	"speechAnalysis/conf"
	"speechAnalysis/constvar"
	"speechAnalysis/models"
	"speechAnalysis/pkg/logx"
	"strings"
	"time"
)

// Response 结构体用于存储响应体的内容
type Response struct {
	Code   int     `json:"code"`
	Msg    string  `json:"msg"`
	Result string  `json:"result"`
	Score  float64 `json:"score"`
}

func AnalysisAudio(filename string, targetURL string) (resp Response, err error) {
	file, err := os.Open(filename)
	if err != nil {
		return
	}
	defer file.Close()

	// 创建一个缓冲区来存储表单数据
	var requestBody bytes.Buffer
	writer := multipart.NewWriter(&requestBody)

	// 创建一个表单字段，用于存储文件
	fileWriter, err := writer.CreateFormFile("audio", filename)
	if err != nil {
		return
	}

	// 将文件内容复制到表单字段中
	_, err = io.Copy(fileWriter, file)
	if err != nil {
		return
	}

	// 关闭表单写入器，以便写入末尾的边界
	writer.Close()

	// 创建POST请求，指定URL和请求体
	request, err := http.NewRequest("POST", targetURL, &requestBody)
	if err != nil {
		return
	}

	// 设置请求头，指定Content-Type为multipart/form-data
	request.Header.Set("Content-Type", writer.FormDataContentType())

	// 发送请求
	client := &http.Client{}
	response, err := client.Do(request)
	if err != nil {
		return
	}
	defer response.Body.Close()

	// 读取响应
	body := &bytes.Buffer{}
	_, err = io.Copy(body, response.Body)
	if err != nil {
		return
	}

	err = json.NewDecoder(body).Decode(&resp)
	if err != nil {
		return
	}

	return
}

func Process(audioId uint) (err error) {
	audio, err := models.NewAudioSearch().SetID(audioId).First()

	if err != nil {
		return errors.New("查找音频失败")
	}

	if audio.AudioStatus != constvar.AudioStatusUploadOk && audio.AudioStatus != constvar.AudioStatusFailed {
		return errors.New("状态不正确")
	}

	err = models.NewAudioSearch().SetID(audioId).UpdateByMap(map[string]interface{}{"audio_status": constvar.AudioStatusProcessing})
	if err != nil {
		return errors.New("DB错误")
	}

	go func() {
		resp, err := AnalysisAudio(audio.FilePath, conf.AanlysisConf.Url)
		if err != nil {
			logx.Errorf("err when AnalysisAudio:%v", err)
			_ = models.NewAudioSearch().SetID(audioId).UpdateByMap(map[string]interface{}{"audio_status": constvar.AudioStatusFailed})
			return
		}
		if resp.Code != 0 {
			logx.Errorf("AnalysisAudio error return:%v", resp)
			_ = models.NewAudioSearch().SetID(audioId).UpdateByMap(map[string]interface{}{"audio_status": constvar.AudioStatusFailed})
			return
		}
		logx.Infof("AnalysisAudio result: %v", resp)
		words := GetWordFromText(resp.Result, audio)

		err = models.WithTransaction(func(db *gorm.DB) error {
			err = models.NewAudioSearch().SetOrm(db).SetID(audioId).UpdateByMap(map[string]interface{}{
				"audio_status": constvar.AudioStatusFinish,
				"score":        resp.Score,
				"tags":         strings.Join(words, ","),
			})
			if err != nil {
				return err
			}
			err = models.NewAudioTextSearch().SetOrm(db).Save(&models.AudioText{
				AudioID:   audio.ID,
				AudioText: resp.Result,
			})
			return err
		})
		if err != nil {
			logx.Infof("AnalysisAudio success but update record failed: %v", err)
			_ = models.NewAudioSearch().SetID(audioId).UpdateByMap(map[string]interface{}{"audio_status": constvar.AudioStatusFailed})
			return
		}
	}()

	return nil
}

func GetWordFromText(text string, audio *models.Audio) (words []string) {
	if audio == nil {
		return nil
	}
	wordRecords, err := models.NewWordSearch().SetLocomotiveNumber(audio.LocomotiveNumber).FindNotTotal()
	if err != nil || len(wordRecords) == 0 {
		return nil
	}
	for _, v := range wordRecords {
		if strings.Contains(text, v.Content) {
			words = append(words, v.Content)
		}
	}
	return words
}

func PreLoad(cxt context.Context) {
	mkdirErr := os.MkdirAll(conf.LocalConf.PreLoadPath, os.ModePerm)
	if mkdirErr != nil {
		logx.Errorf("function os.MkdirAll() err:%v", mkdirErr)
	}
	//文件夹下新增音频文件时触发
	watcher, err := fsnotify.NewWatcher()
	if err != nil {
		log.Fatal(err)
	}
	defer watcher.Close()
	err = watcher.Add(conf.LocalConf.PreLoadPath)
	if err != nil {
		log.Fatal(err)
	}
	for {
		select {
		case <-cxt.Done():
			fmt.Println("preload stop")
		case event, ok := <-watcher.Events:
			if !ok {
				continue
			}
			if event.Op&fsnotify.Create == fsnotify.Create {
				// 判断文件类型是否为.mp3或.wav
				if filepath.Ext(event.Name) == ".mp3" || filepath.Ext(event.Name) == ".wav" {
					// 文件名
					fileName := filepath.Base(event.Name)
					// 文件大小
					bs, _ := os.ReadFile(event.Name)
					size := len(bs)
					//校验文件命名
					arr := strings.Split(fileName, "_")
					if len(arr) != 6 {
						logx.Errorf(fmt.Sprintf("%s:%s", fileName, "文件名称错误"))
						continue
					}
					timeStr := arr[4] + strings.Split(arr[5], ".")[0]
					t, err := time.ParseInLocation("20060102150405", timeStr, time.Local)
					if err != nil {
						logx.Errorf(fmt.Sprintf("%s:%s", fileName, "时间格式不对"))
					}

					//查重
					_, err = models.NewAudioSearch().SetName(fileName).First()
					if err != gorm.ErrRecordNotFound {
						logx.Errorf(fmt.Sprintf("%s:%s", fileName, "重复上传"))
						continue
					}

					//将文件移动到uploads文件夹下
					src := conf.LocalConf.StorePath + "/" + fileName
					err = os.Rename(event.Name, src)
					if err != nil {
						logx.Errorf(fmt.Sprintf("%s:%s", fileName, "移动文件失败"))
						continue
					}

					audio := &models.Audio{
						Name:             fileName,
						Size:             int64(size),
						FilePath:         src,
						AudioStatus:      constvar.AudioStatusUploadOk,
						LocomotiveNumber: arr[0],
						TrainNumber:      arr[1],
						DriverNumber:     arr[2],
						Station:          arr[3],
						OccurrenceAt:     t,
						IsFollowed:       0,
					}

					if err = models.NewAudioSearch().Create(audio); err != nil {
						logx.Errorf(fmt.Sprintf("%s:%s", fileName, "数据库create失败"))
						continue
					}

					go func() {
						var trainInfoNames = []string{arr[0], arr[1], arr[3]} //
						var (
							info   *models.TrainInfo
							err    error
							parent models.TrainInfo
						)
						for i := 0; i < 3; i++ {
							name := trainInfoNames[i]
							class := constvar.Class(i + 1)
							info, err = models.NewTrainInfoSearch().SetName(name).SetClass(class).First()
							if err == gorm.ErrRecordNotFound {
								info = &models.TrainInfo{
									Name:     name,
									Class:    class,
									ParentID: parent.ID,
								}
								_ = models.NewTrainInfoSearch().Create(info)
							}
							parent = *info
						}

					}()
				}
			}
		case err, ok := <-watcher.Errors:
			if !ok {
				logx.Errorf(err.Error())
			}
		}
	}
}