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 }