hz_crm_api/models/data_manage/edb_data_gie.go

package data_manage

import (
	"fmt"
	"github.com/beego/beego/v2/client/orm"
	"hongze/hz_crm_api/utils"
	"strconv"
	"strings"
	"time"
)

type GieData struct {
	InputValue string `orm:"column(DATA_VALUE)" description:"日期"`
	DataTime   string `orm:"column(DATA_DATE)" description:"值"`
}

func GetEdbDataGieMaxOrMinDate(edbCode string) (minDate, maxDate string, err error) {
	o := orm.NewOrmUsingDB("data")
	sql := ` SELECT MIN(data_time) AS minDate,MAX(data_time) AS maxDate FROM edb_data_gie WHERE edb_code=? `
	err = o.Raw(sql, edbCode).QueryRow(&minDate, &maxDate)
	return
}

func GetEdbDataByGie(edbCode, suffix, startDate, endDate string) (searchItem *EdbInfoSearch, err error) {
	o := orm.NewOrmUsingDB("data")
	to, err := o.Begin()
	if err != nil {
		return
	}
	searchItem = new(EdbInfoSearch)
	searchItem.EdbCode = edbCode
	eicBaseDataAll, err := GetBaseFromEicDataAllByIndexCode(edbCode, suffix)
	if err != nil && err.Error() != utils.ErrNoRow() {
		fmt.Println("GetBaseFromEicDataAllByIndexCode err:", err)
		return
	}

	var isAdd bool
	addSql := ` INSERT INTO edb_data_gie(edb_info_id,edb_code,data_time,value,create_time,modify_time,status,data_timestamp) values `
	dataList := make([]*EdbInfoSearchData, 0)
	existMap := make(map[string]string)

	for _, sv := range eicBaseDataAll {
		eDate := sv.GasDayStartedOn
		dataTime, err := time.Parse(utils.FormatDate, eDate)
		if err != nil {
			fmt.Println("time.Parse Err:" + eDate)
			return nil, err
		}
		timestamp := dataTime.UnixNano() / 1e6
		timeStr := fmt.Sprintf("%d", timestamp)
		//var name string
		if _, ok := existMap[eDate]; !ok {
			if suffix == "GS" {
				//name = "gas_in_storage"
				addSql += GetAddSql("0", edbCode, eDate, timeStr, sv.GasInStorage)
				existMap[eDate] = sv.GasInStorage
			} else if suffix == "F" {
				//name = "full"
				addSql += GetAddSql("0", edbCode, eDate, timeStr, sv.Full)
				existMap[eDate] = sv.Full
			} else if suffix == "T" {
				//name = "trend"
				addSql += GetAddSql("0", edbCode, eDate, timeStr, sv.Trend)
				existMap[eDate] = sv.Trend
			} else if suffix == "In" {
				//name = "injection"
				addSql += GetAddSql("0", edbCode, eDate, timeStr, sv.Injection)
				existMap[eDate] = sv.Injection
			} else if suffix == "Out" {
				//name = "withdrawal"
				addSql += GetAddSql("0", edbCode, eDate, timeStr, sv.Withdrawal)
				existMap[eDate] = sv.Withdrawal
			} else if suffix == "WGV" {
				//name = "working_gas_volume"
				addSql += GetAddSql("0", edbCode, eDate, timeStr, sv.WorkingGasVolume)
				existMap[eDate] = sv.WorkingGasVolume
			} else if suffix == "IC" {
				//name = "injection_capacity"
				addSql += GetAddSql("0", edbCode, eDate, timeStr, sv.InjectionCapacity)
				existMap[eDate] = sv.InjectionCapacity
			} else if suffix == "WC" {
				//name = "withdrawal_capacity"
				addSql += GetAddSql("0", edbCode, eDate, timeStr, sv.WithdrawalCapacity)
				existMap[eDate] = sv.WithdrawalCapacity
			}
			isAdd = true
		}
	}
	if isAdd {
		addSql = strings.TrimRight(addSql, ",")
		utils.FileLog.Info("addSql:" + addSql)
		_, err = to.Raw(addSql).Exec()
		if err != nil {
			fmt.Println("addSql err:", err)
			return searchItem, err
		}
	}
	if err != nil {
		_ = to.Rollback()
	} else {
		_ = to.Commit()
	}
	size := utils.EDB_DATA_LIMIT
	dataList, err = GetEdbDataAllByEdbCode(edbCode, utils.DATA_SOURCE_GIE, size)
	if err != nil {
		utils.FileLogData.Info("GetEdbDataGieByCode Err:%s", err.Error())
		return searchItem, err
	}
	minDate, maxDate, err := GetEdbDataGieMaxOrMinDate(edbCode)
	if err != nil {
		return searchItem, err
	}
	searchItem.DataList = dataList
	searchItem.StartDate = minDate
	searchItem.EndDate = maxDate
	if searchItem.DataList == nil {
		searchItem.DataList = make([]*EdbInfoSearchData, 0)
	}
	return
}

