package jobs import ( "context" "path/filepath" "time" "quyun/app/models" "quyun/database/fields" "quyun/database/schemas/public/model" "quyun/pkg/utils" "quyun/providers/ali" "quyun/providers/app" "quyun/providers/job" . "github.com/riverqueue/river" log "github.com/sirupsen/logrus" _ "go.ipao.vip/atom" "go.ipao.vip/atom/contracts" ) var _ contracts.JobArgs = (*VideoStoreShort)(nil) type VideoStoreShort struct { MediaHash string `json:"media_hash"` FilePath string `json:"file_path"` } func (s VideoStoreShort) InsertOpts() InsertOpts { return InsertOpts{ Queue: QueueDefault, Priority: PriorityDefault, } } func (s VideoStoreShort) Kind() string { return "video_store_short" } func (a VideoStoreShort) UniqueID() string { return a.Kind() } var _ Worker[VideoStoreShort] = (*VideoStoreShortWorker)(nil) // @provider(job) type VideoStoreShortWorker struct { WorkerDefaults[VideoStoreShort] oss *ali.OSSClient job *job.Job app *app.Config } func (w *VideoStoreShortWorker) NextRetry(job *Job[VideoStoreShort]) time.Time { return time.Now().Add(30 * time.Second) } func (w *VideoStoreShortWorker) Work(ctx context.Context, job *Job[VideoStoreShort]) error { log := log.WithField("job", job.Args.Kind()) log.Infof("[Start] Working on job with strings: %+v", job.Args) defer log.Infof("[End] Finished %s", job.Args.Kind()) media, err := models.Medias.GetByHash(ctx, job.Args.MediaHash) if err != nil { log.Errorf("Error getting media by ID: %v", err) return JobCancel(err) } duration, err := utils.GetMediaDuration(job.Args.FilePath) if err != nil { log.Errorf("Error getting media duration: %v", err) return JobCancel(err) } // get file md5 log.Infof("pending get file md5 %s", job.Args.FilePath) fileMd5, err := utils.GetFileMd5(job.Args.FilePath) if err != nil { log.Errorf("Error getting file md5: %v", err) return JobCancel(err) } log.Infof("got file md5 %s %s", job.Args.FilePath, fileMd5) filePath := w.oss.GetSavePath(fileMd5 + filepath.Ext(job.Args.FilePath)) // get file size log.Infof("pending get file size %s", job.Args.FilePath) fileSize, err := utils.GetFileSize(job.Args.FilePath) if err != nil { log.Errorf("Error getting file size: %v", err) return JobCancel(err) } log.Infof("got file size %s %d", job.Args.FilePath, fileSize) // save to db and relate to master mediaModel := &model.Medias{ CreatedAt: time.Now(), Name: "[试听]" + media.Name, MimeType: media.MimeType, Size: fileSize, Path: filePath, Hash: fileMd5, Metas: fields.ToJson(fields.MediaMetas{ ParentHash: media.Hash, Short: true, Duration: duration, }), } // upload to oss log.Infof("pending upload file to oss %s", job.Args.FilePath) if err := w.oss.Upload(ctx, job.Args.FilePath, filePath, ali.WithInternal()); err != nil { log.Errorf("Error uploading file to OSS: %v", err) return err } log.Infof("pending create media record %s", job.Args.FilePath) if err := models.Medias.Create(ctx, mediaModel); err != nil { log.Errorf("Error saving media record: %v data: %+v", err, mediaModel) return err } log.Infof("Media record created with path: %s and hash: %s", filePath, fileMd5) log.Infof("pending remove local storage file %s", job.Args.FilePath) if w.job.Add(&RemoveFile{FilePath: job.Args.FilePath}); err != nil { log.Errorf("Error removing original file: %v", err) } return w.NextJob(media.Hash) } func (w *VideoStoreShortWorker) NextJob(hash string) error { if err := w.job.Add(&VideoExtractHeadImage{MediaHash: hash}); err != nil { log.Errorf("Error adding job: %v", err) return err } return nil }