钢联化工终端埋点

models/data_stat/edb_info_update_log.go

@@ -42,3 +42,22 @@ func AddEdbUpdateLog(item *EdbInfoUpdateLog) (lastId int64, err error) {
 	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 source_update_result !=0 order by id desc`
+	_, err = o.Raw(sql, source, startDate, endDate).QueryRows(&item)
+	return
+}

models/data_stat/edb_info_update_stat.go

@@ -7,39 +7,94 @@ import (
 
 // 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
-	LatestDate         string    `description:"数据最新日期"`
-	LatestValue        float64   `description:"数据最新值"`
-	TerminalCode       string    `description:"终端编码，用于配置在机器上"`
-	DataUpdateTime     time.Time `description:"最近一次数据发生变化的时间"`
-	ErDataUpdateDate   time.Time `description:"本次更新，数据发生变化的最早日期"`
-	UpdateResult       int32     `description:"本次刷新结果：1成功，2失败"`
-	UpdateFailedReason string    `description:"数据未正常刷新原因"`
-	ModifyTime         string    `description:"修改时间"`
-	UpdateTime         string    `description:"刷新/修改时间"`
-	IsAdd              int8      `description:"是否是当日新增的指标：1是，2否"`
-	NeedRefresh        int8      `description:"今日是否需要定时刷新，1需要，0不需要"` //todo 确认该字段
-	HasRefresh         int8      `description:"今日是否已刷新，1是，0否"`
+	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:"终端编码，用于配置在机器上"`
+	DataUpdateTime           string    `description:"最近一次数据发生变化的时间"`
+	ErDataUpdateDate         string    `description:"本次更新，数据发生变化的最早日期"`
+	SourceUpdateResult       int       `description:"从终端刷新到数据源的刷新结果：1成功，2失败"`
+	SourceUpdateFailedReason string    `description:"从终端刷新到数据源的数据未正常刷新原因"`
+	SourceUpdateTime         string    `description:"从终端刷新到数据源的 刷新时间"`
+	ModifyTime               time.Time `description:"修改时间"`
+	IsAdd                    int       `description:"是否是当日新增的指标：1是，2否"`
+	NeedRefresh              int       `description:"今日是否需要定时刷新，1需要，0不需要"`
+	HasRefresh               int       `description:"今日是否发起刷新服务，1是，0否"`
 }
 
-func AddEdbUpdateStat(item *EdbInfoUpdateStat) (lastId int64, err error) {
+// Add 新增
+func (r *EdbInfoUpdateStat) Add(list []*EdbInfoUpdateStat) (err error) {
 	o := orm.NewOrm()
-	lastId, err = o.Insert(item)
+	_, err = o.InsertMulti(len(list), list)
+	return
+}
+
+// UpdateEdbUpdateStatMulti 批量更新
+func UpdateEdbUpdateStatMulti(list []*EdbInfoUpdateStat) (err error) {
+	o := orm.NewOrm()
+	// todo 更新
+	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 = ?,
+data_update_time = ?,
+er_data_update_date = ?,
+is_add = ?,
+need_refresh = ?,
+has_refresh = ?,
+source_update_result = ?,
+source_update_failed_reason = ?,
+source_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.DataUpdateTime, v.ErDataUpdateDate, v.IsAdd, v.NeedRefresh, v.HasRefresh, v.SourceUpdateResult, v.SourceUpdateFailedReason, v.SourceUpdateTime, 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/db.go

@@ -111,6 +111,7 @@ func initBaseIndex() {
 // initDataStat 数据源统计管理相关表
 func initDataStat() {
 	orm.RegisterModel(
-		new(data_stat.EdbInfoUpdateLog), // 指标更新/刷新日志列表
+		new(data_stat.EdbInfoUpdateLog),  // 指标更新/刷新日志列表
+		new(data_stat.EdbInfoUpdateStat), // 数据源明细表
 	)
 }

services/base_from_mysteel_chemical.go

@@ -190,7 +190,6 @@ func handleIndex(indexItem *models.HandleMysteelIndex) (err error) {
 			fmt.Println("ModifyMysteelIndexMaxAndMinInfo Err:" + e.Error())
 		}
 	}
-	// todo 钢联数据刷新-ETA指标刷新异常 添加记录
 	// 同步刷新图库钢联的指标
 	go func() {
 		var indexErr error

services/edb_info_stat.go

@@ -3,6 +3,7 @@ package services
 import (
 	"eta/eta_index_lib/models"
 	"eta/eta_index_lib/models/data_stat"
+	"eta/eta_index_lib/utils"
 	"fmt"
 	"time"
 )
@@ -53,3 +54,131 @@ func AddEdbInfoUpdateLog(edbInfoId int, updateResult int, updateFailedReason str
 	}
 	return
 }
+
+// MysteelChemicalEdbInfoUpdateStat 定时统计钢联化工的数据源明细表
+func MysteelChemicalEdbInfoUpdateStat() (err error) {
+	//查询钢联的所有指标信息
+	condition := " and source = ? and edb_info_id=101838"
+	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,
+				//CreateTime:               v.CreateTime,
+				LatestDate:       v.LatestDate,
+				LatestValue:      v.LatestValue,
+				TerminalCode:     v.TerminalCode,
+				DataUpdateTime:   v.DataUpdateTime,
+				ErDataUpdateDate: v.ErDataUpdateDate,
+				/*SourceUpdateResult:       v.SourceUpdateResult,
+				SourceUpdateFailedReason: v.SourceUpdateFailedReason,
+				SourceUpdateTime:         v.SourceUpdateTime,*/
+				ModifyTime: nowTime,
+				//IsAdd:                    v.IsAdd,
+				//NeedRefresh:              v.NeedRefresh,
+				//HasRefresh:               v.HasRefresh,
+			}
+			// todo 判断是否需要当日更新
+			needRefresh, _ := checkEdbInfoNeedRefresh(v.EdbInfoId, 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.SourceUpdateTime = up.SourceUpdateTime
+				tmp.SourceUpdateResult = up.SourceUpdateResult
+				tmp.SourceUpdateFailedReason = up.SourceUpdateFailedReason
+				if up.SourceUpdateFailedReason != "" && up.SourceUpdateFailedReason != "未刷新到数据" {
+					tmp.SourceUpdateFailedReason = "服务异常"
+				}
+				tmp.HasRefresh = 1
+			} else if needRefresh == 1 {
+				tmp.HasRefresh = 0
+				tmp.SourceUpdateResult = 2
+				tmp.SourceUpdateFailedReason = "服务异常"
+			}
+
+			// 判断是否需要新增还是更新
+			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)
+	} else if len(modifyList) > 0 {
+		err = data_stat.UpdateEdbUpdateStatMulti(modifyList)
+	}
+	return
+}
+
+func checkEdbInfoNeedRefresh(edbInfoId int, frequency string) (needRefresh int, err error) {
+	return
+}