jy-data-extract/udpfilterdup/src/main.go

package main

/**
招标信息判重
**/

import (
	"encoding/json"
	"flag"
	"fmt"
	"log"
	mu "mfw/util"
	"net"
	"os"
	"qfw/util"
	"regexp"
	"strings"
	"sync"
	"time"
)

var (
	Sysconfig    map[string]interface{} //配置文件
	mconf        map[string]interface{} //mongodb配置信息
	mgo          *MongodbSim            //mongodb操作对象
	extract      string
	extract_back string
	udpclient    mu.UdpClient             //udp对象
	nextNode     []map[string]interface{} //下节点数组
	dupdays      = 5                      //初始化判重范围
	DM           *datamap                 //

	//正则筛选相关
	FilterRegTitle   = regexp.MustCompile("^_$")
	FilterRegTitle_0 = regexp.MustCompile("^_$")
	FilterRegTitle_1 = regexp.MustCompile("^_$")
	FilterRegTitle_2 = regexp.MustCompile("^_$")

	isMerger       bool                              //是否合并
	threadNum      int                               //线程数量
	SiteMap        map[string]map[string]interface{} //站点map
	LowHeavy       bool                              //低质量数据判重
	TimingTask     bool                              //是否定时任务
	timingSpanDay  int64                             //时间跨度
	timingPubScope int64                             //发布时间周期
	gtid,lteid,lastid string
	IdType         bool   							 //默认object类型
)

func init() {
	flag.StringVar(&lastid, "id", "", "最后加载id") //以小于等于此id开始加载最近几天的数据
	flag.StringVar(&gtid, "gtid", "", "历史的起始id")

	flag.Parse()
	//172.17.145.163:27080
	util.ReadConfig(&Sysconfig)
	nextNode = util.ObjArrToMapArr(Sysconfig["nextNode"].([]interface{}))
	mconf = Sysconfig["mongodb"].(map[string]interface{})
	mgo = &MongodbSim{
		MongodbAddr: mconf["addr"].(string),
		DbName:      mconf["db"].(string),
		Size:        util.IntAllDef(mconf["pool"], 10),
	}
	mgo.InitPool()

	extract = mconf["extract"].(string)
	extract_back = mconf["extract_back"].(string)

	dupdays = util.IntAllDef(Sysconfig["dupdays"], 3)
	//加载数据
	DM = NewDatamap(dupdays, lastid)
	FilterRegTitle = regexp.MustCompile(util.ObjToString(Sysconfig["specialwords"]))
	FilterRegTitle_0 = regexp.MustCompile(util.ObjToString(Sysconfig["specialtitle_0"]))
	FilterRegTitle_1 = regexp.MustCompile(util.ObjToString(Sysconfig["specialtitle_1"]))
	FilterRegTitle_2 = regexp.MustCompile(util.ObjToString(Sysconfig["specialtitle_2"]))
	isMerger = Sysconfig["isMerger"].(bool)
	threadNum = util.IntAllDef(Sysconfig["threads"], 1)
	LowHeavy = Sysconfig["lowHeavy"].(bool)
	TimingTask = Sysconfig["timingTask"].(bool)
	timingSpanDay = util.Int64All(Sysconfig["timingSpanDay"])
	timingPubScope = util.Int64All(Sysconfig["timingPubScope"])

	//站点配置
	site := mconf["site"].(map[string]interface{})
	SiteMap = make(map[string]map[string]interface{}, 0)
	start := int(time.Now().Unix())
	sess_site := mgo.GetMgoConn()
	defer mgo.DestoryMongoConn(sess_site)
	res_site := sess_site.DB(site["dbname"].(string)).C(site["coll"].(string)).Find(map[string]interface{}{}).Sort("_id").Iter()
	for site_dict := make(map[string]interface{}); res_site.Next(&site_dict); {
		data_map := map[string]interface{}{
			"area":     util.ObjToString(site_dict["area"]),
			"city":     util.ObjToString(site_dict["city"]),
			"district": util.ObjToString(site_dict["district"]),
			"sitetype": util.ObjToString(site_dict["sitetype"]),
			"level":    util.ObjToString(site_dict["level"]),
			"weight":   util.ObjToString(site_dict["weight"]),
		}
		SiteMap[util.ObjToString(site_dict["site"])] = data_map
	}
	log.Printf("new站点加载用时：%d秒,%d个\n", int(time.Now().Unix())-start, len(SiteMap))
}


func main() {
	go checkMapJob()
	updport := Sysconfig["udpport"].(string)
	udpclient = mu.UdpClient{Local: updport, BufSize: 1024}
	udpclient.Listen(processUdpMsg)
	log.Println("Udp服务监听", updport)
	if TimingTask {
		go historyTaskDay()
	}
	time.Sleep(99999 * time.Hour)
}

