|
|
@@ -15,6 +15,7 @@ import (
|
|
|
"time"
|
|
|
)
|
|
|
|
|
|
+//之前main方法,只更新
|
|
|
func TaskAgency(mapinfo *map[string]interface{}) {
|
|
|
defer util.Catch()
|
|
|
gtid, lteid := util.ObjToString((*mapinfo)["gtid"]), util.ObjToString((*mapinfo)["lteid"])
|
|
|
@@ -39,208 +40,342 @@ func TaskAgency(mapinfo *map[string]interface{}) {
|
|
|
"$gte": GId,
|
|
|
"$lte": LtId,
|
|
|
},
|
|
|
- }).Select(bson.M{"agency": 1, "agencytel": 1, "agencyperson": 1,
|
|
|
- "topscopeclass": 1, "agencyaddr": 1}).Iter()
|
|
|
+ }).Select(bson.M{"winner": 1, "winnertel": 1, "winnerperson": 1,
|
|
|
+ "topscopeclass": 1, "winneraddr": 1}).Iter()
|
|
|
if cursor == nil {
|
|
|
log.Println(cursor)
|
|
|
return
|
|
|
}
|
|
|
- overid := gtid
|
|
|
- tmp := map[string]interface{}{}
|
|
|
- for cursor.Next(&tmp) {
|
|
|
- overid = tmp["_id"].(primitive.ObjectID).Hex()
|
|
|
- log.Println(tmp["_id"])
|
|
|
- if tmp["agency"] == nil || tmp["agency"] == "" {
|
|
|
- continue
|
|
|
- }
|
|
|
- //redis查询是否存在
|
|
|
- rdb := RedisPool.Get()
|
|
|
- if reply, err := redis.String(rdb.Do("GET", tmp["agency"])); err != nil {
|
|
|
- //redis不存在存到临时表,定时任务处理
|
|
|
- FClient.DbName = Config["mgodb_extract_kf"]
|
|
|
- if tmpid := FClient.Save("agency_new", tmp) ;tmpid==nil{
|
|
|
- log.Println("FClient.Save err",tmpid)
|
|
|
+ //判断是否是存量,是存量走Redis遍历
|
|
|
+ if v, ok := (*mapinfo)["data_info"].(string); ok && v == "save" {
|
|
|
+ //存量处理
|
|
|
+ tmp := map[string]interface{}{}
|
|
|
+ conn := HisRedisPool.Conn()
|
|
|
+ defer conn.Close()
|
|
|
+ //选择redis db
|
|
|
+ conn.Select(1)
|
|
|
+ //遍历bidding表保存到redis
|
|
|
+ // key：_id value:json结构体
|
|
|
+ for cursor.Next(&tmp) {
|
|
|
+ if tmp["winner"] == nil || tmp["winner"] == "" {
|
|
|
+ continue
|
|
|
}
|
|
|
- //log.Println("get redis id err:定时任务处理", err, tmp)
|
|
|
- if err := rdb.Close(); err != nil {
|
|
|
+ bytes, _ := json.Marshal(tmp)
|
|
|
+ if err := conn.Set(tmp["_id"].(primitive.ObjectID).Hex(), string(bytes), 0).Err(); err != nil {
|
|
|
log.Println(err)
|
|
|
}
|
|
|
- continue
|
|
|
+ }
|
|
|
+ //遍历redis
|
|
|
+ if scan := conn.Scan(0, "", 100); scan.Err() != nil {
|
|
|
+ log.Println(scan.Err())
|
|
|
+ return
|
|
|
} else {
|
|
|
- //log.Println("redis get :", reply)
|
|
|
- //redis存在
|
|
|
- //log.Println(reply)
|
|
|
- //reply = "5e0316b998a9abaf6535df3d"
|
|
|
- //id, err := primitive.ObjectIDFromHex(reply)
|
|
|
- //if err != nil {
|
|
|
- // log.Println("get redis id Hex err:", err, tmp)
|
|
|
- // if err := rdb.Close(); err != nil {
|
|
|
- // log.Println(err)
|
|
|
- // }
|
|
|
- // continue
|
|
|
- //}
|
|
|
- if err := rdb.Close(); err != nil {
|
|
|
- log.Println(err)
|
|
|
- }
|
|
|
- //拿到合并后的qyk
|
|
|
- FClient.DbName = Config["mgodb_extract_kf"]
|
|
|
- oldTmp := FClient.FindById(Config["mgo_qyk_c"], reply)
|
|
|
- if oldTmp == nil{
|
|
|
- log.Println("redis id 不存在")
|
|
|
- continue
|
|
|
- }
|
|
|
- //err = FClient.Database(Config["mgodb_extract_kf"]).Collection(Config["mgo_qyk_c"]).
|
|
|
- // FindOne(context.TODO(), bson.M{"_id": id}).Decode(&oldTmp)
|
|
|
- //if err != nil {
|
|
|
- // log.Println("qyk id err:", err, id)
|
|
|
- // continue
|
|
|
- //}
|
|
|
- //比较合并
|
|
|
- //行业类型
|
|
|
- tmpTopscopeclass := []string{}
|
|
|
- tmpTopscopeclassMap := make(map[string]bool)
|
|
|
+ iterator := scan.Iterator()
|
|
|
+ for iterator.Next() {
|
|
|
+ redisId := iterator.Val() //redis key
|
|
|
+ redisvalue := conn.Get(iterator.Val()).String() //redis val
|
|
|
+ tmp := make(map[string]interface{})
|
|
|
+ json.Unmarshal([]byte(redisvalue),&tmp)
|
|
|
+ //重复增量操作
|
|
|
+ //redis查询是否存在
|
|
|
+ rdb := RedisPool.Get()
|
|
|
+ if reply, err := redis.String(rdb.Do("GET", tmp["winner"])); err != nil {
|
|
|
+ //redis不存在,存到临时表,定时任务处理
|
|
|
+ FClient.DbName = Config["mgodb_extract_kf"]
|
|
|
+ if tmpid := FClient.Save("winner_new", tmp); tmpid == nil {
|
|
|
+ log.Println("存量 FClient.Save err", tmpid)
|
|
|
+ }
|
|
|
+ //log.Println("get redis id err:定时任务处理", err, tmp)
|
|
|
+ if err := rdb.Close(); err != nil {
|
|
|
+ log.Println("存量",err)
|
|
|
+ }
|
|
|
+ //删除存量redis
|
|
|
+ conn.Del(redisId)
|
|
|
+ continue
|
|
|
+ } else {
|
|
|
+ if err := rdb.Close(); err != nil {
|
|
|
+ log.Println(err)
|
|
|
+ }
|
|
|
+ //拿到合并后的qyk
|
|
|
+ FClient.DbName = Config["mgodb_extract_kf"]
|
|
|
+ oldTmp := FClient.FindById(Config["mgo_qyk_c"], reply)
|
|
|
+ if oldTmp == nil {
|
|
|
+ log.Println("存量 redis id 不存在",reply,tmp["winner"])
|
|
|
+ continue
|
|
|
+ }
|
|
|
+ tmpTopscopeclass := []string{}
|
|
|
+ tmpTopscopeclassMap := make(map[string]bool)
|
|
|
|
|
|
- if oldTmp["industry"] == nil {
|
|
|
- //log.Println(reflect.ValueOf(tmp["topscopeclass"]))
|
|
|
- if v, ok := tmp["topscopeclass"].(primitive.A); ok {
|
|
|
- for _, vv := range v {
|
|
|
- if vvv, ok := vv.(string); ok && len(vvv) > 1 {
|
|
|
- tmpTopscopeclassMap[vvv[:len(vvv)-1]] = true
|
|
|
+ if oldTmp["industry"] == nil {
|
|
|
+ if v, ok := tmp["topscopeclass"].(primitive.A); ok {
|
|
|
+ for _, vv := range v {
|
|
|
+ if vvv, ok := vv.(string); ok && len(vvv) > 1 {
|
|
|
+ tmpTopscopeclassMap[vvv[:len(vvv)-1]] = true
|
|
|
+ }
|
|
|
+ }
|
|
|
+ for k := range tmpTopscopeclassMap {
|
|
|
+ tmpTopscopeclass = append(tmpTopscopeclass, k)
|
|
|
+ }
|
|
|
+ }
|
|
|
+ } else {
|
|
|
+ if v, ok := oldTmp["industry"].(primitive.A); ok {
|
|
|
+ for _, vv := range v {
|
|
|
+ if vvv, ok := vv.(string); ok {
|
|
|
+ tmpTopscopeclassMap[vvv] = true
|
|
|
+ }
|
|
|
+ }
|
|
|
+ }
|
|
|
+ if v, ok := tmp["topscopeclass"].(primitive.A); ok {
|
|
|
+ for _, vv := range v {
|
|
|
+ if vvv, ok := vv.(string); ok && len(vvv) > 1 {
|
|
|
+ tmpTopscopeclassMap[vvv[:len(vvv)-1]] = true
|
|
|
+ }
|
|
|
+ }
|
|
|
+ for k := range tmpTopscopeclassMap {
|
|
|
+ tmpTopscopeclass = append(tmpTopscopeclass, k)
|
|
|
+ }
|
|
|
}
|
|
|
}
|
|
|
- for k := range tmpTopscopeclassMap {
|
|
|
- tmpTopscopeclass = append(tmpTopscopeclass, k)
|
|
|
+ sort.Strings(tmpTopscopeclass)
|
|
|
+ oldTmp["industry"] = tmpTopscopeclass
|
|
|
+ esId := oldTmp["_id"].(primitive.ObjectID).Hex()
|
|
|
+ //更新行业类型
|
|
|
+ if tmp["winnerperson"] == nil || tmp["winnerperson"] == "" || Reg_xing.MatchString(util.ObjToString(tmp["winnerperson"])) {
|
|
|
+ oldTmp["updatatime"] = time.Now().Unix()
|
|
|
+ //mongo更新
|
|
|
+ FClient.DbName = Config["mgodb_extract_kf"]
|
|
|
+ if !FClient.UpdateById(Config["mgo_qyk_c"], esId, bson.M{"$set": oldTmp}) {
|
|
|
+ log.Println("mongo更新err", esId)
|
|
|
+ }
|
|
|
+
|
|
|
+ //es更新
|
|
|
+ delete(oldTmp, "_id")
|
|
|
+ if _, err := EsConn.Update().Index(Config["elasticsearch_index"]).Type(Config["elasticsearch_type"]).Id(esId).Doc(oldTmp).Refresh(true).Do(); err != nil {
|
|
|
+ log.Println("update es err:", err)
|
|
|
+ }
|
|
|
+ //删除存量redis
|
|
|
+ conn.Del(redisId)
|
|
|
+ continue
|
|
|
}
|
|
|
- }
|
|
|
- } else {
|
|
|
- if v, ok := oldTmp["industry"].(primitive.A); ok {
|
|
|
- for _, vv := range v {
|
|
|
- if vvv, ok := vv.(string); ok {
|
|
|
- tmpTopscopeclassMap[vvv] = true
|
|
|
+ //联系方式合并
|
|
|
+ var tmpperson, winnertel string
|
|
|
+ tmpperson = tmp["winnerperson"].(string)
|
|
|
+ if tmp["winnertel"] == nil || tmp["winnertel"] == "" {
|
|
|
+ winnertel = ""
|
|
|
+ } else {
|
|
|
+ if Reg_xing.MatchString(util.ObjToString(tmp["winnertel"])) || !Reg_tel.MatchString(util.ObjToString(tmp["winnertel"])) {
|
|
|
+ winnertel = ""
|
|
|
+ } else {
|
|
|
+ winnertel = util.ObjToString(tmp["winnertel"])
|
|
|
}
|
|
|
}
|
|
|
- }
|
|
|
- if v, ok := tmp["topscopeclass"].(primitive.A); ok {
|
|
|
- for _, vv := range v {
|
|
|
- if vvv, ok := vv.(string); ok && len(vvv) > 1 {
|
|
|
- tmpTopscopeclassMap[vvv[:len(vvv)-1]] = true
|
|
|
+ contactMaps := make([]interface{}, 0)
|
|
|
+ if oldTmp["contact"] == nil {
|
|
|
+ tmpContact := make(map[string]interface{})
|
|
|
+ tmpContact["infoid"] = redisId
|
|
|
+ tmpContact["contact_person"] = tmpperson
|
|
|
+ tmpContact["contact_type"] = "项目联系人"
|
|
|
+ tmpContact["phone"] = winnertel
|
|
|
+ tmpContact["topscopeclass"] = strings.Join(tmpTopscopeclass, ";")
|
|
|
+ tmpContact["updatetime"] = time.Now().Unix()
|
|
|
+ contactMaps = append(contactMaps, tmpContact)
|
|
|
+ } else {
|
|
|
+ //对比前四项,相等丢弃
|
|
|
+ if v, ok := oldTmp["contact"].(primitive.A); ok {
|
|
|
+ var isNotUpdate bool
|
|
|
+ for _, vv := range v {
|
|
|
+ if vvv, ok := vv.(map[string]interface{}); ok {
|
|
|
+ if vvv["contact_person"] == tmpperson && vvv["contact_type"] == "项目联系人" &&
|
|
|
+ vvv["phone"] == winnertel && vvv["topscopeclass"] == strings.Join(tmpTopscopeclass, ";") {
|
|
|
+ isNotUpdate = true
|
|
|
+ vvv["updatetime"] = time.Now().Unix()
|
|
|
+ }
|
|
|
+ contactMaps = append(contactMaps, vvv)
|
|
|
+ }
|
|
|
+ }
|
|
|
+ if !isNotUpdate {
|
|
|
+ vvv := make(map[string]interface{})
|
|
|
+ vvv["infoid"] = redisId
|
|
|
+ vvv["contact_person"] = tmp["winnerperson"]
|
|
|
+ vvv["contact_type"] = "项目联系人"
|
|
|
+ vvv["phone"] = winnertel
|
|
|
+ vvv["topscopeclass"] = strings.Join(tmpTopscopeclass, ";")
|
|
|
+ vvv["updatetime"] = time.Now().Unix()
|
|
|
+ contactMaps = append(contactMaps, vvv)
|
|
|
+ }
|
|
|
}
|
|
|
}
|
|
|
- for k := range tmpTopscopeclassMap {
|
|
|
- tmpTopscopeclass = append(tmpTopscopeclass, k)
|
|
|
+ oldTmp["contact"] = contactMaps
|
|
|
+ //mongo更新
|
|
|
+ oldTmp["updatatime"] = time.Now().Unix()
|
|
|
+ FClient.DbName = Config["mgodb_extract_kf"]
|
|
|
+ if !FClient.UpdateById(Config["mgo_qyk_c"], esId, bson.M{"$set": oldTmp}) {
|
|
|
+ log.Println("存量 mongo更新 err", esId, oldTmp)
|
|
|
}
|
|
|
+ //es更新
|
|
|
+ delete(oldTmp, "_id")
|
|
|
+ if _, err := EsConn.Update().Index(Config["elasticsearch_index"]).Type(Config["elasticsearch_type"]).Id(esId).Doc(oldTmp).Refresh(true).Do(); err != nil {
|
|
|
+ log.Println("存量 EsConn err :", err)
|
|
|
+ }
|
|
|
+ //最后删除redis
|
|
|
+ conn.Del(redisId)
|
|
|
}
|
|
|
}
|
|
|
- sort.Strings(tmpTopscopeclass)
|
|
|
- oldTmp["industry"] = tmpTopscopeclass
|
|
|
- esId := oldTmp["_id"].(primitive.ObjectID).Hex()
|
|
|
- //更新行业类型
|
|
|
- if tmp["winnerperson"] == nil || tmp["winnerperson"] == "" || Reg_xing.MatchString(util.ObjToString(tmp["winnerperson"])) {
|
|
|
- oldTmp["updatatime"] = time.Now().Unix()
|
|
|
- //mongo更新
|
|
|
- FClient.DbName =Config["mgodb_extract_kf"]
|
|
|
- if !FClient.UpdateById(Config["mgo_qyk_c"],esId,bson.M{"$set": oldTmp}){
|
|
|
- log.Println("mongo更新err",esId)
|
|
|
+ }
|
|
|
+ log.Println("存量历史合并执行完成 ok", gtid, lteid)
|
|
|
+
|
|
|
+ } else {
|
|
|
+ //增量处理
|
|
|
+ overid := gtid
|
|
|
+ tmp := map[string]interface{}{}
|
|
|
+ for cursor.Next(&tmp) {
|
|
|
+ overid = tmp["_id"].(primitive.ObjectID).Hex()
|
|
|
+ log.Println(tmp["_id"])
|
|
|
+ if tmp["winner"] == nil || tmp["winner"] == "" {
|
|
|
+ continue
|
|
|
+ }
|
|
|
+ //redis查询是否存在
|
|
|
+ rdb := RedisPool.Get()
|
|
|
+ if reply, err := redis.String(rdb.Do("GET", tmp["winner"])); err != nil {
|
|
|
+ //redis不存在存到临时表,定时任务处理
|
|
|
+ FClient.DbName = Config["mgodb_extract_kf"]
|
|
|
+ if tmpid := FClient.Save("winner_new", tmp); tmpid == nil {
|
|
|
+ log.Println("FClient.Save err", tmpid)
|
|
|
}
|
|
|
- //if _, err := FClient.Database(Config["mgodb_extract_kf"]).Collection(Config["mgo_qyk_c"]).
|
|
|
- // UpdateOne(context.TODO(), bson.M{"_id": oldTmp["_id"]}, bson.M{"$set": oldTmp}); err != nil {
|
|
|
- // log.Println("mongo更新err:", err)
|
|
|
- //}
|
|
|
- //es更新
|
|
|
- delete(oldTmp, "_id")
|
|
|
- //esConn := elastic.GetEsConn()
|
|
|
- //defer elastic.DestoryEsConn(esConn)
|
|
|
- if _, err := EsConn.Update().Index(Config["elasticsearch_index"]).Type(Config["elasticsearch_type"]).Id(esId).Doc(oldTmp).Refresh(true).Do(); err != nil {
|
|
|
- log.Println("update es err:", err)
|
|
|
+ //log.Println("get redis id err:定时任务处理", err, tmp)
|
|
|
+ if err := rdb.Close(); err != nil {
|
|
|
+ log.Println(err)
|
|
|
}
|
|
|
- //log.Println( err2,err3)
|
|
|
continue
|
|
|
- }
|
|
|
- //联系方式合并
|
|
|
- var tmpperson, winnertel string
|
|
|
- tmpperson = tmp["winnerperson"].(string)
|
|
|
- if tmp["winnertel"] == nil || tmp["winnertel"] == "" {
|
|
|
- winnertel = ""
|
|
|
} else {
|
|
|
- if Reg_xing.MatchString(util.ObjToString(tmp["winnertel"])) || !Reg_tel.MatchString(util.ObjToString(tmp["winnertel"])) {
|
|
|
+ if err := rdb.Close(); err != nil {
|
|
|
+ log.Println(err)
|
|
|
+ }
|
|
|
+ //拿到合并后的qyk
|
|
|
+ FClient.DbName = Config["mgodb_extract_kf"]
|
|
|
+ oldTmp := FClient.FindById(Config["mgo_qyk_c"], reply)
|
|
|
+ if oldTmp == nil {
|
|
|
+ log.Println("redis id 不存在")
|
|
|
+ continue
|
|
|
+ }
|
|
|
+ //比较合并
|
|
|
+ //行业类型
|
|
|
+ tmpTopscopeclass := []string{}
|
|
|
+ tmpTopscopeclassMap := make(map[string]bool)
|
|
|
+
|
|
|
+ if oldTmp["industry"] == nil {
|
|
|
+ if v, ok := tmp["topscopeclass"].(primitive.A); ok {
|
|
|
+ for _, vv := range v {
|
|
|
+ if vvv, ok := vv.(string); ok && len(vvv) > 1 {
|
|
|
+ tmpTopscopeclassMap[vvv[:len(vvv)-1]] = true
|
|
|
+ }
|
|
|
+ }
|
|
|
+ for k := range tmpTopscopeclassMap {
|
|
|
+ tmpTopscopeclass = append(tmpTopscopeclass, k)
|
|
|
+ }
|
|
|
+ }
|
|
|
+ } else {
|
|
|
+ if v, ok := oldTmp["industry"].(primitive.A); ok {
|
|
|
+ for _, vv := range v {
|
|
|
+ if vvv, ok := vv.(string); ok {
|
|
|
+ tmpTopscopeclassMap[vvv] = true
|
|
|
+ }
|
|
|
+ }
|
|
|
+ }
|
|
|
+ if v, ok := tmp["topscopeclass"].(primitive.A); ok {
|
|
|
+ for _, vv := range v {
|
|
|
+ if vvv, ok := vv.(string); ok && len(vvv) > 1 {
|
|
|
+ tmpTopscopeclassMap[vvv[:len(vvv)-1]] = true
|
|
|
+ }
|
|
|
+ }
|
|
|
+ for k := range tmpTopscopeclassMap {
|
|
|
+ tmpTopscopeclass = append(tmpTopscopeclass, k)
|
|
|
+ }
|
|
|
+ }
|
|
|
+ }
|
|
|
+ sort.Strings(tmpTopscopeclass)
|
|
|
+ oldTmp["industry"] = tmpTopscopeclass
|
|
|
+ esId := oldTmp["_id"].(primitive.ObjectID).Hex()
|
|
|
+ //更新行业类型
|
|
|
+ if tmp["winnerperson"] == nil || tmp["winnerperson"] == "" || Reg_xing.MatchString(util.ObjToString(tmp["winnerperson"])) {
|
|
|
+ oldTmp["updatatime"] = time.Now().Unix()
|
|
|
+ //mongo更新
|
|
|
+ FClient.DbName = Config["mgodb_extract_kf"]
|
|
|
+ if !FClient.UpdateById(Config["mgo_qyk_c"], esId, bson.M{"$set": oldTmp}) {
|
|
|
+ log.Println("mongo更新err", esId)
|
|
|
+ }
|
|
|
+
|
|
|
+ //es更新
|
|
|
+ delete(oldTmp, "_id")
|
|
|
+ if _, err := EsConn.Update().Index(Config["elasticsearch_index"]).Type(Config["elasticsearch_type"]).Id(esId).Doc(oldTmp).Refresh(true).Do(); err != nil {
|
|
|
+ log.Println("update es err:", err)
|
|
|
+ }
|
|
|
+ continue
|
|
|
+ }
|
|
|
+ //联系方式合并
|
|
|
+ var tmpperson, winnertel string
|
|
|
+ tmpperson = tmp["winnerperson"].(string)
|
|
|
+ if tmp["winnertel"] == nil || tmp["winnertel"] == "" {
|
|
|
winnertel = ""
|
|
|
} else {
|
|
|
- winnertel = util.ObjToString(tmp["winnertel"])
|
|
|
+ if Reg_xing.MatchString(util.ObjToString(tmp["winnertel"])) || !Reg_tel.MatchString(util.ObjToString(tmp["winnertel"])) {
|
|
|
+ winnertel = ""
|
|
|
+ } else {
|
|
|
+ winnertel = util.ObjToString(tmp["winnertel"])
|
|
|
+ }
|
|
|
}
|
|
|
- }
|
|
|
- contactMaps := make([]interface{}, 0)
|
|
|
- if oldTmp["contact"] == nil {
|
|
|
- tmpContact := make(map[string]interface{})
|
|
|
- tmpContact["contact_person"] = tmpperson
|
|
|
- tmpContact["contact_type"] = "项目联系人"
|
|
|
- tmpContact["phone"] = winnertel
|
|
|
- tmpContact["topscopeclass"] = strings.Join(tmpTopscopeclass, ";")
|
|
|
- tmpContact["updatetime"] = time.Now().Unix()
|
|
|
- contactMaps = append(contactMaps, tmpContact)
|
|
|
- } else {
|
|
|
- //对比前四项,相等丢弃
|
|
|
- if v, ok := oldTmp["contact"].(primitive.A); ok {
|
|
|
- var isNotUpdate bool
|
|
|
- for _, vv := range v {
|
|
|
- if vvv, ok := vv.(map[string]interface{}); ok {
|
|
|
- if vvv["contact_person"] == tmpperson && vvv["contact_type"] == "项目联系人" &&
|
|
|
- vvv["phone"] == winnertel && vvv["topscopeclass"] == strings.Join(tmpTopscopeclass, ";") {
|
|
|
- isNotUpdate = true
|
|
|
- vvv["updatetime"] = time.Now().Unix()
|
|
|
+ contactMaps := make([]interface{}, 0)
|
|
|
+ if oldTmp["contact"] == nil {
|
|
|
+ tmpContact := make(map[string]interface{})
|
|
|
+ tmpContact["infoid"] = overid
|
|
|
+ tmpContact["contact_person"] = tmpperson
|
|
|
+ tmpContact["contact_type"] = "项目联系人"
|
|
|
+ tmpContact["phone"] = winnertel
|
|
|
+ tmpContact["topscopeclass"] = strings.Join(tmpTopscopeclass, ";")
|
|
|
+ tmpContact["updatetime"] = time.Now().Unix()
|
|
|
+ contactMaps = append(contactMaps, tmpContact)
|
|
|
+ } else {
|
|
|
+ //对比前四项,相等丢弃
|
|
|
+ if v, ok := oldTmp["contact"].(primitive.A); ok {
|
|
|
+ var isNotUpdate bool
|
|
|
+ for _, vv := range v {
|
|
|
+ if vvv, ok := vv.(map[string]interface{}); ok {
|
|
|
+ if vvv["contact_person"] == tmpperson && vvv["contact_type"] == "项目联系人" &&
|
|
|
+ vvv["phone"] == winnertel && vvv["topscopeclass"] == strings.Join(tmpTopscopeclass, ";") {
|
|
|
+ isNotUpdate = true
|
|
|
+ vvv["updatetime"] = time.Now().Unix()
|
|
|
+ }
|
|
|
+ contactMaps = append(contactMaps, vvv)
|
|
|
}
|
|
|
+ }
|
|
|
+ if !isNotUpdate {
|
|
|
+ vvv := make(map[string]interface{})
|
|
|
+ vvv["infoid"] = overid
|
|
|
+ vvv["contact_person"] = tmp["winnerperson"]
|
|
|
+ vvv["contact_type"] = "项目联系人"
|
|
|
+ vvv["phone"] = winnertel
|
|
|
+ vvv["topscopeclass"] = strings.Join(tmpTopscopeclass, ";")
|
|
|
+ vvv["updatetime"] = time.Now().Unix()
|
|
|
contactMaps = append(contactMaps, vvv)
|
|
|
}
|
|
|
}
|
|
|
- if !isNotUpdate {
|
|
|
- vvv := make(map[string]interface{})
|
|
|
- vvv["contact_person"] = tmp["winnerperson"]
|
|
|
- vvv["contact_type"] = "项目联系人"
|
|
|
- vvv["phone"] = winnertel
|
|
|
- vvv["topscopeclass"] = strings.Join(tmpTopscopeclass, ";")
|
|
|
- vvv["updatetime"] = time.Now().Unix()
|
|
|
- contactMaps = append(contactMaps, vvv)
|
|
|
- }
|
|
|
+ }
|
|
|
+ oldTmp["contact"] = contactMaps
|
|
|
+ //mongo更新
|
|
|
+ oldTmp["updatatime"] = time.Now().Unix()
|
|
|
+ FClient.DbName = Config["mgodb_extract_kf"]
|
|
|
+ if !FClient.UpdateById(Config["mgo_qyk_c"], esId, bson.M{"$set": oldTmp}) {
|
|
|
+ log.Println("mongo更新 err", esId, oldTmp)
|
|
|
+ }
|
|
|
+ //es更新
|
|
|
+ delete(oldTmp, "_id")
|
|
|
+ if _, err := EsConn.Update().Index(Config["elasticsearch_index"]).Type(Config["elasticsearch_type"]).Id(esId).Doc(oldTmp).Refresh(true).Do(); err != nil {
|
|
|
+ log.Println("EsConn err :", err)
|
|
|
}
|
|
|
}
|
|
|
- oldTmp["contact"] = contactMaps
|
|
|
- //mongo更新
|
|
|
- oldTmp["updatatime"] = time.Now().Unix()
|
|
|
- FClient.DbName=Config["mgodb_extract_kf"]
|
|
|
- if !FClient.UpdateById(Config["mgo_qyk_c"],esId,bson.M{"$set": oldTmp}){
|
|
|
- log.Println("mongo更新 err",esId,oldTmp)
|
|
|
- }
|
|
|
- //if _, err := FClient.Database(Config["mgodb_extract_kf"]).Collection(Config["mgo_qyk_c"]).
|
|
|
- // UpdateOne(context.TODO(), bson.M{"_id": oldTmp["_id"]}, bson.M{"$set": oldTmp}); err != nil {
|
|
|
- // log.Println("mongo更新 err :", err)
|
|
|
- //}
|
|
|
- //es更新
|
|
|
- delete(oldTmp, "_id")
|
|
|
- //esConn := elastic.GetEsConn()
|
|
|
- //defer elastic.DestoryEsConn(esConn)
|
|
|
- if _, err := EsConn.Update().Index(Config["elasticsearch_index"]).Type(Config["elasticsearch_type"]).Id(esId).Doc(oldTmp).Refresh(true).Do(); err != nil {
|
|
|
- log.Println("EsConn err :", err)
|
|
|
- }
|
|
|
- //log.Println( err2,err3)
|
|
|
}
|
|
|
+ log.Println("增量合并执行完成 ok", gtid, lteid, overid)
|
|
|
}
|
|
|
- //defer cursor.Close(context.Background())
|
|
|
- //log.Println("合并执行完成", gtid, lteid, overid)
|
|
|
- //if overid != lteid {
|
|
|
- // by, _ := json.Marshal(map[string]interface{}{
|
|
|
- // "gtid": overid,
|
|
|
- // "lteid": lteid,
|
|
|
- // "stype": "",
|
|
|
- // })
|
|
|
- // if e := udpclient.WriteUdp(by, mu.OP_TYPE_DATA, &net.UDPAddr{
|
|
|
- // IP: net.ParseIP("127.0.0.1"),
|
|
|
- // Port: Updport,
|
|
|
- // }); e != nil {
|
|
|
- // log.Println(e)
|
|
|
- // }
|
|
|
- // log.Println("重新发送udp:", string(by))
|
|
|
- // return
|
|
|
- //}
|
|
|
- log.Println("合并执行完成 ok", gtid, lteid, overid)
|
|
|
|
|
|
}
|
|
|
|
|
|
@@ -384,6 +519,7 @@ func TimedTaskAgency() {
|
|
|
}
|
|
|
contact["topscopeclass"] = "企业公示" //项目类型
|
|
|
contact["updatetime"] = time.Now().Unix() //更新时间
|
|
|
+ contact["infoid"] = "" //招标信息id
|
|
|
contacts = append(contacts, contact)
|
|
|
resulttmp["contact"] = contacts
|
|
|
|
