1090 lines
30 KiB
Go
1090 lines
30 KiB
Go
package main
|
||
|
||
/*
|
||
#include <stdlib.h>
|
||
*/
|
||
import "C"
|
||
import (
|
||
"context"
|
||
"encoding/json"
|
||
"fmt"
|
||
"github.com/elastic/go-elasticsearch/v8"
|
||
"github.com/elastic/go-elasticsearch/v8/esapi"
|
||
"log"
|
||
"strings"
|
||
"unsafe"
|
||
)
|
||
|
||
// ESClient 封装Elasticsearch客户端
|
||
type ESClient struct {
|
||
client *elasticsearch.Client
|
||
}
|
||
|
||
// IndexInfo 索引信息结构
|
||
type IndexInfo struct {
|
||
Name string `json:"index"` // 索引名称
|
||
Health string `json:"health"` // 健康状态 green：所有主分片和副本分片都正常 yellow：主分片正常，但部分副本分片不正常 red：有主分片不正常
|
||
Status string `json:"status"` // 索引状态 open：为打开（可用）
|
||
UUID string `json:"uuid"` // 索引的唯一标识符
|
||
Pri string `json:"pri"` // 主分片数量
|
||
Rep string `json:"rep"` // 副本分片数量
|
||
DocsCount string `json:"docs.count"` // 文档总数
|
||
DocsDeleted string `json:"docs.deleted"` // 已删除文档数
|
||
StoreSize string `json:"store.size"` // 总存储大小
|
||
PriStoreSize string `json:"pri.store.size"` // 主分片存储大小
|
||
Settings map[string]interface{} `json:"settings,omitempty"` // 索引设置
|
||
Mappings map[string]interface{} `json:"mappings,omitempty"` // 索引映射
|
||
Aliases []string `json:"aliases,omitempty"` // 索引别名
|
||
}
|
||
|
||
// ES配置常量
|
||
const (
|
||
esAddress = "http://103.236.91.138:9200" // ES地址
|
||
esUsername = "elastic" // 用户名（如果有认证）
|
||
esPassword = "5mRDIUg52VC0fp14nw-F" // 密码（如果有认证）
|
||
)
|
||
|
||
// newESClient 创建ES客户端
|
||
func newESClient(addresses []string, username, password string) (*ESClient, error) {
|
||
cfg := elasticsearch.Config{
|
||
Addresses: addresses, // ES节点地址
|
||
Username: username, // 用户名（如果有认证）
|
||
Password: password, // 密码（如果有认证）
|
||
}
|
||
// 创建新的Elasticsearch客户端
|
||
client, err := elasticsearch.NewClient(cfg)
|
||
if err != nil {
|
||
return nil, fmt.Errorf("创建ES客户端失败: %w", err)
|
||
}
|
||
|
||
// 测试连接：发送ping请求验证连接是否正常
|
||
res, err := client.Ping()
|
||
if err != nil {
|
||
return nil, fmt.Errorf("连接ES失败: %w", err)
|
||
}
|
||
defer res.Body.Close() // 确保响应体被关闭
|
||
// 检查响应状态
|
||
if res.IsError() {
|
||
return nil, fmt.Errorf("ES响应错误: %s", res.String())
|
||
}
|
||
|
||
log.Println("✅ 成功连接到Elasticsearch")
|
||
return &ESClient{
|
||
client: client, // 返回封装的客户端
|
||
}, nil
|
||
}
|
||
|
||
// 查询所有索引
|
||
func (es *ESClient) listAllIndices() ([]string, error) {
|
||
// 创建索引列表请求
|
||
req := esapi.CatIndicesRequest{
|
||
Format: "json", // 返回JSON格式
|
||
Pretty: true, // 美化输出
|
||
}
|
||
// 执行请求
|
||
res, err := req.Do(context.Background(), es.client)
|
||
if err != nil {
|
||
return nil, fmt.Errorf("查询索引列表失败: %w", err)
|
||
}
|
||
defer res.Body.Close() // 确保响应体被关闭
|
||
// 检查响应状态
|
||
if res.IsError() {
|
||
return nil, fmt.Errorf("查询索引列表失败: %s", res.String())
|
||
}
|
||
|
||
// 解析返回的索引信息
|
||
var indices []map[string]interface{}
|
||
if err := json.NewDecoder(res.Body).Decode(&indices); err != nil {
|
||
return nil, fmt.Errorf("解析索引列表失败: %w", err)
|
||
}
|
||
// 提取索引名称
|
||
var indexNames []string
|
||
for _, idx := range indices {
|
||
if name, ok := idx["index"].(string); ok && name != "" {
|
||
indexNames = append(indexNames, name)
|
||
}
|
||
}
|
||
|
||
return indexNames, nil
|
||
}
|
||
|
||
// 获取所有索引的详细信息
|
||
func (es *ESClient) getIndicesInfo() ([]IndexInfo, error) {
|
||
// 先获取所有索引名
|
||
indexNames, err := es.listAllIndices()
|
||
if err != nil {
|
||
return nil, err
|
||
}
|
||
|
||
// 如果有索引，获取详细信息
|
||
if len(indexNames) == 0 {
|
||
return []IndexInfo{}, nil
|
||
}
|
||
|
||
// 获取索引的统计信息
|
||
statsReq := esapi.IndicesStatsRequest{
|
||
Index: indexNames, // 指定要查询的索引
|
||
Human: true, // 返回人类可读的格式
|
||
}
|
||
|
||
statsRes, err := statsReq.Do(context.Background(), es.client)
|
||
if err != nil {
|
||
return nil, fmt.Errorf("获取索引统计信息失败: %w", err)
|
||
}
|
||
defer statsRes.Body.Close() // 确保响应体被关闭
|
||
|
||
if statsRes.IsError() {
|
||
return nil, fmt.Errorf("获取索引统计信息失败: %s", statsRes.String())
|
||
}
|
||
|
||
var statsResult map[string]interface{}
|
||
if err := json.NewDecoder(statsRes.Body).Decode(&statsResult); err != nil {
|
||
return nil, fmt.Errorf("解析索引统计信息失败: %w", err)
|
||
}
|
||
|
||
// 获取索引的健康状态
|
||
healthReq := esapi.CatIndicesRequest{
|
||
Index: indexNames, // 指定要查询的索引
|
||
Format: "json", // 返回JSON格式
|
||
H: []string{"index,health,status,pri,rep,docs.count,docs.deleted,store.size,pri.store.size"}, // 要返回的字段
|
||
}
|
||
|
||
healthRes, err := healthReq.Do(context.Background(), es.client)
|
||
if err != nil {
|
||
return nil, fmt.Errorf("获取索引健康状态失败: %w", err)
|
||
}
|
||
defer healthRes.Body.Close() // 确保响应体被关闭
|
||
|
||
if healthRes.IsError() {
|
||
return nil, fmt.Errorf("获取索引健康状态失败: %s", healthRes.String())
|
||
}
|
||
|
||
var healthInfo []map[string]interface{}
|
||
if err := json.NewDecoder(healthRes.Body).Decode(&healthInfo); err != nil {
|
||
return nil, fmt.Errorf("解析索引健康状态失败: %w", err)
|
||
}
|
||
|
||
// 组合索引信息
|
||
var indicesInfo []IndexInfo
|
||
for _, health := range healthInfo {
|
||
indexName, _ := health["index"].(string)
|
||
if indexName == "" {
|
||
continue // 跳过空索引名
|
||
}
|
||
// 构建索引信息结构
|
||
info := IndexInfo{
|
||
Name: indexName,
|
||
Health: getString(health, "health", "unknown"),
|
||
Status: getString(health, "status", "unknown"),
|
||
Pri: getString(health, "pri", "0"),
|
||
Rep: getString(health, "rep", "0"),
|
||
DocsCount: getString(health, "docs.count", "0"),
|
||
DocsDeleted: getString(health, "docs.deleted", "0"),
|
||
StoreSize: getString(health, "store.size", "0b"),
|
||
PriStoreSize: getString(health, "pri.store.size", "0b"),
|
||
}
|
||
|
||
indicesInfo = append(indicesInfo, info)
|
||
}
|
||
|
||
return indicesInfo, nil
|
||
}
|
||
|
||
// 获取单个索引的详细信息
|
||
func (es *ESClient) getIndexDetail(indexName string) (map[string]interface{}, error) {
|
||
req := esapi.IndicesGetRequest{
|
||
Index: []string{indexName}, // 指定要查询的索引
|
||
}
|
||
|
||
res, err := req.Do(context.Background(), es.client)
|
||
if err != nil {
|
||
return nil, fmt.Errorf("获取索引详情失败: %w", err)
|
||
}
|
||
defer res.Body.Close() // 确保响应体被关闭
|
||
|
||
// 检查响应状态
|
||
if res.IsError() {
|
||
if res.StatusCode == 404 {
|
||
return nil, fmt.Errorf("索引不存在: %s", indexName)
|
||
}
|
||
return nil, fmt.Errorf("获取索引详情失败: %s", res.String())
|
||
}
|
||
|
||
var result map[string]interface{}
|
||
if err := json.NewDecoder(res.Body).Decode(&result); err != nil {
|
||
return nil, fmt.Errorf("解析索引详情失败: %w", err)
|
||
}
|
||
|
||
return result, nil
|
||
}
|
||
|
||
// 获取索引设置
|
||
func (es *ESClient) getIndexSettings(indexName string) (map[string]interface{}, error) {
|
||
req := esapi.IndicesGetSettingsRequest{
|
||
Index: []string{indexName}, // 指定要查询的索引
|
||
}
|
||
|
||
res, err := req.Do(context.Background(), es.client)
|
||
if err != nil {
|
||
return nil, fmt.Errorf("获取索引设置失败: %w", err)
|
||
}
|
||
defer res.Body.Close() // 确保响应体被关闭
|
||
// 检查响应状态
|
||
if res.IsError() {
|
||
if res.StatusCode == 404 {
|
||
return nil, fmt.Errorf("索引不存在: %s", indexName)
|
||
}
|
||
return nil, fmt.Errorf("获取索引设置失败: %s", res.String())
|
||
}
|
||
|
||
var result map[string]interface{}
|
||
if err := json.NewDecoder(res.Body).Decode(&result); err != nil {
|
||
return nil, fmt.Errorf("解析索引设置失败: %w", err)
|
||
}
|
||
|
||
return result, nil
|
||
}
|
||
|
||
// createIndex 创建索引（如果不存在）
|
||
func (es *ESClient) createIndex(indexName string, mapping string) (map[string]interface{}, error) {
|
||
// 检查索引是否存在
|
||
res, err := es.client.Indices.Exists([]string{indexName})
|
||
if err != nil {
|
||
return nil, fmt.Errorf("检查索引存在失败: %w", err)
|
||
}
|
||
defer res.Body.Close() // 确保响应体被关闭
|
||
|
||
// 如果索引已存在，直接返回
|
||
if res.StatusCode == 200 {
|
||
log.Printf("索引 %s 已存在\n", indexName)
|
||
detail, err := es.getIndexDetail(indexName)
|
||
if err != nil {
|
||
return nil, err
|
||
}
|
||
return detail, nil
|
||
}
|
||
// 创建索引请求
|
||
req := esapi.IndicesCreateRequest{
|
||
Index: indexName, // 索引名称
|
||
Body: strings.NewReader(mapping), // 索引映射配置
|
||
}
|
||
// 执行创建索引请求
|
||
res, err = req.Do(context.Background(), es.client)
|
||
if err != nil {
|
||
return nil, fmt.Errorf("创建索引请求失败: %w", err)
|
||
}
|
||
defer res.Body.Close() // 确保响应体被关闭
|
||
// 检查响应状态
|
||
if res.IsError() {
|
||
return nil, fmt.Errorf("创建索引失败: %s", res.String())
|
||
}
|
||
// 获取新创建索引的详情
|
||
detail, err := es.getIndexDetail(indexName)
|
||
if err != nil {
|
||
return nil, err
|
||
}
|
||
|
||
log.Printf("✅ 成功创建索引: %s\n", indexName)
|
||
return detail, nil
|
||
}
|
||
|
||
// deleteIndex 删除索引
|
||
func (es *ESClient) deleteIndex(indexName string) error {
|
||
req := esapi.IndicesDeleteRequest{
|
||
Index: []string{indexName}, // 要删除的索引
|
||
}
|
||
|
||
res, err := req.Do(context.Background(), es.client)
|
||
if err != nil {
|
||
return fmt.Errorf("删除索引请求失败: %w", err)
|
||
}
|
||
defer res.Body.Close() // 确保响应体被关闭
|
||
// 检查响应状态
|
||
if res.IsError() {
|
||
if res.StatusCode == 404 {
|
||
return fmt.Errorf("索引不存在: %s", indexName)
|
||
}
|
||
return fmt.Errorf("删除索引失败: %s", res.String())
|
||
}
|
||
|
||
log.Printf("✅ 成功删除索引: %s\n", indexName)
|
||
return nil
|
||
}
|
||
|
||
// 更新索引设置
|
||
func (es *ESClient) updateIndexSettings(indexName string, settings map[string]interface{}) error {
|
||
// 构建设置请求体
|
||
requestBody := map[string]interface{}{
|
||
"settings": settings, // 新的设置参数
|
||
}
|
||
|
||
bodyJSON, err := json.Marshal(requestBody)
|
||
if err != nil {
|
||
return fmt.Errorf("序列化设置失败: %w", err)
|
||
}
|
||
|
||
req := esapi.IndicesPutSettingsRequest{
|
||
Index: []string{indexName}, // 要更新的索引
|
||
Body: strings.NewReader(string(bodyJSON)), // 设置内容
|
||
}
|
||
|
||
res, err := req.Do(context.Background(), es.client)
|
||
if err != nil {
|
||
return fmt.Errorf("更新索引设置请求失败: %w", err)
|
||
}
|
||
defer res.Body.Close() // 确保响应体被关闭
|
||
// 检查响应状态
|
||
if res.IsError() {
|
||
if res.StatusCode == 404 {
|
||
return fmt.Errorf("索引不存在: %s", indexName)
|
||
}
|
||
return fmt.Errorf("更新索引设置失败: %s", res.String())
|
||
}
|
||
|
||
log.Printf("✅ 成功更新索引设置: %s\n", indexName)
|
||
return nil
|
||
}
|
||
|
||
// 更新索引映射（添加新字段）
|
||
func (es *ESClient) updateIndexMappings(indexName string, newMappings map[string]interface{}) error {
|
||
// 构建映射请求体
|
||
requestBody := map[string]interface{}{
|
||
"properties": newMappings, // 新的映射字段
|
||
}
|
||
|
||
bodyJSON, err := json.Marshal(requestBody)
|
||
if err != nil {
|
||
return fmt.Errorf("序列化映射失败: %w", err)
|
||
}
|
||
|
||
req := esapi.IndicesPutMappingRequest{
|
||
Index: []string{indexName}, // 要更新的索引
|
||
Body: strings.NewReader(string(bodyJSON)), // 映射内容
|
||
}
|
||
|
||
res, err := req.Do(context.Background(), es.client)
|
||
if err != nil {
|
||
return fmt.Errorf("更新索引映射请求失败: %w", err)
|
||
}
|
||
defer res.Body.Close() // 确保响应体被关闭
|
||
// 检查响应状态
|
||
if res.IsError() {
|
||
if res.StatusCode == 404 {
|
||
return fmt.Errorf("索引不存在: %s", indexName)
|
||
}
|
||
return fmt.Errorf("更新索引映射失败: %s", res.String())
|
||
}
|
||
|
||
log.Printf("✅ 成功更新索引映射: %s\n", indexName)
|
||
return nil
|
||
}
|
||
|
||
// 关闭索引
|
||
func (es *ESClient) closeIndex(indexName string) error {
|
||
req := esapi.IndicesCloseRequest{
|
||
Index: []string{indexName}, // 要关闭的索引
|
||
}
|
||
|
||
res, err := req.Do(context.Background(), es.client)
|
||
if err != nil {
|
||
return fmt.Errorf("关闭索引请求失败: %w", err)
|
||
}
|
||
defer res.Body.Close() // 确保响应体被关闭
|
||
|
||
if res.IsError() {
|
||
if res.StatusCode == 404 {
|
||
return fmt.Errorf("索引不存在: %s", indexName)
|
||
}
|
||
return fmt.Errorf("关闭索引失败: %s", res.String())
|
||
}
|
||
|
||
log.Printf("✅ 已关闭索引: %s\n", indexName)
|
||
return nil
|
||
}
|
||
|
||
// 打开索引
|
||
func (es *ESClient) openIndex(indexName string) error {
|
||
req := esapi.IndicesOpenRequest{
|
||
Index: []string{indexName}, // 要打开的索引
|
||
}
|
||
|
||
res, err := req.Do(context.Background(), es.client)
|
||
if err != nil {
|
||
return fmt.Errorf("打开索引请求失败: %w", err)
|
||
}
|
||
defer res.Body.Close() // 确保响应体被关闭
|
||
// 检查响应状态
|
||
if res.IsError() {
|
||
if res.StatusCode == 404 {
|
||
return fmt.Errorf("索引不存在: %s", indexName)
|
||
}
|
||
return fmt.Errorf("打开索引失败: %s", res.String())
|
||
}
|
||
|
||
log.Printf("✅ 已打开索引: %s\n", indexName)
|
||
return nil
|
||
}
|
||
|
||
// getDocumentCount 获取索引文档数量
|
||
func (es *ESClient) getDocumentCount(indexName string) (int64, error) {
|
||
req := esapi.CountRequest{
|
||
Index: []string{indexName}, // 要计数的索引
|
||
}
|
||
|
||
res, err := req.Do(context.Background(), es.client)
|
||
if err != nil {
|
||
return 0, fmt.Errorf("获取文档数量请求失败: %w", err)
|
||
}
|
||
defer res.Body.Close() // 确保响应体被关闭
|
||
|
||
if res.IsError() {
|
||
if res.StatusCode == 404 {
|
||
return 0, fmt.Errorf("索引不存在: %s", indexName)
|
||
}
|
||
return 0, fmt.Errorf("获取文档数量失败: %s", res.String())
|
||
}
|
||
|
||
var result map[string]interface{}
|
||
if err := json.NewDecoder(res.Body).Decode(&result); err != nil {
|
||
return 0, fmt.Errorf("解析响应失败: %w", err)
|
||
}
|
||
// 提取文档数量
|
||
if count, ok := result["count"].(float64); ok {
|
||
return int64(count), nil
|
||
}
|
||
|
||
return 0, fmt.Errorf("无法获取文档数量")
|
||
}
|
||
|
||
// ========================== 文档相关操作 =============
|
||
|
||
// createDocument 创建文档
|
||
func (es *ESClient) createDocument(indexName string, id string, doc interface{}) error {
|
||
// 将文档序列化为JSON
|
||
docJSON, err := json.Marshal(doc)
|
||
if err != nil {
|
||
return fmt.Errorf("序列化文档失败: %w", err)
|
||
}
|
||
|
||
req := esapi.IndexRequest{
|
||
Index: indexName, // 目标索引
|
||
DocumentID: id, // 文档ID（可选，为空时ES自动生成）
|
||
Body: strings.NewReader(string(docJSON)), // 文档内容
|
||
Refresh: "true", // 立即刷新使文档可搜索
|
||
}
|
||
|
||
res, err := req.Do(context.Background(), es.client)
|
||
if err != nil {
|
||
return fmt.Errorf("创建文档请求失败: %w", err)
|
||
}
|
||
defer res.Body.Close() // 确保响应体被关闭
|
||
// 检查响应状态
|
||
if res.IsError() {
|
||
return fmt.Errorf("创建文档失败: %s", res.String())
|
||
}
|
||
|
||
log.Printf("✅ 成功创建文档: indexName=%s\n", indexName)
|
||
return nil
|
||
}
|
||
|
||
// getDocument 根据ID获取文档
|
||
func (es *ESClient) getDocument(indexName string, id string) (map[string]interface{}, error) {
|
||
req := esapi.GetRequest{
|
||
Index: indexName, // 索引名称
|
||
DocumentID: id, // 文档ID
|
||
}
|
||
|
||
res, err := req.Do(context.Background(), es.client)
|
||
if err != nil {
|
||
return nil, fmt.Errorf("获取文档请求失败: %w", err)
|
||
}
|
||
defer res.Body.Close() // 确保响应体被关闭
|
||
// 检查响应状态
|
||
if res.IsError() {
|
||
if res.StatusCode == 404 {
|
||
return nil, fmt.Errorf("文档不存在: ID=%s", id)
|
||
}
|
||
return nil, fmt.Errorf("获取文档失败: %s", res.String())
|
||
}
|
||
|
||
var result map[string]interface{}
|
||
if err := json.NewDecoder(res.Body).Decode(&result); err != nil {
|
||
return nil, fmt.Errorf("解析响应失败: %w", err)
|
||
}
|
||
|
||
return result, nil
|
||
}
|
||
|
||
// updateDocument 更新文档
|
||
func (es *ESClient) updateDocument(indexName string, id string, updateData map[string]interface{}) error {
|
||
// 构建更新脚本
|
||
updateBody := map[string]interface{}{
|
||
"doc": updateData, // 更新内容放在doc字段中
|
||
}
|
||
|
||
bodyJSON, err := json.Marshal(updateBody)
|
||
if err != nil {
|
||
return fmt.Errorf("序列化更新数据失败: %w", err)
|
||
}
|
||
|
||
req := esapi.UpdateRequest{
|
||
Index: indexName, // 索引名称
|
||
DocumentID: id, // 文档ID
|
||
Body: strings.NewReader(string(bodyJSON)), // 更新内容
|
||
Refresh: "true", // 立即刷新使更新可搜索
|
||
}
|
||
|
||
res, err := req.Do(context.Background(), es.client)
|
||
if err != nil {
|
||
return fmt.Errorf("更新文档请求失败: %w", err)
|
||
}
|
||
defer res.Body.Close() // 确保响应体被关闭
|
||
// 检查响应状态
|
||
if res.IsError() {
|
||
if res.StatusCode == 404 {
|
||
return fmt.Errorf("文档不存在: ID=%s", id)
|
||
}
|
||
return fmt.Errorf("更新文档失败: %s", res.String())
|
||
}
|
||
|
||
log.Printf("✅ 成功更新文档: ID=%s\n", id)
|
||
return nil
|
||
}
|
||
|
||
// deleteDocument 删除文档
|
||
func (es *ESClient) deleteDocument(indexName string, id string) error {
|
||
req := esapi.DeleteRequest{
|
||
Index: indexName, // 索引名称
|
||
DocumentID: id, // 文档ID
|
||
Refresh: "true", // 立即刷新
|
||
}
|
||
|
||
res, err := req.Do(context.Background(), es.client)
|
||
if err != nil {
|
||
return fmt.Errorf("删除文档请求失败: %w", err)
|
||
}
|
||
defer res.Body.Close() // 确保响应体被关闭
|
||
// 检查响应状态
|
||
if res.IsError() {
|
||
if res.StatusCode == 404 {
|
||
return fmt.Errorf("文档不存在: ID=%s", id)
|
||
}
|
||
return fmt.Errorf("删除文档失败: %s", res.String())
|
||
}
|
||
|
||
log.Printf("✅ 成功删除文档: ID=%s\n", id)
|
||
return nil
|
||
}
|
||
|
||
// searchDocuments 搜索文档
|
||
func (es *ESClient) searchDocuments(indexName string, query map[string]interface{}) ([]map[string]interface{}, error) {
|
||
// 将查询条件序列化为JSON
|
||
queryJSON, err := json.Marshal(query)
|
||
if err != nil {
|
||
return nil, fmt.Errorf("序列化查询失败: %w", err)
|
||
}
|
||
|
||
req := esapi.SearchRequest{
|
||
Index: []string{indexName}, // 要搜索的索引
|
||
Body: strings.NewReader(string(queryJSON)), // 查询条件
|
||
}
|
||
|
||
res, err := req.Do(context.Background(), es.client)
|
||
if err != nil {
|
||
return nil, fmt.Errorf("搜索请求失败: %w", err)
|
||
}
|
||
defer res.Body.Close() // 确保响应体被关闭
|
||
// 检查响应状态
|
||
if res.IsError() {
|
||
return nil, fmt.Errorf("搜索失败: %s", res.String())
|
||
}
|
||
|
||
var result map[string]interface{}
|
||
if err := json.NewDecoder(res.Body).Decode(&result); err != nil {
|
||
return nil, fmt.Errorf("解析搜索结果失败: %w", err)
|
||
}
|
||
|
||
// 提取命中的文档
|
||
hits, ok := result["hits"].(map[string]interface{})
|
||
if !ok {
|
||
return []map[string]interface{}{}, nil // 没有命中结果
|
||
}
|
||
|
||
hitsArray, ok := hits["hits"].([]interface{})
|
||
if !ok {
|
||
return []map[string]interface{}{}, nil // 命中结果格式不正确
|
||
}
|
||
// 处理每个命中文档
|
||
var documents []map[string]interface{}
|
||
for _, hit := range hitsArray {
|
||
if hitMap, ok := hit.(map[string]interface{}); ok {
|
||
if source, ok := hitMap["_source"].(map[string]interface{}); ok {
|
||
source["_id"] = hitMap["_id"] // 将文档ID添加到结果中
|
||
documents = append(documents, source)
|
||
}
|
||
}
|
||
}
|
||
|
||
return documents, nil
|
||
}
|
||
|
||
//// BulkCreate 批量创建文档
|
||
//func (es *ESClient) BulkCreate(docs map[string]interface{}) error {
|
||
// var body string
|
||
// for id, doc := range docs {
|
||
// docJSON, err := json.Marshal(doc)
|
||
// if err != nil {
|
||
// return fmt.Errorf("序列化文档失败: %w", err)
|
||
// }
|
||
//
|
||
// // 批量操作格式
|
||
// body += fmt.Sprintf(`{"index":{"_index":"%s","_id":"%s"}}%s`, es.index, id, "\n")
|
||
// body += string(docJSON) + "\n"
|
||
// }
|
||
//
|
||
// req := esapi.BulkRequest{
|
||
// Body: strings.NewReader(body),
|
||
// }
|
||
//
|
||
// res, err := req.Do(context.Background(), es.client)
|
||
// if err != nil {
|
||
// return fmt.Errorf("批量操作请求失败: %w", err)
|
||
// }
|
||
// defer res.Body.Close()
|
||
//
|
||
// if res.IsError() {
|
||
// return fmt.Errorf("批量操作失败: %s", res.String())
|
||
// }
|
||
//
|
||
// log.Printf("✅ 批量创建 %d 个文档成功\n", len(docs))
|
||
// return nil
|
||
//}
|
||
|
||
// ==================== 辅助函数 ====================
|
||
|
||
// getString 从map中安全获取字符串值，避免类型断言失败
|
||
func getString(m map[string]interface{}, key, defaultValue string) string {
|
||
if val, ok := m[key]; ok {
|
||
if str, ok := val.(string); ok {
|
||
return str
|
||
}
|
||
}
|
||
return defaultValue // 如果key不存在或不是字符串类型，返回默认值
|
||
}
|
||
|
||
// =================== C 导入函数 =======================
|
||
|
||
// c响应信息
|
||
type APIResponse struct {
|
||
Success bool `json:"success"` // 操作是否成功
|
||
Message string `json:"message"` // 响应消息（错误时为错误信息）
|
||
Data interface{} `json:"data"` // 响应数据（成功时返回）
|
||
}
|
||
|
||
// ListAllIndices 查询所有索引
|
||
//
|
||
//export ListAllIndices
|
||
func ListAllIndices() *C.char {
|
||
var apiResp APIResponse
|
||
// 创建ES客户端
|
||
client, err := newESClient([]string{esAddress}, esUsername, esPassword)
|
||
if err != nil {
|
||
apiResp = APIResponse{
|
||
Success: false,
|
||
Message: err.Error(),
|
||
}
|
||
apiRespStr, _ := json.Marshal(apiResp)
|
||
return C.CString(string(apiRespStr))
|
||
}
|
||
// 查询所有索引
|
||
info, err := client.listAllIndices()
|
||
if err != nil {
|
||
apiResp = APIResponse{
|
||
Success: false,
|
||
Message: err.Error(),
|
||
}
|
||
apiRespStr, _ := json.Marshal(apiResp)
|
||
return C.CString(string(apiRespStr))
|
||
}
|
||
apiResp = APIResponse{
|
||
Success: true,
|
||
Data: info,
|
||
}
|
||
apiRespStr, _ := json.Marshal(apiResp)
|
||
return C.CString(string(apiRespStr))
|
||
}
|
||
|
||
// GetIndicesInfo 获取所有索引的详细信息
|
||
//
|
||
//export GetIndicesInfo
|
||
func GetIndicesInfo() *C.char {
|
||
var apiResp APIResponse
|
||
// 创建ES客户端
|
||
client, err := newESClient([]string{esAddress}, esUsername, esPassword)
|
||
if err != nil {
|
||
apiResp = APIResponse{
|
||
Success: false,
|
||
Message: err.Error(),
|
||
}
|
||
apiRespStr, _ := json.Marshal(apiResp)
|
||
return C.CString(string(apiRespStr))
|
||
}
|
||
// 获取所有索引的详细信息
|
||
info, err := client.getIndicesInfo()
|
||
if err != nil {
|
||
apiResp = APIResponse{
|
||
Success: false,
|
||
Message: err.Error(),
|
||
}
|
||
apiRespStr, _ := json.Marshal(apiResp)
|
||
return C.CString(string(apiRespStr))
|
||
}
|
||
apiResp = APIResponse{
|
||
Success: true,
|
||
Data: info,
|
||
}
|
||
apiRespStr, _ := json.Marshal(apiResp)
|
||
return C.CString(string(apiRespStr))
|
||
}
|
||
|
||
// GetIndexDetail 获取单个索引的详细信息
|
||
//
|
||
//export GetIndexDetail
|
||
func GetIndexDetail(indexName *C.char) *C.char {
|
||
goIndexName := C.GoString(indexName)
|
||
var apiResp APIResponse
|
||
// 创建ES客户端
|
||
client, err := newESClient([]string{esAddress}, esUsername, esPassword)
|
||
if err != nil {
|
||
apiResp = APIResponse{
|
||
Success: false,
|
||
Message: err.Error(),
|
||
}
|
||
apiRespStr, _ := json.Marshal(apiResp)
|
||
return C.CString(string(apiRespStr))
|
||
}
|
||
info, err := client.getIndexDetail(goIndexName)
|
||
if err != nil {
|
||
apiResp = APIResponse{
|
||
Success: false,
|
||
Message: err.Error(),
|
||
}
|
||
apiRespStr, _ := json.Marshal(apiResp)
|
||
return C.CString(string(apiRespStr))
|
||
}
|
||
apiResp = APIResponse{
|
||
Success: true,
|
||
Data: info,
|
||
}
|
||
apiRespStr, _ := json.Marshal(apiResp)
|
||
return C.CString(string(apiRespStr))
|
||
}
|
||
|
||
// CreateIndex 创建索引（如果不存在）
|
||
//
|
||
//export CreateIndex
|
||
func CreateIndex(indexName, mapping *C.char) *C.char {
|
||
goIndexName := C.GoString(indexName)
|
||
goMapping := C.GoString(mapping)
|
||
var apiResp APIResponse
|
||
// 创建ES客户端
|
||
client, err := newESClient([]string{esAddress}, esUsername, esPassword)
|
||
if err != nil {
|
||
apiResp = APIResponse{
|
||
Success: false,
|
||
Message: err.Error(),
|
||
}
|
||
apiRespStr, _ := json.Marshal(apiResp)
|
||
return C.CString(string(apiRespStr))
|
||
}
|
||
info, err := client.createIndex(goIndexName, goMapping)
|
||
if err != nil {
|
||
apiResp = APIResponse{
|
||
Success: false,
|
||
Message: err.Error(),
|
||
}
|
||
apiRespStr, _ := json.Marshal(apiResp)
|
||
return C.CString(string(apiRespStr))
|
||
}
|
||
apiResp = APIResponse{
|
||
Success: true,
|
||
Data: info,
|
||
}
|
||
apiRespStr, _ := json.Marshal(apiResp)
|
||
return C.CString(string(apiRespStr))
|
||
}
|
||
|
||
// DeleteIndex 删除索引
|
||
//
|
||
//export DeleteIndex
|
||
func DeleteIndex(indexName *C.char) *C.char {
|
||
goIndexName := C.GoString(indexName)
|
||
var apiResp APIResponse
|
||
// 创建ES客户端
|
||
client, err := newESClient([]string{esAddress}, esUsername, esPassword)
|
||
if err != nil {
|
||
apiResp = APIResponse{
|
||
Success: false,
|
||
Message: err.Error(),
|
||
}
|
||
apiRespStr, _ := json.Marshal(apiResp)
|
||
return C.CString(string(apiRespStr))
|
||
}
|
||
err = client.deleteIndex(goIndexName)
|
||
if err != nil {
|
||
apiResp = APIResponse{
|
||
Success: false,
|
||
Message: err.Error(),
|
||
}
|
||
apiRespStr, _ := json.Marshal(apiResp)
|
||
return C.CString(string(apiRespStr))
|
||
}
|
||
apiResp = APIResponse{
|
||
Success: true,
|
||
}
|
||
apiRespStr, _ := json.Marshal(apiResp)
|
||
return C.CString(string(apiRespStr))
|
||
}
|
||
|
||
// GetDocumentCount 获取索引文档数量
|
||
//
|
||
//export GetDocumentCount
|
||
func GetDocumentCount(indexName *C.char) *C.char {
|
||
goIndexName := C.GoString(indexName)
|
||
var apiResp APIResponse
|
||
// 创建ES客户端
|
||
client, err := newESClient([]string{esAddress}, esUsername, esPassword)
|
||
if err != nil {
|
||
apiResp = APIResponse{
|
||
Success: false,
|
||
Message: err.Error(),
|
||
}
|
||
apiRespStr, _ := json.Marshal(apiResp)
|
||
return C.CString(string(apiRespStr))
|
||
}
|
||
info, err := client.getDocumentCount(goIndexName)
|
||
if err != nil {
|
||
apiResp = APIResponse{
|
||
Success: false,
|
||
Message: err.Error(),
|
||
}
|
||
apiRespStr, _ := json.Marshal(apiResp)
|
||
return C.CString(string(apiRespStr))
|
||
}
|
||
apiResp = APIResponse{
|
||
Success: true,
|
||
Data: info,
|
||
}
|
||
apiRespStr, _ := json.Marshal(apiResp)
|
||
return C.CString(string(apiRespStr))
|
||
}
|
||
|
||
// CreateDocument 创建文档
|
||
//
|
||
//export CreateDocument
|
||
func CreateDocument(indexName *C.char, id *C.char, doc *C.char) *C.char {
|
||
goIndexName := C.GoString(indexName)
|
||
goId := C.GoString(id)
|
||
goDoc := C.GoString(doc)
|
||
var apiResp APIResponse
|
||
var newDoc interface{}
|
||
if err := json.Unmarshal([]byte(goDoc), &newDoc); err != nil {
|
||
apiResp = APIResponse{
|
||
Success: false,
|
||
Message: fmt.Sprintf("解析JSON失败: %v", err),
|
||
}
|
||
apiRespStr, _ := json.Marshal(apiResp)
|
||
return C.CString(string(apiRespStr))
|
||
}
|
||
// 创建ES客户端
|
||
client, err := newESClient([]string{esAddress}, esUsername, esPassword)
|
||
if err != nil {
|
||
apiResp = APIResponse{
|
||
Success: false,
|
||
Message: err.Error(),
|
||
}
|
||
apiRespStr, _ := json.Marshal(apiResp)
|
||
return C.CString(string(apiRespStr))
|
||
}
|
||
err = client.createDocument(goIndexName, goId, newDoc)
|
||
if err != nil {
|
||
apiResp = APIResponse{
|
||
Success: false,
|
||
Message: err.Error(),
|
||
}
|
||
apiRespStr, _ := json.Marshal(apiResp)
|
||
return C.CString(string(apiRespStr))
|
||
}
|
||
apiResp = APIResponse{
|
||
Success: true,
|
||
}
|
||
apiRespStr, _ := json.Marshal(apiResp)
|
||
return C.CString(string(apiRespStr))
|
||
}
|
||
|
||
// GetDocument 根据ID获取文档
|
||
//
|
||
//export GetDocument
|
||
func GetDocument(indexName *C.char, id *C.char) *C.char {
|
||
goIndexName := C.GoString(indexName)
|
||
goId := C.GoString(id)
|
||
var apiResp APIResponse
|
||
// 创建ES客户端
|
||
client, err := newESClient([]string{esAddress}, esUsername, esPassword)
|
||
if err != nil {
|
||
apiResp = APIResponse{
|
||
Success: false,
|
||
Message: err.Error(),
|
||
}
|
||
apiRespStr, _ := json.Marshal(apiResp)
|
||
return C.CString(string(apiRespStr))
|
||
}
|
||
info, err := client.getDocument(goIndexName, goId)
|
||
if err != nil {
|
||
apiResp = APIResponse{
|
||
Success: false,
|
||
Message: err.Error(),
|
||
}
|
||
apiRespStr, _ := json.Marshal(apiResp)
|
||
return C.CString(string(apiRespStr))
|
||
}
|
||
apiResp = APIResponse{
|
||
Success: true,
|
||
Data: info,
|
||
}
|
||
apiRespStr, _ := json.Marshal(apiResp)
|
||
return C.CString(string(apiRespStr))
|
||
}
|
||
|
||
// UpdateDocument 更新文档
|
||
//
|
||
//export UpdateDocument
|
||
func UpdateDocument(indexName *C.char, id *C.char, updateData *C.char) *C.char {
|
||
goIndexName := C.GoString(indexName)
|
||
goId := C.GoString(id)
|
||
goUpdateData := C.GoString(updateData)
|
||
var apiResp APIResponse
|
||
// 创建ES客户端
|
||
client, err := newESClient([]string{esAddress}, esUsername, esPassword)
|
||
if err != nil {
|
||
apiResp = APIResponse{
|
||
Success: false,
|
||
Message: err.Error(),
|
||
}
|
||
apiRespStr, _ := json.Marshal(apiResp)
|
||
return C.CString(string(apiRespStr))
|
||
}
|
||
// 转换updateData
|
||
var newUpdateData map[string]interface{}
|
||
if err = json.Unmarshal([]byte(goUpdateData), &newUpdateData); err != nil {
|
||
apiResp = APIResponse{
|
||
Success: false,
|
||
Message: fmt.Sprintf("解析JSON失败: %v", err),
|
||
}
|
||
apiRespStr, _ := json.Marshal(apiResp)
|
||
return C.CString(string(apiRespStr))
|
||
}
|
||
err = client.updateDocument(goIndexName, goId, newUpdateData)
|
||
if err != nil {
|
||
apiResp = APIResponse{
|
||
Success: false,
|
||
Message: err.Error(),
|
||
}
|
||
apiRespStr, _ := json.Marshal(apiResp)
|
||
return C.CString(string(apiRespStr))
|
||
}
|
||
apiResp = APIResponse{
|
||
Success: true,
|
||
}
|
||
apiRespStr, _ := json.Marshal(apiResp)
|
||
return C.CString(string(apiRespStr))
|
||
}
|
||
|
||
// DeleteDocument 删除文档
|
||
//
|
||
//export DeleteDocument
|
||
func DeleteDocument(indexName *C.char, id *C.char) *C.char {
|
||
goIndexName := C.GoString(indexName)
|
||
goId := C.GoString(id)
|
||
var apiResp APIResponse
|
||
// 创建ES客户端
|
||
client, err := newESClient([]string{esAddress}, esUsername, esPassword)
|
||
if err != nil {
|
||
apiResp = APIResponse{
|
||
Success: false,
|
||
Message: err.Error(),
|
||
}
|
||
apiRespStr, _ := json.Marshal(apiResp)
|
||
return C.CString(string(apiRespStr))
|
||
}
|
||
err = client.deleteDocument(goIndexName, goId)
|
||
if err != nil {
|
||
apiResp = APIResponse{
|
||
Success: false,
|
||
Message: err.Error(),
|
||
}
|
||
apiRespStr, _ := json.Marshal(apiResp)
|
||
return C.CString(string(apiRespStr))
|
||
}
|
||
apiResp = APIResponse{
|
||
Success: true,
|
||
}
|
||
apiRespStr, _ := json.Marshal(apiResp)
|
||
return C.CString(string(apiRespStr))
|
||
}
|
||
|
||
// SearchDocuments 搜索文档
|
||
//
|
||
//export SearchDocuments
|
||
func SearchDocuments(indexName *C.char, query *C.char) *C.char {
|
||
goIndexName := C.GoString(indexName)
|
||
goQuery := C.GoString(query)
|
||
var apiResp APIResponse
|
||
// 创建ES客户端
|
||
client, err := newESClient([]string{esAddress}, esUsername, esPassword)
|
||
if err != nil {
|
||
apiResp = APIResponse{
|
||
Success: false,
|
||
Message: err.Error(),
|
||
}
|
||
apiRespStr, _ := json.Marshal(apiResp)
|
||
return C.CString(string(apiRespStr))
|
||
}
|
||
// 转换updateData 解析json
|
||
var newQuery map[string]interface{}
|
||
if err = json.Unmarshal([]byte(goQuery), &newQuery); err != nil {
|
||
apiResp = APIResponse{
|
||
Success: false,
|
||
Message: fmt.Sprintf("解析JSON失败: %v", err),
|
||
}
|
||
apiRespStr, _ := json.Marshal(apiResp)
|
||
return C.CString(string(apiRespStr))
|
||
}
|
||
info, err := client.searchDocuments(goIndexName, newQuery)
|
||
if err != nil {
|
||
apiResp = APIResponse{
|
||
Success: false,
|
||
Message: err.Error(),
|
||
}
|
||
apiRespStr, _ := json.Marshal(apiResp)
|
||
return C.CString(string(apiRespStr))
|
||
}
|
||
apiResp = APIResponse{
|
||
Success: true,
|
||
Data: info,
|
||
}
|
||
apiRespStr, _ := json.Marshal(apiResp)
|
||
return C.CString(string(apiRespStr))
|
||
}
|
||
|
||
// 释放C字符串内存
|
||
//
|
||
//export FreeCString
|
||
func FreeCString(str *C.char) {
|
||
C.free(unsafe.Pointer(str))
|
||
}
|
||
|
||
// main 函数
|
||
//func main() {
|
||
//}
|