diff --git a/internal/biz/ding_talk_bot.go b/internal/biz/ding_talk_bot.go index af040c2..40f99fb 100644 --- a/internal/biz/ding_talk_bot.go +++ b/internal/biz/ding_talk_bot.go @@ -7,6 +7,8 @@ import ( "ai_scheduler/internal/data/constants" "ai_scheduler/internal/data/impl" "ai_scheduler/internal/data/model" + "ai_scheduler/internal/domain/workflow/recharge" + "ai_scheduler/internal/domain/workflow/runtime" "ai_scheduler/internal/entitys" "ai_scheduler/internal/pkg/l_request" "ai_scheduler/internal/pkg/utils_oss" @@ -35,19 +37,20 @@ import ( // AiRouterBiz 智能路由服务 type DingTalkBotBiz struct { - do *do.Do - handle *do.Handle - botConfigImpl *impl.BotConfigImpl - replier *chatbot.ChatbotReplier - log log.Logger - dingTalkUser *dingtalk.User - botTools []model.AiBotTool - botGroupImpl *impl.BotGroupImpl - toolManager *tools.Manager - chatHis *impl.BotChatHisImpl - conf *config.Config - cardSend *dingtalk.SendCardClient - ossClient *utils_oss.Client + do *do.Do + handle *do.Handle + botConfigImpl *impl.BotConfigImpl + replier *chatbot.ChatbotReplier + log log.Logger + dingTalkUser *dingtalk.User + botTools []model.AiBotTool + botGroupImpl *impl.BotGroupImpl + toolManager *tools.Manager + chatHis *impl.BotChatHisImpl + conf *config.Config + cardSend *dingtalk.SendCardClient + ossClient *utils_oss.Client + workflowManager *runtime.Registry } // NewDingTalkBotBiz @@ -63,20 +66,22 @@ func NewDingTalkBotBiz( conf *config.Config, cardSend *dingtalk.SendCardClient, ossClient *utils_oss.Client, + workflowManager *runtime.Registry, ) *DingTalkBotBiz { return &DingTalkBotBiz{ - do: do, - handle: handle, - botConfigImpl: botConfigImpl, - replier: chatbot.NewChatbotReplier(), - dingTalkUser: dingTalkUser, - botTools: tools.BootTools, - botGroupImpl: botGroupImpl, - toolManager: toolManager, - chatHis: chatHis, - conf: conf, - cardSend: cardSend, - ossClient: ossClient, + do: do, + handle: handle, + botConfigImpl: botConfigImpl, + replier: chatbot.NewChatbotReplier(), + dingTalkUser: dingTalkUser, + botTools: tools.BootTools, + botGroupImpl: botGroupImpl, + toolManager: toolManager, + chatHis: chatHis, + conf: conf, + cardSend: cardSend, + ossClient: ossClient, + workflowManager: workflowManager, } } @@ -623,6 +628,38 @@ func (d *DingTalkBotBiz) GetReportLists(ctx context.Context, group *model.AiBotG return } + // 追加电商充值系统统计 - 返回统一使用 []*bbxt.ReportRes + rechargeReports, err := d.rechargeDailyReport(ctx, time.Now(), []string{"官方-爱奇艺-星钻季卡", "官方-爱奇艺-星钻半年卡", "官方--腾讯-年卡", "官方--爱奇艺-月卡"}, d.ossClient) + reports = append(reports, rechargeReports...) + + return +} + +// rechargeDailyReport 获取电商充值系统统计报告 +func (d *DingTalkBotBiz) rechargeDailyReport(ctx context.Context, now time.Time, productNames []string, ossClient *utils_oss.Client) (reports []*bbxt.ReportRes, err error) { + workflowId := recharge.WorkflowIDStatisticsOursProduct + args := &runtime.WorkflowArgs{ + Args: map[string]any{ + "product_names": productNames, + "now": now, + }, + } + res, err := d.workflowManager.Invoke(ctx, workflowId, args) + if err != nil { + return + } + + reports = []*bbxt.ReportRes{ + { + ReportName: "我们的商品统计(电商充值系统)", + Title: fmt.Sprintf("%s 电商充值系统我们的商品统计", now.Format("2006-01-02")), + Path: res["path"].(string), + Url: res["url"].(string), + Data: res["data"].([][]string), + Desc: res["desc"].(string), + }, + } + return } diff --git a/internal/biz/do/handle.go b/internal/biz/do/handle.go index a1a9dcf..afc4b2b 100644 --- a/internal/biz/do/handle.go +++ b/internal/biz/do/handle.go @@ -396,7 +396,7 @@ func (r *Handle) handleEinoWorkflow(ctx context.Context, rec *entitys.Recognize, // 工作流内部输出 workflowId := task.Index - _, err = r.workflowManager.Invoke(ctx, workflowId, rec) + _, err = r.workflowManager.Invoke(ctx, workflowId, &runtime.WorkflowArgs{Recognize: rec}) if err != nil { return err } diff --git a/internal/domain/tools/common/image_converter/client.go b/internal/domain/tools/common/image_converter/client.go index 5eba199..2de57d2 100644 --- a/internal/domain/tools/common/image_converter/client.go +++ b/internal/domain/tools/common/image_converter/client.go @@ -21,7 +21,7 @@ func New(cfg config.ToolConfig) *Client { } // Call 将 Excel 文件转换为图片 -func (c *Client) Call(filename string, fileBytes []byte) ([]byte, error) { +func (c *Client) Call(filename string, fileBytes []byte, scale int) ([]byte, error) { body := &bytes.Buffer{} writer := multipart.NewWriter(body) @@ -33,6 +33,14 @@ func (c *Client) Call(filename string, fileBytes []byte) ([]byte, error) { return nil, err } + // 添加 scale 参数 + if scale <= 0 { + scale = 2 + } + if err = writer.WriteField("scale", fmt.Sprintf("%d", scale)); err != nil { + return nil, err + } + if err = writer.Close(); err != nil { return nil, err } diff --git a/internal/domain/workflow/hyt/goods_add.go b/internal/domain/workflow/hyt/goods_add.go index 91db063..c41b66b 100644 --- a/internal/domain/workflow/hyt/goods_add.go +++ b/internal/domain/workflow/hyt/goods_add.go @@ -8,7 +8,6 @@ import ( "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" @@ -42,7 +41,7 @@ type GoodsAddWorkflowInput struct { func (o *goodsAdd) ID() string { return WorkflowIDGoodsAdd } -func (o *goodsAdd) Invoke(ctx context.Context, rec *entitys.Recognize) (map[string]any, error) { +func (o *goodsAdd) Invoke(ctx context.Context, rec *runtime.WorkflowArgs) (map[string]any, error) { // 构建工作流 runnable, err := o.buildWorkflow(ctx) if err != nil { diff --git a/internal/domain/workflow/hyt/product_upload.go b/internal/domain/workflow/hyt/product_upload.go index 35114ed..b2af877 100644 --- a/internal/domain/workflow/hyt/product_upload.go +++ b/internal/domain/workflow/hyt/product_upload.go @@ -6,7 +6,6 @@ import ( 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" @@ -39,7 +38,7 @@ type ProductUploadWorkflowInput struct { func (o *productUpload) ID() string { return WorkflowIDProductUpload } -func (o *productUpload) Invoke(ctx context.Context, rec *entitys.Recognize) (map[string]any, error) { +func (o *productUpload) Invoke(ctx context.Context, rec *runtime.WorkflowArgs) (map[string]any, error) { // 构建工作流 runnable, err := o.buildWorkflow(ctx) if err != nil { diff --git a/internal/domain/workflow/recharge/statistics_ours_product.go b/internal/domain/workflow/recharge/statistics_ours_product.go index d8bd64a..ebc33a9 100644 --- a/internal/domain/workflow/recharge/statistics_ours_product.go +++ b/internal/domain/workflow/recharge/statistics_ours_product.go @@ -6,7 +6,6 @@ import ( toolManager "ai_scheduler/internal/domain/tools" "ai_scheduler/internal/domain/tools/recharge/statistics_ours_product" "ai_scheduler/internal/domain/workflow/runtime" - "ai_scheduler/internal/entitys" "ai_scheduler/internal/pkg/utils_oss" "context" "errors" @@ -37,28 +36,31 @@ type StatisticsOursProductWorkflowInput struct { } type StatisticsOursProductWorkflowOutput struct { - ImgUrl string `json:"img_url"` + Path string `json:"path"` + Url string `json:"url"` + Data [][]string `json:"data"` + Desc string `json:"desc"` } func (w *statisticsOursProduct) ID() string { return WorkflowIDStatisticsOursProduct } -func (w *statisticsOursProduct) Invoke(ctx context.Context, rec *entitys.Recognize) (map[string]any, error) { +func (w *statisticsOursProduct) Invoke(ctx context.Context, args *runtime.WorkflowArgs) (map[string]any, error) { // 构建工作流 runnable, err := w.buildWorkflow(ctx) if err != nil { return nil, err } - // 解析参数 (假设参数在 rec.Match.Parameters 中，或者根据实际情况解析) - // 这里简化处理，假设需要解析参数 - // 实际上这里应该根据 LLM 解析的结果来填充 Input - // 暂时假设 ParameterResult 是 JSON 字符串 + // 获取参数时间 + now := args.Args["now"].(time.Time) input := &StatisticsOursProductWorkflowInput{ // 默认值，具体应从 rec 解析 - StartTime: time.Now().Format("2006010200"), - EndTime: time.Now().Format("2006010223"), + StartTime: now.Format("2006010200"), + EndTime: now.Format("2006010215"), } + input.StartTime = "2025122300" + // 工作流过程调用 output, err := runnable.Invoke(ctx, input) if err != nil { @@ -70,11 +72,11 @@ func (w *statisticsOursProduct) Invoke(ctx context.Context, rec *entitys.Recogni return nil, errorcode.WorkflowErr(errStr) } - return map[string]any{"img_url": output.ImgUrl}, nil + return output, nil } -func (w *statisticsOursProduct) buildWorkflow(ctx context.Context) (compose.Runnable[*StatisticsOursProductWorkflowInput, *StatisticsOursProductWorkflowOutput], error) { - c := compose.NewChain[*StatisticsOursProductWorkflowInput, *StatisticsOursProductWorkflowOutput]() +func (w *statisticsOursProduct) buildWorkflow(ctx context.Context) (compose.Runnable[*StatisticsOursProductWorkflowInput, map[string]any], error) { + c := compose.NewChain[*StatisticsOursProductWorkflowInput, map[string]any]() // 1. 调用工具统计我们的商品 c.AppendLambda(compose.InvokableLambda(w.callStatisticsTool)) @@ -82,6 +84,9 @@ func (w *statisticsOursProduct) buildWorkflow(ctx context.Context) (compose.Runn // 2. 生成 Excel 并转图片上传 c.AppendLambda(compose.InvokableLambda(w.generateExcelAndUpload)) + // 3. 转map输出 + c.AppendLambda(compose.InvokableLambda(w.convertToMap)) + return c.Compile(ctx) } @@ -98,20 +103,20 @@ func (w *statisticsOursProduct) callStatisticsTool(ctx context.Context, input *S func (w *statisticsOursProduct) generateExcelAndUpload(ctx context.Context, data []statistics_ours_product.StatisticsOursProductItem) (*StatisticsOursProductWorkflowOutput, error) { // 2. 获取模板路径 (假设在项目根目录的 assets/templates 下) cwd, _ := filepath.Abs(".") - templatePath := filepath.Join(cwd, "assets", "templates", "statistics_ours_product.xlsx") + templatePath := filepath.Join(cwd, "tmpl", "excel_temp", "recharge_statistics_ours_product.xlsx") fileName := fmt.Sprintf("statistics_ours_product_%d", time.Now().Unix()) // 3. 转换数据为 [][]string excelData := w.convertDataToExcelFormat(data) // 4. 生成 Excel - excelBytes, err := w.toolManager.Common.ExcelGenerator.Call(templatePath, excelData, 2, 2) + excelBytes, err := w.toolManager.Common.ExcelGenerator.Call(templatePath, excelData, 4, 3) if err != nil { return nil, fmt.Errorf("生成 Excel 失败: %v", err) } // 5. Excel 转图片 - picBytes, err := w.toolManager.Common.ImageConverter.Call(fileName+".xlsx", excelBytes) + picBytes, err := w.toolManager.Common.ImageConverter.Call(fileName+".xlsx", excelBytes, 2) if err != nil { return nil, fmt.Errorf("Excel 转图片失败: %v", err) } @@ -122,7 +127,14 @@ func (w *statisticsOursProduct) generateExcelAndUpload(ctx context.Context, data return nil, fmt.Errorf("上传 OSS 失败: %v", err) } - return &StatisticsOursProductWorkflowOutput{ImgUrl: url}, nil + res := &StatisticsOursProductWorkflowOutput{ + Path: "", + Url: url, + Data: excelData, + Desc: "", + } + + return res, nil } // convertDataToExcelFormat 将业务数据转换为 Excel 生成器需要的二维字符串数组 @@ -131,16 +143,26 @@ func (w *statisticsOursProduct) convertDataToExcelFormat(data []statistics_ours_ for _, item := range data { row := []string{ item.OursProductName, - fmt.Sprintf("%d", item.OursProductId), + // fmt.Sprintf("%d", item.OursProductId), item.Count, - item.TotalPrice, - item.SuccessCount, + // item.TotalPrice, + // item.SuccessCount, item.SuccessPrice, - item.FailCount, - item.FailPrice, + // item.FailCount, + // item.FailPrice, item.Profit, } + result = append(result, row) } return result } + +func (w *statisticsOursProduct) convertToMap(ctx context.Context, output *StatisticsOursProductWorkflowOutput) (map[string]any, error) { + return map[string]any{ + "path": output.Path, + "url": output.Url, + "data": output.Data, + "desc": output.Desc, + }, nil +} diff --git a/internal/domain/workflow/runtime/registry.go b/internal/domain/workflow/runtime/registry.go index f804e1d..1d391f2 100644 --- a/internal/domain/workflow/runtime/registry.go +++ b/internal/domain/workflow/runtime/registry.go @@ -15,7 +15,7 @@ import ( type Workflow interface { ID() string // Schema() map[string]any - Invoke(ctx context.Context, requireData *entitys.Recognize) (map[string]any, error) + Invoke(ctx context.Context, requireData *WorkflowArgs) (map[string]any, error) } type Deps struct { @@ -28,6 +28,11 @@ type Deps struct { type Factory func(deps *Deps) (Workflow, error) +type WorkflowArgs struct { + *entitys.Recognize + Args map[string]any +} + var ( regMu sync.RWMutex factories = map[string]Factory{} @@ -69,7 +74,7 @@ func Default() *Registry { return r } -func (r *Registry) Invoke(ctx context.Context, id string, rec *entitys.Recognize) (map[string]any, error) { +func (r *Registry) Invoke(ctx context.Context, id string, rec *WorkflowArgs) (map[string]any, error) { regMu.RLock() f, ok := factories[id] regMu.RUnlock() diff --git a/internal/domain/workflow/zltx/bug_optimization_submit.bak.go b/internal/domain/workflow/zltx/bug_optimization_submit.bak.go index 6ed6bb4..0ab94ee 100644 --- a/internal/domain/workflow/zltx/bug_optimization_submit.bak.go +++ b/internal/domain/workflow/zltx/bug_optimization_submit.bak.go @@ -42,7 +42,7 @@ func (w *bugOptimizationSubmitBak) ID() string { type BugOptimizationSubmitBakInput struct { Ch chan entitys.Response - RequireData *entitys.Recognize + RequireData *runtime.WorkflowArgs } type BugOptimizationSubmitBakOutput struct { @@ -54,7 +54,7 @@ type contextWithTaskBak struct { TaskID string } -func (w *bugOptimizationSubmitBak) Invoke(ctx context.Context, recognize *entitys.Recognize) (map[string]any, error) { +func (w *bugOptimizationSubmitBak) Invoke(ctx context.Context, recognize *runtime.WorkflowArgs) (map[string]any, error) { chain, err := w.buildWorkflow(ctx) if err != nil { return nil, err diff --git a/internal/domain/workflow/zltx/bug_optimization_submit.go b/internal/domain/workflow/zltx/bug_optimization_submit.go index 30ad0bc..ccf6812 100644 --- a/internal/domain/workflow/zltx/bug_optimization_submit.go +++ b/internal/domain/workflow/zltx/bug_optimization_submit.go @@ -42,7 +42,7 @@ func (w *bugOptimizationSubmit) ID() string { type BugOptimizationSubmitInput struct { Ch chan entitys.Response - RequireData *entitys.Recognize + RequireData *runtime.WorkflowArgs } type BugOptimizationSubmitOutput struct { @@ -54,7 +54,7 @@ type contextWithTask struct { TaskID string } -func (w *bugOptimizationSubmit) Invoke(ctx context.Context, recognize *entitys.Recognize) (map[string]any, error) { +func (w *bugOptimizationSubmit) Invoke(ctx context.Context, recognize *runtime.WorkflowArgs) (map[string]any, error) { chain, err := w.buildWorkflow(ctx) if err != nil { return nil, err diff --git a/internal/domain/workflow/zltx/order_after_reseller_batch.go b/internal/domain/workflow/zltx/order_after_reseller_batch.go index eee022a..620d636 100644 --- a/internal/domain/workflow/zltx/order_after_reseller_batch.go +++ b/internal/domain/workflow/zltx/order_after_reseller_batch.go @@ -78,7 +78,7 @@ type OrderAfterSaleResellerBatchData struct { func (o *orderAfterSaleResellerBatch) ID() string { return "zltx.orderAfterSaleResellerBatch" } // Invoke 调用原有编排工作流并规范化输出 -func (o *orderAfterSaleResellerBatch) Invoke(ctx context.Context, rec *entitys.Recognize) (map[string]any, error) { +func (o *orderAfterSaleResellerBatch) Invoke(ctx context.Context, rec *runtime.WorkflowArgs) (map[string]any, error) { // 构建工作流 chain, err := o.buildWorkflow(ctx) if err != nil { diff --git a/internal/services/capability.go b/internal/services/capability.go index 759433c..d2c1e9b 100644 --- a/internal/services/capability.go +++ b/internal/services/capability.go @@ -198,7 +198,7 @@ func (s *CapabilityService) ProductIngestConfirm(c *fiber.Ctx) error { // 调用eino工作流，实现商品上传到目标系统 rec := &entitys.Recognize{UserContent: &entitys.RecognizeUserContent{Text: req.Confirmed}} - res, err := s.workflowManager.Invoke(ctx, workflowId, rec) + res, err := s.workflowManager.Invoke(ctx, workflowId, &runtime.WorkflowArgs{Recognize: rec}) if err != nil { return err } diff --git a/internal/services/dtalk_bot_test.go b/internal/services/dtalk_bot_test.go index a9a75e0..bd0ceef 100644 --- a/internal/services/dtalk_bot_test.go +++ b/internal/services/dtalk_bot_test.go @@ -12,6 +12,7 @@ import ( "ai_scheduler/internal/domain/component/callback" "ai_scheduler/internal/domain/repo" "ai_scheduler/internal/domain/workflow" + "ai_scheduler/internal/domain/workflow/runtime" "ai_scheduler/internal/pkg" "ai_scheduler/internal/pkg/dingtalk" "ai_scheduler/internal/pkg/utils_ollama" @@ -102,7 +103,9 @@ func run() { handle := do.NewHandle(ollamaService, manager, configConfig, sessionImpl, registry, oldClient, contactClient, notableClient) // 初始化钉钉机器人业务逻辑 utils_ossClient, _ := utils_oss.NewClient(configConfig) - dingTalkBotBiz := biz.NewDingTalkBotBiz(doDo, handle, botConfigImpl, botGroupImpl, user, toolRegis, botChatHisImpl, manager, configConfig, sendCardClient, utils_ossClient) + // 初始化工作流管理器 + workflowManager := runtime.NewRegistry() + dingTalkBotBiz := biz.NewDingTalkBotBiz(doDo, handle, botConfigImpl, botGroupImpl, user, toolRegis, botChatHisImpl, manager, configConfig, sendCardClient, utils_ossClient, workflowManager) // 初始化钉钉机器人服务 cronService = NewCronService(configConfig, dingTalkBotBiz) dingBotService = NewDingBotService(configConfig, dingTalkBotBiz) diff --git a/internal/tools/bbxt/upload.go b/internal/tools/bbxt/upload.go index c485c49..90277e3 100644 --- a/internal/tools/bbxt/upload.go +++ b/internal/tools/bbxt/upload.go @@ -42,7 +42,7 @@ func (u *Uploader) Run(report *ReportRes) (err error) { return fmt.Errorf("write to bytes failed: %v", err) } - picBytes, err := u.excel2picPy(report.Path, excelBytes.Bytes()) + picBytes, err := u.excel2picPy(report.Path, excelBytes.Bytes(), 2) if err != nil { return fmt.Errorf("excel2picPy failed: %v", err) } @@ -62,7 +62,7 @@ func (u *Uploader) Run(report *ReportRes) (err error) { // --header 'Content-Type: multipart/form-data; boundary=--------------------------952147881043913664015069' \ // --form 'file=@"C:\\Users\\Administrator\\Downloads\\销售同比分析2025-12-29 0-12点.xlsx"' \ // --form 'sheet_name="销售同比分析"' -func (u *Uploader) excel2picPy(templatePath string, excelBytes []byte) ([]byte, error) { +func (u *Uploader) excel2picPy(templatePath string, excelBytes []byte, scale int) ([]byte, error) { // 1. 获取 Sheet Name // 尝试从 excelBytes 解析，如果失败则使用默认值 "Sheet1" sheetName := "Sheet1" @@ -99,6 +99,14 @@ func (u *Uploader) excel2picPy(templatePath string, excelBytes []byte) ([]byte, return nil, fmt.Errorf("write field sheet_name failed: %v", err) } + // 添加 scale 字段 + if scale <= 0 { + scale = 2 + } + if err = writer.WriteField("scale", fmt.Sprintf("%d", scale)); err != nil { + return nil, fmt.Errorf("write field scale failed: %v", err) + } + if err = writer.Close(); err != nil { return nil, fmt.Errorf("close writer failed: %v", err) } diff --git a/tmpl/excel_temp/kshj_gt.xlsx b/tmpl/excel_temp/kshj_gt.xlsx index 2b27bfa..5aeeab5 100755 Binary files a/tmpl/excel_temp/kshj_gt.xlsx and b/tmpl/excel_temp/kshj_gt.xlsx differ diff --git a/tmpl/excel_temp/kshj_total.xlsx b/tmpl/excel_temp/kshj_total.xlsx index b29ae47..9074fe0 100755 Binary files a/tmpl/excel_temp/kshj_total.xlsx and b/tmpl/excel_temp/kshj_total.xlsx differ diff --git a/tmpl/excel_temp/recharge_statistics_ours_product.xlsx b/tmpl/excel_temp/recharge_statistics_ours_product.xlsx index 8d5b5a3..8294b55 100755 Binary files a/tmpl/excel_temp/recharge_statistics_ours_product.xlsx and b/tmpl/excel_temp/recharge_statistics_ours_product.xlsx differ