144 lines
3.1 KiB
Go
144 lines
3.1 KiB
Go
package esmanager
|
|
|
|
import (
|
|
"bytes"
|
|
"encoding/json"
|
|
"fmt"
|
|
"io"
|
|
"time"
|
|
|
|
"github.com/elastic/go-elasticsearch/v7/esapi"
|
|
)
|
|
|
|
type M map[string]interface{}
|
|
|
|
type A []map[string]interface{}
|
|
|
|
type MGetBody struct {
|
|
ID string `json:"_id"`
|
|
}
|
|
|
|
// ElasticResp 响应
|
|
type Response = esapi.Response
|
|
|
|
type Options struct {
|
|
Address []string
|
|
MaxIdleConnPerHost int
|
|
IdleConnTimeout time.Duration
|
|
}
|
|
|
|
func unmarshalAggregateBody(bind interface{}, resp *Response) error {
|
|
var v map[string]interface{}
|
|
defer func() { _ = resp.Body.Close() }()
|
|
if bind == nil { //传入空表示只执行 defer
|
|
return nil
|
|
}
|
|
err := json.NewDecoder(resp.Body).Decode(&v)
|
|
if err != nil {
|
|
return err
|
|
}
|
|
if v != nil && len(v) > 0 {
|
|
if v, ok := v["aggregations"].(map[string]interface{}); ok {
|
|
b, _ := json.Marshal(v)
|
|
_ = json.Unmarshal(b, &bind)
|
|
}
|
|
}
|
|
return nil
|
|
}
|
|
|
|
func unmarshalAnalyzeBody(bind interface{}, resp *Response) error {
|
|
var v map[string]interface{}
|
|
defer func() { _ = resp.Body.Close() }()
|
|
if bind == nil { //传入空表示只执行 defer
|
|
return nil
|
|
}
|
|
err := json.NewDecoder(resp.Body).Decode(&v)
|
|
if err != nil {
|
|
return err
|
|
}
|
|
if v != nil && len(v) > 0 {
|
|
b, _ := json.Marshal(v)
|
|
_ = json.Unmarshal(b, &bind)
|
|
}
|
|
return nil
|
|
}
|
|
|
|
func unmarshalBody(bind interface{}, resp *Response) error {
|
|
var data map[string]interface{}
|
|
defer func() { _ = resp.Body.Close() }()
|
|
if bind == nil { //传入空表示只执行 defer
|
|
return nil
|
|
}
|
|
err := json.NewDecoder(resp.Body).Decode(&data)
|
|
if err != nil {
|
|
return err
|
|
}
|
|
if data != nil && len(data) > 0 {
|
|
if hits1, ok := data["hits"].(map[string]interface{}); ok {
|
|
b, _ := json.Marshal(hits1)
|
|
_ = json.Unmarshal(b, &bind)
|
|
} else if _, ok := data["docs"].([]interface{}); ok {
|
|
b, _ := json.Marshal(data["docs"])
|
|
_ = json.Unmarshal(b, &bind)
|
|
} else if _, ok := data["_source"].(map[string]interface{}); ok {
|
|
b, _ := json.Marshal(data)
|
|
_ = json.Unmarshal(b, &bind)
|
|
}
|
|
}
|
|
return nil
|
|
}
|
|
|
|
func marshalSearchM(query M) io.Reader {
|
|
var buf bytes.Buffer
|
|
if query == nil || len(query) == 0 {
|
|
return nil
|
|
}
|
|
err := json.NewEncoder(&buf).Encode(query)
|
|
if err != nil {
|
|
return nil
|
|
}
|
|
return bytes.NewReader(buf.Bytes())
|
|
}
|
|
|
|
func marshalBulkM(source M) io.Reader {
|
|
var buf bytes.Buffer
|
|
for k, v := range source {
|
|
var meta = []byte(fmt.Sprintf(`{ "index" : { "_id" : "%s" } }%s`, k, "\n"))
|
|
var data, _ = json.Marshal(v)
|
|
data = append(data, "\n"...)
|
|
buf.Grow(len(meta) + len(data))
|
|
buf.Write(meta)
|
|
buf.Write(data)
|
|
}
|
|
return bytes.NewReader(buf.Bytes())
|
|
}
|
|
|
|
func marshalStruct(v interface{}) io.Reader {
|
|
var buf bytes.Buffer
|
|
if v == nil {
|
|
return nil
|
|
}
|
|
err := json.NewEncoder(&buf).Encode(v)
|
|
if err != nil {
|
|
return nil
|
|
}
|
|
return bytes.NewReader(buf.Bytes())
|
|
}
|
|
|
|
func unmarshalStruct(bind interface{}, resp *Response) error {
|
|
var v map[string]interface{}
|
|
defer func() { _ = resp.Body.Close() }()
|
|
if bind == nil { //传入空表示只执行 defer
|
|
return nil
|
|
}
|
|
err := json.NewDecoder(resp.Body).Decode(&v)
|
|
if err != nil {
|
|
return err
|
|
}
|
|
if v != nil && len(v) > 0 {
|
|
b, _ := json.Marshal(v)
|
|
_ = json.Unmarshal(b, &bind)
|
|
}
|
|
return nil
|
|
}
|