Initial commit

Co-Authored-By: Claude Opus 5.5 (1M context) <noreply@anthropic.com>

web/service/srv_search/define.go

@@ -0,0 +1,19 @@
+package srv_search
+
+import (
+	"videoweb/pub/mongos"
+
+	"go.mongodb.org/mongo-driver/bson"
+)
+
+// Param 引擎参数
+type Param struct {
+	//数据库
+	Coll *mongos.Collection
+	//过滤器
+	CollFilter bson.M
+	//跟新字段
+	CollFields []string
+	//es索引
+	EsIndex string
+}

web/service/srv_search/es.go

@@ -0,0 +1,177 @@
+package srv_search
+
+import (
+	"context"
+	"errors"
+	"fmt"
+	"strconv"
+	"time"
+	"videoweb/pub/common/log"
+	"videoweb/pub/common/srvdef"
+	"videoweb/pub/common/utils"
+	"videoweb/pub/esmanager"
+	"videoweb/pub/models/esmod"
+	"videoweb/pub/models/videoinformationmod"
+
+	"go.mongodb.org/mongo-driver/bson/primitive"
+)
+
+var syncIsRuining = false
+
+func StartES() {
+	log.Info("srv_search:Start...")
+	defer log.Info("srv_search:Start!!!")
+	// 初始化 ES 索引 mapping（已存在时报错可忽略）
+	//InitArticleESIndex()
+	//InitToolESIndex()
+	startSyncEs()
+	_, _ = skdCron.AddFunc("0 0/17 * * * ?", func() {
+		startSyncEs()
+	})
+	skdCron.Start()
+}
+
+func startSyncEs() {
+	ctx := context.Background()
+	updateEsData(ctx, esmod.GetVideoIndex(), getVideoEsData)
+}
+
+func StopES() {
+	log.Info("srv_search:Stop...")
+	defer log.Info("srv_search:Stop!!!")
+}
+
+func updateEsData(ctx context.Context, esIndex string, fn func(ctx context.Context, startTime, endTime *time.Time, esIndex string) (esmanager.M, *srvdef.SrvErr)) {
+	log.Info("srv_search:do...", log.Any("ESIndex", esIndex))
+	defer log.Info("srv_search:do end", log.Any("EsIndex", esIndex))
+	if fn == nil {
+		return
+	}
+	esRecord, serr := esmod.GetRecordTimeByESIndex(ctx, esIndex)
+	if serr != nil {
+		log.Error(fmt.Sprintf("srv_search:startEngine GetRecordByESIndex failed! Index:%v,err:%+v:", esIndex, serr))
+		return
+	}
+	isNew := false
+	if esRecord == nil {
+		isNew = true
+		esRecord = &esmod.ESRecord{
+			Id:         primitive.NewObjectID(),
+			EsIndex:    esIndex,
+			RecordTime: time.Time{},
+			Timestamp:  esmod.ESTimestamp{},
+			UpdatedAt:  time.Time{},
+			CreatedAt:  time.Now(),
+		}
+	}
+	var now = time.Now()
+	var esData esmanager.M
+	// 跨天凌晨5点重置所有数据
+	if isNew || ((now.Hour() >= 5 && esRecord.UpdatedAt.Local().Hour() < 5) || esRecord.UpdatedAt.Local().Day() < now.Day()) {
+		esData, serr = fn(ctx, nil, nil, esIndex)
+		if serr != nil {
+			log.Warn("updateEsData fn return err", log.Any("index", esIndex), log.E(serr))
+			return
+		}
+		_ = delAllFromES(esIndex)
+		isUpdate, err := commit(esData, esIndex, time.Now())
+		if err != nil {
+			log.Error("Elastic sync video data RunSync commit err! ", log.E(err))
+			return
+		}
+
+		if isUpdate {
+			// 有成功过就算成功
+			if err := esmod.SetOneEsRecordTime(nil, esIndex, now); err != nil {
+				return
+			}
+		}
+	} else {
+		timestamp := esRecord.RecordTime
+		var recentMinute = utils.RecentMinute(now, esmod.FiveMinuteScale)
+		for recentMinute.After(timestamp) {
+			var end = timestamp.Add(esmod.FiveMinuteScale * time.Minute)
+			if end.After(recentMinute) {
+				end = recentMinute
+			}
+			esData, serr = fn(ctx, &timestamp, &end, esIndex)
+			if serr != nil {
+				log.Warn("updateEsData fn return err", log.Any("index", esIndex), log.E(serr))
+				return
+			}
+			if esData != nil && len(esData) > 0 {
+				isUpdate, err := commit(esData, esIndex, time.Now())
+				if err != nil {
+					log.Error("Elastic sync video data RunSync commit err! ", log.E(err))
+					return
+				}
+				if isUpdate {
+					// 有成功过就算成功
+					if err := esmod.SetOneEsRecordTime(nil, esIndex, now); err != nil {
+						return
+					}
+				}
+			}
+			timestamp = end
+		}
+	}
+}
+
+func commit(data esmanager.M, esIndex string, syncTime time.Time) (bool, error) {
+	bulkData := handleBulkCountLimit(data)
+	isHaveSuccess := false
+	if bulkData != nil && len(bulkData) > 0 {
+		for _, doc := range bulkData {
+			if err := esmanager.Instance.Bulk(esIndex, doc); err != nil {
+				log.Warn(fmt.Sprintf("Elastic sync commit video data SyncDataToES failed!, err:%+v\n", err))
+				// 出错继续
+			} else {
+				isHaveSuccess = true
+			}
+		}
+	}
+	if isHaveSuccess {
+		return isHaveSuccess, nil
+	} else {
+		return isHaveSuccess, errors.New("Elastic sync commit video data UpsertOneTrans failed!")
+	}
+}
+
+func handleBulkCountLimit(data esmanager.M) []esmanager.M {
+	if data == nil || len(data) == 0 {
+		return make([]esmanager.M, 0)
+	}
+	rsl := make([]esmanager.M, 0)
+	rsl = append(rsl, make(esmanager.M, batchSize))
+	index := 0
+	for k, v := range data {
+		rsl[index][k] = v
+		if len(rsl[index]) >= batchSize {
+			index++
+			rsl = append(rsl, make(esmanager.M, batchSize))
+		}
+	}
+	return rsl
+}
+
+// DelVideoFromES 从 ES 实时删除单条视频（视频删除/下架时调用）
+func DelVideoFromES(id int64) {
+	if esmanager.Instance == nil {
+		return
+	}
+	strId := strconv.FormatInt(id, 10)
+	if err := batchDelFromES(esmod.GetVideoIndex(), []string{strId}); err != nil {
+		log.Warn("srv_search:DelVideoFromES failed", log.Any("id", id), log.E(err))
+	}
+}
+
+// UpsertVideoToES 将单条视频实时写入/更新 ES（视频上架时调用）
+func UpsertVideoToES(info *videoinformationmod.VideoInformation) {
+	if esmanager.Instance == nil {
+		return
+	}
+	esData := videoToEsData([]*videoinformationmod.VideoInformation{info})
+	if err := AddMultiDataToES(esmod.GetVideoIndex(), esData); err != nil {
+		log.Warn("srv_search:UpsertVideoToES failed", log.Any("id", info.Id), log.E(err))
+	}
+}

