ai-agent/workflow/service/flow/lambda_node.go

package flow

import (
	"ai-agent/gateway"
	"ai-agent/workflow/consts/flow"
	"ai-agent/workflow/consts/model"
	"ai-agent/workflow/consts/node"
	"ai-agent/workflow/consts/public"
	flowDao "ai-agent/workflow/dao/flow"
	nodeDao "ai-agent/workflow/dao/node"
	sessionDao "ai-agent/workflow/dao/session"
	flowDto "ai-agent/workflow/model/dto/flow"
	nodeDto "ai-agent/workflow/model/dto/node"
	sessionDto "ai-agent/workflow/model/dto/session"
	"ai-agent/workflow/model/entity"
	"ai-agent/workflow/service/flow/processor"
	"ai-agent/workflow/service/flow/processor/builtin/media"
	"context"
	"encoding/base64"
	"encoding/json"
	"fmt"
	"strings"
	"sync"

	"gitea.redpowerfuture.com/red-future/common/db/gfdb"
	"gitea.redpowerfuture.com/red-future/common/utils"
	"github.com/cloudwego/eino-examples/compose/batch/batch"
	"github.com/cloudwego/eino/compose"
	"github.com/gogf/gf/v2/database/gdb"
	"github.com/gogf/gf/v2/frame/g"
	"github.com/gogf/gf/v2/util/gconv"
	"github.com/google/uuid"
)

// StartLambda 启动节点
func StartLambda(ctx context.Context, input any) (any, error) {
	return input, nil
}

// FormLambda 表单调用节点
func FormLambda(ctx context.Context, input any) (any, error) {
	nodeInput, ok := input.(*flowDto.NodeExecutionInput)
	if !ok {
		return nil, fmt.Errorf("入参类型错误")
	}
	// 解析 valueSource 引用，填充表单节点输出配置（供下游引用）
	for _, output := range nodeInput.Config.OutputConfig {
		ProcessValueSourceRecursive(output, nodeInput.Global)
	}
	return nodeInput, nil
}

