jy-data-extract/udp_winner/timedTaskBuyer.go

package main

import (
	"encoding/json"
	"fmt"
	"github.com/garyburd/redigo/redis"
	"go.mongodb.org/mongo-driver/bson/primitive"
	"gopkg.in/mgo.v2/bson"
	"log"
	mu "mfw/util"
	"net"
	"qfw/util"
	"sort"
	"strings"
	"time"
)

func TaskBuyer(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{"buyer": 1, "buyertel": 1, "buyerperson": 1,
		"topscopeclass": 1, "buyeraddr": 1}).Iter()
	if cursor == nil {
		log.Println(cursor)
		return
	}
	overid := gtid
	tmp := map[string]interface{}{}
	//binding tmp
	for cursor.Next(&tmp) {
		overid = tmp["_id"].(primitive.ObjectID).Hex()
		log.Println(tmp["_id"])
		if tmp["buyer"] == nil || tmp["buyer"] == "" {
			continue
		}
		//redis查询是否存在
		rdb := RedisPool.Get()
		if reply, err := redis.String(rdb.Do("GET", tmp["buyer"])); err != nil {
			//redis不存在存到临时表,定时任务处理
			FClient.DbName = Config["mgodb_extract_kf"]
			if tmpid := FClient.Save("buyer_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)

}



//定时任务
//1.存异常表
//2.合并原始库新增
func TimedTaskBuyer() {
	//time.Sleep(time.Hour*70)
	t2 := time.NewTimer(time.Second * 5)
	for range t2.C {
		Fcconn := FClient.GetMgoConn()
		defer FClient.DestoryMongoConn(Fcconn)
		tmpLast := map[string]interface{}{}
		if iter := Fcconn.DB(Config["mgodb_extract_kf"]).C("buyer_new").Find(bson.M{}).Sort("-_id").Limit(1).Iter(); iter != nil {
			if !iter.Next(&tmpLast) {
				//临时表无数据
				log.Println("临时表无数据:")
				t2.Reset(time.Minute * 5)
				continue
			} else {
				log.Println("临时表有数据:", tmpLast)
				fconn := FClient.GetMgoConn()
				defer FClient.DestoryMongoConn(fconn)
				cursor := fconn.DB(Config["mgodb_extract_kf"]).C("buyer_new").Find(bson.M{
					"_id": bson.M{
						"$lte": tmpLast["_id"],
					},
				}).Sort("_id").Iter()
				if cursor == nil {
					log.Println("查询失败")
					t2.Reset(time.Second * 5)
					continue
				}
				//遍历临时表数据，匹配不到原始库存入异常表
				tmp := make(map[string]interface{})
				for cursor.Next(&tmp) {
					tmpId := tmp["_id"].(primitive.ObjectID).Hex()
					//再重新查找redis,存在发udp处理，不存在走新增合并
					rdb := RedisPool.Get()
					if _, err := redis.String(rdb.Do("GET", tmp["buyer"])); err == nil {
						//{"gtid":"57d7ad2f61a0721f152d2ad5","lteid":"5e20968d85a9271abf0ad6c2","stype":""}
						//redis存在发送udp进行处理
						by, _ := json.Marshal(map[string]interface{}{
							"gtid":  tmpId,
							"lteid": tmpId,
							"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)
						}
						//存在的话删除tmp mongo表
						FClient.DbName = Config["mgodb_extract_kf"]
						if DeletedCount := FClient.DeleteById("buyer_new", tmpId); DeletedCount == 0 {
							log.Println("删除临时表err:", DeletedCount)
						}
						if err := rdb.Close(); err != nil {
							log.Println(err)
						}
						continue
					} else {
						if err = rdb.Close(); err != nil {
							log.Println(err)
						}
					}
					//查询redis不存在新增
					FClient.DbName = Config["mgodb_enterprise"]
					//qyxy 企业库 两亿条
					resulttmp := FClient.FindOne(Config["mgodb_enterprise_c"], bson.M{"company_name": tmp["buyer"]})
					if resulttmp["_id"] == nil {
						//log.Println(r)
						//匹配不到原始库,存入异常表删除临时表
						FClient.DbName = Config["mgodb_extract_kf"]
						if saveid := FClient.Save("buyer_err", tmp); saveid == nil {
							log.Println("存入异常表错误", tmp)
						}
						FClient.DbName = Config["mgodb_extract_kf"]
						if deleteNum := FClient.DeleteById("buyer_new", tmpId); deleteNum == 0 {
							log.Println("删除临时表错误", deleteNum)
						}
						continue
					} else {
						//log.Println(123)
						//匹配到原始库,新增 resulttmp
						if resulttmp["credit_no"] != nil {
							if credit_no, ok := resulttmp["credit_no"].(string); ok && strings.TrimSpace(credit_no) != "" &&
								len(strings.TrimSpace(credit_no)) > 8 {
								dataNo := strings.TrimSpace(credit_no)[2:8]
								if Addrs[dataNo] != nil {
									if v, ok := Addrs[dataNo].(map[string]interface{}); ok {
										if resulttmp["province"] == nil || resulttmp["province"] == "" {
											resulttmp["province"] = v["province"]
										}
										resulttmp["city"] = v["city"]
										resulttmp["district"] = v["district"]

									}
								}
							}
						}
						contacts := make([]map[string]interface{}, 0)
						contact := make(map[string]interface{}, 0)
						if resulttmp["legal_person"] != nil {
							contact["contact_person"] = resulttmp["legal_person"] //联系人
						} else {
							contact["contact_person"] = "" //联系人
						}
						contact["contact_type"] = "法定代表人" //法定代表人
						//log.Println(1)
						if resulttmp["annual_reports"] != nil {
							bytes, err := json.Marshal(resulttmp["annual_reports"])
							if err != nil {
								log.Println("annual_reports err:", err)
							}
							phonetmp := make([]map[string]interface{}, 0)
							err = json.Unmarshal(bytes, &phonetmp)
							if err != nil {
								log.Println("Unmarshal err:", err)
							}
							for _, vv := range phonetmp {
								if vv["company_phone"] != nil {
									if vv["company_phone"] == "" {
										continue
									} else {
										contact["phone"] = vv["company_phone"] //联系电话
										break
									}
								} else {
									contact["phone"] = "" //联系电话
								}

							}
						}
						//log.Println(k, contact["phone"], resulttmp["_id"])
						//time.Sleep(10 * time.Second)
						if contact["phone"] == nil {
							contact["phone"] = "" //联系电话
						}
						contact["topscopeclass"] = "企业公示"         //项目类型
						contact["updatetime"] = time.Now().Unix() //更新时间
						contacts = append(contacts, contact)
						resulttmp["contact"] = contacts

						savetmp := make(map[string]interface{}, 0)
						for _, sk := range BuyerFields {
							if sk == "_id" {
								savetmp["tmp"+sk] = resulttmp[sk]
								continue
							} else if sk == "area_code" {
								//行政区划代码
								savetmp[sk] = fmt.Sprint(resulttmp[sk])
								continue
							} else if sk == "report_websites" {
								//网址
								if resulttmp["report_websites"] == nil {
									savetmp["website"] = ""
								} else {
									report_websitesArr := []string{}
									if ppms, ok := resulttmp[sk].(primitive.A); ok {
										for _, v := range ppms {
											if vvv, ok := v.(map[string]interface{}); ok {
												if rv, ok := vvv["website_url"].(string); ok {
													report_websitesArr = append(report_websitesArr, rv)
												}
											}
										}
									}
									sort.Strings(report_websitesArr)
									savetmp["website"] = strings.Join(report_websitesArr, ";")
								}
								continue
							} else if sk == "wechat_accounts" {
								savetmp[sk] = []interface{}{}
								continue
							}else if sk=="buyer_name" {
								if resulttmp["company_name"] == nil {
									savetmp[sk] = ""
								}else {
									savetmp[sk] = resulttmp[sk]
								}
								continue
							}else if sk=="address"{
								if resulttmp["company_address"] == nil {
									savetmp[sk] = ""
								}else {
									savetmp[sk] = resulttmp[sk]
								}
								continue
							}



							if resulttmp[sk] == nil && sk != "history_name" && sk != "wechat_accounts" &&
								sk != "buyer_name" && sk != "address" &&
								sk != "contact" && sk != "report_websites" {
								savetmp[sk] = ""
							} else {
								savetmp[sk] = resulttmp[sk]
							}
						}
						//tmps = append(tmps, savetmp)
						savetmp["updatatime"] = time.Now().Unix()
						//保存mongo
						FClient.DbName = Config["mgodb_extract_kf"]
						saveid := FClient.Save(Config["mgo_qyk_c"], savetmp)
						if saveid != nil {
							//保存redis
							rc := RedisPool.Get()
							var _id string
							if v, ok := saveid.(primitive.ObjectID); ok {
								_id = v.Hex()
							}
							if _, err := rc.Do("SET", savetmp["buyer_name"], _id); err != nil {
								log.Println("save redis err:", tmp["_id"], savetmp["_id"], savetmp["buyer_name"], err)
								if err := rc.Close(); err != nil {
									log.Println(err)
								}
							} else {
								//保存es
								delete(savetmp, "_id")
								if err := rc.Close(); err != nil {
									log.Println(err)
								}

								//esConn := elastic.GetEsConn()
								//defer elastic.DestoryEsConn(esConn)
								if _, err := EsConn.Index().Index(Config["elasticsearch_buyer_index"]).Type(Config["elasticsearch_buyer_type"]).Id(_id).BodyJson(savetmp).Refresh(true).Do(); err != nil {
									log.Println("save es err :", tmp["_id"], savetmp["_id"], err)
								} else {
									//删除临时表
									FClient.DbName = Config["mgodb_extract_kf"]
									if deleteNum := FClient.DeleteById("buyer_new", tmpId); deleteNum == 0 {
										log.Println("删除临时表失败", deleteNum)
									}
								}
							}
						} else {
							log.Println("save mongo err:", saveid, tmp["_id"])
						}
					}
				}
			}
		}
		t2.Reset(time.Minute)
	}
}