生成视频并且上传到minio-异步
This commit is contained in:
@@ -8,17 +8,23 @@ import (
|
||||
"io"
|
||||
"mime/multipart"
|
||||
"net/http"
|
||||
"net/url"
|
||||
"os"
|
||||
"os/exec"
|
||||
"path/filepath"
|
||||
"strings"
|
||||
"time"
|
||||
|
||||
dao "media/dao/video"
|
||||
dto "media/model/dto/video"
|
||||
entity "media/model/entity/video"
|
||||
|
||||
"gitea.com/red-future/common/beans"
|
||||
commonHttp "gitea.com/red-future/common/http"
|
||||
|
||||
"github.com/gogf/gf/v2/frame/g"
|
||||
"github.com/gogf/gf/v2/os/glog"
|
||||
"github.com/gogf/gf/v2/util/guid"
|
||||
)
|
||||
|
||||
type concatService struct{}
|
||||
@@ -47,6 +53,9 @@ type ConcatRes struct {
|
||||
|
||||
// Concat 拼接多个视频为一个
|
||||
func (s *concatService) Concat(ctx context.Context, req *ConcatReq) (res *ConcatRes, err error) {
|
||||
g.Log().Infof(ctx, "[Concat] 服务层收到请求: videoPaths=%v, method=%s, upload=%v",
|
||||
req.VideoPaths, req.Method, req.Upload)
|
||||
|
||||
if len(req.VideoPaths) < 2 {
|
||||
return nil, fmt.Errorf("至少需要2个视频才能拼接")
|
||||
}
|
||||
@@ -119,9 +128,10 @@ func (s *concatService) Concat(ctx context.Context, req *ConcatReq) (res *Concat
|
||||
InputFiles: len(req.VideoPaths),
|
||||
}
|
||||
|
||||
// 如果需要上传到 MinIO
|
||||
// 如果需要上传到 MinIO（用独立 context，避免 HTTP 断开后 ctx 被取消）
|
||||
if req.Upload {
|
||||
uploadRes, uploadErr := s.UploadToMinIO(ctx, outputPath)
|
||||
uploadCtx := context.Background()
|
||||
uploadRes, uploadErr := s.UploadToMinIO(uploadCtx, outputPath)
|
||||
if uploadErr != nil {
|
||||
return nil, fmt.Errorf("上传到MinIO失败: %v", uploadErr)
|
||||
}
|
||||
@@ -153,7 +163,9 @@ func (s *concatService) concatByDemuxer(ctx context.Context, ffmpegPath string,
|
||||
output,
|
||||
}
|
||||
|
||||
cmd := exec.CommandContext(ctx, ffmpegPath, args...)
|
||||
// 使用独立 context，避免 HTTP 请求超时导致 ffmpeg 被 SIGKILL
|
||||
bgCtx := context.Background()
|
||||
cmd := exec.CommandContext(bgCtx, ffmpegPath, args...)
|
||||
outputBytes, err := cmd.CombinedOutput()
|
||||
if err != nil {
|
||||
return fmt.Errorf("ffmpeg demuxer 失败: %v\n%s", err, string(outputBytes))
|
||||
@@ -232,7 +244,9 @@ func (s *concatService) concatByFilter(ctx context.Context, ffmpegPath string, i
|
||||
os.WriteFile(filterFile, []byte(filterStr), 0644)
|
||||
defer os.Remove(filterFile)
|
||||
|
||||
cmd := exec.CommandContext(ctx, ffmpegPath, args...)
|
||||
// 使用独立 context，避免 HTTP 请求超时导致 ffmpeg 被 SIGKILL
|
||||
bgCtx := context.Background()
|
||||
cmd := exec.CommandContext(bgCtx, ffmpegPath, args...)
|
||||
outputBytes, err := cmd.CombinedOutput()
|
||||
if err != nil {
|
||||
return fmt.Errorf("ffmpeg filter 失败: %v\n日志:\n%s", err, string(outputBytes))
|
||||
@@ -405,3 +419,263 @@ func CleanupConcat(paths []string) {
|
||||
os.Remove(p)
|
||||
}
|
||||
}
|
||||
|
||||
// ---------- 异步拼接任务管理 ----------
|
||||
|
||||
// CreateAsyncTask 创建异步拼接任务（URL模式），返回 taskId，后台处理
|
||||
func (s *concatService) CreateAsyncTask(ctx context.Context, videoURLs []string, method string, upload bool, callbackURL string) (string, error) {
|
||||
if len(videoURLs) < 2 {
|
||||
return "", fmt.Errorf("至少需要2个视频才能拼接")
|
||||
}
|
||||
|
||||
taskID := "concat_" + guid.S()
|
||||
task := &entity.ConcatTask{
|
||||
TaskID: taskID,
|
||||
Status: "pending",
|
||||
MethodUsed: method,
|
||||
}
|
||||
if _, err := dao.ConcatTask.Insert(ctx, task); err != nil {
|
||||
return "", fmt.Errorf("创建任务失败: %v", err)
|
||||
}
|
||||
|
||||
g.Log().Infof(ctx, "[异步拼接] 创建任务 %s, 视频数=%d, 回调=%s", taskID, len(videoURLs), callbackURL)
|
||||
|
||||
// 异步处理：先下载再拼接
|
||||
go s.processAsyncTask(taskID, videoURLs, method, upload, callbackURL)
|
||||
|
||||
return taskID, nil
|
||||
}
|
||||
|
||||
// CreateAsyncTaskWithFiles 创建异步拼接任务（文件上传模式），直接处理本地文件
|
||||
func (s *concatService) CreateAsyncTaskWithFiles(ctx context.Context, filePaths []string, method string, upload bool, callbackURL string) (string, error) {
|
||||
if len(filePaths) < 2 {
|
||||
return "", fmt.Errorf("至少需要2个视频才能拼接")
|
||||
}
|
||||
|
||||
taskID := "concat_" + guid.S()
|
||||
task := &entity.ConcatTask{
|
||||
TaskID: taskID,
|
||||
Status: "pending",
|
||||
MethodUsed: method,
|
||||
}
|
||||
if _, err := dao.ConcatTask.Insert(ctx, task); err != nil {
|
||||
return "", fmt.Errorf("创建任务失败: %v", err)
|
||||
}
|
||||
|
||||
g.Log().Infof(ctx, "[异步拼接-文件] 创建任务 %s, 文件数=%d, 回调=%s", taskID, len(filePaths), callbackURL)
|
||||
|
||||
// 异步处理：已有本地文件，直接拼接
|
||||
go s.processAsyncTaskWithFiles(taskID, filePaths, method, upload, callbackURL)
|
||||
|
||||
return taskID, nil
|
||||
}
|
||||
|
||||
// GetTaskResult 查询异步任务结果
|
||||
func (s *concatService) GetTaskResult(ctx context.Context, taskID string) (*dto.GetConcatTaskRes, error) {
|
||||
task, err := dao.ConcatTask.GetByTaskID(ctx, taskID)
|
||||
if err != nil {
|
||||
return nil, fmt.Errorf("查询任务失败: %v", err)
|
||||
}
|
||||
if task == nil {
|
||||
return nil, fmt.Errorf("任务不存在: %s", taskID)
|
||||
}
|
||||
|
||||
return dao.EntityToTaskRes(task), nil
|
||||
}
|
||||
|
||||
// processAsyncTaskWithFiles 后台处理异步拼接任务（文件上传模式，文件已在本地）
|
||||
func (s *concatService) processAsyncTaskWithFiles(taskID string, filePaths []string, method string, upload bool, callbackURL string) {
|
||||
bgCtx := context.Background()
|
||||
bgCtx = context.WithValue(bgCtx, "user", &beans.User{UserName: "admin", TenantId: 1})
|
||||
|
||||
dao.ConcatTask.UpdateRunning(bgCtx, taskID)
|
||||
|
||||
defer func() {
|
||||
if r := recover(); r != nil {
|
||||
errMsg := fmt.Sprintf("异步拼接异常: %v", r)
|
||||
g.Log().Errorf(bgCtx, "[异步拼接 %s] %s", taskID, errMsg)
|
||||
dao.ConcatTask.UpdateError(bgCtx, taskID, errMsg)
|
||||
s.concatCallback(bgCtx, taskID, callbackURL)
|
||||
}
|
||||
}()
|
||||
|
||||
concatErr := s.executeConcat(bgCtx, taskID, filePaths, method, upload)
|
||||
if concatErr != nil {
|
||||
dao.ConcatTask.UpdateError(bgCtx, taskID, concatErr.Error())
|
||||
s.concatCallback(bgCtx, taskID, callbackURL)
|
||||
return
|
||||
}
|
||||
|
||||
g.Log().Infof(bgCtx, "[异步拼接 %s] 完成", taskID)
|
||||
|
||||
if callbackURL != "" {
|
||||
s.concatCallback(bgCtx, taskID, callbackURL)
|
||||
}
|
||||
}
|
||||
|
||||
// processAsyncTask 后台处理异步拼接任务（URL模式，需要先下载）
|
||||
func (s *concatService) processAsyncTask(taskID string, videoURLs []string, method string, upload bool, callbackURL string) {
|
||||
bgCtx := context.Background()
|
||||
bgCtx = context.WithValue(bgCtx, "user", &beans.User{UserName: "admin", TenantId: 1})
|
||||
|
||||
dao.ConcatTask.UpdateRunning(bgCtx, taskID)
|
||||
|
||||
defer func() {
|
||||
if r := recover(); r != nil {
|
||||
errMsg := fmt.Sprintf("异步拼接异常: %v", r)
|
||||
g.Log().Errorf(bgCtx, "[异步拼接 %s] %s", taskID, errMsg)
|
||||
dao.ConcatTask.UpdateError(bgCtx, taskID, errMsg)
|
||||
s.concatCallback(bgCtx, taskID, callbackURL)
|
||||
}
|
||||
}()
|
||||
|
||||
// 下载视频
|
||||
var savePaths []string
|
||||
tempDir := g.Cfg().MustGet(bgCtx, "ffmpeg.temp_dir", "resource/temp").String()
|
||||
os.MkdirAll(tempDir, 0755)
|
||||
|
||||
for _, videoURL := range videoURLs {
|
||||
savePath, dlErr := downloadFile(bgCtx, videoURL, tempDir)
|
||||
if dlErr != nil {
|
||||
g.Log().Warningf(bgCtx, "[异步拼接 %s] 下载失败 %s: %v", taskID, videoURL, dlErr)
|
||||
continue
|
||||
}
|
||||
savePaths = append(savePaths, savePath)
|
||||
}
|
||||
|
||||
if len(savePaths) < 2 {
|
||||
errMsg := fmt.Sprintf("成功下载的视频不足2个(共%d)", len(videoURLs))
|
||||
dao.ConcatTask.UpdateError(bgCtx, taskID, errMsg)
|
||||
CleanupConcat(savePaths)
|
||||
s.concatCallback(bgCtx, taskID, callbackURL)
|
||||
return
|
||||
}
|
||||
|
||||
// 执行拼接
|
||||
concatErr := s.executeConcat(bgCtx, taskID, savePaths, method, upload)
|
||||
CleanupConcat(savePaths)
|
||||
|
||||
if concatErr != nil {
|
||||
dao.ConcatTask.UpdateError(bgCtx, taskID, concatErr.Error())
|
||||
s.concatCallback(bgCtx, taskID, callbackURL)
|
||||
return
|
||||
}
|
||||
|
||||
g.Log().Infof(bgCtx, "[异步拼接 %s] 完成", taskID)
|
||||
|
||||
if callbackURL != "" {
|
||||
s.concatCallback(bgCtx, taskID, callbackURL)
|
||||
}
|
||||
}
|
||||
|
||||
// executeConcat 执行拼接并更新任务状态，返回输出路径
|
||||
func (s *concatService) executeConcat(ctx context.Context, taskID string, filePaths []string, method string, upload bool) error {
|
||||
tempDir := filepath.Dir(filePaths[0])
|
||||
outputPath := filepath.Join(tempDir,
|
||||
fmt.Sprintf("concat_%s_x%d_%s.mp4", taskID, len(filePaths), time.Now().Format("150405")))
|
||||
|
||||
res, concatErr := s.Concat(ctx, &ConcatReq{
|
||||
VideoPaths: filePaths,
|
||||
OutputPath: outputPath,
|
||||
Method: method,
|
||||
Upload: upload,
|
||||
})
|
||||
if concatErr != nil {
|
||||
os.Remove(outputPath)
|
||||
return concatErr
|
||||
}
|
||||
|
||||
// 更新数据库为成功
|
||||
fileName := filepath.Base(outputPath)
|
||||
fileFormat := ""
|
||||
if idx := strings.LastIndex(fileName, "."); idx > 0 {
|
||||
fileFormat = fileName[idx+1:]
|
||||
}
|
||||
dao.ConcatTask.UpdateSuccess(ctx, taskID,
|
||||
res.FileURL, res.FileSize, fileName, fileFormat,
|
||||
"", res.MethodUsed, res.DurationStr)
|
||||
|
||||
os.Remove(outputPath)
|
||||
return nil
|
||||
}
|
||||
|
||||
// concatCallback 回调通知（从数据库读取任务结果发送）
|
||||
func (s *concatService) concatCallback(ctx context.Context, taskID, callbackURL string) {
|
||||
if callbackURL == "" {
|
||||
return
|
||||
}
|
||||
|
||||
task, err := dao.ConcatTask.GetByTaskID(ctx, taskID)
|
||||
if err != nil || task == nil {
|
||||
g.Log().Errorf(ctx, "[异步拼接回调 %s] 查询任务失败: %v", taskID, err)
|
||||
return
|
||||
}
|
||||
|
||||
payload := map[string]interface{}{
|
||||
"taskId": task.TaskID,
|
||||
"status": task.Status,
|
||||
}
|
||||
if task.Status == "success" {
|
||||
payload["fileURL"] = task.FileURL
|
||||
payload["fileSize"] = task.FileSize
|
||||
}
|
||||
if task.Status == "failed" {
|
||||
payload["errorMessage"] = task.ErrorMessage
|
||||
}
|
||||
|
||||
body, _ := json.Marshal(payload)
|
||||
g.Log().Infof(ctx, "[异步拼接回调 %s] 状态=%s, 目标=%s", taskID, task.Status, callbackURL)
|
||||
|
||||
req, _ := http.NewRequest("POST", callbackURL, bytes.NewReader(body))
|
||||
req.Header.Set("Content-Type", "application/json")
|
||||
userJSON, _ := json.Marshal(beans.User{UserName: "admin", TenantId: 1})
|
||||
req.Header.Set("X-User-Info", string(userJSON))
|
||||
|
||||
client := &http.Client{Timeout: 30 * time.Second}
|
||||
resp, reqErr := client.Do(req)
|
||||
if reqErr != nil {
|
||||
g.Log().Errorf(ctx, "[异步拼接回调 %s] 请求失败: %v", taskID, reqErr)
|
||||
return
|
||||
}
|
||||
defer resp.Body.Close()
|
||||
respBody, _ := io.ReadAll(resp.Body)
|
||||
g.Log().Infof(ctx, "[异步拼接回调 %s] 响应 status=%d, body=%s", taskID, resp.StatusCode, string(respBody))
|
||||
}
|
||||
|
||||
// downloadFile 下载文件到临时目录
|
||||
func downloadFile(ctx context.Context, rawURL, tempDir string) (string, error) {
|
||||
parsedURL, err := url.Parse(rawURL)
|
||||
if err != nil {
|
||||
return "", err
|
||||
}
|
||||
segments := strings.Split(parsedURL.Path, "/")
|
||||
fileName := segments[len(segments)-1]
|
||||
if fileName == "" {
|
||||
fileName = fmt.Sprintf("video_%d.mp4", time.Now().UnixMilli())
|
||||
}
|
||||
savePath := filepath.Join(tempDir, fmt.Sprintf("%d_%s", time.Now().UnixMilli(), fileName))
|
||||
|
||||
client := &http.Client{Timeout: 10 * time.Minute}
|
||||
resp, err := client.Get(rawURL)
|
||||
if err != nil {
|
||||
return "", err
|
||||
}
|
||||
defer resp.Body.Close()
|
||||
|
||||
if resp.StatusCode != http.StatusOK {
|
||||
return "", fmt.Errorf("HTTP %d", resp.StatusCode)
|
||||
}
|
||||
|
||||
out, err := os.Create(savePath)
|
||||
if err != nil {
|
||||
return "", err
|
||||
}
|
||||
defer out.Close()
|
||||
|
||||
_, err = io.Copy(out, resp.Body)
|
||||
if err != nil {
|
||||
os.Remove(savePath)
|
||||
return "", err
|
||||
}
|
||||
return savePath, nil
|
||||
}
|
||||
|
||||
Reference in New Issue
Block a user