data_field_dispose/data_project_wy/task.go

package main

import (
	"context"
	"fmt"
	"github.com/gogf/gf/v2/util/gconv"
	"go.mongodb.org/mongo-driver/bson"
	"jygit.jydev.jianyu360.cn/data_processing/common_utils/mongodb"
	"strings"
	"sync"
	"time"
)

type Transaction struct {
	Project_Id        string   `bson:"project_id"`
	Project_Name      string   `bson:"project_name"`
	Project_Budget    float64  `bson:"project_budget"`
	Project_Bidamount float64  `bson:"project_bidamount"`
	Project_Money     float64  `bson:"project_money"`
	Business_Type     string   `bson:"business_type"`
	Project_Bidstatus int      `bson:"project_bidstatus"`
	Info_Id           string   `bson:"info_id"`
	Info_Ids          []string `bson:"info_ids"`
	Information_Id    string   `bson:"information_id"`
	Buyer             string   `bson:"buyer"`
	Buyer_Id          string   `bson:"buyer_id"`
	Winner            []string `bson:"winner"`
	Winner_Id         []string `bson:"winner_id"`
	Agency            string   `bson:"agency"`
	Agency_Id         string   `bson:"agency_id"`
	Property_Form     []string `bson:"property_form"`
	SubClass          []string `bson:"subclass"`
	MultiPackage      int      `bson:"multipackage"`
	Area              string   `bson:"area"`
	City              string   `bson:"city"`
	District          string   `bson:"district"`
	ZbTime            int64    `bson:"zbtime"`
	JgTime            int64    `bson:"jgtime"`
	StartTime         int64    `bson:"starttime"`
	EndTime           int64    `bson:"endtime"`
	Create_Time       int64    `bson:"create_time"`
	Update_Time       int64    `bson:"update_time"`
	//
	From string `bson:"from"`
}

func IncTransactionDataFromBidAndPro() {
	IncTransactionDataFromBid() //bidding
	IncTransactionDataFromPro() //project
	return
	IncTransactionDataMgoToCkhAndEs() //mongodb迁移至clickhouse
}

// IncTransactionDataFromBid 增量bidding
func IncTransactionDataFromBid() {
	endTime := GetTime(-1) //前一天凌晨
	fmt.Println("开始执行增量采购意向、拟建信息", BidStartTime, endTime)
	if BidStartTime >= endTime {
		fmt.Println("增量bidding采购意向、拟建查询异常：", BidStartTime, endTime)
		return
	}
	query := map[string]interface{}{
		"comeintime": map[string]interface{}{
			"$gte": BidStartTime,
			"$lt":  endTime,
		},
	}
	fmt.Println("增量bidding采购意向query：", query)
	sess := MgoB.GetMgoConn()
	defer MgoB.DestoryMongoConn(sess)
	ch := make(chan bool, 5)
	wg := &sync.WaitGroup{}
	lock := &sync.Mutex{}
	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,
		//
		"owner":                 1,
		"s_topscopeclass":       1,
		"publishtime":           1,
		"toptype":               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
	count := 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.Int(tmp["extracttype"]) == -1 { //重复数据过滤
				return
			}
			toptype := gconv.String(tmp["toptype"])
			tag_topinformation := gconv.String(tmp["tag_topinformation"])
			tag_topinformation_ai := gconv.String(tmp["tag_topinformation_ai"])
			var business_type string
			var project_bidstatus int
			if toptype == "采购意向" { //采购意向数据
				if !strings.Contains(tag_topinformation, "物业") && !strings.Contains(tag_topinformation_ai, "物业") {
					return
				}
				business_type = "采购意向"
				project_bidstatus = 3
			} else if toptype == "拟建" {
				if strings.Contains(tag_topinformation, "物业") || strings.Contains(tag_topinformation_ai, "物业") {
					return
				}
				business_type = "新增项目"
				project_bidstatus = 4
			} else {
				return
			}
			result := DealTransactionForBid(tmp, business_type, project_bidstatus)
			lock.Lock()
			if len(result) > 0 {
				arr = append(arr, result)
				count++
			}
			if len(arr) > 50 {
				MgoPro.SaveBulk("projectset_wy", arr...)
				arr = []map[string]interface{}{}
			}
			lock.Unlock()
		}(tmp)
		if n%1000 == 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("执行增量采购意向、拟建信息完毕", BidStartTime, endTime, count)
	BidStartTime = endTime //替换
}

