全量

udpfilterdup/src/README.md

@@ -13,6 +13,13 @@
             "coll": "site"
         }
     },
+    "task_mongodb": {
+        "task_addrName": "172.17.4.187:27081",
+        "task_dbName": "qfw",
+        "task_collName": "ocr_flie_over",
+        "pool": 5
+        
+    },
     "jkmail": {
         "to": "zhengkun@topnet.net.cn,zhangjinkun@topnet.net.cn",
         "api": "http://10.171.112.160:19281/_send/_mail"
@@ -95,6 +102,45 @@ func moveOnceTimeOut()  {
 
 
 
+{
+    "udpport": ":1785",
+    "dupdays": 7,
+    "mongodb": {
+        "addr": "172.17.4.85:27080",
+        "pool": 10,
+        "db": "qfw",
+        "extract": "result_20200715",
+        "extract_back": "result_20200714",
+        "site": {
+            "dbname": "qfw",
+            "coll": "site"
+        }
+    },
+    "jkmail": {
+        "to": "zhengkun@topnet.net.cn,zhangjinkun@topnet.net.cn",
+        "api": "http://172.17.145.179:19281/_send/_mail"
+    },
+    "nextNode": [
+        {
+            "addr": "127.0.0.1",
+            "port": 1783,
+            "stype": "bidding",
+            "memo": "创建招标数据索引new"
+        }
+    ],
+    "threads": 1,
+    "isMerger": false,
+    "lowHeavy":true,
+    "timingTask":false,
+    "timingSpanDay": 5,
+    "timingPubScope": 720,
+    "specialwords": "(重招|重新招标|勘察|施工|监理|总承包|土石方|可研)",
+    "specialtitle_0": "(包|标段|标包)[(（]?[0-9a-zA-Z一二三四五六七八九十零１２３４５６７８９][)）]?",
+    "specialtitle_1": "[0-9a-zA-Z一二三四五六七八九十零１２３４５６７８９](次|包|标段|标包|批|期)",
+    "specialtitle_2": "项目[(（][0-9a-zA-Z一二三四五六七八九十零１２３４５６７８９][)）]",
+    "beifen": "[(（]?[0-9一二三四五六七八九十零１２３４５６７８９再][)）]?[子分]?[次批标包]|重招|重新招标|勘察|设计|施工|监理|总承包|土石方|可研"
+}
+
 
 
 

udpfilterdup/src/config.json

@@ -12,6 +12,12 @@
             "coll": "site"
         }
     },
