fieldproject/proposed_project/merge.go

package main

import (
	util "app.yhyue.com/data_processing/common_utils"
	"app.yhyue.com/data_processing/common_utils/log"
	"app.yhyue.com/data_processing/common_utils/mongodb"
	"fmt"
	"go.uber.org/zap"
	"math/rand"
	"reflect"
	"sort"
	"strconv"
	"sync"
	"time"
	"unicode/utf8"
)

// @Description 参与合并流程
// @Author J 2022/10/20 13:57
func startProjectMerge(tmp map[string]interface{}, info *Info) {
	IDArr := getCompareIds(info.ProjectName, info.Site)
	isMerge := false
	var mergeProject *Project
	for _, v := range IDArr {
		mergeProject = v.P
		if info.Area != "" && mergeProject.Area != "" {
			if info.Area != "全国" && mergeProject.Area != "全国" && info.Area != mergeProject.Area {
				continue
			} else {
				isMerge = true
				break
			}
		} else {
			isMerge = true
			break
		}
	}
	if isMerge && mergeProject != nil {
		updateProject(tmp, *info, mergeProject)
	} else {
		_, pinfo := newProject(tmp, *info)
		AllPidMapLock.Lock()
		AllPidMap[info.Id] = &ID{P: pinfo}
		AllPidMapLock.Unlock()
		AllPnMapLock.Lock()
		res := AllPnMap[info.Site]
		if res != nil {
			res.Lock.Lock()
			res.Id[info.ProjectName] = info.Id
			res.Lock.Unlock()
		} else {
			res = &Pname{
				Id:   map[string]string{info.ProjectName: info.Id},
				Lock: sync.Mutex{},
			}
		}
		AllPnMap[info.Site] = res
		AllPnMapLock.Unlock()
	}
}

// @Description 通过项目名称找可以与之合并的项目集合
// @Author J 2022/10/20 14:22
func getCompareIds(pname, site string) (IDArr []*ID) {
	AllPnMapLock.Lock()
	res := AllPnMap[site]
	defer AllPnMapLock.Unlock()

	pid := ""
	IDArr = []*ID{} //项目信息
	if res != nil {
		res.Lock.Lock()
		pid = res.Id[pname]
		res.Lock.Unlock()

		AllPidMapLock.Lock()
		ID := AllPidMap[pid]
		AllPidMapLock.Unlock()
		if ID != nil {
			IDArr = append(IDArr, ID)
		}
	}
	return IDArr

}

func newProject(tmp map[string]interface{}, info Info) (string, *Project) {
	pId := mongodb.StringTOBsonId(info.Id)
	set := make(map[string]interface{})
	set["_id"] = pId
	for _, f := range FIELDS {
		if tmp[f] != nil && tmp[f] != "" {
			if reflect.TypeOf(tmp[f]).String() == "string" {
				if utf8.RuneCountInString(util.ObjToString(tmp[f])) < 1000 {
					set[f] = tmp[f]
				}
			} else {
				set[f] = tmp[f]
			}
		}
	}
	p1 := NewCachePinfo(info)
	set["firsttime"] = util.IntAll(info.Publishtime)
	set["lasttime"] = util.IntAll(info.Publishtime)
	set["ids"] = []string{info.Id}
	set["sourceinfourl"] = tmp["href"]
	set["follow_num"] = 1
	set["pici"] = time.Now().Unix()
	p1.FollowNum = 1
	now := time.Now().Unix()
	st := util.FormatDateByInt64(&now, util.Date_yyyyMMdd)
	parseSt := strconv.FormatInt(util.Int64All(st), 8)                                       // 转8进制
	rd := fmt.Sprintf("%04v", rand.New(rand.NewSource(time.Now().UnixNano())).Int63n(10000)) // 4位随机数
	set["proposed_number"] = fmt.Sprintf("NZJ%s%s", parseSt, rd)
	push := PushListInfo(tmp, info.Id)
	push["follow_num"] = 1
	set["list"] = []map[string]interface{}{
		push,
	}
	updatePool <- []map[string]interface{}{
		{
			"_id": pId,
		},
		{
			"$set": set,
		},
	}
	return info.Id, &p1
}

