fieldproject/fieldproject_com/task.go

package main

import (
	util "app.yhyue.com/data_processing/common_utils"
	"app.yhyue.com/data_processing/common_utils/log"
	"app.yhyue.com/data_processing/common_utils/mongodb"
	"app.yhyue.com/data_processing/common_utils/redis"
	"fieldproject_common/config"
	"fmt"
	uuid "github.com/satori/go.uuid"
	"github.com/spf13/cobra"
	"go.uber.org/zap"
	"strings"
	"sync"
	"time"
	"unicode/utf8"
)

// @Description bidding数据 bid_field
// @Author J 2022/8/30 09:09
func bidding() *cobra.Command {
	cmdClient := &cobra.Command{
		Use:   "bidding",
		Short: "Start processing bidding data",
		Run: func(cmd *cobra.Command, args []string) {
			go updateEsMethod()
			taskBidding()
		},
	}
	//cmdClient.Flags().StringVarP(&cfg, "conf", "c", "", "server config [toml]")
	return cmdClient
}

// @Description 医疗机构数据
// @Author J 2022/8/11 16:50
func institution() *cobra.Command {
	cmdClient := &cobra.Command{
		Use:   "medical_institution",
		Short: "Start processing medical_institutional data",
		Run: func(cmd *cobra.Command, args []string) {
			go SaveEs(config.Conf.DB.Es.IndexM, config.Conf.DB.Es.TypeM)
			taskIterateSql()
		},
	}
	//cmdClient.Flags().StringVarP(&cfg, "conf", "c", "", "server config [toml]")
	return cmdClient
}

// @Description 产品数据
// @Author J 2022/8/11 16:49
func product() *cobra.Command {
	cmdClient := &cobra.Command{
		Use:   "product",
		Short: "Start processing product data",
		Run: func(cmd *cobra.Command, args []string) {
			go SaveFunc("dws_f_product_baseinfo", ProductField)
			//taskIterateSql1()
			taskProduct()
		},
	}
	//cmdClient.Flags().StringVarP(&cfg, "conf", "c", "", "server config [toml]")
	return cmdClient
}

func dealer() *cobra.Command {
	cmdClient := &cobra.Command{
		Use:   "dealer",
		Short: "Start processing dealer data",
		Run: func(cmd *cobra.Command, args []string) {
			go SaveFunc("dwd_f_yl_dealer_baseinfo_new", DealerField)
			//go SaveFuncRc()
			taskDealer()
		},
	}
	return cmdClient
}

func ent() *cobra.Command {
	cmdClient := &cobra.Command{
		Use:   "ent",
		Short: "Start processing ent data",
		Run: func(cmd *cobra.Command, args []string) {
			go SaveFunc("dws_f_ent_pa_baseinfo", EntField)
			taskEnt()
		},
	}
	return cmdClient
}

func register() *cobra.Command {
	cmdClient := &cobra.Command{
		Use:   "register",
		Short: "Start processing register data",
		Run: func(cmd *cobra.Command, args []string) {
			go SaveFunc("dws_f_register_baseinfo", RegField)
			taskRegister()
		},
	}
	return cmdClient
}

func project() *cobra.Command {
	cmdClient := &cobra.Command{
		Use:   "project",
		Short: "Start processing project data",
		Run: func(cmd *cobra.Command, args []string) {
			go SaveFunc("dwd_f_yl_purchasing_baseinfo_new", ProjectField)
			//go SaveFunc1("dwd_f_yl_purchasing_win_baseinfo_new", WinerField)

			InitPoCode()
			InitLvCode()
			redis.InitRedis1("ent_id=172.17.4.189:8379", 6)    // name_id
			redis.InitRedis1("bid_class=172.17.4.189:8379", 7) // class
			taskProject()
		},
	}
	//cmdClient.Flags().StringVarP(&cfg, "conf", "c", "", "server config [toml]")
	return cmdClient
}

func taskBidding() {
	sess := MongoTool.GetMgoConn()
	defer MongoTool.DestoryMongoConn(sess)

	ch := make(chan bool, 3)
	wg := &sync.WaitGroup{}

	//q := map[string]interface{}{"_id": mongodb.StringTOBsonId("5a8d7f4840d2d9bbe8962002")}
	query := sess.DB(config.Conf.DB.Mongo.Dbname).C("bidding").Find(nil).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()
			}()
			if b := util.ObjToString(tmp["bid_field"]); b != "" {
				updateEsPool <- []map[string]interface{}{{
					"_id": mongodb.BsonIdToSId(tmp["_id"]),
				},
					{"bid_field": b},
				}
			}
		}(tmp)
		tmp = make(map[string]interface{})
	}
	wg.Wait()
	log.Info(fmt.Sprintf("over --- %d", count))
}

