eta_api/services/elastic.go

package services

import (
	"context"
	"encoding/json"
	"fmt"
	"github.com/olivere/elastic/v7"
	"hongze/hz_eta_api/models"
	saModel "hongze/hz_eta_api/models/semantic_analysis"
	"hongze/hz_eta_api/services/alarm_msg"
	"hongze/hz_eta_api/utils"
	"strings"
)

func NewClient() (client *elastic.Client, err error) {
	client, err = elastic.NewClient(
		elastic.SetURL(utils.ES_URL),
		elastic.SetBasicAuth(utils.ES_USERNAME, utils.ES_PASSWORD),
		elastic.SetSniff(false))
	return
}

// indexName:索引名称
// mappingJson:表结构
func EsCreateIndex(indexName, mappingJson string) (err error) {
	client, err := NewClient()
	if err != nil {
		return
	}
	//定义表结构
	exists, err := client.IndexExists(indexName).Do(context.Background()) //<5>
	if err != nil {
		return
	}
	if !exists {
		resp, err := client.CreateIndex(indexName).BodyJson(mappingJson).Do(context.Background())
		//BodyJson(bodyJson).Do(context.Background())
		if err != nil {
			fmt.Println("CreateIndex Err:" + err.Error())
			return err
		}
		fmt.Println(resp.Index, resp.ShardsAcknowledged, resp.Acknowledged)
	} else {
		fmt.Println(indexName + " 已存在")
	}
	return
}

// 删除数据
func EsDeleteData(indexName, docId string) (err error) {
	client, err := NewClient()
	if err != nil {
		return
	}
	resp, err := client.Delete().Index(indexName).Id(docId).Do(context.Background())
	fmt.Println(resp)
	if err != nil {
		return
	}
	if resp.Status == 0 {
		fmt.Println("删除成功")
	} else {
		fmt.Println("AddData", resp.Status, resp.Result)
	}
	return
}

func MappingModify(indexName, mappingJson string) {
	client, err := NewClient()
	if err != nil {
		return
	}
	result, err := client.PutMapping().Index(indexName).BodyString(mappingJson).Do(context.Background())
	fmt.Println(err)
	fmt.Println(result)
	return
}

// EsAddOrEditReport 新增编辑es报告
func EsAddOrEditReport(indexName, docId string, item *models.ElasticReportDetail) (err error) {
	defer func() {
		if err != nil {
			fmt.Println("EsAddOrEditReport Err:", err.Error())
		}
	}()
	client, err := NewClient()
	if err != nil {
		return
	}
	// docId为报告ID+章节ID
	searchById, err := client.Get().Index(indexName).Id(docId).Do(context.Background())
	if err != nil && !strings.Contains(err.Error(), "404") {
		fmt.Println("Get Err" + err.Error())
		return
	}
	if searchById != nil && searchById.Found {
		resp, err := client.Update().Index(indexName).Id(docId).Doc(map[string]interface{}{
			"ReportId":           item.ReportId,
			"ReportChapterId":    item.ReportChapterId,
			"Title":              item.Title,
			"Abstract":           item.Abstract,
			"BodyContent":        item.BodyContent,
			"PublishTime":        item.PublishTime,
			"PublishState":       item.PublishState,
			"Author":             item.Author,
			"ClassifyIdFirst":    item.ClassifyIdFirst,
			"ClassifyNameFirst":  item.ClassifyNameFirst,
			"ClassifyIdSecond":   item.ClassifyIdSecond,
			"ClassifyNameSecond": item.ClassifyNameSecond,
			"Categories":         item.Categories,
			"StageStr":           item.StageStr,
		}).Do(context.Background())
		if err != nil {
			return err
		}
		//fmt.Println(resp.Status, resp.Result)
		if resp.Status == 0 {
			fmt.Println("修改成功" + docId)
			err = nil
		} else {
			fmt.Println("EditData", resp.Status, resp.Result)
		}
	} else {
		resp, err := client.Index().Index(indexName).Id(docId).BodyJson(item).Do(context.Background())
		if err != nil {
			fmt.Println("新增失败:", err.Error())
			return err
		}
		if resp.Status == 0 && resp.Result == "created" {
			fmt.Println("新增成功" + docId)
			return nil
		} else {
			fmt.Println("AddData", resp.Status, resp.Result)
		}
	}
	return
}

