jy-data-extract/fullproject/src_dev3/new_project.go

package main

import (
	qu "qfw/util"
	"strings"
	"time"

	"gopkg.in/mgo.v2/bson"
)

var FIELDS = []string{"area", "city", "district", "projectname", "projectcode", "buyer", "winner", "budget", "bidamount", "bidstatus", "agency", "projectscope"}

func NewProject(tmp map[string]interface{}, thisinfo *Info) (string, *ProjectInfo) {
	set := map[string]interface{}{}
	for _, f := range FIELDS {
		if tmp[f] != nil {
			set[f] = tmp[f]
		}
	}
	set["s_projectname"] = thisinfo.ProjectName
	set["createtime"] = time.Now().Unix()
	set["sourceinfoid"] = qu.BsonIdToSId(tmp["_id"])
	set["sourceinfourl"] = tmp["href"]
	set["topscopeclass"] = thisinfo.Topscopeclass
	set["subscopeclass"] = thisinfo.Subscopeclass
	if thisinfo.Buyerperson != "" {
		set["buyerperson"] = thisinfo.Buyerperson
	}
	if thisinfo.Buyertel != "" {
		set["buyertel"] = thisinfo.Buyertel
	}
	if thisinfo.Buyerclass != "" {
		set["buyerclass"] = thisinfo.Buyerclass
	}
	if thisinfo.Bidopentime > 0 {
		set["bidopentime"] = thisinfo.Bidopentime
	}
	if len(thisinfo.Winnerorder) > 0 {
		set["winnerorder"] = thisinfo.Winnerorder
	}
	s_subscopeclass := strings.Join(thisinfo.Subscopeclass, ",")
	set["s_subscopeclass"] = s_subscopeclass
	s_winner := strings.Join(thisinfo.Winners, ",")
	set["s_winner"] = s_winner
	if tmp["package"] != nil {
		set["package"] = tmp["package"] //没定义优先级
	}
	push := NewPushInfo(tmp)
	set["list"] = []bson.M{
		push,
	}
	pId := bson.NewObjectId()
	set["_id"] = pId
	///id := MongoTool.Save(ProjectColl, set)
	p1 := NewPinfo(pId, thisinfo, set)
	return pId.Hex(), &p1
}

func NewPushInfo(tmp map[string]interface{}) bson.M {
	res := bson.M{
		"comeintime":  tmp["comeintime"],
		"publishtime": tmp["publishtime"],
		"title":       tmp["title"],
		"toptype":     tmp["toptype"],
		"subtype":     tmp["subtype"],
		"infoformat":  tmp["infoformat"],
		"infoid":      qu.BsonIdToSId(tmp["_id"]),
		"area":        tmp["area"],
		"city":        tmp["city"],
		"projectname": tmp["projectname"],
		"projectcode": tmp["projectcode"],
		"buyer":       tmp["buyer"],
		"href":        tmp["href"],
	}
	for _, k := range []string{"winner", "budget", "bidamount"} {
		if tmp[k] != nil {
			res[k] = tmp[k]
		}
	}
	return res
}

//生成存放在redis数组中的对象
func NewPinfo(id bson.ObjectId, thisinfo *Info, set map[string]interface{}) ProjectInfo {
	p1 := ProjectInfo{
		Id:            id,
		Ids:           []string{thisinfo.Id},
		Topscopeclass: thisinfo.Topscopeclass,
		Subscopeclass: thisinfo.Subscopeclass,
		Winners:       thisinfo.Winners,
		ProjectName:   thisinfo.ProjectName,
		ProjectCode:   thisinfo.ProjectCode,
		Buyer:         thisinfo.Buyer,
		Agency:        thisinfo.Agency,
		Area:          thisinfo.Area,
		City:          thisinfo.City,
		District:      thisinfo.District,
		MPN:           []string{},
		MPC:           []string{},
		HasPackage:    thisinfo.HasPackage,
		Package:       map[string]interface{}{},
		Buyerclass:    thisinfo.Buyerclass,
		Bidopentime:   thisinfo.Bidopentime,
		Winnerorder:   thisinfo.Winnerorder,
		FirstTime:     thisinfo.Publishtime,
		LastTime:      thisinfo.Publishtime,
		Budget:        thisinfo.Budget,
		Bidamount:     thisinfo.Bidamount,
	}
	if thisinfo.LenPTC > 5 {
		p1.MPC = append(p1.MPC, thisinfo.PTC)
	}
	//savePool <- set
	MongoTool.SaveByOriID(ProjectColl, set)

	return p1
}

var updatePool = make(chan []map[string]interface{}, 30)

//var savePool = make(chan map[string]interface{}, 6)
//func SaveQueue() {
//	arr := []map[string]interface{}{}
//	sp := make(chan bool, 3)
//	for {
//		select {
//		case <-beforUpdate:
//			if len(arr) > 0 {
//				MongoTool.SaveBulk(ProjectColl, arr...)
//				arr = []map[string]interface{}{}
//			}
//		case v := <-savePool:
//			arr = append(arr, v)
//			if len(arr) > 50 {
//				sp <- true
//				go func(arr []map[string]interface{}) {
//					MongoTool.SaveBulk(ProjectColl, arr...)
//					<-sp
//				}(arr)
//				arr = []map[string]interface{}{}
//			}
//		case <-time.After(80 * time.Millisecond):
//			if len(arr) > 0 {
//				sp <- true
//				go func(arr []map[string]interface{}) {
//					MongoTool.SaveBulk(ProjectColl, arr...)
//					<-sp
//				}(arr)
//				arr = []map[string]interface{}{}
//			}
//		}
//	}
//}

func updateQueue() {
	arr := [][]map[string]interface{}{}
	sp := make(chan bool, 3)
	for {
		select {
		case v := <-updatePool:
			arr = append(arr, v)
			if len(arr) > 200 {
				sp <- true
				go func(arr [][]map[string]interface{}) {
					MongoTool.UpdateBulk(ProjectColl, arr...)
					<-sp
				}(arr)
				arr = [][]map[string]interface{}{}
			}
		case <-time.After(500 * time.Millisecond):
			if len(arr) > 0 {
				sp <- true
				go func(arr [][]map[string]interface{}) {
					MongoTool.UpdateBulk(ProjectColl, arr...)
					<-sp
				}(arr)
				arr = [][]map[string]interface{}{}
			}
		}
	}
}