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 }