ai-agent/workflow/service/session/session_service.go

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
}