|
@@ -4,7 +4,6 @@ import (
|
|
|
"encoding/json"
|
|
|
"fmt"
|
|
|
"github.com/garyburd/redigo/redis"
|
|
|
- "go.mongodb.org/mongo-driver/bson/primitive"
|
|
|
"gopkg.in/mgo.v2/bson"
|
|
|
"log"
|
|
|
mu "mfw/util"
|
|
@@ -14,6 +13,7 @@ import (
|
|
|
"strconv"
|
|
|
"strings"
|
|
|
"time"
|
|
|
+ "unicode/utf8"
|
|
|
)
|
|
|
|
|
|
//之前main方法,只更新
|
|
@@ -26,51 +26,68 @@ func TaskWinner(mapinfo *map[string]interface{}) {
|
|
|
log.Println(gtid, lteid, "参数错误")
|
|
|
return
|
|
|
}
|
|
|
- GId, err := primitive.ObjectIDFromHex(gtid)
|
|
|
- LtId, err2 := primitive.ObjectIDFromHex(lteid)
|
|
|
- if err != nil || err2 != nil {
|
|
|
- log.Println(gtid, lteid, "转换_id错误")
|
|
|
+ var GId, LtId bson.ObjectId
|
|
|
+ if bson.IsObjectIdHex(gtid) && bson.IsObjectIdHex(lteid) {
|
|
|
+ GId = bson.ObjectIdHex(gtid)
|
|
|
+ LtId = bson.ObjectIdHex(lteid)
|
|
|
+ } else {
|
|
|
+ log.Println(gtid, lteid, "不是Objectid,转换_id错误", gtid, lteid)
|
|
|
return
|
|
|
}
|
|
|
//udp的id区间查询bidding 中标人 中标联系人 中标联系电话
|
|
|
// topscopeclass项目类型-industry行业类型&&topscopeclass联系人项目类型
|
|
|
// (area地区-province省份 city城市-city城市 district区县-district区县)
|
|
|
// winneraddr-company_address企业地址
|
|
|
- SourceClientcc := SourceClient.GetMgoConn()
|
|
|
- defer SourceClient.DestoryMongoConn(SourceClientcc)
|
|
|
+ SourceClientcc := SourceClient.GetMgoConn(86400)
|
|
|
cursor := SourceClientcc.DB(Config["mgodb_bidding"]).C(Config["mgodb_mgoinit_c"]).Find(bson.M{
|
|
|
"_id": bson.M{
|
|
|
"$gte": GId,
|
|
|
"$lte": LtId,
|
|
|
},
|
|
|
- }).Select(bson.M{"winner": 1, "winnertel": 1, "winnerperson": 1,
|
|
|
- "topscopeclass": 1, "winneraddr": 1}).Iter()
|
|
|
- if cursor == nil {
|
|
|
- log.Println(cursor)
|
|
|
+ }).Select(bson.M{"winner": 1, "winnertel": 1, "winnerperson": 1, "topscopeclass": 1, "winneraddr": 1}).Iter()
|
|
|
+
|
|
|
+ if cursor.Err() != nil {
|
|
|
+ SourceClient.DestoryMongoConn(SourceClientcc)
|
|
|
+ log.Println(cursor.Err())
|
|
|
return
|
|
|
}
|
|
|
//判断是否是存量,是存量走Redis遍历
|
|
|
if v, ok := (*mapinfo)["data_info"].(string); ok && v == "save" {
|
|
|
//存量处理
|
|
|
- tmp := map[string]interface{}{}
|
|
|
conn := HisRedisPool.Conn()
|
|
|
defer conn.Close()
|
|
|
//选择redis db
|
|
|
redis_winner_db, _ := strconv.Atoi(Config["redis_winner_db"])
|
|
|
conn.Select(redis_winner_db)
|
|
|
//遍历bidding表保存到redis
|
|
|
- // key:_id value:json结构体
|
|
|
+ //key:企业名 value:json结构体{"winner": 1, "winnertel": 1, "winnerperson": 1,"topscopeclass": 1, "winneraddr": 1,"_id":1}
|
|
|
+ tmp := make(map[string]interface{})
|
|
|
for cursor.Next(&tmp) {
|
|
|
- if tmp["winner"] == nil || tmp["winner"] == "" {
|
|
|
+ winner, ok := tmp["winner"].(string)
|
|
|
+ if !ok ||utf8.RuneCountInString(winner)<4 {
|
|
|
continue
|
|
|
}
|
|
|
- mgoId := tmp["_id"].(primitive.ObjectID).Hex()
|
|
|
- delete(tmp, "_id")
|
|
|
- bytes, _ := json.Marshal(tmp)
|
|
|
- if err := conn.Set(mgoId, string(bytes), 0).Err(); err != nil {
|
|
|
+ //判断redis key是否存在
|
|
|
+ e_num := conn.Exists(winner).Val()
|
|
|
+ //获取字符串_id
|
|
|
+ mgoId := tmp["_id"].(bson.ObjectId).Hex()
|
|
|
+ //替换_id
|
|
|
+ tmp["_id"] = mgoId
|
|
|
+ //创建value数组
|
|
|
+ tmps := make([]map[string]interface{}, 0)
|
|
|
+ if e_num > 0 {
|
|
|
+ //存量redis的key存在,累加更新
|
|
|
+ bytes, _ := conn.Get(winner).Bytes()
|
|
|
+ json.Unmarshal(bytes, &tmps)
|
|
|
+ }
|
|
|
+ tmps = append(tmps, tmp)
|
|
|
+ bytes, _ := json.Marshal(tmps)
|
|
|
+ //存量redis的key不存在,新增 key :企业名 val :[]map
|
|
|
+ if err := conn.Set(winner, string(bytes), 0).Err(); err != nil {
|
|
|
log.Println(err)
|
|
|
}
|
|
|
}
|
|
|
+ SourceClient.DestoryMongoConn(SourceClientcc)
|
|
|
//遍历redis
|
|
|
if scan := conn.Scan(0, "", 100); scan.Err() != nil {
|
|
|
log.Println(scan.Err())
|
|
@@ -78,166 +95,122 @@ func TaskWinner(mapinfo *map[string]interface{}) {
|
|
|
} else {
|
|
|
iterator := scan.Iterator()
|
|
|
for iterator.Next() {
|
|
|
- redisId := iterator.Val() //redis key
|
|
|
- redisvalue := conn.Get(iterator.Val()).Val() //redis val
|
|
|
- tmp := make(map[string]interface{})
|
|
|
- json.Unmarshal([]byte(redisvalue), &tmp)
|
|
|
- //重复增量操作
|
|
|
+ redisCName := iterator.Val() //redis key 企业名
|
|
|
+ redisvalueBytes, _ := conn.Get(redisCName).Bytes() //redis val []数组
|
|
|
+ rValuesMaps := make([]map[string]interface{}, 0)
|
|
|
+ json.Unmarshal(redisvalueBytes, &rValuesMaps)
|
|
|
//redis查询是否存在
|
|
|
rdb := RedisPool.Get()
|
|
|
rdb.Do("SELECT", Config["redis_winner_db"])
|
|
|
- if reply, err := redis.String(rdb.Do("GET", tmp["winner"])); err != nil {
|
|
|
+ if reply, err := redis.String(rdb.Do("GET", redisCName)); 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 tmpid := FClient.Save("winner_new", tmps); tmpid == nil {
|
|
|
+ // log.Println("存量 FClient.Save err", tmpid)
|
|
|
+ //}
|
|
|
+ for _, vmap := range rValuesMaps {
|
|
|
+ vmap["_id"] = bson.ObjectIdHex(vmap["_id"].(string))
|
|
|
+ if err = FClient.SaveForOld("winner_new", vmap); err != nil {
|
|
|
+ log.Println("存量 FClient.Save err", err, vmap)
|
|
|
+ }
|
|
|
}
|
|
|
//log.Println("get redis id err:定时任务处理", err, tmp)
|
|
|
if err := rdb.Close(); err != nil {
|
|
|
log.Println("存量", err)
|
|
|
}
|
|
|
- //删除存量redis
|
|
|
- conn.Del(redisId)
|
|
|
continue
|
|
|
} else {
|
|
|
+ //redis存在更新合并
|
|
|
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"])
|
|
|
+ oldTmp, b := FClient.FindById(Config["mgo_qyk_c"], reply, nil)
|
|
|
+ if !b || oldTmp == nil {
|
|
|
+ log.Println(redisCName, "存量 redis id 不存在", reply)
|
|
|
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 {
|
|
|
+ if (*oldTmp)["industry"] != nil {
|
|
|
+ if v, ok := (*oldTmp)["industry"].([]interface{}); 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 _, rvaluemaps := range rValuesMaps {
|
|
|
+ if tclasss, ok := rvaluemaps["topscopeclass"].([]string); ok {
|
|
|
+ for _, vv := range tclasss {
|
|
|
+ if len(vv) > 1 {
|
|
|
+ tmpTopscopeclassMap[vv[:len(vv)-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)
|
|
|
+ (*oldTmp)["industry"] = tmpTopscopeclass
|
|
|
+ esId := (*oldTmp)["_id"].(bson.ObjectId).Hex()
|
|
|
+ //联系方式合并
|
|
|
+ contactMaps := make([]interface{}, 0)
|
|
|
+ if (*oldTmp)["contact"] != nil {
|
|
|
+ //直接添加联系人,不再判断
|
|
|
+ if v, ok := (*oldTmp)["contact"].([]interface{}); ok {
|
|
|
+ contactMaps = append(contactMaps, v...)
|
|
|
}
|
|
|
- //删除存量redis
|
|
|
- conn.Del(redisId)
|
|
|
- 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"])) {
|
|
|
- winnertel = ""
|
|
|
+ //遍历redis value联系人
|
|
|
+ for _, rvmap := range rValuesMaps {
|
|
|
+ var tmpperson, winnertel string
|
|
|
+ if rvmapperson, ok := rvmap["winnerperson"].(string); ok && rvmapperson != "" {
|
|
|
+ tmpperson = rvmapperson
|
|
|
} else {
|
|
|
- winnertel = util.ObjToString(tmp["winnertel"])
|
|
|
+ continue
|
|
|
+ }
|
|
|
+ if rvmapwintel, ok := rvmap["winnertel"].(string); ok {
|
|
|
+ winnertel = rvmapwintel
|
|
|
+ } else {
|
|
|
+ winnertel = ""
|
|
|
+ }
|
|
|
+ if Reg_xing.MatchString(winnertel) || !Reg_tel.MatchString(winnertel) {
|
|
|
+ winnertel = ""
|
|
|
}
|
|
|
- }
|
|
|
- contactMaps := make([]interface{}, 0)
|
|
|
- if oldTmp["contact"] == nil {
|
|
|
tmpContact := make(map[string]interface{})
|
|
|
- tmpContact["infoid"] = redisId
|
|
|
+ tmpContact["infoid"] = rvmap["_id"]
|
|
|
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)
|
|
|
+ tmpclass := make([]string, 0)
|
|
|
+ if tclasss, ok := rvmap["topscopeclass"].([]string); ok {
|
|
|
+ for _, vv := range tclasss {
|
|
|
+ if len(vv) > 1 {
|
|
|
+ tmpclass = append(tmpclass, vv[:len(vv)-1])
|
|
|
}
|
|
|
}
|
|
|
- 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)
|
|
|
- }
|
|
|
- }*/
|
|
|
- //直接添加联系人,不再判断
|
|
|
- if v, ok := oldTmp["contact"].(primitive.A); ok {
|
|
|
- contactMaps = append(contactMaps, v...)
|
|
|
}
|
|
|
- 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)
|
|
|
+ tmpContact["topscopeclass"] = strings.Join(tmpclass, ";")
|
|
|
+ tmpContact["updatetime"] = time.Now().Unix()
|
|
|
+ contactMaps = append(contactMaps, tmpContact)
|
|
|
}
|
|
|
- oldTmp["contact"] = contactMaps
|
|
|
+ (*oldTmp)["contact"] = contactMaps
|
|
|
//mongo更新
|
|
|
- oldTmp["updatatime"] = time.Now().Unix()
|
|
|
+ (*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")
|
|
|
+ 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)
|
|
|
}
|
|
|
}
|
|
|
}
|
|
@@ -248,19 +221,20 @@ func TaskWinner(mapinfo *map[string]interface{}) {
|
|
|
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"] == "" {
|
|
|
+ overid = tmp["_id"].(bson.ObjectId).Hex()
|
|
|
+ //log.Println(tmp["_id"])
|
|
|
+ winner, ok := tmp["winner"].(string)
|
|
|
+ if !ok || utf8.RuneCountInString(winner) < 4 {
|
|
|
continue
|
|
|
}
|
|
|
//redis查询是否存在
|
|
|
rdb := RedisPool.Get()
|
|
|
rdb.Do("SELECT", Config["redis_winner_db"])
|
|
|
- if reply, err := redis.String(rdb.Do("GET", tmp["winner"])); err != nil {
|
|
|
+ if reply, err := redis.String(rdb.Do("GET", 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.SaveForOld("winner_new", tmp); err != nil {
|
|
|
+ log.Println("FClient.Save err", err, tmp)
|
|
|
}
|
|
|
//log.Println("get redis id err:定时任务处理", err, tmp)
|
|
|
if err := rdb.Close(); err != nil {
|
|
@@ -273,53 +247,44 @@ func TaskWinner(mapinfo *map[string]interface{}) {
|
|
|
}
|
|
|
//拿到合并后的qyk
|
|
|
FClient.DbName = Config["mgodb_extract_kf"]
|
|
|
- oldTmp := FClient.FindById(Config["mgo_qyk_c"], reply)
|
|
|
- if oldTmp == nil {
|
|
|
+ oldTmp, b := FClient.FindById(Config["mgo_qyk_c"], reply, bson.M{})
|
|
|
+ if !b || oldTmp == nil {
|
|
|
log.Println("redis id 不存在")
|
|
|
continue
|
|
|
}
|
|
|
//比较合并
|
|
|
//行业类型
|
|
|
tmpTopscopeclass := []string{}
|
|
|
+ tmpConTopscopeclass := []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 {
|
|
|
+ if (*oldTmp)["industry"] != nil {
|
|
|
+ if v, ok := (*oldTmp)["industry"].([]interface{}); 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)
|
|
|
+ }
|
|
|
+ if v, ok := tmp["topscopeclass"].([]interface{}); ok {
|
|
|
+ for _, vv := range v {
|
|
|
+ if vvv, ok := vv.(string); ok && len(vvv) > 1 {
|
|
|
+ tmpTopscopeclassMap[vvv[:len(vvv)-1]] = true
|
|
|
+ tmpConTopscopeclass = append(tmpConTopscopeclass, vvv[:len(vvv)-1])
|
|
|
}
|
|
|
}
|
|
|
}
|
|
|
+ for k := range tmpTopscopeclassMap {
|
|
|
+ tmpTopscopeclass = append(tmpTopscopeclass, k)
|
|
|
+ }
|
|
|
sort.Strings(tmpTopscopeclass)
|
|
|
- oldTmp["industry"] = tmpTopscopeclass
|
|
|
+ (*oldTmp)["industry"] = tmpTopscopeclass
|
|
|
|
|
|
- esId := oldTmp["_id"].(primitive.ObjectID).Hex()
|
|
|
+ esId := (*oldTmp)["_id"].(bson.ObjectId).Hex()
|
|
|
//更新行业类型
|
|
|
if tmp["winnerperson"] == nil || tmp["winnerperson"] == "" || Reg_xing.MatchString(util.ObjToString(tmp["winnerperson"])) {
|
|
|
- oldTmp["updatatime"] = time.Now().Unix()
|
|
|
+ (*oldTmp)["updatatime"] = time.Now().Unix()
|
|
|
//mongo更新
|
|
|
FClient.DbName = Config["mgodb_extract_kf"]
|
|
|
if !FClient.UpdateById(Config["mgo_qyk_c"], esId, bson.M{"$set": oldTmp}) {
|
|
@@ -327,7 +292,7 @@ func TaskWinner(mapinfo *map[string]interface{}) {
|
|
|
}
|
|
|
|
|
|
//es更新
|
|
|
- delete(oldTmp, "_id")
|
|
|
+ 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)
|
|
|
}
|
|
@@ -335,78 +300,47 @@ func TaskWinner(mapinfo *map[string]interface{}) {
|
|
|
}
|
|
|
//联系方式合并
|
|
|
var tmpperson, winnertel string
|
|
|
- tmpperson = tmp["winnerperson"].(string)
|
|
|
- if tmp["winnertel"] == nil || tmp["winnertel"] == "" {
|
|
|
+ if tmppersona, ok := tmp["winnerperson"].(string); ok {
|
|
|
+ tmpperson = tmppersona
|
|
|
+ }
|
|
|
+ if winnerteltmp, ok := tmp["winnertel"].(string); ok {
|
|
|
+ winnertel = winnerteltmp
|
|
|
+ }
|
|
|
+ if Reg_xing.MatchString(winnertel) || !Reg_tel.MatchString(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"])
|
|
|
- }
|
|
|
+ winnertel = winnertel
|
|
|
}
|
|
|
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 (*oldTmp)["contact"] != nil {
|
|
|
//直接添加联系人,不再判断
|
|
|
- if v, ok := oldTmp["contact"].(primitive.A); ok {
|
|
|
+ if v, ok := (*oldTmp)["contact"].([]interface{}); ok {
|
|
|
contactMaps = append(contactMaps, v...)
|
|
|
}
|
|
|
- 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)
|
|
|
}
|
|
|
- oldTmp["contact"] = contactMaps
|
|
|
+ vvv := make(map[string]interface{})
|
|
|
+ vvv["infoid"] = overid
|
|
|
+ vvv["contact_person"] = tmpperson
|
|
|
+ vvv["contact_type"] = "项目联系人"
|
|
|
+ vvv["phone"] = winnertel
|
|
|
+ vvv["topscopeclass"] = strings.Join(tmpConTopscopeclass, ";")
|
|
|
+ vvv["updatetime"] = time.Now().Unix()
|
|
|
+ contactMaps = append(contactMaps, vvv)
|
|
|
+ (*oldTmp)["contact"] = contactMaps
|
|
|
//mongo更新
|
|
|
- oldTmp["updatatime"] = time.Now().Unix()
|
|
|
+ (*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")
|
|
|
+ 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)
|
|
|
}
|
|
|
}
|
|
|
}
|
|
|
+ SourceClient.DestoryMongoConn(SourceClientcc)
|
|
|
log.Println("增量合并执行完成 ok", gtid, lteid, overid)
|
|
|
}
|
|
|
|
|
@@ -419,19 +353,18 @@ func TimedTaskWinner() {
|
|
|
//time.Sleep(time.Hour*70)
|
|
|
t2 := time.NewTimer(time.Second * 5)
|
|
|
for range t2.C {
|
|
|
- Fcconn := FClient.GetMgoConn()
|
|
|
- defer FClient.DestoryMongoConn(Fcconn)
|
|
|
+ Fcconn := FClient.GetMgoConn(86400)
|
|
|
tmpLast := map[string]interface{}{}
|
|
|
if iter := Fcconn.DB(Config["mgodb_extract_kf"]).C("winner_new").Find(bson.M{}).Sort("-_id").Limit(1).Iter(); iter != nil {
|
|
|
if !iter.Next(&tmpLast) {
|
|
|
//临时表无数据
|
|
|
log.Println("临时表无数据:")
|
|
|
t2.Reset(time.Minute * 5)
|
|
|
+ FClient.DestoryMongoConn(Fcconn)
|
|
|
continue
|
|
|
} else {
|
|
|
log.Println("临时表有数据:", tmpLast)
|
|
|
- fconn := FClient.GetMgoConn()
|
|
|
- defer FClient.DestoryMongoConn(fconn)
|
|
|
+ fconn := FClient.GetMgoConn(86400)
|
|
|
cursor := fconn.DB(Config["mgodb_extract_kf"]).C("winner_new").Find(bson.M{
|
|
|
"_id": bson.M{
|
|
|
"$lte": tmpLast["_id"],
|
|
@@ -440,17 +373,17 @@ func TimedTaskWinner() {
|
|
|
if cursor == nil {
|
|
|
log.Println("查询失败")
|
|
|
t2.Reset(time.Second * 5)
|
|
|
+ FClient.DestoryMongoConn(fconn)
|
|
|
continue
|
|
|
}
|
|
|
//遍历临时表数据,匹配不到原始库存入异常表
|
|
|
tmp := make(map[string]interface{})
|
|
|
for cursor.Next(&tmp) {
|
|
|
- tmpId := tmp["_id"].(primitive.ObjectID).Hex()
|
|
|
+ tmpId := tmp["_id"].(bson.ObjectId).Hex()
|
|
|
//再重新查找redis,存在发udp处理,不存在走新增合并
|
|
|
rdb := RedisPool.Get()
|
|
|
rdb.Do("SELECT", Config["redis_winner_db"])
|
|
|
if _, err := redis.String(rdb.Do("GET", tmp["winner"])); err == nil {
|
|
|
- //{"gtid":"57d7ad2f61a0721f152d2ad5","lteid":"5e20968d85a9271abf0ad6c2","stype":""}
|
|
|
//redis存在发送udp进行处理
|
|
|
by, _ := json.Marshal(map[string]interface{}{
|
|
|
"gtid": tmpId,
|
|
@@ -467,7 +400,7 @@ func TimedTaskWinner() {
|
|
|
}
|
|
|
//存在的话删除tmp mongo表
|
|
|
FClient.DbName = Config["mgodb_extract_kf"]
|
|
|
- if DeletedCount := FClient.DeleteById("winner_new", tmpId); DeletedCount == 0 {
|
|
|
+ if DeletedCount := FClient.Del("winner_new", bson.M{"_id": bson.ObjectIdHex(tmpId)}); !DeletedCount {
|
|
|
log.Println("删除临时表err:", DeletedCount)
|
|
|
}
|
|
|
if err := rdb.Close(); err != nil {
|
|
@@ -482,33 +415,32 @@ func TimedTaskWinner() {
|
|
|
//查询redis不存在新增
|
|
|
FClient.DbName = Config["mgodb_enterprise"]
|
|
|
|
|
|
- resulttmp := FClient.FindOne(Config["mgodb_enterprise_c"], bson.M{"company_name": tmp["winner"]})
|
|
|
- if resulttmp["_id"] == nil {
|
|
|
+ resulttmp, b := FClient.FindOne(Config["mgodb_enterprise_c"], bson.M{"company_name": tmp["winner"]})
|
|
|
+ if !b || (*resulttmp)["_id"] == nil {
|
|
|
//log.Println(r)
|
|
|
//匹配不到原始库,存入异常表删除临时表
|
|
|
FClient.DbName = Config["mgodb_extract_kf"]
|
|
|
- if saveid := FClient.Save("winner_err", tmp); saveid == nil {
|
|
|
- log.Println("存入异常表错误", tmp)
|
|
|
+ if err := FClient.SaveForOld("winner_err", tmp); err != nil {
|
|
|
+ log.Println("存入异常表错误", err, tmp)
|
|
|
}
|
|
|
- FClient.DbName = Config["mgodb_extract_kf"]
|
|
|
- if deleteNum := FClient.DeleteById("winner_new", tmpId); deleteNum == 0 {
|
|
|
+ if deleteNum := FClient.Del("winner_new", bson.M{"_id": bson.ObjectIdHex(tmpId)}); !b {
|
|
|
log.Println("删除临时表错误", deleteNum)
|
|
|
}
|
|
|
continue
|
|
|
} else {
|
|
|
//log.Println(123)
|
|
|
//匹配到原始库,新增 resulttmp
|
|
|
- if resulttmp["credit_no"] != nil {
|
|
|
- if credit_no, ok := resulttmp["credit_no"].(string); ok && strings.TrimSpace(credit_no) != "" &&
|
|
|
+ if (*resulttmp)["credit_no"] != nil {
|
|
|
+ if credit_no, ok := (*resulttmp)["credit_no"].(string); ok && strings.TrimSpace(credit_no) != "" &&
|
|
|
len(strings.TrimSpace(credit_no)) > 8 {
|
|
|
dataNo := strings.TrimSpace(credit_no)[2:8]
|
|
|
if Addrs[dataNo] != nil {
|
|
|
if v, ok := Addrs[dataNo].(map[string]interface{}); ok {
|
|
|
- if resulttmp["province"] == nil || resulttmp["province"] == "" {
|
|
|
- resulttmp["province"] = v["province"]
|
|
|
+ if (*resulttmp)["province"] == nil || (*resulttmp)["province"] == "" {
|
|
|
+ (*resulttmp)["province"] = v["province"]
|
|
|
}
|
|
|
- resulttmp["city"] = v["city"]
|
|
|
- resulttmp["district"] = v["district"]
|
|
|
+ (*resulttmp)["city"] = v["city"]
|
|
|
+ (*resulttmp)["district"] = v["district"]
|
|
|
|
|
|
}
|
|
|
}
|
|
@@ -516,15 +448,15 @@ func TimedTaskWinner() {
|
|
|
}
|
|
|
contacts := make([]map[string]interface{}, 0)
|
|
|
contact := make(map[string]interface{}, 0)
|
|
|
- if resulttmp["legal_person"] != nil {
|
|
|
- contact["contact_person"] = resulttmp["legal_person"] //联系人
|
|
|
+ if (*resulttmp)["legal_person"] != nil {
|
|
|
+ contact["contact_person"] = (*resulttmp)["legal_person"] //联系人
|
|
|
} else {
|
|
|
contact["contact_person"] = "" //联系人
|
|
|
}
|
|
|
contact["contact_type"] = "法定代表人" //法定代表人
|
|
|
//log.Println(1)
|
|
|
- if resulttmp["annual_reports"] != nil {
|
|
|
- bytes, err := json.Marshal(resulttmp["annual_reports"])
|
|
|
+ if (*resulttmp)["annual_reports"] != nil {
|
|
|
+ bytes, err := json.Marshal((*resulttmp)["annual_reports"])
|
|
|
if err != nil {
|
|
|
log.Println("annual_reports err:", err)
|
|
|
}
|
|
@@ -556,24 +488,49 @@ func TimedTaskWinner() {
|
|
|
contact["updatetime"] = time.Now().Unix() //更新时间
|
|
|
contact["infoid"] = "" //招标信息id
|
|
|
contacts = append(contacts, contact)
|
|
|
- resulttmp["contact"] = contacts
|
|
|
+ //添加临时表匹配到的联系人
|
|
|
+ vvv := make(map[string]interface{})
|
|
|
+ vvv["infoid"] = tmp["_id"].(bson.ObjectId).Hex()
|
|
|
+ if tmp["winnerperson"] != nil {
|
|
|
+ vvv["contact_person"] = tmp["winnerperson"]
|
|
|
+ } else {
|
|
|
+ vvv["contact_person"] = ""
|
|
|
+ }
|
|
|
+ vvv["contact_type"] = "项目联系人"
|
|
|
+ // "winner": 1, "winnertel": 1, "winnerperson": 1, "topscopeclass": 1
|
|
|
+ if tmp["winnertel"] != nil {
|
|
|
+ vvv["phone"] = tmp["winnertel"]
|
|
|
+ } else {
|
|
|
+ vvv["phone"] = ""
|
|
|
+ }
|
|
|
+ tmpclass := make([]string, 0)
|
|
|
+ if tclasss, ok := tmp["topscopeclass"].([]string); ok {
|
|
|
+ for _, vv := range tclasss {
|
|
|
+ if len(vv) > 1 {
|
|
|
+ tmpclass = append(tmpclass, vv[:len(vv)-1])
|
|
|
+ }
|
|
|
+ }
|
|
|
+ }
|
|
|
+ vvv["topscopeclass"] = strings.Join(tmpclass, ";")
|
|
|
+ vvv["updatetime"] = time.Now().Unix()
|
|
|
+ contacts = append(contacts, vvv)
|
|
|
+ (*resulttmp)["contact"] = contacts
|
|
|
|
|
|
savetmp := make(map[string]interface{}, 0)
|
|
|
for _, sk := range Fields {
|
|
|
if sk == "establish_date" {
|
|
|
- if resulttmp[sk] != nil {
|
|
|
- savetmp[sk] = resulttmp[sk].(primitive.DateTime).Time().UTC().Unix()
|
|
|
+ if (*resulttmp)[sk] != nil {
|
|
|
+ savetmp[sk] = (*resulttmp)[sk].(time.Time).UTC().Unix()
|
|
|
continue
|
|
|
}
|
|
|
} else if sk == "capital" {
|
|
|
//log.Println(sk, resulttmp[sk])
|
|
|
- savetmp[sk] = ObjToMoney([]interface{}{resulttmp[sk], ""})[0]
|
|
|
+ savetmp[sk] = ObjToMoney([]interface{}{(*resulttmp)[sk], ""})[0]
|
|
|
continue
|
|
|
} else if sk == "partners" {
|
|
|
//log.Println(sk, resulttmp[sk], )
|
|
|
- //fmt.Println(reflect.TypeOf(resulttmp[sk]))
|
|
|
- if resulttmp[sk] != nil {
|
|
|
- if ppms, ok := resulttmp[sk].(primitive.A); ok {
|
|
|
+ if (*resulttmp)[sk] != nil {
|
|
|
+ if ppms, ok := (*resulttmp)[sk].([]interface{}); ok {
|
|
|
for i, _ := range ppms {
|
|
|
if ppms[i].(map[string]interface{})["stock_type"] != nil {
|
|
|
ppms[i].(map[string]interface{})["stock_type"] = "企业公示"
|
|
@@ -588,19 +545,19 @@ func TimedTaskWinner() {
|
|
|
}
|
|
|
continue
|
|
|
} else if sk == "_id" {
|
|
|
- savetmp["tmp"+sk] = resulttmp[sk]
|
|
|
+ savetmp["tmp"+sk] = (*resulttmp)[sk]
|
|
|
continue
|
|
|
} else if sk == "area_code" {
|
|
|
//行政区划代码
|
|
|
- savetmp[sk] = fmt.Sprint(resulttmp[sk])
|
|
|
+ savetmp[sk] = fmt.Sprint((*resulttmp)[sk])
|
|
|
continue
|
|
|
} else if sk == "report_websites" {
|
|
|
//网址
|
|
|
- if resulttmp["report_websites"] == nil {
|
|
|
+ if (*resulttmp)["report_websites"] == nil {
|
|
|
savetmp["website"] = ""
|
|
|
} else {
|
|
|
report_websitesArr := []string{}
|
|
|
- if ppms, ok := resulttmp[sk].(primitive.A); ok {
|
|
|
+ if ppms, ok := (*resulttmp)[sk].([]interface{}); ok {
|
|
|
for _, v := range ppms {
|
|
|
if vvv, ok := v.(map[string]interface{}); ok {
|
|
|
if rv, ok := vvv["website_url"].(string); ok {
|
|
@@ -616,27 +573,40 @@ func TimedTaskWinner() {
|
|
|
} else if sk == "wechat_accounts" {
|
|
|
savetmp[sk] = []interface{}{}
|
|
|
continue
|
|
|
+ } else if sk == "industry" {
|
|
|
+ tmpTopscopeclass := make([]string, 0)
|
|
|
+ if v, ok := tmp["topscopeclass"].([]interface{}); ok {
|
|
|
+ for _, vv := range v {
|
|
|
+ if vvv, ok := vv.(string); ok && len(vvv) > 1 {
|
|
|
+ tmpTopscopeclass = append(tmpTopscopeclass, vvv[:len(vvv)-1])
|
|
|
+ }
|
|
|
+ }
|
|
|
+ }
|
|
|
+ sort.Strings(tmpTopscopeclass)
|
|
|
+ savetmp[sk] = tmpTopscopeclass
|
|
|
+ continue
|
|
|
}
|
|
|
- if resulttmp[sk] == nil && sk != "history_name" && sk != "wechat_accounts" && sk != "establish_date" && sk != "capital" && sk != "partners" && sk != "contact" && sk != "report_websites" {
|
|
|
+ if (*resulttmp)[sk] == nil && sk != "history_name" && sk != "wechat_accounts" && sk != "establish_date" && sk != "capital" && sk != "partners" && sk != "contact" && sk != "report_websites" {
|
|
|
savetmp[sk] = ""
|
|
|
} else {
|
|
|
- savetmp[sk] = resulttmp[sk]
|
|
|
+ savetmp[sk] = (*resulttmp)[sk]
|
|
|
}
|
|
|
}
|
|
|
//tmps = append(tmps, savetmp)
|
|
|
savetmp["updatatime"] = time.Now().Unix()
|
|
|
//保存mongo
|
|
|
FClient.DbName = Config["mgodb_extract_kf"]
|
|
|
+
|
|
|
saveid := FClient.Save(Config["mgo_qyk_c"], savetmp)
|
|
|
- if saveid != nil {
|
|
|
+ if saveid != "" {
|
|
|
//保存redis
|
|
|
rc := RedisPool.Get()
|
|
|
rc.Do("SELECT", Config["redis_winner_db"])
|
|
|
- var _id string
|
|
|
- if v, ok := saveid.(primitive.ObjectID); ok {
|
|
|
- _id = v.Hex()
|
|
|
- }
|
|
|
- if _, err := rc.Do("SET", savetmp["company_name"], _id); err != nil {
|
|
|
+ //var _id string
|
|
|
+ //if v, ok := saveid.(primitive.ObjectID); ok {
|
|
|
+ // _id = v.Hex()
|
|
|
+ //}
|
|
|
+ if _, err := rc.Do("SET", savetmp["company_name"], saveid); err != nil {
|
|
|
log.Println("save redis err:", tmp["_id"], savetmp["_id"], savetmp["company_name"], err)
|
|
|
if err := rc.Close(); err != nil {
|
|
|
log.Println(err)
|
|
@@ -650,12 +620,12 @@ func TimedTaskWinner() {
|
|
|
|
|
|
//esConn := elastic.GetEsConn()
|
|
|
//defer elastic.DestoryEsConn(esConn)
|
|
|
- if _, err := EsConn.Index().Index(Config["elasticsearch_index"]).Type(Config["elasticsearch_type"]).Id(_id).BodyJson(savetmp).Refresh(true).Do(); err != nil {
|
|
|
+ if _, err := EsConn.Index().Index(Config["elasticsearch_index"]).Type(Config["elasticsearch_type"]).Id(saveid).BodyJson(savetmp).Refresh(true).Do(); err != nil {
|
|
|
log.Println("save es err :", tmp["_id"], savetmp["_id"], err)
|
|
|
} else {
|
|
|
//删除临时表
|
|
|
FClient.DbName = Config["mgodb_extract_kf"]
|
|
|
- if deleteNum := FClient.DeleteById("winner_new", tmpId); deleteNum == 0 {
|
|
|
+ if deleteNum := FClient.Del("winner_new", bson.M{"_id": bson.ObjectIdHex(tmpId)}); !deleteNum {
|
|
|
log.Println("删除临时表失败", deleteNum)
|
|
|
}
|
|
|
}
|
|
@@ -665,8 +635,10 @@ func TimedTaskWinner() {
|
|
|
}
|
|
|
}
|
|
|
}
|
|
|
+ FClient.DestoryMongoConn(fconn)
|
|
|
}
|
|
|
}
|
|
|
+ FClient.DestoryMongoConn(Fcconn)
|
|
|
t2.Reset(time.Minute)
|
|
|
}
|
|
|
}
|