kpt-pasture/module/crontab/cow_cron.go

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()
	for _, indicatorsDetail := range indicatorsDetailsList {
		pastureIndicatorList := map[int64]string{}
		switch indicatorsDetail.Kind {
		case model.AllCow:
			pastureIndicatorList = e.FindPastureAllCow(pastureList)
		case model.CalvingInterval:
			pastureIndicatorList = e.FindCalvingInterval(pastureList, startTime, endTime)
		case model.OutputNumber:
			pastureIndicatorList = e.FindOutputNumber(pastureList, startTime, endTime)
		case model.InputNumber:
			pastureIndicatorList = e.FindInputNumber(pastureList, startTime, endTime)
		case model.SalesVolume:
			pastureIndicatorList = e.FindSalesVolume(pastureList, startTime, endTime)
		case model.CalvingNumber:
			pastureIndicatorList = e.FindCalvingNumber(pastureList, startTime, endTime)
		case model.AdultAbortionRate:
			pastureIndicatorList = e.FindAdultAbortionRate(pastureList, "adult", startTime, endTime)
		case model.YouthAbortionRate:
			pastureIndicatorList = e.FindAdultAbortionRate(pastureList, "youth", startTime, endTime)
		case model.AllDieNumber:
			pastureIndicatorList = e.FindDepartureNumber(pastureList, pasturePb.DepartureType_Death, startTime, endTime)
		case model.DiseaseNumber:
			pastureIndicatorList = e.FindDiseaseNumber(pastureList, startTime, endTime)
		case model.CureNumber:
			pastureIndicatorList = e.FindCureNumber(pastureList, startTime, endTime)
		case model.OutNumber:
			pastureIndicatorList = e.FindDepartureNumber(pastureList, pasturePb.DepartureType_Out, startTime, endTime)
		case model.CalfDieNumber:
			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 {
		weeklyActive := e.UpdateCowWeeklyHigh(cow)
		cow.EventUpdate(weeklyActive)
		if err := e.DB.Model(new(model.Cow)).
			Select("day_age", "calving_age", "pregnancy_age", "admission_age", "abortion_age", "cow_type",
				"weekly_active", "lactation_age", "dry_milk_age").
			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()
	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.Unix())
		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 {
			endDay := nowTime.AddDate(0, 0, model.CalendarTypeEndDaysMap[pasturePb.CalendarType_Immunisation])
			e.CreatedCalendar(plan.PastureId, pasturePb.CalendarType_Immunisation, nowTime.Format(model.LayoutDate2), endDay.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,
			model.DryMilkAge,
		}).Order("pasture_id").
		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 = ?", systemBasic.MinValue-10).
				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)
		case model.DryMilkAge: // 干奶清单
			pref.Where("pregnancy_age = ?", systemBasic.MinValue).
				Where("breed_status = ?", pasturePb.BreedStatus_Pregnant).
				Where("is_pregnant = ?", pasturePb.IsShow_Ok).
				Where("NOT EXISTS (SELECT 1 FROM event_dry_milk WHERE event_dry_milk.cow_id = cow.id AND event_dry_milk.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
}

// UpdateDiseaseToCalendar 更新每天治疗中牛头数到日历表中
func (e *Entry) UpdateDiseaseToCalendar() error {
	pastureList := e.FindPastureList()
	for _, pasture := range pastureList {
		// 更新每天治疗中牛头数到日历表中
		var count int64
		if err := e.DB.Model(new(model.EventCowDisease)).
			Where("pasture_id = ?", pasture.Id).
			Where("health_status IN (?)", []pasturePb.HealthStatus_Kind{pasturePb.HealthStatus_Disease, pasturePb.HealthStatus_Treatment}).
			Count(&count).Error; err != nil {
			zaplog.Error("crontab", zap.Any("UpdateDisease", err))
		}
		if count > 0 {
			calendarTypeName := backend.CalendarTypeMap()[pasturePb.CalendarType_Disease]
			startDay := time.Now().Format(model.LayoutDate2)
			newCalendar := model.NewCalendar(pasture.Id, calendarTypeName, pasturePb.CalendarType_Disease, startDay, startDay, int32(count))
			if err := e.DB.Model(new(model.Calendar)).
				Create(newCalendar).Error; err != nil {
				zaplog.Error("crontab", zap.Any("UpdateDisease", err))
			}
		}
	}

	return nil
}

func (e *Entry) InitEventData(cowList []*model.Cow, systemBasic *model.SystemBasic) {
	calendarType := pasturePb.CalendarType_Invalid
	nowTime := time.Now()
	startDay, endDay := "", ""
	switch systemBasic.Name {
	case model.PregnantCheckForFirst, model.PregnantCheckForSecond:
		penMap, _ := e.GetPenMapList(systemBasic.PastureId)
		calendarType = pasturePb.CalendarType_Pregnancy_Check
		startDay, endDay = nowTime.Format(model.LayoutDate2), nowTime.AddDate(0, 0, model.CalendarTypeEndDaysMap[calendarType]).Format(model.LayoutDate2)
		eventPregnantCheckDataList := model.NewEventPregnantCheckList(systemBasic.PastureId, cowList, penMap, systemBasic.Name, startDay, endDay)
		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
		}
	case model.WeaningAge:
		calendarType = pasturePb.CalendarType_Weaning
		startDay, endDay = nowTime.Format(model.LayoutDate2), nowTime.AddDate(0, 0, model.CalendarTypeEndDaysMap[calendarType]).Format(model.LayoutDate2)
		eventWeaningDataList := model.NewEventWeaningList(systemBasic.PastureId, cowList, startDay, endDay)
		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
		}
	case model.PregnancyAge:
		calendarType = pasturePb.CalendarType_Calving
		startDay, endDay = nowTime.Format(model.LayoutDate2), nowTime.AddDate(0, 0, model.CalendarTypeEndDaysMap[calendarType]).Format(model.LayoutDate2)
		eventCalvingList := model.NewEventCalvingList(systemBasic.PastureId, cowList, startDay, endDay)
		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
		}
	case model.DryMilkAge:
		calendarType = pasturePb.CalendarType_DryMilk
		startDay, endDay = nowTime.Format(model.LayoutDate2), nowTime.AddDate(0, 0, model.CalendarTypeEndDaysMap[calendarType]).Format(model.LayoutDate2)
		eventCalvingList := model.NewEventDryMilkList(systemBasic.PastureId, cowList, startDay, endDay)
		if err := e.DB.Model(new(model.EventDryMilk)).Create(eventCalvingList).Error; err != nil {
			zaplog.Error("crontab", zap.Any("InitEventData", err), zap.Any("eventCalvingList", eventCalvingList))
			return
		}
	}
	e.CreatedCalendar(systemBasic.PastureId, calendarType, startDay, endDay, int32(len(cowList)))
}

