jy-data-extract/udp_winner/timedTaskWinner.go

package main

import (
	"encoding/json"
	"fmt"
	"github.com/garyburd/redigo/redis"
	"gopkg.in/mgo.v2"
	"gopkg.in/mgo.v2/bson"
	"log"
	"qfw/util"
	"sort"
	"strings"
	"time"
	"unicode/utf8"
)

//之前main方法,只更新
func TaskWinner(mapinfo *map[string]interface{}) {
	defer util.Catch()
	gtid, lteid := util.ObjToString((*mapinfo)["gtid"]), util.ObjToString((*mapinfo)["lteid"])
	if gtid == "" || lteid == "" {
		log.Println(gtid, lteid, "参数错误")
		return
	}
	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(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, "package": 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" {
		//存量处理
		conn := HisRedisPool.Conn()
		defer conn.Close()
		//选择redis db
		conn.Select(redis_winner_db)
		//遍历bidding表保存到redis
		//key：企业名  value:json结构体{"winner": 1, "winnertel": 1, "winnerperson": 1,"topscopeclass": 1, "winneraddr": 1,"_id":1}
		tmp := make(map[string]interface{})
		for cursor.Next(&tmp) {
			winner, ok := tmp["winner"].(string)
			if !ok || utf8.RuneCountInString(winner) < 4 {
				continue
			}
			//判断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())
			return
		} else {
			iterator := scan.Iterator()
			for iterator.Next() {
				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", redis_winner_db)
				if reply, err := redis.String(rdb.Do("GET", redisCName)); err != nil {
					//redis不存在,存到临时表,定时任务处理
					FClient.DbName = Config["mgodb_extract_kf"]
					//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)
					}
					continue
				} else {
					//redis存在更新合并
					if err := rdb.Close(); err != nil {
						log.Println(err)
					}
					//拿到合并后的qyk
					FClient.DbName = Config["mgodb_extract_kf"]
					oldTmp, b := FClient.FindById(Config["mgo_qyk_c"], reply, nil)
					if !b || (*oldTmp) == nil || reply == "" || (*oldTmp)["_id"] == nil {
						log.Println(redisCName, "存量 redis id 不存在", reply)
						continue
					}
					tmpTopscopeclass := []string{}
					tmpTopscopeclassMap := make(map[string]bool)

					if (*oldTmp)["industry"] != nil {
						if v, ok := (*oldTmp)["industry"].([]interface{}); ok {
							for _, vv := range v {
								if vvv, ok := vv.(string); ok {
									tmpTopscopeclassMap[vvv] = 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)
					}
					sort.Strings(tmpTopscopeclass)
					(*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 value联系人
					for _, rvmap := range rValuesMaps {
						var tmpperson, winnertel string
						if rvmapperson, ok := rvmap["winnerperson"].(string); ok && rvmapperson != "" {
							tmpperson = rvmapperson
						} else {
							continue
						}
						if rvmapwintel, ok := rvmap["winnertel"].(string); ok {
							winnertel = rvmapwintel
						} else {
							winnertel = ""
						}
						if Reg_xing.MatchString(winnertel) || !Reg_tel.MatchString(winnertel) {
							winnertel = ""
						}
						tmpContact := make(map[string]interface{})
						tmpContact["infoid"] = rvmap["_id"]
						tmpContact["contact_person"] = tmpperson
						tmpContact["contact_type"] = "项目联系人"
						tmpContact["phone"] = winnertel
						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])
								}
							}
						}
						tmpContact["topscopeclass"] = strings.Join(tmpclass, ";")
						tmpContact["updatetime"] = time.Now().Unix()
						contactMaps = append(contactMaps, tmpContact)
					}
					(*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)
					}
				}
			}
		}
		log.Println("存量历史合并执行完成 ok", gtid, lteid)

	} else {
		//增量
		overid := addfunc(gtid, cursor)
		SourceClient.DestoryMongoConn(SourceClientcc)
		log.Println("增量合并执行完成 ok", gtid, lteid, overid)
	}

}

