base_from_mysteel_chemical.go 17 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334335336337338339340341342343344345346347348349350351352353354355356357358359360361362363364365366367368369370371372373374375376377378379380381382383384385386387388389390391392393394395396397398399400401402403404405406407408409410411412413414415416417418419420421422423424425426427428429430431432433434435436437438439440441442443444445446447448449450451452453454455456457458459460461462463464465466467468469470471472473474475476477478479480481482483484485486487488489490491492493494495496497498499500501502503504505506507508509510511
  1. package models
  2. import (
  3. "eta/eta_index_lib/utils"
  4. "fmt"
  5. "strconv"
  6. "strings"
  7. "time"
  8. "github.com/beego/beego/v2/client/orm"
  9. )
  10. //钢联
  11. type MysteelChemicalData struct {
  12. InputValue float64 `orm:"column(value)" description:"值"`
  13. DataTime string `orm:"column(data_time)" description:"日期"`
  14. }
  15. func GetMysteelChemicalDataByCondition(condition string, pars []interface{}) (item []*MysteelChemicalData, err error) {
  16. sql1 := ` SELECT * FROM base_from_mysteel_chemical_data WHERE 1=1 `
  17. o := orm.NewOrm()
  18. if condition != "" {
  19. sql1 += condition
  20. }
  21. sql := `select * from (` + sql1 + ` having 1 order by modify_time DESC ) tmp GROUP BY data_time ORDER BY data_time DESC `
  22. _, err = o.Raw(sql, pars).QueryRows(&item)
  23. return
  24. }
  25. // AddEdbDataFromMysteelChemical 新增钢联指标数据
  26. func AddEdbDataFromMysteelChemical(edbCode string) (err error) {
  27. o := orm.NewOrm()
  28. var condition string
  29. var pars []interface{}
  30. if edbCode != "" {
  31. condition += " AND index_code = ? "
  32. pars = append(pars, edbCode)
  33. }
  34. mysteelChemicalDataList, err := GetMysteelChemicalDataByCondition(condition, pars)
  35. if err != nil {
  36. return
  37. }
  38. dataLen := len(mysteelChemicalDataList)
  39. existMap := make(map[string]string)
  40. if dataLen > 0 {
  41. var isAdd bool
  42. addSql := ` INSERT INTO edb_data_mysteel_chemical (edb_info_id,edb_code,data_time,value,create_time,modify_time,data_timestamp) values `
  43. for i := 0; i < dataLen; i++ {
  44. item := mysteelChemicalDataList[i]
  45. eDate := item.DataTime
  46. sValue := utils.SubFloatToString(item.InputValue, 30)
  47. if sValue != "" {
  48. if _, ok := existMap[eDate]; !ok {
  49. dataTime, err := time.ParseInLocation(utils.FormatDate, eDate, time.Local)
  50. if err != nil {
  51. return err
  52. }
  53. timestamp := dataTime.UnixNano() / 1e6
  54. timeStr := fmt.Sprintf("%d", timestamp)
  55. addSql += GetAddSql("0", edbCode, eDate, timeStr, sValue)
  56. isAdd = true
  57. }
  58. }
  59. existMap[eDate] = eDate
  60. }
  61. if isAdd {
  62. addSql = strings.TrimRight(addSql, ",")
  63. utils.FileLog.Info("addSql:" + addSql)
  64. _, err = o.Raw(addSql).Exec()
  65. if err != nil {
  66. return err
  67. }
  68. }
  69. }
  70. return
  71. }
  72. // RefreshEdbDataFromMysteelChemical 刷新钢联指标数据
  73. func RefreshEdbDataFromMysteelChemical(edbInfoId int, edbCode, startDate string) (err error) {
  74. source := utils.DATA_SOURCE_MYSTEEL_CHEMICAL
  75. o := orm.NewOrm()
  76. if err != nil {
  77. return
  78. }
  79. edbInfoIdStr := strconv.Itoa(edbInfoId)
  80. //计算数据
  81. var condition string
  82. var pars []interface{}
  83. if edbCode != "" {
  84. condition += " AND index_code=? "
  85. pars = append(pars, edbCode)
  86. }
  87. if startDate != "" {
  88. condition += " AND data_time>=? "
  89. pars = append(pars, startDate)
  90. }
  91. mysteelChemicalDataList, err := GetMysteelChemicalDataByCondition(condition, pars)
  92. if err != nil {
  93. return
  94. }
  95. // 真实数据的最大日期 , 插入规则配置的日期
  96. var realDataMaxDate, edbDataInsertConfigDate time.Time
  97. var edbDataInsertConfig *EdbDataInsertConfig
  98. var isFindConfigDateRealData bool //是否找到配置日期的实际数据的值
  99. {
  100. edbDataInsertConfig, err = GetEdbDataInsertConfigByEdbId(edbInfoId)
  101. if err != nil && err.Error() != utils.ErrNoRow() {
  102. return
  103. }
  104. if edbDataInsertConfig != nil {
  105. edbDataInsertConfigDate = edbDataInsertConfig.Date
  106. }
  107. }
  108. var existCondition string
  109. var existPars []interface{}
  110. existCondition += " AND edb_info_id=? "
  111. existPars = append(existPars, edbInfoId)
  112. if startDate != "" {
  113. existCondition += " AND data_time>=? "
  114. existPars = append(existPars, startDate)
  115. }
  116. //获取指标所有数据
  117. existList, err := GetEdbDataByCondition(source, existCondition, existPars)
  118. if err != nil {
  119. return err
  120. }
  121. existMap := make(map[string]*EdbInfoSearchData)
  122. for _, v := range existList {
  123. existMap[v.DataTime] = v
  124. }
  125. addSql := ` INSERT INTO edb_data_mysteel_chemical(edb_info_id,edb_code,data_time,value,create_time,modify_time,data_timestamp) values `
  126. var isAdd bool
  127. addMap := make(map[string]string)
  128. for _, v := range mysteelChemicalDataList {
  129. item := v
  130. eDate := item.DataTime
  131. sValue := utils.SubFloatToString(item.InputValue, 30)
  132. dataTime, err := time.ParseInLocation(utils.FormatDate, eDate, time.Local)
  133. if err != nil {
  134. return err
  135. }
  136. if findItem, ok := existMap[v.DataTime]; !ok {
  137. if sValue != "" {
  138. timestamp := dataTime.UnixNano() / 1e6
  139. timeStr := fmt.Sprintf("%d", timestamp)
  140. saveValue := sValue
  141. if _, addOk := addMap[eDate]; !addOk {
  142. addSql += GetAddSql(edbInfoIdStr, edbCode, eDate, timeStr, saveValue)
  143. isAdd = true
  144. }
  145. }
  146. } else {
  147. if findItem != nil && utils.SubFloatToString(findItem.Value, 30) != sValue {
  148. err = ModifyEdbDataById(source, findItem.EdbDataId, sValue)
  149. if err != nil {
  150. return err
  151. }
  152. }
  153. }
  154. addMap[v.DataTime] = v.DataTime
  155. // 下面代码主要目的是处理掉手动插入的数据判断
  156. {
  157. if realDataMaxDate.IsZero() || dataTime.After(realDataMaxDate) {
  158. realDataMaxDate = dataTime
  159. }
  160. if edbDataInsertConfigDate.IsZero() || dataTime.Equal(edbDataInsertConfigDate) {
  161. isFindConfigDateRealData = true
  162. }
  163. }
  164. }
  165. // 处理手工数据补充的配置
  166. HandleConfigInsertEdbData(realDataMaxDate, edbDataInsertConfig, edbInfoId, source, existMap, isFindConfigDateRealData)
  167. if isAdd {
  168. addSql = strings.TrimRight(addSql, ",")
  169. _, err = o.Raw(addSql).Exec()
  170. if err != nil {
  171. return err
  172. }
  173. }
  174. return
  175. }
  176. // 钢联化工指标数据
  177. type BaseFromMysteelChemicalIndex struct {
  178. BaseFromMysteelChemicalIndexId int64 `orm:"column(base_from_mysteel_chemical_index_id);pk"`
  179. BaseFromMysteelChemicalClassifyId int `description:"分类id"`
  180. IndexCode string `description:"指标编码"`
  181. IndexName string `description:"指标名称"`
  182. Unit string `description:"单位"`
  183. Source string `description:"数据来源"`
  184. Frequency string `description:"频度"`
  185. StartDate time.Time `description:"开始日期"`
  186. EndDate time.Time `description:"结束日期"`
  187. Describe string `description:"指标描述"`
  188. UpdateWeek string `description:"更新周期"`
  189. UpdateTime string `description:"更新时间,多个时间点用英文,隔开"`
  190. UpdateTime2 string `description:"更新时间2"`
  191. SysUserId int `description:"创建人id"`
  192. SysUserRealName string `description:"创建人姓名"`
  193. FilePath string `description:"文件存储路径"`
  194. MergeFilePath string `description:"更新文件"`
  195. FileIndex int `description:"文件索引"`
  196. MergeUpdateWeek string `description:"合并文件的更新周"`
  197. UpdateDate string `description:"更新日期"`
  198. CreateTime time.Time `description:"创建时间"`
  199. ModifyTime time.Time `description:"修改时间"`
  200. IsStop int `description:"是否停更:1:停更,0:未停更"`
  201. TerminalCode string `description:"终端编码"`
  202. }
  203. // GetIndexRefreshAllByMergeFile 根据合并文件去分组查询需要刷新的文件
  204. func (m *BaseFromMysteelChemicalIndex) GetIndexRefreshAllByMergeFile() (items []*BaseFromMysteelChemicalIndex, err error) {
  205. o := orm.NewOrm()
  206. sql := `SELECT * FROM base_from_mysteel_chemical_index WHERE merge_file_path <>'' AND base_from_mysteel_chemical_classify_id NOT IN(54,55,56,57) GROUP BY merge_file_path`
  207. _, err = o.Raw(sql).QueryRows(&items)
  208. return
  209. }
  210. func (m *BaseFromMysteelChemicalIndex) GetIndexItem(indexCode string) (item *BaseFromMysteelChemicalIndex, err error) {
  211. o := orm.NewOrm()
  212. sql := `SELECT * FROM base_from_mysteel_chemical_index WHERE index_code = ? `
  213. err = o.Raw(sql, indexCode).QueryRow(&item)
  214. return
  215. }
  216. func (m *BaseFromMysteelChemicalIndex) GetIndexCreate(terminalCode string) (items []*BaseFromMysteelChemicalIndex, err error) {
  217. o := orm.NewOrm()
  218. endTime := time.Now().Add(-2 * time.Minute).Format(utils.FormatDateTime)
  219. sql := `SELECT * FROM base_from_mysteel_chemical_index WHERE index_name = '' AND create_time <= ? AND terminal_code = ? `
  220. _, err = o.Raw(sql, endTime, terminalCode).QueryRows(&items)
  221. return
  222. }
  223. // Add 新增
  224. func (m *BaseFromMysteelChemicalIndex) Add() (err error) {
  225. o := orm.NewOrm()
  226. _, err = o.Insert(m)
  227. return
  228. }
  229. func (m *BaseFromMysteelChemicalIndex) Update(cols []string) (err error) {
  230. o := orm.NewOrm()
  231. _, err = o.Update(m, cols...)
  232. return
  233. }
  234. // GetNoMergeIndexByFrequencyCount 获取未合并的指标总数量
  235. func (m *BaseFromMysteelChemicalIndex) GetNoMergeIndexByFrequencyCount(frequency string) (total int64, err error) {
  236. o := orm.NewOrm()
  237. sql := `SELECT count(1) AS total FROM base_from_mysteel_chemical_index WHERE frequency = ? AND merge_file_path = '' AND index_name NOT LIKE "%停%" `
  238. err = o.Raw(sql, frequency).QueryRow(&total)
  239. return
  240. }
  241. // GetMaxFileIndexReq 获取最大文件数请求
  242. type GetMaxFileIndexReq struct {
  243. Frequency string
  244. EndDate string
  245. TerminalCode string `description:"终端编码"`
  246. }
  247. // GetMaxFileIndex 获取最大文件的
  248. func (m *BaseFromMysteelChemicalIndex) GetMaxFileIndex(frequency string) (item *BaseFromMysteelChemicalIndex, err error) {
  249. o := orm.NewOrm()
  250. sql := `SELECT * FROM base_from_mysteel_chemical_index WHERE frequency=? AND index_name NOT LIKE "%停%" ORDER BY file_index DESC `
  251. err = o.Raw(sql, frequency).QueryRow(&item)
  252. return
  253. }
  254. // GetNoMergeIndexByFrequencyPageReq 获取未合并的指标列表
  255. type GetNoMergeIndexByFrequencyPageReq struct {
  256. Frequency string
  257. Limit int
  258. }
  259. // GetNoMergeIndexByFrequencyPage 获取未合并的指标列表
  260. func (m *BaseFromMysteelChemicalIndex) GetNoMergeIndexByFrequencyPage(frequency string, limit int) (items []*BaseFromMysteelChemicalIndex, err error) {
  261. o := orm.NewOrm()
  262. sql := `SELECT * FROM base_from_mysteel_chemical_index WHERE frequency=? AND merge_file_path = '' limit ? `
  263. _, err = o.Raw(sql, frequency, limit).QueryRows(&items)
  264. return
  265. }
  266. type GetIndexCountByMergeFilePathReq struct {
  267. MergeFilePath string
  268. }
  269. func (m *BaseFromMysteelChemicalIndex) GetIndexCountByMergeFilePath(mergeFilePath string) (total int64, err error) {
  270. o := orm.NewOrm()
  271. sql := `SELECT count(1) AS total FROM base_from_mysteel_chemical_index WHERE merge_file_path = ?`
  272. err = o.Raw(sql, mergeFilePath).QueryRow(&total)
  273. return
  274. }
  275. // GetIndexRefreshMethanolByMergeFile 根据合并文件去分组查询需要刷新的甲醇文件
  276. func (m *BaseFromMysteelChemicalIndex) GetIndexRefreshMethanolByMergeFile() (items []*BaseFromMysteelChemicalIndex, err error) {
  277. o := orm.NewOrm()
  278. sql := `SELECT * FROM base_from_mysteel_chemical_index WHERE merge_file_path <> '' AND base_from_mysteel_chemical_classify_id IN(54,55,56,57) GROUP BY merge_file_path`
  279. _, err = o.Raw(sql).QueryRows(&items)
  280. return
  281. }
  282. // GetIndexRefreshMethanolByTimely 获取需要及时刷新的文件
  283. func (m *BaseFromMysteelChemicalIndex) GetIndexRefreshMethanolByTimely() (items []*BaseFromMysteelChemicalIndex, err error) {
  284. o := orm.NewOrm()
  285. sql := `SELECT * FROM base_from_mysteel_chemical_index WHERE merge_file_path <> '' AND is_refresh = 1 GROUP BY merge_file_path`
  286. _, err = o.Raw(sql).QueryRows(&items)
  287. return
  288. }
  289. type GetIndexByFrequencyReq struct {
  290. Frequency string
  291. }
  292. // GetIndexByFrequency 根据频率获取指标
  293. func (m *BaseFromMysteelChemicalIndex) GetIndexByFrequency(frequency string) (items []*BaseFromMysteelChemicalIndex, err error) {
  294. o := orm.NewOrm()
  295. sql := `SELECT * FROM base_from_mysteel_chemical_index WHERE frequency = ? AND merge_file_path = '' `
  296. _, err = o.Raw(sql, frequency).QueryRows(&items)
  297. return
  298. }
  299. type GetIndexRefreshWeekReq struct {
  300. NowWeekZn string
  301. StartTime string
  302. EndTime string
  303. }
  304. func (m *BaseFromMysteelChemicalIndex) GetIndexRefreshWeek(nowWeekZn, startTime, endTime string) (items []*BaseFromMysteelChemicalIndex, err error) {
  305. where := `update_week = ? AND ((update_time >= ? AND update_time <= ?) OR (update_time2 >= ? AND update_time2 <= ?)) `
  306. o := orm.NewOrm()
  307. sql := fmt.Sprintf(`SELECT * FROM base_from_mysteel_chemical_index WHERE %s `, where)
  308. _, err = o.Raw(sql, nowWeekZn, startTime, endTime, startTime, endTime).QueryRows(&items)
  309. return
  310. }
  311. // UpdateIndexReq 根据编码更新指标请求体
  312. type UpdateIndexReq struct {
  313. Item *BaseFromMysteelChemicalIndex
  314. UpdateCols []string
  315. }
  316. func (m *BaseFromMysteelChemicalIndex) UpdateIndex(item *BaseFromMysteelChemicalIndex, updateCols []string) (err error) {
  317. if item == nil {
  318. return
  319. }
  320. if len(updateCols) == 0 {
  321. return
  322. }
  323. o := orm.NewOrm()
  324. _, err = o.Update(item, updateCols...)
  325. return
  326. }
  327. // MultiUpdateIndex 批量更新指标
  328. func (m *BaseFromMysteelChemicalIndex) MultiUpdateIndex(items []UpdateIndexReq) (err error) {
  329. if len(items) == 0 {
  330. return
  331. }
  332. o := orm.NewOrm()
  333. for _, v := range items {
  334. if len(v.UpdateCols) == 0 {
  335. continue
  336. }
  337. if v.Item == nil {
  338. continue
  339. }
  340. if _, e := o.Update(v.Item, v.UpdateCols...); e != nil {
  341. err = fmt.Errorf("update err: %s", e.Error())
  342. return
  343. }
  344. }
  345. return
  346. }
  347. type HandleMysteelIndex struct {
  348. IndexName string `description:"指标名称"`
  349. IndexCode string `description:"指标编码"`
  350. Unit string `description:"单位"`
  351. Source string `description:"数据来源"`
  352. Frequency string `description:"频度"`
  353. StartDate string `description:"开始日期"`
  354. EndDate string `description:"结束日期"`
  355. Describe string `description:"指标描述"`
  356. UpdateDate string `description:"更新日期"`
  357. ExcelDataMap map[string]string
  358. }
  359. type HandleMysteelIndexResp struct {
  360. List []*HandleMysteelIndex
  361. }
  362. // 钢联化工指标数据
  363. type BaseFromMysteelChemicalData struct {
  364. BaseFromMysteelChemicalDataId int64 `orm:"column(base_from_mysteel_chemical_data_id);pk"`
  365. BaseFromMysteelChemicalIndexId int64
  366. IndexCode string
  367. DataTime time.Time
  368. Value string
  369. UpdateDate string
  370. CreateTime time.Time `description:"创建时间"`
  371. ModifyTime time.Time `description:"修改时间"`
  372. }
  373. // GetIndexDataList 根据指标编码获取数据
  374. func (d *BaseFromMysteelChemicalData) GetIndexDataList(indexCode string) (item []*BaseFromMysteelChemicalData, err error) {
  375. o := orm.NewOrm()
  376. sql := `SELECT * FROM base_from_mysteel_chemical_data WHERE index_code = ? `
  377. _, err = o.Raw(sql, indexCode).QueryRows(&item)
  378. return
  379. }
  380. // Update 修改
  381. func (r *BaseFromMysteelChemicalData) Update(updateCols []string) (err error) {
  382. o := orm.NewOrm()
  383. _, err = o.Update(r, updateCols...)
  384. return
  385. }
  386. // Add 新增
  387. func (r *BaseFromMysteelChemicalData) Add(list []BaseFromMysteelChemicalData) (err error) {
  388. o := orm.NewOrm()
  389. _, err = o.InsertMulti(len(list), list)
  390. return
  391. }
  392. type AddMysteelIndexResp struct {
  393. EdbCode string `description:"指标编码"`
  394. BaseFromMysteelChemicalClassifyId int `description:"指标分类"`
  395. SysUserId int `description:"操作人id"`
  396. SysUserRealName string `description:"操作人真实名称"`
  397. }
  398. type GetIndexDetailReq struct {
  399. IndexCode string `description:"指标编码"`
  400. }
  401. // GetIndexCountByFrequency 获取未合并的指标总数量
  402. func (m *BaseFromMysteelChemicalIndex) GetIndexCountByFrequency(condition string, pars []interface{}) (total int64, err error) {
  403. o := orm.NewOrm()
  404. sql := `SELECT count(1) AS total FROM base_from_mysteel_chemical_index WHERE 1=1 `
  405. if condition != "" {
  406. sql += condition
  407. }
  408. err = o.Raw(sql, pars).QueryRow(&total)
  409. return
  410. }
  411. // GetNoMergeIndexByFrequencyPage 获取未合并的指标列表
  412. func (m *BaseFromMysteelChemicalIndex) GetIndexLimitByFrequency(frequency string, limit int) (items []*BaseFromMysteelChemicalIndex, err error) {
  413. o := orm.NewOrm()
  414. sql := `SELECT * FROM base_from_mysteel_chemical_index WHERE frequency=? limit ? `
  415. _, err = o.Raw(sql, frequency, limit).QueryRows(&items)
  416. return
  417. }
  418. type GetIndexPageByFrequencyPageReq struct {
  419. Frequency string
  420. EndDate string
  421. StartSize int
  422. PageSize int
  423. TerminalCode string `description:"终端编码"`
  424. }
  425. // GetNoMergeIndexByFrequencyPage 获取未合并的指标列表
  426. func (m *BaseFromMysteelChemicalIndex) GetIndexPageByFrequency(condition string, pars []interface{}, startSize, pageSize int) (items []*BaseFromMysteelChemicalIndex, err error) {
  427. o := orm.NewOrm()
  428. sql := `SELECT * FROM base_from_mysteel_chemical_index WHERE 1=1 `
  429. if condition != "" {
  430. sql += condition
  431. }
  432. sql += ` LIMIT ?,? `
  433. _, err = o.Raw(sql, pars, startSize, pageSize).QueryRows(&items)
  434. return
  435. }
  436. func (d *BaseFromMysteelChemicalData) GetMysteelIndexInfoMaxAndMinInfo(indexCode string) (item *EdbInfoMaxAndMinInfo, err error) {
  437. o := orm.NewOrm()
  438. sql := ` SELECT MIN(data_time) AS min_date,MAX(data_time) AS max_date,MIN(value) AS min_value,MAX(value) AS max_value FROM base_from_mysteel_chemical_data WHERE index_code=? `
  439. err = o.Raw(sql, indexCode).QueryRow(&item)
  440. return
  441. }
  442. func (d *BaseFromMysteelChemicalData) ModifyMysteelIndexMaxAndMinInfo(indexCode string, item *EdbInfoMaxAndMinInfo) (err error) {
  443. o := orm.NewOrm()
  444. sql := ` UPDATE base_from_mysteel_chemical_index SET start_date=?,end_date=?,modify_time=NOW() WHERE index_code=? `
  445. _, err = o.Raw(sql, item.MinDate, item.MaxDate, indexCode).Exec()
  446. return
  447. }