func taskIterateSql() {
	pool := make(chan bool, 10) //控制线程数
	wg := &sync.WaitGroup{}

	finalId := 0
	lastInfo := MysqlM.SelectBySql(fmt.Sprintf("SELECT id FROM %s ORDER BY id DESC LIMIT 1", "institution_baseinfo"))
	if len(*lastInfo) > 0 {
		finalId = util.IntAll((*lastInfo)[0]["id"])
	}
	log.Info("taskIterateSql---", zap.Int("finally id: ", finalId))
	lastid, count := 0, 0
	for {
		util.Debug("重新查询，lastid---", lastid)
		q := fmt.Sprintf("SELECT * FROM %s WHERE id > %d ORDER BY id ASC limit 100000", "institution_baseinfo", lastid)
		rows, err := MysqlM.DB.Query(q)
		if err != nil {
			log.Error("taskIterateSql---", zap.Error(err))
		}
		columns, err := rows.Columns()
		if finalId == lastid {
			util.Debug("----finish----------", 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("taskIterateSql---", zap.Error(err))
				break
			}
			for i, col := range values {
				if v, ok := col.([]uint8); ok {
					ret[columns[i]] = string(v)
				} else {
					ret[columns[i]] = col
				}
			}
			lastid = util.IntAll(ret["id"])
			count++
			if count%500 == 0 {
				util.Debug("current-------", count, lastid)
			}
			pool <- true
			wg.Add(1)
			go func(tmp map[string]interface{}) {
				defer func() {
					<-pool
					wg.Done()
				}()
				EsSaveCache <- method(tmp)
			}(ret)
			ret = make(map[string]interface{})
		}
		_ = rows.Close()
		wg.Wait()
	}
}

func taskDealer() {
	sess := MongoTool2.GetMgoConn()
	defer MongoTool2.DestoryMongoConn(sess)

	ch := make(chan bool, 3)
	wg := &sync.WaitGroup{}

	query := sess.DB(config.Conf.DB.Mongo2.Dbname).C("zktest_mysql_company_info").Find(nil).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()
			}()
			saveM := make(map[string]interface{})
			record := make(map[string]interface{})
			for _, f := range DealerField {
				if f == "name_id" && util.ObjToString(tmp["name_id"]) == "" {
					name_id := strings.ReplaceAll(uuid.NewV4().String(), "-", "")
					saveM[f] = name_id

					saveM["exists_id"] = 0
					record = map[string]interface{}{"name_id": name_id, "name": tmp["company_name"], "type": 2, "createtime": time.Now().Format(util.Date_Full_Layout)}
				} else if f == "dealer_name" {
					saveM[f] = tmp["company_name"]
				} 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 == "business_model" {
					saveM[f] = tmp["business_type"]
				} else if f == "capital" {
					text := util.ObjToString(tmp["capital"])
					capital := ObjToMoney(text)
					capital = capital / 10000
					if capital != 0 {
						capital, _ = util.FormatFloat(capital, 2)
					} else {
						capital = 0
					}

					saveM[f] = capital
					if capital < 100 {
						saveM["capital_code"] = 1
					} else if capital >= 100 && capital < 500 {
						saveM["capital_code"] = 2
					} else if capital >= 500 && capital < 1000 {
						saveM["capital_code"] = 3
					} else if capital >= 1000 {
						saveM["capital_code"] = 4
					}
				} else if f == "createtime" {
					saveM[f] = time.Now().Format(util.Date_Full_Layout)
				} else if f == "website" {
					tid := util.ObjToString(tmp["company_id"])
					std, _ := MongoTool.FindOneByField("qyxy_std", map[string]interface{}{"_id": tid}, map[string]interface{}{"website_url": 1})
					if std != nil && len(*std) > 0 && len(util.ObjToString((*std)["website_url"])) <= 255 {
						saveM[f] = util.ObjToString((*std)["website_url"])
					}
				} else {
					if tmp[f] != nil {
						saveM[f] = tmp[f]
					}
				}
			}

			if len(record) > 0 {
				saveRcPool <- record
			}
			saveBasePool <- saveM
		}(tmp)
		tmp = make(map[string]interface{})
	}
	wg.Wait()
	log.Info(fmt.Sprintf("over --- %d", count))
}

func taskEnt() {
	sess := MongoTool2.GetMgoConn()
	defer MongoTool2.DestoryMongoConn(sess)

	ch := make(chan bool, 3)
	wg := &sync.WaitGroup{}

	query := sess.DB(config.Conf.DB.Mongo2.Dbname).C("zktest_mysql_company_info").Find(nil).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()
			}()
			saveM := make(map[string]interface{})
			for _, f := range EntField {
				if f == "area_code" {
					if tmp["area"] != nil {
						saveM[f] = AreaCode[util.ObjToString(tmp["area"])]
					}
				} else if f == "issue_date" {
					if util.ObjToString(tmp[f]) != "" {
						saveM[f] = tmp[f]
					}
				} else if f == "sourcetype" {
					saveM[f] = 1
				} else if f == "createtime" || f == "updatetime" {
					saveM[f] = time.Now().Format(util.Date_Full_Layout)
				} else if f == "website" {
					tid := util.ObjToString(tmp["company_id"])
					std, _ := MongoTool.FindOneByField("qyxy_std", map[string]interface{}{"_id": tid}, map[string]interface{}{"website_url": 1})
					if std != nil && len(*std) > 0 && len(util.ObjToString((*std)["website_url"])) <= 255 {
						saveM[f] = util.ObjToString((*std)["website_url"])
					}
				} else {
					if tmp[f] != nil {
						saveM[f] = tmp[f]
					}
				}
			}

			saveBasePool <- saveM
		}(tmp)
		tmp = make(map[string]interface{})
	}
	wg.Wait()
	log.Info(fmt.Sprintf("over --- %d", count))
}

