package data_manage import ( "eta/eta_api/utils" "fmt" "github.com/beego/beego/v2/client/orm" "strconv" "strings" "time" ) type CoalData struct { InputValue string `orm:"column(DATA_VALUE)" description:"日期"` DataTime string `orm:"column(DATA_DATE)" description:"值"` } type BaseFromCoalDataSimple struct { Id int `orm:"column(base_from_trade_coal_index_id);pk"` DealCode string BuyCode string SoldCode string DataTime string DealValue string BuyValue string SoldValue string } type BaseInfoFromCoal struct { IndexName string Frequency string Unit string } type BaseFromTradeCoalIndex struct { BaseFromTradeCoalIndexId int `orm:"column(base_from_trade_coal_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 GetEdbDataCoalMaxOrMinDate(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_coal WHERE edb_code=? ` err = o.Raw(sql, edbCode).QueryRow(&minDate, &maxDate) return } // RefreshEdbDataByCoal 刷新煤炭网指标数据 func RefreshEdbDataByCoal(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() } }() var suffix string if strings.Contains(edbCode, "jsm") { suffix = "jsm_index" } else if strings.Contains(edbCode, "company") { suffix = "company_index" } else if strings.Contains(edbCode, "firm") { suffix = "firm_index" } else if strings.Contains(edbCode, "coastal") { suffix = "coastal_index" } else if strings.Contains(edbCode, "inland") { suffix = "inland_index" } edbInfoIdStr := strconv.Itoa(edbInfoId) //计算数据 var condition string var pars []interface{} if edbCode != "" { condition += " AND index_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) } coalDataList, err := GetCoalDataByTradeCode(condition, suffix, pars) addSql := ` INSERT INTO edb_data_coal(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 coalDataList { item := v itemValue := v.DealValue if _, ok := existMap[v.DataTime]; !ok { count, err := GetEdbDataCoalByCodeAndDate(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 = ModifyEdbDataCoal(int64(edbInfoId), v.DataTime, v.DealValue) if err != nil { return err } } } existMap[v.DataTime] = v.DealValue } if isAdd { addSql = strings.TrimRight(addSql, ",") _, err = to.Raw(addSql).Exec() if err != nil { return err } } return } // RefreshAllEdbDataByCoal 全部刷新中金所 func RefreshAllEdbDataByCoal(edbInfoId, source, subSource 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() } }() var suffix string if strings.Contains(edbCode, "jsm") { suffix = "jsm_index" } else if strings.Contains(edbCode, "company") { suffix = "company_index" } else if strings.Contains(edbCode, "firm") { suffix = "firm_index" } else if strings.Contains(edbCode, "coastal") { suffix = "coastal_index" } else if strings.Contains(edbCode, "inland") { suffix = "inland_index" } edbInfoIdStr := strconv.Itoa(edbInfoId) //计算数据 var condition string var pars []interface{} if edbCode != "" { condition += " AND index_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) } coalDataList, err := GetCoalDataByTradeCode(condition, suffix, pars) //获取指标所有数据 dataList := make([]*EdbDataBase, 0) dataTableName := GetEdbDataTableName(source, subSource) 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_coal(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 coalDataList { item := v itemValue := v.DealValue 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 = to.Raw(sql, sValue, edbInfoId, eDate).Exec() if err != nil { return err } } } } } existMap[v.DataTime] = v.DataTime } if isAdd { addSql = strings.TrimRight(addSql, ",") _, err = to.Raw(addSql).Exec() if err != nil { return err } } return } // GetBaseInfoFromCoalByIndexCode 获取指标信息 func GetBaseInfoFromCoalByIndexCode(indexCode, suffix string) (list []*BaseInfoFromCoal, err error) { o := orm.NewOrmUsingDB("data") sql := `SELECT * FROM base_from_coalmine_%s WHERE index_code=? ` sql = fmt.Sprintf(sql, suffix) _, err = o.Raw(sql, indexCode).QueryRows(&list) return } func GetCoalDataByTradeCode(condition, suffix string, pars []interface{}) (item []*BaseFromCoalDataSimple, err error) { sql := ` SELECT * FROM base_from_coalmine_%s WHERE 1=1 ` sql = fmt.Sprintf(sql, suffix) o := orm.NewOrmUsingDB("data") if condition != "" { sql += condition } sql += ` ORDER BY data_time DESC ` _, err = o.Raw(sql, pars).QueryRows(&item) return } func GetEdbDataCoalByCodeAndDate(edbCode string, startDate string) (count int, err error) { o := orm.NewOrmUsingDB("data") sql := ` SELECT COUNT(1) AS count FROM edb_data_coal WHERE edb_code=? AND data_time=? ` err = o.Raw(sql, edbCode, startDate).QueryRow(&count) return } func ModifyEdbDataCoal(edbInfoId int64, dataTime, value string) (err error) { o := orm.NewOrmUsingDB("data") sql := ` UPDATE edb_data_coal SET value=?,modify_time=NOW() WHERE edb_info_id=? AND data_time=? ` _, err = o.Raw(sql, value, edbInfoId, dataTime).Exec() return }