fix:新增凌晨5点的定时任务：定时往追加数据到ETA表格

models/data_manage/table_data.go

@@ -0,0 +1,92 @@
+package data_manage
+
+import (
+	"github.com/rdlucklib/rdluck_tools/orm"
+	"time"
+)
+
+type TableData struct {
+	TableDataId       int       `orm:"column(table_data_id);pk"`
+	TableInfoId       int       `description:"表格id"`
+	Date              time.Time `description:"指标时间"`
+	DataCol1          string    `orm:"column(data_col_1);" description:"第1个表格的数据"`
+	DataCol2          string    `orm:"column(data_col_2);" description:"第2个表格的数据"`
+	DataCol3          string    `orm:"column(data_col_3);" description:"第3个表格的数据"`
+	DataCol4          string    `orm:"column(data_col_4);" description:"第4个表格的数据"`
+	DataCol5          string    `orm:"column(data_col_5);" description:"第5个表格的数据"`
+	DataCol6          string    `orm:"column(data_col_6);" description:"第6个表格的数据"`
+	DataCol7          string    `orm:"column(data_col_7);" description:"第7个表格的数据"`
+	DataCol8          string    `orm:"column(data_col_8);" description:"第8个表格的数据"`
+	DataCol9          string    `orm:"column(data_col_9);" description:"第9个表格的数据"`
+	DataCol10         string    `orm:"column(data_col_10);" description:"第10个表格的数据"`
+	DataType          int8      `description:"数据类型，默认的区间数据是 1；插入数据是 2"`
+	Sort              int       `description:"排序字段，越小越靠前"`
+	AfterTableDataId  int       `description:"在某个表格数据id后面"`
+	BeforeTableDataId int       `description:"在某个表格数据id前面"`
+	ModifyTime        time.Time `description:"修改时间"`
+	CreateTime        time.Time `description:"创建时间"`
+}
+
+// GetBetweenTableDataListByTableInfoId 根据表格id获取数据列表
+func GetBetweenTableDataListByTableInfoId(tableInfoId, dataType int) (items []*TableData, err error) {
+	o := orm.NewOrm()
+	o.Using("data")
+	sql := ` SELECT * FROM table_data WHERE table_info_id = ? and after_table_data_id =0  and before_table_data_id =0 `
+	_, err = o.Raw(sql, tableInfoId).QueryRows(&items)
+	return
+}
+
+// GetInsertTableDataListByTableInfoId 根据表格id获取区间数据列表
+func GetInsertTableDataListByTableInfoId(tableInfoId, dataType int) (items []*TableData, err error) {
+	o := orm.NewOrm()
+	o.Using("data")
+	pars := make([]interface{}, 0)
+	pars = append(pars, tableInfoId)
+
+	sql := ` SELECT * FROM table_data WHERE table_info_id = ? and data_type=2 and (after_table_data_id !=0 and before_table_data_id != 0) `
+	if dataType > 0 {
+		sql += ` and data_type=? `
+		pars = append(pars, dataType)
+	}
+	_, err = o.Raw(sql, pars...).QueryRows(&items)
+	return
+}
+
+// GetTableDataListByTableInfoId 根据表格id获取插入数据列表
+func GetTableDataListByTableInfoId(tableInfoId int) (items []*TableData, err error) {
+	o := orm.NewOrm()
+	o.Using("data")
+	pars := make([]interface{}, 0)
+	pars = append(pars, tableInfoId)
+
+	sql := ` SELECT * FROM table_data WHERE table_info_id = ? order by sort asc `
+
+	_, err = o.Raw(sql, tableInfoId).QueryRows(&items)
+	return
+}
+
+// GetBetweenTableDataByTableInfoIdAndDate 根据表格id获取区间数据
+func GetBetweenTableDataByTableInfoIdAndDate(tableInfoId int, date string) (item *TableData, err error) {
+	o := orm.NewOrm()
+	o.Using("data")
+	sql := ` SELECT * FROM table_data WHERE table_info_id = ? and date = ? and after_table_data_id =0 `
+	err = o.Raw(sql, tableInfoId, date).QueryRow(&item)
+	return
+}
+
+// GetMaxSortByTableInfoId 根据表格id获取最大排序
+func GetMaxSortByTableInfoId(tableInfoId int) (sort int, err error) {
+	o := orm.NewOrm()
+	o.Using("data")
+	sql := ` SELECT max(sort) FROM table_data WHERE table_info_id = ?  `
+	err = o.Raw(sql, tableInfoId).QueryRow(&sort)
+	return
+}
+
+// ExecTableDataSql 执行ETA表格的sql
+func ExecTableDataSql(sql string) (err error) {
+	o := orm.NewOrm()
+	o.Using("data")
+	_, err = o.Raw(sql).Exec()
+	return
+}