// ModelLambda 模型调用节点
func ModelLambda(ctx context.Context, input any) (any, error) {
	nodeInput, ok := input.(*flowDto.NodeExecutionInput)
	if !ok {
		return nil, fmt.Errorf("入参类型错误")
	}

	modelParams, err := BuildModelRequestBody(nodeInput.Config.ModelConfig.ModelRequestParamsPath, nodeInput.Global)
	if err != nil {
		return nil, err
	}

	// 2. 前置工具：决定模型调用入参（单次/多次）
	// 入参统一为扁平模型请求体（BuildModelRequestBody 输出，key 为点分路径）。
	// 分批处理器按默认上限拆分集合字段，其余前置工具（如 split_shots_pipeline）读取扁平参数。
	preToolParams := modelParams
	paramsList, err := invokePreTool(ctx, nodeInput.Config.PreTool, preToolParams)
	if err != nil {
		return nil, err
	}

	// 3. 逐批调用模型，汇总输出（保持请求顺序），累计 token/费用供节点记录落库
	var outputRes []map[string]any
	var totalTokens int64
	var totalCost float64
	if len(paramsList) > 1 {
		// 段级续跑仅在"多段 + 视频模型"启用；非视频分段（批量文本等）走原逻辑零影响。
		// 段身份 = 列表位置（0-based）：paramsList 顺序即段序，concat 按列表顺序拼接；
		// 位置互不重复且跨 reExecute 稳定（参数一致 → 段数/顺序不变）。不依赖 params 里的
		// segment_index——真实链路（上游 split_shots_pipeline 转写 → 下游 split_segment 按
		// __segment_fields 拆分，invokePreTool 剥离 __ 内部键）下 paramsList 只有模型参数。
		// segVideo=false 时走既有非段级合并路径（全量生成、不落库、不复用），全新执行行为不变，
		// 后续自动 concat 判断（独立的 isVideoModel 调用）仍正常执行。
		segVideo := isVideoModel(ctx, nodeInput.Config.ModelConfig.ModelId)

		// 续跑（!ForceNewRun）时读取该节点已成功段；全新执行不查（BuildExecution 已清旧段），saved 为 nil → 全量重生成
		var saved map[int]entity.SegmentRef
		if !nodeInput.Global.ForceNewRun && segVideo {
			saved, err = flowDao.FlowSegmentResultDao.ListByNode(ctx, nodeInput.Global.ExecutionId, nodeInput.Config.Id)
			if err != nil {
				return nil, err
			}
		}

		idxList, needGen := planSegmentResume(paramsList, saved)

		results := make([][]map[string]any, len(paramsList))
		tokenRes := make([]*gateway.ModelCallRes, len(paramsList))
		errs := make([]error, len(paramsList))
		saveErrs := make([]error, len(paramsList))
		isInference := make([]bool, len(paramsList))
		var wg sync.WaitGroup
		for i, params := range paramsList {
			if !needGen[i] {
				continue
			}
			wg.Add(1)
			go func(i int, params map[string]any) {
				defer wg.Done()
				// 视频段每段失败自动重试 1 次（共 2 次尝试）；非视频保持单次调用
				for attempt := 0; attempt < segmentGenerateMaxAttempts; attempt++ {
					results[i], tokenRes[i], isInference[i], errs[i] = ModelCallResultLambda(ctx, nodeInput.Config.ModelConfig.ModelId, nodeInput.Global.SessionId, params, nodeInput.Config.Prompt)
					if errs[i] == nil || !segVideo {
						break
					}
				}
				// 每段成功立即落库：该段刚成功即持久化，其他段仍在跑/重试时已成功段也不丢；
				// 后续段失败或进程崩溃（panic/OOM/kill）时，已完成段已在库中，reExecute 可直接复用
				if segVideo && errs[i] == nil {
					for _, rec := range results[i] {
						key := media.FindVideoKey(rec)
						url := media.FindVideoURL(ctx, rec)
						if key == "" || url == "" {
							continue
						}
						if err := flowDao.FlowSegmentResultDao.Save(ctx, nodeInput.Global.ExecutionId, nodeInput.Config.Id, idxList[i], key, url); err != nil {
							saveErrs[i] = err
						}
					}
				}
			}(i, params)
		}
		wg.Wait()

		// 仍有失败段或落库失败 → 节点失败（成功段已立即落库，供下次 reExecute 复用）
		for i := range results {
			if saveErrs[i] != nil {
				return nil, saveErrs[i]
			}
			if needGen[i] && errs[i] != nil {
				return nil, errs[i]
			}
			if needGen[i] && tokenRes[i] != nil {
				totalTokens += tokenRes[i].TotalTokens
				totalCost += tokenRes[i].Cost
			}
		}

		if segVideo {
			// 复用段 + 新生段按段序号升序合并，concat 按列表顺序拼接 → 顺序保证
			outputRes = mergeSegmentOutputs(idxList, needGen, results, saved)
		} else {
			if isInference[0] {
				outputRes = mergeInferenceBatchResults(results)
			} else {
				for _, res := range results {
					outputRes = append(outputRes, res...)
				}
			}
		}
	} else {
		for _, params := range paramsList {
			res, modelRes, _, err := ModelCallResultLambda(ctx, nodeInput.Config.ModelConfig.ModelId, nodeInput.Global.SessionId, params, nodeInput.Config.Prompt)
			if err != nil {
				return nil, err
			}
			if modelRes != nil {
				totalTokens += modelRes.TotalTokens
				totalCost += modelRes.Cost
			}
			outputRes = append(outputRes, res...)
		}
	}

	// 3.5 把本次节点消耗的 token/费用写入节点执行记录，供汇总节点聚合到 exec_workflow
	if nodeInput.NodeExecutionId > 0 && (totalTokens > 0 || totalCost > 0) {
		if _, err = nodeDao.NodeExecutionDao.Update(ctx, &nodeDto.UpdateNodeExecutionReq{
			Id: nodeInput.NodeExecutionId,
			TokenInfo: []map[string]any{{
				"total_tokens": totalTokens,
				"total_fee":    totalCost,
			}},
		}); err != nil {
			return nil, fmt.Errorf("节点：%v 写入token信息失败: %v", nodeInput.Config.Name, err)
		}
	}

	// 4.5 视频模型节点返回多个视频时，自动调用视频合成工具（concat_videos）合并为单条；
	// 已显式配置 concat_videos 后置工具时跳过，避免重复合并
	if nodeInput.Config.PostTool != media.ProcessorName && len(outputRes) > 1 && isVideoModel(ctx, nodeInput.Config.ModelConfig.ModelId) {
		g.Log().Debugf(ctx, "modelId1:%v ,outputRes: %v", nodeInput.Config.ModelConfig.ModelId, outputRes)
		outputRes, err = invokePostTool(ctx, media.ProcessorName, outputRes, map[string]any{"callback_url": "callback_url", "upload": true})
		g.Log().Debugf(ctx, "modelId2:%v ,outputRes: %v", nodeInput.Config.ModelConfig.ModelId, outputRes)
		if err != nil {
			return nil, err
		}
	} else {
		// 4. 后置工具：加工模型输出（透传原始请求参数，供后置工具读取合并配置等）
		outputRes, err = invokePostTool(ctx, nodeInput.Config.PostTool, outputRes, modelParams)
		if err != nil {
			return nil, err
		}
	}
	g.Log().Debugf(ctx, "modelId3:%v ,outputRes: %v", nodeInput.Config.ModelConfig.ModelId, outputRes)
	nodeInput.Config.OutputResult = outputRes
	return nodeInput, nil
}

