data_clean/src/check.go

package main

import (
	"github.com/uuid"
	"log"
	qu "qfw/util"
	"strings"
	"sync"
)

// 开始审查数据
func startCheckData(sid, eid string) {
	defer qu.Catch()
	q := map[string]interface{}{
		"_id": map[string]interface{}{
			"$gt":  StringTOBsonId(sid),
			"$lte": StringTOBsonId(eid),
		},
	}
	check_pool := make(chan bool, check_thread)
	check_wg := &sync.WaitGroup{}
	sess := data_mgo.GetMgoConn()
	defer data_mgo.DestoryMongoConn(sess)
	it := sess.DB(data_mgo.DbName).C(coll_name).Find(&q).Iter()
	total := 0
	for tmp := make(map[string]interface{}); it.Next(&tmp); total++ {
		if total%10000 == 0 {
			log.Println("当前数量:", total, tmp["_id"])
		}
		update_id := map[string]interface{}{"_id": tmp["_id"]}
		check_pool <- true
		check_wg.Add(1)
		go func(tmp map[string]interface{}, update_id map[string]interface{}) {
			defer func() {
				<-check_pool
				check_wg.Done()
			}()
			//更新-
			update_info := make(map[string]interface{}, 0)
			//审查-发布时间
			getCheckDataPub(tmp, update_info)
			//审查-单位信息
			getCheckDataUnit(tmp, update_info)
			//是否删除
			is_unset := false
			if update_ai {
				is_unset = ChooseCheckDataAI(tmp, &update_info)
				if update_info["com_package"] == nil { //构建单包信息···
					com_package := CreatSingleFieldInfo(tmp, update_info)
					update_info["com_package"] = com_package
				}
			}
			//最终更新
			if len(update_info) > 0 {
				UpdateTask.updatePool <- []map[string]interface{}{
					update_id,
					{"$set": update_info},
				}
			}
			if is_unset {
				UpdateTask.updatePool <- []map[string]interface{}{
					update_id,
					{"$unset": Unset_Check},
				}
			}
		}(tmp, update_id)
		tmp = make(map[string]interface{})
	}
	check_wg.Wait()
	log.Println("data_clean is over ", total)
	sendNextNode(sid, eid)
}

func CreatSingleFieldInfo(tmp map[string]interface{}, update_info map[string]interface{}) []map[string]interface{} {
	pkgArr := []map[string]interface{}{}
	com_package := map[string]interface{}{}
	package_id := uuid.New().String()
	package_id = strings.ReplaceAll(package_id, "-", "")
	com_package["package_id"] = package_id
	com_package["name"] = qu.ObjToString(tmp["projectname"])

	if update_info["budget"] != nil {
		com_package["budget"] = update_info["budget"]
	} else {
		if tmp["budget"] != nil {
			com_package["budget"] = tmp["budget"]
		}
	}
	subtype := qu.ObjToString(update_info["subtype"])
	if subtype == "" {
		subtype = qu.ObjToString(tmp["subtype"])
	}
	if subtype == "单一" || subtype == "中标" || subtype == "成交" || subtype == "合同" {
		if update_info["bidamount"] != nil {
			com_package["bidamount"] = update_info["bidamount"]
		} else {
			if tmp["bidamount"] != nil {
				com_package["bidamount"] = tmp["bidamount"]
			}
		}

		if update_info["winner"] != nil {
			com_package["winner"] = update_info["winner"]
		} else {
			if tmp["winner"] != nil {
				com_package["winner"] = tmp["winner"]
			}
		}
	}
	pkgArr = append(pkgArr, com_package)
	return pkgArr
}