web/service/srv_search/es_playlet.go

@@ -0,0 +1,247 @@
+package srv_search
+
+import (
+	"context"
+	"strconv"
+	"videoweb/pub/common/log"
+	"videoweb/pub/common/srvdef"
+	"videoweb/pub/esmanager"
+	"videoweb/pub/models/esmod"
+	"videoweb/pub/models/playletmod"
+	"videoweb/pub/models/playlettagmod"
+	"videoweb/pub/models/playletthememod"
+
+	"go.mongodb.org/mongo-driver/bson"
+	"go.mongodb.org/mongo-driver/mongo/options"
+)
+
+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.OperateType, 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.ThemeIds, Value: 1},
+	{Key: playletmod.FN.Playlet.Title, Value: 1},
+	{Key: playletmod.FN.Playlet.PlayName, Value: 1},
+	{Key: playletmod.FN.Playlet.Description, Value: 1},
+	{Key: playletmod.FN.Playlet.Slug, 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.HeatValue, 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.CreatedAt, Value: 1},
+	{Key: playletmod.FN.Playlet.UpdatedAt, Value: 1},
+	{Key: "seoInfo", Value: 1},
+}
+
+// SyncPlayletStatusToES 根据短剧上下架状态同步ES搜索数据
+func SyncPlayletStatusToES(ctx context.Context, ids []int64, status playletmod.Status) *srvdef.SrvErr {
+	// ES只保留已上架短剧；上架写入/更新索引，下架从索引删除。
+	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
+	}
+}
+
+// upsertPlayletsToES 批量写入已上架短剧ES文档
+func upsertPlayletsToES(ctx context.Context, ids []int64) *srvdef.SrvErr {
+	// 只查询已上架数据，避免并发状态变化后把已下架短剧重新写进ES。
+	list, err := playletmod.FindManyPlaylet(ctx, bson.M{
+		playletmod.FN.Playlet.Id:     bson.M{"$in": ids},
+		playletmod.FN.Playlet.Status: playletmod.StatusUp,
+	}, options.Find().SetProjection(playletESProjection))
+	if err != nil {
+		return err
+	}
+	if len(list) == 0 {
+		return nil
+	}
+	themeNameMap, err := buildPlayletESThemeNameMap(ctx, list)
+	if err != nil {
+		return err
+	}
+	tagNameMap, err := buildPlayletESTagNameMap(ctx, list)
+	if err != nil {
+		return err
+	}
+	esData := make(esmanager.M, len(list))
+	for _, item := range list {
+		if item == nil {
+			continue
+		}
+		esData[strconv.FormatInt(item.Id, 10)] = playletToESDoc(item, themeNameMap, tagNameMap)
+	}
+	if len(esData) == 0 {
+		return nil
+	}
+	if err = AddMultiDataToES(esmod.GetPlayletIndex(), esData); err != nil {
+		log.Error("批量写入短剧ES失败", log.Any("短剧ID", ids), log.E(err))
+		return err
+	}
+	log.Info("批量写入短剧ES成功", log.Any("数量", len(esData)), log.Any("短剧ID", ids))
+	return nil
+}
+
+// deletePlayletsFromES 批量删除短剧ES文档
+func deletePlayletsFromES(ids []int64) *srvdef.SrvErr {
+	// ES文档ID和Mongo短剧业务ID保持一致，可以直接按ID批量删除。
+	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 err := batchDelFromES(esmod.GetPlayletIndex(), strIds); err != nil {
+		log.Error("批量删除短剧ES失败", log.Any("短剧ID", ids), log.E(err))
+		return err
+	}
+	log.Info("批量删除短剧ES成功", log.Any("数量", len(strIds)), log.Any("短剧ID", ids))
+	return nil
+}
+
+// playletToESDoc 转换短剧Mongo数据为ES文档
+func playletToESDoc(item *playletmod.Playlet, themeNameMap map[int64]string, tagNameMap map[int64]string) *esmod.ESPlaylet {
+	// ES文档保留题材名和标签名，app搜索可直接命中剧名、题材名、简介和标签名。
+	return &esmod.ESPlaylet{
+		Id:           item.Id,
+		Status:       item.Status,
+		IsFeatured:   item.IsFeatured,
+		IsRecommend:  item.IsRecommend,
+		OperateType:  playletmod.NormalizeOperateType(item.OperateType),
+		ShowOnBanner: item.ShowOnBanner,
+		Banner:       item.Banner,
+		OnlineAt:     item.OnlineAt,
+		ThemeIds:     item.ThemeIds,
+		ThemeNames:   playletESNamesByIds(item.ThemeIds, themeNameMap),
+		Title:        item.Title,
+		PlayName:     item.PlayName,
+		Description:  item.Description,
+		Slug:         item.Slug,
+		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.CountBrowse,
+		HeatValue:    item.HeatValue,
+		CountCollect: item.CountCollect,
+		CountLike:    item.CountLike + item.VirtualCountLike,
+		CreatedAt:    item.CreatedAt,
+		UpdatedAt:    item.UpdatedAt,
+		SeoInfo:      item.SeoInfo,
+	}
+}
+
+// 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, err := 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 err != nil {
+		return nil, err
+	}
+	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, err := playlettagmod.FindManyPlayletTag(ctx, bson.M{
+		playlettagmod.FN.PlayletTag.Id: bson.M{"$in": ids},
+	}, options.Find().SetProjection(bson.D{
+		{Key: playlettagmod.FN.PlayletTag.Id, Value: 1},
+		{Key: playlettagmod.FN.PlayletTag.Name, Value: 1},
+	}))
+	if err != nil {
+		return nil, err
+	}
+	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
+}
+
+// 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
+}