// RefreshEdbDataByGie 刷新欧洲天然气指标数据
func RefreshEdbDataByGie(edbInfoId int, edbCode, startDate, endDate string) (err error) {
	o := orm.NewOrmUsingDB("data")
	to, err := o.Begin()
	if err != nil {
		return
	}
	defer func() {
		if err != nil {
			_ = to.Rollback()
		} else {
			_ = to.Commit()
		}
	}()
	if err != nil {
		fmt.Println("refresh err:", err)
		return
	}
	var suffix string
	l := len(edbCode)
	if strings.Contains(edbCode[l-2:], "GS") {
		suffix = "GS"
	} else if strings.Contains(edbCode[l-2:], "CF") {
		suffix = "CF"
	} else if strings.Contains(edbCode[l-1:], "T") {
		suffix = "T"
	} else if strings.Contains(edbCode[l-2:], "In") {
		suffix = "In"
	} else if strings.Contains(edbCode[l-3:], "Out") {
		suffix = "Out"
	} else if strings.Contains(edbCode[l-3:], "WGV") {
		suffix = "WGV"
	} else if strings.Contains(edbCode[l-2:], "IC") {
		suffix = "IC"
	} else if strings.Contains(edbCode[l-2:], "WC") {
		suffix = "WC"
	} else if strings.Contains(edbCode[l-1:], "F") {
		suffix = "F"
	} else if strings.Contains(edbCode[l-1:], "C") {
		suffix = "C"
	} else {
		suffix = ""
	}
	edbInfoIdStr := strconv.Itoa(edbInfoId)
	//计算数据
	var condition string
	var pars []interface{}

	if edbCode != "" {
		condition += " AND eic_code=? "
		pars = append(pars, edbCode[:l-len(suffix)])
	}

	if startDate != "" {
		condition += " AND gas_day_started_on>=? "
		pars = append(pars, startDate)
	}

	if endDate != "" {
		condition += " AND gas_day_started_on<=? "
		pars = append(pars, endDate)
	}

	eicDataList, err := GetGieDataByTradeCodeV2(condition, pars)
	fmt.Println("eicDataList", len(eicDataList))
	addSql := ` INSERT INTO edb_data_gie(edb_info_id,edb_code,data_time,value,create_time,modify_time,status,data_timestamp) values `
	var isAdd bool
	existMap := make(map[string]string)
	for _, v := range eicDataList {
		var value string
		if suffix == "GS" {
			value = v.GasInStorage
		} else if suffix == "C" {
			value = v.Consumption
		} else if suffix == "CF" {
			value = v.ConsumptionFull
		} else if suffix == "F" {
			value = v.Full
		} else if suffix == "T" {
			value = v.Trend
		} else if suffix == "In" {
			value = v.Injection
		} else if suffix == "Out" {
			value = v.Withdrawal
		} else if suffix == "WGV" {
			value = v.WorkingGasVolume
		} else if suffix == "IC" {
			value = v.InjectionCapacity
		} else if suffix == "WC" {
			value = v.WithdrawalCapacity
		}
		item := v
		itemValue := value
		if _, ok := existMap[v.GasDayStart]; !ok {
			count, err := GetEdbDataGieByCodeAndDate(edbCode, v.GasDayStart)
			if err != nil && err.Error() != utils.ErrNoRow() {
				return err
			}
			if count <= 0 {
				eDate := item.GasDayStart
				sValue := itemValue
				if sValue != "" {
					dataTime, err := time.Parse(utils.FormatDate, eDate)
					if err != nil {
						return err
					}
					timestamp := dataTime.UnixNano() / 1e6
					timeStr := fmt.Sprintf("%d", timestamp)
					addSql += GetAddSql(edbInfoIdStr, edbCode, eDate, timeStr, sValue)
					isAdd = true
				}
			} else {
				err = ModifyEdbDataGie(int64(edbInfoId), v.GasDayStart, value)
				if err != nil {
					return err
				}
			}
		}
		existMap[v.GasDayStart] = value
	}
	if isAdd {
		addSql = strings.TrimRight(addSql, ",")
		_, err = to.Raw(addSql).Exec()
		if err != nil {
			return err
		}
	}
	return
}

