ai-agent/workflow/service/flow/processor/builtin/media/media.go

package media

import (
	"ai-agent/workflow/service/flow/processor"
	"context"
	"fmt"
	"strings"
	"time"

	commonHttp "gitea.redpowerfuture.com/red-future/common/http"
	"gitea.redpowerfuture.com/red-future/common/oss"
	"gitea.redpowerfuture.com/red-future/common/utils"
	"github.com/gogf/gf/v2/frame/g"
	"github.com/gogf/gf/v2/util/gconv"
)

// TaskKind 媒体任务类型（拼接/拼接+混音），决定提交与查询的接口路径
type TaskKind string

const (
	TaskKindConcat TaskKind = "concat" // 纯拼接，无 BGM
	TaskKindMerge  TaskKind = "merge"  // 拼接+混音，有 BGM
)

// MergeTask media 服务异步任务状态
type MergeTask struct {
	TaskID       string `json:"taskId"`
	Status       string `json:"status"` // pending/running/success/failed
	FileURL      string `json:"fileURL,omitempty"`
	ErrorMessage string `json:"errorMessage,omitempty"`
	DurationStr  string `json:"durationStr,omitempty"`
}

type mergeSubmitReq struct {
	VideoURLs   []string `json:"video_urls"`
	AudioURLs   []string `json:"audio_urls,omitempty"`
	Method      string   `json:"method,omitempty"`
	Upload      bool     `json:"upload"`
	CallbackURL string   `json:"callback_url"`
}

type mergeSubmitRes struct {
	TaskID string `json:"taskId"`
}

// ProcessorName 处理器注册名，供模型节点前置工具分发按名判定入参形态（扁平请求体）。
const ProcessorName = "concat_videos"

func init() {
	processor.Register(ConcatVideosProcessor())
}

// ConcatVideosProcessor 合并视频
func ConcatVideosProcessor() *processor.Processor {
	return &processor.Processor{
		Name:        ProcessorName,
		Description: "合并视频",
		IsShow:      false,
		Func: func(ctx context.Context, args map[string]any) (any, error) {
			outputRes := parseOutputList(args)
			segments, err := collectSegmentResults(ctx, outputRes)
			if err != nil {
				return nil, err
			}

			videoURLs := make([]string, 0, len(segments))
			for _, seg := range segments {
				videoURLs = append(videoURLs, seg.VideoURL)
			}

			reqParams := gconv.Map(args["request"])
			bgmURLs := gconv.Strings(reqParams["bgm_urls"])
			if len(bgmURLs) == 0 {
				// 兼容串行结果把 bgm 带回 output 的情况
				for _, m := range outputRes {
					bgmURLs = append(bgmURLs, gconv.Strings(m["bgm_urls"])...)
				}
			}
			upload := gconv.Bool(reqParams["upload"])
			callback := gconv.String(reqParams["callback_url"])

			merged, err := MergeSegments(ctx, videoURLs, bgmURLs, upload, callback, 30*time.Minute)
			if err != nil {
				return nil, err
			}
			// 合并结果沿用输入视频的 key 返回，保持 key 不变；
			// 否则下游按原 key 引用（值来源/保存文件映射）会失配
			retKey := "fileURL"
			if len(outputRes) > 0 {
				if k := findVideoKey(outputRes[0]); k != "" {
					retKey = k
				}
			}
			return map[string]any{
				retKey: merged.FileURL,
			}, nil
		},
	}
}

// parseOutputList 兼容不同反序列化形态的 output 列表
func parseOutputList(args map[string]any) []map[string]any {
	switch v := args["output"].(type) {
	case []map[string]any:
		return v
	case []any:
		out := make([]map[string]any, 0, len(v))
		for _, it := range v {
			if m, ok := it.(map[string]any); ok {
				out = append(out, m)
			} else if m := gconv.Map(it); m != nil {
				out = append(out, m)
			}
		}
		return out
	}
	return nil
}

// MergeSegments 把分段视频按序合并：有 BGM 走拼接+混音（merge），否则纯拼接（concat）。
// 返回最终合并结果（FileURL）。
func MergeSegments(ctx context.Context, videoURLs, bgmURLs []string, upload bool, callbackURL string, timeout time.Duration) (*MergeTask, error) {
	var kind TaskKind
	var taskID string
	var err error
	if len(bgmURLs) > 0 {
		kind = TaskKindMerge
		taskID, err = SubmitMergeAsync(ctx, videoURLs, bgmURLs, upload, callbackURL)
	} else {
		kind = TaskKindConcat
		taskID, err = SubmitConcatAsync(ctx, videoURLs, upload, callbackURL)
	}
	if err != nil {
		return nil, err
	}
	return WaitMediaTask(ctx, kind, taskID, timeout)
}

// SubmitConcatAsync 提交纯拼接异步任务
func SubmitConcatAsync(ctx context.Context, videoURLs []string, upload bool, callbackURL string) (string, error) {
	return submitMediaTask(ctx, TaskKindConcat, &mergeSubmitReq{
		VideoURLs:   videoURLs,
		Method:      "auto",
		Upload:      upload,
		CallbackURL: callbackURL,
	})
}

// SubmitMergeAsync 提交拼接+混音异步任务
func SubmitMergeAsync(ctx context.Context, videoURLs, audioURLs []string, upload bool, callbackURL string) (string, error) {
	return submitMediaTask(ctx, TaskKindMerge, &mergeSubmitReq{
		VideoURLs:   videoURLs,
		AudioURLs:   audioURLs,
		Upload:      upload,
		CallbackURL: callbackURL,
	})
}