web/service/srv_search/esmedia.go

@@ -0,0 +1,97 @@
+package srv_search
+
+import (
+	"context"
+	"strconv"
+	"strings"
+	"time"
+	"videoweb/pub/common/srvdef"
+	"videoweb/pub/esmanager"
+	"videoweb/pub/models/esmod"
+	"videoweb/pub/models/videoinformationmod"
+
+	"go.mongodb.org/mongo-driver/bson"
+	"go.mongodb.org/mongo-driver/mongo/options"
+)
+
+func getVideoEsData(ctx context.Context, minTime, endTime *time.Time, esIndex string) (esmanager.M, *srvdef.SrvErr) {
+	data, serr := getVideoData(ctx, minTime, endTime)
+	if serr != nil {
+		return nil, serr
+	}
+	return videoToEsData(data), nil
+}
+func getVideoData(ctx context.Context, minTime, endTime *time.Time) ([]*videoinformationmod.VideoInformation, *srvdef.SrvErr) {
+	pageNumber := 1
+	pageSize := 1000
+	rsl := make([]*videoinformationmod.VideoInformation, 0)
+	filter := bson.M{
+		videoinformationmod.FN.VideoInformation.MediaType: bson.M{
+			"$in": []interface{}{
+				videoinformationmod.MediaType_Video,
+				videoinformationmod.MediaType_VideoV2,
+			},
+		},
+	}
+	if minTime != nil && endTime != nil {
+		filter[videoinformationmod.FN.VideoInformation.UpdatedAt] = bson.M{
+			"$gte": minTime,
+			"$lte": endTime,
+		}
+	}
+	for {
+		opts := &options.FindOptions{}
+		opts.SetSort(bson.M{videoinformationmod.FN.VideoInformation.UpdatedAt: -1})
+		opts.SetSkip(int64(pageSize * (pageNumber - 1)))
+		opts.SetLimit(int64(pageSize))
+		data, serr := videoinformationmod.FindManyVideoInformations(ctx, filter, opts)
+		if serr != nil {
+			return nil, serr
+		}
+		rsl = append(rsl, data...)
+		if len(data) < pageSize {
+			break
+		}
+		pageNumber++
+	}
+	return rsl, nil
+}
+func videoToEsData(data []*videoinformationmod.VideoInformation) esmanager.M {
+	rsl := make(map[string]interface{}, len(data))
+	for _, v := range data {
+		esInfo := esmod.ESVideo{
+			Id:          v.Id,
+			Title:       v.Title,
+			MediaType:   v.MediaType,
+			AddedTime:   v.AddedTime,
+			CreatorName: v.CreatorName,
+			Keyworlds:   "",
+			Desc:        v.Desc,
+			Remark:      v.Remark,
+			Status:      v.Status,
+			IsHot:       v.IsHot,
+			Sort:        v.Sort,
+			WatchTimes:  v.Real.WatchTimes,
+			ShowTimes:   v.Real.ShowTimes,
+			Comments:    v.Real.Comments,
+			UpdatedAt:   v.UpdatedAt,
+			CreatedAt:   v.CreatedAt,
+			Keyworld:    v.Keyworld,
+			CreatorId:   v.CreatorId,
+			FirstTag:    v.FirstTag,
+			SecondTags:  v.SecondTags,
+			CoverImg:    v.CoverImg,
+			Nodes:       v.Nodes,
+			Video:       v.Video,
+			CategoryIds: v.CategoryIds,
+			Tags:        v.Tags,
+			Real:        v.Real,
+			VideoCount:  v.VideoCount,
+		}
+		if v.Keyworld != nil && len(v.Keyworld) > 0 {
+			esInfo.Keyworlds = strings.Join(v.Keyworld, " ")
+		}
+		rsl[strconv.FormatInt(v.Id, 10)] = esInfo
+	}
+	return rsl
+}

