diff --git a/config.yml b/config.yml index 3d1734b..62bea00 100644 --- a/config.yml +++ b/config.yml @@ -33,14 +33,14 @@ database: user: "postgres" pass: "Bjang09@686^*^" name: "model-gateway" - prefix: "" + prefix: "model_gateway_" role: "master" debug: true dryRun: false charset: "utf8" timezone: "Asia/Shanghai" - maxIdle: 5 - maxOpen: 20 + maxIdle: 15 + maxOpen: 60 maxLifetime: "30s" maxIdleConnTime: "30s" createdAt: "created_at" @@ -59,6 +59,10 @@ consul: jaeger: addr: 192.168.0.83:4318 +nats: + addr: 192.168.0.83 + port: 4222 + # 本地调试用：可选自动执行 worker/cleaner（默认关闭） asynch: queryPending: diff --git a/go.mod b/go.mod index 622871d..0202c85 100644 --- a/go.mod +++ b/go.mod @@ -3,20 +3,15 @@ module model-gateway go 1.26.1 require ( - gitea.redpowerfuture.com/red-future/common v0.0.23 - github.com/bjang03/gmq v0.0.1 + gitea.redpowerfuture.com/red-future/common v0.0.31 + github.com/bjang03/gmq v0.0.2 github.com/gogf/gf/contrib/drivers/pgsql/v2 v2.10.2 github.com/gogf/gf/contrib/nosql/redis/v2 v2.10.2 github.com/gogf/gf/v2 v2.10.2 github.com/google/uuid v1.6.0 github.com/tidwall/gjson v1.19.0 - github.com/tidwall/sjson v1.2.5 ) -replace gitea.redpowerfuture.com/red-future/common v0.0.23 => ../common - -replace github.com/bjang03/gmq v0.0.1 => ../gmq - require ( github.com/BurntSushi/toml v1.5.0 // indirect github.com/armon/go-metrics v0.4.1 // indirect @@ -81,7 +76,8 @@ require ( github.com/rivo/uniseg v0.4.7 // indirect github.com/spf13/cast v1.10.0 // indirect github.com/tidwall/match v1.1.1 // indirect - github.com/tidwall/pretty v1.2.0 // indirect + github.com/tidwall/pretty v1.2.1 // indirect + github.com/tidwall/sjson v1.2.5 // indirect github.com/tiger1103/gfast-token v1.0.10 // indirect github.com/vcaesar/cedar v0.30.0 // indirect github.com/vmihailenco/msgpack v4.0.4+incompatible // indirect diff --git a/go.sum b/go.sum index 0433054..ff40f00 100644 --- a/go.sum +++ b/go.sum @@ -1,4 +1,6 @@ cloud.google.com/go v0.26.0/go.mod h1:aQUYkXzVsufM+DwF1aE+0xfcU+56JwCaLick0ClmMTw= +gitea.redpowerfuture.com/red-future/common v0.0.31 h1:9H8nL5Drazcv7Hs9d4j+cXhaB+7uOllIUqEOyZy1Eao= +gitea.redpowerfuture.com/red-future/common v0.0.31/go.mod h1:xPU7aaMxn8rtNnWc2LDUXZL+IkaUkpQeLgflqw9FvdU= github.com/BurntSushi/toml v0.3.1/go.mod h1:xHWCNGjB5oqiDr8zfno3MHue2Ht5sIBksp03qcyfWMU= github.com/BurntSushi/toml v1.5.0 h1:W5quZX/G/csjUnuI8SUYlsHs9M38FC7znL0lIO+DvMg= github.com/BurntSushi/toml v1.5.0/go.mod h1:ukJfTF/6rtPPRCnwkur4qwRxa8vTRFBF0uk2lLoLwho= @@ -19,6 +21,8 @@ github.com/beorn7/perks v1.0.1/go.mod h1:G2ZrVWU2WbWT9wwq4/hrbKbnv/1ERSJQ0ibhJ6r github.com/bgentry/speakeasy v0.1.0/go.mod h1:+zsyZBPWlz7T6j88CTgSN5bM796AkVf0kBD4zp0CCIs= github.com/bitly/go-simplejson v0.5.1 h1:xgwPbetQScXt1gh9BmoJ6j9JMr3TElvuIyjR8pgdoow= github.com/bitly/go-simplejson v0.5.1/go.mod h1:YOPVLzCfwK14b4Sff3oP1AmGhI9T9Vsg84etUnlyp+Q= +github.com/bjang03/gmq v0.0.2 h1:3CcVorDXYoRIN65bbzwRuUxzkBCkEpHWmKHOkfXzUo0= +github.com/bjang03/gmq v0.0.2/go.mod h1:Y7TwWGuV4Cw97WUDaM7x+NC4kyFx1z44WAvNwJV3HV8= github.com/bsm/ginkgo/v2 v2.12.0 h1:Ny8MWAHyOepLGlLKYmXG4IEkioBysk6GpaRTLC8zwWs= github.com/bsm/ginkgo/v2 v2.12.0/go.mod h1:SwYbGRRDovPVboqFv0tPTcG1sN61LM1Z4ARdbAV9g4c= github.com/bsm/gomega v1.27.10 h1:yeMWxP2pV2fG3FgAODIY8EiRE3dy0aeFYt4l7wh6yKA= @@ -323,6 +327,8 @@ github.com/tidwall/match v1.1.1 h1:+Ho715JplO36QYgwN9PGYNhgZvoUSc9X2c80KVTi+GA= github.com/tidwall/match v1.1.1/go.mod h1:eRSPERbgtNPcGhD8UCthc6PmLEQXEWd3PRB5JTxsfmM= github.com/tidwall/pretty v1.2.0 h1:RWIZEg2iJ8/g6fDDYzMpobmaoGh5OLl4AXtGUGPcqCs= github.com/tidwall/pretty v1.2.0/go.mod h1:ITEVvHYasfjBbM0u2Pg8T2nJnzm8xPwvNhhsoaGGjNU= +github.com/tidwall/pretty v1.2.1 h1:qjsOFOWWQl+N3RsoF5/ssm1pHmJJwhjlSbZ51I6wMl4= +github.com/tidwall/pretty v1.2.1/go.mod h1:ITEVvHYasfjBbM0u2Pg8T2nJnzm8xPwvNhhsoaGGjNU= github.com/tidwall/sjson v1.2.5 h1:kLy8mja+1c9jlljvWTlSazM7cKDRfJuR/bOJhcY5NcY= github.com/tidwall/sjson v1.2.5/go.mod h1:Fvgq9kS/6ociJEDnK0Fk1cpYF4FIW6ZF7LAe+6jwd28= github.com/tiger1103/gfast-token v1.0.10 h1:fNiBE/Dq5iTHvTGlCx3DmXa2o4hr0NtumFpffZ39k6s= diff --git a/model/domain/schema_mapping.go b/model/domain/schema_mapping.go index 2274808..e787af4 100644 --- a/model/domain/schema_mapping.go +++ b/model/domain/schema_mapping.go @@ -29,6 +29,7 @@ type ChatFieldsRes struct { // VideoFields 视频模型业务字段映射 // 适用于 视频模型(600) 及其子类型 type VideoFields struct { + SystemPrompt string `json:"system_prompt" dc:"系统提示词,用于描述系统的需求"` MinDuration string `json:"min_duration" dc:"模型支持的最小视频时长（秒）"` MaxDuration string `json:"max_duration" dc:"模型支持的最大视频时长（秒）"` FirstFrame string `json:"first_frame" dc:"视频的首帧/初始画面，传入一张图片作为视频第一帧画面"` diff --git a/model/dto/model_call_dto.go b/model/dto/model_call_dto.go index 27ca8b1..af5beb9 100644 --- a/model/dto/model_call_dto.go +++ b/model/dto/model_call_dto.go @@ -69,6 +69,11 @@ var ModelErrorResp struct { } `json:"error"` } +var ModelError1Resp struct { + Code int `json:"code"` + Message string `json:"message"` +} + type ModelMsg struct { TaskID int64 `json:"id" dc:"任务ID"` State task.Status `json:"state" dc:"状态"` diff --git a/model/entity/model_manage.go b/model/entity/model_manage.go index 9b34d09..0a8170e 100644 --- a/model/entity/model_manage.go +++ b/model/entity/model_manage.go @@ -90,6 +90,7 @@ type ModelManage struct { MinDuration int `orm:"min_duration" json:"minDuration" description:"最小时长（秒）"` MaxDuration int `orm:"max_duration" json:"maxDuration" description:"最大时长（秒）"` LastFrame string `orm:"last_frame" json:"lastFrame" description:"视频的尾帧图像"` + ErrorMessageMapping map[string]any `orm:"error_message_mapping" json:"errorMessageMapping" description:"错误消息映射"` } type TokenMapping struct { @@ -101,6 +102,7 @@ type TokenMapping struct { type AsyncTaskMapping struct { Url string `json:"url" dc:"url"` HttpMethod string `json:"httpMethod" dc:"http方法" d:"POST"` + RequestBodyMapping map[string]any `json:"requestBodyMapping" description:"请求体映射"` RequestHeadMapping map[string]string `json:"requestHeadMapping" description:"请求头映射"` ResponseMapping map[string]any `json:"responseMapping" description:"响应映射"` TaskId string `json:"taskId" dc:"任务id"` diff --git a/service/model_call_service.go b/service/model_call_service.go index 971e954..d3aa2e9 100644 --- a/service/model_call_service.go +++ b/service/model_call_service.go @@ -378,20 +378,39 @@ LOOP: // 3. businessParams 只装业务字段，按业务字段名（RequestBusinessFieldMapping 的 key）传值， // TakeBusinessFields 解析为写入路径，构建完成后由 WriteBusinessFields 按路径写入最终请求体 func buildChatRequestParams(modelInfo *entity.ModelManage, requestParams, businessParams map[string]any) (map[string]any, error) { + + // requestParams 可能混有扁平路径 key（messages.enumValues...）与已是对象/数组的值（stream）。 + // IsFlatMap 遇 map/slice 值即整体返回 false 会跳过 unflatten；sjson.Set 能处理任意值类型作为子树， + // 带点 key 按路径展开、无点 key 直接赋值，故始终 unflatten + var err error + requestParams, err = utils.UnFlatBySjson(requestParams) + if err != nil { + return nil, err + } + rest := make(map[string]any, len(requestParams)) for k, v := range requestParams { rest[k] = v } // 请求结构源：模板字段兜底（模板 value/defaultValue 生效）+ requestParams 覆盖同名； // 保证模板声明的结构字段（如 stream_options 对象）即使 requestParams 未传也进请求体 + // 用户已传字段按模板类型元数据递归合并（补 type 包装、补默认字段），数组字段仅在用户提供时才合并 src := rest for k, v := range modelInfo.RequestBodyMapping { - mapV := gconv.Map(v) - if mapV["type"] == "array" { - continue + tmplMap, _ := v.(map[string]any) + if tmplMap != nil { + if t, _ := tmplMap["type"].(string); t == "array" { + if _, has := src[k]; !has { + continue + } + src[k] = modelUtils.MergeNode(src[k], v) + continue + } } - if _, has := src[k]; !has { - src[k] = v + if userVal, has := src[k]; has { + src[k] = modelUtils.MergeNode(userVal, v) + } else { + src[k] = modelUtils.DeepCopyNode(v) } } if len(requestParams) > 0 { diff --git a/service/model_session_service.go b/service/model_session_service.go index aeb1e96..b0c5fff 100644 --- a/service/model_session_service.go +++ b/service/model_session_service.go @@ -22,7 +22,7 @@ var ModelSession = &modelSessionService{} type modelSessionService struct{} // modelCallMaxRetries 上游调用最大重试次数 -const modelCallMaxRetries = 3 +const modelCallMaxRetries = 15 // CreateSession 创建会话 func (s *modelSessionService) CreateSession(ctx context.Context, req *dto.CallModelSessionReq) (res *dto.ModelCallRes, err error) { @@ -80,7 +80,7 @@ LOOP: if *model.ResponseTypeSync.Code() == *modelInfo.ResponseType { respBodyMap := make(map[string]string, len(modelInfo.ResponseBodyMapping)) for k, _ := range modelInfo.ResponseBodyMapping { - respBodyMap[modelUtils.CleanFieldPath(k)] = modelUtils.CleanFieldPath(k) + respBodyMap[k] = modelUtils.CleanFieldPath(k) } // 基于统一字段路径（GetByPath）按映射取值组装结果 var respObj map[string]any @@ -89,7 +89,7 @@ LOOP: } content := make(map[string]any, len(respBodyMap)) for bizKey, jsonPath := range respBodyMap { - content[bizKey] = modelUtils.GetByPathValue(respObj, jsonPath) + content[bizKey] = uploadTempURLToOSS(ctx, modelUtils.GetByPathValue(respObj, jsonPath)) } businessField := make(map[string]any, len(modelInfo.ResponseBusinessFieldMapping)) @@ -139,7 +139,8 @@ LOOP: return docMsg, nil } -// CreateSessionStreamOnce 流式调用上游模型 → 缓冲全量后一次返回（走 gf 框架正常返回） +// CreateSessionStreamOnce 流式调用上游模型 → 缓冲全量后一次返回（走 gf 框架正常返回）。 +// 与同步请求一致：上游返回可重试错误码（限流/5xx）时按指数退避重试（最多 modelCallMaxRetries 次）。 func (s *modelSessionService) CreateSessionStreamOnce(ctx context.Context, req *dto.CallModelSessionReq) (docMsg *dto.ModelCallRes, err error) { startTime := time.Now() @@ -147,9 +148,23 @@ func (s *modelSessionService) CreateSessionStreamOnce(ctx context.Context, req * modelInfo := req.ModelInfo newRequestParams := req.RequestParams - // 获取上游流式 reader（stream=false → w 不会被使用，传 nil） + attempt := 0 +LOOP: + // 获取上游流式 reader（stream=false → w 不会被使用，传 nil）。 + // 非 2xx 状态/网络错误在此返回；错误含可重试错误码（限流/5xx）时按指数退避重试，与同步请求一致。 streamReader, err := ModelHttpStreamRequest(ctx, nil, modelInfo.BaseURL, modelInfo.RequestHeadMapping, modelInfo.HttpMethod, newRequestParams) if err != nil { + if retryCode := streamRetryCodeOfError(err); retryCode != "" && attempt < modelCallMaxRetries { + attempt++ + wait := time.Duration(1< 0 { realText = gconv.String(arr[0]) @@ -194,8 +218,29 @@ func (s *modelSessionService) CreateSessionStreamOnce(ctx context.Context, req * return nil }) - // 流结束后组装 - docMsg.Content = map[string]any{"respBody": contentBuf.String()} + // 流内返回可重试错误码：丢弃本次部分内容，指数退避后重新请求 + if streamErrCode != "" { + if attempt < modelCallMaxRetries && isRetryableErrorCode(streamErrCode) { + attempt++ + wait := time.Duration(1< 0 { realText = gconv.String(arr[0]) @@ -398,8 +445,57 @@ func (s *modelSessionService) CreateSessionStream(ctx context.Context, w http.Re // ModelHttpNormalRequest 不返回 HTTP status，只能按响应体 error.code 字符串判定。 func isRetryableErrorCode(code string) bool { switch code { - case "429", "500", "501", "502", "503", "limit_requests", "limit_tokens", "rate_limit_exceeded": + case "429", "500", "501", "502", "503", "InvalidParameter", "limit_requests", "limit_tokens", "rate_limit_exceeded": return true } return false } + +// streamErrorOfChunk 从流式分片提取错误码与消息：优先 OpenAI 兼容 error 事件，顶层 code 兜底。 +func streamErrorOfChunk(chunk map[string]any) (code, msg string) { + if errObj := gconv.Map(chunk["error"]); errObj != nil { + code = gconv.String(errObj["code"]) + msg = gconv.String(errObj["message"]) + } + if code == "" { + code = gconv.String(chunk["code"]) + } + return +} + +// streamRetryCodeOfError 从流式请求错误中提取可重试错误码：优先解析错误体 error.code/顶层 code， +// 其次取非 2xx 的 HTTP 状态码；纯网络错误等无错误码场景返回空串（与同步请求一致，不重试）。 +func streamRetryCodeOfError(err error) string { + if err == nil { + return "" + } + msg := err.Error() + // 非 2xx 时 ModelHttpStreamRequest 返回 "[HTTP][Stream] 状态码异常: %d, body={...}" + if idx := strings.Index(msg, "body="); idx >= 0 { + body := msg[idx+len("body="):] + var errResp struct { + Error struct { + Code string `json:"code"` + } `json:"error"` + Code string `json:"code"` + } + if json.Unmarshal([]byte(body), &errResp) == nil { + if errResp.Error.Code != "" { + return errResp.Error.Code + } + if errResp.Code != "" { + return errResp.Code + } + } + } + if idx := strings.Index(msg, "状态码异常: "); idx >= 0 { + codeStr := strings.TrimSpace(msg[idx+len("状态码异常: "):]) + if comma := strings.IndexByte(codeStr, ','); comma >= 0 { + codeStr = codeStr[:comma] + } + if isRetryableErrorCode(codeStr) { + return codeStr + } + } + return "" +} diff --git a/service/model_task_end_service.go b/service/model_task_end_service.go index 6d6b7ae..9e4e781 100644 --- a/service/model_task_end_service.go +++ b/service/model_task_end_service.go @@ -139,12 +139,14 @@ func (s *modelTaskEndService) handleSingleTask(ctx context.Context, item *entity LOOP: // 替换URL占位符 url := urlParamReg.ReplaceAllString(modelInfo.AsyncTaskMapping.Url, item.TaskId) + // 组装查询请求体：POST 查询接口需要 body（从 RequestBodyMapping 出发，替换 {…} 占位符为任务 ID） + reqBody := buildAsyncTaskBody(modelInfo.AsyncTaskMapping.RequestBodyMapping, item.TaskId) // 发起HTTP请求 modelRespBody, err := ModelHttpNormalRequest( asyncCtx, url, modelInfo.AsyncTaskMapping.RequestHeadMapping, - modelInfo.AsyncTaskMapping.HttpMethod, nil, + modelInfo.AsyncTaskMapping.HttpMethod, reqBody, ) if err != nil { g.Log().Errorf(asyncCtx, "模型请求失败: %v", err) @@ -251,6 +253,42 @@ func (s *modelTaskEndService) handleSingleTask(ctx context.Context, item *entity }) } +// buildAsyncTaskBody 组装异步任务查询请求体：从 AsyncTaskMapping.RequestBodyMapping 出发， +// 把 {…} 占位符（如 {taskId}）替换为实际任务 ID，兼容 POST 查询接口需要请求体的场景。 +// 未配置映射时返回 nil（GET 查询/无需 body 的场景）。 +func buildAsyncTaskBody(mapping map[string]any, taskID string) map[string]any { + if len(mapping) == 0 { + return nil + } + out := make(map[string]any, len(mapping)) + for k, v := range mapping { + out[k] = replaceTaskPlaceholder(v, taskID) + } + return out +} + +// replaceTaskPlaceholder 递归替换结构体中的 {…} 占位符为任务 ID +func replaceTaskPlaceholder(v any, taskID string) any { + switch val := v.(type) { + case string: + return urlParamReg.ReplaceAllString(val, taskID) + case map[string]any: + m := make(map[string]any, len(val)) + for k, x := range val { + m[k] = replaceTaskPlaceholder(x, taskID) + } + return m + case []any: + arr := make([]any, len(val)) + for i, x := range val { + arr[i] = replaceTaskPlaceholder(x, taskID) + } + return arr + default: + return v + } +} + // tempDownloadTimeout 临时路径下载超时 const tempDownloadTimeout = 5 * time.Minute @@ -304,7 +342,8 @@ func uploadSingleURL(ctx context.Context, rawURL string) (string, bool) { g.Log().Errorf(ctx, "临时路径转存OSS失败: url=%s err=%v", rawURL, err) return rawURL, false } - return ossRes.FileAddressPrefix + ossRes.FileURL, true + //ossRes.FileAddressPrefix + ossRes.FileURL, true + return ossRes.FileURL, true } func isHTTPURL(s string) bool { diff --git a/service/model_task_start_service.go b/service/model_task_start_service.go index 579a786..2374164 100644 --- a/service/model_task_start_service.go +++ b/service/model_task_start_service.go @@ -49,10 +49,23 @@ func (s *modelTaskStartService) CreateTask(ctx context.Context, req *dto.CallMod if err != nil { return nil, fmt.Errorf("模型返回参数解析失败：%v", err) } - if errMsg.Error.Code != "" { - docMsg.ErrorMsg = errMsg.Error.Message - updateModelReq.ErrorMsg = docMsg.ErrorMsg + if g.IsEmpty(errMsg.Error.Code) { + errMsg1 := new(dto.ModelError1Resp) + err = gconv.Struct(modelRespBody, errMsg1) + if err != nil { + return nil, fmt.Errorf("模型返回参数解析失败：%v", err) + } + if !g.IsEmpty(errMsg1.Code) && errMsg1.Code != 20000000 { + docMsg.ErrorMsg = errMsg1.Message + updateModelReq.ErrorMsg = docMsg.ErrorMsg + } } else { + if errMsg.Error.Code != "" { + docMsg.ErrorMsg = errMsg.Error.Message + updateModelReq.ErrorMsg = docMsg.ErrorMsg + } + } + if docMsg.ErrorMsg == "" { taskIDPath := modelUtils.CleanFieldPath(modelInfo.AsyncTaskMapping.TaskId) docMsg.Content = map[string]any{ "respBody": modelUtils.GetByPathValue(respObj, taskIDPath), diff --git a/service/utils/business_fields.go b/service/utils/business_fields.go index 2e13ea8..d4d046f 100644 --- a/service/utils/business_fields.go +++ b/service/utils/business_fields.go @@ -86,16 +86,16 @@ func isEmptyBizValue(v any) bool { // // SetByPath（写，构建请求体）与 GetByPath（读，解析响应）共用 parsePath； // 读方向语义：数组段非末段取第 0 个元素继续下钻，[*] 为末段返回整个数组，选择器定位匹配元素； -// 写方向语义：数组段非末段作用于最后一个元素，末段追加，选择器 upsert（命中更新/未命中新建）， +// 写方向语义：数组段非末段作用于最后一个元素，末段前置追加（业务值在前），选择器 upsert（命中更新/未命中新建）， // 多值（[]any）仅在叶子选择器展开为多个独立元素（多个参考图/视频等）。 // ============================================================ // SetByPath 按业务字段映射路径把值写入请求结构（请求侧构建）。 // 路径语法与 BuildSchemaMapping 输出一致（干净形态，无需 attrs 剔除）。 // -// 写入语义： -// - 目标字段已存在且是数组 → 追加 -// - 目标字段已存在且非数组 → 覆盖（普通叶子路径） +// 写入语义（业务字段一律前置追加，不覆盖已有值）： +// - 目标字段已存在且是数组 → 业务值（或其元素）前置插入数组头部，原元素依次后移 +// - 目标字段已存在且非数组（普通叶子路径）→ 业务值前置拼接（字符串拼接/数组包裹），原值保留在后 // - 目标字段不存在 → 新建 // - 数组段/选择器段目标字段不是数组（如字符串 content）→ 返回错误，不覆盖已有值 // - 中间路径遇到非对象字段 → 返回错误 @@ -103,7 +103,7 @@ func isEmptyBizValue(v any) bool { // - 数组段带选择器 → 命中则更新目标字段，未命中则按选择器字段构造新元素追加； // 选择器段即使未标 [*] 也按数组处理（如 input.media?type=first_frame&url=#） // - 多级选择器 → 递归：中间级选择器定位/新建容器元素并继续下钻，叶子选择器写值 -// - 值本身是数组 → 叶子选择器逐值追加；普通数组段/点号路径按 appendValues 追加 +// - 值本身是数组 → 叶子选择器逐值追加；普通数组段/点号路径前置追加（业务值在前，原值保留在后，不覆盖） func SetByPath(root map[string]any, path string, value any) error { steps := parsePath(NormalizeFieldPath(path)) if len(steps) == 0 { @@ -146,7 +146,7 @@ func setBySteps(cur map[string]any, steps []step, value any) error { return err } if last { - cur[first.key] = appendValues(arr, value) + cur[first.key] = prependToArray(arr, value) return nil } // 无选择器数组段：作用于最后一个元素（追加语义） @@ -572,23 +572,46 @@ func setLeafPath(m map[string]any, path string, value any) error { return nil } -// setLeaf 叶子写入：目标已是数组 → 追加；非数组/不存在 → 覆盖/新建 +// setLeaf 叶子写入（业务字段语义：值前置追加而非覆盖）： +// 字段不存在 → 直接写入；已有值 → 业务字段值前置到原有值前面（数组前插/字符串拼接），原值保留在后。 func setLeaf(parent map[string]any, key string, value any) { - if existing, ok := parent[key]; ok { - if arr, isArr := existing.([]any); isArr { - parent[key] = appendValues(arr, value) - return - } + existing, ok := parent[key] + if !ok { + parent[key] = value + return } - parent[key] = value + parent[key] = prependValue(value, existing) } -// appendValues 追加值到数组；value 为数组时逐个追加 -func appendValues(arr []any, value any) []any { - if vals, ok := value.([]any); ok { - return append(arr, vals...) +// prependValue 把 value 前置到 existing 前（业务值在前，原值保留在后，不覆盖）： +// - existing 是数组 → value（或其元素）前插到数组头部 +// - value 是数组（existing 为标量）→ value 各元素在前，existing 作为末位元素 +// - 其余标量 → 字符串拼接，业务值在前 +func prependValue(value, existing any) any { + if arr, isArr := existing.([]any); isArr { + return prependToArray(arr, value) } - return append(arr, value) + if vals, isArr := value.([]any); isArr { + out := make([]any, 0, len(vals)+1) + out = append(out, vals...) + out = append(out, existing) + return out + } + return gconv.String(value) + gconv.String(existing) +} + +// prependToArray 把 value（或其元素）插到数组头部，原元素依次后移 +func prependToArray(arr []any, value any) []any { + if vals, isArr := value.([]any); isArr { + out := make([]any, 0, len(vals)+len(arr)) + out = append(out, vals...) + out = append(out, arr...) + return out + } + out := make([]any, 0, len(arr)+1) + out = append(out, value) + out = append(out, arr...) + return out } // ensureMap 确保键对应 map，不存在则新建；已存在但非对象 → 返回错误 diff --git a/service/utils/json_flatten.go b/service/utils/json_flatten.go deleted file mode 100644 index 29af6c5..0000000 --- a/service/utils/json_flatten.go +++ /dev/null @@ -1,43 +0,0 @@ -package utils - -import ( - "encoding/json" - "fmt" - - "github.com/tidwall/sjson" -) - -// IsFlatMap 递归判断 map 是否扁平化 -func IsFlatMap(m map[string]interface{}) bool { - for _, v := range m { - switch val := v.(type) { - case map[string]interface{}: - return false - case []interface{}: - for _, item := range val { - if _, ok := item.(map[string]interface{}); ok { - return false - } - } - } - } - return true -} - -// UnFlatBySjson 将扁平路径映射还原为嵌套 JSON -func UnFlatBySjson(flatMap map[string]interface{}) (map[string]interface{}, error) { - raw := "{}" - for path, val := range flatMap { - var err error - raw, err = sjson.Set(raw, path, val) - if err != nil { - return nil, fmt.Errorf("sjson set path %s failed: %w", path, err) - } - } - - var result map[string]interface{} - if err := json.Unmarshal([]byte(raw), &result); err != nil { - return nil, fmt.Errorf("parse final json failed: %w", err) - } - return result, nil -} diff --git a/service/utils/template_merge.go b/service/utils/template_merge.go new file mode 100644 index 0000000..74272e1 --- /dev/null +++ b/service/utils/template_merge.go @@ -0,0 +1,260 @@ +package utils + +import "github.com/gogf/gf/v2/util/gconv" + +// MergeNode 按模板节点的元数据把用户数据节点合并为模板可解析的节点： +// - 用户节点无 type 键 → 套上模板类型包装（标量取用户值，空则回落模板 value/defaultValue） +// - 用户节点有 type 键 → 递归合并子节点（子节点缺 type 时同样补模板类型，避免模板包装泄漏） +// - 对象：模板 attrs 缺省字段补进用户容器（保留模板默认值） +// - 数组：用户 enumValues/attrs 元素逐个与模板元素原型合并（模板未提供的槽位克隆原型） +func MergeNode(user, tmpl any) any { + tmplMap, ok := tmpl.(map[string]any) + if !ok { + return user + } + fieldType, _ := tmplMap["type"].(string) + + // 用户节点已带 type：按用户类型递归合并子节点；标量叶子已完备，直接返回 + if userMap, ok := user.(map[string]any); ok { + if ut, isTpl := userMap["type"]; isTpl { + switch gconv.String(ut) { + case TypeObject: + return mergeObject(userMap, tmplMap) + case TypeArray: + return mergeArray(userMap, tmplMap) + default: + return userMap + } + } + } + + switch fieldType { + case TypeString, TypeBool, TypeNumber, TypeNumberInt, TypeNumberFloat: + return wrapScalarNode(tmplMap, user) + case TypeObject: + return mergeObject(user, tmplMap) + case TypeArray: + return mergeArray(user, tmplMap) + default: + return user + } +} + +// wrapScalarNode 把用户标量值包装为 {type, value} 节点；用户值为空时回落模板 value/defaultValue +func wrapScalarNode(tmplMap map[string]any, user any) map[string]any { + val := user + if m, ok := user.(map[string]any); ok { + if v, has := m["value"]; has { + val = v + } else { + val = nil + } + } + node := map[string]any{"type": normalizeScalarType(gconv.String(tmplMap["type"]))} + if hasUsableValue(val) { + node["value"] = val + } else if v, has := tmplMap["value"]; has { + node["value"] = v + } else if d, has := tmplMap["defaultValue"]; has { + node["value"] = d + } + return node +} + +// normalizeScalarType integer/float 统一为 number（resolveField 只分发 string/boolean/number） +func normalizeScalarType(t string) string { + if t == TypeNumberInt || t == TypeNumberFloat { + return TypeNumber + } + return t +} + +// mergeObject 合并对象节点：模板 attrs 缺省字段补进用户容器（模板默认值保留） +func mergeObject(user any, tmplMap map[string]any) map[string]any { + tmplAttrs, _ := tmplMap["attrs"].(map[string]any) + node := map[string]any{"type": TypeObject} + if tmplAttrs == nil { + node["attrs"] = user + return node + } + var container map[string]any + switch u := user.(type) { + case map[string]any: + if a, has := u["attrs"]; has { + if am, ok := a.(map[string]any); ok { + container = copyMap(am) + } else { + container = map[string]any{} + } + } else if v, has := u["value"]; has { + if vm, ok := v.(map[string]any); ok { + container = copyMap(vm) + } else { + container = map[string]any{} + } + } else { + container = copyMap(u) + } + default: + container = map[string]any{} + } + for k, subTmpl := range tmplAttrs { + if _, has := container[k]; !has { + container[k] = DeepCopyNode(subTmpl) + } else { + container[k] = MergeNode(container[k], subTmpl) + } + } + node["attrs"] = container + return node +} + +// mergeArray 合并数组节点：用户 enumValues/attrs 元素逐个与模板元素原型合并。 +// 用户未填的槽位（sjson null 填充）用模板对应槽位原型补位，保留下标不塌缩。 +func mergeArray(user any, tmplMap map[string]any) map[string]any { + proto := arrayElementTemplate(tmplMap) + tmplSlots, _ := tmplMap["enumValues"].([]any) // 模板各槽位原型，按下标一一对应 + node := map[string]any{"type": TypeArray} + slotProto := func(i int) any { + if i < len(tmplSlots) { + return tmplSlots[i] + } + return proto + } + switch u := user.(type) { + case map[string]any: + if evs, ok := u["enumValues"].([]any); ok { + out := make([]any, len(evs)) + for i, ev := range evs { + if ev == nil { + // 保留 null 槽位以维持下标；解析阶段 resolveArray 会丢弃空元素， + // 避免用模板原型填充 null 而物化出"幽灵"元素（用户未填的数组槽位） + out[i] = nil + continue + } + out[i] = mergeArrayElement(ev, slotProto(i)) + } + node["enumValues"] = out + } else if a, ok := u["attrs"].([]any); ok { + out := make([]any, len(a)) + for i, item := range a { + out[i] = mergeArrayElement(item, proto) + } + node["attrs"] = out + } else if v, has := u["value"]; has { + node["value"] = v + } else if len(u) > 0 { + node["attrs"] = []any{mergeArrayElement(u, proto)} + } + case []any: + if isSchemaElementList(u) { + out := make([]any, len(u)) + for i, item := range u { + out[i] = mergeArrayElement(item, proto) + } + node["attrs"] = out + } else { + node["value"] = u + } + default: + node["value"] = user + } + return node +} + +// mergeArrayElement 合并单个数组元素：元素缺 type 时按槽位原型补默认字段。 +// slotProto 可为 nil（模板未提供槽位原型时）→ 原样返回用户元素 +func mergeArrayElement(ev any, slotProto any) any { + if ev == nil { + return ev + } + proto, _ := slotProto.(map[string]any) + if proto == nil { + return ev + } + evMap, ok := ev.(map[string]any) + if !ok { + return ev + } + if _, isTpl := evMap["type"]; isTpl { + return ev + } + // 元素原型自身是包装节点（嵌套对象/数组）→ 按模板类型合并 + if t, has := proto["type"].(string); has && t != "" { + return MergeNode(ev, proto) + } + protoAttrs, _ := proto["attrs"].(map[string]any) + var container map[string]any + if a, has := evMap["attrs"]; has { + if am, ok := a.(map[string]any); ok { + container = copyMap(am) + } else { + container = map[string]any{} + } + } else { + container = copyMap(evMap) + } + for k, subTmpl := range protoAttrs { + if _, has := container[k]; !has { + container[k] = DeepCopyNode(subTmpl) + } else { + container[k] = MergeNode(container[k], subTmpl) + } + } + return map[string]any{"attrs": container} +} + +// isSchemaElementList 判断 []any 是否为 schema 元素列表（元素均为 map，且带 type 或 attrs 包装） +func isSchemaElementList(list []any) bool { + if len(list) == 0 { + return false + } + for _, item := range list { + m, ok := item.(map[string]any) + if !ok { + return false + } + if _, hasType := m["type"]; !hasType { + if _, hasAttrs := m["attrs"]; !hasAttrs { + return false + } + } + } + return true +} + +// DeepCopyNode 深拷贝任意嵌套节点（模板节点整体拷贝进 src 时使用，避免共享引用被后续回填修改） +func DeepCopyNode(v any) any { + switch val := v.(type) { + case map[string]any: + out := make(map[string]any, len(val)) + for k, sub := range val { + out[k] = DeepCopyNode(sub) + } + return out + case []any: + out := make([]any, len(val)) + for i, sub := range val { + out[i] = DeepCopyNode(sub) + } + return out + default: + return v + } +} + +// hasUsableValue 值是否可写入节点 value；0/false 视为有效值 +func hasUsableValue(v any) bool { + switch val := v.(type) { + case nil: + return false + case string: + return val != "" + case []any: + return len(val) > 0 + case map[string]any: + return len(val) > 0 + default: + return true + } +} diff --git a/service/utils/template_parse.go b/service/utils/template_parse.go index 9391066..bbba1a4 100644 --- a/service/utils/template_parse.go +++ b/service/utils/template_parse.go @@ -4,6 +4,7 @@ import ( "regexp" "strings" + "gitea.redpowerfuture.com/red-future/common/utils" "github.com/gogf/gf/v2/util/gconv" ) @@ -106,10 +107,10 @@ func CleanMapFieldPath(m map[string]string) map[string]string { // - 数组展开: enumValue 内某叶子字段 value 为数组时，按数组元素展开为多个项 func ParseConfigTemplate(cfg map[string]interface{}) map[string]interface{} { var flattenJSON map[string]interface{} - flatMap := IsFlatMap(cfg) + flatMap := utils.IsFlatMap(cfg) if flatMap { var err error - flattenJSON, err = UnFlatBySjson(cfg) + flattenJSON, err = utils.UnFlatBySjson(cfg) if err != nil { return nil } @@ -152,13 +153,13 @@ func resolveScalar(field map[string]interface{}) interface{} { if vv != "" { return vv } - case float64: - if vv != 0 { - return vv - } case bool: return vv default: + // 数值零值（int/float 各类型）视为未提供，跳过；bool/string 已在上方处理 + if isNumericZero(v) { + return nil + } return vv } } @@ -168,19 +169,50 @@ func resolveScalar(field map[string]interface{}) interface{} { if dv != "" { return dv } - case float64: - if dv != 0 { - return dv - } case bool: return dv default: + if isNumericZero(d) { + return nil + } return dv } } return nil } +// isNumericZero 判断是否为数值零值（模板 value 常为 int 字面量，经 gconv 可能为 float64） +func isNumericZero(v interface{}) bool { + switch vv := v.(type) { + case int: + return vv == 0 + case int8: + return vv == 0 + case int16: + return vv == 0 + case int32: + return vv == 0 + case int64: + return vv == 0 + case uint: + return vv == 0 + case uint8: + return vv == 0 + case uint16: + return vv == 0 + case uint32: + return vv == 0 + case uint64: + return vv == 0 + case float32: + return vv == 0 + case float64: + return vv == 0 + default: + return false + } +} + // resolveObject 解析对象字段，递归处理 attrs // // 特殊处理「参数定义」结构：当 attrs 含 default 字段时，说明该对象是一个 diff --git a/service/utils/template_validate.go b/service/utils/template_validate.go index 2709484..aa91dab 100644 --- a/service/utils/template_validate.go +++ b/service/utils/template_validate.go @@ -4,6 +4,7 @@ import ( "fmt" "model-gateway/model/dto" + "gitea.redpowerfuture.com/red-future/common/utils" "github.com/gogf/gf/v2/frame/g" "github.com/gogf/gf/v2/util/gconv" "github.com/gogf/gf/v2/util/gutil" @@ -40,8 +41,8 @@ func CheckBody(body map[string]interface{}, templateParams map[string]interface{ func checkParams(userParams map[string]interface{}, templateParams map[string]interface{}, strictUnknown, backfill bool) error { // 兼容扁平路径入参：还原为嵌套结构 orig := userParams - if IsFlatMap(userParams) { - nested, err := UnFlatBySjson(userParams) + if utils.IsFlatMap(userParams) { + nested, err := utils.UnFlatBySjson(userParams) if err != nil { return fmt.Errorf("无法解析用户参数: %w", err) } @@ -143,21 +144,29 @@ func validateNode(parent map[string]interface{}, key string, tmplNode interface{ case TypeObject: // 对象元素：以元素 attrs 为容器递归校验子字段（模板对象节点 {type:object,attrs:{...}} 的 // 子字段藏在 attrs 下；纯对象 map 直接以自身为容器） - if attrs, ok := protoTmpl.Attrs.(map[string]interface{}); ok { - for i, elem := range userArr { - elemMap, isMap := elem.(map[string]interface{}) - if !isMap { - continue - } - container, has := userObjectValue(elemMap, true) - if !has { - continue - } - for subKey, subTmpl := range attrs { - subPath := fmt.Sprintf("%s[%d].%s", path, i, subKey) - if err := validateNode(container, subKey, subTmpl, subPath, strictUnknown, backfill); err != nil { - return err - } + for i, elem := range userArr { + elemMap, isMap := elem.(map[string]interface{}) + if !isMap { + continue + } + container, has := userObjectValue(elemMap, true) + if !has { + continue + } + // 元素带 type 键（schema 包裹）：已由 validateEnumValues 按其自身 attrs 校验，跳过， + // 避免用枚举首原型（可能是必填字段模板）误报其他槽位元素缺失 + if _, isWrapped := elemMap["type"]; isWrapped { + continue + } + // 纯对象元素（解析后数组下标可能塌缩）：按 attrs 键集结构匹配槽位原型，避免恒用首原型误报必填 + subAttrs := matchArraySlotProto(&tmpl, container) + if subAttrs == nil { + continue + } + for subKey, subTmpl := range subAttrs { + subPath := fmt.Sprintf("%s[%d].%s", path, i, subKey) + if err := validateNode(container, subKey, subTmpl, subPath, strictUnknown, backfill); err != nil { + return err } } } @@ -369,6 +378,38 @@ func arrayElementPrototype(tmpl *dto.Template) map[string]interface{} { return nil } +// matchArraySlotProto 按元素 attrs 键集与各槽位原型 attrs 键集的重合度匹配最合适的槽位原型。 +// 解析后数组下标可能塌缩（resolveArray 丢弃空元素），不能按 index 对齐，故用结构匹配。 +// 键集完全无重合时返回 nil（跳过该校验，避免用错误原型误报必填）。 +func matchArraySlotProto(tmpl *dto.Template, container map[string]interface{}) map[string]interface{} { + var best map[string]interface{} + bestCount := -1 + for _, ev := range tmpl.EnumValues { + evMap, ok := ev.(map[string]interface{}) + if !ok { + continue + } + attrs, ok := evMap["attrs"].(map[string]interface{}) + if !ok { + continue + } + count := 0 + for k := range container { + if _, has := attrs[k]; has { + count++ + } + } + if count > bestCount { + bestCount = count + best = attrs + } + } + if bestCount <= 0 { + return nil + } + return best +} + // isValueEmptyByType 按 tmpl.Type 判断是否为"业务空值" func isValueEmptyByType(tmpl *dto.Template) bool { switch tmpl.Type {