// isVideoModel 判断模型是否为视频模型（模型类型 TypeVideo=600），用于视频节点多视频自动合成判断
func isVideoModel(ctx context.Context, modelId int64) bool {
	modelInfo, err := gateway.GetModelInfoById(ctx, &gateway.GetModelInfoByIdReq{ModelId: modelId})
	if err != nil {
		g.Log().Warningf(ctx, "查询模型配置失败，跳过自动视频合成 modelId=%d err=%v", modelId, err)
		return false
	}
	return modelInfo.ModelManage.ModelType != nil && *modelInfo.ModelManage.ModelType == model.TypeVideo
}

// mergeInferenceBatchResults 推理模型分批结果拼接为单条输出记录：
// 各批结果按批序对同名 key 的值做字符串拼接（"拼到一个字段"），最终返回单条 {key:值} 记录。
// 非字符串值（如结构/数组字段）取最后一份，避免误拼接。
func mergeInferenceBatchResults(results [][]map[string]any) []map[string]any {
	merged := make(map[string]any)
	for _, res := range results {
		for _, record := range res {
			for key, val := range record {
				prev, has := merged[key]
				if !has {
					merged[key] = val
					continue
				}
				sPrev, pOK := prev.(string)
				sVal, vOK := val.(string)
				if pOK && vOK {
					merged[key] = sPrev + "\n" + sVal
					continue
				}
				merged[key] = val
			}
		}
	}
	return []map[string]any{merged}
}

// invokePreTool 执行前置处理器，把模型请求参数转换为模型调用入参列表。
// 前置处理器契约：入参即模型请求参数本体；返回值：
//   - map[string]any   一次模型调用，入参为返回值
//   - []map[string]any 多次模型调用，逐个入参请求
//   - nil              视为异常，节点失败（不允许静默跳过模型调用）
func invokePreTool(ctx context.Context, processorName string, modelParams map[string]any) (paramsList []map[string]any, err error) {
	if processorName == "" {
		return []map[string]any{stripInternalKeys(modelParams)}, nil
	}
	data, err := processor.Call(ctx, processorName, modelParams)
	if err != nil {
		return nil, fmt.Errorf("执行前置处理器[%s]失败: %v", processorName, err)
	}
	switch v := data.(type) {
	case nil:
		return nil, fmt.Errorf("前置处理器[%s]返回空", processorName)
	case map[string]any:
		return []map[string]any{stripInternalKeys(v)}, nil
	case []map[string]any:
		list := make([]map[string]any, 0, len(v))
		for _, m := range v {
			list = append(list, stripInternalKeys(m))
		}
		return list, nil
	default:
		return nil, fmt.Errorf("前置处理器[%s]返回类型不支持: %T", processorName, data)
	}
}

