300 lines
8.9 KiB
Go
300 lines
8.9 KiB
Go
package srv_aicomment
|
|
|
|
import (
|
|
"context"
|
|
"math/rand"
|
|
"strconv"
|
|
"strings"
|
|
"sync/atomic"
|
|
"time"
|
|
"unicode/utf8"
|
|
|
|
"videoweb/pub/common/log"
|
|
"videoweb/pub/common/srvdef"
|
|
"videoweb/pub/models/aicommentrecord"
|
|
"videoweb/pub/models/commentmod"
|
|
"videoweb/pub/models/playletmod"
|
|
"videoweb/pub/models/usermod"
|
|
"videoweb/pub/mongos"
|
|
"videoweb/pub/pubdef"
|
|
|
|
"go.mongodb.org/mongo-driver/bson"
|
|
"go.mongodb.org/mongo-driver/mongo/options"
|
|
)
|
|
|
|
var pollingRunning atomic.Bool
|
|
|
|
const (
|
|
maxPollTaskCount int64 = 500
|
|
commentLevelFirst int64 = 1
|
|
commentReviewPass int = 1
|
|
)
|
|
|
|
func PollAiCommentResults(ctx context.Context) {
|
|
service := GetAiCommentService()
|
|
if !service.Configured() {
|
|
return
|
|
}
|
|
if !pollingRunning.CompareAndSwap(false, true) {
|
|
log.Debug("自动评论轮询正在执行中，本次跳过")
|
|
return
|
|
}
|
|
defer pollingRunning.Store(false)
|
|
|
|
if serr := expireStaleSubmittedTasks(ctx); serr != nil {
|
|
log.Error("自动评论轮询释放超时任务失败", log.E(serr))
|
|
return
|
|
}
|
|
records, serr := aicommentrecord.FindByStatus(ctx, aicommentrecord.StatusSubmitted, maxPollTaskCount)
|
|
if serr != nil {
|
|
log.Error("自动评论轮询查询任务记录失败", log.E(serr))
|
|
return
|
|
}
|
|
if len(records) == 0 {
|
|
return
|
|
}
|
|
taskIDs := make([]string, 0, len(records))
|
|
recordMap := make(map[string]*aicommentrecord.AiCommentRecord, len(records))
|
|
for _, record := range records {
|
|
if record.TaskId == "" {
|
|
continue
|
|
}
|
|
taskIDs = append(taskIDs, record.TaskId)
|
|
recordMap[record.TaskId] = record
|
|
}
|
|
results, err := service.QueryResults(ctx, taskIDs)
|
|
if err != nil {
|
|
log.Error("自动评论轮询查询生成结果失败", log.E(err), log.Any("taskIDs", taskIDs))
|
|
return
|
|
}
|
|
processAndInsertAiComments(ctx, results, recordMap)
|
|
}
|
|
|
|
func processAndInsertAiComments(ctx context.Context, results []ResultItem, recordMap map[string]*aicommentrecord.AiCommentRecord) {
|
|
users, err := robotUsers(ctx)
|
|
if err != nil || len(users) == 0 {
|
|
log.Error("自动评论机器人用户池为空或查询失败", log.E(err))
|
|
return
|
|
}
|
|
playletMap := loadPlayletsByResults(ctx, results, recordMap)
|
|
successTaskIDs := make([]string, 0)
|
|
|
|
for _, result := range results {
|
|
record, ok := recordMap[result.ID]
|
|
if !ok {
|
|
log.Warn("自动评论轮询未找到任务记录", log.Any("taskID", result.ID))
|
|
continue
|
|
}
|
|
if result.ErrorMsg != "" || result.Status == 2 {
|
|
_ = aicommentrecord.UpdateStatusAndDescByTaskID(ctx, result.ID, aicommentrecord.StatusFailed, result.ErrorMsg)
|
|
log.Error("自动评论生成结果失败", log.Any("taskID", result.ID), log.Any("videoId", result.VideoId), log.Any("errorMsg", result.ErrorMsg))
|
|
continue
|
|
}
|
|
if len(result.Comments) == 0 {
|
|
continue
|
|
}
|
|
playletID, err := strconv.ParseInt(record.Vid, 10, 64)
|
|
if err != nil || playletID <= 0 {
|
|
log.Warn("自动评论短剧ID无效", log.Any("taskID", result.ID), log.Any("vid", record.Vid), log.E(err))
|
|
continue
|
|
}
|
|
playlet := playletMap[playletID]
|
|
if playlet == nil || playlet.Status != playletmod.StatusUp {
|
|
_ = aicommentrecord.UpdateStatusAndDescByTaskID(ctx, result.ID, aicommentrecord.StatusAbandoned, "短剧已下架或不存在")
|
|
continue
|
|
}
|
|
objectType := record.ObjType
|
|
if objectType <= 0 {
|
|
objectType = objectTypePlaylet
|
|
}
|
|
retainCount := record.RetainCount
|
|
if retainCount <= 0 {
|
|
retainCount = len(result.Comments)
|
|
}
|
|
selectedComments := selectAIComments(result.Comments, retainCount)
|
|
currentCount, serr := commentmod.CountComments(ctx, bson.M{
|
|
commentmod.FN.Comment.ObjectId: playletID,
|
|
commentmod.FN.Comment.ObjectType: objectTypePlaylet,
|
|
commentmod.FN.Comment.ReviewStatus: commentReviewPass,
|
|
commentmod.FN.Comment.ParentsId: 0,
|
|
})
|
|
if serr != nil {
|
|
log.Error("自动评论统计现有评论失败", log.E(serr), log.Any("playletID", playletID))
|
|
continue
|
|
}
|
|
insertedCount := 0
|
|
now := time.Now()
|
|
publishedAt := PlayletPublishedAt(playlet)
|
|
objectName := ""
|
|
if playlet != nil {
|
|
objectName = playlet.Title
|
|
}
|
|
if objectName == "" {
|
|
objectName = record.Title
|
|
}
|
|
for index, comment := range selectedComments {
|
|
if currentCount >= int64(aiCommentPlayletTotal) {
|
|
break
|
|
}
|
|
comment.Text = strings.TrimSpace(comment.Text)
|
|
if comment.Text == "" || utf8.RuneCountInString(comment.Text) > 200 {
|
|
continue
|
|
}
|
|
user := users[rand.Intn(len(users))]
|
|
commentID := mongos.Lid.Comment.Get()
|
|
item := &commentmod.CommentModel{
|
|
Id: commentID,
|
|
ObjectId: playletID,
|
|
ObjectType: objectType,
|
|
ObjectName: objectName,
|
|
UserId: user.Id,
|
|
UserName: commentUserName(user),
|
|
UserAvatar: user.Avatar,
|
|
Level: commentLevelFirst,
|
|
Text: comment.Text,
|
|
ReviewStatus: commentReviewPass,
|
|
IsAi: true,
|
|
AiTaskId: result.ID,
|
|
AiIndex: index,
|
|
CreatedAt: ScatteredCommentTime(publishedAt, now, playletID, commentID, index, len(selectedComments)),
|
|
UpdatedAt: now,
|
|
}
|
|
inserted, insertErr := commentmod.InsertAICommentOnce(ctx, item)
|
|
if insertErr != nil {
|
|
log.Error("自动评论写入失败", log.E(insertErr), log.Any("taskID", result.ID), log.Any("index", index))
|
|
return
|
|
}
|
|
if inserted {
|
|
currentCount++
|
|
insertedCount++
|
|
}
|
|
}
|
|
if serr := refreshPlayletCommentCount(ctx, playletID); serr != nil {
|
|
continue
|
|
}
|
|
if currentCount < int64(aiCommentPlayletTotal) {
|
|
_ = aicommentrecord.UpdateStatusAndDescByTaskID(ctx, result.ID, aicommentrecord.StatusFailed, "生成评论不足，等待重试")
|
|
continue
|
|
}
|
|
successTaskIDs = append(successTaskIDs, result.ID)
|
|
log.Info("自动评论写入完成", log.Any("taskID", result.ID), log.Any("vid", record.Vid), log.Any("count", insertedCount))
|
|
}
|
|
|
|
if len(successTaskIDs) > 0 {
|
|
if serr := aicommentrecord.BatchUpdateStatus(ctx, successTaskIDs, aicommentrecord.StatusSuccess); serr != nil {
|
|
log.Error("自动评论更新成功任务状态失败", log.E(serr), log.Any("taskIDs", successTaskIDs))
|
|
}
|
|
}
|
|
}
|
|
|
|
func loadPlayletsByResults(ctx context.Context, results []ResultItem, recordMap map[string]*aicommentrecord.AiCommentRecord) map[int64]*playletmod.Playlet {
|
|
ids := make([]int64, 0, len(results))
|
|
seen := make(map[int64]struct{})
|
|
for _, result := range results {
|
|
if len(result.Comments) == 0 {
|
|
continue
|
|
}
|
|
record := recordMap[result.ID]
|
|
if record == nil {
|
|
continue
|
|
}
|
|
id, err := strconv.ParseInt(record.Vid, 10, 64)
|
|
if err != nil || id <= 0 {
|
|
continue
|
|
}
|
|
if _, ok := seen[id]; ok {
|
|
continue
|
|
}
|
|
seen[id] = struct{}{}
|
|
ids = append(ids, id)
|
|
}
|
|
if len(ids) == 0 {
|
|
return map[int64]*playletmod.Playlet{}
|
|
}
|
|
list, serr := playletmod.FindManyPlaylet(ctx, bson.M{playletmod.FN.Playlet.Id: bson.M{"$in": ids}})
|
|
if serr != nil {
|
|
log.Warn("自动评论查询短剧上架时间失败", log.E(serr), log.Any("ids", ids))
|
|
return map[int64]*playletmod.Playlet{}
|
|
}
|
|
res := make(map[int64]*playletmod.Playlet, len(list))
|
|
for _, playlet := range list {
|
|
res[playlet.Id] = playlet
|
|
}
|
|
return res
|
|
}
|
|
|
|
func robotUsers(ctx context.Context) (usermod.UserList, error) {
|
|
users, serr := usermod.FindManyUser(ctx, bson.M{
|
|
usermod.FN.User.Status: pubdef.UserStatusNormal,
|
|
usermod.FN.User.Id: bson.M{
|
|
"$gte": pubdef.RobotId_Min,
|
|
"$lte": pubdef.RobotId_Max,
|
|
},
|
|
}, options.Find().SetLimit(1000))
|
|
if serr != nil {
|
|
return nil, serr
|
|
}
|
|
if len(users) > 0 {
|
|
return users, nil
|
|
}
|
|
user := createFallbackRobotUser()
|
|
if serr = usermod.InsertOneUser(ctx, user); serr != nil {
|
|
return nil, serr
|
|
}
|
|
return usermod.UserList{user}, nil
|
|
}
|
|
|
|
func createFallbackRobotUser() *usermod.User {
|
|
now := time.Now()
|
|
return &usermod.User{
|
|
Id: mongos.Lid.User.Get(),
|
|
Type: pubdef.UserTypeGuest,
|
|
Status: pubdef.UserStatusNormal,
|
|
NickName: "游客" + strconv.FormatInt(now.Unix(), 10),
|
|
Device: "robot",
|
|
SysType: "ios",
|
|
LoginType: pubdef.LoginTypeDevice,
|
|
LastLoginAt: now,
|
|
CreatedAt: now,
|
|
UpdatedAt: now,
|
|
}
|
|
}
|
|
|
|
func commentUserName(user *usermod.User) string {
|
|
if user == nil {
|
|
return ""
|
|
}
|
|
if name := strings.TrimSpace(user.NickName); name != "" {
|
|
return name
|
|
}
|
|
if name := strings.TrimSpace(user.AccountNo); name != "" {
|
|
return name
|
|
}
|
|
return "用户" + strconv.FormatInt(user.Id, 10)
|
|
}
|
|
|
|
func refreshPlayletCommentCount(ctx context.Context, playletID int64) *srvdef.SrvErr {
|
|
if playletID <= 0 {
|
|
return nil
|
|
}
|
|
count, serr := commentmod.CountComments(ctx, bson.M{
|
|
commentmod.FN.Comment.ObjectId: playletID,
|
|
commentmod.FN.Comment.ObjectType: objectTypePlaylet,
|
|
commentmod.FN.Comment.ReviewStatus: commentReviewPass,
|
|
commentmod.FN.Comment.ParentsId: 0,
|
|
})
|
|
if serr != nil {
|
|
log.Warn("自动评论统计短剧评论数失败", log.E(serr), log.Any("playletId", playletID))
|
|
return serr
|
|
}
|
|
if serr = playletmod.UpdateOnePlaylet(ctx, bson.M{
|
|
playletmod.FN.Playlet.Id: playletID,
|
|
}, bson.M{
|
|
"$set": bson.M{playletmod.FN.Playlet.CountComment: count},
|
|
}); serr != nil {
|
|
log.Warn("自动评论更新短剧评论数失败", log.E(serr), log.Any("playletId", playletID))
|
|
return serr
|
|
}
|
|
return nil
|
|
}
|