data_field_dispose/data_project_wy/history.go

package main

import (
	"fmt"
	"github.com/gogf/gf/v2/util/gconv"
	"jygit.jydev.jianyu360.cn/data_processing/common_utils/mongodb"
	"strings"
	"sync"
)

// HisTransactionDataFromBid 历史bidding（指定截止comeintime，采购意向）
func HisTransactionDataFromBid() {
	sess := MgoB.GetMgoConn()
	defer MgoB.DestoryMongoConn(sess)
	ch := make(chan bool, 10)
	wg := &sync.WaitGroup{}
	lock := &sync.Mutex{}
	query := map[string]interface{}{
		"toptype": "采购意向",
	}
	fields := map[string]interface{}{
		"projectname":   1,
		"budget":        1,
		"bidamount":     1,
		"buyer":         1,
		"s_winner":      1,
		"agency":        1,
		"property_form": 1,
		"multipackage":  1,
		"area":          1,
		"city":          1,
		"district":      1,
		//
		"publishtime":           1,
		"comeintime":            1,
		"extracttype":           1,
		"tag_subinformation":    1,
		"tag_subinformation_ai": 1,
		"tag_topinformation":    1,
		"tag_topinformation_ai": 1,
	}
	arr := []map[string]interface{}{}
	it := sess.DB(MgoB.DbName).C("bidding").Find(&query).Select(&fields).Iter()
	n := 0
	for tmp := make(map[string]interface{}); it.Next(tmp); n++ {
		ch <- true
		wg.Add(1)
		go func(tmp map[string]interface{}) {
			defer func() {
				<-ch
				wg.Done()
			}()
			if gconv.Int64(tmp["comeintime"]) >= 1713196800 { //截止时间1713196800
				return
			}
			if gconv.Int(tmp["extracttype"]) == -1 { //重复数据过滤
				return
			}
			if tmp["tag_topinformation"] == nil && tmp["tag_topinformation_ai"] == nil { //无效数据过滤
				return
			}
			result := DealTransactionForBid(tmp, "采购意向", 3)
			lock.Lock()
			if len(result) > 0 {
				arr = append(arr, result)
			}
			if len(arr) > 50 {
				MgoPro.SaveBulk("projectset_wy", arr...)
				arr = []map[string]interface{}{}
			}
			lock.Unlock()
		}(tmp)
		if n%10000 == 0 {
			fmt.Println("current:", n)
		}
		tmp = map[string]interface{}{}
	}
	wg.Wait()
	if len(arr) > 0 {
		MgoPro.SaveBulk("projectset_wy", arr...)
		arr = []map[string]interface{}{}
	}
	fmt.Println("结束")
}

// HisTransactionDataFromBid2 历史bidding（指定截止comeintime，新增项目）
func HisTransactionDataFromBid2() {
	sess := MgoB.GetMgoConn()
	defer MgoB.DestoryMongoConn(sess)
	ch := make(chan bool, 20)
	wg := &sync.WaitGroup{}
	lock := &sync.Mutex{}
	query := map[string]interface{}{
		"comeintime": map[string]interface{}{
			"$gte": 1713715200,
			"$lt":  1713801600,
		},
		"toptype": "拟建",
	}
	fields := map[string]interface{}{
		"projectname":   1,
		"budget":        1,
		"bidamount":     1,
		"buyer":         1,
		"s_winner":      1,
		"agency":        1,
		"property_form": 1,
		"multipackage":  1,
		"area":          1,
		"city":          1,
		"district":      1,
		//
		"s_topscopeclass":       1,
		"publishtime":           1,
		"toptype":               1,
		"comeintime":            1,
		"extracttype":           1,
		"tag_subinformation":    1,
		"tag_subinformation_ai": 1,
		"tag_topinformation":    1,
		"tag_topinformation_ai": 1,
	}
	arr := []map[string]interface{}{}
	it := sess.DB(MgoB.DbName).C("bidding").Find(&query).Select(&fields).Iter()
	n := 0
	for tmp := make(map[string]interface{}); it.Next(tmp); n++ {
		ch <- true
		wg.Add(1)
		go func(tmp map[string]interface{}) {
			defer func() {
				<-ch
				wg.Done()
			}()
			//comeintime := gconv.Int64(tmp["comeintime"])
			//if comeintime < 1609430400 || comeintime >= 1713715200 {
			//	return
			//}
			if gconv.Int(tmp["extracttype"]) == -1 { //重复数据过滤
				return
			}
			if s_topscopeclass := gconv.String(tmp["s_topscopeclass"]); !strings.Contains(s_topscopeclass, "建筑工程") { //排除非建筑工程
				return
			}
			if tag_topinformation := gconv.String(tmp["tag_topinformation"]); strings.Contains(tag_topinformation, "物业") { //排除物业
				return
			} else if tag_topinformation_ai := gconv.String(tmp["tag_topinformation_ai"]); strings.Contains(tag_topinformation_ai, "物业") {
				return
			}
			//if tmp["tag_topinformation"] != nil || tmp["tag_topinformation_ai"] != nil { //不包含物业
			//	return
			//}
			project_bidstatus := 4 //拟建
			business_type := "新增项目"
			result := DealTransactionForBid(tmp, business_type, project_bidstatus)
			lock.Lock()
			if len(result) > 0 {
				arr = append(arr, result)
			}
			if len(arr) > 50 {
				MgoPro.SaveBulk("projectset_wy_nj", arr...)
				arr = []map[string]interface{}{}
			}
			lock.Unlock()
		}(tmp)
		if n%10000 == 0 {
			fmt.Println("current:", n)
		}
		tmp = map[string]interface{}{}
	}
	wg.Wait()
	if len(arr) > 0 {
		MgoPro.SaveBulk("projectset_wy_nj", arr...)
		arr = []map[string]interface{}{}
	}
	fmt.Println("结束")
}

