eta_api/models/data_manage/edb_data_coal.go

package data_manage

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

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

type BaseFromCoalDataSimple struct {
	Id        int `orm:"column(base_from_trade_coal_index_id);pk"`
	DealCode  string
	BuyCode   string
	SoldCode  string
	DataTime  string
	DealValue string
	BuyValue  string
	SoldValue string
}

type BaseInfoFromCoal struct {
	IndexName string
	Frequency string
	Unit      string
}

type BaseFromTradeCoalIndex struct {
	BaseFromTradeCoalIndexId int `orm:"column(base_from_trade_coal_index_id);pk"`
	Rank                     int
	DealShortName            string
	DealName                 string
	DealCode                 string
	DealValue                string
	DealChange               int
	BuyShortName             string
	BuyName                  string
	BuyCode                  string
	BuyValue                 string
	BuyChange                int
	SoldShortName            string
	SoldName                 string
	SoldCode                 string
	SoldValue                string
	SoldChange               int
	Frequency                string
	ClassifyName             string
	ClassifyType             string
	CreateTime               time.Time
	ModifyTime               time.Time
	DataTime                 string
}

func GetEdbDataCoalMaxOrMinDate(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_coal WHERE edb_code=? `
	err = o.Raw(sql, edbCode).QueryRow(&minDate, &maxDate)
	return
}

// RefreshEdbDataByCoal 刷新煤炭网指标数据
func RefreshEdbDataByCoal(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()
		}
	}()
	var suffix string
	if strings.Contains(edbCode, "jsm") {
		suffix = "jsm_index"
	} else if strings.Contains(edbCode, "company") {
		suffix = "company_index"
	} else if strings.Contains(edbCode, "firm") {
		suffix = "firm_index"
	} else if strings.Contains(edbCode, "coastal") {
		suffix = "coastal_index"
	} else if strings.Contains(edbCode, "inland") {
		suffix = "inland_index"
	}
	edbInfoIdStr := strconv.Itoa(edbInfoId)
	//计算数据
	var condition string
	var pars []interface{}

	if edbCode != "" {
		condition += " AND index_code=? "
		pars = append(pars, edbCode)
	}

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

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

	coalDataList, err := GetCoalDataByTradeCode(condition, suffix, pars)

	addSql := ` INSERT INTO edb_data_coal(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 coalDataList {
		item := v
		itemValue := v.DealValue
		if _, ok := existMap[v.DataTime]; !ok {
			count, err := GetEdbDataCoalByCodeAndDate(edbCode, v.DataTime)
			if err != nil && err.Error() != utils.ErrNoRow() {
				return err
			}
			if count <= 0 {
				eDate := item.DataTime
				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 = ModifyEdbDataCoal(int64(edbInfoId), v.DataTime, v.DealValue)
				if err != nil {
					return err
				}
			}
		}
		existMap[v.DataTime] = v.DealValue
	}
	if isAdd {
		addSql = strings.TrimRight(addSql, ",")
		_, err = to.Raw(addSql).Exec()
		if err != nil {
			return err
		}
	}
	return
}

// RefreshAllEdbDataByCoal 全部刷新中金所
func RefreshAllEdbDataByCoal(edbInfoId, source, subSource 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()
		}
	}()
	var suffix string
	if strings.Contains(edbCode, "jsm") {
		suffix = "jsm_index"
	} else if strings.Contains(edbCode, "company") {
		suffix = "company_index"
	} else if strings.Contains(edbCode, "firm") {
		suffix = "firm_index"
	} else if strings.Contains(edbCode, "coastal") {
		suffix = "coastal_index"
	} else if strings.Contains(edbCode, "inland") {
		suffix = "inland_index"
	}
	edbInfoIdStr := strconv.Itoa(edbInfoId)
	//计算数据
	var condition string
	var pars []interface{}

	if edbCode != "" {
		condition += " AND index_code=? "
		pars = append(pars, edbCode)
	}

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

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

	coalDataList, err := GetCoalDataByTradeCode(condition, suffix, pars)

	//获取指标所有数据
	dataList := make([]*EdbDataBase, 0)
	dataTableName := GetEdbDataTableName(source, subSource)
	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_coal(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 coalDataList {
		item := v
		itemValue := v.DealValue
		if _, ok := existMap[v.DataTime]; !ok {
			eDate := item.DataTime
			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.DataTime] = v.DataTime
	}
	if isAdd {
		addSql = strings.TrimRight(addSql, ",")
		_, err = to.Raw(addSql).Exec()
		if err != nil {
			return err
		}
	}
	return
}

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

func GetCoalDataByTradeCode(condition, suffix string, pars []interface{}) (item []*BaseFromCoalDataSimple, err error) {
	sql := ` SELECT * FROM base_from_coalmine_%s WHERE 1=1 `
	sql = fmt.Sprintf(sql, suffix)
	o := orm.NewOrmUsingDB("data")
	if condition != "" {
		sql += condition
	}
	sql += ` ORDER BY data_time DESC `
	_, err = o.Raw(sql, pars).QueryRows(&item)
	return
}

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

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