func taskRegister() {
	sess := MongoTool2.GetMgoConn()
	defer MongoTool2.DestoryMongoConn(sess)

	ch := make(chan bool, 3)
	wg := &sync.WaitGroup{}

	query := sess.DB(config.Conf.DB.Mongo2.Dbname).C("nmpa_company").Find(nil).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()
			}()
			saveM := make(map[string]interface{})
			for _, f := range RegField {
				if f == "dealer_id" {
					name := util.ObjToString(tmp["company"])
					info := MysqlM.FindOne("dws_f_dealer_baseinfo", map[string]interface{}{"dealer_name": name}, "name_id", "")
					if info == nil || len(*info) == 0 {
						return
					}
					saveM[f] = (*info)["name_id"]
					saveM["website"] = (*info)["website"]
				} else if f == "company_name" {
					saveM[f] = tmp["company"]
				} else if f == "regnum" {
					if tmp["reg_no"] != nil {
						saveM[f] = tmp["reg_no"]
					}
				} else if f == "scope" {
					if util.ObjToString(tmp["class"]) == "生产型" {
						saveM[f] = tmp["product_range"]
					} else if util.ObjToString(tmp["class"]) == "经营型" {
						saveM[f] = tmp["business_range"]
					}
				} else if f == "type" {
					if util.ObjToString(tmp[f]) == "备案企业" {
						saveM[f] = 2
					} else if util.ObjToString(tmp[f]) == "许可企业" {
						saveM[f] = 1
					}
				} else if f == "approve_depart" {
					saveM[f] = tmp["badw"]
				} else if f == "approve_date" {
					if util.ObjToString(tmp["barq"]) != "" && util.ObjToString(tmp["barq"]) != "null" {
						saveM[f] = tmp["barq"]
					}
				} else if f == "validity_date" {
					if util.ObjToString(tmp["yxqx"]) != "" && util.ObjToString(tmp["yxqx"]) != "null" {
						saveM[f] = tmp["yxqx"]
					}
				} else if f == "type_address" {
					if util.ObjToString(tmp["class"]) == "生产型" {
						saveM[f] = tmp["product_address"]
					} else if util.ObjToString(tmp["class"]) == "经营型" {
						saveM[f] = tmp["business_address"]
					}
				} else if f == "createtime" {
					saveM[f] = time.Now().Format(util.Date_Full_Layout)
				} else {
					if tmp[f] != nil {
						saveM[f] = tmp[f]
					}
				}
			}

			saveBasePool <- saveM
		}(tmp)
		tmp = make(map[string]interface{})
	}
	wg.Wait()
	log.Info(fmt.Sprintf("over --- %d", count))
}

func taskProduct() {
	sess := MongoTool2.GetMgoConn()
	defer MongoTool2.DestoryMongoConn(sess)

	ch := make(chan bool, 3)
	wg := &sync.WaitGroup{}

	query := sess.DB(config.Conf.DB.Mongo2.Dbname).C("zktest_mysql_product_info").Find(nil).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()
			}()
			saveM := make(map[string]interface{})
			for _, f := range ProductField {
				if f == "dealer_id" {
					name := util.ObjToString(tmp["company_name"])
					info := MysqlM.FindOne("dws_f_dealer_baseinfo", map[string]interface{}{"dealer_name": name}, "name_id", "")
					if info == nil || len(*info) == 0 {
						MongoTool2.Save("product_err_record", map[string]interface{}{"infoid": mongodb.BsonIdToSId(tmp["_id"])})
						return
					}
					saveM[f] = (*info)["name_id"]
				} else if f == "make_country" || f == "regist_type" {
					saveM[f] = util.IntAll(tmp[f])
				} else if f == "createtime" {
					saveM[f] = time.Now().Format(util.Date_Full_Layout)
				} else if f == "medical_equipment_class1" {
					saveM[f] = tmp["product_class1"]
				} else if f == "medical_equipment_class2" {
					saveM[f] = tmp["product_class2"]
				} else if f == "medical_equipment_class3" {
					saveM[f] = tmp["product_class3"]
				} else if f == "sdproduct_name" {
					saveM[f] = tmp["product_class4"]
				} else if f == "sdequipment_code" {
					if len(util.ObjToString(tmp["product_code"])) > 7 {
						saveM[f] = util.ObjToString(tmp["product_code"])[:7]
					} else {
						saveM[f] = tmp["product_code"]
					}
				} else {
					if tmp[f] != nil {
						saveM[f] = tmp[f]
					}
				}
			}

			saveBasePool <- saveM
		}(tmp)
		tmp = make(map[string]interface{})
	}
	wg.Wait()
	log.Info(fmt.Sprintf("over --- %d", count))
}