+    "task_mongodb": {
+        "task_addrName": "172.17.4.187:27081",
+        "task_dbName": "qfw",
+        "task_collName": "ocr_flie_over",
+        "pool": 5
+    },
     "jkmail": {
         "to": "zhengkun@topnet.net.cn,zhangjinkun@topnet.net.cn",
         "api": "http://10.171.112.160:19281/_send/_mail"

udpfilterdup/src/datamap.go

@@ -125,7 +125,6 @@ func TimedTaskDatamap(days int,lasttime int64,numIndex int) *datamap {
 	return dm
 }
 
-
 //增量
 func NewDatamap(days int, lastid string) *datamap {
 	datelimit = qutil.Float64All(days * 86400 * 2)
@@ -472,7 +471,8 @@ func (d *datamap) update(t int64) {
 	if TimingTask {
 
 	}else {
-		d.keymap = d.GetLatelyFiveDayDouble(t)
+		//d.keymap = d.GetLatelyFiveDayDouble(t)
+		d.keymap = d.GetLatelyFiveDay(t)
 	}
 	m := map[string]bool{}
 	for _, v := range d.keymap {
@@ -497,15 +497,15 @@ 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)
-//	for i := 0; i < d.days; i++ {
-//		array[i] = now.Format(qutil.Date_yyyyMMdd)
-//		now = now.AddDate(0, 0, -1)
-//	}
-//	return array
-//}
+func (d *datamap) GetLatelyFiveDay(t int64) []string  {
+	array := make([]string, d.days)
+	now := time.Unix(t, 0)
+	for i := 0; i < d.days; i++ {
+		array[i] = now.Format(qutil.Date_yyyyMMdd)
+		now = now.AddDate(0, 0, -1)
+	}
+	return array
+}
 
 func (d *datamap) GetLatelyFiveDayDouble(t int64) []string  {//增量-两倍
 	array := make([]string, d.days*2)

udpfilterdup/src/main.go

@@ -26,11 +26,13 @@ var (
 	Sysconfig    map[string]interface{} //配置文件
 	mconf        map[string]interface{} //mongodb配置信息
 	mgo          *MongodbSim            //mongodb操作对象
+	task_mgo     *MongodbSim            //mongodb操作对象
+	task_collName	string
 	extract      string
 	extract_back string
 	udpclient    mu.UdpClient             //udp对象
 	nextNode     []map[string]interface{} //下节点数组
-	dupdays      = 5                      //初始化判重范围
+	dupdays      = 7                      //初始化判重范围
 	DM           *datamap                 //
 
 	//正则筛选相关
@@ -61,6 +63,17 @@ func init() {
 	flag.Parse()
 	//172.17.145.163:27080
 	util.ReadConfig(&Sysconfig)
+
+	task_mconf := Sysconfig["task_mongodb"].(map[string]interface{})
+	task_mgo = &MongodbSim{
+		MongodbAddr: task_mconf["task_addrName"].(string),
+		DbName:      task_mconf["task_dbName"].(string),
+		Size:        util.IntAllDef(task_mconf["task_pool"], 10),
+	}
+	task_mgo.InitPool()
+	task_collName = task_mconf["task_collName"].(string)
+
+
 	nextNode = util.ObjArrToMapArr(Sysconfig["nextNode"].([]interface{}))
 	mconf = Sysconfig["mongodb"].(map[string]interface{})
 	mgo = &MongodbSim{
@@ -130,8 +143,8 @@ func mainT() {
 		time.Sleep(99999 * time.Hour)
 	} else {
 		//IdType = true  //打开id字符串模式
-		sid := "4f16936d52c1d9fbf843c60e"
-		eid := "6f16936d52c1d9fbf843c60e"
+		sid := "1f16936d52c1d9fbf843c60e"
+		eid := "9f16936d52c1d9fbf843c60e"
 		log.Println("正常判重测试开始")
 		log.Println(sid, "---", eid)
 		mapinfo := map[string]interface{}{}
@@ -355,7 +368,7 @@ func task(data []byte, mapInfo map[string]interface{}) {
 	}
 	log.Println("this task over.", n, "repeateN:", repeateN, mapInfo["stop"])
 
-	time.Sleep(60 * time.Second)
+	time.Sleep(30 * time.Second)
 
 	//任务完成，开始发送广播通知下面节点
 	if n >= repeateN && mapInfo["stop"] == nil {
@@ -401,18 +414,25 @@ func historyTaskDay() {
 		}
 
 		//查询表最后一个id
-		sess := mgo.GetMgoConn()
-		defer mgo.DestoryMongoConn(sess)
-		q:=map[string]interface{}{}
+		task_sess := task_mgo.GetMgoConn()
+		defer task_mgo.DestoryMongoConn(task_sess)
+		q:=map[string]interface{}{
+			"isused":true,
+		}
 		between_time := time.Now().Unix() - (86400 * timingPubScope)//两年周期
-		it_last := sess.DB(mgo.DbName).C(extract).Find(&q).Sort("-_id").Iter()
+		it_last := task_sess.DB(task_mgo.DbName).C(task_collName).Find(&q).Sort("-_id").Iter()
 		for tmp := make(map[string]interface{}); it_last.Next(&tmp); {
-			lteid = BsonTOStringId(tmp["_id"])
+			lteid = util.ObjToString(tmp["gtid"])
+			log.Println("查询的最后一个任务Id:",lteid)
 			break
 		}
-
+		//
+		log.Println("查询完毕-先睡眠5分钟",gtid,lteid)
 		time.Sleep(5 * time.Minute)
 
+
+		sess := mgo.GetMgoConn()//连接器
+		defer mgo.DestoryMongoConn(sess)
 		//开始判重
 		q = map[string]interface{}{
 			"_id": map[string]interface{}{