estrus_warning.go 9.7 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300
  1. package crontab
  2. import (
  3. "fmt"
  4. "kpt-pasture/model"
  5. "kpt-pasture/util"
  6. "sort"
  7. "time"
  8. pasturePb "gitee.com/xuyiping_admin/go_proto/proto/go/backend/cow"
  9. "gitee.com/xuyiping_admin/pkg/logger/zaplog"
  10. "gitee.com/xuyiping_admin/pkg/xerr"
  11. "go.uber.org/zap"
  12. )
  13. // NeckRingEstrusWarning 脖环发情预警
  14. func (e *Entry) NeckRingEstrusWarning() (err error) {
  15. pastureList := e.FindPastureList()
  16. if pastureList == nil || len(pastureList) == 0 {
  17. return nil
  18. }
  19. for _, pasture := range pastureList {
  20. if err = e.UpdateNeckRingWarning(pasture.Id); err != nil {
  21. zaplog.Error("UpdateNeckRingWarning", zap.Any("NeckRingEstrusWarning", err), zap.Any("pasture", pasture))
  22. }
  23. }
  24. return nil
  25. }
  26. func (e *Entry) UpdateNeckRingWarning(pastureId int64) (err error) {
  27. // 先删除历史数据
  28. if err = e.DB.Model(new(model.NeckRingEstrusWarning)).
  29. Where("pasture_id = ?", pastureId).
  30. Delete(new(model.NeckRingEstrusWarning)).Error; err != nil {
  31. return xerr.WithStack(err)
  32. }
  33. // 计算时间范围
  34. now := time.Now().Local()
  35. startTime := now.AddDate(0, 0, -1)
  36. neckRingEstrusList := make([]*model.NeckRingEstrus, 0)
  37. if err = e.DB.Table(fmt.Sprintf("%s as a", new(model.NeckRingEstrus).TableName())).
  38. Select("a.*").
  39. Joins("JOIN cow as b ON a.cow_id = b.id AND a.pasture_id = b.pasture_id").
  40. Where("a.pasture_id = ?", pastureId).
  41. Where("a.active_time >= ?", fmt.Sprintf("%s 00:00:00", startTime.Format(model.LayoutDate2))).
  42. Where("a.active_level >= ?", pasturePb.EstrusLevel_Low).
  43. Where("a.check_result IN (?)", []pasturePb.CheckResult_Kind{pasturePb.CheckResult_Pending, pasturePb.CheckResult_Correct}).
  44. Where("a.is_show = ?", pasturePb.IsShow_Ok).
  45. Where("a.is_peak >= ?", pasturePb.IsShow_Ok).
  46. Where("b.admission_status = ?", pasturePb.AdmissionStatus_Admission).
  47. Find(&neckRingEstrusList).Error; err != nil {
  48. return xerr.WithStack(err)
  49. }
  50. zaplog.Info("UpdateNeckRingWarning", zap.Any("neckRingEstrusList", neckRingEstrusList))
  51. if len(neckRingEstrusList) == 0 {
  52. return nil
  53. }
  54. neckRingEstrusWarningList := e.GroupAndProcessData(neckRingEstrusList)
  55. zaplog.Info("UpdateNeckRingWarning", zap.Any("neckRingEstrusWarningList", neckRingEstrusWarningList))
  56. if len(neckRingEstrusWarningList) > 0 {
  57. if err = e.DB.Create(neckRingEstrusWarningList).Error; err != nil {
  58. return xerr.WithStack(err)
  59. }
  60. } else {
  61. return nil
  62. }
  63. minId := e.getMinId(pastureId)
  64. // 更新HighChange字段
  65. // e.UpdateHighChange(pastureId,minId)
  66. // 更新IsPeak字段
  67. e.UpdateNeckRingWarningIsPeak(pastureId, minId)
  68. return nil
  69. }
  70. func (e *Entry) UpdateNeckRingWarningIsPeak(pastureId, minId int64) {
  71. sqlQuery := e.DB.Table(fmt.Sprintf("%s as a", new(model.NeckActiveHabit).TableName())).
  72. Select("1").
  73. Where("a.id >= ?", minId).
  74. Where("a.cow_id = b.cow_id").
  75. Where("a.active_time > b.date_time")
  76. if err := e.DB.Table(fmt.Sprintf("%s as b", new(model.NeckRingEstrusWarning).TableName())).
  77. Where("b.pasture_id = ?", pastureId).
  78. Where("EXISTS (?)", sqlQuery).
  79. Update("is_peak", pasturePb.IsShow_Ok).Error; err != nil {
  80. zaplog.Error("UpdateNeckRingWarningIsPeak", zap.Any("err", err))
  81. }
  82. // UPDATE v_v_hact v JOIN estrusact e ON v.inteaid=e.inteaid SET e.isPeak=1 WHERE v.isPeak=1;
  83. if err := e.DB.Raw(`UPDATE neck_ring_estrus_warning v JOIN neck_ring_estrus e ON v.neck_ring_estrus_id = e.id SET e.is_peak = ? WHERE v.is_peak = ?`,
  84. pasturePb.IsShow_Ok, pasturePb.IsShow_Ok).Error; err != nil {
  85. zaplog.Error("UpdateNeckRingWarningIsPeak", zap.Any("err", err))
  86. }
  87. }
  88. func (e *Entry) UpdateHighChange(pastureId, minId int64) {
  89. estrusWarningList, err := e.GetCowHighChange(pastureId, minId)
  90. if err != nil {
  91. zaplog.Error("UpdateHighChange", zap.Any("err", err))
  92. return
  93. }
  94. zaplog.Info("UpdateHighChange", zap.Any("estrusWarningList", estrusWarningList))
  95. for _, v := range estrusWarningList {
  96. neckRingEstrusWarning := &model.NeckRingEstrusWarning{}
  97. if err = e.DB.Model(new(model.NeckRingEstrusWarning)).
  98. Where("neck_ring_estrus_id = ?", v.NeckRingEstrusId).
  99. Find(neckRingEstrusWarning).Error; err != nil {
  100. zaplog.Error("UpdateHighChange", zap.Any("Find", err), zap.Any("v", v))
  101. continue
  102. }
  103. if v.Nb1 <= model.MinNb1 {
  104. count := e.getCowHigh(pastureId, v.CowId, minId, v.DateTime)
  105. if count <= 0 {
  106. if err = e.DB.Model(new(model.NeckRingEstrusWarning)).
  107. Where("neck_ring_estrus_id = ?", v.NeckRingEstrusId).
  108. Where("cow_id = ?", v.CowId).
  109. Where("pasture_id = ?", pastureId).Delete(new(model.NeckRingEstrusWarning)).Error; err != nil {
  110. zaplog.Error("UpdateHighChange", zap.Any("Delete", err), zap.Any("v", v))
  111. }
  112. continue
  113. }
  114. }
  115. if err = e.DB.Model(new(model.NeckRingEstrusWarning)).
  116. Where("neck_ring_estrus_id = ?", v.NeckRingEstrusId).
  117. Updates(map[string]interface{}{
  118. "warning_kind": pasturePb.Warning_Estrus,
  119. "high_change": v.HighChange,
  120. }).Error; err != nil {
  121. zaplog.Error("UpdateHighChange", zap.Any("Updates", err), zap.Any("v", v))
  122. continue
  123. }
  124. }
  125. }
  126. func (e *Entry) GroupAndProcessData(records []*model.NeckRingEstrus) []*model.NeckRingEstrusWarning {
  127. groups := make(map[int64]*model.GroupEstrusData)
  128. // 分组处理
  129. for _, record := range records {
  130. key := record.CowId
  131. if _, exist := groups[key]; !exist {
  132. groups[key] = &model.GroupEstrusData{
  133. CowId: record.CowId,
  134. PastureId: record.PastureId,
  135. Records: make([]*model.NeckRingEstrus, 0),
  136. EarNumber: record.EarNumber,
  137. NeckRingNumber: record.NeckRingNumber,
  138. Moved: false,
  139. }
  140. }
  141. // 检查是否在移牛列表中
  142. if moved := e.getRecentMovedCows(record.PastureId, record.CowId); moved {
  143. groups[key].Moved = true
  144. }
  145. groups[key].Records = append(groups[key].Records, record)
  146. }
  147. // 处理每个分组
  148. var neckRingEstrusWarningList []*model.NeckRingEstrusWarning
  149. for _, group := range groups {
  150. if len(group.Records) == 0 {
  151. continue
  152. }
  153. // 排序记录
  154. sort.Slice(group.Records, func(i, j int) bool {
  155. return group.Records[i].Id > group.Records[j].Id
  156. })
  157. // 计算字段
  158. latest := group.Records[0]
  159. maxId := findMaxId(group.Records)
  160. firstTime := findMaxTime(group.Records, func(r *model.NeckRingEstrus) string { return r.FirstTime })
  161. dateTime := findMaxTime(group.Records, func(r *model.NeckRingEstrus) string { return r.ActiveTime })
  162. neckRingEstrusWarning := model.NewNeckRingEstrusWarning(
  163. maxId, group.PastureId, group.CowId, group.EarNumber, group.NeckRingNumber,
  164. firstTime, dateTime, latest.LastTime, pasturePb.Warning_Estrus, latest.ActiveLevel,
  165. )
  166. neckRingEstrusWarningList = append(neckRingEstrusWarningList, neckRingEstrusWarning)
  167. }
  168. return neckRingEstrusWarningList
  169. }
  170. func (e *Entry) GetCowHighChange(pastureId, minId int64) ([]*model.EstrusWarning, error) {
  171. nowTime := time.Now().Local().Add(-48 * time.Hour)
  172. estrusWarningList := make([]*model.EstrusWarning, 0)
  173. if err := e.DB.Table(fmt.Sprintf("%s as a", new(model.NeckActiveHabit).TableName())).
  174. Select(
  175. "GROUP_CONCAT(IF(ROUND(a.change_filter*a.filter_correct/100,0)=-99,'',ROUND(a.change_filter*a.filter_correct/100,0)) ) as high_change",
  176. "b.neck_ring_estrus_id", "b.cow_id", "b.date_time", "COUNT(a.change_filter>-99) as nb1",
  177. "COUNT(a.change_filter=-99 AND a.created_at>=(STR_TO_DATE(b.date_time,'%Y-%m-%d %H:%i:%s') -INTERVAL 48 HOUR )) as nb2").
  178. Joins("JOIN neck_ring_estrus_warning as b ON a.cow_id = b.cow_id AND a.pasture_id = b.pasture_id").
  179. Where("a.id > ?", minId).
  180. Where("a.pasture_id = ?", pastureId).
  181. Where("a.created_at > ?", nowTime.Unix()).
  182. Group("b.neck_ring_estrus_id").
  183. Having("nb1 <= ? AND nb2 >= ?", model.Nb1, model.Nb2).
  184. Find(&estrusWarningList).
  185. Error; err != nil {
  186. return nil, xerr.WithStack(err)
  187. }
  188. return estrusWarningList, nil
  189. }
  190. // getRecentMovedCows 辅助函数:检查是否在移牛事件
  191. func (e *Entry) getRecentMovedCows(pastureId, cowId int64) bool {
  192. var count int64
  193. startDate := time.Now().Local().AddDate(0, 0, -2)
  194. table := &model.EventCowLog{CowId: cowId}
  195. if err := e.DB.Table(table.TableName()).
  196. Where("cow_id = ?", cowId).
  197. Where("pasture_id = ?", pastureId).
  198. Where("event_type = ?", pasturePb.EventType_Transfer_Ben).
  199. Where("event_at >= ?", startDate.Unix()).
  200. Count(&count).Error; err != nil {
  201. return false
  202. }
  203. if count > 0 {
  204. return true
  205. }
  206. return false
  207. }
  208. func (e *Entry) getMinId(pastureId int64) int64 {
  209. var minId int64
  210. nowTime := time.Now().Local().AddDate(0, 0, -2).Format(model.LayoutDate2)
  211. if err := e.DB.Model(new(model.NeckActiveHabit)).
  212. Select("MIN(id) as id").
  213. Where("heat_date = ?", nowTime).
  214. Where("pasture_id = ?", pastureId).
  215. Scan(&minId).Error; err != nil {
  216. return 1
  217. }
  218. return minId
  219. }
  220. func (e *Entry) getCowHigh(pastureId, cowId, minId int64, dateTime string) int64 {
  221. dateTimeUnix, _ := util.TimeParseLocal(model.LayoutTime, dateTime)
  222. dateTimeUnixStart := time.Time{}
  223. if dateTimeUnix.IsZero() {
  224. dateTimeUnixStart = time.Now().Local().Add(-22 * time.Hour)
  225. } else {
  226. dateTimeUnixStart = dateTimeUnix.Add(-22 * time.Hour)
  227. }
  228. var count int64
  229. if err := e.DB.Model(new(model.NeckActiveHabit)).
  230. Select("COUNT(0) as count").
  231. Where("created_at BETWEEN ? AND ?", dateTimeUnixStart.Unix(), dateTimeUnix).
  232. Where("pasture_id = ?", pastureId).
  233. Where("cow_id = ?", cowId).
  234. Where("id >= ?", minId).
  235. Scan(&count).Error; err != nil {
  236. return 0
  237. }
  238. return count
  239. }
  240. // 辅助函数:计算最大时间
  241. func findMaxTime(records []*model.NeckRingEstrus, getter func(*model.NeckRingEstrus) string) string {
  242. var max time.Time
  243. for _, r := range records {
  244. t1 := getter(r)
  245. if t1 == "" {
  246. continue
  247. }
  248. t, err := util.TimeParseLocal(model.LayoutTime, t1)
  249. if err != nil {
  250. continue
  251. }
  252. if t.After(max) {
  253. max = t
  254. }
  255. }
  256. if max.IsZero() {
  257. return ""
  258. }
  259. return max.Format(model.LayoutTime)
  260. }
  261. func findMaxId(records []*model.NeckRingEstrus) int64 {
  262. maxId := int64(0)
  263. for _, v := range records {
  264. if v.Id > maxId {
  265. maxId = v.Id
  266. }
  267. }
  268. return maxId
  269. }