// stripInternalKeys 剥离 __ 前缀的内部键（如 __segment_fields/__produced），
// 模型网关做参数严格校验（CheckParams strictUnknown）会拒绝未知字段，内部标记不得随请求体下发。
func stripInternalKeys(params map[string]any) map[string]any {
	if params == nil {
		return params
	}
	for k := range params {
		if strings.HasPrefix(k, "__") {
			delete(params, k)
		}
	}
	return params
}

// invokePostTool 执行后置处理器，加工模型调用结果。
// 后置处理器契约：入参 {"output": 模型输出结果列表, "request": 原始模型请求参数}（列表须包成对象传入）；返回值：
//   - []map[string]any 替换模型输出
//   - map[string]any   替换为单条输出
//   - nil              保留原输出
func invokePostTool(ctx context.Context, processorName string, outputRes []map[string]any, requestParams map[string]any) ([]map[string]any, error) {
	if processorName == "" {
		return outputRes, nil
	}
	data, err := processor.Call(ctx, processorName, map[string]any{"output": outputRes, "request": requestParams})
	if err != nil {
		return nil, fmt.Errorf("执行后置处理器[%s]失败: %v", processorName, err)
	}
	switch v := data.(type) {
	case nil:
		return outputRes, nil
	case []map[string]any:
		return v, nil
	case map[string]any:
		return []map[string]any{v}, nil
	default:
		return nil, fmt.Errorf("后置处理器[%s]返回类型不支持: %T", processorName, data)
	}
}

