修改链接池

修改中标单位标准库逻辑

udp_winner/main.go

@@ -5,16 +5,16 @@ import (
 	"fmt"
 	"github.com/garyburd/redigo/redis"
 	hisRedis "github.com/go-redis/redis"
-	"go.mongodb.org/mongo-driver/bson"
+	"gopkg.in/mgo.v2/bson"
 	es "gopkg.in/olivere/elastic.v1"
 	"log"
 	mu "mfw/util"
 	"net"
 	"qfw/common/src/qfw/util/elastic"
 	"qfw/util"
+	"qfw/util/mongodb"
 	"regexp"
 	"strconv"
-
 	"strings"
 	"time"
 )
@@ -22,7 +22,7 @@ import (
 var (
 	Config                                = make(map[string]string)
 	Fields, BuyerFields, AgencyFields     []string
-	SourceClient, FClient                 *MongodbSim
+	SourceClient, FClient                 *mongodb.MongodbSim
 	RedisPool                             redis.Pool
 	HisRedisPool                          *hisRedis.Client
 	Addrs                                 = make(map[string]interface{}, 0) //省市县
@@ -46,15 +46,15 @@ func init() {
 	log.Println(Config)
 	var err error
 	cpnum, err := strconv.Atoi(Config["chan_pool_num"])
-	if err != nil{
+	if err != nil {
 		log.Fatalln(err)
 	}
-	CPool = make(chan bool,cpnum)
+	CPool = make(chan bool, cpnum)
 	Fields = []string{"_id", "contact", "partners", "business_scope", "company_address",
 		"capital", "establish_date", "legal_person", "company_type",
 		"district", "city", "province", "area_code", "credit_no",
-		"company_name", "history_name", "topscopeclass", "wechat_accounts",
-		"alias", "website", "report_websites"}
+		"company_name", "history_name", "wechat_accounts",
+		"alias", "website", "report_websites","industry"}
 
 	BuyerFields = []string{"_id", "contact", "type", "ranks", "buyerclass",
 		"address", "district", "city", "province", "area_code", "credit_no", "buyer_name",
@@ -66,14 +66,13 @@ func init() {
 	pool_size, _ := strconv.Atoi(Config["pool_size"])
 
 	//mongo init
-	SourceClient = new(MongodbSim)
+	SourceClient = new(mongodb.MongodbSim)
 	SourceClient.MongodbAddr = Config["mgoinit"]
 	SourceClient.Size = pool_size
+	SourceClient.DbName = Config["mgodb_bidding"]
 	//mongodbSim.DbName = "qfw"
 	SourceClient.InitPool()
-	SourceClientmgoConn := SourceClient.GetMgoConn()
-	defer SourceClient.DestoryMongoConn(SourceClientmgoConn)
-	FClient = new(MongodbSim)
+	FClient = new(mongodb.MongodbSim)
 	FClient.MongodbAddr = Config["mgourl"]
 	FClient.Size = pool_size
 	FClient.DbName = Config["mgodb_extract_kf"]
@@ -141,8 +140,8 @@ func main() {
 	log.Println("Udp服务监听", updport)
 	log.Println("发送端口port：", Updport)
 	go TimedTaskWinner() //定时任务
-	go TimedTaskBuyer()  //定时任务
-	go TimedTaskAgency() //定时任务
+	//go TimedTaskBuyer()  //定时任务
+	//go TimedTaskAgency() //定时任务
 	c := make(chan int, 1)
 	<-c
 
@@ -173,9 +172,9 @@ func processUdpMsg(act byte, data []byte, ra *net.UDPAddr) {
 				if key == "winner" {
 					go TaskWinner(tmp)
 				} else if key == "buyer" {
-					go TaskBuyer(tmp)
+					//go TaskBuyer(tmp)
 				} else if key == "agency" {
-					go TaskAgency(tmp)
+					//go TaskAgency(tmp)
 				}
 			}
 		}

udp_winner/mgo.go

@@ -1,321 +0,0 @@
-package main
-
-import (
-	"context"
-	"log"
-	"time"
-
-	"go.mongodb.org/mongo-driver/bson"
-	"go.mongodb.org/mongo-driver/bson/primitive"
-	"go.mongodb.org/mongo-driver/mongo"
-	"go.mongodb.org/mongo-driver/mongo/options"
-)
-
-var Mgo *MongodbSim
-
-type MgoSess struct {
-	Db     string
-	Coll   string
-	Query  interface{}
-	Sorts  []string
-	fields interface{}
-	limit  int64
-	skip   int64
-	M      *MongodbSim
-}
-
-type MgoIter struct {
-	Cursor *mongo.Cursor
-}
-
-func (mt *MgoIter) Next(result interface{}) bool {
-	if mt.Cursor != nil {
-		if mt.Cursor.Next(nil) {
-			err := mt.Cursor.Decode(result)
-			if err != nil {
-				log.Println("mgo cur err", err.Error())
-				mt.Cursor.Close(nil)
-				return false
-			}
-			return true
-		} else {
-			mt.Cursor.Close(nil)
-			return false
-		}
-	} else {
-		return false
-	}
-
-}
-
-func (ms *MgoSess) DB(name string) *MgoSess {
-	ms.Db = name
-	return ms
-}
-
-func (ms *MgoSess) C(name string) *MgoSess {
-	ms.Coll = name
-	return ms
-}
-
-func (ms *MgoSess) Find(q interface{}) *MgoSess {
-	ms.Query = q
-	return ms
-}
-
-func (ms *MgoSess) Select(fields interface{}) *MgoSess {
-	ms.fields = fields
-	return ms
-}
-
-func (ms *MgoSess) Limit(limit int64) *MgoSess {
-	ms.limit = limit
-	return ms
-}
-func (ms *MgoSess) Skip(skip int64) *MgoSess {
-	ms.skip = skip
-	return ms
-}
-
-func (ms *MgoSess) Sort(sorts ...string) *MgoSess {
-	ms.Sorts = sorts
-	return ms
-}
-
-func (ms *MgoSess) Iter() *MgoIter {
-	it := &MgoIter{}
-	find := options.Find()
-	if ms.skip > 0 {
-		find.SetSkip(ms.skip)
-	}
-	if ms.limit > 0 {
-		find.SetLimit(ms.limit)
-	}
-	find.SetBatchSize(100)
-	if len(ms.Sorts) > 0 {
-		sort := bson.M{}
-		for _, k := range ms.Sorts {
-			switch k[:1] {
-			case "-":
-				sort[k[1:]] = -1
-			case "+":
-				sort[k[1:]] = 1
-			default:
-				sort[k] = 1
-			}
-		}
-		find.SetSort(sort)
-	}
-	if ms.fields != nil {
-		find.SetProjection(ms.fields)
-	}
-	cur, err := ms.M.C.Database(ms.Db).Collection(ms.Coll).Find(ms.M.Ctx, ms.Query, find)
-	if err != nil {
-		log.Println("mgo find err", err.Error())
-	} else {
-		it.Cursor = cur
-	}
-	return it
-}
-
-type MongodbSim struct {
-	MongodbAddr string
-	Size        int
-	//	MinSize     int
-	DbName   string
-	C        *mongo.Client
-	Ctx      context.Context
-	ShortCtx context.Context
-	pool     chan bool
-}
-
-func (m *MongodbSim) GetMgoConn() *MgoSess {
-	//m.Open()
-	ms := &MgoSess{}
-	ms.M = m
-	return ms
-}
-
-func (m *MongodbSim) DestoryMongoConn(ms *MgoSess) {
-	//m.Close()
-	ms.M = nil
-	ms = nil
-}
-
-func (m *MongodbSim) InitPool() {
-	opts := options.Client()
-	opts.SetConnectTimeout(3 * time.Second)
-	opts.ApplyURI("mongodb://" + m.MongodbAddr)
-	opts.SetMaxPoolSize(uint16(m.Size))
-	m.pool = make(chan bool, m.Size)
-	opts.SetMaxConnIdleTime(2 * time.Hour)
-	m.Ctx, _ = context.WithTimeout(context.Background(), 99999*time.Hour)
-	m.ShortCtx, _ = context.WithTimeout(context.Background(), 1*time.Minute)
-	client, err := mongo.Connect(m.ShortCtx, opts)
-	if err != nil {
-		log.Println("mgo init error:", err.Error())
-	} else {
-		m.C = client
-		log.Println("init success")
-	}
-}
-
-func (m *MongodbSim) Open() {
-	m.pool <- true
-}
-func (m *MongodbSim) Close() {
-	<-m.pool
-}
-
-//批量插入
-func (m *MongodbSim) UpSertBulk(c string, doc ...[]map[string]interface{}) (map[int64]interface{}, bool) {
-	m.Open()
-	defer m.Close()
-	coll := m.C.Database(m.DbName).Collection(c)
-	var writes []mongo.WriteModel
-	for _, d := range doc {
-		write := mongo.NewUpdateOneModel()
-		write.SetFilter(d[0])
-		write.SetUpdate(d[1])
-		write.SetUpsert(true)
-		writes = append(writes, write)
-	}
-	r, e := coll.BulkWrite(m.Ctx, writes)
-	if e != nil {
-		log.Println("mgo upsert error:", e.Error())
-		return nil, false
-	}
-	//	else {
-	//		if r.UpsertedCount != int64(len(doc)) {
-	//			log.Println("mgo upsert uncomplete:uc/dc", r.UpsertedCount, len(doc))
-	//		}
-	//		return true
-	//	}
-	return r.UpsertedIDs, true
-}
-
-//批量插入
-func (m *MongodbSim) SaveBulk(c string, doc ...map[string]interface{}) bool {
-	m.Open()
-	defer m.Close()
-	coll := m.C.Database(m.DbName).Collection(c)
-	var writes []mongo.WriteModel
-	for _, d := range doc {
-		write := mongo.NewInsertOneModel()
-		write.SetDocument(d)
-		writes = append(writes, write)
-	}
-	_, e := coll.BulkWrite(m.Ctx, writes)
-	if e != nil {
-		log.Println("mgo savebulk error:", e.Error())
-		return false
-	}
-	return true
-}
-
-//保存
-func (m *MongodbSim) Save(c string, doc map[string]interface{}) interface{} {
-	m.Open()
-	defer m.Close()
-	coll := m.C.Database(m.DbName).Collection(c)
-	r, err := coll.InsertOne(m.Ctx, doc)
-	if err != nil {
-		return nil
-	}
-	return r.InsertedID
-}
-
-//更新by Id
-func (m *MongodbSim) UpdateById(c, id string, doc map[string]interface{}) bool {
-	m.Open()
-	defer m.Close()
-	coll := m.C.Database(m.DbName).Collection(c)
-	_, err := coll.UpdateOne(m.Ctx, map[string]interface{}{"_id": StringTOBsonId(id)}, doc)
-	if err != nil {
-		log.Println(err)
-		return false
-	}
-	return true
-}
-
-//删除by id
-func (m *MongodbSim) DeleteById(c, id string) int64 {
-	m.Open()
-	defer m.Close()
-	coll := m.C.Database(m.DbName).Collection(c)
-	r, err := coll.DeleteOne(m.Ctx, map[string]interface{}{"_id": StringTOBsonId(id)})
-	if err != nil {
-		return 0
-	}
-	return r.DeletedCount
-}
-
-//通过条件删除
-func (m *MongodbSim) Delete(c string, query map[string]interface{}) int64 {
-	m.Open()
-	defer m.Close()
-	coll := m.C.Database(m.DbName).Collection(c)
-	r, err := coll.DeleteMany(m.Ctx, query)
-	if err != nil {
-		return 0
-	}
-	return r.DeletedCount
-}
-
-//findbyid
-func (m *MongodbSim) FindById(c, id string) map[string]interface{} {
-	m.Open()
-	defer m.Close()
-	coll := m.C.Database(m.DbName).Collection(c)
-	r := coll.FindOne(m.Ctx, map[string]interface{}{"_id": StringTOBsonId(id)})
-	v := map[string]interface{}{}
-	if e := r.Decode(&v);e != nil{
-		log.Println(e)
-		return nil
-	}
-	return v
-}
-
-//findone
-func (m *MongodbSim) FindOne(c string, query map[string]interface{}) map[string]interface{} {
-	m.Open()
-	defer m.Close()
-	coll := m.C.Database(m.DbName).Collection(c)
-	r := coll.FindOne(m.Ctx, query)
-	v := map[string]interface{}{}
-	r.Decode(&v)
-	return v
-}
-
-//find
-func (m *MongodbSim) Find(c string, query map[string]interface{}, sort, fields interface{}) ([]map[string]interface{}, error) {
-	m.Open()
-	defer m.Close()
-	coll := m.C.Database(m.DbName).Collection(c)
-	op := options.Find()
-	r, err := coll.Find(m.Ctx, query, op.SetSort(sort), op.SetProjection(fields))
-	if err != nil {
-		log.Fatal(err)
-		return nil, err
-	}
-	var results []map[string]interface{}
-	if err = r.All(m.Ctx, &results); err != nil {
-		log.Fatal(err)
-		return nil, err
-	}
-	return results, nil
-}
-
-//创建_id
-func NewObjectId() primitive.ObjectID {
-	return primitive.NewObjectID()
-}
-
-func StringTOBsonId(id string) primitive.ObjectID {
-	objectId, _ := primitive.ObjectIDFromHex(id)
-	return objectId
-}
-
-func BsonTOStringId(id interface{}) string {
-	return id.(primitive.ObjectID).Hex()
-}

udp_winner/timedTaskWinner.go

@@ -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"
@@ -26,10 +25,12 @@ 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  中标人 中标联系人 中标联系电话
@@ -43,31 +44,44 @@ func TaskWinner(mapinfo *map[string]interface{}) {
 			"$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 {
+		log.Println(cursor.Err())
 		return
 	}
 	//判断是否是存量,是存量走Redis遍历
-	if v, ok := (*mapinfo)["data_info"].(string); ok && v == "save" {
+ 	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 || winner == "" {
 				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)
 			}
 		}
@@ -78,166 +92,123 @@ 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)
+					//FClient.DbName = Config["mgodb_extract_kf"]
+					//if tmpid := FClient.Save("winner_new", tmps); tmpid == nil {
+					//	log.Println("存量 FClient.Save err", tmpid)
+					//}
+					fsavec := FClient.GetMgoConn().DB(Config["mgodb_extract_kf"]).C("winner_new")
+					for _, vmap := range rValuesMaps {
+						vmap["_id"] = bson.ObjectIdHex(vmap["_id"].(string))
+						if err = fsavec.Insert(vmap); err != nil{
+							log.Println("存量 FClient.Save err", err)
+						}
 					}
 					//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,8 +219,8 @@ 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"])
+			overid = tmp["_id"].(bson.ObjectId).Hex()
+			//log.Println(tmp["_id"])
 			if tmp["winner"] == nil || tmp["winner"] == "" {
 				continue
 			}
@@ -259,7 +230,7 @@ func TaskWinner(mapinfo *map[string]interface{}) {
 			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 {
+				if tmpid := FClient.Save("winner_new", tmp); tmpid == "" {
 					log.Println("FClient.Save err", tmpid)
 				}
 				//log.Println("get redis id err:定时任务处理", err, tmp)
@@ -273,53 +244,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 +289,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,73 +297,41 @@ 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)
 				}
@@ -445,12 +375,11 @@ func TimedTaskWinner() {
 				//遍历临时表数据，匹配不到原始库存入异常表
 				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 +396,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 +411,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)
+						fdmongo := FClient.GetMgoConn().DB(Config["mgodb_extract_kf"])
+						if err := fdmongo.C("winner_err").Insert( tmp); err != nil {
+							log.Println("存入异常表错误", err,tmp)
 						}
-						FClient.DbName = Config["mgodb_extract_kf"]
-						if deleteNum := FClient.DeleteById("winner_new", tmpId); deleteNum == 0 {
+						if deleteNum := fdmongo.C("winner_new").RemoveId( 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 +444,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 +484,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 +541,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 +569,42 @@ 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 +618,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)
 									}
 								}