//测试组人员使用
func mainT() {
	testRepairData11()
	return

	if TimingTask {
		log.Println("新历史任务测试开始")
		go historyTaskDay()
		//go timedTaskDay()
		time.Sleep(99999 * time.Hour)
	} else {
		//IdType = true  //打开id字符串模式
		sid := "4f16936d52c1d9fbf843c60e"
		eid := "6f16936d52c1d9fbf843c60e"
		log.Println("正常判重测试开始")
		log.Println(sid, "---", eid)
		mapinfo := map[string]interface{}{}
		if sid == "" || eid == "" {
			log.Println("sid,eid参数不能为空")
			os.Exit(0)
		}
		mapinfo["gtid"] = sid
		mapinfo["lteid"] = eid
		mapinfo["stop"] = "true"
		task([]byte{}, mapinfo)
		time.Sleep(99999 * time.Hour)
	}
}

func processUdpMsg(act byte, data []byte, ra *net.UDPAddr) {
	fmt.Println("接受的段数据")
	switch act {
	case mu.OP_TYPE_DATA: //上个节点的数据
		//从表中开始处理
		var mapInfo map[string]interface{}
		err := json.Unmarshal(data, &mapInfo)
		log.Println("err:", err, "mapInfo:", mapInfo)
		if err != nil {
			udpclient.WriteUdp([]byte("err:"+err.Error()), mu.OP_NOOP, ra)
		} else if mapInfo != nil {
			taskType := util.ObjToString(mapInfo["stype"])
			if taskType == "normalTask" {
				//判重流程
				go task(data, mapInfo)
			} else {
				//其他
				go task(data, mapInfo)
			}
			key, _ := mapInfo["key"].(string)
			if key == "" {
				key = "udpok"
			}
			udpclient.WriteUdp([]byte(key), mu.OP_NOOP, ra)
		}
	case mu.OP_NOOP: //下个节点回应
		ok := string(data)
		if ok != "" {
			log.Println("ok:", ok)
			udptaskmap.Delete(ok)
		}
	}
}