web/service/srv_search/keyword.go

@@ -0,0 +1,71 @@
+package srv_search
+
+import (
+	"regexp"
+	"strings"
+)
+
+// 分隔符：空格、英文逗号、中文逗号、顿号、斜杠、竖线
+var kwSplitRe = regexp.MustCompile(`[\s,，、/|]+`)
+
+// ParseKeywords 解析搜索字符串为关键词列表。
+// 规则：去除首尾空格，转小写，按空格/逗号/顿号/斜杠/竖线分割，过滤空串。
+func ParseKeywords(q string) []string {
+	q = strings.TrimSpace(q)
+	if q == "" {
+		return nil
+	}
+	parts := kwSplitRe.Split(strings.ToLower(q), -1)
+	result := make([]string, 0, len(parts))
+	for _, p := range parts {
+		if p = strings.TrimSpace(p); p != "" {
+			result = append(result, p)
+		}
+	}
+	return result
+}
+
+// buildMultiKeywordQuery 构建多关键词 ES bool should 查询（关键词之间 OR 关系）。
+// 每个关键词同时生成两个子句：
+//   - 精确 multi_match（权重更高，boost=2）
+//   - 模糊 multi_match（fuzziness=AUTO，prefix_length=1），覆盖拼写错误/多余字符等场景
+//
+// 示例：ChatGPTT→ChatGPT(1次编辑)、midjorny→midjourney(2次编辑)、数智人→数字人(1次编辑)
+func buildMultiKeywordQuery(keywords []string, fields []string, from, size int) map[string]interface{} {
+	shouldClauses := make([]map[string]interface{}, 0, len(keywords)*2)
+	for _, kw := range keywords {
+		// 精确匹配，分数更高
+		shouldClauses = append(shouldClauses, map[string]interface{}{
+			"multi_match": map[string]interface{}{
+				"query":  kw,
+				"fields": fields,
+				"type":   "best_fields",
+				"boost":  2,
+			},
+		})
+		// 模糊匹配：AUTO fuzziness（长度≥6时允许2次编辑），prefix_length=1 避免过度误召回
+		shouldClauses = append(shouldClauses, map[string]interface{}{
+			"multi_match": map[string]interface{}{
+				"query":         kw,
+				"fields":        fields,
+				"type":          "best_fields",
+				"fuzziness":     "AUTO",
+				"prefix_length": 1,
+				"fuzzy_rewrite": "constant_score_blended",
+			},
+		})
+	}
+	return map[string]interface{}{
+		"from": from,
+		"size": size,
+		"query": map[string]interface{}{
+			"bool": map[string]interface{}{
+				"should":               shouldClauses,
+				"minimum_should_match": 1,
+			},
+		},
+		"sort": []map[string]interface{}{
+			{"_score": map[string]interface{}{"order": "desc"}},
+		},
+	}
+}

