jy-data-extract/standardata/src/standarbuyer.go

// standarbuyer
package main

import (
	"dbutil/mongo"
	"dbutil/redis"
	"encoding/json"
	log "github.com/donnie4w/go-logger/logger"
	"go.mongodb.org/mongo-driver/bson/primitive"
	"gopkg.in/mgo.v2/bson"
	qu "qfw/util"
	"time"
	"unicode/utf8"
)

//增量处理
func buyerStandarData(db string, query map[string]interface{}) {
	defer qu.Catch()
	sess := MongoFrom.GetMgoConn()
	defer MongoFrom.DestoryMongoConn(sess)

	it := sess.DB(db).C(extractcoll).Find(query).Select(bson.M{"repeat": 1, "buyer": 1, "buyertel": 1,
		"buyerperson": 1, "buyerclass": 1, "topscopeclass": 1}).Sort("_id").Iter()
	index := 0
	for tmp := make(map[string]interface{}); it.Next(&tmp); index++ {
		if qu.IntAll(tmp["repeat"]) > 0 { //重复数据跳过
			continue
		}
		buyer := qu.ObjToString(tmp["buyer"])
		if utf8.RuneCountInString(buyer) < 5 {
			if LenReg.MatchString(buyer)&&utf8.RuneCountInString(buyer)==4 {

			}else {
				continue
			}
		}
		infoid := mongo.BsonTOStringId(tmp["_id"])
		buyerclass := qu.ObjToString(tmp["buyerclass"])
		topscopeclass, _ := tmp["topscopeclass"].(primitive.A)
		entid, _ := redis.GetRedisStr("buyer", buyerbd, buyer)
		ps := []map[string]interface{}{}
		buyerperson := qu.ObjToString(tmp["buyerperson"])
		buyertel := qu.ObjToString(tmp["buyertel"])
		if entid == "" {
			savetoerr := true
			if buyerperson != "" || buyertel != "" {
				v := map[string]interface{}{
					"contact_person": buyerperson,
					"phone":          buyertel,
					"buyerclass":     buyerclass,
					"topscopeclass":  comRepTopscopeclass(topscopeclass),
					"infoid":         infoid,
				}
				ps = append(ps, v)
				data := comHisMegerNewData(buyer, "buyer", ps)
				if data != nil {
					province, city, district := "", "", ""
					province = qu.ObjToString(data["province"])
					if province == "" { //省份为空，buyer优先提取区域信息再company_address
						province, city, district = GetProvinceCityDistrict([]string{buyer}) //先buyer
						if province == "" {                                                 //再address
							if address := qu.ObjToString(data["company_address"]); address != "" {
								province, city, district = GetProvinceCityDistrict([]string{address})
							}
						}
						data["province"] = province
						data["city"] = city
						data["district"] = district
					}
					//提取固话和手机号
					contactArr := []interface{}{}
					contactArr = append(contactArr, v)
					latestFixedPhone, latestMobilePhone, timesFixedPhone, timesMobilePhone := getPhone(contactArr)
					data["latestfixedphone"] = latestFixedPhone
					data["latestmobilephone"] = latestMobilePhone
					data["fixedphone"] = timesFixedPhone
					data["mobilephone"] = timesMobilePhone
					data["institute_type"] = "企业"
					_id := MongoTo.Save(buyerent, data)
					if _id!=nil {
						redis.PutRedis("buyer", buyerbd, buyer, mongo.BsonTOStringId(_id), -1)
					}
					savetoerr = false
				}
			}
			if savetoerr {
				t := MongoTo.FindOne(buyererr, map[string]interface{}{"name": buyer})
				if len(t) < 1 {
					province, city, district := GetProvinceCityDistrict([]string{buyer})
					MongoTo.Save(buyererr, map[string]interface{}{
						"name":       buyer,
						"buyerclass": buyerclass,
						"check":      comMarkdata(buyer, "buyer"),
						"updatetime": time.Now().Unix(),
						"province":   province,
						"city":       city,
						"district":   district,
					})
				}
			}
		} else {
			if buyerperson != "" && buyertel != "" {
				v := map[string]interface{}{
					"contact_person": buyerperson,
					"phone":          buyertel,
					"buyerclass":     buyerclass,
					"topscopeclass":  comRepTopscopeclass(topscopeclass),
					"infoid":         infoid,
				}
				data := buyerMegerBuyerclass(entid, v)
				MongoTo.UpdateById(buyerent, entid,
					map[string]interface{}{
						"$set": data,
						//"$push": map[string]interface{}{"contact": v},
					},
				)
			}
		}
		tmp = map[string]interface{}{}
		if index%10000 == 0 {
			log.Debug("buyer index ", index)
		}
	}
	log.Debug("buyer ok index", index)
}

