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

package src1

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

import (
	"encoding/json"
	"flag"
	"fmt"
	"github.com/cron"
	"log"
	mu "mfw/util"
	"net"
	"os"
	"qfw/util"
	"regexp"
	"strings"
	"sync"
	"time"

	"gopkg.in/mgo.v2/bson"
)

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                             //发布时间周期
	sid,eid,lastid string                     		 //测试人员判重使用
	IdType         bool   							 //默认object类型
)

func init() {

	flag.StringVar(&lastid, "id", "", "最后加载id") //以小于等于此id开始加载最近几天的数据
	flag.StringVar(&sid, "sid", "", "开始id")
	flag.StringVar(&eid, "eid", "", "结束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 mainT() {

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



//测试组人员使用
func main() {

	if TimingTask {
		log.Println("定时任务测试开始")
		go timedTaskDay()
		time.Sleep(99999 * time.Hour)
	} else {
		//IdType = true  //打开id字符串模式
		sid = "5ef01220801f744d045f51f1"
		eid = "5ef61eb3801f744d046402dd"
		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{}{}
	ids:=[]string{}
	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 && tmp["attach_text"]!=nil{
			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,
						},
					},
				})
			}
		}(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"])

	//任务完成，开始发送广播通知下面节点
	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,
				"ids":   strings.Join(ids, ","),
			})
			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 timedTaskDay() {
	log.Println("部署定时任务")
	c := cron.New()
	c.AddFunc("0 0 */4 * * ?", func() { timedTaskOnce() })
	c.Start()
}
func timedTaskOnce() {
	defer util.Catch()
	log.Println("开始一次迁移任务")
	movedata()
	log.Println("开始一次任务判重")
	//当前时间-8   -4 小时
	now := time.Now()
	log.Println(now)
	preTime := time.Date(now.Year(), now.Month(), now.Day(), now.Hour()-8, 0, 0, 0, time.Local)
	curTime := time.Date(now.Year(), now.Month(), now.Day(), now.Hour()-4, 0, 0, 0, time.Local)
	log.Println(preTime,curTime)
	task_sid := util.BsonIdToSId(bson.NewObjectIdWithTime(preTime))
	task_eid := util.BsonIdToSId(bson.NewObjectIdWithTime(curTime))
	between_time := curTime.Unix() - (86400 * timingPubScope)
	log.Println("id区间：",task_sid, task_eid,"时间：", between_time)
	//区间id
	q_start := map[string]interface{}{
		"_id": map[string]interface{}{
			"$gte": StringTOBsonId(task_sid),
			"$lte": StringTOBsonId(task_eid),
		},
	}
	sess := mgo.GetMgoConn()
	defer mgo.DestoryMongoConn(sess)
	it_start := sess.DB(mgo.DbName).C(extract).Find(&q_start).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_start.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的数据")
		return
	}

	//测试分组数量是否正确
	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
	for k,v:=range pendAllArr { //每组结束更新一波数据
		//构建当前组的数据池
		log.Println("构建第",k,"组---（数据池）")
		//当前组的第一个发布时间
		first_pt :=util.Int64All(v[0]["publishtime"])
		DM = TimedTaskDatamap(dupdays, first_pt)
		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("无效数据")
					updateExtract = append(updateExtract, []map[string]interface{}{
						map[string]interface{}{
							"_id": tmp["_id"],
						},
						map[string]interface{}{
							"$set": map[string]interface{}{
								"repeat":   -1, //无效数据标签
								"dataging": 0,
							},
						},
					})
					if len(updateExtract) > 50 {
						mgo.UpSertBulk(extract, updateExtract...)
						updateExtract = [][]map[string]interface{}{}
					}
					continue
				}
			}
			b, source, reason := DM.check(info)
			if b { //有重复,生成更新语句，更新抽取和更新招标
				log.Println("判重结果", b, reason,"目标id",info.id)
				repeateN++
				//重复数据打标签
				updateExtract = append(updateExtract, []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 {
				updateExtract = append(updateExtract, []map[string]interface{}{
					map[string]interface{}{
						"_id": tmp["_id"],
					},
					map[string]interface{}{
						"$set": map[string]interface{}{
							"dataging": 0,//符合条件的都为dataging==0
						},
					},
				})
			}
			if len(updateExtract) > 50 {
				mgo.UpSertBulk(extract, updateExtract...)
				updateExtract = [][]map[string]interface{}{}
			}
		}

		//每组数据结束-更新数据
		if len(updateExtract) > 0 {
			mgo.UpSertBulk(extract, updateExtract...)
			updateExtract = [][]map[string]interface{}{}
		}

	}



	if len(updateExtract) > 0 {
		mgo.UpSertBulk(extract, updateExtract...)
		updateExtract = [][]map[string]interface{}{}
	}
	log.Println("this timeTask over.", n, "repeateN:", repeateN)

	//任务完成，开始发送广播通知下面节点 发udp 去升索引待定 + 合并
	if n > repeateN {
		for _, to := range nextNode {
			next_sid := util.BsonIdToSId(task_sid)
			next_eid := util.BsonIdToSId(task_eid)
			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)
		}
	}
}


//迁移数据dupdays+5之前的数据
func movedata() {
	sess := mgo.GetMgoConn()
	defer mgo.DestoryMongoConn(sess)
	year, month, day := time.Now().Date()
	now:=time.Now()
	move_time := time.Date(now.Year(), now.Month(), now.Day(), now.Hour()-8, 0, 0, 0, time.Local).Unix()
	q := map[string]interface{}{
		"comeintime": map[string]interface{}{
			"$lt": move_time,
		},
	}
	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)
}