web/service/srv_search/search.go

@@ -0,0 +1,27 @@
+package srv_search
+
+import (
+	"github.com/robfig/cron/v3"
+	"videoweb/pub/common/log"
+)
+
+var (
+	skdCron   *cron.Cron
+	batchSize int   = 100
+	opMaxDiff int32 = 10
+)
+
+func Start() {
+	skdCron = cron.New(cron.WithSeconds())
+	log.Info("srv_search:Start...")
+	defer log.Info("srv_search:Start!!!")
+	go func() {
+		StartES()
+	}()
+}
+
+func Stop() {
+	log.Info("srv_search:Stop...")
+	defer log.Info("srv_search:Stop!!!")
+	StopES()
+}

web/service/srv_search/seo.go

@@ -0,0 +1,15 @@
+package srv_search
+
+import (
+	"strings"
+	"videoweb/pub/pubdef"
+)
+
+func flattenSeoInfo(seo pubdef.SeoValue) (title, keywords, desc string) {
+	title = seo.Title
+	desc = seo.Description
+	if len(seo.Keyworld) > 0 {
+		keywords = strings.Join(seo.Keyworld, " ")
+	}
+	return
+}

web/service/srv_search/sync_common.go

@@ -0,0 +1,47 @@
+package srv_search
+
+import (
+	"context"
+	"time"
+	"videoweb/pub/esmanager"
+	"videoweb/pub/models/esmod"
+)
+
+const esSyncOverlap = time.Minute
+
+// needFullESSync 判断是否需要全量同步 ES
+func needFullESSync(ctx context.Context, esIndex string) (bool, *esmod.ESRecord) {
+	esRecord, serr := esmod.GetRecordTimeByESIndex(ctx, esIndex)
+	if serr != nil || esRecord == nil {
+		return true, esRecord
+	}
+
+	count, err := esmanager.Instance.Count(esIndex, esmanager.M{
+		"query": esmanager.M{"match_all": esmanager.M{}},
+	})
+	if err != nil || count == 0 {
+		return true, esRecord
+	}
+
+	now := time.Now()
+	if (now.Hour() >= 5 && esRecord.UpdatedAt.Local().Hour() < 5) ||
+		esRecord.UpdatedAt.Local().Day() < now.Day() {
+		return true, esRecord
+	}
+	return false, esRecord
+}
+
+// esSyncSinceTime 取增量同步起点（RecordTime 优先，回退 1 分钟避免边界遗漏）
+func esSyncSinceTime(esRecord *esmod.ESRecord) time.Time {
+	if esRecord == nil {
+		return time.Time{}
+	}
+	since := esRecord.RecordTime
+	if since.IsZero() {
+		since = esRecord.UpdatedAt
+	}
+	if since.IsZero() {
+		return time.Time{}
+	}
+	return since.Add(-esSyncOverlap)
+}

