eta_api/models/data_manage/edb_data_insert_config.go

package data_manage

import (
	"errors"
	"eta_gn/eta_api/global"
	"eta_gn/eta_api/models/mgo"
	"eta_gn/eta_api/utils"
	"fmt"
	"go.mongodb.org/mongo-driver/bson"
	"go.mongodb.org/mongo-driver/mongo"
	"gorm.io/gorm"
	"strconv"
	"time"
)

type EdbDataInsertConfig struct {
	EdbInfoId  int       `orm:"column(edb_info_id);pk" gorm:"primaryKey;autoIncrement:false"  description:"指标id"`
	Date       time.Time `description:"插入的日期"`
	Value      string    `description:"插入的值"`
	RealDate   time.Time `description:"实际数据的值日期"`
	ModifyTime time.Time `description:"数据更新时间"`
	CreateTime time.Time `description:"数据插入的时间"`
}

func GetEdbDataInsertConfigByEdbId(edbInfoId int) (item *EdbDataInsertConfig, err error) {
	sql := ` SELECT * FROM edb_data_insert_config WHERE edb_info_id=? `
	err = global.DmSQL["data"].Raw(sql, edbInfoId).First(&item).Error

	return
}

func GetEdbDataInsertConfigByEdbIdList(edbInfoIdList []int) (items []*EdbDataInsertConfig, err error) {
	num := len(edbInfoIdList)
	if num <= 0 {
		return
	}

	o := global.DmSQL["data"]
	sql := ` SELECT * FROM edb_data_insert_config WHERE edb_info_id in (` + utils.GetOrmInReplace(num) + `) `
	err = o.Raw(sql, edbInfoIdList).Find(&items).Error

	return
}

func CreateEdbDataInsertConfigAndData(edbInfo *EdbInfo, date time.Time, value string) (err error, errMsg string, isSendEmail bool) {
	isSendEmail = true

	tableName := GetEdbDataTableName(edbInfo.Source, edbInfo.SubSource)
	if tableName == `` {
		err = errors.New("找不到该指标的数据表")
		return
	}

	dateStr := date.Format(utils.FormatDate)

	var saveValue string
	if value != "" {
		floatValue, e := strconv.ParseFloat(value, 64)
		if e != nil {
			err = e
			fmt.Println("转换失败:", e.Error())
			return
		}
		saveValue = utils.SubFloatToString(floatValue, 30)
	}

	to := global.DmSQL["data"].Begin()
	defer func() {
		if err != nil {
			_ = to.Rollback()
		} else {
			_ = to.Commit()
		}
	}()

	var oldConfigDate time.Time //之前配置的日期
	{
		var item *EdbDataInsertConfig
		sql := ` SELECT * FROM edb_data_insert_config WHERE edb_info_id=? `
		err = to.Raw(sql, edbInfo.EdbInfoId).First(&item).Error
		if err != nil && !utils.IsErrNoRow(err) {
			return
		}
		err = nil

		if item == nil || item.EdbInfoId <= 0 {
			var currLatestDate time.Time
			currLatestDateStr := edbInfo.LatestDate // 实际日期
			if currLatestDateStr != `` && currLatestDateStr != `0000-00-00` {
				currLatestDate, _ = time.ParseInLocation(utils.FormatDate, currLatestDateStr, time.Local)
			}
			if !currLatestDate.IsZero() && (currLatestDate.After(date) || currLatestDate.Equal(date)) {
				errMsg = `选择日期不能早于/等于数据最新日期`
				err = errors.New("选择日期不能早于/等于数据最新日期,数据最新日期：" + utils.TimeToFormatDate(edbInfo.EndDate) + ";填写日期：" + dateStr)
				isSendEmail = false
				return
			}

			realDate, _ := time.ParseInLocation(utils.FormatDate, edbInfo.LatestDate, time.Local)
			item = &EdbDataInsertConfig{
				EdbInfoId:  edbInfo.EdbInfoId,
				Date:       date,
				Value:      saveValue,
				RealDate:   realDate,
				ModifyTime: time.Now(),
				CreateTime: time.Now(),
			}
			err = to.Create(item).Error
		} else {
			if date.Equal(item.RealDate) || date.Before(item.RealDate) {
				errMsg = `选择日期不能早于/等于数据最新日期`
				err = errors.New("选择日期不能早于/等于数据最新日期,数据最新日期：" + utils.TimeToFormatDate(edbInfo.EndDate) + ";填写日期：" + dateStr)
				isSendEmail = false
				return
			}
			oldConfigDate = item.Date // 之前配置的日期

			item.Date = date
			item.Value = saveValue
			item.ModifyTime = time.Now()
			err = to.Select("Date", "Value", "ModifyTime").Updates(item).Error
		}
		if err != nil {
			return
		}
	}

	if edbInfo.Source == utils.DATA_SOURCE_BUSINESS && utils.UseMongo {
		dateStr, err = updateInsertConfigValueByMongo(to, edbInfo, oldConfigDate, date, value)
	} else if edbInfo.Source == utils.DATA_SOURCE_THS && edbInfo.SubSource == utils.DATA_SUB_SOURCE_HIGH_FREQUENCY && utils.UseMongo {
		dateStr, err = updateThsHfInsertConfigValueByMongo(to, edbInfo, oldConfigDate, date, value)
	} else {
		dateStr, err = updateInsertConfigValueByMysql(to, edbInfo, oldConfigDate, date, value)
	}
	if err != nil {
		return
	}

	if dateStr != `` {
		endDateTime, tmpErr := utils.FormatDateStrToTime(dateStr)
		if tmpErr != nil {
			err = tmpErr
			return
		}
		edbInfo.EndDate = endDateTime
		err = to.Select("EndDate").Updates(edbInfo).Error
	}

	return
}