//增量
func addfunc(gtid string, cursor *mgo.Iter) string {
	//增量处理
	overid := gtid
	tmp := map[string]interface{}{}
	for cursor.Next(&tmp) {
		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", redis_winner_db)
		if reply, err := redis.String(rdb.Do("GET", winner)); err != nil {
			//redis不存在存到临时表,定时任务处理
			FClient.DbName = Config["mgodb_extract_kf"]
			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 {
				log.Println(err)
			}
			continue
		} else {
			if err := rdb.Close(); err != nil {
				log.Println(err)
			}
			//拿到合并后的qyk
			FClient.DbName = Config["mgodb_extract_kf"]
			oldTmp, b := FClient.FindById(Config["mgo_qyk_c"], reply, bson.M{})
			if !b || (*oldTmp) == nil || reply == "" || (*oldTmp)["_id"] == nil {
				log.Println("redis id 不存在")
				continue
			}
			//比较合并 行业类型
			tmpTopscopeclass := []string{}
			tmpConTopscopeclass := []string{}
			tmpTopscopeclassMap := make(map[string]bool)

			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"].([]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

			esId := (*oldTmp)["_id"].(bson.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
			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 {
				winnertel = winnertel
			}
			contactMaps := make([]interface{}, 0)
			if (*oldTmp)["contact"] != nil {
				//直接添加联系人，不再判断
				if v, ok := (*oldTmp)["contact"].([]interface{}); ok {
					contactMaps = append(contactMaps, v...)
				}
			}
			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)
			//分包处理
			PackageDealWith(&contactMaps, tmp)
			(*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)
			}
		}
	}
	return overid
}