web/service/srv_search/updater.go

@@ -0,0 +1,223 @@
+package srv_search
+
+import (
+	"context"
+	"fmt"
+	"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"
+	"strconv"
+	"videoweb/pub/common/log"
+	"videoweb/pub/common/srvdef"
+	"videoweb/pub/common/utils"
+	"videoweb/pub/esmanager"
+	"videoweb/pub/models/esmod"
+	"videoweb/pub/mongos"
+)
+
+// resetAllDocsToES 重设所有文档到数据库
+func resetAllDocsToES(ctx context.Context, param *Param) *srvdef.SrvErr {
+	log.Info("srv_search:resetAllDocsToES...", log.Any("ESIndex", param.EsIndex))
+	defer log.Info("srv_search:startActress!!!")
+	//删除所有记录
+	delAllFromES(param.EsIndex)
+
+	var pageNum int64 = 1
+	var pageSize int64 = 200
+	updSize := 0
+	for {
+		findOpt := constructFindOption(pageNum, pageSize)
+		docs, serr := findManyFromDb(ctx, param, findOpt)
+		if serr != nil {
+			return serr
+		}
+		if len(docs) == 0 {
+			break
+		}
+		updSize = updSize + len(docs)
+		updateESData, serr := sliceMapToJsonMap(docs, param)
+		log.Info("srv_search:resetAllDocsToES", log.Any("updateESData", updateESData))
+		if serr = batchAddToES(param.EsIndex, updateESData); serr != nil {
+			return serr
+		}
+		pageNum++
+		log.Info("resetAllToES>>>>>>>>>>>>>>>>>>", log.Any("updateSize:", updSize))
+	}
+	return nil
+}
+
+// constructMatch 构建match
+func constructMatch(param *Param) bson.M {
+	updateFilter := bson.A{}
+	for _, fieldName := range param.CollFields {
+		filter := bson.M{"updateDescription.updatedFields." + fieldName: bson.M{"$exists": true}}
+		updateFilter = append(updateFilter, filter)
+	}
+	match := bson.M{"$or": bson.A{
+		bson.M{"operationType": bson.M{"$in": []string{"delete", "insert", "replace"}}},
+		bson.M{"$and": bson.A{
+			bson.M{"operationType": bson.M{"$in": []string{"update"}}},
+			bson.M{"$or": updateFilter}},
+		}}}
+	return match
+}
+
+// constructActtressOption 查找选项
+func constructFindOption(pageNum int64, pageSize int64) *options.FindOptions {
+	opt := &options.FindOptions{}
+	opt.SetSort(bson.M{"createdAt": -1})
+	skip := (pageNum - 1) * pageSize
+	opt.SetSkip(skip)
+	opt.SetLimit(pageSize)
+	return opt
+}
+
+// AddMultiDataToES 添加一批Ex
+func AddMultiDataToES(esIndex string, esvs esmanager.M) *srvdef.SrvErr {
+	if err := esmanager.Instance.Bulk(esIndex, esvs); err != nil {
+		info := fmt.Sprintf("(srv_search:AddMultiDataToES) ES Bulk failed! err:%+v:", err)
+		log.Warn(info)
+		return srvdef.New(srvdef.ErrElsOperError, "", "")
+	}
+	return nil
+}
+
+// findManyFromDb 查找一批数据
+func findManyFromDb(ctx context.Context, param *Param,
+	opts ...*options.FindOptions) ([]bson.M, *srvdef.SrvErr) {
+	docs := make([]bson.M, 0)
+	if err := param.Coll.Find(ctx, &docs, param.CollFilter, opts...); err != nil {
+		log.Warn("srv_search:findManyFromDb db.Find failed", log.E(err), log.Any("filter", param.CollFilter))
+		return nil, srvdef.New(srvdef.ErrDbQueryError, "", "")
+	}
+	return docs, nil
+}
+
+// sliceMapToJsonMap 从map切片构建Map对象
+func sliceMapToJsonMap(docs []bson.M, param *Param) (esmanager.M, *srvdef.SrvErr) {
+	esMap := make(esmanager.M)
+	for _, doc := range docs {
+		fieldMap := make(esmanager.M)
+		for key, value := range doc {
+			if utils.IncludeString(param.CollFields, key) {
+				fieldMap[key] = value
+			} else if key == "_id" {
+				fieldMap["id"] = value
+			}
+		}
+		if id, ok := doc["_id"].(int64); ok {
+			strId := strconv.FormatInt(id, 10)
+			esMap[strId] = fieldMap
+		} else {
+			log.Warn("srv_search:sliceMapToJsonMap Key doesn't exist", log.Any("keyId", "_id"))
+		}
+	}
+	return esMap, nil
+}
+
+// getInfoFromChangeData 从改变流中获取数据信息
+func getInfoFromChangeData(changeData bson.M, param *Param) (
+	string, *esmod.ESTimestamp, string, esmanager.M, *srvdef.SrvErr) {
+	operationType, ok := changeData["operationType"].(string)
+	if ok == false {
+		log.Error("srv_search:getInfoFromChangeData operationType type error")
+		return "", nil, "", nil, srvdef.New(srvdef.ErrData, "", "")
+	}
+	ptimestamp, ok := changeData["clusterTime"].(primitive.Timestamp)
+	if ok == false {
+		log.Error("srv_search:getInfoFromChangeData clusterTime type error")
+		return "", nil, "", nil, srvdef.New(srvdef.ErrData, "", "")
+	}
+	estimestamp := &esmod.ESTimestamp{T: ptimestamp.T, I: ptimestamp.I}
+	documentKey, ok := changeData["documentKey"].(primitive.M)
+	if ok == false {
+		log.Error("srv_search:getInfoFromChangeData documentKey type error")
+		return "", nil, "", nil, srvdef.New(srvdef.ErrData, "", "")
+	}
+	docKey, ok := documentKey["_id"].(int64)
+	if ok == false {
+		log.Error("srv_search:getInfoFromChangeData documentKey[_id] type error")
+		return "", nil, "", nil, srvdef.New(srvdef.ErrData, "", "")
+	}
+	strId := strconv.FormatInt(docKey, 10)
+	esDocMap := make(esmanager.M)
+	if changeData["fullDocument"] != nil {
+		docMap, ok := changeData["fullDocument"].(primitive.M)
+		if ok == false {
+			log.Error("srv_search:getInfoFromChangeData fullDocument type error")
+			return "", nil, "", nil, srvdef.New(srvdef.ErrData, "", "")
+		}
+
+		for key, value := range docMap {
+			if utils.IncludeString(param.CollFields, key) {
+				esDocMap[key] = value
+			} else if key == "_id" {
+				esDocMap["id"] = value
+			}
+		}
+	}
+	return operationType, estimestamp, strId, esDocMap, nil
+}
+
+// delAllFromES 删除所有数据
+func delAllFromES(esIndex string) *srvdef.SrvErr {
+	query := esmanager.M{
+		"query": esmanager.M{
+			"match_all": esmanager.M{},
+		},
+	}
+	if err := esmanager.Instance.DelMany([]string{esIndex}, query); err != nil {
+		log.Warn(fmt.Sprintf("(srv_search:DelAllFromEs) Es DelMany failed! err:%+v:", err),
+			log.Any("query", query))
+		return srvdef.New(srvdef.ErrElsOperError, "", "")
+	}
+	return nil
+}
+
+// batchAddToES 添加一批数据到ES
+func batchAddToES(esIndex string, esvs esmanager.M) *srvdef.SrvErr {
+	log.Info("srv_search:resetAllDocsToES", log.Any("esIndex", esIndex), log.Any("esvs", esvs))
+	if err := esmanager.Instance.Bulk(esIndex, esvs); err != nil {
+		info := fmt.Sprintf("(srv_search:batchAddToES) ES Bulk failed! err:%+v:", err)
+		log.Warn(info)
+		return srvdef.New(srvdef.ErrElsOperError, "", "")
+	}
+	return nil
+}
+
+// batchDelFromES 删除一批数据从ES
+func batchDelFromES(esIndex string, ids []string) *srvdef.SrvErr {
+	query := esmanager.M{
+		"query": esmanager.M{
+			"bool": esmanager.M{
+				"must": esmanager.A{
+					{
+						"ids": esmanager.M{
+							"values": ids,
+						},
+					},
+				},
+			},
+		},
+	}
+	if err := esmanager.Instance.DelMany([]string{esIndex}, query); err != nil {
+		info := fmt.Sprintf(
+			"(srv_search:batchDelFromES) Es DelMany failed! err:%+v:", err)
+		log.Warn(info)
+		return srvdef.New(srvdef.ErrElsOperError, "", "")
+	}
+	return nil
+}
+
+// watchOpLog 监控表变化
+func watchOpLog(ctx context.Context, db *mongos.Collection, pipeline []bson.M,
+	opt *options.ChangeStreamOptions) (*mongo.ChangeStream, *srvdef.SrvErr) {
+	changeStream, err := db.WatchOpLog(ctx, pipeline, opt)
+	if err != nil {
+		log.Warn("srv_search:watchOpLog db.WatchOpLog failed",
+			log.E(err), log.Any("pipeline", pipeline))
+		return nil, srvdef.New(srvdef.ErrDbQueryError, "", "")
+	}
+	return changeStream, nil
+}