猪群管理聚合服务

internal/app/service/pig_batch_service.go

@@ -1,55 +1,37 @@
 package service
 
 import (
-	"errors"
-	"fmt"
-
 	"git.huangwc.com/pig/pig-farm-controller/internal/app/dto"
+	domain_pig "git.huangwc.com/pig/pig-farm-controller/internal/domain/pig"
 	"git.huangwc.com/pig/pig-farm-controller/internal/infra/logs"
 	"git.huangwc.com/pig/pig-farm-controller/internal/infra/models"
-	"git.huangwc.com/pig/pig-farm-controller/internal/infra/repository"
-
-	"gorm.io/gorm"
 )
 
-var (
+// PigBatchService 接口定义保持不变，继续作为应用层对外的契约。
-	ErrPigBatchNotFound              = errors.New("指定的猪批次不存在")
-	ErrPigBatchActive                = errors.New("活跃的猪批次不能被删除")
-	ErrPigBatchNotActive             = errors.New("猪批次不处于活跃状态，无法修改关联猪栏")
-	ErrPenOccupiedByOtherBatch       = errors.New("猪栏已被其他批次使用")
-	ErrPenStatusInvalidForAllocation = errors.New("猪栏状态不允许分配")
-	ErrPenNotAssociatedWithBatch     = errors.New("猪栏未与该批次关联")
-)
-
-// PigBatchService 提供了猪批次管理的业务逻辑
 type PigBatchService interface {
 	CreatePigBatch(dto *dto.PigBatchCreateDTO) (*dto.PigBatchResponseDTO, error)
 	GetPigBatch(id uint) (*dto.PigBatchResponseDTO, error)
 	UpdatePigBatch(id uint, dto *dto.PigBatchUpdateDTO) (*dto.PigBatchResponseDTO, error)
 	DeletePigBatch(id uint) error
 	ListPigBatches(isActive *bool) ([]*dto.PigBatchResponseDTO, error)
-	// UpdatePigBatchPens 更新猪批次关联的猪栏
 	UpdatePigBatchPens(batchID uint, desiredPenIDs []uint) error
 }
 
+// pigBatchService 的实现现在依赖于领域服务接口。
 type pigBatchService struct {
 	logger        *logs.Logger
-	pigBatchRepo repository.PigBatchRepository // 猪批次仓库
+	domainService domain_pig.PigBatchService // 依赖注入领域服务
-	pigFarmRepo  repository.PigFarmRepository  // 猪场资产仓库 (包含猪栏操作)
-	uow          repository.UnitOfWork         // 工作单元，用于事务管理
 }
 