// EsAddOrEditEnglishReport 新增编辑es英文报告
func EsAddOrEditEnglishReport(indexName, docId string, item *models.ElasticEnglishReportDetail) (err error) {
	defer func() {
		if err != nil {
			fmt.Println("EsAddOrEditEnglishReport Err:", err.Error())
			go alarm_msg.SendAlarmMsg("新增编辑es英文报告 EsAddOrEditEnglishReport，Err："+err.Error(), 3)
		}
	}()
	client, err := NewClient()
	if err != nil {
		return
	}
	// docId为报告ID
	searchById, err := client.Get().Index(indexName).Id(docId).Do(context.Background())
	if err != nil {
		if strings.Contains(err.Error(), "404") {
			err = nil
		} else {
			fmt.Println("Get Err" + err.Error())
			return
		}
	}
	if searchById != nil && searchById.Found {
		resp, e := client.Update().Index(indexName).Id(docId).Doc(map[string]interface{}{
			"Id":                 item.Id,
			"ReportId":           item.ReportId,
			"VideoId":            item.VideoId,
			"Title":              item.Title,
			"Abstract":           item.Abstract,
			"BodyContent":        item.BodyContent,
			"PublishTime":        item.PublishTime,
			"PublishState":       item.PublishState,
			"Author":             item.Author,
			"ClassifyIdFirst":    item.ClassifyIdFirst,
			"ClassifyNameFirst":  item.ClassifyNameFirst,
			"ClassifyIdSecond":   item.ClassifyIdSecond,
			"ClassifyNameSecond": item.ClassifyNameSecond,
			"CreateTime":         item.CreateTime,
			"Overview":           item.Overview,
			"ReportCode":         item.ReportCode,
			"Frequency":          item.Frequency,
			"StageStr":           item.StageStr,
			"ContentSub":         item.ContentSub,
		}).Do(context.Background())
		if e != nil {
			err = e
			return
		}
		//fmt.Println(resp.Status, resp.Result)
		if resp.Status == 0 {
			fmt.Println("修改成功" + docId)
			err = nil
		} else {
			fmt.Println("EditData", resp.Status, resp.Result)
		}
	} else {
		resp, e := client.Index().Index(indexName).Id(docId).BodyJson(item).Do(context.Background())
		if e != nil {
			err = e
			fmt.Println("新增失败:", err.Error())
			return
		}
		if resp.Status == 0 && resp.Result == "created" {
			fmt.Println("新增成功" + docId)
			return
		} else {
			fmt.Println("AddData", resp.Status, resp.Result)
		}
	}
	return
}

// EsAddOrEditSaCompare 新增编辑es语义分析文档对比
func EsAddOrEditSaCompare(indexName, docId string, item *saModel.SaCompareElastic) (err error) {
	defer func() {
		if err != nil {
			fmt.Println("EsAddOrEditSaCompare Err:", err.Error())
			go alarm_msg.SendAlarmMsg("新增编辑es语义分析文档对比 EsAddOrEditSaCompare，Err："+err.Error(), 3)
		}
	}()
	client := utils.EsClient
	// docId为报告ID
	searchById, err := client.Get().Index(indexName).Id(docId).Do(context.Background())
	if err != nil {
		if strings.Contains(err.Error(), "404") {
			err = nil
		} else {
			fmt.Println("Get Err" + err.Error())
			return
		}
	}
	if searchById != nil && searchById.Found {
		resp, e := client.Update().Index(indexName).Id(docId).Doc(map[string]interface{}{
			"SaCompareId":  item.SaCompareId,
			"ClassifyName": item.ClassifyName,
			"ClassifyId":   item.ClassifyId,
			"Title":        item.Title,
			"ResultImg":    item.ResultImg,
			"CreateTime":   item.CreateTime,
			"ModifyTime":   item.ModifyTime,
			"SysAdminId":   item.SysAdminId,
			"SysAdminName": item.SysAdminName,
		}).Do(context.Background())
		if e != nil {
			err = e
			return
		}
		//fmt.Println(resp.Status, resp.Result)
		if resp.Status == 0 {
			fmt.Println("修改成功" + docId)
			err = nil
		} else {
			fmt.Println("EditData", resp.Status, resp.Result)
		}
	} else {
		resp, e := client.Index().Index(indexName).Id(docId).BodyJson(item).Do(context.Background())
		if e != nil {
			err = e
			fmt.Println("新增失败:", err.Error())
			return
		}
		if resp.Status == 0 && resp.Result == "created" {
			fmt.Println("新增成功" + docId)
			return
		} else {
			fmt.Println("AddData", resp.Status, resp.Result)
		}
	}
	return
}