func SubFlowLambda(ctx context.Context, input any) (any, error) {
	// 1. 类型断言（和其他节点保持一致的入参结构）
	nodeExecInput, ok := input.(*flowDto.NodeExecutionInput)
	if !ok {
		return nil, fmt.Errorf("子流程节点入参类型错误，期望*flowDto.NodeExecutionInput，实际%T", input)
	}
	// 2. 解析子流程配置
	subFlowConfig := nodeExecInput.Config.SubConfig
	if subFlowConfig == nil {
		return nil, fmt.Errorf("子流程节点缺少配置")
	}
	getRes, err := FlowUserService.Get(ctx, &flowDto.GetFlowUserReq{
		Id: subFlowConfig.WorkflowId,
	})
	if err != nil {
		return nil, err
	}
	// 3. 引入参数解析：把首页表单值/上游引用值/静态默认值写入子流程开始节点 outputConfig。
	//    须在 BuildGraph / ExtractFlowNodeFrom 之前执行，batchInputs 深拷贝的才是注入后的开始节点。
	injectSubFlowFields(nodeExecInput.Global, getRes.FlowContent, subFlowConfig.Fields)
	// 4. 并发数：从主流程开始节点 outputConfig 的 maxConcurrency 字段读取（前端把子流程节点生成次数表单字段聚合到主流程开始节点），读不到再用子流程节点配置兜底
	maxConcurrency := mainFlowMaxConcurrency(nodeExecInput.Global, subFlowConfig.MaxConcurrency)
	// 4. 编译子流程Graph（复用现有 BuildGraphFromFlowContent 逻辑）
	nodeList, subGraph := BuildGraph(ctx, getRes.FlowContent)
	// 4. 构建子流程Workflow（绑定START/END，和示例对齐）
	innerWorkflow := compose.NewWorkflow[*flowDto.FlowExecutionInput, *flowDto.FlowExecutionInput]()
	// 挂载子图节点并绑定全局START
	innerWorkflow.AddGraphNode("sub_flow_graph", subGraph).AddInput(compose.START)
	// 绑定子图输出到全局END
	innerWorkflow.End().AddInput("sub_flow_graph")
	// 生成次数（批量条数）：maxConcurrency<=0 时按 1 次兜底
	batchCount := maxConcurrency
	if batchCount <= 0 {
		batchCount = 1
	}
	// 5. 构建BatchNode（批量执行子流程，复用示例逻辑）
	batchNode := batch.NewBatchNode(&batch.NodeConfig[*flowDto.FlowExecutionInput, *flowDto.FlowExecutionInput]{
		Name:           fmt.Sprintf("sub_flow_batch_%s", nodeExecInput.Config.Id),
		InnerTask:      innerWorkflow,
		MaxConcurrency: batchCount,
	})

	// 6. 提取批量输入：按生成次数生成 N 份（每份独立克隆 ConfigMap，避免并发执行时节点输出写串）
	nodeInputParams := ExtractFlowNodeFrom(getRes.FlowContent)
	configMap := make(map[string]*entity.FlowNode)
	for _, cfg := range nodeInputParams {
		configMap[cfg.Id] = cfg
	}
	for _, i := range nodeList {
		configMap[i.Id] = &i
	}
	batchInputs := make([]*flowDto.FlowExecutionInput, 0, batchCount)
	for j := 0; j < batchCount; j++ {
		batchInputs = append(batchInputs, &flowDto.FlowExecutionInput{
			NodeGroupId: nodeExecInput.Global.NodeGroupId,
			ExecutionId: nodeExecInput.Global.ExecutionId,
			FlowId:      nodeExecInput.Global.FlowId,
			ConfigMap:   cloneConfigMap(configMap),
			SessionId:   nodeExecInput.Global.SessionId,
		})
	}
	// 7. 执行批量子流程
	batchOutput, err := batchNode.Invoke(ctx, batchInputs)
	if err != nil {
		return nil, fmt.Errorf("执行子流程BatchNode失败: %v", err)
	}
	// 8. 展平每份子流程执行的节点输出，写回当前节点 OutputResult 供下游引用
	var outputRes []map[string]any
	for _, single := range batchOutput {
		if single == nil {
			continue
		}
		outputRes = append(outputRes, collectFlowNodeResults(single)...)
	}
	g.Log().Info(ctx, fmt.Sprintf("子流程执行完成，共 %d 次，输出 %d 条", batchCount, len(outputRes)))
	nodeExecInput.Config.OutputResult = outputRes
	return nodeExecInput, nil
}

// injectSubFlowFields 将子流程节点引入参数（subConfig.Fields）解析后写入子流程开始节点
// outputConfig，使子流程启动时能读到首页表单值/上游引用值/静态默认值。
// 每个字段的取值优先级：valueSource 引用解析成功 → field.value → field.defaultValue；
// 匹配键为 field（前端约定以 field 为主，不兼容 path）。
func injectSubFlowFields(global *flowDto.FlowExecutionInput, subFlowContent *entity.FlowInfo, fields []map[string]any) {
	if global == nil || subFlowContent == nil || len(fields) == 0 {
		return
	}
	startNode := subFlowStartNode(subFlowContent)
	if startNode == nil {
		return
	}
	byField := make(map[string]map[string]any, len(startNode.OutputConfig))
	for _, output := range startNode.OutputConfig {
		byField[gconv.String(output["field"])] = output
	}
	for _, field := range fields {
		entry := byField[gconv.String(field["field"])]
		if entry == nil {
			continue
		}
		value := field["value"]
		if vs, has := field["valueSource"]; has && vs != nil {
			if vsNodeId, vsField := firstValueSource(vs); vsNodeId != "" && vsField != "" {
				if v, _, ok := resolveValueSource(global, vsNodeId, vsField); ok {
					value = v
				}
			}
		}
		if value == nil {
			value = field["defaultValue"]
		}
		if value != nil {
			entry["value"] = value
		}
	}
}

