jyps/project_portrait/main.go

package main

import (
	"context"
	"encoding/json"
	"fmt"
	"github.com/olivere/elastic/v7"
	"io"
	util "jygit.jydev.jianyu360.cn/data_processing/common_utils"
	es "jygit.jydev.jianyu360.cn/data_processing/common_utils/elastic"
	"jygit.jydev.jianyu360.cn/data_processing/common_utils/mongodb"
	"log"
	"time"
)

var (
	saveEsPool    = make(chan map[string]interface{}, 5000) //保存binding数据到es
	saveEsSp      = make(chan bool, 5)
	EsBulkSize    = 50
	Es            *es.Elastic
	portraitIndex = "project_portrait"     // 画像索引
	portraitMgo   = "wcc_project_portrait" // MongoDB 的表名
)

type PortraitData struct {
	Buyer        string  `json:"buyer"`
	Area         string  `json:"area"`
	City         string  `json:"city"`
	Class        string  `json:"class"`
	BusinessType string  `json:"business_type"`
	Lasttime     int64   `json:"lasttime"`
	ProjectCount int64   `json:"project_count"`
	ProjectMoney float64 `json:"project_money"`
}

func Init() {
	//Es = &es.Elastic{
	//	//S_esurl: "http://127.0.0.1:19908",
	//	S_esurl:  "http://172.17.4.184:19908",
	//	I_size:   5,
	//	Username: "jybid",
	//	Password: "Top2023_JEB01i@31",
	//}
	//Es.InitElasticSize()
	//测试环境
	Es = &es.Elastic{
		S_esurl:  "http://192.168.3.149:9201",
		I_size:   5,
		Username: "",
		Password: "",
	}
	Es.InitElasticSize()
}
func main() {
	Init()
	go SaveEsMethod()
	//dealAllData()
	allDataEs() //处理存量数据到es

	ch := make(chan bool, 1)
	<-ch
}

// dealIncData 处理增量数据
func dealIncData() {

	now := time.Now()
	yesterday := time.Date(now.Year(), now.Month(), now.Day()-1, 0, 0, 0, 0, now.Location())
	today := time.Date(now.Year(), now.Month(), now.Day(), 0, 0, 0, 0, now.Location())
	q := map[string]interface{}{
		"pici": map[string]interface{}{
			"$gt":  yesterday.Unix(),
			"$lte": today.Unix(),
		},
	}
	log.Println(q)

}