func taskProject() {
	sess := MongoTool1.GetMgoConn()
	defer MongoTool1.DestoryMongoConn(sess)

	ch := make(chan bool, 3)
	wg := &sync.WaitGroup{}

	query := sess.DB(config.Conf.DB.Mongo1.Dbname).C("projectset_medical").Find(nil).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()
			}()
			if tmp["jg_plist"] == nil {
				return
			}
			saveM := make(map[string]interface{})
			infoid := util.ObjToString(tmp["sourceinfoid"])
			saveM["projectid"] = mongodb.BsonIdToSId(tmp["_id"])
			saveM["infoid"] = infoid
			saveM["jyhref"] = fmt.Sprintf(config.Conf.Serve.JyHref, util.CommonEncodeArticle("content", infoid))
			saveM["bidstatus"] = Bidstatus[util.ObjToString(tmp["bidstatus"])]
			saveM["bidstype"] = Bidtype[util.ObjToString(tmp["bidstype"])]
			saveM["updatetime"] = time.Now().Format(util.Date_Full_Layout)
			saveM["createtime"] = time.Now().Format(util.Date_Full_Layout)
			if tmp["budget"] != nil {
				saveM["budget"], _ = util.FormatFloat(util.Float64All(tmp["budget"]), 4)
			}
			if tmp["area"] != nil {
				saveM["area_code"] = AreaCode[util.ObjToString(tmp["area"])]
			}
			if tmp["area"] != nil && tmp["city"] != nil {
				c := util.ObjToString(tmp["area"]) + "," + util.ObjToString(tmp["city"])
				saveM["city_code"] = AreaCode[c]
			}
			if tmp["area"] != nil && tmp["city"] != nil && tmp["district"] != nil {
				c := util.ObjToString(tmp["area"]) + "," + util.ObjToString(tmp["city"]) + "," + util.ObjToString(tmp["district"])
				saveM["district_code"] = AreaCode[c]
			}
			for _, f := range []string{"title", "projectname", "projectcode", "purchasing", "agency", "buyer"} {
				if f == "purchasing" {
					if utf8.RuneCountInString(util.ObjToString(tmp[f])) < 20000 {
						saveM[f] = tmp[f]
					}
				} else {
					if util.ObjToString(tmp[f]) != "" {
						saveM[f] = tmp[f]
					}
				}
			}
			if b := util.ObjToString(tmp["buyer"]); b != "" {
				if eid := redis.GetStr("ent_id", b); eid != "" {
					saveM["buyer_id"] = strings.Split(eid, "_")[0]
					saveM["mi_area_code"] = strings.Split(eid, "_")[1]
					if cd := LvCode[strings.Split(eid, "_")[0]]; cd != "" {
						saveM["mi_level_code"] = cd
					}
				}
			}
			if a := util.ObjToString(tmp["agency"]); a != "" {
				if eid := redis.GetStr("ent_id", a); eid != "" {
					saveM["agency_id"] = strings.Split(eid, "_")[0]
				}
			}
			for _, f := range []string{"bidopentime", "zbtime", "jgtime"} {
				if util.IntAll(tmp[f]) > 0 {
					t := util.Int64All(tmp[f])
					saveM[f] = util.FormatDateByInt64(&t, util.Date_Full_Layout)
				}
			}
			if t := util.Int64All(tmp["jgtime"]); t > 0 {
				y := time.Unix(t, 0).Year()
				m := time.Unix(t, 0).Month().String()
				saveM["year_tags"] = y
				saveM["month_tags"] = Month[m]
				saveM["quarter_tags"] = Quarter[m]
			}
			pname := make(map[string]bool) // 标的物名称 去重
			for _, p := range tmp["jg_plist"].([]interface{}) {
				p1 := p.(map[string]interface{})
				if name := util.ObjToString(p1["itemname"]); name != "" && !pname[name] {
					if utf8.RuneCountInString(name) > 255 {
						continue
					}
					pname[name] = true
					saveM1 := util.DeepCopy(saveM).(map[string]interface{})
					saveM1["itemname"] = name

					if code := redis.GetStr("bid_class", fmt.Sprintf("%s_%s", infoid, name)); code != "" {
						if len(code) > 7 {
							saveM1["sdequipment_code"] = code[:7]
						} else {
							saveM1["sdequipment_code"] = code
						}
						for k, v := range PclassCode[code] {
							if k == "class_1" {
								saveM1["medical_equipment_class1"] = v
							} else if k == "class_2" {
								saveM1["medical_equipment_class2"] = v
							} else if k == "class_3" {
								saveM1["medical_equipment_class3"] = v
							} else if k == "class_4" {
								saveM1["sdproduct_name"] = v
							}
						}
					}
					if p1["brandname"] != nil {
						saveM1["brandname"] = p1["brandname"]
					}
					if p1["specs"] != nil {
						saveM1["specs"] = p1["specs"]
					}
					if p1["model"] != nil && utf8.RuneCountInString(util.ObjToString(p1["model"])) < 200 {
						saveM1["model"] = p1["model"]
					}
					if p1["unitname"] != nil {
						saveM1["unitname"] = p1["unitname"]
					}
					if p1["number"] != nil && util.IntAll(p1["number"]) < 100000 {
						saveM1["number"] = util.IntAll(p1["number"])
					}
					if p1["unitprice"] != nil && util.Float64All(p1["unitprice"]) < 100000000 {
						saveM1["unitprice"], _ = util.FormatFloat(util.Float64All(p1["unitprice"]), 2)
					}
					if p1["totalprice"] != nil {
						saveM1["totalprice"], _ = util.FormatFloat(util.Float64All(p1["totalprice"]), 2)
					}
					saveBasePool <- saveM1
				}
			}
			// 中标信息
			saveW := make(map[string]interface{})
			if w := util.ObjToString(tmp["winner"]); w != "" {
				saveW["winner"] = w
				if eid := redis.GetStr("ent_id", w); eid != "" {
					saveW["winner_id"] = strings.Split(eid, "_")[0]
					saveW["winner_area_code"] = strings.Split(eid, "_")[1]
					if ccode := strings.Split(eid, "_")[2]; ccode != "" {
						saveW["winner_city_code"] = ccode
					}
				}
				if util.ObjToString(tmp["winnertel"]) != "" {
					saveW["contact_tel"] = tmp["winnertel"]
				}
				if util.ObjToString(tmp["winnerperson"]) != "" {
					saveW["contact_name"] = tmp["winnerperson"]
				}

				saveW["projectid"] = mongodb.BsonIdToSId(tmp["_id"])
				saveW["infoid"] = util.ObjToString(tmp["sourceinfoid"])
				if tmp["bidamount"] != nil {
					saveW["bidamount"], _ = util.FormatFloat(util.Float64All(tmp["bidamount"]), 4)
				}
				if util.IntAll(tmp["jgtime"]) > 0 {
					t := util.Int64All(tmp["jgtime"])
					saveW["jgtime"] = util.FormatDateByInt64(&t, util.Date_Full_Layout)
				}
				saveW["is_winner"] = 1
				saveW["updatetime"] = time.Now().Format(util.Date_Full_Layout)
				saveW["createtime"] = time.Now().Format(util.Date_Full_Layout)
				saveBasePool1 <- saveW
			}
			//中标候选
			if tmp["winnerorder"] != nil {
				saveW1 := make(map[string]interface{})
				for _, w := range tmp["winnerorder"].([]interface{}) {
					if util.ObjToString(w) == "" {
						continue
					}
					saveW1["winner"] = w
					if eid := redis.GetStr("ent_id", util.ObjToString(w)); eid != "" {
						saveW1["winner_id"] = strings.Split(eid, "_")[0]
						saveW1["winner_area_code"] = strings.Split(eid, "_")[1]
						if ccode := strings.Split(eid, "_")[2]; ccode != "" {
							saveW1["winner_city_code"] = ccode
						}
					}
					saveW1["projectid"] = mongodb.BsonIdToSId(tmp["_id"])
					saveW1["infoid"] = util.ObjToString(tmp["sourceinfoid"])
					if tmp["bidamount"] != nil {
						saveW1["bidamount"], _ = util.FormatFloat(util.Float64All(tmp["bidamount"]), 4)
					}
					if util.IntAll(tmp["jgtime"]) > 0 {
						t := util.Int64All(tmp["jgtime"])
						saveW["jgtime"] = util.FormatDateByInt64(&t, util.Date_Full_Layout)
					}
					saveW["is_winner"] = 2
					saveW["updatetime"] = time.Now().Format(util.Date_Full_Layout)
					saveW["createtime"] = time.Now().Format(util.Date_Full_Layout)
					saveBasePool1 <- saveW
				}
			}
		}(tmp)
		tmp = make(map[string]interface{})
	}
	wg.Wait()
	log.Info(fmt.Sprintf("over --- %d", count))
}

