调试验证历史数据判重相关功能

udpfilterdup/src/datamap.go

@@ -132,17 +132,18 @@ func NewHistorymap(startid string,lastid string,startTime int64,lastTime int64)
 	defer mgo.DestoryMongoConn(sess_start)  //lte  gte
 	it_start := sess_start.DB(mgo.DbName).C(extract).Find(mongodb.ObjToMQ(`{"_id":{"$lte":"`+startid+`"}}`,
 		true)).Sort("-_id").Iter()
-	n := 0
-	for tmp := make(map[string]interface{}); it_start.Next(&tmp); n++ {
-		cm := tmp["comeintime"]
-		comeintime := qutil.Int64All(tmp["comeintime"])
+	m,n := 0,0
+	for tmp_start := make(map[string]interface{}); it_start.Next(&tmp_start);{
+		cm := tmp_start["comeintime"]
+		comeintime := qutil.Int64All(tmp_start["comeintime"])
 		if comeintime == 0 {
-			id := qutil.BsonIdToSId(tmp["_id"])[0:8]
+			id := qutil.BsonIdToSId(tmp_start["_id"])[0:8]
 			comeintime, _ = strconv.ParseInt(id, 16, 64)
 		}
 
-		if qutil.Float64All(startTime-comeintime) < datelimit {
-			info := NewInfo(tmp)
+		if qutil.Float64All(startTime-comeintime) <= datelimit {
+			n++
+			info := NewInfo(tmp_start)
 			dkey := qutil.FormatDateWithObj(&cm, qutil.Date_yyyyMMdd)
 			k := fmt.Sprintf("%s_%s_%s", dkey, info.subtype, info.area)
 			data := hm.data[k]
@@ -152,28 +153,30 @@ func NewHistorymap(startid string,lastid string,startTime int64,lastTime int64)
 			data = append(data, info)
 			hm.data[k] = data
 			hm.keys[dkey] = true
-		} else {
+		}else {
 			break
 		}
-		tmp = make(map[string]interface{})
+		tmp_start = make(map[string]interface{})
 	}
 
-
+	log.Println("load history 前:", n)
 	//取lastid之后5天
 	sess_last := mgo.GetMgoConn()
 	defer mgo.DestoryMongoConn(sess_last)  //lte  gte
-	it_last := sess_start.DB(mgo.DbName).C(extract).Find(mongodb.ObjToMQ(`{"_id":{"$gte":"`+lastid+`"}}`,
+	it_last := sess_last.DB(mgo.DbName).C(extract).Find(mongodb.ObjToMQ(`{"_id":{"$gte":"`+lastid+`"}}`,
 		true)).Sort("_id").Iter()
-	for tmp := make(map[string]interface{}); it_last.Next(&tmp); n++ {
-		cm := tmp["comeintime"]
-		comeintime := qutil.Int64All(tmp["comeintime"])
+
+	for tmp_last := make(map[string]interface{}); it_last.Next(&tmp_last); {
+		cm := tmp_last["comeintime"]
+		comeintime := qutil.Int64All(tmp_last["comeintime"])
 		if comeintime == 0 {
-			id := qutil.BsonIdToSId(tmp["_id"])[0:8]
+			id := qutil.BsonIdToSId(tmp_last["_id"])[0:8]
 			comeintime, _ = strconv.ParseInt(id, 16, 64)
 		}
 
-		if qutil.Float64All(comeintime-lastTime) > datelimit {
-			info := NewInfo(tmp)
+		if qutil.Float64All(comeintime-lastTime) <= datelimit {
+			m++
+			info := NewInfo(tmp_last)
 			dkey := qutil.FormatDateWithObj(&cm, qutil.Date_yyyyMMdd)
 			k := fmt.Sprintf("%s_%s_%s", dkey, info.subtype, info.area)
 			data := hm.data[k]
@@ -183,17 +186,15 @@ func NewHistorymap(startid string,lastid string,startTime int64,lastTime int64)
 			data = append(data, info)
 			hm.data[k] = data
 			hm.keys[dkey] = true
-		} else {
+		}else {
 			break
 		}
-		tmp = make(map[string]interface{})
+		tmp_last = make(map[string]interface{})
 	}
 
+	log.Println("load history 后:", m)
 
 
-
-
-	log.Println("load history:", n)
 	return hm
 }
 
@@ -293,54 +294,54 @@ L:
 						}
 					}
 
-					//前置条件
+					//前置条件2个不重复  一个重复
 					if info.titleSpecialWord&&info.title!=v.title&&v.title!="" {
 						continue
 					}
-
 					if info.buyer != "" &&v.buyer == info.buyer {
 						//满足标题
 						if len([]rune(v.title)) >= 10 && len([]rune(info.title)) >= 10 && v.title != info.title && (info.specialWord || v.specialWord) {
 							continue
 						}
 					}
+					if info.site!=""&&info.site==v.site{
+						if info.href!=""&&info.href==v.href {
+							reason = "href相同"
+							b = true
+							source = v
+							reasons = reason
+							break L
+						}
+					}
 					//代理机构相同-非空相等
 					if v.agency != "" && info.agency != "" && v.agency == info.agency {
-						reason = reason + "同机构,"
-						if info.agency=="" {
-							reason = reason + "指定范围,"
-							//指定该范围内数据判重  jsondata
-							if v.agency=="" {
-								continue
-							}
-						}else {
-							reason = reason + "非指定范围,"
+						reason = reason + "同机构-"
+						if quickHeavyMethodTwo(v,info) {
+							b = true
+							source = v
+							reasons = reason
+							break
+						}
+					}else {
+						reason = reason + "非同机构-"
+						if info.city!=""&&info.city==v.city{
+							reason = reason + "同城-"
 							if quickHeavyMethodTwo(v,info) {
 								b = true
 								source = v
 								reasons = reason
-								break L
-							}
-						}
-					}else {
-						reason = reason + "非同机构,"
-						if info.agency=="" {
-							reason = reason + "指定范围,"
-							//指定该范围内数据判重  jsondata
-							if v.agency=="" {
-								continue
+								break
 							}
 						}else {
-							reason = reason + "非指定范围,"
+							reason = reason + "不同城-"
 							if quickHeavyMethodOne(v,info) {
 								b = true
 								source = v
 								reasons = reason
-								break L
+								break
 							}
 						}
 					}
-
 				}
 			}
 		}
@@ -385,47 +386,78 @@ L:
 		data := h.data[k]
 		if len(data) > 0 { //对比v   找到同类型，同省或全国的数据作对比
 			for _, v := range data {
+				reason = ""
 				if v.id == info.id {//正常重复
 					return false, v,""
 				}
+				//备份  新增发布时间为空-取入库时间-在为空取id
 				if math.Abs(qutil.Float64All(v.accurateTime-info.accurateTime)) > datelimit {
 					continue   //是否为5天内数据
 				}
-				//类型分组-相同类型继续
+				//类型分组
 				if info.subtype==v.subtype {
+					//站点配置--
+					if info.site!="" {
+						dict := SiteMap[info.site].(map[string]string)
+						if dict!=nil{
+							//临时改变--具体值
+							if info.area=="全国" &&dict["area"]!="" {
+								info.area = dict["area"]
+								info.city = dict["city"]
+							}else {
+								if info.city=="" &&dict["city"]!="" {
+									info.area = dict["area"]
+									info.city = dict["city"]
+								}
+							}
+						}
+					}
+
+					//前置条件2个不重复  一个重复
+					if info.titleSpecialWord&&info.title!=v.title&&v.title!="" {
+						continue
+					}
+					if info.buyer != "" &&v.buyer == info.buyer {
+						//满足标题
+						if len([]rune(v.title)) >= 10 && len([]rune(info.title)) >= 10 && v.title != info.title && (info.specialWord || v.specialWord) {
+							continue
+						}
+					}
+					if info.site!=""&&info.site==v.site{
+						if info.href!=""&&info.href==v.href {
+							reason = "href相同"
+							b = true
+							source = v
+							reasons = reason
+							break L
+						}
+					}
 					//代理机构相同-非空相等
 					if v.agency != "" && info.agency != "" && v.agency == info.agency {
-						reason = reason + "同机构,"
-						if info.agency=="" {
-							reason = reason + "指定范围,"
-							//指定该范围内数据判重  jsondata
-							if v.agency=="" {
-								continue
-							}
-						}else {
-							reason = reason + "非指定范围,"
+						reason = reason + "同机构-"
+						if quickHeavyMethodTwo(v,info) {
+							b = true
+							source = v
+							reasons = reason
+							break
+						}
+					}else {
+						reason = reason + "非同机构-"
+						if info.city!=""&&info.city==v.city{
+							reason = reason + "同城-"
 							if quickHeavyMethodTwo(v,info) {
 								b = true
 								source = v
 								reasons = reason
-								break L
-							}
-						}
-					}else {
-						reason = reason + "非同机构,"
-						if info.agency=="" {
-							reason = reason + "指定范围,"
-							//指定该范围内数据判重  jsondata
-							if v.agency=="" {
-								continue
+								break
 							}
 						}else {
-							reason = reason + "非指定范围,"
+							reason = reason + "不同城-"
 							if quickHeavyMethodOne(v,info) {
 								b = true
 								source = v
 								reasons = reason
-								break L
+								break
 							}
 						}
 					}
@@ -657,32 +689,32 @@ func tenderRepeat_A(v *Info ,info *Info) bool {
 	var ss string
 	p1,p2,p3,p4,p9,p10,p11 := false,false,false,false,false,false,false
 	if v.projectname!=""&&v.projectname==info.projectname {
-		ss = fmt.Sprintf(ss,"p1(名称)-")
+		ss = ss+"p1(名称)-"
 		p1 = true
 	}
 	if v.buyer!=""&&v.buyer==info.buyer {
-		ss = fmt.Sprintf(ss,"p2(单位)-")
+		ss = ss+"p2(单位)-"
 		p2 = true
 	}
 	if v.projectcode!=""&&v.projectcode==info.projectcode {
-		ss = fmt.Sprintf(ss,"p3(编号)-")
+		ss = ss+"p3(编号)-"
 		p3 = true
 	}
 	if v.budget!=0&&v.budget==info.budget {
-		ss = fmt.Sprintf(ss,"p4(预算)-")
+		ss = ss+"p4(预算)-"
 		p4 = true
 	}
 	if v.bidopentime!=0&&v.bidopentime==info.bidopentime {
-		ss = fmt.Sprintf(ss,"p9(开标时间)-")
+		ss = ss+"p9(开标时间)-"
 		p9 = true
 	}
 	if v.agencyaddr!=""&&v.agencyaddr==info.agencyaddr {
-		ss = fmt.Sprintf(ss,"p10(开标地点)-")
+		ss = ss+"p10(开标地点)-"
 		p10 = true
 	}
 	if len([]rune(v.title))>10 && len([]rune(info.title))>10&&
 		(strings.Contains(v.title, info.title)||strings.Contains(info.title, v.title)) {
-		ss = fmt.Sprintf(ss,"p11(标题)-")
+		ss = ss+"p11(标题)-"
 		p11 = true
 	}
 
@@ -731,7 +763,7 @@ func tenderRepeat_B(v *Info ,info *Info) bool {
 		if n==2 &&m==2 {
 			return false
 		}else {
-			reason = reason+"满足招标B，七选二,"
+			reason = reason+"满足招标B,七选二,"
 			return true
 		}
 	}
@@ -767,28 +799,28 @@ func winningRepeat_A(v *Info ,info *Info) bool {
 	var ss string
 	p1,p2,p3,p5,p6,p11 := false,false,false,false,false,false
 	if v.projectname!=""&&v.projectname==info.projectname {
-		ss = fmt.Sprintf(ss,"p1(标题)-")
+		ss = ss+"p1(标题)-"
 		p1 = true
 	}
 	if v.buyer!=""&&v.buyer==info.buyer {
-		ss = fmt.Sprintf(ss,"p2(单位)-")
+		ss = ss+"p2(单位)-"
 		p2 = true
 	}
 	if v.projectcode!=""&&v.projectcode==info.projectcode {
-		ss = fmt.Sprintf(ss,"p3(编号)-")
+		ss = ss+"p3(编号)-"
 		p3 = true
 	}
 	if v.bidamount!=0&&v.bidamount==info.bidamount {
-		ss = fmt.Sprintf(ss,"p5(中标金)-")
+		ss = ss+"p5(中标金)-"
 		p5 = true
 	}
 	if v.winner!=""&&v.winner==info.winner {
-		ss = fmt.Sprintf(ss,"p6(中标人)-")
+		ss = ss+"p6(中标人)-"
 		p6 = true
 	}
 	if len([]rune(v.title))>10 && len([]rune(info.title))>10&&
 		(strings.Contains(v.title, info.title)||strings.Contains(info.title, v.title)) {
-		ss = fmt.Sprintf(ss,"p11(标题)-")
+		ss = ss+"p11(标题)-"
 		p11 = true
 	}
 
@@ -798,7 +830,7 @@ func winningRepeat_A(v *Info ,info *Info) bool {
 		(p2&&p5&&p6)||(p2&&p5&&p11)||(p2&&p6&&p11)||
 		(p3&&p5&&p6)||(p3&&p5&&p11)||(p3&&p6&&p11)||
 		(p5&&p6&&p11){
-		reason = reason+"满足中标A，3要素组合-"+ss+","
+		reason = reason+"满足中标A,3要素组合-"+ss+","
 		return true
 	}
 
@@ -833,7 +865,7 @@ func winningRepeat_B(v *Info ,info *Info) bool {
 		if n==2 &&m==2 {
 			return false
 		}else {
-			reason = reason+"满足中标B，六选二,"
+			reason = reason+"满足中标B.六选二,"
 			return true
 		}
 	}
@@ -924,13 +956,6 @@ func (d *datamap) update(t int64) {
 	//log.Println("更新前后数据:", all, all1)
 }
 
-
-
-
-
-
-
-
 func (d *datamap) GetLatelyFiveDay(t int64) []string {
 	array := make([]string, d.days)
 	now := time.Unix(t, 0)

udpfilterdup/src/main.go

@@ -37,9 +37,7 @@ var (
 	DM           *datamap                 //
 	HM           *historymap                 //判重数据
 	lastid       = "5d767728a5cb26b9b7748868"
-	//5da3f2c5a5cb26b9b79847fc
 	//ObjectId("5d767728a5cb26b9b7748868")
-	//5da3f2c5a5cb26b9b79847fe
 	//正则筛选相关
 	FilterRegTitle = regexp.MustCompile("^_$")
 	FilterRegTitle_1 = regexp.MustCompile("^_$")
@@ -74,15 +72,10 @@ func init() {
 	dupdays = util.IntAllDef(Sysconfig["dupdays"], 3)
 	//加载数据
 	DM = NewDatamap(dupdays, lastid)
-	//fmt.Println(DM.keys)
-	//fmt.Println(DM.data)
 	FilterRegTitle = regexp.MustCompile(util.ObjToString(Sysconfig["specialwords"]))
 	FilterRegTitle_1 = regexp.MustCompile(util.ObjToString(Sysconfig["specialtitle_1"]))
 	FilterRegTitle_2 = regexp.MustCompile(util.ObjToString(Sysconfig["specialtitle_2"]))
 
-
-
-
 	//站点相关数据库
 	mongodb.InitMongodbPool(5, "192.168.3.207:27082", "")
 
@@ -272,11 +265,12 @@ func processUdpMsg(act byte, data []byte, ra *net.UDPAddr) {
 		} else if mapInfo != nil {
 
 			//更新流程
-			//go historyTask(data,mapInfo)
+			go historyTask(data,mapInfo)
 
 
 			//判重流程
-			go task(data, mapInfo)
+			//go task(data, mapInfo)
+
 			key, _ := mapInfo["key"].(string)
 			if key == "" {
 				key = "udpok"
@@ -295,8 +289,7 @@ func processUdpMsg(act byte, data []byte, ra *net.UDPAddr) {
 //开始判重程序
 func task(data []byte, mapInfo map[string]interface{}) {
 
-	//
-	fmt.Println("开始判重")
+	fmt.Println("开始数据判重")
 	defer util.Catch()
 	//区间id
 	sess := mgo.GetMgoConn()
@@ -348,19 +341,16 @@ func task(data []byte, mapInfo map[string]interface{}) {
 				}
 				mapLock.Unlock()
 			}else  {
-				//判重原因 reason
-				// tmp["_id"] 对比id   id原始id
+				//判重原因 reason  tmp["_id"] 对比id   id原始id
 				mapLock.Lock()
 				b, source,reason := DM.check(info)
 				if b { //有重复,生成更新语句，更新抽取和更新招标
 					repeateN++
 					var mergeArr  = []int64{} 	//更改合并数组记录
 					var newData  = &Info{}		//更换新的数据池数据
-					if repeateN<0 {
-						fmt.Println(mergeArr)
-					}
 					var id_map  = map[string]interface{}{}
 					repeat_id := ""
+
 					//合并操作--评功权重打分-合并完替换原始数据池
 					basic_bool := basicDataScore(source,info)
 					if basic_bool {
@@ -385,12 +375,11 @@ func task(data []byte, mapInfo map[string]interface{}) {
 						},
 					}
 
+					//合并记录
 					if len(newData.mergemap)>0 {
 						update_map["$set"].(map[string]interface{})["merge"] = newData.mergemap
-						fmt.Println(newData.mergemap,"---",len(newData.mergemap))
 					}
 
-
 					//更新合并后的数据
 					for _,value :=range mergeArr {
 						if value==1 {
@@ -415,9 +404,6 @@ func task(data []byte, mapInfo map[string]interface{}) {
 						}
 					}
 
-
-
-
 					//构建数据库更新用到的
 					updateExtract = append(updateExtract, []map[string]interface{}{
 						id_map,
@@ -429,32 +415,7 @@ func task(data []byte, mapInfo map[string]interface{}) {
 					}
 					mapLock.Unlock()
 
-
-
-					//log.Println("判重具体原因：",reason)
-					//repeateN++
-					//mapLock.Lock()
-					//updateExtract = append(updateExtract, []map[string]interface{}{
-					//	map[string]interface{}{
-					//		"_id": tmp["_id"],
-					//	},
-					//	map[string]interface{}{
-					//		"$set": map[string]interface{}{
-					//			"repeat":   1,
-					//			"repeatid": id,
-					//		},
-					//	},
-					//})
-					//
-					//if len(updateExtract) > 500 {
-					//	mgo.UpdateBulk(extract, updateExtract...)
-					//	updateExtract = [][]map[string]interface{}{}
-					//}
-					//mapLock.Unlock()
-
-
 				} else {
-					//IS.Add("new")
 					mapLock.Unlock()
 				}
 			}
@@ -522,12 +483,12 @@ func historyTask(data []byte, mapInfo map[string]interface{}) {
 			}
 		}
 	}
-	fmt.Println(minTime,maxTime)
-
+	fmt.Println("最小时间==",minTime,"最大时间==",maxTime)
+	//最小时间== 1568087634 最大时间== 1568103381
 	HM = NewHistorymap(util.ObjToString(mapInfo["gtid"]),
 		util.ObjToString(mapInfo["lteid"]),minTime,maxTime)
 
-
+	//return
 	//开始判重...
 	defer util.Catch()
 	sess_task := mgo.GetMgoConn()
@@ -558,7 +519,6 @@ func historyTask(data []byte, mapInfo map[string]interface{}) {
 				wg.Done()
 			}()
 			info := NewInfo(tmp)
-
 			//是否为无效数据
 			if invalidData(info.buyer,info.projectname,info.projectcode) {
 				mapLock.Lock()
@@ -572,22 +532,19 @@ func historyTask(data []byte, mapInfo map[string]interface{}) {
 						},
 					},
 				})
-
 				if len(updateExtract) > 500 {
 					mgo.UpdateBulk(extract, updateExtract...)
 					updateExtract = [][]map[string]interface{}{}
 				}
 				mapLock.Unlock()
 			}else  {
+				mapLock.Lock()
 				b, source,reason := HM.check(info)
 				if b { //有重复,生成更新语句，更新抽取和更新招标
-
 					if reason == "未判重记录" {
+						fmt.Println("未判重记录")
 						//把info的数据判重的标签更换，并新增字段
-						mapLock.Lock()
-						//构建数据库更新用到的
-						//对比的数据打判重标签
-						DM.replaceSourceData(info,info.id) //替换即添加
+						DM.replaceSourceData(info, info.id) //替换即添加
 						updateExtract = append(updateExtract, []map[string]interface{}{
 							map[string]interface{}{
 								"_id": tmp["_id"],
@@ -606,76 +563,39 @@ func historyTask(data []byte, mapInfo map[string]interface{}) {
 						mapLock.Unlock()
 					}else {
 						repeateN++
-						mapLock.Lock()
-
-						var mergeArr = []int64{} 	//更改合并数组记录
+						var mergeArr  = []int64{} 	//更改合并数组记录
 						var newData  = &Info{}		//更换新的数据池数据
-
 						var id_map  = map[string]interface{}{}
+						repeat_id := ""
+
 						//合并操作--评功权重打分-合并完替换原始数据池
 						basic_bool := basicDataScore(source,info)
 						if basic_bool {
 							//已原始数据为标准-对比数据打判重标签
 							newData,mergeArr= mergeDataFields(source,info)
-
-							//if repeateN<2 {
-							//	//
-							//	fmt.Println(newData.mergemap,mergeArr)
-							//}
-
 							DM.replaceSourceData(newData,source.id) //替换
 							id_map["_id"]= util.StringTOBsonId(source.id)
-
-							//对比的数据打判重标签
-							updateExtract = append(updateExtract, []map[string]interface{}{
-								map[string]interface{}{
-									"_id": tmp["_id"],
-								},
-								map[string]interface{}{
-									"$set": map[string]interface{}{
-										"repeat":   1,
-										"repeatid": source.id,
-									},
-								},
-							})
-
-
-
+							repeat_id = source.id
 						}else {
 							//已对比数据为标准 ，数据池的数据打判重标签
 							newData,mergeArr= mergeDataFields(info,source)
 							DM.replaceSourceData(newData,source.id)//替换
 							id_map["_id"]= util.StringTOBsonId(info.id)
-
-							//数据池的数据打判重标签
-							updateExtract = append(updateExtract, []map[string]interface{}{
-								map[string]interface{}{
-									"_id": util.StringTOBsonId(source.id),
-								},
-								map[string]interface{}{
-									"$set": map[string]interface{}{
-										"repeat":   1,
-										"repeatid": info.id,
-									},
-								},
-							})
-
+							repeat_id = info.id
 						}
 
-						//newData.mergemap["mark"] = "合并操作"
-
-
-
-						//
 						var update_map  = map[string]interface{}{
 							"$set": map[string]interface{}{
 								"reason":reason,
-								"merge":newData.mergemap,
+								"repeat":"1",
+								"repeatid":repeat_id,
 							},
 						}
 
-
-
+						//合并记录
+						if len(newData.mergemap)>0 {
+							update_map["$set"].(map[string]interface{})["merge"] = newData.mergemap
+						}
 
 						//更新合并后的数据
 						for _,value :=range mergeArr {
@@ -700,10 +620,6 @@ func historyTask(data []byte, mapInfo map[string]interface{}) {
 
 							}
 						}
-
-
-
-
 						//构建数据库更新用到的
 						updateExtract = append(updateExtract, []map[string]interface{}{
 							id_map,
@@ -715,8 +631,8 @@ func historyTask(data []byte, mapInfo map[string]interface{}) {
 						}
 						mapLock.Unlock()
 					}
-				} else {
-					//IS.Add("new")
+				}else {
+					mapLock.Unlock()
 				}
 			}
 		}(tmp)

udps/main.go

@@ -32,9 +32,14 @@ func main() {
 		9W
 	5d767728a5cb26b9b7748868
 	ObjectId("5d77c881a5cb26b9b7de209d")
+
+
+	//历史中间一段数据
+	ObjectId("5d771e90a5cb26b9b7be7976")
+	ObjectId("5d775be4a5cb26b9b759b5eb")
 	*/
-	flag.StringVar(&sid, "sid", "5d767728a5cb26b9b7748868", "开始id")
-	flag.StringVar(&eid, "eid", "5d77c881a5cb26b9b7de209d", "结束id")
+	flag.StringVar(&sid, "sid", "5d771e90a5cb26b9b7be7976", "开始id")
+	flag.StringVar(&eid, "eid", "5d775be4a5cb26b9b759b5eb", "结束id")
 	flag.StringVar(&startDate, "start", "", "开始日期2006-01-02")
 	flag.StringVar(&endDate, "end", "", "结束日期2006-01-02")
 	flag.StringVar(&ip, "ip", "127.0.0.1", "ip")