kpt-pasture/module/crontab/crontab.go

package crontab

import (
	"context"
	"errors"
	"kpt-pasture/model"
	"kpt-pasture/util"
	"time"

	"gorm.io/gorm"

	"gitee.com/xuyiping_admin/pkg/xerr"

	"gitee.com/xuyiping_admin/pkg/logger/zaplog"
	"go.uber.org/zap"

	pasturePb "gitee.com/xuyiping_admin/go_proto/proto/go/backend/cow"
)

// UpdateCowInfo 牛只基本信息维护
func (e *Entry) UpdateCowInfo() error {
	cowList := make([]*model.Cow, 0)
	if err := e.DB.Where("is_remove = ?", pasturePb.IsShow_Ok).Find(&cowList).Error; err != nil {
		return err
	}
	for _, cow := range cowList {
		dayAge := cow.GetDayAge()
		calvingAge := cow.GetCalvingAge()
		pregnancyAge := cow.GetDaysPregnant()
		admissionAge := cow.GetAdmissionAge()
		if err := e.DB.Where("id = ?", cow.Id).Updates(map[string]interface{}{
			"day_age":       dayAge,
			"calving_at":    calvingAge,
			"pregnancy_age": pregnancyAge,
			"admission_age": admissionAge,
		}).Error; err != nil {
			zaplog.Error("Crontab", zap.Any("UpdateCowDayAge", err))
		}
	}
	return nil
}

// GenerateAsynqWorkOrder 异步生成工作单
func (e *Entry) GenerateAsynqWorkOrder() error {
	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
		}
		execTime := time.Now().Unix() - timeUnix

		if _, err = e.AsynqClient.CtxEnqueue(
			context.Background(),
			model.NewTaskWorkOrderPayload(workOrder.Id, time.Duration(execTime)*time.Second),
		); err != nil {
			zaplog.Error("PushMessage CtxEnqueue", zap.Any("Err", err))
		}
	}
	return nil
}

// ImmunizationPlan 免疫计划，生成工作单
func (e *Entry) ImmunizationPlan() error {
	planList := make([]*model.ImmunizationPlan, 0)
	if err := e.DB.Where("is_show = ?", pasturePb.IsShow_Ok).Find(&planList).Error; err != nil {
		return xerr.WithStack(err)
	}

	for _, plan := range planList {
		cowList := make([]*model.Cow, 0)
		pref := e.DB.Select("id").Where("is_remove = ?", pasturePb.IsShow_Ok).
			Where("cow_type = ?", plan.CowType)

		switch plan.Conditions {
		case pasturePb.ImmunizationConditions_Days_Age:
			pref.Where("day_age >= ?", plan.Value).
				Where("day_age <= ?", plan.Value2)
		case pasturePb.ImmunizationConditions_Days_After_Delivery:
			pref.Where("calving_age >= ?", plan.Value).
				Where("calving_age <= ?", plan.Value2)
		case pasturePb.ImmunizationConditions_Days_Of_Pregnancy:
			pref.Where("pregnancy_age >= ?", plan.Value).
				Where("pregnancy_age <= ?", plan.Value2).
				Where("is_pregnant = ?", pasturePb.IsShow_Ok)
		case pasturePb.ImmunizationConditions_Month:
			// todo 待实现月份
		case pasturePb.ImmunizationConditions_Admission_Days:
			pref.Where("admission_age >= ?", plan.Value).
				Where("admission_age <= ?", plan.Value2)
		}

		if err := pref.Find(&cowList).Error; err != nil {
			return xerr.WithStack(err)
		}
		if len(cowList) <= 0 {
			continue
		}

		if err := e.GenerateCalendarByImmunization(cowList, plan.Name); err != nil {
			zaplog.Error("crontab", zap.Any("GenerateWorkOrderCalendar", err), zap.Any("cowList", cowList))
		}
	}
	return nil
}

