data_field_dispose/data_fusion/main.go

package main

import (
	"data_fusion/config"
	"fmt"
	"go.mongodb.org/mongo-driver/bson"
	"go.uber.org/zap"
	util "jygit.jydev.jianyu360.cn/data_processing/common_utils"
	"jygit.jydev.jianyu360.cn/data_processing/common_utils/log"
	"jygit.jydev.jianyu360.cn/data_processing/common_utils/mongodb"
	"jygit.jydev.jianyu360.cn/data_processing/common_utils/redis"
	"reflect"
	"sync"
	"time"
)

var (
	ArrLock []sync.Mutex
)

func init() {
	config.Init("./common.toml")
	InitLog()
	InitMgo()
	redis.InitRedis1(config.Conf.DB.Redis.Addr, config.Conf.DB.Redis.Db)

	initData()

	updatePool = make(chan []map[string]interface{}, 5000)
	updateSp = make(chan bool, 5)
	recordPool = make(chan []map[string]interface{}, 5000)
	recordSp = make(chan bool, 5)

	for i := 0; i < config.Conf.Serve.Thread; i++ {
		ArrLock = append(ArrLock, sync.Mutex{})
	}
}

func main() {
	go updateMethod()
	go updateMethod1()

	sess := MgoB.GetMgoConn()
	defer MgoB.DestoryMongoConn(sess)

	ch := make(chan bool, config.Conf.Serve.Thread)
	wg := &sync.WaitGroup{}

	//q := map[string]interface{}{"_id": mongodb.StringTOBsonId("635ff15d631ff1ac3d095c41")}
	q := bson.M{"extracttype": -1}
	f := map[string]interface{}{"detail": 0, "contenthtml": 0, "summary": 0, "description": 0}
	it := sess.DB(config.Conf.DB.Mongo.Dbname).C(config.Conf.DB.Mongo.Coll).Find(q).Select(f).Iter()
	count := 0
	for tmp := make(map[string]interface{}); it.Next(tmp); count++ {
		if count%20000 == 0 {
			log.Info("main", zap.Int("current:", count))
		}
		ch <- true
		wg.Add(1)
		go func(tmp map[string]interface{}) {
			defer func() {
				<-ch
				wg.Done()
			}()

			repeatId := util.ObjToString(tmp["repeat_id"])
			ArrLock[util.HashCode(repeatId)%config.Conf.Serve.Thread].Lock()
			if str := redis.GetStr("fusion_id", repeatId); str != "" {
				tmp1 := findData(repeatId)
				var update map[string]interface{}
				var fs []string
				update, fs = mergeTmp(tmp1, tmp)
				update1 := util.DeepCopy(update).(map[string]interface{})
				update1["infoid"] = mongodb.BsonIdToSId(tmp["_id"])
				if fs != nil && len(fs) > 0 {
					updatePool <- []map[string]interface{}{
						{"_id": tmp1["_id"]},
						{"$set": update, "$addToSet": map[string]interface{}{"fusion_fields": map[string]interface{}{"$each": fs}}},
					}
				}
				record := make(map[string]interface{})
				record["$push"] = map[string]interface{}{
					"ids":    mongodb.BsonIdToSId(tmp["_id"]),
					"record": update1,
				}
				recordPool <- []map[string]interface{}{
					{"_id": mongodb.StringTOBsonId(repeatId)},
					record,
				}
				redis.PutCKV("fusion_id", repeatId, fmt.Sprintf("%s-%s", str, mongodb.BsonIdToSId(tmp["_id"])))
			} else {
				tmp1 := findData(repeatId)
				var update map[string]interface{}
				var fs []string
				update, fs = mergeTmp(tmp1, tmp)
				set := util.DeepCopy(update).(map[string]interface{})
				if fs != nil && len(fs) > 0 {
					set["fusion_fields"] = fs
				}
				if len(set) > 0 {
					updatePool <- []map[string]interface{}{
						{"_id": tmp1["_id"]},
						{"$set": set},
					}
				}
				record := make(map[string]interface{})
				record["_id"] = tmp1["_id"]
				record["template_id"] = mongodb.BsonIdToSId(tmp1["_id"])
				record["ids"] = []string{mongodb.BsonIdToSId(tmp["_id"]), mongodb.BsonIdToSId(tmp1["_id"])}
				var recordlist []map[string]interface{}
				update1 := util.DeepCopy(update).(map[string]interface{})
				update1["infoid"] = mongodb.BsonIdToSId(tmp["_id"])
				recordlist = append(recordlist, update1)
				record["record"] = recordlist
				recordPool <- []map[string]interface{}{
					{"_id": tmp1["_id"]},
					{"$set": record},
				}
				redis.PutCKV("fusion_id", repeatId, mongodb.BsonIdToSId(tmp["_id"]))
			}
			ArrLock[util.HashCode(repeatId)%config.Conf.Serve.Thread].Unlock()

		}(tmp)
		tmp = map[string]interface{}{}
	}
	wg.Wait()

	log.Info("fusion over...", zap.Int("count:", count))
	c := make(chan bool, 1)
	<-c
}

func findData(id string) map[string]interface{} {
	tmp, _ := MgoB.FindById(config.Conf.DB.Mongo.Coll, id, nil)
	if tmp != nil && len(*tmp) > 0 {
		return *tmp
	} else {
		tmp, _ = MgoB.FindById("bidding", id, nil)
		if tmp != nil && len(*tmp) > 0 {
			return *tmp
		}
	}
	return nil
}

