data_field_dispose/data_tidb/project.go

package main

import (
	"context"
	"data_tidb/config"
	"encoding/json"
	"fmt"
	es "github.com/olivere/elastic/v7"
	"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 taskR() {
	client := Es.GetEsConn()
	defer Es.DestoryEsConn(client)

	wg := &sync.WaitGroup{}

	query := es.NewBoolQuery().
		//Must(es.NewTermsQuery("id", "64e7685255d5406905c94a64"))
		Must(es.NewRangeQuery("comeintime").Gt(Pici))
	//Must(es.NewExistsQuery("yuceendtime"))
	fsc := es.NewFetchSourceContext(true).Include("buyer", "buyertel", "buyerperson", "agency", "agencytel",
		"agencyperson", "s_winner", "winnertel", "winnerperson", "id", "pici")
	util.Debug(Es.Count("bidding", query))
	countDocs := 0
	res, err := client.Scroll().Index("bidding").Query(query).Sort("comeintime", true).FetchSourceContext(fsc).Scroll("5m").Size(2000).Do(context.Background()) //查询一条获取游标
	if err == nil {
		taskInfoA(res, wg, &countDocs)
		scrollId := res.ScrollId
		for {
			searchResult, err := client.Scroll("1m").Index("bidding").Sort("comeintime", true).ScrollId(scrollId).Size(2000).Do(context.TODO()) //查询
			if err != nil {
				util.Debug("Es Search Data Error：", err.Error())
				break
			}
			taskInfoA(searchResult, wg, &countDocs)
			scrollId = searchResult.ScrollId
		}
		wg.Wait()
		util.Debug("over---", countDocs, Pici)
		_, _ = client.ClearScroll().ScrollId(scrollId).Do(context.Background()) //清理游标
	} else {
		util.Debug(err)
	}
}

func taskInfoA(searchResult *es.SearchResult, wg *sync.WaitGroup, countDocs *int) {
	ch := make(chan bool, 2)
	for _, hit := range searchResult.Hits.Hits {
		//开始处理数据
		wg.Add(1)
		ch <- true
		go func(tmpHit *es.SearchHit) {
			defer func() {
				<-ch
				wg.Done()
			}()
			tmp := make(map[string]interface{})
			if json.Unmarshal(tmpHit.Source, &tmp) == nil {
				if t := util.Int64All(tmp["pici"]); t > Pici {
					piciLock.Lock()
					Pici = t
					piciLock.Unlock()
				}
				taskRelation(tmp)
			}
		}(hit)
		*countDocs += 1

		if *countDocs%10000 == 0 {
			util.Debug("Current：", *countDocs)
		}
	}
}

func taskPAdd() {
	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.MongoB.Dbname).C("projectset_20230904").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))
		}
		if t := util.Int64All(tmp["pici"]); t > Pici {
			Pici = t
		}
		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{}) {
	id := util.ObjToString(tmp["id"])
	pid := ""
	if str := redis.GetStr("project_ids", id); str == "" {
		info, _ := MongoP.FindOneByField("projectset_20230904", bson.M{"ids": id}, bson.M{"ids": 1})
		if len(*info) > 0 {
			pid = mongodb.BsonIdToSId((*info)["_id"])
			for _, s := range util.ObjArrToStringArr((*info)["ids"].([]interface{})) {
				redis.PutCKV("project_ids", s, mongodb.BsonIdToSId((*info)["_id"]))
			}
		} else {
			//log.Info("taskRelation pid err", zap.Any("id", id))
			//return
		}
	} else {
		pid = str
	}
	if b := util.ObjToString(tmp["buyer"]); b != "" {
		saveM := make(map[string]interface{})

		saveM["projectid"] = pid
		saveM["infoid"] = id
		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"] = id
		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
				}
			}
		}
	}

	sw := util.ObjToString(tmp["s_winner"])
	if !strings.Contains(sw, ",") {
		if code := redis.GetStr("qyxy_id", sw); code != "" {
			saveM := make(map[string]interface{})
			saveM["projectid"] = pid
			saveM["infoid"] = id
			saveM["identity_type"] = 2
			saveM["createtime"] = time.Now().Format(util.Date_Full_Layout)
			saveM["name_id"] = code
			if util.ObjToString(tmp["winnertel"]) != "" {
				q := make(map[string]interface{})
				q["name_id"] = code
				q["identity_type"] = 2 // 010
				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"]
					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
		}
	}
}

func taskRedisPid() {
	sess := MongoP.GetMgoConn()
	defer MongoP.DestoryMongoConn(sess)

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

	f := bson.M{"ids": 1, "id": 1}
	query := sess.DB(config.Conf.DB.MongoB.Dbname).C("projectset_20230904").Find(nil).Select(f).Iter()
	count := 0
	for tmp := make(map[string]interface{}); query.Next(tmp); count++ {
		if count%20000 == 0 {
			log.Info(fmt.Sprintf("current --- %d", count))
		}
		if t := util.Int64All(tmp["pici"]); t > Pici {
			Pici = t
		}
		ch <- true
		wg.Add(1)
		go func(tmp map[string]interface{}) {
			defer func() {
				<-ch
				wg.Done()
			}()
			pid := mongodb.BsonIdToSId(tmp["_id"])
			for _, i := range tmp["ids"].([]interface{}) {
				redis.PutCKV("project_ids", util.ObjToString(i), pid)
			}
		}(tmp)
		tmp = make(map[string]interface{})
	}
	wg.Wait()
	log.Info(fmt.Sprintf("over --- %d", count))
}