q792602257/VptPassiveAdapter
0
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

telemetry

Browse Source
This commit is contained in:
Jerry Yan
2025-04-13 11:24:20 +08:00
parent a6c5ba5957
commit 3d989c2f47
14 changed files with 443 additions and 63 deletions
Download Patch File Download Diff File Expand all files Collapse all files

130
util/ffmpeg.go
View File

@ -3,7 +3,10 @@ package util
import (
"ZhenTuLocalPassiveAdapter/dto"
"bytes"
"context"
"fmt"
"go.opentelemetry.io/otel/attribute"
"go.opentelemetry.io/otel/codes"
"log"
"math/rand"
"os"
@ -17,25 +20,36 @@ import (
const FfmpegExec = "ffmpeg"
func RunFfmpegTask(task *dto.FfmpegTask) bool {
func RunFfmpegTask(ctx context.Context, task *dto.FfmpegTask) bool {
_, span := tracer.Start(ctx, "RunFfmpegTask")
defer span.End()
var result bool
if len(task.Files) == 1 {
// 单个文件切割，用简单方法
result = runFfmpegForSingleFile(task)
result = runFfmpegForSingleFile(ctx, task)
} else {
// 多个文件切割，用速度快的
result = runFfmpegForMultipleFile1(task)
result = runFfmpegForMultipleFile1(ctx, task)
}
// 先尝试方法1
if result {
span.SetStatus(codes.Ok, "FFMPEG简易方法成功")
return true
}
log.Printf("FFMPEG简易方法失败，尝试复杂方法转码")
// 不行再尝试方法二
return runFfmpegForMultipleFile2(task)
result = runFfmpegForMultipleFile2(ctx, task)
if result {
span.SetStatus(codes.Ok, "FFMPEG复杂方法成功")
return true
}
span.SetStatus(codes.Error, "FFMPEG复杂方法失败")
return result
}
func runFfmpegForMultipleFile1(task *dto.FfmpegTask) bool {
func runFfmpegForMultipleFile1(ctx context.Context, task *dto.FfmpegTask) bool {
_, span := tracer.Start(ctx, "runFfmpegForMultipleFile1")
defer span.End()
// 多文件，方法一：先转换成ts，然后合并切割
// 步骤一：先转换成ts，并行转换
var wg sync.WaitGroup
@ -47,7 +61,7 @@ func runFfmpegForMultipleFile1(task *dto.FfmpegTask) bool {
go func(file *dto.File) {
defer wg.Done()
tmpFile := path.Join(os.TempDir(), file.Name+".ts")
result, err := convertMp4ToTs(*file, tmpFile)
result, err := convertMp4ToTs(ctx, *file, tmpFile)
if err != nil {
log.Printf("转码出错: %v", err)
mu.Lock()
@ -69,12 +83,15 @@ func runFfmpegForMultipleFile1(task *dto.FfmpegTask) bool {
wg.Wait()
if notOk {
span.SetStatus(codes.Error, "FFMPEG多文件转码失败")
return false
}
// 步骤二：使用concat协议拼接裁切
result, err := QuickConcatVideoCut(task.Files, int64(task.Offset), int64(task.Length), task.OutputFile)
result, err := QuickConcatVideoCut(ctx, task.Files, int64(task.Offset), int64(task.Length), task.OutputFile)
if err != nil {
span.SetAttributes(attribute.String("error", err.Error()))
span.SetStatus(codes.Error, "FFMPEG多文件concat协议转码失败")
return false
}
@ -84,34 +101,49 @@ func runFfmpegForMultipleFile1(task *dto.FfmpegTask) bool {
log.Printf("删除临时文件失败: %v", err)
}
}
if result {
span.SetStatus(codes.Ok, "FFMPEG多文件concat协议转码成功")
} else {
span.SetStatus(codes.Error, "FFMPEG多文件concat协议转码失败")
}
return result
}
func runFfmpegForMultipleFile2(task *dto.FfmpegTask) bool {
func runFfmpegForMultipleFile2(ctx context.Context, task *dto.FfmpegTask) bool {
_, span := tracer.Start(ctx, "runFfmpegForMultipleFile2")
defer span.End()
// 多文件，方法二：使用计算资源编码
result, err := SlowVideoCut(task.Files, int64(task.Offset), int64(task.Length), task.OutputFile)
result, err := SlowVideoCut(ctx, task.Files, int64(task.Offset), int64(task.Length), task.OutputFile)
if err != nil {
return false
}
return result
}
func runFfmpegForSingleFile(task *dto.FfmpegTask) bool {
result, err := QuickVideoCut(task.Files[0].Url, int64(task.Offset), int64(task.Length), task.OutputFile)
func runFfmpegForSingleFile(ctx context.Context, task *dto.FfmpegTask) bool {
_, span := tracer.Start(ctx, "runFfmpegForSingleFile")
defer span.End()
result, err := QuickVideoCut(ctx, task.Files[0].Url, int64(task.Offset), int64(task.Length), task.OutputFile)
if err != nil {
span.SetStatus(codes.Error, "FFMPEG单个文件裁切失败")
return false
}
_, err = os.Stat(task.OutputFile)
stat, err := os.Stat(task.OutputFile)
if err != nil {
span.SetStatus(codes.Error, "文件不存在")
log.Printf("文件不存在：%s", task.OutputFile)
return false
}
span.SetAttributes(attribute.String("file.name", task.OutputFile))
span.SetAttributes(attribute.Int64("file.size", stat.Size()))
return result
}
func CheckFileCoverageAndConstructTask(fileList []dto.File, beginDt, endDt time.Time, task dto.Task) (*dto.FfmpegTask, error) {
func CheckFileCoverageAndConstructTask(ctx context.Context, fileList []dto.File, beginDt, endDt time.Time, task dto.Task) (*dto.FfmpegTask, error) {
_, span := tracer.Start(ctx, "CheckFileCoverageAndConstructTask")
defer span.End()
if fileList == nil || len(fileList) == 0 {
span.SetStatus(codes.Error, "无法根据要求找到对应录制片段")
log.Printf("无法根据要求找到对应录制片段！ID：【%s】，开始时间：【%s】，结束时间：【%s】", task.TaskID, beginDt, endDt)
return nil, fmt.Errorf("无法根据要求找到对应录制片段")
}
@ -126,6 +158,7 @@ func CheckFileCoverageAndConstructTask(fileList []dto.File, beginDt, endDt time.
}
if file.StartTime.Sub(lastFile.EndTime).Milliseconds() > 2000 {
// 片段断开
span.SetStatus(codes.Error, "FFMPEG片段断开")
log.Printf("分析FFMPEG任务失败：ID：【%s】，文件片段：【%s,%s】中间断开【%f】秒(超过2秒)", task.TaskID, lastFile.Name, file.Name, file.StartTime.Sub(lastFile.EndTime).Seconds())
return nil, fmt.Errorf("片段断开")
}
@ -135,6 +168,7 @@ func CheckFileCoverageAndConstructTask(fileList []dto.File, beginDt, endDt time.
// 通过文件列表构造的任务仍然是缺失的
if fileList[len(fileList)-1].EndTime.Before(endDt) {
span.SetStatus(codes.Error, "FFMPEG片段断开")
log.Printf("分析FFMPEG任务失败：ID：【%s】，文件片段：【%s】，无法完整覆盖时间点【%s】", task.TaskID, fileList[len(fileList)-1].Name, endDt)
return nil, fmt.Errorf("片段断开")
}
@ -146,11 +180,16 @@ func CheckFileCoverageAndConstructTask(fileList []dto.File, beginDt, endDt time.
Offset: int(beginDt.Sub(fileList[0].StartTime).Seconds()),
OutputFile: path.Join(os.TempDir(), task.TaskID+".mp4"),
}
span.SetAttributes(attribute.Int("task.files", len(ffmpegTask.Files)))
span.SetAttributes(attribute.Int("task.offset", ffmpegTask.Offset))
span.SetAttributes(attribute.Int("task.length", ffmpegTask.Length))
span.SetStatus(codes.Ok, "FFMPEG任务构造成功")
return ffmpegTask, nil
}
func convertMp4ToTs(file dto.File, outFileName string) (bool, error) {
func convertMp4ToTs(ctx context.Context, file dto.File, outFileName string) (bool, error) {
_, span := tracer.Start(ctx, "convertMp4ToTs")
defer span.End()
ffmpegCmd := []string{
FfmpegExec,
"-hide_banner",
@ -161,10 +200,12 @@ func convertMp4ToTs(file dto.File, outFileName string) (bool, error) {
"-f", "mpegts",
outFileName,
}
return handleFfmpegProcess(ffmpegCmd)
return handleFfmpegProcess(ctx, ffmpegCmd)
}
func convertHevcToTs(file dto.File, outFileName string) (bool, error) {
func convertHevcToTs(ctx context.Context, file dto.File, outFileName string) (bool, error) {
_, span := tracer.Start(ctx, "convertHevcToTs")
defer span.End()
ffmpegCmd := []string{
FfmpegExec,
"-hide_banner",
@ -175,10 +216,12 @@ func convertHevcToTs(file dto.File, outFileName string) (bool, error) {
"-f", "mpegts",
outFileName,
}
return handleFfmpegProcess(ffmpegCmd)
return handleFfmpegProcess(ctx, ffmpegCmd)
}
func QuickVideoCut(inputFile string, offset, length int64, outputFile string) (bool, error) {
func QuickVideoCut(ctx context.Context, inputFile string, offset, length int64, outputFile string) (bool, error) {
_, span := tracer.Start(ctx, "QuickVideoCut")
defer span.End()
ffmpegCmd := []string{
FfmpegExec,
"-hide_banner",
@ -189,16 +232,21 @@ func QuickVideoCut(inputFile string, offset, length int64, outputFile string) (b
"-reset_timestamps", "1",
"-ss", strconv.FormatInt(offset, 10),
"-t", strconv.FormatInt(length, 10),
"-fflags", "+genpts",
"-f", "mp4",
outputFile,
}
return handleFfmpegProcess(ffmpegCmd)
return handleFfmpegProcess(ctx, ffmpegCmd)
}
func QuickConcatVideoCut(inputFiles []dto.File, offset, length int64, outputFile string) (bool, error) {
func QuickConcatVideoCut(ctx context.Context, inputFiles []dto.File, offset, length int64, outputFile string) (bool, error) {
_, span := tracer.Start(ctx, "QuickConcatVideoCut")
defer span.End()
tmpFile := fmt.Sprintf("tmp%.10f.txt", rand.Float64())
tmpFileObj, err := os.Create(tmpFile)
if err != nil {
span.SetAttributes(attribute.String("error", err.Error()))
span.SetStatus(codes.Error, "创建临时文件失败")
log.Printf("创建临时文件失败：%s", tmpFile)
return false, err
}
@ -208,6 +256,8 @@ func QuickConcatVideoCut(inputFiles []dto.File, offset, length int64, outputFile
for _, filePo := range inputFiles {
_, err := tmpFileObj.WriteString(fmt.Sprintf("file '%s'\n", filePo.Url))
if err != nil {
span.SetAttributes(attribute.String("error", err.Error()))
span.SetStatus(codes.Error, "写入临时文件失败")
log.Printf("写入临时文件失败：%s", tmpFile)
return false, err
}
@ -227,10 +277,12 @@ func QuickConcatVideoCut(inputFiles []dto.File, offset, length int64, outputFile
"-f", "mp4",
outputFile,
}
return handleFfmpegProcess(ffmpegCmd)
return handleFfmpegProcess(ctx, ffmpegCmd)
}
func SlowVideoCut(inputFiles []dto.File, offset, length int64, outputFile string) (bool, error) {
func SlowVideoCut(ctx context.Context, inputFiles []dto.File, offset, length int64, outputFile string) (bool, error) {
_, span := tracer.Start(ctx, "SlowVideoCut")
defer span.End()
ffmpegCmd := []string{
FfmpegExec,
"-hide_banner",
@ -259,11 +311,17 @@ func SlowVideoCut(inputFiles []dto.File, offset, length int64, outputFile string
outputFile,
)
return handleFfmpegProcess(ffmpegCmd)
return handleFfmpegProcess(ctx, ffmpegCmd)
}
func handleFfmpegProcess(ffmpegCmd []string) (bool, error) {
func handleFfmpegProcess(ctx context.Context, ffmpegCmd []string) (bool, error) {
_, span := tracer.Start(ctx, "handleFfmpegProcess")
defer span.End()
span.SetAttributes(attribute.String("ffmpeg.cmd", strings.Join(ffmpegCmd, " ")))
startTime := time.Now()
defer func() {
span.SetAttributes(attribute.Int64("ffmpeg.duration", int64(time.Since(startTime).Seconds())))
}()
log.Printf("FFMPEG执行命令：【%s】", strings.Join(ffmpegCmd, " "))
cmd := exec.Command(ffmpegCmd[0], ffmpegCmd[1:]...)
@ -272,6 +330,8 @@ func handleFfmpegProcess(ffmpegCmd []string) (bool, error) {
err := cmd.Start()
if err != nil {
span.SetAttributes(attribute.String("ffmpeg.stderr", stderr.String()))
span.SetStatus(codes.Error, "FFMPEG执行命令失败")
log.Printf("FFMPEG执行命令失败，错误信息：%s，命令：【%s】", stderr.String(), strings.Join(ffmpegCmd, " "))
return false, err
}
@ -284,10 +344,14 @@ func handleFfmpegProcess(ffmpegCmd []string) (bool, error) {
select {
case <-time.After(1 * time.Minute):
span.SetAttributes(attribute.String("ffmpeg.stderr", stderr.String()))
span.SetStatus(codes.Error, "FFMPEG执行命令没有在1分钟内退出")
log.Printf("FFMPEG执行命令没有在1分钟内退出，命令：【%s】", strings.Join(ffmpegCmd, " "))
return false, fmt.Errorf("ffmpeg command timed out")
case err := <-done:
if err != nil {
span.SetAttributes(attribute.String("ffmpeg.stderr", stderr.String()))
span.SetStatus(codes.Error, "FFMPEG执行命令失败")
log.Printf("FFMPEG执行命令失败，错误信息：%s，命令：【%s】", stderr.String(), strings.Join(ffmpegCmd, " "))
return false, err
}
@ -297,7 +361,9 @@ func handleFfmpegProcess(ffmpegCmd []string) (bool, error) {
}
}
func GetVideoDuration(filePath string) (float64, error) {
func GetVideoDuration(ctx context.Context, filePath string) (float64, error) {
_, span := tracer.Start(ctx, "GetVideoDuration")
defer span.End()
ffprobeCmd := []string{
"ffprobe",
"-v", "error",
@ -305,21 +371,25 @@ func GetVideoDuration(filePath string) (float64, error) {
"-of", "default=noprint_wrappers=1:nokey=1",
filePath,
}
span.SetAttributes(attribute.String("ffprobe.cmd", strings.Join(ffprobeCmd, " ")))
cmd := exec.Command(ffprobeCmd[0], ffprobeCmd[1:]...)
var out bytes.Buffer
cmd.Stdout = &out
err := cmd.Run()
if err != nil {
span.SetAttributes(attribute.String("error", err.Error()))
span.SetStatus(codes.Error, "failed to get video duration")
return 0, fmt.Errorf("failed to get video duration: %w", err)
}
span.SetAttributes(attribute.String("ffmpeg.stdout", out.String()))
durationStr := strings.TrimSpace(out.String())
duration, err := strconv.ParseFloat(durationStr, 64)
if err != nil {
span.SetAttributes(attribute.String("error", err.Error()))
span.SetStatus(codes.Error, "failed to parse video duration")
return 0, fmt.Errorf("failed to parse video duration: %w", err)
}
span.SetAttributes(attribute.Float64("video.duration", duration))
return duration, nil
}

5
util/tracer.go Normal file
View File

@ -0,0 +1,5 @@
package util
import "go.opentelemetry.io/otel"
var tracer = otel.Tracer("util")