// RefreshAllEdbDataByGie 全部刷新欧洲天然气
func RefreshAllEdbDataByGie(edbInfoId, source int, edbCode, startDate, endDate string) (err error) {
	o := orm.NewOrmUsingDB("data")
	to, err := o.Begin()
	if err != nil {
		return
	}
	defer func() {
		if err != nil {
			_ = to.Rollback()
		} else {
			_ = to.Commit()
		}
	}()
	if err != nil {
		return
	}
	var suffix string
	l := len(edbCode)
	if strings.Contains(edbCode[l-2:], "GS") {
		suffix = "GS"
	} else if strings.Contains(edbCode[l-2:], "CF") {
		suffix = "CF"
	} else if strings.Contains(edbCode[l-1:], "T") {
		suffix = "T"
	} else if strings.Contains(edbCode[l-2:], "In") {
		suffix = "In"
	} else if strings.Contains(edbCode[l-3:], "Out") {
		suffix = "Out"
	} else if strings.Contains(edbCode[l-3:], "WGV") {
		suffix = "WGV"
	} else if strings.Contains(edbCode[l-2:], "IC") {
		suffix = "IC"
	} else if strings.Contains(edbCode[l-2:], "WC") {
		suffix = "WC"
	} else if strings.Contains(edbCode[l-1:], "F") {
		suffix = "F"
	} else if strings.Contains(edbCode[l-1:], "C") {
		suffix = "C"
	} else {
		suffix = ""
	}
	edbInfoIdStr := strconv.Itoa(edbInfoId)
	//计算数据
	var condition string
	var pars []interface{}

	if edbCode != "" {
		condition += " AND eic_code=? "
		pars = append(pars, edbCode[:l-len(suffix)])
	}

	if startDate != "" {
		condition += " AND gas_day_start>=? "
		pars = append(pars, startDate)
	}

	if endDate != "" {
		condition += " AND gas_day_start<=? "
		pars = append(pars, endDate)
	}

	eicDataList, err := GetGieDataByTradeCodeV2(condition, pars)
	fmt.Println("all eicDataList", len(eicDataList))
	//获取指标所有数据
	dataList := make([]*EdbDataBase, 0)
	dataTableName := GetEdbDataTableName(source)
	sql := `SELECT * FROM %s WHERE edb_info_id=? `
	sql = fmt.Sprintf(sql, dataTableName)
	_, err = to.Raw(sql, edbInfoId).QueryRows(&dataList)
	if err != nil {
		return err
	}
	dataMap := make(map[string]string)
	for _, v := range dataList {
		dataMap[v.DataTime] = v.Value
	}

	addSql := ` INSERT INTO edb_data_gie(edb_info_id,edb_code,data_time,value,create_time,modify_time,status,data_timestamp) values `
	var isAdd bool
	existMap := make(map[string]string)
	for _, v := range eicDataList {
		var value string
		if suffix == "GS" {
			value = v.GasInStorage
		} else if suffix == "C" {
			value = v.Consumption
		} else if suffix == "CF" {
			value = v.ConsumptionFull
		} else if suffix == "F" {
			value = v.Full
		} else if suffix == "T" {
			value = v.Trend
		} else if suffix == "In" {
			value = v.Injection
		} else if suffix == "Out" {
			value = v.Withdrawal
		} else if suffix == "WGV" {
			value = v.WorkingGasVolume
		} else if suffix == "IC" {
			value = v.InjectionCapacity
		} else if suffix == "WC" {
			value = v.WithdrawalCapacity
		}
		item := v
		itemValue := value
		if _, ok := existMap[v.GasDayStart]; !ok {
			eDate := item.GasDayStart
			sValue := itemValue
			if sValue != "" {
				dataTime, err := time.Parse(utils.FormatDate, eDate)
				if err != nil {
					return err
				}
				timestamp := dataTime.UnixNano() / 1e6
				timeStr := fmt.Sprintf("%d", timestamp)
				saveValue := sValue

				if existVal, ok := dataMap[eDate]; !ok {
					addSql += GetAddSql(edbInfoIdStr, edbCode, eDate, timeStr, saveValue)
					isAdd = true
				} else {
					if existVal != saveValue {
						sql := ` UPDATE %s SET value=?,modify_time=NOW() WHERE edb_info_id=? AND data_time=? `
						sql = fmt.Sprintf(sql, dataTableName)
						_, err = to.Raw(sql, sValue, edbInfoId, eDate).Exec()
						if err != nil {
							return err
						}
					}
				}
			}
		}
		existMap[v.GasDayStart] = v.GasDayStart
	}
	if isAdd {
		addSql = strings.TrimRight(addSql, ",")
		_, err = to.Raw(addSql).Exec()
		if err != nil {
			return err
		}
	}
	return
}

