web_hghl_server/web/service/srv_search/es_playlet.go

package srv_search

import (
	"context"
	"strconv"
	"time"
	"videoweb/pub/common/log"
	"videoweb/pub/common/srvdef"
	"videoweb/pub/esmanager"
	"videoweb/pub/models/contenttagmod"
	"videoweb/pub/models/esmod"
	"videoweb/pub/models/playletcategorymod"
	"videoweb/pub/models/playletmod"
	"videoweb/pub/models/playletthememod"

	"go.mongodb.org/mongo-driver/bson"
	"go.mongodb.org/mongo-driver/mongo/options"
)

const playletESPageSize int64 = 1000

var playletESProjection = bson.D{
	{Key: playletmod.FN.Playlet.Id, Value: 1},
	{Key: playletmod.FN.Playlet.Status, Value: 1},
	{Key: playletmod.FN.Playlet.IsFeatured, Value: 1},
	{Key: playletmod.FN.Playlet.IsRecommend, Value: 1},
	{Key: playletmod.FN.Playlet.ShowOnBanner, Value: 1},
	{Key: playletmod.FN.Playlet.Banner, Value: 1},
	{Key: playletmod.FN.Playlet.OnlineAt, Value: 1},
	{Key: playletmod.FN.Playlet.CategoryIds, Value: 1},
	{Key: playletmod.FN.Playlet.ThemeIds, Value: 1},
	{Key: playletmod.FN.Playlet.Title, Value: 1},
	{Key: playletmod.FN.Playlet.Desc, Value: 1},
	{Key: playletmod.FN.Playlet.Anchor, Value: 1},
	{Key: playletmod.FN.Playlet.Cover, Value: 1},
	{Key: playletmod.FN.Playlet.CompressedCover, Value: 1},
	{Key: playletmod.FN.Playlet.TagIds, Value: 1},
	{Key: playletmod.FN.Playlet.TotalEpisode, Value: 1},
	{Key: playletmod.FN.Playlet.UpdateStatus, Value: 1},
	{Key: playletmod.FN.Playlet.CountBrowse, Value: 1},
	{Key: playletmod.FN.Playlet.VirtualCountBrowse, Value: 1},
	{Key: playletmod.FN.Playlet.HeatValue, Value: 1},
	{Key: playletmod.FN.Playlet.VirtualHeatValue, Value: 1},
	{Key: playletmod.FN.Playlet.CountCollect, Value: 1},
	{Key: playletmod.FN.Playlet.CountLike, Value: 1},
	{Key: playletmod.FN.Playlet.VirtualCountLike, Value: 1},
	{Key: playletmod.FN.Playlet.SeoInfo.Title, Value: 1},
	{Key: playletmod.FN.Playlet.SeoInfo.Description, Value: 1},
	{Key: playletmod.FN.Playlet.SeoInfo.Keyworld, Value: 1},
	{Key: playletmod.FN.Playlet.CreatedAt, Value: 1},
	{Key: playletmod.FN.Playlet.UpdatedAt, Value: 1},
}

// getPlayletEsData 查询短剧增量数据并转换为ES批量写入数据
func getPlayletEsData(ctx context.Context, minTime, endTime *time.Time, esIndex string) (esmanager.M, *srvdef.SrvErr) {
	list, err1 := getPlayletData(ctx, minTime, endTime)
	if err1 != nil {
		return nil, err1
	}
	return playletToEsData(ctx, list)
}

// SyncPlayletsToESByLatestStatus 根据短剧最新上下架状态同步ES搜索数据
func SyncPlayletsToESByLatestStatus(ctx context.Context, ids []int64) *srvdef.SrvErr {
	ids = dedupePlayletESIds(ids)
	if len(ids) == 0 {
		return nil
	}
	list, err1 := playletmod.FindManyPlaylet(ctx, bson.M{
		playletmod.FN.Playlet.Id: bson.M{"$in": ids},
	}, options.Find().SetProjection(bson.D{
		{Key: playletmod.FN.Playlet.Id, Value: 1},
		{Key: playletmod.FN.Playlet.Status, Value: 1},
	}))
	if err1 != nil {
		return err1
	}
	foundIds := make(map[int64]struct{}, len(list))
	upIds := make([]int64, 0, len(list))
	downIds := make([]int64, 0, len(ids))
	for _, item := range list {
		if item == nil {
			continue
		}
		foundIds[item.Id] = struct{}{}
		if item.Status == playletmod.StatusUp {
			upIds = append(upIds, item.Id)
		} else {
			downIds = append(downIds, item.Id)
		}
	}
	for _, id := range ids {
		if _, ok := foundIds[id]; !ok {
			downIds = append(downIds, id)
		}
	}
	if err1 := SyncPlayletStatusToES(ctx, upIds, playletmod.StatusUp); err1 != nil {
		return err1
	}
	if err1 := SyncPlayletStatusToES(ctx, downIds, playletmod.StatusDown); err1 != nil {
		return err1
	}
	return nil
}