func updateInsertConfigValueByMysql(to *gorm.DB, edbInfo *EdbInfo, oldConfigDate, date time.Time, value string) (dateStr string, err error) {
	tableName := GetEdbDataTableName(edbInfo.Source, edbInfo.SubSource)
	if tableName == `` {
		err = errors.New("找不到该指标的数据表")
		return
	}

	dateStr = date.Format(utils.FormatDate)
	timestamp := date.UnixNano() / 1e6
	timeStr := fmt.Sprintf("%d", timestamp)
	var saveValue string
	if value != "" {
		floatValue, e := strconv.ParseFloat(value, 64)
		if e != nil {
			err = e
			fmt.Println("转换失败:", e.Error())
			return
		}
		saveValue = utils.SubFloatToString(floatValue, 30)
	}

	var offsetDate string
	{
		var edbDateData *EdbDataList
		if !oldConfigDate.IsZero() {
			sql := `SELECT edb_data_id,edb_info_id,data_time,"value",data_timestamp FROM %s WHERE edb_info_id=? AND data_time = ?`
			sql = fmt.Sprintf(sql, tableName)
			err = to.Raw(sql, edbInfo.EdbInfoId, oldConfigDate.Format(utils.FormatDate)).Scan(&edbDateData).Error
			if err != nil && !utils.IsErrNoRow(err) {
				return
			}
			err = nil
		}

		if edbDateData == nil {
			addSql := ` INSERT INTO %s (edb_info_id,edb_code,data_time,value,create_time,modify_time,data_timestamp) VALUES ( %d, '%s', '%s', '%s', now(), now(), %s) `
			addSql = fmt.Sprintf(addSql, tableName, edbInfo.EdbInfoId, edbInfo.EdbCode, dateStr, saveValue, timeStr)
			err = to.Exec(addSql).Error
		} else if value == "" {
			sql := `SELECT data_time FROM %s WHERE edb_info_id=? ORDER BY data_time DESC LIMIT 1 OFFSET 1`
			sql = fmt.Sprintf(sql, tableName)
			err = to.Raw(sql, edbInfo.EdbInfoId).Scan(&offsetDate).Error
			if err != nil && !utils.IsErrNoRow(err) {
				return
			}

			dateStr = offsetDate

			deleteSql := `DELETE FROM %s WHERE edb_data_id = %d;`
			deleteSql = fmt.Sprintf(deleteSql, tableName, edbDateData.EdbDataId)
			err = to.Exec(deleteSql).Error
		} else {
			updateSql := `UPDATE %s SET data_time = '%s', value = '%s', modify_time= now(), data_timestamp= %s WHERE edb_data_id = %d;`
			updateSql = fmt.Sprintf(updateSql, tableName, dateStr, saveValue, timeStr, edbDateData.EdbDataId)
			err = to.Exec(updateSql).Error
		}
		if err != nil {
			return
		}
	}

	return
}

