hongze_task/models/data_manage/edb_data_dl.go

package data_manage

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

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

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

type BaseInfoFromDl struct {
	DealName string
	BuyName  string
	SoldName string
}

type BaseFromTradeDalianIndex struct {
	BaseFromTradeDalianIndexId int `orm:"column(base_from_trade_dalian_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 GetEdbDataDlMaxOrMinDate(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_dl WHERE edb_code=? `
	err = o.Raw(sql, edbCode).QueryRow(&minDate, &maxDate)
	return
}


// RefreshEdbDataByDl 刷新大商所指标数据
func RefreshEdbDataByDl(edbInfoId int, edbCode, startDate, endDate string) (err error) {
	o := orm.NewOrmUsingDB("data")
	tx,err:=o.Begin()
	if err!=nil {
		return err
	}
	defer func() {
		if err != nil {
			tx.Rollback()
		} else {
			tx.Commit()
		}
	}()
	if err != nil {
		return
	}
	var suffix string
	if strings.Contains(edbCode, "deal") {
		suffix = "deal"
	} else if strings.Contains(edbCode, "buy") {
		suffix = "buy"
	} else if strings.Contains(edbCode, "sold") {
		suffix = "sold"
	}
	edbInfoIdStr := strconv.Itoa(edbInfoId)
	//计算数据
	var condition string
	var pars []interface{}

	if edbCode != "" {
		if suffix == "deal" {
			condition += " AND deal_code=? "
		} else if suffix == "buy" {
			condition += " AND buy_code=? "
		} else {
			condition += " AND sold_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)
	}

	glDataList, err := GetDlDataByTradeCode(condition, pars)

	addSql := ` INSERT INTO edb_data_dl(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 glDataList {
		var value string
		if suffix == "deal" {
			value = v.DealValue
		} else if suffix == "buy" {
			value = v.BuyValue
		} else {
			value = v.SoldValue
		}
		item := v
		itemValue := value
		if _, ok := existMap[v.DataTime]; !ok {
			count, err := GetEdbDataDlByCodeAndDate(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 = ModifyEdbDataDl(int64(edbInfoId), v.DataTime, value)
				if err != nil {
					return err
				}
			}
		}
		existMap[v.DataTime] = value
	}
	if isAdd {
		addSql = strings.TrimRight(addSql, ",")
		_, err = tx.Raw(addSql).Exec()
		if err != nil {
			return err
		}
	}
	return
}

// RefreshAllEdbDataByDl 全部刷新大商所
func RefreshAllEdbDataByDl(edbInfoId, source int, edbCode, startDate, endDate string) (err error) {
	o := orm.NewOrmUsingDB("data")
	tx,err:=o.Begin()
	if err!=nil {
		return err
	}
	defer func() {
		if err != nil {
			tx.Rollback()
		} else {
			tx.Commit()
		}
	}()
	if err != nil {
		return
	}
	var suffix string
	if strings.Contains(edbCode, "deal") {
		suffix = "deal"
	} else if strings.Contains(edbCode, "buy") {
		suffix = "buy"
	} else if strings.Contains(edbCode, "sold") {
		suffix = "sold"
	}
	edbInfoIdStr := strconv.Itoa(edbInfoId)
	//计算数据
	var condition string
	var pars []interface{}

	if edbCode != "" {
		if suffix == "deal" {
			condition += " AND deal_code=? "
		} else if suffix == "buy" {
			condition += " AND buy_code=? "
		} else {
			condition += " AND sold_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)
	}

	glDataList, err := GetDlDataByTradeCode(condition, pars)

	//获取指标所有数据
	dataList := make([]*EdbDataBase, 0)
	dataTableName := GetEdbDataTableName(source)
	sql := `SELECT * FROM %s WHERE edb_info_id=? `
	sql = fmt.Sprintf(sql, dataTableName)
	_, err = o.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_dl(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 glDataList {
		var value string
		if suffix == "deal" {
			value = v.DealValue
		} else if suffix == "buy" {
			value = v.BuyValue
		} else {
			value = v.SoldValue
		}
		item := v
		itemValue := value
		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 = tx.Raw(sql, sValue, edbInfoId, eDate).Exec()
						if err != nil {
							return err
						}
					}
				}
			}
		}
		existMap[v.DataTime] = v.DataTime
	}
	if isAdd {
		addSql = strings.TrimRight(addSql, ",")
		_, err = tx.Raw(addSql).Exec()
		if err != nil {
			return err
		}
	}
	return
}

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

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

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


func GetBaseFromDalianDataAllByIndexCode(indexCode, suffix string) (list []*BaseFromTradeDalianIndex, err error) {
	o := orm.NewOrmUsingDB("data")
	sql := `SELECT * FROM base_from_trade_dalian_index WHERE %s_code=? `
	sql = fmt.Sprintf(sql, suffix)
	_, err = o.Raw(sql, indexCode).QueryRows(&list)
	return
}

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

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