func method(tmp map[string]interface{}) map[string]interface{} {
	m := make(map[string]interface{})
	for k, v := range config.Conf.DB.Es.FieldM {
		if k == "alias" {
			var arr []string
			info := MysqlM.Find("institution_alias", map[string]interface{}{"company_id": tmp["company_id"]}, "", "", -1, -1)
			for _, m2 := range *info {
				arr = append(arr, util.ObjToString(m2["alias"]))
			}
			if len(arr) > 0 {
				m[k] = strings.Join(arr, ",")
			}
		} else if k == "sdequipment" {
			if util.ObjToString(tmp["mi_type_code"]) == "0208" && strings.Contains(util.ObjToString(tmp["level_code"]), "03") {
				m[k] = "多功能电离子手术治疗机,CO2激光治疗仪,半导体激光治疗仪,准分子激光治疗仪,微波治疗仪,生物共振检测治疗仪,过敏原检测仪,紫外线治疗仪,蓝红光痤疮治疗仪,多功能手术仪,显微镜,手术器材,高频电针、电刀、电灼器,病理切片机,红宝石激光美容仪,光子嫩肤仪,半导体激光脱毛机,中药熏洗机"
			} else if util.ObjToString(tmp["mi_type_code"]) == "0201" && strings.Contains(util.ObjToString(tmp["level_code"]), "03") {
				m[k] = "牙科综合治疗台,石膏模拟切边机,抛光机,氦氖激光器,光敏固化灯,种植机,喷砂机,铸造机,石膏振荡器,干燥箱,全瓷/铸造烤瓷设备,超声波洁牙机,手术器械及器械车,牙科技工装置,包装机,纸塑包装封口机,清洗机"
			} else if util.ObjToString(tmp["mi_type_code"]) == "0203" && strings.Contains(util.ObjToString(tmp["level_code"]), "03") {
				m[k] = "小儿多参数心电监护仪,小儿呼吸机,小儿吸痰器,胆红素测定仪,小儿脉氧仪,小儿雾化治疗仪,复合脉冲磁性治疗仪,经皮给药治疗仪,儿童智能测量仪"
			} else if util.ObjToString(tmp["mi_type_code"]) == "12" && strings.Contains(util.ObjToString(tmp["level_code"]), "03") {
				m[k] = "中心负压吸引设备,中心供氧设备,多参数监护设备,心脏起搏器,心脏除颤器,心脏复苏机,呼吸机,儿童用呼吸机,简易呼吸器,自动洗胃机,心电图机,多功能抢救床,气管插管设备,转运车,快速血糖仪,亚低温治疗仪,冰帽,电子冰毯,微量泵,输液泵,营养输注泵,医用冰箱,血压计,体温计,体重计,空气消毒机"
			} else if util.ObjToString(tmp["mi_type_code"]) == "01" && strings.Contains(util.ObjToString(tmp["level_code"]), "03") {
				m[k] = "中心负压吸引设备,中心供氧设备,多参数监护设备,心脏除颤器,心脏复苏机,呼吸机,简易呼吸器,自动洗胃机,心电图机,多功能抢救床,气管插管设备,转运车,快速血糖仪,亚低温治疗仪,冰帽,电子冰毯,微量泵,输液泵,营养输注泵,医用冰箱,空气消毒机,动态心电监测系统,有创呼吸机,动态血压监护仪,便携式血氧饱和度监护仪,多导睡眠呼吸监测仪,床边肺功能仪,便携式血气分析仪,心肺功能监测仪,胃动力检测仪,胃电治疗仪,腹水超滤仪,腹水浓缩机,人工肝,肝病治疗仪,床单位臭氧消毒机,结肠灌洗治疗仪,经皮肾镜,动态血糖监测仪,胰岛素泵,糖尿病足病诊断箱,眼底镜,空气波压力治疗仪,动态脑电监护仪,颅内压监测仪,脑水肿监测仪,半导体激光治疗仪,钴60放射治疗机,超声聚焦刀,氩氦刀,射频肿瘤治疗仪,微波热疗仪,双筒显微镜,相差显微镜,荧光显微镜,倒置显微镜,骨髓活检装置,流式细胞分析仪,细胞分离机,脑细胞介质分析仪,双光能骨密度仪,脑电超慢涨落分析仪,动脉硬化测试仪,移动式负压吸引器,换药床,乳腺微创真空旋切系统,外碎石设备,骨科牵引床,脊柱牵引床,推拿手法床,石膏床,石膏剪,石膏锯,水温箱,足底静脉泵,激光治疗仪,骨科康复设备,吸引设备,供氧设备,监护设备,呼叫系统,心脏除颤仪,抢救车,换药车,转运床,营养输注泵,防褥疮气垫,血压计,体温计,体重计,移动紫外线灯,负压病房设施,层流病房设施,心脏起搏器,便携式呼吸机,雾化器,床边支气管镜,血液动力学检测仪,电冰毯,电子冰帽,脑电图监测仪,脑功能监测仪,振动排痰器,床单元臭氧消毒机,层流净化系统,holter,主动脉球囊反搏泵,便携式超声诊断仪,食道电生理仪,诊察床,肛管直肠压力测定设备,肛门镜,肛门坐浴熏洗设备,结肠灌洗设备,肛肠综合治疗仪,痔科套扎器,肛肠内腔治疗仪,肛门肌电图,多功能电离子手术治疗机,CO2激光治疗仪,准分子激光治疗仪,微波治疗仪,生物共振检测治疗仪,过敏原检测仪,紫外线治疗仪,蓝红光痤疮治疗仪,多功能手术仪,显微镜,手术器材,高频电针,电刀,电灼器,病理切片机,红宝石激光美容仪,光子嫩肤仪,半导体激光脱毛机,中药熏洗机,母婴监护仪,妇科检查台,计划生育手术床,冲洗车,阴道镜,人流吸引器,超声诊断仪,超高频电波刀,超声聚焦治疗仪,盆腔炎治疗设备,产后康复综合治疗仪,胎心监护仪,妇科检查床,综合产床,新生儿抢救台,婴儿辐射保暖台,婴儿培养箱,电动羊水吸引器,经皮给药治疗仪,胆红素测定仪,产妇电脑综合治疗仪,消毒隔离器械柜,小儿多参数心电监护仪,小儿呼吸机,小儿吸痰器,小儿脉氧仪,小儿雾化治疗仪,复合脉冲磁性治疗仪,儿童智能测量仪,眼科治疗床,裂隙灯,眼压计,角膜曲率计,视力灯箱,客观视力仪,电脑验光仪,全自动电脑视野仪,手术显微镜,眼科AB超声仪,超声乳化治疗仪,眼底荧光造影仪,自动焦度仪,沙眼治疗仪,睫毛电解器,视觉诱发电位仪,耳鼻喉综合治疗台,耳科旋转椅,鼓气电窥耳镜,耳钻,动态喉镜,纤维喉镜,间接喉镜,直接喉镜,支撑喉镜,电子喉镜,鼻咽喉镜,间接鼻咽喉镜,前鼻镜,鼻内镜及手术系统,电测听器,前庭检查仪,眼震电图仪,声阻抗仪,五官科多用显微镜,牙科综合治疗台,石膏模拟切边机,抛光机,氦氖激光器,光敏固化灯,种植机,喷砂机,铸造机,石膏振荡器,干燥箱,全瓷/铸造烤瓷设备,超声波洁牙机,手术器械及器械车,牙科技工装置,包装机,纸塑包装封口机,清洗机,普通X光机,洗片机,透视机,移动式X光机,数字X线摄影,CR,DR,干式激光相机,X线电子计算机断层扫描装置,CT,数字减影血管造影X光机,DSA,数字化胃肠X光机,数据传输系统,LIS,全自动血细胞分析仪,尿液分析仪,尿沉渣工作站,冰点渗透压计,凝血检测仪,血糖测定仪,微量血糖测定仪,血气分析仪,干式生化分析仪,生化分析仪,发光免疫分析仪,全自动酶免免疫分析系统,酶标仪,电泳分析仪,血小板聚集仪,全自动细菌培养系统,生物培养箱,微生物鉴定药敏分析仪,血培养仪,菌落计数器,厌氧菌培养箱,幽门螺旋杆菌检测仪,氨基酸分析系统,荧光定量PCR检测系统,TCT液基细胞学检测仪,HPV-DNA检测系统,心梗三项检测仪,脑钠肽检测仪,二氧化碳培养箱,高温灭菌器,生物安全柜,血液流变仪,普通显微镜,生物显微镜,血沉仪,蛋白电泳仪,特种蛋白仪,电解质分析仪,精子分析系统,血栓弹力分析仪,血型鉴定及配血设备,纯水系统,自动洗板机,分析天平,超声清洗器,振荡器,电热培养箱,恒温水浴箱,医用冷库,医用冷藏柜,超低温冰柜,普通离心机,低速冷冻离心机,高速冷冻离心机,快速血糖仪微量泵,输液泵营养,输注泵,负压病房设施,层流病房设施,有创呼吸机便携式呼吸机,心电图机动态心电监测系统,微量泵输液泵,食道电生理仪床,单元臭氧消毒机,肛管直肠,压力测定设备,超声聚焦治疗仪,盆腔炎治疗设备,裂隙灯眼压计,眼底镜角膜曲率计视力,灯箱客观视力仪,睫毛电解器,鼓气电窥耳镜耳钻,动态喉镜纤维喉镜,间接喉镜直接喉镜,前鼻镜鼻,内镜及手术系统,电测听器前庭检查仪,抛光机氦氖激光器,光敏固化灯种植机,喷砂机铸造机石膏振荡器,干燥箱全瓷/铸造烤瓷设备,普通X光机,透视机移动式X光机,X线电子计算机,数字化胃肠X光机,数据传输系统（LIS）全自动血细胞分析仪,尿沉渣工作站冰点渗透压计,全自动细菌培养系统,生物培养箱,血沉仪蛋白电泳仪,分析天平超声清洗器,干燥箱医用冰箱,心脏复苏机,营养输注泵,血透机,自动腹透机,外碎石设备,核磁共振仪,核磁共振成像系统"
			}
			info1, _ := MongoTool.FindOneByField("20220906shebei_buchong", map[string]interface{}{"company_id": tmp["company_id"]}, map[string]interface{}{"itemname_all": 1})
			if info1 != nil && len(*info1) > 0 {
				if m[k] != nil {
					m[k] = fmt.Sprintf("%s,%s", m[k], (*info1)["itemname_all"])
				} else {
					m[k] = (*info1)["itemname_all"]
				}
			}
			info := MysqlM.FindOne("code_sdleveltypeequip", map[string]interface{}{"code": tmp["sdequipment_code"]}, "", "")
			if info != nil && len(*info) > 0 {
				if m[k] != nil {
					m[k] = fmt.Sprintf("%s,%s", m[k], (*info)["equipment"])
				} else {
					m[k] = (*info)["equipment"]
				}
			}
		} else if k == "area_code" {
			m[k] = tmp[k]
			info := MysqlB.FindOne("code_area", map[string]interface{}{"code": tmp["area_code"]}, "", "")
			if info != nil && len(*info) > 0 {
				m["area"] = (*info)["area"]
				if (*info)["city"] != nil {
					m["city"] = (*info)["city"]
				}
				if (*info)["district"] != nil {
					m["district"] = (*info)["district"]
				}
			}
		} else if k == "mi_type_code" {
			m[k] = string([]byte(util.ObjToString(tmp[k]))[:2])
		} else if tmp[k] == nil || util.ObjToString(tmp[k]) == "" {
			continue
		} else {
			if v == "string" {
				m[k] = tmp[k]
			} else {
				m[k] = util.IntAll(tmp[k])
			}
		}
	}
	return m
}

