csxj2026/junhong_cmp_fiber
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

批量收回功能
All checks were successful
构建并部署到测试环境（无 SSH） / build-and-deploy (push) Successful in 8m1s
Details

Browse Source
This commit is contained in:
break
2026-07-28 10:53:15 +08:00
parent 7c5b6ee036
commit 178cc45bc2
16 changed files with 117 additions and 71 deletions
Download Patch File Download Diff File Expand all files Collapse all files

74
internal/task/device_batch_allocation.go
View File

@@ -12,6 +12,7 @@ import (
"github.com/break/junhong_cmp_fiber/internal/model"
"github.com/break/junhong_cmp_fiber/internal/model/dto"
"github.com/break/junhong_cmp_fiber/internal/store/postgres"
"github.com/break/junhong_cmp_fiber/pkg/constants"
"github.com/break/junhong_cmp_fiber/pkg/errors"
"github.com/break/junhong_cmp_fiber/pkg/middleware"
@@ -23,8 +24,8 @@ type deviceBatchAllocationRow struct {
}
func (h *DeviceImportHandler) handleDeviceBatchAllocation(ctx context.Context, task *model.DeviceImportTask) error {
if h.allocationExecutor == nil || task.TargetID == nil || *task.TargetID == 0 {
_ = h.importTaskStore.UpdateStatus(ctx, task.ID, model.ImportTaskStatusFailed, "设备批量分配执行器或目标未配置")
if h.allocationExecutor == nil || (task.OperationType != constants.DeviceImportOperationRecall && (task.TargetID == nil || *task.TargetID == 0)) {
_ = h.importTaskStore.UpdateStatus(ctx, task.ID, model.ImportTaskStatusFailed, "设备CSV批量执行器或目标未配置")
return asynq.SkipRetry
}
rows, err := h.downloadDeviceBatchAllocationCSV(ctx, task)
@@ -32,9 +33,14 @@ func (h *DeviceImportHandler) handleDeviceBatchAllocation(ctx context.Context, t
_ = h.importTaskStore.UpdateStatus(ctx, task.ID, model.ImportTaskStatusFailed, err.Error())
return asynq.SkipRetry
}
shopScope, err := h.resolveDeviceBatchShopScope(ctx, task)
if err != nil {
_ = h.importTaskStore.UpdateStatus(ctx, task.ID, model.ImportTaskStatusFailed, err.Error())
return asynq.SkipRetry
}
workerCtx := middleware.SetUserContext(ctx, &middleware.UserContextInfo{
UserID: task.Creator, UserType: task.OperatorType, Username: task.CreatorName,
ShopID: valueOrZero(task.OperatorShopID), SubordinateShopIDs: operatorDeviceShopScope(task.OperatorShopID),
ShopID: valueOrZero(task.OperatorShopID), SubordinateShopIDs: shopScope,
})
result, err := h.executeDeviceBatchAllocation(workerCtx, task, rows)
if err != nil {
@@ -43,7 +49,7 @@ func (h *DeviceImportHandler) handleDeviceBatchAllocation(ctx context.Context, t
}
_ = h.importTaskStore.UpdateResult(ctx, task.ID, len(rows), result.successCount, result.skipCount, result.failCount, 0, result.skippedItems, result.failedItems, nil)
if result.successCount == 0 && result.failCount > 0 {
_ = h.importTaskStore.UpdateStatus(ctx, task.ID, model.ImportTaskStatusFailed, "所有设备分配均失败")
_ = h.importTaskStore.UpdateStatus(ctx, task.ID, model.ImportTaskStatusFailed, "所有设备操作均失败")
} else {
_ = h.importTaskStore.UpdateStatus(ctx, task.ID, model.ImportTaskStatusCompleted, "")
}
@@ -52,20 +58,20 @@ func (h *DeviceImportHandler) handleDeviceBatchAllocation(ctx context.Context, t
func (h *DeviceImportHandler) downloadDeviceBatchAllocationCSV(ctx context.Context, task *model.DeviceImportTask) ([]deviceBatchAllocationRow, error) {
if h.storageService == nil || task.StorageKey == "" {
return nil, errors.New(errors.CodeServiceUnavailable, "设备批量分配对象存储未配置")
return nil, errors.New(errors.CodeServiceUnavailable, "设备CSV批量对象存储未配置")
}
path, cleanup, err := h.storageService.DownloadToTemp(ctx, task.StorageKey)
if err != nil {
return nil, errors.Wrap(errors.CodeInternalError, err, "下载设备批量分配CSV失败")
return nil, errors.Wrap(errors.CodeInternalError, err, "下载设备批量CSV失败")
}
defer cleanup()
info, err := os.Stat(path)
if err != nil || info.Size() > constants.DeviceBatchAllocationMaxFileSize {
return nil, errors.New(errors.CodeInvalidParam, "设备批量分配CSV不存在或超过10MB")
return nil, errors.New(errors.CodeInvalidParam, "设备批量CSV不存在或超过10MB")
}
data, err := os.ReadFile(path)
if err != nil {
return nil, errors.Wrap(errors.CodeInternalError, err, "读取设备批量分配CSV失败")
return nil, errors.Wrap(errors.CodeInternalError, err, "读取设备批量CSV失败")
}
return parseDeviceBatchAllocationCSV(data)
}
@@ -81,22 +87,22 @@ func parseDeviceBatchAllocationCSV(data []byte) ([]deviceBatchAllocationRow, err
break
}
if err != nil || len(record) != 1 {
return nil, errors.New(errors.CodeInvalidParam, "设备批量分配CSV必须只有一列设备标识")
return nil, errors.New(errors.CodeInvalidParam, "设备批量CSV必须只有一列设备标识")
}
identifier := strings.TrimSpace(record[0])
if line == 1 && (identifier == "device_identifier" || identifier == "设备标识") {
continue
}
if identifier == "" {
return nil, errors.New(errors.CodeInvalidParam, "设备批量分配CSV存在空设备标识")
return nil, errors.New(errors.CodeInvalidParam, "设备批量CSV存在空设备标识")
}
rows = append(rows, deviceBatchAllocationRow{line: line, identifier: identifier})
if len(rows) > constants.DeviceBatchAllocationMaxRows {
return nil, errors.New(errors.CodeInvalidParam, "设备批量分配CSV最多包含1000行设备")
return nil, errors.New(errors.CodeInvalidParam, "设备批量CSV最多包含1000行设备")
}
}
if len(rows) == 0 {
return nil, errors.New(errors.CodeInvalidParam, "设备批量分配CSV没有有效数据行")
return nil, errors.New(errors.CodeInvalidParam, "设备批量CSV没有有效数据行")
}
return rows, nil
}
@@ -135,7 +141,7 @@ func (h *DeviceImportHandler) executeDeviceBatchAllocation(ctx context.Context,
continue
}
seen[device.ID] = struct{}{}
if deviceAlreadyAtAllocationTarget(device, task.OperationType, *task.TargetID) {
if deviceAlreadyAtAllocationTarget(device, task) {
result.successCount++
continue
}
@@ -179,17 +185,34 @@ func (h *DeviceImportHandler) applyDeviceBatchAllocation(ctx context.Context, ta
for _, item := range response.FailedItems {
failed[item.DeviceID] = item.Reason
}
case constants.DeviceImportOperationRecall:
response, err := h.allocationExecutor.RecallDevices(ctx, &dto.RecallDevicesRequest{DeviceIDs: deviceIDs, Remark: "CSV批量回收任务 " + task.TaskNo}, task.Creator, task.OperatorShopID)
if err != nil {
return nil, err
}
for _, item := range response.FailedItems {
failed[item.DeviceID] = item.Reason
}
default:
return nil, errors.New(errors.CodeInvalidParam, "设备批量分配任务类型不支持")
return nil, errors.New(errors.CodeInvalidParam, "设备CSV批量任务类型不支持")
}
return failed, nil
}
func deviceAlreadyAtAllocationTarget(device *model.Device, operation string, targetID uint) bool {
if operation == constants.DeviceImportOperationAssignShop {
return device.ShopID != nil && *device.ShopID == targetID
func deviceAlreadyAtAllocationTarget(device *model.Device, task *model.DeviceImportTask) bool {
switch task.OperationType {
case constants.DeviceImportOperationAssignShop:
return task.TargetID != nil && device.ShopID != nil && *device.ShopID == *task.TargetID
case constants.DeviceImportOperationAssignSeries:
return task.TargetID != nil && device.SeriesID != nil && *device.SeriesID == *task.TargetID
case constants.DeviceImportOperationRecall:
if task.OperatorShopID == nil {
return device.ShopID == nil
}
return device.ShopID != nil && *device.ShopID == *task.OperatorShopID
default:
return false
}
return operation == constants.DeviceImportOperationAssignSeries && device.SeriesID != nil && *device.SeriesID == targetID
}
func deviceAllocationResultItem(row deviceBatchAllocationRow, reason string) model.ImportResultItem {
@@ -203,9 +226,16 @@ func valueOrZero(value *uint) uint {
return *value
}
func operatorDeviceShopScope(shopID *uint) []uint {
if shopID == nil {
return nil
func (h *DeviceImportHandler) resolveDeviceBatchShopScope(ctx context.Context, task *model.DeviceImportTask) ([]uint, error) {
if task.OperatorShopID == nil {
return nil, nil
}
return []uint{*shopID}
if task.OperationType != constants.DeviceImportOperationRecall {
return []uint{*task.OperatorShopID}, nil
}
shopIDs, err := postgres.NewShopStore(h.db, h.redis).GetSubordinateShopIDs(ctx, *task.OperatorShopID)
if err != nil {
return nil, errors.Wrap(errors.CodeDatabaseError, err, "查询代理设备回收范围失败")
}
return shopIDs, nil
}