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

src/res/fieldscore.json

@@ -265,7 +265,7 @@
         "negativewords": [
             {
                 "describe": "包含负分",
-                "regstr": "((中标|候选|成交|代表|投标|代理)人|我公司|定标|通知|异议|法院|要求|代理|详细|test|意见|原因|具体|结果|负责|付款|附件|候选|招标失败|注册表|交易中心|序号内容|不足|公告|变更|采购|招标|废标|废止|流标|中标|投标|评标|开标|供应商|金额|万元|元整|预算|报价|单价|第(\\d|一|二|三|四|五)(名|包)|排名|候选|确定|标段|(标|一|二|三|四|五)包|中选|成交[^通]|包号|(A|B|C|D|E|F|G)包|地址|详情|要求|推荐|名称|评审|得分|合同|平方米|公示期|结果|备注|说明|单位|代表|委托|工作日|营业(执|期)|通过|代码|电话|联系|条件|合理|费率|以上|以下|拟定|注：|\\d[\\s]{0,10}(\\d|元|包|米|平米|平方米|吨|辆|千克|克|毫克|毫升|公升|套|件|瓶|箱|只|台|年|月|日|天|号)|(:|：|；|;|？|￥|\\*|%)|^[a-zA-Z0-9-]{5,100}|^[a-zA-Z0-9-]{1,100}$|[a-zA-Z0-9-]{10,100}\\n)",
+                "regstr": "((中标|候选|成交|代表|投标|代理)人|我公司|定标|通知|异议|法院|要求|代理|详细|test|意见|原因|具体|结果|负责|付款|附件|候选|招标失败|注册表|交易中心|序号内容|不足|公告|变更|采购|招标|废标|废止|流标|投标|评标|开标|供应商|金额|万元|元整|预算|报价|单价|第(\\d|一|二|三|四|五)(名|包)|排名|候选|确定|标段|(标|一|二|三|四|五)包|中选|成交[^通]|包号|(A|B|C|D|E|F|G)包|地址|详情|要求|推荐|名称|评审|得分|合同|平方米|公示期|结果|备注|说明|单位|代表|委托|工作日|营业(执|期)|通过|代码|电话|联系|条件|合理|费率|以上|以下|拟定|注：|\\d[\\s]{0,10}(\\d|元|包|米|平米|平方米|吨|辆|千克|克|毫克|毫升|公升|套|件|瓶|箱|只|台|年|月|日|天|号)|(:|：|；|;|？|￥|\\*|%)|^[a-zA-Z0-9-]{5,100}|^[a-zA-Z0-9-]{1,100}$|[a-zA-Z0-9-]{10,100}\\n)",
                 "score": -20
             },
 			{

standardata/src/config.json

@@ -29,5 +29,7 @@
       "redisdb":3
     }
   },
-  "redis": "winner=172.17.148.44:2679,buyer=172.17.148.44:2679,agency=172.17.148.44:2679"
+  "redis": "winner=172.17.148.44:2679,buyer=172.17.148.44:2679,agency=172.17.148.44:2679",
+  "username": "",
+  "password": ""
 }

standardata/src/main.go

@@ -14,21 +14,21 @@ import (
 )
 
 var (
-	MongoFrom /*抽取原*/, MongoTo /*保存库*/, MongoEnt /*企业库*/, MgoExt/*抽取信息库*/ *mongo.MongodbSim
-	sysconfig                   map[string]interface{}
-	extractcoll                 string
-	Version                     string
-	VersionColl                 string
-	winnerent, winnererr        string
-	buyerent, buyererr          string
-	agencyent, agencyerr        string
-	winnerbd, buyerbd, agencybd int
-	Addrs                       = make(map[string]interface{}, 0) //省市县
-	winchanbool                 = make(chan bool, 3)
-	buyerchanbool               = make(chan bool, 3)
-	agencychanbool              = make(chan bool, 3)
-	gochan                      = make(chan bool, 3)
-	udpclient                   mu.UdpClient
+	MongoFrom /*抽取原*/, MongoTo /*保存库*/, MongoEnt /*企业库*/, MgoExt /*抽取信息库*/ *mongo.MongodbSim
+	sysconfig                                                              map[string]interface{}
+	extractcoll                                                            string
+	Version                                                                string
+	VersionColl                                                            string
+	winnerent, winnererr                                                   string
+	buyerent, buyererr                                                     string
+	agencyent, agencyerr                                                   string
+	winnerbd, buyerbd, agencybd                                            int
+	Addrs                                                                  = make(map[string]interface{}, 0) //省市县
+	winchanbool                                                            = make(chan bool, 3)
+	buyerchanbool                                                          = make(chan bool, 3)
+	agencychanbool                                                         = make(chan bool, 3)
+	gochan                                                                 = make(chan bool, 3)
+	udpclient                                                              mu.UdpClient
 	//异常表正则匹配处理
 	WinnerRegOk, WinnerRegErr, AgencyRegOk, AgencyRegErr, BuyerRegOk, BuyerRegErr []regexp.Regexp
 )
@@ -59,6 +59,8 @@ func init() {
 		MongodbAddr: qu.ObjToString(sysconfig["mgofrom"]),
 		Size:        qu.IntAll(sysconfig["mgofromsize"]),
 		DbName:      qu.ObjToString(sysconfig["mgofromdb"]),
+		UserName:    qu.ObjToString(sysconfig["username"]),
+		PassWord:    qu.ObjToString(sysconfig["password"]),
 	}
 	MongoFrom.InitPool()
 
@@ -66,6 +68,8 @@ func init() {
 		MongodbAddr: qu.ObjToString(sysconfig["mgoto"]),
 		Size:        qu.IntAll(sysconfig["mgotosize"]),
 		DbName:      qu.ObjToString(sysconfig["mgotodb"]),
+		UserName:    qu.ObjToString(sysconfig["username"]),
+		PassWord:    qu.ObjToString(sysconfig["password"]),
 	}
 	MongoTo.InitPool()
 
@@ -73,6 +77,8 @@ func init() {
 		MongodbAddr: qu.ObjToString(sysconfig["mgoent"]),
 		Size:        qu.IntAll(sysconfig["mgoentsize"]),
 		DbName:      qu.ObjToString(sysconfig["mgoentdb"]),
+		UserName:    qu.ObjToString(sysconfig["username"]),
+		PassWord:    qu.ObjToString(sysconfig["password"]),
 	}
 	MongoEnt.InitPool()
 