func taskIterateSql1() {
	pool := make(chan bool, 10) //控制线程数
	wg := &sync.WaitGroup{}

	finalId := 0
	lastInfo := MysqlB.SelectBySql(fmt.Sprintf("SELECT id FROM %s ORDER BY product_id DESC LIMIT 1", "product_baseinfo"))
	if len(*lastInfo) > 0 {
		finalId = util.IntAll((*lastInfo)[0]["product_id"])
	}
	log.Info("taskIterateSql1---", zap.Int("finally id", finalId))
	lastid, count := 0, 0
	for {
		util.Debug("重新查询，lastid---", lastid)
		q := fmt.Sprintf("SELECT * FROM %s WHERE product_id > %d ORDER BY product_id ASC limit 100000", "product_baseinfo", lastid)
		rows, err := MysqlB.DB.Query(q)
		if err != nil {
			log.Error("taskIterateSql1---", zap.Error(err))
		}
		columns, err := rows.Columns()
		if finalId == lastid {
			util.Debug("----finish----------", 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("taskIterateSql1---", zap.Error(err))
				break
			}
			for i, col := range values {
				if v, ok := col.([]uint8); ok {
					ret[columns[i]] = string(v)
				} else {
					ret[columns[i]] = col
				}
			}
			lastid = util.IntAll(ret["product_id"])
			count++
			if count%500 == 0 {
				util.Debug("current-------", count, lastid)
			}
			pool <- true
			wg.Add(1)
			go func(tmp map[string]interface{}) {
				defer func() {
					<-pool
					wg.Done()
				}()
				cid := util.ObjToString(tmp["company_id"])
				info := MysqlM.FindOne("dws_f_dealer_baseinfo", map[string]interface{}{"company_id": cid}, "name_id, dealer_name", "")
				if info == nil || len(*info) == 0 {
					MongoTool2.Save("product_err_record", map[string]interface{}{"product_id": tmp["product_id"]})
					return
				}
				saveM := make(map[string]interface{})
				saveM["dealer_id"] = (*info)["name_id"]
				saveM["company_name"] = (*info)["dealer_name"]
				saveM["sdequipment_code"] = tmp["medical_equipment_code"]
				saveM["createtime"] = time.Now().Format(util.Date_Full_Layout)
				delete(tmp, "medical_equipment_code")
				delete(tmp, "product_id")
				delete(tmp, "comeintime")
				saveBasePool <- tmp
			}(ret)
			ret = make(map[string]interface{})
		}
		_ = rows.Close()
		wg.Wait()
	}
}