// UpdateCowWeeklyHigh 更新牛只周活动量
func (e *Entry) UpdateCowWeeklyHigh(cow *model.Cow) int32 {
	highData, err := e.GetSystemNeckRingConfigure(cow.PastureId, model.High)
	if err != nil || highData == nil || highData.Value <= 0 {
		return 0
	}

	minWeeklyActive, er := e.GetSystemNeckRingConfigure(cow.PastureId, model.MinWeeklyActive)
	if er != nil || minWeeklyActive == nil || minWeeklyActive.Value <= 0 {
		return 0
	}

	weeklyActiveModelList := make([]*model.WeeklyActiveModel, 0)
	startTime := time.Now().AddDate(0, 0, -8).Format(model.LayoutDate2)
	endTime := time.Now().AddDate(0, 0, -2).Format(model.LayoutDate2)
	selectStr := fmt.Sprintf(`cow_id,heat_date,count(1) AS nb,IF(ROUND(AVG(high))>%d, ROUND(AVG(high)), %d) AS high`, minWeeklyActive.Value, minWeeklyActive.Value)
	if err = e.DB.Model(new(model.NeckActiveHabit)).
		Select(selectStr).
		Where("cow_id = ?", cow.Id).
		Where("heat_date BETWEEN ? AND ?", startTime, endTime).
		Where("high > ?", highData.Value).
		Having("nb > ?", 8).
		Order("high").
		Group("heat_date").
		Find(&weeklyActiveModelList).Error; err != nil {
		zaplog.Error("crontab", zap.Any("UpdateCowWeeklyHigh", err))
		return 0
	}

	if len(weeklyActiveModelList) < 3 {
		return 0
	}

	countHigh := len(weeklyActiveModelList)
	sumHigh := int32(0)
	minHigh := weeklyActiveModelList[0].High
	maxHigh := int32(0)
	for _, v := range weeklyActiveModelList {
		sumHigh += v.High
		if v.High > maxHigh {
			maxHigh = v.High
		}

		if v.High < minHigh {
			minHigh = v.High
		}
	}
	diff := 3
	x := int32(0)
	if countHigh == 3 {
		diff = 2
		x = weeklyActiveModelList[1].High
	}
	avgHigh := float64(sumHigh-minHigh-maxHigh-x) / float64(countHigh-diff)
	return int32(avgHigh)
}