fieldproject/tieta_data/task.go

package main

import (
	"encoding/json"
	"github.com/tealeg/xlsx"
	"go.mongodb.org/mongo-driver/bson"
	"go.uber.org/zap"
	util "jygit.jydev.jianyu360.cn/data_processing/common_utils"
	"jygit.jydev.jianyu360.cn/data_processing/common_utils/log"
	"jygit.jydev.jianyu360.cn/data_processing/common_utils/mongodb"
	"regexp"
	"strings"
	"tieta_data/config"
	"time"
)

var (
	queryClose     = make(chan bool)
	queryCloseOver = make(chan bool)
	pool           = make(chan bool, 2)

	StrOrNum   = regexp.MustCompile("^[0-9_-]{1,4}$|^[a-zA-Z_-]{1,4}$")
	_datereg   = regexp.MustCompile("20[0-2][0-9][年-][0-9]{1,2}[月-][0-9]{1,2}[日-]([0-9]{1,2}时[0-9]{0,2})?")
	StrOrNum2  = regexp.MustCompile("^[0-9_-]+$|^[a-zA-Z_-]+$")
	replaceStr = regexp.MustCompile("(工程|采购|项目|[?！、【】（）—()－-]|栏标价|中标候选人|招标代理)")
)

func taskExcelP(tmp map[string]interface{}, row *xlsx.Row) {

	for _, v := range FieldArr {

		if FieldMap[v] != "" {
			if v == "项目id" {
				row.AddCell().SetValue(mongodb.BsonIdToSId(tmp["_id"]))
			} else if v == "预算金额（元）" || v == "中标金额（元）" {
				if tmp[FieldMap[v]] != nil {
					row.AddCell().SetValue(tmp[FieldMap[v]])
				} else {
					row.AddCell().SetValue("")
				}
			} else if v == "招标数据更新时间" || v == "中标数据更新时间" {
				p := util.Int64All(tmp[FieldMap[v]])
				if p > 0 {
					row.AddCell().SetValue(util.FormatDateByInt64(&p, util.Date_Full_Layout))
				} else {
					row.AddCell().SetValue("")
				}
			} else {
				row.AddCell().SetValue(util.ObjToString(tmp[FieldMap[v]]))
			}
		} else {
			row.AddCell().SetValue("")
		}

	}
}

func taskExcelB(tmp map[string]interface{}, row *xlsx.Row) {
	for _, v := range FieldArr1 {
		if FieldMap1[v] != "" {
			if v == "信息id" {
				row.AddCell().SetValue(util.ObjToString(tmp["id"]))
			} else if v == "项目id" {
				info, _ := MongoTool.FindOne(config.Conf.DB.Mongo.Pcoll, bson.M{"ids": util.ObjToString(tmp["id"])})
				if len(*info) > 0 {
					row.AddCell().SetValue(mongodb.BsonIdToSId((*info)["_id"]))
				} else {
					row.AddCell().SetValue("")
				}
			} else if v == "发布时间" || v == "标书获取时间" || v == "标书截止时间" || v == "投标开始时间" || v == "投标截止时间" || v == "开标时间" {
				p := util.Int64All(tmp[FieldMap1[v]])
				if p > 0 {
					row.AddCell().SetValue(util.FormatDateByInt64(&p, util.Date_Full_Layout))
				} else {
					row.AddCell().SetValue("")
				}
			} else if v == "预算金额（元）" || v == "中标金额（元）" {
				if tmp[FieldMap1[v]] != nil {
					row.AddCell().SetValue(tmp[FieldMap1[v]])
				} else {
					row.AddCell().SetValue("")
				}
			} else if v == "是否电子招标" {
				if util.ObjToString(tmp[FieldMap1[v]]) != "" {
					row.AddCell().SetValue("是")
				} else {
					row.AddCell().SetValue("否")
				}
			} else {
				row.AddCell().SetValue(util.ObjToString(tmp[FieldMap1[v]]))
			}
		} else {
			row.AddCell().SetValue("")
		}

	}
}

func loadData() {
	sess := MongoTool.GetMgoConn()
	defer MongoTool.DestoryMongoConn(sess)

	it := sess.DB(config.Conf.DB.Mongo.Dbname).C(config.Conf.DB.Mongo.Pcoll).Find(nil).Select(nil).Iter()
	n := 0
	result := make(map[string]interface{})
	if it.Next(&result) {
		n++
		bys, _ := json.Marshal(result)
		var pro *Project
		_ = json.Unmarshal(bys, &pro)

		id := pro.Id.Hex()
		for _, v := range append([]string{pro.ProjectName}, pro.MPN...) {
			if v != "" {
				//v = pcReplace.ReplaceAllString(v, "")
				if v != "" {
					k := mapPn[v]
					if k == nil {
						k = &Key{Arr: []string{id}}
						mapPn[v] = k
					} else {
						k.Arr = append(k.Arr, id)
					}
				}
			}
		}
		for _, v := range append([]string{pro.ProjectCode}, pro.MPC...) {
			if v != "" {
				if v != "" {
					k := mapPc[v]
					if k == nil {
						k = &Key{Arr: []string{id}}
						mapPc[v] = k
					} else {
						k.Arr = append(k.Arr, id)
					}
				}
			}
		}
		if pro.Buyer != "" && len([]rune(pro.Buyer)) > 2 {
			k := mapPb[pro.Buyer]
			if k == nil {
				k = &Key{Arr: []string{id}}
				mapPb[pro.Buyer] = k
			} else {
				k.Arr = append(k.Arr, id)
			}
		}
		AllIdsMap[id] = &ID{Id: id, P: pro}
	}

	time.Sleep(2 * time.Second)
	log.Info("load project over..", zap.Int("n", n))
}

