fonchain-fiee/pkg/service/asChat/robot/robot.go

// Package robot -----------------------------
// @file      : robot.go
// @author    : JJXu
// @contact   : wavingbear@163.com
// @time      : 2025/6/13 17:41
// -------------------------------------------
package robot

import (
	"context"
	"fmt"
	"fonchain-fiee/api/accountFiee"
	"fonchain-fiee/pkg/common/ws"
	"fonchain-fiee/pkg/service"
	"fonchain-fiee/pkg/service/asChat/consts"
	"log"
	"sync"
	"time"
)

func NewRobot() *Robot {
	ctx := context.Background()
	robotQuery, err := service.AccountFieeProvider.GetChatUserList(ctx, &accountFiee.GetChatUserListRequest{
		Query: &accountFiee.ChatUserData{Role: 3},
		Page:  1, PageSize: 1,
	})
	if err != nil {
		panic("聊天机器人初始化失败,err:" + err.Error())
	}
	var robotInfo *accountFiee.ChatUserData
	if robotQuery.Total > 0 {
		robotInfo = robotQuery.List[0]
	} else {
		robotInfo = &accountFiee.ChatUserData{
			NickName: "阿泰",
			Role:     3,
			Origin:   "fontree",
		}
		createChatUserResp, errs := service.AccountFieeProvider.CreateChatUser(ctx, robotInfo)
		if errs != nil {
			panic("聊天机器人创建失败,err:" + errs.Error())
		}
		robotInfo = createChatUserResp.Data
	}
	r := &Robot{
		Info: robotInfo,
		EventListener: &ws.EventListener{
			Name: "robot1",
			ListenEvents: []ws.ListenEvent{ //只监听消息推送事件
				{ws.EventChatMessage, ws.EventProgressAfter},
			},
			Chan: make(ws.ListenEventChan),
		},
	}
	consts.ChatRoom.RegisterEventListener(r.EventListener)
	go r.Run()
	return r
}

type Robot struct {
	Info      *accountFiee.ChatUserData //机器人信息
	Rules     []Reply                   //回复规则
	DelayTask []RobotTask               //演示任务
	ticker    *time.Ticker              //定时器
	stopChan  chan struct{}             //停止管道
	isRunning bool                      //运行状态
	mu        sync.Mutex
	*ws.EventListener
}

func (r *Robot) Listen(record *accountFiee.ChatRecordData) {
	for _, replyRules := range r.Rules {
		for _, rule := range replyRules.Rules {
			hit, runTime, function := rule.Hit(record)
			if hit && function != nil {
				if runTime.IsZero() {
					go func() {
						err := function(r.Info.ID, replyRules.Response)
						if err != nil {
							log.Printf("聊天机器人[%d]回复消息失败:%v", r.Info.ID, err)
						}
					}()
				} else {
					r.mu.Lock()
					r.DelayTask = append(r.DelayTask, RobotTask{
						RunTime:  runTime,
						Run:      function,
						Response: replyRules.Response,
					})
					r.mu.Unlock()
					// 添加任务后启动定时任务（如果未运行）
					if !r.isRunning {
						go r.Run()
					}
				}
				break
			}
		}
	}
}

func (r *Robot) Run() {
	r.mu.Lock()
	if r.isRunning {
		r.mu.Unlock()
		return
	}
	r.isRunning = true
	r.ticker = time.NewTicker(time.Second)
	r.stopChan = make(chan struct{})
	r.mu.Unlock()

	defer func() {
		r.mu.Lock()
		r.isRunning = false
		if r.ticker != nil {
			r.ticker.Stop()
			r.ticker = nil
		}
		r.stopChan = nil
		r.mu.Unlock()
	}()

	for {
		select {
		case <-r.ticker.C:
			r.mu.Lock()
			if len(r.DelayTask) == 0 {
				r.mu.Unlock()
				break
				//return // 没有任务时退出
			}
			now := time.Now()
			var remainingTasks []RobotTask
			for _, task := range r.DelayTask {
				if now.After(task.RunTime) {
					// 执行任务
					go func() {
						err := task.Run(r.Info.ID, task.Response)
						if err != nil {
							log.Printf("聊天机器人[%d]回复消息失败:%v", r.Info.ID, err)
						}
					}()
				} else {
					// 保留未到期的任务
					remainingTasks = append(remainingTasks, task)
				}
			}
			r.DelayTask = remainingTasks
			r.mu.Unlock()
		case <-r.stopChan:
			return
		case event := <-r.EventListener.Chan:
			fmt.Sprintf("listen event:%#v\n", event)

		}
	}
}

// Stop 主动停止机器人的定时任务
func (r *Robot) Stop() {
	r.mu.Lock()
	if r.stopChan != nil {
		close(r.stopChan)
	}
	r.mu.Unlock()
}