437 lines
13 KiB
Go
437 lines
13 KiB
Go
package video
|
|
|
|
import (
|
|
"bytes"
|
|
"context"
|
|
"encoding/json"
|
|
"fmt"
|
|
"io"
|
|
"mime/multipart"
|
|
"net/http"
|
|
"os"
|
|
"os/exec"
|
|
"path/filepath"
|
|
"strings"
|
|
"sync"
|
|
"time"
|
|
|
|
dao "media/dao/video"
|
|
dto "media/model/dto/video"
|
|
entity "media/model/entity/video"
|
|
|
|
"gitea.redpowerfuture.com/red-future/common/beans"
|
|
commonHttp "gitea.redpowerfuture.com/red-future/common/http"
|
|
|
|
"github.com/gogf/gf/v2/frame/g"
|
|
"github.com/gogf/gf/v2/util/guid"
|
|
)
|
|
|
|
type mergeService struct{}
|
|
|
|
// Merge 视频拼接+混音服务单例
|
|
var Merge = new(mergeService)
|
|
|
|
// mergeSem 并发控制信号量
|
|
var mergeSem chan struct{}
|
|
var mergeSemOnce sync.Once
|
|
|
|
// ---------- 异步任务管理 ----------
|
|
|
|
// CreateAsyncTask 创建异步拼接+混音任务（URL模式），返回 taskId，后台处理
|
|
func (s *mergeService) CreateAsyncTask(ctx context.Context, videoURLs, audioURLs []string, callbackURL string, upload bool) (string, error) {
|
|
if len(videoURLs) < 1 {
|
|
return "", fmt.Errorf("至少需要1个视频")
|
|
}
|
|
if len(audioURLs) < 1 {
|
|
return "", fmt.Errorf("至少需要1个音频")
|
|
}
|
|
|
|
// 将 videoURLs/audioURLs 序列化为 JSON 存入数据库
|
|
videoURLsJSON, _ := json.Marshal(videoURLs)
|
|
audioURLsJSON, _ := json.Marshal(audioURLs)
|
|
|
|
taskID := "merge_" + guid.S()
|
|
task := &entity.VideoAudioMergeTask{
|
|
TaskID: taskID,
|
|
VideoURLs: string(videoURLsJSON),
|
|
AudioURLs: string(audioURLsJSON),
|
|
Status: "pending",
|
|
CallbackURL: callbackURL,
|
|
}
|
|
if _, err := dao.MergeTask.Insert(ctx, task); err != nil {
|
|
return "", fmt.Errorf("创建任务失败: %v", err)
|
|
}
|
|
|
|
// 提取调用方用户信息，传给 goroutine
|
|
user := getUserFromCtx(ctx)
|
|
|
|
g.Log().Infof(ctx, "[拼接混音-异步] 创建任务 %s, 视频数=%d, 音频数=%d, 回调=%s",
|
|
taskID, len(videoURLs), len(audioURLs), callbackURL)
|
|
|
|
// 异步处理：先下载再拼接+混音
|
|
go s.processAsyncTask(user, taskID, videoURLs, audioURLs, upload, callbackURL)
|
|
|
|
return taskID, nil
|
|
}
|
|
|
|
// processAsyncTask 后台处理异步拼接+混音任务（URL模式，需要先下载）
|
|
func (s *mergeService) processAsyncTask(user *beans.User, taskID string, videoURLs, audioURLs []string, upload bool, callbackURL string) {
|
|
bgCtx := context.Background()
|
|
bgCtx = context.WithValue(bgCtx, "user", user)
|
|
|
|
dao.MergeTask.UpdateRunning(bgCtx, taskID)
|
|
|
|
defer func() {
|
|
if r := recover(); r != nil {
|
|
errMsg := fmt.Sprintf("异步拼接混音异常: %v", r)
|
|
g.Log().Errorf(bgCtx, "[拼接混音 %s] %s", taskID, errMsg)
|
|
dao.MergeTask.UpdateError(bgCtx, taskID, errMsg)
|
|
s.callback(bgCtx, taskID, callbackURL)
|
|
}
|
|
}()
|
|
|
|
// 1. 下载所有视频
|
|
tempDir := g.Cfg().MustGet(bgCtx, "ffmpeg.temp_dir", "resource/temp").String()
|
|
os.MkdirAll(tempDir, 0755)
|
|
|
|
var videoPaths []string
|
|
for _, videoURL := range videoURLs {
|
|
savePath, dlErr := downloadFile(bgCtx, videoURL, tempDir)
|
|
if dlErr != nil {
|
|
g.Log().Warningf(bgCtx, "[拼接混音 %s] 视频下载失败 %s: %v", taskID, videoURL, dlErr)
|
|
continue
|
|
}
|
|
videoPaths = append(videoPaths, savePath)
|
|
}
|
|
|
|
if len(videoPaths) < 1 {
|
|
errMsg := fmt.Sprintf("所有视频下载失败(共%d个)", len(videoURLs))
|
|
dao.MergeTask.UpdateError(bgCtx, taskID, errMsg)
|
|
cleanupFiles(videoPaths)
|
|
s.callback(bgCtx, taskID, callbackURL)
|
|
return
|
|
}
|
|
|
|
// 2. 下载所有音频
|
|
var audioPaths []string
|
|
for _, audioURL := range audioURLs {
|
|
savePath, dlErr := downloadFile(bgCtx, audioURL, tempDir)
|
|
if dlErr != nil {
|
|
g.Log().Warningf(bgCtx, "[拼接混音 %s] 音频下载失败 %s: %v", taskID, audioURL, dlErr)
|
|
continue
|
|
}
|
|
audioPaths = append(audioPaths, savePath)
|
|
}
|
|
|
|
if len(audioPaths) < 1 {
|
|
errMsg := fmt.Sprintf("所有音频下载失败(共%d个)", len(audioURLs))
|
|
dao.MergeTask.UpdateError(bgCtx, taskID, errMsg)
|
|
cleanupFiles(videoPaths)
|
|
cleanupFiles(audioPaths)
|
|
s.callback(bgCtx, taskID, callbackURL)
|
|
return
|
|
}
|
|
|
|
// 3. 等待并发许可，执行拼接+混音（FFmpeg 密集操作）
|
|
acquireMergeSem()
|
|
defer releaseMergeSem()
|
|
|
|
mergeErr := s.executeMerge(bgCtx, taskID, videoPaths, audioPaths, upload)
|
|
cleanupFiles(videoPaths)
|
|
cleanupFiles(audioPaths)
|
|
|
|
if mergeErr != nil {
|
|
dao.MergeTask.UpdateError(bgCtx, taskID, mergeErr.Error())
|
|
s.callback(bgCtx, taskID, callbackURL)
|
|
return
|
|
}
|
|
|
|
g.Log().Infof(bgCtx, "[拼接混音 %s] 完成", taskID)
|
|
|
|
if callbackURL != "" {
|
|
s.callback(bgCtx, taskID, callbackURL)
|
|
}
|
|
}
|
|
|
|
// executeMerge 执行拼接+混音，输出最终视频并更新任务状态
|
|
func (s *mergeService) executeMerge(ctx context.Context, taskID string, videoPaths, audioPaths []string, upload bool) error {
|
|
ffmpegPath, err := lookupFFmpegPath()
|
|
if err != nil {
|
|
return err
|
|
}
|
|
|
|
tempDir := filepath.Dir(videoPaths[0])
|
|
|
|
// Step 1: 拼接多视频 → 临时拼接文件
|
|
concatPath := filepath.Join(tempDir, fmt.Sprintf("concat_%s.mp4", taskID))
|
|
defer os.Remove(concatPath)
|
|
|
|
concatRes, concatErr := Concat.Concat(ctx, &ConcatReq{
|
|
VideoPaths: videoPaths,
|
|
OutputPath: concatPath,
|
|
Method: "reencode", // 强制重编码以统一分辨率
|
|
Upload: false,
|
|
})
|
|
if concatErr != nil {
|
|
return fmt.Errorf("视频拼接失败: %v", concatErr)
|
|
}
|
|
|
|
g.Log().Infof(ctx, "[拼接混音 %s] 视频拼接完成: duration=%s, size=%d", taskID, concatRes.DurationStr, concatRes.FileSize)
|
|
|
|
// Step 2: 如果有多段音频，先拼接音频
|
|
audioPath := audioPaths[0]
|
|
if len(audioPaths) > 1 {
|
|
concatAudioPath := filepath.Join(tempDir, fmt.Sprintf("concat_audio_%s.wav", taskID))
|
|
defer os.Remove(concatAudioPath)
|
|
|
|
concatAudioArgs := []string{}
|
|
for _, ap := range audioPaths {
|
|
concatAudioArgs = append(concatAudioArgs, "-i", ap)
|
|
}
|
|
filterStr := fmt.Sprintf("concat=n=%d:v=0:a=1", len(audioPaths))
|
|
concatAudioArgs = append(concatAudioArgs, "-filter_complex", filterStr, "-y", concatAudioPath)
|
|
|
|
g.Log().Debugf(ctx, "[拼接混音 %s] 音频拼接命令: %s %v", taskID, ffmpegPath, concatAudioArgs)
|
|
|
|
cmd := exec.CommandContext(ctx, ffmpegPath, concatAudioArgs...)
|
|
outputBytes, cmdErr := cmd.CombinedOutput()
|
|
if cmdErr != nil {
|
|
return fmt.Errorf("音频拼接失败: %v\n%s", cmdErr, string(outputBytes))
|
|
}
|
|
|
|
audioPath = concatAudioPath
|
|
g.Log().Infof(ctx, "[拼接混音 %s] 音频拼接完成: %d段 → %s", taskID, len(audioPaths), concatAudioPath)
|
|
}
|
|
|
|
// Step 3: 混入音频（以视频时长为准，音频短则静音补全，音频长则截断）
|
|
outputPath := filepath.Join(tempDir, fmt.Sprintf("merge_%s_%s.mp4", taskID, time.Now().Format("150405")))
|
|
|
|
duration := concatRes.Duration
|
|
|
|
args := []string{
|
|
"-i", concatPath,
|
|
"-i", audioPath,
|
|
"-c:v", "copy",
|
|
"-map", "0:v:0",
|
|
"-map", "1:a:0",
|
|
"-af", "apad",
|
|
"-c:a", "aac",
|
|
"-t", fmt.Sprintf("%.3f", duration),
|
|
"-y",
|
|
outputPath,
|
|
}
|
|
|
|
g.Log().Debugf(ctx, "[拼接混音 %s] 混音命令: %s %v", taskID, ffmpegPath, args)
|
|
|
|
bgCtx := context.Background()
|
|
cmd := exec.CommandContext(bgCtx, ffmpegPath, args...)
|
|
outputBytes, err := cmd.CombinedOutput()
|
|
if err != nil {
|
|
os.Remove(outputPath)
|
|
return fmt.Errorf("混音失败: %v\n%s", err, string(outputBytes))
|
|
}
|
|
|
|
// 获取输出文件信息
|
|
stat, statErr := os.Stat(outputPath)
|
|
if statErr != nil {
|
|
os.Remove(outputPath)
|
|
return fmt.Errorf("输出文件异常: %v", statErr)
|
|
}
|
|
|
|
durationStr := formatDuration(duration)
|
|
|
|
// 上传到 MinIO
|
|
fileURL := ""
|
|
if upload {
|
|
uploadCtx := context.WithValue(context.Background(), "user", getUserFromCtx(ctx))
|
|
uploadRes, uploadErr := uploadToMinIO(uploadCtx, outputPath)
|
|
if uploadErr != nil {
|
|
os.Remove(outputPath)
|
|
return fmt.Errorf("上传到MinIO失败: %v", uploadErr)
|
|
}
|
|
fileURL = uploadRes.FileURL
|
|
}
|
|
|
|
// 更新数据库为成功
|
|
fileName := filepath.Base(outputPath)
|
|
fileFormat := ""
|
|
if idx := strings.LastIndex(fileName, "."); idx > 0 {
|
|
fileFormat = fileName[idx+1:]
|
|
}
|
|
dao.MergeTask.UpdateSuccess(ctx, taskID,
|
|
fileURL, stat.Size(), fileName, fileFormat,
|
|
"", durationStr)
|
|
|
|
os.Remove(outputPath)
|
|
return nil
|
|
}
|
|
|
|
// GetTaskResult 查询异步任务结果
|
|
func (s *mergeService) GetTaskResult(ctx context.Context, taskID string) (*dto.GetMergeTaskRes, error) {
|
|
task, err := dao.MergeTask.GetByTaskID(ctx, taskID)
|
|
if err != nil {
|
|
return nil, fmt.Errorf("查询任务失败: %v", err)
|
|
}
|
|
if task == nil {
|
|
return nil, fmt.Errorf("任务不存在: %s", taskID)
|
|
}
|
|
return dao.EntityToMergeTaskRes(task), nil
|
|
}
|
|
|
|
// callback 回调通知（从数据库读取任务结果发送）
|
|
func (s *mergeService) callback(ctx context.Context, taskID, callbackURL string) {
|
|
if callbackURL == "" {
|
|
return
|
|
}
|
|
|
|
task, err := dao.MergeTask.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
|
|
payload["durationStr"] = task.DurationStr
|
|
}
|
|
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")
|
|
cbUser := getUserFromCtx(ctx)
|
|
userJSON, _ := json.Marshal(cbUser)
|
|
req.Header.Set("X-User-Info", string(userJSON))
|
|
|
|
client := &http.Client{Timeout: 2 * time.Minute}
|
|
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))
|
|
}
|
|
|
|
// ---------- 上传到 MinIO ----------
|
|
|
|
// uploadToMinIO 上传文件到 MinIO（复用 concat_service 的逻辑）
|
|
func uploadToMinIO(ctx context.Context, localFilePath string) (*uploadFileRes, error) {
|
|
var buf bytes.Buffer
|
|
mw := multipart.NewWriter(&buf)
|
|
|
|
file, err := os.Open(localFilePath)
|
|
if err != nil {
|
|
return nil, fmt.Errorf("打开文件失败: %v", err)
|
|
}
|
|
defer file.Close()
|
|
|
|
fw, err := mw.CreateFormFile("file", filepath.Base(localFilePath))
|
|
if err != nil {
|
|
return nil, fmt.Errorf("创建表单文件字段失败: %v", err)
|
|
}
|
|
if _, err = io.Copy(fw, file); err != nil {
|
|
return nil, fmt.Errorf("写入文件内容失败: %v", err)
|
|
}
|
|
mw.Close()
|
|
|
|
client := commonHttp.Httpclient.Clone()
|
|
newTransport := http.DefaultTransport.(*http.Transport).Clone()
|
|
newTransport.ResponseHeaderTimeout = 5 * time.Minute
|
|
client.Transport = newTransport
|
|
client.SetTimeout(10 * time.Minute)
|
|
|
|
hasAuthHeader := false
|
|
if r := g.RequestFromCtx(ctx); r != nil {
|
|
for k, v := range r.Header {
|
|
client.SetHeader(k, v[0])
|
|
if strings.EqualFold(k, "Authorization") || strings.EqualFold(k, "X-User-Info") {
|
|
hasAuthHeader = true
|
|
}
|
|
}
|
|
}
|
|
if !hasAuthHeader {
|
|
uploadUser := getUserFromCtx(ctx)
|
|
userJSON, _ := json.Marshal(uploadUser)
|
|
client.SetHeader("X-User-Info", string(userJSON))
|
|
}
|
|
|
|
contentType := mw.FormDataContentType()
|
|
client.SetHeader("Content-Type", contentType)
|
|
|
|
response, err := client.Post(ctx, "oss/file/uploadFile", buf.Bytes())
|
|
if err != nil {
|
|
return nil, fmt.Errorf("调用OSS上传服务失败: %v", err)
|
|
}
|
|
defer response.Close()
|
|
|
|
body := response.ReadAll()
|
|
|
|
var apiResp struct {
|
|
Code int `json:"code"`
|
|
Message string `json:"message"`
|
|
Data *uploadFileRes `json:"data"`
|
|
}
|
|
if err = json.Unmarshal(body, &apiResp); err != nil {
|
|
return nil, fmt.Errorf("响应解析失败: %v", err)
|
|
}
|
|
if apiResp.Code != 200 && apiResp.Code != 0 {
|
|
return nil, fmt.Errorf("OSS上传失败: %s", apiResp.Message)
|
|
}
|
|
return apiResp.Data, nil
|
|
}
|
|
|
|
// acquireMergeSem 获取并发许可（懒初始化信号量）
|
|
func acquireMergeSem() {
|
|
mergeSemOnce.Do(func() {
|
|
concurrency := g.Cfg().MustGet(context.Background(), "merge.concurrency", 1).Int()
|
|
if concurrency < 1 {
|
|
concurrency = 1
|
|
}
|
|
mergeSem = make(chan struct{}, concurrency)
|
|
})
|
|
mergeSem <- struct{}{}
|
|
}
|
|
|
|
// releaseMergeSem 释放并发许可
|
|
func releaseMergeSem() {
|
|
<-mergeSem
|
|
}
|
|
|
|
// cleanupFiles 清理文件列表
|
|
func cleanupFiles(paths []string) {
|
|
for _, p := range paths {
|
|
os.Remove(p)
|
|
}
|
|
}
|
|
|
|
// lookupFFmpegPath 查找 ffmpeg 可执行文件路径
|
|
func lookupFFmpegPath() (string, error) {
|
|
ctx := context.Background()
|
|
ffmpegPath := g.Cfg().MustGet(ctx, "ffmpeg.path", "").String()
|
|
if ffmpegPath != "" {
|
|
if _, err := os.Stat(ffmpegPath); err == nil {
|
|
g.Log().Infof(ctx, "[ffmpeg] ✔ 已安装, 使用配置路径: %s", ffmpegPath)
|
|
return ffmpegPath, nil
|
|
}
|
|
g.Log().Infof(ctx, "[ffmpeg] 配置路径不可用: %s，回退到系统 PATH 查找", ffmpegPath)
|
|
}
|
|
path, err := exec.LookPath("ffmpeg")
|
|
if err != nil {
|
|
g.Log().Error(ctx, "[ffmpeg] ❌ 未找到，启动时已自动尝试安装，若仍缺失请手动安装")
|
|
return "", fmt.Errorf("未找到 ffmpeg")
|
|
}
|
|
g.Log().Infof(ctx, "[ffmpeg] ✔ 已安装, 系统路径: %s", path)
|
|
return path, nil
|
|
}
|