eta_api/controllers/data_source/data_source.go

package data_source

import (
	"eta/eta_api/models"
	"eta/eta_api/models/data_manage"
	dataSourceModel "eta/eta_api/models/data_source"
	"eta/eta_api/services/elastic"
	"eta/eta_api/utils"
	"fmt"
	"github.com/rdlucklib/rdluck_tools/paging"
	"strconv"
)

// SearchByEs
// @Title ES搜索
// @Description ES搜索
// @Param   PageSize   query   int  true       "每页数据条数"
// @Param   CurrentIndex   query   int  true       "当前页页码，从1开始"
// @Param   Keyword   query   string  true       "搜索关键词"
// @Param   Source   query   int  true       "数据源"
// @Success 200 {object} response.ExcelListResp
// @router /common/search_by_es [get]
func (c *DataSourceController) SearchByEs() {
	br := new(models.BaseResponse).Init()
	defer func() {
		if br.ErrMsg == "" {
			br.IsSendEmail = false
		}
		c.Data["json"] = br
		c.ServeJSON()
	}()
	sysUser := c.SysUser
	if sysUser == nil {
		br.Msg = "请登录"
		br.ErrMsg = "请登录，SysUser Is Empty"
		br.Ret = 408
		return
	}

	pageSize, _ := c.GetInt("PageSize")
	currentIndex, _ := c.GetInt("CurrentIndex")
	keyword := c.GetString("KeyWord")
	if keyword == `` {
		keyword = c.GetString("Keyword")
	}
	source, _ := c.GetInt("Source")
	if source <= 0 {
		br.Msg = "来源有误"
		br.ErrMsg = fmt.Sprintf("数据来源有误, Source: %d", source)
		return
	}
	subSource, _ := c.GetInt("SubSource")

	// 以下为兼容各旧接口的额外传参，不为空时修改Resp对应的Key
	primaryIdKey := c.GetString("PrimaryIdKey")
	indexNameKey := c.GetString("IndexNameKey")
	classifyIdKey := c.GetString("ClassifyIdKey")

	var total, startSize int
	if pageSize <= 0 {
		pageSize = utils.PageSize15
	}
	if currentIndex <= 0 {
		currentIndex = 1
	}
	startSize = paging.StartIndex(currentIndex, pageSize)

	// es搜索
	t, list, e := elastic.SearchDataSourceIndex(utils.EsDataSourceIndexName, keyword, source, subSource, []int{}, []int{}, []string{}, startSize, pageSize)
	if e != nil {
		br.Msg = "获取失败"
		br.ErrMsg = fmt.Sprintf("ES-搜索数据源列表失败, %v", e)
		return
	}
	total = int(t)

	// （短期方案）中国煤炭市场网/煤炭江湖额外补充基础字段
	indexModifyTime := make(map[string]string)
	indexBaseInfo := make(map[string]*dataSourceModel.BaseFromCoalmineIndexBase)
	{
		if source == utils.DATA_SOURCE_MTJH {
			var updateCodes []string
			for _, v := range list {
				if v.ModifyTime == "" {
					updateCodes = append(updateCodes, v.IndexCode)
				}
			}
			if len(updateCodes) > 0 {
				baseIndexes, e := dataSourceModel.GetMtjhBaseInfoFromDataTable(updateCodes)
				if e != nil {
					br.Msg = "获取失败"
					br.ErrMsg = fmt.Sprintf("获取煤炭江湖指标基础信息失败, %v", e)
					return
				}
				for _, v := range baseIndexes {
					indexModifyTime[v.IndexCode] = utils.TimeTransferString(utils.FormatDateTime, v.ModifyTime)
				}
			}
		}
		if source == utils.DATA_SOURCE_COAL {
			var updateCodes []string
			for _, v := range list {
				if v.Unit == "" || v.Frequency == "" || v.ModifyTime == "" {
					updateCodes = append(updateCodes, v.IndexCode)
				}
			}
			if len(updateCodes) > 0 {
				baseIndexes, e := dataSourceModel.GetCoalmineBaseInfoFromDataTable(updateCodes)
				if e != nil {
					br.Msg = "获取失败"
					br.ErrMsg = fmt.Sprintf("获取中国煤炭市场网指标基础信息失败, %v", e)
					return
				}
				for _, v := range baseIndexes {
					indexBaseInfo[v.IndexCode] = v
				}
			}
		}
	}

	listMap := make([]map[string]interface{}, 0)
	for _, v := range list {
		// （短期方案）由于start_date、end_date和latest_value字段不全的历史遗留问题，这里查出来并补充进ES里面去
		var updateEs bool
		if v.StartDate == "" || v.EndDate == "" || v.LatestValue == "" || v.LatestValue == "0" {
			minMax, e := dataSourceModel.GetBaseIndexDataMinMax(v.Source, v.SubSource, v.IndexCode)
			if e != nil && e.Error() != utils.ErrNoRow() {
				br.Msg = "获取失败"
				br.ErrMsg = fmt.Sprintf("获取指标开始结束时间失败, %v", e)
				return
			}
			if minMax != nil {
				v.StartDate = minMax.MinDate
				v.EndDate = minMax.MaxDate
				v.LatestValue = minMax.LatestValue
				updateEs = true
			}
		}

		// 煤炭江湖缺ModifyTime
		if source == utils.DATA_SOURCE_MTJH && v.ModifyTime == "" && indexModifyTime[v.IndexCode] != "" {
			v.ModifyTime = indexModifyTime[v.IndexCode]
			updateEs = true
		}

		// 中国煤炭市场网缺Unit, Frequency, ModifyTime
		if source == utils.DATA_SOURCE_COAL && (v.Unit == "" || v.Frequency == "" || v.ModifyTime == "") {
			if indexBaseInfo[v.IndexCode] != nil {
				v.Unit = indexBaseInfo[v.IndexCode].Unit
				v.Frequency = indexBaseInfo[v.IndexCode].Frequency
				v.ModifyTime = utils.TimeTransferString(utils.FormatDateTime, indexBaseInfo[v.IndexCode].ModifyTime)
				updateEs = true
			}
		}

		// 写入ES更新队列
		if updateEs {
			if e := utils.Rc.LPush(utils.CACHE_DATA_SOURCE_ES_HANDLE, v); e != nil {
				br.Msg = "获取失败"
				br.ErrMsg = fmt.Sprintf("写入ES更新队列失败, Source: %d, IndexCode: %s, err: %v", v.Source, v.IndexCode, e)
				return
			}
		}

		// 转换成map返回
		listMap = append(listMap, v.ToMap(primaryIdKey, indexNameKey, classifyIdKey))
	}

	// 美国农业部（多一个ParentClassifyId字段）
	if source == utils.DATA_SOURCE_USDA_FAS {
		// 父级分类ID
		classifyIds := make([]int, 0)
		for _, v := range list {
			classifyIds = append(classifyIds, v.ClassifyId)
		}
		classifyList, e := data_manage.GetBaseFromUsdaFasClassifyByIds(classifyIds)
		if e != nil {
			br.Msg = "获取失败"
			br.ErrMsg = fmt.Sprintf("获取美国农业部分类失败, %v", e)
			return
		}
		classifyMap := make(map[int]int)
		for _, v := range classifyList {
			classifyMap[v.ClassifyId] = v.ParentId
		}
		for _, v := range listMap {
			id, ok := v["ClassifyId"].(int)
			if !ok {
				v["ParentClassifyId"] = 0
				continue
			}
			v["ParentClassifyId"] = classifyMap[id]
		}
	}

	// 煤炭江湖（多一个Area字段）
	if source == utils.DATA_SOURCE_MTJH {
		cond := ``
		pars := make([]interface{}, 0)
		indexes, e := data_manage.GetMtjhItemsByCondition(cond, pars)
		if e != nil {
			br.Msg = "获取失败"
			br.ErrMsg = fmt.Sprintf("获取煤炭江湖指标失败, %v", e)
			return
		}
		indexArea := make(map[string]string)
		for _, v := range indexes {
			indexArea[v.IndexCode] = v.Area
		}
		for _, v := range listMap {
			code, ok := v["IndexCode"].(string)
			if !ok {
				v["Area"] = ""
				continue
			}
			v["Area"] = indexArea[code]
		}
	}

	// 分类的唯一编码(前端定位用)
	if classifyIdKey == "" {
		classifyIdKey = "ClassifyId"
	}
	idUniqueCodeArr := []int{
		utils.DATA_SOURCE_SCI_HQ, utils.DATA_SOURCE_MYSTEEL_CHEMICAL, utils.DATA_SOURCE_YS, utils.DATA_SOURCE_EIA_STEO,
	}
	if utils.InArrayByInt(idUniqueCodeArr, source) {
		for _, v := range listMap {
			classifyId, ok := v[classifyIdKey].(int)
			if !ok {
				v["ClassifyUniqueCode"] = ""
				continue
			}
			v["ClassifyUniqueCode"] = strconv.Itoa(classifyId)
		}
	}
	//if source == utils.DATA_SOURCE_MANUAL {
	//	// 手工指标
	//	var wxUserId int64
	//	wxUserId = int64(sysUser.AdminId)
	//	if sysUser.RoleTypeCode == utils.ROLE_TYPE_CODE_ADMIN {
	//		wxUserId = 0
	//	}
	//	classifies, err := models.GetEdbdataClassify(wxUserId)
	//	if err != nil {
	//		br.Msg = "获取失败"
	//		br.ErrMsg = fmt.Sprintf("获取手工指标分类失败, %v", e)
	//		return
	//	}
	//	unicodeMap := make(map[int]string)
	//	for _, v := range classifies {
	//		unicodeMap[v.ClassifyId] = v.UniqueCode
	//	}
	//	for _, v := range listMap {
	//		classifyId, ok := v["ClassifyId"].(int)
	//		if !ok {
	//			continue
	//		}
	//		v["ClassifyUniqueCode"] = unicodeMap[classifyId]
	//	}
	//}

	page := paging.GetPaging(currentIndex, pageSize, total)
	resp := dataSourceModel.SearchDataSourceResp{
		Paging: page,
		List:   listMap,
	}
	br.Data = resp
	br.Ret = 200
	br.Success = true
	br.Msg = "获取成功"
}