init: web_hgdjc_server

2026-11-07 10:44:09 +08:00
commit 71a4f4e175
713 changed files with 147749 additions and 0 deletions
@@ -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
+}