数据清洗 - 备份

sensitive/src/config.json

@@ -23,5 +23,8 @@
   "nextNode": [],
   "client_es": "http://ela.spdata.jianyu360.com",
   "es_type": "unique_qy",
-  "es_index": "unique_qy"
+  "es_index": "unique_qy",
+  "qyxy_client_es":"http://127.0.0.1:12003",
+  "qyxy_es_type":"qyxy",
+  "qyxy_es_index":"qyxy"
 }

sensitive/src/main.go

@@ -1,6 +1,8 @@
 package main
 
 import (
+	"log"
+	"time"
 	ul "util"
 )
 
@@ -8,9 +10,22 @@ func init()  {
 	ul.InitC()
 }
 func main() {
+	//ul.TestData()
 	//go ul.AddTaskSensitiveWordsData() //增量-新增
 	ul.SentiveUdp()  //数据流程-增量
+
 	lock := make(chan bool)
 	<-lock
 }
 
+
+
+
+func ttttt()  {
+	log.Println("。。。。。。")
+	for {
+		tick := time.Tick(time.Second * 10)
+		<-tick
+		log.Println("执行一次")
+	}
+}

sensitive/src/mark

@@ -1,44 +1,60 @@
 {
-  "85_mgo_addr": "192.168.3.207:27092",
-  "mongodbPoolSize": 10,
-  "collection": "result_20210109",
-  "163_mgo_addr": "172.17.4.187:27082,172.17.145.163:27083",
-  "163_userName": "zhengkun",
-  "163_passWord": "zk@123123",
+  "udpport": ":1762",
+  "save_mgodb": {
+    "addr": "172.17.4.85:27080",
+    "db": "qfw",
+    "coll": "result_20210109",
+    "pool": 10
+  },
+  "qfw_mgodb": {
+    "addr": "172.17.4.187:27082,172.17.145.163:27083",
+    "db": "mixdata",
+    "coll": "qyxy_std",
+    "pool": 10,
+    "username": "zhengkun",
+    "password": "zk@123123"
+  },
   "fields": {
     "buyer": 1,
     "agency": 1,
     "winner": 1,
     "s_winner": 1
   },
-  "buyer_c": "",
-  "agency_c": "",
-  "winner_c": "",
-  "udpport": "1782",
-  "nextNode": [
-    {
-      "addr": "127.0.0.1",
-      "port": 88888,
-      "memo": "抽取城市"
-    }
-  ],
-  "userName": "",
-  "passWord": "",
-  "client_es": "http://ela.spdata.jianyu360.com",
+  "nextNode": [],
+  "client_es": "http://172.17.4.84:9800",
   "es_type": "unique_qy",
-  "es_index": "unique_qy"
+  "es_index": "unique_qy",
+  "qyxy_client_es":"http://172.17.145.170:9800",
+  "qyxy_es_type":"qyxy",
+  "qyxy_es_index":"qyxy"
 }
 
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
 {
   "udpport": ":1762",
   "save_mgodb": {
-    "addr": "172.17.4.85:27080",
+    "addr": "127.0.0.1:27017",
     "db": "qfw",
     "coll": "result_20210109",
     "pool": 10
   },
   "qfw_mgodb": {
-    "addr": "172.17.4.187:27082,172.17.145.163:27083",
+    "addr": "127.0.0.1:27017",
     "db": "mixdata",
     "coll": "qyxy_std",
     "pool": 10,
@@ -52,7 +68,10 @@
     "s_winner": 1
   },
   "nextNode": [],
-  "client_es": "http://172.17.4.84:9800",
+  "client_es": "http://ela.spdata.jianyu360.com",
   "es_type": "unique_qy",
-  "es_index": "unique_qy"
+  "es_index": "unique_qy",
+  "qyxy_client_es":"http://172.17.145.170:9800",
+  "qyxy_es_type":"qyxy",
+  "qyxy_es_index":"qyxy"
 }

sensitive/src/util/increasedata.go