models/data_manage/table_edb_mapping.go

@@ -0,0 +1,52 @@
+package data_manage
+
+import (
+	"github.com/rdlucklib/rdluck_tools/orm"
+	"time"
+)
+
+// TableEdbMapping ETA表格指标关系表
+type TableEdbMapping struct {
+	TableEdbMappingId int       `orm:"column(table_edb_mapping_id);pk"`
+	TableInfoId       int       `description:"表格id"`
+	EdbInfoId         int       `description:"指标id"`
+	UniqueCode        string    `description:"唯一编码"`
+	EdbDataColIndex   int       `description:"指标数据对应的列，从1开始"`
+	EndDate           time.Time `description:"最后一次数据更新日期"`
+	ModifyTime        time.Time `description:"修改时间"`
+	CreateTime        time.Time `description:"创建时间"`
+}
+
+// TableEdbMappingItem ETA表格指标关系表
+type TableEdbMappingItem struct {
+	TableEdbMappingId int       ``
+	TableInfoId       int       `description:"表格id"`
+	EdbInfoId         int       `description:"指标id"`
+	Source            int       `description:"指标来源"`
+	UniqueCode        string    `description:"唯一编码"`
+	EdbDataColIndex   int       `description:"指标数据对应的列，从1开始"`
+	EndDate           time.Time `description:"最后一次数据更新日期"`
+	EdbDataEndDate    time.Time `description:"指标数据更新结束日期"`
+	ModifyTime        time.Time `description:"修改时间"`
+	CreateTime        time.Time `description:"创建时间"`
+}
+
+// GetAllTableEdbMappingItemList 获取所有的 ETA表格指标关系列表数据
+func GetAllTableEdbMappingItemList() (items []*TableEdbMappingItem, err error) {
+	o := orm.NewOrm()
+	o.Using("data")
+	sql := ` select a.*,b.source,b.end_date edb_data_end_date from table_edb_mapping a 
+ join edb_info b on a.edb_info_id=b.edb_info_id `
+	_, err = o.Raw(sql).QueryRows(&items)
+	return
+}
+
+// GetAllTableEdbMappingItemListByTableInfoId 根据表格id获取ETA表格所有的指标关系列表数据
+func GetAllTableEdbMappingItemListByTableInfoId(tableInfoId int) (items []*TableEdbMappingItem, err error) {
+	o := orm.NewOrm()
+	o.Using("data")
+	sql := ` select a.*,b.source,b.end_date edb_data_end_date from table_edb_mapping a 
+ join edb_info b on a.edb_info_id=b.edb_info_id where a.table_info_id = ?`
+	_, err = o.Raw(sql, tableInfoId).QueryRows(&items)
+	return
+}

models/data_manage/table_info.go