// DealTransactionForBid bidding采购意向数据处理
func DealTransactionForBid(tmp map[string]interface{}, business_type string, project_bidstatus int) map[string]interface{} {
	//基本信息封装
	id := mongodb.BsonIdToSId(tmp["_id"])
	buyer := gconv.String(tmp["buyer"])
	if buyer == "" {
		buyer = gconv.String(tmp["owner"])
	}
	winner := gconv.String(tmp["s_winner"])
	agency := gconv.String(tmp["agency"])
	property_form := []string{}
	if tmp["property_form"] != nil {
		property_form = gconv.Strings(tmp["property_form"])
	}
	bidamount := gconv.Float64(tmp["bidamount"])
	budget := gconv.Float64(tmp["budget"])
	money := bidamount
	if money <= 0 {
		money = budget
	}

	//物业分类
	subclass := []string{}
	if tag_subinformation := tmp["tag_subinformation"]; tag_subinformation != nil {
		subclass = gconv.Strings(tag_subinformation)
	} else if tag_subinformation_ai := tmp["tag_subinformation_ai"]; tag_subinformation_ai != nil {
		subclass = gconv.Strings(tag_subinformation_ai)
	}

	//TODO 情报库信息（待补充）
	information_id := ""
	starttime, endtime := int64(0), int64(0)
	//if project_bidstatus == 4 {//补充情报信息
	//	//bidId = "65fbf3f566cf0db42a2a99d2"
	//	info := FindInfomationData(id) //情报信息查询
	//	information_id = info.Id
	//	starttime = info.Starttime
	//	endtime = info.Endtime
	//}

	//TODO 查询法人库信息（待补充）
	winners := []string{}
	if winner != "" {
		winners = strings.Split(winner, ",")
	}
	buyer_id, agency_id := "", ""
	winner_ids := []string{}
	//buyer_id, agency_id, winner_ids := FindEntInfoData(id, buyer, agency, winners)
	//物业信息
	t := &Transaction{
		Project_Id:        id,
		Project_Name:      gconv.String(tmp["projectname"]),
		Project_Budget:    budget,
		Project_Bidamount: bidamount,
		Project_Money:     money,
		Business_Type:     business_type,
		Project_Bidstatus: project_bidstatus,
		Info_Id:           id,
		Info_Ids:          []string{id},
		Information_Id:    information_id,
		Buyer:             buyer,
		Winner:            winners,
		Agency:            agency,
		Buyer_Id:          buyer_id,
		Winner_Id:         winner_ids,
		Agency_Id:         agency_id,
		Property_Form:     property_form,
		SubClass:          subclass,
		MultiPackage:      gconv.Int(tmp["multipackage"]),
		Area:              gconv.String(tmp["area"]),
		City:              gconv.String(tmp["city"]),
		District:          gconv.String(tmp["district"]),
		ZbTime:            gconv.Int64(tmp["publishtime"]),
		JgTime:            int64(0),
		StartTime:         starttime,
		EndTime:           endtime,
		Create_Time:       time.Now().Unix(),
		Update_Time:       time.Now().Unix(),
		//
		From: "bidding",
	}
	result := map[string]interface{}{}
	infomation, _ := bson.Marshal(t)
	bson.Unmarshal(infomation, &result)
	return result
}

// IncTransactionDataFromProject 增量project
func IncTransactionDataFromPro() {
	endTime := GetTime(-1) //前一天凌晨
	fmt.Println("开始执行增量项目信息", ProStartTime, endTime)
	if ProStartTime >= endTime {
		fmt.Println("增量项目信息查询异常：", ProStartTime, endTime)
		return
	}
	query := map[string]interface{}{
		"pici": map[string]interface{}{
			"$gte": ProStartTime,
			"$lt":  endTime,
		},
	}
	fmt.Println("增量项目查询query：", query)
	sess := MgoPro.GetMgoConn()
	defer MgoPro.DestoryMongoConn(sess)
	ch := make(chan bool, 5)
	wg := &sync.WaitGroup{}
	lock := &sync.Mutex{}
	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,
		"pici":                  1,
		"ids":                   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
	count := 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 {
				count++
				update := []map[string]interface{}{
					{"project_id": tmp["_id"]},
					{"$set": result},
				}
				arr = append(arr, update)
			}
			if len(arr) > 50 {
				MgoPro.UpSertBulk("projectset_wy_back", arr...)
				arr = [][]map[string]interface{}{}
			}
			lock.Unlock()
		}(tmp)
		if n%1000 == 0 {
			fmt.Println("current:", n)
		}
		tmp = map[string]interface{}{}
	}
	wg.Wait()
	if len(arr) > 0 {
		MgoPro.UpSertBulk("projectset_wy_back", arr...)
		arr = [][]map[string]interface{}{}
	}
	fmt.Println("执行增量项目信息完毕", ProStartTime, endTime, count)
	ProStartTime = endTime //替换
}

