kpt-pasture/module/backend/cow_more.go

package backend

import (
	"context"
	"fmt"
	"kpt-pasture/model"
	"kpt-pasture/util"
	"net/http"
	"strings"
	"sync"
	"time"

	pasturePb "gitee.com/xuyiping_admin/go_proto/proto/go/backend/cow"
	"gitee.com/xuyiping_admin/pkg/xerr"
	"github.com/nicksnyder/go-i18n/v2/i18n"
)

func (s *StoreEntry) GetNeckActiveHabitsConcurrent(cowInfo *model.Cow, cowIds []int64, dayRange []string) ([]*model.NeckActiveHabit, error) {
	const (
		batchSize   = 100
		workerCount = 5
	)

	resultsChan := make(chan []*model.NeckActiveHabit)
	errChan := make(chan error, 1) // 缓冲 1，避免 goroutine 阻塞
	doneChan := make(chan struct{})

	var (
		neckActiveHabitList []*model.NeckActiveHabit
		mu                  sync.Mutex
	)

	// 结果收集 goroutine
	go func() {
		defer close(doneChan)
		for batchResults := range resultsChan {
			mu.Lock()
			neckActiveHabitList = append(neckActiveHabitList, batchResults...)
			mu.Unlock()
		}
	}()

	sem := make(chan struct{}, workerCount)
	var wg sync.WaitGroup

	// 分发任务
	go func() {
		defer func() {
			wg.Wait()          // 等待所有 goroutine 完成
			close(resultsChan) // 安全关闭
		}()

		for i := 0; i < len(cowIds); i += batchSize {
			end := i + batchSize
			if end > len(cowIds) {
				end = len(cowIds)
			}
			batch := cowIds[i:end]

			wg.Add(1)
			sem <- struct{}{} // 获取令牌（可能阻塞，但不会死锁）

			go func(batch []int64) {
				defer func() {
					<-sem // 释放令牌
					wg.Done()
				}()

				var batchResults []*model.NeckActiveHabit
				if err := s.DB.Table(new(model.NeckActiveHabit).TableName()).
					Select("rumina,intake,inactive,gasp,high,active,active_time").
					Where("pasture_id = ?", cowInfo.PastureId).
					Where("is_show = ?", pasturePb.IsShow_Ok).
					Where("cow_id IN (?)", batch).
					Where("heat_date BETWEEN ? AND ?", dayRange[0], dayRange[len(dayRange)-1]).
					Find(&batchResults).Error; err != nil {
					select {
					case errChan <- fmt.Errorf("batch query error: %v", err):
					default:
					}
					return
				}

				if len(batchResults) > 0 {
					resultsChan <- batchResults
				}
			}(batch)
		}
	}()

	// 等待结果
	select {
	case err := <-errChan:
		return nil, err
	case <-doneChan:
		return neckActiveHabitList, nil
	}
}

func (s *StoreEntry) CowGrowthCurve(ctx context.Context, req *pasturePb.CowGrowthCurveRequest) (*pasturePb.CowGrowthCurveResponse, error) {
	userModel, err := s.GetUserModel(ctx)
	if err != nil {
		return nil, xerr.WithStack(err)
	}
	cowInfo, err := s.GetCowInfoByEarNumber(ctx, userModel.AppPasture.Id, req.EarNumber)
	if err != nil {
		messageId, _ := userModel.LanguageContent.Localize(&i18n.LocalizeConfig{
			MessageID:    "auth.errorCow",
			TemplateData: map[string]interface{}{"earNumber": req.EarNumber},
		})
		return nil, xerr.Customf(messageId)
	}

	weightList := make([]*model.EventWeight, 0)
	if err = s.DB.Table(new(model.EventWeight).TableName()).
		Where("cow_id = ?", cowInfo.Id).
		Where("pasture_id = ?", userModel.AppPasture.Id).
		Order("weight_at").
		Find(&weightList).Error; err != nil {
		return nil, xerr.WithStack(err)
	}

	eventCowLogList := make([]*model.EventCowLog, 0)
	eventCowLog := &model.EventCowLog{CowId: cowInfo.Id}
	if err = s.DB.Table(eventCowLog.TableName()).
		Where("pasture_id = ?", userModel.AppPasture.Id).
		Where("cow_id = ?", cowInfo.Id).
		Order("id desc").
		Find(&eventCowLogList).Error; err != nil {
		return nil, xerr.WithStack(err)
	}

	return &pasturePb.CowGrowthCurveResponse{
		Code: http.StatusOK,
		Msg:  "ok",
		Data: model.EventWeightSlice(weightList).ToPB(eventCowLogList),
	}, nil
}