// firstValueSource 从 valueSource 提取第一个引用源 (nodeId, field)。
// 前端统一发送数组 [{nodeId, field}]（见 serializeSubFlowConfig），旧 DSL 可能是单对象
// {nodeId, fieldName}，两种形态都兼容；子流程字段与引用源一一对应，只取第一个。
func firstValueSource(vs any) (nodeId, field string) {
	if vs == nil {
		return
	}
	switch v := vs.(type) {
	case []any:
		if len(v) > 0 {
			return firstValueSource(v[0])
		}
		return
	case []map[string]any:
		if len(v) > 0 {
			return firstValueSource(v[0])
		}
		return
	}
	m := gconv.Map(vs)
	nodeId = gconv.String(m["nodeId"])
	field = gconv.String(m["fieldName"])
	if field == "" {
		field = gconv.String(m["field"])
	}
	return
}

// subFlowStartNode 返回工作流开始节点
func subFlowStartNode(content *entity.FlowInfo) *entity.FlowNode {
	if content == nil {
		return nil
	}
	for i := range content.Nodes {
		if content.Nodes[i].Id == content.StartNodeId {
			return &content.Nodes[i]
		}
	}
	return nil
}

// mainFlowMaxConcurrency 取子流程批量执行并发数：从主流程开始节点 outputConfig
// 的 maxConcurrency 字段读取（前端把子流程节点的生成次数表单字段聚合到主流程开始节点），
// 读不到再用子流程节点配置的兜底值。
func mainFlowMaxConcurrency(global *flowDto.FlowExecutionInput, fallback int) int {
	if global == nil {
		return fallback
	}
	for _, n := range global.ConfigMap {
		if n == nil || n.NodeCode != node.NodeTypeStart {
			continue
		}
		for _, output := range n.OutputConfig {
			if gconv.String(output["field"]) != "maxConcurrency" && gconv.String(output["path"]) != "maxConcurrency" {
				continue
			}
			if v := gconv.Int(output["value"]); v > 0 {
				return v
			}
		}
		return fallback
	}
	return fallback
}

// cloneConfigMap 深拷贝 ConfigMap，保证各批次子流程并发执行时节点输出互不串扰。
// 浅拷贝会共享 *entity.FlowNode，并发写 OutputResult 产生竞态。
func cloneConfigMap(src map[string]*entity.FlowNode) map[string]*entity.FlowNode {
	dst := make(map[string]*entity.FlowNode, len(src))
	for k, v := range src {
		data, err := json.Marshal(v)
		if err != nil {
			dst[k] = v
			continue
		}
		n := new(entity.FlowNode)
		if err = json.Unmarshal(data, n); err != nil {
			dst[k] = v
			continue
		}
		dst[k] = n
	}
	return dst
}

// collectFlowNodeResults 收集一次子流程执行中所有已执行节点的输出，展平成 {字段:值} 列表
func collectFlowNodeResults(execInput *flowDto.FlowExecutionInput) []map[string]any {
	var res []map[string]any
	for _, executed := range execInput.ExecutedNodes {
		if nodeConfig := execInput.ConfigMap[executed.NodeId]; nodeConfig != nil {
			res = append(res, nodeConfig.OutputResult...)
		}
	}
	return res
}

// HttpLambda 构建HTTP(S)接口
func HttpLambda(ctx context.Context, input any) (any, error) {
	nodeInput, ok := input.(*flowDto.NodeExecutionInput)
	if !ok {
		return nil, fmt.Errorf("入参类型错误")
	}
	outputRes, err := HttpCallResultLambda(ctx, nodeInput)
	if err != nil {
		return nil, err
	}
	nodeInput.Config.OutputResult = outputRes
	return nodeInput, nil
}

func DataMergeLambda(ctx context.Context, input any) (res any, err error) {
	nodeInput, ok := input.(*flowDto.NodeExecutionInput)
	if !ok {
		return nil, fmt.Errorf("参数合并入参类型错误")
	}
	return nodeInput, nil
}