// GetBaseInfoFromEicByIndexCode 获取指标信息
func GetBaseInfoFromEicByIndexCode(indexCode, suffix string) (list []*BaseFromTradeEicIndex, err error) {
	o := orm.NewOrmUsingDB("data")
	sql := `SELECT * FROM base_from_trade_eic_index WHERE %s_code=? `
	sql = fmt.Sprintf(sql, suffix)
	_, err = o.Raw(sql, indexCode).QueryRows(&list)
	return
}

func GetGieDataByTradeCode(condition string, pars []interface{}) (item []*BaseFromTradeEicIndex, err error) {
	sql := ` SELECT * FROM base_from_trade_eic_index WHERE 1=1 `
	o := orm.NewOrmUsingDB("data")
	if condition != "" {
		sql += condition
	}
	sql += ` ORDER BY gas_day_started_on DESC `
	fmt.Println(sql, pars)
	_, err = o.Raw(sql, pars).QueryRows(&item)
	return
}

type EicIndexV2 struct {
	BaseFromEicIndexId     int `orm:"column(base_from_eic_index_id);pk"`
	Type                   string
	EicCode                string
	Name                   string
	Status                 string
	GasDayStart            string
	GasInStorage           string
	GasInStorageCode       string
	Consumption            string
	ConsumptionCode        string
	ConsumptionFull        string
	ConsumptionFullCode    string
	Full                   string
	FullCode               string
	Trend                  string
	TrendCode              string
	Injection              string
	InjectionCode          string
	Withdrawal             string
	WithdrawalCode         string
	WorkingGasVolume       string
	WorkingGasVolumeCode   string
	InjectionCapacity      string
	InjectionCapacityCode  string
	WithdrawalCapacity     string
	WithdrawalCapacityCode string
	Info                   string
	Parent                 string
	CreateTime             time.Time
	ModifyTime             time.Time
	Children               []BaseFromTradeEicIndexV2
}