// SyncPlayletStatusToES 根据短剧上下架状态同步ES搜索数据
func SyncPlayletStatusToES(ctx context.Context, ids []int64, status playletmod.Status) *srvdef.SrvErr {
	ids = dedupePlayletESIds(ids)
	if len(ids) == 0 {
		return nil
	}
	if esmanager.Instance == nil {
		log.Error("短剧ES同步失败，ES未初始化", log.Any("短剧ID", ids), log.Any("状态", status))
		return srvdef.New(srvdef.ErrElsOperError, "", "")
	}
	switch status {
	case playletmod.StatusUp:
		return upsertPlayletsToES(ctx, ids)
	case playletmod.StatusDown:
		return deletePlayletsFromES(ids)
	default:
		return nil
	}
}

func getPlayletData(ctx context.Context, minTime, endTime *time.Time) (playletmod.PlayletList, *srvdef.SrvErr) {
	pageNum := int64(1)
	result := make(playletmod.PlayletList, 0)
	filter := bson.M{
		playletmod.FN.Playlet.Status: playletmod.StatusUp,
	}
	if minTime != nil && endTime != nil {
		filter[playletmod.FN.Playlet.UpdatedAt] = bson.M{
			"$gte": minTime,
			"$lte": endTime,
		}
	}
	for {
		list, err1 := playletmod.FindManyPlaylet(ctx, filter, options.Find().
			SetProjection(playletESProjection).
			SetSort(bson.D{{Key: playletmod.FN.Playlet.UpdatedAt, Value: -1}}).
			SetSkip((pageNum-1)*playletESPageSize).
			SetLimit(playletESPageSize))
		if err1 != nil {
			return nil, err1
		}
		result = append(result, list...)
		if int64(len(list)) < playletESPageSize {
			break
		}
		pageNum++
	}
	return result, nil
}

func playletToEsData(ctx context.Context, list playletmod.PlayletList) (esmanager.M, *srvdef.SrvErr) {
	categoryNameMap, err1 := buildPlayletESCategoryNameMap(ctx, list)
	if err1 != nil {
		return nil, err1
	}
	themeNameMap, err1 := buildPlayletESThemeNameMap(ctx, list)
	if err1 != nil {
		return nil, err1
	}
	tagNameMap, err1 := buildPlayletESTagNameMap(ctx, list)
	if err1 != nil {
		return nil, err1
	}
	result := make(esmanager.M, len(list))
	for _, item := range list {
		if item == nil {
			continue
		}
		result[strconv.FormatInt(item.Id, 10)] = playletToESDoc(item, categoryNameMap, themeNameMap, tagNameMap)
	}
	return result, nil
}

// upsertPlayletsToES 批量写入已上架短剧ES文档
func upsertPlayletsToES(ctx context.Context, ids []int64) *srvdef.SrvErr {
	list, err1 := playletmod.FindManyPlaylet(ctx, bson.M{
		playletmod.FN.Playlet.Id:     bson.M{"$in": ids},
		playletmod.FN.Playlet.Status: playletmod.StatusUp,
	}, options.Find().SetProjection(playletESProjection))
	if err1 != nil {
		return err1
	}
	if len(list) == 0 {
		return nil
	}
	esData, err1 := playletToEsData(ctx, list)
	if err1 != nil {
		return err1
	}
	if len(esData) == 0 {
		return nil
	}
	if err1 = AddMultiDataToES(esmod.GetPlayletIndex(), esData); err1 != nil {
		log.Error("批量写入短剧ES失败", log.Any("短剧ID", ids), log.E(err1))
		return err1
	}
	log.Info("批量写入短剧ES成功", log.Any("数量", len(esData)), log.Any("短剧ID", ids))
	return nil
}

// deletePlayletsFromES 批量删除短剧ES文档
func deletePlayletsFromES(ids []int64) *srvdef.SrvErr {
	strIds := make([]string, 0, len(ids))
	for _, id := range ids {
		if id <= 0 {
			continue
		}
		strIds = append(strIds, strconv.FormatInt(id, 10))
	}
	if len(strIds) == 0 {
		return nil
	}
	if err1 := batchDelFromES(esmod.GetPlayletIndex(), strIds); err1 != nil {
		log.Error("批量删除短剧ES失败", log.Any("短剧ID", ids), log.E(err1))
		return err1
	}
	log.Info("批量删除短剧ES成功", log.Any("数量", len(strIds)), log.Any("短剧ID", ids))
	return nil
}

