base_from_python.go 13 KB

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