func method1(tmp map[string]interface{}) map[string]interface{} {
	m := make(map[string]interface{})
	for k, v := range config.Conf.DB.Es.FieldS {
		if k == "business_model" {
			info := MysqlB.FindOne("company_business_model", map[string]interface{}{"company_id": tmp["company_id"], "company_field_code": "0101"}, "", "")
			if info != nil && len(*info) > 0 {
				m[k] = util.IntAll((*info)["business_model"])
			} else {
				m[k] = 2
			}
		} else if k == "supplier" {
			m[k] = tmp["company_name"]
		} else if k == "productlist" {
			var p = method2(util.ObjToString(tmp["company_id"]))
			if p != nil {
				m[k] = p
			}
		} else if k == "area_code" {
			if tmp[k] != nil {
				m[k] = tmp[k]
				info := MysqlB.FindOne("code_area", map[string]interface{}{"code": tmp["area_code"]}, "", "")
				if info != nil && len(*info) > 0 {
					m["area"] = (*info)["area"]
					if (*info)["city"] != nil {
						m["city"] = (*info)["city"]
					}
					if (*info)["district"] != nil {
						m["district"] = (*info)["district"]
					}
				}
			} else {
				log.Error("area_code", zap.Any("id", tmp["company_id"]))
			}
		} else if tmp[k] == nil || util.ObjToString(tmp[k]) == "" {
			continue
		} else {
			if v == "string" {
				m[k] = tmp[k]
			} else {
				m[k] = util.IntAll(tmp[k])
			}
		}
	}
	return m
}