// playletToESDoc 转换短剧Mongo数据为ES文档
func playletToESDoc(item *playletmod.Playlet, categoryNameMap map[int64]string, themeNameMap map[int64]string, tagNameMap map[int64]string) *esmod.ESPlaylet {
	return &esmod.ESPlaylet{
		Id:            item.Id,
		Status:        item.Status,
		IsFeatured:    item.IsFeatured,
		IsRecommend:   item.IsRecommend,
		ShowOnBanner:  item.ShowOnBanner,
		Banner:        item.Banner,
		OnlineAt:      item.OnlineAt,
		CategoryIds:   item.CategoryIds,
		CategoryNames: playletESNamesByIds(item.CategoryIds, categoryNameMap),
		ThemeIds:      item.ThemeIds,
		ThemeNames:    playletESNamesByIds(item.ThemeIds, themeNameMap),
		Title:         item.Title,
		Desc:          item.Desc,
		Anchor:        item.Anchor,
		Cover:         item.DisplayCover(),
		TagIds:        item.TagIds,
		TagNames:      playletESNamesByIds(item.TagIds, tagNameMap),
		TotalEpisode:  item.TotalEpisode,
		UpdateStatus:  item.UpdateStatus,
		CountBrowse:   item.DisplayCountBrowse(),
		HeatValue:     item.DisplayHeatValue(),
		CountCollect:  item.CountCollect,
		CountLike:     item.CountLike + item.VirtualCountLike,
		SeoInfo:       item.SeoInfo,
		CreatedAt:     item.CreatedAt,
		UpdatedAt:     item.UpdatedAt,
	}
}

// buildPlayletESCategoryNameMap 批量查询短剧分类名称
func buildPlayletESCategoryNameMap(ctx context.Context, list playletmod.PlayletList) (map[int64]string, *srvdef.SrvErr) {
	ids := collectPlayletESIds(list, func(item *playletmod.Playlet) []int64 {
		return item.CategoryIds
	})
	result := make(map[int64]string, len(ids))
	if len(ids) == 0 {
		return result, nil
	}
	categories, err1 := playletcategorymod.FindManyPlayletCategory(ctx, bson.M{
		playletcategorymod.FN.PlayletCategory.Id: bson.M{"$in": ids},
	}, options.Find().SetProjection(bson.D{
		{Key: playletcategorymod.FN.PlayletCategory.Id, Value: 1},
		{Key: playletcategorymod.FN.PlayletCategory.Name, Value: 1},
	}))
	if err1 != nil {
		return nil, err1
	}
	for _, category := range categories {
		if category != nil {
			result[category.Id] = category.Name
		}
	}
	return result, nil
}

// buildPlayletESThemeNameMap 批量查询短剧题材名称
func buildPlayletESThemeNameMap(ctx context.Context, list playletmod.PlayletList) (map[int64]string, *srvdef.SrvErr) {
	ids := collectPlayletESIds(list, func(item *playletmod.Playlet) []int64 {
		return item.ThemeIds
	})
	result := make(map[int64]string, len(ids))
	if len(ids) == 0 {
		return result, nil
	}
	themes, err1 := playletthememod.FindManyPlayletTheme(ctx, bson.M{
		playletthememod.FN.PlayletTheme.Id: bson.M{"$in": ids},
	}, options.Find().SetProjection(bson.D{
		{Key: playletthememod.FN.PlayletTheme.Id, Value: 1},
		{Key: playletthememod.FN.PlayletTheme.Name, Value: 1},
	}))
	if err1 != nil {
		return nil, err1
	}
	for _, theme := range themes {
		if theme != nil {
			result[theme.Id] = theme.Name
		}
	}
	return result, nil
}

// buildPlayletESTagNameMap 批量查询短剧标签名称
func buildPlayletESTagNameMap(ctx context.Context, list playletmod.PlayletList) (map[int64]string, *srvdef.SrvErr) {
	ids := collectPlayletESIds(list, func(item *playletmod.Playlet) []int64 {
		return item.TagIds
	})
	result := make(map[int64]string, len(ids))
	if len(ids) == 0 {
		return result, nil
	}
	tags, err1 := contenttagmod.GetMapTags(ctx, ids)
	if err1 != nil {
		return nil, err1
	}
	for _, tag := range tags {
		if tag != nil {
			result[tag.Id] = tag.Name
		}
	}
	return result, nil
}

// collectPlayletESIds 汇总并去重短剧关联ID
func collectPlayletESIds(list playletmod.PlayletList, pick func(item *playletmod.Playlet) []int64) []int64 {
	result := make([]int64, 0)
	seen := make(map[int64]struct{})
	for _, item := range list {
		if item == nil {
			continue
		}
		for _, id := range pick(item) {
			if id <= 0 {
				continue
			}
			if _, ok := seen[id]; ok {
				continue
			}
			seen[id] = struct{}{}
			result = append(result, id)
		}
	}
	return result
}

func dedupePlayletESIds(ids []int64) []int64 {
	result := make([]int64, 0, len(ids))
	seen := make(map[int64]struct{}, len(ids))
	for _, id := range ids {
		if id <= 0 {
			continue
		}
		if _, ok := seen[id]; ok {
			continue
		}
		seen[id] = struct{}{}
		result = append(result, id)
	}
	return result
}

// playletESNamesByIds 按ID顺序返回ES文档名称集合
func playletESNamesByIds(ids []int64, nameMap map[int64]string) []string {
	result := make([]string, 0, len(ids))
	for _, id := range ids {
		if name := nameMap[id]; name != "" {
			result = append(result, name)
		}
	}
	return result
}