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

fullproject/src/project.go

@@ -362,17 +362,19 @@ func (p *ProjectTask) compareBCTABB(info *Info, cp *ProjectInfo, diffTime int64,
 		compareBudget = "A"
 		score += 1
 		score2 += 1
-	} else if info.Budget == 0 && cp.Budget == 0 {
-		compareBudget = "B"
 	}
+	//	else if info.Budget == 0 && cp.Budget == 0 {
+	//		compareBudget = "B"
+	//	}
 	compareBidmount = "C"
 	if info.Bidamount > 0 && (info.Bidamount == cp.Bidamount || (cp.Budget > 0 && cp.Budget > info.Bidamount && (cp.Budget-info.Bidamount) < 0.1*cp.Budget)) {
 		compareBidmount = "A"
 		score += 1
 		score2 += 1
-	} else if info.Bidamount == 0 && cp.Bidamount == 0 {
-		compareBidmount = "B"
 	}
+	//	else if info.Bidamount == 0 && cp.Bidamount == 0 {
+	//		compareBidmount = "B"
+	//	}
 
 	cp.score = score
 	return
@@ -431,7 +433,7 @@ func (p *ProjectTask) NewProject(tmp map[string]interface{}, thisinfo *Info) (st
 	set["sourceinfourl"] = tmp["href"]
 	set["firsttime"] = tmp["publishtime"]
 	set["lasttime"] = tmp["publishtime"]
-	set["pici"] = now
+	set["pici"] = p.pici
 	set["ids"] = []string{thisinfo.Id}
 	if thisinfo.TopType == "招标" {
 		set["zbtime"] = tmp["publishtime"]
@@ -637,7 +639,7 @@ func (p *ProjectTask) UpdateProject(tmp map[string]interface{}, thisinfo *Info,
 		set["bidamount"] = pInfo.Bidamount
 	}
 
-	if thisinfo.Budget > 0 && pInfo.Budget < 1 {
+	if thisinfo.Budget > 0 && pInfo.Budget < 1 { //多包的会有问题，没有进行合计。
 		pInfo.Budget = thisinfo.Budget
 		set["budget"] = pInfo.Budget
 	}
@@ -679,6 +681,7 @@ func (p *ProjectTask) UpdateProject(tmp map[string]interface{}, thisinfo *Info,
 
 	set["mpn"] = pInfo.MPN
 	set["mpc"] = pInfo.MPC
+	set["pici"] = p.pici
 
 	if thisinfo.HasPackage {
 		set["multipackage"] = 1
@@ -693,7 +696,7 @@ func (p *ProjectTask) UpdateProject(tmp map[string]interface{}, thisinfo *Info,
 	//保留原数据吧
 	push := p.PushListInfo(tmp)
 	push["compareStr"] = comStr
-	push["resVal"] = pInfo.pjVal
+	push["resVal"] = pInfo.resVal
 	push["pjVal"] = pInfo.pjVal
 	update["$push"] = map[string]interface{}{
 		"list": push,

fullproject/src/task.go

@@ -47,19 +47,20 @@ type ProjectTask struct {
 	currentTime int64
 	//保存长度
 	saveSize int
+	pici     int64
 }
 
 func NewPT() *ProjectTask {
 	return &ProjectTask{
 		InitMinTime: int64(1325347200),
 		name:        "全/增量对象",
-		thread:      3,
+		thread:      4,
 		updatePool:  make(chan []map[string]interface{}, 2000),
 		wg:          sync.WaitGroup{},
-		AllIdsMap:   make(map[string]*ID, 5000000),
-		mapPb:       make(map[string]*Key, 5000000),
-		mapPn:       make(map[string]*Key, 5000000),
-		mapPc:       make(map[string]*Key, 5000000),
+		AllIdsMap:   make(map[string]*ID, 10000000),
+		mapPb:       make(map[string]*Key, 3000000),
+		mapPn:       make(map[string]*Key, 10000000),
+		mapPc:       make(map[string]*Key, 10000000),
 		saveSize:    200,
 		coll:        ProjectColl,
 	}
@@ -197,15 +198,34 @@ func (p *ProjectTask) taskQl(udpInfo map[string]interface{}) {
 	//		log.Println("publishtime_1索引不存在")
 	//	}
 	//	MongoTool.DestoryMongoConn(sess)
-	thread := util.IntAllDef(udpInfo["thread"], 3)
+	thread := util.IntAllDef(udpInfo["thread"], 4)
 	if thread > 0 {
 		p.thread = thread
 	}
-	bcon := true
-	if bcon {
-		//生成查询语句执行
-		p.enter(db, coll, map[string]interface{}{})
+	q, _ := udpInfo["query"].(map[string]interface{})
+	if q == nil {
+		q = map[string]interface{}{}
+		lteid, _ := udpInfo["lteid"].(string)
+		var idmap map[string]interface{}
+		if len(lteid) > 15 {
+			idmap = map[string]interface{}{
+				"$lte": util.StringTOBsonId(lteid),
+			}
+		}
+		gtid, _ := udpInfo["gtid"].(string)
+		if len(gtid) > 15 {
+			if idmap == nil {
+				idmap = map[string]interface{}{}
+			}
+			idmap["$gt"] = util.StringTOBsonId(gtid)
+		}
+		if idmap != nil {
+			q["_id"] = idmap
+		}
 	}
+	//生成查询语句执行
+	p.enter(db, coll, q)
+
 }
 
 //增量合并
@@ -236,12 +256,11 @@ func (p *ProjectTask) taskZl(udpInfo map[string]interface{}) {
 			},
 		}
 	}
-	pici := time.Now().Unix()
 	if q != nil {
 		//生成查询语句执行
 		p.enter(db, coll, q)
 	}
-	nextNode(gtid, lteid, "project", pici)
+	nextNode(gtid, lteid, "project", p.pici)
 }
 
 //通知下个节点nextNode