// SameTimePlan 同期计划，生成工作单
func (e *Entry) SameTimePlan() error {
	sameTimeList := make([]*model.SameTime, 0)
	if err := e.DB.Where("is_show = ?", pasturePb.IsShow_Ok).Find(&sameTimeList).Error; err != nil {
		return xerr.WithStack(err)
	}

	cowSameTimeList := make([]*model.SameTimeCow, 0)
	if err := e.DB.Select("cow_id").Where("status = ?", pasturePb.IsShow_Ok).Find(&cowSameTimeList).Error; err != nil {
		if !errors.Is(err, gorm.ErrRecordNotFound) {
			return xerr.WithStack(err)
		}
	}
	cowIds := make([]int64, 0)
	if len(cowSameTimeList) > 0 {
		for _, cowSameTime := range cowSameTimeList {
			cowIds = append(cowIds, cowSameTime.CowId)
		}
	}

	for _, plan := range sameTimeList {
		cowList := make([]*model.Cow, 0)
		pref := e.DB.Select("id").
			Where("is_remove = ?", pasturePb.IsShow_Ok).
			Where("is_pregnant = ?", pasturePb.IsShow_No).
			Where("calving_age >= ?", plan.PostpartumDaysStart).
			Where("calving_age <= ?", plan.PostpartumDaysEnd)
		if len(cowIds) > 0 {
			pref.Where("id NOT IN ?", cowIds)
		}

		if err := pref.Find(&cowList); err != nil {
			zaplog.Error("crontab", zap.Any("SameTimePlan", err), zap.Any("plan", plan))
			continue
		}
		if len(cowList) <= 0 {
			continue
		}
		if err := e.GenerateCalendarBySameTimePlan(cowList, plan); err != nil {
			zaplog.Error("crontab", zap.Any("GenerateCalendarBySameTimePlan", err), zap.Any("cowList", cowList), zap.Any("plan", plan))
			continue
		}
	}
	return nil
}

func (e *Entry) GenerateCalendarByImmunization(cowList []*model.Cow, name string) error {
	workOrderCalendarList := e.getWorkOrderCalendar(name)
	newCowList := make([]*model.Cow, 0)
	if len(workOrderCalendarList) > 0 {
		// 过滤已经存在的牛只数据，避免重复生成工作单
		calendarIds := make([]int64, 0)
		if err := e.DB.Model(&model.WorkOrderCalendar{}).Select("id").
			Where("name = ?", name).
			Where("is_show = ?", pasturePb.IsShow_Ok).
			Order("id DESC").
			Limit(100). // todo 默认取100条数据
			Find(&calendarIds).Error; err != nil {
			return xerr.WithStack(err)
		}

		workOrderList := make([]*model.WorkOrderList, 0)
		if err := e.DB.Where("calendar_id IN ?", calendarIds).
			Where("is_show  = ?", pasturePb.IsShow_Ok).
			Where("is_completion = ?", pasturePb.IsShow_No).
			Find(&workOrderList).Error; err != nil {
			return xerr.WithStack(err)
		}

		if len(workOrderList) > 0 {
			for _, cow := range cowList {
				for _, workOrder := range workOrderList {
					if workOrder.CowId == cow.Id {
						continue
					}
				}
				newCowList = append(newCowList, cow)
			}
		}
	}
	count := len(newCowList)
	if err := e.DB.Transaction(func(tx *gorm.DB) error {
		newWorkOrderCalendar := model.NewWorkOrderCalendar(name, int32(count))
		if err := tx.Create(newWorkOrderCalendar).Error; err != nil {
			return xerr.WithStack(err)
		}
		newWorkOrderList := make([]*model.WorkOrderList, 0)
		for _, cow := range newCowList {
			newWorkOrderList = append(newWorkOrderList, model.NewWorkOrderList(name, newWorkOrderCalendar.Id, cow.Id))
		}
		if err := tx.Create(newWorkOrderList).Error; err != nil {
			return xerr.WithStack(err)
		}
		return nil
	}); err != nil {
		return xerr.WithStack(err)
	}
	return nil
}

func (e *Entry) GenerateCalendarBySameTimePlan(cowList []*model.Cow, sameTime *model.SameTime) error {
	if len(cowList) <= 0 {
		return nil
	}

	cowSameTimeList := make([]*model.SameTimeCow, 0)
	for _, cow := range cowList {
		cowSameTimeList = append(cowSameTimeList, &model.SameTimeCow{
			CowId:      cow.Id,
			SameTimeId: sameTime.Id,
			Status:     pasturePb.IsShow_Ok,
			StartAt:    time.Now().Unix(),
			EndAt:      0,
		})
	}

	if err := e.DB.Create(cowSameTimeList).Error; err != nil {
		return xerr.WithStack(err)
	}

	return nil
}

func (e *Entry) getWorkOrderCalendar(name string) []*model.WorkOrderCalendar {
	res := make([]*model.WorkOrderCalendar, 0)
	if err := e.DB.Where("name = ?", name).
		Where("is_show = ?", pasturePb.IsShow_Ok).
		Find(&res).Error; err != nil {
		zaplog.Error("getWorkOrderCalendar", zap.Any("err", err))
	}
	return res
}