// dealAllData 处理存量数据，
func dealAllData() {
	/**
	循环采购单位存量数据，
	*/
	url := "http://172.17.4.184:19908"
	//url := "http://127.0.0.1:19908"
	username := "jybid"
	password := "Top2023_JEB01i@31"
	index := "buyer" //索引名称
	//index := "projectset" //索引名称
	// 创建 Elasticsearch 客户端
	client, err := elastic.NewClient(
		elastic.SetURL(url),
		elastic.SetBasicAuth(username, password),
		elastic.SetSniff(false),
	)
	if err != nil {
		log.Fatalf("创建 Elasticsearch 客户端失败：%s", err)
	}

	MgoB := &mongodb.MongodbSim{
		MongodbAddr: "172.17.189.140:27080",
		//MongodbAddr: "127.0.0.1:27083",
		Size:     10,
		DbName:   "qfw",
		UserName: "SJZY_RWbid_ES",
		Password: "SJZY@B4i4D5e6S",
		//Direct:      true,
	}
	MgoB.InitPool()

	//query := elastic.NewBoolQuery().
	//	Must(elastic.NewTermQuery("toptype", "结果")).
	//	Must(elastic.NewTermQuery("subtype", "招标"))

	ctx := context.Background()
	//开始滚动搜索
	scrollID := ""
	scroll := "10m"
	searchSource := elastic.NewSearchSource().
		//Query(query).
		Size(10000).
		Sort("_doc", true) //升序排序
	//Sort("_doc", false) //降序排序

	searchService := client.Scroll(index).
		Size(10000).
		Scroll(scroll).
		SearchSource(searchSource)

	res, err := searchService.Do(ctx)

	if err != nil {
		if err == io.EOF {
			fmt.Println("没有数据")
		} else {
			panic(err)
		}

	}
	//defer client.ClearScroll().ScrollId(scrollID).Do(ctx) // 在退出时清理资源
	fmt.Println("总数是：", res.TotalHits())
	total := 0
	for len(res.Hits.Hits) > 0 {
		for k, hit := range res.Hits.Hits {
			if k%1000 == 0 {
				log.Println("当前：", k)
			}
			var doc map[string]interface{}
			err := json.Unmarshal(hit.Source, &doc)
			if err != nil {
				log.Printf("解析文档失败：%s", err)
				continue
			}

			//处理查询结果
			portrait := PortraitData{
				Buyer:        util.ObjToString(doc["name"]),
				BusinessType: getStr(util.ObjToString(doc["buyerclass"])),
				Class:        "情报_物业",
			}

			// 构建查询
			query := elastic.NewBoolQuery().
				Must(
					//elastic.NewTermQuery("buyer", "泸州市龙马潭区人民医院"),
					elastic.NewTermQuery("buyer", util.ObjToString(doc["name"])),
					elastic.NewTermQuery("tag_topinformation", "情报_物业"),
				)
			// 创建搜索服务
			searchService2 := client.Search().
				Index("projectset"). // 替换为你的索引名称
				Query(query).
				Sort("lasttime", false). // false表示降序
				Size(1).
				Aggregation("total_price", elastic.NewSumAggregation().Field("sortprice"))

			// 执行查询
			searchResult, err := searchService2.Do(context.Background())
			if err != nil {
				log.Fatalf("Error getting response: %s", err)
			}
			// 处理结果
			if searchResult.Hits.TotalHits.Value > 0 {
				portrait.ProjectCount = searchResult.TotalHits()
				for _, hit := range searchResult.Hits.Hits {
					var doc2 map[string]interface{}
					err := json.Unmarshal(hit.Source, &doc2)
					if err != nil {
						log.Printf("解析文档失败：%s", err)
						continue
					}
					portrait.Lasttime = util.Int64All(doc2["lasttime"])
					portrait.Area = util.ObjToString(doc2["area"])
					portrait.City = util.ObjToString(doc2["city"])
				}
				// 处理聚合结果
				if agg, found := searchResult.Aggregations.Sum("total_price"); found {
					portrait.ProjectMoney = *agg.Value
				} else {
					log.Println("Aggregation not found")
				}
				//写入MongoDB
				MgoB.Save("wcc_project_portrait", structToMap(portrait))
			} else {
				continue
			}
		}

		total = total + len(res.Hits.Hits)
		scrollID = res.ScrollId
		res, err = client.Scroll().ScrollId(scrollID).Scroll(scroll).Do(ctx)
		log.Println("current count:", total)
		if err != nil {
			if err == io.EOF {
				// 滚动到最后一批数据，退出循环
				break
			}
			log.Println("滚动搜索失败：", err, res)
			break // 处理错误时退出循环
		}
	}
	// 在循环外调用 ClearScroll
	_, err = client.ClearScroll().ScrollId(scrollID).Do(ctx)
	if err != nil {
		log.Printf("清理滚动搜索失败：%s", err)
	}

	fmt.Println("结束~~~~~~~~~~~~~~~")
}

// allDataEs 处理存量数据到es
func allDataEs() {
	//MgoB := &mongodb.MongodbSim{
	//	MongodbAddr: "172.17.189.140:27080",
	//	//MongodbAddr: "127.0.0.1:27083",
	//	Size:     10,
	//	DbName:   "qfw",
	//	UserName: "SJZY_RWbid_ES",
	//	Password: "SJZY@B4i4D5e6S",
	//	//Direct:      true,
	//}
	//MgoB.InitPool()
	//
	MgoB := &mongodb.MongodbSim{
		MongodbAddr: "192.168.3.206:27002",
		Size:        10,
		DbName:      "qfw_data",
		UserName:    "root",
		Password:    "root",
		//Direct:      true,
	}
	MgoB.InitPool()

	defer util.Catch()
	sess := MgoB.GetMgoConn()
	defer MgoB.DestoryMongoConn(sess)
	count := 0
	it := sess.DB("qfw_data").C(portraitMgo).Find(nil).Select(nil).Iter()
	for tmp := make(map[string]interface{}); it.Next(&tmp); count++ {
		if count%5000 == 0 {
			log.Println("current:", count, tmp["_id"])
		}
		id := mongodb.BsonIdToSId(tmp["_id"])
		tmp["id"] = id
		tmp["_id"] = id

		saveEsPool <- tmp
		//Es.Save("project_portrait", tmp)
	}
	log.Println("数据处理完毕")
}

func SaveEsMethod() {
	arru := make([]map[string]interface{}, EsBulkSize)
	indexu := 0
	for {
		select {
		case v := <-saveEsPool:
			arru[indexu] = v
			indexu++
			if indexu == EsBulkSize {
				saveEsSp <- true
				go func(arru []map[string]interface{}) {
					defer func() {
						<-saveEsSp
					}()
					Es.BulkSave(portraitIndex, arru)
				}(arru)
				arru = make([]map[string]interface{}, EsBulkSize)
				indexu = 0
			}
		case <-time.After(1000 * time.Millisecond):
			if indexu > 0 {
				saveEsSp <- true
				go func(arru []map[string]interface{}) {
					defer func() {
						<-saveEsSp
					}()
					Es.BulkSave(portraitIndex, arru)
				}(arru[:indexu])
				arru = make([]map[string]interface{}, EsBulkSize)
				indexu = 0
			}
		}
	}
}