func GetGieDataByTradeCodeV2(condition string, pars []interface{}) (item []*EicIndexV2, err error) {
	sql := ` SELECT * FROM base_from_trade_eic_index_v2 WHERE 1=1 `
	o := orm.NewOrmUsingDB("data")
	if condition != "" {
		sql += condition
	}
	sql += ` ORDER BY gas_day_start DESC `
	fmt.Println(sql, pars)
	_, err = o.Raw(sql, pars).QueryRows(&item)
	return
}

func AddEdbDataGieBySql(sqlStr string) (err error) {
	o := orm.NewOrmUsingDB("data")
	_, err = o.Raw(sqlStr).Exec()
	return
}

func GetEdbDataGieByCode(edbCode string) (items []*EdbInfoSearchData, err error) {
	o := orm.NewOrmUsingDB("data")
	sql := ` SELECT * FROM edb_data_gie WHERE edb_code=? ORDER BY data_time DESC LIMIT ? `
	_, err = o.Raw(sql, edbCode, utils.EDB_DATA_LIMIT).QueryRows(&items)
	return
}

func GetBaseFromEicDataAllByIndexCode(indexCode, suffix string) (list []*BaseFromTradeEicIndex, err error) {
	o := orm.NewOrmUsingDB("data")
	var name string
	if suffix == "" {
		name = "eic_code"
	} else if suffix == "GS" {
		name = "gas_in_storage_code"
	} else if suffix == "C" {
		name = "consumption_code"
	} else if suffix == "CF" {
		name = "consumption_full_code"
	} else if suffix == "F" {
		name = "full_code"
	} else if suffix == "T" {
		name = "trend_code"
	} else if suffix == "In" {
		name = "injection_code"
	} else if suffix == "Out" {
		name = "withdrawal_code"
	} else if suffix == "WGV" {
		name = "working_gas_volume_code"
	} else if suffix == "IC" {
		name = "injection_capacity_code"
	} else if suffix == "WC" {
		name = "withdrawal_capacity_code"
	}
	sql := `SELECT * FROM base_from_trade_eic_index_v2 WHERE %s=? `
	sql = fmt.Sprintf(sql, name)
	_, err = o.Raw(sql, indexCode).QueryRows(&list)
	return
}

func GetBaseFromEicDataAllByIndexCodeV2(indexCode, suffix string) (list []*BaseFromTradeEicIndexV2, err error) {
	o := orm.NewOrmUsingDB("data")
	var name string
	if suffix == "" {
		name = "eic_code"
	} else if suffix == "GS" {
		name = "gas_in_storage_code"
	} else if suffix == "C" {
		name = "consumption_code"
	} else if suffix == "CF" {
		name = "consumption_full_code"
	} else if suffix == "F" {
		name = "full_code"
	} else if suffix == "T" {
		name = "trend_code"
	} else if suffix == "In" {
		name = "injection_code"
	} else if suffix == "Out" {
		name = "withdrawal_code"
	} else if suffix == "WGV" {
		name = "working_gas_volume_code"
	} else if suffix == "IC" {
		name = "injection_capacity_code"
	} else if suffix == "WC" {
		name = "withdrawal_capacity_code"
	}
	sql := `SELECT * FROM base_from_trade_eic_index_v2 WHERE %s=? `
	sql = fmt.Sprintf(sql, name)
	_, err = o.Raw(sql, indexCode).QueryRows(&list)
	return
}

func GetEdbDataGieByCodeAndDate(edbCode string, startDate string) (count int, err error) {
	o := orm.NewOrmUsingDB("data")
	sql := ` SELECT COUNT(1) AS count FROM edb_data_gie WHERE edb_code=? AND data_time=? `
	err = o.Raw(sql, edbCode, startDate).QueryRow(&count)
	return
}

func ModifyEdbDataGie(edbInfoId int64, dataTime, value string) (err error) {
	o := orm.NewOrmUsingDB("data")
	sql := ` UPDATE edb_data_gie SET value=?,modify_time=NOW() WHERE edb_info_id=? AND data_time=? `
	_, err = o.Raw(sql, value, edbInfoId, dataTime).Exec()
	return
}