func getWeight(tmp map[string]interface{}) (int, string) {
	var w int
	if util.IntAll(tmp["publishtime"]) <= 0 {
		return 0, "发布时间小于0"
	}
	if BinarySearch(CompeteSite, util.ObjToString(tmp["site"])) > -1 {
		return 0, "竞品网站数据"
	}
	for k, v := range config.Conf.Serve.Weight {
		if tmp[k] != nil {
			if k == "attachments" {
				if pinfo, ok := tmp["projectinfo"].(map[string]interface{}); ok {
					if atts, o2 := pinfo["attachments"].(map[string]interface{}); o2 {
						if len(atts) > 0 {
							w += v
						}
					}
				}
			} else {
				if reflect.TypeOf(tmp[k]).String() == "string" {
					if util.ObjToString(tmp[k]) != "" {
						w += v
					}
				} else if reflect.TypeOf(tmp[k]).String() == "float64" {
					if util.Float64All(tmp[k]) > 0 {
						w += v
					}
				} else if reflect.TypeOf(tmp[k]).String() == "[]interface {}" {
					if len(tmp[k].([]interface{})) > 0 {
						w += v
					}
				} else {
					w += v
				}
			}
		}
	}
	return w, ""
}

// @Description tmp模版数据， tmp1补充数据
// @Author J 2023/1/3 11:31
func mergeTmp(tmp map[string]interface{}, tmp1 map[string]interface{}) (map[string]interface{}, []string) {
	update := make(map[string]interface{})
	var fs []string
	for _, v := range config.Conf.Serve.Fields {
		if v == "attachments" {
			if pinfo1, ok1 := tmp1["projectinfo"].(map[string]interface{}); ok1 {
				if ats, ok2 := pinfo1[v].(map[string]interface{}); ok2 {
					if pinfo1[v] != nil && len(ats) > 0 {
						if pinfo, ok := tmp["projectinfo"].(map[string]interface{}); ok {
							if pinfo[v] == nil {
								pinfo[v] = pinfo1[v]
								update["projectinfo"] = pinfo
								update["attach_text"] = tmp1["attach_text"] // 补充附件文本
								fs = append(fs, v)
								fs = append(fs, "attach_text")
							}
						} else {
							update["projectinfo"] = map[string]interface{}{v: pinfo1[v]}
							update["attach_text"] = tmp1["attach_text"]
							fs = append(fs, v)
							fs = append(fs, "attach_text")
						}
					}
				} else {
					if pinfo1[v] != nil {
						log.Error("mergeTmp err...", zap.Any("id", mongodb.BsonIdToSId(tmp1["_id"])))
					}
				}
			}
		} else if v == "city" {
			if util.ObjToString(tmp["area"]) == util.ObjToString(tmp1["area"]) {
				if tmp[v] == nil && tmp1[v] != nil {
					update[v] = util.ObjToString(tmp1[v])
					fs = append(fs, v)
				}
			}
		} else if v == "district" {
			if util.ObjToString(tmp["city"]) == util.ObjToString(tmp1["city"]) {
				if tmp[v] == nil && tmp1[v] != nil {
					update[v] = util.ObjToString(tmp1[v])
					fs = append(fs, v)
				}
			}
		} else {
			if tmp[v] == nil && tmp1[v] != nil {
				if reflect.TypeOf(tmp1[v]).String() == "string" && util.ObjToString(tmp1[v]) != "" {
					update[v] = util.ObjToString(tmp1[v])
					fs = append(fs, v)
				} else if reflect.TypeOf(tmp1[v]).String() == "[]interface {}" && len(tmp1[v].([]interface{})) > 0 {
					update[v] = tmp1[v]
					fs = append(fs, v)
				} else {
					update[v] = tmp1[v]
					fs = append(fs, v)
				}
			}
		}
	}
	return update, fs
}

func updateMethod() {
	arru := make([][]map[string]interface{}, 500)
	indexu := 0
	for {
		select {
		case v := <-recordPool:
			arru[indexu] = v
			indexu++
			if indexu == 500 {
				recordSp <- true
				go func(arru [][]map[string]interface{}) {
					defer func() {
						<-recordSp
					}()
					MgoB.UpSertBulk(config.Conf.DB.Mongo.Record, arru...)
				}(arru)
				arru = make([][]map[string]interface{}, 500)
				indexu = 0
			}
		case <-time.After(1000 * time.Millisecond):
			if indexu > 0 {
				recordSp <- true
				go func(arru [][]map[string]interface{}) {
					defer func() {
						<-recordSp
					}()
					MgoB.UpSertBulk(config.Conf.DB.Mongo.Record, arru...)
				}(arru[:indexu])
				arru = make([][]map[string]interface{}, 500)
				indexu = 0
			}
		}
	}
}

func updateMethod1() {
	arru := make([][]map[string]interface{}, 500)
	indexu := 0
	for {
		select {
		case v := <-updatePool:
			arru[indexu] = v
			indexu++
			if indexu == 500 {
				updateSp <- true
				go func(arru [][]map[string]interface{}) {
					defer func() {
						<-updateSp
					}()
					MgoB.UpdateBulk(config.Conf.DB.Mongo.Coll, arru...)
				}(arru)
				arru = make([][]map[string]interface{}, 500)
				indexu = 0
			}
		case <-time.After(1000 * time.Millisecond):
			if indexu > 0 {
				updateSp <- true
				go func(arru [][]map[string]interface{}) {
					defer func() {
						<-updateSp
					}()
					MgoB.UpdateBulk(config.Conf.DB.Mongo.Coll, arru...)
				}(arru[:indexu])
				arru = make([][]map[string]interface{}, 500)
				indexu = 0
			}
		}
	}
}