Merge branch 'dev3.3' of http://192.168.3.207:10080/qmx/jy-data-extract into dev3.3

dataclean/src/config.json

@@ -1,5 +1,5 @@
 {
-    "udpport": "1501",
+    "udpport": ":1501",
     "mongodbs": {
         "qfw": {
             "servers": "192.168.3.207:27081",
@@ -10,8 +10,8 @@
         "extract": {
             "servers": "192.168.3.207:27082",
             "size": 5,
-            "name": "extract_kf",
-            "coll": "result"
+            "name": "mxs",
+            "coll": "bidamount"
         },
         "project": {
             "servers": "192.168.3.207:27081",

dataclean/src/main.go

@@ -6,7 +6,8 @@ import (
 	mu "mfw/util"
 	qu "qfw/util"
 	"qfw/util/mongodb"
-	"service"
+
+	"./service"
 )
 
 func init() {

dataclean/src/service/bidamount.go

@@ -2,17 +2,109 @@
 package service
 
 import (
-	"log"
+	"math"
+	qu "qfw/util"
 )
 
 //返回
 func (d *ResultInfo) BidAmount(data map[string]interface{}) {
-	log.Println("执行：BidAmout")
+	//log.Println("执行：BidAmout")
+	if d.Extract == nil {
+		d.Extract = &ExtractInfo{}
+	}
 	state := false //值未发生改变
 	//todo 清洗操作
+	bidamount_tab := make(map[string]interface{}) //记录每个流程的判断情况 0:无法判断;1:不满足要求;2:满足要求
+	bidamount_by := ""                            //val：file通过字段修正;val：logic通过逻辑判断修正
+	_id := qu.BsonIdToSId(data["_id"])
+	bidamount, _ := data["bidamount"].(float64)           //中标金额
+	result := bidamount                                   //结果
+	budget, _ := data["budget"].(float64)                 //预算
+	supervisorrate, _ := data["supervisorrate"].(float64) //费率
+	//log.Println(bidamount, budget, supervisorrate)
+	//1.注册资本
+	b1, n1 := ClearByRegisterCapital(bidamount, 0, 20.0)
+	//qu.Debug("1111", b1, n1)
+	bidamount_tab["fund"] = n1
+	//2.与预算的比例
+	b2, n2 := ClearByBudretProportion(budget, bidamount)
+	//qu.Debug("2222", b2, n2)
+	bidamount_tab["ratio"] = n2
+	//3.中标金额费率
+	b3, n3, val3 := ClearBySupervisorrate(budget, bidamount, supervisorrate)
+	//qu.Debug("3333", b3, n3, val3)
+	bidamount_tab["rate"] = n3
+
+	if !b1 && !b2 && !b3 { //均不满足要求，更改中标金额 false:1 false:0
+		if val3 > 0 { //首先应用费率算出的值
+			result = val3
+			bidamount_by = "file"
+		} else if n1 == 1 || n2 == 1 {
+			result = 0
+			bidamount_by = "logic"
+		}
+		//qu.Debug("result====", result, bidamount)
+
+		if bidamount != result { //排除抽取数据为空的情况
+			if result == 0 {
+				data["bidamount"] = ""
+			} else {
+				data["bidamount"] = result
+			}
+			data["bidamount_by"] = bidamount_by
+			data["bidamount_tab"] = bidamount_tab
+			if bidamount == 0 {
+				data["bidamount_back"] = ""
+			} else {
+				data["bidamount_back"] = bidamount
+			}
+			//qu.Debug("bidamount---", data["bidamount"], "bidamount_back---", data["bidamount_back"])
+			d.Extract.Data = data
+			d.Extract.Id = _id
+			state = true
+		}
+	}
 
 	if state {
 		d.Flag++
-		d.Extract.UpParam = append(d.Extract.UpParam, "bidamount") //更新属性
+		d.Extract.UpParam = append(d.Extract.UpParam, "bidamount")      //更新属性
+		d.Extract.UpParam = append(d.Extract.UpParam, "bidamount_by")   //更新属性
+		d.Extract.UpParam = append(d.Extract.UpParam, "bidamount_tab")  //更新属性
+		d.Extract.UpParam = append(d.Extract.UpParam, "bidamount_back") //更新属性
+	}
+}
+
+func ClearByRegisterCapital(bidamount, fund float64, thresholdVal float64) (bool, int) { //m为注册资本，thresholdVal为阈值
+	if fund == 0 || bidamount == 0 {
+		return false, 0
+	}
+	if bidamount <= fund*thresholdVal { //中标金额满足20倍的阈值
+		return true, 2
+	} else {
+		return false, 1
+	}
+}
+func ClearByBudretProportion(budget, bidamount float64) (bool, int) {
+	if budget == 0 || bidamount == 0 {
+		return false, 0
+	}
+	if bidamount <= budget*1.1 && bidamount >= budget*0.7 { //数据在范围内
+		return true, 2
+	} else {
+		return false, 1
+	}
+}
+func ClearBySupervisorrate(budget, bidamount, supervisorrate float64) (bool, int, float64) {
+	if budget == 0 || supervisorrate == 0 {
+		return false, 0, 0
+	}
+	tb := budget * math.Pow10(6)
+	ts := supervisorrate * math.Pow10(6)
+	qu.Debug(tb, ts)
+	tmpBidAmount := tb * ts / math.Pow10(12) //预算*费率
+	if bidamount == tmpBidAmount {
+		return true, 2, bidamount
+	} else {
+		return false, 1, tmpBidAmount
 	}
 }

dataclean/src/service/fieldmain.go

@@ -30,6 +30,7 @@ func DataClean(udpInfo map[string]interface{}) {
 	wg := &sync.WaitGroup{}
 	for tmp := make(map[string]interface{}); query.Next(tmp); index++ {
 		//清洗数据流开始
+		wg.Add(1)
 		ClearThread <- true
 		go func(data map[string]interface{}) {
 			defer func() {
@@ -64,9 +65,9 @@ func DataClean(udpInfo map[string]interface{}) {
 
 func Fields(tmp map[string]interface{}) {
 	data := &ResultInfo{}
-	data.BidAmount(tmp)
+	data.BidAmount(tmp) //中标金额清理
 	//其他字段清洗...
-
+	qu.Debug("flag=====", data.Flag)
 	if data.Flag > 0 { //有字段发生变化
 		ResultInfos = append(ResultInfos, data)
 	}

dataclean/src/service/update.go

@@ -104,5 +104,5 @@ func up() {
 		db := Mogdbs["project"]
 		db.Mongodb.UpSertBulk(db.Coll, updateP...)
 	}
-	log.Println("更新完毕", "updateB:", len(updateE), "updateE:", len(updateB), "updateP:", len(updateP))
+	log.Println("更新完毕", "updateB:", len(updateB), "updateE:", len(updateE), "updateP:", len(updateP))
 }

dataclean/src/service/var.go

@@ -28,18 +28,21 @@ type ResultInfo struct {
 	Flag    int          //更新标记
 }
 
+//bidding数据
 type BidInfo struct {
 	Id      string
 	Data    map[string]interface{} //原始信息
 	UpParam []string               //更新属性集
 }
 
+//抽取结果数据
 type ExtractInfo struct {
 	Id      string
 	Data    map[string]interface{} //原始信息
 	UpParam []string               //更新属性集
 }
 
+//项目数据
 type ProjectInfo struct {
 	Id      string
 	Data    map[string]interface{} //原始信息