-// NewPigBatchService 创建一个新的 PigBatchService 实例
+// NewPigBatchService 构造函数被修改，以注入领域服务。
-func NewPigBatchService(pigBatchRepo repository.PigBatchRepository, pigFarmRepo repository.PigFarmRepository, uow repository.UnitOfWork, logger *logs.Logger) PigBatchService {
+func NewPigBatchService(domainService domain_pig.PigBatchService, logger *logs.Logger) PigBatchService {
 	return &pigBatchService{
 		logger:        logger,
-		pigBatchRepo: pigBatchRepo,
+		domainService: domainService,
-		pigFarmRepo:  pigFarmRepo,
-		uow:          uow,
 	}
 }
 
-// toPigBatchResponseDTO 将 models.PigBatch 转换为 dto.PigBatchResponseDTO
+// toPigBatchResponseDTO 负责将领域模型转换为应用层DTO，这个职责保留在应用层。
 func (s *pigBatchService) toPigBatchResponseDTO(batch *models.PigBatch) *dto.PigBatchResponseDTO {
 	if batch == nil {
 		return nil
@@ -62,14 +44,15 @@ func (s *pigBatchService) toPigBatchResponseDTO(batch *models.PigBatch) *dto.Pig
 		EndDate:      batch.EndDate,
 		InitialCount: batch.InitialCount,
 		Status:       batch.Status,
-		IsActive:     batch.IsActive(), // 使用模型自带的 IsActive 方法
+		IsActive:     batch.IsActive(),
 		CreateTime:   batch.CreatedAt,
 		UpdateTime:   batch.UpdatedAt,
 	}
 }
 
-// CreatePigBatch 处理创建猪批次的业务逻辑
+// CreatePigBatch 现在将请求委托给领域服务处理。
 func (s *pigBatchService) CreatePigBatch(dto *dto.PigBatchCreateDTO) (*dto.PigBatchResponseDTO, error) {
+	// 1. DTO -> 领域模型
 	batch := &models.PigBatch{
 		BatchNumber:  dto.BatchNumber,
 		OriginType:   dto.OriginType,
@@ -78,41 +61,38 @@ func (s *pigBatchService) CreatePigBatch(dto *dto.PigBatchCreateDTO) (*dto.PigBa
 		Status:       dto.Status,
 	}
 
-	createdBatch, err := s.pigBatchRepo.CreatePigBatch(batch)
+	// 2. 调用领域服务
+	createdBatch, err := s.domainService.CreatePigBatch(batch)
 	if err != nil {
-		s.logger.Errorf("创建猪批次失败: %v", err)
+		s.logger.Errorf("应用层: 创建猪批次失败: %v", err)
-		return nil, err
+		return nil, err // 将领域层的错误传递上去
 	}
 
+	// 3. 领域模型 -> DTO
 	return s.toPigBatchResponseDTO(createdBatch), nil
 }
 
-// GetPigBatch 处理获取单个猪批次的业务逻辑
+// GetPigBatch 从领域服务获取数据并转换为DTO。
 func (s *pigBatchService) GetPigBatch(id uint) (*dto.PigBatchResponseDTO, error) {
-	batch, err := s.pigBatchRepo.GetPigBatchByID(id)
+	batch, err := s.domainService.GetPigBatch(id)
 	if err != nil {
-		if errors.Is(err, gorm.ErrRecordNotFound) {
+		s.logger.Warnf("应用层: 获取猪批次失败, ID: %d, 错误: %v", id, err)
-			return nil, ErrPigBatchNotFound
-		}
-		s.logger.Errorf("获取猪批次失败，ID: %d, 错误: %v", id, err)
 		return nil, err
 	}
 
 	return s.toPigBatchResponseDTO(batch), nil
 }
 
-// UpdatePigBatch 处理更新猪批次的业务逻辑
+// UpdatePigBatch 协调获取、更新和保存的流程。
 func (s *pigBatchService) UpdatePigBatch(id uint, dto *dto.PigBatchUpdateDTO) (*dto.PigBatchResponseDTO, error) {
-	existingBatch, err := s.pigBatchRepo.GetPigBatchByID(id)
+	// 1. 先获取最新的领域模型
+	existingBatch, err := s.domainService.GetPigBatch(id)
 	if err != nil {
-		if errors.Is(err, gorm.ErrRecordNotFound) {
+		s.logger.Warnf("应用层: 更新猪批次失败，获取原批次信息错误, ID: %d, 错误: %v", id, err)
-			return nil, ErrPigBatchNotFound
-		}
-		s.logger.Errorf("更新猪批次失败，获取原批次信息错误，ID: %d, 错误: %v", id, err)
 		return nil, err
 	}
 
-	// 根据 DTO 中的非空字段更新模型
+	// 2. 将DTO中的变更应用到模型上
 	if dto.BatchNumber != nil {
 		existingBatch.BatchNumber = *dto.BatchNumber
 	}
@@ -132,55 +112,32 @@ func (s *pigBatchService) UpdatePigBatch(id uint, dto *dto.PigBatchUpdateDTO) (*
 		existingBatch.Status = *dto.Status
 	}
 
-	updatedBatch, rowsAffected, err := s.pigBatchRepo.UpdatePigBatch(existingBatch)
+	// 3. 调用领域服务执行更新
+	updatedBatch, err := s.domainService.UpdatePigBatch(existingBatch)
 	if err != nil {
-		s.logger.Errorf("更新猪批次失败，ID: %d, 错误: %v", id, err)
+		s.logger.Errorf("应用层: 更新猪批次失败, ID: %d, 错误: %v", id, err)
 		return nil, err
 	}
-	// 如果没有行受影响，则认为猪批次不存在
-	if rowsAffected == 0 {
-		return nil, ErrPigBatchNotFound
-	}
 
+	// 4. 转换并返回结果
 	return s.toPigBatchResponseDTO(updatedBatch), nil
 }
 
-// DeletePigBatch 处理删除猪批次的业务逻辑
+// DeletePigBatch 将删除操作委托给领域服务。
 func (s *pigBatchService) DeletePigBatch(id uint) error {
-	// 1. 获取猪批次信息
+	err := s.domainService.DeletePigBatch(id)
-	batch, err := s.pigBatchRepo.GetPigBatchByID(id)
 	if err != nil {
-		if errors.Is(err, gorm.ErrRecordNotFound) {
+		s.logger.Errorf("应用层: 删除猪批次失败, ID: %d, 错误: %v", id, err)
-			return ErrPigBatchNotFound
-		}
-		s.logger.Errorf("删除猪批次失败，获取批次信息错误，ID: %d, 错误: %v", id, err)
 		return err
 	}
-
-	// 2. 检查猪批次是否活跃
-	if batch.IsActive() {
-		return ErrPigBatchActive // 如果活跃，则不允许删除
-	}
-
-	// 3. 执行删除操作
-	rowsAffected, err := s.pigBatchRepo.DeletePigBatch(id)
-	if err != nil {
-		s.logger.Errorf("删除猪批次失败，ID: %d, 错误: %v", id, err)
-		return err
-	}
-	// 如果没有行受影响，则认为猪批次不存在
-	if rowsAffected == 0 {
-		return ErrPigBatchNotFound
-	}
-
 	return nil
 }
 
-// ListPigBatches 处理批量查询猪批次的业务逻辑
+// ListPigBatches 从领域服务获取列表并进行转换。
 func (s *pigBatchService) ListPigBatches(isActive *bool) ([]*dto.PigBatchResponseDTO, error) {
-	batches, err := s.pigBatchRepo.ListPigBatches(isActive)
+	batches, err := s.domainService.ListPigBatches(isActive)
 	if err != nil {
-		s.logger.Errorf("批量查询猪批次失败，错误: %v", err)
+		s.logger.Errorf("应用层: 批量查询猪批次失败: %v", err)
 		return nil, err
 	}
 
@@ -192,111 +149,12 @@ func (s *pigBatchService) ListPigBatches(isActive *bool) ([]*dto.PigBatchRespons
 	return responseDTOs, nil
 }
 
-// UpdatePigBatchPens 更新猪批次关联的猪栏
+// UpdatePigBatchPens 将关联猪栏的复杂操作委托给领域服务。
 func (s *pigBatchService) UpdatePigBatchPens(batchID uint, desiredPenIDs []uint) error {
-	// 使用工作单元执行事务
+	err := s.domainService.UpdatePigBatchPens(batchID, desiredPenIDs)
-	return s.uow.ExecuteInTransaction(func(tx *gorm.DB) error {
-		// 1. 验证猪批次
-		pigBatch, err := s.pigFarmRepo.GetPigBatchByIDTx(tx, batchID)
 	if err != nil {
-			if errors.Is(err, gorm.ErrRecordNotFound) {
+		s.logger.Errorf("应用层: 更新猪批次猪栏关联失败, 批次ID: %d, 错误: %v", batchID, err)
-				return ErrPigBatchNotFound
+		return err
 	}
-			s.logger.Errorf("更新猪批次猪栏失败: 获取猪批次信息错误，ID: %d, 错误: %v", batchID, err)
-			return fmt.Errorf("获取猪批次信息失败: %w", err)
-		}
-
-		if !pigBatch.IsActive() {
-			return ErrPigBatchNotActive
-		}
-
-		// 2. 获取当前关联的猪栏
-		currentPens, err := s.pigFarmRepo.GetPensByBatchID(tx, batchID)
-		if err != nil {
-			s.logger.Errorf("更新猪批次猪栏失败: 获取当前关联猪栏错误，批次ID: %d, 错误: %v", batchID, err)
-			return fmt.Errorf("获取当前关联猪栏失败: %w", err)
-		}
-
-		currentPenMap := make(map[uint]models.Pen)
-		currentPenIDsSet := make(map[uint]struct{})
-		for _, pen := range currentPens {
-			currentPenMap[pen.ID] = pen
-			currentPenIDsSet[pen.ID] = struct{}{} // 用于快速查找
-		}
-
-		// 3. 构建期望猪栏集合
-		desiredPenIDsSet := make(map[uint]struct{})
-		for _, penID := range desiredPenIDs {
-			desiredPenIDsSet[penID] = struct{}{} // 用于快速查找
-		}
-
-		// 4. 计算需要添加和移除的猪栏
-		var pensToRemove []uint
-		for penID := range currentPenIDsSet {
-			if _, found := desiredPenIDsSet[penID]; !found {
-				pensToRemove = append(pensToRemove, penID)
-			}
-		}
-
-		var pensToAdd []uint
-		for _, penID := range desiredPenIDs {
-			if _, found := currentPenIDsSet[penID]; !found {
-				pensToAdd = append(pensToAdd, penID)
-			}
-		}
-
-		// 5. 处理移除猪栏
-		for _, penID := range pensToRemove {
-			currentPen := currentPenMap[penID]
-			// 验证：确保猪栏确实与当前批次关联
-			if currentPen.PigBatchID == nil || *currentPen.PigBatchID != batchID {
-				s.logger.Warnf("尝试移除未与批次 %d 关联的猪栏 %d", batchID, penID)
-				return fmt.Errorf("猪栏 %d 未与该批次关联，无法移除", penID)
-			}
-
-			updates := make(map[string]interface{})
-			updates["pig_batch_id"] = nil // 总是将 PigBatchID 设为 nil
-
-			// 只有当猪栏当前状态是“使用中”时，才将其状态改回“空闲”
-			if currentPen.Status == models.PenStatusOccupied {
-				updates["status"] = models.PenStatusEmpty
-			}
-
-			if err := s.pigFarmRepo.UpdatePenFields(tx, penID, updates); err != nil {
-				s.logger.Errorf("更新猪批次猪栏失败: 移除猪栏 %d 失败: %v", penID, err)
-				return fmt.Errorf("移除猪栏 %d 失败: %w", penID, err)
-			}
-		}
-
-		// 6. 处理添加猪栏
-		for _, penID := range pensToAdd {
-			actualPen, err := s.pigFarmRepo.GetPenByIDTx(tx, penID)
-			if err != nil {
-				if errors.Is(err, gorm.ErrRecordNotFound) {
-					return fmt.Errorf("猪栏 %d 不存在: %w", penID, ErrPenNotFound)
-				}
-				s.logger.Errorf("更新猪批次猪栏失败: 获取猪栏 %d 信息错误: %v", penID, err)
-				return fmt.Errorf("获取猪栏 %d 信息失败: %w", penID, err)
-			}
-
-			// 验证：猪栏必须是“空闲”状态且未被任何批次使用，才能被分配
-			if actualPen.Status != models.PenStatusEmpty {
-				return fmt.Errorf("猪栏 %s 状态为 %s，无法分配: %w", actualPen.PenNumber, actualPen.Status, ErrPenStatusInvalidForAllocation)
-			}
-			if actualPen.PigBatchID != nil {
-				return fmt.Errorf("猪栏 %s 已被其他批次 %d 使用，无法分配: %w", actualPen.PenNumber, *actualPen.PigBatchID, ErrPenOccupiedByOtherBatch)
-			}
-
-			updates := map[string]interface{}{
-				"pig_batch_id": &batchID,                 // 将 PigBatchID 设为当前批次ID的指针
-				"status":       models.PenStatusOccupied, // 分配后，状态变为“使用中”
-			}
-			if err := s.pigFarmRepo.UpdatePenFields(tx, penID, updates); err != nil {
-				s.logger.Errorf("更新猪批次猪栏失败: 添加猪栏 %d 失败: %v", penID, err)
-				return fmt.Errorf("添加猪栏 %d 失败: %w", penID, err)
-			}
-		}
-
 	return nil
-	})
 }

internal/core/application.go

@@ -12,6 +12,7 @@ import (
 	"git.huangwc.com/pig/pig-farm-controller/internal/app/webhook"
 	"git.huangwc.com/pig/pig-farm-controller/internal/domain/audit"
 	"git.huangwc.com/pig/pig-farm-controller/internal/domain/device"
+	"git.huangwc.com/pig/pig-farm-controller/internal/domain/pig"
 	"git.huangwc.com/pig/pig-farm-controller/internal/domain/task"
 	"git.huangwc.com/pig/pig-farm-controller/internal/domain/token"
 	"git.huangwc.com/pig/pig-farm-controller/internal/infra/config"
@@ -77,9 +78,12 @@ func NewApplication(configPath string) (*Application, error) {
 	// 初始化事务管理器
 	unitOfWork := repository.NewGormUnitOfWork(storage.GetDB(), logger)
 
+	// 初始化猪群管理服务
+	pigBatchDomain := pig.NewPigBatchService(pigBatchRepo, pigFarmRepo, unitOfWork)
+
 	// --- 业务逻辑处理器初始化 ---
 	pigFarmService := service.NewPigFarmService(pigFarmRepo, unitOfWork, logger)
-	pigBatchService := service.NewPigBatchService(pigBatchRepo, pigFarmRepo, unitOfWork, logger)
+	pigBatchService := service.NewPigBatchService(pigBatchDomain, logger)
 
 	// 初始化审计服务
 	auditService := audit.NewService(userActionLogRepo, logger)

internal/domain/pig/pig_batch.go

@@ -0,0 +1,50 @@
+package pig
+
+import (
+	"errors"
+
+	"git.huangwc.com/pig/pig-farm-controller/internal/infra/models"
+)
+
+// --- 业务错误定义 ---
+
+var (
+	// ErrPigBatchNotFound 表示当尝试访问一个不存在的猪批次时发生的错误。
+	ErrPigBatchNotFound = errors.New("指定的猪批次不存在")
+	// ErrPigBatchActive 表示当尝试对一个活跃的猪批次执行不允许的操作（如删除）时发生的错误。
+	ErrPigBatchActive = errors.New("活跃的猪批次不能被删除")
+	// ErrPigBatchNotActive 表示当猪批次不处于活跃状态，但执行了需要其活跃的操作时发生的错误。
+	ErrPigBatchNotActive = errors.New("猪批次不处于活跃状态，无法修改关联猪栏")
+	// ErrPenOccupiedByOtherBatch 表示当尝试将一个已经被其他批次占用的猪栏分配给新批次时发生的错误。
+	ErrPenOccupiedByOtherBatch = errors.New("猪栏已被其他批次使用")
+	// ErrPenStatusInvalidForAllocation 表示猪栏的当前状态（例如，'维修中'）不允许被分配。
+	ErrPenStatusInvalidForAllocation = errors.New("猪栏状态不允许分配")
+	// ErrPenNotFound 表示猪栏不存在
+	ErrPenNotFound = errors.New("指定的猪栏不存在")
+)
+
+// --- 领域服务接口 ---
+
+// PigBatchService 定义了猪批次管理的核心业务逻辑接口。
+// 它抽象了所有与猪批次相关的操作，使得应用层可以依赖于此接口，而不是具体的实现。
+type PigBatchService interface {
+	// CreatePigBatch 创建一个新的猪批次。
+	CreatePigBatch(batch *models.PigBatch) (*models.PigBatch, error)
+
+	// GetPigBatch 根据ID获取单个猪批次的详细信息。
+	GetPigBatch(id uint) (*models.PigBatch, error)
+
+	// UpdatePigBatch 更新一个已存在的猪批次信息。
+	UpdatePigBatch(batch *models.PigBatch) (*models.PigBatch, error)
+
+	// DeletePigBatch 删除一个指定的猪批次。
+	// 实现时需要包含业务规则校验，例如，活跃的批次不能被删除。
+	DeletePigBatch(id uint) error
+
+	// ListPigBatches 根据是否活跃的状态，列出所有符合条件的猪批次。
+	ListPigBatches(isActive *bool) ([]*models.PigBatch, error)
+
+	// UpdatePigBatchPens 负责原子性地更新一个猪批次所关联的所有猪栏。
+	// 它会处理猪栏的添加、移除，并确保数据的一致性。
+	UpdatePigBatchPens(batchID uint, desiredPenIDs []uint) error
+}

internal/domain/pig/pig_batch_service.go

@@ -0,0 +1,192 @@
+package pig
+
+import (
+	"errors"
+	"fmt"
+
+	"git.huangwc.com/pig/pig-farm-controller/internal/infra/models"
+	"git.huangwc.com/pig/pig-farm-controller/internal/infra/repository"
+	"gorm.io/gorm"
+)
+
+// --- 领域服务实现 ---
+
+// pigBatchService 是 PigBatchService 接口的具体实现。
+// 它封装了业务逻辑所需的所有依赖，如数据库仓库和工作单元。
+type pigBatchService struct {
+	pigBatchRepo repository.PigBatchRepository // 猪批次仓库
+	pigFarmRepo  repository.PigFarmRepository  // 猪场资产仓库 (包含猪栏操作)
+	uow          repository.UnitOfWork         // 工作单元，用于管理事务
+}
+
+// NewPigBatchService 是 pigBatchService 的构造函数。
+// 它通过依赖注入的方式，创建并返回一个 PigBatchService 接口的实例。
+func NewPigBatchService(
+	pigBatchRepo repository.PigBatchRepository,
+	pigFarmRepo repository.PigFarmRepository,
+	uow repository.UnitOfWork,
+) PigBatchService {
+	return &pigBatchService{
+		pigBatchRepo: pigBatchRepo,
+		pigFarmRepo:  pigFarmRepo,
+		uow:          uow,
+	}
+}
+
+// CreatePigBatch 实现了创建猪批次的逻辑。
+func (s *pigBatchService) CreatePigBatch(batch *models.PigBatch) (*models.PigBatch, error) {
+	// 业务规则可以在这里添加，例如检查批次号是否唯一等
+	return s.pigBatchRepo.CreatePigBatch(batch)
+}
+
+// GetPigBatch 实现了获取单个猪批次的逻辑。
+func (s *pigBatchService) GetPigBatch(id uint) (*models.PigBatch, error) {
+	batch, err := s.pigBatchRepo.GetPigBatchByID(id)
+	if err != nil {
+		if errors.Is(err, gorm.ErrRecordNotFound) {
+			return nil, ErrPigBatchNotFound
+		}
+		return nil, err
+	}
+	return batch, nil
+}
+
+// UpdatePigBatch 实现了更新猪批次的逻辑。
+func (s *pigBatchService) UpdatePigBatch(batch *models.PigBatch) (*models.PigBatch, error) {
+	// 可以在这里添加更新前的业务校验
+	updatedBatch, rowsAffected, err := s.pigBatchRepo.UpdatePigBatch(batch)
+	if err != nil {
+		return nil, err
+	}
+	if rowsAffected == 0 {
+		return nil, ErrPigBatchNotFound // 如果没有行被更新，可能意味着记录不存在
+	}
+	return updatedBatch, nil
+}
+
+// DeletePigBatch 实现了删除猪批次的逻辑，并包含业务规则校验。
+func (s *pigBatchService) DeletePigBatch(id uint) error {
+	// 1. 获取猪批次信息
+	batch, err := s.GetPigBatch(id) // 复用 GetPigBatch 方法
+	if err != nil {
+		return err // GetPigBatch 已经处理了 ErrRecordNotFound 的情况
+	}
+
+	// 2. 核心业务规则：检查猪批次是否为活跃状态
+	if batch.IsActive() {
+		return ErrPigBatchActive // 如果活跃，则不允许删除
+	}
+
+	// 3. 执行删除
+	rowsAffected, err := s.pigBatchRepo.DeletePigBatch(id)
+	if err != nil {
+		return err
+	}
+	if rowsAffected == 0 {
+		return ErrPigBatchNotFound
+	}
+
+	return nil
+}
+
+// ListPigBatches 实现了批量查询猪批次的逻辑。
+func (s *pigBatchService) ListPigBatches(isActive *bool) ([]*models.PigBatch, error) {
+	return s.pigBatchRepo.ListPigBatches(isActive)
+}
+
+// UpdatePigBatchPens 实现了在事务中更新猪批次关联猪栏的复杂逻辑。
+func (s *pigBatchService) UpdatePigBatchPens(batchID uint, desiredPenIDs []uint) error {
+	// 使用工作单元来确保操作的原子性
+	return s.uow.ExecuteInTransaction(func(tx *gorm.DB) error {
+		// 1. 验证猪批次是否存在且活跃
+		pigBatch, err := s.pigFarmRepo.GetPigBatchByIDTx(tx, batchID)
+		if err != nil {
+			if errors.Is(err, gorm.ErrRecordNotFound) {
+				return ErrPigBatchNotFound
+			}
+			return fmt.Errorf("获取猪批次信息失败: %w", err)
+		}
+
+		if !pigBatch.IsActive() {
+			return ErrPigBatchNotActive
+		}
+
+		// 2. 获取当前关联的猪栏
+		currentPens, err := s.pigFarmRepo.GetPensByBatchID(tx, batchID)
+		if err != nil {
+			return fmt.Errorf("获取当前关联猪栏失败: %w", err)
+		}
+
+		currentPenMap := make(map[uint]models.Pen)
+		currentPenIDsSet := make(map[uint]struct{})
+		for _, pen := range currentPens {
+			currentPenMap[pen.ID] = pen
+			currentPenIDsSet[pen.ID] = struct{}{}
+		}
+
+		// 3. 构建期望猪栏ID集合
+		desiredPenIDsSet := make(map[uint]struct{})
+		for _, penID := range desiredPenIDs {
+			desiredPenIDsSet[penID] = struct{}{}
+		}
+
+		// 4. 计算需要添加和移除的猪栏
+		var pensToRemove []uint
+		for penID := range currentPenIDsSet {
+			if _, found := desiredPenIDsSet[penID]; !found {
+				pensToRemove = append(pensToRemove, penID)
+			}
+		}
+
+		var pensToAdd []uint
+		for _, penID := range desiredPenIDs {
+			if _, found := currentPenIDsSet[penID]; !found {
+				pensToAdd = append(pensToAdd, penID)
+			}
+		}
+
+		// 5. 处理移除猪栏的逻辑
+		for _, penID := range pensToRemove {
+			currentPen := currentPenMap[penID]
+			updates := make(map[string]interface{})
+			updates["pig_batch_id"] = nil
+
+			if currentPen.Status == models.PenStatusOccupied {
+				updates["status"] = models.PenStatusEmpty
+			}
+
+			if err := s.pigFarmRepo.UpdatePenFields(tx, penID, updates); err != nil {
+				return fmt.Errorf("移除猪栏 %d 失败: %w", penID, err)
+			}
+		}
+
+		// 6. 处理添加猪栏的逻辑
+		for _, penID := range pensToAdd {
+			actualPen, err := s.pigFarmRepo.GetPenByIDTx(tx, penID)
+			if err != nil {
+				if errors.Is(err, gorm.ErrRecordNotFound) {
+					return fmt.Errorf("猪栏 %d 不存在: %w", penID, ErrPenNotFound)
+				}
+				return fmt.Errorf("获取猪栏 %d 信息失败: %w", penID, err)
+			}
+
+			// 核心业务规则：校验猪栏是否可被分配
+			if actualPen.Status != models.PenStatusEmpty {
+				return fmt.Errorf("猪栏 %s 状态为 %s，无法分配: %w", actualPen.PenNumber, actualPen.Status, ErrPenStatusInvalidForAllocation)
+			}
+			if actualPen.PigBatchID != nil {
+				return fmt.Errorf("猪栏 %s 已被其他批次 %d 使用: %w", actualPen.PenNumber, *actualPen.PigBatchID, ErrPenOccupiedByOtherBatch)
+			}
+
+			updates := map[string]interface{}{
+				"pig_batch_id": &batchID,
+				"status":       models.PenStatusOccupied,
+			}
+			if err := s.pigFarmRepo.UpdatePenFields(tx, penID, updates); err != nil {
+				return fmt.Errorf("添加猪栏 %d 失败: %w", penID, err)
+			}
+		}
+
+		return nil
+	})
+}