//开始判重程序
func task(data []byte, mapInfo map[string]interface{}) {
	log.Println("开始数据判重")
	defer util.Catch()
	//区间id
	q := map[string]interface{}{
		"_id": map[string]interface{}{
			"$gt":  StringTOBsonId(mapInfo["gtid"].(string)),
			"$lte": StringTOBsonId(mapInfo["lteid"].(string)),
		},
	}
	if IdType {
		q = map[string]interface{}{
			"_id": map[string]interface{}{
				"$gt":  mapInfo["gtid"].(string),
				"$lte": mapInfo["lteid"].(string),
			},
		}
	}


	log.Println(mgo.DbName, extract, q)
	sess := mgo.GetMgoConn()
	defer mgo.DestoryMongoConn(sess)
	it := sess.DB(mgo.DbName).C(extract).Find(&q).Sort("publishtime").Iter()
	updateExtract := [][]map[string]interface{}{}
	pool := make(chan bool, threadNum)
	wg := &sync.WaitGroup{}
	n, repeateN := 0, 0
	for tmp := make(map[string]interface{}); it.Next(&tmp); n++ {
		if n%10000 == 0 {
			log.Println("current:", n, tmp["_id"], "repeateN:", repeateN)
		}
		source := util.ObjToMap(tmp["jsondata"]) //前置-jsondata判重
		if util.IntAll((*source)["sourcewebsite"]) == 1 {
			repeateN++
			updateExtract = append(updateExtract, []map[string]interface{}{
				map[string]interface{}{
					"_id": tmp["_id"],
				},
				map[string]interface{}{
					"$set": map[string]interface{}{
						"repeat": 1,
						"dataging":0,
						"repeat_reason": "sourcewebsite为1,重复",
					},
				},
			})
			if len(updateExtract) >= 200 {
				mgo.UpSertBulk(extract, updateExtract...)
				updateExtract = [][]map[string]interface{}{}
			}
			tmp = make(map[string]interface{})
			continue
		}

		if util.IntAll(tmp["repeat"]) == 1 || util.IntAll(tmp["repeat"]) == -1||
			util.IntAll(tmp["dataging"]) == 1 {
			if util.IntAll(tmp["repeat"]) == 1 {
				repeateN++
			}
			tmp = make(map[string]interface{})
			continue
		}
		pool <- true
		wg.Add(1)
		go func(tmp map[string]interface{}) {
			defer func() {
				<-pool
				wg.Done()
			}()
			info := NewInfo(tmp)
			if !LowHeavy { //是否进行低质量数据判重
				if invalidData(info.buyer, info.projectname, info.projectcode, info.contractnumber) {
					updateExtract = append(updateExtract, []map[string]interface{}{
						map[string]interface{}{
							"_id": tmp["_id"],
						},
						map[string]interface{}{
							"$set": map[string]interface{}{
								"repeat": -1, //无效数据标签
							},
						},
					})
					if len(updateExtract) >= 200 {
						mgo.UpSertBulk(extract, updateExtract...)
						updateExtract = [][]map[string]interface{}{}
					}
					return
				}
			}
			//正常判重
			b, source, reason := DM.check(info)
			if b { //有重复,生成更新语句，更新抽取和更新招标
				repeateN++
				var updateID = map[string]interface{}{} //记录更新判重的
				updateID["_id"] = StringTOBsonId(info.id)
				if IdType {
					updateID["_id"] = info.id
				}

				updateExtract = append(updateExtract, []map[string]interface{}{//重复数据打标签
					updateID,
					map[string]interface{}{
						"$set": map[string]interface{}{
							"repeat":        1,
							"repeat_reason": reason,
							"repeat_id":     source.id,
						},
					},
				})

				//是否合并-低质量数据不合并
				if isMerger && !strings.Contains(reason,"低质量"){
					newData, update_map ,isReplace := mergeDataFields(source, info)
					if isReplace {//替换-数据池
						fmt.Println("合并更新的id：",source.id)
						//数据池 - 替换
						DM.replacePoolData(newData)
						//mongo更新 - 具体字段 - merge
						mgo.UpdateById(extract,source.id,update_map)
						//发udp  更新索引
						//for _, to := range nextNode {
						//	key := source.id + "-" + source.id + "-" + util.ObjToString(to["stype"])
						//	by, _ := json.Marshal(map[string]interface{}{
						//		"gtid":  source.id,
						//		"lteid": source.id,
						//		"stype": "biddingall",
						//		"key":   key,
						//	})
						//	addr := &net.UDPAddr{
						//		IP:   net.ParseIP(to["addr"].(string)),
						//		Port: util.IntAll(to["port"]),
						//	}
						//	node := &udpNode{by, addr, time.Now().Unix(), 0}
						//	udptaskmap.Store(key, node)
						//	udpclient.WriteUdp(by, mu.OP_TYPE_DATA, addr)
						//}
					}
				}
			}
		}(tmp)

		if len(updateExtract) >= 200 {
			mgo.UpSertBulk(extract, updateExtract...)
			updateExtract = [][]map[string]interface{}{}
		}
		tmp = make(map[string]interface{})
	}
	wg.Wait()
	if len(updateExtract) > 0 {
		mgo.UpSertBulk(extract, updateExtract...)
	}
	log.Println("this task over.", n, "repeateN:", repeateN, mapInfo["stop"])

	time.Sleep(60 * time.Second)

	//任务完成，开始发送广播通知下面节点
	if n > repeateN && mapInfo["stop"] == nil {
		log.Println("判重任务完成发送udp")
		for _, to := range nextNode {
			sid, _ := mapInfo["gtid"].(string)
			eid, _ := mapInfo["lteid"].(string)
			key := sid + "-" + eid + "-" + util.ObjToString(to["stype"])
			by, _ := json.Marshal(map[string]interface{}{
				"gtid":  sid,
				"lteid": eid,
				"stype": util.ObjToString(to["stype"]),
				"key":   key,
			})
			addr := &net.UDPAddr{
				IP:   net.ParseIP(to["addr"].(string)),
				Port: util.IntAll(to["port"]),
			}
			node := &udpNode{by, addr, time.Now().Unix(), 0}
			udptaskmap.Store(key, node)
			udpclient.WriteUdp(by, mu.OP_TYPE_DATA, addr)
		}
	}
}


