3521 lines
108 KiB
Go
3521 lines
108 KiB
Go
package vidmod
|
||
|
||
import (
|
||
"bytes"
|
||
"context"
|
||
"encoding/json"
|
||
"errors"
|
||
"fmt"
|
||
"math"
|
||
"sort"
|
||
"strings"
|
||
"sync"
|
||
"time"
|
||
|
||
"91porn-server/app/appg"
|
||
"91porn-server/common"
|
||
"91porn-server/common/db"
|
||
"91porn-server/common/localcache"
|
||
"91porn-server/common/log"
|
||
"91porn-server/common/pageopt"
|
||
"91porn-server/common/stderr"
|
||
"91porn-server/models"
|
||
"91porn-server/models/commod"
|
||
"91porn-server/models/v/vidpopmod"
|
||
|
||
"github.com/shopspring/decimal"
|
||
"go.mongodb.org/mongo-driver/bson"
|
||
"go.mongodb.org/mongo-driver/bson/primitive"
|
||
"go.mongodb.org/mongo-driver/mongo"
|
||
"go.mongodb.org/mongo-driver/mongo/options"
|
||
)
|
||
|
||
var mdb *db.MongoDB
|
||
|
||
// 推荐将选比例
|
||
var chosonRate = 0.8
|
||
|
||
const (
|
||
ToppingLimit = 5 // ToppingLimit 置顶、力荐、置精数量限制
|
||
table = models.VideoInfo
|
||
vidPopCfgKey = "vidPopCfgKey"
|
||
allVidWithPop = "allVidWithPop"
|
||
)
|
||
|
||
func coll(t *db.MongoTool) *db.MongoTool {
|
||
if t == nil {
|
||
return mdb.Coll(table)
|
||
}
|
||
return t.Coll(table)
|
||
}
|
||
|
||
// initIndex 初始化索引
|
||
func initIndex() {
|
||
many := []mongo.IndexModel{
|
||
{
|
||
Keys: bson.D{{Key: "sourceID", Value: 1}},
|
||
Options: options.Index().SetUnique(true).SetPartialFilterExpression(bson.M{"sourceID": bson.M{"$gt": ""}}),
|
||
},
|
||
// skd 同步数据到es的时候需要按照更新时间筛选
|
||
{
|
||
Keys: bson.D{{"updatedAt", -1}},
|
||
},
|
||
{
|
||
Keys: bson.D{{Key: "commentCount", Value: -1}},
|
||
},
|
||
{
|
||
Keys: bson.D{{Key: "playCount", Value: -1}},
|
||
},
|
||
{
|
||
// 针对视频热度排序查询
|
||
Keys: bson.D{{Key: "newsType", Value: 1}, {Key: "status", Value: 1}, {Key: "hot", Value: -1}, {Key: "reviewAt", Value: -1}},
|
||
},
|
||
{
|
||
// 针对视频播放最多排序查询
|
||
Keys: bson.D{{Key: "newsType", Value: 1}, {Key: "status", Value: 1}, {Key: "playCount", Value: -1}, {Key: "reviewAt", Value: -1}},
|
||
},
|
||
{
|
||
// 针对视频点赞最多排序查询 2抖音-推荐排序
|
||
Keys: bson.D{{Key: "newsType", Value: 1}, {Key: "status", Value: 1}, {Key: "likeCount", Value: -1}, {Key: "reviewAt", Value: -1}},
|
||
},
|
||
{
|
||
Keys: bson.D{{Key: "isTopping", Value: -1}, {Key: "status", Value: -1}, {Key: "createdAt", Value: -1}},
|
||
},
|
||
{
|
||
Keys: bson.D{{Key: "publisherID", Value: 1}, {Key: "status", Value: 1}, {Key: "worksSort", Value: -1}, {Key: "createdAt", Value: -1}},
|
||
},
|
||
{
|
||
Keys: bson.D{{Key: "createdAt", Value: 1}},
|
||
},
|
||
{
|
||
Keys: bson.D{{Key: "status", Value: 1}, {Key: "playCount", Value: -1}, {Key: "likeCount", Value: -1}, {Key: "reviewAt", Value: -1}},
|
||
},
|
||
{
|
||
Keys: bson.D{{Key: "coins", Value: 1}, {Key: "status", Value: 1}, {Key: "reviewAt", Value: 1}},
|
||
},
|
||
// 由于废除了isTopping字段置顶，所以上线后这个索引要删掉
|
||
//{
|
||
// Keys: bson.D{{Key: "mId", Value: 1}, {Key: "status", Value: 1}, {Key: "isTopping", Value: 1}, {Key: "reviewAt", Value: 1}},
|
||
//},
|
||
// 置顶改为有数值的置顶，废除isTopping字段
|
||
// 首页最新展示样式模块
|
||
{
|
||
Keys: bson.D{{Key: "newsType", Value: 1}, {Key: "status", Value: 1}, {Key: "reviewAt", Value: -1}},
|
||
},
|
||
// 首页-最新
|
||
{
|
||
Keys: bson.D{{Key: "mId", Value: 1}, {Key: "status", Value: 1}, {Key: "reviewAt", Value: -1}},
|
||
},
|
||
// 首页-最多观看
|
||
{
|
||
Keys: bson.D{{Key: "mId", Value: 1}, {Key: "status", Value: 1}, {Key: "playCount", Value: -1}, {Key: "reviewAt", Value: -1}},
|
||
},
|
||
// 首页-最多观看
|
||
{
|
||
Keys: bson.D{{Key: "mId", Value: 1}, {Key: "status", Value: 1}, {Key: "hot", Value: -1}, {Key: "reviewAt", Value: -1}},
|
||
},
|
||
// 首页-热门推荐
|
||
{
|
||
Keys: bson.D{{Key: "mId", Value: 1}, {Key: "status", Value: 1}, {Key: "liaoBaTopSort", Value: -1}, {Key: "likeCount", Value: -1}, {Key: "reviewAt", Value: -1}},
|
||
},
|
||
// 首页-最多收藏
|
||
{
|
||
Keys: bson.D{{Key: "mId", Value: 1}, {Key: "status", Value: 1}, {Key: "collectCount", Value: -1}, {Key: "reviewAt", Value: -1}},
|
||
},
|
||
// 黄游-畅销（最多购买）
|
||
{
|
||
Keys: bson.D{{Key: "mId", Value: 1}, {Key: "status", Value: 1}, {Key: "purchaseCount", Value: -1}, {Key: "reviewAt", Value: -1}},
|
||
},
|
||
{
|
||
Keys: bson.D{{Key: "isTopping", Value: 1}, {Key: "reviewAt", Value: 1}},
|
||
},
|
||
// 由于废除了isTopping字段置顶，所以上线后这些个索引要删掉
|
||
//{
|
||
// Keys: bson.D{{Key: "tags", Value: 1}, {Key: "status", Value: 1}, {Key: "isTopping", Value: 1}, {Key: "reviewAt", Value: 1}},
|
||
//},
|
||
// 置顶改为有数值的置顶，废除isTopping字段
|
||
// 社区-最新/标签新
|
||
{
|
||
Keys: bson.D{{Key: "tags", Value: 1}, {Key: "status", Value: 1}, {Key: "reviewAt", Value: -1}},
|
||
},
|
||
// 社区-推荐/标签最多收藏
|
||
{
|
||
Keys: bson.D{{Key: "tags", Value: 1}, {Key: "status", Value: 1}, {Key: "liaoBaTopSort", Value: -1}, {Key: "collectCount", Value: -1}, {Key: "reviewAt", Value: -1}},
|
||
},
|
||
// 社区-最热/标签最多观看
|
||
{
|
||
Keys: bson.D{{Key: "tags", Value: 1}, {Key: "status", Value: 1}, {Key: "playCount", Value: -1}, {Key: "reviewAt", Value: -1}},
|
||
},
|
||
// 社区-精选
|
||
{
|
||
Keys: bson.D{{Key: "tags", Value: 1}, {Key: "status", Value: 1}, {Key: "likeCount", Value: -1}, {Key: "reviewAt", Value: -1}},
|
||
},
|
||
// 社区-视频/标签-最多观看--按照类型查询
|
||
{
|
||
Keys: bson.D{{Key: "tags", Value: 1}, {Key: "newsType", Value: 1}, {Key: "status", Value: 1}, {Key: "playCount", Value: -1}, {Key: "reviewAt", Value: -1}},
|
||
},
|
||
// 标签-最多收藏
|
||
{
|
||
Keys: bson.D{{Key: "tags", Value: 1}, {Key: "status", Value: 1}, {Key: "collectCount", Value: -1}, {Key: "reviewAt", Value: -1}},
|
||
},
|
||
// 标签-最新-按照类型查询
|
||
{
|
||
Keys: bson.D{{Key: "tags", Value: 1}, {Key: "newsType", Value: 1}, {Key: "status", Value: 1}, {Key: "reviewAt", Value: -1}},
|
||
},
|
||
// /标签-最多收藏-按照类型查询
|
||
{
|
||
Keys: bson.D{{Key: "tags", Value: 1}, {Key: "newsType", Value: 1}, {Key: "status", Value: 1}, {Key: "liaoBaTopSort", Value: -1}, {Key: "collectCount", Value: -1}, {Key: "reviewAt", Value: -1}},
|
||
},
|
||
// /标签-最多收藏-按照类型查询
|
||
{
|
||
Keys: bson.D{{Key: "tags", Value: 1}, {Key: "newsType", Value: 1}, {Key: "status", Value: 1}, {Key: "liaoBaTopSort", Value: -1}, {Key: "playCount", Value: -1}, {Key: "reviewAt", Value: -1}},
|
||
},
|
||
{ // 热评
|
||
Keys: bson.D{{Key: "tags", Value: 1}, {Key: "newsType", Value: 1}, {Key: "status", Value: 1}, {Key: "hot", Value: -1}, {Key: "reviewAt", Value: -1}},
|
||
},
|
||
{
|
||
Keys: bson.D{{Key: "status", Value: 1}, {Key: "isTopping", Value: -1}, {Key: "reviewAt", Value: 1}},
|
||
},
|
||
{ // 各类视频热度排序查询
|
||
Keys: bson.D{{Key: "status", Value: 1}, {Key: "hot", Value: -1}, {Key: "reviewAt", Value: -1}},
|
||
},
|
||
{ // 每日短视频推荐分数按 _id 稳定分批扫描
|
||
Keys: bson.D{
|
||
{Key: "newsType", Value: 1},
|
||
{Key: "status", Value: 1},
|
||
{Key: "deleteAt", Value: 1},
|
||
{Key: "_id", Value: 1},
|
||
},
|
||
Options: options.Index().SetName(shortRecommendRefreshIndexName),
|
||
},
|
||
// 折扣专区使用的索引
|
||
{
|
||
Keys: bson.D{{Key: "discountAreaId", Value: 1}, {Key: "status", Value: 1}, {Key: "reviewAt", Value: -1}},
|
||
},
|
||
{
|
||
Keys: bson.D{{Key: "discountAreaId", Value: 1}, {Key: "status", Value: 1}, {Key: "likeCount", Value: -1}, {Key: "reviewAt", Value: -1}},
|
||
},
|
||
|
||
// 2025--- 新版本 ---------------------
|
||
{
|
||
Keys: bson.D{{Key: "publisherID", Value: 1}, {Key: "status", Value: 1}, {Key: "reviewAt", Value: -1}},
|
||
}, { // 排行榜总榜查询
|
||
Keys: bson.D{{Key: "newsType", Value: 1}, {Key: "status", Value: 1}, {Key: "reviewAt", Value: -1}, {Key: "collectCount", Value: -1}},
|
||
}, { // 短视频推荐(精选靠前)
|
||
Keys: bson.D{{Key: "newsType", Value: 1}, {Key: "status", Value: 1}, {"chosen", -1}, {Key: "likeCount", Value: -1}, {Key: "reviewAt", Value: -1}},
|
||
},
|
||
{
|
||
Keys: bson.D{{Key: "showType", Value: 1}},
|
||
},
|
||
{
|
||
Keys: bson.D{{Key: "liaoBaTopSort", Value: 1}, {Key: "reviewAt", Value: -1}},
|
||
},
|
||
{
|
||
// H.265 等待队列：按状态和入队时间读取待提交任务。
|
||
Keys: bson.D{{Key: "h265Status", Value: 1}, {Key: "h265QueuedAt", Value: 1}},
|
||
},
|
||
{
|
||
// H.265 云端处理中队列：使用专用时间，避免普通业务更新影响超时判断。
|
||
Keys: bson.D{{Key: "h265Status", Value: 1}, {Key: "h265PendingAt", Value: 1}},
|
||
},
|
||
{
|
||
// H.265 失败重试：按状态、失败次数和审核时间读取可重试任务。
|
||
Keys: bson.D{{Key: "h265Status", Value: 1}, {Key: "h265FailCount", Value: 1}, {Key: "reviewAt", Value: -1}},
|
||
},
|
||
}
|
||
if _, err := coll(nil).CreateIndex(many); err != nil {
|
||
panic(fmt.Sprintf("%s model set index err ==>[%+v]", table, err))
|
||
}
|
||
}
|
||
|
||
type RecommendInteraction string
|
||
|
||
const (
|
||
RecommendInteractionLike RecommendInteraction = "like"
|
||
RecommendInteractionCollect RecommendInteraction = "collect"
|
||
RecommendInteractionComment RecommendInteraction = "comment"
|
||
RecommendInteractionShare RecommendInteraction = "share"
|
||
)
|
||
|
||
// IncrementRecommendInteraction 累加真实互动推荐值。取消、删除和运营假数据不得调用。
|
||
func IncrementRecommendInteraction(id ObjectID, interaction RecommendInteraction) error {
|
||
return IncrementRecommendInteractionContext(context.Background(), id, interaction)
|
||
}
|
||
|
||
// IncrementRecommendInteractionContext 累加真实互动推荐值，并继承调用方的取消和超时。
|
||
func IncrementRecommendInteractionContext(
|
||
ctx context.Context,
|
||
id ObjectID,
|
||
interaction RecommendInteraction,
|
||
) error {
|
||
field := ""
|
||
switch interaction {
|
||
case RecommendInteractionLike:
|
||
field = "recommendLikeCount"
|
||
case RecommendInteractionCollect:
|
||
field = "recommendCollectCount"
|
||
case RecommendInteractionComment:
|
||
field = "recommendCommentCount"
|
||
case RecommendInteractionShare:
|
||
field = "recommendShareCount"
|
||
default:
|
||
return fmt.Errorf("unknown recommend interaction: %s", interaction)
|
||
}
|
||
initialized := bson.M{
|
||
"_id": id, "status": CheckPass, "newsType": SHORT,
|
||
"recommendInitialized": true,
|
||
}
|
||
collection := coll(mdb.ToolCtx(ctx))
|
||
result, err := collection.UpdateOne(initialized, bson.M{"$inc": bson.M{field: 1}})
|
||
if err != nil || result.MatchedCount > 0 {
|
||
return err
|
||
}
|
||
|
||
// 初始化只会由一个并发请求成功；失败者随后重试普通 $inc，避免丢互动。
|
||
initializing := bson.M{
|
||
"_id": id, "status": CheckPass, "newsType": SHORT,
|
||
"recommendInitialized": bson.M{"$ne": true},
|
||
}
|
||
result, err = collection.UpdateOne(initializing, recommendInteractionInitializationPipeline(field))
|
||
if err != nil || result.MatchedCount > 0 {
|
||
return err
|
||
}
|
||
_, err = collection.UpdateOne(initialized, bson.M{"$inc": bson.M{field: 1}})
|
||
return err
|
||
}
|
||
|
||
func recommendInteractionInitializationPipeline(incrementField string) mongo.Pipeline {
|
||
countFields := []struct {
|
||
recommend string
|
||
legacy string
|
||
}{
|
||
{recommend: "recommendLikeCount", legacy: "likeCount"},
|
||
{recommend: "recommendCollectCount", legacy: "collectCount"},
|
||
{recommend: "recommendCommentCount", legacy: "commentCount"},
|
||
{recommend: "recommendShareCount", legacy: "shareCount"},
|
||
}
|
||
set := bson.M{"recommendInitialized": true}
|
||
for _, countField := range countFields {
|
||
current := bson.M{"$ifNull": bson.A{"$" + countField.recommend, int64(0)}}
|
||
baseline := bson.M{"$max": bson.A{
|
||
current,
|
||
bson.M{"$max": bson.A{
|
||
bson.M{"$ifNull": bson.A{"$" + countField.legacy, int64(0)}},
|
||
int64(0),
|
||
}},
|
||
}}
|
||
if countField.recommend == incrementField {
|
||
baseline = bson.M{"$add": bson.A{baseline, int64(1)}}
|
||
}
|
||
set[countField.recommend] = baseline
|
||
}
|
||
return mongo.Pipeline{bson.D{{Key: "$set", Value: set}}}
|
||
}
|
||
|
||
const (
|
||
shortRecommendRefreshIndexName = "idx_short_recommend_scan"
|
||
recommendScoreBatchSize = 1000
|
||
recommendScoreCursorBatchSize = 4000
|
||
recommendScoreWorkerCount = 2
|
||
recommendScoreCursorCloseTimeout = 5 * time.Second
|
||
)
|
||
|
||
// RecommendCandidate 是生成全局队列所需的最小视频快照。
|
||
type RecommendCandidate struct {
|
||
ID ObjectID
|
||
MID string
|
||
ReviewAt time.Time
|
||
RecommendScore int64
|
||
}
|
||
|
||
// recommendScoreDocument 仅在单个扫描批次内存活，避免全局队列长期持有互动明细。
|
||
type recommendScoreDocument struct {
|
||
ID ObjectID `bson:"_id"`
|
||
MID string `bson:"mId"`
|
||
ReviewAt time.Time `bson:"reviewAt"`
|
||
LikeCount int `bson:"likeCount"`
|
||
CollectCount int `bson:"collectCount"`
|
||
CommentCount int `bson:"commentCount"`
|
||
ShareCount int `bson:"shareCount"`
|
||
RecommendLikeCount int64 `bson:"recommendLikeCount"`
|
||
RecommendCollectCount int64 `bson:"recommendCollectCount"`
|
||
RecommendCommentCount int64 `bson:"recommendCommentCount"`
|
||
RecommendShareCount int64 `bson:"recommendShareCount"`
|
||
RecommendScore *int64 `bson:"recommendScore"`
|
||
RecommendInitialized bool `bson:"recommendInitialized"`
|
||
}
|
||
|
||
type recommendScoreStore interface {
|
||
MaxEligibleID(ctx context.Context) (ObjectID, error)
|
||
OpenCursor(ctx context.Context, maxID ObjectID, batchSize int32) (recommendScoreCursor, error)
|
||
BulkWrite(ctx context.Context, writes []mongo.WriteModel) error
|
||
}
|
||
|
||
type recommendInitializationCounter interface {
|
||
CountUninitialized(ctx context.Context, maxID ObjectID, limit int64) (int64, error)
|
||
}
|
||
|
||
type recommendScoreCursor interface {
|
||
Next(context.Context) bool
|
||
Decode(interface{}) error
|
||
Err() error
|
||
Close(context.Context) error
|
||
}
|
||
|
||
type recommendScoreScope struct {
|
||
generatedAt time.Time
|
||
excludedModuleIDs []string
|
||
}
|
||
|
||
func newRecommendScoreScope(generatedAt time.Time, excludedModuleIDs []string) recommendScoreScope {
|
||
seen := make(map[string]struct{}, len(excludedModuleIDs))
|
||
normalized := make([]string, 0, len(excludedModuleIDs))
|
||
for _, moduleID := range excludedModuleIDs {
|
||
moduleID = strings.TrimSpace(moduleID)
|
||
if moduleID == "" {
|
||
continue
|
||
}
|
||
if _, ok := seen[moduleID]; ok {
|
||
continue
|
||
}
|
||
seen[moduleID] = struct{}{}
|
||
normalized = append(normalized, moduleID)
|
||
}
|
||
return recommendScoreScope{
|
||
generatedAt: generatedAt,
|
||
excludedModuleIDs: normalized,
|
||
}
|
||
}
|
||
|
||
func (s recommendScoreScope) filter() bson.M {
|
||
filter := bson.M{
|
||
"status": CheckPass,
|
||
"newsType": SHORT,
|
||
"deleteAt": nil,
|
||
"recoWeight": bson.M{"$ne": -1},
|
||
// 审核通过时间晚于本次构建快照的视频不能进入新视频池或高分池。
|
||
"reviewAt": bson.M{"$lte": s.generatedAt},
|
||
}
|
||
if len(s.excludedModuleIDs) > 0 {
|
||
// mId $nin 的选择性通常较低，不为它单独增加索引；Mongo 继续使用
|
||
// newsType/status/deleteAt/_id 稳定扫描索引，并在服务端尽早做残余过滤。
|
||
filter["mId"] = bson.M{"$nin": append([]string(nil), s.excludedModuleIDs...)}
|
||
}
|
||
return filter
|
||
}
|
||
|
||
type mongoRecommendScoreStore struct {
|
||
scope recommendScoreScope
|
||
}
|
||
|
||
func (s mongoRecommendScoreStore) MaxEligibleID(ctx context.Context) (ObjectID, error) {
|
||
var latest struct {
|
||
ID ObjectID `bson:"_id"`
|
||
}
|
||
filter, opts := recommendScoreMaxEligibleIDQuery(s.scope)
|
||
err := coll(mdb.ToolCtx(ctx)).FindOne(&latest, filter, opts)
|
||
return recommendScoreMaxEligibleIDResult(latest.ID, err)
|
||
}
|
||
|
||
func recommendScoreMaxEligibleIDResult(id ObjectID, err error) (ObjectID, error) {
|
||
if errors.Is(err, mongo.ErrNoDocuments) {
|
||
return primitive.NilObjectID, nil
|
||
}
|
||
if err != nil {
|
||
return primitive.NilObjectID, err
|
||
}
|
||
return id, nil
|
||
}
|
||
|
||
func recommendScoreMaxEligibleIDQuery(
|
||
scope recommendScoreScope,
|
||
) (bson.M, *options.FindOneOptions) {
|
||
opts := options.FindOne().
|
||
SetProjection(bson.M{"_id": 1}).
|
||
SetSort(bson.D{{Key: "_id", Value: -1}}).
|
||
SetHint(shortRecommendRefreshIndexName)
|
||
return scope.filter(), opts
|
||
}
|
||
|
||
func recommendScoreCursorQuery(
|
||
maxID ObjectID,
|
||
batchSize int32,
|
||
scope recommendScoreScope,
|
||
) (bson.M, *options.FindOptions) {
|
||
filter := scope.filter()
|
||
filter["_id"] = bson.M{"$lte": maxID}
|
||
opts := options.Find().
|
||
SetProjection(bson.M{
|
||
"_id": 1, "mId": 1, "reviewAt": 1, "likeCount": 1, "collectCount": 1, "commentCount": 1, "shareCount": 1,
|
||
"recommendLikeCount": 1, "recommendCollectCount": 1, "recommendCommentCount": 1,
|
||
"recommendShareCount": 1, "recommendScore": 1, "recommendInitialized": 1,
|
||
}).
|
||
SetSort(bson.D{{Key: "_id", Value: 1}}).
|
||
SetBatchSize(batchSize).
|
||
SetHint(shortRecommendRefreshIndexName)
|
||
return filter, opts
|
||
}
|
||
|
||
func (s mongoRecommendScoreStore) OpenCursor(
|
||
ctx context.Context,
|
||
maxID ObjectID,
|
||
batchSize int32,
|
||
) (recommendScoreCursor, error) {
|
||
filter, opts := recommendScoreCursorQuery(maxID, batchSize, s.scope)
|
||
return coll(mdb.ToolCtx(ctx)).FindCursor(filter, opts)
|
||
}
|
||
|
||
func (mongoRecommendScoreStore) BulkWrite(ctx context.Context, writes []mongo.WriteModel) error {
|
||
_, err := coll(mdb.ToolCtx(ctx)).Bulk(writes, options.BulkWrite().SetOrdered(false))
|
||
return err
|
||
}
|
||
|
||
func (s mongoRecommendScoreStore) CountUninitialized(
|
||
ctx context.Context,
|
||
maxID ObjectID,
|
||
limit int64,
|
||
) (int64, error) {
|
||
if limit <= 0 {
|
||
return 0, nil
|
||
}
|
||
filter, opts := recommendInitializationCountQuery(maxID, limit, s.scope)
|
||
return coll(mdb.ToolCtx(ctx)).Count(filter, opts)
|
||
}
|
||
|
||
func recommendInitializationCountQuery(
|
||
maxID ObjectID,
|
||
limit int64,
|
||
scope recommendScoreScope,
|
||
) (bson.M, *options.CountOptions) {
|
||
filter := scope.filter()
|
||
filter["_id"] = bson.M{"$lte": maxID}
|
||
filter["recommendInitialized"] = bson.M{"$ne": true}
|
||
opts := options.Count().
|
||
SetHint(shortRecommendRefreshIndexName).
|
||
SetLimit(limit)
|
||
return filter, opts
|
||
}
|
||
|
||
func nonNegativeInt(v int) int64 {
|
||
if v < 0 {
|
||
return 0
|
||
}
|
||
return int64(v)
|
||
}
|
||
|
||
func maxInt64(a, b int64) int64 {
|
||
if a > b {
|
||
return a
|
||
}
|
||
return b
|
||
}
|
||
|
||
func CalculateRecommendScore(like, collect, comment, share int64) int64 {
|
||
return like + collect*2 + comment*3 + share*5
|
||
}
|
||
|
||
func compareObjectID(a, b ObjectID) int {
|
||
return bytes.Compare(a[:], b[:])
|
||
}
|
||
|
||
func prepareRecommendScoreBatch(
|
||
documents []recommendScoreDocument,
|
||
now time.Time,
|
||
) ([]RecommendCandidate, []mongo.WriteModel) {
|
||
candidates := make([]RecommendCandidate, 0, len(documents))
|
||
var writes []mongo.WriteModel
|
||
var updatePipeline mongo.Pipeline
|
||
for i := range documents {
|
||
document := &documents[i]
|
||
like := document.RecommendLikeCount
|
||
collect := document.RecommendCollectCount
|
||
comment := document.RecommendCommentCount
|
||
share := document.RecommendShareCount
|
||
// 传统真实计数是增量链路的每日补偿源；取max既能补偿瞬时漏写，
|
||
// 又保证取消点赞/收藏或删除评论时推荐累计分不回退。
|
||
like = maxInt64(like, nonNegativeInt(document.LikeCount))
|
||
collect = maxInt64(collect, nonNegativeInt(document.CollectCount))
|
||
comment = maxInt64(comment, nonNegativeInt(document.CommentCount))
|
||
share = maxInt64(share, nonNegativeInt(document.ShareCount))
|
||
score := CalculateRecommendScore(like, collect, comment, share)
|
||
candidates = append(candidates, RecommendCandidate{
|
||
ID: document.ID, MID: document.MID, ReviewAt: document.ReviewAt, RecommendScore: score,
|
||
})
|
||
if document.RecommendInitialized &&
|
||
document.RecommendScore != nil &&
|
||
*document.RecommendScore == score {
|
||
continue
|
||
}
|
||
if updatePipeline == nil {
|
||
updatePipeline = recommendScoreUpdatePipeline(now)
|
||
}
|
||
writes = append(writes, mongo.NewUpdateOneModel().
|
||
SetFilter(bson.M{"_id": document.ID}).
|
||
SetUpdate(updatePipeline))
|
||
}
|
||
return candidates, writes
|
||
}
|
||
|
||
func recommendScoreUpdatePipeline(now time.Time) mongo.Pipeline {
|
||
counts := []struct {
|
||
recommend string
|
||
legacy string
|
||
}{
|
||
{recommend: "recommendLikeCount", legacy: "likeCount"},
|
||
{recommend: "recommendCollectCount", legacy: "collectCount"},
|
||
{recommend: "recommendCommentCount", legacy: "commentCount"},
|
||
{recommend: "recommendShareCount", legacy: "shareCount"},
|
||
}
|
||
initializeSet := bson.M{"recommendInitialized": true}
|
||
for _, count := range counts {
|
||
current := bson.M{"$ifNull": bson.A{"$" + count.recommend, int64(0)}}
|
||
baseline := bson.M{"$max": bson.A{
|
||
current,
|
||
bson.M{"$max": bson.A{
|
||
bson.M{"$ifNull": bson.A{"$" + count.legacy, int64(0)}},
|
||
int64(0),
|
||
}},
|
||
}}
|
||
initializeSet[count.recommend] = baseline
|
||
}
|
||
score := bson.M{"$add": bson.A{
|
||
"$recommendLikeCount",
|
||
bson.M{"$multiply": bson.A{"$recommendCollectCount", int64(2)}},
|
||
bson.M{"$multiply": bson.A{"$recommendCommentCount", int64(3)}},
|
||
bson.M{"$multiply": bson.A{"$recommendShareCount", int64(5)}},
|
||
}}
|
||
return mongo.Pipeline{
|
||
bson.D{{Key: "$set", Value: initializeSet}},
|
||
bson.D{{Key: "$set", Value: bson.M{
|
||
"recommendScore": score, "recommendScoreAt": now,
|
||
}}},
|
||
}
|
||
}
|
||
|
||
func streamRecommendScoreBatches(
|
||
ctx context.Context,
|
||
maxID ObjectID,
|
||
batchSize int,
|
||
cursorBatchSize int32,
|
||
store recommendScoreStore,
|
||
yield func([]recommendScoreDocument) error,
|
||
) (err error) {
|
||
cursor, err := store.OpenCursor(ctx, maxID, cursorBatchSize)
|
||
if err != nil {
|
||
return err
|
||
}
|
||
if cursor == nil {
|
||
return errors.New("recommend score cursor is nil")
|
||
}
|
||
defer func() {
|
||
closeCtx, cancel := context.WithTimeout(context.Background(), recommendScoreCursorCloseTimeout)
|
||
defer cancel()
|
||
if closeErr := cursor.Close(closeCtx); err == nil && closeErr != nil {
|
||
err = closeErr
|
||
}
|
||
}()
|
||
|
||
batch := make([]recommendScoreDocument, 0, batchSize)
|
||
previousID := primitive.NilObjectID
|
||
for cursor.Next(ctx) {
|
||
var document recommendScoreDocument
|
||
if err := cursor.Decode(&document); err != nil {
|
||
return err
|
||
}
|
||
if document.ID.IsZero() ||
|
||
compareObjectID(document.ID, previousID) <= 0 ||
|
||
compareObjectID(document.ID, maxID) > 0 {
|
||
return fmt.Errorf("recommend score cursor returned an invalid _id sequence")
|
||
}
|
||
previousID = document.ID
|
||
batch = append(batch, document)
|
||
if len(batch) < batchSize {
|
||
continue
|
||
}
|
||
if err := yield(batch); err != nil {
|
||
return err
|
||
}
|
||
// Worker 异步持有已提交批次，必须换一块底层数组，不能 batch[:0] 复用。
|
||
batch = make([]recommendScoreDocument, 0, batchSize)
|
||
}
|
||
if err := cursor.Err(); err != nil {
|
||
return err
|
||
}
|
||
if len(batch) > 0 {
|
||
if err := yield(batch); err != nil {
|
||
return err
|
||
}
|
||
}
|
||
return nil
|
||
}
|
||
|
||
// refreshRecommendScores 固定扫描高水位，并使用有界 worker 分批刷新推荐分。
|
||
func refreshRecommendScores(
|
||
ctx context.Context,
|
||
now time.Time,
|
||
batchSize int,
|
||
cursorBatchSize int32,
|
||
workerCount int,
|
||
store recommendScoreStore,
|
||
initializationLimits ...int64,
|
||
) ([]RecommendCandidate, error) {
|
||
startedAt := time.Now()
|
||
if batchSize <= 0 {
|
||
return nil, fmt.Errorf("recommend score batch size must be positive")
|
||
}
|
||
if cursorBatchSize <= 0 {
|
||
return nil, fmt.Errorf("recommend score cursor batch size must be positive")
|
||
}
|
||
if workerCount <= 0 {
|
||
return nil, fmt.Errorf("recommend score worker count must be positive")
|
||
}
|
||
maxID, err := store.MaxEligibleID(ctx)
|
||
if err != nil {
|
||
return nil, err
|
||
}
|
||
if maxID.IsZero() {
|
||
return []RecommendCandidate{}, nil
|
||
}
|
||
initializationLimit := int64(0)
|
||
if len(initializationLimits) > 0 {
|
||
initializationLimit = initializationLimits[0]
|
||
}
|
||
if initializationLimit > 0 {
|
||
counter, ok := store.(recommendInitializationCounter)
|
||
if !ok {
|
||
return nil, fmt.Errorf("recommend score store does not support initialization guard")
|
||
}
|
||
count, countErr := counter.CountUninitialized(ctx, maxID, initializationLimit+1)
|
||
if countErr != nil {
|
||
return nil, countErr
|
||
}
|
||
if count > initializationLimit {
|
||
return nil, RecommendInitializationLimitError{
|
||
ObservedAtLeast: count,
|
||
Limit: initializationLimit,
|
||
}
|
||
}
|
||
}
|
||
|
||
workCtx, cancel := context.WithCancel(ctx)
|
||
defer cancel()
|
||
jobs := make(chan []recommendScoreDocument, workerCount)
|
||
candidates := make([]RecommendCandidate, 0, batchSize*workerCount)
|
||
scoreWriteCount := 0
|
||
bulkWriteCount := 0
|
||
var candidatesMu sync.Mutex
|
||
var firstErr error
|
||
var errOnce sync.Once
|
||
fail := func(err error) {
|
||
if err == nil {
|
||
return
|
||
}
|
||
errOnce.Do(func() {
|
||
firstErr = err
|
||
cancel()
|
||
})
|
||
}
|
||
|
||
var workers sync.WaitGroup
|
||
workers.Add(workerCount)
|
||
for i := 0; i < workerCount; i++ {
|
||
go func() {
|
||
defer workers.Done()
|
||
for {
|
||
select {
|
||
case <-workCtx.Done():
|
||
return
|
||
case batch, ok := <-jobs:
|
||
if !ok {
|
||
return
|
||
}
|
||
batchCandidates, writes := prepareRecommendScoreBatch(batch, now)
|
||
if len(writes) > 0 {
|
||
if err := store.BulkWrite(workCtx, writes); err != nil {
|
||
fail(err)
|
||
return
|
||
}
|
||
}
|
||
candidatesMu.Lock()
|
||
candidates = append(candidates, batchCandidates...)
|
||
scoreWriteCount += len(writes)
|
||
if len(writes) > 0 {
|
||
bulkWriteCount++
|
||
}
|
||
candidatesMu.Unlock()
|
||
}
|
||
}
|
||
}()
|
||
}
|
||
|
||
var producer sync.WaitGroup
|
||
producer.Add(1)
|
||
go func() {
|
||
defer producer.Done()
|
||
defer close(jobs)
|
||
err := streamRecommendScoreBatches(
|
||
workCtx,
|
||
maxID,
|
||
batchSize,
|
||
cursorBatchSize,
|
||
store,
|
||
func(batch []recommendScoreDocument) error {
|
||
select {
|
||
case jobs <- batch:
|
||
return nil
|
||
case <-workCtx.Done():
|
||
return workCtx.Err()
|
||
}
|
||
},
|
||
)
|
||
if err != nil {
|
||
fail(err)
|
||
}
|
||
}()
|
||
|
||
producer.Wait()
|
||
workers.Wait()
|
||
if firstErr != nil {
|
||
return nil, firstErr
|
||
}
|
||
if err := ctx.Err(); err != nil {
|
||
return nil, err
|
||
}
|
||
log.Info("short recommend scores refreshed",
|
||
log.Any("candidateCount", len(candidates)),
|
||
log.Any("scoreWriteCount", scoreWriteCount),
|
||
log.Any("bulkWriteCount", bulkWriteCount),
|
||
log.Any("batchSize", batchSize),
|
||
log.Any("cursorBatchSize", cursorBatchSize),
|
||
log.Any("workerCount", workerCount),
|
||
log.Any("durationMs", time.Since(startedAt).Milliseconds()))
|
||
return candidates, nil
|
||
}
|
||
|
||
// RefreshRecommendScores 全量计算短视频推荐分，仅回写首次初始化或分数变化的视频。
|
||
func RefreshRecommendScores(ctx context.Context, now time.Time) ([]RecommendCandidate, error) {
|
||
scope := newRecommendScoreScope(now, nil)
|
||
return refreshRecommendScores(
|
||
ctx,
|
||
now,
|
||
recommendScoreBatchSize,
|
||
recommendScoreCursorBatchSize,
|
||
recommendScoreWorkerCount,
|
||
mongoRecommendScoreStore{scope: scope},
|
||
)
|
||
}
|
||
|
||
// RecommendInitializationLimitError 表示首次推荐字段初始化量超过发布保护阈值。
|
||
// 该错误在任何BulkWrite之前返回，避免生产首发产生不可控的海量写入。
|
||
type RecommendInitializationLimitError struct {
|
||
ObservedAtLeast int64
|
||
Limit int64
|
||
}
|
||
|
||
func (e RecommendInitializationLimitError) Error() string {
|
||
return fmt.Sprintf(
|
||
"short recommend initialization requires at least %d writes, limit is %d",
|
||
e.ObservedAtLeast,
|
||
e.Limit,
|
||
)
|
||
}
|
||
|
||
// RefreshRecommendScoresWithInitializationLimit 在刷新前限制首次初始化写入量。
|
||
// excludedModuleIDs 会进入初始化计数与扫描共用的Mongo过滤器，确保被排除模块
|
||
// 不占用保护额度、不生成候选，也不会触发推荐分BulkWrite。
|
||
// limit<=0时关闭保护，行为与RefreshRecommendScores一致。
|
||
func RefreshRecommendScoresWithInitializationLimit(
|
||
ctx context.Context,
|
||
now time.Time,
|
||
limit int64,
|
||
excludedModuleIDs ...string,
|
||
) ([]RecommendCandidate, error) {
|
||
scope := newRecommendScoreScope(now, excludedModuleIDs)
|
||
return refreshRecommendScores(
|
||
ctx,
|
||
now,
|
||
recommendScoreBatchSize,
|
||
recommendScoreCursorBatchSize,
|
||
recommendScoreWorkerCount,
|
||
mongoRecommendScoreStore{scope: scope},
|
||
limit,
|
||
)
|
||
}
|
||
|
||
// GetTotalCnt 获取标签视频总数
|
||
func GetTotalCnt(cond bson.M) (int64, error) {
|
||
total, err := coll(nil).Count(cond)
|
||
if err != nil {
|
||
log.Error(fmt.Sprintf("[METHOD-%s]==> Model %s %s fail error:%+v:", "getTotalCnt", table, "Count", err),
|
||
log.Any("cond", cond),
|
||
)
|
||
return 0, err
|
||
}
|
||
return total, nil
|
||
}
|
||
|
||
// getPublishedVideoList 条件获取视频列表
|
||
func getPublishedVideoList(page, size uint64, cond bson.M, sort bson.D) ([]*VideoModel, int, int64, bool, error) {
|
||
hasNext := false
|
||
cond["status"] = 1
|
||
opts := options.FindOptions{}
|
||
if sort != nil {
|
||
opts.SetSort(sort)
|
||
}
|
||
opts.SetSkip(int64((page - 1) * size)).SetLimit(int64(size + 1))
|
||
var back []*VideoModel
|
||
if err := coll(nil).Find(&back, cond, &opts); err != nil {
|
||
log.Error(fmt.Sprintf("[METHOD-%s]==> Model %s %s fail error:%+v:", "getPublishedVideoList", table, "Find", err),
|
||
log.Any("page", page),
|
||
log.Any("size", size),
|
||
log.Any("cond", cond),
|
||
log.Any("sort", sort),
|
||
)
|
||
return nil, 0, 0, false, err
|
||
}
|
||
if uint64(len(back)) > size {
|
||
hasNext = true
|
||
back = back[:size]
|
||
}
|
||
return back, 1, 1, hasNext, nil
|
||
}
|
||
|
||
// GetVideoListByCond 条件获取视频列表
|
||
func GetVideoListByCond(filter primitive.M, opts ...*options.FindOptions) ([]*VideoModel, error) {
|
||
var out []*VideoModel
|
||
if err := coll(nil).Find(&out, filter, opts...); err != nil {
|
||
log.Error(fmt.Sprintf("[METHOD-QueryAllDocument]==> Model %s Find fail error:%+v:", table, err),
|
||
log.Any("filter", filter),
|
||
)
|
||
return nil, err
|
||
}
|
||
return out, nil
|
||
}
|
||
|
||
// getPublishedVideoList 条件获取视频列表，供推荐使用
|
||
func getPublishedVideoList4Reco(size uint64, cond bson.M, sort bson.D) ([]*RecoVideoModel, error) {
|
||
var back []*RecoVideoModel
|
||
if size == 0 {
|
||
return back, nil
|
||
}
|
||
cond["status"] = 1
|
||
cond["playTime"] = bson.M{"$lt": 600}
|
||
opts := options.FindOptions{}
|
||
if sort != nil {
|
||
opts.SetSort(sort)
|
||
}
|
||
opts.SetProjection(bson.M{"_id": 1, "createdAt": 1, "chosenDate": 1})
|
||
opts.SetLimit(int64(size))
|
||
if err := coll(nil).Find(&back, cond, &opts); err != nil {
|
||
log.Error(fmt.Sprintf("[METHOD-%s]==> Model %s %s fail error:%+v:", "getPublishedVideoList4Reco", table, "Find", err),
|
||
log.Any("size", size),
|
||
log.Any("cond", cond),
|
||
log.Any("sort", sort),
|
||
)
|
||
return nil, err
|
||
}
|
||
return back, nil
|
||
}
|
||
|
||
// getDistinctVideoID 获取推荐的视频id，发布者去重
|
||
func getDistinctVideoID(page, size uint64, cond bson.M, sort bson.D) ([]primitive.ObjectID, time.Time, time.Time, error) {
|
||
var maxChosenLine time.Time
|
||
var maxCreateLine time.Time
|
||
cond["recoWeight"] = bson.M{"$gte": 0}
|
||
videoInfos, err := getPublishedVideoList4Reco(size, cond, sort)
|
||
if err != nil {
|
||
return []primitive.ObjectID{}, maxChosenLine, maxCreateLine, err
|
||
}
|
||
videoInfosLen := len(videoInfos)
|
||
if videoInfosLen == 0 {
|
||
return []primitive.ObjectID{}, maxChosenLine, maxCreateLine, nil
|
||
}
|
||
vids := make([]primitive.ObjectID, 0, videoInfosLen)
|
||
for _, i := range videoInfos {
|
||
if i == nil {
|
||
continue
|
||
}
|
||
if i.ChosenDate.After(maxChosenLine) {
|
||
maxChosenLine = i.ChosenDate
|
||
}
|
||
if i.CreatedAt.After(maxCreateLine) {
|
||
maxCreateLine = i.CreatedAt
|
||
}
|
||
vids = append(vids, i.ID)
|
||
}
|
||
return vids, maxChosenLine, maxCreateLine, nil
|
||
}
|
||
|
||
func getChosenVideoID(line time.Time, page, size uint64, cond bson.M, sort bson.D) ([]primitive.ObjectID, time.Time, error) {
|
||
cond["chosenDate"] = bson.M{"$gt": line}
|
||
cond["chosen"] = true
|
||
ids, chonseLine, _, err := getDistinctVideoID(page, size, cond, sort)
|
||
return ids, chonseLine, err
|
||
}
|
||
|
||
func getUnChosenVideoID(line time.Time, page, size uint64, cond bson.M, sort bson.D) ([]primitive.ObjectID, time.Time, error) {
|
||
cond["createdAt"] = bson.M{"$gt": line}
|
||
cond["chosen"] = false
|
||
ids, _, createLine, err := getDistinctVideoID(page, size, cond, sort)
|
||
return ids, createLine, err
|
||
}
|
||
|
||
func getRecommendVideoIDs(chosenLine time.Time, unpopLine time.Time, cond bson.M, page, size uint64, sort bson.D) ([]primitive.ObjectID, time.Time, time.Time, error) {
|
||
chosenSize := uint64(float64(size) * chosonRate)
|
||
if chosenSize == 0 {
|
||
chosenSize = 1
|
||
}
|
||
var maxChoseLine time.Time
|
||
var maxCreateLine time.Time
|
||
idsChosen, maxChoseLine, err := getChosenVideoID(chosenLine, page, chosenSize, cond, sort)
|
||
if err != nil {
|
||
return nil, maxChoseLine, maxCreateLine, err
|
||
}
|
||
unChosenSize := size - uint64(len(idsChosen))
|
||
if unChosenSize == 0 {
|
||
return idsChosen, maxChoseLine, maxCreateLine, nil
|
||
}
|
||
delete(cond, "chosenDate")
|
||
idsUnchosen, maxCreateLine, err := getUnChosenVideoID(unpopLine, page, unChosenSize, cond, sort)
|
||
if err != nil {
|
||
return nil, maxChoseLine, maxCreateLine, err
|
||
}
|
||
idsChosen = append(idsChosen, idsUnchosen...)
|
||
return idsChosen, maxChoseLine, maxCreateLine, nil
|
||
}
|
||
|
||
// getVideoList 条件获取视频列表
|
||
func getVideoList(page, size uint64, cond bson.M, s bson.D) ([]*VideoModel, int64, bool, error) {
|
||
total, err := coll(nil).Count(cond)
|
||
if err != nil {
|
||
log.Error(fmt.Sprintf("[METHOD-%s]==> Model %s %s fail error:%+v:", "getVideoList", table, "Count", err),
|
||
log.Any("page", page),
|
||
log.Any("size", size),
|
||
log.Any("cond", cond),
|
||
log.Any("sort", s),
|
||
)
|
||
return nil, 0, false, err
|
||
}
|
||
var (
|
||
back []*VideoModel
|
||
opts = options.Find().SetSkip(int64(page-1) * int64(size)).SetLimit(int64(size))
|
||
)
|
||
if s != nil {
|
||
opts.SetSort(s)
|
||
}
|
||
if err = coll(nil).Find(&back, cond, opts); err != nil {
|
||
log.Error(fmt.Sprintf("[METHOD-%s]==> Model %s %s fail error:%+v:", "getVideoList", table, "Find", err),
|
||
log.Any("page", page),
|
||
log.Any("size", size),
|
||
log.Any("cond", cond),
|
||
log.Any("sort", s),
|
||
)
|
||
return nil, 0, false, err
|
||
}
|
||
if uint64(len(back)) > size {
|
||
return back[:size], total, true, nil
|
||
}
|
||
return back, total, false, nil
|
||
}
|
||
|
||
func GetHotList(skip, limit int64, cond bson.M) (back []*VideoModel, total int64, hasNext bool, err error) {
|
||
// 获取总条数
|
||
total, err = coll(nil).Count(cond)
|
||
if err != nil {
|
||
log.Error(fmt.Sprintf("[METHOD-%s]==> Model %s %s fail error:%+v:", "getVideoList", table, "Find", err),
|
||
log.Any("skip", skip),
|
||
log.Any("limit", limit),
|
||
log.Any("cond", cond),
|
||
)
|
||
return
|
||
}
|
||
if total == 0 {
|
||
return
|
||
}
|
||
res := make([]*VideoModel, 0)
|
||
if err = coll(nil).Find(&res, cond); err != nil {
|
||
log.Error(fmt.Sprintf("[METHOD-%s]==> Model %s %s fail error:%+v:", "getVideoList", table, "Find", err),
|
||
log.Any("skip", skip),
|
||
log.Any("limit", limit),
|
||
log.Any("cond", cond),
|
||
)
|
||
return
|
||
}
|
||
cfg, err := getVideoPopularityConfig()
|
||
if err != nil {
|
||
return
|
||
}
|
||
for _, v := range res {
|
||
v.Hot = calcHot(v, cfg)
|
||
}
|
||
sort.Slice(res, func(i, j int) bool {
|
||
return res[i].Hot > res[j].Hot
|
||
})
|
||
if len(res) <= int(skip) {
|
||
hasNext = false
|
||
return
|
||
}
|
||
if len(res) <= int(skip+limit) {
|
||
back = res[skip:]
|
||
} else {
|
||
back = res[skip : skip+limit]
|
||
hasNext = true
|
||
}
|
||
return
|
||
}
|
||
|
||
// 主动设置缓存
|
||
func SetAllVidWithPopCache() {
|
||
res := make([]*VideoModel, 0)
|
||
filter := bson.M{
|
||
"newsType": SP,
|
||
"status": 1,
|
||
}
|
||
if err := coll(nil).Find(&res, filter); err != nil {
|
||
log.Error(fmt.Sprintf("[METHOD-%s]==> Model %s %s fail error:%+v:", "setAllVidWithPopCache", table, "Find", err),
|
||
log.Any("cond", filter),
|
||
)
|
||
return
|
||
}
|
||
cfg, err := getVideoPopularityConfig()
|
||
if err != nil {
|
||
return
|
||
}
|
||
for _, v := range res {
|
||
v.Hot = calcHot(v, cfg)
|
||
}
|
||
sort.Slice(res, func(i, j int) bool {
|
||
return res[i].Hot > res[j].Hot
|
||
})
|
||
localcache.C.Set(allVidWithPop, res, 30*time.Minute)
|
||
}
|
||
|
||
func GetHotListFromCache(skip, limit int64, listType ListType, tag string, paymentEnum PaymentEnum) (back []*VideoModel, hasNext bool, err error) {
|
||
res := make([]*VideoModel, 0)
|
||
if cfg, ok := localcache.C.Get(allVidWithPop); ok {
|
||
res = cfg.([]*VideoModel)
|
||
} else {
|
||
filter := bson.M{
|
||
"newsType": SP,
|
||
"status": 1,
|
||
}
|
||
if err = coll(nil).Find(&res, filter); err != nil {
|
||
log.Error(fmt.Sprintf("[METHOD-%s]==> Model %s %s fail error:%+v:", "getVideoList", table, "Find", err),
|
||
log.Any("skip", skip),
|
||
log.Any("limit", limit),
|
||
log.Any("cond", filter),
|
||
)
|
||
return
|
||
}
|
||
cfg, err := getVideoPopularityConfig()
|
||
if err != nil {
|
||
return nil, false, err
|
||
}
|
||
for _, v := range res {
|
||
v.Hot = calcHot(v, cfg)
|
||
}
|
||
sort.Slice(res, func(i, j int) bool {
|
||
return res[i].Hot > res[j].Hot
|
||
})
|
||
localcache.C.Set(allVidWithPop, res, 30*time.Minute)
|
||
}
|
||
if listType != Video || tag != "" || paymentEnum != PaymentDefault {
|
||
outIndexSli := make([]int, 0)
|
||
for i, v := range res {
|
||
if listType == LongVideo && v.PlayTime < 600 {
|
||
outIndexSli = append([]int{i}, outIndexSli...)
|
||
continue
|
||
}
|
||
if listType == ShortVideo && v.PlayTime >= 600 {
|
||
outIndexSli = append([]int{i}, outIndexSli...)
|
||
continue
|
||
}
|
||
if tag != "" {
|
||
hasTag := false
|
||
for _, t := range v.Tags {
|
||
if t.Hex() == tag {
|
||
hasTag = true
|
||
break
|
||
}
|
||
}
|
||
if !hasTag {
|
||
outIndexSli = append([]int{i}, outIndexSli...)
|
||
continue
|
||
}
|
||
}
|
||
if paymentEnum == PaymentVIP && v.Coins > 0 {
|
||
outIndexSli = append([]int{i}, outIndexSli...)
|
||
continue
|
||
}
|
||
if paymentEnum == PaymentGold && v.Coins == 0 {
|
||
outIndexSli = append([]int{i}, outIndexSli...)
|
||
continue
|
||
}
|
||
}
|
||
if len(outIndexSli) > 0 {
|
||
for oi := range outIndexSli {
|
||
res = append(res[:outIndexSli[oi]], res[outIndexSli[oi]+1:]...)
|
||
}
|
||
}
|
||
}
|
||
if len(res) <= int(skip) {
|
||
hasNext = false
|
||
return
|
||
}
|
||
if len(res) <= int(skip+limit) {
|
||
back = res[skip:]
|
||
} else {
|
||
back = res[skip : skip+limit]
|
||
hasNext = true
|
||
}
|
||
return
|
||
}
|
||
|
||
func calcHot(video *VideoModel, cfg *vidpopmod.VideoPopularityConfig) float64 {
|
||
if video.Status == CheckPass || video.Status == Free {
|
||
qualityScore := float64(video.PlayCount + cfg.EffectivePlayCountMultiplier*video.EffectivePlayCount +
|
||
cfg.LikeCountMultiplier*video.LikeCount)
|
||
hot := (qualityScore + float64(cfg.InitialPopularity)) / math.Pow(1+time.Since(video.ReviewAt).Hours(),
|
||
cfg.ReviewTimePower)
|
||
return hot
|
||
}
|
||
return 0
|
||
}
|
||
|
||
func getVideoPopularityConfig() (*vidpopmod.VideoPopularityConfig, error) {
|
||
cfg, ok := localcache.C.Get(vidPopCfgKey)
|
||
if ok {
|
||
return cfg.(*vidpopmod.VideoPopularityConfig), nil
|
||
}
|
||
res, err := vidpopmod.FindOne()
|
||
if err != nil {
|
||
return nil, err
|
||
}
|
||
localcache.C.Set(vidPopCfgKey, &res, 15*time.Minute)
|
||
return &res, nil
|
||
}
|
||
|
||
// 获取作品列表
|
||
func getWorksList(skip, limit int64, cond bson.M, sort bson.D) (back []*VideoModel, total int64, hasNext bool, err error) {
|
||
opts := options.FindOptions{}
|
||
if sort != nil {
|
||
opts.SetSort(sort)
|
||
}
|
||
opts.SetSkip(skip).SetLimit(limit + 1)
|
||
// 获取总条数
|
||
total, err = coll(nil).Count(cond)
|
||
if err != nil {
|
||
log.Error(fmt.Sprintf("[METHOD-%s]==> Model %s %s fail error:%+v:", "getVideoList", table, "Find", err),
|
||
log.Any("skip", skip),
|
||
log.Any("limit", limit),
|
||
log.Any("cond", cond),
|
||
log.Any("sort", sort),
|
||
)
|
||
return
|
||
}
|
||
if total == 0 {
|
||
return
|
||
}
|
||
if err = coll(nil).Find(&back, cond, &opts); err != nil {
|
||
log.Error(fmt.Sprintf("[METHOD-%s]==> Model %s %s fail error:%+v:", "getVideoList", table, "Find", err),
|
||
log.Any("skip", skip),
|
||
log.Any("limit", limit),
|
||
log.Any("cond", cond),
|
||
log.Any("sort", sort),
|
||
)
|
||
return
|
||
}
|
||
if len(back) > int(limit) {
|
||
hasNext = true
|
||
back = back[:limit]
|
||
}
|
||
return
|
||
}
|
||
|
||
// ModifyMany 批量修改视屏信息
|
||
func ModifyMany(ids []primitive.ObjectID, doc SetDoc) (int64, error) {
|
||
if len(ids) == 0 {
|
||
return 0, nil
|
||
}
|
||
filter := M{
|
||
"_id": bson.M{"$in": ids},
|
||
}
|
||
docM, err := common.ToBsonM(doc)
|
||
if err != nil {
|
||
return 0, err
|
||
}
|
||
update := M{
|
||
"$set": docM,
|
||
}
|
||
ret, err := coll(nil).UpdateMany(filter, update)
|
||
if err != nil {
|
||
log.Error(fmt.Sprintf("[METHOD-%s]==> Model %s %s fail error:%+v:", "RemoveVideoFromMy", table, "UpdateMany", err),
|
||
log.Any("updateDoc", doc),
|
||
)
|
||
return 0, err
|
||
}
|
||
return ret.ModifiedCount, nil
|
||
}
|
||
|
||
// CountByCreatedTime CountByCreatedTime
|
||
func CountByCreatedTime(start time.Time, end time.Time) (int64, error) {
|
||
count, err := coll(nil).Count(bson.M{"createdAt": bson.M{"$gte": start, "$lt": end}})
|
||
if err != nil {
|
||
log.Error(fmt.Sprintf("[METHOD-%s]==> Model %s %s fail error:%+v:", "CountByCreatedTime", table, "Count", err))
|
||
return count, err
|
||
}
|
||
return count, nil
|
||
}
|
||
|
||
// CountTotalLikeByUID 查询用户被赞总数
|
||
func CountTotalLikeByUID(uid uint64) (int64, error) {
|
||
pipeline := []bson.M{
|
||
{"$match": bson.M{"publisherID": uid}},
|
||
{"$group": bson.M{"_id": nil, "total": bson.M{"$sum": "$likeCount"}}},
|
||
}
|
||
type res struct {
|
||
Total int64 `json:"total" bson:"total"`
|
||
}
|
||
resA := make([]*res, 0)
|
||
if err := coll(nil).Aggregate(&resA, pipeline); err != nil {
|
||
log.Error(fmt.Sprintf("[METHOD-%s]==> Model %s %s fail error:%+v:", "CountTotalLikeByUID", table, "Aggregate", err),
|
||
log.Any("uid", uid),
|
||
)
|
||
return 0, err
|
||
}
|
||
if len(resA) > 0 {
|
||
return resA[0].Total, nil
|
||
}
|
||
return 0, nil
|
||
}
|
||
|
||
// 获取广告帖子
|
||
func GetAdVideoIdsList(size uint64) []string {
|
||
cond := bson.M{"status": CheckPass, "newsType": bson.M{"$in": []string{AD_COVER, AD_SP}}, "recoWeight": bson.M{"$gte": 0}}
|
||
sort := bson.D{{Key: "sortCode", Value: -1}, {Key: "createdAt", Value: -1}}
|
||
videoInfos, err := getPublishedVideoList4Reco(size, cond, sort)
|
||
if err != nil {
|
||
return []string{}
|
||
}
|
||
videoInfosLen := len(videoInfos)
|
||
if videoInfosLen == 0 {
|
||
return []string{}
|
||
}
|
||
ids := make([]string, videoInfosLen)
|
||
for i := range videoInfos {
|
||
ids[i] = videoInfos[i].ID.Hex()
|
||
}
|
||
return ids
|
||
}
|
||
|
||
// GetVideoListByPublishers 通过发布者获取视频列表
|
||
func GetVideoListByPublishers(uids []uint64, page, size uint64) ([]*VideoModel, int, bool, error) {
|
||
if len(uids) == 0 {
|
||
return []*VideoModel{}, 0, false, nil
|
||
}
|
||
cond := bson.M{"publisherID": bson.M{"$in": uids}} // 下层函数已经设置过滤status
|
||
sort := bson.D{{Key: "reviewAt", Value: -1}}
|
||
infos, totalPages, _, hasNext, err := getPublishedVideoList(page, size, cond, sort)
|
||
return infos, totalPages, hasNext, err
|
||
}
|
||
|
||
// GetVideoListRecommend 获取推荐视频列表
|
||
func GetVideoListRecommend(chosenLine time.Time, unpopLine time.Time, newsType, quality, direction string, page, size uint64) ([]*VideoModel, []primitive.ObjectID, int, time.Time, time.Time, error) {
|
||
cond := bson.M{"newsType": newsType, "coins": 0}
|
||
if quality != "" {
|
||
cond["quality"] = quality
|
||
}
|
||
if direction != "" {
|
||
cond["direction"] = direction
|
||
}
|
||
sort := bson.D{{Key: "createdAt", Value: 1}}
|
||
ids, maxChoseLine, maxCreateLine, err := getRecommendVideoIDs(chosenLine, unpopLine, cond, page, size, sort)
|
||
if err != nil {
|
||
return nil, nil, 0, maxChoseLine, maxCreateLine, err
|
||
}
|
||
newIDs := common.Deduplication(ids)
|
||
infos, err := GetVideoListByIDs(newIDs)
|
||
if err != nil {
|
||
return nil, nil, 0, maxChoseLine, maxCreateLine, err
|
||
}
|
||
return infos, newIDs, 1, maxChoseLine, maxCreateLine, nil
|
||
}
|
||
|
||
func FindOneByUid(cond bson.M) (vidInfo *VideoModel, err error) {
|
||
sort := D{
|
||
{"reviewAt", -1},
|
||
}
|
||
opt := (&options.FindOneOptions{}).SetSort(sort)
|
||
err = coll(nil).FindOne(&vidInfo, cond, opt)
|
||
return
|
||
}
|
||
|
||
// GetUnpopularVideos 获取冷门视频列表
|
||
func GetUnpopularVideos(line time.Time, newsType, quality, direction string, page, size uint64) ([]string, time.Time, error) {
|
||
cond := bson.M{"createdAt": bson.M{"$gt": line}, "coins": 0, "newsType": newsType}
|
||
if quality != "" {
|
||
cond["quality"] = quality
|
||
}
|
||
if direction != "" {
|
||
cond["direction"] = direction
|
||
}
|
||
sort := bson.D{{Key: "createdAt", Value: 1}}
|
||
ids, maxLine, err := getUnChosenVideoID(line, page, size, cond, sort)
|
||
if err != nil {
|
||
return nil, maxLine, err
|
||
}
|
||
if len(ids) == 0 { //如果没有视频拉取了，重置时间游标
|
||
var newRound time.Time
|
||
ids, maxLine, err = getUnChosenVideoID(newRound, page, size, cond, sort)
|
||
}
|
||
arr := common.ObjectIDs2String(ids)
|
||
return arr, maxLine, err
|
||
}
|
||
|
||
// GetChargeVideos 获取收费视频
|
||
func GetChargeVideos(line time.Time, newsType, quality, direction string, page, size uint64) ([]string, time.Time, error) {
|
||
cond := bson.M{"coins": bson.M{"$gt": 0}, "newsType": newsType}
|
||
if quality != "" {
|
||
cond["quality"] = quality
|
||
}
|
||
if direction != "" {
|
||
cond["direction"] = direction
|
||
}
|
||
sort := bson.D{{Key: "createdAt", Value: 1}}
|
||
ids, maxLine, err := getUnChosenVideoID(line, page, size, cond, sort)
|
||
if err != nil {
|
||
return nil, maxLine, err
|
||
}
|
||
if len(ids) == 0 { //如果没有视频拉取了，重置时间游标
|
||
var newRound time.Time
|
||
ids, maxLine, err = getUnChosenVideoID(newRound, page, size, cond, sort)
|
||
}
|
||
arr := common.ObjectIDs2String(ids)
|
||
return arr, maxLine, err
|
||
}
|
||
|
||
// GetLatestUploadForRecommend 获取最新上传视频列表
|
||
func GetLatestUploadForRecommend(line time.Time, newsType, quality, direction string, page, size uint64) ([]string, time.Time, error) {
|
||
sort := bson.D{{Key: "createdAt", Value: 1}}
|
||
filter := bson.M{"createdAt": bson.M{"$gt": line}, "coins": 0, "newsType": newsType}
|
||
if quality != "" {
|
||
filter["quality"] = quality
|
||
}
|
||
if direction != "" {
|
||
filter["direction"] = direction
|
||
}
|
||
oids, _, maxLine, err := getDistinctVideoID(page, size, filter, sort)
|
||
ids := common.ObjectIDs2String(oids)
|
||
return ids, maxLine, err
|
||
}
|
||
|
||
// GetBloggerVideos 获取原创博主视频
|
||
func GetBloggerVideos(bloggerUID uint64, skip, limit int64) (videos []*VideoModel, hasNext bool, err error) {
|
||
opts := options.Find().SetSkip(skip).SetLimit(limit + 1).SetSort(bson.D{{Key: "createdAt", Value: -1}})
|
||
cond := bson.M{"publisherID": bloggerUID, "status": 1, "newsType": SP}
|
||
if err = coll(nil).Find(&videos, cond, opts); err != nil {
|
||
return
|
||
}
|
||
if len(videos) == int(limit)+1 {
|
||
hasNext = true
|
||
videos = videos[:limit]
|
||
}
|
||
return
|
||
}
|
||
|
||
func GetVideoListByIDsSortByPlayCnt(vids []ObjectID, limit int64) (videos []*VideoModel, err error) {
|
||
cond := M{"_id": bson.M{"$in": vids}}
|
||
sort := D{{Key: "playCount", Value: -1}}
|
||
opts := options.FindOptions{}
|
||
opts.SetSort(sort).SetLimit(limit)
|
||
err = coll(nil).Find(&videos, cond, &opts)
|
||
return
|
||
}
|
||
|
||
// GetVideoListByIDs 通过视频id获取视频列表
|
||
func GetVideoListByIDs(vids []ObjectID) ([]*VideoModel, error) {
|
||
if vids == nil {
|
||
vids = []ObjectID{}
|
||
}
|
||
cond := M{"_id": bson.M{"$in": vids}}
|
||
infos, _, _, _, err := getPublishedVideoList(1, uint64(len(vids)), cond, nil)
|
||
return infos, err
|
||
}
|
||
|
||
// GetVideoListByIDs 通过视频id获取视频列表
|
||
func GetVideoListByIDsAndPlayTime(vids []ObjectID, page, size uint64, playTimeType int) ([]*VideoModel, error) {
|
||
if vids == nil {
|
||
vids = []ObjectID{}
|
||
}
|
||
cond := M{"_id": bson.M{"$in": vids}}
|
||
switch playTimeType {
|
||
case 1:
|
||
cond["playTime"] = bson.M{"$gte": 60 * 10}
|
||
case 2:
|
||
cond["playTime"] = bson.M{"$lt": 60 * 10}
|
||
}
|
||
infos, _, _, _, err := getPublishedVideoList(page, size+1, cond, nil)
|
||
return infos, err
|
||
}
|
||
|
||
// GetVideoByID 根据id获取一条视频记录
|
||
func GetVideoByID(id primitive.ObjectID) (videoInfo *VideoModel, err error) {
|
||
err = coll(nil).FindOne(&videoInfo, bson.M{"_id": id})
|
||
return
|
||
}
|
||
|
||
func GetVideoListByIDsNew(vids []ObjectID) ([]*VideoModel, bool, error) {
|
||
if vids == nil {
|
||
vids = []ObjectID{}
|
||
}
|
||
cond := M{"_id": bson.M{"$in": vids}}
|
||
sort := bson.D{{Key: "reviewAt", Value: -1}, {Key: "createdAt", Value: -1}}
|
||
infos, _, _, hasNext, err := getPublishedVideoList(1, uint64(len(vids)), cond, sort)
|
||
return infos, hasNext, err
|
||
}
|
||
|
||
// GetVideoListByIDs 通过视频id获取视频列表 底层不判断视频状态, 由上次业务保证ID的正确性
|
||
func GetVideoListByIDsNoStatus(vids []ObjectID) ([]*VideoModel, error) {
|
||
if vids == nil {
|
||
vids = []ObjectID{}
|
||
}
|
||
cond := M{"_id": bson.M{"$in": vids}}
|
||
sort := D{{Key: "reviewAt", Value: -1}}
|
||
infos, _, _, err := getVideoList(1, uint64(len(vids)), cond, sort)
|
||
return infos, err
|
||
}
|
||
|
||
// GetVideoListByIDsUnPublish 通过视频id获取状态不为审核通过的视频
|
||
func GetVideoListByIDsUnPublish(vids []ObjectID) (back []*RecoVideoModel, err error) {
|
||
opts := options.Find()
|
||
opts.SetProjection(bson.M{"_id": 1})
|
||
cond := M{"_id": bson.M{"$in": vids}, "status": bson.M{"$ne": CheckPass}}
|
||
if err = coll(nil).Find(&back, cond, opts); err != nil {
|
||
log.Error(fmt.Sprintf("[METHOD-%s]==> Model %s %s fail error:%+v:", "getVideoListByIDsUnPublish", table, "Find", err))
|
||
}
|
||
return
|
||
}
|
||
|
||
// VideoMap VideoMap
|
||
func VideoMap(vids []ObjectID) (map[ObjectID]*VideoModel, error) {
|
||
filter := M{
|
||
"_id": bson.M{"$in": vids},
|
||
"status": 1, //状态，0 未审核 1通过 2审核失败 3视为免费 默认为0
|
||
}
|
||
list := make([]VideoModel, 0, len(vids))
|
||
if err := coll(nil).Find(&list, filter); err != nil {
|
||
return nil, err
|
||
}
|
||
m := make(map[ObjectID]*VideoModel)
|
||
for _, v := range list {
|
||
vid := v
|
||
m[vid.ID] = &vid
|
||
}
|
||
return m, nil
|
||
}
|
||
|
||
// SearchVideoMap 搜索
|
||
func SearchVideoMap(vids []ObjectID) (map[ObjectID]*VideoModel, []VideoModel, error) {
|
||
filter := M{
|
||
"_id": bson.M{"$in": vids},
|
||
"status": 1, //状态，0 未审核 1通过 2审核失败 3视为免费 默认为0
|
||
}
|
||
list := make([]VideoModel, 0, len(vids))
|
||
if err := coll(nil).Find(&list, filter); err != nil {
|
||
return nil, nil, err
|
||
}
|
||
m := make(map[ObjectID]*VideoModel)
|
||
for _, v := range list {
|
||
vid := v
|
||
m[vid.ID] = &vid
|
||
}
|
||
return m, list, nil
|
||
}
|
||
|
||
// GetHisWorkList 获取作品列表
|
||
func GetHisWorkList(uid uint64, sikp, limit int64, status *int, sortType string, playTimeType int) ([]*VideoModel, int64, bool, error) {
|
||
cond := bson.M{"publisherID": uid}
|
||
switch playTimeType {
|
||
case 1:
|
||
cond["newsType"] = SP
|
||
case 2:
|
||
cond["newsType"] = SHORT
|
||
case 3:
|
||
cond["newsType"] = COVER
|
||
case 4:
|
||
cond["newsType"] = PIC
|
||
default:
|
||
cond["newsType"] = bson.M{"$in": []string{SP, SHORT, COVER, PIC}}
|
||
}
|
||
if status != nil {
|
||
cond["status"] = status
|
||
}
|
||
var sort bson.D
|
||
switch sortType {
|
||
case "hot":
|
||
return GetHotList(sikp, limit, cond)
|
||
case "watch":
|
||
sort = bson.D{{Key: "playCount", Value: -1}, {Key: "reviewAt", Value: -1}}
|
||
case "like":
|
||
sort = bson.D{{Key: "likeCount", Value: -1}, {Key: "reviewAt", Value: -1}}
|
||
case "new", "":
|
||
sort = bson.D{{Key: "reviewAt", Value: -1}}
|
||
default: //兼容旧版，传入参数为其他值
|
||
sort = bson.D{{Key: "playCount", Value: -1}, {Key: "reviewAt", Value: -1}}
|
||
}
|
||
// infos, _, total, hasNext, err := getPublishedVideoList(page, size, cond, sort)
|
||
return getWorksList(sikp, limit, cond, sort)
|
||
}
|
||
|
||
// GetMyWorkList 获取我自己作品列表
|
||
func GetMyWorkList(uid uint64, sortType string, sikp, limit uint64, playTimeType int) ([]*VideoModel, int64, bool, error) {
|
||
cond := bson.M{"publisherID": uid}
|
||
switch playTimeType {
|
||
case 1:
|
||
cond["newsType"] = SP
|
||
case 2:
|
||
cond["newsType"] = SHORT
|
||
case 3:
|
||
cond["newsType"] = COVER
|
||
case 4:
|
||
cond["newsType"] = PIC
|
||
default:
|
||
cond["newsType"] = bson.M{"$in": []string{SP, SHORT, COVER, PIC}}
|
||
}
|
||
sort := bson.D{{Key: "createdAt", Value: -1}}
|
||
switch sortType {
|
||
case "hot":
|
||
sort = bson.D{{"likeCount", -1}, {"reviewAt", -1}}
|
||
case "new", "":
|
||
sort = bson.D{{Key: "reviewAt", Value: -1}}
|
||
default:
|
||
}
|
||
// infos, _, total, hasNext, err := getPublishedVideoList(page, size, cond, sort)
|
||
return getWorksList(int64(sikp), int64(limit), cond, sort)
|
||
}
|
||
|
||
// GetWorksSortList 获取排序作品---默认最多存在5个作品
|
||
func GetWorksSortList(uid uint64, status *int) ([]*VideoModel, int64, bool, error) {
|
||
cond := bson.M{"publisherID": uid, "worksSort": bson.M{"$gt": 0}}
|
||
if status != nil {
|
||
cond["status"] = status
|
||
}
|
||
sort := bson.D{{Key: "worksSort", Value: 1}}
|
||
return getWorksList(0, 5, cond, sort)
|
||
}
|
||
|
||
func GetWorksSortCount(uid uint64, status *int) (int64, error) {
|
||
cond := bson.M{"publisherID": uid, "worksSort": bson.M{"$gt": 0}}
|
||
if status != nil {
|
||
cond["status"] = status
|
||
}
|
||
// 获取总条数
|
||
return coll(nil).Count(cond)
|
||
}
|
||
|
||
func UpdateWorksSort(uid uint64, new, old int) error {
|
||
cond := bson.M{"publisherID": uid, "worksSort": old}
|
||
if _, err := coll(nil).UpdateOne(cond, bson.M{"$set": bson.M{"worksSort": new}}); err != nil {
|
||
log.Error(fmt.Sprintf("[METHOD-%s]==> Model %s %s fail error:%+v:",
|
||
"UpdateWorksSort", table, "UpdateOne", err), log.Any("cond", cond), log.Any("new", new))
|
||
return err
|
||
}
|
||
return nil
|
||
}
|
||
|
||
// GetVideoListByLocation 根据位置ID获取视频帖子列表,按照置顶，力荐，置精,播放量，审核通过时间排序
|
||
func GetVideoListByLocation(locationID primitive.ObjectID, page, limit int64) ([]*VideoModel, error) {
|
||
var out []*VideoModel
|
||
filter := bson.M{"location": locationID, "status": 1, "newsType": SP}
|
||
opts := options.Find().SetSkip((page - 1) * limit).SetLimit(limit).
|
||
SetSort(bson.D{{Key: "isChoosen", Value: -1}, {Key: "playCount", Value: -1}, {Key: "reviewAt", Value: -1}})
|
||
return out, coll(nil).Find(&out, filter, opts)
|
||
}
|
||
|
||
// GetImageListByLocation 根据位置ID获取图片帖子列表,按照置顶，力荐，置精,播放量，审核通过时间排序
|
||
func GetImageListByLocation(locationID primitive.ObjectID, page, limit int64) ([]*VideoModel, error) {
|
||
var out []*VideoModel
|
||
filter := bson.M{"location": locationID, "status": 1, "newsType": COVER}
|
||
opts := options.Find().SetSkip((page - 1) * limit).SetLimit(limit).
|
||
SetSort(bson.D{{Key: "isChoosen", Value: -1}, {Key: "playCount", Value: -1}, {Key: "reviewAt", Value: -1}})
|
||
return out, coll(nil).Find(&out, filter, opts)
|
||
}
|
||
|
||
// GetAllVideoListByMapCond 根据条件获取所有视频，不分页！
|
||
func GetAllVideoListByMapCond(cond map[string]interface{}, sort primitive.D) ([]*VideoModel, stderr.Code) {
|
||
opts := options.FindOptions{}
|
||
if sort != nil {
|
||
opts.SetSort(sort)
|
||
}
|
||
var back []*VideoModel
|
||
if err := coll(nil).Find(&back, bson.M(cond), &opts); err != nil {
|
||
log.Error(fmt.Sprintf("[METHOD-%s]==> Model %s %s fail error:%+v:", "getVideoList", table, "Find", err),
|
||
log.Any("cond", cond),
|
||
log.Any("sort", sort),
|
||
)
|
||
return nil, stderr.ErrDbQueryError
|
||
}
|
||
return back, stderr.Success
|
||
}
|
||
|
||
// GetvideoListByMapCond web端通过组合条件获取视频列表
|
||
func GetvideoListByMapCond(cond map[string]interface{}, sort primitive.D, page, size uint64) ([]*VideoModel, int64, error) {
|
||
infos, total, _, err := getVideoList(page, size, bson.M(cond), sort)
|
||
return infos, total, err
|
||
}
|
||
|
||
// IncForwardCount 转发次数增加
|
||
func IncForwardCount(inc int, ids ...ObjectID) error {
|
||
var query = bson.M{}
|
||
if len(ids) == 1 {
|
||
query["_id"] = ids[0]
|
||
} else {
|
||
query["_id"] = bson.M{"$in": ids}
|
||
}
|
||
update := bson.M{"$set": bson.M{"updatedAt": time.Now()}, "$inc": bson.M{"forwardCount": inc, "collectCount": inc}}
|
||
_, err := coll(nil).UpdateOne(query, update)
|
||
if err != nil {
|
||
log.Error(fmt.Sprintf("[METHOD-%s]==> Model %s %s fail error:%+v:", "IncLikeCount", table, "UpdateOne", err),
|
||
log.Any("ids", ids),
|
||
)
|
||
return err
|
||
}
|
||
return nil
|
||
}
|
||
|
||
// TotalWorks 用户总作品数
|
||
func TotalWorks(filter bson.M) ([]WorkCount, error) {
|
||
pip := []bson.M{
|
||
{"$match": filter}, // 过滤条件 由外部决定
|
||
{"$group": bson.M{"_id": "$publisherID", "count": bson.M{"$sum": 1}}}, // 统计用户作品总数
|
||
{"$sort": bson.M{"count": -1}}, // 按照作品数排序
|
||
//{"$project": bson.M{"uid": "$_id.publisherID", "count": 1, "_id": 0}}, // 输出结果不包含_id
|
||
}
|
||
data := make([]WorkCount, 0)
|
||
if err := coll(nil).Aggregate(&data, pip); err != nil {
|
||
return nil, fmt.Errorf("TotalWorks err: %s", err.Error())
|
||
}
|
||
return data, nil
|
||
}
|
||
|
||
// IncBatchForwardCount 转发次数增加
|
||
func IncBatchForwardCount(ids []ObjectID) error {
|
||
query := bson.M{"_id": bson.M{"$in": ids}}
|
||
update := bson.M{"$set": bson.M{"updatedAt": time.Now()}, "$inc": bson.M{"forwardCount": 1}}
|
||
_, err := coll(nil).UpdateMany(query, update)
|
||
if err != nil {
|
||
log.Error(fmt.Sprintf("[METHOD-%s]==> Model %s %s fail error:%+v:", "IncBatchForwardCount", table, "UpdateMany", err),
|
||
log.Any("ids", ids),
|
||
)
|
||
return err
|
||
}
|
||
return nil
|
||
}
|
||
|
||
// GetPayVidList GetPayVidList
|
||
func GetPayVidList(sort bson.D, skip int64, limit int64) ([]ObjectID, error) {
|
||
filter := bson.M{
|
||
"status": CheckPass, //状态，0 未审核 1通过 2审核失败 3视为免费 默认为0
|
||
"newsType": SP,
|
||
"coins": bson.M{"$gt": 0},
|
||
}
|
||
opt := &options.FindOptions{}
|
||
if len(sort) != 0 {
|
||
opt.SetSort(sort)
|
||
}
|
||
opt.SetSkip(skip)
|
||
opt.SetLimit(limit)
|
||
payVideoList := make([]VideoModel, 0, limit)
|
||
if err := coll(nil).Find(&payVideoList, filter, opt); err != nil {
|
||
return nil, err
|
||
}
|
||
vidList := make([]ObjectID, len(payVideoList))
|
||
for i, v := range payVideoList {
|
||
vidList[i] = v.ID
|
||
}
|
||
return vidList, nil
|
||
}
|
||
|
||
// GetFreeVidList 获取免费视频
|
||
func GetFreeVidList(sort bson.D, skip int64, limit int64) ([]ObjectID, error) {
|
||
filter := bson.M{
|
||
"freeArea": true,
|
||
"coins": 0,
|
||
"status": 1, //状态，0 未审核 1通过 2审核失败 3视为免费 默认为0
|
||
}
|
||
opt := &options.FindOptions{}
|
||
opt.SetSort(bson.D{{Key: "fakePlayCount", Value: -1}, {Key: "createdAt", Value: -1}})
|
||
opt.SetSkip(skip)
|
||
opt.SetLimit(limit)
|
||
list := make([]VideoModel, 0, limit)
|
||
if err := coll(nil).Find(&list, filter, opt); err != nil {
|
||
return nil, err
|
||
}
|
||
listLen := len(list)
|
||
if listLen == 0 {
|
||
return []ObjectID{}, nil
|
||
}
|
||
vidList := make([]ObjectID, listLen)
|
||
for i := range list {
|
||
vidList[i] = list[i].ID
|
||
}
|
||
return vidList, nil
|
||
}
|
||
|
||
// GetVideoInfo 通过视频id获取视频详细信息
|
||
func GetVideoInfo(videoID string) (VideoModel, error) {
|
||
v := VideoModel{}
|
||
oid, err := primitive.ObjectIDFromHex(videoID)
|
||
if err != nil {
|
||
log.Error(fmt.Sprintf("[METHOD-%s]==> Model %s %s fail error:%+v:", "GetVideoInfo", table, "ObjectIDFromHex", err),
|
||
log.Any("videoID", videoID),
|
||
)
|
||
return v, err
|
||
}
|
||
if err := coll(nil).FindOne(&v, bson.M{"_id": oid}); err != nil {
|
||
log.Error(fmt.Sprintf("[METHOD-%s]==> Model %s %s fail error:%+v:", "GetVideoInfo", table, "FindOne", err),
|
||
log.Any("videoID", videoID),
|
||
)
|
||
return v, err
|
||
}
|
||
return v, nil
|
||
}
|
||
|
||
// IsPublisher 判断是否是视频发布者
|
||
func IsPublisher(uid uint64, vid ObjectID) (bool, error) {
|
||
info, err := GetVideoInfo(vid.Hex())
|
||
if err != nil {
|
||
return false, err
|
||
}
|
||
return info.PublisherID == uid, nil
|
||
}
|
||
|
||
// GetShareCount 获取视频分享数
|
||
func GetShareCount(vid string) (int, error) {
|
||
info, err := GetVideoInfo(vid)
|
||
if err != nil {
|
||
return 0, err
|
||
}
|
||
return info.ShareCount, nil
|
||
}
|
||
|
||
// GetShareList 获取分享视频列表
|
||
func getShareList(status *int) ([]ShareInfo, error) {
|
||
filter := bson.M{"shareSort": bson.M{"$gt": 0}}
|
||
if status != nil {
|
||
filter["status"] = *status
|
||
}
|
||
opts := options.Find().SetSort(bson.D{{Key: "shareSort", Value: -1}}).SetLimit(20)
|
||
var sis []ShareInfo
|
||
return sis, coll(nil).Find(&sis, filter, opts)
|
||
}
|
||
|
||
// GetVideoTag 获取视频标签
|
||
func GetVideoTag(vid string) ([]ObjectID, error) {
|
||
info, err := GetVideoInfo(vid)
|
||
if err != nil {
|
||
return nil, err
|
||
}
|
||
return info.Tags, nil
|
||
}
|
||
|
||
// GetVideoCover 获取视频封面
|
||
func GetVideoCover(vid string) (string, error) {
|
||
info, err := GetVideoInfo(vid)
|
||
if err != nil {
|
||
return "", err
|
||
}
|
||
return info.Cover, nil
|
||
}
|
||
|
||
// FakeLikeGteLike 是否假点赞数大于真点赞数
|
||
func FakeLikeGteLike(vid string, fakeLike int) (bool, error) {
|
||
info, err := GetVideoInfo(vid)
|
||
if err != nil {
|
||
return false, err
|
||
}
|
||
return fakeLike >= info.LikeCount, nil
|
||
}
|
||
|
||
// GetVidStatus 获取视频状态
|
||
func GetVidStatus(vid string) (int, error) {
|
||
info, err := GetVideoInfo(vid)
|
||
if err != nil {
|
||
return 0, err
|
||
}
|
||
return info.Status, nil
|
||
}
|
||
|
||
// GetVideoCntByPublisher 获取某人发布的视频数
|
||
func GetVideoCntByPublisher(uid uint64, t string) (int64, error) {
|
||
total, err := coll(nil).Count(bson.M{"publisherID": uid, "newsType": t})
|
||
if err != nil {
|
||
log.Error(fmt.Sprintf("[METHOD-%s]==> Model %s %s fail error:%+v:", "GetVideoCntByPublisher", table, "Count", err),
|
||
log.Any("uid", uid),
|
||
)
|
||
return 0, err
|
||
}
|
||
return total, nil
|
||
}
|
||
|
||
// GetPublishedVideoCntByPublisher 获取某人过审发布的视频数
|
||
func GetPublishedVideoCntByPublisher(uid uint64) (int64, error) {
|
||
total, err := coll(nil).Count(bson.M{"publisherID": uid, "status": 1})
|
||
if err != nil {
|
||
log.Error(fmt.Sprintf("[METHOD-%s]==> Model %s %s fail error:%+v:", "GetPublishedVideoCntByPublisher", table, "Count", err),
|
||
log.Any("uid", uid),
|
||
)
|
||
return 0, err
|
||
}
|
||
return total, nil
|
||
}
|
||
|
||
// VideoListByTagID VideoListByTagID
|
||
func VideoListByTagID(tid ObjectID, sort bson.D, vidSkipPer int64, vidLimitPer int64) ([]VideoModel, error) {
|
||
filter := M{
|
||
"tags": M{
|
||
"$in": A{tid},
|
||
},
|
||
"status": 1, //状态，0 未审核 1通过 2审核失败 3视为免费 默认为0
|
||
}
|
||
opt := &options.FindOptions{}
|
||
if len(sort) != 0 {
|
||
opt.SetSort(sort)
|
||
}
|
||
opt.SetSkip(vidSkipPer)
|
||
opt.SetLimit(vidLimitPer)
|
||
list := make([]VideoModel, 0, vidLimitPer)
|
||
return list, coll(nil).Find(&list, filter, opt)
|
||
}
|
||
|
||
// VideoListByTagIDSort 排序
|
||
func VideoListByTagIDSort(tid ObjectID, newsType string, sort int, skip, limit uint64) (list []RecoVideoModel, err error) {
|
||
filter := M{
|
||
"tags": tid,
|
||
"newsType": newsType,
|
||
"status": 1, //状态，0 未审核 1通过 2审核失败 3视为免费 默认为0
|
||
}
|
||
sortCond := bson.D{{Key: "tagSort." + tid.Hex(), Value: -1}, {Key: "reviewAt", Value: -1}}
|
||
switch sort {
|
||
case 1:
|
||
sortCond = bson.D{{Key: "liaoBaTopSort", Value: -1}, {Key: "playCount", Value: -1}, {Key: "likeCount", Value: -1}, {Key: "reviewAt", Value: -1}}
|
||
case 2:
|
||
sortCond = bson.D{{Key: "reviewAt", Value: -1}}
|
||
case 3:
|
||
sortCond = bson.D{{Key: "likeCount", Value: -1}, {Key: "reviewAt", Value: -1}}
|
||
case 4:
|
||
sortCond = bson.D{{Key: "isChoosen", Value: -1}, {Key: "chosen", Value: -1}, {Key: "commentCount", Value: -1}, {Key: "reviewAt", Value: -1}}
|
||
case 5:
|
||
sortCond = bson.D{{Key: "reviewAt", Value: -1}}
|
||
filter["playTime"] = bson.M{"$gte": 600}
|
||
filter["newsType"] = SP
|
||
default:
|
||
}
|
||
opt := options.Find().SetSkip(int64(skip)).SetLimit(int64(limit))
|
||
opt.SetProjection(bson.M{"chosenDate": 1, "createdAt": 1}) // 默认展示_id字段
|
||
opt.SetSort(sortCond) //内嵌动态字段无法设置索引,建议不要这么设计
|
||
return list, coll(nil).Find(&list, filter, opt)
|
||
}
|
||
|
||
// VideoListByTagIDForRecommend 根据标签推荐
|
||
func VideoListByTagIDForRecommend(tid, NewsType, quality, direction string, line time.Time, vidPage, vidLimitPer uint64) ([]string, time.Time, error) {
|
||
oid, _ := primitive.ObjectIDFromHex(tid)
|
||
filter := M{"tags": oid, "newsType": NewsType, "createdAt": bson.M{"$gt": line}, "coins": 0, "status": 1} //状态，0 未审核 1通过 2审核失败 3视为免费 默认为0
|
||
if quality != "" {
|
||
filter["quality"] = quality
|
||
}
|
||
if direction != "" {
|
||
filter["direction"] = direction
|
||
}
|
||
sort := bson.D{{Key: "createdAt", Value: 1}}
|
||
ids, _, maxCreateLine, err := getDistinctVideoID(vidPage, vidLimitPer, filter, sort)
|
||
if err != nil {
|
||
return nil, maxCreateLine, err
|
||
}
|
||
vidIDs := common.ObjectIDs2String(ids)
|
||
return vidIDs, maxCreateLine, nil
|
||
}
|
||
|
||
// InsertBase 提交视频基本信息
|
||
func InsertBase(info VideoModel) (primitive.ObjectID, error) {
|
||
vid, err := coll(nil).InsertOne(&info)
|
||
if err != nil {
|
||
log.Error(fmt.Sprintf("[METHOD-%s]==> Model %s %s fail error:%+v:", "InsertBase", table, "InsertOne", err),
|
||
log.Any("info", info),
|
||
)
|
||
return primitive.NilObjectID, err
|
||
}
|
||
return vid.InsertedID.(primitive.ObjectID), nil
|
||
}
|
||
|
||
// 根据md5判断该视频是否重复
|
||
func IsExist(md5 string, size int) bool {
|
||
suc, err := coll(nil).Exists(bson.M{"md5": md5, "size": size})
|
||
if err != nil {
|
||
log.Error(fmt.Sprintf("[METHOD-%s]==> Model %s %s fail error:%+v:", "IsExist", table, "Exists", err),
|
||
log.Any("md5", md5),
|
||
log.Any("size", size),
|
||
)
|
||
return suc
|
||
}
|
||
return suc
|
||
}
|
||
|
||
// 根据sourceID 判断文件视频文件是否存在
|
||
func IsExistBySocID(sourceID string) bool {
|
||
suc, err := coll(nil).Exists(bson.M{"sourceID": sourceID})
|
||
if err != nil {
|
||
log.Error(fmt.Sprintf("[METHOD-%s]==> Model %s %s fail error:%+v:", "IsExistBySocID", table, "Exists", err),
|
||
log.Any("sourceID", sourceID),
|
||
)
|
||
return suc
|
||
}
|
||
return suc
|
||
}
|
||
|
||
// GetFreeTime 获取免费时长
|
||
func GetFreeTime(tm uint) int {
|
||
if tm < second30 {
|
||
return 0
|
||
}
|
||
if tm < minite10 {
|
||
return free10
|
||
}
|
||
if tm < minite30 {
|
||
return free60
|
||
}
|
||
if tm < minite60 {
|
||
return free120
|
||
}
|
||
return free120
|
||
}
|
||
|
||
// IncCommentCount 评论统计次数加加
|
||
func IncCommentCount(id ObjectID) error {
|
||
query := bson.M{"_id": id}
|
||
update := bson.M{"$inc": bson.M{"commentCount": 1, "fakeCommentCount": 1}}
|
||
if _, err := coll(nil).UpdateOne(query, update); err != nil {
|
||
log.Error(fmt.Sprintf("[METHOD-%s]==> Model %s %s fail error:%+v:", "IncCommentCount", table, "UpdateOne", err),
|
||
log.Any("id", id),
|
||
)
|
||
return err
|
||
}
|
||
return nil
|
||
}
|
||
|
||
// IncCommentCount 评论统计次数减少
|
||
func DecCommentCount(id ObjectID, n int64) error {
|
||
query := bson.M{"_id": id}
|
||
update := bson.M{"$inc": bson.M{"commentCount": n, "fakeCommentCount": n}}
|
||
if _, err := coll(nil).UpdateOne(query, update); err != nil {
|
||
log.Error(fmt.Sprintf("[METHOD-%s]==> Model %s %s fail error:%+v:", "DecCommentCount", table, "UpdateOne", err),
|
||
log.Any("id", id),
|
||
)
|
||
return err
|
||
}
|
||
return nil
|
||
}
|
||
|
||
// IncLikeCount 点赞统计次数加加
|
||
func IncLikeCount(id ObjectID) error {
|
||
query := bson.M{"_id": id}
|
||
update := bson.M{"$set": bson.M{"updatedAt": time.Now()}, "$inc": bson.M{"likeCount": 1, "fakeLikeCount": 1}}
|
||
if _, err := coll(nil).UpdateOne(query, update); err != nil {
|
||
log.Error(fmt.Sprintf("[METHOD-%s]==> Model %s %s fail error:%+v:", "IncLikeCount", table, "UpdateOne", err),
|
||
log.Any("id", id),
|
||
)
|
||
return err
|
||
}
|
||
return nil
|
||
}
|
||
|
||
// DecLikeCount 点赞统计次数减减
|
||
func DecLikeCount(ids ...ObjectID) error {
|
||
var query = bson.M{}
|
||
if len(ids) == 1 {
|
||
query["_id"] = ids[0]
|
||
} else {
|
||
query["_id"] = bson.M{"$in": ids}
|
||
}
|
||
update := bson.M{"$set": bson.M{"updatedAt": time.Now()}, "$inc": bson.M{"likeCount": -1, "fakeLikeCount": -1}}
|
||
if _, err := coll(nil).UpdateOne(query, update); err != nil {
|
||
log.Error(fmt.Sprintf("[METHOD-%s]==> Model %s %s fail error:%+v:", "DecLikeCount", table, "UpdateOne", err),
|
||
log.Any("ids", ids),
|
||
)
|
||
return err
|
||
}
|
||
return nil
|
||
}
|
||
|
||
// IncPlayCount 播发统计次数加加
|
||
func IncPlayCount(id ObjectID, req PlayReq, popCfg vidpopmod.VideoPopularityConfig) error {
|
||
query := bson.M{"_id": id}
|
||
vid, err := GetVideoInfo(id.Hex())
|
||
if err != nil {
|
||
log.Error("IncPlayCount GetVideoInfo", log.Any("video id", id), log.E(err))
|
||
return err
|
||
}
|
||
now := time.Now()
|
||
update := bson.M{"$set": bson.M{"updatedAt": now}, "$inc": bson.M{"playCount": 1, "fakePlayCount": 1}}
|
||
if req.Longer > vid.FreeTime {
|
||
vid.EffectivePlayCount += calcVideoEffectivePlayCount(req.Longer, int(vid.PlayTime), popCfg)
|
||
vid.PlayCount += 1
|
||
vid.FakePlayCount += 1
|
||
update = bson.M{
|
||
"$set": bson.M{
|
||
"updatedAt": now,
|
||
//"hot": calcHot(&vid, &popCfg),
|
||
},
|
||
"$inc": bson.M{
|
||
"playCount": 1,
|
||
"fakePlayCount": 1,
|
||
"effectivePlayCount": calcVideoEffectivePlayCount(req.Longer, int(vid.PlayTime), popCfg),
|
||
}}
|
||
}
|
||
if _, err = coll(nil).UpdateOne(query, update); err != nil {
|
||
log.Error(fmt.Sprintf("[METHOD-%s]==> Model %s %s fail error:%+v:", "IncPlayCount", table, "UpdateOne", err),
|
||
log.Any("id", id),
|
||
)
|
||
return err
|
||
}
|
||
return nil
|
||
}
|
||
|
||
func calcVideoEffectivePlayCount(longer, total int, popCfg vidpopmod.VideoPopularityConfig) int {
|
||
for _, v := range popCfg.PlayTimePercentage {
|
||
if total >= v.Min && total < v.Max {
|
||
eff := float64(total) * float64(v.Percentage) / 100
|
||
if float64(longer) > eff {
|
||
return 1
|
||
}
|
||
}
|
||
}
|
||
return 0
|
||
}
|
||
|
||
// IncPurchaseCount 购买次数增加
|
||
func IncPurchaseCount(id ObjectID) error {
|
||
query := bson.M{"_id": id}
|
||
update := bson.M{"$inc": bson.M{"purchaseCount": 1}}
|
||
if _, err := coll(nil).UpdateOne(query, update); err != nil {
|
||
log.Error(fmt.Sprintf("[METHOD-%s]==> Model %s %s fail error:%+v:", "IncPurchaseCount", table, "UpdateOne", err),
|
||
log.Any("id", id),
|
||
)
|
||
return err
|
||
}
|
||
return nil
|
||
}
|
||
|
||
// UpdateVideoByMapCond web通过map视频更新
|
||
func UpdateVideoByMapCond(id ObjectID, set bson.M) (VideoModel, error) {
|
||
set["updatedAt"] = time.Now()
|
||
var v VideoModel
|
||
if err := coll(nil).FindOneAndUpdate(&v, bson.M{"_id": id}, bson.M{"$set": set}); err != nil {
|
||
log.Error(fmt.Sprintf("[METHOD-%s]==> Model %s %s fail error:%+v:", "UpdateVideoByMapCond", table, "UpdateMany", err),
|
||
log.Any("id", id),
|
||
log.Any("update", set),
|
||
)
|
||
return v, err
|
||
}
|
||
return v, nil
|
||
}
|
||
|
||
func UpdateOneByID(id ObjectID, set bson.M) (res *mongo.UpdateResult, err error) {
|
||
return coll(nil).UpdateOne(bson.M{"_id": id}, bson.M{"$set": set})
|
||
}
|
||
|
||
func BatchUpdateReco(ids []ObjectID, reco bool) error {
|
||
if len(ids) == 0 {
|
||
return nil
|
||
}
|
||
query := bson.M{"_id": bson.M{"$in": ids}}
|
||
var update = bson.M{}
|
||
if reco {
|
||
update["recoWeight"] = 0
|
||
} else {
|
||
update["recoWeight"] = -1
|
||
}
|
||
_, err := coll(nil).UpdateMany(query, bson.M{"$set": update})
|
||
return err
|
||
}
|
||
|
||
// UpdateVideoBatch web批量更新视频状态
|
||
func UpdateVideoBatch(ids []ObjectID, field string, status bool) (int64, error) {
|
||
if len(ids) == 0 {
|
||
return 0, nil
|
||
}
|
||
query := bson.M{"_id": bson.M{"$in": ids}}
|
||
update := bson.M{field: status}
|
||
now := time.Now()
|
||
if field == "chosen" {
|
||
update["chosenDate"] = now
|
||
}
|
||
if field == "freeArea" {
|
||
update["freeAreaDate"] = now
|
||
update["recoWeight"] = -1
|
||
}
|
||
update["updatedAt"] = now
|
||
//if field == "liaoBaTop" && status {
|
||
// n, err := CountNum(bson.M{"liaoBaTop": true})
|
||
// if err != nil {
|
||
// return 0, err
|
||
// }
|
||
//if n >= 20 {
|
||
// return 0, errors.New("overLiaoBaLimit")
|
||
//}
|
||
//}
|
||
result, err := coll(nil).UpdateMany(query, bson.M{"$set": update})
|
||
if err != nil {
|
||
log.Error(fmt.Sprintf("[METHOD-%s]==> Model %s %s fail error:%+v:", "UpdateVideoBatch", table, "UpdateMany", err),
|
||
log.Any("ids", ids),
|
||
log.Any("field", field),
|
||
log.Any("status", status),
|
||
)
|
||
return 0, err
|
||
}
|
||
return result.ModifiedCount, nil
|
||
}
|
||
|
||
// UpdateManyVideo web批量更新视频状态
|
||
func UpdateManyVideo(ids []ObjectID, update bson.M) (int64, error) {
|
||
if len(ids) == 0 {
|
||
return 0, nil
|
||
}
|
||
query := bson.M{"_id": bson.M{"$in": ids}}
|
||
result, err := coll(nil).UpdateMany(query, bson.M{"$set": update})
|
||
if err != nil {
|
||
log.Error(fmt.Sprintf("[METHOD-%s]==> Model %s %s fail error:%+v:", "UpdateManyVideo", table, "UpdateMany", err),
|
||
log.Any("ids", ids),
|
||
)
|
||
return 0, err
|
||
}
|
||
return result.ModifiedCount, nil
|
||
}
|
||
|
||
// UpdateVideosByMapCond web通过map视频更新
|
||
func UpdateVideosByMapCond(ids []ObjectID, set bson.M) (err error) {
|
||
set["updatedAt"] = time.Now()
|
||
|
||
if _, err = coll(nil).UpdateMany(bson.M{"_id": bson.M{"$in": ids}}, bson.M{"$set": set}); err != nil {
|
||
log.Error(fmt.Sprintf("[METHOD-%s]==> Model %s %s fail error:%+v:", "UpdateVideoByMapCond", table, "UpdateMany", err),
|
||
log.Any("ids", ids),
|
||
log.Any("update", set),
|
||
)
|
||
return err
|
||
}
|
||
return nil
|
||
}
|
||
|
||
// PassVids web批量更新视频状态
|
||
func PassVidsOnlineTime(id primitive.ObjectID, pass int, reason string, manager string, onlineTime time.Time, tagSort primitive.M) (VideoModel, error) {
|
||
var vid VideoModel
|
||
query := bson.M{"_id": id}
|
||
var update bson.M
|
||
now := time.Now()
|
||
if pass == Free {
|
||
update = bson.M{"status": CheckPass, "coins": 0, "updatedAt": now, "reviewAt": onlineTime, "reviewAccount": manager}
|
||
} else if pass == CheckPass {
|
||
update = bson.M{"status": CheckPass, "updatedAt": now, "reviewAt": onlineTime, "reviewAccount": manager, "tagSort": tagSort}
|
||
} else if pass == CheckFailure {
|
||
update = bson.M{"status": CheckFailure, "reason": reason, "updatedAt": now, "reviewAt": onlineTime, "reviewAccount": manager}
|
||
}
|
||
if err := coll(nil).FindOneAndUpdate(&vid, query, bson.M{"$set": update}); err != nil {
|
||
log.Error(fmt.Sprintf("[METHOD-%s]==> Model %s %s fail error:%+v:", "PassVids", table, "FindOneAndUpdate", err),
|
||
log.Any("id", id),
|
||
log.Any("pass", pass),
|
||
log.Any("reason", reason),
|
||
)
|
||
return vid, err
|
||
}
|
||
return vid, nil
|
||
}
|
||
|
||
// PassVids web批量更新视频状态
|
||
func PassVids(id primitive.ObjectID, pass int, reason string, manager string, tagSort primitive.M) (VideoModel, error) {
|
||
var vid VideoModel
|
||
query := bson.M{"_id": id}
|
||
var update bson.M
|
||
now := time.Now()
|
||
if pass == Free {
|
||
update = bson.M{"status": CheckPass, "coins": 0, "updatedAt": now, "reviewAt": now, "reviewAccount": manager}
|
||
} else if pass == CheckPass {
|
||
update = bson.M{"status": CheckPass, "updatedAt": now, "reviewAt": now, "reviewAccount": manager, "tagSort": tagSort}
|
||
} else if pass == CheckFailure {
|
||
update = bson.M{"status": CheckFailure, "reason": reason, "updatedAt": now, "reviewAt": now, "reviewAccount": manager}
|
||
}
|
||
if err := coll(nil).FindOneAndUpdate(&vid, query, bson.M{"$set": update}); err != nil {
|
||
log.Error(fmt.Sprintf("[METHOD-%s]==> Model %s %s fail error:%+v:", "PassVids", table, "FindOneAndUpdate", err),
|
||
log.Any("id", id),
|
||
log.Any("pass", pass),
|
||
log.Any("reason", reason),
|
||
)
|
||
return vid, err
|
||
}
|
||
return vid, nil
|
||
}
|
||
|
||
// UpdateVideoResolutionPlayTime 更新视频
|
||
func UpdateVideoResolutionPlayTime(id ObjectID, update WebVideoUpdateDoc) (int64, error) {
|
||
query := bson.M{"_id": id}
|
||
doc, _ := common.ToBsonM(update)
|
||
result, err := coll(nil).UpdateOne(query, bson.M{"$set": doc})
|
||
if err != nil {
|
||
log.Error(fmt.Sprintf("[METHOD-%s]==> Model %s %s fail error:%+v:", "UpdateVideoResolutionPlayTime", table, "UpdateOne", err),
|
||
log.Any("id", id),
|
||
log.Any("update", update),
|
||
)
|
||
return 0, err
|
||
}
|
||
return result.ModifiedCount, nil
|
||
}
|
||
|
||
// sumFakePlayCount sumFakePlayCount
|
||
func sumFakePlayCount(sort D, skip, limit int64, mats ...pageopt.Matcher) (int64, error) {
|
||
videoList := make([]VideoModel, 0, limit)
|
||
opt := (&options.FindOptions{}).
|
||
SetSort(sort).
|
||
SetSkip(skip).
|
||
SetLimit(limit)
|
||
filter := pageopt.MergeM(mats)
|
||
filter["status"] = 1 //状态，0 未审核 1通过 2审核失败 3视为免费 默认为0
|
||
if err := coll(nil).Find(&videoList, filter, opt); err != nil {
|
||
return 0, err
|
||
}
|
||
total := int64(0)
|
||
for _, v := range videoList {
|
||
total += int64(v.FakePlayCount)
|
||
}
|
||
return total, nil
|
||
}
|
||
|
||
func vidList(sort D, skip, limit int64) ([]ObjectID, error) {
|
||
videoList := make([]VideoModel, 0, limit)
|
||
opt := (&options.FindOptions{}).
|
||
SetSort(sort).
|
||
SetSkip(skip).
|
||
SetLimit(limit)
|
||
filter := M{
|
||
"status": 1, //状态，0 未审核 1通过 2审核失败 3视为免费 默认为0
|
||
}
|
||
if err := coll(nil).Find(&videoList, filter, opt); err != nil {
|
||
return nil, err
|
||
}
|
||
vidList := make([]ObjectID, 0, len(videoList))
|
||
for _, video := range videoList {
|
||
vidList = append(vidList, video.ID)
|
||
}
|
||
return vidList, nil
|
||
}
|
||
|
||
// VIDListByCreatedTime VIDListByCreatedTime
|
||
func VIDListByCreatedAt(skip, limit int64) ([]ObjectID, error) {
|
||
sort := D{
|
||
{Key: "createdAt", Value: -1},
|
||
}
|
||
return vidList(sort, skip, limit)
|
||
}
|
||
|
||
func SumFakePlayCountSortByCreatedAt(limit int64) (int64, error) {
|
||
sort := D{
|
||
{Key: "createdAt", Value: -1},
|
||
}
|
||
return sumFakePlayCount(sort, 0, limit)
|
||
}
|
||
|
||
func SumFakePlayCountSortByFakeLikeCount(limit int64) (int64, error) {
|
||
sort := D{
|
||
{Key: "fakeLikeCount", Value: -1},
|
||
}
|
||
return sumFakePlayCount(sort, 0, limit)
|
||
}
|
||
|
||
// VIDListByFakePlayCount VIDListByFakePlayCount
|
||
func VIDListByFakePlayCount(skip, limit int64) ([]ObjectID, error) {
|
||
sort := D{
|
||
{Key: "fakePlayCount", Value: -1},
|
||
}
|
||
return vidList(sort, skip, limit)
|
||
}
|
||
|
||
func SumFakePlayCountSortByFakePlayCount(limit int64) (int64, error) {
|
||
sort := D{
|
||
{Key: "fakePlayCount", Value: -1},
|
||
}
|
||
return sumFakePlayCount(sort, 0, limit)
|
||
}
|
||
|
||
// VIDListByFakeLikeCount VIDListByFakeLikeCount
|
||
func VIDListByFakeLikeCount(skip, limit int64) ([]ObjectID, error) {
|
||
sort := D{
|
||
{Key: "fakeLikeCount", Value: -1},
|
||
}
|
||
return vidList(sort, skip, limit)
|
||
}
|
||
|
||
func SumFakePlayCountByFakeLikeCount(limit int64) (int64, error) {
|
||
sort := D{
|
||
{Key: "fakeLikeCount", Value: -1},
|
||
}
|
||
return sumFakePlayCount(sort, 0, limit)
|
||
}
|
||
|
||
// VIDListSortByFakeCommentCount VIDListSortByFakeCommentCount
|
||
func VIDListSortByFakeCommentCount(skip, limit int64) ([]ObjectID, error) {
|
||
sort := D{
|
||
{Key: "fakeCommentCount", Value: -1},
|
||
}
|
||
return vidList(sort, skip, limit)
|
||
}
|
||
|
||
func SumFakePlayCountSortByFakeCommentCount(limit int64) (int64, error) {
|
||
sort := D{
|
||
{Key: "fakeCommentCount", Value: -1},
|
||
}
|
||
return sumFakePlayCount(sort, 0, limit)
|
||
}
|
||
|
||
func SumFakePlayCountSortByFreeArea(limit int64) (int64, error) {
|
||
sort := D{{Key: "fakeCommentCount", Value: -1}}
|
||
freeAreaMatch := &pageopt.AssignMatch{Key: "freeArea", Val: true}
|
||
coinsMatch := &pageopt.AssignMatch{Key: "coins", Val: 0}
|
||
return sumFakePlayCount(sort, 0, limit, freeAreaMatch, coinsMatch)
|
||
}
|
||
|
||
func VIDListToneOfficialRecom(skip, limit int64) ([]primitive.ObjectID, error) {
|
||
sort := D{
|
||
{Key: "fakeCommentCount", Value: -1},
|
||
}
|
||
opt := (&options.FindOptions{}).
|
||
SetSort(sort).
|
||
SetSkip(skip).
|
||
SetLimit(limit)
|
||
filter := M{
|
||
"status": 1, //状态，0 未审核 1通过 2审核失败 3视为免费 默认为0
|
||
"chosen": true,
|
||
}
|
||
var videoList []VideoModel
|
||
if err := coll(nil).Find(&videoList, filter, opt); err != nil {
|
||
return nil, err
|
||
}
|
||
vidList := make([]ObjectID, len(videoList))
|
||
for i := range videoList {
|
||
vidList[i] = videoList[i].ID
|
||
}
|
||
return vidList, nil
|
||
}
|
||
|
||
func SumFakePlayCountSortByToneOfficialRecom(limit int64) (int64, error) {
|
||
sort := D{
|
||
{Key: "fakeCommentCount", Value: -1},
|
||
}
|
||
opt := (&options.FindOptions{}).
|
||
SetSort(sort).
|
||
SetLimit(limit)
|
||
filter := M{
|
||
"status": 1, //状态，0 未审核 1通过 2审核失败 3视为免费 默认为0
|
||
"chosen": true,
|
||
}
|
||
videoList := make([]VideoModel, 0, limit)
|
||
if err := coll(nil).Find(&videoList, filter, opt); err != nil {
|
||
return 0, err
|
||
}
|
||
total := int64(0)
|
||
for _, v := range videoList {
|
||
total += int64(v.FakePlayCount)
|
||
}
|
||
return total, nil
|
||
}
|
||
|
||
// 通过关键字或tid匹配title获取VID List
|
||
func VIDListByKeywordOrTags(keyword string, tids []ObjectID, skip int64, limit int64) ([]ObjectID, error) {
|
||
opt := (&options.FindOptions{}).
|
||
SetSkip(skip).
|
||
SetLimit(limit).
|
||
SetSort(D{{Key: "fakePlayCount", Value: -1}, {Key: "createdAt", Value: -1}}) //索引 因为业务没有明确要求顺序 排序放最后提升效率
|
||
filter := M{
|
||
"status": 1, //状态，0 未审核 1通过 2审核失败 3视为免费 默认为0
|
||
"title": M{
|
||
"$regex": fmt.Sprintf("^%s", keyword),
|
||
},
|
||
}
|
||
if len(tids) != 0 {
|
||
filter = M{
|
||
"status": 1, //状态，0 未审核 1通过 2审核失败 3视为免费 默认为0
|
||
"$or": A{
|
||
M{"title": M{
|
||
"$regex": fmt.Sprintf("^%s", keyword),
|
||
}},
|
||
M{"tags": M{"$in": tids}},
|
||
},
|
||
}
|
||
}
|
||
docList := make([]struct {
|
||
ID ObjectID `bson:"_id"`
|
||
}, 0, limit)
|
||
if err := coll(nil).Find(&docList, filter, opt); err != nil {
|
||
return nil, err
|
||
}
|
||
idList := make([]ObjectID, len(docList))
|
||
for i := range docList {
|
||
idList[i] = docList[i].ID
|
||
}
|
||
return idList, nil
|
||
}
|
||
|
||
// GetCityPlayCount2Map 获取城市的访问量和播放量(热度)
|
||
func GetCityPlayCount2Map(locIDs []ObjectID) (map[ObjectID]int, error) {
|
||
if locIDs == nil {
|
||
locIDs = []ObjectID{}
|
||
}
|
||
m := make(map[ObjectID]int)
|
||
var data []CityCount
|
||
p := []bson.M{
|
||
{"$match": bson.M{"location": bson.M{"$in": locIDs}, "status": 1}},
|
||
{"$group": bson.M{"_id": "$location", "count": bson.M{"$sum": "$playCount"}}},
|
||
}
|
||
if err := coll(nil).Aggregate(&data, p); err != nil {
|
||
log.Error(fmt.Sprintf("[METHOD-%s]==> Model %s %s fail error:%+v:", "GetCityPlayCount2Map", table, "Aggregate", err),
|
||
log.Any("locIDs", locIDs),
|
||
)
|
||
return nil, err
|
||
}
|
||
for _, d := range data {
|
||
m[d.ID] = d.Count
|
||
}
|
||
return m, nil
|
||
}
|
||
|
||
// GetCityPublishedVideoCount2Map 获取城市的视频数
|
||
func GetCityPublishedVideoCount2Map(locIDs []ObjectID) (map[ObjectID]int, error) {
|
||
if locIDs == nil {
|
||
locIDs = []ObjectID{}
|
||
}
|
||
m := make(map[ObjectID]int)
|
||
var data []CityCount
|
||
p := []bson.M{
|
||
{"$match": bson.M{"location": bson.M{"$in": locIDs}, "status": 1}},
|
||
{"$group": bson.M{"_id": "$location", "count": bson.M{"$sum": 1}}},
|
||
}
|
||
if err := coll(nil).Aggregate(&data, p); err != nil {
|
||
log.Error(fmt.Sprintf("[METHOD-%s]==> Model %s %s fail error:%+v:", "GetCityPublishedVideoCount2Map", table, "Aggregate", err),
|
||
log.Any("locIDs", locIDs),
|
||
)
|
||
return nil, err
|
||
}
|
||
for _, d := range data {
|
||
m[d.ID] = d.Count
|
||
}
|
||
return m, nil
|
||
}
|
||
|
||
// GetTitleTimeByIDs 获取视频标题
|
||
func GetTitleTimeByIDs(videoIDs []primitive.ObjectID) (map[primitive.ObjectID]TitleTime, error) {
|
||
m := make(map[primitive.ObjectID]TitleTime)
|
||
if videoIDs == nil {
|
||
videoIDs = []primitive.ObjectID{}
|
||
}
|
||
infos, err := GetVideoListByIDs(videoIDs)
|
||
if err != nil {
|
||
return m, err
|
||
}
|
||
for _, i := range infos {
|
||
info := TitleTime{
|
||
Title: i.Title,
|
||
PlayTime: i.PlayTime,
|
||
}
|
||
m[i.ID] = info
|
||
}
|
||
return m, nil
|
||
}
|
||
|
||
// uid获取视频并按播放量降序
|
||
func GetVideosByPublisherID(publisherID uint64, stdQuery commod.StdQuery) (data []VideoModel, err error) {
|
||
if err = coll(nil).Find(&data, bson.M{"publisherID": publisherID}, commod.ConvertToListQuery(stdQuery)); err != nil {
|
||
log.Error(fmt.Sprintf("[METHOD-%s]==> Model %s %s fail error:%+v:", "GetVideosByPublisherID", table, "Find", err),
|
||
log.Any("publisherID", publisherID),
|
||
)
|
||
return
|
||
}
|
||
return
|
||
}
|
||
|
||
func CountPayVidsByUIDS(ids []primitive.ObjectID) (total int64, err error) {
|
||
total, err = coll(nil).Count(bson.M{"_id": bson.M{"$in": ids}})
|
||
if err != nil {
|
||
log.Warn(fmt.Sprintf("[METHOD-%s]==> Model %s %s fail error:%+v:", "CountPayVidsByUIDS", table, "Count", err),
|
||
log.Any("ids", ids),
|
||
)
|
||
return
|
||
}
|
||
return
|
||
}
|
||
|
||
// RecommendVidoeByCity 通过城市推荐视频
|
||
func RecommendVidoeByCity(locID primitive.ObjectID, line time.Time, newsType, quality, direction string, page, size uint64) ([]string, time.Time, error) {
|
||
cond := bson.M{"location": locID, "coins": 0, "createdAt": bson.M{"$gt": line}, "newsType": newsType}
|
||
if quality != "" {
|
||
cond["quality"] = quality
|
||
}
|
||
if direction != "" {
|
||
cond["direction"] = direction
|
||
}
|
||
sort := bson.D{{Key: "createdAt", Value: 1}}
|
||
ids, _, maxLine, err := getDistinctVideoID(page, size, cond, sort)
|
||
if err != nil {
|
||
return nil, maxLine, err
|
||
}
|
||
arr := common.ObjectIDs2String(ids)
|
||
return arr, maxLine, err
|
||
}
|
||
|
||
// RecommendVidoeByHotWords 通过热词推荐视频
|
||
func RecommendVidoeByHotWords(words []string, page, size uint64) []string {
|
||
pt := make([]string, len(words))
|
||
for _, w := range words {
|
||
es, err := Search(w, int64((page-1)*size), int64(size))
|
||
if err != nil {
|
||
log.Warn(fmt.Sprintf("[METHOD-%s]==> Model %s %s fail error:%+v:", "RecommendVidoeByHotWords", table, "Search", err),
|
||
log.Any("words", words),
|
||
log.Any("page", page),
|
||
log.Any("size", size),
|
||
log.E(err))
|
||
continue
|
||
}
|
||
for _, v := range es {
|
||
pt = append(pt, v.ID.Hex())
|
||
}
|
||
}
|
||
return pt
|
||
}
|
||
|
||
// GetChosenVideoID 获取精选视频
|
||
func GetChosenVideoID(line time.Time, newsType string, page, size uint64) ([]string, time.Time, error) {
|
||
cond := bson.M{"coins": 0, "newsType": newsType}
|
||
sort := bson.D{{Key: "createdAt", Value: 1}}
|
||
ids, maxLine, err := getChosenVideoID(line, page, size, cond, sort)
|
||
if err != nil {
|
||
return []string{}, maxLine, err
|
||
}
|
||
arr := common.ObjectIDs2String(ids)
|
||
return arr, maxLine, err
|
||
}
|
||
|
||
// GetPublishedVideoListForSyncCdn 条件获取视频列表 分页获取
|
||
func GetPublishedVideoListForSyncCdn(page int, size int, lastId string) ([]*VideoModel, error) {
|
||
cond := bson.M{}
|
||
cond["status"] = 1
|
||
if lastId != "" {
|
||
oid, _ := primitive.ObjectIDFromHex(lastId)
|
||
cond["_id"] = bson.M{"$gt": oid}
|
||
}
|
||
opts := options.FindOptions{}
|
||
opts.SetSort(bson.D{{Key: "createdAt", Value: 1}})
|
||
opts.SetSkip(int64((page - 1) * size)).SetLimit(int64(size))
|
||
var back []*VideoModel
|
||
if err := coll(nil).Find(&back, cond, &opts); err != nil {
|
||
log.Error(fmt.Sprintf("[METHOD-%s]==> Model %s %s fail error:%+v:", "GetPublishedVideoListForSyncCdn", table, "Find", err),
|
||
log.Any("page", page),
|
||
log.Any("size", size),
|
||
log.Any("lastId", lastId),
|
||
)
|
||
return nil, err
|
||
}
|
||
return back, nil
|
||
}
|
||
|
||
func GetVideoListByUpdateTimeRange(start time.Time, end time.Time, page int, size int) (data []*VideoModel, hasNext bool, err error) {
|
||
var query = bson.M{
|
||
"updatedAt": bson.M{"$gte": start, "$lt": end},
|
||
}
|
||
opts := options.FindOptions{}
|
||
opts.SetSort(bson.D{{Key: "_id", Value: 1}})
|
||
opts.SetSkip(int64((page - 1) * size)).SetLimit(int64(size) + 1)
|
||
if err = coll(nil).Find(&data, query, &opts); err != nil {
|
||
log.Error(fmt.Sprintf("[METHOD-%s]==> Model %s %s fail error:%+v:", "GetVideoListByUpdateTimeRange", table, "Find", err),
|
||
log.Any("start", start),
|
||
log.Any("end", end),
|
||
)
|
||
return
|
||
}
|
||
if len(data) > size {
|
||
hasNext = true
|
||
data = data[:size]
|
||
}
|
||
|
||
return
|
||
}
|
||
|
||
func GetVideoListLtMinPlayLikeCount(minFakeLikesCnt, minFakePlayCnt, skip, limit int64) (data []*VideoModel, err error) {
|
||
filter := M{
|
||
"$or": A{
|
||
M{"fakeLikeCount": M{"$lt": minFakeLikesCnt}},
|
||
M{"fakePlayCount": M{"$lt": minFakePlayCnt}},
|
||
},
|
||
}
|
||
opts := options.Find().SetSkip(skip).SetLimit(limit)
|
||
err = coll(nil).Find(&data, filter, opts)
|
||
return
|
||
}
|
||
|
||
func GetVideoListByCreateTimeRange(start time.Time, end time.Time, maxCommentNum int) (data []*VideoModel, err error) {
|
||
var query = bson.M{
|
||
"createdAt": bson.M{"$gte": start, "$lt": end},
|
||
"commentCount": bson.M{"$lt": maxCommentNum},
|
||
"status": 1,
|
||
}
|
||
opts := options.Find().SetSkip(0).SetLimit(5000)
|
||
if err = coll(nil).Find(&data, query, opts); err != nil {
|
||
log.Error(fmt.Sprintf("[METHOD-%s]==> Model %s %s fail error:%+v:", "GetVideoListByCreateTimeRange", table, "Find", err),
|
||
log.Any("start", start),
|
||
log.Any("end", end),
|
||
)
|
||
return
|
||
}
|
||
return
|
||
}
|
||
|
||
func FindMany(mats ...Matcher) (VIDSlice, error) {
|
||
return List(nil, nil, nil, mats...)
|
||
}
|
||
|
||
// GetHisVideoCount 获取多个用户视频总数
|
||
func GetHisVideoCount(uids []uint64) (map[uint64]int, error) {
|
||
if uids == nil {
|
||
uids = []uint64{}
|
||
}
|
||
var data []UserVideoCount
|
||
p := []bson.M{
|
||
{"$match": bson.M{"publisherID": bson.M{"$in": uids}, "status": 1}},
|
||
{"$group": bson.M{"_id": "$publisherID", "count": bson.M{"$sum": 1}}},
|
||
}
|
||
if err := coll(nil).Aggregate(&data, p); err != nil {
|
||
log.Error(fmt.Sprintf("[METHOD-%s]==> Model %s %s fail error:%+v:", "GetHisVideoCount", table, "Aggregate", err),
|
||
log.Any("uids", uids),
|
||
)
|
||
return nil, err
|
||
}
|
||
m := make(map[uint64]int)
|
||
for _, d := range data {
|
||
m[d.ID] = d.Count
|
||
}
|
||
return m, nil
|
||
}
|
||
|
||
// 获取新导入的视频
|
||
func GetNewImportVideo(newUpdatedAt string, startTime, endTime time.Time, page, pageSize int64) (data []*VideoModel, err error) {
|
||
opts := options.Find().SetSkip((page - 1) * pageSize).SetLimit(pageSize).SetSort(bson.D{{Key: "createdAt", Value: 1}})
|
||
query := bson.M{}
|
||
if newUpdatedAt != "" {
|
||
query["newUpdatedAt"] = newUpdatedAt
|
||
}
|
||
if !startTime.IsZero() {
|
||
query["createdAt"] = bson.M{"$gte": startTime}
|
||
}
|
||
if !endTime.IsZero() {
|
||
query["createdAt"] = bson.M{"$lt": endTime}
|
||
}
|
||
if err = coll(nil).Find(&data, query, opts); err != nil {
|
||
log.Error(fmt.Sprintf("[METHOD-%s]==> Model %s %s fail error:%+v:", "GetNewImportVideo", table, "Find", err),
|
||
log.Any("newUpdatedAt", newUpdatedAt),
|
||
log.Any("startTime", startTime),
|
||
log.Any("endTime", endTime),
|
||
)
|
||
return
|
||
}
|
||
return
|
||
}
|
||
|
||
func UpdateManyForContentCount(m map[primitive.ObjectID]int) (updateCount int64, err error) {
|
||
models := make([]mongo.WriteModel, len(m))
|
||
i := 0
|
||
for k, v := range m {
|
||
models[i] = mongo.NewUpdateOneModel().SetFilter(bson.M{"_id": k}).SetUpdate(bson.M{"$inc": bson.M{"commentCount": v}})
|
||
i++
|
||
}
|
||
res, err := coll(nil).Bulk(models)
|
||
if err != nil {
|
||
log.Error(fmt.Sprintf("[METHOD-%s]==> Model %s %s fail error:%+v:", "UpdateMany", table, "Bulk", err),
|
||
log.Any("map", m))
|
||
return
|
||
}
|
||
updateCount = res.ModifiedCount
|
||
return
|
||
}
|
||
|
||
// DelVideo 删除视频
|
||
func DelVideo(ids []primitive.ObjectID) error {
|
||
if ids == nil {
|
||
return nil
|
||
}
|
||
if _, err := coll(nil).DeleteMany(bson.M{"_id": bson.M{"$in": ids}}); err != nil {
|
||
log.Error(fmt.Sprintf("[METHOD-%s]==> Model %s %s fail error:%+v:", "DeleteMany", table, "err", err),
|
||
log.Any("ids", ids))
|
||
return err
|
||
}
|
||
return nil
|
||
}
|
||
|
||
// GetVideosByUIDs 通过uids获取视频
|
||
func GetVideosByUIDs(ids []uint64) (map[uint64]int, map[uint64]int, error) {
|
||
mTotal := make(map[uint64]int)
|
||
mUndeal := make(map[uint64]int)
|
||
if ids == nil {
|
||
return mTotal, mUndeal, nil
|
||
}
|
||
var vInfos []VideoModel
|
||
if err := coll(nil).Find(&vInfos, bson.M{"publisherID": bson.M{"$in": ids}}); err != nil {
|
||
log.Error(fmt.Sprintf("[METHOD-%s]==> Model %s %s fail error:%+v:", "GetVideosByUIDs", table, "err", err),
|
||
log.Any("ids", ids))
|
||
return mTotal, mUndeal, err
|
||
}
|
||
for _, v := range vInfos {
|
||
if v.Status == 0 {
|
||
mUndeal[v.PublisherID]++
|
||
}
|
||
mTotal[v.PublisherID]++
|
||
}
|
||
return mTotal, mUndeal, nil
|
||
}
|
||
|
||
// getPublishedNewsList 条件获取帖子列表
|
||
func getPublishedNewsList(page, size uint64, cond bson.M, sort bson.D) ([]*VideoModel, bool, error) {
|
||
hasNext := false
|
||
cond["status"] = 1
|
||
skip := (page - 1) * size
|
||
opts := options.FindOptions{}
|
||
if sort != nil {
|
||
opts.SetSort(sort)
|
||
}
|
||
opts.SetSkip(int64(skip)).SetLimit(int64(size + 1))
|
||
var back []*VideoModel
|
||
if err := coll(nil).Find(&back, cond, &opts); err != nil {
|
||
log.Error(fmt.Sprintf("[METHOD-%s]==> Model %s %s fail error:%+v:", "getPublishedVideoList", table, "Find", err),
|
||
log.Any("page", page),
|
||
log.Any("size", size),
|
||
log.Any("cond", cond),
|
||
log.Any("sort", sort),
|
||
)
|
||
return nil, hasNext, err
|
||
}
|
||
if uint64(len(back)) > size {
|
||
hasNext = true
|
||
back = back[:size]
|
||
}
|
||
return back, hasNext, nil
|
||
}
|
||
|
||
// GetNewestNews 获取最新帖子列表
|
||
func GetNewestNews(page, size uint64, recentMinute time.Time) ([]*VideoModel, bool, error) {
|
||
cond := bson.M{"reviewAt": bson.M{"$lte": recentMinute}}
|
||
sort := bson.D{{Key: "reviewAt", Value: -1}}
|
||
return getPublishedNewsList(page, size, cond, sort)
|
||
}
|
||
|
||
// GetNewestNews 获取最新帖子列表
|
||
func GetNewestNews_old(page, size uint64) ([]*VideoModel, bool, error) {
|
||
cond := bson.M{}
|
||
sort := bson.D{{Key: "reviewAt", Value: -1}}
|
||
return getPublishedNewsList(page, size, cond, sort)
|
||
}
|
||
|
||
// GetNewestNews 获取最新帖子列表
|
||
func GetNewestShortVideo(page, size uint64) ([]*VideoModel, bool, error) {
|
||
cond := bson.M{"newsType": SP, "playTime": bson.M{"$lte": 600}}
|
||
sort := bson.D{{Key: "reviewAt", Value: -1}}
|
||
return getPublishedNewsList(page, size, cond, sort)
|
||
}
|
||
|
||
// GetList 获取帖子列表
|
||
func GetList(filter primitive.M, opts ...*options.FindOptions) (out []*VideoModel, err error) {
|
||
if err = coll(nil).Find(&out, filter, opts...); err != nil {
|
||
log.Error(fmt.Sprintf("[METHOD-%s]==> Model %s %s fail error:%+v:", "GetList", table, "Find", err),
|
||
log.Any("filter", filter),
|
||
log.Any("opts", opts),
|
||
)
|
||
return
|
||
}
|
||
return
|
||
}
|
||
|
||
// GetNewsByIDs2M 通过id获取视频信息到map
|
||
func GetNewsByIDs2M(ids []primitive.ObjectID) (map[primitive.ObjectID]*VideoModel, error) {
|
||
mInfo := make(map[primitive.ObjectID]*VideoModel)
|
||
cond := bson.M{"_id": bson.M{"$in": ids}}
|
||
infos, _, err := getPublishedNewsList(1, uint64(len(ids)), cond, nil)
|
||
if err != nil {
|
||
return mInfo, err
|
||
}
|
||
for _, v := range infos {
|
||
mInfo[v.ID] = v
|
||
}
|
||
return mInfo, err
|
||
}
|
||
|
||
// GetNewsListLocation 获取附近的帖子列表
|
||
func GetNewsListLocation(lid ObjectID, page, size uint64) ([]*VideoModel, bool, error) {
|
||
cond := bson.M{"location": lid}
|
||
sort := bson.D{{Key: "playCount", Value: -1}, {Key: "createdAt", Value: -1}}
|
||
return getPublishedNewsList(page, size, cond, sort)
|
||
}
|
||
|
||
// GetCoinsNews 获取金币专区帖子列表
|
||
func GetCoinsNews(page, size uint64, typ int) ([]*VideoModel, bool, error) {
|
||
cond := bson.M{"coins": bson.M{"$gt": 0}}
|
||
if typ == CommonUp {
|
||
cond["$or"] = bson.A{bson.M{"isMadou": bson.M{"$exists": false}}, bson.M{"isMadou": false}}
|
||
}
|
||
if typ == MadouUp {
|
||
cond["isMadou"] = true
|
||
}
|
||
sort := bson.D{{Key: "createdAt", Value: -1}}
|
||
return getPublishedNewsList(page, size, cond, sort)
|
||
}
|
||
|
||
// 获取原创新版帖子列表
|
||
func GetOriginals(uid uint64) ([]*VideoModel, error) {
|
||
cond := bson.M{"publisherID": uid, "newsType": SP}
|
||
sort := bson.D{{Key: "worksSort", Value: -1}}
|
||
infos, _, err := getPublishedNewsList(1, 3, cond, sort)
|
||
return infos, err
|
||
}
|
||
|
||
// GetTotalNewsList 统计获取帖子
|
||
func GetTotalNewsList() ([]*VideoModel, error) {
|
||
cond := bson.M{"status": 1}
|
||
var back []*VideoModel
|
||
if err := coll(nil).Find(&back, cond); err != nil {
|
||
log.Error(fmt.Sprintf("[METHOD-%s]==> Model %s %s fail error:%+v:", "getPublishedVideoList", table, "Find", err))
|
||
return back, err
|
||
}
|
||
return back, nil
|
||
}
|
||
|
||
// IsNewsCover 获取帖子类型
|
||
func IsNewsCover(id string) (bool, error) {
|
||
info, err := GetVideoInfo(id)
|
||
if err != nil {
|
||
return false, err
|
||
}
|
||
return info.NewsType == COVER || info.NewsType == SEED_LINK || info.NewsType == PIC || info.MDSID != "", nil
|
||
}
|
||
|
||
// GetTopping 获取置顶
|
||
func GetTopping(typ int, lid primitive.ObjectID, isMadou bool) ([]*VideoModel, error) {
|
||
cond := bson.M{"isTopping": true}
|
||
sort := bson.D{{Key: "reviewAt", Value: -1}}
|
||
if typ == TypeSameCity {
|
||
cond["location"] = lid
|
||
}
|
||
if typ == TypePay {
|
||
cond["coins"] = bson.M{"$gt": 0}
|
||
cond["isMadou"] = isMadou
|
||
}
|
||
infos, _, err := getPublishedNewsList(1, ToppingLimit, cond, sort)
|
||
return infos, err
|
||
}
|
||
|
||
// GetReco 获取力荐
|
||
func GetReco(typ int, lid primitive.ObjectID, isMadou bool) ([]*VideoModel, error) {
|
||
cond := bson.M{"isRecommend": true}
|
||
sort := bson.D{{Key: "reviewAt", Value: -1}}
|
||
if typ == TypeSameCity {
|
||
cond["location"] = lid
|
||
}
|
||
if typ == TypePay {
|
||
cond["coins"] = bson.M{"$gt": 0}
|
||
cond["isMadou"] = isMadou
|
||
}
|
||
infos, _, err := getPublishedNewsList(1, ToppingLimit, cond, sort)
|
||
return infos, err
|
||
}
|
||
|
||
// GetChosen 获取置精
|
||
func GetChosen(typ int, lid primitive.ObjectID, isMadou bool) ([]*VideoModel, error) {
|
||
cond := bson.M{"isChoosen": true}
|
||
sort := bson.D{{Key: "reviewAt", Value: -1}}
|
||
if typ == TypeSameCity {
|
||
cond["location"] = lid
|
||
}
|
||
if typ == TypePay {
|
||
cond["coins"] = bson.M{"$gt": 0}
|
||
cond["isMadou"] = isMadou
|
||
}
|
||
infos, _, err := getPublishedNewsList(1, ToppingLimit, cond, sort)
|
||
return infos, err
|
||
}
|
||
|
||
// IsSP 是否是短视频
|
||
func IsSP(newsType string) bool {
|
||
return newsType == SP
|
||
}
|
||
|
||
// IsCover 是否是图片帖子
|
||
func IsCover(newsType string) bool {
|
||
return newsType == COVER || newsType == PIC
|
||
}
|
||
|
||
// IncRewarded 增加打赏金额
|
||
func IncRewarded(vid primitive.ObjectID, decimal decimal.Decimal) error {
|
||
cond := bson.M{"_id": vid}
|
||
update := bson.M{"$inc": bson.M{"rewarded": decimal}}
|
||
_, err := coll(nil).UpdateOne(cond, update)
|
||
return err
|
||
}
|
||
|
||
// IncSpecFakeRewarded 增加指定的(假)打赏金额
|
||
func IncSpecFakeRewarded(vid primitive.ObjectID, decimal decimal.Decimal) error {
|
||
cond := bson.M{"_id": vid}
|
||
update := bson.M{"$inc": bson.M{"fakeRewarded": decimal}}
|
||
_, err := coll(nil).UpdateOne(cond, update)
|
||
return err
|
||
}
|
||
|
||
func GetManyPrefetchVideos(page, pageSize int64) ([]*PrefetchVideoModel, error) {
|
||
opt := options.Find().
|
||
SetLimit(pageSize).
|
||
SetSkip((page - 1) * pageSize).
|
||
SetSort(bson.M{"createdAt": 1})
|
||
var data []*PrefetchVideoModel
|
||
return data, coll(nil).Find(&data, bson.M{"status": 1}, opt)
|
||
}
|
||
|
||
func GetLiaoBaTop() ([]*VideoModel, error) {
|
||
var data []*VideoModel
|
||
opts := options.Find().SetSort(bson.D{
|
||
{Key: "liaoBaTopSort", Value: -1},
|
||
{Key: "reviewAt", Value: -1},
|
||
{Key: "_id", Value: -1},
|
||
})
|
||
return data, coll(nil).Find(&data, bson.M{"liaoBaTop": true, "status": bson.M{"$nin": []int{0, 2}}}, opts)
|
||
}
|
||
|
||
func GetVipList(page, pageSize uint64) ([]*VideoModel, bool, error) {
|
||
cond := bson.M{"coins": 0}
|
||
sort := bson.D{{Key: "liaoBaTopSort", Value: -1}, {Key: "reviewAt", Value: -1}}
|
||
return getPublishedNewsList(page, pageSize, cond, sort)
|
||
}
|
||
|
||
// FindShortVidByIDs 获取长视频
|
||
func FindShortVidByIDs(vids []primitive.ObjectID) ([]primitive.ObjectID, error) {
|
||
var data struct {
|
||
IDS []primitive.ObjectID `bson:"ids"`
|
||
}
|
||
p := []bson.M{
|
||
{"$match": bson.M{"_id": bson.M{"$in": vids}, "playTime": bson.M{"$gt": 600}}},
|
||
{"$group": bson.M{"_id": nil, "ids": bson.M{"$push": "$_id"}}},
|
||
}
|
||
if err := coll(nil).AggregateDecode(&data, p); err != nil {
|
||
log.Error(fmt.Sprintf("[METHOD-%s]==> Model %s %s fail error:%+v:", "FindShortVidByIDs", table, "Aggregate", err),
|
||
log.Any("vids", vids),
|
||
)
|
||
return nil, err
|
||
}
|
||
return data.IDS, nil
|
||
}
|
||
|
||
// GetRecommIDs 获取推荐视频
|
||
func GetRecommIDs(limit int64) ([]*VideoModel, error) {
|
||
cond := bson.M{"newsType": SP, "status": 1, "quality": bson.M{"$in": []string{Middle, High}}}
|
||
var back []*VideoModel
|
||
opt := options.Find().SetLimit(limit)
|
||
if err := coll(nil).Find(&back, cond, opt); err != nil {
|
||
log.Error(fmt.Sprintf("[METHOD-%s]==> Model %s %s fail error:%+v:", "getVideoIDsList", table, "AggregateDecode", err),
|
||
log.Any("cond", cond),
|
||
)
|
||
return nil, err
|
||
}
|
||
return back, nil
|
||
}
|
||
|
||
// GetHotVideoByOnlineTimeRange 获取时间区间上架的最热视频
|
||
func GetHotVideoByOnlineTimeRange(start time.Time, end time.Time, page, size int64) (back []*VideoModel, hasNext bool, err error) {
|
||
var query = bson.M{
|
||
"newsType": SP,
|
||
"reviewAt": bson.M{"$gte": start, "$lt": end},
|
||
"status": CheckPass,
|
||
}
|
||
opt := &options.FindOptions{}
|
||
opt.SetSort(bson.D{{Key: "playCount", Value: -1}, {Key: "reviewAt", Value: -1}}).SetSkip((page - 1) * size).SetLimit(size + 1)
|
||
|
||
if err = coll(nil).Find(&back, query, opt); err != nil {
|
||
log.Error(fmt.Sprintf("[METHOD-%s]==> Model %s %s fail error:%+v:", "GetHotVideoByOnlineTimeRange", table, "Find", err),
|
||
log.Any("start", start),
|
||
log.Any("end", end),
|
||
)
|
||
return
|
||
}
|
||
|
||
if int64(len(back)) > size {
|
||
hasNext = true
|
||
back = back[:size]
|
||
}
|
||
|
||
return
|
||
}
|
||
|
||
func IncVideoPageView(vids []primitive.ObjectID) error {
|
||
if len(vids) == 0 {
|
||
return nil
|
||
}
|
||
query := bson.M{"_id": bson.M{"$in": vids}}
|
||
update := bson.M{"$inc": bson.M{"pageViewCount": 1}}
|
||
|
||
_, err := coll(nil).UpdateMany(query, update)
|
||
if err != nil {
|
||
log.Error(fmt.Sprintf("[METHOD-%s]==> Model %s %s fail error:%+v:", "IncVideoPageView", table, "UpdateMany", err), log.Any("vids", vids))
|
||
return err
|
||
}
|
||
return nil
|
||
}
|
||
|
||
func Bulk(models []mongo.WriteModel) error {
|
||
_, err := coll(nil).Bulk(models, options.BulkWrite().SetOrdered(false))
|
||
return err
|
||
}
|
||
|
||
// GetCountByPublish 获取用户视频数量 status 0 未审核 1通过 2审核失败 3视为免费 默认为0
|
||
func GetCountByPublish(publisherID uint64, status *int) (int64, error) {
|
||
var filter = bson.M{"publisherID": publisherID}
|
||
if status != nil {
|
||
filter["status"] = status
|
||
}
|
||
return coll(nil).Count(filter)
|
||
}
|
||
|
||
// WorkLeaderboard 作品榜单
|
||
func WorkLeaderboard(bind interface{}, filter bson.M, limit int) error {
|
||
filter["publisherID"] = bson.M{"$nin": []uint64{100007, 100008}} // 排除官方上传
|
||
pip := []bson.M{
|
||
{"$match": filter}, // 过滤条件 由外部决定
|
||
{"$group": bson.M{"_id": "$publisherID", "count": bson.M{"$sum": 1}}}, // 统计用户作品总数
|
||
{"$sort": bson.M{"count": -1}}, // 按照作品数排序
|
||
{"$limit": limit}, // 限制返回条数
|
||
}
|
||
return coll(nil).Aggregate(bind, pip)
|
||
}
|
||
|
||
// GetCreatorNumber 获取创作者数量(查询非常慢,需要配合redis使用)
|
||
func GetCreatorNumber() (int64, error) {
|
||
var result []struct {
|
||
Count int64 `json:"count" bson:"count"`
|
||
}
|
||
pip := []bson.M{
|
||
{"$group": bson.M{"_id": "$publisherID"}}, // 按照创作者ID分组
|
||
{"$group": bson.M{"_id": "null", "count": bson.M{"$sum": 1}}}, // 统计创作者总数
|
||
{"$project": bson.M{"_id": 0}}, // 输出结果不包含_id
|
||
}
|
||
if err := coll(nil).Aggregate(&result, pip); err != nil {
|
||
return 0, err
|
||
}
|
||
// 容错判定
|
||
if len(result) > 0 {
|
||
return result[0].Count, nil
|
||
}
|
||
return 0, nil
|
||
}
|
||
|
||
func IsSubmitByPublisherID(publisherID uint64) (bool, error) {
|
||
return coll(nil).Exists(bson.M{"publisherID": publisherID})
|
||
}
|
||
|
||
// GetRecommendation 获取推荐帖子==>幸福广场使用
|
||
func GetRecommendation() (out []*VideoModel, err error) {
|
||
filter := bson.M{
|
||
"newsType": COVER,
|
||
"status": 1,
|
||
//"$or": bson.A{
|
||
// bson.M{"isTopping": true},
|
||
// bson.M{"isRecommend": true},
|
||
// bson.M{"isChoosen": true},
|
||
//},
|
||
}
|
||
if err = coll(nil).Find(&out, filter, options.Find().SetLimit(10)); err != nil {
|
||
log.Error(fmt.Sprintf("[METHOD-%s]==> Model %s %s fail error:%+v:", "GetRecommendation", table, "Find", err),
|
||
log.Any("filter", filter),
|
||
)
|
||
return nil, err
|
||
}
|
||
return
|
||
}
|
||
|
||
// VideoListByTagID
|
||
func VideoListByTagIDAndPlayTime(tid ObjectID, playDuration int, outVids []ObjectID) ([]*VideoModel, error) {
|
||
opt := &options.FindOptions{}
|
||
opt.SetSort(bson.D{{Key: "playCount", Value: -1}})
|
||
filter := M{
|
||
"tags": tid,
|
||
"status": 1, //状态，0 未审核 1通过 2审核失败 3视为免费 默认为0
|
||
"coins": 0,
|
||
}
|
||
if playDuration == 1 {
|
||
filter["playTime"] = bson.M{"$gte": 600}
|
||
opt.SetLimit(1)
|
||
} else if playDuration == 2 {
|
||
filter["playTime"] = bson.M{"$lt": 600}
|
||
opt.SetLimit(2)
|
||
}
|
||
if len(outVids) > 0 {
|
||
filter["_id"] = bson.M{"$nin": outVids}
|
||
}
|
||
list := make([]*VideoModel, 0)
|
||
return list, coll(nil).Find(&list, filter, opt)
|
||
}
|
||
|
||
// HotVideo 热门视频
|
||
func HotVideo() ([]*VideoModel, error) {
|
||
opt := &options.FindOptions{}
|
||
opt.SetSort(bson.D{{Key: "playCount", Value: -1}}).SetLimit(20)
|
||
filter := M{
|
||
"status": 1, //状态，0 未审核 1通过 2审核失败 3视为免费 默认为0
|
||
"coins": 0,
|
||
}
|
||
list := make([]*VideoModel, 0)
|
||
return list, coll(nil).Find(&list, filter, opt)
|
||
}
|
||
|
||
func GetVideosByIDs(ids []primitive.ObjectID) ([]*VideoModel, error) {
|
||
var out []*VideoModel
|
||
return out, coll(nil).Find(&out, bson.M{"_id": bson.M{"$in": ids}, "status": 1})
|
||
}
|
||
|
||
// GetRecommendVideosByIDsContext 获取仍满足短视频推荐条件的视频完整快照。
|
||
// 队列生成后视频可能被下架、删除或改变类型，因此请求侧必须再次校验当前状态。
|
||
func GetRecommendVideosByIDsContext(ctx context.Context, ids []primitive.ObjectID) ([]*VideoModel, error) {
|
||
if len(ids) == 0 {
|
||
return []*VideoModel{}, nil
|
||
}
|
||
var out []*VideoModel
|
||
return out, coll(mdb.ToolCtx(ctx)).Find(&out, recommendVideosByIDsFilter(ids))
|
||
}
|
||
|
||
func recommendVideosByIDsFilter(ids []primitive.ObjectID) bson.M {
|
||
return bson.M{
|
||
"_id": bson.M{"$in": ids},
|
||
"status": CheckPass,
|
||
"newsType": SHORT,
|
||
"deleteAt": nil,
|
||
"recoWeight": bson.M{"$ne": -1},
|
||
}
|
||
}
|
||
|
||
func DeleteBeforeDeleteAt(t *db.MongoTool, tm time.Time) error {
|
||
_, err := coll(t).DeleteMany(bson.M{"deletedAt": bson.M{"$lt": tm}})
|
||
return err
|
||
}
|
||
|
||
func GetVideosRecommandInVids(vids []primitive.ObjectID) ([]*VideoModel, error) {
|
||
var vms []*VideoModel
|
||
opts := options.Find().SetSort(bson.D{{Key: "effectivePlayCount", Value: -1}}).SetLimit(20)
|
||
return vms, coll(nil).Find(&vms, bson.M{"_id": bson.M{"$in": vids}, "status": 1, "reviewAt": bson.M{"$gte": time.Now().Add(time.Hour * 24 * -200)}}, opts)
|
||
}
|
||
|
||
func GetSearchRecommand() (data []*VideoModel, err error) {
|
||
//加一个搜索缓存
|
||
key := fmt.Sprintf("searchRecommend_search")
|
||
expaire := time.Minute * 2
|
||
s, err := appg.Redis.Get(key)
|
||
if err != nil {
|
||
return
|
||
}
|
||
if s != nil && *s != "" {
|
||
err = json.Unmarshal([]byte(*s), &data)
|
||
return data, err
|
||
}
|
||
if data == nil {
|
||
var rechargeVms []*VideoModel
|
||
var freeVms []*VideoModel
|
||
opts := options.Find().SetSort(bson.D{{Key: "effectivePlayCount", Value: -1}}).SetLimit(10)
|
||
tm := time.Now().Add(time.Hour * 24 * -200)
|
||
var wg sync.WaitGroup
|
||
wg.Add(2)
|
||
common.Go(func() {
|
||
defer wg.Done()
|
||
if err := coll(nil).Find(&rechargeVms, bson.M{"coins": bson.M{"$gt": 0}, "status": 1, "reviewAt": bson.M{"$gte": tm}}, opts); err != nil {
|
||
return
|
||
}
|
||
})
|
||
common.Go(func() {
|
||
defer wg.Done()
|
||
if err := coll(nil).Find(&freeVms, bson.M{"coins": 0, "status": 1, "reviewAt": bson.M{"$gte": tm}}, opts); err != nil {
|
||
return
|
||
}
|
||
})
|
||
wg.Wait()
|
||
data = append(data, rechargeVms...)
|
||
data = append(data, freeVms...)
|
||
sort.Slice(data, func(i, j int) bool {
|
||
return data[i].FakePlayCount > data[j].FakePlayCount
|
||
})
|
||
|
||
common.Go(func() {
|
||
b, _ := json.Marshal(data)
|
||
_ = appg.Redis.Set(key, string(b), expaire)
|
||
})
|
||
|
||
}
|
||
return data, nil
|
||
}
|
||
|
||
func FindOneByFilter(cond bson.M) (vidInfo *VideoModel, err error) {
|
||
err = coll(nil).FindOne(&vidInfo, cond)
|
||
return
|
||
}
|
||
|
||
// ExportFindMany 导出的时候查询所有
|
||
func ExportFindMany(filter bson.M, skip int64, size int64) ([]*VideoModel, error) {
|
||
sort := bson.D{{Key: "createdAt", Value: 1}}
|
||
opts := options.FindOptions{}
|
||
opts.SetSort(sort).SetSkip(skip).SetLimit(size)
|
||
data := make([]*VideoModel, 0)
|
||
if err := coll(nil).Find(&data, filter, &opts); err != nil {
|
||
log.ZapLog.Warn(" ExportFindMany Find fail", log.E(err))
|
||
return data, err
|
||
}
|
||
return data, nil
|
||
}
|
||
|
||
func GetShortVideoListByUpdateTimeRange(start time.Time, end time.Time, page int, size int) (data []*VideoModel, hasNext bool, err error) {
|
||
var query = bson.M{
|
||
//"newsType": SHORT, 防止改类型
|
||
"playTime": bson.M{"$lte": 300},
|
||
"updatedAt": bson.M{"$gte": start, "$lt": end},
|
||
}
|
||
opts := options.Find().SetSort(bson.D{{Key: "_id", Value: 1}}).SetSkip(int64((page - 1) * size)).SetLimit(int64(size))
|
||
data, hasNext, err = FindList(query, opts)
|
||
if err != nil {
|
||
log.Error(fmt.Sprintf("[METHOD-%s]==> Model %s %s fail error:%+v:", "GetVideoListByUpdateTimeRange", table, "Find", err),
|
||
log.Any("start", start),
|
||
log.Any("end", end),
|
||
)
|
||
return
|
||
}
|
||
return
|
||
}
|
||
|
||
//====================================================================================================
|
||
//====================================================================================================
|
||
//====================================================================================================
|
||
//====================================================================================================
|
||
//====================================================================================================
|
||
|
||
type FilterType interface {
|
||
string | primitive.ObjectID | bson.M
|
||
}
|
||
|
||
func makeFilter[T FilterType](cond T) (bson.M, error) {
|
||
var (
|
||
filter bson.M
|
||
c any = cond
|
||
)
|
||
|
||
switch vidStr := c.(type) {
|
||
case string:
|
||
oid, err := primitive.ObjectIDFromHex(vidStr)
|
||
if err != nil {
|
||
log.Error(fmt.Sprintf("[METHOD-Delete]==> Model %s %s fail error:%+v", table, "ObjectIDFromHex", err), log.Any("oid", oid))
|
||
return nil, err
|
||
}
|
||
filter = bson.M{"_id": oid}
|
||
|
||
case primitive.ObjectID:
|
||
filter = bson.M{"_id": c}
|
||
|
||
default: // bson.M
|
||
filter = c.(bson.M)
|
||
}
|
||
|
||
return filter, nil
|
||
}
|
||
|
||
// Create 创建数据
|
||
func Create(t *db.MongoTool, data ...*VideoModel) error {
|
||
r, err := coll(t).InsertMany(data)
|
||
if err != nil {
|
||
log.Error(fmt.Sprintf("[METHOD-Create]==> Model %s %s fail error:%+v", table, "InsertMany", err), log.Any("data", data))
|
||
return err
|
||
}
|
||
if len(r.InsertedIDs) == len(data) {
|
||
for i, d := range r.InsertedIDs {
|
||
data[i].ID = d.(primitive.ObjectID)
|
||
}
|
||
}
|
||
|
||
return nil
|
||
}
|
||
|
||
// DeleteOne 根据条件删除一条
|
||
func DeleteOne[T FilterType](cond T, opt ...*options.DeleteOptions) error {
|
||
filter, err := makeFilter(cond)
|
||
if err != nil {
|
||
return err
|
||
}
|
||
_, err = coll(nil).DeleteOne(filter, opt...)
|
||
if err != nil {
|
||
log.Error(fmt.Sprintf("[METHOD-Delete]==> Model %s %s fail error:%+v", table, "DeleteOne", err), log.Any("filter", filter))
|
||
}
|
||
return err
|
||
}
|
||
|
||
// DeleteMany 根据条件删除一条或者多条
|
||
func DeleteMany[T FilterType](cond T, opt ...*options.DeleteOptions) error {
|
||
filter, err := makeFilter(cond)
|
||
if err != nil {
|
||
return err
|
||
}
|
||
_, err = coll(nil).DeleteMany(filter, opt...)
|
||
if err != nil {
|
||
log.Error(fmt.Sprintf("[METHOD-Delete]==> Model %s %s fail error:%+v", table, "DeleteMany", err), log.Any("filter", filter))
|
||
}
|
||
return err
|
||
}
|
||
|
||
// FindList 条件获取列表
|
||
func FindList(filter bson.M, opts *options.FindOptions, count ...*int64) (out []*VideoModel, hasNext bool, err error) {
|
||
if opts == nil {
|
||
opts = options.Find()
|
||
}
|
||
|
||
if opts.Limit == nil {
|
||
opts.SetLimit(1000)
|
||
}
|
||
|
||
if opts.Sort == nil {
|
||
opts.SetSort(bson.D{{Key: "_id", Value: -1}})
|
||
}
|
||
|
||
// 不需要统计总条数 就不要创建count,避免无用的查询
|
||
if len(count) == 1 {
|
||
*count[0], err = coll(nil).Count(filter)
|
||
if err != nil {
|
||
|
||
return nil, false, err
|
||
}
|
||
}
|
||
|
||
limit := int(*opts.Limit)
|
||
opts.SetLimit(int64(limit + 1))
|
||
|
||
err = coll(nil).Find(&out, filter, opts)
|
||
if err != nil {
|
||
log.Error(fmt.Sprintf("[METHOD-FetchList]==> Model %s Find fail error:%+v:", table, err), log.Any("filter", filter))
|
||
return out, false, err
|
||
}
|
||
|
||
hasNext = len(out) > limit
|
||
if hasNext {
|
||
out = out[:limit]
|
||
}
|
||
|
||
return out, hasNext, nil
|
||
}
|
||
|
||
// FindOne 通过id获取详细信息
|
||
func FindOne[T FilterType](cond T, opt ...*options.FindOneOptions) (v *VideoModel, err error) {
|
||
filter, err := makeFilter(cond)
|
||
if err != nil {
|
||
return nil, err
|
||
}
|
||
err = coll(nil).FindOne(&v, filter, opt...)
|
||
if err != nil {
|
||
log.Error(fmt.Sprintf("[METHOD-FetchInfoById]==> Model %s %s fail error:%+v:", table, "FindOne", err), log.Any("filter", filter))
|
||
return v, err
|
||
}
|
||
return v, nil
|
||
}
|
||
|
||
// UpdateOne 单条更新
|
||
func UpdateOne[T FilterType](cond T, data bson.M) (err error) {
|
||
filter, err := makeFilter(cond)
|
||
if err != nil {
|
||
return err
|
||
}
|
||
|
||
_, err = coll(nil).UpdateOne(filter, data)
|
||
if err != nil {
|
||
log.Error(fmt.Sprintf("[METHOD-Update]==> Model %s %s fail error:%+v:", table, "UpdateOne", err), log.Any("filter", filter))
|
||
return err
|
||
}
|
||
return nil
|
||
}
|
||
|
||
// UpdateMany 批量更新
|
||
func UpdateMany[T FilterType](cond T, data bson.M) (err error) {
|
||
filter, err := makeFilter(cond)
|
||
if err != nil {
|
||
return err
|
||
}
|
||
_, err = coll(nil).UpdateMany(filter, data)
|
||
if err != nil {
|
||
log.Error(fmt.Sprintf("[METHOD-Update]==> Model %s %s fail error:%+v:", table, "UpdateOne", err), log.Any("filter", filter))
|
||
return err
|
||
}
|
||
return nil
|
||
}
|
||
|
||
// GetVideoListByIDsPublish 通过视频id获取状态为审核通过的视频
|
||
func GetVideoListByIDsPublish(vIds []ObjectID) (back []*VideoModel, err error) {
|
||
opts := options.Find().SetSort(bson.D{{Key: "reviewAt", Value: -1}, {Key: "createdAt", Value: -1}})
|
||
cond := M{"_id": bson.M{"$in": vIds}, "status": CheckPass}
|
||
if err = coll(nil).Find(&back, cond, opts); err != nil {
|
||
log.Error(fmt.Sprintf("[METHOD-%s]==> Model %s %s fail error:%+v:", "GetVideoListByIDsPublish", table, "Find", err))
|
||
}
|
||
return
|
||
}
|
||
|
||
// FindForReviewBatch 按 _id 游标分页扫描上架视频，用于内容审查任务
|
||
// 仅扫描 status ∈ {CheckPass(1), Free(3)}（与"已上架"语义一致）
|
||
// lastID 传 primitive.NilObjectID 表示从头开始；limit 控制每批数量
|
||
func FindForReviewBatch(lastID primitive.ObjectID, limit int64) ([]*VideoModel, error) {
|
||
cond := bson.M{"status": bson.M{"$in": []int{CheckPass, Free}}}
|
||
if !lastID.IsZero() {
|
||
cond["_id"] = bson.M{"$gt": lastID}
|
||
}
|
||
opts := options.Find().
|
||
SetSort(bson.D{{Key: "_id", Value: 1}}).
|
||
SetLimit(limit)
|
||
var list []*VideoModel
|
||
if err := coll(nil).Find(&list, cond, opts); err != nil {
|
||
return nil, err
|
||
}
|
||
return list, nil
|
||
}
|
||
|
||
// CountForReview 上架视频总数（用于任务初始化）；条件与 FindForReviewBatch 保持一致
|
||
func CountForReview() (int64, error) {
|
||
return coll(nil).Count(bson.M{"status": bson.M{"$in": []int{CheckPass, Free}}})
|
||
}
|
||
|
||
// UpdateForReview 内容审查通过后回写文本字段
|
||
// title/content/richText 任一非空则更新对应字段
|
||
func UpdateForReview(id primitive.ObjectID, title, content, richText string) error {
|
||
set := bson.M{}
|
||
if title != "" {
|
||
set["title"] = title
|
||
}
|
||
if content != "" {
|
||
set["content"] = content
|
||
}
|
||
if richText != "" {
|
||
set["richText"] = richText
|
||
}
|
||
if len(set) == 0 {
|
||
return nil
|
||
}
|
||
set["updatedAt"] = time.Now()
|
||
_, err := coll(nil).UpdateOne(bson.M{"_id": id}, bson.M{"$set": set})
|
||
return err
|
||
}
|
||
|
||
// OffShelfManyForReview 内容审查命中后批量下架视频（status -> OffShelf）
|
||
// 仅当前状态为 CheckPass/Free 的会被改动，已删除/已下架等状态保持不变
|
||
func OffShelfManyForReview(ids []primitive.ObjectID) error {
|
||
if len(ids) == 0 {
|
||
return nil
|
||
}
|
||
_, err := coll(nil).UpdateMany(
|
||
bson.M{"_id": bson.M{"$in": ids}, "status": bson.M{"$in": []int{CheckPass, Free}}},
|
||
bson.M{"$set": bson.M{"status": OffShelf, "updatedAt": time.Now()}},
|
||
)
|
||
return err
|
||
}
|