jy-data-extract/udp_winner/main.go

package main

import (
	"encoding/json"
	"fmt"
	"github.com/garyburd/redigo/redis"
	"go.mongodb.org/mongo-driver/bson"
	"go.mongodb.org/mongo-driver/bson/primitive"
	es "gopkg.in/olivere/elastic.v1"
	"log"
	mu "mfw/util"
	"net"
	"qfw/common/src/qfw/util/elastic"
	"qfw/util"
	"regexp"
	"strconv"

	"sort"
	"strings"
	"time"
)

var (
	Config                                = make(map[string]string)
	Fields                                []string
	SourceClient, FClient                 *MongodbSim
	RedisPool                             redis.Pool
	Addrs                                 = make(map[string]interface{}, 0) //省市县
	udpclient                             mu.UdpClient                      //udp对象
	ElasticClientIndex, ElasticClientType string
	Reg_xing                              = regexp.MustCompile(`\*{1,}`)
	Reg_person                            = regexp.MustCompile("[\u4E00-\u9FA5\\s]+")
	Reg_tel                               = regexp.MustCompile(`^[0-9\-\s]*$`)
	EsConn                                *es.Client
	Updport                               int
)

/**
新增
初始化
*/
func init() {
	log.SetFlags(log.Ldate | log.Ltime | log.Lshortfile)
	util.ReadConfig(&Config)
	log.Println(Config)
	Fields = []string{"_id", "contact", "partners", "business_scope", "company_address", "capital",
		"establish_date", "legal_person", "company_type", "district", "city", "province", "area_code", "credit_no",
		"company_name", "history_name", "topscopeclass", "wechat_accounts", "alias", "website", "report_websites"}
	var err error
	pool_size, _ := strconv.Atoi(Config["pool_size"])

	//mongo init
	SourceClient = new(MongodbSim)
	SourceClient.MongodbAddr = Config["mgoinit"]
	SourceClient.Size = pool_size
	//mongodbSim.DbName = "qfw"
	SourceClient.InitPool()
	SourceClientmgoConn := SourceClient.GetMgoConn()
	defer SourceClient.DestoryMongoConn(SourceClientmgoConn)
	FClient = new(MongodbSim)
	FClient.MongodbAddr = Config["mgourl"]
	FClient.Size = pool_size
	FClient.DbName =Config["mgodb_extract_kf"]
	//mongodbSim.DbName = "qfw"
	FClient.InitPool()
	FClientmgoConn := FClient.GetMgoConn()
	defer FClient.DestoryMongoConn(FClientmgoConn)
	//加载省市县代码
	cursor2 := FClientmgoConn.DB(Config["mgodb_extract_kf"]).C("address").Find(bson.M{}).Select(bson.M{"province": 1, "code": 1, "city": 1, "district": 1}).Iter()
	//defer FClient.Connect(cc)
	if cursor2 == nil {
		log.Fatalln(cursor2)
	}
	tmp := make(map[string]interface{})
	for cursor2.Next(&tmp) {
		code := tmp["code"]
		if code != nil && strings.TrimSpace(code.(string)) != "" {
			Addrs[fmt.Sprint(code)] = tmp
		}
	}
	log.Println(len(Addrs))
	//es.NewClient(es.SetURL(addrs...), es.SetMaxRetries(2), es.SetSniff(false))
	//es init
	elastic.InitElasticSize(Config["elasticsearch"], 50)
	EsConn = elastic.GetEsConn()
	defer elastic.DestoryEsConn(EsConn)
	//redis
	RedisPool = redis.Pool{
		MaxIdle:     50,
		IdleTimeout: 10 * time.Second,
		Dial: func() (redis.Conn, error) {
			conn, e := redis.Dial("tcp", Config["redis"])
			if e != nil {
				return conn, e
			}
			_, err = conn.Do("SELECT", "1")
			if err != nil {
				return nil, err
			}
			return conn, nil
		}}
	c := RedisPool.Get()
	if _, err := c.Do("PING"); err != nil {
		log.Fatalln("redis err:", err)
	}
	c.Close()
}

