diff --git a/cmd/server/main.go b/cmd/server/main.go index e0c9ab3..d7da1db 100644 --- a/cmd/server/main.go +++ b/cmd/server/main.go @@ -10,7 +10,7 @@ import ( ) func main() { - configPath := flag.String("config", "./config/config_test.yaml", "Path to configuration file") + configPath := flag.String("config", "./config/config.yaml", "Path to configuration file") onBot := flag.String("bot", "", "bot start") cron := flag.String("cron", "", "close") runJob := flag.String("runJob", "", "run single job and exit") diff --git a/config/config.yaml b/config/config.yaml index 560758b..c57e203 100644 --- a/config/config.yaml +++ b/config/config.yaml @@ -4,23 +4,29 @@ server: host: "0.0.0.0" ollama: - base_url: "http://172.17.0.1:11434" - # model: "qwen3:8b" - # generate_model: "qwen3:8b" - # mapping_model: "qwen3:8b" - model: "qwen3-coder:480b-cloud" - generate_model: "qwen3-coder:480b-cloud" - mapping_model: "deepseek-v3.2:cloud" + base_url: "http://192.168.6.115:11434" + model: "qwen3:8b" + generate_model: "qwen3:8b" + mapping_model: "qwen3:8b" +# model: "qwen3-coder:480b-cloud" +# generate_model: "qwen3-coder:480b-cloud" +# mapping_model: "deepseek-v3.2:cloud" vl_model: "qwen2.5vl:3b" timeout: "120s" level: "info" format: "json" vllm: - base_url: "http://172.17.0.1:8001/v1" - vl_model: "qwen2.5-vl-3b-awq" - timeout: "120s" - level: "info" + vl_model: + base_url: "http://192.168.6.115:8001/v1" + model: "qwen2.5-vl-3b-awq" + timeout: "120s" + level: "info" + text_model: + base_url: "http://192.168.6.115:8002/v1" + model: "qwen3-8b-fp8" + timeout: "120s" + level: "info" coze: base_url: "https://api.coze.cn" @@ -157,7 +163,7 @@ eino_tools: # == 通用工具 == # 表格转图片 excel2pic: - base_url: "http://excel2pic:8000/api/v1/convert" + base_url: "http://192.168.6.115:8010/api/v1/convert" dingtalk: api_key: "dingsbbntrkeiyazcfdg" diff --git a/config/config_env.yaml b/config/config_env.yaml index 9327cb8..6c106a1 100644 --- a/config/config_env.yaml +++ b/config/config_env.yaml @@ -17,10 +17,16 @@ ollama: format: "json" vllm: - base_url: "http://117.175.169.61:16001/v1" - vl_model: "qwen2.5-vl-3b-awq" - timeout: "120s" - level: "info" + vl_model: + base_url: "http://192.168.6.115:8001/v1" + model: "qwen2.5-vl-3b-awq" + timeout: "120s" + level: "info" + text_model: + base_url: "http://192.168.6.115:8002/v1" + model: "qwen3-8b-fp8" + timeout: "120s" + level: "info" coze: base_url: "https://api.coze.cn" @@ -115,6 +121,7 @@ tools: api_secret: "pat_eEN0BdLNDughEtABjJJRYTW71olvDU0qUbfQUeaPc2NnYWO8HeyNoui5aR9z0sSZ" zltxResellerAuthProductToManagerAndDefaultLossReason: base_url: "https://revcl.1688sup.com/api/admin/reseller/resellerAuthProduct/getManagerAndDefaultLossReason" + # eino tool 配置 eino_tools: # == 货易通 hyt == diff --git a/config/config_test.yaml b/config/config_test.yaml index 03b0bdf..dd4a63a 100644 --- a/config/config_test.yaml +++ b/config/config_test.yaml @@ -14,10 +14,16 @@ ollama: format: "json" vllm: - base_url: "http://host.docker.internal:8001/v1" - vl_model: "qwen2.5-vl-3b-awq" - timeout: "120s" - level: "info" + vl_model: + base_url: "http://192.168.6.115:8001/v1" + model: "qwen2.5-vl-3b-awq" + timeout: "120s" + level: "info" + text_model: + base_url: "http://192.168.6.115:8002/v1" + model: "qwen3-8b-fp8" + timeout: "120s" + level: "info" coze: base_url: "https://api.coze.cn" @@ -26,12 +32,11 @@ coze: lsxd: # 统一登录 - login_url: "http://api.test.user.1688sup.com/v1/login/phone" - phone: "OFJ8UpqOlI7+w3Qklf36ZA==" - password: "tEbFegH/DRRh6LutFb7o3g==" - code: "123456" - check_token_url: "http://api.test.user.1688sup.com/v1/user/welcome" - + login_url: "https://api.user.1688sup.com/v1/login/phone" + phone: "ORlviZN7N06W2+WKLe76xg==" + password: "V5Uh8C4bamEM6UQZh4TCeQ==" + code: "456789" + check_token_url: "https://api.user.1688sup.com/v1/user/welcome" sys: session_len: 6 @@ -151,7 +156,7 @@ eino_tools: # == 通用工具 == # 表格转图片 excel2pic: - base_url: "http://192.168.6.109:8010/api/v1/convert" + base_url: "http://192.168.6.115:8010/api/v1/convert" dingtalk: api_key: "dingsbbntrkeiyazcfdg" diff --git a/internal/biz/ding_talk_bot.go b/internal/biz/ding_talk_bot.go index 55ad22a..56d38c7 100644 --- a/internal/biz/ding_talk_bot.go +++ b/internal/biz/ding_talk_bot.go @@ -623,6 +623,38 @@ func (d *DingTalkBotBiz) SendReport(ctx context.Context, groupInfo *model.AiBotG return } +// SendReports 发送多个报告 +func (d *DingTalkBotBiz) SendReports(ctx context.Context, groupInfo *model.AiBotGroup, reports []*bbxt.ReportRes) (err error) { + if len(reports) == 0 { + return errors.New("report is empty") + } + + title := fmt.Sprintf("截止%s日报", time.Now().Format("1月2日15点")) + reportChan := make(chan string, len(reports)*2) + writeCount := 0 + for _, v := range reports { + if v == nil { + continue + } + reportChan <- fmt.Sprintf("**%s**", v.Title) + reportChan <- fmt.Sprintf("![图片](%s)", v.Url) + writeCount += 2 + } + close(reportChan) + if writeCount == 0 { + return errors.New("report is empty") + } + err = d.HandleStreamRes(ctx, &chatbot.BotCallbackDataModel{ + RobotCode: groupInfo.RobotCode, + ConversationType: constants.ConversationTypeGroup, + ConversationId: groupInfo.ConversationID, + Text: chatbot.BotCallbackDataTextModel{ + Content: title, + }, + }, reportChan) + return +} + func (d *DingTalkBotBiz) GetGroupInfo(ctx context.Context, groupId int) (group model.AiBotGroup, err error) { cond := builder.NewCond() diff --git a/internal/biz/do/handle.go b/internal/biz/do/handle.go index 0cee873..3d06ccc 100644 --- a/internal/biz/do/handle.go +++ b/internal/biz/do/handle.go @@ -37,6 +37,7 @@ import ( type Handle struct { Ollama *llm_service.OllamaService + Vllm *llm_service.VllmService toolManager *tools.Manager conf *config.Config sessionImpl *impl.SessionImpl @@ -48,6 +49,7 @@ type Handle struct { func NewHandle( Ollama *llm_service.OllamaService, + Vllm *llm_service.VllmService, toolManager *tools.Manager, conf *config.Config, sessionImpl *impl.SessionImpl, @@ -58,6 +60,7 @@ func NewHandle( ) *Handle { return &Handle{ Ollama: Ollama, + Vllm: Vllm, toolManager: toolManager, conf: conf, sessionImpl: sessionImpl, @@ -73,7 +76,8 @@ func (r *Handle) Recognize(ctx context.Context, rec *entitys.Recognize, promptPr prompt, err := promptProcessor.CreatePrompt(ctx, rec) //意图识别 - recognizeMsg, err := r.Ollama.IntentRecognize(ctx, &entitys.ToolSelect{ + // recognizeMsg, err := r.Ollama.IntentRecognize(ctx, &entitys.ToolSelect{ + recognizeMsg, err := r.Vllm.IntentRecognize(ctx, &entitys.ToolSelect{ Prompt: prompt, Tools: rec.Tasks, }) @@ -819,7 +823,7 @@ func handleCozeWorkflowEvents(ctx context.Context, resp coze.Stream[coze.Workflo case coze.WorkflowEventTypeMessage: entitys.ResStream(ch, index, event.Message.Content) case coze.WorkflowEventTypeError: - entitys.ResError(ch, index, fmt.Sprintf("工作流执行错误: %s", event.Error)) + entitys.ResError(ch, index, fmt.Sprintf("工作流执行错误: %v", event.Error)) case coze.WorkflowEventTypeDone: entitys.ResEnd(ch, index, "工作流执行完成") case coze.WorkflowEventTypeInterrupt: diff --git a/internal/biz/do/macro.go b/internal/biz/do/macro.go index fe1ffc5..0d99485 100644 --- a/internal/biz/do/macro.go +++ b/internal/biz/do/macro.go @@ -26,6 +26,7 @@ import ( type Macro struct { botGroupImpl *impl.BotGroupImpl + botGroupConfigImpl *impl.BotGroupConfigImpl reportDailyCacheImpl *impl.ReportDailyCacheImpl config *config.Config rdb *utils.Rdb @@ -36,12 +37,14 @@ func NewMacro( reportDailyCacheImpl *impl.ReportDailyCacheImpl, config *config.Config, rdb *utils.Rdb, + botGroupConfigImpl *impl.BotGroupConfigImpl, ) *Macro { return &Macro{ botGroupImpl: botGroupImpl, reportDailyCacheImpl: reportDailyCacheImpl, config: config, rdb: rdb, + botGroupConfigImpl: botGroupConfigImpl, } } @@ -193,7 +196,7 @@ func (m *Macro) ProductModify(ctx context.Context, content string, groupConfig * groupConfig.ProductName = itemInfo cond := builder.NewCond() cond = cond.And(builder.Eq{"config_id": groupConfig.ConfigID}) - err = m.botGroupImpl.UpdateByCond(&cond, groupConfig) + err = m.botGroupConfigImpl.UpdateByCond(&cond, groupConfig) if err != nil { err = fmt.Errorf("修改失败:%v", err) return diff --git a/internal/biz/do/macro_test.go b/internal/biz/do/macro_test.go new file mode 100644 index 0000000..80445be --- /dev/null +++ b/internal/biz/do/macro_test.go @@ -0,0 +1,32 @@ +package do + +import ( + "ai_scheduler/internal/config" + "ai_scheduler/internal/data/impl" + "ai_scheduler/internal/data/model" + + "ai_scheduler/utils" + "context" + "testing" +) + +func Test_report(t *testing.T) { + con := "[利润同比报表]商品修改：官方–优酷周卡,官方–优酷月卡,官方–优酷季卡,官方–优酷年卡,,官方–爱奇艺-月卡,官方–爱奇艺-季卡,官方–爱奇艺-年卡,官方–芒果-PC周卡,官方–芒果-PC月卡,官方–芒果-PC季卡,官方–QQ音乐-绿钻月卡,官方–饿了么超级会员月卡,官方–网易云黑胶vip月卡,官方–喜马拉雅巅峰会员月卡，剪映会员7天卡，剪映会员月卡，剪映会员年卡，剪映SVIP会员7天卡，剪映SVIP会员月卡，剪映SVIP会员年卡" + run() + + chatId, err, i := ma.ProductModify(context.Background(), con, &model.AiBotGroupConfig{ConfigID: 1, ToolList: "8,9,10,11,12,13,16"}) + t.Log(chatId, err, i) +} + +var ma *Macro + +func run() { + configConfig, _ := config.LoadConfigWithTest() + db, _ := utils.NewGormDb(configConfig) + + rdb := utils.NewRdb(configConfig) + reportDailyCacheImpl := impl.NewReportDailyCacheImpl(db) + botGroupImpl := impl.NewBotGroupImpl(db) + botGroupConfigImpl := impl.NewBotGroupConfigImpl(db) + ma = NewMacro(botGroupImpl, reportDailyCacheImpl, configConfig, rdb, botGroupConfigImpl) +} diff --git a/internal/biz/group_config.go b/internal/biz/group_config.go index d478b25..22d1795 100644 --- a/internal/biz/group_config.go +++ b/internal/biz/group_config.go @@ -112,14 +112,12 @@ func (g *GroupConfigBiz) GetReportLists(ctx context.Context, groupConfig *model. if err != nil { return } - //追加电商充值系统统计 - 返回统一使用[]*bbxt.ReportRes rechargeReports, err := g.rechargeDailyReport(ctx, time.Now(), nil, g.ossClient) if err != nil || len(rechargeReports) == 0 { return } - - reports = append(reports, rechargeReports...) + reports = append(rechargeReports, reports...) return } @@ -189,7 +187,7 @@ func (g *GroupConfigBiz) handleReport(ctx context.Context, rec *entitys.Recogniz if _err != nil { return _err } - reports = append(reports, repo...) + reports = append(reports, repo) case "report_sales_analysis": product := strings.Split(groupConfig.ProductName, ",") repo, _err := rep.GetStatisOfficialProductSum(t, product) @@ -214,8 +212,9 @@ func (g *GroupConfigBiz) handleReport(ctx context.Context, rec *entitys.Recogniz if _err != nil { return _err } - reports = append(reports, repo...) reports = append(reports, rechargeReport...) + reports = append(reports, repo...) + case "report_daily_recharge": product := strings.Split(groupConfig.ProductName, ",") repo, _err := g.rechargeDailyReport(ctx, t, product, nil) diff --git a/internal/biz/handle/dingtalk/send_card.go b/internal/biz/handle/dingtalk/send_card.go index 4660f33..71de592 100644 --- a/internal/biz/handle/dingtalk/send_card.go +++ b/internal/biz/handle/dingtalk/send_card.go @@ -20,7 +20,7 @@ import ( ) const DefaultInterval = 100 * time.Millisecond -const HeardBeatX = 1000 +const HeardBeatX = 100 type SendCardClient struct { Auth *Auth @@ -115,6 +115,7 @@ func (s *SendCardClient) NewCard(ctx context.Context, cardSend *CardSend) error s.processContentChannel(ctx, cardSend, cardInstanceId.String(), client) }() wg.Wait() + log.Info("处理通道结束") } return nil @@ -163,7 +164,7 @@ func (s *SendCardClient) processContentChannel(ctx context.Context, cardSend *Ca var ( contentBuilder strings.Builder - lastUpdate time.Time + lastUpdate = time.Now() ) for { @@ -173,6 +174,7 @@ func (s *SendCardClient) processContentChannel(ctx context.Context, cardSend *Ca // 通道关闭，发送最终内容 if contentBuilder.Len() > 0 { if err := s.updateCardContent(ctx, cardSend, cardInstanceId, contentBuilder.String(), client); err != nil { + log.Info("contentBuilder.Len()修改失败1") s.logger.Errorf("更新卡片失败1:%s", err.Error()) } } @@ -181,6 +183,7 @@ func (s *SendCardClient) processContentChannel(ctx context.Context, cardSend *Ca contentBuilder.WriteString(content) if contentBuilder.Len() > 0 { if err := s.updateCardContent(ctx, cardSend, cardInstanceId, contentBuilder.String(), client); err != nil { + log.Info("contentBuilder.Len()修改失败2") s.logger.Errorf("更新卡片失败2：%s", err.Error()) } } @@ -188,10 +191,12 @@ func (s *SendCardClient) processContentChannel(ctx context.Context, cardSend *Ca case <-heartbeatTicker.C: if time.Now().Unix()-lastUpdate.Unix() >= HeardBeatX { + log.Infof("心跳超时,当前时间：%d,最后时间：%d", time.Now().Unix(), lastUpdate.Unix()) return } case <-ctx.Done(): + log.Info("send_card上下文失效") s.logger.Info("context canceled, stop channel processing") return } diff --git a/internal/biz/llm_service/vllm.go b/internal/biz/llm_service/vllm.go new file mode 100644 index 0000000..c054f27 --- /dev/null +++ b/internal/biz/llm_service/vllm.go @@ -0,0 +1,153 @@ +package llm_service + +import ( + "ai_scheduler/internal/config" + "ai_scheduler/internal/entitys" + "ai_scheduler/internal/pkg" + "ai_scheduler/internal/pkg/utils_vllm" + "context" + "encoding/base64" + "errors" + "fmt" + "strings" + + "github.com/cloudwego/eino/schema" + "github.com/ollama/ollama/api" +) + +type VllmService struct { + client *utils_vllm.Client + config *config.Config +} + +func NewVllmService( + client *utils_vllm.Client, + config *config.Config, +) *VllmService { + return &VllmService{ + client: client, + config: config, + } +} + +func (s *VllmService) IntentRecognize(ctx context.Context, req *entitys.ToolSelect) (msg string, err error) { + msgs := s.convertMessages(req.Prompt) + tools := s.convertTools(req.Tools) + + resp, err := s.client.ToolSelect(ctx, msgs, tools) + if err != nil { + return + } + + if resp.Content == "" { + if len(resp.ToolCalls) > 0 { + call := resp.ToolCalls[0] + var matchFromTools = &entitys.Match{ + Confidence: 1, + Index: call.Function.Name, + Parameters: call.Function.Arguments, + IsMatch: true, + } + msg = pkg.JsonStringIgonErr(matchFromTools) + } else { + err = errors.New("不太明白你想表达的意思呢，可以在仔细描述一下您所需要的内容吗，感谢感谢") + return + } + } else { + msg = resp.Content + } + return +} + +func (s *VllmService) convertMessages(prompts []api.Message) []*schema.Message { + msgs := make([]*schema.Message, 0, len(prompts)) + for _, p := range prompts { + msg := &schema.Message{ + Role: schema.RoleType(p.Role), + Content: p.Content, + } + + // 这里实际应该不会走进来 + if len(p.Images) > 0 { + parts := []schema.MessageInputPart{ + {Type: schema.ChatMessagePartTypeText, Text: p.Content}, + } + for _, imgData := range p.Images { + b64 := base64.StdEncoding.EncodeToString(imgData) + mimeType := "image/jpeg" + parts = append(parts, schema.MessageInputPart{ + Type: schema.ChatMessagePartTypeImageURL, + Image: &schema.MessageInputImage{ + MessagePartCommon: schema.MessagePartCommon{ + MIMEType: mimeType, + Base64Data: &b64, + }, + }, + }) + } + msg.UserInputMultiContent = parts + } + msgs = append(msgs, msg) + } + return msgs +} + +func (s *VllmService) convertTools(tasks []entitys.RegistrationTask) []*schema.ToolInfo { + tools := make([]*schema.ToolInfo, 0, len(tasks)) + for _, task := range tasks { + params := make(map[string]*schema.ParameterInfo) + for k, v := range task.TaskConfigDetail.Param.Properties { + dt := schema.String + + // Handle v.Type dynamically to support both string and []string (compiler suggests []string) + // Using fmt.Sprint handles both cases safely without knowing exact type structure + typeStr := fmt.Sprintf("%v", v.Type) + typeStr = strings.Trim(typeStr, "[]") // normalize "[string]" -> "string" + + switch typeStr { + case "string": + dt = schema.String + case "integer", "int": + dt = schema.Integer + case "number", "float": + dt = schema.Number + case "boolean", "bool": + dt = schema.Boolean + case "object": + dt = schema.Object + case "array": + dt = schema.Array + } + + required := false + for _, r := range task.TaskConfigDetail.Param.Required { + if r == k { + required = true + break + } + } + + desc := v.Description + if len(v.Enum) > 0 { + var enumStrs []string + for _, e := range v.Enum { + enumStrs = append(enumStrs, fmt.Sprintf("%v", e)) + } + desc += " Enum: " + strings.Join(enumStrs, ", ") + } + + params[k] = &schema.ParameterInfo{ + Type: dt, + Desc: desc, + Required: required, + } + } + + tools = append(tools, &schema.ToolInfo{ + Name: task.Name, + Desc: task.Desc, + ParamsOneOf: schema.NewParamsOneOfByParams(params), + }) + } + return tools +} diff --git a/internal/biz/provider_set.go b/internal/biz/provider_set.go index bf1f48f..f3b7003 100644 --- a/internal/biz/provider_set.go +++ b/internal/biz/provider_set.go @@ -14,6 +14,7 @@ var ProviderSetBiz = wire.NewSet( NewChatHistoryBiz, //llm_service.NewLangChainGenerate, llm_service.NewOllamaGenerate, + llm_service.NewVllmService, //handle.NewHandle, do.NewDo, do.NewHandle, diff --git a/internal/config/config.go b/internal/config/config.go index c8b5c70..b3e234a 100644 --- a/internal/config/config.go +++ b/internal/config/config.go @@ -154,8 +154,13 @@ type OllamaConfig struct { } type VllmConfig struct { + VLModel VllmModel `mapstructure:"vl_model"` + TextModel VllmModel `mapstructure:"text_model"` +} + +type VllmModel struct { BaseURL string `mapstructure:"base_url"` - VlModel string `mapstructure:"vl_model"` + Model string `mapstructure:"model"` Timeout time.Duration `mapstructure:"timeout"` Level string `mapstructure:"level"` } @@ -366,7 +371,7 @@ func LoadConfigWithEnv() (*Config, error) { if err != nil { return nil, err } - viper.SetConfigFile(modularDir + "/config/config_test.yaml") + viper.SetConfigFile(modularDir + "/config/config.yaml") viper.SetConfigType("yaml") // 读取配置文件 if err := viper.ReadInConfig(); err != nil { diff --git a/internal/pkg/utils_vllm/client.go b/internal/pkg/utils_vllm/client.go index c8c4aec..6926b21 100644 --- a/internal/pkg/utils_vllm/client.go +++ b/internal/pkg/utils_vllm/client.go @@ -7,33 +7,63 @@ import ( "encoding/base64" "github.com/cloudwego/eino-ext/components/model/openai" + "github.com/cloudwego/eino/components/model" "github.com/cloudwego/eino/schema" ) type Client struct { - model *openai.ChatModel - config *config.Config + vlModel *openai.ChatModel + generateModel *openai.ChatModel + config *config.Config } func NewClient(config *config.Config) (*Client, func(), error) { - m, err := openai.NewChatModel(context.Background(), &openai.ChatModelConfig{ - BaseURL: config.Vllm.BaseURL, - Model: config.Vllm.VlModel, - Timeout: config.Vllm.Timeout, + // 初始化视觉模型 + vl, err := openai.NewChatModel(context.Background(), &openai.ChatModelConfig{ + BaseURL: config.Vllm.VLModel.BaseURL, + Model: config.Vllm.VLModel.Model, + Timeout: config.Vllm.VLModel.Timeout, }) if err != nil { return nil, nil, err } - c := &Client{model: m, config: config} + + // 初始化生成模型 + gen, err := openai.NewChatModel(context.Background(), &openai.ChatModelConfig{ + BaseURL: config.Vllm.TextModel.BaseURL, + Model: config.Vllm.TextModel.Model, + Timeout: config.Vllm.TextModel.Timeout, + ExtraFields: map[string]any{ + "chat_template_kwargs": map[string]any{ + "enable_thinking": false, + }, + }, + }) + if err != nil { + return nil, nil, err + } + + c := &Client{ + vlModel: vl, + generateModel: gen, + config: config, + } cleanup := func() {} return c, cleanup, nil } func (c *Client) Chat(ctx context.Context, msgs []*schema.Message) (*schema.Message, error) { - return c.model.Generate(ctx, msgs) + // 默认聊天使用生成模型 + return c.generateModel.Generate(ctx, msgs) +} + +func (c *Client) ToolSelect(ctx context.Context, msgs []*schema.Message, tools []*schema.ToolInfo) (*schema.Message, error) { + // 工具选择使用生成模型 + return c.generateModel.Generate(ctx, msgs, model.WithTools(tools)) } func (c *Client) RecognizeWithImg(ctx context.Context, systemPrompt, userPrompt string, imgURLs []string) (*schema.Message, error) { + // 图片识别使用视觉模型 in := []*schema.Message{ { Role: schema.System, @@ -58,11 +88,12 @@ func (c *Client) RecognizeWithImg(ctx context.Context, systemPrompt, userPrompt } in[1].UserInputMultiContent = parts - return c.model.Generate(ctx, in) + return c.vlModel.Generate(ctx, in) } // 识别图片by二进制文件 func (c *Client) RecognizeWithImgBytes(ctx context.Context, systemPrompt, userPrompt string, imgBytes []byte, imgType string) (*schema.Message, error) { + // 图片识别使用视觉模型 in := []*schema.Message{ { Role: schema.System, @@ -82,9 +113,10 @@ func (c *Client) RecognizeWithImgBytes(ctx context.Context, systemPrompt, userPr MIMEType: imgType, Base64Data: util.AnyToPoint(base64.StdEncoding.EncodeToString(imgBytes)), }, + Detail: schema.ImageURLDetailHigh, }, }) in[1].UserInputMultiContent = parts - return c.model.Generate(ctx, in) + return c.vlModel.Generate(ctx, in) } diff --git a/internal/services/cron.go b/internal/services/cron.go index e3c5604..b1f1023 100644 --- a/internal/services/cron.go +++ b/internal/services/cron.go @@ -50,12 +50,18 @@ func (d *CronService) CronReportSendDingTalk(ctx context.Context) error { return err } - for _, report := range reports { - err = d.dingTalkBotBiz.SendReport(ctx, &groupInfo, report) - if err != nil { - log.Error(err) - continue - } + // for _, report := range reports { + // err = d.dingTalkBotBiz.SendReport(ctx, &groupInfo, report) + // if err != nil { + // log.Error(err) + // continue + // } + // } + + err = d.dingTalkBotBiz.SendReports(ctx, &groupInfo, reports) + if err != nil { + log.Error(err) + return err } return nil } diff --git a/internal/services/dtalk_bot.go b/internal/services/dtalk_bot.go index 7245f18..39f561c 100644 --- a/internal/services/dtalk_bot.go +++ b/internal/services/dtalk_bot.go @@ -69,9 +69,15 @@ func (d *DingBotService) runBackgroundTasks(ctx context.Context, data *chatbot.B g.Go(func() error { defer func() { // 确保通道最终关闭 + log.Println("流式处理协程关闭") + close(resChan) }() - return d.dingTalkBotBiz.HandleStreamRes(ctx, data, resChan) + err := d.dingTalkBotBiz.HandleStreamRes(ctx, data, resChan) + if err != nil { + log.Println("流式回复产生错误，错误：", err.Error()) + } + return err }) // 2. 业务处理协程（负责关闭requireData.Ch） diff --git a/internal/services/dtalk_bot_test.go b/internal/services/dtalk_bot_test.go index 8df3265..1f7e207 100644 --- a/internal/services/dtalk_bot_test.go +++ b/internal/services/dtalk_bot_test.go @@ -123,7 +123,9 @@ func Run() { group := qywx.NewGroup(botGroupQywxImpl, qywxAuth) other := qywx.NewOther(qywxAuth) qywxAppBiz := biz.NewQywxAppBiz(configConfig, botGroupQywxImpl, group, other) - groupConfigBiz := biz.NewGroupConfigBiz(toolRegis, utils_ossClient, botGroupConfigImpl, registry, configConfig, reportDailyCacheImpl, rdb, macro, manager, handle) + reportDailyCacheImpl := impl.NewReportDailyCacheImpl(db) + macro := do.NewMacro(botGroupImpl, reportDailyCacheImpl, configConfig, rdb) + groupConfigBiz := biz.NewGroupConfigBiz(toolRegis, utils_ossClient, botGroupConfigImpl, registry, configConfig, impl.NewReportDailyCacheImpl(db), rdb, macro, manager, handle) dingTalkBotBiz := biz.NewDingTalkBotBiz(doDo, handle, botConfigImpl, botGroupImpl, user, botChatHisImpl, reportDailyCacheImpl, manager, configConfig, sendCardClient, groupConfigBiz, macro) // 初始化钉钉机器人服务 cronService = NewCronService(configConfig, dingTalkBotBiz, qywxAppBiz, groupConfigBiz) diff --git a/internal/tools/bbxt/bbxt.go b/internal/tools/bbxt/bbxt.go index 7a03cce..be068bc 100644 --- a/internal/tools/bbxt/bbxt.go +++ b/internal/tools/bbxt/bbxt.go @@ -13,6 +13,8 @@ import ( "sort" "time" + + "github.com/shopspring/decimal" ) const ( @@ -37,6 +39,8 @@ var ResellerBlackListProduct = []string{ "蓝星严选连续包月", "通钱-2025年12月", "彦浩同行", + "运营部测试专用", + "彦浩直客商户", } type BbxtTools struct { @@ -91,9 +95,7 @@ func (b *BbxtTools) DailyReport( if err != nil { return } - reports = append(reports, productLossReport...) - reports = append(reports, profitRankingSum, statisOfficialProductSum, statisOfficialProductSumDecline) - + reports = append(reports, profitRankingSum, statisOfficialProductSum, statisOfficialProductSumDecline, productLossReport) if ossClient != nil { uploader := NewUploader(ossClient, b.config) for _, report := range reports { @@ -135,7 +137,9 @@ func (b *BbxtTools) ResellerLossSort(ctx context.Context, now time.Time) ([]*Res reseller := resellerMap[info.ResellerId] // 累加经销商总亏损 - reseller.Total += info.Loss + num1 := decimal.NewFromFloat(reseller.Total) + num2 := decimal.NewFromFloat(info.Loss) + reseller.Total, _ = num1.Add(num2).Round(2).Float64() // 检查产品是否已存在 if _, ok := reseller.ProductLoss[info.OursProductId]; !ok { @@ -209,61 +213,66 @@ func (b *BbxtTools) ResellerLossToMapResellerLossSumProductRelation(totalDetail } // StatisOursProductLossSum 负利润分析 -func (b *BbxtTools) StatisOursProductLossSum(ctx context.Context, now time.Time, initFunc LossSumInitFunc) (report []*ReportRes, err error) { +func (b *BbxtTools) StatisOursProductLossSum(ctx context.Context, now time.Time, initFunc LossSumInitFunc) (report *ReportRes, err error) { resellers, err := b.ResellerLossSort(ctx, now) var ( - total [][]string - gt []*ResellerLoss + total [][]string + //gt []*ResellerLoss totalDetail []*ResellerLoss - totalSum float64 - totalSum500 float64 + totalSum = decimal.NewFromFloat(0) + //totalSum500 float64 ) // 构建分组 for _, v := range resellers { - if v.Total <= -100 && !slices.Contains(ResellerBlackListProduct, v.ResellerName) { + if slices.Contains(ResellerBlackListProduct, v.ResellerName) { + continue + } + if v.Total <= -100 { total = append(total, []string{ fmt.Sprintf("%s", v.ResellerName), fmt.Sprintf("%.2f", v.Total), }) - totalSum += v.Total + totalDetail = append(totalDetail, v) } - if v.Total <= -500 && !slices.Contains(ResellerBlackListProduct, v.ResellerName) { - gt = append(gt, v) - totalSum500 += v.Total - } + num := decimal.NewFromFloat(v.Total) + totalSum = totalSum.Add(num).Round(2) + //if v.Total <= -500 && !slices.Contains(ResellerBlackListProduct, v.ResellerName) { + // gt = append(gt, v) + // totalSum500 += v.Total + //} } - report = make([]*ReportRes, 3) + //report = make([]*ReportRes, 3) timeCh := now.Format("1月2日15点") //总量生成excel - if len(total) > 0 { - filePath := b.cacheDir + "/kshj_total" + fmt.Sprintf("%d%d", time.Now().Unix(), rand.Intn(1000)) + ".xlsx" - err = b.SimpleFillExcelWithTitle(b.excelTempDir+"/"+"kshj_total.xlsx", filePath, total, "") - if err != nil { - return - } - report[0] = &ReportRes{ - ReportName: "分销商负利润统计", - Title: "截至" + timeCh + "利润累计亏损" + fmt.Sprintf("%.2f", totalSum), - Path: filePath, - Data: total, - } - } - - if len(gt) > 0 { - filePath := b.cacheDir + "/kshj_gt" + fmt.Sprintf("%d%d", time.Now().Unix(), rand.Intn(1000)) + ".xlsx" - title := "截至" + timeCh + "亏损500以上的分销商和产品" - err = b.resellerDetailFillExcelV2(b.excelTempDir+"/"+"kshj_gt.xlsx", filePath, gt, title) - if err != nil { - return - } - report[1] = &ReportRes{ - ReportName: "负利润分析(亏损500以上)", - Title: "截至" + timeCh + "亏损500以上利润累计亏损" + fmt.Sprintf("%.2f", totalSum500), - Path: filePath, - Data: total, - } - } + //if len(total) > 0 { + // filePath := b.cacheDir + "/kshj_total" + fmt.Sprintf("%d%d", time.Now().Unix(), rand.Intn(1000)) + ".xlsx" + // err = b.SimpleFillExcelWithTitle(b.excelTempDir+"/"+"kshj_total.xlsx", filePath, total, "") + // if err != nil { + // return + // } + // report[0] = &ReportRes{ + // ReportName: "分销商负利润统计", + // Title: "截至" + timeCh + "利润累计亏损" + fmt.Sprintf("%.2f", totalSum), + // Path: filePath, + // Data: total, + // } + //} + // + //if len(gt) > 0 { + // filePath := b.cacheDir + "/kshj_gt" + fmt.Sprintf("%d%d", time.Now().Unix(), rand.Intn(1000)) + ".xlsx" + // title := "截至" + timeCh + "亏损500以上的分销商和产品" + // err = b.resellerDetailFillExcelV2(b.excelTempDir+"/"+"kshj_gt.xlsx", filePath, gt, title) + // if err != nil { + // return + // } + // report[1] = &ReportRes{ + // ReportName: "负利润分析(亏损500以上)", + // Title: "截至" + timeCh + "亏损500以上利润累计亏损" + fmt.Sprintf("%.2f", totalSum500), + // Path: filePath, + // Data: total, + // } + //} if len(totalDetail) > 0 { err = initFunc(ctx, now, totalDetail, b) @@ -271,14 +280,15 @@ func (b *BbxtTools) StatisOursProductLossSum(ctx context.Context, now time.Time, return } filePath := b.cacheDir + "/kshj_total_ana" + fmt.Sprintf("%d%d", time.Now().Unix(), rand.Intn(1000)) + ".xlsx" - title := "截至" + timeCh + "亏损100以上的分销商&产品负利润原因" + totalSumFloat64, _ := totalSum.Float64() + title := "截至" + timeCh + "利润累计亏损" + fmt.Sprintf("%.2f", totalSumFloat64) + ",亏损100以上利润原因如下" err = b.resellerDetailFillExcelAna(b.excelTempDir+"/"+"kshj_total_ana.xlsx", filePath, totalDetail, title) if err != nil { return } - report[2] = &ReportRes{ + report = &ReportRes{ ReportName: "负利润分析(亏损100以上)", - Title: "截至" + timeCh + "亏损100以上利润原因", + Title: title, Path: filePath, Data: total, } diff --git a/internal/tools/bbxt/bbxt_test.go b/internal/tools/bbxt/bbxt_test.go index 3d0f623..b2e58b1 100644 --- a/internal/tools/bbxt/bbxt_test.go +++ b/internal/tools/bbxt/bbxt_test.go @@ -106,7 +106,7 @@ var ( ) func run() { - configConfig, _ = config.LoadConfigWithTest() + configConfig, _ = config.LoadConfigWithEnv() // 初始化数据库连接 db, _ := utils.NewGormDb(configConfig) reportDailyCacheImpl = impl.NewReportDailyCacheImpl(db) diff --git a/internal/tools/bbxt/excel.go b/internal/tools/bbxt/excel.go index 91915dc..c10906d 100644 --- a/internal/tools/bbxt/excel.go +++ b/internal/tools/bbxt/excel.go @@ -343,14 +343,17 @@ func (b *BbxtTools) resellerDetailFillExcelAna(templatePath, outputPath string, styleD3, err := f.GetCellStyle(sheet, fmt.Sprintf("D%d", tplRowData)) if err != nil { - styleC3 = 0 + styleD3 = 0 } styleE3, err := f.GetCellStyle(sheet, fmt.Sprintf("E%d", tplRowData)) if err != nil { - styleC3 = 0 + styleE3 = 0 + } + styleF3, err := f.GetCellStyle(sheet, fmt.Sprintf("F%d", tplRowData)) + if err != nil { + styleF3 = 0 } - rowHeightData, err := f.GetRowHeight(sheet, tplRowData) if err != nil { rowHeightData = 20 @@ -372,11 +375,15 @@ func (b *BbxtTools) resellerDetailFillExcelAna(templatePath, outputPath string, } styleTotalD, err := f.GetCellStyle(sheet, fmt.Sprintf("D%d", tplRowTotal)) if err != nil { - styleTotalC = 0 + styleTotalD = 0 } styleTotalE, err := f.GetCellStyle(sheet, fmt.Sprintf("E%d", tplRowTotal)) if err != nil { - styleTotalC = 0 + styleTotalE = 0 + } + styleTotalF, err := f.GetCellStyle(sheet, fmt.Sprintf("F%d", tplRowTotal)) + if err != nil { + styleTotalF = 0 } rowHeightTotal, err := f.GetRowHeight(sheet, tplRowTotal) if err != nil { @@ -408,8 +415,9 @@ func (b *BbxtTools) resellerDetailFillExcelAna(templatePath, outputPath string, f.SetCellValue(sheet, fmt.Sprintf("A%d", currentRow), reseller.ResellerName) f.SetCellValue(sheet, fmt.Sprintf("B%d", currentRow), p.ProductName) f.SetCellValue(sheet, fmt.Sprintf("C%d", currentRow), p.Loss) - f.SetCellValue(sheet, fmt.Sprintf("D%d", currentRow), reseller.Manager) - f.SetCellValue(sheet, fmt.Sprintf("E%d", currentRow), p.LossReason) + f.SetCellValue(sheet, fmt.Sprintf("D%d", currentRow), reseller.Total) + f.SetCellValue(sheet, fmt.Sprintf("E%d", currentRow), reseller.Manager) + f.SetCellValue(sheet, fmt.Sprintf("F%d", currentRow), p.LossReason) // 设置样式 if styleA3 != 0 { f.SetCellStyle(sheet, fmt.Sprintf("A%d", currentRow), fmt.Sprintf("A%d", currentRow), styleA3) @@ -426,7 +434,9 @@ func (b *BbxtTools) resellerDetailFillExcelAna(templatePath, outputPath string, if styleE3 != 0 { f.SetCellStyle(sheet, fmt.Sprintf("E%d", currentRow), fmt.Sprintf("E%d", currentRow), styleE3) } - + if styleF3 != 0 { + f.SetCellStyle(sheet, fmt.Sprintf("F%d", currentRow), fmt.Sprintf("F%d", currentRow), styleF3) + } totalLoss += p.Loss currentRow++ } @@ -436,6 +446,7 @@ func (b *BbxtTools) resellerDetailFillExcelAna(templatePath, outputPath string, if endRow > startRow { f.MergeCell(sheet, fmt.Sprintf("A%d", startRow), fmt.Sprintf("A%d", endRow)) f.MergeCell(sheet, fmt.Sprintf("D%d", startRow), fmt.Sprintf("D%d", endRow)) + f.MergeCell(sheet, fmt.Sprintf("E%d", startRow), fmt.Sprintf("E%d", endRow)) } } @@ -447,7 +458,7 @@ func (b *BbxtTools) resellerDetailFillExcelAna(templatePath, outputPath string, f.SetCellValue(sheet, fmt.Sprintf("A%d", currentRow), "合计") // B列留空，C列填充总亏损 - f.SetCellValue(sheet, fmt.Sprintf("C%d", currentRow), totalLoss) + f.SetCellValue(sheet, fmt.Sprintf("D%d", currentRow), totalLoss) // 设置合计行样式 if styleTotalA != 0 { @@ -465,7 +476,9 @@ func (b *BbxtTools) resellerDetailFillExcelAna(templatePath, outputPath string, if styleTotalE != 0 { f.SetCellStyle(sheet, fmt.Sprintf("E%d", currentRow), fmt.Sprintf("E%d", currentRow), styleTotalE) } - + if styleTotalF != 0 { + f.SetCellStyle(sheet, fmt.Sprintf("F%d", currentRow), fmt.Sprintf("F%d", currentRow), styleTotalF) + } // 取消合并合计行的A、B列 // f.MergeCell(sheet, fmt.Sprintf("A%d", currentRow), fmt.Sprintf("B%d", currentRow)) diff --git a/tmpl/excel_temp/kshj_total_ana.xlsx b/tmpl/excel_temp/kshj_total_ana.xlsx index a2b8178..e4e7277 100644 Binary files a/tmpl/excel_temp/kshj_total_ana.xlsx and b/tmpl/excel_temp/kshj_total_ana.xlsx differ