jyps/getEs/qyxy.go

package main

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

// getNeqData 获取id _id 不相等数据
func getNeqData() {
	// 本地
	//mgo := &mongodb.MongodbSim{
	//	MongodbAddr: "127.0.0.1:27017",
	//	DbName:      "wcc",
	//	Size:        10,
	//}
	//mgo.InitPool()

	mgo := &mongodb.MongodbSim{
		MongodbAddr: "172.17.189.140:27080",
		DbName:      "qfw",
		Size:        10,
		UserName:    "SJZY_RWbid_ES",
		Password:    "SJZY@B4i4D5e6S",
	}
	mgo.InitPool()

	//url := "http://127.0.0.1:19908"
	url := "http://172.17.4.184:19908"
	username := "jybid"
	password := "Top2023_JEB01i@31"

	// 创建 Elasticsearch 客户端
	client, err := elastic.NewClient(
		elastic.SetURL(url),
		elastic.SetBasicAuth(username, password),
		elastic.SetSniff(false),
	)
	if err != nil {
		log.Fatalf("创建 Elasticsearch 客户端失败：%s", err)
	}

	//// 创建查询条件
	//q := elastic.NewBoolQuery().
	//	//Must(elastic.NewMatchQuery("toptype", "预告")).
	//	MustNot(elastic.NewTermQuery("_id", "id"))
	//Filter(elastic.NewRangeQuery("comeintime").Gte(1694707200).Lte(1695121200))

	// 构建查询
	query := elastic.NewBoolQuery().
		Must(
			elastic.NewScriptQuery(elastic.NewScript(`doc['_id'].value != doc['id'].value`)),
			//elastic.NewRangeQuery("comeintime").Gte(1694707200),
		)

	//query := elastic.NewBoolQuery().
	//	Must(elastic.NewMatchQuery("title", "租公租房提取公积金")).
	//	Must(elastic.NewTermQuery("toptype", "拟建")).Must(elastic.NewMatchPhraseQuery())

	count := 0
	// 执行Count请求来获取文档总数
	countResult, err := client.Count().Index("qyxy").Query(query).Do(context.Background())
	if err != nil {
		log.Fatalf("执行Count请求失败：%s", err)
	}

	// 获取符合条件的文档总数
	total := countResult
	fmt.Printf("符合条件的文档总数：%d\n", total)

	//
	//开始滚动搜索
	scrollService := client.Scroll("qyxy").Query(query).Size(10000).FetchSource(true)
	for {
		results, err := scrollService.Do(context.Background())
		if err != nil {
			log.Fatalf("滚动搜索失败：%s", err)
		}

		fmt.Println("current count:", count)
		if len(results.Hits.Hits) == 0 {
			// 没有更多的文档了，退出循环
			break
		}

		for _, hit := range results.Hits.Hits {
			// 处理每个文档
			// ...
			item := make(map[string]interface{})
			if err := json.Unmarshal(hit.Source, &item); err != nil {
				log.Printf("解码文档失败：%s\n", err)
				continue
			}
			save := map[string]interface{}{
				"id":           item["id"],
				"company_name": item["company_name"],
			}
			mgo.Save("wcc_es_id_err_0428", save)
		}
		count += len(results.Hits.Hits)
	}
	fmt.Println("结束~~~~~~~~~~~~~~~")
}

// deleteEs 删除 es 数据
func deleteEs() {
	//url := "http://127.0.0.1:19905"
	////url := "http://172.17.4.184:19905"
	//username := "jybid"
	//password := "Top2023_JEB01i@31"

	url := "http://127.0.0.1:19805"
	//url := "http://172.17.4.184:19905"
	username := "es_all"
	password := "TopJkO2E_d1x"

	// 创建 Elasticsearch 客户端
	client, err := elastic.NewClient(
		elastic.SetURL(url),
		elastic.SetBasicAuth(username, password),
		elastic.SetSniff(false),
	)
	if err != nil {
		log.Fatalf("创建 Elasticsearch 客户端失败：%s", err)
	}
	// 构建查询
	query := elastic.NewBoolQuery().
		Must(
			elastic.NewScriptQuery(elastic.NewScript(`doc['_id'].value != doc['id'].value`)),
			elastic.NewRangeQuery("comeintime").Gte(1672416000),
		)

	// 执行Count请求来获取文档总数
	countResult, err := client.Count().Index("bidding").Query(query).Do(context.Background())
	if err != nil {
		log.Fatalf("执行Count请求失败：%s", err)
	}

	// 获取符合条件的文档总数
	total := countResult
	fmt.Printf("符合条件的文档总数：%d\n", total)

	// 创建删除请求
	deleteService := client.DeleteByQuery().Index("bidding").Query(query)

	// 执行删除操作
	response, err := deleteService.Do(context.Background())
	if err != nil {
		log.Fatalf("执行删除操作失败：%s", err)
	}

	// 检查删除操作的结果
	if response != nil {
		fmt.Printf("已删除文档数：%d\n", response.Deleted)
	} else {
		fmt.Println("删除操作没有返回结果。")
	}

}

type CreditLabel struct {
	ZhimaToptype string `json:"zhima_toptype"`
	ZhimaSubtype string `json:"zhima_subtype"`
	ZhimaName    string `json:"zhima_name"`
}

// getZhiMa 芝麻标签存在
func getZhiMa() {
	client, err := elastic.NewClient(elastic.SetURL("http://192.168.3.149:9201"))
	if err != nil {
		panic(err)
	}

	// 查询 zhima_labels 字段存在的数据
	query := elastic.NewExistsQuery("zhima_labels")
	searchResult, err := client.Search().
		Index("qyxy").
		Query(query).
		Do(context.Background())
	if err != nil {
		panic(err)
	}

	for _, hit := range searchResult.Hits.Hits {
		var label CreditLabel
		err := json.Unmarshal(hit.Source, &label)
		if err != nil {
			panic(err)
		}
		fmt.Println(label)
	}

}