eta_index_lib/services/base_from_wind.go

package services

import (
	"encoding/json"
	"errors"
	"eta/eta_index_lib/models"
	"eta/eta_index_lib/services/alarm_msg"
	"eta/eta_index_lib/utils"
	"fmt"
	"github.com/rdlucklib/rdluck_tools/http"
	"time"
)

const (
	WindNoAuthCode = -40522015
)

// GetEdbDataFromWind 获取wind数据
func GetEdbDataFromWind(edbCode, startDate, endDate string) (item *models.EdbDataFromWind, errorCode int, apiErrMsg string, err error) {
	windUrl, err := GetWindUrl(edbCode)
	if err != nil {
		errorCode = 421
		go alarm_msg.SendAlarmMsg(fmt.Sprintf("获取wind服务器地址失败，err:%s", err.Error()), 3)
		return
	}
	thsUrl := windUrl + `edbInfo/wind?EdbCode=%s&StartDate=%s&EndDate=%s`
	thsUrl = fmt.Sprintf(thsUrl, edbCode, startDate, endDate)
	utils.FileLog.Info(fmt.Sprintf("windUrl:%s", thsUrl))
	body, err := http.Get(thsUrl)
	fmt.Println("GetEdbDataByWind body:")
	fmt.Println(string(body))
	utils.FileLog.Info(fmt.Sprint("指标编码：", edbCode, ";wind result:", string(body)))

	if err != nil {
		apiErrMsg = err.Error()
		return
	}
	item = new(models.EdbDataFromWind)
	err = json.Unmarshal(body, &item)

	//异常的话，需要邮件通知
	if len(item.ErrorCode) > 0 {
		if item.ErrorCode["0"] != 0 {
			if item.ErrorCode["0"] == -40522017 {
				//{
				//DT: {
				//0: 1654646400000
				//},
				//CLOSE: {
				//0: "CEDBService:: quota exceeded."
				//},
				//ErrorCode: {
				//0: -40522017
				//}
				//}
				// 设置服务器已超限
				SetIsLimitEdbCodeInWindUrl(windUrl)
				err = DeleteEdbCodeInWindUrl(edbCode)
				if err != nil {
					return
				}
				apiErrMsg = fmt.Sprintf("ErrorCode:%d, wind数据接口超限", item.ErrorCode["0"])
				return GetEdbDataFromWind(edbCode, startDate, endDate)
			} else if item.ErrorCode["0"] == -40520005 {
				//.ErrorCode=-40520005
				//.Data=[No Python API Authority
				SetIsLimitEdbCodeInWindUrl(windUrl)
				err = DeleteEdbCodeInWindUrl(edbCode)
				if err != nil {
					return
				}
				apiErrMsg = fmt.Sprintf("ErrorCode:%d, 无API接口权限", item.ErrorCode["0"])
				go alarm_msg.SendAlarmMsg(fmt.Sprintf("wind数据服务异常，edbCode：%s，ErrorCode:%d，result:%s", edbCode, item.ErrorCode["0"], string(body)), 3)
				return GetEdbDataFromWind(edbCode, startDate, endDate)
			} else if item.ErrorCode["0"] == WindNoAuthCode {
				// 指标下架, 无权限
				apiErrMsg = fmt.Sprintf("ErrorCode:%d, 指标已下架", item.ErrorCode["0"])
				return nil, WindNoAuthCode, apiErrMsg, nil
			} else {
				apiErrMsg = fmt.Sprintf("ErrorCode:%d, wind数据服务异常", item.ErrorCode["0"])
				go alarm_msg.SendAlarmMsg(fmt.Sprintf("wind数据服务异常，edbCode：%s，ErrorCode:%d，result:%s", edbCode, item.ErrorCode["0"], string(body)), 3)
			}
		}
	}
	return
}

// GetEdbDataFromWindUrl 通过url获取wind数据
func GetEdbDataFromWindUrl(windUrl, edbCode, startDate, endDate string) (item *models.EdbDataFromWind, errorCode int, apiErrMsg string, err error) {
	// todo 刷新后的处理
	if windUrl == `` {
		return GetEdbDataFromWind(edbCode, startDate, endDate)
	}

	requestWindUrl := windUrl + `edbInfo/wind?EdbCode=%s&StartDate=%s&EndDate=%s`
	requestWindUrl = fmt.Sprintf(requestWindUrl, edbCode, startDate, endDate)
	utils.FileLog.Info(fmt.Sprintf("windUrl:%s", requestWindUrl))
	body, err := http.Get(requestWindUrl)
	fmt.Println("GetEdbDataByWind body:")
	fmt.Println(string(body))
	utils.FileLog.Info(fmt.Sprint("wind result:", string(body)))

	if err != nil {
		return
	}
	item = new(models.EdbDataFromWind)
	err = json.Unmarshal(body, &item)

	//异常的话，需要邮件通知
	if len(item.ErrorCode) > 0 {
		if item.ErrorCode["0"] != 0 {
			if item.ErrorCode["0"] == -40522017 {
				//{
				//DT: {
				//0: 1654646400000
				//},
				//CLOSE: {
				//0: "CEDBService:: quota exceeded."
				//},
				//ErrorCode: {
				//0: -40522017
				//}
				//}
				// 设置服务器已超限
				errorCode = 421
				apiErrMsg = fmt.Sprintf("ErrorCode:%d, wind数据接口超限", item.ErrorCode["0"])
				go alarm_msg.SendAlarmMsg(fmt.Sprintf("wind数据接口超限，地址：%s", requestWindUrl), 2)
				//go utils.SendEmail("wind数据接口超限", "地址:"+requestWindUrl, utils.EmailSendToUsers)
				return
			} else if item.ErrorCode["0"] == -40520005 {
				//.ErrorCode=-40520005
				//.Data=[No Python API Authority
				err = errors.New("No Python API Authority")
				apiErrMsg = fmt.Sprintf("ErrorCode:%d, 无API接口权限", item.ErrorCode["0"])
				go alarm_msg.SendAlarmMsg(fmt.Sprintf("wind数据服务异常，edbCode：%s，ErrorCode:%d，result:%s", edbCode, item.ErrorCode["0"], string(body)), 3)
				return
			} else if item.ErrorCode["0"] == WindNoAuthCode {
				// 指标下架, 无权限
				apiErrMsg = fmt.Sprintf("ErrorCode:%d, 指标已下架", item.ErrorCode["0"])
				return nil, WindNoAuthCode, apiErrMsg, nil
			} else {
				apiErrMsg = fmt.Sprintf("ErrorCode:%d, wind数据服务异常", item.ErrorCode["0"])
				go alarm_msg.SendAlarmMsg(fmt.Sprintf("wind数据服务异常，edbCode：%s，ErrorCode:%d，result:%s", edbCode, item.ErrorCode["0"], string(body)), 3)
			}
		}
	}
	return
}