func taskProject(mapinfo map[string]interface{}) {
	q := make(map[string]interface{})
	gtid := util.ObjToString(mapinfo["gtid"])
	lteid := util.ObjToString(mapinfo["lteid"])
	q["_id"] = map[string]interface{}{
		"$lte": mongodb.StringTOBsonId(lteid),
		"$gte": mongodb.StringTOBsonId(gtid),
	}
	//生成查询语句执行
	log.Info("查询语句：", zap.Any("q", q))
	count := 0

	sess := MongoTool.GetMgoConn()
	defer MongoTool.DestoryMongoConn(sess)
	infoPool := make(chan map[string]interface{}, 2000)
	over := make(chan bool)
	go func() {
	L:
		for {
			select {
			case tmp := <-infoPool:
				pool <- true
				go func(tmp map[string]interface{}) {
					defer func() {
						<-pool
					}()
					info := ParseInfo(tmp)
					startProjectMerge(info, tmp)
				}(tmp)
			default:
				select {
				case tmp := <-infoPool:
					pool <- true
					go func(tmp map[string]interface{}) {
						defer func() {
							<-pool
						}()
						info := ParseInfo(tmp)
						startProjectMerge(info, tmp)
					}(tmp)
				case <-over:
					break L
				}
			}
		}
	}()
	fields := map[string]interface{}{"kvtext": 0, "repeat_reason": 0, "field_source": 0, "detail": 0, "contenthtml": 0, "jsondata": 0}
	ms := sess.DB(config.Conf.DB.Mongo.Dbname).C(config.Conf.DB.Mongo.Coll).Find(q).Select(fields)
	query := ms.Iter()
	var lastid interface{}
L:
	for {
		select {
		case <-queryClose:
			log.Error("receive interrupt sign")
			queryCloseOver <- true
			break L
		default:
			tmp := make(map[string]interface{})
			if query.Next(&tmp) {
				lastid = tmp["_id"]
				if count%2000 == 0 {
					log.Info("current", zap.Int("count", count), zap.Any("lastid", lastid))
				}
				infoPool <- tmp
				count++
			} else {
				break L
			}
		}
	}
	time.Sleep(5 * time.Second)
	over <- true
	////阻塞
	for n := 0; n < 1; n++ {
		pool <- true
	}
	log.Info("所有线程执行完成...", zap.Int("count:", count))
}

func ParseInfo(tmp map[string]interface{}) (info *Info) {
	bys, _ := json.Marshal(tmp)
	var thisinfo *Info
	_ = json.Unmarshal(bys, &thisinfo)
	if thisinfo == nil {
		return nil
	}

	if len(thisinfo.Topscopeclass) == 0 {
		thisinfo.Topscopeclass = []string{}
	}
	if len(thisinfo.Subscopeclass) == 0 {
		thisinfo.Subscopeclass = []string{}
	}
	if thisinfo.SubType == "" {
		thisinfo.SubType = util.ObjToString(tmp["bidstatus"])
	}

	if thisinfo.ProjectName != "" && len([]rune(thisinfo.ProjectName)) > 0 {
		//thisinfo.ProjectName = pcReplace.ReplaceAllString(thisinfo.ProjectName, "")
		//if thisinfo.ProjectName != "" {
		thisinfo.pnbval++
		//}
	}

	if thisinfo.ProjectCode != "" || thisinfo.PTC != "" {
		if thisinfo.ProjectCode != "" {
			//thisinfo.ProjectCode = pcReplace.ReplaceAllString(thisinfo.ProjectCode, "")
			if thisinfo.pnbval == 0 && len([]rune(thisinfo.ProjectCode)) < 5 {
				thisinfo.ProjectCode = StrOrNum.ReplaceAllString(thisinfo.ProjectCode, "")
			}
		} else {
			//thisinfo.PTC = pcReplace.ReplaceAllString(thisinfo.PTC, "")
			if thisinfo.pnbval == 0 && len([]rune(thisinfo.PTC)) < 5 {
				thisinfo.PTC = StrOrNum.ReplaceAllString(thisinfo.PTC, "")
			}
		}
		thisinfo.pnbval++
	}
	if thisinfo.ProjectCode == thisinfo.PTC || strings.Index(thisinfo.ProjectCode, thisinfo.PTC) > -1 {
		thisinfo.PTC = ""
	}

	if thisinfo.Buyer != "" && len([]rune(thisinfo.Buyer)) > 2 {
		thisinfo.pnbval++
	} else {
		thisinfo.Buyer = ""
	}
	//winners整理、清理
	winner := util.ObjToString(tmp["winner"])
	m1 := map[string]bool{}
	winners := []string{}
	if winner != "" {
		m1[winner] = true
		winners = append(winners, winner)
	}
	thisinfo.Winners = winners

	thisinfo.LenPC = len([]rune(thisinfo.ProjectCode))
	thisinfo.LenPTC = len([]rune(thisinfo.PTC))
	thisinfo.LenPN = len([]rune(thisinfo.ProjectName))
	return thisinfo
}