// EsAddOrEditSaDoc 新增编辑语义分析文档
func EsAddOrEditSaDoc(indexName, docId string, item *saModel.ElasticSaDoc) (err error) {
	defer func() {
		if err != nil {
			fmt.Println("EsAddOrEditSaDoc Err:", err.Error())
		}
	}()

	client, e := NewClient()
	if e != nil {
		err = e
		return
	}

	// docId为语义分析文档ID+段落ID
	searchById, e := client.Get().Index(indexName).Id(docId).Do(context.Background())
	if e != nil && !strings.Contains(e.Error(), "404") {
		err = fmt.Errorf("query sa doc err: %s", e.Error())
		return
	}

	// 更新
	if searchById != nil && searchById.Found {
		docMap := map[string]interface{}{
			"SaDocId":        item.SaDocId,
			"SaDocSectionId": item.SaDocSectionId,
			"ClassifyId":     item.ClassifyId,
			"ClassifyName":   item.ClassifyName,
			"Title":          item.Title,
			"Theme":          item.Theme,
			"BodyContent":    item.BodyContent,
			"Author":         item.Author,
			"CoverImg":       item.CoverImg,
			"CreateTime":     item.CreateTime,
		}

		resp, e := client.Update().Index(indexName).Id(docId).Doc(docMap).Do(context.Background())
		if e != nil {
			err = fmt.Errorf("update sa doc err: %s", e.Error())
			return
		}
		if resp.Status == 0 {
			return
		}
		fmt.Println("EditData", resp.Status, resp.Result)
		return
	}

	// 新增
	resp, e := client.Index().Index(indexName).Id(docId).BodyJson(item).Do(context.Background())
	if e != nil {
		err = fmt.Errorf("insert sa doc err: %s", e.Error())
		return
	}
	if resp.Status == 0 && resp.Result == "created" {
		return
	}
	fmt.Println("AddData", resp.Status, resp.Result)
	return
}

// EsSearchSaCompareData 查询es中的文档对比信息
func EsSearchSaCompareData(keywordStr string, from, size int) (total int, list []*saModel.SaCompareElastic, err error) {
	list = make([]*saModel.SaCompareElastic, 0)
	indexName := utils.EsSemanticAnalysisCompareIndexName
	var tmpTotal int64
	defer func() {
		if err != nil {
			fmt.Println("EsSearchSaCompareData Err:", err.Error())
		}
	}()
	client := utils.EsClient

	mustMap := make([]interface{}, 0)
	mustNotMap := make([]interface{}, 0)

	//关键字匹配
	shouldMap := map[string]interface{}{
		"should": []interface{}{
			map[string]interface{}{
				"match_phrase": map[string]interface{}{
					"Title": keywordStr,
					//"Frequency.keyword": "月度",
				},
			},
		},
	}
	mustMap = append(mustMap, map[string]interface{}{
		"bool": shouldMap,
	})

	queryMap := map[string]interface{}{
		"query": map[string]interface{}{
			"bool": map[string]interface{}{
				"must":     mustMap,
				"must_not": mustNotMap,
				//"should":   shouldMap,
			},
		},
	}

	//根据条件数量统计
	requestTotalHits := client.Count(indexName).BodyJson(queryMap)
	tmpTotal, err = requestTotalHits.Do(context.Background())
	if err != nil {
		return
	}
	total = int(tmpTotal)
	// 分页查询
	queryMap["from"] = from
	queryMap["size"] = size
	queryMap["sort"] = []map[string]interface{}{
		map[string]interface{}{
			"CreateTime.keyword": map[string]interface{}{
				"order": "desc",
			},
		},
		map[string]interface{}{
			"_score": map[string]interface{}{
				"order": "desc",
			},
		},
	}
	jsonBytes, _ := json.Marshal(queryMap)
	fmt.Println(string(jsonBytes))

	request := client.Search(indexName).Source(queryMap) // sets the JSON request

	searchMap := make(map[string]string)

	searchResp, err := request.Do(context.Background())
	if err != nil {
		return
	}
	fmt.Println(searchResp)
	fmt.Println(searchResp.Status)
	if searchResp.Status != 0 {
		return
	}

	if searchResp.Hits != nil {
		for _, v := range searchResp.Hits.Hits {
			if _, ok := searchMap[v.Id]; !ok {
				itemJson, tmpErr := v.Source.MarshalJSON()
				if tmpErr != nil {
					err = tmpErr
					fmt.Println("movieJson err:", err)
					return
				}
				compareItem := new(saModel.SaCompareElastic)
				tmpErr = json.Unmarshal(itemJson, &compareItem)
				if err != nil {
					fmt.Println("json.Unmarshal itemJson err:", err)
					err = tmpErr
					return
				}

				list = append(list, compareItem)
				searchMap[v.Id] = v.Id
			}
		}
	}
	return
}