package crontab import ( "context" "fmt" "kpt-pasture/model" "kpt-pasture/module/backend" "kpt-pasture/util" "time" pasturePb "gitee.com/xuyiping_admin/go_proto/proto/go/backend/cow" "gitee.com/xuyiping_admin/pkg/logger/zaplog" "gitee.com/xuyiping_admin/pkg/xerr" "go.uber.org/zap" ) const ( UpdateCowInfo = "UpdateCowInfo" ImmunizationPlan = "ImmunizationPlan" SameTimePlan = "SameTimePlan" WorkOrderMaster = "WorkOrderMaster" SystemBasicCrontab = "SystemBasicCrontab" NeckRingOriginal = "NeckRingOriginal" ) // GenerateAsynqWorkOrder 异步生成工作单 func (e *Entry) GenerateAsynqWorkOrder() error { if ok := e.IsExistCrontabLog(WorkOrderMaster); ok { return nil } defer func() { e.CreateCrontabLog(WorkOrderMaster) }() workOrderList := make([]*model.WorkOrderMaster, 0) if err := e.DB.Where("is_show = ?", pasturePb.IsShow_Ok).Find(&workOrderList).Error; err != nil { return err } for _, workOrder := range workOrderList { timeUnix, err := util.ConvertParseLocalUnix(workOrder.ExecTime) if timeUnix <= 0 || err != nil { zaplog.Error("crontab", zap.Any("GenerateWorkOrder", err), zap.Any("execTime", workOrder.ExecTime)) continue } nowTime := time.Now().Unix() if timeUnix < nowTime { continue } execTime := timeUnix - nowTime task := model.NewTaskWorkOrderPayload(workOrder.Id, time.Duration(execTime)*time.Second) if _, err = e.AsynqClient.CtxEnqueue(context.Background(), task); err != nil { zaplog.Error("PushMessage CtxEnqueue", zap.Any("Err", err)) } } return nil } // Indicators 月度指标维护 func (e *Entry) Indicators() error { indicatorsDetailsList := make([]*model.IndicatorsDetails, 0) if err := e.DB.Model(new(model.IndicatorsDetails)). Find(&indicatorsDetailsList).Error; err != nil { return err } pastureList := e.FindPastureList() startTime, endTime := util.GetMonthStartAndEndTimestamp() zaplog.Info("Indicators", zap.Any("startTime", startTime), zap.Any("indicatorsDetailsList", indicatorsDetailsList)) for _, indicatorsDetail := range indicatorsDetailsList { pastureIndicatorList := map[int64]string{} switch indicatorsDetail.Kind { case "all_cow": pastureIndicatorList = e.FindPastureAllCow(pastureList) case "calving_interval": pastureIndicatorList = e.FindCalvingInterval(pastureList, startTime, endTime) case "output_number": pastureIndicatorList = e.FindOutputNumber(pastureList, startTime, endTime) case "input_number": pastureIndicatorList = e.FindInputNumber(pastureList, startTime, endTime) case "sales_volume": pastureIndicatorList = e.FindSalesVolume(pastureList, startTime, endTime) case "calving_number": pastureIndicatorList = e.FindCalvingNumber(pastureList, startTime, endTime) case "adult_abortion_rate": pastureIndicatorList = e.FindAdultAbortionRate(pastureList, "adult", startTime, endTime) case "youth_abortion_rate": pastureIndicatorList = e.FindAdultAbortionRate(pastureList, "youth", startTime, endTime) case "all_die_number": pastureIndicatorList = e.FindDepartureNumber(pastureList, pasturePb.DepartureType_Death, startTime, endTime) case "disease_number": pastureIndicatorList = e.FindDiseaseNumber(pastureList, startTime, endTime) case "cure_number": pastureIndicatorList = e.FindCureNumber(pastureList, startTime, endTime) case "out_number": pastureIndicatorList = e.FindDepartureNumber(pastureList, pasturePb.DepartureType_Out, startTime, endTime) case "calf_die_number": pastureIndicatorList = e.FindCalfDieNumber(pastureList, pasturePb.DepartureType_Death, startTime, endTime) } for pastureId, value := range pastureIndicatorList { e.UpdatePastureIndicators(pastureId, indicatorsDetail, startTime, value) } } return nil } // UpdateCowInfo 牛只基本信息维护 func (e *Entry) UpdateCowInfo() error { cowList := make([]*model.Cow, 0) if err := e.DB.Model(new(model.Cow)). Where("admission_status = ?", pasturePb.AdmissionStatus_Admission). Find(&cowList).Error; err != nil { return err } if ok := e.IsExistCrontabLog(UpdateCowInfo); ok { return nil } defer func() { e.CreateCrontabLog(UpdateCowInfo) }() for _, cow := range cowList { cow.EventInfoUpdate() if err := e.DB.Model(new(model.Cow)). Select("day_age", "calving_age", "pregnancy_age", "admission_age", "abortion_age", "cow_type"). Where("id = ?", cow.Id). Updates(cow).Error; err != nil { zaplog.Error("Crontab", zap.Any("UpdateCowDayAge", err)) } } return nil } // ImmunizationPlan 免疫计划,生成工作单 func (e *Entry) ImmunizationPlan() error { if ok := e.IsExistCrontabLog(ImmunizationPlan); ok { return nil } planList := make([]*model.ImmunizationPlan, 0) if err := e.DB.Model(new(model.ImmunizationPlan)). Where("is_show = ?", pasturePb.IsShow_Ok). Order("pasture_id"). Find(&planList).Error; err != nil { return xerr.WithStack(err) } var todayCount int32 = 0 nowTime := time.Now().Unix() for _, plan := range planList { cowList := make([]*model.Cow, 0) pref := e.DB.Table(fmt.Sprintf("%s as a", new(model.Cow).TableName())). Select("a.*"). Where("a.pasture_id = ?", plan.PastureId). Where("a.admission_status = ?", pasturePb.AdmissionStatus_Admission). Where("NOT EXISTS ( select 1 from event_immunization_plan b where b.pen_id = a.id and b.status = ? and b.plan_day > ?)", pasturePb.IsShow_No, nowTime) if plan.CowType > 0 { pref.Where("a.cow_type = ?", plan.CowType) } switch plan.Conditions { case pasturePb.ImmunizationConditions_Days_Age: pref.Where("a.day_age = ?", plan.Value) case pasturePb.ImmunizationConditions_Days_After_Delivery: pref.Where("a.calving_age = ?", plan.Value) case pasturePb.ImmunizationConditions_Days_Of_Pregnancy: pref.Where("a.pregnancy_age = ?", plan.Value). Where("a.is_pregnant = ?", pasturePb.IsShow_Ok) case pasturePb.ImmunizationConditions_Month: continue case pasturePb.ImmunizationConditions_Admission_Days: pref.Where("a.admission_age = ?", plan.Value) case pasturePb.ImmunizationConditions_Other_Vaccine_After: if plan.ImmunizationPlanId > 0 { pref.Joins("INNER JOIN event_immunization_plan as b ON b.plan_id = ? ", plan.ImmunizationPlanId). Where("b.cow_id = a.id"). Where("DATE_ADD(b.reality_day, INTERVAL ? DAY) = ?", plan.Value, time.Now().Format(model.LayoutDate2)). Where("b.status = ?", pasturePb.IsShow_Ok) } } if err := pref.Find(&cowList).Debug().Error; err != nil { return xerr.WithStack(err) } if len(cowList) <= 0 { continue } newImmunizationPlanCowList := model.NewCowImmunizationPlanList(cowList, plan) if err := e.DB.Model(new(model.EventImmunizationPlan)).Create(newImmunizationPlanCowList).Error; err != nil { zaplog.Error("ImmunizationPlan", zap.Any("err", err), zap.Any("plan", plan), zap.Any("cowList", cowList), zap.Any("newImmunizationPlanCowList", newImmunizationPlanCowList), ) return xerr.WithStack(err) } todayCount = int32(len(newImmunizationPlanCowList)) if todayCount > 0 { e.CreatedCalendar(plan.PastureId, pasturePb.CalendarType_Immunisation, time.Now().Format(model.LayoutDate2), todayCount) } e.CreateCrontabLog(ImmunizationPlan) } return nil } // SameTimePlan 同期计划,生成工作单 func (e *Entry) SameTimePlan() error { if ok := e.IsExistCrontabLog(SameTimePlan); ok { return nil } sameTimeList := make([]*model.SameTime, 0) if err := e.DB.Model(new(model.SameTime)). Where("is_show = ?", pasturePb.IsShow_Ok). Order("pasture_id"). Find(&sameTimeList).Error; err != nil { return xerr.WithStack(err) } if len(sameTimeList) < 0 { return nil } defer func() { e.CreateCrontabLog(SameTimePlan) }() currWeek := time.Now().Weekday() for _, sameTime := range sameTimeList { if time.Weekday(sameTime.WeekType) != currWeek { continue } cowList := make([]*model.Cow, 0) pref := e.DB.Model(new(model.Cow)). Where("admission_status = ?", pasturePb.AdmissionStatus_Admission). Where("sex = ?", pasturePb.Genders_Female). Where("pasture_id = ?", sameTime.PastureId). Where("is_pregnant = ?", pasturePb.IsShow_No) switch sameTime.CowType { case pasturePb.SameTimeCowType_Breeding_Calf: pref.Where("calving_age BETWEEN ? AND ?", sameTime.PostpartumDaysStart, sameTime.PostpartumDaysEnd). Where("lact > ?", 0) case pasturePb.SameTimeCowType_Empty: pref.Where( e.DB.Where("breed_status = ?", pasturePb.BreedStatus_Empty). Or("breed_status = ?", pasturePb.BreedStatus_Abort), ) default: continue } if err := pref.Where(`NOT EXISTS (SELECT 1 FROM event_cow_same_time WHERE event_cow_same_time.cow_id = cow.id AND event_cow_same_time.status = ?)`, pasturePb.IsShow_No). Find(&cowList).Error; err != nil { zaplog.Error("crontab", zap.Any("err", err), zap.Any("sameTime", sameTime)) return xerr.WithStack(err) } if len(cowList) <= 0 { continue } if err := e.GenerateCalendarBySameTimePlan(cowList, sameTime); err != nil { zaplog.Error("crontab", zap.Any("GenerateCalendarBySameTimePlan", err), zap.Any("cowList", cowList), zap.Any("plan", sameTime), ) continue } } return nil } // UpdateSameTime 更新每天同期数据 func (e *Entry) UpdateSameTime() error { calendarTypeList := backend.CalendarTypeEnumList("") showDay := time.Now().Format(model.LayoutDate2) for _, v := range calendarTypeList { count := int64(0) if err := e.DB.Model(new(model.EventCowSameTime)). Where("same_time_type = ?", v.Value). Where("status = ?", pasturePb.IsShow_No). Count(&count).Error; err != nil { zaplog.Error("crontab", zap.Any("UpdateSameTime", err), zap.Any("count", count)) } if count >= 0 { continue } isExist := int64(0) if err := e.DB.Model(new(model.Calendar)). Where("calendar_type = ?", v.Value). Where("show_day = ?", showDay). Count(&isExist).Error; err != nil { continue } if isExist <= 0 { continue } if err := e.DB.Model(new(model.Calendar)). Where("calendar_type = ?", v.Value). Where("show_day = ?", showDay). Updates(map[string]interface{}{ "count": count, }).Error; err != nil { continue } } return nil } // SystemBasicCrontab 基础配置计划任务 func (e *Entry) SystemBasicCrontab() error { if ok := e.IsExistCrontabLog(SystemBasicCrontab); ok { return nil } defer func() { e.CreateCrontabLog(SystemBasicCrontab) }() systemBasicList := make([]*model.SystemBasic, 0) if err := e.DB.Model(new(model.SystemBasic)). Where("is_show = ?", pasturePb.IsShow_Ok). Where("name IN ?", []string{ model.PregnantCheckForFirst, model.PregnantCheckForSecond, model.WeaningAge, model.PregnancyAge, }).Find(&systemBasicList).Error; err != nil { zaplog.Error("crontab", zap.Any("PregnancyCheck", err)) return xerr.WithStack(err) } currWeekValue := time.Now().Weekday() for _, systemBasic := range systemBasicList { // 周执行 if systemBasic.Name == model.PregnantCheckForFirst && systemBasic.WeekValue >= 0 && time.Weekday(systemBasic.WeekValue) != currWeekValue { continue } cowList := make([]*model.Cow, 0) pref := e.DB.Model(new(model.Cow)). Where("admission_status = ? ", pasturePb.AdmissionStatus_Admission). Where("pasture_id = ?", systemBasic.PastureId) switch systemBasic.Name { case model.PregnantCheckForFirst: // 初检清单 pref.Where("breed_status = ?", pasturePb.BreedStatus_Breeding). Where("last_mating_at > ?", 0). Where("DATE(FROM_UNIXTIME(last_mating_at)) BETWEEN DATE_SUB(CURDATE(), INTERVAL ? DAY) AND DATE_SUB(CURDATE(), INTERVAL ? DAY)", systemBasic.MaxValue, systemBasic.MinValue). Where(`NOT EXISTS (SELECT 1 FROM event_pregnant_check WHERE event_pregnant_check.cow_id = cow.id AND event_pregnant_check.status = ? AND event_pregnant_check.pregnant_check_name = ?)`, pasturePb.IsShow_No, model.PregnantCheckForFirst) case model.PregnantCheckForSecond: // 复检清单 过滤初检空怀的牛只 pref.Where("breed_status IN (?)", []pasturePb.BreedStatus_Kind{pasturePb.BreedStatus_Pregnant}). Where("last_mating_at > ?", 0). Where("DATE(FROM_UNIXTIME(last_mating_at)) = DATE_SUB(CURDATE(), INTERVAL ? DAY)", systemBasic.MinValue). Where(`NOT EXISTS (SELECT 1 FROM event_pregnant_check WHERE event_pregnant_check.cow_id = cow.id AND event_pregnant_check.status = ? AND event_pregnant_check.pregnant_check_name = ? )`, pasturePb.IsShow_No, model.PregnantCheckForSecond) case model.WeaningAge: // 断奶清单 pref.Where("day_age = ?", systemBasic.MinValue). Where("NOT EXISTS (SELECT 1 FROM event_weaning WHERE event_weaning.cow_id = cow.id AND event_weaning.status = ?)", pasturePb.IsShow_No) case model.PregnancyAge: // 产犊清单 pref.Where("pregnancy_age BETWEEN ? AND ?", systemBasic.MinValue, systemBasic.MaxValue). Where("breed_status = ?", pasturePb.BreedStatus_Pregnant). Where("NOT EXISTS (SELECT 1 FROM event_calving WHERE event_calving.cow_id = cow.id AND event_calving.status = ?)", pasturePb.IsShow_No) default: continue } if err := pref.Find(&cowList).Error; err != nil { zaplog.Error("crontab", zap.Any("PregnancyCheck", err), zap.Any("cowList", cowList)) continue } if len(cowList) <= 0 { continue } e.InitEventData(cowList, systemBasic) } return nil } func (e *Entry) DeleteOldOriginal() error { if err := e.DB.Model(new(model.NeckRingOriginal)). Where("active_date <= ?", time.Now().AddDate(0, 0, -7).Format(model.LayoutDate2)). Delete(&model.NeckRingOriginal{}).Error; err != nil { zaplog.Error("crontab", zap.Any("DeleteOldOriginal", err)) } return nil } func (e *Entry) InitEventData(cowList []*model.Cow, systemBasic *model.SystemBasic) { calendarType := pasturePb.CalendarType_Invalid switch systemBasic.Name { case model.PregnantCheckForFirst, model.PregnantCheckForSecond: penMap, _ := e.GetPenMapList(systemBasic.PastureId) eventPregnantCheckDataList := model.NewEventPregnantCheckList(systemBasic.PastureId, cowList, penMap, systemBasic.Name) if err := e.DB.Model(new(model.EventPregnantCheck)).Create(eventPregnantCheckDataList).Error; err != nil { zaplog.Error("crontab", zap.Any("InitEventData", err), zap.Any("eventPregnantCheckDataList", eventPregnantCheckDataList)) return } calendarType = pasturePb.CalendarType_Pregnancy_Check case model.WeaningAge: eventWeaningDataList := model.NewEventWeaningList(systemBasic.PastureId, cowList) if err := e.DB.Model(new(model.EventWeaning)).Create(eventWeaningDataList).Error; err != nil { zaplog.Error("crontab", zap.Any("InitEventData", err), zap.Any("eventWeaningDataList", eventWeaningDataList)) return } calendarType = pasturePb.CalendarType_Weaning case model.PregnancyAge: eventCalvingList := model.NewEventCalvingList(systemBasic.PastureId, cowList) if err := e.DB.Model(new(model.EventCalving)).Create(eventCalvingList).Error; err != nil { zaplog.Error("crontab", zap.Any("InitEventData", err), zap.Any("eventCalvingList", eventCalvingList)) return } calendarType = pasturePb.CalendarType_Calving } e.CreatedCalendar(systemBasic.PastureId, calendarType, time.Now().Format(model.LayoutDate2), int32(len(cowList))) }