更新索引biddingall,添加主要字段

createEsIndex/bidding_es.go

@@ -67,6 +67,12 @@ func biddingTask(mapInfo map[string]interface{}) {
 		}
 		ch <- true
 		wg.Add(1)
+		// 创建一个新的map用于goroutine，避免重用
+		docCopy := make(map[string]interface{})
+		for k, v := range tmp {
+			docCopy[k] = v
+		}
+
 		go func(tmp map[string]interface{}) {
 			defer func() {
 				<-ch
@@ -159,7 +165,7 @@ func biddingTask(mapInfo map[string]interface{}) {
 					go UdpMethod(mongodb.BsonIdToSId(newTmp["_id"]))
 				}
 			}
-		}(tmp)
+		}(docCopy)
 		tmp = map[string]interface{}{}
 	}
 	wg.Wait()
@@ -171,9 +177,9 @@ func biddingTask(mapInfo map[string]interface{}) {
 			"lteid": bson.M{"$lte": util.ObjToString(mapInfo["lteid"])}}
 
 		if config.Conf.Env.Ai {
-			MgoBOld.Update("bidding_processing_ids", uq, bson.M{"$set": bson.M{"dataprocess_ai": 7, "updatetime": time.Now().Unix(), "index_num": index}}, false, true)
+			MgoBOld.Update("bidding_processing_ids", uq, bson.M{"$set": bson.M{"dataprocess_ai": 8, "updatetime": time.Now().Unix(), "index_num": index}}, false, true)
 		} else {
-			MgoB.Update("bidding_processing_ids", uq, bson.M{"$set": bson.M{"dataprocess": 8, "updatetime": time.Now().Unix(), "index_num": index}}, false, true)
+			MgoB.Update("bidding_processing_ids", uq, bson.M{"$set": bson.M{"dataprocess": 9, "updatetime": time.Now().Unix(), "index_num": index}}, false, true)
 		}
 	}
 

createEsIndex/biddingdata.go

@@ -7,6 +7,7 @@ import (
 	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"
+	"time"
 )
 
 func biddingDataTask(data []byte, mapInfo map[string]interface{}) {
@@ -45,6 +46,17 @@ func biddingDataTask(data []byte, mapInfo map[string]interface{}) {
 		"spidercode":  1,
 		"extracttype": 1,
 		"comeintime":  1,
+		"buyer":       1, //采购单位
+		"s_winner":    1, //中标单位
+		"projectcode": 1, //项目标号
+		"agency":      1, //代理机构
+		"budget":      1, //预算
+		"area":        1, //省份
+		"city":        1, //城市
+		"district":    1, //区县
+		"bidamount":   1, //中标金额
+		"toptype":     1, //一级分类
+		"subtype":     1, //二级分类
 	}).Sort("_id").Iter()
 	n := 0
 	//更新数组
@@ -60,6 +72,7 @@ func biddingDataTask(data []byte, mapInfo map[string]interface{}) {
 			}()
 			newTmp, _ := GetEsField(tmp, stype)
 			newTmp["extracttype"] = util.IntAll(tmp["extracttype"])
+			newTmp["pici"] = time.Now().Unix()
 			saveEsAllPool <- newTmp
 		}(tmp)
 		tmp = make(map[string]interface{})

createEsIndex/main.go

@@ -221,7 +221,7 @@ func processUdpMsg(act byte, data []byte, ra *net.UDPAddr) {
 					}()
 					projectAllData()
 				}()
-			case "biddingdata": //es 单机版，采集判重
+			case "biddingdata": //es 单机版，采集判重,对应索引 biddingalll; 172.17.4.184:19905
 				pool <- true
 				go func() {
 					defer func() {