func submitMediaTask(ctx context.Context, kind TaskKind, req *mergeSubmitReq) (string, error) {
	path := "media/video/" + string(kind) + "/async"
	res := new(mergeSubmitRes)
	if err := commonHttp.Post(ctx, path, utils.HeadersFromCtx(ctx), res, req); err != nil {
		return "", fmt.Errorf("提交%s任务失败: %v", kind, err)
	}
	if res.TaskID == "" {
		return "", fmt.Errorf("media 返回空 taskId")
	}
	return res.TaskID, nil
}

// WaitMediaTask 轮询媒体任务直到 success/failed，或超时。
func WaitMediaTask(ctx context.Context, kind TaskKind, taskID string, timeout time.Duration) (*MergeTask, error) {
	deadline := time.Now().Add(timeout)
	ticker := time.NewTicker(15 * time.Second)
	defer ticker.Stop()
	for {
		t, err := getMediaTask(ctx, kind, taskID)
		if err == nil {
			switch t.Status {
			case "success":
				if t.FileURL == "" {
					return nil, fmt.Errorf("%s任务成功但未返回文件URL", kind)
				}
				return t, nil
			case "failed":
				return nil, fmt.Errorf("%s任务失败: %s", kind, t.ErrorMessage)
			}
		}
		if time.Now().After(deadline) {
			return nil, fmt.Errorf("%s任务[%s]超时", kind, taskID)
		}
		select {
		case <-ctx.Done():
			return nil, ctx.Err()
		case <-ticker.C:
		}
	}
}

func getMediaTask(ctx context.Context, kind TaskKind, taskID string) (*MergeTask, error) {
	path := "media/video/" + string(kind) + "/task/" + taskID
	res := new(MergeTask)
	if err := commonHttp.Get(ctx, path, utils.HeadersFromCtx(ctx), res); err != nil {
		return nil, err
	}
	return res, nil
}

// segmentResult 一段视频生成的产出（原 ai-agent/video/plan.SegmentResult，plan 包已并入处理器树）。
type segmentResult struct {
	SegmentIndex int    `json:"segment_index"`
	VideoURL     string `json:"video_url"`
	Duration     int    `json:"duration"`
}

// collectSegmentResults 把模型节点/串行工具的产出（[]map[string]any）收敛为有序的分段结果列表。
// 兼容两种形状：串行工具产出的 {segment_index, video_url, duration}；
// 并行模型调用产出的 {<url字段>: url}（按列表顺序对应各段）。
func collectSegmentResults(ctx context.Context, outputRes []map[string]any) ([]segmentResult, error) {
	if len(outputRes) == 0 {
		return nil, fmt.Errorf("没有可合并的分段视频")
	}
	var segs []segmentResult
	for i, m := range outputRes {
		seg := segmentResult{
			SegmentIndex: i,
			Duration:     gconv.Int(m["duration"]),
			VideoURL:     findVideoURL(ctx, m),
		}
		if idx := gconv.Int(m["segment_index"]); len(outputRes) > 1 && idx > 0 {
			seg.SegmentIndex = idx
		}
		if seg.VideoURL == "" {
			return nil, fmt.Errorf("第 %d 段未获取到视频URL", i)
		}
		segs = append(segs, seg)
	}
	return segs, nil
}

// findVideoURL 从模型返回参数中提取视频 URL：优先命中常见键，再兼容扁平点号键（content.attrs.video_url 等）任意含 url 的字段。
func findVideoURL(ctx context.Context, params map[string]any) string {
	key := findVideoKey(params)
	if key == "" {
		return ""
	}
	return normalizeVideoURL(ctx, gconv.String(params[key]))
}

// findVideoKey 返回视频 URL 所在字段的 key（命中规则与 findVideoURL 一致），
// 供视频合并后以原 key 返回结果，避免下游按原 key 引用（值来源/保存文件映射）失配。
func findVideoKey(params map[string]any) string {
	if params == nil {
		return ""
	}
	for _, key := range []string{"video_url", "video_oss_url", "http_file_url", "file_url", "url"} {
		if gconv.String(params[key]) != "" {
			return key
		}
	}
	// 兼容扁平点号键（如 content.attrs.video_url）：优先命中含 video 的键，再兜底任意含 url 的键
	var fallback string
	for k, v := range params {
		if !strings.Contains(strings.ToLower(k), "url") {
			continue
		}
		if gconv.String(v) == "" {
			continue
		}
		if strings.Contains(strings.ToLower(k), "video") {
			return k
		}
		if fallback == "" {
			fallback = k
		}
	}
	return fallback
}

// normalizeVideoURL 统一视频地址：已是完整 http(s) 链接原样返回；相对路径（MinIO 对象路径）补上文件前缀，供 media 服务下载
func normalizeVideoURL(ctx context.Context, url string) string {
	if url == "" || strings.HasPrefix(url, "http") {
		return url
	}
	prefix, err := oss.GetFileAddressPrefix(ctx)
	if err != nil {
		g.Log().Warningf(ctx, "获取文件前缀失败，视频地址保持相对路径: %s err=%v", url, err)
		return url
	}
	return prefix + url
}

// FindVideoKey 返回视频 URL 所在字段的 key（规则同 findVideoKey），供工作流段级续跑重建输出记录保持 key 一致
func FindVideoKey(params map[string]any) string {
	return findVideoKey(params)
}

// FindVideoURL 从模型返回参数中提取视频 URL（规则同 findVideoURL），供段级续跑落库
func FindVideoURL(ctx context.Context, params map[string]any) string {
	return findVideoURL(ctx, params)
}