@@ -0,0 +1,108 @@
+package data_manage
+
+import (
+	"github.com/rdlucklib/rdluck_tools/orm"
+	"time"
+)
+
+type TableInfo struct {
+	TableInfoId     int       `orm:"column(table_info_id);pk"`
+	TableName       string    `description:"表格名称"`
+	UniqueCode      string    `description:"表格唯一编码"`
+	TableClassifyId int       `description:"表格分类id"`
+	SysUserId       int       `description:"操作人id"`
+	SysUserRealName string    `description:"操作人真实姓名"`
+	StartDate       time.Time `description:"开始日期"`
+	EdbInfoIds      string    `description:"指标id"`
+	TableType       int       `description:"表格类型，1：指标+日期"`
+	TableImage      string    `description:"图表图片"`
+	Sort            int       `description:"排序字段，数字越小越排前面"`
+	EdbEndDate      time.Time `description:"指标最后更新日期"`
+	ModifyTime      time.Time
+	CreateTime      time.Time
+}
+
+type TableInfoItem struct {
+	TableInfoId     int    `description:"图表id"`
+	TableName       string `description:"图表名称"`
+	TableClassifyId int    `description:"图表分类"`
+}
+
+//图表检索数据
+type TableInfoSearch struct {
+	EdbCode   string `description:"图表编码"`
+	StartDate string `description:"起始日期"`
+	EndDate   string `description:"终止日期"`
+	DataList  []*EdbInfoSearchData
+}
+
+// GetTableInfoById 根据id获取表格详情
+func GetTableInfoById(tableInfoId int) (item *TableInfo, err error) {
+	o := orm.NewOrm()
+	o.Using("data")
+	sql := ` SELECT * FROM table_info WHERE table_info_id=? `
+	err = o.Raw(sql, tableInfoId).QueryRow(&item)
+	return
+}
+
+// GetAllTableInfoList 获取所有的 ETA表格信息
+func GetAllTableInfoList() (items []*TableInfo, err error) {
+	o := orm.NewOrm()
+	o.Using("data")
+	sql := ` select * from table_info order by table_info_id desc`
+	_, err = o.Raw(sql).QueryRows(&items)
+	return
+}
+
+func GetTableInfoByCondition(condition string, pars []interface{}) (item *TableInfo, err error) {
+	o := orm.NewOrm()
+	o.Using("data")
+	sql := ` SELECT * FROM table_info WHERE 1=1 `
+	if condition != "" {
+		sql += condition
+	}
+	err = o.Raw(sql, pars).QueryRow(&item)
+	return
+}
+
+func ModifyTableInfoModifyTime(tableInfoId int64) (err error) {
+	o := orm.NewOrm()
+	o.Using("data")
+	sql := ` UPDATE  table_info SET modify_time = NOW() WHERE table_info_id = ? `
+	_, err = o.Raw(sql, tableInfoId).Exec()
+	return
+}
+
+// TableDataList 表格数据
+type TableDataList struct {
+	Date      string `description:"指标日期"`
+	DataCol1  string `description:"第1个表格的数据"`
+	DataCol2  string `description:"第2个表格的数据"`
+	DataCol3  string `description:"第3个表格的数据"`
+	DataCol4  string `description:"第4个表格的数据"`
+	DataCol5  string `description:"第5个表格的数据"`
+	DataCol6  string `description:"第6个表格的数据"`
+	DataCol7  string `description:"第7个表格的数据"`
+	DataCol8  string `description:"第8个表格的数据"`
+	DataCol9  string `description:"第9个表格的数据"`
+	DataCol10 string `description:"第10个表格的数据"`
+	DataType  int8   `description:"数据类型，默认的区间数据是 1；插入数据是 2"`
+	Sort      int    `description:"排序字段，越小越靠前"`
+}
+
+// Update 更新图表基础信息
+func (tableInfo *TableInfo) Update(cols []string) (err error) {
+	o := orm.NewOrm()
+	o.Using("data")
+	_, err = o.Update(tableInfo, cols...)
+	return
+}
+
+// GetTableInfoByClassifyIdAndName 根据分类id和图表名获取图表信息
+func GetTableInfoByClassifyIdAndName(classifyId int, tableName string) (item *TableInfo, err error) {
+	o := orm.NewOrm()
+	o.Using("data")
+	sql := ` SELECT * FROM table_info WHERE table_classify_id = ? and table_name=? `
+	err = o.Raw(sql, classifyId, tableName).QueryRow(&item)
+	return
+}

models/db.go

@@ -95,3 +95,11 @@ func init() {
 		new(data_manage.EdbDataLt),               //路透指标数据表
 	)
 }
+
+func initEdb() {
+	orm.RegisterModel(
+		new(data_manage.TableData),
+		new(data_manage.TableEdbMapping),
+		new(data_manage.TableInfo),
+	)
+}

services/data/table_info.go

