base_from_python.go 13 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334335336337338339340341342343344345346347348349350351352353354355356357358359360361362363364365366367368369370371372373374375376377378379380381382383384385386387388389390391392393394395396397398399400401402403
  1. package models
  2. import (
  3. "errors"
  4. "eta/eta_index_lib/global"
  5. "eta/eta_index_lib/services/alarm_msg"
  6. "eta/eta_index_lib/utils"
  7. "fmt"
  8. "github.com/shopspring/decimal"
  9. "gorm.io/gorm"
  10. "strings"
  11. "time"
  12. )
  13. // EdbDataPython python指标数据结构体
  14. type EdbDataPython struct {
  15. EdbDataId int `gorm:"column:edb_data_id;primaryKey"`
  16. //EdbDataId int `orm:"column(edb_data_id);pk"`
  17. EdbInfoId int
  18. EdbCode string
  19. DataTime string
  20. Value float64
  21. CreateTime time.Time
  22. ModifyTime time.Time
  23. DataTimestamp int64
  24. }
  25. func (e *EdbDataPython) AfterFind(db *gorm.DB) (err error) {
  26. e.DataTime = utils.GormDateStrToDateStr(e.DataTime)
  27. return
  28. }
  29. // EdbDataFromPython 通过python代码获取到的指标数据
  30. type EdbDataFromPython struct {
  31. Date map[int]string `json:"date"`
  32. Value map[int]float64 `json:"value"`
  33. }
  34. // AddPythonEdb 新增python运算指标
  35. func AddPythonEdb(edbInfo *EdbInfo, item EdbDataFromPython, edbInfoList []*EdbInfo) (err error) {
  36. //o := orm.NewOrm()
  37. //添加指标关系
  38. for _, tmpEdbInfo := range edbInfoList {
  39. calculateMappingItem := new(EdbInfoCalculateMapping)
  40. calculateMappingItem.CreateTime = time.Now()
  41. calculateMappingItem.ModifyTime = time.Now()
  42. calculateMappingItem.Sort = 1
  43. calculateMappingItem.EdbCode = edbInfo.EdbCode
  44. calculateMappingItem.EdbInfoId = edbInfo.EdbInfoId
  45. calculateMappingItem.FromEdbInfoId = tmpEdbInfo.EdbInfoId
  46. calculateMappingItem.FromEdbCode = tmpEdbInfo.EdbCode
  47. calculateMappingItem.FromEdbName = tmpEdbInfo.EdbName
  48. calculateMappingItem.FromSource = tmpEdbInfo.Source
  49. calculateMappingItem.FromSourceName = tmpEdbInfo.SourceName
  50. calculateMappingItem.Source = edbInfo.Source
  51. calculateMappingItem.SourceName = edbInfo.SourceName
  52. //_, err = o.Insert(calculateMappingItem)
  53. err = global.DEFAULT_DB.Create(&calculateMappingItem).Error
  54. if err != nil {
  55. return
  56. }
  57. }
  58. var isAdd bool
  59. addSql := ` INSERT INTO edb_data_python (edb_info_id,edb_code,data_time,value,create_time,modify_time,data_timestamp) values `
  60. addSql = utils.ReplaceDriverKeywords("", addSql)
  61. for k, dateTimeStr := range item.Date {
  62. //格式化时间
  63. currentDate, tmpErr := time.ParseInLocation(utils.FormatDate, dateTimeStr, time.Local)
  64. if tmpErr != nil {
  65. err = tmpErr
  66. return
  67. }
  68. timestamp := currentDate.UnixNano() / 1e6
  69. timestampStr := fmt.Sprintf("%d", timestamp)
  70. //值
  71. val := item.Value[k]
  72. saveVal := utils.SubFloatToString(val, 20)
  73. addSql += GetAddSql(fmt.Sprint(edbInfo.EdbInfoId), edbInfo.EdbCode, dateTimeStr, timestampStr, saveVal)
  74. isAdd = true
  75. }
  76. if isAdd {
  77. addSql = strings.TrimRight(addSql, ",")
  78. //_, err = o.Raw(addSql).Exec()
  79. err = global.DEFAULT_DB.Exec(addSql).Error
  80. if err != nil {
  81. return
  82. }
  83. }
  84. return
  85. }
  86. // EditPythonEdb 编辑python运算指标
  87. func EditPythonEdb(edbInfoId int, edbCode string, item EdbDataFromPython) (err error) {
  88. //o := orm.NewOrm()
  89. var isAdd bool
  90. addSql := ` INSERT INTO edb_data_python (edb_info_id,edb_code,data_time,value,create_time,modify_time,data_timestamp) values `
  91. addSql = utils.ReplaceDriverKeywords("", addSql)
  92. for k, dateTimeStr := range item.Date {
  93. //格式化时间
  94. currentDate, tmpErr := time.ParseInLocation(utils.FormatDate, dateTimeStr, time.Local)
  95. if tmpErr != nil {
  96. err = tmpErr
  97. return
  98. }
  99. timestamp := currentDate.UnixNano() / 1e6
  100. timestampStr := fmt.Sprintf("%d", timestamp)
  101. //值
  102. val := item.Value[k]
  103. saveVal := utils.SubFloatToString(val, 20)
  104. addSql += GetAddSql(fmt.Sprint(edbInfoId), edbCode, dateTimeStr, timestampStr, saveVal)
  105. isAdd = true
  106. }
  107. if isAdd {
  108. addSql = strings.TrimRight(addSql, ",")
  109. //_, err = o.Raw(addSql).Exec()
  110. err = global.DEFAULT_DB.Exec(addSql).Error
  111. if err != nil {
  112. return
  113. }
  114. }
  115. return
  116. }
  117. // RefreshAllPythonEdb 刷新所有 python运算指标
  118. func RefreshAllPythonEdb(edbInfo *EdbInfo, item EdbDataFromPython) (err error) {
  119. //o := orm.NewOrm()
  120. //to, err := o.Begin()
  121. to := global.DEFAULT_DB.Begin()
  122. if err != nil {
  123. return
  124. }
  125. defer func() {
  126. if err != nil {
  127. fmt.Println("RefreshAllPythonEdb,Err:" + err.Error())
  128. _ = to.Rollback()
  129. } else {
  130. _ = to.Commit()
  131. }
  132. }()
  133. pythonDataMap := make(map[string]float64)
  134. pythonDate := make([]string, 0)
  135. for k, dateTimeStr := range item.Date {
  136. pythonDataMap[dateTimeStr] = item.Value[k]
  137. pythonDate = append(pythonDate, dateTimeStr)
  138. }
  139. //查询当前指标现有的数据
  140. var condition string
  141. var pars []interface{}
  142. condition += " AND edb_info_id=? "
  143. pars = append(pars, edbInfo.EdbInfoId)
  144. //所有的数据
  145. dataList, err := GetAllEdbDataPythonByEdbInfoId(edbInfo.EdbInfoId)
  146. if err != nil {
  147. return err
  148. }
  149. //待修改的指标数据map(index:日期,value:值)
  150. updateEdbDataMap := make(map[string]float64)
  151. removeDateList := make([]string, 0) //需要删除的日期
  152. for _, v := range dataList {
  153. currDataTime := v.DataTime
  154. pythonData, ok := pythonDataMap[currDataTime]
  155. if !ok {
  156. // 如果python运算出来的数据中没有该日期,那么需要移除该日期的数据
  157. removeDateList = append(removeDateList, currDataTime)
  158. } else {
  159. currValue, _ := decimal.NewFromFloat(pythonData).Truncate(4).Float64() //保留4位小数
  160. //如果计算出来的值与库里面的值不匹配,那么就去修改该值
  161. if v.Value != currValue {
  162. //将计算后的数据存入待拼接指标map里面,以便后续计算
  163. updateEdbDataMap[currDataTime] = currValue
  164. }
  165. }
  166. //移除python指标数据中当天的日期
  167. delete(pythonDataMap, currDataTime)
  168. }
  169. //sort.Strings(tbzEdbDataTimeList)
  170. //新增的数据入库
  171. {
  172. addDataList := make([]*EdbDataPython, 0)
  173. for dataTime, dataValue := range pythonDataMap {
  174. //时间戳
  175. currentDate, _ := time.ParseInLocation(utils.FormatDate, dataTime, time.Local)
  176. timestamp := currentDate.UnixNano() / 1e6
  177. edbDataPython := &EdbDataPython{
  178. EdbInfoId: edbInfo.EdbInfoId,
  179. EdbCode: edbInfo.EdbCode,
  180. DataTime: dataTime,
  181. Value: dataValue,
  182. CreateTime: time.Now(),
  183. ModifyTime: time.Now(),
  184. DataTimestamp: timestamp,
  185. }
  186. addDataList = append(addDataList, edbDataPython)
  187. }
  188. //最后如果还有需要新增的数据,那么就统一入库
  189. if len(addDataList) > 0 {
  190. //_, tmpErr := o.InsertMulti(len(addDataList), addDataList)
  191. tmpErr := to.CreateInBatches(addDataList, len(addDataList)).Error
  192. if tmpErr != nil {
  193. err = tmpErr
  194. return
  195. }
  196. }
  197. }
  198. //删除已经不存在的累计同比拼接指标数据(由于同比值当日的数据删除了)
  199. {
  200. if len(removeDateList) > 0 {
  201. removeDateStr := strings.Join(removeDateList, `','`)
  202. removeDateStr = `'` + removeDateStr + `'`
  203. //如果拼接指标变更了,那么需要删除所有的指标数据
  204. tableName := GetEdbDataTableName(edbInfo.Source, edbInfo.SubSource)
  205. sql := fmt.Sprintf(` DELETE FROM %s WHERE edb_info_id = ? and data_time in (%s) `, tableName, removeDateStr)
  206. //_, err = o.Raw(sql, edbInfo.EdbInfoId).Exec()
  207. err = to.Exec(sql, edbInfo.EdbInfoId).Error
  208. if err != nil {
  209. err = errors.New("删除不存在的Python运算指标数据失败,Err:" + err.Error())
  210. return
  211. }
  212. }
  213. }
  214. //修改现有的数据中对应的值
  215. {
  216. tableName := GetEdbDataTableName(edbInfo.Source, edbInfo.SubSource)
  217. for edbDate, edbDataValue := range updateEdbDataMap {
  218. sql := fmt.Sprintf(` UPDATE %s set value = ?,modify_time=now() WHERE edb_info_id = ? and data_time = ? `, tableName)
  219. //_, err = o.Raw(sql, edbDataValue, edbInfo.EdbInfoId, edbDate).Exec()
  220. err = to.Exec(sql, edbDataValue, edbInfo.EdbInfoId, edbDate).Error
  221. if err != nil {
  222. err = errors.New("更新现有的Python运算指标数据失败,Err:" + err.Error())
  223. return
  224. }
  225. }
  226. }
  227. return
  228. }
  229. // EditEdbInfoCalculateMapping 更新关联关系表
  230. func EditEdbInfoCalculateMapping(edbInfo *EdbInfo, edbInfoList []*EdbInfo) (err error) {
  231. //o := orm.NewOrm()
  232. var existCondition string
  233. var existPars []interface{}
  234. existCondition += " AND edb_info_id=? "
  235. existPars = append(existPars, edbInfo.EdbInfoId)
  236. //查询出所有的关联指标
  237. existList, err := GetEdbInfoCalculateListByCondition(existCondition, existPars)
  238. if err != nil {
  239. err = fmt.Errorf("判断指标是否改变失败,Err:" + err.Error())
  240. return
  241. }
  242. existEdbInfoIdMap := make(map[int]int)
  243. isOpEdbInfoIdMap := make(map[int]int)
  244. for _, v := range existList {
  245. existEdbInfoIdMap[v.FromEdbInfoId] = v.FromEdbInfoId
  246. }
  247. //添加指标关系
  248. for _, tmpEdbInfo := range edbInfoList {
  249. //如果该指标id已经处理过了,那么就不处理了
  250. if _, ok := isOpEdbInfoIdMap[tmpEdbInfo.EdbInfoId]; ok {
  251. continue
  252. }
  253. if _, ok := existEdbInfoIdMap[tmpEdbInfo.EdbInfoId]; ok {
  254. //如果存在,那么就移除map里面的东西
  255. delete(existEdbInfoIdMap, tmpEdbInfo.EdbInfoId)
  256. isOpEdbInfoIdMap[tmpEdbInfo.EdbInfoId] = tmpEdbInfo.EdbInfoId
  257. } else {
  258. calculateMappingItem := new(EdbInfoCalculateMapping)
  259. calculateMappingItem.CreateTime = time.Now()
  260. calculateMappingItem.ModifyTime = time.Now()
  261. calculateMappingItem.Sort = 1
  262. calculateMappingItem.EdbCode = edbInfo.EdbCode
  263. calculateMappingItem.EdbInfoId = edbInfo.EdbInfoId
  264. calculateMappingItem.FromEdbInfoId = tmpEdbInfo.EdbInfoId
  265. calculateMappingItem.FromEdbCode = tmpEdbInfo.EdbCode
  266. calculateMappingItem.FromEdbName = tmpEdbInfo.EdbName
  267. calculateMappingItem.FromSource = tmpEdbInfo.Source
  268. calculateMappingItem.FromSourceName = tmpEdbInfo.SourceName
  269. calculateMappingItem.Source = edbInfo.Source
  270. calculateMappingItem.SourceName = edbInfo.SourceName
  271. //_, err = o.Insert(calculateMappingItem)
  272. err = global.DEFAULT_DB.Create(&calculateMappingItem).Error
  273. if err != nil {
  274. return
  275. }
  276. }
  277. }
  278. for _, v := range existEdbInfoIdMap {
  279. //删除,计算指标关联的,基础指标的关联关系
  280. sql := ` DELETE FROM edb_info_calculate_mapping WHERE edb_info_id = ? and from_edb_info_id=?`
  281. //_, err = o.Raw(sql, edbInfo.EdbInfoId, v).Exec()
  282. err = global.DEFAULT_DB.Exec(sql, edbInfo.EdbInfoId, v).Error
  283. if err != nil {
  284. err = errors.New("删除计算指标关联关系失败,Err:" + err.Error())
  285. return
  286. }
  287. }
  288. return
  289. }
  290. // GetAllEdbDataPythonByEdbInfoId 根据指标id获取全部的数据
  291. func GetAllEdbDataPythonByEdbInfoId(edbInfoId int) (items []*EdbDataPython, err error) {
  292. //o := orm.NewOrm()
  293. sql := ` SELECT * FROM edb_data_python WHERE edb_info_id=? ORDER BY data_time DESC `
  294. //_, err = o.Raw(sql, edbInfoId).QueryRows(&items)
  295. err = global.DEFAULT_DB.Raw(sql, edbInfoId).Find(&items).Error
  296. return
  297. }
  298. // EdbInfoPythonSaveReq 计算(运算)指标请求参数
  299. type EdbInfoPythonSaveReq struct {
  300. AdminId int `description:"添加人id"`
  301. AdminName string `description:"添加人名称"`
  302. EdbName string `description:"指标名称"`
  303. Frequency string `description:"频率"`
  304. Unit string `description:"单位"`
  305. ClassifyId int `description:"分类id"`
  306. CalculateFormula string `description:"计算公式"`
  307. EdbInfoIdArr []struct {
  308. EdbInfoId int `description:"指标id"`
  309. FromTag string `description:"指标对应标签"`
  310. }
  311. }
  312. // ExecPythonEdbReq 执行python代码运算指标的请求参数
  313. type ExecPythonEdbReq struct {
  314. PythonCode string `description:"python代码"`
  315. }
  316. // AddPythonEdbReq 添加python代码运算指标的请求参数
  317. type AddPythonEdbReq struct {
  318. AdminId int `description:"添加人id"`
  319. AdminName string `description:"添加人名称"`
  320. EdbInfoId int `description:"指标id"`
  321. EdbName string `description:"指标名称"`
  322. Frequency string `description:"频度"`
  323. Unit string `description:"单位"`
  324. ClassifyId int `description:"分类id"`
  325. PythonCode string `description:"python代码"`
  326. }
  327. // AnalysisPythonCode 解析Python代码,获取关联code
  328. func AnalysisPythonCode(pythonCode, edbName string) (edbInfoList []*EdbInfo) {
  329. tmpEdbCodeList := make([]string, 0) //临时指标code
  330. edbCodeLen := 0 //指标数
  331. tmpList := strings.Split(pythonCode, "\n")
  332. for _, v := range tmpList {
  333. if strings.Contains(v, "edb_code") {
  334. edbCodeLen++
  335. tmpCodeStrList := strings.Split(v, "edb_code")
  336. if len(tmpCodeStrList) > 1 {
  337. //根据单引号获取
  338. tmpCodeStrList2 := strings.Split(tmpCodeStrList[1], "'")
  339. if len(tmpCodeStrList2) > 1 {
  340. if tmpCodeStrList2[1] != "" {
  341. tmpEdbCodeList = append(tmpEdbCodeList, tmpCodeStrList2[1])
  342. }
  343. }
  344. //根据双引号获取
  345. tmpCodeStrList3 := strings.Split(tmpCodeStrList[1], `"`)
  346. if len(tmpCodeStrList3) > 1 {
  347. if tmpCodeStrList3[1] != "" {
  348. tmpEdbCodeList = append(tmpEdbCodeList, tmpCodeStrList3[1])
  349. }
  350. }
  351. }
  352. }
  353. }
  354. for _, v := range tmpEdbCodeList {
  355. fmt.Println(v)
  356. item, _ := GetEdbInfoOnlyByEdbCode(v)
  357. if item != nil {
  358. edbInfoList = append(edbInfoList, item)
  359. }
  360. }
  361. if len(edbInfoList) != edbCodeLen {
  362. //code匹配失败,需要短信提醒
  363. go alarm_msg.SendAlarmMsg(fmt.Sprintf("python代码关联指标匹配失败,指标名称:%s;实际关联%d个,匹配上%d个", edbName, edbCodeLen, len(edbInfoList)), 3)
  364. }
  365. return
  366. }