@@ -80,6 +86,8 @@ func init() {
 		MongodbAddr: qu.ObjToString(sysconfig["mgoto"]),
 		Size:        qu.IntAll(sysconfig["mgotosize"]),
 		DbName:      qu.ObjToString(sysconfig["extractdb"]),
+		UserName:    qu.ObjToString(sysconfig["username"]),
+		PassWord:    qu.ObjToString(sysconfig["password"]),
 	}
 	MgoExt.InitPool()
 	redis.InitRedis(qu.ObjToString(sysconfig["redis"]))

udpfilterdup/src/dataMethodMerge.go

@@ -185,6 +185,7 @@ func mergeDataFieldsArr(source *Info, info *Info) (*Info, []int64, bool) {
 		mergeArr = append(mergeArr, 6)
 		is_replace = true
 	}
+
 	//7、中标金额
 	if source.bidamount == 0 && info.bidamount != 0 {
 		merge_recordMap["bidamount"] = source.bidamount

udpfusion/src/config.json

@@ -3,11 +3,11 @@
   "mongodb": {
     "addrName": "192.168.3.207:27092",
     "dbName": "zhengkun",
-    "collName": "all_test",
+    "collName": "fusion_test",
     "pool": 10,
     "site": {
-      "site_dbname": "qfw",
-      "site_coll": "site"
+      "dbname": "zhengkun",
+      "coll": "site"
     }
   },
   "notFusionKey": {

udpfusion/src/main.go

@@ -2,13 +2,14 @@ package main
 
 import (
 	"encoding/json"
+	"go.mongodb.org/mongo-driver/bson/primitive"
 	"log"
 	mu "mfw/util"
 	"net"
 	"os"
 	"qfw/common/src/qfw/util"
 	qu "qfw/util"
-	"strconv"
+	"sync"
 	"time"
 )
 
@@ -24,11 +25,12 @@ var (
 	NoNeedFusionKey 	map[string]interface{}   	//不需要融合的key
 	UpdateFusion		*updateFusionInfo
 	UpdateRecord		*updateRecordInfo			//更新池
+	siteJsonData		map[string]string			//站点池
 )
 
 
 
-func initMgo()  {
+func initMgoAndSite()  {
 	mconf := sysconfig["mongodb"].(map[string]interface{})
 	log.Println(mconf)
 	mgo = &MongodbSim{
@@ -45,13 +47,25 @@ func initMgo()  {
 	NoNeedFusionKey = sysconfig["notFusionKey"].(map[string]interface{})
 
 
+	site := mconf["site"].(map[string]interface{})
+	siteJsonData = make(map[string]string, 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); {
+		siteJsonData[util.ObjToString(site_dict["site"])] = util.ObjToString(site_dict["sitetype"])
+	}
+	log.Printf("new站点加载用时：%d秒,%d个\n", int(time.Now().Unix())-start, len(siteJsonData))
 }
 
 
+
+
 func init() {
 	//加载配置文件
 	qu.ReadConfig(&sysconfig)
-	initMgo()
+	initMgoAndSite()
 
 	//更新池
 	UpdateFusion = newUpdateFusionPool()
@@ -62,6 +76,8 @@ func init() {
 
 
 
+
+
 	log.Println("采用udp模式")
 }
 
@@ -78,8 +94,9 @@ func mainT() {
 //快速测试使用
 func main() {
 
-	sid := "1f0000000000000000000000"
-	eid := "9f0000000000000000000000"
+
+	sid := "100000000000000000000000"
+	eid := "900000000000000000000000"
 	//log.Println(sid, "---", eid)
 	mapinfo := map[string]interface{}{}
 	if sid == "" || eid == "" {
@@ -100,12 +117,7 @@ func main() {
 
 //融合具体方法
 func startTask(data []byte, mapInfo map[string]interface{}) {
-
-	//遍历数据
 	log.Println("开始融合流程")
-
-
-
 	defer qu.Catch()
 	//区间id
 	q := map[string]interface{}{
@@ -119,8 +131,14 @@ func startTask(data []byte, mapInfo map[string]interface{}) {
 	defer mgo.DestoryMongoConn(sess)
 	it := sess.DB(mgo.DbName).C(coll_name).Find(&q).Iter()
 	//编译不同的融合组,如何划分组
-	fusionDataGroupArr := make([][]string,0) //待融合组
-	addOrUpdateArr := make([]bool,0) //新增-bool-记录
+	/***********************/
+	/***********************/
+	/***y
+	********************/
+	/***********************/
+	fusionDataGroupArr := make([][]string,0) 			//待融合组
+	addOrUpdateArr := make([]bool,0) 					//新增-bool-记录-组新增，组更新
+	infoFusionArr := make([]map[string]interface{},0) 	//记录取融合表的数据
 
 	repeatArr,sourceArr,index := make([]string,0),make([]string,0),0 //重复数据组
 	for tmp := make(map[string]interface{}); it.Next(&tmp); index++ {
@@ -136,14 +154,14 @@ func startTask(data []byte, mapInfo map[string]interface{}) {
 		}else {
 			fusionDataGroupArr = append(fusionDataGroupArr,[]string{tmpId})
 			addOrUpdateArr = append(addOrUpdateArr,false)
+			infoFusionArr = append(infoFusionArr, map[string]interface{}{})
 		}
 		tmp = make(map[string]interface{})
 	}
 
 	log.Println("task first:",index,len(fusionDataGroupArr),"+",len(repeatArr))
-	log.Println("状态记录：",len(addOrUpdateArr))
+
 	//根据重复组，重新划分新的组别
-	num1,num2:=0,0
 	for i:=0;i<len(repeatArr);i++ {
 		sourceid := sourceArr[i]
 		isAddExist,index := false,0
@@ -163,92 +181,172 @@ func startTask(data []byte, mapInfo map[string]interface{}) {
 			arr = fusionDataGroupArr[index]
 			arr = append(arr,repeatArr[i])//组拼接当前id
 			fusionDataGroupArr[index] = arr
-			num1++
 		}else {//当前段落未找到-需要查询融合表，,遍历融合表
-			arr := make([]string,0)
-			arr = dealWithFindFusionDataArr(sourceid)
+			arr,fusionTmpData := make([]string,0),make(map[string]interface{},0)
+			arr,fusionTmpData = dealWithFindFusionDataArr(sourceid)
+
 			arr = append(arr,repeatArr[i])//组拼接当前id
-			if len(arr)<1 {
-				log.Println("数据异常,融合表找不到数据",repeatArr[i])
-			}else { //新增
-				log.Println("数据融合新增")
+			if len(arr)<1 { //异常错误，新增
+				log.Println("... ... 数据异常异常,融合表，当前组均找不到数据",repeatArr[i])
+				arr_error := make([]string,0)
+				arr_error = append(arr_error,repeatArr[i])//组拼接当前id
+				fusionDataGroupArr = append(fusionDataGroupArr,arr_error)
+				addOrUpdateArr = append(addOrUpdateArr,false)
+				infoFusionArr = append(infoFusionArr, map[string]interface{}{})
+			}else { //正常更新
 				fusionDataGroupArr = append(fusionDataGroupArr,arr)
 				addOrUpdateArr = append(addOrUpdateArr,true)
+				infoFusionArr = append(infoFusionArr,fusionTmpData)
 			}
-			num2++
 
 		}
 		//不断改变中
-		log.Println("当前分组数量：",len(fusionDataGroupArr))
+		//log.Println("当前分组数量：",len(fusionDataGroupArr))
 	}
+	log.Println("最终待融合分组数量：",len(fusionDataGroupArr))
+	log.Println("********************分割线********************")
+	log.Println("********************分割线********************")
+	log.Println("********************分割线********************")
+	log.Println("开始处理新增分组... ...")
 
-	log.Println("分组完毕：","重复新增数量：",num1,"重复更新数量：",num2,len(repeatArr))
-	log.Println("最终带融合分组：",len(fusionDataNewGroupArr))
-	//分组细节需要修改 - 带测试
-	return
 
 
-	log.Println("开始处理新增分组... ...")
 	start := int(time.Now().Unix())
-	//进行分组融合
-	for i:=0;i<len(fusionDataNewGroupArr);i++ {
-		fusionArr := fusionDataNewGroupArr[i]
-		//构建数据
-		log.Println("构建第一组数据...",fusionArr)
-		weight :=NewWeightData(fusionArr)
-		//整理数据-筛选排名，模板
-		weight.analyzeBuildStandardData()
-		if len(fusionArr)<=1 {
-			//更新数据（融合表）   日志数据（日志记录表）
-		 	//updateData,_ := weight.dealWithAddFusionStruct()
-			//log.Println("新增：更新数据",len(updateData))
-		 	//mgo.Save(fusion_coll_name,updateData) //新增
-		}else {
-			//updateData,_ := weight.dealWithMultipleFusionStruct()
-			//log.Println("多组新增：更新数据",len(updateData))
-			//mgo.Save(fusion_coll_name,updateData)
-		}
-	}
-	log.Println("新增融合over ：",len(fusionDataNewGroupArr),"用时：",int(time.Now().Unix())-start)
 
+	//多线程 - 处理数据
+	pool := make(chan bool, 3)
+	wg := &sync.WaitGroup{}
+
+	for i:=0;i<len(fusionDataGroupArr);i++ {
+		fusionArr := fusionDataGroupArr[i]
+		pool <- true
+		wg.Add(1)
+		go func(fusionArr []string,i int) {
+			defer func() {
+				<-pool
+				wg.Done()
+			}()
+			//构建数据
+			log.Println("构建第",i+1,"组数据...","数量：",len(fusionArr),fusionArr)
+			weight :=NewWeightData(fusionArr)
+			////整理数据-筛选排名，模板
+			weight.analyzeBuildStandardData()
+
+			if len(fusionArr)<=1 {
+				//log.Println("单组生成... ...")
+				saveFusionData,saveRecordData := weight.dealWithAddFusionStruct()
+				saveid:=mgo.Save(fusion_coll_name,saveFusionData)
+				saveRecordData["_id"] = saveid
+				mgo.Save(record_coll_name,saveRecordData)
+			}else {
+				if addOrUpdateArr[i] {
+					//log.Println("多组更新... ...")
+					tmpdata:=infoFusionArr[i]
+					updateFusionData,updateRecordData := weight.dealWithMultipleUpdateFusionStruct(tmpdata)
+
+					UpdateFusion.updatePool <- []map[string]interface{}{
+						map[string]interface{}{
+							"_id": tmpdata["_id"],
+						},
+						updateFusionData,
+					}
+					UpdateRecord.updatePool <- []map[string]interface{}{
+						map[string]interface{}{
+							"_id": tmpdata["_id"],
+						},
+						updateRecordData,
+					}
+				}else {
+					//log.Println("多组生成... ...")
+					saveFusionData,saveRecordData := weight.dealWithMultipleAddFusionStruct()
+					saveid:=mgo.Save(fusion_coll_name,saveFusionData)
+					saveRecordData["_id"] = saveid
+					mgo.Save(record_coll_name,saveRecordData)
+				}
+			}
 
-	//多组-融合表更新
-	//UpdateFusion.updatePool <- []map[string]interface{}{//原始数据打标签
-	//	map[string]interface{}{},
-	//	updateData,
-	//}
 
+		}(fusionArr,i)
 
-	time.Sleep(30 * time.Second)
 
+
+	}
+
+	wg.Wait()
+
+	log.Println("fusion is over ：",len(fusionDataGroupArr),"用时：",int(time.Now().Unix())-start,"秒")
+	log.Println("睡眠30秒，然后在发广播")
+	time.Sleep(30 * time.Second)
 	//任务完成，开始发送广播通知下面节点
 	taskSendFusionUdp(mapInfo)
 
 }
 
 //查询融合表数据-找到对应组id
-func dealWithFindFusionDataArr(sourceid string) []string {
-	arr := make([]string,0)
+func dealWithFindFusionDataArr(sourceid string) ([]string,map[string]interface{}) {
+	newArr ,arr := make([]string,0),make(primitive.A,0)
+	tmpData:=make(map[string]interface{},0)
 	q := map[string]interface{}{}
 	sess := mgo.GetMgoConn()
 	defer mgo.DestoryMongoConn(sess)
-	it := sess.DB(mgo.DbName).C(coll_name).Find(&q).Iter()
+	it := sess.DB(mgo.DbName).C(fusion_coll_name).Find(&q).Iter()
+
 	for tmp := make(map[string]interface{}); it.Next(&tmp); {
-		fusion_allids := tmp["fusion_allids"].([]string)
-		for _,v:=range fusion_allids {
-			if v==sourceid {
-				//找到目标组-
-				arr = fusion_allids
-				tmp = make(map[string]interface{})
-				break
+		//log.Println(reflect.TypeOf(tmp["fusion_allids"]))
+		if fusion_allids,b := tmp["fusion_allids"].(primitive.A);b {
+			for _,v:=range fusion_allids {
+				if v==sourceid {
+					//找到目标组-
+					arr = fusion_allids
+					tmpData = tmp
+					tmp = make(map[string]interface{})
+					break
+				}
 			}
 		}
+
 		tmp = make(map[string]interface{})
 	}
-	return arr
+
+	for _,v:=range  arr{
+		newArr = append(newArr,qu.ObjToString(v))
+	}
+
+	return newArr,tmpData
 }
 
+//查询记录1表数据-找到对应的id ， 更新用到
+func dealWithFindRecordData(sourceid string) string {
+	newArr ,arr := make([]string,0),make(primitive.A,0)
+	//tmpData:=make(map[string]interface{},0)
+	q := map[string]interface{}{}
+	sess := mgo.GetMgoConn()
+	defer mgo.DestoryMongoConn(sess)
+	it := sess.DB(mgo.DbName).C(fusion_coll_name).Find(&q).Iter()
 
+	for tmp := make(map[string]interface{}); it.Next(&tmp); {
+		//log.Println(reflect.TypeOf(tmp["fusion_allids"]))
+		if fusion_allids,b := tmp["fusion_allids"].(primitive.A);b {
+			for _,v:=range fusion_allids {
+				if v==sourceid {
+					//找到目标组-
+					arr = fusion_allids
+					//tmpData = tmp
+					tmp = make(map[string]interface{})
+					break
+				}
+			}
+		}
+
+		tmp = make(map[string]interface{})
+	}
+
+	for _,v:=range  arr{
+		newArr = append(newArr,qu.ObjToString(v))
+	}
+
+	return ""
+}
 
 //udp 监听
 func processUdpMsg(act byte, data []byte, ra *net.UDPAddr) {
@@ -282,7 +380,7 @@ func processUdpMsg(act byte, data []byte, ra *net.UDPAddr) {
 	}
 }
 
-
+//结束发送udp
 func taskSendFusionUdp(mapinfo map[string]interface{})  {
 
 	//log.Println("信息融合结束-发送udp")
@@ -314,14 +412,5 @@ func taskSendFusionUdp(mapinfo map[string]interface{})  {
 
 
 
-//判断是否在当前id段落
-func judgeIsCurIds (gtid string,lteid string,curid string) bool {
 
-	gt_time, _ := strconv.ParseInt(gtid[:8], 16, 64)
-	lte_time, _ := strconv.ParseInt(lteid[:8], 16, 64)
-	cur_time, _ := strconv.ParseInt(curid[:8], 16, 64)
-	if cur_time>gt_time&&cur_time<=lte_time {
-		return true
-	}
-	return false
-}
+

udpfusion/src/updateFusion.go

@@ -20,13 +20,14 @@ type updateFusionInfo struct {
 var sp_f = make(chan bool, 5)
 
 func newUpdateFusionPool() *updateFusionInfo {
-	update:=&updateFusionInfo{make(chan []map[string]interface{}, 5000),100}
+	update:=&updateFusionInfo{make(chan []map[string]interface{}, 50000),100}
 	return update
 }
 
 
 func (update *updateFusionInfo) updateFusionData() {
-	log.Println("开始不断监听--待更新数据")
+	log.Println("监听--融合更新数据")
+
 	tmpArr := make([][]map[string]interface{}, update.saveSize)
 	tmpIndex := 0
 	for {

udpfusion/src/updateRecord.go

@@ -26,7 +26,7 @@ func newUpdateRecordPool() *updateRecordInfo {
 
 
 func (update *updateRecordInfo) updateRecordData() {
-	log.Println("开始不断监听--待更新数据")
+	log.Println("监听日志--待更新数据")
 	tmpArr := make([][]map[string]interface{}, update.saveSize)
 	tmpIndex := 0
 	for {

udpfusion/src/weightFusion.go

@@ -2,10 +2,9 @@ package main
 
 import (
 	"fmt"
-	"log"
+	"go.mongodb.org/mongo-driver/bson/primitive"
 	qu "qfw/util"
 	"time"
-	"go.mongodb.org/mongo-driver/bson/primitive"
 )
 
 //处理融合数据-返回，融合新数据数据-新增
@@ -13,10 +12,6 @@ func (weight *weightDataMap) dealWithAddFusionStruct ()(map[string]interface{},m
 
 	//指定模板数据dict-单条数据
 	dict :=weight.data[weight.templateid].data
-
-	//采用新增id
-	delete(dict,"_id")
-
 	//最早发布时间 （小）
 	dict["early_publishtime"] = qu.IntAll(dict["publishtime"])
 	//最近发布时间  (大)
@@ -34,64 +29,131 @@ func (weight *weightDataMap) dealWithAddFusionStruct ()(map[string]interface{},m
 	//融合保存相关联ids
 	dict["fusion_saveids"] = weight.saveids
 
+	//采用新增id
+	delete(dict,"_id")
+
+
+	recordDict := make(map[string]interface{},0)
+	recordDict["1"] = map[string]interface{}{
+		"data":map[string]interface{}{},
+		"snapshot":map[string]interface{}{},
+	}
+	recordDict["number"] = qu.Int64All(1)
 
 
-	return dict,dict
+	return dict,recordDict
 }
 
-//处理多条融合数据-返回融合新数据，融合细节数据
-func (weight *weightDataMap) dealWithMultipleFusionStruct ()(map[string]interface{},map[string]interface{}){
+//处理多条融合数据 - 多组新增类
+func (weight *weightDataMap) dealWithMultipleAddFusionStruct ()(map[string]interface{},map[string]interface{}){
 
 	//指定模板数据dict
 	dict :=weight.data[weight.templateid].data
-
 	//最早|近发布时间
 	dict["early_publishtime"],dict["lately_publishtime"] = weight.dealWithTimeData("publishtime")
 	//最早|近入库时间
 	dict["early_comeintime"],dict["lately_comeintime"] = weight.dealWithTimeData("comeintime")
+	//所有相关联ids
+	dict["fusion_allids"] = weight.allids
+	//融合保存相关联ids
+	dict["fusion_saveids"] = weight.saveids
+
+	//日志记录
+	recordDict := make(map[string]interface{},0)
+	//结构体字段逻辑处理
+	structData := weight.dealWithStructData(&recordDict)
+	for k,v:=range structData {
+		dict[k] = v
+	}
+	//非空新增字段
+	otherFieldData := weight.dealWithOtherFieldData(&recordDict)
+	for k,v:=range otherFieldData {
+		dict[k] = v
+	}
+
+	//if len(recordDict)>0 {
+	//	log.Println("有更新数据:",len(recordDict))
+	//}
+
 	//当前更新时间
-	dict["current_updatetime"] = qu.IntAll(time.Now().Unix())
+	dict["fusion_time"] = qu.IntAll(time.Now().Unix())
+	//融合生成时间
+	dict["fusion_updatetime"] = qu.IntAll(time.Now().Unix())
+
+
+	//采用新增id
+	delete(dict,"_id")
+
+	newRecordDict := make(map[string]interface{},0)
+	newRecordDict["1"] = map[string]interface{}{
+		"data":recordDict,
+		"snapshot":map[string]interface{}{},
+	}
+	newRecordDict["number"] = qu.Int64All(1)
 
-	//融合生成时间-取融合表融合数据
-	dict["current_updatetime"] = ""
+	//返回，更新数据，日志记录数据
+	return dict,newRecordDict
+}
+
+//处理多条融合数据-返回融合新数据，融合更新细节数据
+func (weight *weightDataMap) dealWithMultipleUpdateFusionStruct (tmpData map[string]interface{})(map[string]interface{},map[string]interface{}){
+
+	//指定模板数据dict
+	dict :=weight.data[weight.templateid].data
+	//最早|近发布时间
+	dict["early_publishtime"],dict["lately_publishtime"] = weight.dealWithTimeData("publishtime")
+	//最早|近入库时间
+	dict["early_comeintime"],dict["lately_comeintime"] = weight.dealWithTimeData("comeintime")
 	//所有相关联ids
 	dict["fusion_allids"] = weight.allids
 	//融合保存相关联ids
 	dict["fusion_saveids"] = weight.saveids
 
 
+
 	//日志记录-还有快照页面 等等
 	recordDict := make(map[string]interface{},0)
-
-
 	//结构体字段逻辑处理
 	structData := weight.dealWithStructData(&recordDict)
 	for k,v:=range structData {
-		log.Println("key:",k,"value",v)
-		dict["k"] = v
+		dict[k] = v
 	}
-
 	//非空新增字段
 	otherFieldData := weight.dealWithOtherFieldData(&recordDict)
 	for k,v:=range otherFieldData {
-		//log.Println("key:",k,"value",v)
 		dict[k] = v
 	}
 
-	dict["repeat"] = 0
-
-
-	//log.Println("待更新数据：",dict)
-	//log.Println("待更新日志：",recordDict)
+	//if len(recordDict)>0 {
+	//	log.Println("有更新数据:",len(recordDict))
+	//}
 
 
+	//融合生成时间-取融合表第一次融合时间
+	dict["fusion_time"] = tmpData["fusion_time"]
+	//当前更新时间
+	dict["fusion_updatetime"] = qu.IntAll(time.Now().Unix())
+	//删除_id
+	delete(dict,"_id")
 
+	newRecordDict := make(map[string]interface{},0)
+	newRecordDict = mgo.FindById(record_coll_name,BsonTOStringId(tmpData["_id"]))
+	number:=qu.Int64All(newRecordDict["number"])
+	number++
+	key:=fmt.Sprintf("%d",number)
+	newRecordDict[key] = map[string]interface{}{
+		"data":recordDict,
+		"snapshot":tmpData,
+	}
+	newRecordDict["number"] = number
 
 	//返回，更新数据，日志记录数据
-	return dict,recordDict
+	return dict,newRecordDict
 }
 
 
+
+
 //处理其他字段数据
 func (weight *weightDataMap)dealWithOtherFieldData(recordDict *map[string]interface{}) map[string]interface{} {
 
@@ -143,7 +205,7 @@ func (weight *weightDataMap)dealWithOtherFieldData(recordDict *map[string]interf
 	}
 
 
-	log.Println("待替换key:",arr,"修改后:",modifyData)
+	//log.Println("待替换key:",arr,"修改后:",modifyData)
 
 	//第二步-集合最大化
 	isRank := 2
@@ -161,7 +223,7 @@ func (weight *weightDataMap)dealWithOtherFieldData(recordDict *map[string]interf
 						continue
 					}
 					if judgeIsEffectiveData(newValue,key) {
-						log.Println("最大化有效-",key)
+						//log.Println("最大化有效-",key)
 						templateTmp[key] = newValue
 						modifyData[key] = newValue
 						(*recordDict)[key] = map[string]interface{}{
@@ -179,7 +241,7 @@ func (weight *weightDataMap)dealWithOtherFieldData(recordDict *map[string]interf
 		}
 	}
 
-	log.Println("isRank:",isRank,len(modifyData))
+	//log.Println("isRank:",isRank,len(modifyData))
 
 	return modifyData
 }
@@ -215,7 +277,7 @@ func (weight *weightDataMap)dealWithStructData(recordDict *map[string]interface{
 	if tmp_arr,b := templateTmp["attach_text"].(map[string]interface{});b {
 		//有值符合-
 		attach_text = tmp_arr
-		log.Println("默认初始：",attach_text)
+		//log.Println("默认初始：",attach_text)
 	}
 	//附件判重-并合并新增
 	keyIndex := -1
@@ -226,7 +288,7 @@ func (weight *weightDataMap)dealWithStructData(recordDict *map[string]interface{
 			keyIndex = key
 		}
 	}
-	log.Println("当前keyIndex",keyIndex)
+	//log.Println("当前keyIndex",keyIndex)
 	for _,value_id :=range weight.saveids {
 		if templateid == value_id {
 			continue
@@ -236,13 +298,13 @@ func (weight *weightDataMap)dealWithStructData(recordDict *map[string]interface{
 			if len(attachData)>0  { //有值
 				for _,v:=range attachData { //子元素
 					if attach,isOK := v.(map[string]interface{});isOK {
-						log.Println(attach)
+						//log.Println(attach)
 						if !dealWithRepeatAttachData(attach_text,attach) {
 							//符合条件-不重复直接添加
 							keyIndex++
 							saveKey := fmt.Sprintf("%v",keyIndex)
 							attach_text[saveKey] = attach //key累加
-							log.Println(attach_text)
+							//log.Println(attach_text)
 							isAttach = true
 
 							//多条情况-融合
@@ -338,3 +400,4 @@ func (weight *weightDataMap)dealWithStructData(recordDict *map[string]interface{
 	return modifyData
 }
 
+

udpfusion/src/weightFusionMethod.go

@@ -3,6 +3,7 @@ package main
 import (
 	qu "qfw/util"
 	"reflect"
+	"strconv"
 )
 
 //处理-重复附件数据
@@ -131,6 +132,11 @@ func judgeIsEffectiveData(value interface{},key string) bool  {
 		return true
 	}
 
+	if valueType==reflect.Bool {
+		return true
+	}
+
+
 	//其他类型采用
 	valueLen := reflect.ValueOf(value).Len()
 	if valueLen>0 {
@@ -142,4 +148,17 @@ func judgeIsEffectiveData(value interface{},key string) bool  {
 	}
 
 
+}
+
+
+//判断是否在当前id段落
+func judgeIsCurIds (gtid string,lteid string,curid string) bool {
+
+	gt_time, _ := strconv.ParseInt(gtid[:8], 16, 64)
+	lte_time, _ := strconv.ParseInt(lteid[:8], 16, 64)
+	cur_time, _ := strconv.ParseInt(curid[:8], 16, 64)
+	if cur_time>gt_time&&cur_time<=lte_time {
+		return true
+	}
+	return false
 }

udpfusion/src/weightValue.go

@@ -1,10 +1,5 @@
 package main
 
-import (
-	"log"
-	"sync"
-)
-
 type weightInfo struct {
 	maxLevel 		bool
 	minLevel		bool
@@ -16,7 +11,6 @@ type weightInfo struct {
 
 
 type weightDataMap struct {
-	lock   sync.Mutex //锁
 	data   map[string]*weightInfo
 	allids    []string
 	saveids    []string
@@ -24,9 +18,8 @@ type weightDataMap struct {
 }
 
 func NewWeightData(arr []string) *weightDataMap {
-	//测试-默认第一个
 
-	weight := &weightDataMap{sync.Mutex{},map[string]*weightInfo{},[]string{},[]string{},""}
+	weight := &weightDataMap{map[string]*weightInfo{},[]string{},[]string{},""}
 
 	data := make(map[string]*weightInfo,0)
 	for _,v:=range arr {
@@ -36,25 +29,6 @@ func NewWeightData(arr []string) *weightDataMap {
 		}
 	}
 
-	//测试模拟分数
-	//weight := &weightDataMap{sync.Mutex{},map[string]*weightInfo{},[]string{},[]string{},templateid}
-	//data := make(map[string]*weightInfo,0)
-	//max :=[]bool{false,false,false,false,false,false,false,false,false,false}
-	//min :=[]bool{false,false,false,false,false,false,false,false,false,false}
-	//site :=[]int{2,1,5,3,4,2,3,5,1,0}
-	//qua :=[]int{15,11,11,11,22,19,22,44,22,66}
-	//rank :=[]int{-1,-1,-1,-1,-1,-1,-1,-1,-1,-1}
-	//for k,v:=range arr {
-	//	data[v] = &weightInfo{
-	//		max[k],
-	//		min[k],
-	//		site[k],
-	//		qua[k],
-	//		rank[k],
-	//	}
-	//}
-
-
 	weight.data = data
 
 	return weight
@@ -75,9 +49,9 @@ func analyzeTheSoureData(tmp map[string]interface{}) *weightInfo {
 
 
 	//测试 指定模板-数据-最高权重
-	if BsonTOStringId(tmp["_id"])=="5638baccaf53745d9a000998" {
-		maxLevel = true
-	}
+	//if BsonTOStringId(tmp["_id"])=="602686f5f021652bdea41ea1" {
+	//	maxLevel = true
+	//}
 
 	delete(tmp,"_id")
 
@@ -94,7 +68,7 @@ func analyzeTheSoureData(tmp map[string]interface{}) *weightInfo {
 //分析模板数据-打标记构建数据结构
 func (weight *weightDataMap) analyzeBuildStandardData() {
 
-	weight.lock.Lock()
+
 	//分析里面的打分，以及是否参与融合来决定
 	data:=weight.data
 	//先构建
@@ -131,9 +105,8 @@ func (weight *weightDataMap) analyzeBuildStandardData() {
 		}
 	}
 	rank_s :=1
-
 	if len(isMaxIndexArr)>=1 {
-		log.Println("进行最大权重...")
+		//log.Println("进行最大权重...")
 		rankIndexArr := dealWithGroupScores(isMaxIndexArr,isMaxIndexValueArr,arrSiteLevel)
 		for _,v:=range rankIndexArr {
 			arrRanking[v] = rank_s
@@ -193,6 +166,6 @@ func (weight *weightDataMap) analyzeBuildStandardData() {
 	weight.allids = arrAllIds
 	weight.saveids = arrSaveIds
 
-	weight.lock.Unlock()
+
 }
 

udpfusion/src/weightValueMethod.go

@@ -1,8 +1,7 @@
 package main
-
 import (
-	"math/rand"
-	"time"
+	qu "qfw/util"
+
 )
 
 func dealWithGroupScores(indexArr []int, scoreArr []int,siteArr []int) []int {
@@ -80,19 +79,314 @@ func sortGroupInt(arrValue []int,arrIndex []int) ([]int ,[]int){
 //分析站点评分
 func analyzeTheSite(tmp map[string]interface{}) int {
 	/*
-		站点评分1-5级
+		站点评分1-N级
+		5、政府采购
+		4、公共资源
+		3、人民政府/政府机构/学校/医院/公司官方网站
+		2、社会公共招标平台/企业公共平台
+		1、其他
+		0、""
+
 	*/
-	//测试随机分
-	rand.Seed(time.Now().UnixNano()) //以当前系统时间作为种子参数
-	return rand.Intn(10)
+	level := 0
+	site := qu.ObjToString(tmp["site"])
+	if site == "政府采购" {
+		level = 5
+	}else if site == "公共资源"  {
+		level = 4
+	}else if site=="人民政府"||site=="政府机构"||site=="学校"||site=="医院"||site=="公司官方网站"  {
+		level = 3
+	}else if site == "社会公共招标平台"||site == "企业公共平台"  {
+		level = 2
+	}else if site == "其他"  {
+		level = 1
+	}else {
+
+	}
+	return level
 }
 
 //分析要素评分
 func analyzeTheElements(tmp map[string]interface{}) int {
+	//质量评分总分
+	score,subtype := 0,qu.ObjToString(tmp["subtype"])
+	if subtype=="预告"||subtype=="招标"||subtype=="邀标"||subtype=="询价"||
+		subtype=="竞谈"||subtype=="竞价"||subtype=="单一"||subtype=="变更"||subtype=="其它"{
+		score = dealWithCallBidTypeScore(tmp)
+	}else if subtype=="中标"||subtype=="成交"{
+		score = dealWithWinBidTypeScore(tmp)
+	}else if subtype=="废标"||subtype=="流标"{
+		score = dealWithFailBidTypeScore(tmp)
+	}else if subtype=="合同"{
+		score = dealWithContractTypeScore(tmp)
+	}else {
+
+	}
+	return score
+}
+
+//招标类型
+func dealWithCallBidTypeScore(tmp map[string]interface{}) int {
+	/*
+	预告、招标类型：
+	公告类型：预告、招标、邀标、询价、竞谈、竞价、单一、变更、其它
+	2分：项目名称、采购单位
+	1分：代理机构、预算、项目编号、省份、城市
+	0.5分：采购单位联系人、采购单位电话、采购单位地址、采购单位邮编、招标代理机构联系人、招标代理机构联系方式、招标代理机构地址、投标截止日期、开标日期
+	*/
+	score :=0
+	//projectname buyer
+	projectname := qu.ObjToString(tmp["projectname"])
+	if tmp["projectname"]!=nil && projectname !="" {
+		score = score+4
+	}
+	buyer := qu.ObjToString(tmp["buyer"])
+	if tmp["buyer"]!=nil && buyer !="" {
+		score = score+4
+	}
+
+	//projectcode agency area city budget
+	projectcode := qu.ObjToString(tmp["projectcode"])
+	if tmp["projectcode"]!=nil && projectcode !="" {
+		score = score+2
+	}
+	agency := qu.ObjToString(tmp["agency"])
+	if tmp["agency"]!=nil && agency !="" {
+		score = score+2
+	}
+	area := qu.ObjToString(tmp["area"])
+	if tmp["area"]!=nil && area !="" {
+		score = score+2
+	}
+	city := qu.ObjToString(tmp["city"])
+	if tmp["city"]!=nil && city !="" {
+		score = score+2
+	}
+	budget := qu.Float64All(tmp["budget"])
+	if tmp["budget"]!=nil && budget !=0.0 {
+		score = score+2
+	}
+
+	score = score+dealWithLowScoreData(tmp)
+
+
+	return score
+}
+//中标类型
+func dealWithWinBidTypeScore(tmp map[string]interface{}) int {
+	/*
+	中标结果类型：
+	公告类型：中标、成交、
+	2分：项目名称、采购单位、中标单位
+	1分：代理机构、中标金额、项目编号、省份、城市、预算
+	0.5分：采购单位联系人、采购单位电话、采购单位地址、采购单位邮编、招标代理机构联系人、招标代理机构联系方式、招标代理机构地址、投标截止日期、开标日期
+	*/
+	score :=0
+	//projectname buyer winner
+	projectname := qu.ObjToString(tmp["projectname"])
+	if tmp["projectname"]!=nil && projectname !="" {
+		score = score+4
+	}
+	buyer := qu.ObjToString(tmp["buyer"])
+	if tmp["buyer"]!=nil && buyer !="" {
+		score = score+4
+	}
+	winner := qu.ObjToString(tmp["winner"])
+	if tmp["winner"]!=nil && winner !="" {
+		score = score+4
+	}
+
+	//projectcode agency area city budget bidamount
+	projectcode := qu.ObjToString(tmp["projectcode"])
+	if tmp["projectcode"]!=nil && projectcode !="" {
+		score = score+2
+	}
+	agency := qu.ObjToString(tmp["agency"])
+	if tmp["agency"]!=nil && agency !="" {
+		score = score+2
+	}
+	area := qu.ObjToString(tmp["area"])
+	if tmp["area"]!=nil && area !="" {
+		score = score+2
+	}
+	city := qu.ObjToString(tmp["city"])
+	if tmp["city"]!=nil && city !="" {
+		score = score+2
+	}
+	budget := qu.Float64All(tmp["budget"])
+	if tmp["budget"]!=nil && budget !=0.0 {
+		score = score+2
+	}
+	bidamount := qu.Float64All(tmp["bidamount"])
+	if tmp["bidamount"]!=nil && bidamount !=0.0 {
+		score = score+2
+	}
+
+	score = score+dealWithLowScoreData(tmp)
+
+
+	return score
+}
+
+func dealWithFailBidTypeScore(tmp map[string]interface{}) int {
+	/*
+	流标废标类型：
+	公告类型：废标、流标
+	2分：项目名称、采购单位
+	1分：代理机构、项目编号、省份、城市、预算
+	0.5分：采购单位联系人、采购单位电话、采购单位地址、采购单位邮编、招标代理机构联系人、招标代理机构联系方式、招标代理机构地址、投标截止日期、开标日期
+	*/
+	score :=0
+	//projectname buyer
+	projectname := qu.ObjToString(tmp["projectname"])
+	if tmp["projectname"]!=nil && projectname !="" {
+		score = score+4
+	}
+	buyer := qu.ObjToString(tmp["buyer"])
+	if tmp["buyer"]!=nil && buyer !="" {
+		score = score+4
+	}
+
+	//projectcode agency area city budget
+	projectcode := qu.ObjToString(tmp["projectcode"])
+	if tmp["projectcode"]!=nil && projectcode !="" {
+		score = score+2
+	}
+	agency := qu.ObjToString(tmp["agency"])
+	if tmp["agency"]!=nil && agency !="" {
+		score = score+2
+	}
+	area := qu.ObjToString(tmp["area"])
+	if tmp["area"]!=nil && area !="" {
+		score = score+2
+	}
+	city := qu.ObjToString(tmp["city"])
+	if tmp["city"]!=nil && city !="" {
+		score = score+2
+	}
+	budget := qu.Float64All(tmp["budget"])
+	if tmp["budget"]!=nil && budget !=0.0 {
+		score = score+2
+	}
+
+	score = score+dealWithLowScoreData(tmp)
+
+
+	return score
+}
+
+func dealWithContractTypeScore(tmp map[string]interface{}) int {
 	/*
-		质量评分总分
+	公告类型：合同
+	2分：项目名称、采购单位、合同编号、中标单位
+	1分：代理机构、中标金额、项目编号、省份、城市、合同名称、合同签订日期、预算
+	0.5分：采购单位联系人、采购单位电话、采购单位地址、采购单位邮编、招标代理机构联系人、招标代理机构联系方式、招标代理机构地址、投标截止日期、开标日期
+
+	其它类型：拟建、结果变更、违规、验收、其它
+	暂无
 	*/
-	//测试随机分
-	rand.Seed(time.Now().UnixNano()) //以当前系统时间作为种子参数
-	return rand.Intn(100)
+	score :=0
+	//projectname buyer winner contractnumber
+	projectname := qu.ObjToString(tmp["projectname"])
+	if tmp["projectname"]!=nil && projectname !="" {
+		score = score+4
+	}
+	buyer := qu.ObjToString(tmp["buyer"])
+	if tmp["buyer"]!=nil && buyer !="" {
+		score = score+4
+	}
+	winner := qu.ObjToString(tmp["winner"])
+	if tmp["winner"]!=nil && winner !="" {
+		score = score+4
+	}
+	contractnumber := qu.ObjToString(tmp["contractnumber"])
+	if tmp["contractnumber"]!=nil && contractnumber !="" {
+		score = score+4
+	}
+
+	//projectcode agency area city contractname bidamount budget  signaturedate
+	projectcode := qu.ObjToString(tmp["projectcode"])
+	if tmp["projectcode"]!=nil && projectcode !="" {
+		score = score+2
+	}
+	agency := qu.ObjToString(tmp["agency"])
+	if tmp["agency"]!=nil && agency !="" {
+		score = score+2
+	}
+	area := qu.ObjToString(tmp["area"])
+	if tmp["area"]!=nil && area !="" {
+		score = score+2
+	}
+	city := qu.ObjToString(tmp["city"])
+	if tmp["city"]!=nil && city !="" {
+		score = score+2
+	}
+	contractname := qu.ObjToString(tmp["contractname"])
+	if tmp["contractname"]!=nil && contractname !="" {
+		score = score+2
+	}
+	budget := qu.Float64All(tmp["budget"])
+	if tmp["budget"]!=nil && budget !=0.0 {
+		score = score+2
+	}
+	bidamount := qu.Float64All(tmp["bidamount"])
+	if tmp["bidamount"]!=nil && bidamount !=0.0 {
+		score = score+2
+	}
+	signaturedate := qu.Int64All(tmp["signaturedate"])
+	if tmp["signaturedate"]!=nil && signaturedate !=0 {
+		score = score+2
+	}
+
+
+
+	score = score+dealWithLowScoreData(tmp)
+
+
+	return score
+}
+
+//公用低分项
+func dealWithLowScoreData(tmp map[string]interface{}) int  {
+	score :=0
+	//buyerperson buyertel buyeraddr buyerzipcode
+	//agencyperson  agencytel agencyaddr
+	//bidopentime bidendtime
+	buyerperson := qu.ObjToString(tmp["buyerperson"])
+	if tmp["buyerperson"]!=nil && buyerperson !="" {
+	}
+	buyertel := qu.ObjToString(tmp["buyertel"])
+	if tmp["buyertel"]!=nil && buyertel !="" {
+		score = score+1
+	}
+	buyeraddr := qu.ObjToString(tmp["buyeraddr"])
+	if tmp["buyeraddr"]!=nil && buyeraddr !="" {
+		score = score+1
+	}
+	buyerzipcode := qu.ObjToString(tmp["buyerzipcode"])
+	if tmp["buyerzipcode"]!=nil && buyerzipcode !="" {
+		score = score+1
+	}
+	agencyperson := qu.ObjToString(tmp["agencyperson"])
+	if tmp["agencyperson"]!=nil && agencyperson !="" {
+		score = score+1
+	}
+	agencytel := qu.ObjToString(tmp["agencytel"])
+	if tmp["agencytel"]!=nil && agencytel !="" {
+		score = score+1
+	}
+	agencyaddr := qu.ObjToString(tmp["agencyaddr"])
+	if tmp["agencyaddr"]!=nil && agencyaddr !="" {
+		score = score+1
+	}
+	bidopentime := qu.Int64All(tmp["bidopentime"])
+	if tmp["bidopentime"]!=nil && bidopentime !=0 {
+		score = score+1
+	}
+	bidendtime := qu.Int64All(tmp["bidendtime"])
+	if tmp["bidendtime"]!=nil && bidendtime !=0 {
+		score = score+1
+	}
+
+	return score
 }

util/src/dbutil/mongo/mgo.go

@@ -127,6 +127,8 @@ type MongodbSim struct {
 	Ctx      context.Context
 	ShortCtx context.Context
 	pool     chan bool
+	UserName string
+	PassWord string
 }
 
 func (m *MongodbSim) GetMgoConn() *MgoSess {
@@ -149,6 +151,9 @@ func (m *MongodbSim) InitPool() {
 	opts.SetMaxPoolSize(uint64(m.Size))
 	m.pool = make(chan bool, m.Size)
 	opts.SetMaxConnIdleTime(2 * time.Hour)
+	if m.UserName != "" && m.PassWord != "" {
+		opts.SetAuth(options.Credential{Username: m.UserName, Password: m.PassWord})
+	}
 	m.Ctx, _ = context.WithTimeout(context.Background(), 99999*time.Hour)
 	m.ShortCtx, _ = context.WithTimeout(context.Background(), 1*time.Minute)
 	client, err := mongo.Connect(m.ShortCtx, opts)