package data_manage import ( "eta/eta_mobile/utils" "fmt" "github.com/beego/beego/v2/client/orm" "strconv" "strings" "time" ) type GieData struct { InputValue string `orm:"column(DATA_VALUE)" description:"日期"` DataTime string `orm:"column(DATA_DATE)" description:"值"` } func GetEdbDataGieMaxOrMinDate(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_gie WHERE edb_code=? ` err = o.Raw(sql, edbCode).QueryRow(&minDate, &maxDate) return } //func GetEdbDataByGie(edbCode, suffix, startDate, endDate string) (searchItem *EdbInfoSearch, err error) { // o := orm.NewOrmUsingDB("data") // to, err := o.Begin() // if err != nil { // return // } // searchItem = new(EdbInfoSearch) // searchItem.EdbCode = edbCode // eicBaseDataAll, err := GetBaseFromEicDataAllByIndexCode(edbCode, suffix) // if err != nil && err.Error() != utils.ErrNoRow() { // fmt.Println("GetBaseFromEicDataAllByIndexCode err:", err) // return // } // // var isAdd bool // addSql := ` INSERT INTO edb_data_gie(edb_info_id,edb_code,data_time,value,create_time,modify_time,status,data_timestamp) values ` // dataList := make([]*EdbInfoSearchData, 0) // existMap := make(map[string]string) // // for _, sv := range eicBaseDataAll { // eDate := sv.GasDayStartedOn // dataTime, err := time.Parse(utils.FormatDate, eDate) // if err != nil { // fmt.Println("time.Parse Err:" + eDate) // return nil, err // } // timestamp := dataTime.UnixNano() / 1e6 // timeStr := fmt.Sprintf("%d", timestamp) // //var name string // if _, ok := existMap[eDate]; !ok { // if suffix == "GS" { // //name = "gas_in_storage" // addSql += GetAddSql("0", edbCode, eDate, timeStr, sv.GasInStorage) // existMap[eDate] = sv.GasInStorage // } else if suffix == "F" { // //name = "full" // addSql += GetAddSql("0", edbCode, eDate, timeStr, sv.Full) // existMap[eDate] = sv.Full // } else if suffix == "T" { // //name = "trend" // addSql += GetAddSql("0", edbCode, eDate, timeStr, sv.Trend) // existMap[eDate] = sv.Trend // } else if suffix == "In" { // //name = "injection" // addSql += GetAddSql("0", edbCode, eDate, timeStr, sv.Injection) // existMap[eDate] = sv.Injection // } else if suffix == "Out" { // //name = "withdrawal" // addSql += GetAddSql("0", edbCode, eDate, timeStr, sv.Withdrawal) // existMap[eDate] = sv.Withdrawal // } else if suffix == "WGV" { // //name = "working_gas_volume" // addSql += GetAddSql("0", edbCode, eDate, timeStr, sv.WorkingGasVolume) // existMap[eDate] = sv.WorkingGasVolume // } else if suffix == "IC" { // //name = "injection_capacity" // addSql += GetAddSql("0", edbCode, eDate, timeStr, sv.InjectionCapacity) // existMap[eDate] = sv.InjectionCapacity // } else if suffix == "WC" { // //name = "withdrawal_capacity" // addSql += GetAddSql("0", edbCode, eDate, timeStr, sv.WithdrawalCapacity) // existMap[eDate] = sv.WithdrawalCapacity // } // isAdd = true // } // } // if isAdd { // addSql = strings.TrimRight(addSql, ",") // utils.FileLog.Info("addSql:" + addSql) // _, err = to.Raw(addSql).Exec() // if err != nil { // fmt.Println("addSql err:", err) // return searchItem, err // } // } // if err != nil { // _ = to.Rollback() // } else { // _ = to.Commit() // } // size := utils.EDB_DATA_LIMIT // dataList, err = GetEdbDataAllByEdbCode(edbCode, utils.DATA_SOURCE_GIE, size) // if err != nil { // utils.FileLogData.Info("GetEdbDataGieByCode Err:%s", err.Error()) // return searchItem, err // } // minDate, maxDate, err := GetEdbDataGieMaxOrMinDate(edbCode) // if err != nil { // return searchItem, err // } // searchItem.DataList = dataList // searchItem.StartDate = minDate // searchItem.EndDate = maxDate // if searchItem.DataList == nil { // searchItem.DataList = make([]*EdbInfoSearchData, 0) // } // return //} // RefreshEdbDataByGie 刷新欧洲天然气指标数据 func RefreshEdbDataByGie(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() } }() if err != nil { fmt.Println("refresh err:", err) return } var suffix string l := len(edbCode) if strings.Contains(edbCode[l-2:], "GS") { suffix = "GS" } else if strings.Contains(edbCode[l-2:], "CF") { suffix = "CF" } else if strings.Contains(edbCode[l-1:], "T") { suffix = "T" } else if strings.Contains(edbCode[l-2:], "In") { suffix = "In" } else if strings.Contains(edbCode[l-3:], "Out") { suffix = "Out" } else if strings.Contains(edbCode[l-3:], "WGV") { suffix = "WGV" } else if strings.Contains(edbCode[l-2:], "IC") { suffix = "IC" } else if strings.Contains(edbCode[l-2:], "WC") { suffix = "WC" } else if strings.Contains(edbCode[l-1:], "F") { suffix = "F" } else if strings.Contains(edbCode[l-1:], "C") { suffix = "C" } else { suffix = "" } edbInfoIdStr := strconv.Itoa(edbInfoId) //计算数据 var condition string var pars []interface{} if edbCode != "" { condition += " AND eic_code=? " pars = append(pars, edbCode[:l-len(suffix)]) } if startDate != "" { condition += " AND gas_day_started_on>=? " pars = append(pars, startDate) } if endDate != "" { condition += " AND gas_day_started_on<=? " pars = append(pars, endDate) } eicDataList, err := GetGieDataByTradeCodeV2(condition, pars) fmt.Println("eicDataList", len(eicDataList)) addSql := ` INSERT INTO edb_data_gie(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 eicDataList { var value string if suffix == "GS" { value = v.GasInStorage } else if suffix == "C" { value = v.Consumption } else if suffix == "CF" { value = v.ConsumptionFull } else if suffix == "F" { value = v.Full } else if suffix == "T" { value = v.Trend } else if suffix == "In" { value = v.Injection } else if suffix == "Out" { value = v.Withdrawal } else if suffix == "WGV" { value = v.WorkingGasVolume } else if suffix == "IC" { value = v.InjectionCapacity } else if suffix == "WC" { value = v.WithdrawalCapacity } item := v itemValue := value if _, ok := existMap[v.GasDayStart]; !ok { count, err := GetEdbDataGieByCodeAndDate(edbCode, v.GasDayStart) if err != nil && err.Error() != utils.ErrNoRow() { return err } if count <= 0 { eDate := item.GasDayStart 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 = ModifyEdbDataGie(int64(edbInfoId), v.GasDayStart, value) if err != nil { return err } } } existMap[v.GasDayStart] = value } if isAdd { addSql = strings.TrimRight(addSql, ",") _, err = to.Raw(addSql).Exec() if err != nil { return err } } return } // RefreshAllEdbDataByGie 全部刷新欧洲天然气 func RefreshAllEdbDataByGie(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() } }() if err != nil { return } var suffix string l := len(edbCode) if strings.Contains(edbCode[l-2:], "GS") { suffix = "GS" } else if strings.Contains(edbCode[l-2:], "CF") { suffix = "CF" } else if strings.Contains(edbCode[l-1:], "T") { suffix = "T" } else if strings.Contains(edbCode[l-2:], "In") { suffix = "In" } else if strings.Contains(edbCode[l-3:], "Out") { suffix = "Out" } else if strings.Contains(edbCode[l-3:], "WGV") { suffix = "WGV" } else if strings.Contains(edbCode[l-2:], "IC") { suffix = "IC" } else if strings.Contains(edbCode[l-2:], "WC") { suffix = "WC" } else if strings.Contains(edbCode[l-1:], "F") { suffix = "F" } else if strings.Contains(edbCode[l-1:], "C") { suffix = "C" } else { suffix = "" } edbInfoIdStr := strconv.Itoa(edbInfoId) //计算数据 var condition string var pars []interface{} if edbCode != "" { condition += " AND eic_code=? " pars = append(pars, edbCode[:l-len(suffix)]) } if startDate != "" { condition += " AND gas_day_start>=? " pars = append(pars, startDate) } if endDate != "" { condition += " AND gas_day_start<=? " pars = append(pars, endDate) } eicDataList, err := GetGieDataByTradeCodeV2(condition, pars) fmt.Println("all eicDataList", len(eicDataList)) //获取指标所有数据 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 } addSql := ` INSERT INTO edb_data_gie(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 eicDataList { var value string if suffix == "GS" { value = v.GasInStorage } else if suffix == "C" { value = v.Consumption } else if suffix == "CF" { value = v.ConsumptionFull } else if suffix == "F" { value = v.Full } else if suffix == "T" { value = v.Trend } else if suffix == "In" { value = v.Injection } else if suffix == "Out" { value = v.Withdrawal } else if suffix == "WGV" { value = v.WorkingGasVolume } else if suffix == "IC" { value = v.InjectionCapacity } else if suffix == "WC" { value = v.WithdrawalCapacity } item := v itemValue := value if _, ok := existMap[v.GasDayStart]; !ok { eDate := item.GasDayStart 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.GasDayStart] = v.GasDayStart } if isAdd { addSql = strings.TrimRight(addSql, ",") _, err = to.Raw(addSql).Exec() if err != nil { return err } } return } // GetBaseInfoFromEicByIndexCode 获取指标信息 func GetBaseInfoFromEicByIndexCode(indexCode, suffix string) (list []*BaseFromTradeEicIndex, err error) { o := orm.NewOrmUsingDB("data") sql := `SELECT * FROM base_from_trade_eic_index WHERE %s_code=? ` sql = fmt.Sprintf(sql, suffix) _, err = o.Raw(sql, indexCode).QueryRows(&list) return } func GetGieDataByTradeCode(condition string, pars []interface{}) (item []*BaseFromTradeEicIndex, err error) { sql := ` SELECT * FROM base_from_trade_eic_index WHERE 1=1 ` o := orm.NewOrmUsingDB("data") if condition != "" { sql += condition } sql += ` ORDER BY gas_day_started_on DESC ` fmt.Println(sql, pars) _, err = o.Raw(sql, pars).QueryRows(&item) return } type EicIndexV2 struct { BaseFromEicIndexId int `orm:"column(base_from_eic_index_id);pk"` Type string EicCode string Name string Status string GasDayStart string GasInStorage string GasInStorageCode string Consumption string ConsumptionCode string ConsumptionFull string ConsumptionFullCode string Full string FullCode string Trend string TrendCode string Injection string InjectionCode string Withdrawal string WithdrawalCode string WorkingGasVolume string WorkingGasVolumeCode string InjectionCapacity string InjectionCapacityCode string WithdrawalCapacity string WithdrawalCapacityCode string Info string Parent string CreateTime time.Time ModifyTime time.Time Children []BaseFromTradeEicIndexV2 } func GetGieDataByTradeCodeV2(condition string, pars []interface{}) (item []*EicIndexV2, err error) { sql := ` SELECT * FROM base_from_trade_eic_index_v2 WHERE 1=1 ` o := orm.NewOrmUsingDB("data") if condition != "" { sql += condition } sql += ` ORDER BY gas_day_start DESC ` fmt.Println(sql, pars) _, err = o.Raw(sql, pars).QueryRows(&item) return } func AddEdbDataGieBySql(sqlStr string) (err error) { o := orm.NewOrmUsingDB("data") _, err = o.Raw(sqlStr).Exec() return } func GetEdbDataGieByCode(edbCode string) (items []*EdbInfoSearchData, err error) { o := orm.NewOrmUsingDB("data") sql := ` SELECT * FROM edb_data_gie WHERE edb_code=? ORDER BY data_time DESC LIMIT ? ` _, err = o.Raw(sql, edbCode, utils.EDB_DATA_LIMIT).QueryRows(&items) return } func GetBaseFromEicDataAllByIndexCode(indexCode, suffix string) (list []*BaseFromTradeEicIndex, err error) { o := orm.NewOrmUsingDB("data") var name string if suffix == "" { name = "eic_code" } else if suffix == "GS" { name = "gas_in_storage_code" } else if suffix == "C" { name = "consumption_code" } else if suffix == "CF" { name = "consumption_full_code" } else if suffix == "F" { name = "full_code" } else if suffix == "T" { name = "trend_code" } else if suffix == "In" { name = "injection_code" } else if suffix == "Out" { name = "withdrawal_code" } else if suffix == "WGV" { name = "working_gas_volume_code" } else if suffix == "IC" { name = "injection_capacity_code" } else if suffix == "WC" { name = "withdrawal_capacity_code" } sql := `SELECT * FROM base_from_trade_eic_index_v2 WHERE %s=? ` sql = fmt.Sprintf(sql, name) _, err = o.Raw(sql, indexCode).QueryRows(&list) return } func GetBaseFromEicDataAllByIndexCodeV2(indexCode, suffix string) (list []*BaseFromTradeEicIndexV2, err error) { o := orm.NewOrmUsingDB("data") var name string if suffix == "" { name = "eic_code" } else if suffix == "GS" { name = "gas_in_storage_code" } else if suffix == "C" { name = "consumption_code" } else if suffix == "CF" { name = "consumption_full_code" } else if suffix == "F" { name = "full_code" } else if suffix == "T" { name = "trend_code" } else if suffix == "In" { name = "injection_code" } else if suffix == "Out" { name = "withdrawal_code" } else if suffix == "WGV" { name = "working_gas_volume_code" } else if suffix == "IC" { name = "injection_capacity_code" } else if suffix == "WC" { name = "withdrawal_capacity_code" } sql := `SELECT * FROM base_from_trade_eic_index_v2 WHERE %s=? ` sql = fmt.Sprintf(sql, name) _, err = o.Raw(sql, indexCode).QueryRows(&list) return } func GetEdbDataGieByCodeAndDate(edbCode string, startDate string) (count int, err error) { o := orm.NewOrmUsingDB("data") sql := ` SELECT COUNT(1) AS count FROM edb_data_gie WHERE edb_code=? AND data_time=? ` err = o.Raw(sql, edbCode, startDate).QueryRow(&count) return } func ModifyEdbDataGie(edbInfoId int64, dataTime, value string) (err error) { o := orm.NewOrmUsingDB("data") sql := ` UPDATE edb_data_gie SET value=?,modify_time=NOW() WHERE edb_info_id=? AND data_time=? ` _, err = o.Raw(sql, value, edbInfoId, dataTime).Exec() return }