流式-数据保存

savebidding/file.go

@@ -40,6 +40,7 @@ type DFA struct {
 func (d *DFA) AddWord(keys ...string) {
 	d.AddWordAll(true, keys...)
 }
+
 func (d *DFA) AddWordAll(haskey bool, keys ...string) {
 	if d.Link == nil {
 		d.Link = make(map[string]interface{})

savebidding/main.go

@@ -96,7 +96,7 @@ func SubscribeNats() {
 // 保存
 func SaveDealData(data map[string]interface{}) {
 	//分类及部分字段处理
-	fieldFun(data)
+	fieldFun(data, data)
 
 	//补充publishtime
 	if u.IntAll(data["publishtime"]) == -1 {
@@ -114,7 +114,7 @@ func SaveDealData(data map[string]interface{}) {
 		}
 	}
 	//剑鱼发布信息分类处理
-	//typeFunc(data)//单独数据流处理
+	//typeFunc(data)//暂无剑鱼发布数据流
 
 	// 附件有效字段
 	if r := validFile(data); r != 0 {
@@ -135,7 +135,7 @@ func SaveDealData(data map[string]interface{}) {
 }
 
 // 更新
-func UpdateDealData(id string, data map[string]interface{}) {
+func UpdateDealData(id string, newdata map[string]interface{}) {
 	fields := map[string]interface{}{
 		"detail":      0,
 		"contenthtml": 0,
@@ -151,13 +151,13 @@ func UpdateDealData(id string, data map[string]interface{}) {
 		update := map[string]interface{}{}
 		del := map[string]interface{}{}
 		for _, k := range conf.Config.Fields {
-			tmpV := data[k]   //extract v1
-			bidV := (*bid)[k] //bidding v2
-			if bidV == nil && tmpV != nil {
-				update[k] = tmpV
-			} else if bidV != nil && tmpV != nil && !modifyinfo[k] {
-				update[k] = tmpV
-			} else if bidV != nil && tmpV == nil && !modifyinfo[k] {
+			newV := newdata[k] //extract v1
+			bidV := (*bid)[k]  //bidding v2
+			if bidV == nil && newV != nil {
+				update[k] = newV
+			} else if bidV != nil && newV != nil && !modifyinfo[k] {
+				update[k] = newV
+			} else if bidV != nil && newV == nil && !modifyinfo[k] {
 				if k == "s_subscopeclass" && del["subscopeclass"] == nil {
 					continue
 				} else if k == "s_topscopeclass" && del["topscopeclass"] == nil {
@@ -168,47 +168,48 @@ func UpdateDealData(id string, data map[string]interface{}) {
 		}
 
 		//分类及部分字段处理
-		fieldFun(data)
+		fieldFun(newdata, update)
 
 		//publishtime
-		if u.IntAll(data["publishtime"]) == -1 {
-			methodPb(data) //修正发布时间
-		}
+		//if u.IntAll(data["publishtime"]) == -1 {
+		//	methodPb(data) //修正发布时间
+		//}
 
 		// entidlist
-		if s_winner, ok := data["s_winner"].(string); ok && s_winner != "" {
+		if s_winner, ok := newdata["s_winner"].(string); ok && s_winner != "" {
 			cid := companyFun(s_winner)
 			if len(cid) > 0 {
-				data["entidlist"] = cid
+				update["entidlist"] = cid
 			}
 		}
 		//剑鱼发布信息分类处理
 		//typeFunc(data)
 
 		// 附件有效字段
-		//if r := validFile(data); r != 0 {
-		//	if r == -1 {
-		//		data["isValidFile"] = false
-		//	} else {
-		//		data["isValidFile"] = true
-		//	}
-		//}
+		if r := validFile(newdata); r != 0 {
+			if r == -1 {
+				update["isValidFile"] = false
+			} else {
+				update["isValidFile"] = true
+			}
+		}
 
 		//情报标签字段
 		//if data["tag_topinformation"] != nil {
 		//	data["tag_set"] = getTagSet(data)
 		//}
-
-		//放入通道
-		if len(del) > 0 {
-			DataUpdateCache <- []map[string]interface{}{
-				{"_id": mongodb.StringTOBsonId(id)},
-				{"$set": update, "$unset": del},
-			}
-		} else {
-			DataUpdateCache <- []map[string]interface{}{
-				{"_id": mongodb.StringTOBsonId(id)},
-				{"$set": update, "$unset": del},
+		if len(update) > 0 {
+			//放入通道
+			if len(del) > 0 {
+				DataUpdateCache <- []map[string]interface{}{
+					{"_id": mongodb.StringTOBsonId(id)},
+					{"$set": update, "$unset": del},
+				}
+			} else {
+				DataUpdateCache <- []map[string]interface{}{
+					{"_id": mongodb.StringTOBsonId(id)},
+					{"$set": update},
+				}
 			}
 		}
 	} else { //未找到数据

savebidding/util.go

@@ -21,17 +21,17 @@ func InitEs() {
 // @Description subscopeclass、topscopeclass、package
 // 20230523 多包处理 subpackage = 1
 // @Author J 2022/6/7 5:54 PM
-func fieldFun(data map[string]interface{}) {
+func fieldFun(data map[string]interface{}, update map[string]interface{}) {
 	// 附件重采，数据同步时不更新判重标识（正常抽取判重）
 	if u.IntAll(data["repeat"]) == 1 {
-		data["extracttype"] = -1
+		update["extracttype"] = -1
 		//update["repeat_id"] = compare["repeat_id"]
 	} else {
-		data["extracttype"] = 1
+		update["extracttype"] = 1
 	}
 	//未抽取、判重处理
 	if u.IntAll(data["dataging"]) == 1 { //修改未抽取的bidding数据的dataging
-		data["dataging"] = 0 //
+		update["dataging"] = 0 //
 	}
 	if subscopeclass, ok := data["subscopeclass"].(primitive.A); ok && len(subscopeclass) > 0 { //subscopeclass
 		m1 := map[string]bool{}
@@ -43,8 +43,8 @@ func fieldFun(data map[string]interface{}) {
 				newclass = append(newclass, sclass)
 			}
 		}
-		//data["subscopeclass"] = newclass
-		data["s_subscopeclass"] = strings.Join(newclass, ",")
+		update["subscopeclass"] = newclass //subscopeclass去重
+		update["s_subscopeclass"] = strings.Join(newclass, ",")
 	}
 	if topscopeclass, ok := data["topscopeclass"].(primitive.A); ok && len(topscopeclass) > 0 { //topscopeclass
 		m2 := map[string]bool{}
@@ -57,22 +57,22 @@ func fieldFun(data map[string]interface{}) {
 				newclass = append(newclass, tclass)
 			}
 		}
-		//data["topscopeclass"] = topscopeclass
-		data["s_topscopeclass"] = strings.Join(newclass, ",")
+		update["topscopeclass"] = topscopeclass //topscopeclass保持原有数据
+		update["s_topscopeclass"] = strings.Join(newclass, ",")
 	}
 	//package
 	if package1 := data["package"]; package1 != nil {
 		packageM, _ := package1.(map[string]interface{})
-		//data["package"] = packageM
+		update["package"] = packageM
 		for _, p := range packageM {
 			pm, _ := p.(map[string]interface{})
 			if u.ObjToString(pm["winner"]) != "" || u.Float64All(pm["budget"]) > 0 || u.Float64All(pm["bidamount"]) > 0 {
-				data["multipackage"] = 1
+				update["multipackage"] = 1
 				break
 			}
 		}
 	} else {
-		data["multipackage"] = 0
+		update["multipackage"] = 0
 	}
 
 	// subpackage
@@ -101,7 +101,7 @@ func fieldFun(data map[string]interface{}) {
 						bidamount += f
 					}
 					if bidamount == u.Float64All(data["bidamount"]) {
-						data["subpackage"] = 1
+						update["subpackage"] = 1
 					}
 				}
 			}