//历史数据处理
func historybuyer(db, fromcoll string) {
	defer qu.Catch()
	log.Debug("history  start")
	sess := MongoFrom.GetMgoConn()
	defer MongoFrom.Close()
	it := sess.DB(db).C(fromcoll).Find(map[string]interface{}{}).Select(bson.M{"repeat": 1, "buyer": 1, "buyertel": 1, "buyerperson": 1, "buyerclass": 1, "topscopeclass": 1}).Sort("_id").Iter()
	index := 0
	for tmp := make(map[string]interface{}); it.Next(&tmp); index++ {
		if qu.IntAll(tmp["repeat"]) > 0 { //重复数据跳过
			continue
		}
		_id := mongo.BsonTOStringId(tmp["_id"])
		buyerchanbool <- true
		go func(tmp map[string]interface{}) {
			defer func() {
				<-buyerchanbool
			}()
			buyer := qu.ObjToString(tmp["buyer"])
			buyerclass :=""
			if tb ,ok := tmp["buyerclass"].(primitive.A);ok{
				tbn := len(tb)
				if tbn>0{
					buyerclass = qu.ObjToString(tb[tbn-1])
				}
			}else if tbs ,ok :=tmp["buyerclass"].(string);ok{
				buyerclass = tbs

			}
			topscopeclass, _ := tmp["topscopeclass"].(primitive.A)
			if buyer != "" && (utf8.RuneCountInString(buyer) > 4 || (LenReg.MatchString(buyer)&&utf8.RuneCountInString(buyer)==4)) {
				buyerperson := qu.ObjToString(tmp["buyerperson"])
				buyertel := qu.ObjToString(tmp["buyertel"])
				b, _ := redis.ExistRedis("buyer", buyerbd, buyer)
				if b {
					if buyerperson != "" || buyertel != "" {
						strs, _ := redis.GetRedisStr("buyer", buyerbd, buyer)
						ps := []interface{}{}
						err := json.Unmarshal([]byte(strs), &ps)
						if err == nil {
							v := map[string]interface{}{
								"contact_person": buyerperson,
								"phone":          buyertel,
								"buyerclass":     buyerclass,
								"topscopeclass":  comRepTopscopeclass(topscopeclass),
								"infoid":         _id,
							}
							ps = append(ps, v)
							bs, _ := json.Marshal(ps)
							redis.PutRedis("buyer", buyerbd, buyer, bs, -1)
						}
					}
					return
				}
				val := []map[string]interface{}{}
				if buyerperson != "" || buyertel != "" {
					tmp := map[string]interface{}{
						"contact_person": buyerperson,
						"phone":          buyertel,
						"buyerclass":     buyerclass,
						"topscopeclass":  comRepTopscopeclass(topscopeclass),
						"infoid":         _id,
					}
					val = append(val, tmp)
				}
				bs, _ := json.Marshal(val)
				redis.PutRedis("buyer", buyerbd, buyer, bs, -1)
				MongoTo.Save(buyererr, map[string]interface{}{
					"name":       buyer,
					"buyerclass": buyerclass,
					"updatetime": time.Now().Unix(),
				})

			}
		}(tmp)
		tmp = map[string]interface{}{}
		if index%10000 == 0 {
			log.Debug("index", index, _id)
		}
	}
	log.Debug("history  ok  index", index)
	buyerStandarHistory(qu.ObjToString(sysconfig["mgotodb"]))
}

//查询buyererr标准化历史数据
func buyerStandarHistory(db string) {
	defer qu.Catch()
	log.Debug("开始标准化数据--buyer", db)
	sessto := MongoTo.GetMgoConn()
	defer MongoTo.Close()
	it := sessto.DB(db).C(buyererr).Find(map[string]interface{}{}).Iter()
	index := 0
	entnum := 0
	for tmp := make(map[string]interface{}); it.Next(&tmp); index++ {
		err_id := mongo.BsonTOStringId(tmp["_id"])
		name := qu.ObjToString(tmp["name"])
		buyerchanbool <- true
		go func(tmp map[string]interface{}) {
			defer func() {
				<-buyerchanbool
			}()
			strs, err := redis.GetRedisStr("buyer", buyerbd, name)
			if err != nil {
				return
			}
			ps := []map[string]interface{}{}
			err = json.Unmarshal([]byte(strs), &ps)
			if err == nil {
				data := comHisMegerNewData(name, "buyer", ps)
				if data != nil {
					MongoTo.Save(buyerent, data)
					MongoTo.DeleteById(buyererr, err_id)
					entnum++
				} else { //未查询到企业，打标记并存表
					num := comMarkdata(name, "buyer")
					tmp["check"] = num
					MongoTo.UpdateById(buyererr, err_id, map[string]interface{}{"$set": map[string]interface{}{"check": num}})
				}
			} else {
				log.Debug("jsonErr", name, err)
			}
		}(tmp)
		if index%1000 == 0 {
			log.Debug("标准化历史数据--buyer", index, err_id, entnum)
		}
		tmp = map[string]interface{}{}
	}
	log.Debug("标准化数据完成--buyer", index, entnum)
}