func (s *StoreEntry) CowLactCurve(ctx context.Context, req *pasturePb.CowLactCurveRequest) (*pasturePb.CowLactCurveResponse, error) {
	userModel, err := s.GetUserModel(ctx)
	if err != nil {
		return nil, xerr.WithStack(err)
	}
	cowInfo, err := s.GetCowInfoByCowId(ctx, userModel.AppPasture.Id, int64(req.CowId))
	if err != nil {
		messageId, _ := userModel.LanguageContent.Localize(&i18n.LocalizeConfig{
			MessageID:    "auth.errorCow",
			TemplateData: map[string]interface{}{"earNumber": req.EarNumber},
		})
		return nil, xerr.Customf(messageId)
	}

	cowLactList := make([]*model.CowLact, 0)
	if err = s.DB.Table(new(model.CowLact).TableName()).
		Where("cow_id = ?", req.CowId).
		Where("pasture_id = ?", userModel.AppPasture.Id).
		Order("lact").
		Find(&cowLactList).Error; err != nil {
		return nil, xerr.WithStack(err)
	}

	data := &pasturePb.CowLactCurveData{
		DateTime:            make(map[int32]string),
		WeekAvgMilk:         make([]float32, 0),
		DayMilk:             make([]float32, 0),
		DHI:                 make([]float32, 0),
		MilkProductionTrend: make([]float32, 0),
		DayHigh:             make([]int32, 0),
		DayRumina:           make([]int32, 0),
		DayIntake:           make([]int32, 0),
		DayInactive:         make([]int32, 0),
		DayChew:             make([]int32, 0),
		DayImmobility:       make([]int32, 0),
		EstrusWarning:       make(map[int32]int32),
	}

	if len(cowLactList) <= 0 {
		return &pasturePb.CowLactCurveResponse{
			Code: http.StatusOK,
			Msg:  "ok",
			Data: data,
		}, nil
	}
	starTime := ""
	endTime := ""
	cowLactMap := make(map[int32]*model.CowLact)
	for _, v := range cowLactList {
		cowLactMap[v.Lact] = v
		if v.Lact == req.Lact {
			starTime = v.StartTime
		}
	}
	if st, ok := cowLactMap[req.Lact+1]; ok {
		et, _ := util.TimeParseLocal(model.LayoutDate2, st.StartTime)
		endTime = et.AddDate(0, 0, -1).Format(model.LayoutDate2)
	} else {
		endTime = time.Now().Local().Format(model.LayoutDate2)
	}

	neckRingList := make([]*model.NeckActiveHabit, 0)
	if err = s.DB.Model(new(model.NeckActiveHabit)).
		Where("neck_ring_number = ?", cowInfo.NeckRingNumber).
		Where("pasture_id = ?", userModel.AppPasture.Id).
		Where("cow_id > ?", 0).
		Where("heat_date BETWEEN ? AND ?", starTime, endTime).
		Order("heat_date, frameid").Find(&neckRingList).Error; err != nil {
		return nil, xerr.WithStack(err)
	}

	return &pasturePb.CowLactCurveResponse{
		Code: http.StatusOK,
		Msg:  "ok",
		Data: data,
	}, nil
}