func updateInsertConfigValueByMongo(to *gorm.DB, edbInfo *EdbInfo, oldConfigDate, newDate time.Time, value string) (dateStr string, err error) {
	tableName := GetEdbDataTableName(edbInfo.Source, edbInfo.SubSource)
	if tableName == `` {
		err = errors.New("找不到该指标的数据表")
		return
	}

	dateStr = newDate.Format(utils.FormatDate)
	timestamp := newDate.UnixNano() / 1e6
	var floatValue float64
	if value != "" {
		floatValue, err = strconv.ParseFloat(value, 64)
		if err != nil {
			fmt.Println("转换失败:", err.Error())
			return
		}
	}

	mogDataObj := mgo.EdbDataBusiness{}
	coll := mogDataObj.GetCollection()

	var edbDateData *mgo.EdbDataBusiness
	if !oldConfigDate.IsZero() {
		queryConditions := bson.M{
			"edb_info_id": edbInfo.EdbInfoId,
			"data_time":   oldConfigDate,
		}

		edbDateData, err = mogDataObj.GetItem(queryConditions)
		if err != nil && err != mongo.ErrNoDocuments {
			return
		}
		err = nil

	}

	if edbDateData == nil {
		addDataItem := mgo.EdbDataBusiness{
			EdbInfoId:     edbInfo.EdbInfoId,
			EdbCode:       edbInfo.EdbCode,
			DataTime:      newDate,
			Value:         floatValue,
			CreateTime:    time.Now(),
			ModifyTime:    time.Now(),
			DataTimestamp: timestamp,
		}
		err = mogDataObj.InsertDataByColl(coll, addDataItem)
		if err != nil {
			fmt.Println("mogDataObj.BatchInsertData() Err:" + err.Error())
			return
		}
		return
	}

	if value == "" {
		queryConditions := bson.M{
			"edb_info_id": edbInfo.EdbInfoId,
		}

		tmpDataList, tmpErr := mogDataObj.GetLimitDataList(queryConditions, 2, []string{"-data_time"})
		if tmpErr != nil {
			fmt.Println("mogDataObj.GetLimitDataList() Err:" + tmpErr.Error())
			return
		}
		if len(tmpDataList) < 2 {
			return
		}

		dateStr = tmpDataList[1].DataTime.Format(utils.FormatDate)

		err = mogDataObj.RemoveManyByColl(coll, bson.M{"_id": tmpDataList[0].ID})

		return
	}

	updateData := bson.M{"$set": bson.M{
		"value":          floatValue,
		"modify_time":    time.Now(),
		"data_time":      newDate,
		"data_timestamp": timestamp,
	}}
	err = mogDataObj.UpdateDataByColl(coll, bson.M{"_id": edbDateData.ID}, updateData)

	return
}

func updateThsHfInsertConfigValueByMongo(to *gorm.DB, edbInfo *EdbInfo, oldConfigDate, newDate time.Time, value string) (dateStr string, err error) {
	tableName := GetEdbDataTableName(edbInfo.Source, edbInfo.SubSource)
	if tableName == `` {
		err = errors.New("找不到该指标的数据表")
		return
	}

	dateStr = newDate.Format(utils.FormatDate)
	timestamp := newDate.UnixNano() / 1e6
	var floatValue float64
	if value != "" {
		floatValue, err = strconv.ParseFloat(value, 64)
		if err != nil {
			fmt.Println("转换失败:", err.Error())
			return
		}
	}

	mogDataObj := mgo.EdbDataThsHf{}
	coll := mogDataObj.GetCollection()

	var edbDateData *mgo.EdbDataThsHf
	if !oldConfigDate.IsZero() {
		queryConditions := bson.M{
			"edb_info_id": edbInfo.EdbInfoId,
			"data_time":   oldConfigDate,
		}

		edbDateData, err = mogDataObj.GetItem(queryConditions)
		if err != nil && err != mongo.ErrNoDocuments {
			return
		}
		err = nil

	}

	if edbDateData == nil {
		addDataItem := mgo.EdbDataThsHf{
			EdbInfoId:     edbInfo.EdbInfoId,
			EdbCode:       edbInfo.EdbCode,
			DataTime:      newDate,
			Value:         floatValue,
			CreateTime:    time.Now(),
			ModifyTime:    time.Now(),
			DataTimestamp: timestamp,
		}
		err = mogDataObj.InsertDataByColl(coll, addDataItem)
		if err != nil {
			fmt.Println("mogDataObj.BatchInsertData() Err:" + err.Error())
			return
		}
		return
	}

	if value == "" {
		queryConditions := bson.M{
			"edb_info_id": edbInfo.EdbInfoId,
		}

		tmpDataList, tmpErr := mogDataObj.GetLimitDataList(queryConditions, 2, []string{"-data_time"})
		if tmpErr != nil {
			fmt.Println("mogDataObj.GetLimitDataList() Err:" + tmpErr.Error())
			return
		}
		if len(tmpDataList) < 2 {
			return
		}

		dateStr = tmpDataList[1].DataTime.Format(utils.FormatDate)

		err = mogDataObj.RemoveManyByColl(coll, bson.M{"_id": tmpDataList[0].ID})

		return
	}

	updateData := bson.M{"$set": bson.M{
		"value":          floatValue,
		"modify_time":    time.Now(),
		"data_time":      newDate,
		"data_timestamp": timestamp,
	}}
	err = mogDataObj.UpdateDataByColl(coll, bson.M{"_id": edbDateData.ID}, updateData)

	return
}