// DealTransactionForPro project数据处理
func DealTransactionForPro(data map[string]interface{}) map[string]interface{} {
	//基本信息封装
	id := mongodb.BsonIdToSId(data["_id"])
	buyer := gconv.String(data["buyer"])
	winner := gconv.String(data["s_winner"])
	agency := gconv.String(data["agency"])
	zbtime := gconv.Int64(data["zbtime"])
	if zbtime == 0 {
		zbtime = gconv.Int64(data["firsttime"])
	}
	property_form := []string{}
	if data["property_form"] != nil {
		property_form = gconv.Strings(data["property_form"])
	}
	bidamount := gconv.Float64(data["bidamount"])
	budget := gconv.Float64(data["budget"])
	money := bidamount
	if money <= 0 {
		money = budget
	}

	//物业分类
	subclass := []string{}
	if tag_subinformation := data["tag_subinformation"]; tag_subinformation != nil {
		subclass = gconv.Strings(tag_subinformation)
	} else if tag_subinformation_ai := data["tag_subinformation_ai"]; tag_subinformation_ai != nil {
		subclass = gconv.Strings(tag_subinformation_ai)
	}

	//项目状态、商机类型
	business_type := ""
	project_bidstatus := 2
	bidstatus := gconv.String(data["bidstatus"])
	if bidstatus == "中标" || bidstatus == "成交" || bidstatus == "合同" {
		project_bidstatus = 1
		business_type = "存量项目"
	} else if bidstatus == "废标" || bidstatus == "流标" {
		project_bidstatus = 0
	} else if bidstatus == "拟建" {
		project_bidstatus = 4
	} else if bidstatus == "招标" {
		business_type = "招标项目"
	}
	//查询情报信息
	//bidId = "65fbf3f566cf0db42a2a99d2"
	ids := gconv.Strings(data["ids"])
	info := FindInfomationData(ids...) //情报信息查询

	//TODO 查询法人库信息（待补充）
	winners := []string{}
	if winner != "" {
		winners = strings.Split(winner, ",")
	}
	buyer_id, agency_id := "", ""
	winner_ids := []string{}
	//buyer_id, agency_id, winner_ids = FindEntInfoData(id, buyer, agency, winners)
	//物业信息
	t := &Transaction{
		Project_Id:        id,
		Project_Name:      gconv.String(data["projectname"]),
		Project_Budget:    budget,
		Project_Bidamount: bidamount,
		Project_Money:     money,
		Business_Type:     business_type,
		Project_Bidstatus: project_bidstatus,
		Info_Id:           gconv.String(data["sourceinfoid"]),
		Info_Ids:          ids,
		Information_Id:    info.Id,
		Buyer:             buyer,
		Winner:            winners,
		Agency:            agency,
		Buyer_Id:          buyer_id,
		Winner_Id:         winner_ids,
		Agency_Id:         agency_id,
		Property_Form:     property_form,
		SubClass:          subclass,
		MultiPackage:      gconv.Int(data["multipackage"]),
		Area:              gconv.String(data["area"]),
		City:              gconv.String(data["city"]),
		District:          gconv.String(data["district"]),
		ZbTime:            zbtime,
		JgTime:            gconv.Int64(data["jgtime"]),
		StartTime:         info.Starttime,
		EndTime:           info.Endtime,
		Create_Time:       time.Now().Unix(),
		Update_Time:       time.Now().Unix(),
		//
		From: "project",
	}
	result := map[string]interface{}{}
	infomation, _ := bson.Marshal(t)
	bson.Unmarshal(infomation, &result)
	return result
}

