融合-多线程-正式流程质量-站点完善

udpfusion/src/config.json

@@ -6,8 +6,8 @@
     "collName": "zk_data",
     "pool": 10,
     "site": {
-      "site_dbname": "qfw",
-      "site_coll": "site"
+      "dbname": "zhengkun",
+      "coll": "site"
     }
   },
   "notFusionKey": {

udpfusion/src/main.go

@@ -9,7 +9,6 @@ import (
 	"os"
 	"qfw/common/src/qfw/util"
 	qu "qfw/util"
-	"strconv"
 	"sync"
 	"time"
 )
@@ -26,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{
@@ -47,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()
@@ -64,6 +76,8 @@ func init() {
 
 
 
+
+
 	log.Println("采用udp模式")
 }
 
@@ -80,6 +94,7 @@ func mainT() {
 //快速测试使用
 func main() {
 
+
 	sid := "100000000000000000000000"
 	eid := "900000000000000000000000"
 	//log.Println(sid, "---", eid)
@@ -102,9 +117,7 @@ func main() {
 
 //融合具体方法
 func startTask(data []byte, mapInfo map[string]interface{}) {
-
 	log.Println("开始融合流程")
-
 	defer qu.Catch()
 	//区间id
 	q := map[string]interface{}{
@@ -191,25 +204,18 @@ func startTask(data []byte, mapInfo map[string]interface{}) {
 		//log.Println("当前分组数量：",len(fusionDataGroupArr))
 	}
 	log.Println("最终待融合分组数量：",len(fusionDataGroupArr))
-
-
-
-
 	log.Println("********************分割线********************")
 	log.Println("********************分割线********************")
 	log.Println("********************分割线********************")
 	log.Println("开始处理新增分组... ...")
 	start := int(time.Now().Unix())
-	//进行分组融合
-	pool := make(chan bool, 1)
+
+	//多线程 - 处理数据
+	pool := make(chan bool, 3)
 	wg := &sync.WaitGroup{}
 
 	for i:=0;i<len(fusionDataGroupArr);i++ {
 		fusionArr := fusionDataGroupArr[i]
-		//构建数据
-		log.Println("构建第",i+1,"组数据...","数量：",len(fusionArr),fusionArr)
-		//多线程 - 处理数据
-
 		pool <- true
 		wg.Add(1)
 		go func(fusionArr []string,i int) {
@@ -217,8 +223,8 @@ func startTask(data []byte, mapInfo map[string]interface{}) {
 				<-pool
 				wg.Done()
 			}()
-
-
+			//构建数据
+			log.Println("构建第",i+1,"组数据...","数量：",len(fusionArr),fusionArr)
 			weight :=NewWeightData(fusionArr)
 			////整理数据-筛选排名，模板
 			weight.analyzeBuildStandardData()
@@ -265,8 +271,8 @@ func startTask(data []byte, mapInfo map[string]interface{}) {
 
 	wg.Wait()
 
-	log.Println("新增融合over ：",len(fusionDataGroupArr),"用时：",int(time.Now().Unix())-start,"秒")
-
+	log.Println("fusion is over ：",len(fusionDataGroupArr),"用时：",int(time.Now().Unix())-start,"秒")
+	log.Println("睡眠30秒，然后在发广播")
 	time.Sleep(30 * time.Second)
 	//任务完成，开始发送广播通知下面节点
 	taskSendFusionUdp(mapInfo)
@@ -395,17 +401,7 @@ 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

@@ -27,6 +27,7 @@ func newUpdateFusionPool() *updateFusionInfo {
 
 func (update *updateFusionInfo) updateFusionData() {
 	log.Println("监听--融合更新数据")
+
 	tmpArr := make([][]map[string]interface{}, update.saveSize)
 	tmpIndex := 0
 	for {

udpfusion/src/weightFusionMethod.go

@@ -3,6 +3,7 @@ package main
 import (
 	qu "qfw/util"
 	"reflect"
+	"strconv"
 )
 
 //处理-重复附件数据
@@ -147,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

@@ -18,7 +18,6 @@ type weightDataMap struct {
 }
 
 func NewWeightData(arr []string) *weightDataMap {
-	//测试-默认第一个
 
 	weight := &weightDataMap{map[string]*weightInfo{},[]string{},[]string{},""}
 

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,278 @@ 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分：采购单位联系人、采购单位电话、采购单位地址、采购单位邮编、招标代理机构联系人、招标代理机构联系方式、招标代理机构地址、投标截止日期、开标日期
 	*/
-	//测试随机分
-	rand.Seed(time.Now().UnixNano()) //以当前系统时间作为种子参数
-	return rand.Intn(100)
+	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分：采购单位联系人、采购单位电话、采购单位地址、采购单位邮编、招标代理机构联系人、招标代理机构联系方式、招标代理机构地址、投标截止日期、开标日期
+
+	其它类型：拟建、结果变更、违规、验收、其它
+	暂无
+	*/
+	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
+	}
+
+	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
 }