标讯数仓-备份

data_mgo_to_tidb/bidding.go

@@ -27,8 +27,8 @@ func TaskBidding() {
 	wg := &sync.WaitGroup{}
 	q := map[string]interface{}{
 		"_id": map[string]interface{}{
-			"$gt":  mongodb.StringTOBsonId("100000000000000000000000"),
-			"$lte": mongodb.StringTOBsonId("900000000000000000000000"),
+			"$gt":  mongodb.StringTOBsonId("6558df800000000000000000"),
+			"$lte": mongodb.StringTOBsonId("655a31000000000000000000"),
 		},
 	}
 	query := sess.DB(config.Conf.DB.MongoB.Dbname).C("bidding").Find(q).Sort("_id").Iter()
@@ -45,12 +45,12 @@ func TaskBidding() {
 				wg.Done()
 			}()
 			if util.IntAll(tmp["extracttype"]) != -1 {
-				//taskBase(tmp)    //基础标讯数据
-				//taskExpand(tmp)  //扩展数据
-				//taskDetail(tmp)  //正文信息
-				//taskAtts(tmp)    //附件信息
-				//taskIntent(tmp) //采购意向
-				//taskPackage(tmp) //分包
+				taskBase(tmp)    //基础标讯数据
+				taskExpand(tmp)  //扩展数据
+				taskDetail(tmp)  //正文信息
+				taskAtts(tmp)    //附件信息
+				taskIntent(tmp)  //采购意向
+				taskPackage(tmp) //分包
 			}
 		}(tmp)
 		tmp = make(map[string]interface{})
@@ -62,7 +62,6 @@ func TaskBidding() {
 // @Description 基本信息
 func taskBase(tmp map[string]interface{}) {
 	saveM := make(map[string]interface{})
-	var errf []string // 异常字段
 	for _, nf := range BaseField {
 		f := nf[2:]
 		if f == "info_id" {
@@ -121,21 +120,14 @@ func taskBase(tmp map[string]interface{}) {
 		} else {
 			if tmp[f] != nil {
 				if BaseVMap[f] != nil {
-					var b bool
-					saveM[nf], b = verifyF(f, tmp[f], BaseVMap[f])
-					if b { // 保存异常字段数据
-						errf = append(errf, f)
-					}
+					saveM[nf], _ = verifyF(f, tmp[f], BaseVMap[f])
 				} else {
-					saveM[nf] = tmp[f]
+					saveM[nf] = util.ObjToString(tmp[f])
 				}
 			}
 		}
 	}
-	saveBasePool <- saveM
-	if len(errf) > 0 {
-		saveErrPool <- map[string]interface{}{"infoid": mongodb.BsonIdToSId(tmp["_id"]), "f": strings.Join(errf, ",")}
-	}
+	InsertGlobalMysqlData("dwd_f_bid_baseinfo", saveM, mongodb.BsonIdToSId(tmp["_id"]))
 }
 
 func getNameId(name string) string {
@@ -150,7 +142,6 @@ func getNameId(name string) string {
 // @Description 扩展信息
 func taskExpand(tmp map[string]interface{}) {
 	saveM := make(map[string]interface{})
-	var errf []string // 异常字段
 	for _, nf := range ExpandField {
 		f := nf[2:]
 		if f == "info_id" {
@@ -206,35 +197,29 @@ func taskExpand(tmp map[string]interface{}) {
 		} else {
 			if tmp[f] != nil {
 				if ExpandVMap[f] != nil {
-					var b bool
-					saveM[nf], b = verifyF(f, tmp[f], ExpandVMap[f])
-					if b { // 保存异常字段数据
-						errf = append(errf, f)
-					}
+					saveM[nf], _ = verifyF(f, tmp[f], ExpandVMap[f])
 				} else {
-					saveM[nf] = tmp[f]
+					saveM[nf] = util.ObjToString(tmp[f])
 				}
 			}
 		}
 	}
-	saveExpandPool <- saveM
-	if len(errf) > 0 {
-		saveErrPool <- map[string]interface{}{"infoid": mongodb.BsonIdToSId(tmp["_id"]), "f": strings.Join(errf, ",")}
-	}
+	InsertGlobalMysqlData("dwd_f_bid_expand_baseinfo", saveM, mongodb.BsonIdToSId(tmp["_id"]))
 }
 
 // @Description 正文信息
 func taskDetail(tmp map[string]interface{}) {
-	id := mongodb.BsonIdToSId(tmp["_id"])
+	tmpid := mongodb.BsonIdToSId(tmp["_id"])
 	s_detail := util.ObjToString(tmp["detail"])
 	s_contenthtml := util.ObjToString(tmp["contenthtml"])
-	saveDetailPool <- map[string]interface{}{
-		"s_info_id":     id,
+	info := map[string]interface{}{
+		"s_info_id":     tmpid,
 		"s_detail":      s_detail,
 		"s_contenthtml": s_contenthtml,
 		"d_createtime":  time.Now().Format(util.Date_Full_Layout),
 		"d_updatetime":  time.Now().Format(util.Date_Full_Layout),
 	}
+	InsertGlobalMysqlData("dwd_f_bid_detail", info, tmpid)
 }
 
 // @Description 附件
@@ -290,7 +275,7 @@ func taskAttsAttach(att_info map[string]interface{}, tmpid string, f_id int64) {
 				//}
 				//info["s_file_text"] = bs
 			}
-			saveAttrPool <- info
+			InsertGlobalMysqlData("dwd_f_bid_file_text", info, tmpid)
 		}
 	}
 }
@@ -326,8 +311,7 @@ func taskIntent(tmp map[string]interface{}) {
 			}
 		}
 		info["d_createtime"] = time.Now().Format(util.Date_Full_Layout)
-		//InsertGlobalMysqlData("dwd_f_bid_intention_baseinfo", info, tmpid)
-		saveIntentPool <- info
+		InsertGlobalMysqlData("dwd_f_bid_intention_baseinfo", info, tmpid)
 	}
 }
 