// IncTransactionDataMgoToCkhAndEs 数据迁移
func IncTransactionDataMgoToCkhAndEs() {
	/*
		数据根据update_time查询
		1、采购意向数据（from=bidding）只插入
		2、项目信息先查，有则更新，无则插入
	*/
	fmt.Println("开始执行迁移...")
	sess := MgoPro.GetMgoConn()
	defer MgoPro.DestoryMongoConn(sess)
	ch := make(chan bool, 2)
	wg := &sync.WaitGroup{}
	query := map[string]interface{}{
		"update_time": map[string]interface{}{
			"$gte": GetTime(0),
		},
	}
	it := sess.DB(MgoPro.DbName).C("projectset_wy").Find(&query).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()
			}()
			from := gconv.String(tmp["from"])
			delete(tmp, "from")     //无用字段删除
			delete(tmp, "_id")      //无用字段删除
			if !SaveDataToEs(tmp) { //保存、更新es
				fmt.Println("数据保存es失败，项目project_id", tmp["project_id"])
			}
			if from == "bidding" { //采购意向、拟建，插入
				SaveDataToClickHouse(tmp)
			} else { //项目信息，更新，插入
				UpdateOrSaveDataToClickHouse(tmp)
			}
		}(tmp)
		if n%100 == 0 {
			fmt.Println("current:", n)
		}
		tmp = map[string]interface{}{}
	}
	wg.Wait()
	fmt.Println("迁移结束...")
}

type Infomation struct {
	Id        string
	Starttime int64
	Endtime   int64
}

// FindInfomationData 情报信息查询
func FindInfomationData(ids ...string) (info Infomation) {
	for _, id := range ids {
		query := fmt.Sprintf(`SELECT id,starttime,endtime FROM %s WHERE datajson_id = ?`, Config.ClickHouse.DataBase+".information")
		rows, err := CkhTool.Query(context.Background(), query, id)
		if err != nil {
			continue
		}
		for rows.Next() {
			info = Infomation{}
			if err := rows.Scan(&info.Id, &info.Starttime, &info.Endtime); err != nil {
				fmt.Println("查询情报信息异常：", id, err)
			}
			if info.Id != "" {
				return
			}
			//break //目前只有一条结果
		}
	}
	return
}

// FindEntInfoData 法人信息查询
func FindEntInfoData(bid, buyer, agency string, winners []string) (buyer_id, agency_id string, winner_ids []string) {
	winner_ids = []string{}
	winnerMap := map[string]bool{} //记录所有中标单位
	values := []interface{}{}
	placeholders := []string{}
	if buyer != "" {
		placeholders = append(placeholders, "?")
		values = append(values, buyer)
	}
	if len(winners) > 0 {
		for _, w := range winners {
			winnerMap[w] = true
			placeholders = append(placeholders, "?")
			values = append(values, w)
		}
	}
	if agency != "" {
		placeholders = append(placeholders, "?")
		values = append(values, agency)
	}
	if len(values) == 0 {
		return
	}
	query := fmt.Sprintf(`SELECT id,company_name FROM %s WHERE company_name IN (%s)`, Config.ClickHouse.DataBase+".ent_info", strings.Join(placeholders, ","))
	rows, err := CkhTool.Query(context.Background(), query, values...)
	if err != nil {
		return
	}
	for rows.Next() {
		var id, company_name string
		if err := rows.Scan(&id, &company_name); err == nil {
			if company_name == buyer {
				buyer_id = id
			} else if company_name == agency {
				agency_id = id
			} else if winnerMap[company_name] {
				winner_ids = append(winner_ids, id)
			}
		} else {
			fmt.Println("查询情报信息异常：", err, bid)
		}
	}
	return
}

// UpdateOrSaveDataToClickHouse 判断clickhouse更新or保存
func UpdateOrSaveDataToClickHouse(data map[string]interface{}) (err error) {
	project_id := gconv.String(data["project_id"])
	count := FindClickHouseByProjectId(project_id) //查询
	if count > 0 {                                 //更新
		delete(data, "create_time") //不更新创建时间
		delete(data, "project_id")  //不更新项目id（主键）
		err = UpdateDataToClickHouse(data, map[string]interface{}{"project_id": project_id})
		if err != nil {
			fmt.Println("clickhouse更新失败", project_id, data)
		}
	} else { //插入
		err = SaveDataToClickHouse(data)
		if err != nil {
			fmt.Println("clickhouse保存失败", project_id, data)
		}
	}
	return
}

