timed_tasks/yg/main.go

package main

import (
	"log"
	"reflect"
	"regexp"
	"strconv"
	"strings"
	"time"

	"go.mongodb.org/mongo-driver/bson/primitive"

	elastic "app.yhyue.com/moapp/jybase/es"
	"github.com/gogf/gf/v2/util/gconv"
	"github.com/robfig/cron"
	common "jygit.jydev.jianyu360.cn/data_processing/common_utils"
)

var (
	Bidding, Mgo      *MongodbSim
	Es                elastic.Es
	Es2               elastic.Es
	cfg               = new(Config)
	ClearHtml         = regexp.MustCompile("<[^>]*>")
	BiddingLevelField = make(map[string]map[string]string)
	TimeV1            = regexp.MustCompile("^(\\d{4})[年.]?$")
	TimeV2            = regexp.MustCompile("^(\\d{4})[年./-]?(\\d{1,2})[月./-]?$")
	TimeV3            = regexp.MustCompile("^(\\d{4})[年./-]?(\\d{1,2})[月./-]?(\\d{1,2})[日]?$")
	TimeClear         = regexp.MustCompile("[年|月|日|/|.|-]")
)

func init() {
	common.ReadConfig(&cfg)
	Mgo = &MongodbSim{
		// MongodbAddr: "172.17.4.86:27080",
		MongodbAddr: "192.168.3.166:27082",
		DbName:      "yantianlei",
		Size:        20,
		UserName:    "",
		Password:    "",
	}
	Mgo.InitPool()
	Bidding = &MongodbSim{
		// MongodbAddr: "172.17.4.86:27080",
		MongodbAddr: "192.168.3.166:27082",
		DbName:      "yantianlei",
		Size:        20,
		UserName:    "",
		Password:    "",
	}
	Bidding.InitPool()
	// Es = elastic.NewEs("07", "http://127.0.0.1:9801", 20, "jybid", "Top2023_JEB01i@31")
	Es2 = elastic.NewEs("07", "http://192.168.3.241:9205,http://192.168.3.149:9200", 20, "", "")
}

func main() {
	c := cron.New()
	c.AddFunc("0 */10 * * * ?", run)
	c.Start()
	ch := make(chan bool, 1)
	<-ch
}

func run() {
	session := Mgo.GetMgoConn()
	lastId := cfg.LastId
	query := map[string]interface{}{}
	if lastId != "" {
		query["_id"] = map[string]interface{}{"$gt": StringTOBsonId(lastId)}
	}
	log.Println("query :", query)
	defer Mgo.DestoryMongoConn(session)
	count := 0
	iter := session.DB("qfw").C("bidding_yg").Find(&query).Sort("_id").Iter()
	thisData := map[string]interface{}{}
	for {
		if !iter.Next(&thisData) {
			break
		}
		count++
		if count%500 == 0 {
			log.Println("COUNT ", count)
		}
		id := common.ObjToString(thisData["id"])
		source := common.ObjToString(thisData["source"])
		data := Bidding.FindById("bidding", id)
		if data != nil && len(data) > 0 {
			public_type := "平台发布"
			domain_firsttype, domain_secondtype, domain_thirdtype := "", "", ""
			if data["gov_classify"] != nil {
				if gov_classify, ok := data["gov_classify"].(map[string]interface{}); ok {
					root := common.ObjToString(gov_classify["root"])
					if root != "" {
						rootArr := strings.Split(root, "/")
						domain_firsttype = rootArr[0]
						if len(rootArr) > 1 {
							domain_secondtype = rootArr[1]
						}
						if len(rootArr) > 2 {
							domain_thirdtype = rootArr[2]
						}
					}
				}
			}
			deliver_area, deliver_city, deliver_district := data["area"], data["city"], data["district"]
			if source == "user" {
				public_type = "用户发布"
				deliver_area, deliver_city, deliver_district = data["deliver_area"], data["deliver_city"], data["deliver_district"]
			}
			newData := GetEsField(data)
			newData["deliver_area"] = deliver_area
			newData["deliver_city"] = deliver_city
			newData["deliver_district"] = deliver_district
			newData["domain_firsttype"] = domain_firsttype
			newData["domain_secondtype"] = domain_secondtype
			newData["domain_thirdtype"] = domain_thirdtype
			newData["public_type"] = public_type
			newData["source_id"] = id
			Bidding.UpdateById("bidding", id, map[string]interface{}{"$set": map[string]interface{}{
				"deliver_area":      deliver_area,
				"deliver_city":      deliver_city,
				"deliver_district":  deliver_district,
				"domain_firsttype":  domain_firsttype,
				"domain_secondtype": domain_secondtype,
				"domain_thirdtype":  domain_thirdtype,
			}})
			if newData["purchasinglist"] != nil {
				purchasinglist := common.ObjArrToMapArr(newData["purchasinglist"].(primitive.A))
				if source == "user" && len(purchasinglist) > 1 {
					newDatas := newData
					itemMap := map[string]string{}
					itemArr := []string{}
					for _, v := range purchasinglist {
						itemname := common.ObjToString(v["itemname"])
						if itemname != "" {
							itemMap[itemname] = "1"
						}
					}
					for k, _ := range itemMap {
						itemArr = append(itemArr, k)
					}
					citys := gconv.String(deliver_city)
					if citys == "" {
						citys = gconv.String(deliver_area)
					}
					newDatas["title"] = strings.Join(itemArr, "/") + "-" + citys
					mid := primitive.NewObjectID()
					newDatas["_id"] = mid
					Mgo.Save("bidding_yg_info", newDatas)
					newDatas["_id"] = mid.Hex()
					Es.Save("bidding_yg", "", newDatas)
				} else {
					for _, v := range purchasinglist {
						newDatas := newData
						itemname := common.ObjToString(v["itemname"])
						if itemname != "" {
							citys := gconv.String(deliver_city)
							if citys == "" {
								citys = gconv.String(deliver_area)
							}
							newDatas["title"] = itemname + "-" + gconv.String(v["number"]) + gconv.String(v["unitname"]) + "-" + citys
						}
						mid := primitive.NewObjectID()
						newDatas["_id"] = mid
						Mgo.Save("bidding_yg_info", newDatas)
						newDatas["_id"] = mid.Hex()
						Es.Save("bidding_yg", "", newDatas)
					}
				}
			} else {
				mid := primitive.NewObjectID()
				newData["_id"] = mid
				Mgo.Save("bidding_yg_info", newData)
				newData["_id"] = mid.Hex()
				Es.Save("bidding_yg", "", newData)
			}
		}
		cfg.LastId = BsonTOStringId(thisData["_id"])
		thisData = map[string]interface{}{}
	}
	common.WriteSysConfig(&cfg)
}