//定时任务  新增
//1.存异常表
//2.合并原始库新增
func TimedTaskWinner() {
	//time.Sleep(time.Hour*70)
	t2 := time.NewTimer(time.Second * 5)
	for range t2.C {
		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(86400)
				cursor := fconn.DB(Config["mgodb_extract_kf"]).C("winner_new").Find(bson.M{
					"_id": bson.M{
						"$lte": tmpLast["_id"],
					},
				}).Sort("_id").Iter()
				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"].(bson.ObjectId).Hex()
					errwinner, ok := tmp["winner"].(string)
					if !ok || errwinner == "" {
						continue
					}
					//再重新查找redis,存在发udp处理，不存在走新增合并
					rdb := RedisPool.Get()
					rdb.Do("SELECT", redis_winner_db)
					if _, err := redis.String(rdb.Do("GET", errwinner)); err == nil {
						//redis存在发送udp进行处理
						TaskWinner(&map[string]interface{}{
							"gtid":  tmpId,
							"lteid": tmpId,
						})
						//存在的话删除tmp mongo表
						FClient.DbName = Config["mgodb_extract_kf"]
						if DeletedCount := FClient.Del("winner_new", bson.M{"_id": bson.ObjectIdHex(tmpId)}); !DeletedCount {
							log.Println("删除临时表err:", DeletedCount)
						}
						if err := rdb.Close(); err != nil {
							log.Println(err)
						}
						continue
					} else {
						if err = rdb.Close(); err != nil {
							log.Println(err)
						}
					}
					//查询redis不存在新增
					FClient.DbName = Config["mgodb_enterprise"]

					resulttmp, b := FClient.FindOne(Config["mgodb_enterprise_c"], bson.M{"company_name": errwinner})
					if !b || (*resulttmp)["_id"] == nil {
						//log.Println(r)
						//人工审核正则
						var isok bool
						//先遍历ok
						for _, v := range WinnerRegOk {
							isok = v.MatchString(errwinner)
							if isok {
								//匹配ok完,匹配err
								for _, vRegErr := range WinnerRegErr {
									isok = vRegErr.MatchString(errwinner)
									//匹配到ok 也匹配到err 按err算
									if isok {
										tmp["winner_err"] = 1
										break
									}
								}
								//匹配ok，没匹配err 按ok算
								if tmp["winner_err"] == nil {
									tmp["winner_ok"] = 1
									break
								}
							}
						}
						//都没匹配
						if tmp["winner_ok"] == nil  && tmp["winner_err"] == nil{
							tmp["winner_err"] = 1
						}
						//匹配不到原始库,存入异常表删除临时表
						FClient.DbName = Config["mgodb_extract_kf"]
						if err := FClient.SaveForOld("winner_err", tmp); err != nil {
							log.Println("存入异常表错误", err, tmp)
						}
						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) != "" &&
								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"]
										}
										(*resulttmp)["city"] = v["city"]
										(*resulttmp)["district"] = v["district"]

									}
								}
							}
						}
						contacts := make([]map[string]interface{}, 0)
						contact := make(map[string]interface{}, 0)
						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 err != nil {
								log.Println("annual_reports err:", err)
							}
							phonetmp := make([]map[string]interface{}, 0)
							err = json.Unmarshal(bytes, &phonetmp)
							if err != nil {
								log.Println("Unmarshal err:", err)
							}
							for _, vv := range phonetmp {
								if vv["company_phone"] != nil {
									if vv["company_phone"] == "" {
										continue
									} else {
										contact["phone"] = vv["company_phone"] //联系电话
										break
									}
								} else {
									contact["phone"] = "" //联系电话
								}

							}
						}
						//log.Println(k, contact["phone"], resulttmp["_id"])
						//time.Sleep(10 * time.Second)
						if contact["phone"] == nil {
							contact["phone"] = "" //联系电话
						}
						contact["topscopeclass"] = "企业公示"         //项目类型
						contact["updatetime"] = time.Now().Unix() //更新时间
						contact["infoid"] = ""                    //招标信息id
						contacts = append(contacts, contact)
						//添加临时表匹配到的联系人
						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].(time.Time).UTC().Unix()
									continue
								}
							} else if sk == "capital" {
								//log.Println(sk, resulttmp[sk])
								savetmp[sk] = ObjToMoney([]interface{}{(*resulttmp)[sk], ""})[0]
								continue
							} else if sk == "partners" {
								//log.Println(sk, resulttmp[sk], )
								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"] = "企业公示"
											}
											delete(ppms[i].(map[string]interface{}), "identify_type")
										}
										savetmp[sk] = ppms

									}
								} else {
									savetmp[sk] = []interface{}{}
								}
								continue
							} else if sk == "_id" {
								savetmp["tmp"+sk] = (*resulttmp)[sk]
								continue
							} else if sk == "area_code" {
								//行政区划代码
								savetmp[sk] = fmt.Sprint((*resulttmp)[sk])
								continue
							} else if sk == "report_websites" {
								//网址
								if (*resulttmp)["report_websites"] == nil {
									savetmp["website"] = ""
								} else {
									report_websitesArr := []string{}
									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 {
													report_websitesArr = append(report_websitesArr, rv)
												}
											}
										}
									}
									sort.Strings(report_websitesArr)
									savetmp["website"] = strings.Join(report_websitesArr, ";")
								}
								continue
							} 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" {
								savetmp[sk] = ""
							} else {
								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 != "" {
							//保存redis
							rc := RedisPool.Get()
							rc.Do("SELECT", redis_winner_db)
							//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)
								}
							} else {
								//保存es
								delete(savetmp, "_id")
								if err := rc.Close(); err != nil {
									log.Println(err)
								}

								//esConn := elastic.GetEsConn()
								//defer elastic.DestoryEsConn(esConn)
								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.Del("winner_new", bson.M{"_id": bson.ObjectIdHex(tmpId)}); !deleteNum {
										log.Println("删除临时表失败", deleteNum)
									}
								}
							}
						} else {
							log.Println("save mongo err:", saveid, tmp["_id"])
						}
					}
				}
				FClient.DestoryMongoConn(fconn)
				log.Println("winner_new,遍历完成")
			}
		}
		FClient.DestoryMongoConn(Fcconn)
		t2.Reset(time.Minute)
	}
}

//分包处理
func PackageDealWith(contactMaps *[]interface{}, tmp map[string]interface{}) {

}