func SummaryLambda(ctx context.Context, input any) (any, error) {
	execInput, ok := input.(*flowDto.NodeExecutionInput)
	if !ok {
		return nil, fmt.Errorf("汇总节点入参类型错误，实际是 %T", input)
	}

	// 聚合所有已执行节点中需入库的文件结果（两层规则）
	summaryResult := collectSaveFileResults(ctx, execInput.Global)

	// 把汇总结果存入当前节点的输出
	g.Log().Info(ctx, fmt.Sprintf("结果汇总完成，汇总数据：%+v", summaryResult))

	err := gfdb.DB(ctx, public.DbNameBlackDeacon).Transaction(ctx, func(ctx context.Context, tx gdb.TX) error {
		res, _, err := nodeDao.NodeExecutionDao.ListByFlowExecutionId(ctx, &nodeDto.ListNodeExecutionByFlowReq{
			NodeGroupId: execInput.Global.NodeGroupId,
		}, entity.NodeExecutionCol.TokenInfo)
		if err != nil {
			return err
		}
		var totalTokens int
		var totalFee float64
		for _, item := range res {
			for _, itemToken := range item.TokenInfo {
				m := gconv.Map(itemToken)
				totalTokens += gconv.Int(m["total_tokens"])
				totalFee += gconv.Float64(m["total_fee"])
			}
		}
		_, err = sessionDao.ExecWorkflowDao.Update(ctx, &sessionDto.UpdateWorkflowReq{
			Id:          execInput.Global.ExecutionId,
			Status:      flow.FlowExecutionStatusSuccess.Code(),
			TotalTokens: totalTokens,
			TotalFee:    totalFee,
		})
		if err != nil {
			return err
		}
		if len(summaryResult) > 0 {
			_, err = sessionDao.ExecWorkflowResultDao.BatchInsert(ctx, summaryResult)
			if err != nil {
				return err
			}
		}
		return nil
	})

	return execInput, err
}

// collectSaveFileResults 按两层规则收集需入库的文件结果：
// 第一层：节点须开启"保存文件"（IsSaveFile）；
// 第二层：key 取自节点 OutputResult 的各字段，命中 ModelResponseBodyMapping 才入库；
// 原始响应体 key（respBody）恒入库（不要求映射声明）；HTTP 节点产出以 http_file_url:{key}
// 标记的字段（IsSaveFile 时由 HttpCallResultLambda 生成）恒入库（无模型响应映射可查）。
// 结果值为 http(s) URL 或 MinIO 对象裸路径直接使用；非路径值（base64 图片/文本）先上传 OSS 换取 URL，
// 文本内容以 .inc 扩展名存储。
func collectSaveFileResults(ctx context.Context, execInput *flowDto.FlowExecutionInput) []*sessionDto.CreateWorkflowResultReq {
	if execInput == nil {
		return nil
	}
	var summaryResult []*sessionDto.CreateWorkflowResultReq
	for _, executedNode := range execInput.ExecutedNodes {
		nodeConfig := execInput.ConfigMap[executedNode.NodeId]
		if nodeConfig == nil || len(nodeConfig.OutputResult) == 0 || !nodeConfig.IsSaveFile {
			continue
		}
		// 第二层：key 取自节点 OutputResult 的各字段，
		// 命中 ModelResponseBodyMapping 才入库；respBody 与 HTTP 节点 http_file_url:{key} 标记恒入库
		saveKeys := nodeConfig.ModelConfig.ModelResponseBodyMapping
		for _, respBody := range nodeConfig.OutputResult {
			for key, val := range gconv.Map(respBody) {
				isHTTPFile := strings.HasPrefix(key, "http_file_url:")
				if !isHTTPFile {
					if _, ok := saveKeys[key]; !ok && key != "respBody" {
						continue
					}
				}
				fileUrl, err := resolveSaveFileResult(ctx, val)
				if err != nil {
					g.Log().Warningf(ctx, "collectSaveFileResults 上传结果文件失败 key=%s err=%v", key, err)
					continue
				}
				summaryResult = append(summaryResult, &sessionDto.CreateWorkflowResultReq{
					SessionId:     execInput.SessionId,
					FlowId:        execInput.FlowId,
					ExecId:        execInput.ExecutionId,
					ResultFileUrl: fileUrl,
				})
			}
		}
	}
	return summaryResult
}