//企业数据整合(已有标注信息)
func buyerMegerBuyerclass(id string, ps map[string]interface{}) map[string]interface{} {
	tmp := MongoEnt.FindById(buyerent, id, bson.M{"buyerclass": 1, "contact": 1})
	if len(tmp) < 1 {
		return nil
	}
	data := map[string]interface{}{}
	if buyerclass, ok := tmp["buyerclass"].(primitive.A); ok {
		bn := len(buyerclass)
		if bn > 0 {
			data["buyerclass"] = qu.ObjToString(buyerclass[bn-1])
		}
	} else if sb, ok := tmp["buyerclass"].(string); ok {
		data["buyerclass"] = sb
	}
	data["updatetime"] = time.Now().Unix()

	//contact
	contact :=make(primitive.A,0)
	if tmp["contact"]!=nil {
		contact = tmp["contact"].(primitive.A)
	}
	contact = append(contact, ps)
	//bid_contact
	bid_contacts, contacts := bid_contact(contact)
	if len(bid_contacts) > 0 {
		data["bid_contact"] = bid_contacts
	}
	//sort 200
	contact = contacts
	data["contact"] = contact
	//提取固话和手机号
	latestFixedPhone, latestMobilePhone, timesFixedPhone, timesMobilePhone := getPhone(contact)
	data["latestfixedphone"] = latestFixedPhone
	data["latestmobilephone"] = latestMobilePhone
	data["fixedphone"] = timesFixedPhone
	data["mobilephone"] = timesMobilePhone
	return data
}

func putbuyerreids() {
	sess := MongoTo.GetMgoConn()
	defer MongoTo.Close()
	it := sess.DB("mixdata").C("buyer_enterprise").Find(map[string]interface{}{}).Select(bson.M{"buyer_name": 1}).Iter()
	index := 0
	for tmp := make(map[string]interface{}); it.Next(&tmp); index++ {
		id := mongo.BsonTOStringId(tmp["_id"])
		buyer_name, _ := tmp["buyer_name"].(string)
		redis.PutRedis("buyer", 2, buyer_name, id, -1)
		tmp = map[string]interface{}{}
		if index%1000 == 0 {
			log.Debug(id, index)
		}
	}
}

func putbuyerreids__s() {
	sess := MongoTo.GetMgoConn()
	defer MongoTo.Close()
	it := sess.DB("mixdata").C("buyer_enterprise").Find(map[string]interface{}{}).Select(bson.M{"buyer_name": 1}).Iter()
	index := 0
	for tmp := make(map[string]interface{}); it.Next(&tmp); index++ {
		buyer_name, _ := tmp["buyer_name"].(string)
		ts, _ := MongoTo.Find("buyer_enterprise", map[string]interface{}{"buyer_name": buyer_name}, nil, bson.M{"buyer_name": 1})
		if len(ts) > 1 {
			id := mongo.BsonTOStringId(ts[0]["_id"])
			log.Debug(buyer_name, id)
			MongoTo.DeleteById("buyer_enterprise", id)
		}
		if index%10000 == 0 {
			log.Debug(index)
		}
		tmp = map[string]interface{}{}
	}
}

func modifybuyer() {
	sess := MongoTo.GetMgoConn()
	defer MongoTo.Close()
	it := sess.DB("mixdata").C("buyer_enterprise").Find(map[string]interface{}{"institute_type": "事业单位"}).Iter()
	index := 0
	for tmp := make(map[string]interface{}); it.Next(&tmp); index++ {
		id := mongo.BsonTOStringId(tmp["_id"])
		buyer_name, _ := tmp["buyer_name"].(string)
		bys, _ := MongoTo.Find(
			"buyer_enterprise",
			map[string]interface{}{"buyer_name": buyer_name}, nil,
			bson.M{"institute_type": 1, "buyer_name": 1},
		)
		for _, by := range bys {
			byid := mongo.BsonTOStringId(by["_id"])
			institute_type, _ := by["institute_type"].(string)
			if institute_type == "企业" {
				MongoTo.Save("buyer_tmp", tmp)
				delete(tmp, "_id")
				MongoTo.UpdateById("buyer_enterprise", byid, map[string]interface{}{
					"$set": tmp,
				})
				MongoTo.DeleteById("buyer_enterprise", id)
				log.Debug(id)
			}
		}
		tmp = map[string]interface{}{}
	}
}