func method2(cid string) []map[string]interface{} {
	var pmap []map[string]interface{}
	mc := make(map[string]bool) // 记录name 去重
	pinfo1 := MysqlM.Find("product_baseinfo", map[string]interface{}{"company_id": cid}, "", "", -1, -1)
	for _, m2 := range *pinfo1 {
		m := make(map[string]interface{})
		m["name"] = m2["product_name"]
		pmap = append(pmap, m)
		mc[util.ObjToString(m2["product_name"])] = true
	}
	pinfo2, _ := MongoTool.Find("bidding_p_list_0907", map[string]interface{}{"company_id": cid}, nil, nil, false, -1, -1)
	if len(*pinfo2) > 0 {
		for _, m2 := range *pinfo2 {
			key := util.ObjToString(m2["itemname"]) + util.ObjToString(m2["brand"]) + util.ObjToString(m2["model"])
			if mc[key] {
				continue
			} else {
				m := make(map[string]interface{})
				m["name"] = util.ObjToString(m2["itemname"])
				if m2["model"] != nil {
					m["model"] = m2["model"]
				}
				if m2["brand"] != nil {
					m["brand"] = m2["brand"]
				}
				pmap = append(pmap, m)
				mc[key] = true
			}

		}
	}
	return pmap
}

func SaveEs(i, t string) {
	log.Info("SaveEs---", zap.String("i", i), zap.String("t", t))
	arru := make([]map[string]interface{}, 500)
	indexu := 0
	for {
		select {
		case v := <-EsSaveCache:
			arru[indexu] = v
			indexu++
			if indexu == 500 {
				SP <- true
				go func(arru []map[string]interface{}) {
					defer func() {
						<-SP
					}()
					Es.BulkSave(i, t, &arru, false)
				}(arru)
				arru = make([]map[string]interface{}, 500)
				indexu = 0
			}
		case <-time.After(1000 * time.Millisecond):
			if indexu > 0 {
				SP <- true
				go func(arru []map[string]interface{}) {
					defer func() {
						<-SP
					}()
					Es.BulkSave(i, t, &arru, false)
				}(arru[:indexu])
				arru = make([]map[string]interface{}, 500)
				indexu = 0
			}
		}
	}
}

func updateEsMethod() {
	arru := make([][]map[string]interface{}, 200)
	indexu := 0
	for {
		select {
		case v := <-updateEsPool:
			arru[indexu] = v
			indexu++
			if indexu == 200 {
				updateEsSp <- true
				go func(arru [][]map[string]interface{}) {
					defer func() {
						<-updateEsSp
					}()
					Es.UpdateBulk("bidding", "bidding", arru...)
				}(arru)
				arru = make([][]map[string]interface{}, 200)
				indexu = 0
			}
		case <-time.After(1000 * time.Millisecond):
			if indexu > 0 {
				updateEsSp <- true
				go func(arru [][]map[string]interface{}) {
					defer func() {
						<-updateEsSp
					}()
					Es.UpdateBulk("bidding", "bidding", arru...)
				}(arru[:indexu])
				arru = make([][]map[string]interface{}, 200)
				indexu = 0
			}
		}
	}
}