watch preloads file to autoload

yinbangzhong

2024-06-14 777978a2a2ac2232050ba44521f107b85646081b

watch preloads file to autoload

 conf/config.go

@@ -39,7 +39,8 @@
    }

    localConf struct {
        StorePath string // 本地文件存储路径
        StorePath   string // 本地文件存储路径
        PreLoadPath string // 本地文件预加载路径
    }
    Analysis struct {
        Url string // 本地文件存储路径

 conf/config.yaml

@@ -18,5 +18,6 @@
  RotateDays: 5  # 日志文件的最大保留天数
local:
  storePath: uploads
  preLoadPath: preloads
analysis:
  url: http://192.168.20.116:5000/recognition

 go.mod

@@ -5,6 +5,7 @@
require (
    basic.com/aps/nsqclient.git v0.0.0-20230517072415-37491f4a5d25
    github.com/dgrijalva/jwt-go v3.2.0+incompatible
    github.com/fsnotify/fsnotify v1.7.0
    github.com/gin-gonic/gin v1.10.0
    github.com/nsqio/go-nsq v1.1.0
    github.com/spf13/viper v1.18.2
@@ -14,6 +15,7 @@
    go.uber.org/zap v1.27.0
    golang.org/x/crypto v0.23.0
    google.golang.org/grpc v1.63.2
    gopkg.in/fsnotify.v1 v1.4.7
    gopkg.in/natefinch/lumberjack.v2 v2.2.1
    gorm.io/driver/mysql v1.5.6
    gorm.io/gorm v1.25.10
@@ -28,7 +30,6 @@
    github.com/bytedance/sonic/loader v0.1.1 // indirect
    github.com/cloudwego/base64x v0.1.4 // indirect
    github.com/cloudwego/iasm v0.2.0 // indirect
    github.com/fsnotify/fsnotify v1.7.0 // indirect
    github.com/gabriel-vasile/mimetype v1.4.3 // indirect
    github.com/gin-contrib/sse v0.1.0 // indirect
    github.com/go-openapi/jsonpointer v0.19.5 // indirect

 go.sum

@@ -225,6 +225,8 @@
gopkg.in/check.v1 v1.0.0-20180628173108-788fd7840127/go.mod h1:Co6ibVJAznAaIkqp8huTwlJQCZ016jof/cbN4VW5Yz0=
gopkg.in/check.v1 v1.0.0-20200227125254-8fa46927fb4f h1:BLraFXnmrev5lT+xlilqcH8XK9/i0At2xKjWk4p6zsU=
gopkg.in/check.v1 v1.0.0-20200227125254-8fa46927fb4f/go.mod h1:Co6ibVJAznAaIkqp8huTwlJQCZ016jof/cbN4VW5Yz0=
gopkg.in/fsnotify.v1 v1.4.7 h1:xOHLXZwVvI9hhs+cLKq5+I5onOuwQLhQwiu63xxlHs4=
gopkg.in/fsnotify.v1 v1.4.7/go.mod h1:Tz8NjZHkW78fSQdbUxIjBTcgA1z1m8ZHf0WmKUhAMys=
gopkg.in/ini.v1 v1.67.0 h1:Dgnx+6+nfE+IfzjUEISNeydPJh9AXNNsWbGP9KzCsOA=
gopkg.in/ini.v1 v1.67.0/go.mod h1:pNLf8WUiyNEtQjuu5G5vTm06TEv9tsIgeAvK8hOrP4k=
gopkg.in/natefinch/lumberjack.v2 v2.2.1 h1:bBRl1b0OH9s/DuPhuXpNl+VtCaJXFZ5/uEFST95x9zc=

 main.go

@@ -10,6 +10,7 @@
    "speechAnalysis/models"
    "speechAnalysis/pkg/logx"
    "speechAnalysis/router"
    "speechAnalysis/service"
    "syscall"
    "time"
)
@@ -31,6 +32,11 @@
        return
    }

    //监控预加载音频文件
    cxt, cancel := context.WithCancel(context.Background())
    defer cancel()
    go service.PreLoad(cxt)

    logx.Infof("server start serve...")
    server := &http.Server{
        Addr:         ":" + conf.WebConf.Port,

 service/process.go

@@ -2,18 +2,24 @@

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 结构体用于存储响应体的内容
@@ -153,3 +159,112 @@
    }
    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())
            }
        }
    }
}