jy-data-extract/udpcreateindex/src/projectindex.go

package main

import (
	"log"
	"math"
	mgov "mongodb"
	"qfw/util"
	"qfw/util/elastic"
	"strconv"
)

func projectTask(data []byte, project, mapInfo map[string]interface{}) {
	defer util.Catch()
	q, _ := mapInfo["query"].(map[string]interface{})
	if q == nil {
		q = map[string]interface{}{
			"_id": map[string]interface{}{
				"$gt":  mgov.StringTOBsonId(mapInfo["gtid"].(string)),
				"$lte": mgov.StringTOBsonId(mapInfo["lteid"].(string)),
			},
		}
	} else {
		if q["pici"] == nil {
			idMap, _ := q["_id"].(map[string]interface{})
			if idMap != nil {
				tmpQ := map[string]interface{}{}
				for c, id := range idMap {
					if idStr, ok := id.(string); ok && id != "" {
						tmpQ[c] = mgov.StringTOBsonId(idStr)
					}
				}
				q["_id"] = tmpQ
			}
		}
	}

	var session *mgov.MgoSess
	if project["addr"] != nil {
		session = project2db.GetMgoConn()
		defer project2db.DestoryMongoConn(session)
	} else {
		session = extractmgo.GetMgoConn()
		defer extractmgo.DestoryMongoConn(session)
	}
	c, _ := project["collect"].(string)
	db, _ := project["db"].(string)
	index, _ := project["index"].(string)
	itype, _ := project["type"].(string)
	count, _ := session.DB(db).C(c).Find(&q).Count()
	savepool := make(chan bool, 10)

	log.Println(db, c, "查询语句:", q, "同步总数:", count, "elastic库:", index)
	query := session.DB(db).C(c).Find(q).Iter()
	arr := make([]map[string]interface{}, savesizei)
	var n int
	i := 0
	for tmp := make(map[string]interface{}); query.Next(tmp); i = i + 1 {
		pp := map[string]map[string]interface{}{}
		if packages, ok := tmp["package"].(map[string]interface{}); ok {
			for _, pks := range packages {
				if pk, ok := pks.([]interface{}); ok {
					for _, v := range pk {
						if p, ok := v.(map[string]interface{}); ok {
							winner := util.ObjToString(p["winner"])
							bidamount := util.Float64All((p["bidamount"]))
							if len(winner) > 4 && bidamount > 0 {
								p := map[string]interface{}{
									"winner":    winner,
									"bidamount": bidamount,
								}
								pp[winner] = p
							}
						}
					}
				}
			}
		} else {
			winner := util.ObjToString(tmp["winner"])
			bidamount := util.Float64All(tmp["bidamount"])
			if len(winner) > 4 && bidamount > 0 {
				p := map[string]interface{}{
					"winner":    winner,
					"bidamount": bidamount,
				}
				pp[winner] = p
			}
		}

		pk1 := []map[string]interface{}{}
		for _, v := range pp {
			pk1 = append(pk1, v)
		}
		if len(pk1) > 0 {
			tmp["package1"] = pk1
		}
		budget := util.Float64All(tmp["budget"])
		bidamount := util.Float64All(tmp["bidamount"])
		if float64(budget) > 0 && float64(bidamount) > 0 {
			rate := float64(1) - float64(bidamount)/float64(budget)
			f, _ := strconv.ParseFloat(strconv.FormatFloat(rate, 'f', 4, 64), 64)
			//不在0~0.6之间，不生成费率；只生成预算，中标金额舍弃，索引增加折扣率异常标识
			if f < 0 || f > 0.6 {
				delete(tmp, "bidamount")
				tmp["prate_flag"] = 1
			} else {
				tmp["project_rate"] = f
			}
		}
		if topscopeclass, ok := tmp["topscopeclass"].([]interface{}); ok {
			tc := []string{}
			m2 := map[string]bool{}
			for _, v := range topscopeclass {
				str := util.ObjToString(v)
				str = reg_letter.ReplaceAllString(str, "") // 去除字母
				if !m2[str] {
					m2[str] = true
					tc = append(tc, str)
				}
			}
			tmp["topscopeclass"] = tc
		}
		//不生索引字段
		delete(tmp, "package")
		delete(tmp, "infofield")

		bidopentime := util.Int64All(tmp["bidopentime"]) //开标日期
		fzb_publishtime := int64(0)                      //记录第一个招标信息的publishtime
		bidcycle_flag := false                           //判断是否已计算出标书表编制周期
		list := tmp["list"].([]interface{})
		for _, m := range list {
			tmpM := m.(map[string]interface{})
			//删除purchasing,review_experts
			delete(tmpM, "purchasing")
			delete(tmpM, "review_experts")
			if bidamount, ok := tmpM["bidamount"].(string); ok && len(bidamount) > 0 { //bidamount为string类型，转成float
				tmpB := util.Float64All(tmpM["bidamount"])
				tmpM["bidamount"] = tmpB
			}
			//projectscope截断
			listProjectscopeRune := []rune(util.ObjToString(tmpM["projectscope"]))
			if len(listProjectscopeRune) > 1000 {
				tmpM["projectscope"] = string(listProjectscopeRune[:1000])
			}
			//计算bidcycle标书表编制周期字段
			if !bidcycle_flag && bidopentime > 0 { //bidopentime>0证明list中有bidopentime，无则不用计算bidcycle
				if toptype := util.ObjToString(tmpM["toptype"]); toptype == "招标" {
					zb_bidopentime := util.Int64All(tmpM["bidopentime"])
					zb_publishtime := util.Int64All(tmpM["publishtime"])
					if zb_publishtime > 0 {
						if zb_bidopentime > 0 {
							if tmpTime := zb_bidopentime - zb_publishtime; tmpTime > 0 {
								f_day := float64(tmpTime) / float64(86400)
								day := math.Ceil(f_day)
								tmp["bidcycle"] = int(day)
								bidcycle_flag = true
							}
						}
						if fzb_publishtime == 0 { //仅赋值第一个招标信息的publishtime
							fzb_publishtime = zb_publishtime
						}
					}
				}
			}
		}
		//计算bidcycle标书表编制周期字段
		//list中招标信息中未能计算出bidcycle,用第一个招标信息的fzb_publishtime和外围bidopentime计算
		if !bidcycle_flag && bidopentime > 0 && fzb_publishtime > 0 {
			if tmpTime := bidopentime - fzb_publishtime; tmpTime > 0 {
				f_day := float64(tmpTime) / float64(86400)
				day := math.Ceil(f_day)
				tmp["bidcycle"] = int(day)
			}
		}
		//projectscope截断
		projectscopeRune := []rune(util.ObjToString(tmp["projectscope"]))
		if len(projectscopeRune) > 1000 {
			tmp["projectscope"] = string(projectscopeRune[:1000])
		}
		//		if s_budget := fmt.Sprint(tmp["budget"]); s_budget == "" || s_budget == "<nil>" || s_budget == "null" {
		//			tmp["budget"] = nil
		//		}
		//		if s_bidamount := fmt.Sprint(tmp["bidamount"]); s_bidamount == "" || s_bidamount == "<nil>" || s_bidamount == "null" {
		//			tmp["bidamount"] = nil
		//		}
		//go IS.Add("project")
		tmp["s_projectname"] = tmp["projectname"]
		arr[i] = tmp
		n++
		if i == savesizei-1 {
			savepool <- true
			tmps := arr
			go func(tmpn *[]map[string]interface{}) {
				defer func() {
					<-savepool
				}()
				elastic.BulkSave(index, itype, tmpn, true)
			}(&tmps)
			i = 0
			arr = make([]map[string]interface{}, savesizei)
		}
		if n%savesizei == 1000 {
			log.Println("当前:", n)
		}
		tmp = make(map[string]interface{})
	}

	if i > 0 {
		//util.Debug(arr)
		elastic.BulkSave(index, itype, &arr, true)
	}
	log.Println(mapInfo, "create project index...over", n)
}