red-future/media
4
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

修复异常被吞的问题

Browse Source
This commit is contained in:
lmk
2026-06-24 15:51:02 +08:00
parent a3b8005175
commit 4ded242e66
9 changed files with 205 additions and 52 deletions
Download Patch File Download Diff File Expand all files Collapse all files
go.mod
-4
View File
@@ -30,7 +30,6 @@ require (
github.com/goccy/go-json v0.10.6 // indirect
github.com/gogf/gf/contrib/nosql/redis/v2 v2.10.2 // indirect
github.com/gogf/gf/contrib/registry/consul/v2 v2.9.5 // indirect
github.com/gogf/gf/contrib/trace/otlphttp/v2 v2.9.5 // indirect
github.com/gogo/protobuf v1.3.2 // indirect
github.com/golang-jwt/jwt/v5 v5.3.1 // indirect
github.com/golang/glog v1.2.5 // indirect
@@ -50,17 +49,14 @@ require (
github.com/hashicorp/go-immutable-radix v1.3.1 // indirect
github.com/hashicorp/go-multierror v1.1.1 // indirect
github.com/hashicorp/go-rootcerts v1.0.2 // indirect
github.com/hashicorp/go-version v1.9.0 // indirect
github.com/hashicorp/golang-lru v1.0.2 // indirect
github.com/hashicorp/serf v0.10.1 // indirect
github.com/klauspost/compress v1.18.6 // indirect
github.com/klauspost/cpuid/v2 v2.3.0 // indirect
github.com/lib/pq v1.12.3 // indirect
github.com/magiconair/properties v1.8.10 // indirect
github.com/mattn/go-colorable v0.1.15 // indirect
github.com/mattn/go-isatty v0.0.22 // indirect
github.com/mattn/go-runewidth v0.0.24 // indirect
github.com/miekg/dns v1.1.72 // indirect
github.com/mitchellh/go-homedir v1.1.0 // indirect
github.com/olekukonko/cat v0.0.0-20250911104152-50322a0618f6 // indirect
github.com/olekukonko/errors v1.3.0 // indirect
go.sum
-2
View File
@@ -89,8 +89,6 @@ github.com/gogf/gf/contrib/nosql/redis/v2 v2.10.2 h1:iTQegT+lEg/wDKvj2mi3W1wrdrw
github.com/gogf/gf/contrib/nosql/redis/v2 v2.10.2/go.mod h1:ZRw3GNz5cq4uYrW4TPSVyrYWaoqzujKdWro/AOcGBaE=
github.com/gogf/gf/contrib/registry/consul/v2 v2.9.5 h1:eUqwJ/qNH8lJ6yssiqskazgp1ACQuNU6zXlLOZVuXTQ=
github.com/gogf/gf/contrib/registry/consul/v2 v2.9.5/go.mod h1:sjQyMry9+0POYZCA6lHXBxO77WoNKkruJpRB4xKqk5k=
github.com/gogf/gf/contrib/trace/otlphttp/v2 v2.9.5 h1:tHUEZYB5GTqEYYVDYnlGobf1xISARKDE4KHVlgjwTec=
github.com/gogf/gf/contrib/trace/otlphttp/v2 v2.9.5/go.mod h1:cfzTn2HS9RDX8f5pUVkbGxUWcSosouqfNQ1G6cY0V88=
github.com/gogf/gf/v2 v2.10.2 h1:46IO0Uc8e85/FqdftJFskfDejJLBL0JBnGS5qOftUu8=
github.com/gogf/gf/v2 v2.10.2/go.mod h1:Svl1N+E8G/QshU2DUbh/3J/AJauqCgUnxHurXWR4Qx0=
github.com/gogo/protobuf v1.1.1/go.mod h1:r8qH/GZQm5c6nD/R0oafs1akxWv10x8SbQlK7atdtwQ=
service/asr/task_service.go
+40 -10
View File
@@ -53,8 +53,14 @@ func (s *audioTaskService) Create(ctx context.Context, params *CreateTaskParams)
params.Threshold = 0.3
}
inputBytes, _ := json.Marshal(params.InputData)
fnBytes, _ := json.Marshal(params.FileNames)
inputBytes, err := json.Marshal(params.InputData)
if err != nil {
return nil, fmt.Errorf("序列化输入数据失败: %v", err)
}
fnBytes, err := json.Marshal(params.FileNames)
if err != nil {
return nil, fmt.Errorf("序列化文件名列表失败: %v", err)
}
now := time.Now()
task := &entity.TranscribeTask{
@@ -229,18 +235,30 @@ func (s *audioTaskService) callback(ctx context.Context, taskID, status, errMsg,
DetailList: detailItems,
}
body, _ := json.Marshal(payload)
body, err := json.Marshal(payload)
if err != nil {
g.Log().Errorf(ctx, "[回调 %s] 序列化payload失败: %v", taskID, err)
return
}
g.Log().Infof(ctx, "[回调 %s] 触发回调, 状态=%s, 成功=%d 失败=%d, 错误=%s, 目标=%s",
taskID, taskInfo.Status, taskInfo.SuccessFiles, taskInfo.FailFiles, errMsg, callbackURL)
g.Log().Debugf(ctx, "[回调 %s] 回调载荷长度=%d字节, 明细条数=%d",
taskID, len(body), len(detailItems))
// 透传调用方的用户信息，供回调方 GetUserInfo 从 X-User-Info 头获取
cbUser := getUserFromCtx(ctx)
userJSON, _ := json.Marshal(cbUser)
userJSON, je := json.Marshal(cbUser)
if je != nil {
g.Log().Errorf(ctx, "[回调 %s] 序列化用户信息失败: %v", taskID, je)
return
}
g.Log().Infof(ctx, "[回调 %s] curl -X POST '%s' -H 'Content-Type: application/json' -H 'X-User-Info: %s' -d '%s'",
taskID, callbackURL, string(userJSON), strings.ReplaceAll(string(body), "'", "'\\''"))
req, _ := http.NewRequest("POST", callbackURL, bytes.NewReader(body))
req, reqErr := http.NewRequest("POST", callbackURL, bytes.NewReader(body))
if reqErr != nil {
g.Log().Errorf(ctx, "[回调 %s] 创建请求失败: %v", taskID, reqErr)
return
}
req.Header.Set("Content-Type", "application/json")
req.Header.Set("X-User-Info", string(userJSON))
@@ -251,7 +269,11 @@ func (s *audioTaskService) callback(ctx context.Context, taskID, status, errMsg,
}
defer resp.Body.Close()
respBody, _ := io.ReadAll(resp.Body)
respBody, readErr := io.ReadAll(resp.Body)
if readErr != nil {
g.Log().Errorf(ctx, "[回调 %s] 读取响应失败: %v", taskID, readErr)
return
}
g.Log().Infof(ctx, "[回调 %s] 响应 status=%d, body=%s", taskID, resp.StatusCode, string(respBody))
}
@@ -300,8 +322,12 @@ func (s *audioTaskService) processSingleVideo(ctx context.Context, taskID, saveP
var scenesJSON string
if scenes != nil {
sb, _ := json.Marshal(scenes)
scenesJSON = string(sb)
sb, je := json.Marshal(scenes)
if je != nil {
g.Log().Warningf(ctx, "[任务 %s] 序列化场景信息失败: %v", taskID, je)
} else {
scenesJSON = string(sb)
}
}
s.saveDetail(ctx, taskID, fileIndex, fileName,
@@ -414,8 +440,12 @@ func enrichDetailsFromResult(resultJSON string, details []dto.TranscribeTaskDeta
continue
}
if r.FileName == d.FileName {
sb, _ := json.Marshal(r.Result.Scenes)
details[i].Scenes = string(sb)
sb, je := json.Marshal(r.Result.Scenes)
if je != nil {
g.Log().Warningf(context.TODO(), "[回调] 序列化场景信息失败: %v", je)
} else {
details[i].Scenes = string(sb)
}
// 同时补全其他可能缺失的字段
if d.AudioDuration == "" {
details[i].AudioDuration = r.Result.AudioDuration
service/video/analysis_service.go
+27 -5
View File
@@ -137,7 +137,13 @@ func (s *analysisService) processAsyncTask(user *beans.User, taskID string, vide
g.Log().Infof(bgCtx, "[视频分析 %s] Caption接口调用成功: %s", taskID, videoURL)
// 3. 保存结果到数据库（视频不删除，永久保留）
captionJSON, _ := json.Marshal(captionResult)
captionJSON, je := json.Marshal(captionResult)
if je != nil {
g.Log().Errorf(bgCtx, "[视频分析 %s] 序列化结果失败: %v", taskID, je)
failedCount++
analysisDao.AnalysisTask.UpdateProgress(bgCtx, taskID, successCount, failedCount)
continue
}
analysisDao.AnalysisTaskDetail.UpdateSuccess(bgCtx, taskID, videoURL, savePath, string(captionJSON))
successCount++
analysisDao.AnalysisTask.UpdateProgress(bgCtx, taskID, successCount, failedCount)
@@ -364,14 +370,26 @@ func (s *analysisService) analysisCallback(ctx context.Context, taskID, callback
"results": results,
}
body, _ := json.Marshal(payload)
body, err := json.Marshal(payload)
if err != nil {
g.Log().Errorf(ctx, "[视频分析回调 %s] 序列化payload失败: %v", taskID, err)
return
}
g.Log().Infof(ctx, "[视频分析回调 %s] 状态=%s, 目标=%s", taskID, task.Status, callbackURL)
cbReq, _ := http.NewRequest("POST", callbackURL, bytes.NewReader(body))
cbReq, reqErr := http.NewRequest("POST", callbackURL, bytes.NewReader(body))
if reqErr != nil {
g.Log().Errorf(ctx, "[视频分析回调 %s] 创建请求失败: %v", taskID, reqErr)
return
}
cbReq.Header.Set("Content-Type", "application/json")
// 透传调用方用户信息
cbUser := getUserFromCtx(ctx)
userJSON, _ := json.Marshal(cbUser)
userJSON, je := json.Marshal(cbUser)
if je != nil {
g.Log().Errorf(ctx, "[视频分析回调 %s] 序列化用户信息失败: %v", taskID, je)
return
}
cbReq.Header.Set("X-User-Info", string(userJSON))
// 打印 curl 命令方便调试
@@ -386,6 +404,10 @@ func (s *analysisService) analysisCallback(ctx context.Context, taskID, callback
return
}
defer resp.Body.Close()
respBody, _ := io.ReadAll(resp.Body)
respBody, readErr := io.ReadAll(resp.Body)
if readErr != nil {
g.Log().Errorf(ctx, "[视频分析回调 %s] 读取响应失败: %v", taskID, readErr)
return
}
g.Log().Infof(ctx, "[视频分析回调 %s] 响应 status=%d, body=%s", taskID, resp.StatusCode, string(respBody))
}
service/video/caption_service.go
+28 -6
View File
@@ -46,8 +46,14 @@ func (s *captionService) CreateAsyncTask(ctx context.Context, videoURLs []string
subtitles = []dto.SubtitleSegment{}
}
videoURLsJSON, _ := json.Marshal(videoURLs)
elementsJSON, _ := json.Marshal(elements)
videoURLsJSON, err := json.Marshal(videoURLs)
if err != nil {
return "", fmt.Errorf("序列化视频URL失败: %v", err)
}
elementsJSON, err := json.Marshal(elements)
if err != nil {
return "", fmt.Errorf("序列化元素失败: %v", err)
}
taskID := "cap_" + guid.S()
task := &entity.VideoCaptionTask{
@@ -827,13 +833,25 @@ func (s *captionService) callback(ctx context.Context, taskID, callbackURL strin
}
}
body, _ := json.Marshal(payload)
body, err := json.Marshal(payload)
if err != nil {
g.Log().Errorf(ctx, "[字幕回调 %s] 序列化回调payload失败: %v", taskID, err)
return
}
g.Log().Infof(ctx, "[字幕回调 %s] 状态=%s, 目标=%s, body=%s", taskID, task.Status, callbackURL, string(body))
req, _ := http.NewRequest("POST", callbackURL, bytes.NewReader(body))
req, reqErr := http.NewRequest("POST", callbackURL, bytes.NewReader(body))
if reqErr != nil {
g.Log().Errorf(ctx, "[字幕回调 %s] 创建请求失败: %v", taskID, reqErr)
return
}
req.Header.Set("Content-Type", "application/json")
cbUser := getUserFromCtx(ctx)
userJSON, _ := json.Marshal(cbUser)
userJSON, je := json.Marshal(cbUser)
if je != nil {
g.Log().Errorf(ctx, "[字幕回调 %s] 序列化用户信息失败: %v", taskID, je)
return
}
req.Header.Set("X-User-Info", string(userJSON))
client := &http.Client{Timeout: 2 * time.Minute}
@@ -843,7 +861,11 @@ func (s *captionService) callback(ctx context.Context, taskID, callbackURL strin
return
}
defer resp.Body.Close()
respBody, _ := io.ReadAll(resp.Body)
respBody, readErr := io.ReadAll(resp.Body)
if readErr != nil {
g.Log().Errorf(ctx, "[字幕回调 %s] 读取响应失败: %v", taskID, readErr)
return
}
g.Log().Infof(ctx, "[字幕回调 %s] 响应 status=%d, body=%s", taskID, resp.StatusCode, string(respBody))
}
service/video/concat_service.go
+24 -5
View File
@@ -702,14 +702,26 @@ func (s *concatService) concatCallback(ctx context.Context, taskID, callbackURL
payload["errorMessage"] = task.ErrorMessage
}
body, _ := json.Marshal(payload)
body, err := json.Marshal(payload)
if err != nil {
g.Log().Errorf(ctx, "[异步拼接回调 %s] 序列化payload失败: %v", taskID, err)
return
}
g.Log().Infof(ctx, "[异步拼接回调 %s] 状态=%s, 目标=%s", taskID, task.Status, callbackURL)
req, _ := http.NewRequest("POST", callbackURL, bytes.NewReader(body))
req, reqErr := http.NewRequest("POST", callbackURL, bytes.NewReader(body))
if reqErr != nil {
g.Log().Errorf(ctx, "[异步拼接回调 %s] 创建请求失败: %v", taskID, reqErr)
return
}
req.Header.Set("Content-Type", "application/json")
// 透传调用方用户信息
cbUser := getUserFromCtx(ctx)
userJSON, _ := json.Marshal(cbUser)
userJSON, je := json.Marshal(cbUser)
if je != nil {
g.Log().Errorf(ctx, "[异步拼接回调 %s] 序列化用户信息失败: %v", taskID, je)
return
}
req.Header.Set("X-User-Info", string(userJSON))
client := &http.Client{Timeout: 2 * time.Minute}
@@ -719,7 +731,11 @@ func (s *concatService) concatCallback(ctx context.Context, taskID, callbackURL
return
}
defer resp.Body.Close()
respBody, _ := io.ReadAll(resp.Body)
respBody, readErr := io.ReadAll(resp.Body)
if readErr != nil {
g.Log().Errorf(ctx, "[异步拼接回调 %s] 读取响应失败: %v", taskID, readErr)
return
}
g.Log().Infof(ctx, "[异步拼接回调 %s] 响应 status=%d, body=%s", taskID, resp.StatusCode, string(respBody))
}
@@ -757,7 +773,10 @@ func downloadFile(ctx context.Context, rawURL, tempDir string) (string, error) {
if resp.StatusCode != http.StatusOK {
// 读取响应体用于错误诊断
bodyBytes, _ := io.ReadAll(io.LimitReader(resp.Body, 1024))
bodyBytes, readErr := io.ReadAll(io.LimitReader(resp.Body, 1024))
if readErr != nil {
g.Log().Warningf(ctx, "[下载] 读取HTTP错误响应失败: %v", readErr)
}
// 尝试检测签名 URL 是否过期
if parsedURL.Query().Get("x-expires") != "" || parsedURL.Query().Get("expires") != "" {
return "", fmt.Errorf("HTTP %d (签名URL可能已过期，请使用新鲜地址) body=%s", resp.StatusCode, string(bodyBytes))
service/video/cut_service.go
+26 -6
View File
@@ -442,8 +442,12 @@ func (s *cutService) UploadToMinIO(ctx context.Context, localFilePath string) (*
// 无 HTTP 请求时（异步 goroutine），从 context 的用户信息构造 header
if !hasAuthHeader {
uploadUser := getUserFromCtx(ctx)
userJSON, _ := json.Marshal(uploadUser)
client.SetHeader("X-User-Info", string(userJSON))
userJSON, je := json.Marshal(uploadUser)
if je != nil {
g.Log().Errorf(ctx, "[UploadToMinIO] 序列化用户信息失败: %v", je)
} else {
client.SetHeader("X-User-Info", string(userJSON))
}
}
// 设置 multipart Content-Type（含 boundary）
@@ -650,14 +654,26 @@ func (s *cutService) cutCallback(ctx context.Context, taskID, callbackURL string
payload["errorMessage"] = task.ErrorMessage
}
body, _ := json.Marshal(payload)
body, err := json.Marshal(payload)
if err != nil {
g.Log().Errorf(ctx, "[异步剪切回调 %s] 序列化payload失败: %v", taskID, err)
return
}
g.Log().Infof(ctx, "[异步剪切回调 %s] 状态=%s, 目标=%s", taskID, task.Status, callbackURL)
req, _ := http.NewRequest("POST", callbackURL, bytes.NewReader(body))
req, reqErr := http.NewRequest("POST", callbackURL, bytes.NewReader(body))
if reqErr != nil {
g.Log().Errorf(ctx, "[异步剪切回调 %s] 创建请求失败: %v", taskID, reqErr)
return
}
req.Header.Set("Content-Type", "application/json")
// 透传调用方用户信息
cbUser := getUserFromCtx(ctx)
userJSON, _ := json.Marshal(cbUser)
userJSON, je := json.Marshal(cbUser)
if je != nil {
g.Log().Errorf(ctx, "[异步剪切回调 %s] 序列化用户信息失败: %v", taskID, je)
return
}
req.Header.Set("X-User-Info", string(userJSON))
client := &http.Client{Timeout: 2 * time.Minute}
@@ -667,6 +683,10 @@ func (s *cutService) cutCallback(ctx context.Context, taskID, callbackURL string
return
}
defer resp.Body.Close()
respBody, _ := io.ReadAll(resp.Body)
respBody, readErr := io.ReadAll(resp.Body)
if readErr != nil {
g.Log().Errorf(ctx, "[异步剪切回调 %s] 读取响应失败: %v", taskID, readErr)
return
}
g.Log().Infof(ctx, "[异步剪切回调 %s] 响应 status=%d, body=%s", taskID, resp.StatusCode, string(respBody))
}
service/video/merge_service.go
+34 -8
View File
@@ -47,8 +47,14 @@ func (s *mergeService) CreateAsyncTask(ctx context.Context, videoURLs, audioURLs
}
// 将 videoURLs/audioURLs 序列化为 JSON 存入数据库
videoURLsJSON, _ := json.Marshal(videoURLs)
audioURLsJSON, _ := json.Marshal(audioURLs)
videoURLsJSON, err := json.Marshal(videoURLs)
if err != nil {
return "", fmt.Errorf("序列化视频URL失败: %v", err)
}
audioURLsJSON, err := json.Marshal(audioURLs)
if err != nil {
return "", fmt.Errorf("序列化音频URL失败: %v", err)
}
taskID := "merge_" + guid.S()
task := &entity.VideoAudioMergeTask{
@@ -314,13 +320,25 @@ func (s *mergeService) callback(ctx context.Context, taskID, callbackURL string)
payload["errorMessage"] = task.ErrorMessage
}
body, _ := json.Marshal(payload)
body, err := json.Marshal(payload)
if err != nil {
g.Log().Errorf(ctx, "[拼接混音回调 %s] 序列化payload失败: %v", taskID, err)
return
}
g.Log().Infof(ctx, "[拼接混音回调 %s] 状态=%s, 目标=%s", taskID, task.Status, callbackURL)
req, _ := http.NewRequest("POST", callbackURL, bytes.NewReader(body))
req, reqErr := http.NewRequest("POST", callbackURL, bytes.NewReader(body))
if reqErr != nil {
g.Log().Errorf(ctx, "[拼接混音回调 %s] 创建请求失败: %v", taskID, reqErr)
return
}
req.Header.Set("Content-Type", "application/json")
cbUser := getUserFromCtx(ctx)
userJSON, _ := json.Marshal(cbUser)
userJSON, je := json.Marshal(cbUser)
if je != nil {
g.Log().Errorf(ctx, "[拼接混音回调 %s] 序列化用户信息失败: %v", taskID, je)
return
}
req.Header.Set("X-User-Info", string(userJSON))
client := &http.Client{Timeout: 2 * time.Minute}
@@ -330,7 +348,11 @@ func (s *mergeService) callback(ctx context.Context, taskID, callbackURL string)
return
}
defer resp.Body.Close()
respBody, _ := io.ReadAll(resp.Body)
respBody, readErr := io.ReadAll(resp.Body)
if readErr != nil {
g.Log().Errorf(ctx, "[拼接混音回调 %s] 读取响应失败: %v", taskID, readErr)
return
}
g.Log().Infof(ctx, "[拼接混音回调 %s] 响应 status=%d, body=%s", taskID, resp.StatusCode, string(respBody))
}
@@ -373,8 +395,12 @@ func uploadToMinIO(ctx context.Context, localFilePath string) (*uploadFileRes, e
}
if !hasAuthHeader {
uploadUser := getUserFromCtx(ctx)
userJSON, _ := json.Marshal(uploadUser)
client.SetHeader("X-User-Info", string(userJSON))
userJSON, je := json.Marshal(uploadUser)
if je != nil {
g.Log().Errorf(ctx, "[上传MinIO] 序列化用户信息失败: %v", je)
} else {
client.SetHeader("X-User-Info", string(userJSON))
}
}
contentType := mw.FormDataContentType()
service/video/scene_split_service.go
+26 -6
View File
@@ -190,8 +190,12 @@ func (s *sceneSplitService) processTask(user *beans.User, taskID, videoURL strin
}
// 9. 更新数据库为成功
segmentsJSON, _ := json.Marshal(segments)
dao.SceneSplitTask.UpdateSuccess(bgCtx, taskID, string(segmentsJSON), audioURL, len(segments), audioDuration, videoDuration)
segmentsJSON, je := json.Marshal(segments)
if je != nil {
g.Log().Errorf(bgCtx, "[场景分割 %s] 序列化分片信息失败: %v", taskID, je)
} else {
dao.SceneSplitTask.UpdateSuccess(bgCtx, taskID, string(segmentsJSON), audioURL, len(segments), audioDuration, videoDuration)
}
g.Log().Infof(bgCtx, "[场景分割 %s] 完成! 分片数=%d, 音频=%s", taskID, len(segments), audioURL)
@@ -427,15 +431,27 @@ func (s *sceneSplitService) callback(ctx context.Context, taskID, callbackURL st
payload["errorMessage"] = task.ErrorMessage
}
body, _ := json.Marshal(payload)
body, err := json.Marshal(payload)
if err != nil {
g.Log().Errorf(ctx, "[场景分割回调 %s] 序列化payload失败: %v", taskID, err)
return
}
cbUser := getUserFromCtx(ctx)
userJSON, _ := json.Marshal(cbUser)
userJSON, je := json.Marshal(cbUser)
if je != nil {
g.Log().Errorf(ctx, "[场景分割回调 %s] 序列化用户信息失败: %v", taskID, je)
return
}
g.Log().Infof(ctx, "[场景分割回调 %s] 状态=%s, 目标=%s", taskID, task.Status, callbackURL)
g.Log().Infof(ctx, "[场景分割回调 %s] curl: curl -X POST '%s' -H 'Content-Type: application/json' -H 'X-User-Info: %s' -d '%s'",
taskID, callbackURL, string(userJSON), string(body))
req, _ := http.NewRequest("POST", callbackURL, bytes.NewReader(body))
req, reqErr := http.NewRequest("POST", callbackURL, bytes.NewReader(body))
if reqErr != nil {
g.Log().Errorf(ctx, "[场景分割回调 %s] 创建请求失败: %v", taskID, reqErr)
return
}
req.Header.Set("Content-Type", "application/json")
req.Header.Set("X-User-Info", string(userJSON))
@@ -446,6 +462,10 @@ func (s *sceneSplitService) callback(ctx context.Context, taskID, callbackURL st
return
}
defer resp.Body.Close()
respBody, _ := io.ReadAll(resp.Body)
respBody, readErr := io.ReadAll(resp.Body)
if readErr != nil {
g.Log().Errorf(ctx, "[场景分割回调 %s] 读取响应失败: %v", taskID, readErr)
return
}
g.Log().Infof(ctx, "[场景分割回调 %s] 响应 status=%d, body=%s", taskID, resp.StatusCode, string(respBody))
}