func historyTaskDay() {
	defer util.Catch()

	for {
		start:=time.Now().Unix()

		if gtid=="" {
			log.Println("请传gtid,否则无法运行")
			os.Exit(0)
			return
		}
		if lteid!="" {
			//先进行数据迁移
			log.Println("开启一次迁移任务",gtid,lteid)
			moveHistoryData(gtid,lteid)
			gtid = lteid //替换数据
		}

		//查询表最后一个id
		sess := mgo.GetMgoConn()
		defer mgo.DestoryMongoConn(sess)
		q:=map[string]interface{}{}
		between_time := time.Now().Unix() - (86400 * timingPubScope)//两年周期
		it_last := sess.DB(mgo.DbName).C(extract).Find(&q).Sort("-_id").Iter()
		for tmp := make(map[string]interface{}); it_last.Next(&tmp); {
			lteid = BsonTOStringId(tmp["_id"])
			break
		}

		//开始判重
		q = map[string]interface{}{
			"_id": map[string]interface{}{
				"$gt": StringTOBsonId(gtid),
				"$lte": StringTOBsonId(lteid),
			},
		}
		log.Println("历史判重查询条件：",q,"时间：", between_time)

		it := sess.DB(mgo.DbName).C(extract).Find(&q).Sort("publishtime").Iter()
		num,oknum, deterTime:= int64(0),int64(0),int64(0) //计数
		updateExtract := [][]map[string]interface{}{}//批量更新mongo数组
		pendAllArr:=[][]map[string]interface{}{}//待处理数组
		dayArr := []map[string]interface{}{}
		for tmp := make(map[string]interface{}); it.Next(&tmp); num++ {
			if num%10000 == 0 {
				log.Println("正序遍历:", num)
			}
			source := util.ObjToMap(tmp["jsondata"])
			if util.IntAll((*source)["sourcewebsite"]) == 1 {
				updateExtract = append(updateExtract, []map[string]interface{}{
					map[string]interface{}{
						"_id": tmp["_id"],
					},
					map[string]interface{}{
						"$set": map[string]interface{}{
							"repeat": 1,
							"dataging": 0,
							"repeat_reason": "sourcewebsite为1 重复",
						},
					},
				})
				if len(updateExtract) > 50 {
					mgo.UpSertBulk(extract, updateExtract...)
					updateExtract = [][]map[string]interface{}{}
				}
				tmp = make(map[string]interface{})
				continue
			}

			//取-符合-发布时间X年内的数据
			if util.IntAll(tmp["dataging"]) == 1 {
				pubtime := util.Int64All(tmp["publishtime"])
				if pubtime > 0 && pubtime >= between_time {
					oknum++
					if deterTime==0 {
						log.Println("找到第一条符合条件的数据")
						deterTime = util.Int64All(tmp["publishtime"])
						dayArr = append(dayArr,tmp)
					}else {
						if pubtime-deterTime >timingSpanDay*86400 {
							//新数组重新构建，当前组数据加到全部组数据
							pendAllArr = append(pendAllArr,dayArr)
							dayArr = []map[string]interface{}{}
							deterTime = util.Int64All(tmp["publishtime"])
							dayArr = append(dayArr,tmp)
						}else {
							dayArr = append(dayArr,tmp)
						}
					}
				}else {
					//不在两年内的也清标记
					updateExtract = append(updateExtract, []map[string]interface{}{
						map[string]interface{}{
							"_id": tmp["_id"],
						},
						map[string]interface{}{
							"$set": map[string]interface{}{
								"dataging": 0,
							},
						},
					})
					if len(updateExtract) > 50 {
						mgo.UpSertBulk(extract, updateExtract...)
						updateExtract = [][]map[string]interface{}{}
					}

				}
			}
			tmp = make(map[string]interface{})
		}


		//批量更新标记
		if len(updateExtract) > 0 {
			mgo.UpSertBulk(extract, updateExtract...)
			updateExtract = [][]map[string]interface{}{}
		}

		if len(dayArr)>0 {
			pendAllArr = append(pendAllArr,dayArr)
			dayArr = []map[string]interface{}{}
		}

		log.Println("查询数量：",num,"符合条件：",oknum)

		if len(pendAllArr) <= 0 {
			log.Println("没找到dataging==1的数据")
		}

		//测试分组数量是否正确
		testNum:=0
		for k,v:=range pendAllArr {
			log.Println("第",k,"组--","数量：",len(v))
			testNum = testNum+len(v)
		}
		log.Println("本地构建分组完成:",len(pendAllArr),"组","测试-总计数量：",testNum)

		n, repeateN := 0, 0
		log.Println("线程数：",threadNum)
		pool := make(chan bool, threadNum)
		wg := &sync.WaitGroup{}
		for k,v:=range pendAllArr { //每组结束更新一波数据
			pool <- true
			wg.Add(1)
			go func(k int, v []map[string]interface{}) {
				defer func() {
					<-pool
					wg.Done()
				}()
				//每组临时数组 -  互不干扰
				groupUpdateExtract := [][]map[string]interface{}{}
				//构建当前组的数据池
				log.Println("构建第", k, "组---（数据池）")
				//当前组的第一个发布时间
				first_pt := util.Int64All(v[len(v)-1]["publishtime"])
				curTM := TimedTaskDatamap(dupdays+int(timingSpanDay)+1, first_pt+86400, int(k))
				log.Println("开始遍历判重第", k, "组  共计数量：", len(v))
				n = n + len(v)
				log.Println("统计目前总数量：", n, "重复数量：", repeateN)
				for _, tmp := range v {
					info := NewInfo(tmp)
					if !LowHeavy { //是否进行低质量数据判重
						if invalidData(info.buyer, info.projectname, info.projectcode, info.contractnumber) {
							log.Println("无效数据")
							groupUpdateExtract = append(groupUpdateExtract, []map[string]interface{}{
								map[string]interface{}{
									"_id": tmp["_id"],
								},
								map[string]interface{}{
									"$set": map[string]interface{}{
										"repeat":   -1, //无效数据标签
										"dataging": 0,
									},
								},
							})
							if len(groupUpdateExtract) > 50 {
								mgo.UpSertBulk(extract, groupUpdateExtract...)
								groupUpdateExtract = [][]map[string]interface{}{}
							}
							return
						}
					}
					b, source, reason := curTM.check(info)
					if b { //有重复,生成更新语句，更新抽取和更新招标
						repeateN++
						//重复数据打标签
						groupUpdateExtract = append(groupUpdateExtract, []map[string]interface{}{
							map[string]interface{}{
								"_id": tmp["_id"],
							},
							map[string]interface{}{
								"$set": map[string]interface{}{
									"repeat":        1,
									"repeat_reason": reason,
									"repeat_id":     source.id,
									"dataging":      0,
								},
							},
						})
					} else {
						groupUpdateExtract = append(groupUpdateExtract, []map[string]interface{}{
							map[string]interface{}{
								"_id": tmp["_id"],
							},
							map[string]interface{}{
								"$set": map[string]interface{}{
									"dataging": 0, //符合条件的都为dataging==0
								},
							},
						})
					}
					if len(groupUpdateExtract) > 50 {
						mgo.UpSertBulk(extract, groupUpdateExtract...)
						groupUpdateExtract = [][]map[string]interface{}{}
					}
				}
				//每组数据结束-更新数据
				if len(groupUpdateExtract) > 0 {
					mgo.UpSertBulk(extract, groupUpdateExtract...)
				}
			}(k, v)

		}

		wg.Wait()


		//任务完成，开始发送广播通知下面节点 发udp 去升索引待定 + 合并
		if n > repeateN {
			for _, to := range nextNode {
				next_sid := util.BsonIdToSId(gtid)
				next_eid := util.BsonIdToSId(lteid)
				key := next_sid + "-" + next_eid + "-" + util.ObjToString(to["stype"])
				by, _ := json.Marshal(map[string]interface{}{
					"gtid":  next_sid,
					"lteid": next_eid,
					"stype": util.ObjToString(to["stype"]),
					"key":   key,
				})
				addr := &net.UDPAddr{
					IP:   net.ParseIP(to["addr"].(string)),
					Port: util.IntAll(to["port"]),
				}
				node := &udpNode{by, addr, time.Now().Unix(), 0}
				udptaskmap.Store(key, node)
				udpclient.WriteUdp(by, mu.OP_TYPE_DATA, addr)
			}
		}

		end:=time.Now().Unix()

		log.Println("this timeTask over.", n, "repeateN:", repeateN,gtid,lteid)
		log.Println(gtid,lteid)
		if end-start<60*5 {
			log.Println("睡眠.............")
			time.Sleep(5 * time.Minute)
		}
		log.Println("继续下一段的历史判重")
	}


}





//迁移上一段数据
func moveHistoryData(startid string,endid string) {
	sess := mgo.GetMgoConn()
	defer mgo.DestoryMongoConn(sess)
	year, month, day := time.Now().Date()
	q := map[string]interface{}{
		"_id": map[string]interface{}{
			"$gt": StringTOBsonId(startid),
			"$lte": StringTOBsonId(endid),
		},
	}
	log.Println(q)
	it := sess.DB(mgo.DbName).C(extract).Find(&q).Iter()
	index := 0
	for tmp := make(map[string]interface{}); it.Next(&tmp); index++ {
		mgo.Save(extract_back, tmp)
		tmp = map[string]interface{}{}
		if index%1000 == 0 {
			log.Println("index", index)
		}
	}
	log.Println("save to", extract_back, " ok index", index)

	qv := map[string]interface{}{
		"comeintime": map[string]interface{}{
			"$lt": time.Date(year, month, day, 0, 0, 0, 0, time.Local).Add(-time.Duration(dupdays) * 24 * time.Hour*2).Unix(),
		},
	}
	delnum := mgo.Delete(extract, qv)
	log.Println("remove from ", extract, delnum)

}