rogee/qvyun
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

fix: jobs

Browse Source
This commit is contained in:
Rogee
2025-01-16 15:59:21 +08:00
parent 2cd7f2a2b8
commit 3a8bb03cb7
12 changed files with 619 additions and 12 deletions
Download Patch File Download Diff File Expand all files Collapse all files

12
backend/app/events/subscribers/post_created.go
View File

@@ -88,6 +88,18 @@ func (e *PostCreated) Handler(msg *message.Message) ([]*message.Message, error)
Hash: video.Hash,
TenantID: post.TenantID,
UserID: post.UserID,
Mark: "audio-preview",
}, nil)
if err != nil {
return nil, err
}
_, err = job.Insert(context.Background(), jobs.PostVideoExtractAudioJob{
PostID: post.ID,
Hash: video.Hash,
TenantID: post.TenantID,
UserID: post.UserID,
Mark: "audio",
}, nil)
if err != nil {
return nil, err

36
backend/app/http/medias/service.go
View File

@@ -3,6 +3,7 @@ package medias
import (
"context"
"database/sql"
"time"
"backend/database/models/qvyun_v2/public/model"
"backend/database/models/qvyun_v2/public/table"
@@ -31,6 +32,14 @@ func (svc *Service) Create(ctx context.Context, m *model.Medias) (*model.Medias,
_, span := otel.Start(ctx, "medias.service.Create")
defer span.End()
if m.CreatedAt.IsZero() {
m.CreatedAt = time.Now()
}
if m.UpdatedAt.IsZero() {
m.UpdatedAt = time.Now()
}
tbl := table.Medias
stmt := tbl.INSERT(tbl.MutableColumns).MODEL(m).RETURNING(tbl.AllColumns)
span.SetAttributes(semconv.DBStatementKey.String(stmt.DebugSql()))
@@ -72,3 +81,30 @@ func (svc *Service) GetMediasByHash(ctx context.Context, tenantID, userID int64,
return &item
}), nil
}
func (svc *Service) GetMediaByHash(ctx context.Context, tenantID, userID int64, hash string) (*model.Medias, error) {
_, span := otel.Start(ctx, "medias.service.GetMediasByHash")
defer span.End()
tbl := table.Medias
stmt := tbl.
SELECT(tbl.AllColumns).
LIMIT(1).
WHERE(
tbl.TenantID.
EQ(Int64(tenantID)).
AND(
tbl.UserID.EQ(Int64(userID)),
).
AND(
tbl.Hash.EQ(String(hash)),
),
)
span.SetAttributes(semconv.DBStatementKey.String(stmt.DebugSql()))
var ret model.Medias
if err := stmt.QueryContext(ctx, svc.db, &ret); err != nil {
return nil, err
}
return &ret, nil
}

86
backend/app/http/posts/service.go
View File

@@ -7,6 +7,7 @@ import (
"backend/app/requests"
"backend/database"
"backend/database/fields"
"backend/database/models/qvyun_v2/public/model"
"backend/database/models/qvyun_v2/public/table"
"backend/providers/hashids"
@@ -308,8 +309,10 @@ func (svc *Service) Update(ctx context.Context, tenantID, userID, postID int64,
attribute.Int64("user.id", userID),
attribute.Int64("post.id", postID),
)
tbl := table.Posts
post.UpdatedAt = time.Now()
tbl := table.Posts
stmt := tbl.
UPDATE(
tbl.MutableColumns.Except(
@@ -332,3 +335,84 @@ func (svc *Service) Update(ctx context.Context, tenantID, userID, postID int64,
return nil
}
// AttachAssets to post
func (svc *Service) AttachAssets(ctx context.Context, tenantID, userID, postID int64, mediaAssets []fields.MediaAsset) error {
_, span := otel.Start(ctx, "users.service.AttachAssets")
defer span.End()
span.SetAttributes(
attribute.Int64("tenant.id", tenantID),
attribute.Int64("user.id", userID),
attribute.Int64("post.id", postID),
)
post, err := svc.ForceGetPostByID(ctx, postID)
if err != nil {
return err
}
assets := append(post.Assets.Data, mediaAssets...)
post.Assets.Data = assets
tbl := table.Posts
stmt := tbl.
UPDATE(tbl.UpdatedAt, tbl.Assets).
SET(
tbl.UpdatedAt.SET(TimestampT(time.Now())),
tbl.Assets.SET(Json(post.Assets)),
).
WHERE(
tbl.ID.EQ(Int64(postID)).AND(
tbl.TenantID.EQ(Int64(tenantID)).AND(
tbl.UserID.EQ(Int64(userID)),
),
),
)
span.SetAttributes(semconv.DBStatementKey.String(stmt.DebugSql()))
if _, err := stmt.ExecContext(ctx, svc.db); err != nil {
return err
}
return svc.Update(ctx, tenantID, userID, postID, post)
}
// UpdateMeta
func (svc *Service) UpdateMeta(ctx context.Context, tenantID, userID, postID int64, meta fields.PostMeta) error {
_, span := otel.Start(ctx, "users.service.UpdateMeta")
defer span.End()
span.SetAttributes(
attribute.Int64("tenant.id", tenantID),
attribute.Int64("user.id", userID),
attribute.Int64("post.id", postID),
)
post, err := svc.ForceGetPostByID(ctx, postID)
if err != nil {
return err
}
post.Meta = meta
tbl := table.Posts
stmt := tbl.
UPDATE(tbl.UpdatedAt, tbl.Meta).
SET(
tbl.UpdatedAt.SET(TimestampT(time.Now())),
tbl.Meta.SET(Json(post.Meta)),
).
WHERE(
tbl.ID.EQ(Int64(postID)).AND(
tbl.TenantID.EQ(Int64(tenantID)).AND(
tbl.UserID.EQ(Int64(userID)),
),
),
)
span.SetAttributes(semconv.DBStatementKey.String(stmt.DebugSql()))
if _, err := stmt.ExecContext(ctx, svc.db); err != nil {
return err
}
return svc.Update(ctx, tenantID, userID, postID, post)
}

137
backend/app/jobs/post_video_cut.go
View File

@@ -1,12 +1,27 @@
package jobs
import (
"bytes"
"context"
"os"
"os/exec"
"strings"
"time"
"backend/app/http/medias"
"backend/app/http/posts"
"backend/app/http/storages"
"backend/database/fields"
"backend/database/models/qvyun_v2/public/model"
"backend/pkg/utils"
"backend/pkg/utils/fs"
_ "git.ipao.vip/rogeecn/atom"
_ "git.ipao.vip/rogeecn/atom/contracts"
"github.com/pkg/errors"
. "github.com/riverqueue/river"
"github.com/samber/lo"
"github.com/sirupsen/logrus"
)
var (
@@ -26,9 +41,9 @@ func (s PostVideoCutJob) InsertOpts() InsertOpts {
return InsertOpts{
Queue: QueueDefault,
Priority: PriorityDefault,
// UniqueOpts: UniqueOpts{
// ByArgs: true,
// },
UniqueOpts: UniqueOpts{
ByArgs: true,
},
}
}
@@ -43,6 +58,17 @@ var _ Worker[PostVideoCutJob] = (*PostVideoCutJobWorker)(nil)
// @provider(job)
type PostVideoCutJobWorker struct {
WorkerDefaults[PostVideoCutJob]
log *logrus.Entry `inject:"false"`
postSvc *posts.Service
mediaSvc *medias.Service
storageSvc *storages.Service
}
func (w *PostVideoCutJobWorker) Prepare() error {
w.log = logrus.WithField("worker", "PostVideoCutJobWorker")
return nil
}
func (w *PostVideoCutJobWorker) NextRetry(job *Job[PostVideoCutJob]) time.Time {
@@ -50,5 +76,110 @@ func (w *PostVideoCutJobWorker) NextRetry(job *Job[PostVideoCutJob]) time.Time {
}
func (w *PostVideoCutJobWorker) Work(ctx context.Context, job *Job[PostVideoCutJob]) error {
post, err := w.postSvc.GetPostByID(ctx, job.Args.PostID)
if err != nil {
return errors.Wrapf(err, "get post(%d) failed", job.Args.PostID)
}
media, err := w.mediaSvc.GetMediaByHash(ctx, job.Args.TenantID, job.Args.UserID, job.Args.Hash)
if err != nil {
return errors.Wrapf(err, "get media by hash(%s) failed", job.Args.Hash)
}
videoPath := media.Path
// 获取全长度的音频
_, ok := lo.Find(post.Assets.Data, func(asset fields.MediaAsset) bool {
return asset.Type == fields.MediaAssetTypeAudio && asset.Mark != nil && *asset.Mark == "audio-preview"
})
if ok {
return nil
}
previewVideoPath := strings.Replace(videoPath, ".mp4", "-preview.mp4", -1)
duration := lo.ToPtr("00:01:00")
if err := w.extractVideoFromVideo(ctx, videoPath, previewVideoPath, duration); err != nil {
return errors.Wrapf(err, "extract preview video from video failed")
}
fileMd5, err := utils.FileMd5(previewVideoPath)
if err != nil {
return errors.Wrapf(err, "get preview video(%s) file md5 failed", previewVideoPath)
}
if err := os.Rename(previewVideoPath, strings.Replace(videoPath, job.Args.Hash, fileMd5, 1)); err != nil {
return errors.Wrapf(err, "rename video(%s) file failed", videoPath)
}
storage, err := w.storageSvc.GetDefault(ctx)
if err != nil {
return err
}
// save to medias
_, err = w.mediaSvc.Create(ctx, &model.Medias{
TenantID: job.Args.TenantID,
UserID: job.Args.UserID,
PostID: post.ID,
StorageID: storage.ID,
Hash: fileMd5,
Name: post.Title,
MimeType: "video/mp4",
Size: fs.FileSize(previewVideoPath),
Path: previewVideoPath,
})
if err != nil {
return errors.Wrapf(err, "create media failed")
}
assets := []fields.MediaAsset{
{
Type: fields.MediaAssetTypeVideo,
Hash: fileMd5,
Mark: lo.ToPtr("video-preview"),
},
}
if err := w.postSvc.AttachAssets(ctx, job.Args.TenantID, job.Args.UserID, post.ID, assets); err != nil {
return errors.Wrapf(err, "attach video(%s) to post(%d) failed", videoPath, post.ID)
}
post.Meta.WorkerMark = post.Meta.WorkerMark & 1 << 0
if err := w.postSvc.UpdateMeta(ctx, job.Args.TenantID, job.Args.UserID, post.ID, post.Meta); err != nil {
return errors.Wrapf(err, "update post(%d) meta failed", post.ID)
}
return nil
}
// extractVideoFromVideo
func (w *PostVideoCutJobWorker) extractVideoFromVideo(ctx context.Context, videoPath, previewVideoPath string, duration *string) error {
args := []string{
"-i", videoPath,
"-c:v", "copy",
"-c:a", "copy",
}
if duration != nil {
args = append(args, "-t", *duration)
}
args = append(args, previewVideoPath)
w.log.Infof("extractVideoFromVideo: ffmpeg %s", strings.Join(args, " "))
cmd := exec.CommandContext(ctx, "ffmpeg", args...)
var buf bytes.Buffer
logWriter := utils.NewLogBuffer(func(line string) {
w.log.Info(line)
})
cmd.Stdout = utils.NewCombinedBuffer(&buf, logWriter)
cmd.Stderr = utils.NewCombinedBuffer(&buf, logWriter)
if err := cmd.Run(); err != nil {
return errors.Wrapf(err, "extract video failed: %s\n%s", videoPath, buf.String())
}
return nil
}

150
backend/app/jobs/post_video_extract_audio.go
View File

@@ -1,12 +1,27 @@
package jobs
import (
"bytes"
"context"
"os"
"os/exec"
"strings"
"time"
"backend/app/http/medias"
"backend/app/http/posts"
"backend/app/http/storages"
"backend/database/fields"
"backend/database/models/qvyun_v2/public/model"
"backend/pkg/utils"
"backend/pkg/utils/fs"
_ "git.ipao.vip/rogeecn/atom"
_ "git.ipao.vip/rogeecn/atom/contracts"
"github.com/pkg/errors"
. "github.com/riverqueue/river"
"github.com/samber/lo"
"github.com/sirupsen/logrus"
)
var (
@@ -19,6 +34,7 @@ type PostVideoExtractAudioJob struct {
TenantID int64
UserID int64
Hash string
Mark string
}
// InsertOpts implements JobArgsWithInsertOpts.
@@ -26,9 +42,9 @@ func (s PostVideoExtractAudioJob) InsertOpts() InsertOpts {
return InsertOpts{
Queue: QueueDefault,
Priority: PriorityDefault,
// UniqueOpts: UniqueOpts{
// ByArgs: true,
// },
UniqueOpts: UniqueOpts{
ByArgs: true,
},
}
}
@@ -41,6 +57,17 @@ var _ Worker[PostVideoExtractAudioJob] = (*PostVideoExtractAudioJobWorker)(nil)
// @provider(job)
type PostVideoExtractAudioJobWorker struct {
WorkerDefaults[PostVideoExtractAudioJob]
log *logrus.Entry `inject:"false"`
postSvc *posts.Service
mediaSvc *medias.Service
storageSvc *storages.Service
}
func (w *PostVideoExtractAudioJobWorker) Prepare() error {
w.log = logrus.WithField("worker", "PostVideoExtractAudioJobWorker")
return nil
}
func (w *PostVideoExtractAudioJobWorker) NextRetry(job *Job[PostVideoExtractAudioJob]) time.Time {
@@ -48,5 +75,122 @@ func (w *PostVideoExtractAudioJobWorker) NextRetry(job *Job[PostVideoExtractAudi
}
func (w *PostVideoExtractAudioJobWorker) Work(ctx context.Context, job *Job[PostVideoExtractAudioJob]) error {
post, err := w.postSvc.GetPostByID(ctx, job.Args.PostID)
if err != nil {
return errors.Wrapf(err, "get post(%d) failed", job.Args.PostID)
}
media, err := w.mediaSvc.GetMediaByHash(ctx, job.Args.TenantID, job.Args.UserID, job.Args.Hash)
if err != nil {
return errors.Wrapf(err, "get media by hash(%s) failed", job.Args.Hash)
}
videoPath := media.Path
// 获取全长度的音频
_, ok := lo.Find(post.Assets.Data, func(asset fields.MediaAsset) bool {
return asset.Type == fields.MediaAssetTypeAudio && asset.Mark != nil && *asset.Mark == job.Args.Mark
})
if ok {
return nil
}
audioPath := strings.Replace(videoPath, ".mp4", ".mp3", -1)
var duration *string
if job.Args.Mark == "audio-preview" {
duration = lo.ToPtr("00:01:00")
}
if err := w.extractAudioFromVideo(ctx, videoPath, audioPath, duration); err != nil {
return errors.Wrapf(err, "extract audio from video failed")
}
fileMd5, err := utils.FileMd5(audioPath)
if err != nil {
return errors.Wrapf(err, "get audio(%s) file md5 failed", audioPath)
}
if err := os.Rename(audioPath, strings.Replace(audioPath, job.Args.Hash, fileMd5, 1)); err != nil {
return errors.Wrapf(err, "rename audio(%s) file failed", audioPath)
}
storage, err := w.storageSvc.GetDefault(ctx)
if err != nil {
return err
}
// save to medias
_, err = w.mediaSvc.Create(ctx, &model.Medias{
TenantID: job.Args.TenantID,
UserID: job.Args.UserID,
PostID: post.ID,
StorageID: storage.ID,
Hash: fileMd5,
Name: post.Title,
MimeType: "audio/mp3",
Size: fs.FileSize(audioPath),
Path: audioPath,
})
if err != nil {
return errors.Wrapf(err, "create media failed")
}
assets := []fields.MediaAsset{
{
Type: fields.MediaAssetTypeAudio,
Hash: fileMd5,
Mark: lo.ToPtr(job.Args.Mark),
},
}
if err := w.postSvc.AttachAssets(ctx, job.Args.TenantID, job.Args.UserID, post.ID, assets); err != nil {
return errors.Wrapf(err, "attach audio(%s) to post(%d) failed", audioPath, post.ID)
}
if job.Args.Mark == "audio-preview" {
post.Meta.WorkerMark = post.Meta.WorkerMark & 1 << 1
} else if job.Args.Mark == "audio" {
post.Meta.WorkerMark = post.Meta.WorkerMark & 1 << 2
}
if err := w.postSvc.UpdateMeta(ctx, job.Args.TenantID, job.Args.UserID, post.ID, post.Meta); err != nil {
return errors.Wrapf(err, "update post(%d) meta failed", post.ID)
}
return nil
}
// extractAudioFromVideo extracts audio from video.
func (w *PostVideoExtractAudioJobWorker) extractAudioFromVideo(ctx context.Context, videoPath, audioPath string, duration *string) error {
args := []string{
"-i", videoPath,
"-vn",
"-acodec", "libmp3lame",
"-ar", "44100",
"-b:a", "128k",
"-q:a", "2",
}
if duration != nil {
args = append(args, "-t", *duration)
}
args = append(args, audioPath)
w.log.Infof("extractAudioFromVideo: ffmpeg %s", strings.Join(args, " "))
cmd := exec.CommandContext(ctx, "ffmpeg", args...)
var buf bytes.Buffer
logWriter := utils.NewLogBuffer(func(line string) {
w.log.Info(line)
})
cmd.Stdout = utils.NewCombinedBuffer(&buf, logWriter)
cmd.Stderr = utils.NewCombinedBuffer(&buf, logWriter)
if err := cmd.Run(); err != nil {
return errors.Wrapf(err, "extract audio failed: %s\n%s", videoPath, buf.String())
}
return nil
}