Merge branch 'feature/eta1.2.2_edb_source' of eta_server/eta_index_lib into master

controllers/base_from_calculate.go

@@ -1069,6 +1069,8 @@ func (this *CalculateController) BatchEdit() {
 		}
 	}
 
+	//dataUpdateTime := time.Now().Format(utils.FormatDateTime)
+
 	var sourName, errMsg string
 	var edbInfoId int
 	var baseEdbInfoModel models.BaseEdbInfoInterface
@@ -1299,6 +1301,7 @@ func (this *CalculateController) BatchEdit() {
 	}
 
 	// 更新指标最大最小值
+	//err, errMsg = models.UnifiedModifyEdbInfoMaxAndMinInfoTmp(edbInfo, dataUpdateTime)
 	err, errMsg = models.UnifiedModifyEdbInfoMaxAndMinInfo(edbInfo)
 	if err != nil {
 		br.Msg = errMsg

controllers/base_from_lt.go

@@ -43,7 +43,7 @@ func (this *LtController) Add() {
 	cacheKey = utils.CACHE_EDB_DATA_ADD + strconv.Itoa(source) + "_" + req.EdbCode
 	if !utils.Rc.IsExist(cacheKey) {
 		utils.Rc.SetNX(cacheKey, 1, 1*time.Minute)
-		dataItem, err := services.GetEdbDataFromLt(req.EdbCode, utils.BASE_START_DATE, utils.BASE_END_DATE)
+		dataItem, err := services.GetEdbDataFromLt(req.EdbCode, utils.BASE_START_DATE, utils.BASE_END_DATE, "")
 		if err != nil {
 			br.Msg = "获取指标信息失败！"
 			br.ErrMsg = "获取指标信息失败 GetEdbDataFromLt，Err:" + err.Error()
@@ -112,7 +112,7 @@ func (this *LtController) Refresh() {
 		return
 	}
 
-	dataItem, err := services.GetEdbDataFromLt(req.EdbCode, req.StartDate, utils.BASE_END_DATE)
+	dataItem, err := services.GetEdbDataFromLt(req.EdbCode, req.StartDate, utils.BASE_END_DATE, edbInfo.TerminalCode)
 	if err != nil {
 		br.Msg = "获取指标信息失败！"
 		br.ErrMsg = "获取指标信息失败 GetEdbDataFromLt，Err:" + err.Error()

controllers/base_from_mysteel_chemical.go

@@ -107,7 +107,7 @@ func (this *MySteelChemicalController) Refresh() {
 		br.Msg = "系统处理中，请稍后重试"
 		return
 	}
-
+	dataUpdateTime := time.Now().Format(utils.FormatDateTime)
 	utils.Rc.SetNX(cacheKey, 1, 1*time.Minute)
 	err = models.RefreshEdbDataFromMysteelChemical(req.EdbInfoId, req.EdbCode, req.StartDate)
 	if err != nil && err.Error() != utils.ErrNoRow() {
@@ -115,13 +115,20 @@ func (this *MySteelChemicalController) Refresh() {
 		br.ErrMsg = "刷新指标信息失败 RefreshEdbDataFromMysteel，Err:" + err.Error()
 		return
 	}
+
 	// 更新指标最大最小值
-	err, errMsg := models.UnifiedModifyEdbInfoMaxAndMinInfo(edbInfo)
+	erDataUpdateDate, err, errMsg := models.UnifiedModifyEdbInfoMaxAndMinInfoDataUpdate(edbInfo, dataUpdateTime)
 	if err != nil {
 		br.Msg = errMsg
 		br.ErrMsg = err.Error()
 		return
 	}
+	// 添加指标刷新成功日志
+	if erDataUpdateDate != "" {
+		_ = services.AddEdbInfoUpdateLog(edbInfo.EdbInfoId, 1, "", 1, "", 0)
+	} else {
+		_ = services.AddEdbInfoUpdateLog(edbInfo.EdbInfoId, 1, "", 2, "未刷新到数据", 0)
+	}
 
 	// 更新ES
 	go logic.UpdateEs(edbInfo.EdbInfoId)

controllers/base_from_pb.go

@@ -43,7 +43,7 @@ func (this *PbController) Add() {
 	cacheKey = utils.CACHE_EDB_DATA_ADD + strconv.Itoa(source) + "_" + req.EdbCode
 	if !utils.Rc.IsExist(cacheKey) {
 		utils.Rc.SetNX(cacheKey, 1, 1*time.Minute)
-		dataItem, err := services.GetEdbDataFromPb(req.EdbCode, utils.BASE_START_DATE_UnSpace, utils.BASE_END_DATE_UnSpace)
+		dataItem, err := services.GetEdbDataFromPb(req.EdbCode, utils.BASE_START_DATE_UnSpace, utils.BASE_END_DATE_UnSpace, "")
 		if err != nil {
 			br.Msg = "获取指标信息失败！"
 			br.ErrMsg = "获取指标信息失败 GetEdbDataFromPb，Err:" + err.Error()
@@ -115,7 +115,7 @@ func (this *PbController) Refresh() {
 
 	//开始日期格式化
 	startDateTime, _ := time.ParseInLocation(utils.FormatDate, req.StartDate, time.Local)
-	dataItem, err := services.GetEdbDataFromPb(req.EdbCode, startDateTime.Format(utils.FormatDateUnSpace), utils.BASE_END_DATE_UnSpace)
+	dataItem, err := services.GetEdbDataFromPb(req.EdbCode, startDateTime.Format(utils.FormatDateUnSpace), utils.BASE_END_DATE_UnSpace, edbInfo.TerminalCode)
 	if err != nil {
 		br.Msg = "获取指标信息失败！"
 		br.ErrMsg = "获取指标信息失败 GetEdbDataFromPb，Err:" + err.Error()
@@ -127,7 +127,6 @@ func (this *PbController) Refresh() {
 		br.ErrMsg = "刷新指标信息失败 RefreshEdbDataFromPb，Err:" + err.Error()
 		return
 	}
-	// 更新指标最大最小值
 	err, errMsg := models.UnifiedModifyEdbInfoMaxAndMinInfo(edbInfo)
 	if err != nil {
 		br.Msg = errMsg

controllers/base_from_pb_finance.go

@@ -54,7 +54,7 @@ func (this *PbFinanceController) Add() {
 			br.ErrMsg = "指标编码异常，请传入客户编码"
 			return
 		}
-		dataItem, err := services.GetEdbDataFromPbFinance(edbCodeArr[0], edbCodeArr[1], utils.BASE_START_DATE_UnSpace, utils.BASE_END_DATE_UnSpace)
+		dataItem, err := services.GetEdbDataFromPbFinance(edbCodeArr[0], edbCodeArr[1], utils.BASE_START_DATE_UnSpace, utils.BASE_END_DATE_UnSpace, "")
 		if err != nil {
 			br.Msg = "获取指标信息失败！"
 			br.ErrMsg = "获取指标信息失败 GetEdbDataFromPb，Err:" + err.Error()
@@ -136,7 +136,7 @@ func (this *PbFinanceController) Refresh() {
 		br.ErrMsg = "指标编码异常，请传入客户编码"
 		return
 	}
-	dataItem, err := services.GetEdbDataFromPbFinance(edbCodeArr[0], edbCodeArr[1], startDateTime.Format(utils.FormatDateUnSpace), utils.BASE_END_DATE_UnSpace)
+	dataItem, err := services.GetEdbDataFromPbFinance(edbCodeArr[0], edbCodeArr[1], startDateTime.Format(utils.FormatDateUnSpace), utils.BASE_END_DATE_UnSpace, edbInfo.TerminalCode)
 	if err != nil {
 		br.Msg = "获取指标信息失败！"
 		br.ErrMsg = "获取指标信息失败 GetEdbDataFromPb，Err:" + err.Error()

controllers/base_from_ths.go

@@ -43,7 +43,7 @@ func (this *ThsController) Add() {
 	cacheKey = utils.CACHE_EDB_DATA_ADD + strconv.Itoa(source) + "_" + req.EdbCode
 	if !utils.Rc.IsExist(cacheKey) {
 		utils.Rc.SetNX(cacheKey, 1, 1*time.Minute)
-		dataItem, err := services.GetEdbDataFromThs(req.EdbCode, utils.BASE_START_DATE, utils.BASE_END_DATE)
+		dataItem, err := services.GetEdbDataFromThs(req.EdbCode, utils.BASE_START_DATE, utils.BASE_END_DATE, "")
 		if err != nil {
 			br.Msg = "获取指标信息失败！"
 			br.ErrMsg = "获取指标信息失败 GetEdbDataFromThs，Err:" + err.Error()
@@ -114,7 +114,7 @@ func (this *ThsController) Refresh() {
 	defer func() {
 		utils.Rc.Delete(cacheKey)
 	}()
-	dataItem, err := services.GetEdbDataFromThs(req.EdbCode, req.StartDate, utils.BASE_END_DATE)
+	dataItem, err := services.GetEdbDataFromThs(req.EdbCode, req.StartDate, utils.BASE_END_DATE, edbInfo.TerminalCode)
 	if err != nil {
 		br.Msg = "获取指标信息失败！"
 		br.ErrMsg = "获取指标信息失败 GetEdbDataFromThs，Err:" + err.Error()

controllers/base_from_wind.go

@@ -53,10 +53,11 @@ func (this *WindController) Add() {
 		}
 
 		windUrl := ``
-
+		terminalCode := ""
 		// 试用平台的话，需要额外从弘则这边获取下地址
 		if utils.BusinessCode == utils.HZ_TRIAL_BUSSINESS_CODE {
-			tmpWindUrl, err, errMsg := services.GetHzWindUrl(req.EdbCode, source)
+			// todo 从体验版中获取对应的终端信息
+			tmpWindUrl, tmpTerminalCode, err, errMsg := services.GetHzWindUrl(req.EdbCode, source)
 			if err != nil {
 				br.Msg = "添加失败！"
 				br.ErrMsg = "添加失败，Err：" + err.Error()
@@ -66,20 +67,23 @@ func (this *WindController) Add() {
 				return
 			}
 			windUrl = tmpWindUrl
+			terminalCode = tmpTerminalCode
 		}
 
-		if windUrl == `` {
-			windUrl, err = GetServerUrl()
+		if terminalCode == `` {
+			terminalInfo, err := services.GetTerminal(utils.DATA_SOURCE_WIND, "")
 			if err != nil {
 				br.Msg = "获取可以使用的wind地址失败！"
 				br.ErrMsg = "获取可以使用的wind地址失败，err:" + err.Error()
 				return
 			}
+			windUrl = terminalInfo.ServerUrl
+			terminalCode = terminalInfo.TerminalCode
 		}
 
-		if windUrl == `` {
-			br.Msg = "没有配置wind地址！"
-			br.ErrMsg = "没有配置wind地址"
+		if terminalCode == `` {
+			br.Msg = "没有配置wind终端！"
+			br.ErrMsg = "没有配置wind终端"
 			return
 		}
 
@@ -87,6 +91,10 @@ func (this *WindController) Add() {
 		windUrlCacheKey := utils.CACHE_WIND_URL + ":" + req.EdbCode
 		_ = utils.Rc.SetNX(windUrlCacheKey, windUrl, utils.GetTodayLastSecond())
 
+		// 设置指标与终端关系的缓存
+		terminalCodeCacheKey := utils.CACHE_EDB_TERMINAL_CODE_URL + req.EdbCode
+		_ = utils.Rc.Put(terminalCodeCacheKey, terminalCode, utils.GetTodayLastSecond())
+
 		//windUrl = utils.Hz_Wind_Data_Url_LIST[1].Url
 		dataItem, errCode, err := services.GetEdbDataFromWindUrl(windUrl, req.EdbCode, utils.BASE_START_DATE, endDate)
 		//dataItem, errCode, err := services.GetEdbDataFromWind(req.EdbCode, utils.BASE_START_DATE, endDate)
@@ -178,8 +186,13 @@ func (this *WindController) Refresh() {
 	if strings.Contains(req.EdbCode, ".") {
 		endDate = time.Now().Format(utils.FormatDate)
 	}
-
-	dataItem, errCode, err := services.GetEdbDataFromWindUrl(edbInfo.ServerUrl, req.EdbCode, req.StartDate, endDate)
+	//获取对应的终端信息
+	terminal, err := services.GetTerminal(utils.DATA_SOURCE_WIND, edbInfo.TerminalCode)
+	if err != nil {
+		br.Msg = "终端未配置"
+		return
+	}
+	dataItem, errCode, err := services.GetEdbDataFromWindUrl(terminal.ServerUrl, req.EdbCode, req.StartDate, endDate)
 	if errCode == 421 { //指标超限
 		br.Ret = 421
 		br.Msg = "指标超限！"
@@ -216,7 +229,6 @@ func (this *WindController) Refresh() {
 		br.ErrMsg = err.Error()
 		return
 	}
-
 	// 更新ES
 	go logic.UpdateEs(edbInfo.EdbInfoId)
 

controllers/edb_info_stat.go

@@ -0,0 +1,30 @@
+package controllers
+
+import (
+	"eta/eta_index_lib/models"
+	"eta/eta_index_lib/services"
+)
+
+type EdbInfoStatController struct {
+	BaseAuthController
+}
+
+// SetEdbSourceStat
+// @Title 指标终端更新情况定时汇总
+// @Description 指标终端情况定时汇总
+// @Success 200 {object}
+// @router /source_update [post]
+func (this *EdbInfoStatController) SetEdbSourceStat() {
+	br := new(models.BaseResponse).Init()
+	defer func() {
+		this.Data["json"] = br
+		this.ServeJSON()
+	}()
+	// 钢联终端统计汇总
+	_ = services.SetMysteelChemicalEdbInfoUpdateStat()
+	_ = services.SetEdbSourceStat()
+
+	br.Ret = 200
+	br.Success = true
+	br.Msg = "获取成功"
+}

controllers/future_good/future_good_edb_info.go

@@ -70,7 +70,7 @@ func (this *FutureGoodEdbInfoController) Add() {
 		}
 		endDate := time.Now().Format(utils.FormatDate)
 
-		dataItem, err := services.GetFutureGoodDataFromThs(req.EdbCode, utils.BASE_START_DATE, endDate)
+		dataItem, err := services.GetFutureGoodDataFromThs(req.EdbCode, utils.BASE_START_DATE, endDate, "")
 		if err != nil {
 			br.Msg = "获取指标信息失败！"
 			br.ErrMsg = "获取指标信息失败 GetFutureGoodDataFromThs，Err:" + err.Error()
@@ -169,7 +169,7 @@ func (this *FutureGoodEdbInfoController) Refresh() {
 	if startDate == `` { // 如果传入的日期为空的话,那么就默认兼容一周的数据吧
 		startDate = endDateTime.AddDate(0, 0, -7).Format(utils.FormatDate)
 	}
-	dataItem, err := services.GetFutureGoodDataFromThs(req.FutureGoodEdbCode, req.StartDate, endDate)
+	dataItem, err := services.GetFutureGoodDataFromThs(req.FutureGoodEdbCode, req.StartDate, endDate, futureGoodEdbInfo.TerminalCode)
 	if err != nil {
 		br.Msg = "获取指标信息失败！"
 		br.ErrMsg = "获取指标信息失败 GetFutureGoodDataFromThs，Err:" + err.Error()

logic/base_edb_info.go

@@ -35,14 +35,22 @@ func RefreshBaseEdbInfo(edbInfo *models.EdbInfo, startDate string) (isHandling b
 		err = errors.New("刷新指标信息失败 RefreshEdbDataFromSci，Err:" + err.Error())
 		return
 	}
+
 	// 更新指标最大最小值
-	err, errMsg = models.UnifiedModifyEdbInfoMaxAndMinInfo(edbInfo)
-	if err != nil {
-		return
+	if edbInfo.Source == utils.DATA_SOURCE_MYSTEEL_CHEMICAL {
+		dataUpdateTime := time.Now().Format(utils.FormatDateTime)
+		_, err, errMsg = models.UnifiedModifyEdbInfoMaxAndMinInfoDataUpdate(edbInfo, dataUpdateTime)
+		if err != nil {
+			return
+		}
+	} else {
+		err, errMsg = models.UnifiedModifyEdbInfoMaxAndMinInfo(edbInfo)
+		if err != nil {
+			return
+		}
 	}
 
 	// 更新ES
 	go UpdateEs(edbInfo.EdbInfoId)
-
 	return
 }

models/data_stat/edb_info_delete_log.go

@@ -0,0 +1,60 @@
+package data_stat
+
+import (
+	"github.com/beego/beego/v2/client/orm"
+)
+
+// EdbInfoDeleteLog 指标删除日志表
+type EdbInfoDeleteLog struct {
+	Id                 uint64  `orm:"column(id);pk"`
+	EdbInfoId          int     `description:"指标ID"`
+	EdbInfoType        int     `description:"指标类型，0：普通指标，1：预测指标"`
+	SourceName         string  `description:"来源名称"`
+	Source             int     `description:"来源id"`
+	EdbCode            string  `description:"指标编码"`
+	EdbName            string  `description:"指标名称"`
+	EdbNameEn          string  `description:"英文指标名称"`
+	EdbNameSource      string  `description:"指标名称来源"`
+	Frequency          string  `description:"频率"`
+	Unit               string  `description:"单位"`
+	UnitEn             string  `description:"英文单位"`
+	StartDate          string  `description:"起始日期"`
+	EndDate            string  `description:"终止日期"`
+	SysUserId          int     `description:"创建人ID"`
+	SysUserRealName    string  `description:"创建人姓名"`
+	UniqueCode         string  `description:"指标唯一编码"`
+	EdbCreateTime      string  `description:"指标创建时间"`
+	EdbModifyTime      string  `description:"指标修改时间"`
+	CreateTime         string  `description:"创建时间即删除时间"`
+	MinValue           float64 `description:"指标最小值"`
+	MaxValue           float64 `description:"指标最大值"`
+	CalculateFormula   string  `description:"计算公式"`
+	EdbType            int     `description:"指标类型:1:基础指标,2:计算指标"`
+	LatestDate         string  `description:"数据最新日期"`
+	LatestValue        float64 `description:"数据最新值"`
+	MoveType           int     `description:"移动方式:1:领先(默认),2:滞后"`
+	MoveFrequency      string  `description:"移动频度"`
+	NoUpdate           int8    `description:"是否停止更新，0：继续更新；1：停止更新"`
+	ChartImage         string  `description:"图表图片"`
+	Calendar           string  `description:"公历/农历" orm:"default(公历);"`
+	DataDateType       string  `orm:"column(data_date_type);size(255);null;default(交易日)"`
+	ManualSave         int     `description:"是否有手动保存过上下限: 0-否; 1-是"`
+	TerminalCode       string  `description:"终端编码，用于配置在机器上"`
+	DelSysUserId       int     `description:"删除人ID"`
+	DelSysUserRealName string  `description:"删除人姓名"`
+	DataUpdateTime     string  `description:"最近一次数据发生变化的时间"`
+	ErDataUpdateDate   string  `description:"本次更新，数据发生变化的最早日期"`
+}
+
+type EdbInfoDeleteLogNum struct {
+	Source       int    `description:"来源id"`
+	TerminalCode string `description:"终端编码，用于配置在机器上"`
+	Num          int    `description:"被删除的指标数据总数"`
+}
+
+func GetEdbDeleteLogNumByCreateTime(startDate, endDate string) (item []*EdbInfoDeleteLogNum, err error) {
+	o := orm.NewOrm()
+	sql := ` SELECT source, terminal_code, count(*) as num FROM edb_info_delete_log WHERE create_time >= ? and create_time < ? group by source, terminal_code `
+	_, err = o.Raw(sql, startDate, endDate).QueryRows(&item)
+	return
+}

models/data_stat/edb_info_update_log.go

@@ -0,0 +1,64 @@
+package data_stat
+
+import (
+	"github.com/beego/beego/v2/client/orm"
+	"time"
+)
+
+// EdbInfoUpdateLog 指标更新/刷新日志列表
+type EdbInfoUpdateLog struct {
+	Id                     uint64    `orm:"column(id);pk"`
+	EdbInfoId              int       `description:"指标ID"`
+	SourceName             string    `description:"来源名称"`
+	Source                 int       `description:"来源id"`
+	EdbCode                string    `description:"指标编码"`
+	EdbName                string    `description:"指标名称"`
+	EdbNameSource          string    `description:"指标名称来源"`
+	Frequency              string    `description:"频率"`
+	Unit                   string    `description:"单位"`
+	StartDate              string    `description:"起始日期"`
+	EndDate                string    `description:"终止日期"`
+	SysUserId              int       `description:"创建人ID"`
+	SysUserRealName        string    `description:"创建人姓名"`
+	UniqueCode             string    `description:"指标唯一编码"`
+	EdbCreateTime          time.Time `description:"指标创建时间"`
+	EdbModifyTime          time.Time `description:"指标修改时间"`
+	CreateTime             time.Time `description:"创建时间"`
+	UpdateSysUserId        int       `description:"变更人ID"`
+	UpdateSysUserRealName  string    `description:"变更人姓名"`
+	LatestDate             string    `description:"数据最新日期"`
+	LatestValue            float64   `description:"数据最新值"`
+	TerminalCode           string    `description:"终端编码，用于配置在机器上"`
+	ErDataUpdateDate       string    `description:"本次更新，数据发生变化的最早日期"`
+	UpdateResult           int       `description:"本次刷新操作结果：1成功，2失败"`
+	UpdateFailedReason     string    `description:"刷新失败原因"`
+	DataUpdateResult       int       `description:"数据更新结果：1成功，2失败"`
+	DataUpdateFailedReason string    `description:"数据未正常更新原因"`
+	DataUpdateTime         string    `description:"数据更新时间"`
+	IsSourceRefresh        int       `description:"是否为终端刷新到数据源的刷新操作：0否，1是"`
+}
+
+func AddEdbUpdateLog(item *EdbInfoUpdateLog) (lastId int64, err error) {
+	o := orm.NewOrm()
+	lastId, err = o.Insert(item)
+	return
+}
+
+// GetEdbUpdateLogByCondition 获取指标更新记录
+func GetEdbUpdateLogByCondition(condition string, pars []interface{}) (item []*EdbInfoUpdateLog, err error) {
+	o := orm.NewOrm()
+	sql := ` SELECT * FROM edb_info_update_log WHERE 1=1 `
+	if condition != "" {
+		sql += condition
+	}
+	_, err = o.Raw(sql, pars).QueryRows(&item)
+	return
+}
+
+// GetEdbUpdateSourceLogByCreateDate 获取指标终端更新记录
+func GetEdbUpdateSourceLogByCreateDate(source int, startDate, endDate string) (item []*EdbInfoUpdateLog, err error) {
+	o := orm.NewOrm()
+	sql := ` SELECT * FROM edb_info_update_log WHERE source = ? and create_time >= ? and create_time < ? and is_source_refresh =1 order by data_update_result asc, id desc`
+	_, err = o.Raw(sql, source, startDate, endDate).QueryRows(&item)
+	return
+}

models/data_stat/edb_info_update_stat.go

@@ -0,0 +1,103 @@
+package data_stat
+
+import (
+	"github.com/beego/beego/v2/client/orm"
+	"time"
+)
+
+// EdbInfoUpdateStat 指标更新/刷新 汇总表（数据源明细表）
+type EdbInfoUpdateStat struct {
+	Id                     uint64    `orm:"column(id);pk"`
+	EdbInfoId              int       `description:"指标ID"`
+	SourceName             string    `description:"来源名称"`
+	Source                 int       `description:"来源id"`
+	EdbCode                string    `description:"指标编码"`
+	EdbName                string    `description:"指标名称"`
+	EdbNameSource          string    `description:"指标名称来源"`
+	Frequency              string    `description:"频率"`
+	Unit                   string    `description:"单位"`
+	StartDate              string    `description:"起始日期"`
+	EndDate                string    `description:"终止日期"`
+	SysUserId              int       `description:"创建人ID"`
+	SysUserRealName        string    `description:"创建人姓名"`
+	UniqueCode             string    `description:"指标唯一编码"`
+	EdbCreateTime          time.Time `description:"指标创建时间"`
+	EdbModifyTime          time.Time `description:"指标修改时间"`
+	CreateTime             time.Time `description:"记录创建时间"`
+	LatestDate             string    `description:"数据最新日期"`
+	LatestValue            float64   `description:"数据最新值"`
+	TerminalCode           string    `description:"终端编码，用于配置在机器上"`
+	ErDataUpdateDate       string    `description:"本次更新，数据发生变化的最早日期"`
+	DataUpdateResult       int       `description:"数据更新结果：1成功，2失败"`
+	DataUpdateFailedReason string    `description:"数据未正常更新原因"`
+	DataUpdateTime         string    `description:"数据更新时间"`
+	UpdateResult           int       `description:"本次刷新操作结果：1成功，2失败"`
+	UpdateFailedReason     string    `description:"刷新失败原因"`
+	UpdateTime             string    `description:"刷新时间"`
+	ModifyTime             time.Time `description:"修改时间"`
+	IsAdd                  int       `description:"是否是当日新增的指标：1是，2否"`
+	NeedRefresh            int       `description:"今日是否需要定时刷新，1需要，0不需要"`
+	HasRefresh             int       `description:"今日是否发起刷新服务，1是，0否"`
+}
+
+// Add 新增
+func (r *EdbInfoUpdateStat) Add(list []*EdbInfoUpdateStat) (err error) {
+	o := orm.NewOrm()
+	_, err = o.InsertMulti(len(list), list)
+	return
+}
+
+// UpdateEdbUpdateStatMulti 批量更新
+func UpdateEdbUpdateStatMulti(list []*EdbInfoUpdateStat) (err error) {
+	o := orm.NewOrm()
+	sql := `UPDATE edb_info_update_stat 
+SET 
+edb_name = ?,
+edb_name_source = ?,
+frequency = ?,
+unit = ?,
+start_date = ?,
+end_date = ?,
+edb_modify_time = ?,
+latest_date = ?,
+latest_value = ?,
+terminal_code = ?,
+modify_time = ?,
+er_data_update_date = ?,
+is_add = ?,
+need_refresh = ?,
+has_refresh = ?,
+data_update_result = ?,
+data_update_failed_reason = ?,
+data_update_time = ?, 
+update_result = ?,
+update_failed_reason = ?,
+update_time = ?
+WHERE
+	id = ?`
+	p, err := o.Raw(sql).Prepare()
+	if err != nil {
+		return
+	}
+	defer func() {
+		_ = p.Close() // 别忘记关闭 statement
+	}()
+	for _, v := range list {
+		_, err = p.Exec(v.EdbName, v.EdbNameSource, v.Frequency, v.Unit, v.StartDate, v.EndDate, v.EdbModifyTime, v.LatestDate, v.LatestValue, v.TerminalCode, v.ModifyTime, v.ErDataUpdateDate, v.IsAdd, v.NeedRefresh, v.HasRefresh, v.DataUpdateResult, v.DataUpdateFailedReason, v.DataUpdateTime, v.UpdateResult, v.UpdateFailedReason, v.UpdateTime, v.Id)
+		if err != nil {
+			return
+		}
+	}
+	return
+}
+
+// GetEdbUpdateStatByCondition 获取指标更新汇总记录
+func GetEdbUpdateStatByCondition(condition string, pars []interface{}) (item []*EdbInfoUpdateStat, err error) {
+	o := orm.NewOrm()
+	sql := ` SELECT * FROM edb_info_update_stat WHERE 1=1 `
+	if condition != "" {
+		sql += condition
+	}
+	_, err = o.Raw(sql, pars).QueryRows(&item)
+	return
+}

models/data_stat/edb_source_stat.go

@@ -0,0 +1,75 @@
+package data_stat
+
+import (
+	"github.com/beego/beego/v2/client/orm"
+	"time"
+)
+
+// EdbSourceStat 数据源统计表
+type EdbSourceStat struct {
+	Id                int       `orm:"column(id);pk"`
+	SourceName        string    `description:"来源名称"`
+	Source            int       `description:"来源id"`
+	TerminalCode      string    `description:"终端编码，用于配置在机器上"`
+	EdbNum            int       `description:"指标总数"`
+	EdbNewNum         int       `description:"今日新增指标数"`
+	EdbDelNum         int       `description:"今日删除指标数"`
+	NeedRefreshNum    int       `description:"今日需刷新指标"`
+	HasRefreshNum     int       `description:"今日发起刷新任务指标数"`
+	UpdateSuccessNum  int       `description:"今日已更新成功指标数"`
+	UpdateFailedNum   int       `description:"今日已更新失败指标数"`
+	RefreshSuccessNum int       `description:"今日已刷新成功指标数"`
+	RefreshFailedNum  int       `description:"今日已刷新失败指标数"`
+	CreateTime        time.Time `description:"创建时间"`
+	ModifyTime        time.Time `description:"修改时间"`
+}
+
+func GetEdbSourceStatByCondition(condition string, pars []interface{}) (item []*EdbSourceStat, err error) {
+	o := orm.NewOrm()
+	sql := ` SELECT * FROM edb_source_stat WHERE 1=1 `
+	if condition != "" {
+		sql += condition
+	}
+	_, err = o.Raw(sql, pars).QueryRows(&item)
+	return
+}
+
+// Add 新增
+func (r *EdbSourceStat) Add(list []*EdbSourceStat) (err error) {
+	o := orm.NewOrm()
+	_, err = o.InsertMulti(len(list), list)
+	return
+}
+
+// UpdateEdbSourceStatMulti 批量更新
+func UpdateEdbSourceStatMulti(list []*EdbSourceStat) (err error) {
+	o := orm.NewOrm()
+	sql := `UPDATE edb_source_stat 
+SET 
+edb_num=?,                              
+edb_new_num=?,                          
+edb_del_num=?,                          
+need_refresh_num=?,                      
+has_refresh_num=?,                       
+update_success_num=?,                   
+update_failed_num=?,          
+refresh_success_num=?,                   
+refresh_failed_num=?, 
+modify_time=?                           
+WHERE
+	id = ?`
+	p, err := o.Raw(sql).Prepare()
+	if err != nil {
+		return
+	}
+	defer func() {
+		_ = p.Close() // 别忘记关闭 statement
+	}()
+	for _, v := range list {
+		_, err = p.Exec(v.EdbNum, v.EdbNewNum, v.EdbDelNum, v.NeedRefreshNum, v.HasRefreshNum, v.UpdateSuccessNum, v.UpdateFailedNum, v.RefreshSuccessNum, v.RefreshFailedNum, v.ModifyTime, v.Id)
+		if err != nil {
+			return
+		}
+	}
+	return
+}

models/db.go

@@ -1,6 +1,7 @@
 package models
 
 import (
+	"eta/eta_index_lib/models/data_stat"
 	"eta/eta_index_lib/models/excel"
 	"eta/eta_index_lib/models/future_good"
 	"eta/eta_index_lib/models/supply_analysis"
@@ -66,6 +67,8 @@ func init() {
 	// 基础指标表
 	initBaseIndex()
 
+	// initDataStat 数据源统计管理相关表
+	initDataStat()
 	// Eta表格相关
 	initExcel()
 
@@ -111,6 +114,15 @@ func initBaseIndex() {
 	)
 }
 
+// initDataStat 数据源统计管理相关表
+func initDataStat() {
+	orm.RegisterModel(
+		new(data_stat.EdbInfoUpdateLog),  // 指标更新/刷新日志列表
+		new(data_stat.EdbInfoUpdateStat), // 数据源明细表
+		new(data_stat.EdbSourceStat),     // 数据源统计表
+	)
+}
+
 // initExcel Excel
 func initExcel() {
 	orm.RegisterModel(

models/edb_classify.go

@@ -1,6 +1,7 @@
 package models
 
 import (
+	"errors"
 	"eta/eta_index_lib/utils"
 	"github.com/beego/beego/v2/client/orm"
 	"strconv"
@@ -12,6 +13,7 @@ type EdbClassify struct {
 	ClassifyType    uint8     `description:"分类类型，0：普通指标分类，1：预测指标分类"`
 	ClassifyName    string    `description:"分类名称"`
 	ParentId        int       `description:"父级id"`
+	RootId          int       `description:"顶级id"`
 	HasData         int       `description:"是否含有指标数据"`
 	CreateTime      time.Time `description:"创建时间"`
 	ModifyTime      time.Time `description:"修改时间"`
@@ -28,6 +30,13 @@ func AddEdbClassify(item *EdbClassify) (lastId int64, err error) {
 	return
 }
 
+// Update 更新分类基础信息
+func (edbClassify *EdbClassify) Update(cols []string) (err error) {
+	o := orm.NewOrm()
+	_, err = o.Update(edbClassify, cols...)
+	return
+}
+
 func GetEdbClassifyById(classifyId int) (item *EdbClassify, err error) {
 	o := orm.NewOrm()
 	sql := `SELECT * FROM edb_classify WHERE classify_id=? `
@@ -107,14 +116,34 @@ func SaveEdbClassify(classifyName string, parentId, level int, classifyType uint
 	if isAdd {
 		sysUserIdInt, _ := strconv.Atoi(sysUserId)
 		//获取该层级下最大的排序数
-		maxSort, err := GetEdbClassifyMaxSort(parentId, classifyType)
-
+		maxSort, e := GetEdbAndClassifyMaxSort(parentId, classifyType)
+		if e != nil {
+			errMsg = "查询分类信息出错" + e.Error()
+			err = errors.New(errMsg)
+			return
+		}
+		rootId := 0
+		if parentId > 0 {
+			parentClassify, tErr := GetEdbClassifyById(parentId)
+			if tErr != nil {
+				if tErr.Error() == utils.ErrNoRow() {
+					errMsg = "父级分类不存在"
+					err = errors.New(errMsg)
+					return
+				}
+				errMsg = "获取失败"
+				err = errors.New("获取分类信息失败，Err:" + tErr.Error())
+				return
+			}
+			rootId = parentClassify.RootId
+		}
 		timestamp := strconv.FormatInt(time.Now().UnixNano(), 10)
 		classifyInfo = &EdbClassify{
 			//ClassifyId:      0,
 			ClassifyType:    classifyType,
 			ClassifyName:    classifyName,
 			ParentId:        parentId,
+			RootId:          rootId,
 			HasData:         0,
 			CreateTime:      time.Now(),
 			ModifyTime:      time.Now(),
@@ -122,13 +151,23 @@ func SaveEdbClassify(classifyName string, parentId, level int, classifyType uint
 			SysUserRealName: sysUserName,
 			Level:           level + 1,
 			UniqueCode:      utils.MD5(utils.DATA_PREFIX + "_" + timestamp),
-			Sort:            maxSort,
+			Sort:            maxSort + 1,
 		}
-		classifyId, err := AddEdbClassify(classifyInfo)
+		var classifyId int64
+		classifyId, err = AddEdbClassify(classifyInfo)
 		if err != nil {
 			errMsg = "保存分类失败"
+			return
 		}
 		classifyInfo.ClassifyId = int(classifyId)
+		if parentId == 0 { //一级目录的rootId等于自己本身
+			classifyInfo.RootId = int(classifyId)
+			err = classifyInfo.Update([]string{"RootId"})
+			if err != nil {
+				errMsg = "更新分类失败"
+				return
+			}
+		}
 	}
 	return
 }

models/edb_info.go

@@ -47,6 +47,10 @@ type EdbInfo struct {
 	Calendar         string  `description:"公历/农历" orm:"default(公历);"`
 	DataDateType     string  `orm:"column(data_date_type);size(255);null;default(交易日)"`
 	ManualSave       int     `description:"是否有手动保存过上下限: 0-否; 1-是"`
+	TerminalCode     string  `description:"终端编码，用于配置在机器上"`
+	DataUpdateTime   string  `description:"最近一次数据发生变化的时间"`
+	ErDataUpdateDate string  `description:"本次更新，数据发生变化的最早日期"`
+	SourceIndexName  string  `description:"数据源中的指标名称"`
 }
 
 func (e *EdbInfo) Add() (err error) {
@@ -230,6 +234,21 @@ func GetEdbInfoMaxAndMinInfo(source int, edbCode string) (item *EdbInfoMaxAndMin
 	return
 }
 
+type EdbIndoDataUpdateTime struct {
+	MinDate string `description:"本次更新，数据发生变化的最早日期"`
+}
+
+// GetEdbDataUpdateTimeByModify 获取指标的刷新时间获取最早的一条记录
+func GetEdbDataUpdateTimeByModify(source int, edbCode string, dataUpdateTime string) (item *EdbIndoDataUpdateTime, err error) {
+	o := orm.NewOrm()
+	sql := ``
+	tableName := GetEdbDataTableName(source)
+	sql = ` SELECT MIN(data_time) AS min_date FROM %s WHERE edb_code=? and modify_time>=?`
+	sql = fmt.Sprintf(sql, tableName)
+	err = o.Raw(sql, edbCode, dataUpdateTime).QueryRow(&item)
+	return
+}
+
 // ModifyEdbInfoMaxAndMinInfo 修改指标的最新数据信息
 func ModifyEdbInfoMaxAndMinInfo(edbInfoId int, item *EdbInfoMaxAndMinInfo) (err error) {
 	o := orm.NewOrm()
@@ -246,6 +265,14 @@ func ModifyEdbTimeAndLastInfo(edbInfoId int, item *EdbInfoMaxAndMinInfo) (err er
 	return
 }
 
+// ModifyEdbDataUpdateTime 修改指标刷新，本次数据刷新的最早日期
+func ModifyEdbDataUpdateTime(edbInfoId int, dataUpdateTime, erDataUpdateDate string) (err error) {
+	o := orm.NewOrm()
+	sql := `UPDATE edb_info SET data_update_time = ?, er_data_update_date = ?, modify_time = NOW() WHERE edb_info_id = ?`
+	_, err = o.Raw(sql, dataUpdateTime, erDataUpdateDate, edbInfoId).Exec()
+	return
+}
+
 // GetEdbDataCount 获取edb指标数据的数量； order:1升序，其余值为降序
 func GetEdbDataCount(condition string, pars []interface{}, source int) (count int, err error) {
 	o := orm.NewOrm()
@@ -417,6 +444,32 @@ func UnifiedModifyEdbInfoMaxAndMinInfo(edbInfo *EdbInfo) (err error, errMsg stri
 	return
 }
 
+// UnifiedModifyEdbInfoMaxAndMinInfoDataUpdate 统一修改指标的最大最小值以及数据更新时间点
+func UnifiedModifyEdbInfoMaxAndMinInfoDataUpdate(edbInfo *EdbInfo, dataUpdateTime string) (erDataUpdateDate string, err error, errMsg string) {
+	err, errMsg = UnifiedModifyEdbInfoMaxAndMinInfo(edbInfo)
+	if err != nil {
+		return
+	}
+	// 获取本次刷新，指标数据更新的最早日期
+	erDataUpdateTime, tErr := GetEdbDataUpdateTimeByModify(edbInfo.Source, edbInfo.EdbCode, dataUpdateTime)
+	if tErr != nil {
+		if tErr.Error() != utils.ErrNoRow() { //本次刷新指标数据无变化
+			errMsg = "刷新指标失败！"
+			err = errors.New("查询指标刷新，数据更新的最早日期失败，err:" + tErr.Error())
+			return
+		}
+	} else if erDataUpdateTime.MinDate != "" {
+		err = ModifyEdbDataUpdateTime(edbInfo.EdbInfoId, dataUpdateTime, erDataUpdateTime.MinDate)
+		if err != nil {
+			errMsg = "刷新指标失败！"
+			err = errors.New("修改指标数据更新的最早日期失败，err:" + tErr.Error())
+			return
+		}
+		erDataUpdateDate = erDataUpdateTime.MinDate
+	}
+	return
+}
+
 // UnifiedModifyPredictEdbInfoMaxAndMinInfo 统一修改预测运算指标的最大最小值
 func UnifiedModifyPredictEdbInfoMaxAndMinInfo(edbInfo *EdbInfo, latestDateStr string, latestValue float64) (err error, errMsg string) {
 	// 修改最大最小值
@@ -1001,14 +1054,15 @@ type EdbInfoDetailResp struct {
 }
 
 type AddEdbInfoParams struct {
-	Source     int    `description:"来源id"`
-	EdbCode    string `description:"指标编码"`
-	EdbName    string `description:"指标名称"`
-	Frequency  string `description:"频率"`
-	Unit       string `description:"单位"`
-	ClassifyId int    `description:"分类id"`
-	AdminId    int    `description:"创建人ID"`
-	AdminName  string `description:"创建人姓名"`
+	Source       int    `description:"来源id"`
+	EdbCode      string `description:"指标编码"`
+	EdbName      string `description:"指标名称"`
+	Frequency    string `description:"频率"`
+	Unit         string `description:"单位"`
+	ClassifyId   int    `description:"分类id"`
+	TerminalCode string `description:"终端编码"`
+	AdminId      int    `description:"创建人ID"`
+	AdminName    string `description:"创建人姓名"`
 }
 
 // EdbInfoAdd 添加指标到指标库
@@ -1089,6 +1143,8 @@ func EdbInfoAdd(req *AddEdbInfoParams, serverUrl string, sysUserId int, sysUserR
 		edbType = 2 //计算指标
 	}
 
+	var sourceIndexName string
+
 	// 钢联化工需要校验下信息是否完整
 	if source == utils.DATA_SOURCE_MYSTEEL_CHEMICAL {
 		indexObj := new(BaseFromMysteelChemicalIndex)
@@ -1102,6 +1158,12 @@ func EdbInfoAdd(req *AddEdbInfoParams, serverUrl string, sysUserId int, sysUserR
 			return
 		}
 	}
+	//获取该层级下最大的排序数
+	maxSort, err := GetEdbAndClassifyMaxSort(req.ClassifyId, 0)
+	if err != nil {
+		err = errors.New("查询排序信息失败，Err:" + err.Error())
+		return
+	}
 
 	edbInfo.EdbCode = req.EdbCode
 	edbInfo.EdbName = req.EdbName
@@ -1115,6 +1177,8 @@ func EdbInfoAdd(req *AddEdbInfoParams, serverUrl string, sysUserId int, sysUserR
 	edbInfo.CreateTime = time.Now()
 	edbInfo.ModifyTime = time.Now()
 	edbInfo.ServerUrl = serverUrl
+	edbInfo.TerminalCode = req.TerminalCode
+	edbInfo.Sort = maxSort + 1
 	timestamp := strconv.FormatInt(time.Now().UnixNano(), 10)
 	edbInfo.UniqueCode = utils.MD5(utils.DATA_PREFIX + "_" + req.EdbCode + timestamp)
 	itemVal, err := GetEdbInfoMaxAndMinInfo(source, req.EdbCode)
@@ -1123,6 +1187,7 @@ func EdbInfoAdd(req *AddEdbInfoParams, serverUrl string, sysUserId int, sysUserR
 		edbInfo.MinValue = itemVal.MinValue
 	}
 	edbInfo.EdbType = edbType
+	edbInfo.SourceIndexName = sourceIndexName
 	edbInfoId, err := AddEdbInfo(edbInfo)
 	if err != nil {
 		err = errors.New("保存失败，Err:" + err.Error())
@@ -1152,3 +1217,28 @@ type ResetEdbClassifyReq struct {
 	ClassifySecond string `description:"二级分类名称"`
 	ClassifyThird  string `description:"三级分类名称"`
 }
+
+// GetEdbInfoMaxSortByClassifyId 获取分类下指标的最大的排序数
+func GetEdbInfoMaxSortByClassifyId(classifyId int) (sort int, err error) {
+	o := orm.NewOrm()
+	sql := `SELECT Max(sort) AS sort FROM edb_info WHERE classify_id=? `
+	err = o.Raw(sql, classifyId).QueryRow(&sort)
+	return
+}
+
+func GetEdbAndClassifyMaxSort(parentId int, classifyType uint8) (maxSort int, err error) {
+	//获取该层级下最大的排序数
+	classifyMaxSort, err := GetEdbClassifyMaxSort(parentId, classifyType)
+	if err != nil {
+		return
+	}
+	maxSort = classifyMaxSort
+	edbMaxSort, err := GetEdbInfoMaxSortByClassifyId(parentId)
+	if err != nil {
+		return
+	}
+	if maxSort < edbMaxSort {
+		maxSort = edbMaxSort
+	}
+	return
+}

models/edb_terminal.go

@@ -0,0 +1,71 @@
+package models
+
+import (
+	"fmt"
+	"github.com/beego/beego/v2/client/orm"
+	"time"
+)
+
+// EdbTerminal 指标终端
+type EdbTerminal struct {
+	TerminalId   int       `orm:"column(terminal_id);pk"`
+	Source       int       `orm:"column(source)" description:"指标来源类型"`
+	Name         string    `description:"终端别名"`
+	TerminalCode string    `description:"终端编码，用于配置在机器上"`
+	ServerUrl    string    `description:"终端地址"`
+	Num          int       `description:"终端最大指标数"`
+	Status       int       `description:"状态，1启用，2禁用"`
+	Value        string    `description:"终端相关的凭证"`
+	ModifyTime   time.Time `description:"修改时间"`
+	CreateTime   time.Time `description:"创建时间"`
+}
+
+// GetEdbTerminalListBySource 根据指标来源类型获取所有的终端列表
+func GetEdbTerminalListBySource(source int) (items []*EdbTerminal, err error) {
+	o := orm.NewOrm()
+	sql := ` SELECT *  FROM edb_terminal WHERE source = ? and status=1 ORDER BY terminal_id ASC `
+	_, err = o.Raw(sql, source).QueryRows(&items)
+	return
+}
+
+// GetEdbTerminalFirstBySource 根据指标来源类型获取配置的首个终端信息
+func GetEdbTerminalFirstBySource(source int) (item *EdbTerminal, err error) {
+	o := orm.NewOrm()
+	sql := ` SELECT *  FROM edb_terminal WHERE source = ? and status=1 ORDER BY terminal_id ASC Limit 1 `
+	err = o.Raw(sql, source).QueryRow(&item)
+	return
+}
+
+type TerminalCodeCountGroup struct {
+	TerminalCode string
+	Total        int
+}
+
+// GetEdbTerminalByCode 根据终端编码获取终端信息
+func GetEdbTerminalByCode(terminalCode string) (item *EdbTerminal, err error) {
+	o := orm.NewOrm()
+	sql := ` SELECT *  FROM edb_terminal WHERE terminal_code = ? `
+	err = o.Raw(sql, terminalCode).QueryRow(&item)
+	return
+}
+
+// GetEdbCountGroupByTerminal 获取终端code分组总数
+func GetEdbCountGroupByTerminal(source int) (list []TerminalCodeCountGroup, err error) {
+	o := orm.NewOrm()
+	sql := `select terminal_code,count(1) total from edb_info where source = ? AND no_update=0 AND terminal_code != "" group by terminal_code; `
+	_, err = o.Raw(sql, source).QueryRows(&list)
+	return
+}
+
+type BaseIndexTerminalCode struct {
+	TerminalCode string `description:"终端编码，用于配置在机器上"`
+	IndexName    string
+}
+
+// GetBaseIndexTerminalCode 获取数据源的终端code
+func GetBaseIndexTerminalCode(edbCode, tableName string) (item BaseIndexTerminalCode, err error) {
+	o := orm.NewOrm()
+	sql := fmt.Sprintf(`select terminal_code, index_name from %s where index_code = ? `, tableName)
+	err = o.Raw(sql, edbCode).QueryRow(&item)
+	return
+}

models/future_good/future_good_edb_info.go

@@ -26,6 +26,7 @@ type FutureGoodEdbInfo struct {
 	LatestValue         float64   `description:"数据最新的值"`
 	LatestDate          time.Time `description:"数据最新的日期"`
 	ServerUrl           string    `description:"服务器地址"`
+	TerminalCode        string    `description:"终端编码，用于配置在机器上"`
 	CreateTime          time.Time
 	ModifyTime          time.Time
 }

routers/commentsRouter.go

@@ -295,6 +295,15 @@ func init() {
             Filters: nil,
             Params: nil})
 
+    beego.GlobalControllerRouter["eta/eta_index_lib/controllers:EdbInfoStatController"] = append(beego.GlobalControllerRouter["eta/eta_index_lib/controllers:EdbInfoStatController"],
+        beego.ControllerComments{
+            Method: "SetEdbSourceStat",
+            Router: `/source_update`,
+            AllowHTTPMethods: []string{"post"},
+            MethodParams: param.Make(),
+            Filters: nil,
+            Params: nil})
+
     beego.GlobalControllerRouter["eta/eta_index_lib/controllers:EiaSteoController"] = append(beego.GlobalControllerRouter["eta/eta_index_lib/controllers:EiaSteoController"],
         beego.ControllerComments{
             Method: "Add",

routers/router.go

@@ -202,6 +202,11 @@ func init() {
 				&controllers.JiaYueController{},
 			),
 		),
+		beego.NSNamespace("/edb_stat",
+			beego.NSInclude(
+				&controllers.EdbInfoStatController{},
+			),
+		),
 	)
 	beego.AddNamespace(ns)
 }

services/base_from_lt.go

@@ -2,7 +2,6 @@ package services
 
 import (
 	"encoding/json"
-	"errors"
 	"eta/eta_index_lib/utils"
 	"fmt"
 
@@ -15,13 +14,27 @@ type EdbDataFromLt struct {
 }
 
 // GetEdbDataFromLt 获取路透数据
-func GetEdbDataFromLt(edbCode, startDate, endDate string) (dataList map[int64]interface{}, err error) {
-	if utils.Hz_LT_Data_Url == `` {
+func GetEdbDataFromLt(edbCode, startDate, endDate, edbTerminalCode string) (dataList map[int64]interface{}, err error) {
+	/*if utils.Hz_LT_Data_Url == `` {
 		err = errors.New("路透社接口未配置")
 		return
+	}*/
+	terminal, err := GetTerminal(utils.DATA_SOURCE_LT, edbTerminalCode)
+	if err != nil {
+		err = fmt.Errorf("获取路透社接口配置出错 Err: %s", err)
+		return
+	}
+	if terminal.ServerUrl == "" {
+		err = fmt.Errorf("路透社接口未配置")
+		return
+	}
+	if edbTerminalCode == "" {
+		// 设置指标与终端关系的缓存
+		terminalCodeCacheKey := utils.CACHE_EDB_TERMINAL_CODE_URL + edbCode
+		_ = utils.Rc.Put(terminalCodeCacheKey, terminal.TerminalCode, utils.GetTodayLastSecond())
 	}
 	dataList = make(map[int64]interface{})
-	ltUrl := utils.Hz_LT_Data_Url + `edbInfo/ek?EdbCode=%s&StartDate=%s&EndDate=%s`
+	ltUrl := terminal.ServerUrl + `edbInfo/ek?EdbCode=%s&StartDate=%s&EndDate=%s`
 	ltUrl = fmt.Sprintf(ltUrl, edbCode, startDate, endDate)
 	utils.FileLog.Info("ltUrl:%s", ltUrl)
 	body, err := http.Get(ltUrl)

services/base_from_mysteel_chemical.go

@@ -20,11 +20,29 @@ func HandleMysteelIndex(req *models.HandleMysteelIndexResp) (err error) {
 			return
 		}
 	}
+
+	go func() {
+		// 钢联终端统计汇总
+		_ = SetMysteelChemicalEdbInfoUpdateStat()
+		_ = SetEdbSourceStat()
+	}()
 	return
 }
 
 func handleIndex(indexItem *models.HandleMysteelIndex) (err error) {
-	//return
+	defer func() {
+		if err != nil {
+			// 添加刷新失败日志
+			dataUpdateResult := 2
+			dataUpdateFailedReason := "服务异常"
+			edbInfo, e := models.GetEdbInfoByEdbCode(utils.DATA_SOURCE_MYSTEEL_CHEMICAL, indexItem.IndexCode)
+			if e == nil {
+				//查询指标存在，才添加刷新日志
+				_ = AddEdbInfoUpdateLog(edbInfo.EdbInfoId, 2, err.Error(), dataUpdateResult, dataUpdateFailedReason, 1)
+			}
+		}
+	}()
+
 	indexObj := new(models.BaseFromMysteelChemicalIndex)
 	var indexId int64
 
@@ -125,6 +143,7 @@ func handleIndex(indexItem *models.HandleMysteelIndex) (err error) {
 	}
 
 	dataObj := new(models.BaseFromMysteelChemicalData)
+	var hasUpdate bool
 	// 遍历excel数据，然后跟现有的数据做校验，不存在则入库
 	for date, value := range indexItem.ExcelDataMap {
 		if findData, ok := exitDataMap[date]; !ok {
@@ -154,6 +173,7 @@ func handleIndex(indexItem *models.HandleMysteelIndex) (err error) {
 				updateDataColsArr = append(updateDataColsArr, "value")
 				updateDataColsArr = append(updateDataColsArr, "modify_time")
 				dataObj.Update(updateDataColsArr)
+				hasUpdate = true
 			}
 		}
 	}
@@ -162,7 +182,9 @@ func handleIndex(indexItem *models.HandleMysteelIndex) (err error) {
 		err = dataObj.Add(addDataList)
 		if err != nil {
 			fmt.Println("dataObj.Add() Err:" + err.Error())
+			return
 		}
+		hasUpdate = true
 	}
 
 	//修改最大最小日期
@@ -173,26 +195,48 @@ func handleIndex(indexItem *models.HandleMysteelIndex) (err error) {
 			fmt.Println("ModifyMysteelIndexMaxAndMinInfo Err:" + e.Error())
 		}
 	}
-	
 	// 同步刷新图库钢联的指标
-	go func() {
-		var indexErr error
-		defer func() {
-			if indexErr != nil {
-				tips := fmt.Sprintf("钢联数据刷新-ETA指标刷新异常, 指标编码: %s, err: %s", indexObj.IndexCode, indexErr.Error())
-				alarm_msg.SendAlarmMsg(tips, 3)
-			}
-		}()
+	//go func() {
+	var indexErr error
+	var lErr error
+	defer func() {
+		if indexErr != nil {
+			tips := fmt.Sprintf("钢联数据刷新-ETA指标刷新异常, 指标编码: %s, err: %s", indexObj.IndexCode, indexErr.Error())
+			alarm_msg.SendAlarmMsg(tips, 3)
+		}
+
+		if lErr != nil {
+			tips := fmt.Sprintf("钢联数据刷新-ETA指标刷新统计异常, 指标编码: %s, err: %s", indexObj.IndexCode, lErr.Error())
+			alarm_msg.SendAlarmMsg(tips, 3)
+		}
+	}()
+
+	edbInfo, e := models.GetEdbInfoByEdbCode(utils.DATA_SOURCE_MYSTEEL_CHEMICAL, indexObj.IndexCode)
+	if e != nil && e.Error() != utils.ErrNoRow() {
+		indexErr = e
+		return
+	}
 
-		edbInfo, e := models.GetEdbInfoByEdbCode(utils.DATA_SOURCE_MYSTEEL_CHEMICAL, indexObj.IndexCode)
-		if e != nil && e.Error() != utils.ErrNoRow() {
-			indexErr = e
+	if edbInfo != nil {
+		dataUpdateResult := 2
+		dataUpdateFailedReason := "服务异常"
+		_, logErrMsg, logErr := logic.RefreshBaseEdbInfo(edbInfo, ``)
+		if logErr != nil {
+			lErr = AddEdbInfoUpdateLog(edbInfo.EdbInfoId, 2, logErrMsg+err.Error(), dataUpdateResult, dataUpdateFailedReason, 1)
 			return
 		}
-		if edbInfo != nil {
-			logic.RefreshBaseEdbInfo(edbInfo, ``)
+
+		if hasUpdate {
+			dataUpdateResult = 1
+			dataUpdateFailedReason = ""
+		} else {
+			dataUpdateFailedReason = "未刷新到数据"
 		}
-	}()
+
+		// 添加刷新成功日志
+		lErr = AddEdbInfoUpdateLog(edbInfo.EdbInfoId, 1, "", dataUpdateResult, dataUpdateFailedReason, 1)
+	}
+	//}()
 
 	return
 }

services/base_from_pb.go

@@ -2,7 +2,6 @@ package services
 
 import (
 	"encoding/json"
-	"errors"
 	"eta/eta_index_lib/models"
 	"eta/eta_index_lib/utils"
 	"fmt"
@@ -11,13 +10,27 @@ import (
 )
 
 // GetEdbDataFromPb 获取Pb数据
-func GetEdbDataFromPb(edbCode, startDate, endDate string) (item *models.EdbDataFromPb, err error) {
-	if utils.Hz_Pb_Data_Url == `` {
+func GetEdbDataFromPb(edbCode, startDate, endDate, edbTerminalCode string) (item *models.EdbDataFromPb, err error) {
+	/*	if utils.Hz_Pb_Data_Url == `` {
 		err = errors.New("彭博接口未配置")
 		return
+	}*/
+	terminal, err := GetTerminal(utils.DATA_SOURCE_PB, edbTerminalCode)
+	if err != nil {
+		err = fmt.Errorf("获取彭博接口配置出错 Err: %s", err)
+		return
+	}
+	if terminal.ServerUrl == "" {
+		err = fmt.Errorf("彭博接口未配置")
+		return
+	}
+	if edbTerminalCode == "" {
+		// 设置指标与终端关系的缓存
+		terminalCodeCacheKey := utils.CACHE_EDB_TERMINAL_CODE_URL + edbCode
+		_ = utils.Rc.Put(terminalCodeCacheKey, terminal.TerminalCode, utils.GetTodayLastSecond())
 	}
 	edbCode = url.QueryEscape(edbCode)
-	bpUrl := utils.Hz_Pb_Data_Url + `edbInfo/pb?EdbCode=%s&StartDate=%s&EndDate=%s`
+	bpUrl := terminal.ServerUrl + `edbInfo/pb?EdbCode=%s&StartDate=%s&EndDate=%s`
 	bpUrl = fmt.Sprintf(bpUrl, edbCode, startDate, endDate)
 	utils.FileLog.Info("bpUrl:" + bpUrl)
 	body, err := http.Get(bpUrl)
@@ -37,14 +50,28 @@ func GetEdbDataFromPb(edbCode, startDate, endDate string) (item *models.EdbDataF
 }
 
 // GetEdbDataFromPbFinance 获取Pb财务数据
-func GetEdbDataFromPbFinance(companyCode, edbCode, startDate, endDate string) (item *models.EdbDataFromPb, err error) {
-	if utils.Hz_Pb_Data_Url == `` {
+func GetEdbDataFromPbFinance(companyCode, edbCode, startDate, endDate, edbTerminalCode string) (item *models.EdbDataFromPb, err error) {
+	/*if utils.Hz_Pb_Data_Url == `` {
 		err = errors.New("彭博接口未配置")
 		return
+	}*/
+	terminal, err := GetTerminal(utils.DATA_SOURCE_PB_FINANCE, edbTerminalCode)
+	if err != nil {
+		err = fmt.Errorf("获取彭博接口配置出错 Err: %s", err)
+		return
+	}
+	if terminal.ServerUrl == "" {
+		err = fmt.Errorf("彭博接口未配置")
+		return
+	}
+	if edbTerminalCode == "" {
+		// 设置指标与终端关系的缓存
+		terminalCodeCacheKey := utils.CACHE_EDB_TERMINAL_CODE_URL + companyCode + "&&" + edbCode
+		_ = utils.Rc.Put(terminalCodeCacheKey, terminal.TerminalCode, utils.GetTodayLastSecond())
 	}
 	companyCode = url.QueryEscape(companyCode)
 	edbCode = url.QueryEscape(edbCode)
-	bpUrl := utils.Hz_Pb_Data_Url + `edbInfo/pb/finance?CompanyCode=%s&EdbCode=%s&StartDate=%s&EndDate=%s`
+	bpUrl := terminal.ServerUrl + `edbInfo/pb/finance?CompanyCode=%s&EdbCode=%s&StartDate=%s&EndDate=%s`
 	bpUrl = fmt.Sprintf(bpUrl, companyCode, edbCode, startDate, endDate)
 	utils.FileLog.Info("bpUrl:" + bpUrl)
 	body, err := http.Get(bpUrl)

services/base_from_smm.go

@@ -96,7 +96,6 @@ func SmmIndexHandle(baseFilePath, renameFilePath, indexName, indexCode, unit, fr
 		updateParams["frequency"] = frequency
 		updateParams["unit"] = unit
 		updateParams["interface"] = source
-		updateParams["terminal_code"] = terminalCode
 		updateParams["modify_time"] = time.Now()
 
 		if baseFilePath != `` && item.BaseFileName != baseFilePath {

services/base_from_ths.go

@@ -25,25 +25,43 @@ type EdbDataFromThsInterface struct {
 	} `json:"tables"`
 }
 
-func GetEdbDataFromThs(edbCode, startDate, endDate string) (item models.EdbDataFromThs, err error) {
-	//if utils.BusinessCode == utils.HZ_TRIAL_BUSSINESS_CODE { // 体验版也走客户端
-	//	return getEdbDataFromThsApp(edbCode, startDate, endDate, 0)
-	//}
+func GetEdbDataFromThs(edbCode, startDate, endDate, edbTerminalCode string) (item models.EdbDataFromThs, err error) {
+	terminal, err := GetTerminal(utils.DATA_SOURCE_THS, edbTerminalCode)
+	if err != nil {
+		err = fmt.Errorf("获取同花顺接口配置出错 Err: %s", err)
+		return
+	}
+	if terminal.ServerUrl == "" {
+		err = fmt.Errorf("同花顺接口未配置")
+		return
+	}
+
+	if edbTerminalCode == "" {
+		// 设置指标与终端关系的缓存
+		terminalCodeCacheKey := utils.CACHE_EDB_TERMINAL_CODE_URL + edbCode
+		_ = utils.Rc.Put(terminalCodeCacheKey, terminal.TerminalCode, utils.GetTodayLastSecond())
+	}
+
 	// 如果没有配置，获取配置的方式是api，那么就走官方接口
 	if utils.ThsDataMethod == "" || utils.ThsDataMethod == "api" {
-		return getEdbDataFromThsHttp(edbCode, startDate, endDate)
+		var token string
+		token, err = GetAccessToken(false, terminal.Value)
+		if err != nil {
+			return
+		}
+		return getEdbDataFromThsHttp(edbCode, startDate, endDate, terminal.Value, token)
 	}
 
-	return getEdbDataFromThsApp(edbCode, startDate, endDate, 0)
+	return getEdbDataFromThsApp(edbCode, startDate, endDate, 0, terminal.ServerUrl)
 }
 
 // getEdbDataFromThs 获取同花顺接口数据
-func getEdbDataFromThsApp(edbCode, startDate, endDate string, num int) (item models.EdbDataFromThs, err error) {
-	if utils.Hz_Ths_Data_Url == `` {
+func getEdbDataFromThsApp(edbCode, startDate, endDate string, num int, serverUrl string) (item models.EdbDataFromThs, err error) {
+	if serverUrl == `` {
 		err = errors.New("同花顺接口未配置")
 		return
 	}
-	thsUrl := utils.Hz_Ths_Data_Url + `edbInfo/ths?EdbCode=%s&StartDate=%s&EndDate=%s`
+	thsUrl := serverUrl + `edbInfo/ths?EdbCode=%s&StartDate=%s&EndDate=%s`
 	thsUrl = fmt.Sprintf(thsUrl, edbCode, startDate, endDate)
 	utils.FileLog.Info("thsUrl:" + thsUrl)
 	body, err := http.Get(thsUrl)
@@ -68,7 +86,7 @@ func getEdbDataFromThsApp(edbCode, startDate, endDate string, num int) (item mod
 		//session has expired,please re-login after using the system
 		//如果是同花顺登录session失效了，那么就重新请求获取数据
 		if tmpItems.Errorcode == -1020 && num == 0 {
-			return getEdbDataFromThsApp(edbCode, startDate, endDate, 1)
+			return getEdbDataFromThsApp(edbCode, startDate, endDate, 1, serverUrl)
 		}
 
 		err = errors.New(string(body))
@@ -158,21 +176,45 @@ type FutureGoodDataFromThsInterface struct {
 	} `json:"tables"`
 }
 
-func GetFutureGoodDataFromThs(edbCode, startDate, endDate string) (item future_good.FutureGoodDataFromThs, err error) {
+func GetFutureGoodDataFromThs(edbCode, startDate, endDate, edbTerminalCode string) (item future_good.FutureGoodDataFromThs, err error) {
+	terminal, err := GetFirstTerminal(utils.DATA_SOURCE_THS, edbTerminalCode)
+	if err != nil {
+		err = fmt.Errorf("获取同花顺接口配置出错 Err: %s", err)
+		return
+	}
+	if terminal.ServerUrl == "" {
+		err = fmt.Errorf("同花顺接口未配置")
+		return
+	}
+
+	if edbTerminalCode == "" {
+		terminalCodeCacheKey := utils.CACHE_EDB_TERMINAL_CODE_GOOD_URL + edbCode
+		_ = utils.Rc.Put(terminalCodeCacheKey, terminal.TerminalCode, utils.GetTodayLastSecond())
+	}
+
 	if utils.ThsDataMethod == "" || utils.ThsDataMethod == "api" { // 生产环境走官方http请求，测试环境走终端
-		return getFutureGoodDataFromThsHttp(edbCode, startDate, endDate)
+		var token string
+		token, err = GetAccessToken(false, terminal.Value)
+		if err != nil {
+			return
+		}
+		return getFutureGoodDataFromThsHttp(edbCode, startDate, endDate, terminal.Value, token)
 	} else {
-		return getFutureGoodDataFromThsApp(edbCode, startDate, endDate, 0)
+		return getFutureGoodDataFromThsApp(edbCode, startDate, endDate, 0, terminal.ServerUrl)
 	}
 }
 
 // getFutureGoodDataFromThsApp 通过终端获取wind的商品数据
-func getFutureGoodDataFromThsApp(edbCode, startDate, endDate string, num int) (item future_good.FutureGoodDataFromThs, err error) {
-	if utils.Hz_Ths_Data_Url == `` {
+func getFutureGoodDataFromThsApp(edbCode, startDate, endDate string, num int, serverUrl string) (item future_good.FutureGoodDataFromThs, err error) {
+	/*if utils.Hz_Ths_Data_Url == `` {
+		err = errors.New("同花顺接口未配置")
+		return
+	}*/
+	if serverUrl == "" {
 		err = errors.New("同花顺接口未配置")
 		return
 	}
-	thsUrl := utils.Hz_Ths_Data_Url + `edbInfo/ths/future_good?EdbCode=%s&StartDate=%s&EndDate=%s`
+	thsUrl := serverUrl + `edbInfo/ths/future_good?EdbCode=%s&StartDate=%s&EndDate=%s`
 	thsUrl = fmt.Sprintf(thsUrl, edbCode, startDate, endDate)
 	utils.FileLog.Info("thsUrl:" + thsUrl)
 	body, err := http.Get(thsUrl)
@@ -197,7 +239,7 @@ func getFutureGoodDataFromThsApp(edbCode, startDate, endDate string, num int) (i
 		//session has expired,please re-login after using the system
 		//如果是同花顺登录session失效了，那么就重新请求获取数据
 		if tmpItems.Errorcode == -1020 && num == 0 {
-			return getFutureGoodDataFromThsApp(edbCode, startDate, endDate, 1)
+			return getFutureGoodDataFromThsApp(edbCode, startDate, endDate, 1, serverUrl)
 		}
 
 		err = errors.New(string(body))

services/base_from_ths_http.go

@@ -19,7 +19,7 @@ import (
 var tokenRedisKey = `THS_SERVER_ACCESS_TOKEN`
 
 // getEdbDataFromThsHttp 通过url获取同花顺的普通数据
-func getEdbDataFromThsHttp(edbCode, startDate, endDate string) (item models.EdbDataFromThs, err error) {
+func getEdbDataFromThsHttp(edbCode, startDate, endDate, thsRefreshToken, token string) (item models.EdbDataFromThs, err error) {
 	thsUrl := `https://quantapi.51ifind.com/api/v1/edb_service`
 	//indicators 是 半角逗号分隔的所有指标，宏观指标过多，推荐使用Windows超级命令生成。 "indicators":"M001620326,M002822183"
 	//functionpara 否 key-value格式,省略时不进行更新时间筛选。两个时间控件更新起始时间（startrtime）和更新结束时间（endrtime），不勾选时省略见下方代码块
@@ -32,7 +32,7 @@ func getEdbDataFromThsHttp(edbCode, startDate, endDate string) (item models.EdbD
 		"enddate":    endDate,
 	}
 
-	body, err, _ := postCurl(thsUrl, dataMap, 0)
+	body, err, _ := postCurl(thsUrl, dataMap, 0, thsRefreshToken, token)
 	if err != nil {
 		return
 	}
@@ -97,7 +97,7 @@ func getEdbDataFromThsHttp(edbCode, startDate, endDate string) (item models.EdbD
 }
 
 // getFutureGoodDataFromThsHttp 通过url获取同花顺的商品数据
-func getFutureGoodDataFromThsHttp(edbCode, startDate, endDate string) (item future_good.FutureGoodDataFromThs, err error) {
+func getFutureGoodDataFromThsHttp(edbCode, startDate, endDate, thsRefreshToken, token string) (item future_good.FutureGoodDataFromThs, err error) {
 	thsUrl := `https://quantapi.51ifind.com/api/v1/cmd_history_quotation`
 
 	//codes 是 半角逗号分隔的所有代码 "codes":"300033.SZ,600030.SH"
@@ -114,7 +114,7 @@ func getFutureGoodDataFromThsHttp(edbCode, startDate, endDate string) (item futu
 		"enddate":    endDate,
 	}
 
-	body, err, _ := postCurl(thsUrl, dataMap, 0)
+	body, err, _ := postCurl(thsUrl, dataMap, 0, thsRefreshToken, token)
 	if err != nil {
 		return
 	}
@@ -162,7 +162,7 @@ type BaseThsInterface struct {
 }
 
 // postCurl post请求同花顺接口
-func postCurl(urlStr string, dataMap map[string]interface{}, num int) (body []byte, err error, errMsg string) {
+func postCurl(urlStr string, dataMap map[string]interface{}, num int, thsRefreshToken, token string) (body []byte, err error, errMsg string) {
 	logMsg := ``
 	defer func() {
 		if err != nil {
@@ -172,11 +172,6 @@ func postCurl(urlStr string, dataMap map[string]interface{}, num int) (body []by
 			}
 		}
 	}()
-	token, err := GetAccessToken(false)
-	if err != nil {
-		return
-	}
-
 	jsonStrByte, err := json.Marshal(dataMap)
 	if err != nil {
 		return
@@ -215,12 +210,12 @@ func postCurl(urlStr string, dataMap map[string]interface{}, num int) (body []by
 	//如果是token失效，同时只是第一次请求（没有尝试强制刷新token，那么重新请求）
 	if utils.InArrayByInt([]int{-1010, -1302}, int(response.ErrorCode)) && num <= 0 {
 		//token失效
-		_, tmpErr := refreshAccessToken()
-		if tmpErr != nil {
-			err = tmpErr
+		token, err = refreshAccessToken(thsRefreshToken)
+		if err != nil {
+			return
 		}
 		num++
-		return postCurl(urlStr, dataMap, num)
+		return postCurl(urlStr, dataMap, num, thsRefreshToken, token)
 	} else if response.ErrorCode != 0 {
 		utils.FileLog.Info(fmt.Sprint("post data err:", response.ErrMsg, ";url:", urlStr, ";params:", reqStr, ";response:", string(body)))
 		err = errors.New(response.ErrMsg)
@@ -231,28 +226,30 @@ func postCurl(urlStr string, dataMap map[string]interface{}, num int) (body []by
 }
 
 // GetAccessToken 获取accessToken
-func GetAccessToken(isRefresh bool) (token string, err error) {
+func GetAccessToken(isRefresh bool, thsRefreshToken string) (token string, err error) {
 	defer func() {
 		if err != nil {
 			go alarm_msg.SendAlarmMsg("获取同花顺的token失败，ERR："+err.Error(), 3)
 		}
 	}()
-	token, redisErr := utils.Rc.RedisString(tokenRedisKey)
+	redisKey := utils.CACHE_EDB_THS_SERVER_TOKEN + thsRefreshToken + ":"
+	token, redisErr := utils.Rc.RedisString(redisKey)
 	//如果从redis中accessToken 获取失败或者token为空了，再或者需要强制刷新了，那么重新获取accessToken
 	if redisErr != nil || token == `` || isRefresh {
-		return refreshAccessToken()
+		token, _ = refreshAccessToken(thsRefreshToken)
+		return
 	}
 	return
 }
 
 // refreshAccessToken 强制刷新获取accessToken
-func refreshAccessToken() (token string, err error) {
+func refreshAccessToken(thsRefreshToken string) (token string, err error) {
 	defer func() {
 		if err != nil {
 			go alarm_msg.SendAlarmMsg("刷新同花顺的token失败；ERR："+err.Error(), 3)
 		}
 	}()
-	tokenInfo, tmpErr := getAccessToken()
+	tokenInfo, tmpErr := getAccessToken(thsRefreshToken)
 	if tmpErr != nil {
 		err = tmpErr
 		return
@@ -261,16 +258,18 @@ func refreshAccessToken() (token string, err error) {
 
 	expireTime, err := time.ParseInLocation(utils.FormatDateTime, tokenInfo.ExpiredTime, time.Local)
 	if err != nil {
-		go alarm_msg.SendAlarmMsg("获取同花顺的token失败；同花顺token截止日期转换失败，ERR："+err.Error(), 3)
+		err = fmt.Errorf("获取同花顺的token失败；同花顺token截止日期转换失败，ERR：%s", err.Error())
 		return
 	}
 
 	//token存入redis
 	//err = utils.Rc.Put(tokenRedisKey, token, time.Duration(expireTime.Unix()-600)*time.Second)
 	// 本来是要设置下600s的过期时间，但因为不是强制刷新token，就不获取了
-	err = utils.Rc.Put(tokenRedisKey, token, time.Duration(expireTime.Unix())*time.Second)
+	redisKey := utils.CACHE_EDB_THS_SERVER_TOKEN + thsRefreshToken + ":"
+	err = utils.Rc.Put(redisKey, token, time.Duration(expireTime.Unix())*time.Second)
 	if err != nil {
-		go alarm_msg.SendAlarmMsg("获取同花顺的token成功；同花顺token存入redis失败，ERR："+err.Error(), 3)
+		err = fmt.Errorf("获取同花顺的token成功；同花顺token存入redis失败，ERR：%s", err.Error())
+		return
 	}
 	return
 }
@@ -288,22 +287,28 @@ type TokenData struct {
 }
 
 // getAccessToken token内部请求接口
-func getAccessToken() (tokenData TokenData, err error) {
-	if utils.ThsRefreshToken == `` {
+func getAccessToken(thsRefreshToken string) (tokenData TokenData, err error) {
+	/*if utils.ThsRefreshToken == `` {
 		err = errors.New("同花顺token未配置")
 		return
-	}
+	}*/
 	defer func() {
 		if err != nil {
 			go alarm_msg.SendAlarmMsg("更新同花顺的token失败；ERR："+err.Error(), 3)
 		}
 	}()
+
+	if thsRefreshToken == "" {
+		err = errors.New("同花顺token未配置")
+		return
+	}
+
 	//getUrl := `https://quantapi.51ifind.com/api/v1/update_access_token`	// 强制刷新token；目前因为生产和测试都是使用的一个账号，所以token不能因为环境的不同而改变
 	getUrl := `https://quantapi.51ifind.com/api/v1/get_access_token` // 获取当前token
 
 	req, _ := netHttp.NewRequest("GET", getUrl, nil)
 	req.Header.Add("Content-Type", "application/json")
-	req.Header.Add("refresh_token", utils.ThsRefreshToken)
+	req.Header.Add("refresh_token", thsRefreshToken)
 
 	res, err := netHttp.DefaultClient.Do(req)
 	if err != nil {
@@ -329,5 +334,6 @@ func getAccessToken() (tokenData TokenData, err error) {
 		return
 	}
 	tokenData = tokenResp.Data
+
 	return
 }

services/base_from_wind.go

@@ -82,6 +82,7 @@ func GetEdbDataFromWind(edbCode, startDate, endDate string) (item *models.EdbDat
 
 // GetEdbDataFromWindUrl 通过url获取wind数据
 func GetEdbDataFromWindUrl(windUrl, edbCode, startDate, endDate string) (item *models.EdbDataFromWind, errorCode int, err error) {
+	// todo 刷新后的处理
 	if windUrl == `` {
 		return GetEdbDataFromWind(edbCode, startDate, endDate)
 	}
@@ -150,8 +151,12 @@ func GetWindUrl(edbCode string) (windUrl string, err error) {
 	if windUrl != "" {
 		return
 	}
+	terminalList, err := models.GetEdbTerminalListBySource(utils.DATA_SOURCE_WIND)
+	if err != nil {
+		return
+	}
 	//如果缓存中没有的话，那么从配置中获取
-	for _, windUrlMap := range utils.Hz_Wind_Data_Url_LIST {
+	for _, t := range terminalList {
 		//判断该url是否被占满了
 		//count, tmpErr := GetCountEdbCodeInWindUrl(windUrlMap.Url)
 		//if tmpErr != nil && tmpErr.Error() != "nil returned" {
@@ -163,15 +168,15 @@ func GetWindUrl(edbCode string) (windUrl string, err error) {
 		//	AddEdbCodeInWindUrl(windUrlMap.Url, edbCode)
 		//	return
 		//}
-		//如果超限了，那么进入下一循环
-		isLimit, tmpErr := GetIsLimitEdbCodeInWindUrl(windUrlMap.Url)
+		//如果超限了，那么进入下一循环,
+		isLimit, tmpErr := GetIsLimitEdbCodeInWindUrl(t.ServerUrl)
 		if isLimit {
 			err = tmpErr
 			continue
 		}
 
-		windUrl = windUrlMap.Url
-		AddEdbCodeInWindUrl(windUrlMap.Url, edbCode)
+		windUrl = t.ServerUrl
+		AddEdbCodeInWindUrl(windUrl, t.TerminalCode, edbCode)
 		return
 	}
 	return
@@ -210,12 +215,16 @@ func SetIsLimitEdbCodeInWindUrl(windUrl string) {
 
 // AddEdbCodeInWindUrl 将指标插入到缓存key中
 // @return isInsert bool 是否插入数据，true时为插入数据，false表示数据已存在
-func AddEdbCodeInWindUrl(windUrl, edbCode string) (isInsert bool) {
+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
 }
 
@@ -223,5 +232,8 @@ func AddEdbCodeInWindUrl(windUrl, edbCode string) (isInsert bool) {
 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
 }

services/edb_info_stat.go

@@ -0,0 +1,326 @@
+package services
+
+import (
+	"eta/eta_index_lib/models"
+	"eta/eta_index_lib/models/data_stat"
+	"eta/eta_index_lib/services/alarm_msg"
+	"eta/eta_index_lib/utils"
+	"fmt"
+	"time"
+)
+
+// AddEdbInfoUpdateLog 添加指标编辑/刷新日志
+func AddEdbInfoUpdateLog(edbInfoId int, updateResult int, updateFailedReason string, dataUpdateResult int, dataUpdateFailedReason string, isSourceRefresh int) (err error) {
+	var edbInfo *models.EdbInfo
+	if edbInfoId > 0 {
+		// 获取指标详情
+		edbInfo, err = models.GetEdbInfoById(edbInfoId)
+		if err != nil {
+			err = fmt.Errorf("指标不存在")
+			return
+		}
+		log := new(data_stat.EdbInfoUpdateLog)
+		log.EdbInfoId = edbInfo.EdbInfoId
+		log.SourceName = edbInfo.SourceName
+		log.Source = edbInfo.Source
+		log.EdbCode = edbInfo.EdbCode
+		log.EdbName = edbInfo.EdbName
+		log.EdbNameSource = edbInfo.EdbNameSource
+		log.Frequency = edbInfo.Frequency
+		log.Unit = edbInfo.Unit
+		log.StartDate = edbInfo.StartDate
+		log.EndDate = edbInfo.EndDate
+		log.SysUserId = edbInfo.SysUserId
+		log.SysUserRealName = edbInfo.SysUserRealName
+		log.UniqueCode = edbInfo.UniqueCode
+		log.EdbCreateTime = edbInfo.CreateTime
+		log.EdbModifyTime = edbInfo.ModifyTime
+		log.CreateTime = time.Now()
+		log.LatestDate = edbInfo.LatestDate
+		log.LatestValue = edbInfo.LatestValue
+		log.TerminalCode = edbInfo.TerminalCode
+		log.UpdateResult = updateResult
+		log.UpdateFailedReason = updateFailedReason
+		log.DataUpdateTime = edbInfo.DataUpdateTime
+		log.ErDataUpdateDate = edbInfo.ErDataUpdateDate
+		log.DataUpdateResult = dataUpdateResult
+		log.DataUpdateFailedReason = dataUpdateFailedReason
+		log.IsSourceRefresh = isSourceRefresh
+		_, err = data_stat.AddEdbUpdateLog(log)
+		if err != nil {
+			err = fmt.Errorf("新增指标更新日志失败，Err: %s", err)
+			return
+		}
+	}
+	return
+}
+
+// SetMysteelChemicalEdbInfoUpdateStat 定时统计钢联化工的数据源明细表
+func SetMysteelChemicalEdbInfoUpdateStat() (err error) {
+	defer func() {
+		if err != nil {
+			tips := fmt.Sprintf("钢联数据刷新-ETA指标刷新统计异常 Err: %s", err.Error())
+			alarm_msg.SendAlarmMsg(tips, 3)
+		}
+	}()
+	//查询钢联的所有指标信息
+	condition := " and source = ? "
+	var pars []interface{}
+	pars = append(pars, utils.DATA_SOURCE_MYSTEEL_CHEMICAL)
+	edbList, err := models.GetEdbInfoByCondition(condition, pars, 0)
+	if err != nil {
+		err = fmt.Errorf("查询钢联化工指标信息出错，err: %s", err)
+		return
+	}
+	nowTime := time.Now()
+	today := time.Now().Format(utils.FormatDate)
+	todayT, _ := time.ParseInLocation(utils.FormatDate, today, time.Local)
+	nextDay := time.Now().AddDate(0, 0, 1).Format(utils.FormatDate)
+	//查询当日所有钢联指标的终端更新记录
+	updateLogList, err := data_stat.GetEdbUpdateSourceLogByCreateDate(utils.DATA_SOURCE_MYSTEEL_CHEMICAL, today, nextDay)
+	if err != nil {
+		err = fmt.Errorf("查询钢联化工指标终端更新日志报错，err: %s", err)
+		return
+	}
+	updateLogMap := make(map[int]*data_stat.EdbInfoUpdateLog)
+	if len(updateLogList) > 0 {
+		for _, v := range updateLogList {
+			if _, ok := updateLogMap[v.EdbInfoId]; !ok {
+				updateLogMap[v.EdbInfoId] = v
+			}
+		}
+	}
+	statCond := "  and source = ? and create_time >= ? and create_time < ?"
+	var statPars []interface{}
+	statPars = append(statPars, utils.DATA_SOURCE_MYSTEEL_CHEMICAL, today, nextDay)
+	//查询当日钢联所有的刷新记录
+	updateStatList, err := data_stat.GetEdbUpdateStatByCondition(statCond, statPars)
+	if err != nil {
+		err = fmt.Errorf("查询钢联化工数据源明细记录统计报错，err: %s", err)
+		return
+	}
+	updateStatMap := make(map[int]*data_stat.EdbInfoUpdateStat)
+	if len(updateStatList) > 0 {
+		for _, v := range updateStatList {
+			updateStatMap[v.EdbInfoId] = v
+		}
+	}
+	logStat := new(data_stat.EdbInfoUpdateStat)
+	//组装新增数据
+	addList := make([]*data_stat.EdbInfoUpdateStat, 0)
+	modifyList := make([]*data_stat.EdbInfoUpdateStat, 0)
+	if len(edbList) > 0 {
+		for _, v := range edbList {
+			tmp := &data_stat.EdbInfoUpdateStat{
+				EdbInfoId:        v.EdbInfoId,
+				SourceName:       v.SourceName,
+				Source:           v.Source,
+				EdbCode:          v.EdbCode,
+				EdbName:          v.EdbName,
+				EdbNameSource:    v.EdbNameSource,
+				Frequency:        v.Frequency,
+				Unit:             v.Unit,
+				StartDate:        v.StartDate,
+				EndDate:          v.EndDate,
+				SysUserId:        v.SysUserId,
+				SysUserRealName:  v.SysUserRealName,
+				UniqueCode:       v.UniqueCode,
+				EdbCreateTime:    v.CreateTime,
+				EdbModifyTime:    v.ModifyTime,
+				LatestDate:       v.LatestDate,
+				LatestValue:      v.LatestValue,
+				TerminalCode:     v.TerminalCode,
+				DataUpdateTime:   v.DataUpdateTime,
+				ErDataUpdateDate: v.ErDataUpdateDate,
+				ModifyTime:       nowTime,
+			}
+			needRefresh, _ := checkMySteelEdbInfoNeedRefresh(v.Frequency)
+			tmp.NeedRefresh = needRefresh
+
+			// 判断是否当日新增
+			if v.CreateTime.After(todayT) || v.CreateTime == todayT {
+				tmp.IsAdd = 1
+			} else {
+				tmp.IsAdd = 2
+			}
+			if up, ok := updateLogMap[v.EdbInfoId]; ok {
+				tmp.DataUpdateTime = up.DataUpdateTime
+				tmp.ErDataUpdateDate = up.ErDataUpdateDate
+				tmp.DataUpdateResult = up.DataUpdateResult
+				tmp.DataUpdateFailedReason = up.DataUpdateFailedReason
+				tmp.HasRefresh = 1
+				tmp.UpdateResult = up.UpdateResult
+				tmp.UpdateFailedReason = up.UpdateFailedReason
+				tmp.UpdateTime = up.CreateTime.Format(utils.FormatDateTime)
+			} else if needRefresh == 1 {
+				tmp.HasRefresh = 0
+				tmp.DataUpdateResult = 2
+				tmp.DataUpdateFailedReason = "服务异常"
+			}
+
+			// 判断是否需要新增还是更新
+			if exist, ok := updateStatMap[v.EdbInfoId]; ok {
+				tmp.Id = exist.Id
+				modifyList = append(modifyList, tmp)
+			} else {
+				tmp.CreateTime = nowTime
+				addList = append(addList, tmp)
+			}
+		}
+	}
+
+	//判断当日指标统计数据是否存在，如果存在则更新，不存在则新增
+	if len(addList) > 0 {
+		err = logStat.Add(addList)
+	}
+	if len(modifyList) > 0 {
+		err = data_stat.UpdateEdbUpdateStatMulti(modifyList)
+	}
+	return
+}
+
+func checkMySteelEdbInfoNeedRefresh(frequency string) (needRefresh int, err error) {
+	now := time.Now()
+	week := int(now.Weekday())
+	//日度
+	if week >= 1 && week <= 6 {
+		if frequency == "日度" {
+			needRefresh = 1
+			return
+		}
+	}
+	//周度
+	if week >= 3 && week <= 6 {
+		if frequency == "周度" {
+			needRefresh = 1
+			return
+		}
+	}
+
+	day := now.Day() //季度，月度，年度都是每个月1号刷新
+	if day == 1 {
+		needRefresh = 1
+	}
+	return
+}
+
+// SetEdbSourceStat 定时统计数据源汇总表
+func SetEdbSourceStat() (err error) {
+	defer func() {
+		if err != nil {
+			tips := fmt.Sprintf("钢联数据刷新-ETA指标刷新统计异常 Err: %s", err.Error())
+			alarm_msg.SendAlarmMsg(tips, 3)
+		}
+	}()
+	//查询钢联的所有指标信息
+	nowTime := time.Now()
+	today := time.Now().Format(utils.FormatDate)
+	nextDay := time.Now().AddDate(0, 0, 1).Format(utils.FormatDate)
+
+	statCond := "  and  create_time >= ? and create_time < ?"
+	var statPars []interface{}
+	statPars = append(statPars, today, nextDay)
+	//查询当日钢联所有的统计数据
+	updateStatList, err := data_stat.GetEdbUpdateStatByCondition(statCond, statPars)
+	if err != nil {
+		err = fmt.Errorf("查询钢联化工数据源明细记录统计报错，err: %s", err)
+		return
+	}
+	updateStatMap := make(map[string][]*data_stat.EdbInfoUpdateStat)
+	if len(updateStatList) > 0 {
+		for _, v := range updateStatList {
+			updateStatMap[v.TerminalCode] = append(updateStatMap[v.TerminalCode], v)
+		}
+	}
+
+	cond := "  and  create_time >= ? and create_time < ?"
+	var pars []interface{}
+	pars = append(pars, today, nextDay)
+	//查询当日钢联所有的统计数据
+	statList, err := data_stat.GetEdbSourceStatByCondition(cond, pars)
+	if err != nil {
+		err = fmt.Errorf("查询钢联化工数据源统计报错，err: %s", err)
+		return
+	}
+	statMap := make(map[string]*data_stat.EdbSourceStat)
+	if len(statList) > 0 {
+		for _, v := range statList {
+			statMap[v.TerminalCode] = v
+		}
+	}
+
+	// 查询今日被删除的指标数
+	delList, err := data_stat.GetEdbDeleteLogNumByCreateTime(today, nextDay)
+	if err != nil {
+		err = fmt.Errorf("查询今日被删除指标数目报错，err: %s", err)
+		return
+	}
+	delMap := make(map[string]int)
+	if len(delList) > 0 {
+		for _, v := range delList {
+			delMap[v.TerminalCode] = v.Num
+		}
+	}
+	logStat := new(data_stat.EdbSourceStat)
+	//组装新增数据
+	addList := make([]*data_stat.EdbSourceStat, 0)
+	modifyList := make([]*data_stat.EdbSourceStat, 0)
+
+	for terminalCode, list := range updateStatMap {
+		tmp := new(data_stat.EdbSourceStat)
+		for k, v := range list {
+			if k == 0 {
+				tmp.SourceName = v.SourceName
+				tmp.Source = v.Source
+				tmp.TerminalCode = v.TerminalCode
+				tmp.ModifyTime = nowTime
+			}
+
+			tmp.EdbNum = tmp.EdbNum + 1
+			if v.IsAdd == 1 {
+				tmp.EdbNewNum = tmp.EdbNewNum + 1
+			}
+			if v.NeedRefresh == 1 {
+				tmp.NeedRefreshNum = tmp.NeedRefreshNum + 1
+			}
+
+			if v.HasRefresh == 1 {
+				tmp.HasRefreshNum = tmp.HasRefreshNum + 1
+			}
+
+			// 区分刷新成功和更新成功
+			if v.DataUpdateResult == 1 { //处理更新结果
+				tmp.UpdateSuccessNum = tmp.UpdateSuccessNum + 1
+			} else if v.NeedRefresh == 1 {
+				tmp.UpdateFailedNum = tmp.UpdateFailedNum + 1
+			}
+
+			if v.UpdateResult == 1 { //刷新结果
+				tmp.RefreshSuccessNum = tmp.RefreshSuccessNum + 1
+			} else if v.NeedRefresh == 1 {
+				tmp.RefreshFailedNum = tmp.RefreshFailedNum + 1
+			}
+		}
+		// 处理今天删除的指标数量
+		if dn, ok := delMap[terminalCode]; ok {
+			tmp.EdbDelNum = dn
+		}
+		// 判断是否需要新增还是更新
+		if exist, ok := statMap[terminalCode]; ok {
+			tmp.Id = exist.Id
+			modifyList = append(modifyList, tmp)
+		} else {
+			tmp.CreateTime = nowTime
+			addList = append(addList, tmp)
+		}
+	}
+
+	//判断当日指标统计数据是否存在，如果存在则更新，不存在则新增
+	if len(addList) > 0 {
+		err = logStat.Add(addList)
+	}
+	if len(modifyList) > 0 {
+		err = data_stat.UpdateEdbSourceStatMulti(modifyList)
+	}
+	return
+}

services/edb_terminal.go

@@ -0,0 +1,93 @@
+package services
+
+import (
+	"errors"
+	"eta/eta_index_lib/models"
+	"eta/eta_index_lib/utils"
+)
+
+// GetTerminal 获取终端信息
+func GetTerminal(source int, oldTerminalCode string) (edbTerminal *models.EdbTerminal, err error) {
+	if source == utils.DATA_SOURCE_WIND && oldTerminalCode == "" {
+		tmpConfig := new(models.EdbConfig)
+		tmpConfig, err = models.GetConfigDetailByCode("wind_terminal_code")
+		if err != nil && err.Error() != utils.ErrNoRow() {
+			err = errors.New("获取配置的wind地址失败，err:" + err.Error())
+			return
+		}
+		err = nil
+		if tmpConfig != nil {
+			oldTerminalCode = tmpConfig.ConfigValue
+			//return
+		}
+	}
+	if oldTerminalCode != "" {
+		edbTerminal, err = models.GetEdbTerminalByCode(oldTerminalCode)
+		if err != nil {
+			return
+		}
+		return
+	}
+
+	// 配置中没有的话，那么就从分组获取,
+	list, err := models.GetEdbCountGroupByTerminal(source)
+	if err != nil {
+		return
+	}
+	//windUrlNum := 0
+
+	//获取对应的配置url
+	terminalNumMap := make(map[string]int)
+	for _, v := range list {
+		terminalNumMap[v.TerminalCode] = v.Total
+	}
+
+	terminalList, err := models.GetEdbTerminalListBySource(source)
+	if err != nil {
+		return
+	}
+	if len(terminalList) == 0 {
+		err = errors.New("终端地址未配置")
+		return
+	}
+	num := 0
+	for _, v := range terminalList {
+		tmpNum := terminalNumMap[v.TerminalCode]
+		if edbTerminal == nil {
+			edbTerminal = v
+			num = tmpNum
+		} else if tmpNum < num {
+			edbTerminal = v
+			num = tmpNum
+		}
+	}
+	if edbTerminal == nil {
+		err = errors.New("获取配置的终端地址失败")
+		return
+	}
+	/*if edbTerminal != nil {
+		windUrl = edbTerminal.ServerUrl
+		terminalCode = edbTerminal.TerminalCode
+	}*/
+	return
+}
+
+func GetFirstTerminal(source int, oldTerminalCode string) (edbTerminal *models.EdbTerminal, err error) {
+	if oldTerminalCode != "" {
+		edbTerminal, err = models.GetEdbTerminalByCode(oldTerminalCode)
+		if err != nil {
+			return
+		}
+		return
+	}
+
+	edbTerminal, err = models.GetEdbTerminalFirstBySource(source)
+	if err != nil {
+		if err.Error() == utils.ErrNoRow() {
+			err = errors.New("终端未配置")
+			return
+		}
+		return
+	}
+	return
+}

services/hongze_sync_trial.go

@@ -28,14 +28,15 @@ type SyncTrialEdbInfoResp struct {
 	Code int    `json:"code"`
 	Msg  string `json:"msg"`
 	Data struct {
-		EdbCode   string `json:"edb_code"`
-		Source    int    `json:"source"`
-		ServerUrl string `json:"server_url"`
+		EdbCode      string `json:"edb_code"`
+		Source       int    `json:"source"`
+		ServerUrl    string `json:"server_url"`
+		TerminalCode string `json:"terminal_code"`
 	} `json:"data"`
 }
 
 // GetHzWindUrl 获取wind的url
-func GetHzWindUrl(edbCode string, source int) (windUrl string, err error, errMsg string) {
+func GetHzWindUrl(edbCode string, source int) (windUrl string, terminalCode string, err error, errMsg string) {
 	requestSyncTrialUrl := syncTrialUrl + `common/edb_info/edb_code?edb_code=%s&source=%d`
 	requestSyncTrialUrl = fmt.Sprintf(requestSyncTrialUrl, edbCode, source)
 
@@ -73,6 +74,6 @@ func GetHzWindUrl(edbCode string, source int) (windUrl string, err error, errMsg
 		return
 	}
 	windUrl = item.Data.ServerUrl
-
+	terminalCode = item.Data.TerminalCode
 	return
 }

utils/constants.go

@@ -197,11 +197,14 @@ const (
 )
 
 const (
-	CACHE_EDB_DATA_ADD          = "CACHE_EDB_DATA_ADD_"
-	CACHE_EDB_DATA_REFRESH      = "CACHE_EDB_DATA_REFRESH_"
-	CACHE_WIND_URL              = "CACHE_WIND_URL"
-	CACHE_CHART_INFO_DATA       = "chart:info:data:"             //图表数据
-	CACHE_STOCK_PLANT_CALCULATE = "CACHE_STOCK_PLANT_CALCULATE_" // 库存装置减产计算
+	CACHE_EDB_DATA_ADD               = "CACHE_EDB_DATA_ADD_"
+	CACHE_EDB_DATA_REFRESH           = "CACHE_EDB_DATA_REFRESH_"
+	CACHE_WIND_URL                   = "CACHE_WIND_URL"
+	CACHE_CHART_INFO_DATA            = "chart:info:data:"                 //图表数据
+	CACHE_STOCK_PLANT_CALCULATE      = "CACHE_STOCK_PLANT_CALCULATE_"     // 库存装置减产计算
+	CACHE_EDB_TERMINAL_CODE_URL      = "edb:terminal_code:edb_code:"      // 指标与终端关系的缓存
+	CACHE_EDB_TERMINAL_CODE_GOOD_URL = "edb:terminal_code:good:edb_code:" // 指标与终端关系的缓存, 商品期货
+	CACHE_EDB_THS_SERVER_TOKEN       = "edb:ths_server_token:"            //同花顺调用凭证
 )
 
 // 图表类型