diff --git a/backend/app/jobs/download_from_alioss.go b/backend/app/jobs/download_from_alioss.go index e14bec2..83b489a 100644 --- a/backend/app/jobs/download_from_alioss.go +++ b/backend/app/jobs/download_from_alioss.go @@ -2,10 +2,12 @@ package jobs import ( "context" + "errors" "os" "path/filepath" "time" + "quyun/app/models" "quyun/providers/ali" . "github.com/riverqueue/river" @@ -18,7 +20,7 @@ import ( var _ contracts.JobArgs = (*WechatCallback)(nil) type DownloadFromAliOSS struct { - Path string + MediaID int64 `json:"media_id"` } func (s DownloadFromAliOSS) InsertOpts() InsertOpts { @@ -50,13 +52,18 @@ func (w *DownloadFromAliOSSWorker) Work(ctx context.Context, job *Job[DownloadFr log.Infof("[Start] Working on job with strings: %+v", job.Args) defer log.Infof("[End] Finished %s", job.Args.Kind()) - dst := filepath.Join("/Users/rogee/Projects/self/quyun/backend/fixtures/oss/", job.Args.Path) + media, err := models.Medias.GetByID(ctx, job.Args.MediaID) + if err != nil { + log.Errorf("Error getting media by ID: %v", err) + return JobCancel(err) + } + + dst := filepath.Join("/Users/rogee/Projects/self/quyun/backend/fixtures/oss/", media.Path) // check is path exist - _, err := os.Stat(dst) + st, err := os.Stat(dst) if os.IsNotExist(err) { log.Infof("File not exists: %s", dst) - // midir err := os.MkdirAll(filepath.Dir(dst), os.ModePerm) if err != nil { log.Errorf("Error creating directory: %v", err) @@ -64,23 +71,23 @@ func (w *DownloadFromAliOSSWorker) Work(ctx context.Context, job *Job[DownloadFr } } - if err == nil { - log.Infof("File already exists: %s", dst) - err := os.Remove(dst) - if err != nil { + if st.Size() == media.Size { + return JobCancel(errors.New("file already downloaded")) + } else { + // remove file + if err := os.Remove(dst); err != nil { log.Errorf("Error removing file: %v", err) return err } - - return nil } - if err := w.oss.Download(ctx, job.Args.Path, dst); err != nil { + log.Infof("Starting download for file: %s", media.Path) + if err := w.oss.Download(ctx, media.Path, dst); err != nil { log.Errorf("Error downloading file: %v", err) return err } - log.Infof("Successfully downloaded file: %s", job.Args.Path) + log.Infof("Successfully downloaded file: %s", media.Path) return nil } diff --git a/backend/app/jobs/download_from_alioss_test.go b/backend/app/jobs/download_from_alioss_test.go index b050df6..f4fba77 100644 --- a/backend/app/jobs/download_from_alioss_test.go +++ b/backend/app/jobs/download_from_alioss_test.go @@ -4,6 +4,7 @@ import ( "context" "testing" + "quyun/app/models" "quyun/app/service/testx" "quyun/providers/ali" "quyun/providers/job" @@ -31,7 +32,7 @@ type DownloadFromAliOSSSuite struct { } func Test_DownloadFromAliOSS(t *testing.T) { - providers := testx.Default().With(Provide) + providers := testx.Default().With(Provide, models.Provide) testx.Serve(providers, t, func(p DownloadFromAliOSSSuiteInjectParams) { suite.Run(t, &DownloadFromAliOSSSuite{DownloadFromAliOSSSuiteInjectParams: p}) @@ -43,7 +44,7 @@ func (t *DownloadFromAliOSSSuite) Test_Work() { Convey("step 1", func() { job := &Job[DownloadFromAliOSS]{ Args: DownloadFromAliOSS{ - Path: "quyun/959e5310105c96e653f10b74e5bdc36b.mp4", + MediaID: 3, }, } diff --git a/backend/app/models/medias.go b/backend/app/models/medias.go index ec3766c..2e4215a 100644 --- a/backend/app/models/medias.go +++ b/backend/app/models/medias.go @@ -231,3 +231,21 @@ func (m *mediasModel) Update(ctx context.Context, hash string, model *model.Medi m.log.Infof("media item updated successfully") return nil } + +// GetByID +func (m *mediasModel) GetByID(ctx context.Context, id int64) (*model.Medias, error) { + tbl := table.Medias + stmt := tbl. + SELECT(tbl.AllColumns). + WHERE(tbl.ID.EQ(Int64(id))) + m.log.Infof("sql: %s", stmt.DebugSql()) + + var media model.Medias + err := stmt.QueryContext(ctx, db, &media) + if err != nil { + m.log.Errorf("error querying media item by ID: %v", err) + return nil, err + } + + return &media, nil +}