edb_info_stat.go 9.7 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324
  1. package services
  2. import (
  3. "eta/eta_index_lib/models"
  4. "eta/eta_index_lib/models/data_stat"
  5. "eta/eta_index_lib/services/alarm_msg"
  6. "eta/eta_index_lib/utils"
  7. "fmt"
  8. "time"
  9. )
  10. // AddEdbInfoUpdateLog 添加指标编辑/刷新日志
  11. func AddEdbInfoUpdateLog(edbInfoId int, updateResult int, updateFailedReason string, dataUpdateResult int, dataUpdateFailedReason string, isSourceRefresh int) (err error) {
  12. var edbInfo *models.EdbInfo
  13. if edbInfoId > 0 {
  14. // 获取指标详情
  15. edbInfo, err = models.GetEdbInfoById(edbInfoId)
  16. if err != nil {
  17. err = fmt.Errorf("指标不存在")
  18. return
  19. }
  20. log := new(data_stat.EdbInfoUpdateLog)
  21. log.EdbInfoId = edbInfo.EdbInfoId
  22. log.SourceName = edbInfo.SourceName
  23. log.Source = edbInfo.Source
  24. log.EdbCode = edbInfo.EdbCode
  25. log.EdbName = edbInfo.EdbName
  26. log.EdbNameSource = edbInfo.EdbNameSource
  27. log.Frequency = edbInfo.Frequency
  28. log.Unit = edbInfo.Unit
  29. log.StartDate = edbInfo.StartDate
  30. log.EndDate = edbInfo.EndDate
  31. log.SysUserId = edbInfo.SysUserId
  32. log.SysUserRealName = edbInfo.SysUserRealName
  33. log.UniqueCode = edbInfo.UniqueCode
  34. log.EdbCreateTime = edbInfo.CreateTime
  35. log.EdbModifyTime = edbInfo.ModifyTime
  36. log.CreateTime = time.Now()
  37. log.LatestDate = edbInfo.LatestDate
  38. log.LatestValue = edbInfo.LatestValue
  39. log.TerminalCode = edbInfo.TerminalCode
  40. log.UpdateResult = updateResult
  41. log.UpdateFailedReason = updateFailedReason
  42. log.DataUpdateTime = edbInfo.DataUpdateTime
  43. log.ErDataUpdateDate = edbInfo.ErDataUpdateDate
  44. log.DataUpdateResult = dataUpdateResult
  45. log.DataUpdateFailedReason = dataUpdateFailedReason
  46. log.IsSourceRefresh = isSourceRefresh
  47. _, err = data_stat.AddEdbUpdateLog(log)
  48. if err != nil {
  49. err = fmt.Errorf("新增指标更新日志失败,Err: %s", err)
  50. return
  51. }
  52. }
  53. return
  54. }
  55. // SetMysteelChemicalEdbInfoUpdateStat 定时统计钢联化工的数据源明细表
  56. func SetMysteelChemicalEdbInfoUpdateStat() (err error) {
  57. defer func() {
  58. if err != nil {
  59. tips := fmt.Sprintf("钢联数据刷新-ETA指标刷新统计异常 Err: %s", err.Error())
  60. alarm_msg.SendAlarmMsg(tips, 3)
  61. }
  62. }()
  63. //查询钢联的所有指标信息
  64. condition := " and source = ? "
  65. var pars []interface{}
  66. pars = append(pars, utils.DATA_SOURCE_MYSTEEL_CHEMICAL)
  67. edbList, err := models.GetEdbInfoByCondition(condition, pars, 0)
  68. if err != nil {
  69. err = fmt.Errorf("查询钢联化工指标信息出错,err: %s", err)
  70. return
  71. }
  72. nowTime := time.Now()
  73. today := time.Now().Format(utils.FormatDate)
  74. todayT, _ := time.ParseInLocation(utils.FormatDate, today, time.Local)
  75. nextDay := time.Now().AddDate(0, 0, 1).Format(utils.FormatDate)
  76. //查询当日所有钢联指标的终端更新记录
  77. updateLogList, err := data_stat.GetEdbUpdateSourceLogByCreateDate(utils.DATA_SOURCE_MYSTEEL_CHEMICAL, today, nextDay)
  78. if err != nil {
  79. err = fmt.Errorf("查询钢联化工指标终端更新日志报错,err: %s", err)
  80. return
  81. }
  82. updateLogMap := make(map[int]*data_stat.EdbInfoUpdateLog)
  83. if len(updateLogList) > 0 {
  84. for _, v := range updateLogList {
  85. if _, ok := updateLogMap[v.EdbInfoId]; !ok {
  86. updateLogMap[v.EdbInfoId] = v
  87. }
  88. }
  89. }
  90. statCond := " and source = ? and create_time >= ? and create_time < ?"
  91. var statPars []interface{}
  92. statPars = append(statPars, utils.DATA_SOURCE_MYSTEEL_CHEMICAL, today, nextDay)
  93. //查询当日钢联所有的刷新记录
  94. updateStatList, err := data_stat.GetEdbUpdateStatByCondition(statCond, statPars)
  95. if err != nil {
  96. err = fmt.Errorf("查询钢联化工数据源明细记录统计报错,err: %s", err)
  97. return
  98. }
  99. updateStatMap := make(map[int]*data_stat.EdbInfoUpdateStat)
  100. if len(updateStatList) > 0 {
  101. for _, v := range updateStatList {
  102. updateStatMap[v.EdbInfoId] = v
  103. }
  104. }
  105. logStat := new(data_stat.EdbInfoUpdateStat)
  106. //组装新增数据
  107. addList := make([]*data_stat.EdbInfoUpdateStat, 0)
  108. modifyList := make([]*data_stat.EdbInfoUpdateStat, 0)
  109. if len(edbList) > 0 {
  110. for _, v := range edbList {
  111. tmp := &data_stat.EdbInfoUpdateStat{
  112. EdbInfoId: v.EdbInfoId,
  113. SourceName: v.SourceName,
  114. Source: v.Source,
  115. EdbCode: v.EdbCode,
  116. EdbName: v.EdbName,
  117. EdbNameSource: v.EdbNameSource,
  118. Frequency: v.Frequency,
  119. Unit: v.Unit,
  120. StartDate: v.StartDate,
  121. EndDate: v.EndDate,
  122. SysUserId: v.SysUserId,
  123. SysUserRealName: v.SysUserRealName,
  124. UniqueCode: v.UniqueCode,
  125. EdbCreateTime: v.CreateTime,
  126. EdbModifyTime: v.ModifyTime,
  127. LatestDate: v.LatestDate,
  128. LatestValue: v.LatestValue,
  129. TerminalCode: v.TerminalCode,
  130. ModifyTime: nowTime,
  131. }
  132. needRefresh, _ := checkMySteelEdbInfoNeedRefresh(v.Frequency)
  133. tmp.NeedRefresh = needRefresh
  134. // 判断是否当日新增
  135. if v.CreateTime.After(todayT) || v.CreateTime == todayT {
  136. tmp.IsAdd = 1
  137. } else {
  138. tmp.IsAdd = 2
  139. }
  140. if up, ok := updateLogMap[v.EdbInfoId]; ok {
  141. tmp.DataUpdateTime = up.DataUpdateTime
  142. tmp.ErDataUpdateDate = up.ErDataUpdateDate
  143. tmp.DataUpdateResult = up.DataUpdateResult
  144. tmp.DataUpdateFailedReason = up.DataUpdateFailedReason
  145. tmp.HasRefresh = 1
  146. tmp.UpdateResult = up.UpdateResult
  147. tmp.UpdateFailedReason = up.UpdateFailedReason
  148. } else if needRefresh == 1 {
  149. tmp.HasRefresh = 0
  150. tmp.DataUpdateResult = 2
  151. tmp.DataUpdateFailedReason = "服务异常"
  152. }
  153. // 判断是否需要新增还是更新
  154. if exist, ok := updateStatMap[v.EdbInfoId]; ok {
  155. tmp.Id = exist.Id
  156. modifyList = append(modifyList, tmp)
  157. } else {
  158. tmp.CreateTime = nowTime
  159. addList = append(addList, tmp)
  160. }
  161. }
  162. }
  163. //判断当日指标统计数据是否存在,如果存在则更新,不存在则新增
  164. if len(addList) > 0 {
  165. err = logStat.Add(addList)
  166. }
  167. if len(modifyList) > 0 {
  168. err = data_stat.UpdateEdbUpdateStatMulti(modifyList)
  169. }
  170. return
  171. }
  172. func checkMySteelEdbInfoNeedRefresh(frequency string) (needRefresh int, err error) {
  173. now := time.Now()
  174. week := int(now.Weekday())
  175. //日度
  176. if week >= 1 && week <= 6 {
  177. if frequency == "日度" {
  178. needRefresh = 1
  179. return
  180. }
  181. }
  182. time.Sleep(1 * time.Minute)
  183. //周度
  184. if week >= 3 && week <= 6 {
  185. if frequency == "周度" {
  186. needRefresh = 1
  187. return
  188. }
  189. }
  190. day := now.Day() //季度,月度,年度都是每个月1号刷新
  191. if day == 1 {
  192. needRefresh = 1
  193. }
  194. return
  195. }
  196. // SetEdbSourceStat 定时统计数据源汇总表
  197. func SetEdbSourceStat() (err error) {
  198. defer func() {
  199. if err != nil {
  200. tips := fmt.Sprintf("钢联数据刷新-ETA指标刷新统计异常 Err: %s", err.Error())
  201. alarm_msg.SendAlarmMsg(tips, 3)
  202. }
  203. }()
  204. //查询钢联的所有指标信息
  205. nowTime := time.Now()
  206. today := time.Now().Format(utils.FormatDate)
  207. nextDay := time.Now().AddDate(0, 0, 1).Format(utils.FormatDate)
  208. statCond := " and create_time >= ? and create_time < ?"
  209. var statPars []interface{}
  210. statPars = append(statPars, today, nextDay)
  211. //查询当日钢联所有的统计数据
  212. updateStatList, err := data_stat.GetEdbUpdateStatByCondition(statCond, statPars)
  213. if err != nil {
  214. err = fmt.Errorf("查询钢联化工数据源明细记录统计报错,err: %s", err)
  215. return
  216. }
  217. updateStatMap := make(map[string][]*data_stat.EdbInfoUpdateStat)
  218. if len(updateStatList) > 0 {
  219. for _, v := range updateStatList {
  220. updateStatMap[v.TerminalCode] = append(updateStatMap[v.TerminalCode], v)
  221. }
  222. }
  223. cond := " and create_time >= ? and create_time < ?"
  224. var pars []interface{}
  225. pars = append(pars, today, nextDay)
  226. //查询当日钢联所有的统计数据
  227. statList, err := data_stat.GetEdbSourceStatByCondition(cond, pars)
  228. if err != nil {
  229. err = fmt.Errorf("查询钢联化工数据源统计报错,err: %s", err)
  230. return
  231. }
  232. statMap := make(map[string]*data_stat.EdbSourceStat)
  233. if len(statList) > 0 {
  234. for _, v := range statList {
  235. statMap[v.TerminalCode] = v
  236. }
  237. }
  238. // 查询今日被删除的指标数
  239. delList, err := data_stat.GetEdbDeleteLogNumByCreateTime(today, nextDay)
  240. if err != nil {
  241. err = fmt.Errorf("查询今日被删除指标数目报错,err: %s", err)
  242. return
  243. }
  244. delMap := make(map[string]int)
  245. if len(delList) > 0 {
  246. for _, v := range delList {
  247. delMap[v.TerminalCode] = v.Num
  248. }
  249. }
  250. logStat := new(data_stat.EdbSourceStat)
  251. //组装新增数据
  252. addList := make([]*data_stat.EdbSourceStat, 0)
  253. modifyList := make([]*data_stat.EdbSourceStat, 0)
  254. for terminalCode, list := range updateStatMap {
  255. tmp := new(data_stat.EdbSourceStat)
  256. for k, v := range list {
  257. if k == 0 {
  258. tmp.SourceName = v.SourceName
  259. tmp.Source = v.Source
  260. tmp.TerminalCode = v.TerminalCode
  261. tmp.ModifyTime = nowTime
  262. }
  263. tmp.EdbNum = tmp.EdbNum + 1
  264. if v.IsAdd == 1 {
  265. tmp.EdbNewNum = tmp.EdbNewNum + 1
  266. }
  267. if v.NeedRefresh == 1 {
  268. tmp.NeedRefreshNum = tmp.NeedRefreshNum + 1
  269. }
  270. if v.HasRefresh == 1 {
  271. tmp.HasRefreshNum = tmp.HasRefreshNum + 1
  272. }
  273. // 区分刷新成功和更新成功
  274. if v.DataUpdateResult == 1 { //处理更新结果
  275. tmp.UpdateSuccessNum = tmp.UpdateSuccessNum + 1
  276. } else if v.NeedRefresh == 1 {
  277. tmp.UpdateFailedNum = tmp.UpdateFailedNum + 1
  278. }
  279. if v.UpdateResult == 1 { //刷新结果
  280. tmp.RefreshSuccessNum = tmp.RefreshSuccessNum + 1
  281. } else if v.NeedRefresh == 1 {
  282. tmp.RefreshFailedNum = tmp.RefreshFailedNum + 1
  283. }
  284. }
  285. // 处理今天删除的指标数量
  286. if dn, ok := delMap[terminalCode]; ok {
  287. tmp.EdbDelNum = dn
  288. }
  289. // 判断是否需要新增还是更新
  290. if exist, ok := statMap[terminalCode]; ok {
  291. tmp.Id = exist.Id
  292. modifyList = append(modifyList, tmp)
  293. } else {
  294. tmp.CreateTime = nowTime
  295. addList = append(addList, tmp)
  296. }
  297. }
  298. //判断当日指标统计数据是否存在,如果存在则更新,不存在则新增
  299. if len(addList) > 0 {
  300. err = logStat.Add(addList)
  301. }
  302. if len(modifyList) > 0 {
  303. err = data_stat.UpdateEdbSourceStatMulti(modifyList)
  304. }
  305. return
  306. }