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) } }