@@ -345,7 +329,7 @@ func taskPackage(tmp map[string]interface{}) {
 			//标的物信息
 			new_purlist := CPBiddingPackageGoodsBaseInfo(tmp, tmpid, pid)
 			for _, v := range new_purlist {
-				saveGoodsPool <- v
+				InsertGlobalMysqlData("dwd_f_bid_package_goods_baseinfo", v, tmpid)
 			}
 		}
 	} else { //多包...具体源信息
@@ -357,7 +341,7 @@ func taskPackage(tmp map[string]interface{}) {
 					CPBidderPackageBaseInfo(v, tmp, tmpid, pid, true)
 					new_purlist := CPBiddingPackageGoodsBaseInfo(tmp, tmpid, pid)
 					for _, v1 := range new_purlist {
-						saveGoodsPool <- v1
+						InsertGlobalMysqlData("dwd_f_bid_package_goods_baseinfo", v1, tmpid)
 					}
 				} else {
 					CPBidderPackageBaseInfo(v, tmp, tmpid, pid, false)

data_mgo_to_tidb/init.go

@@ -23,50 +23,17 @@ var ( //通用
 	SubScopeCode   = make(map[string]interface{}, 70)
 )
 var ( //标讯
-	saveSize       = 200
-	saveBasePool   = make(chan map[string]interface{}, 5000)
-	saveBaseSp     = make(chan bool, 1)
-	saveExpandPool = make(chan map[string]interface{}, 5000)
-	saveExpandSp   = make(chan bool, 1)
-	saveDetailPool = make(chan map[string]interface{}, 5000)
-	saveDetailSp   = make(chan bool, 1)
-	saveAttrPool   = make(chan map[string]interface{}, 5000)
-	saveAttrSp     = make(chan bool, 1)
-	saveIntentPool = make(chan map[string]interface{}, 5000)
-	saveIntentSp   = make(chan bool, 1)
-	saveBidderPool = make(chan map[string]interface{}, 5000)
-	saveBidderSp   = make(chan bool, 1)
-	saveGoodsPool  = make(chan map[string]interface{}, 5000)
-	saveGoodsSp    = make(chan bool, 1)
-
 	BaseField   []string
 	BaseVMap    map[string]interface{}
 	ExpandField []string
 	ExpandVMap  map[string]interface{}
-	DetailField = []string{"s_info_id", "s_detail", "s_contenthtml", "d_updatetime", "d_createtime"}
-	AttrField   = []string{"s_info_id", "s_file_id", "s_file_text", "i_is_bidding_doc", "d_updatetime", "d_createtime"}
-	IntentField = []string{"s_info_id", "s_intention_name", "s_intention_demand", "s_item", "s_buyer_id", "f_totalprice", "s_expurasingtime", "s_reserved_amount", "d_updatetime", "d_createtime"}
-	BidderField = []string{"s_info_id", "i_package_id", "s_bidder_id", "s_bidder_name", "s_other_bidder_id", "s_other_bidder_name", "i_is_bidders", "i_is_winner", "d_updatetime", "d_createtime"}
-	GoodsField  = []string{"s_info_id", "i_package_id", "s_purchasing_tag", "s_itemname", "s_brand", "s_model", "s_specs", "i_num", "f_unit", "f_totalprice", "d_updatetime", "d_createtime"}
 )
 var ( //项目
-	saveProPool      = make(chan map[string]interface{}, 5000)
-	saveProSp        = make(chan bool, 1)
-	saveProbPool     = make(chan map[string]interface{}, 5000)
-	saveProbSp       = make(chan bool, 1)
-	saveProTagPool   = make(chan map[string]interface{}, 5000)
-	saveProTagSp     = make(chan bool, 1)
-	saveRelationPool = make(chan map[string]interface{}, 5000)
-	saveRelationSp   = make(chan bool, 1)
-	saveErrPool      = make(chan map[string]interface{}, 5000)
-	saveErrSp        = make(chan bool, 1)
-	ProField         []string
-	ProVMap          map[string]interface{}
-	ProBusField      []string
-	ProBusVMap       map[string]interface{}
-	ProTagsField     = []string{"projectid", "labelcode", "labelvalues", "labelweight", "createtime"}
-	RelationField    = []string{"projectid", "infoid", "name_id", "contact_id", "identity_type", "createtime"}
-	FileTypeArr      = []string{"pdf", "doc", "docx", "xlsx", "xls", "jpg", "zip", "rar", "txt", "gif", "png", "bmp", "swf", "html"}
+	ProField    []string
+	ProVMap     map[string]interface{}
+	ProBusField []string
+	ProBusVMap  map[string]interface{}
+	FileTypeArr = []string{"pdf", "doc", "docx", "xlsx", "xls", "jpg", "zip", "rar", "txt", "gif", "png", "bmp", "swf", "html"}
 )
 
 func InitInfo() {

data_mgo_to_tidb/main.go

@@ -1,12 +1,11 @@
 package main
 
 func init() {
-	InitInfo()       //初始化...
-	RunSaveService() //保存服务
+	InitInfo() //初始化...
 }
 
 func main() {
-	//标讯信息
+	//标讯信息-正式版先测试一天的数据
 	TaskBidding()
 
 	c := make(chan bool, 1)

data_mgo_to_tidb/pkg.go

@@ -44,7 +44,7 @@ func CPBidderBiddingBaseInfo(tmp map[string]interface{}, tmpid string, pid int64
 		info["i_is_bidders"] = 1
 	}
 	//保存服务...
-	saveBidderPool <- info
+	InsertGlobalMysqlData("dwd_f_bid_package_bidder_baseinfo", info, tmpid)
 	//候选人相关情况
 	o_win_ids, o_win_names := getWinerorder(tmp, wins[0])
 	for k, v := range o_win_ids {
@@ -56,7 +56,7 @@ func CPBidderBiddingBaseInfo(tmp map[string]interface{}, tmpid string, pid int64
 		o_info["i_is_winner"] = 0
 		o_info["d_updatetime"] = time.Now().Format(util.Date_Full_Layout)
 		//保存服务...
-		saveBidderPool <- info
+		InsertGlobalMysqlData("dwd_f_bid_package_bidder_baseinfo", info, tmpid)
 	}
 }
 
@@ -127,7 +127,7 @@ func CPBidderPackageBaseInfo(pinfos map[string]interface{}, tmp map[string]inter
 			info["i_is_bidders"] = 1
 		}
 		//保存服务...
-		saveBidderPool <- info
+		InsertGlobalMysqlData("dwd_f_bid_package_bidder_baseinfo", info, tmpid)
 		//候选人相关情况
 		if isF {
 			o_win_ids, o_win_names := getWinerorder(tmp, wins[0])
@@ -139,7 +139,7 @@ func CPBidderPackageBaseInfo(pinfos map[string]interface{}, tmp map[string]inter
 				o_info["i_is_winner"] = 0
 				o_info["d_updatetime"] = time.Now().Format(util.Date_Full_Layout)
 				//保存服务...
-				saveBidderPool <- info
+				InsertGlobalMysqlData("dwd_f_bid_package_bidder_baseinfo", info, tmpid)
 			}
 		}
 	}

data_mgo_to_tidb/project.go

@@ -1,486 +1,975 @@
 package main
 
-import (
-	"data_tidb/config"
-	"fmt"
-	"go.mongodb.org/mongo-driver/bson"
-	"go.uber.org/zap"
-	util "jygit.jydev.jianyu360.cn/data_processing/common_utils"
-	"jygit.jydev.jianyu360.cn/data_processing/common_utils/log"
-	"jygit.jydev.jianyu360.cn/data_processing/common_utils/mongodb"
-	"jygit.jydev.jianyu360.cn/data_processing/common_utils/redis"
-	"strings"
-	"sync"
-	"time"
-)
-
-func taskP() {
-	sess := MongoP.GetMgoConn()
-	defer MongoP.DestoryMongoConn(sess)
-
-	ch := make(chan bool, 20)
-	wg := &sync.WaitGroup{}
-
-	//q := map[string]interface{}{"_id": mongodb.StringTOBsonId("64e5a63855d5406905c574e6")}
-	query := sess.DB(config.Conf.DB.MongoP.Dbname).C("projectset_20230407").Find(nil).Sort("-_id").Skip(100000).Iter()
-	count := 0
-	for tmp := make(map[string]interface{}); query.Next(tmp); count++ {
-		if count%20000 == 0 {
-			log.Info(fmt.Sprintf("current --- %d", count))
-		}
-		ch <- true
-		wg.Add(1)
-		go func(tmp map[string]interface{}) {
-			defer func() {
-				<-ch
-				wg.Done()
-			}()
-			//taskPro(tmp)
-			//taskBusiness(tmp)
-			//taskProTag(tmp)
-
-			taskRelation(tmp)
-
-		}(tmp)
-		tmp = make(map[string]interface{})
-	}
-	wg.Wait()
-	log.Info(fmt.Sprintf("over --- %d", count))
-}
-
-func taskPAdd(pici int64) {
-	sess := MongoP.GetMgoConn()
-	defer MongoP.DestoryMongoConn(sess)
-
-	ch := make(chan bool, 20)
-	wg := &sync.WaitGroup{}
-
-	q := bson.M{"pici": bson.M{"$gt": pici}}
-	query := sess.DB(config.Conf.DB.MongoP.Dbname).C("projectset_20230407").Find(q).Iter()
-	count := 0
-	for tmp := make(map[string]interface{}); query.Next(tmp); count++ {
-		if count%20000 == 0 {
-			log.Info(fmt.Sprintf("current --- %d", count))
-		}
-		ch <- true
-		wg.Add(1)
-		go func(tmp map[string]interface{}) {
-			defer func() {
-				<-ch
-				wg.Done()
-			}()
-			//taskPro(tmp)
-			//taskBusiness(tmp)
-			//taskProTag(tmp)
-
-			taskRelation2(tmp)
-
-		}(tmp)
-		tmp = make(map[string]interface{})
-	}
-	wg.Wait()
-	log.Info(fmt.Sprintf("over --- %d", count))
-}
-
-var BidStatus = map[string]int{
-	"预告": 0,
-	"拟建": 1,
-	"招标": 2,
-	"中标": 3,
-	"成交": 4,
-	"废标": 5,
-	"流标": 6,
-	"合同": 7,
-	"其它": 8,
-}
-var BidType = map[string]int{
-	"招标": 0,
-	"邀标": 1,
-	"单一": 2,
-	"竞价": 3,
-	"竞谈": 4,
-	"询价": 5,
-}
-
-// @Description 基础信息
-// @Author J 2022/9/22 18:32
-func taskPro(tmp map[string]interface{}) {
-	saveM := make(map[string]interface{})
-	for _, f := range ProField {
-		if f == "projectid" {
-			saveM[f] = mongodb.BsonIdToSId(tmp["_id"])
-		} else if f == "area_code" {
-			if tmp["area"] != nil {
-				saveM[f] = AreaCode[util.ObjToString(tmp["area"])]
-			}
-		} else if f == "city_code" {
-			if tmp["area"] != nil && tmp["city"] != nil {
-				c := util.ObjToString(tmp["area"]) + "," + util.ObjToString(tmp["city"])
-				saveM[f] = AreaCode[c]
-			}
-		} else if f == "district_code" {
-			if tmp["area"] != nil && tmp["city"] != nil && tmp["district"] != nil {
-				c := util.ObjToString(tmp["area"]) + "," + util.ObjToString(tmp["city"]) + "," + util.ObjToString(tmp["district"])
-				saveM[f] = AreaCode[c]
-			}
-		} else if f == "updatetime" {
-			saveM[f] = time.Now().Format(util.Date_Full_Layout)
-		} else if f == "buyerclass_code" {
-			if obj := util.ObjToString(tmp["buyerclass"]); obj != "" {
-				saveM[f] = BuyerCode[obj]
-			}
-		} else if f == "firsttime" || f == "zbtime" || f == "jgtime" || f == "lasttime" || f == "bidopentime" || f == "createtime" {
-			if tmp[f] != nil && util.Int64All(tmp[f]) > 0 {
-				t := util.Int64All(tmp[f])
-				saveM[f] = util.FormatDateByInt64(&t, util.Date_Full_Layout)
-			}
-		} else if f == "bidstatus" {
-			if b := util.ObjToString(tmp[f]); b != "" {
-				tmp[f] = BidStatus[b]
-			}
-		} else if f == "bidtype" {
-			if b := util.ObjToString(tmp[f]); b != "" {
-				tmp[f] = BidType[b]
-			}
-		} else if f == "multipackage" {
-			if tmp[f] == nil {
-				saveM[f] = 0
-			} else {
-				saveM[f] = tmp[f]
-			}
-		} else if f == "buyer_id" {
-			if b := util.ObjToString(tmp["buyer"]); b != "" {
-				if code := redis.GetStr("qyxy_id", b); code != "" {
-					saveM[f] = code
-				}
-			}
-		} else if f == "agency_id" {
-			if b := util.ObjToString(tmp["agency"]); b != "" {
-				if code := redis.GetStr("qyxy_id", b); code != "" {
-					saveM[f] = code
-				}
-			}
-		} else {
-			if tmp[f] != nil {
-				if ProVMap[f] != nil {
-					saveM[f], _ = verifyF(f, tmp[f], ProVMap[f])
-				} else {
-					saveM[f] = tmp[f]
-				}
-			}
-		}
-	}
-	saveProPool <- saveM
-}
-
-// @Description 项目业务表
-// @Author J 2022/9/30 13:40
-func taskBusiness(tmp map[string]interface{}) {
-	warr := strings.Split(util.ObjToString(tmp["s_winner"]), ",")
-	if BinarySearch(warr, util.ObjToString(tmp["winner"])) == -1 {
-		warr = append(warr, util.ObjToString(tmp["winner"]))
-	}
-	for _, s := range warr {
-		saveM := make(map[string]interface{})
-		for _, f := range ProBusField {
-			if f == "projectid" {
-				saveM[f] = mongodb.BsonIdToSId(tmp["_id"])
-			} else if f == "area_code" {
-				if tmp["area"] != nil {
-					saveM[f] = AreaCode[util.ObjToString(tmp["area"])]
-				}
-			} else if f == "city_code" {
-				if tmp["area"] != nil && tmp["city"] != nil {
-					c := util.ObjToString(tmp["area"]) + "," + util.ObjToString(tmp["city"])
-					saveM[f] = AreaCode[c]
-				}
-			} else if f == "district_code" {
-				if tmp["area"] != nil && tmp["city"] != nil && tmp["district"] != nil {
-					c := util.ObjToString(tmp["area"]) + "," + util.ObjToString(tmp["city"]) + "," + util.ObjToString(tmp["district"])
-					saveM[f] = AreaCode[c]
-				}
-			} else if f == "updatetime" {
-				saveM[f] = time.Now().Format(util.Date_Full_Layout)
-			} else if f == "buyerclass_code" {
-				if obj := util.ObjToString(tmp["buyerclass"]); obj != "" {
-					saveM[f] = BuyerCode[obj]
-				}
-			} else if f == "firsttime" || f == "zbtime" || f == "jgtime" || f == "lasttime" || f == "bidopentime" || f == "createtime" {
-				if tmp[f] != nil && util.Int64All(tmp[f]) > 0 {
-					t := util.Int64All(tmp[f])
-					saveM[f] = util.FormatDateByInt64(&t, util.Date_Full_Layout)
-				}
-			} else if f == "bidstatus" {
-				if b := util.ObjToString(tmp[f]); b != "" {
-					tmp[f] = BidStatus[b]
-				}
-			} else if f == "bidtype" {
-				if b := util.ObjToString(tmp[f]); b != "" {
-					tmp[f] = BidType[b]
-				}
-			} else if f == "buyer_id" {
-				if b := util.ObjToString(tmp["buyer"]); b != "" {
-					saveM["buyer"] = b
-					if code := redis.GetStr("qyxy_id", b); code != "" {
-						saveM[f] = code
-					}
-				}
-			} else if f == "agency_id" {
-				if b := util.ObjToString(tmp["agency"]); b != "" {
-					saveM["agency"] = b
-					if code := redis.GetStr("qyxy_id", b); code != "" {
-						saveM[f] = code
-					}
-				}
-			} else if f == "winner_id" {
-				if s != "" {
-					saveM["winner"] = s
-					if code := redis.GetStr("qyxy_id", s); code != "" {
-						saveM[f] = code
-					}
-				}
-			} else {
-				if tmp[f] != nil {
-					if ProBusVMap[f] != nil {
-						saveM[f], _ = verifyF(f, tmp[f], ProBusVMap[f])
-					} else {
-						saveM[f] = tmp[f]
-					}
-				}
-			}
-		}
-		saveProbPool <- saveM
-	}
-
-}
-
-// @Description 项目信息标签
-// @Author J 2022/9/30 13:54
-func taskProTag(tmp map[string]interface{}) {
-	id := mongodb.BsonIdToSId(tmp["_id"])
-	if topArr, ok := tmp["topscopeclass"].([]interface{}); ok {
-		for _, i2 := range topArr {
-			tclass := regLetter.ReplaceAllString(util.ObjToString(i2), "") // 去除字母
-			code := TopScopeCode[tclass]
-			saveProTagPool <- map[string]interface{}{"projectid": id, "labelcode": "1", "labelvalues": code, "labelweight": 1, "createtime": time.Now().Format(util.Date_Full_Layout)}
-		}
-	}
-	if subArr, ok := tmp["subscopeclass"].([]interface{}); ok {
-		for _, i2 := range subArr {
-			sc := strings.Split(util.ObjToString(i2), "_")
-			code := SubScopeCode[sc[1]]
-			saveProTagPool <- map[string]interface{}{"projectid": id, "labelcode": "2", "labelvalues": code, "labelweight": 1, "createtime": time.Now().Format(util.Date_Full_Layout)}
-		}
-	}
-}
-
-// @Description 关系表
-// @Author J 2022/9/30 13:56
-func taskRelation(tmp map[string]interface{}) {
-	pid := mongodb.BsonIdToSId(tmp["_id"])
-	if tmp["ids"] == nil {
-		log.Info("taskRelation ids err", zap.Any("id", pid))
-		return
-	}
-	ids := util.ObjArrToStringArr(tmp["ids"].([]interface{}))
-	lid := ids[len(ids)-1]
-
-	//if b := util.ObjToString(tmp["buyer"]); b != "" {
-	//	saveM := make(map[string]interface{})
-	//
-	//	saveM["projectid"] = pid
-	//	saveM["infoid"] = lid
-	//	saveM["identity_type"] = 1
-	//	saveM["createtime"] = time.Now().Format(util.Date_Full_Layout)
-	//	if code := redis.GetStr("qyxy_id", b); code != "" {
-	//		saveM["name_id"] = code
-	//		if util.ObjToString(tmp["buyertel"]) != "" {
-	//			q := make(map[string]interface{})
-	//			q["name_id"] = code
-	//			q["identity_type"] = 1
-	//			q["contact_tel"] = util.ObjToString(tmp["buyertel"])
-	//			if util.ObjToString(tmp["buyerperson"]) != "" {
-	//				q["contact_name"] = util.ObjToString(tmp["buyerperson"])
-	//			}
-	//			cinfo := MysqlTool.FindOne("dws_f_ent_contact", q, "", "")
-	//			if cinfo != nil && len(*cinfo) > 0 {
-	//				saveM["contact_id"] = (*cinfo)["id"]
-	//				saveRelationPool <- saveM
-	//			}
-	//		}
-	//	}
-	//}
-
-	//if a := util.ObjToString(tmp["agency"]); a != "" {
-	//	saveM := make(map[string]interface{})
-	//	saveM["projectid"] = pid
-	//	saveM["infoid"] = lid
-	//	saveM["identity_type"] = 4
-	//	saveM["createtime"] = time.Now().Format(util.Date_Full_Layout)
-	//	if code := redis.GetStr("qyxy_id", a); code != "" {
-	//		saveM["name_id"] = code
-	//		if util.ObjToString(tmp["agencytel"]) != "" {
-	//			q := make(map[string]interface{})
-	//			q["name_id"] = code
-	//			q["identity_type"] = 4 // 100
-	//			q["contact_tel"] = util.ObjToString(tmp["agencytel"])
-	//			if util.ObjToString(tmp["agencyperson"]) != "" {
-	//				q["contact_name"] = util.ObjToString(tmp["agencyperson"])
-	//			}
-	//			cinfo := MysqlTool.FindOne("dws_f_ent_contact", q, "", "")
-	//			if cinfo != nil && len(*cinfo) > 0 {
-	//				saveM["contact_id"] = (*cinfo)["id"]
-	//				saveRelationPool <- saveM
-	//			}
-	//		}
-	//	}
-	//}
-
-	for _, item := range tmp["list"].([]interface{}) {
-		item1 := item.(map[string]interface{})
-		sw := util.ObjToString(item1["s_winner"])
-		if !strings.Contains(sw, ",") {
-			if code := redis.GetStr("qyxy_id", sw); code != "" {
-				saveM := make(map[string]interface{})
-				saveM["projectid"] = pid
-				saveM["infoid"] = lid
-				saveM["identity_type"] = 2
-				saveM["createtime"] = time.Now().Format(util.Date_Full_Layout)
-				saveM["name_id"] = code
-				if util.ObjToString(item1["winnertel"]) != "" {
-					q := make(map[string]interface{})
-					q["name_id"] = code
-					q["identity_type"] = 2 // 010
-					q["contact_tel"] = util.ObjToString(item1["winnertel"])
-					if util.ObjToString(item1["winnerperson"]) != "" {
-						q["contact_name"] = util.ObjToString(item1["winnerperson"])
-					}
-					cinfo := MysqlTool.FindOne("dws_f_ent_contact", q, "", "")
-					if cinfo != nil && len(*cinfo) > 0 {
-						saveM["contact_id"] = (*cinfo)["id"]
-						saveRelationPool <- saveM
-					}
-				}
-			}
-		}
-	}
-}
-
-func taskRelation2(tmp map[string]interface{}) {
-
-	pid := mongodb.BsonIdToSId(tmp["_id"])
-	if tmp["ids"] == nil {
-		log.Info("taskRelation ids err", zap.Any("id", pid))
-		return
-	}
-	info := MysqlTool.Find("dws_f_bpmc_relation", bson.M{"projectid": pid}, "", "", -1, -1)
-	if len(*info) > 0 {
-
-	} else {
-		ids := util.ObjArrToStringArr(tmp["ids"].([]interface{}))
-		lid := ids[len(ids)-1]
-
-		if b := util.ObjToString(tmp["buyer"]); b != "" {
-			saveM := make(map[string]interface{})
-			for _, f := range RelationField {
-				if f == "projectid" {
-					saveM[f] = pid
-				} else if f == "infoid" {
-					saveM[f] = lid
-				} else if f == "name_id" {
-					if code := redis.GetStr("qyxy_id", b); code != "" {
-						saveM[f] = code
-						if util.ObjToString(tmp["buyertel"]) != "" {
-							q := make(map[string]interface{})
-							q["name_id"] = code
-							q["identity_type"] = 1
-							q["contact_tel"] = util.ObjToString(tmp["buyertel"])
-							if util.ObjToString(tmp["buyerperson"]) != "" {
-								q["contact_name"] = util.ObjToString(tmp["buyerperson"])
-							}
-							cinfo := MysqlTool.FindOne("dws_f_ent_contact", q, "", "")
-							if cinfo != nil && len(*cinfo) > 0 {
-								saveM["contact_id"] = (*cinfo)["id"]
-							}
-						}
-					}
-				} else if f == "identity_type" {
-					saveM[f] = 1 // 001
-				} else if f == "createtime" {
-					saveM[f] = time.Now().Format(util.Date_Full_Layout)
-				}
-			}
-			saveRelationPool <- saveM
-		}
-
-		if a := util.ObjToString(tmp["agency"]); a != "" {
-			saveM := make(map[string]interface{})
-			for _, f := range RelationField {
-				if f == "projectid" {
-					saveM[f] = pid
-				} else if f == "infoid" {
-					saveM[f] = lid
-				} else if f == "name_id" {
-					if code := redis.GetStr("qyxy_id", a); code != "" {
-						saveM[f] = code
-						if util.ObjToString(tmp["buyertel"]) != "" {
-							q := make(map[string]interface{})
-							q["name_id"] = code
-							q["identity_type"] = 4
-							q["contact_tel"] = util.ObjToString(tmp["agencytel"])
-							if util.ObjToString(tmp["agencyperson"]) != "" {
-								q["contact_name"] = util.ObjToString(tmp["agencyperson"])
-							}
-							cinfo := MysqlTool.FindOne("dws_f_ent_contact", q, "", "")
-							if cinfo != nil && len(*cinfo) > 0 {
-								saveM["contact_id"] = (*cinfo)["id"]
-							}
-						}
-					}
-				} else if f == "identity_type" {
-					saveM[f] = 4 // 100
-				} else if f == "createtime" {
-					saveM[f] = time.Now().Format(util.Date_Full_Layout)
-				}
-			}
-			saveRelationPool <- saveM
-		}
-
-		warr := strings.Split(util.ObjToString(tmp["s_winner"]), ",")
-		if BinarySearch(warr, util.ObjToString(tmp["winner"])) == -1 {
-			warr = append(warr, util.ObjToString(tmp["winner"]))
-		}
-		for _, ws := range warr {
-			saveM := make(map[string]interface{})
-			for _, f := range RelationField {
-				if f == "projectid" {
-					saveM[f] = pid
-				} else if f == "infoid" {
-					saveM[f] = lid
-				} else if f == "name_id" {
-					if code := redis.GetStr("qyxy_id", ws); code != "" {
-						saveM[f] = code
-						if util.ObjToString(tmp["buyertel"]) != "" {
-							q := make(map[string]interface{})
-							q["name_id"] = code
-							q["identity_type"] = 2
-							q["contact_tel"] = util.ObjToString(tmp["winnertel"])
-							if util.ObjToString(tmp["winnerperson"]) != "" {
-								q["contact_name"] = util.ObjToString(tmp["winnerperson"])
-							}
-							cinfo := MysqlTool.FindOne("dws_f_ent_contact", q, "", "")
-							if cinfo != nil && len(*cinfo) > 0 {
-								saveM["contact_id"] = (*cinfo)["id"]
-							}
-						}
-					}
-				} else if f == "identity_type" {
-					saveM[f] = 2 // 010
-				} else if f == "createtime" {
-					saveM[f] = time.Now().Format(util.Date_Full_Layout)
-				}
-			}
-			saveRelationPool <- saveM
-		}
-	}
-}
+//
+//import (
+//	"data_tidb/config"
+//	"fmt"
+//	"go.mongodb.org/mongo-driver/bson"
+//	"go.uber.org/zap"
+//	util "jygit.jydev.jianyu360.cn/data_processing/common_utils"
+//	"jygit.jydev.jianyu360.cn/data_processing/common_utils/log"
+//	"jygit.jydev.jianyu360.cn/data_processing/common_utils/mongodb"
+//	"jygit.jydev.jianyu360.cn/data_processing/common_utils/redis"
+//	"strings"
+//	"sync"
+//	"time"
+//)
+//
+//func taskP() {
+//	sess := MongoP.GetMgoConn()
+//	defer MongoP.DestoryMongoConn(sess)
+//
+//	ch := make(chan bool, 20)
+//	wg := &sync.WaitGroup{}
+//
+//	//q := map[string]interface{}{"_id": mongodb.StringTOBsonId("64e5a63855d5406905c574e6")}
+//	query := sess.DB(config.Conf.DB.MongoP.Dbname).C("projectset_20230407").Find(nil).Sort("-_id").Skip(100000).Iter()
+//	count := 0
+//	for tmp := make(map[string]interface{}); query.Next(tmp); count++ {
+//		if count%20000 == 0 {
+//			log.Info(fmt.Sprintf("current --- %d", count))
+//		}
+//		ch <- true
+//		wg.Add(1)
+//		go func(tmp map[string]interface{}) {
+//			defer func() {
+//				<-ch
+//				wg.Done()
+//			}()
+//			//taskPro(tmp)
+//			//taskBusiness(tmp)
+//			//taskProTag(tmp)
+//			//taskRelation(tmp)
+//
+//		}(tmp)
+//		tmp = make(map[string]interface{})
+//	}
+//	wg.Wait()
+//	log.Info(fmt.Sprintf("over --- %d", count))
+//}
+//
+//func taskPAdd(pici int64) {
+//	sess := MongoP.GetMgoConn()
+//	defer MongoP.DestoryMongoConn(sess)
+//
+//	ch := make(chan bool, 20)
+//	wg := &sync.WaitGroup{}
+//
+//	q := bson.M{"pici": bson.M{"$gt": pici}}
+//	query := sess.DB(config.Conf.DB.MongoP.Dbname).C("projectset_20230407").Find(q).Iter()
+//	count := 0
+//	for tmp := make(map[string]interface{}); query.Next(tmp); count++ {
+//		if count%20000 == 0 {
+//			log.Info(fmt.Sprintf("current --- %d", count))
+//		}
+//		ch <- true
+//		wg.Add(1)
+//		go func(tmp map[string]interface{}) {
+//			defer func() {
+//				<-ch
+//				wg.Done()
+//			}()
+//			//taskPro(tmp)
+//			//taskBusiness(tmp)
+//			//taskProTag(tmp)
+//			//taskRelation2(tmp)
+//
+//		}(tmp)
+//		tmp = make(map[string]interface{})
+//	}
+//	wg.Wait()
+//	log.Info(fmt.Sprintf("over --- %d", count))
+//}
+//
+//var BidStatus = map[string]int{
+//	"预告": 0,
+//	"拟建": 1,
+//	"招标": 2,
+//	"中标": 3,
+//	"成交": 4,
+//	"废标": 5,
+//	"流标": 6,
+//	"合同": 7,
+//	"其它": 8,
+//}
+//var BidType = map[string]int{
+//	"招标": 0,
+//	"邀标": 1,
+//	"单一": 2,
+//	"竞价": 3,
+//	"竞谈": 4,
+//	"询价": 5,
+//}
+//
+//// @Description 基础信息
+//// @Author J 2022/9/22 18:32
+//func taskPro(tmp map[string]interface{}) {
+//	saveM := make(map[string]interface{})
+//	for _, f := range ProField {
+//		if f == "projectid" {
+//			saveM[f] = mongodb.BsonIdToSId(tmp["_id"])
+//		} else if f == "area_code" {
+//			if tmp["area"] != nil {
+//				saveM[f] = AreaCode[util.ObjToString(tmp["area"])]
+//			}
+//		} else if f == "city_code" {
+//			if tmp["area"] != nil && tmp["city"] != nil {
+//				c := util.ObjToString(tmp["area"]) + "," + util.ObjToString(tmp["city"])
+//				saveM[f] = AreaCode[c]
+//			}
+//		} else if f == "district_code" {
+//			if tmp["area"] != nil && tmp["city"] != nil && tmp["district"] != nil {
+//				c := util.ObjToString(tmp["area"]) + "," + util.ObjToString(tmp["city"]) + "," + util.ObjToString(tmp["district"])
+//				saveM[f] = AreaCode[c]
+//			}
+//		} else if f == "updatetime" {
+//			saveM[f] = time.Now().Format(util.Date_Full_Layout)
+//		} else if f == "buyerclass_code" {
+//			if obj := util.ObjToString(tmp["buyerclass"]); obj != "" {
+//				saveM[f] = BuyerCode[obj]
+//			}
+//		} else if f == "firsttime" || f == "zbtime" || f == "jgtime" || f == "lasttime" || f == "bidopentime" || f == "createtime" {
+//			if tmp[f] != nil && util.Int64All(tmp[f]) > 0 {
+//				t := util.Int64All(tmp[f])
+//				saveM[f] = util.FormatDateByInt64(&t, util.Date_Full_Layout)
+//			}
+//		} else if f == "bidstatus" {
+//			if b := util.ObjToString(tmp[f]); b != "" {
+//				tmp[f] = BidStatus[b]
+//			}
+//		} else if f == "bidtype" {
+//			if b := util.ObjToString(tmp[f]); b != "" {
+//				tmp[f] = BidType[b]
+//			}
+//		} else if f == "multipackage" {
+//			if tmp[f] == nil {
+//				saveM[f] = 0
+//			} else {
+//				saveM[f] = tmp[f]
+//			}
+//		} else if f == "buyer_id" {
+//			if b := util.ObjToString(tmp["buyer"]); b != "" {
+//				if code := redis.GetStr("qyxy_id", b); code != "" {
+//					saveM[f] = code
+//				}
+//			}
+//		} else if f == "agency_id" {
+//			if b := util.ObjToString(tmp["agency"]); b != "" {
+//				if code := redis.GetStr("qyxy_id", b); code != "" {
+//					saveM[f] = code
+//				}
+//			}
+//		} else {
+//			if tmp[f] != nil {
+//				if ProVMap[f] != nil {
+//					saveM[f], _ = verifyF(f, tmp[f], ProVMap[f])
+//				} else {
+//					saveM[f] = tmp[f]
+//				}
+//			}
+//		}
+//	}
+//	saveProPool <- saveM
+//}
+//
+//// @Description 项目业务表
+//// @Author J 2022/9/30 13:40
+//func taskBusiness(tmp map[string]interface{}) {
+//	warr := strings.Split(util.ObjToString(tmp["s_winner"]), ",")
+//	if BinarySearch(warr, util.ObjToString(tmp["winner"])) == -1 {
+//		warr = append(warr, util.ObjToString(tmp["winner"]))
+//	}
+//	for _, s := range warr {
+//		saveM := make(map[string]interface{})
+//		for _, f := range ProBusField {
+//			if f == "projectid" {
+//				saveM[f] = mongodb.BsonIdToSId(tmp["_id"])
+//			} else if f == "area_code" {
+//				if tmp["area"] != nil {
+//					saveM[f] = AreaCode[util.ObjToString(tmp["area"])]
+//				}
+//			} else if f == "city_code" {
+//				if tmp["area"] != nil && tmp["city"] != nil {
+//					c := util.ObjToString(tmp["area"]) + "," + util.ObjToString(tmp["city"])
+//					saveM[f] = AreaCode[c]
+//				}
+//			} else if f == "district_code" {
+//				if tmp["area"] != nil && tmp["city"] != nil && tmp["district"] != nil {
+//					c := util.ObjToString(tmp["area"]) + "," + util.ObjToString(tmp["city"]) + "," + util.ObjToString(tmp["district"])
+//					saveM[f] = AreaCode[c]
+//				}
+//			} else if f == "updatetime" {
+//				saveM[f] = time.Now().Format(util.Date_Full_Layout)
+//			} else if f == "buyerclass_code" {
+//				if obj := util.ObjToString(tmp["buyerclass"]); obj != "" {
+//					saveM[f] = BuyerCode[obj]
+//				}
+//			} else if f == "firsttime" || f == "zbtime" || f == "jgtime" || f == "lasttime" || f == "bidopentime" || f == "createtime" {
+//				if tmp[f] != nil && util.Int64All(tmp[f]) > 0 {
+//					t := util.Int64All(tmp[f])
+//					saveM[f] = util.FormatDateByInt64(&t, util.Date_Full_Layout)
+//				}
+//			} else if f == "bidstatus" {
+//				if b := util.ObjToString(tmp[f]); b != "" {
+//					tmp[f] = BidStatus[b]
+//				}
+//			} else if f == "bidtype" {
+//				if b := util.ObjToString(tmp[f]); b != "" {
+//					tmp[f] = BidType[b]
+//				}
+//			} else if f == "buyer_id" {
+//				if b := util.ObjToString(tmp["buyer"]); b != "" {
+//					saveM["buyer"] = b
+//					if code := redis.GetStr("qyxy_id", b); code != "" {
+//						saveM[f] = code
+//					}
+//				}
+//			} else if f == "agency_id" {
+//				if b := util.ObjToString(tmp["agency"]); b != "" {
+//					saveM["agency"] = b
+//					if code := redis.GetStr("qyxy_id", b); code != "" {
+//						saveM[f] = code
+//					}
+//				}
+//			} else if f == "winner_id" {
+//				if s != "" {
+//					saveM["winner"] = s
+//					if code := redis.GetStr("qyxy_id", s); code != "" {
+//						saveM[f] = code
+//					}
+//				}
+//			} else {
+//				if tmp[f] != nil {
+//					if ProBusVMap[f] != nil {
+//						saveM[f], _ = verifyF(f, tmp[f], ProBusVMap[f])
+//					} else {
+//						saveM[f] = tmp[f]
+//					}
+//				}
+//			}
+//		}
+//		saveProbPool <- saveM
+//	}
+//
+//}
+//
+//// @Description 项目信息标签
+//// @Author J 2022/9/30 13:54
+//func taskProTag(tmp map[string]interface{}) {
+//	id := mongodb.BsonIdToSId(tmp["_id"])
+//	if topArr, ok := tmp["topscopeclass"].([]interface{}); ok {
+//		for _, i2 := range topArr {
+//			tclass := regLetter.ReplaceAllString(util.ObjToString(i2), "") // 去除字母
+//			code := TopScopeCode[tclass]
+//			saveProTagPool <- map[string]interface{}{"projectid": id, "labelcode": "1", "labelvalues": code, "labelweight": 1, "createtime": time.Now().Format(util.Date_Full_Layout)}
+//		}
+//	}
+//	if subArr, ok := tmp["subscopeclass"].([]interface{}); ok {
+//		for _, i2 := range subArr {
+//			sc := strings.Split(util.ObjToString(i2), "_")
+//			code := SubScopeCode[sc[1]]
+//			saveProTagPool <- map[string]interface{}{"projectid": id, "labelcode": "2", "labelvalues": code, "labelweight": 1, "createtime": time.Now().Format(util.Date_Full_Layout)}
+//		}
+//	}
+//}
+//
+//// @Description 关系表
+//// @Author J 2022/9/30 13:56
+//func taskRelation(tmp map[string]interface{}) {
+//	pid := mongodb.BsonIdToSId(tmp["_id"])
+//	if tmp["ids"] == nil {
+//		log.Info("taskRelation ids err", zap.Any("id", pid))
+//		return
+//	}
+//	ids := util.ObjArrToStringArr(tmp["ids"].([]interface{}))
+//	lid := ids[len(ids)-1]
+//
+//	//if b := util.ObjToString(tmp["buyer"]); b != "" {
+//	//	saveM := make(map[string]interface{})
+//	//
+//	//	saveM["projectid"] = pid
+//	//	saveM["infoid"] = lid
+//	//	saveM["identity_type"] = 1
+//	//	saveM["createtime"] = time.Now().Format(util.Date_Full_Layout)
+//	//	if code := redis.GetStr("qyxy_id", b); code != "" {
+//	//		saveM["name_id"] = code
+//	//		if util.ObjToString(tmp["buyertel"]) != "" {
+//	//			q := make(map[string]interface{})
+//	//			q["name_id"] = code
+//	//			q["identity_type"] = 1
+//	//			q["contact_tel"] = util.ObjToString(tmp["buyertel"])
+//	//			if util.ObjToString(tmp["buyerperson"]) != "" {
+//	//				q["contact_name"] = util.ObjToString(tmp["buyerperson"])
+//	//			}
+//	//			cinfo := MysqlTool.FindOne("dws_f_ent_contact", q, "", "")
+//	//			if cinfo != nil && len(*cinfo) > 0 {
+//	//				saveM["contact_id"] = (*cinfo)["id"]
+//	//				saveRelationPool <- saveM
+//	//			}
+//	//		}
+//	//	}
+//	//}
+//
+//	//if a := util.ObjToString(tmp["agency"]); a != "" {
+//	//	saveM := make(map[string]interface{})
+//	//	saveM["projectid"] = pid
+//	//	saveM["infoid"] = lid
+//	//	saveM["identity_type"] = 4
+//	//	saveM["createtime"] = time.Now().Format(util.Date_Full_Layout)
+//	//	if code := redis.GetStr("qyxy_id", a); code != "" {
+//	//		saveM["name_id"] = code
+//	//		if util.ObjToString(tmp["agencytel"]) != "" {
+//	//			q := make(map[string]interface{})
+//	//			q["name_id"] = code
+//	//			q["identity_type"] = 4 // 100
+//	//			q["contact_tel"] = util.ObjToString(tmp["agencytel"])
+//	//			if util.ObjToString(tmp["agencyperson"]) != "" {
+//	//				q["contact_name"] = util.ObjToString(tmp["agencyperson"])
+//	//			}
+//	//			cinfo := MysqlTool.FindOne("dws_f_ent_contact", q, "", "")
+//	//			if cinfo != nil && len(*cinfo) > 0 {
+//	//				saveM["contact_id"] = (*cinfo)["id"]
+//	//				saveRelationPool <- saveM
+//	//			}
+//	//		}
+//	//	}
+//	//}
+//
+//	for _, item := range tmp["list"].([]interface{}) {
+//		item1 := item.(map[string]interface{})
+//		sw := util.ObjToString(item1["s_winner"])
+//		if !strings.Contains(sw, ",") {
+//			if code := redis.GetStr("qyxy_id", sw); code != "" {
+//				saveM := make(map[string]interface{})
+//				saveM["projectid"] = pid
+//				saveM["infoid"] = lid
+//				saveM["identity_type"] = 2
+//				saveM["createtime"] = time.Now().Format(util.Date_Full_Layout)
+//				saveM["name_id"] = code
+//				if util.ObjToString(item1["winnertel"]) != "" {
+//					q := make(map[string]interface{})
+//					q["name_id"] = code
+//					q["identity_type"] = 2 // 010
+//					q["contact_tel"] = util.ObjToString(item1["winnertel"])
+//					if util.ObjToString(item1["winnerperson"]) != "" {
+//						q["contact_name"] = util.ObjToString(item1["winnerperson"])
+//					}
+//					cinfo := MysqlTool.FindOne("dws_f_ent_contact", q, "", "")
+//					if cinfo != nil && len(*cinfo) > 0 {
+//						saveM["contact_id"] = (*cinfo)["id"]
+//						saveRelationPool <- saveM
+//					}
+//				}
+//			}
+//		}
+//	}
+//}
+//
+//func taskRelation2(tmp map[string]interface{}) {
+//
+//	pid := mongodb.BsonIdToSId(tmp["_id"])
+//	if tmp["ids"] == nil {
+//		log.Info("taskRelation ids err", zap.Any("id", pid))
+//		return
+//	}
+//	info := MysqlTool.Find("dws_f_bpmc_relation", bson.M{"projectid": pid}, "", "", -1, -1)
+//	if len(*info) > 0 {
+//
+//	} else {
+//		ids := util.ObjArrToStringArr(tmp["ids"].([]interface{}))
+//		lid := ids[len(ids)-1]
+//
+//		if b := util.ObjToString(tmp["buyer"]); b != "" {
+//			saveM := make(map[string]interface{})
+//			for _, f := range RelationField {
+//				if f == "projectid" {
+//					saveM[f] = pid
+//				} else if f == "infoid" {
+//					saveM[f] = lid
+//				} else if f == "name_id" {
+//					if code := redis.GetStr("qyxy_id", b); code != "" {
+//						saveM[f] = code
+//						if util.ObjToString(tmp["buyertel"]) != "" {
+//							q := make(map[string]interface{})
+//							q["name_id"] = code
+//							q["identity_type"] = 1
+//							q["contact_tel"] = util.ObjToString(tmp["buyertel"])
+//							if util.ObjToString(tmp["buyerperson"]) != "" {
+//								q["contact_name"] = util.ObjToString(tmp["buyerperson"])
+//							}
+//							cinfo := MysqlTool.FindOne("dws_f_ent_contact", q, "", "")
+//							if cinfo != nil && len(*cinfo) > 0 {
+//								saveM["contact_id"] = (*cinfo)["id"]
+//							}
+//						}
+//					}
+//				} else if f == "identity_type" {
+//					saveM[f] = 1 // 001
+//				} else if f == "createtime" {
+//					saveM[f] = time.Now().Format(util.Date_Full_Layout)
+//				}
+//			}
+//			saveRelationPool <- saveM
+//		}
+//
+//		if a := util.ObjToString(tmp["agency"]); a != "" {
+//			saveM := make(map[string]interface{})
+//			for _, f := range RelationField {
+//				if f == "projectid" {
+//					saveM[f] = pid
+//				} else if f == "infoid" {
+//					saveM[f] = lid
+//				} else if f == "name_id" {
+//					if code := redis.GetStr("qyxy_id", a); code != "" {
+//						saveM[f] = code
+//						if util.ObjToString(tmp["buyertel"]) != "" {
+//							q := make(map[string]interface{})
+//							q["name_id"] = code
+//							q["identity_type"] = 4
+//							q["contact_tel"] = util.ObjToString(tmp["agencytel"])
+//							if util.ObjToString(tmp["agencyperson"]) != "" {
+//								q["contact_name"] = util.ObjToString(tmp["agencyperson"])
+//							}
+//							cinfo := MysqlTool.FindOne("dws_f_ent_contact", q, "", "")
+//							if cinfo != nil && len(*cinfo) > 0 {
+//								saveM["contact_id"] = (*cinfo)["id"]
+//							}
+//						}
+//					}
+//				} else if f == "identity_type" {
+//					saveM[f] = 4 // 100
+//				} else if f == "createtime" {
+//					saveM[f] = time.Now().Format(util.Date_Full_Layout)
+//				}
+//			}
+//			saveRelationPool <- saveM
+//		}
+//
+//		warr := strings.Split(util.ObjToString(tmp["s_winner"]), ",")
+//		if BinarySearch(warr, util.ObjToString(tmp["winner"])) == -1 {
+//			warr = append(warr, util.ObjToString(tmp["winner"]))
+//		}
+//		for _, ws := range warr {
+//			saveM := make(map[string]interface{})
+//			for _, f := range RelationField {
+//				if f == "projectid" {
+//					saveM[f] = pid
+//				} else if f == "infoid" {
+//					saveM[f] = lid
+//				} else if f == "name_id" {
+//					if code := redis.GetStr("qyxy_id", ws); code != "" {
+//						saveM[f] = code
+//						if util.ObjToString(tmp["buyertel"]) != "" {
+//							q := make(map[string]interface{})
+//							q["name_id"] = code
+//							q["identity_type"] = 2
+//							q["contact_tel"] = util.ObjToString(tmp["winnertel"])
+//							if util.ObjToString(tmp["winnerperson"]) != "" {
+//								q["contact_name"] = util.ObjToString(tmp["winnerperson"])
+//							}
+//							cinfo := MysqlTool.FindOne("dws_f_ent_contact", q, "", "")
+//							if cinfo != nil && len(*cinfo) > 0 {
+//								saveM["contact_id"] = (*cinfo)["id"]
+//							}
+//						}
+//					}
+//				} else if f == "identity_type" {
+//					saveM[f] = 2 // 010
+//				} else if f == "createtime" {
+//					saveM[f] = time.Now().Format(util.Date_Full_Layout)
+//				}
+//			}
+//			saveRelationPool <- saveM
+//		}
+//	}
+//}
+//package main
+//
+//import (
+//"data_tidb/config"
+//"fmt"
+//"go.mongodb.org/mongo-driver/bson"
+//"go.uber.org/zap"
+//util "jygit.jydev.jianyu360.cn/data_processing/common_utils"
+//"jygit.jydev.jianyu360.cn/data_processing/common_utils/log"
+//"jygit.jydev.jianyu360.cn/data_processing/common_utils/mongodb"
+//"jygit.jydev.jianyu360.cn/data_processing/common_utils/redis"
+//"strings"
+//"sync"
+//"time"
+//)
+//
+//func taskP() {
+//	sess := MongoP.GetMgoConn()
+//	defer MongoP.DestoryMongoConn(sess)
+//
+//	ch := make(chan bool, 20)
+//	wg := &sync.WaitGroup{}
+//
+//	//q := map[string]interface{}{"_id": mongodb.StringTOBsonId("64e5a63855d5406905c574e6")}
+//	query := sess.DB(config.Conf.DB.MongoP.Dbname).C("projectset_20230407").Find(nil).Sort("-_id").Skip(100000).Iter()
+//	count := 0
+//	for tmp := make(map[string]interface{}); query.Next(tmp); count++ {
+//		if count%20000 == 0 {
+//			log.Info(fmt.Sprintf("current --- %d", count))
+//		}
+//		ch <- true
+//		wg.Add(1)
+//		go func(tmp map[string]interface{}) {
+//			defer func() {
+//				<-ch
+//				wg.Done()
+//			}()
+//			//taskPro(tmp)
+//			//taskBusiness(tmp)
+//			//taskProTag(tmp)
+//			//taskRelation(tmp)
+//
+//		}(tmp)
+//		tmp = make(map[string]interface{})
+//	}
+//	wg.Wait()
+//	log.Info(fmt.Sprintf("over --- %d", count))
+//}
+//
+//func taskPAdd(pici int64) {
+//	sess := MongoP.GetMgoConn()
+//	defer MongoP.DestoryMongoConn(sess)
+//
+//	ch := make(chan bool, 20)
+//	wg := &sync.WaitGroup{}
+//
+//	q := bson.M{"pici": bson.M{"$gt": pici}}
+//	query := sess.DB(config.Conf.DB.MongoP.Dbname).C("projectset_20230407").Find(q).Iter()
+//	count := 0
+//	for tmp := make(map[string]interface{}); query.Next(tmp); count++ {
+//		if count%20000 == 0 {
+//			log.Info(fmt.Sprintf("current --- %d", count))
+//		}
+//		ch <- true
+//		wg.Add(1)
+//		go func(tmp map[string]interface{}) {
+//			defer func() {
+//				<-ch
+//				wg.Done()
+//			}()
+//			//taskPro(tmp)
+//			//taskBusiness(tmp)
+//			//taskProTag(tmp)
+//			//taskRelation2(tmp)
+//
+//		}(tmp)
+//		tmp = make(map[string]interface{})
+//	}
+//	wg.Wait()
+//	log.Info(fmt.Sprintf("over --- %d", count))
+//}
+//
+//var BidStatus = map[string]int{
+//	"预告": 0,
+//	"拟建": 1,
+//	"招标": 2,
+//	"中标": 3,
+//	"成交": 4,
+//	"废标": 5,
+//	"流标": 6,
+//	"合同": 7,
+//	"其它": 8,
+//}
+//var BidType = map[string]int{
+//	"招标": 0,
+//	"邀标": 1,
+//	"单一": 2,
+//	"竞价": 3,
+//	"竞谈": 4,
+//	"询价": 5,
+//}
+//
+//// @Description 基础信息
+//// @Author J 2022/9/22 18:32
+//func taskPro(tmp map[string]interface{}) {
+//	saveM := make(map[string]interface{})
+//	for _, f := range ProField {
+//		if f == "projectid" {
+//			saveM[f] = mongodb.BsonIdToSId(tmp["_id"])
+//		} else if f == "area_code" {
+//			if tmp["area"] != nil {
+//				saveM[f] = AreaCode[util.ObjToString(tmp["area"])]
+//			}
+//		} else if f == "city_code" {
+//			if tmp["area"] != nil && tmp["city"] != nil {
+//				c := util.ObjToString(tmp["area"]) + "," + util.ObjToString(tmp["city"])
+//				saveM[f] = AreaCode[c]
+//			}
+//		} else if f == "district_code" {
+//			if tmp["area"] != nil && tmp["city"] != nil && tmp["district"] != nil {
+//				c := util.ObjToString(tmp["area"]) + "," + util.ObjToString(tmp["city"]) + "," + util.ObjToString(tmp["district"])
+//				saveM[f] = AreaCode[c]
+//			}
+//		} else if f == "updatetime" {
+//			saveM[f] = time.Now().Format(util.Date_Full_Layout)
+//		} else if f == "buyerclass_code" {
+//			if obj := util.ObjToString(tmp["buyerclass"]); obj != "" {
+//				saveM[f] = BuyerCode[obj]
+//			}
+//		} else if f == "firsttime" || f == "zbtime" || f == "jgtime" || f == "lasttime" || f == "bidopentime" || f == "createtime" {
+//			if tmp[f] != nil && util.Int64All(tmp[f]) > 0 {
+//				t := util.Int64All(tmp[f])
+//				saveM[f] = util.FormatDateByInt64(&t, util.Date_Full_Layout)
+//			}
+//		} else if f == "bidstatus" {
+//			if b := util.ObjToString(tmp[f]); b != "" {
+//				tmp[f] = BidStatus[b]
+//			}
+//		} else if f == "bidtype" {
+//			if b := util.ObjToString(tmp[f]); b != "" {
+//				tmp[f] = BidType[b]
+//			}
+//		} else if f == "multipackage" {
+//			if tmp[f] == nil {
+//				saveM[f] = 0
+//			} else {
+//				saveM[f] = tmp[f]
+//			}
+//		} else if f == "buyer_id" {
+//			if b := util.ObjToString(tmp["buyer"]); b != "" {
+//				if code := redis.GetStr("qyxy_id", b); code != "" {
+//					saveM[f] = code
+//				}
+//			}
+//		} else if f == "agency_id" {
+//			if b := util.ObjToString(tmp["agency"]); b != "" {
+//				if code := redis.GetStr("qyxy_id", b); code != "" {
+//					saveM[f] = code
+//				}
+//			}
+//		} else {
+//			if tmp[f] != nil {
+//				if ProVMap[f] != nil {
+//					saveM[f], _ = verifyF(f, tmp[f], ProVMap[f])
+//				} else {
+//					saveM[f] = tmp[f]
+//				}
+//			}
+//		}
+//	}
+//	saveProPool <- saveM
+//}
+//
+//// @Description 项目业务表
+//// @Author J 2022/9/30 13:40
+//func taskBusiness(tmp map[string]interface{}) {
+//	warr := strings.Split(util.ObjToString(tmp["s_winner"]), ",")
+//	if BinarySearch(warr, util.ObjToString(tmp["winner"])) == -1 {
+//		warr = append(warr, util.ObjToString(tmp["winner"]))
+//	}
+//	for _, s := range warr {
+//		saveM := make(map[string]interface{})
+//		for _, f := range ProBusField {
+//			if f == "projectid" {
+//				saveM[f] = mongodb.BsonIdToSId(tmp["_id"])
+//			} else if f == "area_code" {
+//				if tmp["area"] != nil {
+//					saveM[f] = AreaCode[util.ObjToString(tmp["area"])]
+//				}
+//			} else if f == "city_code" {
+//				if tmp["area"] != nil && tmp["city"] != nil {
+//					c := util.ObjToString(tmp["area"]) + "," + util.ObjToString(tmp["city"])
+//					saveM[f] = AreaCode[c]
+//				}
+//			} else if f == "district_code" {
+//				if tmp["area"] != nil && tmp["city"] != nil && tmp["district"] != nil {
+//					c := util.ObjToString(tmp["area"]) + "," + util.ObjToString(tmp["city"]) + "," + util.ObjToString(tmp["district"])
+//					saveM[f] = AreaCode[c]
+//				}
+//			} else if f == "updatetime" {
+//				saveM[f] = time.Now().Format(util.Date_Full_Layout)
+//			} else if f == "buyerclass_code" {
+//				if obj := util.ObjToString(tmp["buyerclass"]); obj != "" {
+//					saveM[f] = BuyerCode[obj]
+//				}
+//			} else if f == "firsttime" || f == "zbtime" || f == "jgtime" || f == "lasttime" || f == "bidopentime" || f == "createtime" {
+//				if tmp[f] != nil && util.Int64All(tmp[f]) > 0 {
+//					t := util.Int64All(tmp[f])
+//					saveM[f] = util.FormatDateByInt64(&t, util.Date_Full_Layout)
+//				}
+//			} else if f == "bidstatus" {
+//				if b := util.ObjToString(tmp[f]); b != "" {
+//					tmp[f] = BidStatus[b]
+//				}
+//			} else if f == "bidtype" {
+//				if b := util.ObjToString(tmp[f]); b != "" {
+//					tmp[f] = BidType[b]
+//				}
+//			} else if f == "buyer_id" {
+//				if b := util.ObjToString(tmp["buyer"]); b != "" {
+//					saveM["buyer"] = b
+//					if code := redis.GetStr("qyxy_id", b); code != "" {
+//						saveM[f] = code
+//					}
+//				}
+//			} else if f == "agency_id" {
+//				if b := util.ObjToString(tmp["agency"]); b != "" {
+//					saveM["agency"] = b
+//					if code := redis.GetStr("qyxy_id", b); code != "" {
+//						saveM[f] = code
+//					}
+//				}
+//			} else if f == "winner_id" {
+//				if s != "" {
+//					saveM["winner"] = s
+//					if code := redis.GetStr("qyxy_id", s); code != "" {
+//						saveM[f] = code
+//					}
+//				}
+//			} else {
+//				if tmp[f] != nil {
+//					if ProBusVMap[f] != nil {
+//						saveM[f], _ = verifyF(f, tmp[f], ProBusVMap[f])
+//					} else {
+//						saveM[f] = tmp[f]
+//					}
+//				}
+//			}
+//		}
+//		saveProbPool <- saveM
+//	}
+//
+//}
+//
+//// @Description 项目信息标签
+//// @Author J 2022/9/30 13:54
+//func taskProTag(tmp map[string]interface{}) {
+//	id := mongodb.BsonIdToSId(tmp["_id"])
+//	if topArr, ok := tmp["topscopeclass"].([]interface{}); ok {
+//		for _, i2 := range topArr {
+//			tclass := regLetter.ReplaceAllString(util.ObjToString(i2), "") // 去除字母
+//			code := TopScopeCode[tclass]
+//			saveProTagPool <- map[string]interface{}{"projectid": id, "labelcode": "1", "labelvalues": code, "labelweight": 1, "createtime": time.Now().Format(util.Date_Full_Layout)}
+//		}
+//	}
+//	if subArr, ok := tmp["subscopeclass"].([]interface{}); ok {
+//		for _, i2 := range subArr {
+//			sc := strings.Split(util.ObjToString(i2), "_")
+//			code := SubScopeCode[sc[1]]
+//			saveProTagPool <- map[string]interface{}{"projectid": id, "labelcode": "2", "labelvalues": code, "labelweight": 1, "createtime": time.Now().Format(util.Date_Full_Layout)}
+//		}
+//	}
+//}
+//
+//// @Description 关系表
+//// @Author J 2022/9/30 13:56
+//func taskRelation(tmp map[string]interface{}) {
+//	pid := mongodb.BsonIdToSId(tmp["_id"])
+//	if tmp["ids"] == nil {
+//		log.Info("taskRelation ids err", zap.Any("id", pid))
+//		return
+//	}
+//	ids := util.ObjArrToStringArr(tmp["ids"].([]interface{}))
+//	lid := ids[len(ids)-1]
+//
+//	//if b := util.ObjToString(tmp["buyer"]); b != "" {
+//	//	saveM := make(map[string]interface{})
+//	//
+//	//	saveM["projectid"] = pid
+//	//	saveM["infoid"] = lid
+//	//	saveM["identity_type"] = 1
+//	//	saveM["createtime"] = time.Now().Format(util.Date_Full_Layout)
+//	//	if code := redis.GetStr("qyxy_id", b); code != "" {
+//	//		saveM["name_id"] = code
+//	//		if util.ObjToString(tmp["buyertel"]) != "" {
+//	//			q := make(map[string]interface{})
+//	//			q["name_id"] = code
+//	//			q["identity_type"] = 1
+//	//			q["contact_tel"] = util.ObjToString(tmp["buyertel"])
+//	//			if util.ObjToString(tmp["buyerperson"]) != "" {
+//	//				q["contact_name"] = util.ObjToString(tmp["buyerperson"])
+//	//			}
+//	//			cinfo := MysqlTool.FindOne("dws_f_ent_contact", q, "", "")
+//	//			if cinfo != nil && len(*cinfo) > 0 {
+//	//				saveM["contact_id"] = (*cinfo)["id"]
+//	//				saveRelationPool <- saveM
+//	//			}
+//	//		}
+//	//	}
+//	//}
+//
+//	//if a := util.ObjToString(tmp["agency"]); a != "" {
+//	//	saveM := make(map[string]interface{})
+//	//	saveM["projectid"] = pid
+//	//	saveM["infoid"] = lid
+//	//	saveM["identity_type"] = 4
+//	//	saveM["createtime"] = time.Now().Format(util.Date_Full_Layout)
+//	//	if code := redis.GetStr("qyxy_id", a); code != "" {
+//	//		saveM["name_id"] = code
+//	//		if util.ObjToString(tmp["agencytel"]) != "" {
+//	//			q := make(map[string]interface{})
+//	//			q["name_id"] = code
+//	//			q["identity_type"] = 4 // 100
+//	//			q["contact_tel"] = util.ObjToString(tmp["agencytel"])
+//	//			if util.ObjToString(tmp["agencyperson"]) != "" {
+//	//				q["contact_name"] = util.ObjToString(tmp["agencyperson"])
+//	//			}
+//	//			cinfo := MysqlTool.FindOne("dws_f_ent_contact", q, "", "")
+//	//			if cinfo != nil && len(*cinfo) > 0 {
+//	//				saveM["contact_id"] = (*cinfo)["id"]
+//	//				saveRelationPool <- saveM
+//	//			}
+//	//		}
+//	//	}
+//	//}
+//
+//	for _, item := range tmp["list"].([]interface{}) {
+//		item1 := item.(map[string]interface{})
+//		sw := util.ObjToString(item1["s_winner"])
+//		if !strings.Contains(sw, ",") {
+//			if code := redis.GetStr("qyxy_id", sw); code != "" {
+//				saveM := make(map[string]interface{})
+//				saveM["projectid"] = pid
+//				saveM["infoid"] = lid
+//				saveM["identity_type"] = 2
+//				saveM["createtime"] = time.Now().Format(util.Date_Full_Layout)
+//				saveM["name_id"] = code
+//				if util.ObjToString(item1["winnertel"]) != "" {
+//					q := make(map[string]interface{})
+//					q["name_id"] = code
+//					q["identity_type"] = 2 // 010
+//					q["contact_tel"] = util.ObjToString(item1["winnertel"])
+//					if util.ObjToString(item1["winnerperson"]) != "" {
+//						q["contact_name"] = util.ObjToString(item1["winnerperson"])
+//					}
+//					cinfo := MysqlTool.FindOne("dws_f_ent_contact", q, "", "")
+//					if cinfo != nil && len(*cinfo) > 0 {
+//						saveM["contact_id"] = (*cinfo)["id"]
+//						saveRelationPool <- saveM
+//					}
+//				}
+//			}
+//		}
+//	}
+//}
+//
+//func taskRelation2(tmp map[string]interface{}) {
+//
+//	pid := mongodb.BsonIdToSId(tmp["_id"])
+//	if tmp["ids"] == nil {
+//		log.Info("taskRelation ids err", zap.Any("id", pid))
+//		return
+//	}
+//	info := MysqlTool.Find("dws_f_bpmc_relation", bson.M{"projectid": pid}, "", "", -1, -1)
+//	if len(*info) > 0 {
+//
+//	} else {
+//		ids := util.ObjArrToStringArr(tmp["ids"].([]interface{}))
+//		lid := ids[len(ids)-1]
+//
+//		if b := util.ObjToString(tmp["buyer"]); b != "" {
+//			saveM := make(map[string]interface{})
+//			for _, f := range RelationField {
+//				if f == "projectid" {
+//					saveM[f] = pid
+//				} else if f == "infoid" {
+//					saveM[f] = lid
+//				} else if f == "name_id" {
+//					if code := redis.GetStr("qyxy_id", b); code != "" {
+//						saveM[f] = code
+//						if util.ObjToString(tmp["buyertel"]) != "" {
+//							q := make(map[string]interface{})
+//							q["name_id"] = code
+//							q["identity_type"] = 1
+//							q["contact_tel"] = util.ObjToString(tmp["buyertel"])
+//							if util.ObjToString(tmp["buyerperson"]) != "" {
+//								q["contact_name"] = util.ObjToString(tmp["buyerperson"])
+//							}
+//							cinfo := MysqlTool.FindOne("dws_f_ent_contact", q, "", "")
+//							if cinfo != nil && len(*cinfo) > 0 {
+//								saveM["contact_id"] = (*cinfo)["id"]
+//							}
+//						}
+//					}
+//				} else if f == "identity_type" {
+//					saveM[f] = 1 // 001
+//				} else if f == "createtime" {
+//					saveM[f] = time.Now().Format(util.Date_Full_Layout)
+//				}
+//			}
+//			saveRelationPool <- saveM
+//		}
+//
+//		if a := util.ObjToString(tmp["agency"]); a != "" {
+//			saveM := make(map[string]interface{})
+//			for _, f := range RelationField {
+//				if f == "projectid" {
+//					saveM[f] = pid
+//				} else if f == "infoid" {
+//					saveM[f] = lid
+//				} else if f == "name_id" {
+//					if code := redis.GetStr("qyxy_id", a); code != "" {
+//						saveM[f] = code
+//						if util.ObjToString(tmp["buyertel"]) != "" {
+//							q := make(map[string]interface{})
+//							q["name_id"] = code
+//							q["identity_type"] = 4
+//							q["contact_tel"] = util.ObjToString(tmp["agencytel"])
+//							if util.ObjToString(tmp["agencyperson"]) != "" {
+//								q["contact_name"] = util.ObjToString(tmp["agencyperson"])
+//							}
+//							cinfo := MysqlTool.FindOne("dws_f_ent_contact", q, "", "")
+//							if cinfo != nil && len(*cinfo) > 0 {
+//								saveM["contact_id"] = (*cinfo)["id"]
+//							}
+//						}
+//					}
+//				} else if f == "identity_type" {
+//					saveM[f] = 4 // 100
+//				} else if f == "createtime" {
+//					saveM[f] = time.Now().Format(util.Date_Full_Layout)
+//				}
+//			}
+//			saveRelationPool <- saveM
+//		}
+//
+//		warr := strings.Split(util.ObjToString(tmp["s_winner"]), ",")
+//		if BinarySearch(warr, util.ObjToString(tmp["winner"])) == -1 {
+//			warr = append(warr, util.ObjToString(tmp["winner"]))
+//		}
+//		for _, ws := range warr {
+//			saveM := make(map[string]interface{})
+//			for _, f := range RelationField {
+//				if f == "projectid" {
+//					saveM[f] = pid
+//				} else if f == "infoid" {
+//					saveM[f] = lid
+//				} else if f == "name_id" {
+//					if code := redis.GetStr("qyxy_id", ws); code != "" {
+//						saveM[f] = code
+//						if util.ObjToString(tmp["buyertel"]) != "" {
+//							q := make(map[string]interface{})
+//							q["name_id"] = code
+//							q["identity_type"] = 2
+//							q["contact_tel"] = util.ObjToString(tmp["winnertel"])
+//							if util.ObjToString(tmp["winnerperson"]) != "" {
+//								q["contact_name"] = util.ObjToString(tmp["winnerperson"])
+//							}
+//							cinfo := MysqlTool.FindOne("dws_f_ent_contact", q, "", "")
+//							if cinfo != nil && len(*cinfo) > 0 {
+//								saveM["contact_id"] = (*cinfo)["id"]
+//							}
+//						}
+//					}
+//				} else if f == "identity_type" {
+//					saveM[f] = 2 // 010
+//				} else if f == "createtime" {
+//					saveM[f] = time.Now().Format(util.Date_Full_Layout)
+//				}
+//			}
+//			saveRelationPool <- saveM
+//		}
+//	}
+//}
+//
+//
+//
+//
+//
+//

data_mgo_to_tidb/save.go

@@ -1,438 +0,0 @@
-package main
-
-import "time"
-
-func RunSaveService() {
-	go SaveFunc()
-	go SaveExpandFunc()
-	go SaveDetailFunc()
-	go SaveAttrFunc()
-	go SaveIntentFunc()
-	go SaveBidderFunc()
-	go SaveGoodsFunc()
-	//go saveErrMethod()
-}
-
-func SaveFunc() {
-	arru := make([]map[string]interface{}, saveSize)
-	indexu := 0
-	for {
-		select {
-		case v := <-saveBasePool:
-			arru[indexu] = v
-			indexu++
-			if indexu == saveSize {
-				saveBaseSp <- true
-				go func(arru []map[string]interface{}) {
-					defer func() {
-						<-saveBaseSp
-					}()
-					MysqlTool.InsertBulk("dwd_f_bid_baseinfo", BaseField, arru...)
-				}(arru)
-				arru = make([]map[string]interface{}, saveSize)
-				indexu = 0
-			}
-		case <-time.After(1000 * time.Millisecond):
-			if indexu > 0 {
-				saveBaseSp <- true
-				go func(arru []map[string]interface{}) {
-					defer func() {
-						<-saveBaseSp
-					}()
-					MysqlTool.InsertBulk("dwd_f_bid_baseinfo", BaseField, arru...)
-				}(arru[:indexu])
-				arru = make([]map[string]interface{}, saveSize)
-				indexu = 0
-			}
-		}
-	}
-}
-
-func SaveExpandFunc() {
-	arru := make([]map[string]interface{}, saveSize)
-	indexu := 0
-	for {
-		select {
-		case v := <-saveExpandPool:
-			arru[indexu] = v
-			indexu++
-			if indexu == saveSize {
-				saveExpandSp <- true
-				go func(arru []map[string]interface{}) {
-					defer func() {
-						<-saveExpandSp
-					}()
-					MysqlTool.InsertBulk("dwd_f_bid_expand_baseinfo", ExpandField, arru...)
-				}(arru)
-				arru = make([]map[string]interface{}, saveSize)
-				indexu = 0
-			}
-		case <-time.After(1000 * time.Millisecond):
-			if indexu > 0 {
-				saveExpandSp <- true
-				go func(arru []map[string]interface{}) {
-					defer func() {
-						<-saveExpandSp
-					}()
-					MysqlTool.InsertBulk("dwd_f_bid_expand_baseinfo", ExpandField, arru...)
-				}(arru[:indexu])
-				arru = make([]map[string]interface{}, saveSize)
-				indexu = 0
-			}
-		}
-	}
-}
-
-func SaveDetailFunc() {
-	arru := make([]map[string]interface{}, saveSize)
-	indexu := 0
-	for {
-		select {
-		case v := <-saveDetailPool:
-			arru[indexu] = v
-			indexu++
-			if indexu == saveSize {
-				saveDetailSp <- true
-				go func(arru []map[string]interface{}) {
-					defer func() {
-						<-saveDetailSp
-					}()
-					MysqlTool.InsertBulk("dwd_f_bid_detail", DetailField, arru...)
-				}(arru)
-				arru = make([]map[string]interface{}, saveSize)
-				indexu = 0
-			}
-		case <-time.After(1000 * time.Millisecond):
-			if indexu > 0 {
-				saveDetailSp <- true
-				go func(arru []map[string]interface{}) {
-					defer func() {
-						<-saveDetailSp
-					}()
-					MysqlTool.InsertBulk("dwd_f_bid_detail", DetailField, arru...)
-				}(arru[:indexu])
-				arru = make([]map[string]interface{}, saveSize)
-				indexu = 0
-			}
-		}
-	}
-}
-
-func SaveAttrFunc() {
-	arru := make([]map[string]interface{}, saveSize)
-	indexu := 0
-	for {
-		select {
-		case v := <-saveAttrPool:
-			arru[indexu] = v
-			indexu++
-			if indexu == saveSize {
-				saveAttrSp <- true
-				go func(arru []map[string]interface{}) {
-					defer func() {
-						<-saveAttrSp
-					}()
-					MysqlTool.InsertBulk("dwd_f_bid_file_text", AttrField, arru...)
-				}(arru)
-				arru = make([]map[string]interface{}, saveSize)
-				indexu = 0
-			}
-		case <-time.After(1000 * time.Millisecond):
-			if indexu > 0 {
-				saveAttrSp <- true
-				go func(arru []map[string]interface{}) {
-					defer func() {
-						<-saveAttrSp
-					}()
-					MysqlTool.InsertBulk("dwd_f_bid_file_text", AttrField, arru...)
-				}(arru[:indexu])
-				arru = make([]map[string]interface{}, saveSize)
-				indexu = 0
-			}
-		}
-	}
-}
-
-func SaveIntentFunc() {
-	arru := make([]map[string]interface{}, saveSize)
-	indexu := 0
-	for {
-		select {
-		case v := <-saveIntentPool:
-			arru[indexu] = v
-			indexu++
-			if indexu == saveSize {
-				saveIntentSp <- true
-				go func(arru []map[string]interface{}) {
-					defer func() {
-						<-saveIntentSp
-					}()
-					MysqlTool.InsertBulk("dwd_f_bid_intention_baseinfo", IntentField, arru...)
-				}(arru)
-				arru = make([]map[string]interface{}, saveSize)
-				indexu = 0
-			}
-		case <-time.After(1000 * time.Millisecond):
-			if indexu > 0 {
-				saveIntentSp <- true
-				go func(arru []map[string]interface{}) {
-					defer func() {
-						<-saveIntentSp
-					}()
-					MysqlTool.InsertBulk("dwd_f_bid_intention_baseinfo", IntentField, arru...)
-				}(arru[:indexu])
-				arru = make([]map[string]interface{}, saveSize)
-				indexu = 0
-			}
-		}
-	}
-}
-
-func SaveBidderFunc() {
-	arru := make([]map[string]interface{}, saveSize)
-	indexu := 0
-	for {
-		select {
-		case v := <-saveBidderPool:
-			arru[indexu] = v
-			indexu++
-			if indexu == saveSize {
-				saveBidderSp <- true
-				go func(arru []map[string]interface{}) {
-					defer func() {
-						<-saveBidderSp
-					}()
-					MysqlTool.InsertBulk("dwd_f_bid_package_bidder_baseinfo", BidderField, arru...)
-				}(arru)
-				arru = make([]map[string]interface{}, saveSize)
-				indexu = 0
-			}
-		case <-time.After(1000 * time.Millisecond):
-			if indexu > 0 {
-				saveBidderSp <- true
-				go func(arru []map[string]interface{}) {
-					defer func() {
-						<-saveBidderSp
-					}()
-					MysqlTool.InsertBulk("dwd_f_bid_package_bidder_baseinfo", BidderField, arru...)
-				}(arru[:indexu])
-				arru = make([]map[string]interface{}, saveSize)
-				indexu = 0
-			}
-		}
-	}
-}
-
-func SaveGoodsFunc() {
-	arru := make([]map[string]interface{}, saveSize)
-	indexu := 0
-	for {
-		select {
-		case v := <-saveGoodsPool:
-			arru[indexu] = v
-			indexu++
-			if indexu == saveSize {
-				saveGoodsSp <- true
-				go func(arru []map[string]interface{}) {
-					defer func() {
-						<-saveGoodsSp
-					}()
-					MysqlTool.InsertBulk("dwd_f_bid_package_goods_baseinfo", GoodsField, arru...)
-				}(arru)
-				arru = make([]map[string]interface{}, saveSize)
-				indexu = 0
-			}
-		case <-time.After(1000 * time.Millisecond):
-			if indexu > 0 {
-				saveGoodsSp <- true
-				go func(arru []map[string]interface{}) {
-					defer func() {
-						<-saveGoodsSp
-					}()
-					MysqlTool.InsertBulk("dwd_f_bid_package_goods_baseinfo", GoodsField, arru...)
-				}(arru[:indexu])
-				arru = make([]map[string]interface{}, saveSize)
-				indexu = 0
-			}
-		}
-	}
-}
-
-/*
-以下项目相关涉及
-*/
-func SaveProFunc() {
-	arru := make([]map[string]interface{}, saveSize)
-	indexu := 0
-	for {
-		select {
-		case v := <-saveProPool:
-			arru[indexu] = v
-			indexu++
-			if indexu == saveSize {
-				saveProSp <- true
-				go func(arru []map[string]interface{}) {
-					defer func() {
-						<-saveProSp
-					}()
-					MysqlTool.InsertBulk("dws_f_project_baseinfo", ProField, arru...)
-				}(arru)
-				arru = make([]map[string]interface{}, saveSize)
-				indexu = 0
-			}
-		case <-time.After(1000 * time.Millisecond):
-			if indexu > 0 {
-				saveProSp <- true
-				go func(arru []map[string]interface{}) {
-					defer func() {
-						<-saveProSp
-					}()
-					MysqlTool.InsertBulk("dws_f_project_baseinfo", ProField, arru...)
-				}(arru[:indexu])
-				arru = make([]map[string]interface{}, saveSize)
-				indexu = 0
-			}
-		}
-	}
-}
-
-func SaveProbFunc() {
-	arru := make([]map[string]interface{}, saveSize)
-	indexu := 0
-	for {
-		select {
-		case v := <-saveProbPool:
-			arru[indexu] = v
-			indexu++
-			if indexu == saveSize {
-				saveProbSp <- true
-				go func(arru []map[string]interface{}) {
-					defer func() {
-						<-saveProbSp
-					}()
-					MysqlTool.InsertBulk("dws_f_project_business", ProBusField, arru...)
-				}(arru)
-				arru = make([]map[string]interface{}, saveSize)
-				indexu = 0
-			}
-		case <-time.After(1000 * time.Millisecond):
-			if indexu > 0 {
-				saveProbSp <- true
-				go func(arru []map[string]interface{}) {
-					defer func() {
-						<-saveProbSp
-					}()
-					MysqlTool.InsertBulk("dws_f_project_business", ProBusField, arru...)
-				}(arru[:indexu])
-				arru = make([]map[string]interface{}, saveSize)
-				indexu = 0
-			}
-		}
-	}
-}
-
-func SaveProTagFunc() {
-	arru := make([]map[string]interface{}, saveSize)
-	indexu := 0
-	for {
-		select {
-		case v := <-saveProTagPool:
-			arru[indexu] = v
-			indexu++
-			if indexu == saveSize {
-				saveProTagSp <- true
-				go func(arru []map[string]interface{}) {
-					defer func() {
-						<-saveProTagSp
-					}()
-					MysqlTool.InsertBulk("dws_f_project_tags", ProTagsField, arru...)
-				}(arru)
-				arru = make([]map[string]interface{}, saveSize)
-				indexu = 0
-			}
-		case <-time.After(1000 * time.Millisecond):
-			if indexu > 0 {
-				saveProTagSp <- true
-				go func(arru []map[string]interface{}) {
-					defer func() {
-						<-saveProTagSp
-					}()
-					MysqlTool.InsertBulk("dws_f_project_tags", ProTagsField, arru...)
-				}(arru[:indexu])
-				arru = make([]map[string]interface{}, saveSize)
-				indexu = 0
-			}
-		}
-	}
-}
-
-func SaveRelationFunc() {
-	arru := make([]map[string]interface{}, saveSize)
-	indexu := 0
-	for {
-		select {
-		case v := <-saveRelationPool:
-			arru[indexu] = v
-			indexu++
-			if indexu == saveSize {
-				saveRelationSp <- true
-				go func(arru []map[string]interface{}) {
-					defer func() {
-						<-saveRelationSp
-					}()
-					MysqlTool.InsertBulk("dws_f_bpmc_relation", RelationField, arru...)
-				}(arru)
-				arru = make([]map[string]interface{}, saveSize)
-				indexu = 0
-			}
-		case <-time.After(1000 * time.Millisecond):
-			if indexu > 0 {
-				saveRelationSp <- true
-				go func(arru []map[string]interface{}) {
-					defer func() {
-						<-saveRelationSp
-					}()
-					MysqlTool.InsertBulk("dws_f_bpmc_relation", RelationField, arru...)
-				}(arru[:indexu])
-				arru = make([]map[string]interface{}, saveSize)
-				indexu = 0
-			}
-		}
-	}
-}
-
-// 字段错误数据
-func saveErrMethod() {
-	arru := make([]map[string]interface{}, 200)
-	indexu := 0
-	for {
-		select {
-		case v := <-saveErrPool:
-			arru[indexu] = v
-			indexu++
-			if indexu == saveSize {
-				saveErrSp <- true
-				go func(arru []map[string]interface{}) {
-					defer func() {
-						<-saveErrSp
-					}()
-					MongoB.SaveBulk("bidding_mgo_to_tidb_f_err", arru...)
-				}(arru)
-				arru = make([]map[string]interface{}, saveSize)
-				indexu = 0
-			}
-		case <-time.After(1000 * time.Millisecond):
-			if indexu > 0 {
-				saveErrSp <- true
-				go func(arru []map[string]interface{}) {
-					defer func() {
-						<-saveErrSp
-					}()
-					MongoB.SaveBulk("bidding_mgo_to_tidb_f_err", arru...)
-				}(arru[:indexu])
-				arru = make([]map[string]interface{}, saveSize)
-				indexu = 0
-			}
-		}
-	}
-}

data_mgo_to_tidb/test.go

@@ -1,212 +0,0 @@
-package main
-
-import (
-	"fmt"
-	"go.mongodb.org/mongo-driver/bson"
-	"go.uber.org/zap"
-	util "jygit.jydev.jianyu360.cn/data_processing/common_utils"
-	"jygit.jydev.jianyu360.cn/data_processing/common_utils/log"
-	"jygit.jydev.jianyu360.cn/data_processing/common_utils/mongodb"
-	"jygit.jydev.jianyu360.cn/data_processing/common_utils/redis"
-	"sync"
-	"time"
-)
-
-func taskMysql() {
-	pool := make(chan bool, 5) //控制线程数
-	wg := &sync.WaitGroup{}
-
-	finalId := 0
-	lastInfo := MysqlTool.SelectBySql(fmt.Sprintf("SELECT * FROM %s ORDER BY id DESC LIMIT 1", "dws_f_bpmc_relation"))
-	//lastInfo := MysqlTool.SelectBySql(fmt.Sprintf("SELECT id, projectid, infoid, name_id, identity_type+0 FROM %s ORDER BY id DESC LIMIT 1", "dws_f_bpmc_relation_new"))
-	if len(*lastInfo) > 0 {
-		finalId = util.IntAll((*lastInfo)[0]["id"])
-	}
-	log.Info("查询最后id---", zap.Int("finally id: ", finalId))
-	lastid, count := 0, 0
-	for {
-		log.Info("重新查询，lastid---", zap.Int("lastid: ", lastid))
-		q := fmt.Sprintf("SELECT * FROM %s WHERE id > %d ORDER BY id ASC limit 100000", "dws_f_bpmc_relation", lastid)
-		//q := fmt.Sprintf("SELECT id, projectid, infoid, name_id, identity_type+0 FROM %s WHERE id=61771536 ORDER BY id ASC limit 1000000", "dws_f_bpmc_relation")
-		//q := fmt.Sprintf("SELECT id, name, name_id FROM %s WHERE id>%d ORDER BY id ASC limit 1000000", "dws_f_ent_baseinfo", lastid)
-		rows, err := MysqlTool.DB.Query(q)
-		if err != nil {
-			log.Error("mysql query err ", zap.Error(err))
-		}
-		columns, err := rows.Columns()
-		if finalId == lastid {
-			log.Info("----finish-----", zap.Int("count: ", count))
-			break
-		}
-		for rows.Next() {
-			scanArgs := make([]interface{}, len(columns))
-			values := make([]interface{}, len(columns))
-			ret := make(map[string]interface{})
-			for k := range values {
-				scanArgs[k] = &values[k]
-			}
-			err = rows.Scan(scanArgs...)
-			if err != nil {
-				log.Error("mysql scan err ", zap.Error(err))
-				break
-			}
-			for i, col := range values {
-				if col == nil {
-					ret[columns[i]] = nil
-				} else {
-					switch val := (*scanArgs[i].(*interface{})).(type) {
-					case byte:
-						ret[columns[i]] = val
-						break
-					case []byte:
-						v := string(val)
-						switch v {
-						case "\x00": // 处理数据类型为bit的情况
-							ret[columns[i]] = 0
-						case "\x01": // 处理数据类型为bit的情况
-							ret[columns[i]] = 1
-						default:
-							ret[columns[i]] = v
-							break
-						}
-						break
-					case time.Time:
-						if val.IsZero() {
-							ret[columns[i]] = nil
-						} else {
-							ret[columns[i]] = val.Format("2006-01-02 15:04:05")
-						}
-						break
-					default:
-						ret[columns[i]] = val
-					}
-				}
-			}
-			lastid = util.IntAll(ret["id"])
-			count++
-			if count%20000 == 0 {
-				log.Info("current----", zap.Int("count: ", count), zap.Int("lastid: ", lastid))
-			}
-			pool <- true
-			wg.Add(1)
-			go func(tmp map[string]interface{}) {
-				defer func() {
-					<-pool
-					wg.Done()
-				}()
-				//cid := util.Int64All(tmp["id"])
-				//iid := util.ObjToString(tmp["infoid"])
-				//name_id := util.ObjToString(tmp["name_id"])
-				//identity_type := util.Int64All(tmp["identity_type+0"])
-				//if name_id != "" {
-				//	coll := "bidding"
-				//	if iid > "5a862e7040d2d9bbe88e3b1f" {
-				//		coll = "bidding"
-				//	} else {
-				//		coll = "bidding_back"
-				//	}
-				//	info, _ := MongoB.FindById(coll, iid, bson.M{"agencytel": 1, "agencyperson": 1, "buyertel": 1, "buyerperson": 1, "winnertel": 1, "winnerperson": 1})
-				//	if len(*info) > 0 {
-				//		if identity_type == 1 {
-				//			if util.ObjToString((*info)["buyertel"]) != "" {
-				//				q := make(map[string]interface{})
-				//				q["name_id"] = name_id
-				//				q["identity_type"] = identity_type
-				//				q["contact_tel"] = util.ObjToString((*info)["buyertel"])
-				//				if util.ObjToString((*info)["buyerperson"]) != "" {
-				//					q["contact_name"] = util.ObjToString((*info)["buyerperson"])
-				//				}
-				//				cinfo := MysqlTool.FindOne("dws_f_ent_contact", q, "", "")
-				//				if cinfo != nil && len(*cinfo) > 0 {
-				//					MysqlTool.Update("dws_f_bpmc_relation_new", bson.M{"id": cid}, bson.M{"contact_id": (*cinfo)["id"]})
-				//				}
-				//			}
-				//		} else if identity_type == 2 {
-				//			if util.ObjToString((*info)["winnertel"]) != "" {
-				//				q := make(map[string]interface{})
-				//				q["name_id"] = name_id
-				//				q["identity_type"] = identity_type
-				//				q["contact_tel"] = util.ObjToString((*info)["winnertel"])
-				//				if util.ObjToString((*info)["winnerperson"]) != "" {
-				//					q["contact_name"] = util.ObjToString((*info)["winnerperson"])
-				//				}
-				//				cinfo := MysqlTool.FindOne("dws_f_ent_contact", q, "", "")
-				//				if cinfo != nil && len(*cinfo) > 0 {
-				//					MysqlTool.Update("dws_f_bpmc_relation_new", bson.M{"id": cid}, bson.M{"contact_id": (*cinfo)["id"]})
-				//				}
-				//			}
-				//		} else if identity_type == 4 {
-				//			if util.ObjToString((*info)["agencytel"]) != "" {
-				//				q := make(map[string]interface{})
-				//				q["name_id"] = name_id
-				//				q["identity_type"] = identity_type
-				//				q["contact_tel"] = util.ObjToString((*info)["agencytel"])
-				//				if util.ObjToString((*info)["agencyperson"]) != "" {
-				//					q["contact_name"] = util.ObjToString((*info)["agencyperson"])
-				//				}
-				//				cinfo := MysqlTool.FindOne("dws_f_ent_contact", q, "", "")
-				//				if cinfo != nil && len(*cinfo) > 0 {
-				//					MysqlTool.Update("dws_f_bpmc_relation_new", bson.M{"id": cid}, bson.M{"contact_id": (*cinfo)["id"]})
-				//				}
-				//			}
-				//		}
-				//	}
-				//}
-
-				//redis.PutCKV("qyxy_id", util.ObjToString(tmp["name"]), util.ObjToString(tmp["name_id"]))
-
-				saveM := make(map[string]interface{})
-				if util.ObjToString(tmp["name_id"]) != "" {
-					saveM["name_id"] = util.ObjToString(tmp["name_id"])
-				} else {
-					return
-				}
-				if util.ObjToString(tmp["contact_id"]) != "" {
-					saveM["contact_id"] = util.IntAll(tmp["contact_id"])
-				} else {
-					return
-				}
-				saveM["projectid"] = util.ObjToString(tmp["projectid"])
-				saveM["infoid"] = util.ObjToString(tmp["infoid"])
-				saveM["identity_type"] = tmp["identity_type"]
-				saveRelationPool <- saveM
-			}(ret)
-			ret = make(map[string]interface{})
-		}
-		_ = rows.Close()
-		wg.Wait()
-	}
-}
-
-func taskMgo() {
-	sess := MongoP.GetMgoConn()
-	defer MongoP.DestoryMongoConn(sess)
-
-	ch := make(chan bool, 3)
-	wg := &sync.WaitGroup{}
-	q := bson.M{"_id": bson.M{"$lte": mongodb.StringTOBsonId("63411488911e1eb3459fb87e")}}
-	field := map[string]interface{}{"ids": 1}
-	query := sess.DB("qfw").C("projectset_20220721").Find(q).Select(field).Iter()
-	count := 0
-	for tmp := make(map[string]interface{}); query.Next(tmp); count++ {
-		if count%20000 == 0 {
-			util.Debug("current ---", count, tmp["_id"])
-		}
-		ch <- true
-		wg.Add(1)
-		go func(tmp map[string]interface{}) {
-			defer func() {
-				<-ch
-				wg.Done()
-			}()
-			id := mongodb.BsonIdToSId(tmp["_id"])
-			for _, i := range util.ObjArrToStringArr(tmp["ids"].([]interface{})) {
-				redis.PutCKV("s_id", i, id)
-			}
-		}(tmp)
-		tmp = make(map[string]interface{})
-	}
-	wg.Wait()
-
-	util.Debug("over ---", count)
-}