// SaveDataToClickHouse 数据保存clickhouse
func SaveDataToClickHouse(data map[string]interface{}) error {
	fields, placeholders := []string{}, []string{}
	values := []interface{}{}
	for k, v := range data {
		fields = append(fields, k)
		values = append(values, v)
		placeholders = append(placeholders, "?")
	}
	query := fmt.Sprintf("INSERT INTO %s (%s) VALUES (%s)", Config.ClickHouse.DataBase+".transaction_info", strings.Join(fields, ","), strings.Join(placeholders, ","))
	return CkhTool.Exec(context.Background(), query, values...)
}

// FindClickHouseByProjectId 根据条件count clickhouse
func FindClickHouseByProjectId(project_id string) int {
	query := fmt.Sprintf(`SELECT COUNT(1) FROM %s WHERE project_id = ?`, Config.ClickHouse.DataBase+".transaction_info")
	row := CkhTool.QueryRow(context.Background(), query, project_id)
	var count uint64
	row.Scan(&count)
	return gconv.Int(count)
}

// UpdateDataToClickHouse 数据更新clickhouse
func UpdateDataToClickHouse(data, querys map[string]interface{}) error {
	sets := []string{}
	values := []interface{}{}
	for k, v := range data {
		sets = append(sets, fmt.Sprintf("%s=?", k))
		values = append(values, v)
	}
	qs := []string{}
	for k, v := range querys {
		qs = append(qs, fmt.Sprintf("%s=?", k))
		values = append(values, v)
	}
	query := fmt.Sprintf("ALTER TABLE %s UPDATE %s WHERE %s", Config.ClickHouse.DataBase+".transaction_info", strings.Join(sets, ","), strings.Join(qs, ","))
	//query := `ALTER TABLE information.transaction_info UPDATE update_time = ? WHERE project_id = '5c9ee78ca5cb26b9b7fd0b57'`
	return CkhTool.Exec(context.Background(), query, values...)
}

// SaveDataToEs es存储
func SaveDataToEs(data map[string]interface{}) bool {
	tmp := map[string]interface{}{}
	for k, v := range data {
		if k == "project_id" {
			k = "_id"
		}
		tmp[k] = v
	}
	err, result := Es.GetById(Config.Es.Index, gconv.String(tmp["_id"]))
	if err == nil && len(result) > 0 { //存在，更新
		tmp["create_time"] = result["create_time"] //不更新create_time
	}
	return Es.Save(Config.Es.Index, tmp)
}

func FindEntInfoData2(bid, buyer, agency string, winners []string) (buyer_id, agency_id string, winner_ids []string) {
	query := fmt.Sprintf(`SELECT id FROM %s WHERE company_name = ?`, Config.ClickHouse.DataBase+".ent_info")
	if buyer != "" {
		buyer_id = GetClickHouseData(bid, query, buyer)
	}
	if agency != "" {
		agency_id = GetClickHouseData(bid, query, agency)
	}
	if len(winners) > 0 {
		for _, w := range winners {
			winner_id := GetClickHouseData(bid, query, w)
			if winner_id != "" {
				winner_ids = append(winner_ids, winner_id)
			}
		}
	}
	return
}

func GetClickHouseData(bid, query, value string) string {
	rows, err := CkhTool.Query(context.Background(), query, value)
	if err != nil {
		return ""
	}
	for rows.Next() {
		var id string
		if err := rows.Scan(&id); err == nil {
			return id
		} else {
			fmt.Println("查询情报信息异常：", err, bid)
		}
	}
	return ""
}

/*// SaveTransactionData 保存增量物业信息
func SaveTransactionData() {
	fmt.Println("save projectset_wy...")
	savearr := make([]map[string]interface{}, 100)
	indexdb := 0
	for {
		select {
		case v := <-TransactionSaveCache:
			savearr[indexdb] = v
			indexdb++
			if indexdb == 100 {
				Transaction_Ch <- true
				go func(tmp []map[string]interface{}) {
					defer func() {
						<-Transaction_Ch
					}()
					MgoPro.SaveBulk("projectset_wy", tmp...)
				}(savearr)
				savearr = make([]map[string]interface{}, 100)
				indexdb = 0
			}
		case <-time.After(30 * time.Second):
			if indexdb > 0 {
				Transaction_Ch <- true
				go func(tmp []map[string]interface{}) {
					defer func() {
						<-Transaction_Ch
					}()
					MgoPro.SaveBulk("projectset_wy", tmp...)
				}(savearr[:indexdb])
				savearr = make([]map[string]interface{}, 100)
				indexdb = 0
			}
		}
	}
}*/