// resolveSaveFileResult 解析结果值为可入库的 URL：
//   - 已是 http(s) URL 或 MinIO 对象裸路径 → 直接返回
//   - 非路径（base64 图片/文本）→ 上传 OSS 换取 URL
func resolveSaveFileResult(ctx context.Context, val any) (string, error) {
	isPath, path, fileBytes, ext := resolveFileContent(val)
	if isPath {
		return path, nil
	}
	if ext == "" {
		ext = ".png"
	}
	fileUrl, err := gateway.Upload(ctx, fmt.Sprintf("workflow_result_%s%s", uuid.NewString(), ext), fileBytes)
	if err != nil {
		return "", err
	}
	return fileUrl, nil
}

// resolveFileContent 判断结果值形态：
//   - 已是 URL 路径（http/https 开头）→ 直接使用
//   - data URI（data:<mime>;base64,<data>）→ 解码为字节，扩展名按 mime 推断
//   - 纯 base64（可解码且长度足以认为是编码数据）→ 解码为字节，默认 .png
//   - 其余（文本）→ 以 .inc 扩展名上传原文
func resolveFileContent(val any) (isPath bool, path string, fileBytes []byte, ext string) {
	s := gconv.String(val)
	if isFileURL(s) {
		return true, s, nil, ""
	}
	// MinIO 对象裸路径（无 http 前缀，模型网关转存 OSS 后返回）
	if utils.IsOSSPath(s) {
		return true, s, nil, ""
	}
	// data URI：data:<mime>;base64,<payload>
	if b, mime, ok := parseDataURI(s); ok {
		return false, "", b, extOfMime(mime)
	}
	// 纯 base64：可解码且长度足够，视为编码后的文件内容
	trimmed := strings.TrimSpace(s)
	if len(trimmed) >= 64 {
		if b, err := base64.StdEncoding.DecodeString(trimmed); err == nil && len(b) > 0 {
			return false, "", b, ".png"
		}
	}
	// 文本：以 .inc 存储
	return false, "", []byte(s), ".inc"
}

// isFileURL 判断字符串是否已是对外可访问的 URL 路径（http/https 开头）
func isFileURL(s string) bool {
	lower := strings.ToLower(s)
	return strings.HasPrefix(lower, "http://") || strings.HasPrefix(lower, "https://")
}

// extOfMime 按 MIME 类型推断文件扩展名
func extOfMime(mime string) string {
	switch strings.ToLower(strings.TrimSpace(mime)) {
	case "image/png", "png":
		return ".png"
	case "image/jpeg", "image/jpg", "jpeg", "jpg":
		return ".jpg"
	case "image/webp":
		return ".webp"
	case "image/gif":
		return ".gif"
	case "audio/mpeg", "audio/mp3", "mp3":
		return ".mp3"
	case "audio/wav", "wav":
		return ".wav"
	case "video/mp4", "mp4":
		return ".mp4"
	case "application/json", "json":
		return ".json"
	default:
		return ""
	}
}

// parseDataURI 解析 data URI：data:<mime>;base64,<payload>，返回解码字节与 mime
func parseDataURI(s string) ([]byte, string, bool) {
	const prefix = "data:"
	if !strings.HasPrefix(s, prefix) {
		return nil, "", false
	}
	rest := s[len(prefix):]
	comma := strings.Index(rest, ",")
	if comma < 0 {
		return nil, "", false
	}
	mime := rest[:comma]
	if semicolon := strings.Index(mime, ";"); semicolon >= 0 {
		mime = mime[:semicolon]
	}
	payload := strings.TrimPrefix(rest[comma+1:], "base64,")
	b, err := base64.StdEncoding.DecodeString(payload)
	if err != nil {
		return nil, "", false
	}
	return b, mime, true
}