// HisTransactionDataFromProject 历史project（指定截止pici:1713196800）
func HisTransactionDataFromProject() {
	sess := MgoPro.GetMgoConn()
	defer MgoPro.DestoryMongoConn(sess)
	ch := make(chan bool, 20)
	wg := &sync.WaitGroup{}
	lock := &sync.Mutex{}
	query := map[string]interface{}{
		"pici": map[string]interface{}{
			"$lt": 1713196800,
			//"$gt": 1711900800,
		},
	}
	fields := map[string]interface{}{
		"projectname":   1,
		"budget":        1,
		"bidamount":     1,
		"buyer":         1,
		"s_winner":      1,
		"agency":        1,
		"property_form": 1,
		"multipackage":  1,
		"area":          1,
		"city":          1,
		"district":      1,
		"zbtime":        1,
		"jgtime":        1,
		"bidstatus":     1,
		//
		"firsttime":             1,
		"ids":                   1,
		"pici":                  1,
		"sourceinfoid":          1,
		"tag_subinformation":    1,
		"tag_subinformation_ai": 1,
		"tag_topinformation":    1,
		"tag_topinformation_ai": 1,
	}
	arr := []map[string]interface{}{}
	it := sess.DB(MgoPro.DbName).C("projectset_20230904").Find(&query).Select(&fields).Iter()
	n := 0
	for tmp := make(map[string]interface{}); it.Next(tmp); n++ {
		ch <- true
		wg.Add(1)
		go func(tmp map[string]interface{}) {
			defer func() {
				<-ch
				wg.Done()
			}()
			if tmp["tag_topinformation"] == nil && tmp["tag_topinformation_ai"] == nil { //无效数据过滤
				return
			}
			result := DealTransactionForPro(tmp)
			lock.Lock()
			if len(result) > 0 {
				arr = append(arr, result)
			}
			if len(arr) > 50 {
				MgoPro.SaveBulk("projectset_wy_back", arr...)
				arr = []map[string]interface{}{}
			}
			lock.Unlock()
		}(tmp)
		if n%10000 == 0 {
			fmt.Println("current:", n)
		}
		tmp = map[string]interface{}{}
	}
	wg.Wait()
	if len(arr) > 0 {
		MgoPro.SaveBulk("projectset_wy_back", arr...)
		arr = []map[string]interface{}{}
	}
	fmt.Println("结束")
}

// HisTransactionDataAddInformation 补充字段信息
func HisTransactionDataAddInformation() {
	sess := MgoPro.GetMgoConn()
	defer MgoPro.DestoryMongoConn(sess)
	ch := make(chan bool, 5)
	wg := &sync.WaitGroup{}
	lock := &sync.Mutex{}
	query := map[string]interface{}{
		//"_id": map[string]interface{}{
		//	"$gte": mongodb.StringTOBsonId("6627289319c5408c478125d4"),
		//},
	}
	it := sess.DB(MgoPro.DbName).C("projectset_wy").Find(&query).Iter()
	n := 0
	arr := [][]map[string]interface{}{}
	for tmp := make(map[string]interface{}); it.Next(tmp); n++ {
		ch <- true
		wg.Add(1)
		go func(tmp map[string]interface{}) {
			defer func() {
				<-ch
				wg.Done()
			}()
			id := mongodb.BsonIdToSId(tmp["_id"])
			update := []map[string]interface{}{
				{"_id": mongodb.StringTOBsonId(id)},
			}
			set := map[string]interface{}{}
			//法人信息
			buyer_id, agency_id, winner_ids := FindEntInfoData(id, gconv.String(tmp["buyer"]), gconv.String(tmp["agency"]), gconv.Strings(tmp["winner"]))
			set["buyer_id"] = buyer_id
			set["agency_id"] = agency_id
			set["winner_id"] = winner_ids

			//项目信息补充业态
			//if from := gconv.String(tmp["from"]); from == "project" {
			//	project_id := gconv.String(tmp["project_id"])
			//	pro, _ := MgoPro.FindById("projectset_20230904", project_id, map[string]interface{}{"property_form": 1})
			//	if len(*pro) > 0 && (*pro)["property_form"] != nil {
			//		set["property_form"] = (*pro)["property_form"]
			//	}
			//}
			delete(tmp, "from") //无用字段删除
			delete(tmp, "_id")  //无用字段删除
			tmp["buyer_id"] = buyer_id
			tmp["agency_id"] = agency_id
			tmp["winner_id"] = winner_ids
			if !SaveDataToEs(tmp) { //保存、更新es
				fmt.Println("数据保存es失败，数据类型  项目project_id", tmp["project_id"])
			}
			var err error
			err = UpdateOrSaveDataToClickHouse(tmp)
			if err != nil {
				fmt.Println("数据迁移失败，数据类型 项目project_id", tmp["project_id"], err)
			}
			//更新
			update = append(update, map[string]interface{}{"$set": set})
			lock.Lock()
			arr = append(arr, update)
			if len(arr) > 100 {
				MgoPro.UpdateBulk("projectset_wy", arr...)
				arr = [][]map[string]interface{}{}
			}
			lock.Unlock()
		}(tmp)
		if n%100 == 0 {
			fmt.Println("current:", n)
		}
		tmp = map[string]interface{}{}
	}
	wg.Wait()
	if len(arr) > 0 {
		MgoPro.UpdateBulk("projectset_wy", arr...)
		arr = [][]map[string]interface{}{}
	}
	fmt.Println("迁移结束...")
}