// @Description 合并字段信息
// @Author J 2022/10/20 13:57
func updateProject(tmp map[string]interface{}, info Info, pInfo *Project) {
	if BinarySearch(pInfo.Ids, info.Id) > -1 {
		log.Error("updateProject", zap.String("repeat", info.Id), zap.String("pid", mongodb.BsonIdToSId(pInfo.Id)))
		return
	}
	set := make(map[string]interface{})
	pInfo.Ids = append(pInfo.Ids, info.Id)

	pInfo.LastTime = info.Publishtime
	set["lasttime"] = info.Publishtime
	pInfo.FollowNum += 1
	set["follow_num"] = pInfo.FollowNum
	if pInfo.Owner == "" && info.Owner != "" {
		pInfo.Owner = info.Owner
		set["owner"] = info.Owner
	}
	if pInfo.ApproveCode == "" && info.ApproveCode != "" {
		pInfo.ApproveCode = info.ApproveCode
		set["approvecode"] = info.ApproveCode
	}
	if pInfo.ApproveNumber == "" && info.ApproveNumber != "" {
		pInfo.ApproveNumber = info.ApproveNumber
		set["approvenumber"] = info.ApproveNumber
	}
	if pInfo.TotalInvestment == "" && info.TotalInvestment != "" && utf8.RuneCountInString(info.TotalInvestment) < 1000 {
		pInfo.TotalInvestment = info.TotalInvestment
		set["total_investment"] = info.TotalInvestment
	}
	if pInfo.Funds == "" && info.Funds != "" && utf8.RuneCountInString(info.Funds) < 1000 {
		pInfo.Funds = info.Funds
		set["funds"] = info.Funds
	}
	if pInfo.ProjectAddr == "" && info.ProjectAddr != "" && utf8.RuneCountInString(info.ProjectAddr) < 1000 {
		pInfo.ProjectAddr = info.ProjectAddr
		set["projectaddr"] = info.ProjectAddr
	}
	if pInfo.ProjectPeriod == "" && info.ProjectPeriod != "" && utf8.RuneCountInString(info.ProjectPeriod) < 1000 {
		pInfo.ProjectPeriod = info.ProjectPeriod
		set["projectperiod"] = info.ProjectPeriod
	}
	if pInfo.ProjectScale == "" && info.ProjectScale != "" && utf8.RuneCountInString(info.ProjectScale) < 1000 {
		pInfo.ProjectPeriod = info.ProjectScale
		set["project_scale"] = info.ProjectScale
	}
	if pInfo.ProjectStartDate == 0 && info.ProjectStartDate > 0 {
		pInfo.ProjectStartDate = info.ProjectStartDate
		set["project_startdate"] = info.ProjectStartDate
	}
	if pInfo.ProjectCompleteDate == 0 && info.ProjectCompleteDate > 0 {
		pInfo.ProjectCompleteDate = info.ProjectCompleteDate
		set["project_completedate"] = info.ProjectCompleteDate
	}
	if pInfo.ConstructionArea == "" && info.ConstructionArea != "" && utf8.RuneCountInString(info.ConstructionArea) < 1000 {
		pInfo.ConstructionArea = info.ConstructionArea
		set["construction_area"] = info.ConstructionArea
	}
	if pInfo.FloorArea == "" && info.FloorArea != "" && utf8.RuneCountInString(info.FloorArea) < 1000 {
		pInfo.FloorArea = info.FloorArea
		set["floor_area"] = info.FloorArea
	}
	if pInfo.ProjectPerson == "" && info.ProjectPerson != "" {
		pInfo.ProjectPerson = info.ProjectPerson
		set["project_person"] = info.ProjectPerson
	}
	if pInfo.ProjectPhone == "" && info.ProjectPhone != "" {
		pInfo.ProjectPhone = info.ProjectPhone
		set["project_phone"] = info.ProjectPhone
	}

	for _, f := range []string{"ownerclass_code", "category_code", "nature_code", "project_stage_code"} {
		if c := util.ObjToString(tmp[f]); c != "" {
			set[f] = c
		}
	}
	set["pici"] = time.Now().Unix()
	push := PushListInfo(tmp, info.Id)
	push["follow_num"] = pInfo.FollowNum
	update := map[string]interface{}{}
	set["updatetime"] = time.Now().Unix()
	update["$set"] = set
	update["$push"] = map[string]interface{}{
		"list": push,
		"ids":  info.Id,
	}
	updateInfo := []map[string]interface{}{
		{
			"_id": mongodb.StringTOBsonId(pInfo.Id),
		},
		update,
	}
	updatePool <- updateInfo
}

// @Description list中存放的信息
// @Author J 2022/10/23 10:16
func PushListInfo(tmp map[string]interface{}, infoid string) map[string]interface{} {
	res := map[string]interface{}{
		"infoid": infoid,
	}
	for _, k := range INFOFIELDS {
		if tmp[k] != nil {
			res[k] = tmp[k]
		}
	}
	return res
}

func NewCachePinfo(info Info) Project {
	p1 := Project{
		Id:          info.Id,
		Ids:         []string{info.Id},
		ProjectName: info.ProjectName,
		ApproveCode: info.ApproveCode,
		Area:        info.Area,
		City:        info.City,
		District:    info.District,
	}
	return p1
}

//二分字符串查找
func BinarySearch(s []string, k string) int {
	sort.Strings(s)
	lo, hi := 0, len(s)-1
	for lo <= hi {
		m := (lo + hi) >> 1
		if s[m] < k {
			lo = m + 1
		} else if s[m] > k {
			hi = m - 1
		} else {
			return m
		}
	}
	return -1
}