package data_manage import ( "encoding/json" "fmt" "github.com/beego/beego/v2/client/orm" "github.com/rdlucklib/rdluck_tools/http" "hongze/hz_crm_api/utils" "net/url" "strconv" "strings" "time" ) type EdbDataPb struct { EdbDataId int `orm:"column(edb_data_id);pk"` EdbInfoId int EdbCode string DataTime string Value float64 Status int CreateTime time.Time ModifyTime time.Time Ticker string Field string DataTimestamp int64 } func AddEdbDataPb(items []*EdbDataPb) (err error) { o := orm.NewOrmUsingDB("data") _, err = o.InsertMulti(1, items) return } func AddEdbDataPbBySql(sqlStr string) (err error) { o := orm.NewOrmUsingDB("data") _, err = o.Raw(sqlStr).Exec() return } func ModifyEdbDataPb(edbInfoId int, dataTime string, value float64) (err error) { o := orm.NewOrmUsingDB("data") sql := ` UPDATE edb_data_pb SET value=?,modify_time=NOW() WHERE edb_info_id=? AND data_time=? ` _, err = o.Raw(sql, value, edbInfoId, dataTime).Exec() return } func GetEdbDataPbMaxOrMinDate(edbCode string) (min_date, max_date string, err error) { o := orm.NewOrmUsingDB("data") sql := ` SELECT MIN(data_time) AS min_date,MAX(data_time) AS max_date FROM edb_data_pb WHERE edb_code=? ` err = o.Raw(sql, edbCode).QueryRow(&min_date, &max_date) return } func GetEdbDataPbByCode(edbCode string, size int) (items []*EdbInfoSearchData, err error) { o := orm.NewOrmUsingDB("data") sql := ` SELECT * FROM edb_data_pb WHERE edb_code=? ORDER BY data_time DESC LIMIT ? ` _, err = o.Raw(sql, edbCode, size).QueryRows(&items) return } func GetEdbDataPbByCodeAndDate(edbCode string, startDate string) (count int, err error) { o := orm.NewOrmUsingDB("data") sql := ` SELECT COUNT(1) AS count FROM edb_data_pb WHERE edb_code=? AND data_time>=? ` err = o.Raw(sql, edbCode, startDate).QueryRow(&count) return } type EdbDataFromPb struct { Date map[string]int64 `json:"date"` Ticker map[string]string `json:"ticker"` Field map[string]string `json:"field"` Value map[string]float64 `json:"value"` } // 全部刷新 func RefreshAllEdbDataByPb(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() } }() refreshEdbCode := edbCode edbCode = url.QueryEscape(edbCode) bpUrl := utils.Hz_Data_PB_Url + `edbInfo/pb?EdbCode=%s&StartDate=%s&EndDate=%s` bpUrl = fmt.Sprintf(bpUrl, edbCode, startDate, endDate) utils.FileLog.Info("thsUrl:%s", bpUrl) body, err := http.Get(bpUrl) fmt.Println("RefreshAllEdbDataByPb body:") fmt.Println(string(body)) if err != nil { return } item := new(EdbDataFromPb) err = json.Unmarshal(body, &item) if err != nil { return } //获取指标所有数据 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 } edbInfoIdStr := strconv.Itoa(edbInfoId) if len(item.Date) > 0 { dateMap := item.Date var isAdd bool addSql := ` INSERT INTO edb_data_pb(edb_info_id,edb_code,data_time,value,create_time,modify_time,status,ticker,field,data_timestamp) values ` nowStr := time.Now().Format(utils.FormatDateTime) for k, v := range dateMap { timeStr := fmt.Sprintf("%d", v) v = v / 1000 t := time.Unix(v, 0) dateTime := t.Format(utils.FormatDate) val := item.Value[k] field := item.Field[k] ticker := item.Ticker[k] saveValue := utils.SubFloatToString(val, 30) if field == "PX_LAST" { if existVal, ok := dataMap[dateTime]; !ok { addSql += "(" addSql += edbInfoIdStr + "," + "'" + refreshEdbCode + "'" + "," + "'" + dateTime + "'" + "," + saveValue + "," + "'" + nowStr + "'" + "," + "'" + nowStr + "'" + "," + "1" + "," + "'" + ticker + "'" + "," + "'" + field + "'" + "," + "'" + timeStr + "'" addSql += ")," 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, saveValue, edbInfoId, dateTime).Exec() if err != nil { return err } } } } } if isAdd { addSql = strings.TrimRight(addSql, ",") _, err = to.Raw(addSql).Exec() if err != nil { fmt.Println("RefreshAllEdbDataByPb add Err", err.Error()) return } } } return }