edb_data_rzd.go 6.2 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230
  1. // @Author gmy 2024/9/14 16:13:00
  2. package models
  3. import (
  4. "eta/eta_index_lib/utils"
  5. "fmt"
  6. "github.com/beego/beego/v2/client/orm"
  7. "strconv"
  8. "strings"
  9. "time"
  10. )
  11. type EdbDataRzd struct {
  12. edbDataId int `orm:"column(edb_data_id);pk"`
  13. CreateTime string `orm:"column(create_time)"`
  14. ModifyTime string `orm:"column(modify_time)"`
  15. EdbInfoId int `orm:"column(edb_info_id)"`
  16. EdbCode string `orm:"column(edb_code)"`
  17. DataTime string `orm:"column(data_time)"`
  18. Value float64 `orm:"column(value)"`
  19. DataTimestamp uint64 `orm:"column(data_timestamp)"`
  20. }
  21. func init() {
  22. orm.RegisterModel(new(EdbDataRzd))
  23. }
  24. // GetRzdEdbDataByIndexCodeAndDataTime 根据指标编码和时间获取指标库数据
  25. func GetRzdEdbDataByIndexCodeAndDataTime(indexCode string, dataTime string) (items []EdbDataRzd, err error) {
  26. o := orm.NewOrm()
  27. sql := `SELECT * FROM edb_data_rzd WHERE edb_code=? AND data_time=?`
  28. _, err = o.Raw(sql, indexCode, dataTime).QueryRows(&items)
  29. if utils.IsErrNoRow(err) {
  30. return nil, nil
  31. }
  32. return
  33. }
  34. // UpdateRzdEdbDataById 更新指标库数据 须根据指标编码和日期更新 仅适合月度数据
  35. func UpdateRzdEdbDataById(id int, value float64) (err error) {
  36. o := orm.NewOrm()
  37. sql := `UPDATE edb_data_rzd SET value=? WHERE edb_data_id=?`
  38. _, err = o.Raw(sql, value, id).Exec()
  39. return
  40. }
  41. // UpdateRzdEdbDataByIndexCodeAndDataTime 根据指标编码和时间更新数据
  42. func UpdateRzdEdbDataByIndexCodeAndDataTime(indexCode int, dataTime string, value float64) (err error) {
  43. o := orm.NewOrm()
  44. sql := `UPDATE edb_data_rzd SET value=? WHERE edb_info_id=? AND data_time=?`
  45. _, err = o.Raw(sql, value, indexCode, dataTime).Exec()
  46. return
  47. }
  48. // AddRzdEdbDataList 新增指标库数据
  49. func AddRzdEdbDataList(items []EdbDataRzd) (err error) {
  50. o := orm.NewOrm()
  51. _, err = o.InsertMulti(len(items), items)
  52. return
  53. }
  54. // AddEdbDataFromRzd 新增指标数据
  55. func AddEdbDataFromRzd(edbCode string) (err error) {
  56. o := orm.NewOrm()
  57. var condition string
  58. var pars []interface{}
  59. if edbCode != "" {
  60. condition += " AND index_code = ? "
  61. pars = append(pars, edbCode)
  62. }
  63. dataAll, err := GetBaseFromRzdDataByCondition(condition, pars)
  64. if err != nil {
  65. return err
  66. }
  67. dataLen := len(dataAll)
  68. existMap := make(map[string]string)
  69. if dataLen > 0 {
  70. var isAdd bool
  71. addSql := ` INSERT INTO edb_data_rzd (edb_info_id,edb_code,data_time,value,create_time,modify_time,data_timestamp) values `
  72. for i := 0; i < dataLen; i++ {
  73. item := dataAll[i]
  74. eDate := item.DataTime
  75. sValue := utils.SubFloatToString(item.Value, 4)
  76. if sValue != "" {
  77. if _, ok := existMap[eDate]; !ok {
  78. dataTime, err := time.ParseInLocation(utils.FormatDate, eDate, time.Local)
  79. if err != nil {
  80. return err
  81. }
  82. timestamp := dataTime.UnixNano() / 1e6
  83. timeStr := fmt.Sprintf("%d", timestamp)
  84. addSql += GetAddSql("0", edbCode, eDate, timeStr, sValue)
  85. isAdd = true
  86. }
  87. }
  88. existMap[eDate] = eDate
  89. }
  90. if isAdd {
  91. addSql = strings.TrimRight(addSql, ",")
  92. utils.FileLog.Info("addSql:" + addSql)
  93. _, err = o.Raw(addSql).Exec()
  94. if err != nil {
  95. return err
  96. }
  97. }
  98. }
  99. return
  100. }
  101. func RefreshEdbDataFromRzd(edbInfoId int, edbCode, startDate string) (err error) {
  102. source := utils.DATA_SOURCE_RZD
  103. subSource := utils.DATA_SUB_SOURCE_EDB
  104. o := orm.NewOrm()
  105. if err != nil {
  106. return
  107. }
  108. edbInfoIdStr := strconv.Itoa(edbInfoId)
  109. //计算数据
  110. var condition string
  111. var pars []interface{}
  112. if edbCode != "" {
  113. condition += " AND index_code=? "
  114. pars = append(pars, edbCode)
  115. }
  116. if startDate != "" {
  117. condition += " AND data_time>=? "
  118. pars = append(pars, startDate)
  119. }
  120. dataList, err := GetBaseFromRzdDataByCondition(condition, pars)
  121. if err != nil {
  122. return
  123. }
  124. // 真实数据的最大日期 , 插入规则配置的日期
  125. var realDataMaxDate, edbDataInsertConfigDate time.Time
  126. var edbDataInsertConfig *EdbDataInsertConfig
  127. var isFindConfigDateRealData bool //是否找到配置日期的实际数据的值
  128. {
  129. edbDataInsertConfig, err = GetEdbDataInsertConfigByEdbId(edbInfoId)
  130. if err != nil && !utils.IsErrNoRow(err) {
  131. return
  132. }
  133. if edbDataInsertConfig != nil {
  134. edbDataInsertConfigDate = edbDataInsertConfig.Date
  135. }
  136. }
  137. var existCondition string
  138. var existPars []interface{}
  139. existCondition += " AND edb_info_id=? "
  140. existPars = append(existPars, edbInfoId)
  141. if startDate != "" {
  142. existCondition += " AND data_time>=? "
  143. existPars = append(existPars, startDate)
  144. }
  145. //获取指标所有数据
  146. existList, err := GetEdbDataByCondition(source, subSource, existCondition, existPars)
  147. if err != nil {
  148. return err
  149. }
  150. existMap := make(map[string]*EdbInfoSearchData)
  151. for _, v := range existList {
  152. existMap[v.DataTime] = v
  153. }
  154. addSql := ` INSERT INTO edb_data_rzd(edb_info_id,edb_code,data_time,value,create_time,modify_time,data_timestamp) values `
  155. var isAdd bool
  156. addMap := make(map[string]string)
  157. for _, v := range dataList {
  158. item := v
  159. eDate := item.DataTime
  160. sValue := utils.SubFloatToString(item.Value, 4)
  161. dataTime, err := time.ParseInLocation(utils.FormatDate, eDate, time.Local)
  162. if err != nil {
  163. return err
  164. }
  165. if findItem, ok := existMap[v.DataTime]; !ok {
  166. if sValue != "" {
  167. timestamp := dataTime.UnixNano() / 1e6
  168. timeStr := fmt.Sprintf("%d", timestamp)
  169. saveValue := sValue
  170. if _, addOk := addMap[eDate]; !addOk {
  171. addSql += GetAddSql(edbInfoIdStr, edbCode, eDate, timeStr, saveValue)
  172. isAdd = true
  173. }
  174. }
  175. } else {
  176. if findItem != nil && utils.SubFloatToString(findItem.Value, 4) != sValue {
  177. err = ModifyEdbDataById(source, subSource, findItem.EdbDataId, sValue)
  178. if err != nil {
  179. return err
  180. }
  181. }
  182. }
  183. addMap[v.DataTime] = v.DataTime
  184. // 下面代码主要目的是处理掉手动插入的数据判断
  185. {
  186. if realDataMaxDate.IsZero() || dataTime.After(realDataMaxDate) {
  187. realDataMaxDate = dataTime
  188. }
  189. if edbDataInsertConfigDate.IsZero() || dataTime.Equal(edbDataInsertConfigDate) {
  190. isFindConfigDateRealData = true
  191. }
  192. }
  193. }
  194. // 处理手工数据补充的配置
  195. HandleConfigInsertEdbData(realDataMaxDate, edbDataInsertConfig, edbInfoId, source, subSource, existMap, isFindConfigDateRealData)
  196. if isAdd {
  197. addSql = strings.TrimRight(addSql, ",")
  198. _, err = o.Raw(addSql).Exec()
  199. if err != nil {
  200. return err
  201. }
  202. }
  203. return
  204. }