@@ -0,0 +1,126 @@
+package data
+
+import (
+	"context"
+	"fmt"
+	"hongze/hongze_task/models/data_manage"
+	"hongze/hongze_task/utils"
+	"strings"
+)
+
+// AppendDataToEdbTable 追加数据到ETA表格
+func AppendDataToEdbTable(cont context.Context) (err error) {
+	errMsgList := make([]string, 0)
+	defer func() {
+		if len(errMsgList) > 0 {
+			go utils.SendEmail(utils.APPNAME+"【"+utils.RunMode+"】"+"失败提醒", "追加数据到ETA表格; ErrMsg:"+strings.Join(errMsgList, "\n"), utils.EmailSendToUsers)
+		}
+	}()
+	tableInfoList, err := data_manage.GetAllTableInfoList()
+	if err != nil {
+		errMsgList = append(errMsgList, fmt.Sprint("获取ETA表格列表失败，Err:", err.Error()))
+		return
+	}
+
+	for _, tableInfo := range tableInfoList {
+		edbMappingList, tmpErr := data_manage.GetAllTableEdbMappingItemListByTableInfoId(tableInfo.TableInfoId)
+		if tmpErr != nil {
+			errMsgList = append(errMsgList, fmt.Sprint(tableInfo.TableInfoId, "获取关联关系失败，Err:", tmpErr.Error()))
+			continue
+		}
+
+		dataMap := make(map[string]map[string]float64)
+		dateList := make([]string, 0)
+		mappingMap := make(map[int]string) //指标map的最后一次更新时间
+		for _, v := range edbMappingList {
+			//如果最近更新eta表格数据的时间早于最新数据时间，那么需要查询出该数据并插入到eta表格中
+			if v.EdbDataEndDate.After(v.EndDate) {
+				dataList, tmpErr := data_manage.GetEdbDataListAll(" and edb_info_id=? and data_time > ?", []interface{}{v.EdbInfoId, v.EndDate.Format(utils.FormatDate)}, v.Source, 1)
+				if tmpErr != nil {
+					errMsg := fmt.Sprint(v.TableInfoId, "获取指标数据失败，Err:", tmpErr.Error())
+					errMsgList = append(errMsgList, errMsg)
+					continue
+				}
+				if len(dataList) <= 0 {
+					continue
+				}
+				for _, data := range dataList {
+					//指标map的最后一次更新时间
+					mappingMap[v.TableEdbMappingId] = data.DataTime
+
+					tmpDataMap, ok := dataMap[data.DataTime]
+					if !ok {
+						tmpDataMap = make(map[string]float64)
+						dateList = append(dateList, data.DataTime)
+					}
+					key := fmt.Sprint("data_col_", v.EdbDataColIndex)
+					tmpDataMap[key] = data.Value
+					dataMap[data.DataTime] = tmpDataMap
+				}
+			}
+		}
+
+		fmt.Println(dataMap)
+		fmt.Println(dateList)
+		for _, date := range dateList {
+			tmpErr = insertOrUpdateTableData(tableInfo.TableInfoId, date, dataMap[date])
+			if tmpErr != nil {
+				errMsgList = append(errMsgList, fmt.Sprint(tableInfo.TableInfoId, "插入更新", date, "数据失败，Err:", tmpErr.Error()))
+				continue
+			}
+		}
+
+		for tableEdbMappingId, date := range mappingMap {
+			tmpErr = updateTableEdbMappingEndDate(tableEdbMappingId, date)
+			if tmpErr != nil {
+				errMsgList = append(errMsgList, fmt.Sprint(tableEdbMappingId, "更新最后一次数据更新日期数据失败，Err:", tmpErr.Error()))
+				continue
+			}
+		}
+	}
+	return
+}
+
+// insertOrUpdateTableData 插入/更新ETA表格数据
+func insertOrUpdateTableData(tableInfoId int, date string, dataMap map[string]float64) (err error) {
+	item, err := data_manage.GetBetweenTableDataByTableInfoIdAndDate(tableInfoId, date)
+	if err != nil && err.Error() != utils.ErrNoRow() {
+		return
+	}
+	err = nil
+	sql := ``
+	//如果找不到数据，那么是插入数据
+	if item == nil {
+		sort, tmpErr := data_manage.GetMaxSortByTableInfoId(tableInfoId)
+		if tmpErr != nil {
+			err = tmpErr
+			return
+		}
+
+		columnStr := ``
+		valueStr := ``
+		for columnValue, value := range dataMap {
+			columnStr += columnValue + ","
+			valueStr += fmt.Sprint(value, ",")
+		}
+		sort = sort + 1
+		sql = fmt.Sprintf("INSERT INTO `table_data`( `table_info_id`, `date`, %s `data_type`,  `sort`, `modify_time`, `create_time`) VALUES ( %d, '%s', %s 1, %d, now(), now());", columnStr, tableInfoId, date, valueStr, sort)
+
+	} else {
+		updateStr := ``
+		for columnValue, value := range dataMap {
+			updateStr += fmt.Sprint(columnValue, "=", value, ",")
+		}
+		sql = fmt.Sprintf("UPDATE `table_data` SET %s `modify_time` = now() WHERE `table_data_id` = %d;", updateStr, item.TableDataId)
+	}
+	fmt.Println(sql)
+	err = data_manage.ExecTableDataSql(sql)
+	return
+}
+
+func updateTableEdbMappingEndDate(tableEdbMappingId int, endDate string) (err error) {
+	sql := fmt.Sprintf("UPDATE `table_edb_mapping` SET  `end_date` = '%s', `modify_time` = now() WHERE `table_edb_mapping_id` = %d;", endDate, tableEdbMappingId)
+	fmt.Println(sql)
+	err = data_manage.ExecTableDataSql(sql)
+	return
+}

services/task.go

@@ -85,6 +85,10 @@ func Task() {
 	modifyRsCalendarStatus := task.NewTask("modifyRsCalendarStatus", "0 */1 * * * * ", roadshow.ModifyRsCalendarResearcherStatus)
 	task.AddTask("modifyRsCalendarStatus", modifyRsCalendarStatus)
 
+	// 定时往追加数据到ETA表格（凌晨5点）
+	appendDataToEdbTable := task.NewTask("appendDataToEdbTable", "0 1 5 * * * ", data.AppendDataToEdbTable)
+	task.AddTask("定时往追加数据到ETA表格", appendDataToEdbTable)
+
 	task.StartTask()
 
 	fmt.Println("task end")