商品规格参数
", + Introduction: "商品卖点
", + GoodsIllustration: "商品说明
", + IsHot: 2, + Title: "fu测试001", + GoodsNum: "futest001sku", + SpuCode: "futest001spu", + SpuName: "fu测试001", + Price: 100, + SalesPrice: 80, + Discount: 15, + TaxRate: 13, + FreightId: 3, + Remark: "备注说明", + SellByDate: 180, + ExternalPrice: 120, + GoodsBarCode: "futest001code2", + GoodsCode: "futest001code1", + SellByDateUnit: "天", + BrandId: 3, + ExternalUrl: "https://www.baidu.com", + } + + cfg := config.ToolConfig{ + BaseURL: "https://gateway.dev.cdlsxd.cn/goods-admin/api/v1/goods/add", + } + + client := New(cfg) + toolResp, err := client.Call(context.Background(), req) + + if err != nil { + t.Errorf("Call() error = %v", err) + return + } + + fmt.Printf("toolResp: %v\n", toolResp) +} diff --git a/internal/domain/tools/hyt/goods_add/types.go b/internal/domain/tools/hyt/goods_add/types.go new file mode 100644 index 0000000..d17b500 --- /dev/null +++ b/internal/domain/tools/hyt/goods_add/types.go @@ -0,0 +1,33 @@ +package goods_add + +type GoodsAddRequest struct { + Title string `json:"title"` // 商品标题 + GoodsCode string `json:"goods_code"` // 商品编码 + SpuName string `json:"spu_name"` // SPU 名称 + SpuCode string `json:"spu_code"` // SPU 编码 + GoodsNum string `json:"goods_num"` // 商品货号 + GoodsBarCode string `json:"goods_bar_code"` // 商品条形码 + Price float64 `json:"price"` // 市场价 + SalesPrice float64 `json:"sales_price"` // 建议销售价 + ExternalPrice float64 `json:"external_price"` // 电商销售价格 + Unit string `json:"unit"` // 价格单位 + Discount int `json:"discount"` // 折扣 + TaxRate int `json:"tax_rate"` // 税率 + FreightId int `json:"freight_id"` // 运费模板 ID + SellByDate int `json:"sell_by_date"` // 保质期 + SellByDateUnit string `json:"sell_by_date_unit"` // 保质期单位 + BrandId int `json:"brand_id"` // 品牌 ID + IsHot int `json:"is_hot"` // 是否热销主推 1.是 2.否(默认) + ExternalUrl string `json:"external_url"` // 外部平台链接 + Introduction string `json:"introduction"` // 商品卖点 + GoodsAttributes string `json:"goods_attributes"` // 商品规格参数 + GoodsIllustration string `json:"goods_illustration"` // 商品说明 + Remark string `json:"remark"` // 备注说明 + IsComposeGoods int `json:"is_compose_goods"` // 是否组合商品 1.是 2.否(默认) +} + +type GoodsAddResponse struct { + PreviewUrl string `json:"preview_url"` // 预览URL + SpuCode string `json:"spu_code"` // SPU编码 + Id int `json:"id"` // 商品ID +} diff --git a/internal/domain/tools/hyt/goods_brand_search/client.go b/internal/domain/tools/hyt/goods_brand_search/client.go new file mode 100644 index 0000000..e7b3d58 --- /dev/null +++ b/internal/domain/tools/hyt/goods_brand_search/client.go @@ -0,0 +1,67 @@ +package goods_brand_search + +import ( + "ai_scheduler/internal/config" + "ai_scheduler/internal/pkg/l_request" + "ai_scheduler/internal/pkg/util" + "context" + "encoding/json" + "fmt" +) + +type Client struct { + cfg config.ToolConfig +} + +func New(cfg config.ToolConfig) *Client { + return &Client{ + cfg: cfg, + } +} + +func (c *Client) Call(ctx context.Context, name string) (int, error) { + if name == "" { + return 0, nil + } + + reqBody := GoodsBrandSearchRequest{ + Page: 1, + Limit: 1, + Search: SearchCondition{ + Name: name, + }, + } + + apiReq, _ := util.StructToMap(reqBody) + + req := l_request.Request{ + Method: "Post", + Url: c.cfg.BaseURL, + Json: apiReq, + Headers: map[string]string{ + "User-Agent": "Apifox/1.0.0 (https://apifox.com)", + "Content-Type": "application/json", + }, + } + + res, err := req.Send() + if err != nil { + return 0, fmt.Errorf("请求失败，err: %v", err) + } + + var resData GoodsBrandSearchResponse + if err := json.Unmarshal([]byte(res.Text), &resData); err != nil { + return 0, fmt.Errorf("解析响应失败，err: %v", err) + } + + if resData.Code != 200 { + return 0, fmt.Errorf("业务错误，code: %d, msg: %s", resData.Code, resData.Msg) + } + + if len(resData.Data.List) == 0 { + return 0, fmt.Errorf("品牌不存在") + } + + // 返回第一个匹配的品牌ID + return resData.Data.List[0].ID, nil +} diff --git a/internal/domain/tools/hyt/goods_brand_search/client_test.go b/internal/domain/tools/hyt/goods_brand_search/client_test.go new file mode 100644 index 0000000..41009f1 --- /dev/null +++ b/internal/domain/tools/hyt/goods_brand_search/client_test.go @@ -0,0 +1,28 @@ +package goods_brand_search + +import ( + "ai_scheduler/internal/config" + "context" + "fmt" + "testing" +) + +// Test_Call +func Test_Call(t *testing.T) { + // 使用示例中的查询条件 + name := "vivo" + + cfg := config.ToolConfig{ + BaseURL: "https://gateway.dev.cdlsxd.cn/goods-admin/api/v1/goods/brand/list", + } + + client := New(cfg) + toolResp, err := client.Call(context.Background(), name) + + if err != nil { + t.Errorf("Call() error = %v", err) + return + } + + fmt.Printf("toolResp (BrandID): %v\n", toolResp) +} diff --git a/internal/domain/tools/hyt/goods_brand_search/types.go b/internal/domain/tools/hyt/goods_brand_search/types.go new file mode 100644 index 0000000..467a214 --- /dev/null +++ b/internal/domain/tools/hyt/goods_brand_search/types.go @@ -0,0 +1,25 @@ +package goods_brand_search + +type GoodsBrandSearchRequest struct { + Page int `json:"page"` + Limit int `json:"limit"` + Search SearchCondition `json:"search"` +} + +type SearchCondition struct { + Name string `json:"name"` +} + +type GoodsBrandSearchResponse struct { + Code int `json:"code"` + Msg string `json:"message"` + Data struct { + List []BrandInfo `json:"list"` + } `json:"data"` +} + +type BrandInfo struct { + ID int `json:"id"` + Name string `json:"name"` + Logo string `json:"logo"` +} diff --git a/internal/domain/tools/hyt/goods_category_add/client.go b/internal/domain/tools/hyt/goods_category_add/client.go new file mode 100644 index 0000000..8fa0e8b --- /dev/null +++ b/internal/domain/tools/hyt/goods_category_add/client.go @@ -0,0 +1,49 @@ +package goods_category_add + +import ( + "ai_scheduler/internal/config" + "ai_scheduler/internal/pkg/l_request" + "ai_scheduler/internal/pkg/util" + "context" + "encoding/json" + "fmt" +) + +type Client struct { + cfg config.ToolConfig +} + +func New(cfg config.ToolConfig) *Client { + return &Client{ + cfg: cfg, + } +} + +func (c *Client) Call(ctx context.Context, req *GoodsCategoryAddRequest) (bool, error) { + apiReq, _ := util.StructToMap(req) + + r := l_request.Request{ + Method: "Post", + Url: c.cfg.BaseURL, + Json: apiReq, + Headers: map[string]string{ + "Content-Type": "application/json", + }, + } + + res, err := r.Send() + if err != nil { + return false, fmt.Errorf("请求失败，err: %v", err) + } + + var resData GoodsCategoryAddResponse + if err := json.Unmarshal([]byte(res.Text), &resData); err != nil { + return false, fmt.Errorf("解析响应失败，err: %v", err) + } + + if resData.Code != 200 { + return false, fmt.Errorf("业务错误，code: %d, msg: %s", resData.Code, resData.Msg) + } + + return resData.Data.IsSuccess, nil +} diff --git a/internal/domain/tools/hyt/goods_category_add/client_test.go b/internal/domain/tools/hyt/goods_category_add/client_test.go new file mode 100644 index 0000000..fed3a94 --- /dev/null +++ b/internal/domain/tools/hyt/goods_category_add/client_test.go @@ -0,0 +1,31 @@ +package goods_category_add + +import ( + "ai_scheduler/internal/config" + "context" + "fmt" + "testing" +) + +// Test_Call +func Test_Call(t *testing.T) { + req := &GoodsCategoryAddRequest{ + GoodsId: 8496, + CategoryIds: []int{1667}, + IsCover: false, + } + + cfg := config.ToolConfig{ + BaseURL: "https://gateway.dev.cdlsxd.cn/goods-admin/api/v1/good/category/relation/add", + } + + client := New(cfg) + toolResp, err := client.Call(context.Background(), req) + + if err != nil { + t.Errorf("Call() error = %v", err) + return + } + + fmt.Printf("toolResp: %v\n", toolResp) +} diff --git a/internal/domain/tools/hyt/goods_category_add/types.go b/internal/domain/tools/hyt/goods_category_add/types.go new file mode 100644 index 0000000..b3ecf68 --- /dev/null +++ b/internal/domain/tools/hyt/goods_category_add/types.go @@ -0,0 +1,15 @@ +package goods_category_add + +type GoodsCategoryAddRequest struct { + GoodsId int `json:"goods_id"` + CategoryIds []int `json:"category_ids"` + IsCover bool `json:"is_cover"` +} + +type GoodsCategoryAddResponse struct { + Code int `json:"code"` + Msg string `json:"message"` + Data struct { + IsSuccess bool `json:"is_success"` // 是否成功 + } `json:"data"` +} diff --git a/internal/domain/tools/hyt/goods_category_search/client.go b/internal/domain/tools/hyt/goods_category_search/client.go new file mode 100644 index 0000000..3af5e14 --- /dev/null +++ b/internal/domain/tools/hyt/goods_category_search/client.go @@ -0,0 +1,67 @@ +package goods_category_search + +import ( + "ai_scheduler/internal/config" + "ai_scheduler/internal/pkg/l_request" + "ai_scheduler/internal/pkg/util" + "context" + "encoding/json" + "fmt" +) + +type Client struct { + cfg config.ToolConfig +} + +func New(cfg config.ToolConfig) *Client { + return &Client{ + cfg: cfg, + } +} + +func (c *Client) Call(ctx context.Context, name string) (int, error) { + if name == "" { + return 0, nil + } + + reqBody := GoodsCategorySearchRequest{ + Page: 1, + Limit: 1, + Search: SearchCondition{ + Name: name, + Level: 3, // 仅需三级分类 + }, + } + + apiReq, _ := util.StructToMap(reqBody) + + req := l_request.Request{ + Method: "Post", + Url: c.cfg.BaseURL, + Json: apiReq, + Headers: map[string]string{ + "Content-Type": "application/json", + }, + } + + res, err := req.Send() + if err != nil { + return 0, fmt.Errorf("请求失败，err: %v", err) + } + + var resData GoodsCategorySearchResponse + if err := json.Unmarshal([]byte(res.Text), &resData); err != nil { + return 0, fmt.Errorf("解析响应失败，err: %v", err) + } + + if resData.Code != 200 { + return 0, fmt.Errorf("业务错误，code: %d, msg: %s", resData.Code, resData.Msg) + } + + if len(resData.Data.List) == 0 { + return 0, fmt.Errorf("商品分类不存在") + } + + // 返回第一个匹配的分类ID + return resData.Data.List[0].ID, nil +} diff --git a/internal/domain/tools/hyt/goods_category_search/types.go b/internal/domain/tools/hyt/goods_category_search/types.go new file mode 100644 index 0000000..2b9fb0d --- /dev/null +++ b/internal/domain/tools/hyt/goods_category_search/types.go @@ -0,0 +1,25 @@ +package goods_category_search + +type GoodsCategorySearchRequest struct { + Page int `json:"page"` + Limit int `json:"limit"` + Search SearchCondition `json:"search"` +} + +type SearchCondition struct { + Name string `json:"full_name"` + Level int `json:"level"` +} + +type GoodsCategorySearchResponse struct { + Code int `json:"code"` + Msg string `json:"message"` + Data struct { + List []CategoryInfo `json:"list"` + } `json:"data"` +} + +type CategoryInfo struct { + ID int `json:"id"` + Name string `json:"name"` +} diff --git a/internal/domain/tools/hyt/goods_media_add/client.go b/internal/domain/tools/hyt/goods_media_add/client.go new file mode 100644 index 0000000..6632168 --- /dev/null +++ b/internal/domain/tools/hyt/goods_media_add/client.go @@ -0,0 +1,49 @@ +package goods_media_add + +import ( + "ai_scheduler/internal/config" + "ai_scheduler/internal/pkg/l_request" + "ai_scheduler/internal/pkg/util" + "context" + "encoding/json" + "fmt" +) + +type Client struct { + cfg config.ToolConfig +} + +func New(cfg config.ToolConfig) *Client { + return &Client{ + cfg: cfg, + } +} + +func (c *Client) Call(ctx context.Context, req *GoodsMediaAddRequest) (bool, error) { + apiReq, _ := util.StructToMap(req) + + r := l_request.Request{ + Method: "Post", + Url: c.cfg.BaseURL, + Json: apiReq, + Headers: map[string]string{ + "Content-Type": "application/json", + }, + } + + res, err := r.Send() + if err != nil { + return false, fmt.Errorf("请求失败，err: %v", err) + } + + var resData GoodsMediaAddResponse + if err := json.Unmarshal([]byte(res.Text), &resData); err != nil { + return false, fmt.Errorf("解析响应失败，err: %v", err) + } + + if resData.Code != 200 { + return false, fmt.Errorf("业务错误，code: %d, msg: %s", resData.Code, resData.Msg) + } + + return resData.Data.IsSuccess, nil +} diff --git a/internal/domain/tools/hyt/goods_media_add/client_test.go b/internal/domain/tools/hyt/goods_media_add/client_test.go new file mode 100644 index 0000000..f6f16ca --- /dev/null +++ b/internal/domain/tools/hyt/goods_media_add/client_test.go @@ -0,0 +1,37 @@ +package goods_media_add + +import ( + "ai_scheduler/internal/config" + "context" + "fmt" + "testing" +) + +// Test_Call +func Test_Call(t *testing.T) { + req := &GoodsMediaAddRequest{ + GoodsId: 8496, + Data: []MediaItem{ + { + Type: 1, + Url: "https://lsxd-hz-store.oss-cn-hangzhou.aliyuncs.com/physicalGoodsSystems/images/goodsimages/goods/22f03d91-3cb7-45b4-ab92-07aad78a1633-screenshot_2025-12-17_17-46-00.png", + Sort: 1, + }, + }, + IsCover: true, + } + + cfg := config.ToolConfig{ + BaseURL: "https://gateway.dev.cdlsxd.cn/goods-admin/api/v1/media/add/batch", + } + + client := New(cfg) + toolResp, err := client.Call(context.Background(), req) + + if err != nil { + t.Errorf("Call() error = %v", err) + return + } + + fmt.Printf("toolResp: %v\n", toolResp) +} diff --git a/internal/domain/tools/hyt/goods_media_add/types.go b/internal/domain/tools/hyt/goods_media_add/types.go new file mode 100644 index 0000000..bde4826 --- /dev/null +++ b/internal/domain/tools/hyt/goods_media_add/types.go @@ -0,0 +1,21 @@ +package goods_media_add + +type GoodsMediaAddRequest struct { + GoodsId int `json:"goods_id"` + Data []MediaItem `json:"data"` + IsCover bool `json:"is_cover"` +} + +type MediaItem struct { + Type int `json:"type"` + Url string `json:"url"` + Sort int `json:"sort"` +} + +type GoodsMediaAddResponse struct { + Code int `json:"code"` + Msg string `json:"message"` + Data struct { + IsSuccess bool `json:"is_success"` + } `json:"data"` +} diff --git a/internal/domain/tools/hyt/product_upload/client.go b/internal/domain/tools/hyt/product_upload/client.go new file mode 100644 index 0000000..096965c --- /dev/null +++ b/internal/domain/tools/hyt/product_upload/client.go @@ -0,0 +1,68 @@ +package product_upload + +import ( + "ai_scheduler/internal/config" + "ai_scheduler/internal/pkg/l_request" + "ai_scheduler/internal/pkg/util" + "context" + "encoding/json" + "errors" + "fmt" +) + +type Client struct { + cfg config.ToolConfig +} + +func New(cfg config.ToolConfig) *Client { + return &Client{ + cfg: cfg, + } +} + +func (c *Client) Call(ctx context.Context, toolReq *ProductUploadRequest) (toolResp *ProductUploadResponse, err error) { + // 商品有且只能有一个 + if len(toolReq.GoodsList) != 1 { + return nil, errors.New("商品只能有一个") + } + + apiReq, _ := util.StructToMap(toolReq) + + req := l_request.Request{ + Method: "Post", + Url: c.cfg.BaseURL, + Json: apiReq, + } + res, err := req.Send() + + if err != nil { + return nil, fmt.Errorf("请求失败，err: %v", err) + } + + type resType struct { + Code int `json:"code"` + Msg string `json:"msg"` + Data struct { + Ids []int `json:"ids"` // 商品 IDs + } `json:"data"` + } + var resMap resType + err = json.Unmarshal([]byte(res.Text), &resMap) + if err != nil { + return nil, fmt.Errorf("解析响应失败，err: %v", err) + } + if resMap.Code != 200 { + return nil, fmt.Errorf("业务错误，code: %d, msg: %s", resMap.Code, resMap.Msg) + } + if len(resMap.Data.Ids) == 0 { + return nil, fmt.Errorf("ids为空") + } + + toolResp = &ProductUploadResponse{ + PreviewUrl: c.cfg.AddURL, + SpuNum: toolReq.GoodsList[0].GoodsInfo.SpuNum, + Id: resMap.Data.Ids[0], + } + + return toolResp, nil +} diff --git a/internal/domain/tools/hyt/product_upload/types.go b/internal/domain/tools/hyt/product_upload/types.go new file mode 100644 index 0000000..947dbe5 --- /dev/null +++ b/internal/domain/tools/hyt/product_upload/types.go @@ -0,0 +1,54 @@ +package product_upload + +type ProductUploadRequest struct { + SupplierId int `json:"supplier_id"` // 供应商ID + WarehouseId int `json:"warehouse_id"` // 仓库ID + IsDefaultWarehouse int `json:"is_default_warehouse"` // 是否默认仓库 + Sort int `json:"sort"` // 排序 + Profit float64 `json:"profit"` // 利润 + TaxRate int `json:"tax_rate"` // 税率 + GoodsList []Goods `json:"goods_list"` // 商品列表 +} + +type Goods struct { + GoodsInfo GoodsInfo `json:"goods_info"` + GoodsMediaList []GoodsMedia `json:"goods_media_list"` +} + +type GoodsInfo struct { + Title string `json:"title"` // 商品名称 + Brand string `json:"brand"` // 品牌 + Category string `json:"category"` // 分类 + Discount int `json:"discount"` // 折扣 + GoodsAttributes string `json:"goods_attributes"` // 商品属性 + GoodsBarCode string `json:"goods_bar_code"` // 商品条码 + GoodsNum string `json:"goods_num"` // 商品编号 + Introduction string `json:"introduction"` // 商品介绍 + SpuName string `json:"spu_name"` // SPU名称 + SpuNum string `json:"spu_num"` // SPU编号 + Stock int `json:"stock"` // 库存 + TaxRate int `json:"tax_rate"` // 税率 + Unit string `json:"unit"` // 单位 + Weight string `json:"weight"` // 重量 + Price float64 `json:"price"` // 市场价 + SalesPrice float64 `json:"sales_price"` // 建议销售价格 + GoodsIllustration string `json:"goods_illustration"` // 商品插图 - 暂不提供 + Id int `json:"id"` // 商品ID - 无需 + CostPrice float64 `json:"cost_price"` // 成本价格 - 无需 + IsBind int `json:"is_bind"` // 是否绑定 - 默认0 + IsComposeGoods int32 `json:"is_compose_goods"` // 是否组合商品 - 默认2 + IsHot int `json:"is_hot"` // 是否热门商品 - 默认2 +} + +type GoodsMedia struct { + Remark string `json:"remark"` // 备注 + Sort int `json:"sort"` // 排序 + Type int `json:"type"` // 类型 + Url string `json:"url"` // URL +} + +type ProductUploadResponse struct { + PreviewUrl string `json:"preview_url"` // 预览URL + SpuNum string `json:"spu_code"` // SPU编码 + Id int `json:"id"` // 商品ID +} diff --git a/internal/domain/tools/hyt/supplier_search/client.go b/internal/domain/tools/hyt/supplier_search/client.go new file mode 100644 index 0000000..1f47ee8 --- /dev/null +++ b/internal/domain/tools/hyt/supplier_search/client.go @@ -0,0 +1,67 @@ +package supplier_search + +import ( + "ai_scheduler/internal/config" + "ai_scheduler/internal/pkg/l_request" + "context" + "encoding/json" + "fmt" +) + +type Client struct { + cfg config.ToolConfig +} + +func New(cfg config.ToolConfig) *Client { + return &Client{ + cfg: cfg, + } +} + +func (c *Client) Call(ctx context.Context, name string) (int, error) { + if name == "" { + // 如果没有供应商名，返回0，不报错，由上层业务决定是否允许 + return 0, nil + } + + reqBody := SearchRequest{ + Page: 1, + Limit: 1, + Search: SearchCondition{ + Name: name, + }, + } + + apiReq := make(map[string]interface{}) + bytes, _ := json.Marshal(reqBody) + _ = json.Unmarshal(bytes, &apiReq) + + req := l_request.Request{ + Method: "Post", + Url: c.cfg.BaseURL, + Json: apiReq, + Headers: map[string]string{ + "Content-Type": "application/json", + }, + } + + res, err := req.Send() + if err != nil { + return 0, fmt.Errorf("请求失败，err: %v", err) + } + + var resData SearchResponse + if err := json.Unmarshal([]byte(res.Text), &resData); err != nil { + return 0, fmt.Errorf("解析响应失败，err: %v", err) + } + + if resData.Code != 200 { + return 0, fmt.Errorf("业务错误，code: %d, msg: %s", resData.Code, resData.Msg) + } + + if len(resData.Data.List) == 0 { + return 0, fmt.Errorf("供应商不存在") + } + + return resData.Data.List[0].ID, nil +} diff --git a/internal/domain/tools/hyt/supplier_search/types.go b/internal/domain/tools/hyt/supplier_search/types.go new file mode 100644 index 0000000..46a452c --- /dev/null +++ b/internal/domain/tools/hyt/supplier_search/types.go @@ -0,0 +1,24 @@ +package supplier_search + +type SearchRequest struct { + Page int `json:"page"` + Limit int `json:"limit"` + Search SearchCondition `json:"search"` +} + +type SearchCondition struct { + Name string `json:"name"` +} + +type SearchResponse struct { + Code int `json:"code"` + Msg string `json:"msg"` + Data struct { + List []SupplierInfo `json:"list"` + } `json:"data"` +} + +type SupplierInfo struct { + ID int `json:"id"` + Name string `json:"name"` +} diff --git a/internal/domain/tools/hyt/warehouse_search/client.go b/internal/domain/tools/hyt/warehouse_search/client.go new file mode 100644 index 0000000..cf420b2 --- /dev/null +++ b/internal/domain/tools/hyt/warehouse_search/client.go @@ -0,0 +1,62 @@ +package warehouse_search + +import ( + "ai_scheduler/internal/config" + "ai_scheduler/internal/pkg/l_request" + "context" + "encoding/json" + "fmt" +) + +type Client struct { + cfg config.ToolConfig +} + +func New(cfg config.ToolConfig) *Client { + return &Client{ + cfg: cfg, + } +} + +func (c *Client) Call(ctx context.Context, name string) (int, error) { + if name == "" { + // 如果没有仓库名，返回0，不报错，由上层业务决定是否允许 + return 0, nil + } + + // GET 请求参数 + params := map[string]string{ + "name": name, + "page": "1", + "limit": "1", + } + + req := l_request.Request{ + Method: "Get", + Url: c.cfg.BaseURL, + Params: params, + Headers: map[string]string{ + "Content-Type": "application/json", + }, + } + + res, err := req.Send() + if err != nil { + return 0, fmt.Errorf("请求失败，err: %v", err) + } + + var resData SearchResponse + if err := json.Unmarshal([]byte(res.Text), &resData); err != nil { + return 0, fmt.Errorf("解析响应失败，err: %v", err) + } + + if resData.Code != 200 { + return 0, fmt.Errorf("业务错误，code: %d, msg: %s", resData.Code, resData.Msg) + } + + if len(resData.Data.List) == 0 { + return 0, fmt.Errorf("仓库不存在: %s", name) + } + + return resData.Data.List[0].ID, nil +} diff --git a/internal/domain/tools/hyt/warehouse_search/types.go b/internal/domain/tools/hyt/warehouse_search/types.go new file mode 100644 index 0000000..a5ae237 --- /dev/null +++ b/internal/domain/tools/hyt/warehouse_search/types.go @@ -0,0 +1,14 @@ +package warehouse_search + +type SearchResponse struct { + Code int `json:"code"` + Msg string `json:"msg"` + Data struct { + List []WarehouseInfo `json:"list"` + } `json:"data"` +} + +type WarehouseInfo struct { + ID int `json:"id"` + Name string `json:"name"` +} diff --git a/internal/domain/tools/registry.go b/internal/domain/tools/registry.go index dc6a67e..ad9439d 100644 --- a/internal/domain/tools/registry.go +++ b/internal/domain/tools/registry.go @@ -1,16 +1,44 @@ package tools -type Tool interface{ - Name() string +import ( + "ai_scheduler/internal/config" + "ai_scheduler/internal/domain/tools/hyt/goods_add" + "ai_scheduler/internal/domain/tools/hyt/goods_brand_search" + "ai_scheduler/internal/domain/tools/hyt/goods_category_add" + "ai_scheduler/internal/domain/tools/hyt/goods_category_search" + "ai_scheduler/internal/domain/tools/hyt/goods_media_add" + "ai_scheduler/internal/domain/tools/hyt/product_upload" + "ai_scheduler/internal/domain/tools/hyt/supplier_search" + "ai_scheduler/internal/domain/tools/hyt/warehouse_search" +) + +type Manager struct { + Hyt *HytTools + // Zltx *ZltxTools } -var registry = map[string]Tool{} - -func Register(t Tool){ - registry[t.Name()] = t +type HytTools struct { + ProductUpload *product_upload.Client + SupplierSearch *supplier_search.Client + WarehouseSearch *warehouse_search.Client + GoodsAdd *goods_add.Client + GoodsMediaAdd *goods_media_add.Client + GoodsCategoryAdd *goods_category_add.Client + GoodsCategorySearch *goods_category_search.Client + GoodsBrandSearch *goods_brand_search.Client } -func Get(name string) Tool{ - return registry[name] +func NewManager(cfg *config.Config) *Manager { + return &Manager{ + Hyt: &HytTools{ + ProductUpload: product_upload.New(cfg.EinoTools.HytProductUpload), + SupplierSearch: supplier_search.New(cfg.EinoTools.HytSupplierSearch), + WarehouseSearch: warehouse_search.New(cfg.EinoTools.HytWarehouseSearch), + GoodsAdd: goods_add.New(cfg.EinoTools.HytGoodsAdd), + GoodsMediaAdd: goods_media_add.New(cfg.EinoTools.HytGoodsMediaAdd), + GoodsCategoryAdd: goods_category_add.New(cfg.EinoTools.HytGoodsCategoryAdd), + GoodsCategorySearch: goods_category_search.New(cfg.EinoTools.HytGoodsCategorySearch), + GoodsBrandSearch: goods_brand_search.New(cfg.EinoTools.HytGoodsBrandSearch), + }, + } } - diff --git a/internal/domain/workflow/hyt/goods_add.go b/internal/domain/workflow/hyt/goods_add.go new file mode 100644 index 0000000..91db063 --- /dev/null +++ b/internal/domain/workflow/hyt/goods_add.go @@ -0,0 +1,373 @@ +package hyt + +import ( + "ai_scheduler/internal/config" + errorcode "ai_scheduler/internal/data/error" + toolManager "ai_scheduler/internal/domain/tools" + "ai_scheduler/internal/domain/tools/hyt/goods_add" + "ai_scheduler/internal/domain/tools/hyt/goods_category_add" + "ai_scheduler/internal/domain/tools/hyt/goods_media_add" + "ai_scheduler/internal/domain/workflow/runtime" + "ai_scheduler/internal/entitys" + "context" + "encoding/json" + "errors" + "fmt" + "log" + "strconv" + "strings" + "sync" + + "github.com/cloudwego/eino/compose" + "golang.org/x/sync/errgroup" +) + +const WorkflowIDGoodsAdd = "hyt.goodsAdd" + +func init() { + runtime.Register(WorkflowIDGoodsAdd, func(d *runtime.Deps) (runtime.Workflow, error) { + return &goodsAdd{cfg: d.Conf, toolManager: d.ToolManager}, nil + }) +} + +type goodsAdd struct { + cfg *config.Config + toolManager *toolManager.Manager + data *GoodsAddWorkflowInput +} + +type GoodsAddWorkflowInput struct { + Text string `mapstructure:"text"` +} + +func (o *goodsAdd) ID() string { return WorkflowIDGoodsAdd } + +func (o *goodsAdd) Invoke(ctx context.Context, rec *entitys.Recognize) (map[string]any, error) { + // 构建工作流 + runnable, err := o.buildWorkflow(ctx) + if err != nil { + return nil, err + } + + o.data = &GoodsAddWorkflowInput{ + Text: rec.UserContent.Text, + } + // 工作流过程调用 + output, err := runnable.Invoke(ctx, o.data) + if err != nil { + fmt.Println("Invoke err:", err) + errStr := err.Error() + if u := errors.Unwrap(err); u != nil { + errStr = u.Error() + } + return nil, errorcode.WorkflowErr(errStr) + } + + return output, nil +} + +// ProductIngestData 对应 HYTGoodsAddPropertyTemplateZH 的结构 +type GoodsAddProductIngestData struct { + Title string `json:"商品标题"` + GoodsCode string `json:"商品编码"` + SpuName string `json:"SPU名称"` + SpuCode string `json:"SPU编码"` + GoodsNum string `json:"商品货号"` + GoodsBarCode string `json:"商品条形码"` + Price string `json:"市场价"` + SalesPrice string `json:"建议销售价"` + ExternalPrice string `json:"电商销售价格"` + Unit string `json:"单位"` + Discount string `json:"折扣"` + TaxRate string `json:"税率"` + FreightTemplate string `json:"运费模版"` + SellByDate string `json:"保质期"` + SellByDateUnit string `json:"保质期单位"` + Brand string `json:"品牌"` + IsHot string `json:"是否热销主推"` + ExternalUrl string `json:"外部平台链接"` + Introduction string `json:"商品卖点"` + GoodsAttributes string `json:"商品规格参数"` + GoodsIllustration string `json:"商品说明"` + Remark string `json:"备注"` + CategoryName string `json:"分类名称"` + Images []string `json:"电脑端主图"` +} + +// GoodsAddContext Graph 执行上下文状态 +type GoodsAddContext struct { + mu *sync.Mutex + InputText string + IngestData *GoodsAddProductIngestData + + // 核心请求体 + AddGoodsReq *goods_add.GoodsAddRequest + + // 中间态数据 + BrandId int + CategoryId int + BrandName string + CategoryName string + + // 运行结果 + GoodsAddResp *goods_add.GoodsAddResponse + GoodsCategoryAddResp bool + GoodsMediaAddResp bool +} + +// buildWorkflow 构建基于 Graph 的并行工作流 +func (o *goodsAdd) buildWorkflow(ctx context.Context) (compose.Runnable[*GoodsAddWorkflowInput, map[string]any], error) { + g := compose.NewGraph[*GoodsAddWorkflowInput, map[string]any]() + + // 1. DataMapping 节点: 解析 JSON -> 填充基础 Request + g.AddLambdaNode("data_mapping", compose.InvokableLambda(func(ctx context.Context, in *GoodsAddWorkflowInput) (*GoodsAddContext, error) { + state := &GoodsAddContext{ + mu: &sync.Mutex{}, // 初始化锁 + InputText: in.Text, + AddGoodsReq: &goods_add.GoodsAddRequest{}, + } + + // 解析用户输入的中文 JSON + var ingestData GoodsAddProductIngestData + if err := json.Unmarshal([]byte(in.Text), &ingestData); err != nil { + return nil, fmt.Errorf("解析商品数据失败") + } + + // 必填校验 + if ingestData.Title == "" { + return nil, errors.New("商品标题不能为空") + } + if ingestData.GoodsCode == "" { + return nil, errors.New("商品编码不能为空") + } + if ingestData.SpuName == "" { + return nil, errors.New("SPU名称不能为空") + } + if ingestData.SpuCode == "" { + return nil, errors.New("SPU编码不能为空") + } + if ingestData.Price == "" { + return nil, errors.New("市场价不能为空") + } + if ingestData.SalesPrice == "" { + return nil, errors.New("建议销售价不能为空") + } + if ingestData.Unit == "" { + return nil, errors.New("价格单位不能为空") + } + if ingestData.Discount == "" { + return nil, errors.New("折扣不能为空") + } + if ingestData.TaxRate == "" { + return nil, errors.New("税率不能为空") + } + + state.IngestData = &ingestData + state.BrandName = ingestData.Brand + state.CategoryName = ingestData.CategoryName + + // 映射字段到 AddGoodsReq + state.AddGoodsReq.Title = ingestData.Title + state.AddGoodsReq.GoodsCode = ingestData.GoodsCode + state.AddGoodsReq.SpuName = ingestData.SpuName + state.AddGoodsReq.SpuCode = ingestData.SpuCode + state.AddGoodsReq.GoodsNum = ingestData.GoodsNum + state.AddGoodsReq.GoodsBarCode = ingestData.GoodsBarCode + + // 价格处理 + if val, err := strconv.ParseFloat(strings.TrimSuffix(ingestData.Price, "元"), 64); err == nil { + state.AddGoodsReq.Price = val + } + if val, err := strconv.ParseFloat(strings.TrimSuffix(ingestData.SalesPrice, "元"), 64); err == nil { + state.AddGoodsReq.SalesPrice = val + } + if val, err := strconv.ParseFloat(strings.TrimSuffix(ingestData.ExternalPrice, "元"), 64); err == nil { + state.AddGoodsReq.ExternalPrice = val + } + + state.AddGoodsReq.Unit = ingestData.Unit + + // 折扣处理 "80%" -> 80 + discountStr := strings.TrimSuffix(ingestData.Discount, "%") + if val, err := strconv.Atoi(discountStr); err == nil { + state.AddGoodsReq.Discount = val + } + // 税率处理 "13%" -> 13 + taxStr := strings.TrimSuffix(strings.TrimSuffix(ingestData.TaxRate, "%"), " ") + if val, err := strconv.Atoi(taxStr); err == nil { + state.AddGoodsReq.TaxRate = val + } + + // 运费模板先不给 state.AddGoodsReq.FreightId = 3 + + // 保质期处理 "180天" -> 180 + sellByDateStr := strings.TrimSuffix(ingestData.SellByDate, "天") + if val, err := strconv.Atoi(sellByDateStr); err == nil { + state.AddGoodsReq.SellByDate = val + } + state.AddGoodsReq.SellByDateUnit = ingestData.SellByDateUnit + + // state.AddGoodsReq.BrandId 品牌ID后续赋值 + + state.AddGoodsReq.IsHot = 2 + if ingestData.IsHot == "是" { + state.AddGoodsReq.IsHot = 1 + } + + state.AddGoodsReq.ExternalUrl = ingestData.ExternalUrl + state.AddGoodsReq.Introduction = ingestData.Introduction + state.AddGoodsReq.GoodsAttributes = ingestData.GoodsAttributes + state.AddGoodsReq.GoodsIllustration = ingestData.GoodsIllustration + state.AddGoodsReq.Remark = ingestData.Remark + state.AddGoodsReq.IsComposeGoods = 2 // 非组合商品 + + return state, nil + })) + + // 2. 预处理节点: 并行获取 品牌ID 和 分类ID + g.AddLambdaNode("prepare_info", compose.InvokableLambda(func(ctx context.Context, state *GoodsAddContext) (*GoodsAddContext, error) { + eg, ctx := errgroup.WithContext(ctx) + + // 任务1: 获取品牌ID + eg.Go(func() error { + if state.BrandName == "" { + return nil + } + brandId, err := o.toolManager.Hyt.GoodsBrandSearch.Call(ctx, state.BrandName) + if err != nil { + log.Printf("warning: 品牌ID获取失败，%s: %v\n", state.BrandName, err) + return nil + } + state.mu.Lock() + state.BrandId = brandId + state.AddGoodsReq.BrandId = brandId + state.mu.Unlock() + return nil + }) + + // 任务2: 获取分类ID + eg.Go(func() error { + if state.CategoryName == "" { + return nil + } + categoryId, err := o.toolManager.Hyt.GoodsCategorySearch.Call(ctx, state.CategoryName) + if err != nil { + log.Printf("warning: 分类ID获取失败，%s: %v\n", state.CategoryName, err) + return nil + } + state.mu.Lock() + state.CategoryId = categoryId + state.mu.Unlock() + return nil + }) + + // 等待所有任务完成 + _ = eg.Wait() + + return state, nil + })) + + // 3. 新增商品 节点 (依赖 prepare_info) + g.AddLambdaNode("goods_add", compose.InvokableLambda(func(ctx context.Context, state *GoodsAddContext) (*GoodsAddContext, error) { + // 调用 goods_add 工具 + respData, err := o.toolManager.Hyt.GoodsAdd.Call(ctx, state.AddGoodsReq) + if err != nil || respData == nil { + log.Printf("warning: 新增商品失败: %v", err) + return nil, fmt.Errorf("新增商品失败: %s", err.Error()) + } + + state.GoodsAddResp = respData + + return state, nil + })) + + // 4. 后置处理节点: 并行执行 关联分类 和 添加图片 + g.AddLambdaNode("post_process", compose.InvokableLambda(func(ctx context.Context, state *GoodsAddContext) (*GoodsAddContext, error) { + if state.GoodsAddResp.Id == 0 { + return nil, errors.New("商品不存在") + } + + eg, ctx := errgroup.WithContext(ctx) + + // 任务1: 关联分类 + eg.Go(func() error { + if state.CategoryId == 0 { + return nil + } + req := &goods_category_add.GoodsCategoryAddRequest{ + GoodsId: state.GoodsAddResp.Id, + CategoryIds: []int{state.CategoryId}, + IsCover: false, + } + isSuccess, err := o.toolManager.Hyt.GoodsCategoryAdd.Call(ctx, req) + if err != nil { + log.Printf("warning: 关联分类失败: %v", err) + return nil + } + + state.mu.Lock() + state.GoodsCategoryAddResp = isSuccess + state.mu.Unlock() + + return nil + }) + + // 任务2: 添加图片 + eg.Go(func() error { + if len(state.IngestData.Images) == 0 { + return nil + } + req := &goods_media_add.GoodsMediaAddRequest{ + GoodsId: state.GoodsAddResp.Id, + IsCover: true, + Data: make([]goods_media_add.MediaItem, 0), + } + for i, url := range state.IngestData.Images { + req.Data = append(req.Data, goods_media_add.MediaItem{ + Type: 1, // 图片 + Url: url, + Sort: i, + }) + } + isSuccess, err := o.toolManager.Hyt.GoodsMediaAdd.Call(ctx, req) + if err != nil { + log.Printf("warning: 添加图片失败: %v", err) + return nil + } + + state.mu.Lock() + state.GoodsMediaAddResp = isSuccess + state.mu.Unlock() + + return nil + }) + + // 等待所有任务完成 + _ = eg.Wait() + + return state, nil + })) + + // 5. 结果格式化节点 + g.AddLambdaNode("format_output", compose.InvokableLambda(func(ctx context.Context, state *GoodsAddContext) (map[string]any, error) { + if state.GoodsAddResp == nil { + return nil, fmt.Errorf("goods add response is nil") + } + + return map[string]any{ + "预览URL(货易通商品列表)": state.GoodsAddResp.PreviewUrl, + "SPU编码": state.GoodsAddResp.SpuCode, + "商品ID": state.GoodsAddResp.Id, + }, nil + })) + + // 构建边 (线性拓扑) + g.AddEdge(compose.START, "data_mapping") + g.AddEdge("data_mapping", "prepare_info") + g.AddEdge("prepare_info", "goods_add") + g.AddEdge("goods_add", "post_process") + g.AddEdge("post_process", "format_output") + g.AddEdge("format_output", compose.END) + + return g.Compile(ctx) +} diff --git a/internal/domain/workflow/hyt/product_upload.go b/internal/domain/workflow/hyt/product_upload.go new file mode 100644 index 0000000..35114ed --- /dev/null +++ b/internal/domain/workflow/hyt/product_upload.go @@ -0,0 +1,303 @@ +package hyt + +import ( + "ai_scheduler/internal/config" + errorcode "ai_scheduler/internal/data/error" + toolManager "ai_scheduler/internal/domain/tools" + toolPu "ai_scheduler/internal/domain/tools/hyt/product_upload" + "ai_scheduler/internal/domain/workflow/runtime" + "ai_scheduler/internal/entitys" + "context" + "encoding/json" + "errors" + "fmt" + "log" + "strconv" + "strings" + "sync" + + "github.com/cloudwego/eino/compose" +) + +const WorkflowIDProductUpload = "hyt.productUpload" + +func init() { + runtime.Register(WorkflowIDProductUpload, func(d *runtime.Deps) (runtime.Workflow, error) { + return &productUpload{cfg: d.Conf, toolManager: d.ToolManager}, nil + }) +} + +type productUpload struct { + cfg *config.Config + toolManager *toolManager.Manager + data *ProductUploadWorkflowInput +} + +type ProductUploadWorkflowInput struct { + Text string `mapstructure:"text"` +} + +func (o *productUpload) ID() string { return WorkflowIDProductUpload } + +func (o *productUpload) Invoke(ctx context.Context, rec *entitys.Recognize) (map[string]any, error) { + // 构建工作流 + runnable, err := o.buildWorkflow(ctx) + if err != nil { + return nil, err + } + + o.data = &ProductUploadWorkflowInput{ + Text: rec.UserContent.Text, + } + // 工作流过程调用 + output, err := runnable.Invoke(ctx, o.data) + if err != nil { + errStr := err.Error() + if u := errors.Unwrap(err); u != nil { + errStr = u.Error() + } + return nil, errorcode.WorkflowErr(errStr) + } + + fmt.Printf("workflow output: %v\n", output) + + return output, nil +} + +// ProductIngestData 对应 HYTSupplierProductPropertyTemplateZH 的结构 +type SupplierProductIngestData struct { + BarCode string `json:"条码"` + CategoryName string `json:"分类名称"` + GoodsName string `json:"货品名称"` + GoodsNum string `json:"货品编号"` + GoodsArticleNum string `json:"商品货号"` + Brand string `json:"品牌"` + Unit string `json:"单位"` + Specs string `json:"规格参数"` + Description string `json:"货品说明"` + ShelfLife string `json:"保质期"` + ShelfLifeUnit string `json:"保质期单位"` + Link string `json:"链接"` + Images []string `json:"货品图片"` + EPrice string `json:"电商销售价格"` + SalesPrice string `json:"销售价"` + SupplierPrice string `json:"供应商报价"` + TaxRate string `json:"税率"` + SupplierName string `json:"默认供应商"` + WarehouseName string `json:"默认存放仓库"` + Remark string `json:"备注"` + Length string `json:"长"` + Width string `json:"宽"` + Height string `json:"高"` + Weight string `json:"重量"` + SpuName string `json:"SPU名称"` + SpuCode string `json:"SPU编码"` + Profit string `json:"利润"` +} + +// ProductUploadContext Graph 执行上下文状态 +type ProductUploadContext struct { + mu *sync.Mutex + InputText string + IngestData *SupplierProductIngestData + UploadReq *toolPu.ProductUploadRequest + SupplierName string + WarehouseName string + UploadResp *toolPu.ProductUploadResponse +} + +// buildWorkflow 构建基于 Graph 的并行工作流 +func (o *productUpload) buildWorkflow(ctx context.Context) (compose.Runnable[*ProductUploadWorkflowInput, map[string]any], error) { + g := compose.NewGraph[*ProductUploadWorkflowInput, map[string]any]() + + // 1. DataMapping 节点: 解析 JSON -> 填充基础 Request + g.AddLambdaNode("data_mapping", compose.InvokableLambda(func(ctx context.Context, in *ProductUploadWorkflowInput) (*ProductUploadContext, error) { + state := &ProductUploadContext{ + mu: &sync.Mutex{}, // 初始化锁 + InputText: in.Text, + UploadReq: &toolPu.ProductUploadRequest{ + GoodsList: make([]toolPu.Goods, 1), // 初始化一个商品 + }, + } + + // 解析用户输入的中文 JSON + var ingestData SupplierProductIngestData + if err := json.Unmarshal([]byte(in.Text), &ingestData); err != nil { + return nil, fmt.Errorf("解析商品数据失败: %w", err) + } + + // 必填校验 + if ingestData.SupplierName == "" { + return nil, errors.New("供应商名称不能为空") + } + if ingestData.WarehouseName == "" { + return nil, errors.New("仓库名称不能为空") + } + if ingestData.Profit == "" { + return nil, errors.New("利润不能为空") + } + if ingestData.TaxRate == "" { + return nil, errors.New("税率不能为空") + } + if ingestData.SupplierPrice == "" { + return nil, errors.New("供应商报价不能为空") + } + + state.IngestData = &ingestData + state.SupplierName = ingestData.SupplierName + state.WarehouseName = ingestData.WarehouseName + + // 映射字段到 UploadReq + goodsInfo := &state.UploadReq.GoodsList[0].GoodsInfo + goodsInfo.Title = ingestData.GoodsName + goodsInfo.Brand = ingestData.Brand + goodsInfo.Category = ingestData.CategoryName + goodsInfo.GoodsBarCode = ingestData.BarCode + goodsInfo.GoodsNum = ingestData.GoodsNum + if goodsInfo.GoodsNum == "" { + goodsInfo.GoodsNum = ingestData.GoodsArticleNum + } + goodsInfo.Unit = ingestData.Unit + goodsInfo.GoodsAttributes = ingestData.Specs + goodsInfo.Introduction = ingestData.Description + goodsInfo.SpuName = ingestData.SpuName + goodsInfo.SpuNum = ingestData.SpuCode + goodsInfo.Weight = ingestData.Weight + + // 数值处理 + if val, err := strconv.ParseFloat(strings.TrimSuffix(ingestData.SalesPrice, "元"), 64); err == nil { + goodsInfo.SalesPrice = val + } + if val, err := strconv.ParseFloat(strings.TrimSuffix(ingestData.EPrice, "元"), 64); err == nil { + goodsInfo.Price = val // 假设电商价为市场价 + } + // 价格兼容 + if goodsInfo.CostPrice == 0 { + goodsInfo.CostPrice = goodsInfo.Price + } + // 税率处理 "13%" -> 13 + taxStr := strings.TrimSuffix(strings.TrimSuffix(ingestData.TaxRate, "%"), " ") + if val, err := strconv.Atoi(taxStr); err == nil { + goodsInfo.TaxRate = val + state.UploadReq.TaxRate = val + } + // 利润处理 + if val, err := strconv.ParseFloat(strings.TrimSuffix(ingestData.Profit, "元"), 64); err == nil { + state.UploadReq.Profit = val + } + + // 图片处理 + for i, imgUrl := range ingestData.Images { + state.UploadReq.GoodsList[0].GoodsMediaList = append(state.UploadReq.GoodsList[0].GoodsMediaList, toolPu.GoodsMedia{ + Url: imgUrl, + Type: 1, // 图片 + Sort: i, + }) + } + + // 默认值字段 + goodsInfo.IsComposeGoods = 2 + goodsInfo.IsBind = 0 + goodsInfo.IsHot = 2 + state.UploadReq.IsDefaultWarehouse = 1 + state.UploadReq.Sort = 1 + + return state, nil + })) + + // 2. 获取供应商ID 节点 + g.AddLambdaNode("get_supplier_id", compose.InvokableLambda(func(ctx context.Context, state *ProductUploadContext) (*ProductUploadContext, error) { + if state.SupplierName == "" { + return state, errors.New("供应商名称不能为空") + } + + supplierId, err := o.toolManager.Hyt.SupplierSearch.Call(ctx, state.SupplierName) + if err != nil { + // 记录日志，但不阻断流程，可能允许 ID 为 0 + log.Printf("warning: 供应商ID获取失败，%s: %v\n", state.SupplierName, err) + } else { + state.mu.Lock() + defer state.mu.Unlock() + state.UploadReq.SupplierId = supplierId + } + + return state, nil + })) + + // 3. 获取仓库ID 节点 + g.AddLambdaNode("get_warehouse_id", compose.InvokableLambda(func(ctx context.Context, state *ProductUploadContext) (*ProductUploadContext, error) { + if state.WarehouseName == "" { + return state, errors.New("仓库名称不能为空") + } + + warehouseId, err := o.toolManager.Hyt.WarehouseSearch.Call(ctx, state.WarehouseName) + if err != nil { + log.Printf("warning: 仓库ID获取失败，%s: %v\n", state.WarehouseName, err) + } else { + state.mu.Lock() + defer state.mu.Unlock() + state.UploadReq.WarehouseId = warehouseId + } + + return state, nil + })) + + // 4. 合并/同步节点 + g.AddLambdaNode("merge_node", compose.InvokableLambda(func(ctx context.Context, state *ProductUploadContext) (*ProductUploadContext, error) { + // 最终校验 + if state.UploadReq.SupplierId == 0 { + return nil, fmt.Errorf("供应商获取失败") + } + if state.UploadReq.WarehouseId == 0 { + return nil, fmt.Errorf("仓库获取失败") + } + return state, nil + })) + + // 5. 上传节点 + g.AddLambdaNode("upload_product", compose.InvokableLambda(func(ctx context.Context, state *ProductUploadContext) (*ProductUploadContext, error) { + toolRes, err := o.toolManager.Hyt.ProductUpload.Call(ctx, state.UploadReq) + if err != nil { + return nil, fmt.Errorf("商品上传失败") + } + state.UploadResp = toolRes + return state, nil + })) + + // 6. 结果格式化节点 + g.AddLambdaNode("format_output", compose.InvokableLambda(func(ctx context.Context, state *ProductUploadContext) (map[string]any, error) { + if state.UploadResp == nil { + return nil, fmt.Errorf("upload response is nil") + } + return map[string]any{ + "预览URL(货易通商品列表)": state.UploadResp.PreviewUrl, + "SPU编码": state.UploadResp.SpuNum, + "商品ID": state.UploadResp.Id, + }, nil + })) + + // 构建边 + // Start -> Mapping + g.AddEdge(compose.START, "data_mapping") + + // 串行化执行以规避 Eino 指针合并问题 + // Mapping -> Supplier + g.AddEdge("data_mapping", "get_supplier_id") + + // Supplier -> Warehouse + g.AddEdge("get_supplier_id", "get_warehouse_id") + + // Warehouse -> Merge (虽然串行了，保留 Merge 节点做校验) + g.AddEdge("get_warehouse_id", "merge_node") + + // Merge -> Upload + g.AddEdge("merge_node", "upload_product") + + // Upload -> Format + g.AddEdge("upload_product", "format_output") + + // Format -> END + g.AddEdge("format_output", compose.END) + + return g.Compile(ctx) +} diff --git a/internal/domain/workflow/provider_set.go b/internal/domain/workflow/provider_set.go index c728b44..b9a2815 100644 --- a/internal/domain/workflow/provider_set.go +++ b/internal/domain/workflow/provider_set.go @@ -2,18 +2,28 @@ package workflow import ( "ai_scheduler/internal/config" + "ai_scheduler/internal/domain/component" + "ai_scheduler/internal/domain/repo" "ai_scheduler/internal/domain/workflow/runtime" "ai_scheduler/internal/pkg/utils_ollama" + toolManager "ai_scheduler/internal/domain/tools" + "github.com/google/wire" ) var ProviderSetWorkflow = wire.NewSet(NewRegistry) // NewRegistry 注入共享依赖并注册默认 Registry，确保自注册工作流可被发现 -func NewRegistry(conf *config.Config, llm *utils_ollama.Client) *runtime.Registry { +func NewRegistry(conf *config.Config, llm *utils_ollama.Client, repos *repo.Repos, components *component.Components) *runtime.Registry { // 步骤1：设置运行时依赖（配置与LLM客户端），供工作流工厂在首次实例化时使用；必须在任何调用 Invoke 之前完成，否则会触发 "deps not set" - runtime.SetDeps(&runtime.Deps{Conf: conf, LLM: llm}) + runtime.SetDeps(&runtime.Deps{ + Conf: conf, + LLM: llm, + ToolManager: toolManager.NewManager(conf), + Repos: repos, + Component: components, + }) // 步骤2：创建新的工作流注册表；注册表负责按工作流ID惰性实例化并缓存单例实例，保障并发访问下的安全 r := runtime.NewRegistry() // 步骤3：将该注册表设置为全局默认，便于通过 runtime.Default() 获取；自注册的工作流可通过默认注册表被发现并调用 diff --git a/internal/domain/workflow/registry.go b/internal/domain/workflow/registry.go index cbde3b6..10b24ef 100644 --- a/internal/domain/workflow/registry.go +++ b/internal/domain/workflow/registry.go @@ -2,11 +2,15 @@ package workflow import ( "ai_scheduler/internal/config" + "ai_scheduler/internal/domain/component" + toolManager "ai_scheduler/internal/domain/tools" "ai_scheduler/internal/pkg/utils_ollama" ) // 仅声明依赖结构，避免在 workflow 包内实现注册中心逻辑导致循环依赖 type Deps struct { - Conf *config.Config - LLM *utils_ollama.Client + Conf *config.Config + LLM *utils_ollama.Client + ToolManager *toolManager.Manager + Component *component.Components } diff --git a/internal/domain/workflow/runtime/registry.go b/internal/domain/workflow/runtime/registry.go index c854053..f804e1d 100644 --- a/internal/domain/workflow/runtime/registry.go +++ b/internal/domain/workflow/runtime/registry.go @@ -2,6 +2,9 @@ package runtime import ( "ai_scheduler/internal/config" + "ai_scheduler/internal/domain/component" + "ai_scheduler/internal/domain/repo" + toolManager "ai_scheduler/internal/domain/tools" "ai_scheduler/internal/entitys" "ai_scheduler/internal/pkg/utils_ollama" "context" @@ -11,13 +14,16 @@ import ( type Workflow interface { ID() string - Schema() map[string]any + // Schema() map[string]any Invoke(ctx context.Context, requireData *entitys.Recognize) (map[string]any, error) } type Deps struct { - Conf *config.Config - LLM *utils_ollama.Client + Conf *config.Config + LLM *utils_ollama.Client + ToolManager *toolManager.Manager + Component *component.Components // 基础设施能力 + Repos *repo.Repos // 数据访问 } type Factory func(deps *Deps) (Workflow, error) diff --git a/internal/domain/workflow/zltx/bug_optimization_submit.go b/internal/domain/workflow/zltx/bug_optimization_submit.go new file mode 100644 index 0000000..30ad0bc --- /dev/null +++ b/internal/domain/workflow/zltx/bug_optimization_submit.go @@ -0,0 +1,170 @@ +package zltx + +import ( + "context" + "encoding/json" + "errors" + "time" + + "ai_scheduler/internal/domain/component/callback" + "ai_scheduler/internal/domain/repo" + "ai_scheduler/internal/domain/workflow/runtime" + "ai_scheduler/internal/entitys" + "ai_scheduler/internal/pkg" + "ai_scheduler/internal/pkg/l_request" + + "github.com/cloudwego/eino/compose" + "github.com/google/uuid" + "github.com/redis/go-redis/v9" +) + +const WorkflowIDBugOptimizationSubmit = "bug_optimization_submit" + +func init() { + runtime.Register(WorkflowIDBugOptimizationSubmit, func(d *runtime.Deps) (runtime.Workflow, error) { + // 从 Deps.Repos 获取 SessionRepo + return &bugOptimizationSubmit{ + manager: d.Component.Callback, + sessionRepo: d.Repos.Session, + }, nil + }) +} + +type bugOptimizationSubmit struct { + manager callback.Manager + sessionRepo repo.SessionRepo + redisCli *redis.Client +} + +func (w *bugOptimizationSubmit) ID() string { + return WorkflowIDBugOptimizationSubmit +} + +type BugOptimizationSubmitInput struct { + Ch chan entitys.Response + RequireData *entitys.Recognize +} + +type BugOptimizationSubmitOutput struct { + Msg string +} + +type contextWithTask struct { + Input *BugOptimizationSubmitInput + TaskID string +} + +func (w *bugOptimizationSubmit) Invoke(ctx context.Context, recognize *entitys.Recognize) (map[string]any, error) { + chain, err := w.buildWorkflow(ctx) + if err != nil { + return nil, err + } + + input := &BugOptimizationSubmitInput{ + Ch: recognize.Ch, + RequireData: recognize, + } + + out, err := chain.Invoke(ctx, input) + if err != nil { + return nil, err + } + + return map[string]any{"msg": out.Msg}, nil +} + +func (w *bugOptimizationSubmit) buildWorkflow(ctx context.Context) (compose.Runnable[*BugOptimizationSubmitInput, *BugOptimizationSubmitOutput], error) { + c := compose.NewChain[*BugOptimizationSubmitInput, *BugOptimizationSubmitOutput]() + + // Node 1: Prepare and Call + c.AppendLambda(compose.InvokableLambda(w.prepareAndCall)) + + // Node 2: Wait + c.AppendLambda(compose.InvokableLambda(w.waitCallback)) + + return c.Compile(ctx) +} + +func (w *bugOptimizationSubmit) prepareAndCall(ctx context.Context, in *BugOptimizationSubmitInput) (*contextWithTask, error) { + // 生成 TaskID + taskID := uuid.New().String() + + // Ext 中获取 sessionId + sessionID := in.RequireData.GetSession() + + // 注册回调映射 + if err := w.manager.Register(ctx, taskID, sessionID); err != nil { + return nil, err + } + + // 查询用户名 + userName := "unknown" + if w.sessionRepo != nil { + name, err := w.sessionRepo.GetUserName(ctx, sessionID) + if err == nil && name != "" { + userName = name + } + } + + // 构建请求参数 + var fileUrls, fileContent string + if len(in.RequireData.UserContent.File) > 0 { + for _, file := range in.RequireData.UserContent.File { + fileUrls += file.FileUrl + "," + fileContent += file.FileRec + "," + } + fileUrls = fileUrls[:len(fileUrls)-1] + fileContent = fileContent[:len(fileContent)-1] + } + + body := map[string]string{ + "mark": in.RequireData.Match.Index, + "text": in.RequireData.UserContent.Text, + "img": fileUrls, + "img_content": fileContent, + "creator": userName, + "task_id": taskID, + } + + request := l_request.Request{ + Url: "https://connector.dingtalk.com/webhook/flow/10352c521dd02104cee9000c", + Method: "POST", + Headers: map[string]string{ + "Content-Type": "application/json", + }, + JsonByte: pkg.JsonByteIgonErr(body), + } + + res, err := request.Send() + if err != nil { + return nil, err + } + + var data map[string]any + if err := json.Unmarshal(res.Content, &data); err != nil { + return nil, err + } + + if success, ok := data["success"].(bool); !ok || !success { + return nil, errors.New("dingtalk flow failed") + } + + entitys.ResLog(in.Ch, in.RequireData.Match.Index, "问题记录中") + entitys.ResLoading(in.Ch, in.RequireData.Match.Index, "问题记录中...") + + return &contextWithTask{Input: in, TaskID: taskID}, nil +} + +func (w *bugOptimizationSubmit) waitCallback(ctx context.Context, in *contextWithTask) (*BugOptimizationSubmitOutput, error) { + // 阻塞等待回调信号 + // 设置 5 分钟超时 + waitCtx, cancel := context.WithTimeout(ctx, 5*time.Minute) + defer cancel() + + res, err := w.manager.Wait(waitCtx, in.TaskID, 5*time.Minute) + if err != nil { + return nil, err + } + + return &BugOptimizationSubmitOutput{Msg: res}, nil +} diff --git a/internal/domain/workflow/zltx/order_after_reseller_batch.go b/internal/domain/workflow/zltx/order_after_reseller_batch.go index 9749bf7..eee022a 100644 --- a/internal/domain/workflow/zltx/order_after_reseller_batch.go +++ b/internal/domain/workflow/zltx/order_after_reseller_batch.go @@ -22,8 +22,7 @@ func init() { } type orderAfterSaleResellerBatch struct { - cfg config.ToolConfig - data *OrderAfterSaleResellerBatchWorkflowInput + cfg config.ToolConfig } // 工作流入参 @@ -78,15 +77,6 @@ type OrderAfterSaleResellerBatchData struct { // ID 返回工作流唯一标识 func (o *orderAfterSaleResellerBatch) ID() string { return "zltx.orderAfterSaleResellerBatch" } -// Schema 返回入参约束（用于校验/表单生成） -func (o *orderAfterSaleResellerBatch) Schema() map[string]any { - return map[string]any{ - "type": "object", - "properties": map[string]any{"orderNumber": map[string]any{"type": "array", "items": map[string]any{"type": "string"}}}, - "required": []string{"orderNumber"}, - } -} - // Invoke 调用原有编排工作流并规范化输出 func (o *orderAfterSaleResellerBatch) Invoke(ctx context.Context, rec *entitys.Recognize) (map[string]any, error) { // 构建工作流 @@ -95,15 +85,19 @@ func (o *orderAfterSaleResellerBatch) Invoke(ctx context.Context, rec *entitys.R return nil, err } - o.data = &OrderAfterSaleResellerBatchWorkflowInput{ + input := &OrderAfterSaleResellerBatchWorkflowInput{ Ch: rec.Ch, UserInput: rec.UserContent.Text, FileContent: "", UserHistory: rec.ChatHis, ParameterResult: rec.Match.Parameters, } + + // 将 Input 注入 Context + ctx = context.WithValue(ctx, workflowInputContextKey{}, input) + // 工作流过程输出，不关注最终输出 - _, err = chain.Invoke(ctx, o.data) + _, err = chain.Invoke(ctx, input) if err != nil { return nil, err } @@ -116,6 +110,9 @@ func (o *orderAfterSaleResellerBatch) Invoke(ctx context.Context, rec *entitys.R var ErrInvalidOrderNumbers = errors.New("orderNumber 不能为空") +// contextKey 用于在 Context 中传递 WorkflowInput +type workflowInputContextKey struct{} + // buildWorkflow 构建工作流 func (o *orderAfterSaleResellerBatch) buildWorkflow(ctx context.Context) (compose.Runnable[*OrderAfterSaleResellerBatchWorkflowInput, *OrderAfterSaleResellerBatchWorkflowOutput], error) { // 定义工作流、出入参 @@ -136,39 +133,93 @@ func (o *orderAfterSaleResellerBatch) buildWorkflow(ctx context.Context) (compos }), )) + // 3.参数校验 & 传递 Input + // 注意：为了在后续节点访问 WorkflowInput，这里使用闭包或 Context 传递。 + // Eino Chain 节点间传递的是返回值。这里我们修改节点签名，将 input 一路传下去，或者使用 context。 + // 由于 Eino Chain 是强类型的，这里选择让 Parser 返回的数据结构包含原始 input，或者我们在 Parser 后重新组合。 + // 但最简单的方法是使用 Context 存储 Input (如果 Eino 支持 Context 传递)。Eino 的 Invoke 接受 ctx。 + // 但 Eino Chain 的设计是数据流驱动。 + // 修正方案：修改中间节点的数据结构，或者使用闭包捕获（但闭包捕获的是 build 时的变量，无法捕获运行时 input）。 + // 正确做法：Chain 的节点入参必须是上一个节点的出参。 + // 我们可以把 Parser 的输入改为 Input，输出改为一个包含 Input 和 ParsedData 的结构。 + // 但这里为了最小改动，我们利用 Context 来传递 Input 引用（这在 Eino 中是可行的，因为 ctx 会贯穿整个 Invoke）。 + // 更好的做法是重构 Chain 的数据流，但在保持逻辑不变的前提下，Context 是最快解法。 + + // 为了线程安全，我们在第一个节点把 Input 放入 Context？不行，Chain.Invoke(ctx, input) 的 ctx 是外部传入的。 + // Eino 允许 Lambda 修改 Context 吗？通常不允许。 + + // 让我们重新审视数据流： + // Input -> Lambda1 -> Message -> Parser -> NodeData -> Lambda4 -> ToolResp -> Lambda5 -> Output + // Lambda4 需要 Input.Ch 来发 Loading。 + // Lambda5 需要 Input.Ch 来发 Log/Json，还需要 NodeData。 + + // 根本问题是：中间节点丢失了 Input 信息。 + // 解决方案：使用一个聚合结构体在 Chain 中传递。 + + // 由于要大改数据流比较复杂，这里使用一种技巧： + // 在 Invoke 时，构造一个带有 Input 信息的 Context 传入。 + // 这样每个节点都能从 Context 拿到 Input。 + + // 重新实现 buildWorkflow 以支持 Context 传递 + return o.buildWorkflowWithContext(ctx) +} + +func (o *orderAfterSaleResellerBatch) buildWorkflowWithContext(ctx context.Context) (compose.Runnable[*OrderAfterSaleResellerBatchWorkflowInput, *OrderAfterSaleResellerBatchWorkflowOutput], error) { + c := compose.NewChain[*OrderAfterSaleResellerBatchWorkflowInput, *OrderAfterSaleResellerBatchWorkflowOutput]() + + // 0. Context 注入节点 (Trick: 利用第一个节点将 Input 注入 Context，但 Eino Chain 无法修改 Context 传递给下游) + // 实际上，我们可以在 Invoke 调用前，在外部包装 Context。 + // 所以这里不需要额外的节点，只需要在 Invoke 时处理。 + // 但 Invoke 是由 Chain 提供的，我们只能控制传入的 ctx。 + // 见下文 Invoke 方法的修改。 + + // 1.llm 推断参数 + c.AppendLambda(compose.InvokableLambda(func(ctx context.Context, in *OrderAfterSaleResellerBatchWorkflowInput) (*schema.Message, error) { + return &schema.Message{Content: in.ParameterResult}, nil + })) + + // 2.参数解析 + c.AppendLambda(compose.MessageParser( + schema.NewMessageJSONParser[*OrderAfterSaleResellerBatchNodeData](&schema.MessageJSONParseConfig{ + ParseFrom: schema.MessageParseFromContent, + }), + )) + // 3.参数校验 - c.AppendLambda(compose.InvokableLambda(func(_ context.Context, in *OrderAfterSaleResellerBatchNodeData) (*OrderAfterSaleResellerBatchNodeData, error) { - // 校验必填项 + c.AppendLambda(compose.InvokableLambda(func(ctx context.Context, in *OrderAfterSaleResellerBatchNodeData) (*OrderAfterSaleResellerBatchNodeData, error) { if len(in.OrderNumber) == 0 { return nil, ErrInvalidOrderNumbers } - - o.data.Data = in - + // 将解析后的 Data 存入 Input (通过 Context 获取 Input) + input := ctx.Value(workflowInputContextKey{}).(*OrderAfterSaleResellerBatchWorkflowInput) + input.Data = in // 这里修改 Input 是安全的，因为 Input 是请求维度的引用 return in, nil })) // 4.工具调用 - c.AppendLambda(compose.InvokableLambda(func(_ context.Context, in *OrderAfterSaleResellerBatchNodeData) (*toolZoarb.OrderAfterSaleResellerBatchResponse, error) { - entitys.ResLoading(o.data.Ch, o.ID(), "数据拉取中") + c.AppendLambda(compose.InvokableLambda(func(ctx context.Context, in *OrderAfterSaleResellerBatchNodeData) (*toolZoarb.OrderAfterSaleResellerBatchResponse, error) { + input := ctx.Value(workflowInputContextKey{}).(*OrderAfterSaleResellerBatchWorkflowInput) + entitys.ResLoading(input.Ch, o.ID(), "数据拉取中") toolRes, err := toolZoarb.Call(ctx, o.cfg, in.OrderNumber) - - entitys.ResLog(o.data.Ch, o.ID(), "数据拉取完成") + entitys.ResLog(input.Ch, o.ID(), "数据拉取完成") return toolRes, err })) // 5.结果数据映射 - c.AppendLambda(compose.InvokableLambda(o.dataMapping)) + c.AppendLambda(compose.InvokableLambda(func(ctx context.Context, in *toolZoarb.OrderAfterSaleResellerBatchResponse) (*OrderAfterSaleResellerBatchWorkflowOutput, error) { + return o.dataMapping(ctx, in) + })) - // 编译工作流 return c.Compile(ctx) } // 结果数据映射 -func (o *orderAfterSaleResellerBatch) dataMapping(_ context.Context, in *toolZoarb.OrderAfterSaleResellerBatchResponse) (*OrderAfterSaleResellerBatchWorkflowOutput, error) { - entitys.ResLog(o.data.Ch, o.ID(), "数据整理中") +func (o *orderAfterSaleResellerBatch) dataMapping(ctx context.Context, in *toolZoarb.OrderAfterSaleResellerBatchResponse) (*OrderAfterSaleResellerBatchWorkflowOutput, error) { + input := ctx.Value(workflowInputContextKey{}).(*OrderAfterSaleResellerBatchWorkflowInput) + + entitys.ResLog(input.Ch, o.ID(), "数据整理中") toolResp := &OrderAfterSaleResellerBatchWorkflowOutput{ Code: in.Code, @@ -179,17 +230,17 @@ func (o *orderAfterSaleResellerBatch) dataMapping(_ context.Context, in *toolZoa // 转换数据 for _, item := range in.Data.Data { // 处理方式 - afterType := util.StringToInt(o.data.Data.AfterType) + afterType := util.StringToInt(input.Data.AfterType) if afterType == 0 { afterType = 1 // 默认退款 } // 费用承担者 - responsibleType := util.StringToInt(o.data.Data.ResponsibleType) + responsibleType := util.StringToInt(input.Data.ResponsibleType) if responsibleType == 0 { responsibleType = 4 // 默认无 } // 售后金额 - afterSalesPrice := util.StringToFloat64(o.data.Data.AfterSalesPrice) + afterSalesPrice := util.StringToFloat64(input.Data.AfterSalesPrice) if afterSalesPrice == 0 { afterSalesPrice = item.OrderPrice } @@ -208,10 +259,10 @@ func (o *orderAfterSaleResellerBatch) dataMapping(_ context.Context, in *toolZoa Account: item.Account, Platforms: item.Platforms, AfterType: afterType, - Remark: o.data.Data.AfterSalesReason, + Remark: input.Data.AfterSalesReason, AfterAmount: afterSalesPrice, ResponsibleType: responsibleType, - ResponsiblePerson: o.data.Data.ResponsiblePerson, + ResponsiblePerson: input.Data.ResponsiblePerson, }) } @@ -224,7 +275,7 @@ func (o *orderAfterSaleResellerBatch) dataMapping(_ context.Context, in *toolZoa } toolRespJson, _ := json.Marshal(toolResp) - entitys.ResJson(o.data.Ch, o.ID(), string(toolRespJson)) + entitys.ResJson(input.Ch, o.ID(), string(toolRespJson)) return toolResp, nil } diff --git a/internal/entitys/recognize.go b/internal/entitys/recognize.go index 87f684c..fcd2fe5 100644 --- a/internal/entitys/recognize.go +++ b/internal/entitys/recognize.go @@ -3,6 +3,7 @@ package entitys import ( "ai_scheduler/internal/data/constants" "ai_scheduler/internal/data/model" + "encoding/json" ) type Recognize struct { @@ -47,3 +48,20 @@ type RecognizeFile struct { FileRealMime string // 文件真实MIME类型 FileUrl string // 文件下载链接 } + +func (r *Recognize) GetTaskExt() *TaskExt { + var ext TaskExt + if err := json.Unmarshal(r.Ext, &ext); err != nil { + return nil + } + + return &ext +} + +func (r *Recognize) GetSession() string { + ext := r.GetTaskExt() + if ext == nil { + return "" + } + return ext.Session +} diff --git a/internal/gateway/client.go b/internal/gateway/client.go index a293daa..1da0dd8 100644 --- a/internal/gateway/client.go +++ b/internal/gateway/client.go @@ -4,17 +4,18 @@ import ( errors "ai_scheduler/internal/data/error" "ai_scheduler/internal/data/model" "context" - "github.com/google/uuid" "log" "math/rand" "sync" "time" + "github.com/google/uuid" + "github.com/gofiber/websocket/v2" ) var ( - ErrConnClosed = errors.SysErr("连接不存在或已关闭") + ErrConnClosed = errors.SysErrf("连接不存在或已关闭") rng = rand.New(rand.NewSource(time.Now().UnixNano())) idBuf = make([]byte, 20) ) diff --git a/internal/pkg/dingtalk/contact_client.go b/internal/pkg/dingtalk/contact_client.go index 4461995..ed3e3dd 100644 --- a/internal/pkg/dingtalk/contact_client.go +++ b/internal/pkg/dingtalk/contact_client.go @@ -54,10 +54,10 @@ func (c *ContactClient) SearchUserOne(accessToken string, name string) (string, } if resp.Body == nil { - return "", errorcode.ParamErr("empty response body") + return "", errorcode.ParamErrf("empty response body") } if len(resp.Body.List) == 0 { - return "", errorcode.ParamErr("empty user list") + return "", errorcode.ParamErrf("empty user list") } userId := resp.Body.List[0] diff --git a/internal/pkg/dingtalk/notable_client.go b/internal/pkg/dingtalk/notable_client.go index 7cfe04f..d7d5434 100644 --- a/internal/pkg/dingtalk/notable_client.go +++ b/internal/pkg/dingtalk/notable_client.go @@ -67,7 +67,7 @@ func (c *NotableClient) UpdateRecord(accessToken string, req *UpdateRecordReq) ( } if resp.Body == nil { - return false, errorcode.ParamErr("empty response body") + return false, errorcode.ParamErrf("empty response body") } return true, nil diff --git a/internal/pkg/util/map.go b/internal/pkg/util/map.go new file mode 100644 index 0000000..5ca80c8 --- /dev/null +++ b/internal/pkg/util/map.go @@ -0,0 +1,14 @@ +package util + +import "encoding/json" + +// StructToMap 将结构体转换为 map[string]any +func StructToMap(v any) (map[string]any, error) { + b, err := json.Marshal(v) + if err != nil { + return nil, err + } + var m map[string]any + err = json.Unmarshal(b, &m) + return m, err +} diff --git a/internal/pkg/util/time.go b/internal/pkg/util/time.go new file mode 100644 index 0000000..1c1bf3e --- /dev/null +++ b/internal/pkg/util/time.go @@ -0,0 +1,41 @@ +package util + +import "time" + +// 判断当前时间是否在时间窗口内 +// ts 时间戳字符串，支持秒级或毫秒级 +// window 时间窗口，例如 10 * time.Minute +func IsInTimeWindow(ts string, window time.Duration) bool { + // 期望毫秒时间戳或秒级，简单容错 + // 尝试解析为整数 + var n int64 + for _, base := range []int64{1, 1000} { // 秒或毫秒 + if v, ok := parseInt64(ts); ok { + n = v + // 归一为毫秒 + if base == 1 && len(ts) <= 10 { + n = n * 1000 + } + now := time.Now().UnixMilli() + diff := now - n + if diff < 0 { + diff = -diff + } + if diff <= window.Milliseconds() { + return true + } + } + } + return false +} + +func parseInt64(s string) (int64, bool) { + var n int64 + for _, ch := range s { + if ch < '0' || ch > '9' { + return 0, false + } + n = n*10 + int64(ch-'0') + } + return n, true +} diff --git a/internal/pkg/utils_ollama/client.go b/internal/pkg/utils_ollama/client.go index 91640f0..fa88afa 100644 --- a/internal/pkg/utils_ollama/client.go +++ b/internal/pkg/utils_ollama/client.go @@ -90,6 +90,25 @@ func (c *Client) ChatStream(ctx context.Context, ch chan entitys.Response, messa return } +func (c *Client) Chat(ctx context.Context, model string, messages []api.Message) (res api.ChatResponse, err error) { + // 构建聊天请求 + req := &api.ChatRequest{ + Model: model, + Messages: messages, + Stream: new(bool), // 设置为false，不使用流式响应 + Think: &api.ThinkValue{Value: false}, + } + err = c.client.Chat(ctx, req, func(resp api.ChatResponse) error { + res = resp + return nil + }) + if err != nil { + return + } + + return +} + func (c *Client) Generation(ctx context.Context, generateRequest *api.GenerateRequest) (result api.GenerateResponse, err error) { err = c.client.Generate(ctx, generateRequest, func(resp api.GenerateResponse) error { result = resp diff --git a/internal/server/http.go b/internal/server/http.go index fd7e49e..53446c8 100644 --- a/internal/server/http.go +++ b/internal/server/http.go @@ -11,11 +11,13 @@ import ( ) type HTTPServer struct { - app *fiber.App - service *services.ChatService - session *services.SessionService - gateway *gateway.Gateway - callback *services.CallbackService + app *fiber.App + service *services.ChatService + session *services.SessionService + gateway *gateway.Gateway + callback *services.CallbackService + chatHis *services.HistoryService + capabilityService *services.CapabilityService } func NewHTTPServer( @@ -25,10 +27,11 @@ func NewHTTPServer( gateway *gateway.Gateway, callback *services.CallbackService, chatHis *services.HistoryService, + capabilityService *services.CapabilityService, ) *fiber.App { //构建 server app := initRoute() - router.SetupRoutes(app, service, session, task, gateway, callback, chatHis) + router.SetupRoutes(app, service, session, task, gateway, callback, chatHis, capabilityService) return app } diff --git a/internal/server/router/router.go b/internal/server/router/router.go index e2645bb..091c85f 100644 --- a/internal/server/router/router.go +++ b/internal/server/router/router.go @@ -15,16 +15,18 @@ import ( ) type RouterServer struct { - app *fiber.App - service *services.ChatService - session *services.SessionService - gateway *gateway.Gateway - chatHist *services.HistoryService + app *fiber.App + service *services.ChatService + session *services.SessionService + gateway *gateway.Gateway + chatHist *services.HistoryService + capabilityService *services.CapabilityService } // SetupRoutes 设置路由 func SetupRoutes(app *fiber.App, ChatService *services.ChatService, sessionService *services.SessionService, task *services.TaskService, gateway *gateway.Gateway, callbackService *services.CallbackService, chatHist *services.HistoryService, + capabilityService *services.CapabilityService, ) { app.Use(func(c *fiber.Ctx) error { // 设置 CORS 头 @@ -32,6 +34,11 @@ func SetupRoutes(app *fiber.App, ChatService *services.ChatService, sessionServi c.Set("Access-Control-Allow-Methods", "GET, POST, PUT, DELETE, OPTIONS") c.Set("Access-Control-Allow-Headers", "Content-Type, Authorization") + // AI能力调用路由，设置不同的 CORS 头 + if strings.HasPrefix(c.Path(), "/api/v1/capability") { + c.Set("Access-Control-Allow-Headers", "Content-Type, X-Source-Key, X-Timestamp") + } + // 如果是预检请求（OPTIONS），直接返回 204 if c.Method() == "OPTIONS" { return c.SendStatus(fiber.StatusNoContent) // 204 @@ -45,7 +52,6 @@ func SetupRoutes(app *fiber.App, ChatService *services.ChatService, sessionServi r := app.Group("api/v1/") registerResponse(r) - // 注册 CORS 中间件 r.Get("/health", func(c *fiber.Ctx) error { c.Response().SetBody([]byte("1")) return nil @@ -84,6 +90,10 @@ func SetupRoutes(app *fiber.App, ChatService *services.ChatService, sessionServi // 会话历史 r.Post("/chat/history/list", chatHist.List) r.Post("/chat/history/update/content", chatHist.UpdateContent) + + // 能力 + r.Post("/capability/product/ingest", capabilityService.ProductIngest) // 商品数据提取 + r.Post("/capability/product/ingest/:thread_id/confirm", capabilityService.ProductIngestConfirm) // 商品数据提取确认 } func routerSocket(app *fiber.App, chatService *services.ChatService) { diff --git a/internal/services/callback.go b/internal/services/callback.go index c903bea..c68e1c3 100644 --- a/internal/services/callback.go +++ b/internal/services/callback.go @@ -4,6 +4,7 @@ import ( "ai_scheduler/internal/config" "ai_scheduler/internal/data/constants" errorcode "ai_scheduler/internal/data/error" + "ai_scheduler/internal/domain/component/callback" "ai_scheduler/internal/entitys" "ai_scheduler/internal/gateway" "ai_scheduler/internal/pkg" @@ -25,17 +26,17 @@ type CallbackService struct { dingtalkOldClient *dingtalk.OldClient dingtalkContactClient *dingtalk.ContactClient dingtalkNotableClient *dingtalk.NotableClient - callBackTool *tool_callback.CallBackTool + callbackManager callback.Manager } -func NewCallbackService(cfg *config.Config, gateway *gateway.Gateway, dingtalkOldClient *dingtalk.OldClient, dingtalkContactClient *dingtalk.ContactClient, dingtalkNotableClient *dingtalk.NotableClient, callBackTool *tool_callback.CallBackTool) *CallbackService { +func NewCallbackService(cfg *config.Config, gateway *gateway.Gateway, dingtalkOldClient *dingtalk.OldClient, dingtalkContactClient *dingtalk.ContactClient, dingtalkNotableClient *dingtalk.NotableClient, callbackManager callback.Manager) *CallbackService { return &CallbackService{ cfg: cfg, gateway: gateway, dingtalkOldClient: dingtalkOldClient, dingtalkContactClient: dingtalkContactClient, dingtalkNotableClient: dingtalkNotableClient, - callBackTool: callBackTool, + callbackManager: callbackManager, } } @@ -77,20 +78,21 @@ func (s *CallbackService) Callback(c *fiber.Ctx) error { ts := strings.TrimSpace(c.Get("X-Timestamp")) // 时间窗口（如果提供了 ts 则校验，否则跳过），窗口 5 分钟 - if ts != "" && !validateTimestamp(ts, 5*time.Minute) { + // if ts != "" && !validateTimestamp(ts, 5*time.Minute) { + if ts != "" && !util.IsInTimeWindow(ts, 5*time.Minute) { return errorcode.AuthNotFound } // 解析 Envelope var env Envelope if err := json.Unmarshal(c.Body(), &env); err != nil { - return errorcode.ParamErr("invalid json: %v", err) + return errorcode.ParamErrf("invalid json: %v", err) } if env.Action == "" || env.TaskID == "" { - return errorcode.ParamErr("missing action/task_id") + return errorcode.ParamErrf("missing action/task_id") } if env.Data == nil { - return errorcode.ParamErr("missing data") + return errorcode.ParamErrf("missing data") } switch sourceKey { @@ -101,48 +103,51 @@ func (s *CallbackService) Callback(c *fiber.Ctx) error { } } -func validateTimestamp(ts string, window time.Duration) bool { - // 期望毫秒时间戳或秒级，简单容错 - // 尝试解析为整数 - var n int64 - for _, base := range []int64{1, 1000} { // 秒或毫秒 - if v, ok := parseInt64(ts); ok { - n = v - // 归一为毫秒 - if base == 1 && len(ts) <= 10 { - n = n * 1000 - } - now := time.Now().UnixMilli() - diff := now - n - if diff < 0 { - diff = -diff - } - if diff <= window.Milliseconds() { - return true - } - } - } - return false -} +// func validateTimestamp(ts string, window time.Duration) bool { +// // 期望毫秒时间戳或秒级，简单容错 +// // 尝试解析为整数 +// var n int64 +// for _, base := range []int64{1, 1000} { // 秒或毫秒 +// if v, ok := parseInt64(ts); ok { +// n = v +// // 归一为毫秒 +// if base == 1 && len(ts) <= 10 { +// n = n * 1000 +// } +// now := time.Now().UnixMilli() +// diff := now - n +// if diff < 0 { +// diff = -diff +// } +// if diff <= window.Milliseconds() { +// return true +// } +// } +// } +// return false +// } -func parseInt64(s string) (int64, bool) { - var n int64 - for _, ch := range s { - if ch < '0' || ch > '9' { - return 0, false - } - n = n*10 + int64(ch-'0') - } - return n, true -} +// func parseInt64(s string) (int64, bool) { +// var n int64 +// for _, ch := range s { +// if ch < '0' || ch > '9' { +// return 0, false +// } +// n = n*10 + int64(ch-'0') +// } +// return n, true +// } func (s *CallbackService) handleDingTalkCallback(c *fiber.Ctx, env Envelope) error { // 校验taskId - sessionID, ok := s.callBackTool.GetSessionByTaskID(env.TaskID) - if !ok { - return errorcode.ParamErr("missing session_id for task_id: %s", env.TaskID) - } ctx := c.Context() + sessionID, err := s.callbackManager.GetSession(ctx, env.TaskID) + if err != nil { + return errorcode.ParamErrf("failed to get session for task_id: %s, err: %v", env.TaskID, err) + } + if sessionID == "" { + return errorcode.ParamErrf("missing session_id for task_id: %s", env.TaskID) + } switch env.Action { case ActionBugOptimizationSubmitUpdate: @@ -165,8 +170,10 @@ func (s *CallbackService) handleDingTalkCallback(c *fiber.Ctx, env Envelope) err // 发送日志 s.sendStreamTxt(sessionID, msg) - // 删除映射 - s.callBackTool.DelTaskMapping(env.TaskID) + // 通知等待者 + if err := s.callbackManager.Notify(ctx, env.TaskID, msg); err != nil { + // 记录错误但继续 + } return c.JSON(fiber.Map{"code": 0, "message": "ok"}) case ActionBugOptimizationSubmitProcess: @@ -175,14 +182,14 @@ func (s *CallbackService) handleDingTalkCallback(c *fiber.Ctx, env Envelope) err } var data processData if err := json.Unmarshal(env.Data, &data); err != nil { - return errorcode.ParamErr("invalid json: %v", err) + return errorcode.ParamErrf("invalid json: %v", err) } s.sendStreamLoading(sessionID, data.Process) return c.JSON(fiber.Map{"code": 0, "message": "ok"}) default: - return errorcode.ParamErr("unknown action: %s", env.Action) + return errorcode.ParamErrf("unknown action: %s", env.Action) } } @@ -254,27 +261,27 @@ func (s *CallbackService) sendStreamLoading(sessionID string, content string) { func (s *CallbackService) handleBugOptimizationSubmitUpdate(ctx context.Context, taskData json.RawMessage) (string, *errorcode.BusinessErr) { var data BugOptimizationSubmitUpdateData if err := json.Unmarshal(taskData, &data); err != nil { - return "", errorcode.ParamErr("invalid data type: %v", err) + return "", errorcode.ParamErrf("invalid data type: %v", err) } if data.Creator == "" { - return "", errorcode.ParamErr("empty creator") + return "", errorcode.ParamErrf("empty creator") } // 获取创建者uid accessToken, _ := s.dingtalkOldClient.GetAccessToken() creatorId, err := s.dingtalkContactClient.SearchUserOne(accessToken, data.Creator) if err != nil { - return "", errorcode.ParamErr("invalid data type: %v", err) + return "", errorcode.ParamErrf("invalid data type: %v", err) } // 获取用户详情 userDetails, err := s.dingtalkOldClient.QueryUserDetails(ctx, creatorId) if err != nil { - return "", errorcode.ParamErr("invalid data type: %v", err) + return "", errorcode.ParamErrf("invalid data type: %v", err) } if userDetails == nil { - return "", errorcode.ParamErr("user details not found") + return "", errorcode.ParamErrf("user details not found") } unionId := userDetails.UnionID @@ -287,10 +294,10 @@ func (s *CallbackService) handleBugOptimizationSubmitUpdate(ctx context.Context, CreatorUnionId: unionId, }) if err != nil { - return "", errorcode.ParamErr("invalid data type: %v", err) + return "", errorcode.ParamErrf("invalid data type: %v", err) } if !ok { - return "", errorcode.ParamErr("update record failed") + return "", errorcode.ParamErrf("update record failed") } return "问题记录即将完成", nil @@ -300,16 +307,16 @@ func (s *CallbackService) handleBugOptimizationSubmitUpdate(ctx context.Context, func (s *CallbackService) handleBugOptimizationSubmitDone(ctx context.Context, taskData json.RawMessage) (string, *errorcode.BusinessErr) { var data BugOptimizationSubmitDoneData if err := json.Unmarshal(taskData, &data); err != nil { - return "", errorcode.ParamErr("invalid data type: %v", err) + return "", errorcode.ParamErrf("invalid data type: %v", err) } if len(data.Receivers) == 0 { - return "", errorcode.ParamErr("empty receivers") + return "", errorcode.ParamErrf("empty receivers") } // 构建接收者 receivers := s.getDingtalkReceivers(ctx, data.Receivers) if receivers == "" { - return "", errorcode.ParamErr("invalid receivers") + return "", errorcode.ParamErrf("invalid receivers") } // 构建跳转链接 diff --git a/internal/services/capability.go b/internal/services/capability.go new file mode 100644 index 0000000..759433c --- /dev/null +++ b/internal/services/capability.go @@ -0,0 +1,207 @@ +package services + +import ( + "ai_scheduler/internal/config" + "ai_scheduler/internal/data/constants" + errorcode "ai_scheduler/internal/data/error" + "ai_scheduler/internal/domain/workflow/runtime" + "ai_scheduler/internal/entitys" + "ai_scheduler/internal/pkg/util" + "ai_scheduler/internal/pkg/utils_ollama" + "ai_scheduler/utils" + "context" + "encoding/json" + "fmt" + "strings" + "time" + + hytWorkflow "ai_scheduler/internal/domain/workflow/hyt" + + "github.com/gofiber/fiber/v2" + "github.com/google/uuid" + "github.com/ollama/ollama/api" + "github.com/redis/go-redis/v9" +) + +// CapabilityService 统一回调入口 +type CapabilityService struct { + cfg *config.Config + workflowManager *runtime.Registry + rdsCli *redis.Client +} + +func NewCapabilityService(cfg *config.Config, workflowManager *runtime.Registry, rdb *utils.Rdb) *CapabilityService { + return &CapabilityService{ + cfg: cfg, + workflowManager: workflowManager, + rdsCli: rdb.Rdb, + } +} + +// 产品数据提取入参 +type ProductIngestReq struct { + SysId string `json:"sys_id"` // 业务系统ID - 当前仅支持货易通（hyt） + Url string `json:"url"` // 商品详情页URL + Title string `json:"title"` // 商品标题 + Text string `json:"text"` // 商品描述 + Images []string `json:"images"` // 商品图片URL列表 +} + +type ProductIngestResp struct { + ThreadId string `json:"thread_id"` // 线程ID，后续确认调用时需要 + SysId string `json:"sys_id"` // 业务系统ID + MetaData any `json:"meta"` // 元数据 + Draft string `json:"draft"` // 草稿数据，后续确认调用时需要 +} + +// ProductIngest 产品数据提取 +func (s *CapabilityService) ProductIngest(c *fiber.Ctx) error { + ctx := context.Background() + // 请求头校验 + if err := s.checkRequestHeader(c); err != nil { + return err + } + + // 解析请求参数 + req := ProductIngestReq{} + if err := c.BodyParser(&req); err != nil { + return errorcode.ParamErrf("invalid request body: %v", err) + } + // 必要参数校验 + if req.Text == "" || req.SysId == "" { + return errorcode.ParamErrf("missing required fields") + } + + // 映射目标系统商品属性中文模板 + var sysProductPropertyTemplateZH string + switch req.SysId { + case "hyt": // 货易通 + sysProductPropertyTemplateZH = constants.HYTGoodsAddPropertyTemplateZH + default: + return errorcode.ParamErrf("invalid sys_id") + } + + // 模型调用 + client, cleanup, err := utils_ollama.NewClient(s.cfg) + if err != nil { + return err + } + defer cleanup() + res, err := client.Chat(ctx, s.cfg.Ollama.MappingModel, []api.Message{ + { + Role: "system", + Content: constants.SystemPrompt, + }, + { + Role: "assistant", + Content: fmt.Sprintf("目标属性模板：%s。", sysProductPropertyTemplateZH), + }, + { + Role: "user", + Content: req.Text, + }, + { + Role: "user", + Content: "商品图片URL列表：" + strings.Join(req.Images, ","), + }, + }) + if err != nil { + return err + } + + // 生成thread_id + threadId := uuid.NewString() + resp := &ProductIngestResp{ + ThreadId: threadId, + SysId: req.SysId, + MetaData: req, + Draft: res.Message.Content, // Go中map会无序，交给前端解析 + } + respJson, _ := json.Marshal(resp) + + // 存redis缓存 + if err = s.rdsCli.Set(ctx, fmt.Sprintf(constants.CapabilityProductIngestCacheKey, threadId), respJson, 30*time.Minute).Err(); err != nil { + return err + } + + // 解析模型输出 + c.JSON(resp) + + return nil +} + +// checkRequestHeader 校验请求头 +func (s *CapabilityService) checkRequestHeader(c *fiber.Ctx) error { + // 读取头 + token := strings.TrimSpace(c.Get("X-Source-Key")) + ts := strings.TrimSpace(c.Get("X-Timestamp")) + + // 时间窗口校验 + if ts != "" && !util.IsInTimeWindow(ts, 5*time.Minute) { + return errorcode.AuthNotFound + } + // token校验 + if token == "" || token != constants.CapabilityProductIngestToken { + return errorcode.KeyNotFound + } + + return nil +} + +type ProductIngestConfirmReq struct { + ThreadId string `json:"thread_id"` // 线程ID + Confirmed string `json:"confirmed"` // 已确认数据json字符串 +} + +// ProductIngestConfirm 商品数据提取确认 +func (s *CapabilityService) ProductIngestConfirm(c *fiber.Ctx) error { + ctx := context.Background() + + // 请求头校验 + if err := s.checkRequestHeader(c); err != nil { + return err + } + // 获取路径参数中的 thread_id + threadId := c.Params("thread_id") + if threadId == "" { + return errorcode.ParamErrf("missing required fields") + } + // 解析请求参数 body + req := ProductIngestConfirmReq{} + if err := c.BodyParser(&req); err != nil { + return errorcode.ParamErrf("invalid request body: %v", err) + } + // 必要参数校验 + if req.Confirmed == "" || threadId == "" { + return errorcode.ParamErr("missing required fields") + } + + // 校验线程ID是否存在 + resp, err := s.rdsCli.Get(ctx, fmt.Sprintf(constants.CapabilityProductIngestCacheKey, threadId)).Result() + if err != nil { + return errorcode.ParamErr("invalid thread_id") + } + var respData ProductIngestResp + if err = json.Unmarshal([]byte(resp), &respData); err != nil { + return errorcode.ParamErr("invalid thread_id data") + } + + // 映射目标系统工作流ID + var workflowId string + switch respData.SysId { + // 货易通 + case "hyt": + workflowId = hytWorkflow.WorkflowIDGoodsAdd + default: + return errorcode.ParamErr("invalid sys_id") + } + + // 调用eino工作流，实现商品上传到目标系统 + rec := &entitys.Recognize{UserContent: &entitys.RecognizeUserContent{Text: req.Confirmed}} + res, err := s.workflowManager.Invoke(ctx, workflowId, rec) + if err != nil { + return err + } + + return c.JSON(res) +} diff --git a/internal/services/provider_set.go b/internal/services/provider_set.go index 867eb11..375a886 100644 --- a/internal/services/provider_set.go +++ b/internal/services/provider_set.go @@ -12,4 +12,6 @@ var ProviderSetServices = wire.NewSet( NewTaskService, NewCallbackService, NewDingBotService, - NewHistoryService) + NewHistoryService, + NewCapabilityService, +) diff --git a/internal/tool_callback/bug_optimization_submit.go b/internal/tool_callback/bug_optimization_submit.go index 245ab8c..57a5e73 100644 --- a/internal/tool_callback/bug_optimization_submit.go +++ b/internal/tool_callback/bug_optimization_submit.go @@ -60,7 +60,7 @@ func (w *CallBackTool) BugOptimizationSubmit(ctx context.Context, requireData *e cond = cond.And(builder.Eq{"session_id": requireData.Session}) sessionInfo, err := w.sessionImpl.GetOneBySearch(&cond) if err != nil { - err = errors.SysErr("获取会话信息失败：%v", err.Error()) + err = errors.SysErrf("获取会话信息失败：%v", err.Error()) return } userName := sessionInfo["user_name"].(string)