test 嘉悦数据源-新增刷新

controllers/base_from_jiayue.go

@@ -7,7 +7,6 @@ import (
 	"eta/eta_index_lib/services"
 	"eta/eta_index_lib/utils"
 	"strconv"
-	"strings"
 	"time"
 )
 
@@ -32,22 +31,25 @@ func (this *JiaYueController) Add() {
 		this.Data["json"] = br
 		this.ServeJSON()
 	}()
-
-	// TODO:桥接
-	source := utils.DATA_SOURCE_WIND
 	var req models.AddEdbInfoReq
-	err := json.Unmarshal(this.Ctx.Input.RequestBody, &req)
-	if err != nil {
-		br.Msg = "参数解析异常！"
-		br.ErrMsg = "参数解析失败，Err:" + err.Error()
+	if e := json.Unmarshal(this.Ctx.Input.RequestBody, &req); e != nil {
+		br.Msg = "操作失败"
+		br.ErrMsg = "参数解析异常, err: " + e.Error()
 		return
 	}
 	if req.EdbCode == "" {
-		br.Msg = "请输入指标编码！"
-		br.ErrMsg = "请输入指标编码，指标编码为空"
+		br.Msg = "请输入指标编码"
+		br.ErrMsg = "指标编码为空"
+		return
+	}
+	if req.Source <= 0 {
+		br.Msg = "请输入指标来源"
+		br.ErrMsg = "指标来源为空"
 		return
 	}
-	cacheKey = utils.CACHE_EDB_DATA_ADD + strconv.Itoa(source) + "_" + req.EdbCode
+
+	// 加锁
+	cacheKey = utils.CACHE_EDB_DATA_ADD + strconv.Itoa(req.Source) + "_" + req.EdbCode
 	if utils.Rc.IsExist(cacheKey) {
 		br.Ret = 501
 		br.Success = true
@@ -55,38 +57,50 @@ func (this *JiaYueController) Add() {
 		return
 	}
 	utils.Rc.SetNX(cacheKey, 1, 1*time.Minute)
+	defer func() {
+		_ = utils.Rc.Delete(cacheKey)
+	}()
 
-	// 期货数据，就默认到今天，特殊处理下
-	endDate := utils.BASE_END_DATE
-	if strings.Contains(req.EdbCode, ".") {
-		endDate = time.Now().Format(utils.FormatDate)
+	// 校验数据源
+	cond := ` AND edb_source_id = ?`
+	pars := make([]interface{}, 0)
+	pars = append(pars, req.Source)
+	sourceItem, e := models.GetEdbSourceItemByCondition(cond, pars)
+	if e != nil {
+		br.Msg = "数据源有误"
+		br.ErrMsg = "获取数据源失败, Err: " + e.Error()
+		return
 	}
 
-	// 走桥接服务
-	windUrl, _ := services.GetWindBridgeUrl()
-	if windUrl == "" {
-		br.Msg = "wind服务地址有误"
-		br.ErrMsg = "wind服务地址有误"
+	// 从桥接服务获取指标和数据
+	var params models.BridgeJiaYueIndexDataParams
+	params.IndexCode = req.EdbCode
+	params.SourceExtend = sourceItem.SourceExtend
+	params.StartDate = utils.BASE_START_DATE
+	params.EndDate = utils.BASE_END_DATE
+	indexData, e := services.GetJiaYueIndexDataFromBridge(params)
+	if e != nil {
+		br.Msg = "获取指标失败"
+		br.ErrMsg = "获取指标数据失败, Err: " + e.Error()
 		return
 	}
-
-	var dataList []*models.EdbDataFromWindSimple
-	var errMsg string
-	dataList, err, errMsg = services.GetEdbDataFromBridge(windUrl, req, endDate)
-	if err != nil {
-		br.Msg = "获取指标信息失败！"
-		br.ErrMsg = "获取桥接服务指标信息失败，Err：" + errMsg + err.Error()
+	if indexData.Id <= 0 {
+		br.Msg = "指标不存在"
+		br.Success = true
 		return
 	}
-	err = models.AddEdbDataFromWind(req.EdbCode, dataList)
-	if err != nil {
-		br.Msg = "获取指标信息失败！"
-		br.ErrMsg = "获取指标信息失败 AddEdbDataFromWind，Err:" + err.Error()
+
+	// 新增指标数据
+	e = models.AddEdbDataFromJiaYue(sourceItem.TableName, req.EdbCode, indexData.IndexData)
+	if e != nil {
+		br.Msg = "操作失败"
+		br.ErrMsg = "新增指标数据失败, Err: " + e.Error()
 		return
 	}
+
 	br.Ret = 200
 	br.Success = true
-	br.Msg = "获取成功"
+	br.Msg = "操作成功"
 }
 
 // Refresh
@@ -104,7 +118,6 @@ func (this *JiaYueController) Refresh() {
 		this.Data["json"] = br
 		this.ServeJSON()
 	}()
-	source := utils.DATA_SOURCE_WIND
 	var req models.RefreshEdbInfoReq
 	err := json.Unmarshal(this.Ctx.Input.RequestBody, &req)
 	if err != nil {
@@ -122,82 +135,77 @@ func (this *JiaYueController) Refresh() {
 		br.ErrMsg = "请输入指标ID"
 		return
 	}
+	if req.Source <= 0 {
+		br.Msg = "请输入指标来源"
+		br.ErrMsg = "指标来源为空"
+		return
+	}
+	startDate := req.StartDate
+	if startDate == "" {
+		startDate = utils.BASE_START_DATE
+	}
 
-	cacheKey = utils.CACHE_EDB_DATA_REFRESH + strconv.Itoa(source) + "_" + req.EdbCode
+	// 加锁
+	cacheKey = utils.CACHE_EDB_DATA_REFRESH + strconv.Itoa(req.Source) + "_" + req.EdbCode
 	if utils.Rc.IsExist(cacheKey) {
 		br.Ret = 501
 		br.Success = true
 		br.Msg = "系统处理中，请稍后重试"
 		return
 	}
-
 	utils.Rc.SetNX(cacheKey, 1, 1*time.Minute)
 	defer func() {
-		utils.Rc.Delete(cacheKey)
+		_ = utils.Rc.Delete(cacheKey)
 	}()
 
-	//获取指标信息
-	edbInfo, err := models.GetEdbInfoById(req.EdbInfoId)
-	if err != nil {
-		br.Msg = "查询指标信息失败！"
-		br.ErrMsg = "查询指标信息失败 Err:" + err.Error()
-		return
-	}
-
-	//期货数据，就默认到今天，特殊处理下
-	endDate := utils.BASE_END_DATE
-	if strings.Contains(req.EdbCode, ".") {
-		endDate = time.Now().Format(utils.FormatDate)
-	}
-	// 判断是否调用桥接服务
-	windBridgeUrl, _ := services.GetWindBridgeJYUrl()
-	var dataList []*models.EdbDataFromWindSimple
-	if windBridgeUrl != "" && windBridgeUrl == edbInfo.ServerUrl {
-		var errMsg string
-		dataList, err, errMsg = services.RefreshEdbDataFromWindBridge(edbInfo.ServerUrl, req, edbInfo, endDate)
-		if err != nil {
-			br.Msg = "获取指标信息失败！"
-			br.ErrMsg = "获取桥接服务指标信息失败，Err：" + errMsg + err.Error()
-			return
-		}
-	} else {
-		dataItem, errCode, err := services.GetEdbDataFromWindUrl(edbInfo.ServerUrl, req.EdbCode, req.StartDate, endDate)
-		if errCode == 421 { //指标超限
-			br.Ret = 421
-			br.Msg = "指标超限！"
-			return
-		}
-		// 忽略掉指标下架的错误, 并更新指标为停止更新
-		if errCode == services.WindNoAuthCode {
-			edbInfo.NoUpdate = 1
-			edbInfo.ModifyTime = time.Now().Local()
-			if e := edbInfo.Update([]string{"NoUpdate", "ModifyTime"}); e != nil {
-				br.Msg = "刷新失败"
-				br.ErrMsg = "更新wind指标停更失败, Err: " + e.Error()
-				return
-			}
-			br.Ret = 200
-			br.Msg = "操作成功"
-			return
-		}
-		if err != nil {
-			br.Msg = "获取指标信息失败！"
-			br.ErrMsg = "获取指标信息失败 GetEdbDataFromWind，Err:" + err.Error()
-			return
-		}
-		dataList = models.DealWindData(dataItem)
+	// 校验指标和数据源
+	edbInfo, e := models.GetEdbInfoById(req.EdbInfoId)
+	if e != nil {
+		br.Msg = "指标信息有误"
+		br.ErrMsg = "查询指标信息失败, Err: " + e.Error()
+		return
 	}
-	err = models.RefreshEdbDataFromWind(req.EdbInfoId, req.EdbCode, req.StartDate, dataList)
-	if err != nil && err.Error() != utils.ErrNoRow() {
-		br.Msg = "刷新指标信息失败！"
-		br.ErrMsg = "刷新指标信息失败 RefreshEdbDataFromWind，Err:" + err.Error()
+	cond := ` AND edb_source_id = ?`
+	pars := make([]interface{}, 0)
+	pars = append(pars, req.Source)
+	sourceItem, e := models.GetEdbSourceItemByCondition(cond, pars)
+	if e != nil {
+		br.Msg = "数据源有误"
+		br.ErrMsg = "获取数据源失败, Err: " + e.Error()
 		return
 	}
+
+	// 从桥接服务获取指标和数据
+	var params models.BridgeJiaYueIndexDataParams
+	params.IndexCode = req.EdbCode
+	params.SourceExtend = sourceItem.SourceExtend
+	params.StartDate = startDate
+	params.EndDate = utils.BASE_END_DATE
+	indexData, e := services.GetJiaYueIndexDataFromBridge(params)
+	if e != nil {
+		br.Msg = "获取指标失败"
+		br.ErrMsg = "获取指标数据失败, Err: " + e.Error()
+		return
+	}
+	if indexData.Id <= 0 {
+		br.Msg = "指标不存在"
+		br.Success = true
+		return
+	}
+
+	// 刷新指标数据
+	e = models.RefreshEdbDataFromJiaYue(req.Source, req.EdbInfoId, sourceItem.TableName, req.EdbCode, startDate, indexData.IndexData)
+	if e != nil {
+		br.Msg = "刷新失败"
+		br.ErrMsg = "刷新嘉悦指标失败, Err: " + e.Error()
+		return
+	}
+
 	// 更新指标最大最小值
-	err, errMsg := models.UnifiedModifyEdbInfoMaxAndMinInfo(edbInfo)
-	if err != nil {
+	e, errMsg := models.UnifiedModifyEdbInfoMaxAndMinInfo(edbInfo)
+	if e != nil {
 		br.Msg = errMsg
-		br.ErrMsg = err.Error()
+		br.ErrMsg = e.Error()
 		return
 	}
 
@@ -206,5 +214,5 @@ func (this *JiaYueController) Refresh() {
 
 	br.Ret = 200
 	br.Success = true
-	br.Msg = "获取成功"
+	br.Msg = "操作成功"
 }

models/base_from_jiayue.go

@@ -0,0 +1,191 @@
+package models
+
+import (
+	"eta/eta_index_lib/utils"
+	"fmt"
+	"github.com/beego/beego/v2/client/orm"
+	"strconv"
+	"strings"
+	"time"
+)
+
+// BridgeJiaYueIndexDataParams 桥接服务-获取嘉悦指标数据入参
+type BridgeJiaYueIndexDataParams struct {
+	IndexCode    string `json:"index_code" form:"index_code" description:"指标编码"`
+	SourceExtend string `json:"source_extend" form:"source_extend" description:"来源"`
+	StartDate    string `json:"start_date" form:"start_date" description:"开始日期"`
+	EndDate      string `json:"end_date" form:"end_date" description:"结束日期"`
+}
+
+// BridgeJiaYueResultIndexData 桥接服务-获取嘉悦指标数据响应体
+type BridgeJiaYueResultIndexData struct {
+	Code int                      `json:"code" description:"状态码"`
+	Msg  string                   `json:"msg" description:"提示信息"`
+	Data BridgeJiaYueIndexAndData `json:"data" description:"返回数据"`
+}
+
+// BridgeJiaYueIndexAndData 桥接服务-嘉悦指标和数据
+type BridgeJiaYueIndexAndData struct {
+	Id             int                     `description:"指标自增ID" json:"id"`
+	IndexCode      string                  `description:"指标编码" json:"index_code"`
+	IndexName      string                  `description:"指标名称" json:"index_name"`
+	Unit           string                  `description:"单位" json:"unit"`
+	Frequency      string                  `description:"频度" json:"frequency"`
+	LastDate       time.Time               `description:"指标最新时间" json:"last_date"`
+	LastUpdateTime time.Time               `description:"最新更新时间" json:"last_update_time"`
+	Status         int                     `description:"指标状态" json:"status"`
+	IndexData      []BridgeJiaYueIndexData `description:"指标数据" json:"index_data"`
+}
+
+// BridgeJiaYueIndexData 桥接服务-嘉悦指标数据
+type BridgeJiaYueIndexData struct {
+	Val        float64   `json:"val"`
+	DataTime   time.Time `json:"data_time"`
+	UpdateTime time.Time `json:"update_time"`
+}
+
+// AddEdbDataFromJiaYue 新增嘉悦指标数据
+func AddEdbDataFromJiaYue(tableName, edbCode string, dataList []BridgeJiaYueIndexData) (err error) {
+	if tableName == "" {
+		err = fmt.Errorf("数据表名为空")
+		return
+	}
+	if edbCode == "" {
+		err = fmt.Errorf("指标编码为空")
+		return
+	}
+	if len(dataList) == 0 {
+		return
+	}
+
+	sql := fmt.Sprintf(`INSERT INTO %s(edb_info_id, edb_code, data_time, value, create_time, modify_time, data_timestamp) VALUES `, tableName)
+	for _, v := range dataList {
+		val := utils.SubFloatToString(v.Val, 20)
+		stamp := fmt.Sprint(v.DataTime.UnixMilli())
+		sql += GetAddSql("0", edbCode, v.DataTime.Format(utils.FormatDate), stamp, val)
+	}
+	sql = strings.TrimRight(sql, ",")
+
+	// 新增入库
+	o := orm.NewOrm()
+	_, e := o.Raw(sql).Exec()
+	if e != nil {
+		err = fmt.Errorf("insert data err: %s", e.Error())
+	}
+	return
+}
+
+// RefreshEdbDataFromJiaYue 刷新嘉悦指标数据
+func RefreshEdbDataFromJiaYue(source, edbInfoId int, tableName, edbCode, startDate string, dataList []BridgeJiaYueIndexData) (err error) {
+	if source <= 0 {
+		err = fmt.Errorf("指标来源有误")
+		return
+	}
+	if edbInfoId <= 0 {
+		err = fmt.Errorf("指标ID有误")
+		return
+	}
+	if tableName == "" {
+		err = fmt.Errorf("数据表名为空")
+		return
+	}
+
+	// 真实数据的最大日期, 插入规则配置的日期
+	var realDataMaxDate, edbDataInsertConfigDate time.Time
+	var edbDataInsertConfig *EdbDataInsertConfig
+	var isFindConfigDateRealData bool //是否找到配置日期的实际数据的值
+	{
+		edbDataInsertConfig, err = GetEdbDataInsertConfigByEdbId(edbInfoId)
+		if err != nil && err.Error() != utils.ErrNoRow() {
+			return
+		}
+		if edbDataInsertConfig != nil {
+			edbDataInsertConfigDate = edbDataInsertConfig.Date
+		}
+	}
+
+	// 获取已有数据
+	cond := ` AND edb_info_id = ?`
+	pars := make([]interface{}, 0)
+	pars = append(pars, edbInfoId)
+	var startDateTime time.Time
+	if startDate != "" {
+		cond += ` AND data_time >= ?`
+		pars = append(pars, startDate)
+		startDateTime, _ = time.ParseInLocation(utils.FormatDate, startDate, time.Local)
+	}
+	existList, e := GetEdbDataByCondition(source, cond, pars)
+	if e != nil {
+		err = fmt.Errorf("获取指标已有数据失败, Err: %s", e.Error())
+		return
+	}
+	existMap := make(map[string]*EdbInfoSearchData)
+	for _, v := range existList {
+		existMap[v.DataTime] = v
+	}
+
+	// 比对数据
+	hasNew := false
+	strEdbInfoId := strconv.Itoa(edbInfoId)
+	addExists := make(map[string]bool)
+	sqlInsert := fmt.Sprintf(`INSERT INTO %s(edb_info_id, edb_code, data_time, value, create_time, modify_time, data_timestamp) VALUES `, tableName)
+	for _, v := range dataList {
+		val := utils.SubFloatToString(v.Val, 30)
+		stamp := fmt.Sprint(v.DataTime.UnixMilli())
+		dataTime := v.DataTime.Format(utils.FormatDate)
+
+		// 如果传入的开始时间是空的, 且当前数据日期早于传入的开始日期, 那么需要判断下当前日期的数据是否存在
+		if !startDateTime.IsZero() && v.DataTime.Before(startDateTime) {
+			t, e := GetEdbDataByDate(source, edbCode, dataTime)
+			if e == nil && t != nil {
+				existMap[t.DataTime] = t
+			}
+		}
+
+		// 下面代码主要目的是处理掉手动插入的数据判断
+		{
+			if realDataMaxDate.IsZero() || v.DataTime.After(realDataMaxDate) {
+				realDataMaxDate = v.DataTime
+			}
+			if edbDataInsertConfigDate.IsZero() || v.DataTime.Equal(edbDataInsertConfigDate) {
+				isFindConfigDateRealData = true
+			}
+		}
+
+		// 新增数据
+		exist, ok := existMap[dataTime]
+		if !ok {
+			// 不在历史数据中且与新增中的数据不重复
+			if _, o := addExists[dataTime]; !o {
+				hasNew = true
+				sqlInsert += GetAddSql(strEdbInfoId, edbCode, dataTime, stamp, val)
+				addExists[dataTime] = true
+			}
+			continue
+		}
+
+		// 更新数据
+		if exist != nil && utils.SubFloatToString(exist.Value, 30) != val {
+			if e = ModifyEdbDataById(source, exist.EdbDataId, val); e != nil {
+				err = fmt.Errorf("modify edb data err: %s", e.Error())
+				return
+			}
+		}
+	}
+
+	// 处理手工数据补充的配置
+	HandleConfigInsertEdbData(realDataMaxDate, edbDataInsertConfig, edbInfoId, source, existMap, isFindConfigDateRealData)
+
+	// 执行新增
+	if !hasNew {
+		return
+	}
+	o := orm.NewOrm()
+	sqlInsert = strings.TrimRight(sqlInsert, ",")
+	_, e = o.Raw(sqlInsert).Exec()
+	if e != nil {
+		err = fmt.Errorf("insert edb data err: %s", e.Error())
+		return
+	}
+	return
+}

models/edb_data_base.go

@@ -44,6 +44,7 @@ func GetAddSql(edbInfoId, edbCode, dataTime, timestampStr string, value string)
 
 type AddEdbInfoReq struct {
 	EdbCode string `description:"指标编码"`
+	Source  int    `description:"指标来源ID"`
 }
 
 // GetEdbInfoCountByCondition 获取指标数量
@@ -120,6 +121,7 @@ type RefreshEdbInfoReq struct {
 	EdbInfoId int    `description:"指标ID"`
 	EdbCode   string `description:"指标编码"`
 	StartDate string `description:"开始日期"`
+	Source    int    `description:"指标来源ID"`
 }
 
 // GetAllEdbDataList 获取所有的指标数据列表

models/edb_data_table.go

@@ -4,7 +4,24 @@ import (
 	"eta/eta_index_lib/utils"
 )
 
-// 指标数据->存储表
+var (
+	EdbDataTableNameMap map[int]string // 指标来源对应数据表名
+)
+
+// 初始化时加载指标来源对应表名, 避免循环中查库, 注意edb_source表修改table_name的话需要重启服务
+func init() {
+	EdbDataTableNameMap = make(map[int]string)
+	sources, e := GetEdbSourceItemsByCondition(``, make([]interface{}, 0), []string{}, "")
+	if e != nil {
+		utils.FileLog.Info("init source table err: %s", e.Error())
+		return
+	}
+	for _, v := range sources {
+		EdbDataTableNameMap[v.EdbSourceId] = v.TableName
+	}
+}
+
+// GetEdbDataTableName 指标数据->存储表
 func GetEdbDataTableName(source int) (tableName string) {
 	switch source {
 	case utils.DATA_SOURCE_THS:
@@ -150,7 +167,7 @@ func GetEdbDataTableName(source int) (tableName string) {
 	case utils.DATA_SOURCE_PREDICT_CALCULATE_ZSXY:
 		tableName = "edb_data_predict_calculate_zsxy" // 预测指数修匀->73
 	default:
-		tableName = ""
+		tableName = EdbDataTableNameMap[source] // 没有对应的从edb_source中取
 	}
 	return
 }

models/edb_source.go

@@ -0,0 +1,44 @@
+package models
+
+import (
+	"fmt"
+	"github.com/beego/beego/v2/client/orm"
+	"strings"
+)
+
+// EdbSource 指标来源表
+type EdbSource struct {
+	EdbSourceId      int    `orm:"column(edb_source_id);pk"`
+	SourceName       string `description:"指标来源名称"`
+	TableName        string `description:"数据表名"`
+	EdbAddMethod     string `description:"指标新增接口"`
+	EdbRefreshMethod string `description:"指标刷新接口"`
+	IsBase           int    `description:"是否为基础指标: 0-否; 1-是"`
+	FromBridge       int    `description:"是否来源于桥接服务: 0-否; 1-是"`
+	BridgeFlag       string `description:"桥接服务对象标识"`
+	SourceExtend     string `description:"扩展字段做查询用"`
+}
+
+// GetEdbSourceItemsByCondition 获取指标来源列表
+func GetEdbSourceItemsByCondition(condition string, pars []interface{}, fieldArr []string, orderRule string) (items []*EdbSource, err error) {
+	o := orm.NewOrm()
+	fields := strings.Join(fieldArr, ",")
+	if len(fieldArr) == 0 {
+		fields = `*`
+	}
+	order := `ORDER BY edb_source_id ASC`
+	if orderRule != "" {
+		order = ` ORDER BY ` + orderRule
+	}
+	sql := fmt.Sprintf(`SELECT %s FROM edb_source WHERE 1=1 %s %s`, fields, condition, order)
+	_, err = o.Raw(sql, pars).QueryRows(&items)
+	return
+}
+
+// GetEdbSourceItemByCondition 获取指标来源
+func GetEdbSourceItemByCondition(condition string, pars []interface{}) (item *EdbSource, err error) {
+	o := orm.NewOrm()
+	sql := fmt.Sprintf(`SELECT * FROM edb_source WHERE 1=1 %s`, condition)
+	err = o.Raw(sql, pars).QueryRow(&item)
+	return
+}

routers/commentsRouter.go

@@ -358,6 +358,24 @@ func init() {
             Filters: nil,
             Params: nil})
 
+    beego.GlobalControllerRouter["eta/eta_index_lib/controllers:JiaYueController"] = append(beego.GlobalControllerRouter["eta/eta_index_lib/controllers:JiaYueController"],
+        beego.ControllerComments{
+            Method: "Add",
+            Router: `/add`,
+            AllowHTTPMethods: []string{"post"},
+            MethodParams: param.Make(),
+            Filters: nil,
+            Params: nil})
+
+    beego.GlobalControllerRouter["eta/eta_index_lib/controllers:JiaYueController"] = append(beego.GlobalControllerRouter["eta/eta_index_lib/controllers:JiaYueController"],
+        beego.ControllerComments{
+            Method: "Refresh",
+            Router: `/refresh`,
+            AllowHTTPMethods: []string{"post"},
+            MethodParams: param.Make(),
+            Filters: nil,
+            Params: nil})
+
     beego.GlobalControllerRouter["eta/eta_index_lib/controllers:LtController"] = append(beego.GlobalControllerRouter["eta/eta_index_lib/controllers:LtController"],
         beego.ControllerComments{
             Method: "Add",
@@ -835,6 +853,15 @@ func init() {
             Filters: nil,
             Params: nil})
 
+    beego.GlobalControllerRouter["eta/eta_index_lib/controllers:SmmController"] = append(beego.GlobalControllerRouter["eta/eta_index_lib/controllers:SmmController"],
+        beego.ControllerComments{
+            Method: "GetIndexDetailFromDataSourceByCode",
+            Router: `/index_detail/from_data_source`,
+            AllowHTTPMethods: []string{"post"},
+            MethodParams: param.Make(),
+            Filters: nil,
+            Params: nil})
+
     beego.GlobalControllerRouter["eta/eta_index_lib/controllers:SmmController"] = append(beego.GlobalControllerRouter["eta/eta_index_lib/controllers:SmmController"],
         beego.ControllerComments{
             Method: "Refresh",

routers/router.go

@@ -191,6 +191,11 @@ func init() {
 				&controllers.ExchangeCrawler{},
 			),
 		),
+		beego.NSNamespace("/jiayue_index",
+			beego.NSInclude(
+				&controllers.JiaYueController{},
+			),
+		),
 	)
 	beego.AddNamespace(ns)
 }

services/base_from_jiayue.go

@@ -0,0 +1,81 @@
+package services
+
+import (
+	"encoding/json"
+	"eta/eta_index_lib/models"
+	"eta/eta_index_lib/services/alarm_msg"
+	"eta/eta_index_lib/utils"
+	"fmt"
+	"io/ioutil"
+	"net/http"
+	"strings"
+)
+
+var (
+	BridgeApiJiaYueIndexDataUrl = "/api/index_data/jiayue/index_data" // 获取指标数据API
+)
+
+// GetJiaYueIndexDataFromBridge 从桥接服务获取指标数据
+func GetJiaYueIndexDataFromBridge(param models.BridgeJiaYueIndexDataParams) (indexData models.BridgeJiaYueIndexAndData, err error) {
+	defer func() {
+		if err != nil {
+			b, _ := json.Marshal(param)
+			tips := fmt.Sprintf("桥接服务-获取嘉悦指标数据, err: %s, params: %s", err.Error(), string(b))
+			go alarm_msg.SendAlarmMsg(tips, 3)
+		}
+	}()
+
+	url := fmt.Sprint(utils.EtaBridgeUrl, BridgeApiJiaYueIndexDataUrl)
+	data, e := json.Marshal(param)
+	if e != nil {
+		err = fmt.Errorf("data json marshal err: %s", e.Error())
+		return
+	}
+	body := ioutil.NopCloser(strings.NewReader(string(data)))
+	client := &http.Client{}
+	req, e := http.NewRequest("POST", url, body)
+	if e != nil {
+		err = fmt.Errorf("http create request err: %s", e.Error())
+		return
+	}
+
+	checkToken := utils.MD5(utils.EtaBridgeAppNameEn + utils.EtaBridgeMd5Key)
+	contentType := "application/json;charset=utf-8"
+	req.Header.Set("Content-Type", contentType)
+	req.Header.Set("Authorization", checkToken)
+	resp, e := client.Do(req)
+	if e != nil {
+		err = fmt.Errorf("http client do err: %s", e.Error())
+		return
+	}
+	defer func() {
+		_ = resp.Body.Close()
+	}()
+	b, e := ioutil.ReadAll(resp.Body)
+	if e != nil {
+		err = fmt.Errorf("resp body read err: %s", e.Error())
+		return
+	}
+	if len(b) == 0 {
+		err = fmt.Errorf("resp body is empty")
+		return
+	}
+	// 生产环境解密
+	if utils.RunMode == "release" {
+		str := string(b)
+		str = strings.Trim(str, `"`)
+		b = utils.DesBase64Decrypt([]byte(str), utils.EtaBridgeDesKey)
+	}
+
+	result := new(models.BridgeJiaYueResultIndexData)
+	if e = json.Unmarshal(b, &result); e != nil {
+		err = fmt.Errorf("result unmarshal err: %s\nresult: %s", e.Error(), string(b))
+		return
+	}
+	if result.Code != 200 {
+		err = fmt.Errorf("result: %s", string(b))
+		return
+	}
+	indexData = result.Data
+	return
+}

services/base_from_wind.go

@@ -225,67 +225,3 @@ func DeleteEdbCodeInWindUrl(edbCode string) (err error) {
 	err = utils.Rc.Delete(cacheKey)
 	return
 }
-
-func GetEdbDataFromBridge(windUrl string, req models.AddEdbInfoReq, endDate string) (list []*models.EdbDataFromWindSimple, err error, errMsg string) {
-	dataItem, err := GetEdbDataFromBridgeJy(windUrl, req.EdbCode, utils.BASE_START_DATE, endDate)
-	if err != nil {
-		errMsg = "桥接服务里获取指标信息失败 GetEdbDataFromBridgeJy，Err:" + err.Error()
-		err = errors.New("获取指标信息失败！")
-		return
-	}
-	for _, v := range dataItem.IndexData {
-		tmp := new(models.EdbDataFromWindSimple)
-		dataTime, tErr := time.Parse(utils.FormatDate, v.DataTime)
-		if tErr != nil {
-			err = tErr
-			return
-		}
-		timestamp := dataTime.UnixNano() / 1e6
-		timeStr := fmt.Sprintf("%d", timestamp)
-		saveVal := utils.SubFloatToString(v.Val, 30)
-
-		tmp.TimestampStr = timeStr
-		tmp.DateTime = v.DataTime
-		tmp.Val = saveVal
-		list = append(list, tmp)
-	}
-	return
-}
-
-// RefreshEdbDataFromWindBridge 桥接服务刷新wind指标数据
-func RefreshEdbDataFromWindBridge(windUrl string, req models.RefreshEdbInfoReq, edbInfo *models.EdbInfo, endDate string) (list []*models.EdbDataFromWindSimple, err error, errMsg string) {
-	dataItem, err := GetEdbDataFromBridgeJy(windUrl, edbInfo.EdbCode, req.StartDate, endDate)
-	if err != nil {
-		errMsg = "获取指标信息失败 GetEdbDataFromWind，Err:" + err.Error()
-		err = errors.New("获取指标信息失败！")
-		return
-	}
-
-	// 忽略掉指标下架的错误, 并更新指标为停止更新
-	if dataItem.Status == -1 {
-		edbInfo.NoUpdate = 1
-		edbInfo.ModifyTime = time.Now().Local()
-		if e := edbInfo.Update([]string{"NoUpdate", "ModifyTime"}); e != nil {
-			errMsg = "更新wind指标停更失败, Err: " + e.Error()
-			err = errors.New("刷新失败")
-			return
-		}
-		return
-	}
-	for _, v := range dataItem.IndexData {
-		tmp := new(models.EdbDataFromWindSimple)
-		dataTime, tErr := time.Parse(utils.FormatDate, v.DataTime)
-		if tErr != nil {
-			err = tErr
-		}
-		timestamp := dataTime.UnixNano() / 1e6
-		timeStr := fmt.Sprintf("%d", timestamp)
-		saveVal := utils.SubFloatToString(v.Val, 20)
-
-		tmp.TimestampStr = timeStr
-		tmp.DateTime = v.DataTime
-		tmp.Val = saveVal
-		list = append(list, tmp)
-	}
-	return
-}

services/base_from_wind_bridge.go

@@ -1,181 +0,0 @@
-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"
-)
-
-// GetEdbDataFromBridgeJy 从桥接服务里的嘉悦物产数据中心获取数据
-func GetEdbDataFromBridgeJy(windUrl, edbCode, startDate, endDate string) (index *models.EdbDataFromWindBridgeIndex, err error) {
-	if windUrl == `` {
-		windUrl, err = GetWindBridgeUrl()
-		if err != nil {
-			return
-		}
-	}
-	source := "wind"
-	index, err = getEdbDataFromBridgeJy(windUrl, source, edbCode, startDate, endDate)
-	return
-}
-
-// GetWindBridgeUrl 获取桥接服务地址
-func GetWindBridgeUrl() (windUrl string, err error) {
-	defer func() {
-		if err == nil && windUrl == "" {
-			err = errors.New("wind桥接服务器地址为空")
-		}
-	}()
-	// 先配置中获取，如果有指定配置的话，那么取配置中的数据
-	tmpConfig, err := models.GetConfigDetailByCode("wind_from_bridge")
-	if err != nil && err.Error() != utils.ErrNoRow() {
-		err = errors.New("获取配置的wind地址失败，err:" + err.Error())
-		return
-	}
-	err = nil
-	if tmpConfig != nil {
-		windFrom := tmpConfig.ConfigValue
-		if windFrom != "" {
-			tmpConfig, err = models.GetConfigDetailByCode(windFrom)
-			if err != nil && err.Error() != utils.ErrNoRow() {
-				err = errors.New("获取配置的wind地址失败，err:" + err.Error())
-				return
-			}
-			if err == nil {
-				windUrl = tmpConfig.ConfigValue
-			}
-			err = nil
-		}
-		return
-	}
-	return
-}
-
-// GetWindBridgeJYUrl 获取嘉悦桥接服务地址
-func GetWindBridgeJYUrl() (windUrl string, err error) {
-	defer func() {
-		if err == nil && windUrl == "" {
-			err = errors.New("wind桥接服务器地址为空")
-		}
-	}()
-	// 先配置中获取，如果有指定配置的话，那么取配置中的数据
-	tmpConfig, err := models.GetConfigDetailByCode("wind_from_bridge_jy")
-	if err != nil && err.Error() != utils.ErrNoRow() {
-		err = errors.New("获取配置的wind地址失败，err:" + err.Error())
-		return
-	}
-	err = nil
-	if tmpConfig != nil {
-		windUrl = tmpConfig.ConfigValue
-	}
-	return
-}
-
-// GetManualBridgeJYUrl 获取嘉悦桥接服务地址
-func GetManualBridgeJYUrl() (windUrl string, err error) {
-	defer func() {
-		if err == nil && windUrl == "" {
-			err = errors.New("wind桥接服务器地址为空")
-		}
-	}()
-	// 先配置中获取，如果有指定配置的话，那么取配置中的数据
-	tmpConfig, err := models.GetConfigDetailByCode("manual_from_bridge_jy")
-	if err != nil && err.Error() != utils.ErrNoRow() {
-		err = errors.New("获取配置的wind地址失败，err:" + err.Error())
-		return
-	}
-	err = nil
-	if tmpConfig != nil {
-		windUrl = tmpConfig.ConfigValue
-	}
-	return
-}
-
-// GetManualBridgeUrl 获取桥接服务地址
-func GetManualBridgeUrl() (manualUrl string, err error) {
-	defer func() {
-		if err == nil && manualUrl == "" {
-			err = errors.New("wind桥接服务器地址为空")
-		}
-	}()
-	// 先配置中获取，如果有指定配置的话，那么取配置中的数据
-	tmpConfig, err := models.GetConfigDetailByCode("manual_from_bridge")
-	if err != nil && err.Error() != utils.ErrNoRow() {
-		err = errors.New("获取配置的wind地址失败，err:" + err.Error())
-		return
-	}
-	err = nil
-	if tmpConfig != nil {
-		windFrom := tmpConfig.ConfigValue
-		tmpConfig, err = models.GetConfigDetailByCode(windFrom)
-		if err != nil && err.Error() != utils.ErrNoRow() {
-			err = errors.New("获取配置的wind地址失败，err:" + err.Error())
-			return
-		}
-		manualUrl = tmpConfig.ConfigValue
-		return
-	}
-	return
-}
-
-// GetManualEdbDataFromBridgeJy 从桥接服务里的嘉悦物产数据中心获取数据
-func GetManualEdbDataFromBridgeJy(windUrl, edbCode, startDate, endDate string) (list []*models.ManualEdbdata, err error) {
-	if windUrl == `` {
-		windUrl, err = GetManualBridgeUrl()
-		if err != nil {
-			return
-		}
-	}
-	source := "manual"
-	index, err := getEdbDataFromBridgeJy(windUrl, source, edbCode, startDate, endDate)
-	if err != nil {
-		return
-	}
-	for _, v := range index.IndexData {
-		tmp := new(models.ManualEdbdata)
-		t, _ := time.ParseInLocation(utils.FormatDateTime, v.UpdateTime, time.Local)
-		tmp = &models.ManualEdbdata{
-			TradeCode:  index.IndexCode,
-			Dt:         v.DataTime,
-			Close:      fmt.Sprintf("%f", v.Val),
-			ModifyTime: t,
-		}
-		list = append(list, tmp)
-	}
-	return
-}
-
-// getEdbDataFromBridgeJy 从桥接服务里的嘉悦物产数据中心获取数据
-func getEdbDataFromBridgeJy(url, source, edbCode, startDate, endDate string) (index *models.EdbDataFromWindBridgeIndex, err error) {
-	requestWindUrl := url + `?index_code=%s&start_date=%s&end_date=%s&source=%s`
-	requestWindUrl = fmt.Sprintf(requestWindUrl, edbCode, startDate, endDate, source)
-	utils.FileLog.Info(fmt.Sprintf("windUrl:%s", requestWindUrl))
-	body, err := http.Get(requestWindUrl)
-	fmt.Println("GetEdbDataFromBridgeJy body:")
-	fmt.Println(string(body))
-	utils.FileLog.Info(fmt.Sprint("GetEdbDataFromBridgeJy result:", string(body)))
-
-	if err != nil {
-		return
-	}
-	ret := new(models.EdbDataFromWindBridge)
-	err = json.Unmarshal(body, &ret)
-	if err != nil {
-		err = errors.New("返回值格式错误，json解析失败，Err: " + err.Error())
-		return
-	}
-	//异常的话，需要邮件通知
-	if ret.Code != 200 {
-		errMsg := fmt.Sprintf("	桥接服务 指标数据接口报错，地址：%s, Code: %d, Err: %s", requestWindUrl, ret.Code, ret.Msg)
-		err = errors.New(errMsg)
-		go alarm_msg.SendAlarmMsg(errMsg, 3)
-		return
-	}
-	index = &ret.Data
-	return
-}

utils/config.go

@@ -50,6 +50,11 @@ var (
 
 	// AlarmMsgUrl 报警服务地址
 	AlarmMsgUrl string
+
+	EtaBridgeUrl       string // 桥接服务地址
+	EtaBridgeAppNameEn string // 桥接服务英文名称-鉴权用
+	EtaBridgeMd5Key    string // 桥接服务Md5密钥-鉴权用
+	EtaBridgeDesKey    string // 桥接服务Des密钥-解密数据用
 )
 
 // 第三方服务地址配置
@@ -169,6 +174,11 @@ func init() {
 
 		// AlarmMsgUrl 报警服务地址
 		AlarmMsgUrl = config["alarm_msg_url"]
+
+		EtaBridgeUrl = config["eta_bridge_url"]               // 桥接服务地址
+		EtaBridgeAppNameEn = config["eta_bridge_app_name_en"] // 桥接服务英文名称-鉴权用
+		EtaBridgeMd5Key = config["eta_bridge_md5_key"]        // 桥接服务Md5密钥-鉴权用
+		EtaBridgeDesKey = config["eta_bridge_des_key"]        // 桥接服务Des密钥-解密数据用
 	}
 
 	// 第三方服务地址配置

utils/des3.go

@@ -0,0 +1,187 @@
+// 加密工具类，用了3des和base64
+package utils
+
+import (
+	"bytes"
+	"crypto/cipher"
+	"crypto/des"
+	"encoding/base64"
+	"encoding/hex"
+	"errors"
+	"strings"
+)
+
+// des3 + base64 encrypt
+func DesBase64Encrypt(origData []byte, desKey string) []byte {
+	result, err := TripleDesEncrypt(origData, []byte(desKey))
+	if err != nil {
+		panic(any(err))
+	}
+	return []byte(base64.StdEncoding.EncodeToString(result))
+}
+
+func DesBase64Decrypt(crypted []byte, desKey string) []byte {
+	result, _ := base64.StdEncoding.DecodeString(string(crypted))
+	remain := len(result) % 8
+	if remain > 0 {
+		mod := 8 - remain
+		for i := 0; i < mod; i++ {
+			result = append(result, 0)
+		}
+	}
+	origData, err := TripleDesDecrypt(result, []byte(desKey))
+	if err != nil {
+		panic(any(err))
+	}
+	return origData
+}
+
+// 3DES加密
+func TripleDesEncrypt(origData, key []byte) ([]byte, error) {
+	block, err := des.NewTripleDESCipher(key)
+	if err != nil {
+		return nil, err
+	}
+	origData = PKCS5Padding(origData, block.BlockSize())
+	// origData = ZeroPadding(origData, block.BlockSize())
+	blockMode := cipher.NewCBCEncrypter(block, key[:8])
+	crypted := make([]byte, len(origData))
+	blockMode.CryptBlocks(crypted, origData)
+	return crypted, nil
+}
+
+// 3DES解密
+func TripleDesDecrypt(crypted, key []byte) ([]byte, error) {
+	block, err := des.NewTripleDESCipher(key)
+	if err != nil {
+		return nil, err
+	}
+	blockMode := cipher.NewCBCDecrypter(block, key[:8])
+	origData := make([]byte, len(crypted))
+	// origData := crypted
+	blockMode.CryptBlocks(origData, crypted)
+	origData = PKCS5UnPadding(origData)
+	// origData = ZeroUnPadding(origData)
+	return origData, nil
+}
+
+func ZeroPadding(ciphertext []byte, blockSize int) []byte {
+	padding := blockSize - len(ciphertext)%blockSize
+	padtext := bytes.Repeat([]byte{0}, padding)
+	return append(ciphertext, padtext...)
+}
+
+func ZeroUnPadding(origData []byte) []byte {
+	length := len(origData)
+	unpadding := int(origData[length-1])
+	return origData[:(length - unpadding)]
+}
+
+func PKCS5Padding(ciphertext []byte, blockSize int) []byte {
+	padding := blockSize - len(ciphertext)%blockSize
+	padtext := bytes.Repeat([]byte{byte(padding)}, padding)
+	return append(ciphertext, padtext...)
+}
+
+func PKCS5UnPadding(origData []byte) []byte {
+	length := len(origData)
+	// 去掉最后一个字节 unpadding 次
+	unpadding := int(origData[length-1])
+	return origData[:(length - unpadding)]
+}
+
+// DES加密
+func DesEncrypt(content string, key string) string {
+	contents := []byte(content)
+	keys := []byte(key)
+	block, err := des.NewCipher(keys)
+	if err != nil {
+		return ""
+	}
+	contents = PKCS5Padding(contents, block.BlockSize())
+	blockMode := cipher.NewCBCEncrypter(block, keys)
+	crypted := make([]byte, len(contents))
+	blockMode.CryptBlocks(crypted, contents)
+	return byteToHexString(crypted)
+}
+
+func byteToHexString(bytes []byte) string {
+	str := ""
+	for i := 0; i < len(bytes); i++ {
+		sTemp := hex.EncodeToString([]byte{bytes[i]})
+		if len(sTemp) < 2 {
+			str += string(0)
+		}
+		str += strings.ToUpper(sTemp)
+	}
+	return str
+}
+
+// DES解密
+func DesDecrypt(content string, key string) string {
+	contentBytes, err := hex.DecodeString(content)
+	if err != nil {
+		return "字符串转换16进制数组失败" + err.Error()
+	}
+	keys := []byte(key)
+	block, err := des.NewCipher(keys)
+	if err != nil {
+		return "解密失败" + err.Error()
+	}
+	blockMode := cipher.NewCBCDecrypter(block, keys)
+	origData := contentBytes
+	blockMode.CryptBlocks(origData, contentBytes)
+	origData = ZeroUnPadding(origData)
+	return string(origData)
+}
+
+// DES ECB PKCK5Padding
+func EntryptDesECB(data, key []byte) (string, error) {
+	if len(key) > 8 {
+		key = key[:8]
+	}
+	block, err := des.NewCipher(key)
+	if err != nil {
+		return "", errors.New("des.NewCipher " + err.Error())
+	}
+	bs := block.BlockSize()
+	data = PKCS5Padding(data, bs)
+	if len(data)%bs != 0 {
+		return "", errors.New("EntryptDesECB Need a multiple of the blocksize")
+	}
+	out := make([]byte, len(data))
+	dst := out
+	for len(data) > 0 {
+		block.Encrypt(dst, data[:bs])
+		data = data[bs:]
+		dst = dst[bs:]
+	}
+	return base64.StdEncoding.EncodeToString(out), nil
+}
+
+func DecryptDESECB(d string, key []byte) ([]byte, error) {
+	data, err := base64.StdEncoding.DecodeString(d)
+	if err != nil {
+		return nil, errors.New("decodebase64 " + err.Error())
+	}
+	if len(key) > 8 {
+		key = key[:8]
+	}
+	block, err := des.NewCipher(key)
+	if err != nil {
+		return nil, errors.New("des.NewCipher " + err.Error())
+	}
+	bs := block.BlockSize()
+	if len(data)%bs != 0 {
+		return nil, errors.New("DecryptDES crypto/cipher: input not full blocks")
+	}
+	out := make([]byte, len(data))
+	dst := out
+	for len(data) > 0 {
+		block.Decrypt(dst, data[:bs])
+		data = data[bs:]
+		dst = dst[bs:]
+	}
+	out = PKCS5UnPadding(out)
+	return out, nil
+}