func main() {
	//udp
	updport := Config["udpport"]
	Updport, _ = strconv.Atoi(Config["port"])
	udpclient = mu.UdpClient{Local: updport, BufSize: 1024}
	udpclient.Listen(processUdpMsg)
	log.Println("Udp服务监听", updport)
	log.Println("发送端口port：", Updport)
	go TimedTask() //定时任务
	c := make(chan int, 1)
	<-c

}

func processUdpMsg(act byte, data []byte, ra *net.UDPAddr) {
	log.Println(act, string(data), ra)
	switch act {
	case mu.OP_TYPE_DATA: //上个节点的数据
		//从表中开始处理生成企业数据
		tmp := new(map[string]interface{})
		err := json.Unmarshal(data, &tmp)
		if err != nil {
			log.Println("err:", err)
			udpclient.WriteUdp([]byte("err:"+err.Error()), mu.OP_NOOP, ra)
			return
		} else if tmp != nil {
			if key,ok := (*tmp)["key"].(string);ok{
				udpclient.WriteUdp([]byte(key), mu.OP_NOOP, ra)
			}else {
				udpclient.WriteUdp([]byte("udpok"), mu.OP_NOOP, ra)
			}
			go task(tmp)
		}
	case mu.OP_NOOP: //下个节点回应
		log.Println("发送成功", string(data))
	}
}
func task(mapinfo *map[string]interface{}) {
	defer util.Catch()
	gtid, lteid := util.ObjToString((*mapinfo)["gtid"]), util.ObjToString((*mapinfo)["lteid"])
	if gtid == "" || lteid == "" {
		log.Println(gtid, lteid, "参数错误")
		return
	}
	GId, err := primitive.ObjectIDFromHex(gtid)
	LtId, err2 := primitive.ObjectIDFromHex(lteid)
	if err != nil || err2 != nil {
		log.Println(gtid, lteid, "转换_id错误")
		return
	}
	//udp的id区间查询bidding  中标人 中标联系人 中标联系电话
	// topscopeclass项目类型-industry行业类型&&topscopeclass联系人项目类型
	// (area地区-province省份 city城市-city城市 district区县-district区县)
	// winneraddr-company_address企业地址
	SourceClientcc := SourceClient.GetMgoConn()
	defer SourceClient.DestoryMongoConn(SourceClientcc)
	cursor := SourceClientcc.DB(Config["mgodb_bidding"]).C(Config["mgodb_mgoinit_c"]).Find(bson.M{
		"_id": bson.M{
			"$gte": GId,
			"$lte": LtId,
		},
	}).Select(bson.M{"winner": 1, "winnertel": 1, "winnerperson": 1,
		"topscopeclass": 1, "winneraddr": 1}).Iter()
	if cursor == nil {
		log.Println(cursor)
		return
	}
	overid := gtid
	tmp := map[string]interface{}{}
	for cursor.Next(&tmp) {
		overid = tmp["_id"].(primitive.ObjectID).Hex()
		log.Println(tmp["_id"])
		if tmp["winner"] == nil || tmp["winner"] == "" {
			continue
		}
		//redis查询是否存在
		rdb := RedisPool.Get()
		if reply, err := redis.String(rdb.Do("GET", tmp["winner"])); err != nil {
			//redis不存在存到临时表,定时任务处理
			FClient.DbName = Config["mgodb_extract_kf"]
			if tmpid := FClient.Save("winner_new", tmp) ;tmpid==nil{
				log.Println("FClient.Save err",tmpid)
			}
			//log.Println("get redis id err:定时任务处理", err, tmp)
			if err := rdb.Close(); err != nil {
				log.Println(err)
			}
			continue
		} else {
			//log.Println("redis get :", reply)
			//redis存在
			//log.Println(reply)
			//reply = "5e0316b998a9abaf6535df3d"
			//id, err := primitive.ObjectIDFromHex(reply)
			//if err != nil {
			//	log.Println("get redis id  Hex err:", err, tmp)
			//	if err := rdb.Close(); err != nil {
			//		log.Println(err)
			//	}
			//	continue
			//}
			if err := rdb.Close(); err != nil {
				log.Println(err)
			}
			//拿到合并后的qyk
			FClient.DbName = Config["mgodb_extract_kf"]
			oldTmp := FClient.FindById(Config["mgo_qyk_c"], reply)
			if oldTmp == nil{
				log.Println("redis id 不存在")
				continue
			}
			//err = FClient.Database(Config["mgodb_extract_kf"]).Collection(Config["mgo_qyk_c"]).
			//	FindOne(context.TODO(), bson.M{"_id": id}).Decode(&oldTmp)
			//if err != nil {
			//	log.Println("qyk id err:", err, id)
			//	continue
			//}
			//比较合并
			//行业类型
			tmpTopscopeclass := []string{}
			tmpTopscopeclassMap := make(map[string]bool)

			if oldTmp["industry"] == nil {
				//log.Println(reflect.ValueOf(tmp["topscopeclass"]))
				if v, ok := tmp["topscopeclass"].(primitive.A); ok {
					for _, vv := range v {
						if vvv, ok := vv.(string); ok && len(vvv) > 1 {
							tmpTopscopeclassMap[vvv[:len(vvv)-1]] = true
						}
					}
					for k := range tmpTopscopeclassMap {
						tmpTopscopeclass = append(tmpTopscopeclass, k)
					}
				}
			} else {
				if v, ok := oldTmp["industry"].(primitive.A); ok {
					for _, vv := range v {
						if vvv, ok := vv.(string); ok {
							tmpTopscopeclassMap[vvv] = true
						}
					}
				}
				if v, ok := tmp["topscopeclass"].(primitive.A); ok {
					for _, vv := range v {
						if vvv, ok := vv.(string); ok && len(vvv) > 1 {
							tmpTopscopeclassMap[vvv[:len(vvv)-1]] = true
						}
					}
					for k := range tmpTopscopeclassMap {
						tmpTopscopeclass = append(tmpTopscopeclass, k)
					}
				}
			}
			sort.Strings(tmpTopscopeclass)
			oldTmp["industry"] = tmpTopscopeclass
			esId := oldTmp["_id"].(primitive.ObjectID).Hex()
			//更新行业类型
			if tmp["winnerperson"] == nil || tmp["winnerperson"] == "" || Reg_xing.MatchString(util.ObjToString(tmp["winnerperson"])) {
				oldTmp["updatatime"] = time.Now().Unix()
				//mongo更新
				FClient.DbName =Config["mgodb_extract_kf"]
				if !FClient.UpdateById(Config["mgo_qyk_c"],esId,bson.M{"$set": oldTmp}){
					log.Println("mongo更新err",esId)
				}
				//if _, err := FClient.Database(Config["mgodb_extract_kf"]).Collection(Config["mgo_qyk_c"]).
				//	UpdateOne(context.TODO(), bson.M{"_id": oldTmp["_id"]}, bson.M{"$set": oldTmp}); err != nil {
				//	log.Println("mongo更新err:", err)
				//}
				//es更新
				delete(oldTmp, "_id")
				//esConn := elastic.GetEsConn()
				//defer elastic.DestoryEsConn(esConn)
				if _, err := EsConn.Update().Index(Config["elasticsearch_index"]).Type(Config["elasticsearch_type"]).Id(esId).Doc(oldTmp).Refresh(true).Do(); err != nil {
					log.Println("update es err:", err)
				}
				//log.Println( err2,err3)
				continue
			}
			//联系方式合并
			var tmpperson, winnertel string
			tmpperson = tmp["winnerperson"].(string)
			if tmp["winnertel"] == nil || tmp["winnertel"] == "" {
				winnertel = ""
			} else {
				if Reg_xing.MatchString(util.ObjToString(tmp["winnertel"])) || !Reg_tel.MatchString(util.ObjToString(tmp["winnertel"])) {
					winnertel = ""
				} else {
					winnertel = util.ObjToString(tmp["winnertel"])
				}
			}
			contactMaps := make([]interface{}, 0)
			if oldTmp["contact"] == nil {
				tmpContact := make(map[string]interface{})
				tmpContact["contact_person"] = tmpperson
				tmpContact["contact_type"] = "项目联系人"
				tmpContact["phone"] = winnertel
				tmpContact["topscopeclass"] = strings.Join(tmpTopscopeclass, ";")
				tmpContact["updatetime"] = time.Now().Unix()
				contactMaps = append(contactMaps, tmpContact)
			} else {
				//对比前四项,相等丢弃
				if v, ok := oldTmp["contact"].(primitive.A); ok {
					var isNotUpdate bool
					for _, vv := range v {
						if vvv, ok := vv.(map[string]interface{}); ok {
							if vvv["contact_person"] == tmpperson && vvv["contact_type"] == "项目联系人" &&
								vvv["phone"] == winnertel && vvv["topscopeclass"] == strings.Join(tmpTopscopeclass, ";") {
								isNotUpdate = true
								vvv["updatetime"] = time.Now().Unix()
							}
							contactMaps = append(contactMaps, vvv)
						}
					}
					if !isNotUpdate {
						vvv := make(map[string]interface{})
						vvv["contact_person"] = tmp["winnerperson"]
						vvv["contact_type"] = "项目联系人"
						vvv["phone"] = winnertel
						vvv["topscopeclass"] = strings.Join(tmpTopscopeclass, ";")
						vvv["updatetime"] = time.Now().Unix()
						contactMaps = append(contactMaps, vvv)
					}
				}
			}
			oldTmp["contact"] = contactMaps
			//mongo更新
			oldTmp["updatatime"] = time.Now().Unix()
			FClient.DbName=Config["mgodb_extract_kf"]
			if !FClient.UpdateById(Config["mgo_qyk_c"],esId,bson.M{"$set": oldTmp}){
				log.Println("mongo更新 err",esId,oldTmp)
			}
			//if _, err := FClient.Database(Config["mgodb_extract_kf"]).Collection(Config["mgo_qyk_c"]).
			//	UpdateOne(context.TODO(), bson.M{"_id": oldTmp["_id"]}, bson.M{"$set": oldTmp}); err != nil {
			//	log.Println("mongo更新 err :", err)
			//}
			//es更新
			delete(oldTmp, "_id")
			//esConn := elastic.GetEsConn()
			//defer elastic.DestoryEsConn(esConn)
			if _, err := EsConn.Update().Index(Config["elasticsearch_index"]).Type(Config["elasticsearch_type"]).Id(esId).Doc(oldTmp).Refresh(true).Do(); err != nil {
				log.Println("EsConn err :", err)
			}
			//log.Println( err2,err3)
		}
	}
	//defer cursor.Close(context.Background())
	//log.Println("合并执行完成", gtid, lteid, overid)
	//if overid != lteid {
	//	by, _ := json.Marshal(map[string]interface{}{
	//		"gtid":  overid,
	//		"lteid": lteid,
	//		"stype": "",
	//	})
	//	if e := udpclient.WriteUdp(by, mu.OP_TYPE_DATA, &net.UDPAddr{
	//		IP:   net.ParseIP("127.0.0.1"),
	//		Port: Updport,
	//	}); e != nil {
	//		log.Println(e)
	//	}
	//	log.Println("重新发送udp:", string(by))
	//	return
	//}
	log.Println("合并执行完成 ok", gtid, lteid, overid)

}