// GetWindUrl 获取wind的url
func GetWindUrl(edbCode string) (windUrl string, err error) {
	defer func() {
		if err == nil && windUrl == "" {
			err = errors.New("获取wind服务器地址失败，指标超限了")
		}
	}()
	//从缓存中获取
	cacheKey := utils.CACHE_WIND_URL + ":" + edbCode
	windUrl, _ = utils.Rc.RedisString(cacheKey)
	if windUrl != "" {
		return
	}
	terminalList, err := models.GetEdbTerminalListBySource(utils.DATA_SOURCE_WIND)
	if err != nil {
		return
	}
	//如果缓存中没有的话，那么从配置中获取
	for _, t := range terminalList {
		//判断该url是否被占满了
		//count, tmpErr := GetCountEdbCodeInWindUrl(windUrlMap.Url)
		//if tmpErr != nil && tmpErr.Error() != "nil returned" {
		//	err = tmpErr
		//	return
		//}
		//if count < windUrlMap.Num {
		//	windUrl = windUrlMap.Url
		//	AddEdbCodeInWindUrl(windUrlMap.Url, edbCode)
		//	return
		//}
		//如果超限了，那么进入下一循环,
		isLimit, tmpErr := GetIsLimitEdbCodeInWindUrl(t.ServerUrl)
		if isLimit {
			err = tmpErr
			continue
		}

		windUrl = t.ServerUrl
		AddEdbCodeInWindUrl(windUrl, t.TerminalCode, edbCode)
		return
	}
	return
}

// GetCountEdbCodeInWindUrl 从缓存key中获取已经插入入的指标数
func GetCountEdbCodeInWindUrl(windUrl string) (num int, err error) {
	cacheKey := utils.CACHE_WIND_URL + time.Now().Format(utils.FormatDateUnSpace) + ":" + utils.MD5(windUrl)
	num, err = utils.Rc.RedisInt(cacheKey)
	if err != nil && err.Error() == "redigo: nil returned" {
		err = nil
	}
	return
}

// GetIsLimitEdbCodeInWindUrl 从缓存key中获取是否超限
func GetIsLimitEdbCodeInWindUrl(windUrl string) (isLimit bool, err error) {
	cacheKey := utils.CACHE_WIND_URL + ":limit:" + time.Now().Format(utils.FormatDateUnSpace) + ":" + utils.MD5(windUrl)

	num, err := utils.Rc.RedisInt(cacheKey)
	if err != nil && err.Error() == "redigo: nil returned" {
		err = nil
	}
	if num > 0 {
		isLimit = true
	}
	return
}

// SetIsLimitEdbCodeInWindUrl 设置服务器已超限
func SetIsLimitEdbCodeInWindUrl(windUrl string) {
	cacheKey := utils.CACHE_WIND_URL + ":limit:" + time.Now().Format(utils.FormatDateUnSpace) + ":" + utils.MD5(windUrl)
	_ = utils.Rc.SetNX(cacheKey, 1, utils.GetTodayLastSecond())
	return
}

// AddEdbCodeInWindUrl 将指标插入到缓存key中
// @return isInsert bool 是否插入数据，true时为插入数据，false表示数据已存在
func AddEdbCodeInWindUrl(windUrl, terminalCode, edbCode string) (isInsert bool) {
	cacheKey := utils.CACHE_WIND_URL + ":" + edbCode
	isInsert = utils.Rc.SetNX(cacheKey, windUrl, utils.GetTodayLastSecond())

	cacheKey2 := utils.CACHE_WIND_URL + time.Now().Format(utils.FormatDateUnSpace) + ":" + utils.MD5(windUrl)
	utils.Rc.Incrby(cacheKey2, 1)

	// 设置指标与终端关系的缓存
	terminalCodeCacheKey := utils.CACHE_EDB_TERMINAL_CODE_URL + edbCode
	_ = utils.Rc.Put(terminalCodeCacheKey, terminalCode, utils.GetTodayLastSecond())
	return
}

// DeleteEdbCodeInWindUrl 删除指标编码 服务器归属 缓存
func DeleteEdbCodeInWindUrl(edbCode string) (err error) {
	cacheKey := utils.CACHE_WIND_URL + ":" + edbCode
	err = utils.Rc.Delete(cacheKey)

	terminalCodeCacheKey := utils.CACHE_EDB_TERMINAL_CODE_URL + edbCode
	err = utils.Rc.Delete(terminalCodeCacheKey)
	return
}