func (s *StoreEntry) BehaviorRate(ctx context.Context, req *pasturePb.CowBehaviorRateRequest) (*pasturePb.CowBehaviorRateResponse, error) {
	userModel, err := s.GetUserModel(ctx)
	if err != nil {
		return nil, xerr.WithStack(err)
	}
	cowInfo, err := s.GetCowInfoByEarNumber(ctx, userModel.AppPasture.Id, req.EarNumber)
	if err != nil {
		messageId, _ := userModel.LanguageContent.Localize(&i18n.LocalizeConfig{
			MessageID:    "auth.errorCow",
			TemplateData: map[string]interface{}{"earNumber": req.EarNumber},
		})
		return nil, xerr.Customf(messageId)
	}

	if req.EndAt <= 0 || req.StartAt <= 0 || req.EndAt < req.StartAt {
		messageId, _ := userModel.LanguageContent.Localize(&i18n.LocalizeConfig{
			MessageID: "auth.errorDateRange",
		})
		return nil, xerr.Customf(messageId)
	}

	t1 := time.Unix(int64(req.StartAt), 0).Local().Format(model.LayoutDate2)
	t2 := time.Unix(int64(req.EndAt), 0).Local().Format(model.LayoutDate2)

	dataBetween, err := util.GetDaysBetween(t1, t2)
	if err != nil {
		return nil, xerr.WithStack(err)
	}

	neckActiveHabitList := make([]*model.NeckActiveHabit, 0)
	if err = s.DB.Model(new(model.NeckActiveHabit)).
		Select("heat_date,SUM(rumina) as rumina,SUM(intake) as intake,SUM(inactive) as inactive,SUM(gasp) AS gasp").
		Where("neck_ring_number = ?", cowInfo.NeckRingNumber).
		Where("pasture_id = ?", userModel.AppPasture.Id).
		Where("cow_id > ?", 0).
		Where("heat_date BETWEEN ? AND ?", t1, t2).
		Order("heat_date").
		Group("heat_date").
		Find(&neckActiveHabitList).Error; err != nil {
	}

	cowList := make([]*model.Cow, 0)
	if err = s.DB.Model(new(model.Cow)).
		Where("neck_ring_number != ?", "").
		Where("pasture_id = ?", userModel.AppPasture.Id).
		Where("pen_id = ?", cowInfo.PenId).
		Find(&cowList).Error; err != nil {
	}

	neckRingNumbers := make([]string, 0)
	for _, v := range cowList {
		neckRingNumbers = append(neckRingNumbers, v.NeckRingNumber)
	}

	groupNeckActiveHabitList := make([]*model.NeckActiveHabit, 0)
	if err = s.DB.Model(new(model.NeckActiveHabit)).
		Select("heat_date,SUM(rumina) as rumina,SUM(intake) as intake,SUM(inactive) as inactive,SUM(gasp) AS gasp").
		Where("neck_ring_number IN ?", neckRingNumbers).
		Where("pasture_id = ?", userModel.AppPasture.Id).
		Where("cow_id > ?", 0).
		Where("heat_date BETWEEN ? AND ?", t1, t2).
		Order("heat_date").
		Group("heat_date").
		Find(&groupNeckActiveHabitList).Error; err != nil {
	}

	return &pasturePb.CowBehaviorRateResponse{
		Code: http.StatusOK,
		Msg:  "ok",
		Data: model.NeckActiveHabitSlice(neckActiveHabitList).ToPB2(dataBetween, groupNeckActiveHabitList),
	}, nil
}

