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 }