1037 lines
29 KiB
Go
1037 lines
29 KiB
Go
package flow
|
||
|
||
import (
|
||
"ai-agent/workflow/consts/node"
|
||
nodeDao "ai-agent/workflow/dao/node"
|
||
"ai-agent/workflow/model/dto"
|
||
flowDto "ai-agent/workflow/model/dto/flow"
|
||
nodeDto "ai-agent/workflow/model/dto/node"
|
||
"ai-agent/workflow/model/entity"
|
||
"bytes"
|
||
"context"
|
||
"errors"
|
||
"fmt"
|
||
"io"
|
||
"mime/multipart"
|
||
"net/http"
|
||
"net/url"
|
||
"path"
|
||
"path/filepath"
|
||
"regexp"
|
||
"strconv"
|
||
"strings"
|
||
"sync"
|
||
"time"
|
||
|
||
commonHttp "gitea.redpowerfuture.com/red-future/common/http"
|
||
"gitea.redpowerfuture.com/red-future/common/utils"
|
||
"github.com/gogf/gf/v2/errors/gerror"
|
||
"github.com/gogf/gf/v2/frame/g"
|
||
"github.com/gogf/gf/v2/net/ghttp"
|
||
"github.com/gogf/gf/v2/util/gconv"
|
||
"github.com/tidwall/sjson"
|
||
)
|
||
|
||
// 全局等待任务回调的工具
|
||
var (
|
||
asyncMu sync.Mutex
|
||
asyncTasks = make(map[string]chan any)
|
||
)
|
||
|
||
// Wait 阻塞等待回调结果
|
||
// 调用后会一直卡住，直到 Notify 唤醒 或 超时/取消
|
||
func Wait(ctx context.Context, taskId string) (any, error) {
|
||
asyncMu.Lock()
|
||
ch := make(chan any, 1)
|
||
asyncTasks[taskId] = ch
|
||
asyncMu.Unlock()
|
||
|
||
defer close(ch)
|
||
for {
|
||
select {
|
||
case result := <-ch:
|
||
return result, nil
|
||
case <-ctx.Done():
|
||
asyncMu.Lock()
|
||
delete(asyncTasks, taskId)
|
||
asyncMu.Unlock()
|
||
return nil, ctx.Err()
|
||
}
|
||
}
|
||
}
|
||
|
||
// Notify 回调时调用，唤醒等待的任务
|
||
func Notify(taskId string, result any) {
|
||
asyncMu.Lock()
|
||
defer asyncMu.Unlock()
|
||
|
||
ch, exist := asyncTasks[taskId]
|
||
if !exist {
|
||
return
|
||
}
|
||
ch <- result
|
||
delete(asyncTasks, taskId)
|
||
}
|
||
|
||
func GetIsChatModel(ctx context.Context) (res *flowDto.GetIsChatModelRes, err error) {
|
||
headers := make(map[string]string)
|
||
if r := g.RequestFromCtx(ctx); r != nil {
|
||
for k, v := range r.Request.Header {
|
||
if len(v) > 0 {
|
||
headers[k] = v[0]
|
||
}
|
||
}
|
||
}
|
||
res = new(flowDto.GetIsChatModelRes)
|
||
err = commonHttp.Get(ctx, "model-gateway/model/getIsChatModel", headers, res, nil)
|
||
return
|
||
}
|
||
|
||
func GetModelInfo(ctx context.Context, req *flowDto.GetModelInfoReq) (res *flowDto.GetModelInfoRes, err error) {
|
||
headers := make(map[string]string)
|
||
if r := g.RequestFromCtx(ctx); r != nil {
|
||
for k, v := range r.Request.Header {
|
||
if len(v) > 0 {
|
||
headers[k] = v[0]
|
||
}
|
||
}
|
||
}
|
||
res = new(flowDto.GetModelInfoRes)
|
||
err = commonHttp.Get(ctx, "model-gateway/model/getModel", headers, res, req)
|
||
return
|
||
}
|
||
|
||
func GetComposeResult(ctx context.Context, nodeExecutionId int64, buildType int, modelName, promptContent, skillName string, form []map[string]any, userForm []map[string]any, fileUrl []string, sessionId, nodeId string, cause string) (res *flowDto.ComposeCallbackReq, err error) {
|
||
var callbackUrl = utils.GetCallbackURL(ctx, "/flow/execution/composeCallBack")
|
||
var consult = make([]flowDto.Consult, 0)
|
||
var collectFileUrls func(val any) (fullyConsumed bool)
|
||
collectFileUrls = func(val any) (fullyConsumed bool) {
|
||
switch {
|
||
case g.NewVar(val).IsSlice():
|
||
slice := gconv.SliceAny(val)
|
||
allConsumed := false
|
||
for _, item := range slice {
|
||
if collectFileUrls(item) {
|
||
allConsumed = true
|
||
}
|
||
}
|
||
return allConsumed
|
||
case g.NewVar(val).IsMap():
|
||
m := gconv.Map(val)
|
||
allConsumed := false
|
||
for _, item := range m {
|
||
if collectFileUrls(item) {
|
||
allConsumed = true
|
||
}
|
||
}
|
||
return allConsumed
|
||
default:
|
||
s := gconv.String(val)
|
||
if s != "" {
|
||
getFileTypeByPath := GetFileTypeByPath(s)
|
||
if getFileTypeByPath != "" {
|
||
consult = append(consult, flowDto.Consult{
|
||
Type: getFileTypeByPath,
|
||
Url: s,
|
||
})
|
||
return true
|
||
}
|
||
}
|
||
return false
|
||
}
|
||
}
|
||
var newUserForm []map[string]any
|
||
for _, m := range userForm {
|
||
// 先替换字段
|
||
if val, ok := m["audioDuration"]; ok {
|
||
delete(m, "audioDuration")
|
||
m["视频总时长"] = val
|
||
}
|
||
if val, ok := m["videoDuration"]; ok {
|
||
delete(m, "videoDuration")
|
||
m["视频总时长"] = val
|
||
}
|
||
// 收集待删除 key
|
||
var delKeys []string
|
||
for k, v := range m {
|
||
if collectFileUrls(v) {
|
||
delKeys = append(delKeys, k)
|
||
}
|
||
}
|
||
// 统一删除
|
||
for _, k := range delKeys {
|
||
delete(m, k)
|
||
}
|
||
|
||
if len(m) > 0 {
|
||
newUserForm = append(newUserForm, m)
|
||
}
|
||
}
|
||
for _, v := range fileUrl {
|
||
getFileTypeByPath := GetFileTypeByPath(gconv.String(v))
|
||
if getFileTypeByPath != "" {
|
||
consult = append(consult, flowDto.Consult{
|
||
Type: getFileTypeByPath,
|
||
Url: gconv.String(v),
|
||
})
|
||
}
|
||
}
|
||
msgReq := flowDto.ComposeMessagesReq{
|
||
BuildType: buildType,
|
||
ModelName: modelName,
|
||
SkillName: skillName,
|
||
CallbackUrl: callbackUrl,
|
||
Cause: cause,
|
||
Form: form,
|
||
UserForm: newUserForm,
|
||
UserPrompt: promptContent,
|
||
Consult: consult,
|
||
SessionId: sessionId,
|
||
NodeId: nodeId,
|
||
}
|
||
msgRes := new(flowDto.ComposeMessagesRes)
|
||
|
||
// 1. 隔离上游取消（防止节点执行被中断时下游请求被 cancel）+ 设置独立超时
|
||
baseCtx := context.WithoutCancel(ctx)
|
||
postCtx, cancel := context.WithTimeout(baseCtx, 30*time.Minute)
|
||
defer cancel() // 必须释放，防止上下文泄露
|
||
|
||
// 2. 克隆 commonHttp 客户端（保留 Consul 服务发现），显式设置超时和 ResponseHeaderTimeout
|
||
client := commonHttp.Httpclient.Clone()
|
||
client.SetTimeout(30 * time.Minute)
|
||
if tr, ok := client.Transport.(*http.Transport); ok {
|
||
tr.ResponseHeaderTimeout = 30 * time.Minute
|
||
}
|
||
if r := g.RequestFromCtx(ctx); r != nil {
|
||
for k, v := range r.Request.Header {
|
||
if len(v) > 0 {
|
||
client.SetHeader(k, v[0])
|
||
}
|
||
}
|
||
}
|
||
resp, err := client.ContentJson().Post(postCtx, "prompts-core/prompt/composeMessages", &msgReq)
|
||
if err != nil {
|
||
return
|
||
}
|
||
defer resp.Close()
|
||
result, err := io.ReadAll(resp.Body)
|
||
if err != nil {
|
||
return nil, fmt.Errorf("读取composeMessages响应失败: %w", err)
|
||
}
|
||
// 统一处理内部API响应格式：{code:200,message:"",data:{...}}
|
||
resultStrut := &ghttp.DefaultHandlerResponse{}
|
||
|
||
if err = gconv.Struct(result, &resultStrut); err != nil { // 修复：增加err检查
|
||
return nil, fmt.Errorf("响应解析失败: " + err.Error())
|
||
}
|
||
|
||
// 添加调试日志：打印解析后的结构
|
||
g.Log().Debugf(ctx, "[HTTP] 解析后结构: Code=%d, Message=%s, Data类型=%T, Data值=%+v",
|
||
resultStrut.Code, resultStrut.Message, resultStrut.Data, resultStrut.Data)
|
||
|
||
if resultStrut.Code == 200 || resultStrut.Code == 0 {
|
||
if err = gconv.Struct(resultStrut.Data, &msgRes); err != nil { // 修复：增加err检查
|
||
return nil, fmt.Errorf("数据解析失败: " + err.Error())
|
||
}
|
||
// 添加调试日志：打印最终的target
|
||
g.Log().Debugf(ctx, "[HTTP] 最终target: %+v", &msgRes)
|
||
} else {
|
||
err = errors.New(resultStrut.Message)
|
||
}
|
||
if g.IsEmpty(msgRes.TaskId) {
|
||
return nil, fmt.Errorf("msg is empty")
|
||
}
|
||
waitRes, err := Wait(ctx, msgRes.TaskId)
|
||
if err != nil {
|
||
return nil, err
|
||
}
|
||
msg := new(flowDto.ComposeCallbackReq)
|
||
if err = gconv.Struct(waitRes, msg); err != nil {
|
||
return nil, err
|
||
}
|
||
updateTokenCount(ctx, nodeExecutionId, msg.BillingData)
|
||
if !g.IsEmpty(msg.ErrorMsg) {
|
||
return nil, fmt.Errorf(msg.ErrorMsg)
|
||
}
|
||
return msg, nil
|
||
}
|
||
|
||
func CreateGatewayTask(ctx context.Context, nodeExecutionId int64, epicycleId int64, model string, content map[string]any) (map[string]any, error) {
|
||
taskId, err := createGatewayTaskOnly(ctx, epicycleId, model, content)
|
||
if err != nil {
|
||
return nil, err
|
||
}
|
||
return waitGatewayResult(ctx, nodeExecutionId, taskId)
|
||
}
|
||
|
||
// createGatewayTaskOnly creates a gateway task and returns the taskId only
|
||
// doesn't wait for completion
|
||
func createGatewayTaskOnly(ctx context.Context, epicycleId int64, model string, content map[string]any) (string, error) {
|
||
callbackUrl := utils.GetCallbackURL(ctx, "/flow/execution/modelCallback")
|
||
req := flowDto.ModelGatewayReq{
|
||
ModelName: model,
|
||
BizName: g.Cfg().MustGet(ctx, "server.name").String(),
|
||
CallbackUrl: callbackUrl,
|
||
RequestPayload: content,
|
||
EpicycleId: epicycleId,
|
||
}
|
||
|
||
res := new(flowDto.ModelGatewayRes)
|
||
|
||
// 1. 隔离上游取消（防止节点执行被中断时下游请求被 cancel）+ 设置独立超时
|
||
baseCtx := context.WithoutCancel(ctx)
|
||
postCtx, cancel := context.WithTimeout(baseCtx, 30*time.Minute)
|
||
defer cancel() // 必须释放，防止上下文泄露
|
||
|
||
// 2. 克隆 commonHttp 客户端（保留 Consul 服务发现），显式设置超时和 ResponseHeaderTimeout
|
||
client := commonHttp.Httpclient.Clone()
|
||
client.SetTimeout(30 * time.Minute)
|
||
if tr, ok := client.Transport.(*http.Transport); ok {
|
||
tr.ResponseHeaderTimeout = 30 * time.Minute
|
||
}
|
||
if r := g.RequestFromCtx(ctx); r != nil {
|
||
for k, v := range r.Request.Header {
|
||
if len(v) > 0 {
|
||
client.SetHeader(k, v[0])
|
||
}
|
||
}
|
||
}
|
||
rpcResp, err := client.ContentJson().Post(postCtx, "model-gateway/task/createTask", &req)
|
||
if err != nil {
|
||
return "", err
|
||
}
|
||
defer rpcResp.Close()
|
||
result, err := io.ReadAll(rpcResp.Body)
|
||
if err != nil {
|
||
return "", fmt.Errorf("读取createTask响应失败: %w", err)
|
||
}
|
||
// 统一处理内部API响应格式：{code:200,message:"",data:{...}}
|
||
resultStrut := &ghttp.DefaultHandlerResponse{}
|
||
|
||
if err = gconv.Struct(result, &resultStrut); err != nil { // 修复：增加err检查
|
||
return "", fmt.Errorf("响应解析失败: " + err.Error())
|
||
}
|
||
|
||
// 添加调试日志：打印解析后的结构
|
||
g.Log().Debugf(ctx, "[HTTP] 解析后结构: Code=%d, Message=%s, Data类型=%T, Data值=%+v",
|
||
resultStrut.Code, resultStrut.Message, resultStrut.Data, resultStrut.Data)
|
||
|
||
if resultStrut.Code == 200 || resultStrut.Code == 0 {
|
||
if err = gconv.Struct(resultStrut.Data, &res); err != nil { // 修复：增加err检查
|
||
return "", fmt.Errorf("数据解析失败: " + err.Error())
|
||
}
|
||
// 添加调试日志：打印最终的target
|
||
g.Log().Debugf(ctx, "[HTTP] 最终target: %+v", &res)
|
||
} else {
|
||
err = errors.New(resultStrut.Message)
|
||
}
|
||
if g.IsEmpty(res.TaskId) {
|
||
return "", fmt.Errorf("创建模型任务失败，taskId为空")
|
||
}
|
||
return res.TaskId, nil
|
||
}
|
||
|
||
// waitGatewayResult waits for a created gateway task to complete and returns the result
|
||
func waitGatewayResult(ctx context.Context, nodeExecutionId int64, taskId string) (map[string]any, error) {
|
||
waitRes, err := Wait(ctx, taskId)
|
||
if err != nil {
|
||
return nil, err
|
||
}
|
||
|
||
task := new(flowDto.ModelCallbackReq)
|
||
if err = gconv.Struct(waitRes, task); err != nil {
|
||
return nil, err
|
||
}
|
||
updateTokenCount(ctx, nodeExecutionId, task.BillingData)
|
||
if task.State == 3 || !g.IsEmpty(task.ErrorMsg) {
|
||
return nil, fmt.Errorf("模型执行失败：%s", task.ErrorMsg)
|
||
}
|
||
if g.IsEmpty(task.OssFile) {
|
||
return nil, fmt.Errorf("模型返回结果为空")
|
||
}
|
||
// 获取远程文件内容
|
||
file, err := GetFileBytesFromURL(ctx, task.OssFile)
|
||
if err != nil {
|
||
return nil, err
|
||
}
|
||
return gconv.Map(file), nil
|
||
}
|
||
|
||
// updateTokenCount updates the token count in node execution
|
||
func updateTokenCount(ctx context.Context, nodeExecutionId int64, tokenInfo []map[string]any) {
|
||
res, err := nodeDao.NodeExecutionDao.Get(ctx, &nodeDto.GetNodeExecutionReq{
|
||
Id: nodeExecutionId,
|
||
}, entity.NodeExecutionCol.TokenInfo)
|
||
if err != nil {
|
||
return
|
||
}
|
||
var t []map[string]any
|
||
for _, item := range res.TokenInfo {
|
||
t = append(t, item)
|
||
}
|
||
for _, item := range tokenInfo {
|
||
t = append(t, item)
|
||
}
|
||
_, _ = nodeDao.NodeExecutionDao.Update(ctx, &nodeDto.UpdateNodeExecutionReq{
|
||
Id: nodeExecutionId,
|
||
TokenInfo: t,
|
||
})
|
||
}
|
||
|
||
func GetModelResult(ctx context.Context, sessionId string, nodeInput *flowDto.NodeExecutionInput, skillName string, form []map[string]any, userForm []map[string]any) (mapTaskResult []map[string]any, err error) {
|
||
buildType := 1
|
||
if nodeInput.Config.NodeCode == node.NodeTypeDataConversionModel {
|
||
buildType = 3
|
||
}
|
||
|
||
if !nodeInput.Global.IsDialogue {
|
||
sessionId = ""
|
||
}
|
||
needSequential := false
|
||
for _, item := range userForm {
|
||
if g.NewVar(item).IsMap() {
|
||
valMap := gconv.Map(item)
|
||
for _, v := range valMap {
|
||
if g.NewVar(v).IsMap() {
|
||
vv := gconv.Map(v)
|
||
for kk, vvv := range vv {
|
||
if kk == "return_last_frame" {
|
||
needSequential = vvv.(bool)
|
||
}
|
||
}
|
||
}
|
||
}
|
||
}
|
||
}
|
||
composeResult, err := GetComposeResult(ctx, nodeInput.NodeExecutionId, buildType, nodeInput.Config.ModelConfig.ModelName, nodeInput.Config.PromptContent, skillName, form, userForm, nodeInput.Global.FileUrl, sessionId, nodeInput.Config.Id, nodeInput.Config.Name)
|
||
if err != nil {
|
||
return nil, err
|
||
}
|
||
if composeResult.Status != "success" {
|
||
return nil, fmt.Errorf("模型提示词构建错误")
|
||
}
|
||
|
||
modelInfo, err := GetModelInfo(ctx, &flowDto.GetModelInfoReq{ModelName: nodeInput.Config.ModelConfig.ModelName})
|
||
if err != nil {
|
||
return nil, err
|
||
}
|
||
|
||
mapTaskResult = make([]map[string]any, len(composeResult.Messages.Rounds))
|
||
var taskResultMap map[string]any
|
||
|
||
if buildType == 1 {
|
||
if needSequential {
|
||
for idx, item := range composeResult.Messages.Rounds {
|
||
if !g.IsEmpty(taskResultMap) {
|
||
var set string
|
||
set, err = sjson.Set(gconv.String(item), modelInfo.Model.FirstFrame, gconv.String(taskResultMap["content"]))
|
||
if err != nil {
|
||
return nil, err
|
||
}
|
||
item = gconv.Map(set)
|
||
}
|
||
|
||
var taskResult map[string]any
|
||
taskResult, err = CreateGatewayTask(ctx, nodeInput.NodeExecutionId, composeResult.EpicycleId, nodeInput.Config.ModelConfig.ModelName, item)
|
||
if err != nil {
|
||
return nil, err
|
||
}
|
||
if g.IsEmpty(taskResult) {
|
||
return nil, fmt.Errorf("模型返回结果为空")
|
||
}
|
||
|
||
if nodeInput.Config.NodeCode == node.NodeTypeVideoModel {
|
||
ext := GetFileTypeByPath(gconv.String(taskResult["content"]))
|
||
if ext == "image" {
|
||
taskResultMap = taskResult
|
||
} else {
|
||
taskResultMap = make(map[string]any)
|
||
}
|
||
} else {
|
||
taskResultMap = make(map[string]any)
|
||
}
|
||
|
||
mapTaskResult[idx] = taskResult
|
||
//updateTokenCount(ctx, nodeInput.NodeExecutionId, modelInfo.Model.ResponseTokenField, taskResult)
|
||
}
|
||
} else {
|
||
taskIdList := make([]string, len(composeResult.Messages.Rounds))
|
||
|
||
for idx, item := range composeResult.Messages.Rounds {
|
||
taskId, err := createGatewayTaskOnly(ctx, composeResult.EpicycleId, nodeInput.Config.ModelConfig.ModelName, item)
|
||
if err != nil {
|
||
return nil, err
|
||
}
|
||
taskIdList[idx] = taskId
|
||
}
|
||
|
||
// 全局共享子上下文，实现一处报错全部终止
|
||
subCtx, globalCancel := context.WithCancel(ctx)
|
||
defer globalCancel() // 函数退出兜底释放
|
||
|
||
var wg sync.WaitGroup
|
||
errChan := make(chan error, len(taskIdList))
|
||
|
||
// 加互斥锁保护结果map
|
||
var mu sync.Mutex
|
||
|
||
for idx, taskId := range taskIdList {
|
||
wg.Add(1)
|
||
|
||
go func(idx int, taskId string) {
|
||
defer wg.Done()
|
||
|
||
taskResult, err := waitGatewayResult(subCtx, nodeInput.NodeExecutionId, taskId)
|
||
if err != nil {
|
||
errChan <- err
|
||
globalCancel() // 全局取消，所有协程收到ctx取消信号快速退出
|
||
return
|
||
}
|
||
|
||
// 加锁写入map，解决并发竞态
|
||
mu.Lock()
|
||
mapTaskResult[idx] = taskResult
|
||
mu.Unlock()
|
||
|
||
//updateTokenCount(ctx, nodeInput.NodeExecutionId, modelInfo.Model.ResponseTokenField, taskResult)
|
||
}(idx, taskId)
|
||
}
|
||
|
||
wg.Wait()
|
||
close(errChan)
|
||
|
||
// 收集全部错误，而非只读一条
|
||
var errs []error
|
||
for len(errChan) > 0 {
|
||
errs = append(errs, <-errChan)
|
||
}
|
||
|
||
if len(errs) > 0 {
|
||
// 返回第一个错误；如需汇总所有错误可拼接
|
||
return nil, errs[0]
|
||
}
|
||
}
|
||
} else {
|
||
for idx, item := range composeResult.Messages.Rounds {
|
||
mapTaskResult[idx] = item
|
||
//updateTokenCount(ctx, nodeInput.NodeExecutionId, modelInfo.Model.ResponseTokenField, item)
|
||
}
|
||
}
|
||
return mapTaskResult, nil
|
||
}
|
||
|
||
func BuildNestedJson(body g.Map, global *flowDto.FlowExecutionInput) g.Map {
|
||
jsonStr := "{}"
|
||
for originKey, originItem := range body {
|
||
if originKey == "templates" && !g.IsEmpty(global.Templates) {
|
||
jsonStr, _ = sjson.Set(jsonStr, originKey, global.Templates)
|
||
continue
|
||
}
|
||
bodyItemMap := gconv.Map(originItem)
|
||
val := bodyItemMap["value"]
|
||
if v, ok := bodyItemMap["value"]; ok {
|
||
jsonStr, _ = sjson.Set(jsonStr, originKey, v)
|
||
}
|
||
// 判断 value 是不是引用结构（map）
|
||
if g.NewVar(val).IsMap() {
|
||
valMap := gconv.Map(val)
|
||
nodeId := gconv.String(valMap["nodeId"])
|
||
fieldName := gconv.String(valMap["field"])
|
||
if configValue, ok := global.ConfigMap[nodeId]; ok {
|
||
if !g.IsEmpty(configValue.OutputResult) {
|
||
for _, v := range configValue.OutputResult {
|
||
if strings.Contains(v.Field, fieldName) {
|
||
if configValue.NodeCode == node.NodeTypeDataConversionModel {
|
||
switch {
|
||
case g.NewVar(v.Value).IsSlice() || g.NewVar(v.Value).IsMap():
|
||
// 核心：自动判断两种结构，精准赋值
|
||
vm := gconv.Map(v.Value)
|
||
// 先判断是否是 单个key包裹的对象（如 {"subtitle_style": {...}}）
|
||
if len(vm) == 1 {
|
||
// 遍历取出唯一的 key 和 真实值
|
||
for innerKey, innerVal := range vm {
|
||
// 直接用 innerKey（subtitle_style）赋值
|
||
jsonStr, _ = sjson.Set(jsonStr, innerKey, innerVal)
|
||
}
|
||
} else {
|
||
// 直接是对象，用 originKey 赋值
|
||
jsonStr, _ = sjson.Set(jsonStr, originKey, v.Value)
|
||
}
|
||
default:
|
||
jsonStr, _ = sjson.Set(jsonStr, originKey, v.Value)
|
||
}
|
||
} else {
|
||
jsonStr, _ = sjson.Set(jsonStr, originKey, v.Value)
|
||
}
|
||
}
|
||
}
|
||
}
|
||
if !g.IsEmpty(configValue.FormConfig) {
|
||
for _, v := range configValue.FormConfig {
|
||
if v.Field == fieldName {
|
||
if v.Type == "uploadMultiple" {
|
||
if g.NewVar(v.FieldConstraint).IsMap() {
|
||
mapFieldConstraint := gconv.Map(v.FieldConstraint)
|
||
for key, value := range mapFieldConstraint {
|
||
if key == "maxFileCount" {
|
||
if gconv.Int(value) == 1 {
|
||
// 如果是单文件上传，则替换成字符串重新赋值给v.Value
|
||
if g.NewVar(v.Value).IsSlice() {
|
||
sliceVal := gconv.SliceAny(v.Value)
|
||
if len(sliceVal) > 0 {
|
||
v.Value = sliceVal[0]
|
||
}
|
||
}
|
||
}
|
||
}
|
||
}
|
||
}
|
||
}
|
||
jsonStr, _ = sjson.Set(jsonStr, originKey, v.Value)
|
||
}
|
||
}
|
||
}
|
||
}
|
||
}
|
||
}
|
||
return gconv.Map(jsonStr)
|
||
}
|
||
|
||
func VideoConcat(ctx context.Context, videoUrls []string) (r any, err error) {
|
||
var httpUrl = "media/video/concat/async"
|
||
headers := make(map[string]string)
|
||
if r := g.RequestFromCtx(ctx); r != nil {
|
||
for k, v := range r.Request.Header {
|
||
if len(v) > 0 {
|
||
headers[k] = v[0]
|
||
}
|
||
}
|
||
}
|
||
var callbackUrl = utils.GetCallbackURL(ctx, "/flow/execution/videoCallback")
|
||
var newBody = flowDto.VideoConcatReq{
|
||
VideoUrls: videoUrls,
|
||
Method: "auto",
|
||
Upload: true,
|
||
CallbackUrl: callbackUrl,
|
||
}
|
||
res := new(flowDto.VideoConcatRes)
|
||
err = commonHttp.Post(ctx, httpUrl, headers, &res, newBody)
|
||
if err != nil {
|
||
return nil, err
|
||
}
|
||
return Wait(ctx, res.TaskId)
|
||
}
|
||
|
||
func GetFileBytesFromURL(ctx context.Context, fileUrl string) ([]byte, error) {
|
||
newS := strings.ReplaceAll(fileUrl, g.Cfg().MustGet(ctx, "filePrefix").String(), g.Cfg().MustGet(ctx, "minioPrefix").String())
|
||
// 使用 GoFrame 客户端（自带超时、追踪、日志等能力）
|
||
resp, err := g.Client().Get(ctx, newS)
|
||
if err != nil {
|
||
return nil, gerror.Wrapf(err, "failed to request url: %s", newS)
|
||
}
|
||
defer resp.Close()
|
||
|
||
// 校验状态码
|
||
if resp.StatusCode != http.StatusOK {
|
||
return nil, gerror.Newf("request failed with status code: %d, url: %s", resp.StatusCode, newS)
|
||
}
|
||
|
||
// 读取全部内容
|
||
allBytes, err := io.ReadAll(resp.Body)
|
||
if err != nil {
|
||
return nil, gerror.Wrapf(err, "failed to read response body, url: %s", fileUrl)
|
||
}
|
||
|
||
return allBytes, nil
|
||
}
|
||
|
||
func Upload(ctx context.Context, req *dto.UploadFileBytesReq) (*dto.UploadFileBytesRes, error) {
|
||
body := &bytes.Buffer{}
|
||
writer := multipart.NewWriter(body)
|
||
|
||
part, err := writer.CreateFormFile("file", req.FileName)
|
||
if err != nil {
|
||
return nil, err
|
||
}
|
||
if _, err = part.Write(req.FileBytes); err != nil {
|
||
return nil, err
|
||
}
|
||
if err = writer.Close(); err != nil {
|
||
return nil, err
|
||
}
|
||
|
||
headers := make(map[string]string)
|
||
headers["Content-Type"] = writer.FormDataContentType()
|
||
if r := g.RequestFromCtx(ctx); r != nil {
|
||
if auth := r.Header.Get("Authorization"); auth != "" {
|
||
headers["Authorization"] = auth
|
||
}
|
||
}
|
||
|
||
// 发起上传请求
|
||
res := &dto.UploadFileBytesRes{}
|
||
httpUrl := "oss/file/uploadFile"
|
||
if err = commonHttp.Post(ctx, httpUrl, headers, res, body.Bytes()); err != nil {
|
||
return nil, err
|
||
}
|
||
|
||
g.Log().Infof(ctx, "[Upload] success url=%s size=%d", res.FileURL, res.FileSize)
|
||
return res, nil
|
||
}
|
||
|
||
func GetFileTypeByPath(filePath string) string {
|
||
if filePath == "" {
|
||
return ""
|
||
}
|
||
|
||
// 解析 URL，获取真实路径（兼容 http 链接）
|
||
u, err := url.Parse(filePath)
|
||
if err == nil {
|
||
filePath = u.Path
|
||
}
|
||
|
||
// 获取后缀（小写）
|
||
ext := filepath.Ext(filePath)
|
||
ext = strings.ToLower(ext)
|
||
|
||
// 判断类型
|
||
switch ext {
|
||
case ".jpg", ".jpeg", ".png", ".gif", ".bmp", ".webp":
|
||
return "image"
|
||
case ".mp4", ".mov", ".avi", ".flv", ".wmv", ".mkv":
|
||
return "video"
|
||
case ".mp3", ".wav", ".m4a", ".flac", ".aac", ".ogg":
|
||
return "audio"
|
||
case ".txt", ".md", ".log", ".json", ".xml", ".inc":
|
||
return "text"
|
||
case ".html":
|
||
return "html"
|
||
case ".pdf", ".doc", ".docx", ".xls", ".xlsx", ".ppt", ".pptx":
|
||
return "document"
|
||
default:
|
||
return ""
|
||
}
|
||
}
|
||
|
||
// GetUrlSuffix 获取URL文件后缀
|
||
// rawUrl: 原始链接
|
||
// withDot: true 返回 .mp4 false 返回 mp4
|
||
func GetUrlSuffix(rawUrl string, withDot bool) string {
|
||
// 解析URL，剥离查询参数
|
||
u, err := url.Parse(rawUrl)
|
||
if err != nil {
|
||
return ""
|
||
}
|
||
|
||
// 提取路径部分
|
||
filePath := u.Path
|
||
// 获取文件名
|
||
fileName := path.Base(filePath)
|
||
if fileName == "" || !strings.Contains(fileName, ".") {
|
||
return ""
|
||
}
|
||
|
||
// 截取后缀
|
||
suffix := path.Ext(fileName)
|
||
if !withDot {
|
||
suffix = strings.TrimPrefix(suffix, ".")
|
||
}
|
||
return suffix
|
||
}
|
||
|
||
// ProcessPath 处理请求路径
|
||
// 1. 判断是否为合法完整请求路径(以/开头)
|
||
// 2. 包含 tenantId-1 则截断其及前面所有内容
|
||
func ProcessPath(ctx context.Context, path string) string {
|
||
// 判断是否是完整请求路径：以 / 开头
|
||
if strings.HasPrefix(path, "/") {
|
||
return path
|
||
}
|
||
|
||
target, err := utils.GetBucketName(ctx)
|
||
if err != nil {
|
||
return path
|
||
}
|
||
idx := strings.Index(path, target)
|
||
if idx == -1 {
|
||
// 不包含目标字符串，原样返回
|
||
return path
|
||
}
|
||
|
||
// 截取 tenantId-1 后面的内容
|
||
newPath := path[idx+len(target):]
|
||
return newPath
|
||
}
|
||
|
||
func BuildText(text string) string {
|
||
// 生成单条HTML
|
||
var htmlBuilder strings.Builder
|
||
htmlBuilder.WriteString(`
|
||
<!DOCTYPE html>
|
||
<html lang="zh-CN">
|
||
<head>
|
||
<meta charset="UTF-8">
|
||
<meta name="viewport" content="width=device-width, initial-scale=1.0">
|
||
<style>
|
||
* {
|
||
margin: 0;
|
||
padding: 0;
|
||
box-sizing: border-box;
|
||
}
|
||
body {
|
||
font-family: "Microsoft YaHei", "PingFang SC", Arial, sans-serif;
|
||
background: #f5f5f5;
|
||
color: #333;
|
||
line-height: 1.8;
|
||
padding: 20px;
|
||
}
|
||
.container {
|
||
max-width: 900px;
|
||
margin: 0 auto;
|
||
background: #fff;
|
||
border-radius: 12px;
|
||
box-shadow: 0 2px 12px rgba(0, 0, 0, 0.08);
|
||
overflow: hidden;
|
||
}
|
||
.item {
|
||
padding: 30px;
|
||
}
|
||
.image-group img {
|
||
width: 100%;
|
||
height: auto;
|
||
display: block;
|
||
margin-bottom: 6px;
|
||
border-radius: 8px;
|
||
box-shadow: 0 2px 8px rgba(0, 0, 0, 0.1);
|
||
}
|
||
.image-group img:last-child {
|
||
margin-bottom: 0;
|
||
}
|
||
.image-group {
|
||
margin-bottom: 25px;
|
||
}
|
||
.text {
|
||
padding: 0;
|
||
font-size: 15px;
|
||
line-height: 1.4;
|
||
color: #555;
|
||
}
|
||
.text h2 {
|
||
font-size: 28px;
|
||
font-weight: bold;
|
||
color: #1a1a1a;
|
||
margin-bottom: 15px;
|
||
line-height: 1.2;
|
||
}
|
||
.text h3 {
|
||
font-size: 20px;
|
||
font-weight: 600;
|
||
color: #2c3e50;
|
||
margin: 20px 0 12px;
|
||
padding-left: 12px;
|
||
border-left: 4px solid #409eff;
|
||
}
|
||
.text p {
|
||
margin-bottom: 12px;
|
||
text-align: justify;
|
||
}
|
||
.text strong {
|
||
color: #e74c3c;
|
||
font-weight: 600;
|
||
}
|
||
.text ul {
|
||
list-style: none;
|
||
padding: 0;
|
||
margin: 8px 0;
|
||
}
|
||
.text ul li {
|
||
padding: 10px 0 10px 30px;
|
||
position: relative;
|
||
line-height: 1.2;
|
||
}
|
||
.text ul li:before {
|
||
content: "●";
|
||
color: #409eff;
|
||
font-size: 12px;
|
||
position: absolute;
|
||
left: 12px;
|
||
top: 12px;
|
||
}
|
||
@media (max-width: 768px) {
|
||
body {
|
||
padding: 10px;
|
||
}
|
||
.text h2 {
|
||
font-size: 24px;
|
||
}
|
||
.text h3 {
|
||
font-size: 18px;
|
||
}
|
||
}
|
||
</style>
|
||
</head>
|
||
<body>
|
||
<div class="container">
|
||
<div class="item">
|
||
`)
|
||
// 🔥 写入文案前：删除 <p class="image-count">需要配图：X 张</p>
|
||
if text != "" {
|
||
// 写入清理后的文案
|
||
htmlBuilder.WriteString(fmt.Sprintf(`<div class="text">%s</div>`, ImageTagRegex(text)))
|
||
}
|
||
htmlBuilder.WriteString(`</div>
|
||
</div>
|
||
</body>
|
||
</html>`)
|
||
|
||
return htmlBuilder.String()
|
||
}
|
||
|
||
func BuildHtml(text string, images []string) string {
|
||
var htmlBuilder strings.Builder
|
||
htmlBuilder.WriteString(`<!DOCTYPE html>
|
||
<html lang="zh-CN">
|
||
<head>
|
||
<meta charset="UTF-8">
|
||
<meta name="viewport" content="width=device-width, initial-scale=1.0">
|
||
<style>
|
||
* {
|
||
margin: 0;
|
||
padding: 0;
|
||
box-sizing: border-box;
|
||
}
|
||
body {
|
||
font-family: "Microsoft YaHei", sans-serif;
|
||
padding: 20px;
|
||
background-color: #f6f6f6;
|
||
line-height: 1.7;
|
||
font-size: 16px;
|
||
color: #333;
|
||
}
|
||
.container {
|
||
max-width: 750px;
|
||
margin: 0 auto;
|
||
background: #fff;
|
||
padding: 30px;
|
||
border-radius: 12px;
|
||
box-shadow: 0 2px 10px rgba(0, 0, 0, 0.06);
|
||
}
|
||
|
||
</style>
|
||
</head>
|
||
<body>
|
||
<div class="container">
|
||
`)
|
||
// 写入图片（支持0张、1张、多张）
|
||
if len(images) > 0 {
|
||
htmlBuilder.WriteString(`<div class="image-group">`)
|
||
for _, imgUrl := range images {
|
||
htmlBuilder.WriteString(fmt.Sprintf(`<img src="%s" alt="图片"/>`, imgUrl))
|
||
}
|
||
htmlBuilder.WriteString(`</div>`)
|
||
}
|
||
htmlBuilder.WriteString(`
|
||
<div id="content">加载中...</div>
|
||
</div>
|
||
|
||
<script>
|
||
const incUrl = "` + text + `";
|
||
fetch(incUrl)
|
||
.then(res => {
|
||
if (!res.ok) throw new Error("加载失败");
|
||
return res.text();
|
||
})
|
||
.then(text => {
|
||
document.getElementById("content").innerHTML = text;
|
||
})
|
||
.catch(err => {
|
||
document.getElementById("content").innerHTML = "加载失败：" + err.message;
|
||
});
|
||
</script>
|
||
</body>
|
||
</html>`)
|
||
|
||
return htmlBuilder.String()
|
||
}
|
||
|
||
// ExtractImageCount 修复：支持单引号/双引号 + 换行 + 空格
|
||
func ExtractImageCount(content string) int {
|
||
// 🔥 关键：支持 class='image-count' （单引号）
|
||
re := regexp.MustCompile(`<p class=['"]image-count['"][^>]*>.*?(\d+).*?</p>`)
|
||
match := re.FindStringSubmatch(content)
|
||
if len(match) >= 2 {
|
||
num, err := strconv.Atoi(match[1])
|
||
if err == nil {
|
||
return num
|
||
}
|
||
}
|
||
return 0
|
||
}
|
||
|
||
func ImageTagRegex(html string) string {
|
||
// 🔥 修复：支持单引号、双引号、空格、换行，100% 删除 <p class='image-count'>
|
||
imageTagRegex := regexp.MustCompile(`<p class=['"]image-count['"][^>]*>[\s\S]*?</p>`)
|
||
return imageTagRegex.ReplaceAllString(html, "")
|
||
}
|
||
|
||
// StripHtmlTags 去掉所有HTML标签，保留换行和文本结构，并删除配图标记行
|
||
func StripHtmlTags(html string) string {
|
||
// 1. 替换块级标签为换行，保证排版
|
||
blockTags := regexp.MustCompile(`</?(div|p|h1|h2|h3|h4|h5|h6|li|ul|ol|br|tr|td|th)[^>]*>`)
|
||
text := blockTags.ReplaceAllString(html, "\n")
|
||
|
||
// 2. 去掉所有剩余的 HTML 标签
|
||
allTags := regexp.MustCompile(`<[^>]+>`)
|
||
text = allTags.ReplaceAllString(text, "")
|
||
|
||
// 4. 清理多余空行（多个换行只保留一个）
|
||
text = regexp.MustCompile(`\n\s*\n`).ReplaceAllString(text, "\n")
|
||
|
||
// 5. 只去掉首尾空白，中间换行保留
|
||
text = strings.TrimSpace(text)
|
||
|
||
return text
|
||
}
|
||
|
||
// SplitMultiContents 拆分模型返回的多条文案（基于HTML标签分隔）
|
||
func SplitMultiContents(htmlContent string) []string {
|
||
var contents []string
|
||
// 正则匹配<div class="content-item" id="content-{序号}">包裹的内容
|
||
re := regexp.MustCompile(`<div class="content-item" id="content-\d+">([\s\S]*?)</div>`)
|
||
matches := re.FindAllStringSubmatch(htmlContent, -1)
|
||
for _, match := range matches {
|
||
if len(match) > 1 {
|
||
// 清理空内容
|
||
trimmed := strings.TrimSpace(match[1])
|
||
if trimmed != "" {
|
||
contents = append(contents, trimmed)
|
||
}
|
||
}
|
||
}
|
||
// 兜底：如果没有匹配到结构化内容，按换行/分隔符拆分
|
||
if len(contents) == 0 {
|
||
contents = strings.Split(htmlContent, "===分隔符===") // 提示词中可新增此兜底规则
|
||
}
|
||
return contents
|
||
}
|
||
|
||
// GetAllImgSrcFromHtml 先把提取img src的工具方法放在外面
|
||
func GetAllImgSrcFromHtml(html string) []string {
|
||
var imgSrcList []string
|
||
re := regexp.MustCompile(`<img[^>]*src\s*=\s*["']([^"']+)["']`)
|
||
submatch := re.FindAllStringSubmatch(html, -1)
|
||
for _, match := range submatch {
|
||
if len(match) >= 2 {
|
||
imgSrcList = append(imgSrcList, match[1])
|
||
}
|
||
}
|
||
return imgSrcList
|
||
}
|
||
|
||
// ReplaceImgSrc 替换img src的方法
|
||
func ReplaceImgSrc(html string, oldSrc string, newSrc string) string {
|
||
// 精准替换：找到 <img xxx src="oldSrc" xxx>
|
||
re := regexp.MustCompile(`(<img[^>]*src\s*=\s*["'])` + regexp.QuoteMeta(oldSrc) + `(["'])`)
|
||
return re.ReplaceAllString(html, `${1}`+newSrc+`${2}`)
|
||
}
|