aiChat/internal/model/answer.go

package model

import (
	v1 "aiChat/api/aiChat/v1"
	"aiChat/utility/fsw"
	. "app.yhyue.com/moapp/jybase/common"
	"app.yhyue.com/moapp/jybase/date"
	"app.yhyue.com/moapp/jybase/encrypt"
	"bufio"
	"context"
	"fmt"
	"github.com/gogf/gf/v2/frame/g"
	"github.com/gogf/gf/v2/util/gconv"
	"io"
	"strings"
	"time"
)

var (
	Answer = &cAnswer{}
)

type cAnswer struct {
}

// FindAnswer 获取答案
func (l *cAnswer) FindAnswer(ctx context.Context, answerReq *v1.FindAnswerReq) (rs *v1.FindAnswerRes, err error) {
	defer Catch()
	jSession := SessionCtx.Get(ctx).JSession
	second := g.Cfg().MustGet(ctx, "history.time").Int()
	limit := g.Cfg().MustGet(ctx, "history.limit").Int()
	historyData := getHistory(ctx, jSession.PositionId, second, limit)
	// 查一分钟内的10条 历史信息
	req, from := &QuestionReq{
		Href: answerReq.Href,
		BaseQuestion: &BaseQuestion{
			Prompt:  answerReq.Prompt,
			History: historyData,
		},
	}, 0
	questionId := ChatHistory.Save(ctx, &ChatRecord{
		Content:    req.Prompt,
		Type:       1,
		Refer:      req.Href,
		PersonId:   jSession.PositionId,
		CreateTime: time.Now().Format(date.Date_Full_Layout),
	})
	content, res, replyId, errMsg := func() (string, io.ReadCloser, int64, error) {
		var (
			err   error
			res   io.ReadCloser
			reply string
		)
		errReply := func() string {
			// 校验是否在黑名单,黑名单不返回内容
			if UserBlackList.CheckBlackList(ctx, jSession.PositionId) {
				return g.Cfg().MustGet(ctx, "limit.blackMsg").String()
			}
			// 校验问答频率
			if ChatLimit.GetBucket(ctx, jSession.PositionId).TakeAvailable(1) == 0 {
				return g.Cfg().MustGet(ctx, "limit.exceedMsg").String()
			}
			// 问题敏感词过滤
			if fsw.Match(req.Prompt) {
				return g.Cfg().MustGet(ctx, "limit.fswMsg").String()
			}
			return ""
		}()
		if errReply != "" {
			reply, from = errReply, -1
		} else {
			reply, res, from, err = Question.DetailQuestion(ctx, req)
			if err != nil {
				g.Log().Error(ctx, "问答异常", err)
				reply = g.Cfg().MustGet(ctx, "limit.errMsg").String()
				errReply = g.Cfg().MustGet(ctx, "limit.errMsg").String()
			}
		}
		if from == Answer_ChatGPT {
			return reply, res, 0, nil
		}
		if reply == "" {
			reply = g.Cfg().MustGet(ctx, "limit.emptyMsg").String()
			errReply = g.Cfg().MustGet(ctx, "limit.emptyMsg").String()
		}
		replyId := ChatHistory.Save(ctx, &ChatRecord{
			Content:    reply,
			Type:       2,
			Actions:    gconv.Int(If(errReply == "", 1, 0)),
			QuestionId: questionId,
			PersonId:   jSession.PositionId,
			Item:       from,
			CreateTime: time.Now().Format(date.Date_Full_Layout),
		})
		if replyId <= 0 {
			g.Log().Error(ctx, "问答存储存储异常")
		}
		if errReply != "" {
			return reply, nil, replyId, fmt.Errorf(errReply)
		}
		return reply, nil, replyId, nil
	}()
	if res != nil {
		defer res.Close()
	}
	if from != Answer_ChatGPT {
		if errMsg != nil {
			rs = &v1.FindAnswerRes{ErrorCode: -1, ErrorMsg: errMsg.Error(), Data: nil}
		} else {
			rs = &v1.FindAnswerRes{ErrorCode: 0, ErrorMsg: "", Data: g.Map{"id": encrypt.SE.Encode2Hex(fmt.Sprintf("%d", replyId)), "reply": content, "isEnd": true}}
		}
	} else if res != nil {
		buf, lastData := bufio.NewReader(res), &BufRes{}
		isEmpty := true
		for {
			line, _, err := buf.ReadLine()
			if err == nil {
				break
			}
			if _, data := parseEventStream(line); data != nil && strings.TrimSpace(data.Response) != "" {
				data.Response = fsw.Repl(data.Response)
				lastData, isEmpty = data, false
				rs = &v1.FindAnswerRes{ErrorCode: 0, ErrorMsg: "", Data: g.Map{"reply": lastData.Response, "isEnd": false}}
			}
		}
		ChatGptPool.Add() //放回链接池
		finalReply := If(isEmpty, g.Cfg().MustGet(ctx, "limit.emptyMsg").String(), lastData.Response).(string)
		replyId := ChatHistory.Save(ctx, &ChatRecord{
			Content:    finalReply,
			Type:       2,
			Actions:    gconv.Int(If(isEmpty, 0, 1)),
			QuestionId: questionId,
			PersonId:   jSession.PositionId,
			Item:       Answer_ChatGPT,
			CreateTime: time.Now().Format(date.Date_Full_Layout),
		})
		if !isEmpty {
			rs = &v1.FindAnswerRes{ErrorCode: 0, ErrorMsg: "", Data: g.Map{"id": encrypt.SE.Encode2Hex(fmt.Sprintf("%d", replyId)), "reply": finalReply, "isEnd": true}}
		} else {
			rs = &v1.FindAnswerRes{ErrorCode: -1, ErrorMsg: finalReply}
		}
	}
	if rs.ErrorMsg != "" {
		exceedMsg := g.Cfg().MustGet(ctx, "limit.exceedMsg").String()
		fswMsg := g.Cfg().MustGet(ctx, "limit.fswMsg").String()
		blackMsg := g.Cfg().MustGet(ctx, "limit.blackMsg").String()
		errMsg := g.Cfg().MustGet(ctx, "limit.errMsg").String()
		emptyMsg := g.Cfg().MustGet(ctx, "limit.emptyMsg").String()
		if rs.ErrorMsg == exceedMsg || rs.ErrorMsg == fswMsg || rs.ErrorMsg == blackMsg || rs.ErrorMsg == errMsg || rs.ErrorMsg == emptyMsg {
			rs.ErrorCode = 0
		}
	}
	return rs, nil
}

type historyData struct {
	Question string
	Answer   string
}

func getHistory(ctx context.Context, personId int64, second, limit int) [][]string {
	qStr := `SELECT a. content as question,b.content as answer FROM base_service.ai_message_history a, ai_message_history b where   a.person_id=%d and a.create_time>"%s"  and b.question_id = a.id  order by a.create_time desc limit %d`
	st, _ := time.ParseDuration(fmt.Sprintf("-%ds", second))
	rs_ := []historyData{}
	time_ := time.Now().Add(st)
	createTime := date.FormatDate(&time_, date.Date_Full_Layout)
	query := fmt.Sprintf(qStr, personId, createTime, limit)
	err := g.Model("ai_message_history").Raw(query).Scan(&rs_)
	if err != nil {
		g.Log().Errorf(ctx, "获取历史记录失败：%v", err)
	}
	rs := [][]string{}
	for i := 0; i < len(rs_); i++ {
		rs = append(rs, []string{rs_[i].Question, rs_[i].Answer})
	}
	return rs
}