func GetEsField(tmp map[string]interface{}) map[string]interface{} {
	newTmp := tmp
	for field, _ := range tmp {
		if tmp[field] != nil { //
			if field == "purchasinglist" { //标的物处理
				purchasinglist_new := []map[string]interface{}{}
				if pcl, _ := tmp[field].([]interface{}); len(pcl) > 0 {
					for _, ls := range pcl {
						lsm_new := make(map[string]interface{})
						lsm := ls.(map[string]interface{})
						for pf, pftype := range BiddingLevelField[field] {
							lsmv := lsm[pf]
							if lsmv != nil && reflect.TypeOf(lsmv).String() == pftype {
								lsm_new[pf] = lsm[pf]
							}
						}
						if lsm_new != nil && len(lsm_new) > 0 {
							purchasinglist_new = append(purchasinglist_new, lsm_new)
						}
					}
				}
				if len(purchasinglist_new) > 0 {
					newTmp[field] = purchasinglist_new
				}
			} else if field == "procurementlist" {
				if tmp["procurementlist"] != nil {
					var arr []interface{}
					plist := tmp["procurementlist"].([]interface{})
					for _, p := range plist {
						p1 := p.(map[string]interface{})
						p2 := make(map[string]interface{})
						for k, v := range BiddingLevelField[field] {
							if k == "projectname" && common.ObjToString(p1[k]) == "" {
								p2[k] = common.ObjToString(tmp["projectname"])
							} else if k == "buyer" && common.ObjToString(p1[k]) == "" && common.ObjToString(tmp["buyer"]) != "" {
								p2[k] = common.ObjToString(tmp["buyer"])
							} else if k == "expurasingtime" && common.ObjToString(p1[k]) != "" {
								res := getMethod(common.ObjToString(p1[k]))
								if res != 0 {
									p2[k] = res
								}
							} else if p1[k] != nil && reflect.TypeOf(p1[k]).String() == v {
								p2[k] = p1[k]
							}

						}
						arr = append(arr, p2)
					}
					if len(arr) > 0 {
						newTmp[field] = arr
					}
				}
			} else if field == "projectscope" {
				ps, _ := tmp["projectscope"].(string)
				newTmp["projectscope"] = ps
			} else if field == "winnerorder" { //中标候选
				winnerorder_new := []map[string]interface{}{}
				if winnerorder, _ := tmp[field].([]interface{}); len(winnerorder) > 0 {
					for _, win := range winnerorder {
						winMap_new := make(map[string]interface{})
						winMap := win.(map[string]interface{})
						for wf, wftype := range BiddingLevelField[field] {
							wfv := winMap[wf]
							if wfv != nil && reflect.TypeOf(wfv).String() == wftype {
								if wf == "sort" && common.Int64All(wfv) > 100 {
									continue
								}
								winMap_new[wf] = winMap[wf]
							}
						}
						if winMap_new != nil && len(winMap_new) > 0 {
							winnerorder_new = append(winnerorder_new, winMap_new)
						}
					}
				}
				if len(winnerorder_new) > 0 {
					newTmp[field] = winnerorder_new
				}
			} else if field == "qualifies" {
				//项目资质
				qs := []string{}
				if q, _ := tmp[field].([]interface{}); len(q) > 0 {
					for _, v := range q {
						v1 := v.(map[string]interface{})
						qs = append(qs, common.ObjToString(v1["key"]))
					}
				}
				if len(qs) > 0 {
					newTmp[field] = strings.Join(qs, ",")
				}
			} else if field == "bidopentime" {
				if tmp[field] != nil && tmp["bidendtime"] == nil {
					newTmp["bidendtime"] = tmp[field]
					newTmp[field] = tmp[field]
				} else if tmp[field] == nil && tmp["bidendtime"] != nil {
					newTmp["bidendtime"] = tmp[field]
					newTmp[field] = tmp["bidendtime"]
				} else {
					if tmp["bidopentime"] != nil {
						newTmp[field] = tmp["bidopentime"]
					}
				}
			} else if field == "detail" { //过滤
				detail, _ := tmp[field].(string)
				detail = ClearHtml.ReplaceAllString(detail, "")
				newTmp[field] = common.ObjToString(tmp["title"]) + " " + detail
			} else if field == "topscopeclass" || field == "entidlist" {
				newTmp[field] = tmp[field]
			} else if field == "_id" {
				newTmp["_id"] = BsonTOStringId(tmp["_id"])
				newTmp["id"] = BsonTOStringId(tmp["_id"])
			} else if field == "publishtime" || field == "comeintime" {
				//字段类型不正确，特别处理
				if tmp[field] != nil && common.Int64All(tmp[field]) > 0 {
					newTmp[field] = common.Int64All(tmp[field])
				}
			} else if field == "package" {
				delete(newTmp, "package")
			} else if field == "infoformat" {
				newTmp[field] = tmp[field]
			}
		}
	}
	newTmp["pici"] = time.Now().Unix()
	return newTmp
}

