kpt-pasture/module/crontab/cow_cron.go

package crontab

import (
	"context"
	"fmt"
	"kpt-pasture/model"
	"kpt-pasture/module/backend"
	"kpt-pasture/util"
	"time"

	"gorm.io/gorm"

	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
		}

		todayCount += int32(len(cowList))
		penList, _ := e.GetPenMapList(plan.PastureId)
		newImmunizationPlanCowList := model.NewCowImmunizationPlanList(cowList, penList, plan)
		newEventItemList := model.NewEventItemList(cowList, pasturePb.CalendarType_Immunisation)

		if err := e.DB.Transaction(func(tx *gorm.DB) error {
			if err := tx.Create(newImmunizationPlanCowList).Error; err != nil {
				return xerr.WithStack(err)

			}
			if err := tx.Create(newEventItemList).Error; err != nil {
				return xerr.WithStack(err)
			}
			return nil
		}); err != nil {
			zaplog.Error("ImmunizationPlan", zap.Any("newImmunizationPlanCowList", err), zap.Any("plan", plan), zap.Any("cowList", cowList))
		}

		var count int64 = 0
		if err := e.DB.Model(new(model.EventImmunizationPlan)).
			Where("status = ?", pasturePb.IsShow_Ok).
			Where("pasture_id = ?", plan.PastureId).
			Count(&count).Error; err != nil {
			zaplog.Error("Crontab", zap.Any("ImmunizationPlanDefer", err))
		}
		todayCount += int32(count)
		if todayCount > 0 {
			e.CreatedCalendar(plan.PastureId, pasturePb.CalendarType_Immunisation, time.Now().Format(model.LayoutDate2), todayCount)
		}
		e.CreateCrontabLog(ImmunizationPlan)
	}

	zaplog.Info("ImmunizationPlan", zap.Any("todayCount", todayCount))
	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 sameTime == nil {
			continue
		}

		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_Male).
			Where("pasture_id = ?", sameTime.PastureId)

		switch sameTime.CowType {
		case pasturePb.SameTimeCowType_Breeding_Calf:
			pref.Where("calving_age >= ?", sameTime.PostpartumDaysStart).
				Where("calving_age <= ?", sameTime.PostpartumDaysEnd).
				Where("is_pregnant = ?", pasturePb.IsShow_No)
		case pasturePb.SameTimeCowType_Empty:
			pref.Where(
				e.DB.Where("breed_status = ?", pasturePb.BreedStatus_Empty).
					Or("breed_status = ?", pasturePb.BreedStatus_Abort),
			).Where("is_pregnant = ?", pasturePb.IsShow_No)
		default:
			continue
		}

		if err := pref.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.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 (?)", fmt.Sprintf("SELECT 1 FROM event_pregnant_check WHERE event_pregnant_check.cow_id = cow.id AND event_pregnant_check.status = %d", pasturePb.IsShow_No))
		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)) = ?", fmt.Sprintf("DATE_SUB(CURDATE(), INTERVAL %d DAY))", systemBasic.MinValue)).
				Where("NOT EXISTS (?)", fmt.Sprintf("SELECT 1 FROM event_pregnant_check WHERE event_pregnant_check.cow_id = cow.id AND event_pregnant_check.status = %d", pasturePb.IsShow_No))
		case model.WeaningAge: // 断奶清单
			pref.Where("day_age = ?", systemBasic.MinValue).
				Where("NOT EXISTS (?)", fmt.Sprintf("SELECT 1 FROM event_weaning WHERE event_weaning.cow_id = cow.id AND event_weaning.status = %d", pasturePb.IsShow_No))
		case model.PregnancyAge: // 产犊清单
			pref.Where("pregnancy_age = ?", systemBasic.MinValue).
				Where("breed_status = ?", pasturePb.BreedStatus_Pregnant).
				Where("NOT EXISTS (?)", fmt.Sprintf("SELECT 1 FROM event_calving WHERE event_calving.cow_id = cow.id AND event_calving.status = %d", 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) {
	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))
		}
	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))
		}
	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))
		}
	}
}