eta_api/services/llm/daf_service.go

package llm

import (
	"eta/eta_api/models/llm/dfa"
	"eta/eta_api/models/rag"
	"eta/eta_api/utils"
	"eta/eta_api/utils/llm"
	"fmt"
	"time"
)

const (
	fetchSize            = 500
	ETA_ARTICLE_TABLE    = "rag_eta_report"
	WECHAT_ARTICLE_TABLE = "wechat_article"
)

var (
	DAFHandler = llm.GetDAFHandlerInstance()
)

func DealHistoryArticleDafTags() {
	utils.FileLog.Info("自动处理历史文章算法标签任务开始")
	//获取还未处理的标签的文章
	ArticleParts, err := getDealDafTagsArticleList()
	if err != nil {
		utils.FileLog.Error("自动处理历史文章算法标签任务错误退出:err", err.Error())
		return
	}
	go func() {
		etaErr := DAFTagDeal(dfa.ArticleSourceEta, ArticleParts[dfa.ArticleSourceEta])
		if etaErr != nil {

		}
	}()
	go func() {
		wechatErr := DAFTagDeal(dfa.ArticleSourceWechat, ArticleParts[dfa.ArticleSourceWechat])
		if wechatErr != nil {

		}
	}()
	utils.FileLog.Info("自动处理历史文章算法标签任务结束")
}

func getDealDafTagsArticleList() (articleIdsMap map[dfa.ArticleSource][]int, err error) {
	list, err := dfa.GetArticleDafTagMappingList()
	if err != nil {
		utils.FileLog.Error("获取DAF算法标签文章列表失败,err:", err)
		err = fmt.Errorf("获取DAF算法标签文章列表失败,err:%v", err)
		return
	}
	articleIdsMap = make(map[dfa.ArticleSource][]int)
	for _, item := range list {
		if _, ok := articleIdsMap[item.Source]; !ok {
			articleIdsMap[item.Source] = make([]int, 0)
		}
		articleIdsMap[item.Source] = append(articleIdsMap[item.Source], item.ArticleID)
	}
	return
}
func DAFTagDeal(source dfa.ArticleSource, excludeArticleIds []int) (err error) {
	switch source {
	case dfa.ArticleSourceWechat:
		return wechatDeal(excludeArticleIds)
	case dfa.ArticleSourceEta:
		return
	default:
		utils.FileLog.Warn("位置的文章来源,无法进行DAF标签处理,自动退出")
		return
	}
}
func wechatDeal(excludeArticleIds []int) (err error) {
	var id = 0
	for {
		var articleList []*rag.DafWechatArticleItem
		var dafTagList []*dfa.ArticleDfaTagMapping
		articleList, id, err = rag.FetchArticleDafTagMappingList(id, excludeArticleIds, fetchSize)
		if err != nil {
			utils.FileLog.Error("获取微信文章列表失败,err:", err)
			return
		}
		if len(articleList) == 0 {
			break
		}
		for _, article := range articleList {
			dafTagList = append(dafTagList, &dfa.ArticleDfaTagMapping{
				ArticleID:      article.WechatArticleId,
				Source:         dfa.ArticleSourceWechat,
				ArticleTitle:   article.Title,
				ArticleContent: article.TextContent,
				TagName:        "微信文章",
				CreatedTime:    time.Now(),
			})
		}
		DAFHandler.BatchSubmitTasks(dafTagList)
	}
	return
}