func (s *StoreEntry) BehaviorCurveApp(ctx context.Context, req *pasturePb.CowBehaviorCurveRequest) (*pasturePb.CowNeckRingAppResponse, error) {
	userModel, err := s.GetUserModel(ctx)
	if err != nil {
		return nil, xerr.WithStack(err)
	}
	if req.Days > model.NeckRingDataDays+10 {
		messageId, _ := userModel.LanguageContent.Localize(&i18n.LocalizeConfig{
			MessageID: "validate.neckRingAppDays",
		})
		return nil, xerr.Customf(messageId)
	}

	cowInfo, err := s.GetCowEventByEarNumber(ctx, userModel.AppPasture.Id, req.EarNumber)
	if err != nil {
		messageId, _ := userModel.LanguageContent.Localize(&i18n.LocalizeConfig{
			MessageID:    "auth.errorCow",
			TemplateData: map[string]interface{}{"earNumber": req.EarNumber},
		})
		return nil, xerr.Customf(messageId)
	}

	nowTime := time.Now().Local()
	nowDayZero := util.TimeParseLocalUnix(nowTime.Format(model.LayoutDate2))
	endDataTime := nowTime.Format(model.LayoutDate2)
	startDataTime := nowTime.AddDate(0, 0, int(-req.Days)+1).Format(model.LayoutDate2)

	dayRange, err := util.GetDaysBetween(startDataTime, endDataTime)
	if err != nil {
		return nil, xerr.WithStack(err)
	}

	if len(dayRange) <= 0 {
		messageId, _ := userModel.LanguageContent.Localize(&i18n.LocalizeConfig{
			MessageID: "auth.errorDateRange",
		})
		return nil, xerr.Customf(messageId)
	}

	// 行为曲线数据
	neckActiveHabitList := make([]*model.NeckActiveHabit, 0)
	if err = s.DB.Table(new(model.NeckActiveHabit).TableName()).
		Where("neck_ring_number = ?", cowInfo.NeckRingNumber).
		Where("pasture_id = ?", userModel.AppPasture.Id).
		Where("is_show = ?", pasturePb.IsShow_Ok).
		Where("cow_id > ?", 0).
		Where("heat_date IN (?)", dayRange).
		Order("heat_date, frameid").
		Find(&neckActiveHabitList).Error; err != nil {
		return nil, xerr.WithStack(err)
	}

	data := model.NeckActiveHabitSlice(neckActiveHabitList).ToPBApp()
	eventMapList := s.EventTypeMap(userModel)
	for k, v := range eventMapList {
		if k == pasturePb.EventType_Enter || k == pasturePb.EventType_Body_Score || k == pasturePb.EventType_Birth ||
			k == pasturePb.EventType_Weaning || k == pasturePb.EventType_Sale || k == pasturePb.EventType_Weight ||
			k == pasturePb.EventType_Castrated || k == pasturePb.EventType_Perinatal_Transition || k == pasturePb.EventType_Insect_Repellent {
			continue
		}
		data.EventMap = append(data.EventMap, &pasturePb.EventMap{
			EventTypeKind: k,
			EventTypeName: v,
		})
	}
	data.LowActivity = model.LowActivity
	data.MiddleActivity = model.MiddleActivity
	// 牛只事件列表
	eventLogList := make([]*model.EventCowLog, 0)
	eventLog := &model.EventCowLog{CowId: cowInfo.Id}
	if err = s.DB.Table(eventLog.TableName()).
		Where("cow_id = ?", cowInfo.Id).
		Where("pasture_id = ?", userModel.AppPasture.Id).
		Where("event_at BETWEEN ? AND ?", nowDayZero-(int64(req.Days+1)*86400), nowDayZero).
		Order("event_at").
		Find(&eventLogList).Error; err != nil {
		return nil, xerr.WithStack(err)
	}

	for _, v := range eventLogList {
		if v.EventType == pasturePb.EventType_Enter || v.EventType == pasturePb.EventType_Body_Score ||
			v.EventType == pasturePb.EventType_Birth || v.EventType == pasturePb.EventType_Weaning ||
			v.EventType == pasturePb.EventType_Sale || v.EventType == pasturePb.EventType_Weight ||
			v.EventType == pasturePb.EventType_Castrated {
			continue
		}

		if v.EventAt <= 0 {
			continue
		}

		eventAt := time.Unix(v.EventAt, 0).Local()
		data.EventList = append(data.EventList, &pasturePb.EventDetail{
			EventTypeKind:    v.EventType,
			EventTypeName:    v.EventTypeName,
			EventDescription: v.EventDescription,
			EventAtFormat:    fmt.Sprintf("%s 09", eventAt.Format(model.LayoutDate2)),
		})
	}

	// 发情数据
	estrusList := make([]*model.NeckRingEstrus, 0)
	if err = s.DB.Model(new(model.NeckRingEstrus)).
		Select("id,active_level,MAX(active_time) AS active_time,is_peak").
		Where("cow_id = ?", cowInfo.Id).
		Where("pasture_id = ?", userModel.AppPasture.Id).
		Where("active_time BETWEEN ? AND ?", fmt.Sprintf("%s 00:00:00", startDataTime), fmt.Sprintf("%s 23:59:59", endDataTime)).
		Where("is_peak = ?", pasturePb.IsShow_Ok).
		Group("first_time").
		Find(&estrusList).Error; err != nil {
		return nil, xerr.WithStack(err)
	}

	for _, v := range estrusList {
		if data.EstrusList == nil {
			data.EstrusList = make([]string, 0)
		}
		data.EstrusList = append(data.EstrusList, strings.TrimSuffix(v.ActiveTime, ":00:00")+fmt.Sprintf("/%d", v.ActiveLevel))
	}

	return &pasturePb.CowNeckRingAppResponse{
		Code: http.StatusOK,
		Msg:  "ok",
		Data: data,
	}, nil
}