@@ -10,11 +10,13 @@ import (
 
 //定时增量数据处理
 func AddTaskSensitiveWordsData() {
-	log.Println("部署增量......")
+	log.Println("部署定时新增...")
+	runStartOnce()
 	for {
 		tick := time.Tick(time.Hour * 24 * 7) //查询七天前
 		ctime := <-tick
-		cronData := time.Date(ctime.Year(), ctime.Month(), ctime.Day()-7, ctime.Hour(), ctime.Minute(), ctime.Second(), 0, time.Local)
+		log.Println("相隔一周~~~执行一次")
+		cronData := time.Date(ctime.Year(), ctime.Month(), ctime.Day()-7, 0, 0, 0, 0, time.Local)
 		q := map[string]interface{}{
 			"updatetime": map[string]interface{}{
 				"$gte":  cronData.Unix(),
@@ -27,8 +29,11 @@ func AddTaskSensitiveWordsData() {
 			"updatetime": 1,
 			"company_type": 1,
 		}).Iter()
-		total:=0
+		total,isok:=0,0
 		for tmp := map[string]interface{}{}; iter.Next(&tmp); total++{
+			if total%10000==0 {
+				log.Println("cur index ",total,isok)
+			}
 			if company_name, ok := tmp["company_name"].(string); ok {
 				if reglen.MatchString(company_name) || !unstart_strReg.MatchString(company_name) ||
 					con_strReg.MatchString(company_name) {
@@ -43,14 +48,62 @@ func AddTaskSensitiveWordsData() {
 					"qy_name": company_name,
 				})
 				if new_id!=nil {
-					dealWithEsData(company_name, BsonTOStringId(BsonTOStringId(new_id)))
+					isok++
+					dealWithEsData(company_name, BsonTOStringId(new_id))
 				}
 			}
 			tmp = map[string]interface{}{}
 		}
-		log.Println("tick ok", cronData)
+		log.Println("is over ",total)
+	}
+}
+
+//执行一次
+func runStartOnce()  {
+	log.Println("立即执行一次...")
+	nowtime := time.Now()
+	cronData := time.Date(nowtime.Year(), nowtime.Month(), nowtime.Day()-7, 0, 0, 0, 0, time.Local)
+	q := map[string]interface{}{
+		"updatetime": map[string]interface{}{
+			"$gte":  cronData.Unix(),
+		},
+	}
+	sess := Qfw_Mgo.GetMgoConn()
+	defer Qfw_Mgo.DestoryMongoConn(sess)
+	iter := sess.DB(Qfw_Mgo.DbName).C(QfwCollName).Find(&q).Select(map[string]interface{}{
+		"company_name": 1,
+		"updatetime": 1,
+		"company_type": 1,
+	}).Iter()
+	total,isok:=0,0
+	for tmp := map[string]interface{}{}; iter.Next(&tmp); total++{
+		if total%10000==0 {
+			log.Println("cur index ",total,isok)
+		}
+		if company_name, ok := tmp["company_name"].(string); ok {
+			if reglen.MatchString(company_name) || !unstart_strReg.MatchString(company_name) ||
+				con_strReg.MatchString(company_name) {
+				continue
+			}
+			company_type:= qu.ObjToString(tmp["company_type"])
+			if strings.Contains(company_type,"个人")||strings.Contains(company_type,"个体"){
+				continue
+			}
+			//存mgo
+			new_id :=Qfw_Mgo.Save("unique_qyxy", map[string]interface{}{
+				"qy_name": company_name,
+			})
+			if new_id!=nil {
+				isok++
+				dealWithEsData(company_name, BsonTOStringId(new_id))
+			}
+		}
+		tmp = map[string]interface{}{}
 	}
+	log.Println("is over ",total)
 }
+
+
 //处理是否新增es
 func dealWithEsData(name string, tmpid string) {
 	query := `{"query":{"bool":{"must":[{"term":{"` + Es_index + `.name":"` + name + `"}}],"must_not":[],"should":[]}},"from":0,"size":1,"sort":[],"facets":{}}`

sensitive/src/util/init.go

@@ -31,18 +31,23 @@ func InitC() {
 	qfwconf := Sysconfig["qfw_mgodb"].(map[string]interface{})
 	QfwCollName = qu.ObjToString(qfwconf["coll"])
 	Qfw_Mgo = &MongodbSim{
-		MongodbAddr: saveconf["addr"].(string),
-		DbName:      saveconf["db"].(string),
-		Size:        qu.IntAllDef(saveconf["pool"], 5),
+		MongodbAddr: qfwconf["addr"].(string),
+		DbName:      qfwconf["db"].(string),
+		Size:        qu.IntAllDef(qfwconf["pool"], 5),
 		UserName:    qfwconf["username"].(string),
-		PassWord:    qfwconf["username"].(string),
+		PassWord:    qfwconf["password"].(string),
 	}
 	Qfw_Mgo.InitPool()
 
+	Fields = Sysconfig["fields"].(map[string]interface{})
+
 	//初始化es http://172.17.4.84:9800
 	Client_Es, _ = elastic.NewClient(http.DefaultClient, Sysconfig["client_es"].(string))
 	Es_type, Es_index = Sysconfig["es_type"].(string), Sysconfig["es_index"].(string)
-	Fields = Sysconfig["fields"].(map[string]interface{})
+
+	//初始化Es qyxy
+	Qyxy_Client_Es, _ = elastic.NewClient(http.DefaultClient, Sysconfig["qyxy_client_es"].(string))
+	Qyxy_Es_type, Qyxy_Es_index = Sysconfig["qyxy_es_type"].(string), Sysconfig["qyxy_es_index"].(string)
 }
 
 var Sysconfig map[string]interface{}
@@ -52,4 +57,6 @@ var FindBuyerC, FindAgencyC, FindWinnerC string
 var SaveCollName,QfwCollName string
 var Es_type, Es_index string
 var Client_Es *elastic.Client
+var Qyxy_Es_type, Qyxy_Es_index string
+var Qyxy_Client_Es *elastic.Client
 var BuyerFilter = sensitive.New()

sensitive/src/util/test.go

@@ -4,7 +4,11 @@ import "log"
 
 //测试方法
 func TestData() {
-	name:="上海熊猫机械(集团)有限公司昆明分公司"
+	//
+	log.Println("测试一下...")
+	name:="四川灵猫王有害生物防治有限公司"
+	fok, flog, fname := cheakname(name)
+	log.Println(fok,flog,fname)
 	rname, isok, _ ,datas := dealWithNameScoreRules(name)
 	log.Println(rname,isok,len(datas))
 }

sensitive/src/util/udpdata.go

@@ -52,7 +52,13 @@ func processUdpMsg(act byte, data []byte, ra *net.UDPAddr) {
 			go udpclient.WriteUdp([]byte("get:"+sid+"_"+eid), mu.OP_NOOP, ra)
 			log.Println("Udp回应上节点~id段")
 			QuerySensitiveWords(sid, eid)
-			log.Println("")
+			log.Println("...计划发送udp~统计下一节点...")
+			//
+
+
+
+
+
 		}
 	case mu.OP_NOOP: //下个节点回应
 		log.Println(string(data))
@@ -176,10 +182,11 @@ func cheakname(name string) (up bool, log, rname string) {
 	return false, "", name
 }
 func termQuery(name string) bool {
-	query := `{"query":{"bool":{"must":[{"term":{"` + Es_index + `.name":"` + name + `"}}],"must_not":[],"should":[]}},"from":0,"size":1,"sort":[],"facets":{}}`
+
+	query := `{"query":{"bool":{"must":[{"term":{"company_name":"` + name + `"}}],"must_not":[],"should":[]}},"from":0,"size":1,"sort":[],"facets":{}}`
 	tmp := make(map[string]interface{})
 	json.Unmarshal([]byte(query), &tmp)
-	searchResult, err := Client_Es.Search().Index(Es_index).Type(Es_type).Source(tmp).Do()
+	searchResult, err := Qyxy_Client_Es.Search().Index(Qyxy_Es_index).Type(Qyxy_Es_type).Source(tmp).Do()
 	if err != nil {
 		log.Println("从ES查询出错", err.Error(), name)
 		return false