eta_task/services/wechat_platform.go

package services

import (
	"context"
	"eta/eta_task/models/rag"
	"eta/eta_task/utils"
	"fmt"
	"time"
)

type WechatArticleOp struct {
	Source           string
	WechatPlatformId int
}

// AddWechatArticleOpToCache
// @Description: 将公众号文章操作加入缓存
// @param wechatPlatformId
// @param source
// @return bool
func AddWechatArticleOpToCache(wechatPlatformId int, source string) bool {
	record := new(WechatArticleOp)
	record.Source = source
	record.WechatPlatformId = wechatPlatformId
	if utils.Re == nil {
		err := utils.Rc.LPush(utils.CACHE_WECHAT_PLATFORM_ARTICLE, record)

		utils.FileLog.Info(fmt.Sprintf("将公众号文章操作 加入缓存 AddWechatArticleOpToCache LPush: 操作类型：%s，公众号id：%d", source, wechatPlatformId))
		if err != nil {
			fmt.Println("AddWechatArticleOpToCache LPush Err:" + err.Error())
		}
		return true
	}
	return false
}

// RefreshWechatPlatform
// @Description: 刷新公众号
// @author: Roc
// @datetime 2025-03-14 18:46:05
// @param cont context.Context
// @return err error
func RefreshWechatPlatform(cont context.Context) (err error) {
	utils.FileLog.Debug("RefreshWechatPlatform：", time.Now().Format(utils.FormatDateTime))
	obj := new(rag.WechatPlatform)
	platformList, tmpErr := obj.GetListByCondition(` AND enabled = 1 `, []interface{}{}, 0, 100000)
	if tmpErr != nil {
		err = tmpErr
		return
	}

	for _, v := range platformList {
		AddWechatArticleOpToCache(v.WechatPlatformId, "refresh")
		time.Sleep(1 * time.Minute)
	}

	return
}

// AiTaskRecordOp
// @Description: AI模块任务操作记录
type AiTaskRecordOp struct {
	AiTaskRecordId int
}

// AddAiTaskRecordOpToCache
// @Description: AI任务操作调度入队列
// @author: Roc
// @datetime 2025-04-24 09:41:11
// @param aiTaskRecordId int
// @return bool
func AddAiTaskRecordOpToCache(aiTaskRecordId int) bool {
	// 如果不在发布和调试模式，那么就不加入缓存
	if !utils.InArrayByStr([]string{utils.BusinessCodeRelease, utils.BusinessCodeDebug}, utils.BusinessCode) {
		return true
	}
	record := new(AiTaskRecordOp)
	record.AiTaskRecordId = aiTaskRecordId
	if utils.Re == nil {
		err := utils.Rc.LPush(utils.CACHE_AI_ARTICLE_ABSTRACT_LLM_TASK, record)

		utils.FileLog.Info(fmt.Sprintf("将AI任务操作调度入队列 加入缓存 AddAiTaskRecordOpToCache LPush: 记录id：%d", aiTaskRecordId))
		if err != nil {
			fmt.Println("AddAiTaskRecordOpToCache LPush Err:" + err.Error())
		}
		return true
	}
	return false
}

func CheckAiTask(cont context.Context) (err error) {
	defer func() {
		if err != nil {
			utils.FileLog.Error("CheckAiTask err:", err)
		}
	}()
	count, err := utils.Rc.LLen(utils.CACHE_AI_ARTICLE_ABSTRACT_LLM_TASK)
	if err != nil {
		fmt.Println("CheckAiTask err:", err)
	}
	if count > 0 {
		return
	}

	// 队列没数据了，那么需要重新检查下任务状态，并将状态为待处理且未完成的任务加入到队列中
	aiTaskObj := rag.AiTask{}
	taskList, err := aiTaskObj.GetListByCondition(``, ` AND status = ? `, []interface{}{`processing`}, 0, 10)
	if err != nil {
		fmt.Println("CheckAiTask err:", err)
		return
	}

	if len(taskList) <= 0 {
		return
	}

	aiTaskRecordObj := rag.AiTaskRecord{}
	for _, v := range taskList {
		// 查找具体记录
		recordList, tmpErr := aiTaskRecordObj.GetAllListByCondition(``, ` AND ai_task_id = ? AND status = ? `, []interface{}{v.AiTaskID, `待处理`})
		if tmpErr != nil {
			fmt.Println(v.AiTaskID, "获取待处理记录失败； err:", tmpErr)
			continue
		}

		// 将具体记录加入到队列中
		for _, record := range recordList {
			AddAiTaskRecordOpToCache(record.AiTaskRecordID)
		}
	}

	return
}