func (s *StoreEntry) BehaviorCurveGroupComparison(ctx context.Context, req *pasturePb.CowBehaviorCurveRequest) (*pasturePb.BehaviorCurveGroupComparisonResponse, error) {
	userModel, err := s.GetUserModel(ctx)
	if err != nil {
		return nil, xerr.WithStack(err)
	}
	if req.Days > model.NeckRingDataDays+10 {
		messageId, _ := userModel.LanguageContent.Localize(&i18n.LocalizeConfig{
			MessageID: "validate.neckRingAppDays",
		})
		return nil, xerr.Customf(messageId)
	}

	cowInfo, err := s.GetCowEventByEarNumber(ctx, userModel.AppPasture.Id, req.EarNumber)
	if err != nil {
		messageId, _ := userModel.LanguageContent.Localize(&i18n.LocalizeConfig{
			MessageID:    "auth.errorCow",
			TemplateData: map[string]interface{}{"earNumber": req.EarNumber},
		})
		return nil, xerr.Customf(messageId)
	}

	nowTime := time.Now().Local()
	endDataTime := nowTime.Format(model.LayoutDate2)
	startDataTime := nowTime.AddDate(0, 0, int(-req.Days)+1).Format(model.LayoutDate2)

	dayRange, err := util.GetDaysBetween(startDataTime, endDataTime)
	if err != nil {
		return nil, xerr.WithStack(err)
	}

	if len(dayRange) <= 0 {
		messageId, _ := userModel.LanguageContent.Localize(&i18n.LocalizeConfig{
			MessageID: "auth.errorDateRange",
		})
		return nil, xerr.Customf(messageId)
	}

	// 行为曲线数据
	neckActiveHabitList := make([]*model.NeckActiveHabit, 0)
	if err = s.DB.Table(new(model.NeckActiveHabit).TableName()).
		Where("neck_ring_number = ?", cowInfo.NeckRingNumber).
		Where("pasture_id = ?", userModel.AppPasture.Id).
		Where("is_show = ?", pasturePb.IsShow_Ok).
		Where("cow_id > ?", 0).
		Where("heat_date IN (?)", dayRange).
		Order("heat_date, frameid").
		Find(&neckActiveHabitList).Error; err != nil {
		return nil, xerr.WithStack(err)
	}

	originalDataList, dataTimeList := model.NeckActiveHabitSlice(neckActiveHabitList).ToOriginalDataList(req.CurveName)
	q1, q3 := s.CowIQR(cowInfo, req.CurveName, dayRange, dataTimeList)
	return &pasturePb.BehaviorCurveGroupComparisonResponse{
		Code: http.StatusOK,
		Msg:  "ok",
		Data: &pasturePb.CowNeckRingGroupComparisonData{
			DateTime:         dataTimeList,
			OriginalDataList: originalDataList,
			IQR1:             q1,
			IQR3:             q3,
		},
	}, nil
}