package session import ( "ai-agent/gateway" "ai-agent/workflow/consts/flow" flowDao "ai-agent/workflow/dao/flow" sessionDao "ai-agent/workflow/dao/session" flowDto "ai-agent/workflow/model/dto/flow" sessionDto "ai-agent/workflow/model/dto/session" "ai-agent/workflow/model/entity" flowService "ai-agent/workflow/service/flow" "context" "fmt" "sort" "strings" "gitea.redpowerfuture.com/red-future/common/beans" "gitea.redpowerfuture.com/red-future/common/oss" "gitea.redpowerfuture.com/red-future/common/utils" "github.com/gogf/gf/v2/os/glog" "github.com/gogf/gf/v2/util/gconv" ) var SessionService = &sessionService{} type sessionService struct{} // 结果状态(与 workflow_session_result.status / VOSessionInfoResult.Status 一致) const ( resultStatusSuccess = 2 resultStatusFailed = 3 resultStatusCancel = 4 ) func (s *sessionService) List(ctx context.Context, req *sessionDto.ListSessionReq) (res *sessionDto.ListSessionRes, err error) { user, err := utils.GetUserInfo(ctx) if err != nil { return } var page *beans.Page if req.PageSize > 0 { page = &beans.Page{PageNum: req.PageNum, PageSize: req.PageSize} } list, total, err := sessionDao.SessionDao.List(ctx, user.UserName, page) if err != nil { return } res = &sessionDto.ListSessionRes{Total: total} for _, item := range list { res.List = append(res.List, &sessionDto.VOSession{ SessionId: item.SessionId, SessionName: item.SessionName, CreatedAt: item.CreatedAt, }) } return } func (s *sessionService) Delete(ctx context.Context, req *sessionDto.DeleteSessionReq) (err error) { _, err = sessionDao.SessionDao.Delete(ctx, req) return } func (s *sessionService) DeleteRecord(ctx context.Context, req *sessionDto.DeleteSessionRecordReq) (err error) { var chatIds, wfIds []int64 for _, item := range req.Ids { if item.Type == "chat" { chatIds = append(chatIds, item.Id) } else { wfIds = append(wfIds, item.Id) } } if len(chatIds) > 0 { if _, e := sessionDao.ExecChatDao.Delete(ctx, &sessionDto.DeleteExecChatReq{Id: chatIds}); e != nil { return e } } if len(wfIds) > 0 { if _, e := sessionDao.ExecWorkflowDao.Delete(ctx, &sessionDto.DeleteExecWorkflowReq{Id: wfIds}); e != nil { return e } } return } // Get 会话内结果:普通对话 + 工作流执行混排,按创建时间倒序,分页 func (s *sessionService) Get(ctx context.Context, req *sessionDto.GetSessionInfoReq) (res *sessionDto.GetSessionInfoRes, err error) { chatList, err := sessionDao.ExecChatDao.ListBySession(ctx, req.SessionId) if err != nil { return } wfList, err := sessionDao.ExecWorkflowDao.ListBySession(ctx, req.SessionId) if err != nil { return } wfResultList, err := sessionDao.ExecWorkflowResultDao.ListBySession(ctx, req.SessionId) if err != nil { return } prefix, _ := oss.GetFileAddressPrefix(ctx) // 工作流结果按 exec_id 分组,合并到对应执行记录的结果文件URL resultByExec := make(map[int64][]string) for _, wr := range wfResultList { if wr.ResultFileUrl != "" { resultByExec[wr.ExecId] = append(resultByExec[wr.ExecId], prefix+wr.ResultFileUrl) } } res = &sessionDto.GetSessionInfoRes{} for _, c := range chatList { c.ResultFileUrl = prefix + c.ResultFileUrl res.List = append(res.List, chatExecVO(c)) } for _, w := range wfList { res.List = append(res.List, workflowExecVO(w, strings.Join(resultByExec[w.Id], ","))) } sort.Slice(res.List, func(i, j int) bool { ci, cj := res.List[i].CreatedAt, res.List[j].CreatedAt if ci == nil { return false } if cj == nil { return true } return ci.After(cj) }) res.Total = len(res.List) if req.PageSize > 0 { start := int((req.PageNum - 1) * req.PageSize) if start < 0 { start = 0 } if start >= res.Total { res.List = nil return } end := start + int(req.PageSize) if end > res.Total { end = res.Total } res.List = res.List[start:end] } // 读取结果文件内容(仅 .txt)放入 ResultContent(仅当前页),供前端直接展示;路径仍保留在 ResultFileUrl for _, vo := range res.List { vo.ResultContent = readResultFileContent(ctx, vo.ResultFileUrl) } return } // readResultFileContent 读取结果 txt 文件内容(支持逗号分隔的多个 URL),仅 .txt 文件被读取,多个内容用换行连接 func readResultFileContent(ctx context.Context, fileUrl string) string { if fileUrl == "" { return "" } var parts []string for _, u := range strings.Split(fileUrl, ",") { u = strings.TrimSpace(u) if u == "" || !strings.HasSuffix(strings.ToLower(u), ".txt") { continue } fileBytes, err := gateway.GetFileBytesFromURL(ctx, u) if err != nil { glog.Warningf(ctx, "读取结果 txt 文件失败: %v", err) continue } parts = append(parts, string(fileBytes)) } return strings.Join(parts, "\n") } func chatExecVO(c *entity.ExecChat) *sessionDto.VOSessionInfoResult { status := resultStatusSuccess if c.ErrorMessage != "" { status = resultStatusFailed } return &sessionDto.VOSessionInfoResult{ Id: c.Id, Type: "chat", Status: status, RequestParams: map[string]any{"question": c.RequestParams.Question}, ResultFileUrl: c.ResultFileUrl, TotalTokens: c.TotalTokens, TotalFee: c.TotalFee, ErrorMsg: c.ErrorMessage, Error: c.Error, CreatedAt: c.CreatedAt, } } func workflowExecVO(w *entity.ExecWorkflow, resultFileUrl string) *sessionDto.VOSessionInfoResult { status := 1 // FlowExecutionStatus 是 *int8 别名,Code() 返回包级指针,直接 == 是地址比较恒为 false, // 需解引用按值比较,否则所有执行记录在前端都会误显示为"运行中" if w.Status != nil { if *w.Status == *flow.FlowExecutionStatusFailed.Code() { status = resultStatusFailed } else if *w.Status == *flow.FlowExecutionStatusSuccess.Code() { status = resultStatusSuccess } else if *w.Status == *flow.FlowExecutionStatusCancel.Code() { status = resultStatusCancel } } return &sessionDto.VOSessionInfoResult{ Id: w.Id, Type: "workflow", Status: status, FlowId: w.FlowId, RequestParams: gconv.Map(w.RequestParams), ResultFileUrl: resultFileUrl, TotalTokens: w.TotalTokens, TotalFee: w.TotalFee, ActualAmount: w.ActualAmount, ErrorMsg: w.ErrorMessage, Error: w.Error, CreatedAt: w.CreatedAt, } } // ResultList 工作流执行结果树:按创建人分页查询工作流结果记录(exec_workflow_result), // 返回按天分组的树结构(日期→流程→结果文件)。 // 只依赖结果表,不关联 exec_workflow 执行记录——即使执行记录被删除/清理,产出文件仍可查看。 // 分页单位为"天":每页返回 pageSize 个完整日期,同一天内的流程与文件不会被拆到不同页;pageSize 为 0 时返回全部。 func (s *sessionService) ResultList(ctx context.Context, req *sessionDto.ListWorkflowResultReq) (res *flowDto.ListFlowExecutionTreeRes, err error) { user, err := utils.GetUserInfo(ctx) if err != nil { return } dates, err := sessionDao.ExecWorkflowResultDao.ListDates(ctx, user.UserName, req.Page) if err != nil { return } res = &flowDto.ListFlowExecutionTreeRes{} res.ImgAddressPrefix, _ = oss.GetFileAddressPrefix(ctx) if len(dates) == 0 { return } // 仅查结果表:同一执行的多个结果按 exec_id 归为一个流程节点 results, err := sessionDao.ExecWorkflowResultDao.ListByDates(ctx, user.UserName, dates) if err != nil { return } // 按 exec_id 分组(ListByDates 按创建时间倒序,同执行内结果最后反转回正序,保持旧列表输出顺序) type resultGroup struct { ExecId int64 SessionId string FlowId int64 Date string Results []*entity.ExecWorkflowResult } groupByExec := make(map[int64]*resultGroup) var groups []*resultGroup flowIdSet := make(map[int64]struct{}) for _, r := range results { g := groupByExec[r.ExecId] if g == nil { g = &resultGroup{ExecId: r.ExecId, SessionId: r.SessionId, FlowId: r.FlowId} groupByExec[r.ExecId] = g groups = append(groups, g) if r.CreatedAt != nil { g.Date = r.CreatedAt.Format("Y-m-d") } } g.Results = append(g.Results, r) flowIdSet[r.FlowId] = struct{}{} } for _, g := range groups { for i, j := 0, len(g.Results)-1; i < j; i, j = i+1, j-1 { g.Results[i], g.Results[j] = g.Results[j], g.Results[i] } } flowNameMap := make(map[int64]string) for fid := range flowIdSet { if fu, e := flowDao.FlowUserDao.Get(ctx, &flowDto.GetFlowUserReq{Id: fid}); e == nil && fu != nil && fu.FlowName != "" { flowNameMap[fid] = fu.FlowName } } // 按日期分组构建树(flow 顺序即各组首次出现的倒序,与旧行为一致) flowsByDate := make(map[string][]flowDto.FlowNode) for _, g := range groups { flowName := flowNameMap[g.FlowId] if flowName == "" { flowName = "工作流" } var items []flowDto.OutputItem suffixCount := make(map[string]int) for _, rf := range g.Results { if rf.ResultFileUrl == "" { continue } content := rf.ResultFileUrl ext := flowService.GetFileTypeByPath(content) suffix := outputItemSuffix(ext) suffixCount[suffix]++ items = append(items, flowDto.OutputItem{ Content: content, Type: ext, Label: fmt.Sprintf("%s_%d", suffix, suffixCount[suffix]), }) } if len(items) == 0 { continue } flowsByDate[g.Date] = append(flowsByDate[g.Date], flowDto.FlowNode{ FlowName: flowName, Id: g.ExecId, SessionId: g.SessionId, Items: items, }) } for _, d := range dates { if fs := flowsByDate[d]; len(fs) > 0 { res.Tree = append(res.Tree, flowDto.DateNode{CreateDate: d, Flows: fs}) } } return } // outputItemSuffix 按文件类型映射结果项的中文标签前缀(与 flow 侧旧逻辑保持一致) func outputItemSuffix(ext string) string { switch ext { case "image": return "图片" case "video": return "视频" case "audio": return "音频" case "text": return "文案" case "html": return "HTML" default: return "内容" } } func (s *sessionService) ResultDelete(ctx context.Context, req *sessionDto.DeleteWorkflowResultReq) (err error) { _, err = sessionDao.ExecWorkflowResultDao.Delete(ctx, req) return }