eta_index_lib/models/base_from_wind_wsd.go

package models

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

var windWsd = "wsd"

func AddEdbDataFromWindWsd(stockCode string, item map[string]map[string]interface{}) (err error) {
	o := orm.NewOrm()

	var isAdd bool
	addSql := ` INSERT INTO edb_data_wind_wsd(edb_info_id,edb_code,data_time,value,create_time,modify_time,data_timestamp) values `

	indexCodeMap := make(map[string]string)
	for wk, wv := range item {
		for vk, vv := range wv {
			if vv == nil {
				continue
			}

			var indexCode string
			wkInt, err := strconv.ParseInt(wk, 10, 64)
			if err != nil {
				fmt.Println("ParseInt Err:" + err.Error() + ";wk:" + wk)
				return err
			}

			if vk == "OUTMESSAGE" {
				utils.FileLog.Info("OUTMESSAGE:" + vv.(string))
				err = errors.New("OUTMESSAGE:" + vv.(string))
				return err
			}
			
			vk = strings.ToLower(vk)
			indexCode = windWsd + stockCode + vk

			indexCodeMap[indexCode] = indexCode

			wkInt = wkInt / 1000
			t := time.Unix(wkInt, 0)
			if t.After(time.Now()) {
				continue
			}
			dateTime := t.Format(utils.FormatDate)

			dataTime, err := time.ParseInLocation(utils.FormatDate, dateTime, time.Local)
			if err != nil {
				return err
			}
			timestamp := dataTime.UnixNano() / 1e6
			timeStr := fmt.Sprintf("%d", timestamp)

			//saveVal := utils.SubFloatToString(val, 20)

			var saveVal string
			switch vt := vv.(type) {
			case int:
				saveVal = strconv.Itoa(vt)
			case float64:
				saveVal = utils.SubFloatToFloatStr(vt, 20)
			case string:
				saveVal = vt
			}
			addSql += GetAddSql("0", indexCode, dateTime, timeStr, saveVal)
			isAdd = true
		}
	}

	if isAdd {
		for _, v := range indexCodeMap {
			var count int
			sql := ` SELECT COUNT(1) FROM edb_data_wind_wsd WHERE edb_code=? `
			err = o.Raw(sql, v).QueryRow(&count)
			if err != nil {
				return err
			}
			if count > 0 {
				sql = ` DELETE FROM edb_data_wind_wsd WHERE edb_code=? `
				_, err = o.Raw(sql, v).Exec()
				if err != nil {
					return err
				}
			}
		}

		addSql = strings.TrimRight(addSql, ",")
		_, err = o.Raw(addSql).Exec()
		if err != nil {
			fmt.Println("sql exec err:" + err.Error())
			return
		}
	}
	return
}

// RefreshEdbDataFromWind 刷新wind指标数据
func RefreshEdbDataFromWindWsd(edbInfoId int, edbCode, startDate string, item map[string]map[string]interface{}) (err error) {
	o := orm.NewOrm()
	source := utils.DATA_SOURCE_WIND
	subSource := utils.DATA_SUB_SOURCE_DATE

	// 真实数据的最大日期  ， 插入规则配置的日期
	var realDataMaxDate, edbDataInsertConfigDate time.Time
	var edbDataInsertConfig *EdbDataInsertConfig
	var isFindConfigDateRealData bool //是否找到配置日期的实际数据的值
	{
		edbDataInsertConfig, err = GetEdbDataInsertConfigByEdbId(edbInfoId)
		if err != nil && err.Error() != utils.ErrNoRow() {
			return
		}
		if edbDataInsertConfig != nil {
			edbDataInsertConfigDate = edbDataInsertConfig.Date
		}
	}

	var condition string
	var pars []interface{}

	condition += " AND edb_info_id=? "
	pars = append(pars, edbInfoId)

	var startDateTime time.Time
	if startDate != "" {
		condition += " AND data_time>=? "
		pars = append(pars, startDate)
		startDateTime, _ = time.ParseInLocation(utils.FormatDate, startDate, time.Local)
	}

	existList, err := GetEdbDataByCondition(source, subSource, condition, pars)
	existMap := make(map[string]*EdbInfoSearchData)
	for _, v := range existList {
		existMap[v.DataTime] = v
	}
	addSql := ` INSERT INTO edb_data_wind_wsd(edb_info_id,edb_code,data_time,value,create_time,modify_time,data_timestamp) values `
	var isAdd bool
	addMap := make(map[string]string)
	edbInfoIdStr := strconv.Itoa(edbInfoId)

	for wk, wv := range item {
		for vk, vv := range wv {
			if vv == nil {
				continue
			}

			wkInt, err := strconv.ParseInt(wk, 10, 64)
			if err != nil {
				fmt.Println("ParseInt Err:" + err.Error() + ";wk:" + wk)
				return err
			}

			vk = strings.ToLower(vk)

			wkInt = wkInt / 1000
			t := time.Unix(wkInt, 0)
			if t.After(time.Now()) {
				continue
			}
			dateTime := t.Format(utils.FormatDate)

			dataTime, err := time.ParseInLocation(utils.FormatDate, dateTime, time.Local)
			if err != nil {
				return err
			}
			timestamp := dataTime.UnixNano() / 1e6
			timeStr := fmt.Sprintf("%d", timestamp)

			var saveVal string
			switch vt := vv.(type) {
			case int:
				saveVal = strconv.Itoa(vt)
			case float64:
				saveVal = utils.SubFloatToFloatStr(vt, 20)
			case string:
				saveVal = vt
			}

			//如果传入的开始时间是空的，且当前数据日期早于传入的开始日期，那么需要判断下当前日期的数据是否存在
			if !startDateTime.IsZero() && t.Before(startDateTime) {
				tmpItem, tmpErr := GetEdbDataByDate(source, subSource, edbCode, dateTime)
				if tmpErr == nil && tmpItem != nil {
					existMap[tmpItem.DataTime] = tmpItem
				}
			}

			if findItem, ok := existMap[dateTime]; !ok {
				if _, existOk := addMap[dateTime]; !existOk {
					isAdd = true
					addSql += GetAddSql(edbInfoIdStr, edbCode, dateTime, timeStr, saveVal)
					addMap[dateTime] = saveVal
				}
			} else {
				if findItem != nil && utils.SubFloatToString(findItem.Value, 30) != saveVal {
					err = ModifyEdbDataById(source, subSource, findItem.EdbDataId, saveVal)
					if err != nil {
						return err
					}
				}
			}

			// 下面代码主要目的是处理掉手动插入的数据判断
			{
				if realDataMaxDate.IsZero() || t.After(realDataMaxDate) {
					realDataMaxDate = t
				}
				if edbDataInsertConfigDate.IsZero() || t.Equal(edbDataInsertConfigDate) {
					isFindConfigDateRealData = true
				}
			}
		}
	}

	// 处理手工数据补充的配置
	HandleConfigInsertEdbData(realDataMaxDate, edbDataInsertConfig, edbInfoId, source, subSource, existMap, isFindConfigDateRealData)

	if isAdd {
		addSql = strings.TrimRight(addSql, ",")
		_, err = o.Raw(addSql).Exec()
		if err != nil {
			fmt.Println("RefreshEdbDataFromWind add Err", err.Error())
			return
		}
	}

	return
}