func InitEsBiddingField() {
	info, _ := Bidding.Find("bidding_processing_field", map[string]interface{}{"stype": "bidding"}, nil, nil)
	if len(info) > 0 {
		for _, m := range info {
			if common.IntAll(m["level"]) == 2 {
				pfield := common.ObjToString(m["pfield"])
				pfieldMap := BiddingLevelField[pfield]
				if pfieldMap == nil {
					pfieldMap = make(map[string]string, 0)
				}
				pfieldMap[common.ObjToString(m["field"])] = common.ObjToString(m["ftype"])
				BiddingLevelField[pfield] = pfieldMap
			}
		}
	}
	log.Println("BiddingLevelField es 二级字段数量", len(BiddingLevelField))
}

func getMethod(str string) int64 {
	// Handle "YYYY" format
	if TimeV1.MatchString(str) {
		arr := TimeV1.FindStringSubmatch(str)
		st := arr[1] + "0000"
		parseInt, err := strconv.ParseInt(st, 10, 64)
		if err == nil {
			return parseInt
		}
	}

	// Handle "YYYYMM" or "YYYY/MM" or "YYYY-MM" or "YYYY.MM" format
	if TimeV2.MatchString(str) {
		arr := TimeV2.FindStringSubmatch(str)
		year := arr[1]
		month := arr[2]
		if len(month) == 1 {
			month = "0" + month
		}
		str2 := year + month + "00"
		parseInt, err := strconv.ParseInt(str2, 10, 64)
		if err == nil {
			return parseInt
		}
	}

	// Handle "YYYYMMDD" or "YYYY/MM/DD" or "YYYY-MM-DD" or "YYYY.MM.DD" format
	if TimeV3.MatchString(str) {
		match := TimeV3.FindStringSubmatch(str)
		if len(match) >= 4 {
			year := match[1]
			month := match[2]
			day := match[3]
			if len(month) == 1 {
				month = "0" + month
			}
			if len(day) == 1 {
				day = "0" + day
			}
			dateStr := year + month + day
			parseInt, err := strconv.ParseInt(dateStr, 10, 64)
			if err == nil {
				return parseInt
			}
		}
	}

	return 0
}