jy-data-extract/dataclean/src/service/update.go

// update
package service

import (
	"log"
	qu "qfw/util"
	"time"
)

func init() {
	go update()
}

func update() {
	for {
		if len(ResultInfos) > 0 {
			UpdateLock.Lock()
			UpResults = ResultInfos
			ResultInfos = []*ResultInfo{}
			UpdateLock.Unlock()
			up()
			UpResults = []*ResultInfo{}
		}
		time.Sleep(5 * time.Second)
	}
}

func up() {
	log.Println("开始更新")
	binfos := []*BidInfo{}
	einfos := []*ExtractInfo{}
	pinfos := []*ProjectInfo{}
	for _, v := range UpResults {
		if v.Bidding != nil {
			binfos = append(binfos, v.Bidding) //招标信息清洗结果
		}
		if v.Extract != nil {
			einfos = append(einfos, v.Extract) //抽取清洗结果
		}
		if v.Project != nil {
			pinfos = append(pinfos, v.Project) //项目清洗结果
		}
	}

	updateB := [][]map[string]interface{}{}
	updateE := [][]map[string]interface{}{}
	updateP := [][]map[string]interface{}{}
	if len(binfos) > 0 { //招标信息：拼装更新语句
		for _, tmp := range binfos {
			set := map[string]interface{}{}
			for _, v := range tmp.UpParam {
				set[v] = tmp.Data[v]
			}
			tmparr := []map[string]interface{}{
				map[string]interface{}{
					"_id": qu.StringTOBsonId(tmp.Id),
				},
				map[string]interface{}{"$set": set},
			}
			updateB = append(updateB, tmparr)
		}
	}
	if len(einfos) > 0 { //抽取信息：拼装更新语句
		for _, tmp := range einfos {
			set := map[string]interface{}{}
			for _, v := range tmp.UpParam {
				set[v] = tmp.Data[v]
			}
			tmparr := []map[string]interface{}{
				map[string]interface{}{
					"_id": qu.StringTOBsonId(tmp.Id),
				},
				map[string]interface{}{"$set": set},
			}
			updateE = append(updateE, tmparr)
		}
	}
	if len(pinfos) > 0 { //项目信息：拼装更新语句
		for _, tmp := range pinfos {
			set := map[string]interface{}{}
			for _, v := range tmp.UpParam {
				set[v] = tmp.Data[v]
			}
			tmparr := []map[string]interface{}{
				map[string]interface{}{
					"_id": qu.StringTOBsonId(tmp.Id),
				},
				map[string]interface{}{"$set": set},
			}
			updateP = append(updateP, tmparr)
		}
	}

	//执行更新
	if len(updateB) > 0 {
		db := Mogdbs["qfw"]
		db.Mongodb.UpSertBulk(db.Coll, updateB...)
	}
	if len(updateE) > 0 {
		db := Mogdbs["extract"]
		db.Mongodb.UpSertBulk(db.Coll, updateE...)
	}
	if len(updateP) > 0 {
		db := Mogdbs["project"]
		db.Mongodb.UpSertBulk(db.Coll, updateP...)
	}
	log.Println("更新完毕", "updateB:", len(updateE), "updateE:", len(updateB), "updateP:", len(updateP))
}