package task import ( "context" stderrors "errors" "fmt" "path/filepath" "strconv" "strings" "time" "github.com/bytedance/sonic" "github.com/hibiken/asynq" "github.com/redis/go-redis/v9" "go.uber.org/zap" "gorm.io/gorm" "github.com/break/junhong_cmp_fiber/internal/infrastructure/audit" "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/auditcontext" "github.com/break/junhong_cmp_fiber/pkg/constants" pkgerrors "github.com/break/junhong_cmp_fiber/pkg/errors" "github.com/break/junhong_cmp_fiber/pkg/storage" "github.com/break/junhong_cmp_fiber/pkg/utils" ) const deviceBatchSize = 100 type DeviceImportPayload struct { TaskID uint `json:"task_id"` } // DeviceBatchAllocationExecutor 定义 Worker 复用现有设备分配与系列绑定规则的最小接口。 type DeviceBatchAllocationExecutor interface { AllocateDevices(ctx context.Context, req *dto.AllocateDevicesRequest, operatorID uint, operatorShopID *uint) (*dto.AllocateDevicesResponse, error) BatchSetSeriesBinding(ctx context.Context, req *dto.BatchSetDeviceSeriesBindngRequest, operatorShopID *uint) (*dto.BatchSetDeviceSeriesBindngResponse, error) RecallDevices(ctx context.Context, req *dto.RecallDevicesRequest, operatorID uint, operatorShopID *uint) (*dto.RecallDevicesResponse, error) } type DeviceImportHandler struct { db *gorm.DB redis *redis.Client importTaskStore *postgres.DeviceImportTaskStore deviceStore *postgres.DeviceStore deviceSimBindingStore *postgres.DeviceSimBindingStore iotCardStore *postgres.IotCardStore assetWalletStore *postgres.AssetWalletStore assetIdentifierStore *postgres.AssetIdentifierStore storageService *storage.Service auditWriter *audit.Writer logger *zap.Logger allocationExecutor DeviceBatchAllocationExecutor } func NewDeviceImportHandler( db *gorm.DB, redis *redis.Client, importTaskStore *postgres.DeviceImportTaskStore, deviceStore *postgres.DeviceStore, deviceSimBindingStore *postgres.DeviceSimBindingStore, iotCardStore *postgres.IotCardStore, assetWalletStore *postgres.AssetWalletStore, assetIdentifierStore *postgres.AssetIdentifierStore, storageSvc *storage.Service, auditWriter *audit.Writer, logger *zap.Logger, allocationExecutor DeviceBatchAllocationExecutor, ) *DeviceImportHandler { return &DeviceImportHandler{ db: db, redis: redis, importTaskStore: importTaskStore, deviceStore: deviceStore, deviceSimBindingStore: deviceSimBindingStore, iotCardStore: iotCardStore, assetWalletStore: assetWalletStore, assetIdentifierStore: assetIdentifierStore, storageService: storageSvc, auditWriter: auditWriter, logger: logger, allocationExecutor: allocationExecutor, } } func (h *DeviceImportHandler) HandleDeviceImport(ctx context.Context, task *asynq.Task) error { var payload DeviceImportPayload if err := sonic.Unmarshal(task.Payload(), &payload); err != nil { h.logger.Error("解析设备导入任务载荷失败", zap.Error(err), zap.String("task_id", task.ResultWriter().TaskID()), ) return asynq.SkipRetry } importTask, err := h.importTaskStore.GetByID(ctx, payload.TaskID) if err != nil { h.logger.Error("获取导入任务失败", zap.Uint("task_id", payload.TaskID), zap.Error(err), ) return asynq.SkipRetry } ctx = auditcontext.With(ctx, auditcontext.Context{ ActorKind: constants.AuditActorSystemTask, ActorID: constants.TaskTypeDeviceImport, ActorName: "设备导入任务", Source: constants.AuditSourceWorker, CorrelationID: importTask.TaskNo, ParentEventID: audit.TaskEventID(constants.AuditResourceDeviceImportTask, importTask.ID, "completed"), }) switch importTask.Status { case model.ImportTaskStatusPending: // 正常首次处理 case model.ImportTaskStatusProcessing: // 上次 worker 中途中断（重启/崩溃），重置计数重新处理 // 已入库的设备会被 ExistsByDeviceNoBatch 识别为 skip，不会重复写入 h.logger.Warn("检测到导入任务上次处理中途中断，重置进度重新处理", zap.Uint("task_id", payload.TaskID), ) h.importTaskStore.UpdateProgress(ctx, importTask.ID, 0, 0, 0) default: // 已完成或失败，不重复处理 h.logger.Info("导入任务已处理，跳过", zap.Uint("task_id", payload.TaskID), zap.Int("status", importTask.Status), ) return nil } h.importTaskStore.UpdateStatus(ctx, importTask.ID, model.ImportTaskStatusProcessing, "") h.logger.Info("开始处理设备导入任务", zap.Uint("task_id", importTask.ID), zap.String("task_no", importTask.TaskNo), zap.String("storage_key", importTask.StorageKey), ) if importTask.OperationType != "" && importTask.OperationType != constants.DeviceImportOperationCreate { return h.handleDeviceBatchAllocation(ctx, importTask) } parseResult, err := h.downloadAndParse(ctx, importTask) if err != nil { h.logger.Error("下载或解析 Excel 失败", zap.Uint("task_id", importTask.ID), zap.Error(err), ) if finishErr := h.finishDeviceImportTask(ctx, importTask, 0, 0, 1, model.ImportTaskStatusFailed, err.Error()); finishErr != nil { return finishErr } return asynq.SkipRetry } result := h.processImport(ctx, importTask, parseResult.Rows) for _, pe := range parseResult.ParseErrors { result.failedItems = append(result.failedItems, model.ImportResultItem{ Line: pe.Line, ICCID: pe.ICCID, Reason: pe.Reason, }) result.failCount++ } importTask.TotalCount = parseResult.TotalCount status, errorMessage := model.ImportTaskStatusCompleted, "" if result.failCount > 0 && result.successCount == 0 { status, errorMessage = model.ImportTaskStatusFailed, "所有导入均失败" } if err := h.finishDeviceImportTask(ctx, importTask, result.successCount, result.skipCount, result.failCount, status, errorMessage, result.skippedItems, result.failedItems); err != nil { return err } h.logger.Info("设备导入任务完成", zap.Uint("task_id", importTask.ID), zap.Int("success_count", result.successCount), zap.Int("skip_count", result.skipCount), zap.Int("fail_count", result.failCount), ) return nil } func (h *DeviceImportHandler) downloadAndParse(ctx context.Context, task *model.DeviceImportTask) (*utils.DeviceParseResult, error) { if h.storageService == nil { return nil, ErrStorageNotConfigured } if task.StorageKey == "" { return nil, ErrStorageKeyEmpty } localPath, cleanup, err := h.storageService.DownloadToTemp(ctx, task.StorageKey) if err != nil { return nil, err } defer cleanup() if !strings.HasSuffix(strings.ToLower(task.FileName), ".xlsx") { ext := filepath.Ext(task.FileName) return nil, fmt.Errorf("不支持的文件格式 %s,请上传Excel文件(.xlsx)", ext) } return utils.ParseDeviceExcel(localPath) } type deviceImportResult struct { successCount int skipCount int failCount int skippedItems model.ImportResultItems failedItems model.ImportResultItems } func (h *DeviceImportHandler) processImport(ctx context.Context, task *model.DeviceImportTask, rows []utils.DeviceRow) *deviceImportResult { result := &deviceImportResult{ skippedItems: make(model.ImportResultItems, 0), failedItems: make(model.ImportResultItems, 0), } if len(rows) == 0 { return result } for i := 0; i < len(rows); i += deviceBatchSize { end := min(i+deviceBatchSize, len(rows)) batch := rows[i:end] h.processBatch(ctx, task, batch, result) // 每批完成后实时更新进度计数，让前端可以看到处理进度 h.importTaskStore.UpdateProgress(ctx, task.ID, result.successCount, result.skipCount, result.failCount) } return result } func (h *DeviceImportHandler) processBatch(ctx context.Context, task *model.DeviceImportTask, batch []utils.DeviceRow, result *deviceImportResult) { deviceNos := make([]string, 0, len(batch)) allICCIDs := make([]string, 0) for _, row := range batch { deviceNos = append(deviceNos, row.VirtualNo) for _, slotICCID := range row.SlotICCIDs { allICCIDs = append(allICCIDs, slotICCID.ICCID) } } existingDevices, err := h.deviceStore.ExistsByDeviceNoBatch(ctx, deviceNos) if err != nil { h.logger.Error("检查设备是否存在失败", zap.Error(err)) for _, row := range batch { result.failedItems = append(result.failedItems, model.ImportResultItem{ Line: row.Line, ICCID: row.VirtualNo, Reason: "数据库查询失败", }) result.failCount++ } return } var existingCards map[string]*model.IotCard var boundCards map[string]bool if len(allICCIDs) > 0 { cards, err := h.iotCardStore.GetByICCIDs(ctx, allICCIDs) if err != nil { h.logger.Error("查询卡信息失败", zap.Error(err)) } else { existingCards = make(map[string]*model.IotCard) for _, card := range cards { existingCards[card.ICCID] = card } } boundCards, err = h.deviceSimBindingStore.GetBoundICCIDs(ctx, allICCIDs) if err != nil { h.logger.Error("查询卡绑定状态失败", zap.Error(err)) } } for _, row := range batch { if existingDevices[row.VirtualNo] { result.skippedItems = append(result.skippedItems, model.ImportResultItem{ Line: row.Line, ICCID: row.VirtualNo, Reason: "设备号已存在", }) result.skipCount++ continue } rowFailed := false for _, slotICCID := range row.SlotICCIDs { if slotICCID.SlotPosition <= row.MaxSimSlots { continue } result.failedItems = append(result.failedItems, model.ImportResultItem{ Line: row.Line, ICCID: row.VirtualNo, Reason: "卡槽填写超出最大SIM槽数", }) result.failCount++ rowFailed = true break } if rowFailed { continue } var validCards []utils.DeviceSlotICCID var cardIssues []string for _, slotICCID := range row.SlotICCIDs { card, exists := existingCards[slotICCID.ICCID] if !exists { cardIssues = append(cardIssues, slotICCID.ICCID+"不存在") continue } if boundCards[slotICCID.ICCID] { cardIssues = append(cardIssues, slotICCID.ICCID+"已绑定其他设备") continue } if card.ShopID != nil { cardIssues = append(cardIssues, slotICCID.ICCID+"已分配给店铺,不能绑定到平台库存设备") continue } validCards = append(validCards, utils.DeviceSlotICCID{ SlotPosition: slotICCID.SlotPosition, ICCID: slotICCID.ICCID, }) } if len(row.SlotICCIDs) > 0 && len(cardIssues) > 0 { result.failedItems = append(result.failedItems, model.ImportResultItem{ Line: row.Line, ICCID: row.VirtualNo, Reason: "卡验证失败: " + strings.Join(cardIssues, ", "), }) result.failCount++ continue } err := h.db.Transaction(func(tx *gorm.DB) error { txDeviceStore := postgres.NewDeviceStore(tx, nil) txBindingStore := postgres.NewDeviceSimBindingStore(tx, nil) txIotCardStore := postgres.NewIotCardStore(tx, nil) device := &model.Device{ VirtualNo: row.VirtualNo, SN: row.SN, IMEI: row.IMEI, DeviceName: row.DeviceName, DeviceModel: row.DeviceModel, DeviceType: row.DeviceType, MaxSimSlots: row.MaxSimSlots, Manufacturer: row.Manufacturer, BatchNo: task.BatchNo, Status: constants.DeviceStatusInStock, RealnamePolicy: task.RealnamePolicy, } device.Creator = task.Creator device.Updater = task.Creator if err := txDeviceStore.Create(ctx, device); err != nil { return err } txIdentifierStore := postgres.NewAssetIdentifierStore(tx) if err := txIdentifierStore.Register(ctx, device.VirtualNo, model.AssetTypeDevice, device.ID); err != nil { return fmt.Errorf("虚拟号已被占用: %s", device.VirtualNo) } now := time.Now() validCardIDs := make([]uint, 0, len(validCards)) for _, slotICCID := range validCards { cardID := existingCards[slotICCID.ICCID].ID validCardIDs = append(validCardIDs, cardID) binding := &model.DeviceSimBinding{ DeviceID: device.ID, IotCardID: cardID, SlotPosition: slotICCID.SlotPosition, BindStatus: 1, BindTime: &now, } if err := txBindingStore.Create(ctx, binding); err != nil { return err } } // 批量更新卡的设备虚拟号快照（与绑定记录在同一事务中执行） if len(validCardIDs) > 0 { if err := txIotCardStore.BatchUpdateDeviceVirtualNo(ctx, validCardIDs, device.VirtualNo); err != nil { return err } } txWalletStore := postgres.NewAssetWalletStore(tx, nil) deviceWallet := &model.AssetWallet{ ResourceType: constants.AssetWalletResourceTypeDevice, ResourceID: device.ID, Balance: 0, FrozenBalance: 0, Currency: "CNY", Status: constants.AssetWalletStatusNormal, Version: 0, } if err := txWalletStore.Create(ctx, deviceWallet); err != nil { return err } return h.appendDeviceCreateAudit(ctx, tx, task, device) }) if err != nil { h.logger.Error("创建设备失败", zap.String("virtual_no", row.VirtualNo), zap.Error(err), ) result.failedItems = append(result.failedItems, model.ImportResultItem{ Line: row.Line, ICCID: row.VirtualNo, Reason: "数据库写入失败: " + err.Error(), }) result.failCount++ continue } if len(validCards) > 0 && h.iotCardStore != nil { h.iotCardStore.InvalidateListCountCache(ctx) } for _, slotICCID := range row.SlotICCIDs { if card, exists := existingCards[slotICCID.ICCID]; exists && !boundCards[slotICCID.ICCID] && card.ShopID == nil { boundCards[slotICCID.ICCID] = true } } result.successCount++ } } func (h *DeviceImportHandler) appendDeviceCreateAudit(ctx context.Context, tx *gorm.DB, task *model.DeviceImportTask, device *model.Device) error { if h.auditWriter == nil || task == nil || device == nil || device.ID == 0 { return pkgerrors.New(pkgerrors.CodeInvalidStatus, "设备导入统一审计接缝未配置或资源不完整") } resourceID := strconv.FormatUint(uint64(device.ID), 10) resources := []audit.ResourceInput{{ Type: constants.AuditResourceDevice, ID: &resourceID, Key: audit.DeviceResourceKey(device), DisplayName: device.VirtualNo, Relation: constants.AuditResourceRelationPrimary, Role: constants.AuditResourceRoleDeviceTarget, IdentitySnapshot: audit.DeviceIdentitySnapshot(device), AfterData: map[string]any{"created": true}, SubjectVisibility: constants.AuditSubjectResult, SubjectSummary: "设备已导入", }} var bindings []*model.DeviceSimBinding if err := tx.WithContext(ctx).Where("device_id = ? AND bind_status = ?", device.ID, constants.BindStatusBound).Order("slot_position ASC").Find(&bindings).Error; err != nil { return pkgerrors.Wrap(pkgerrors.CodeDatabaseError, err, "查询设备导入卡槽关系失败") } cardIDs := make([]uint, 0, len(bindings)) for _, binding := range bindings { cardIDs = append(cardIDs, binding.IotCardID) } cardByID := make(map[uint]*model.IotCard, len(cardIDs)) if len(cardIDs) > 0 { var cards []*model.IotCard if err := tx.WithContext(ctx).Where("id IN ?", cardIDs).Find(&cards).Error; err != nil { return pkgerrors.Wrap(pkgerrors.CodeDatabaseError, err, "查询设备导入绑定卡失败") } for _, card := range cards { cardByID[card.ID] = card } } for index, binding := range bindings { card := cardByID[binding.IotCardID] if card != nil { cardID := strconv.FormatUint(uint64(card.ID), 10) resources = append(resources, audit.ResourceInput{ Type: constants.AuditResourceIotCard, ID: &cardID, Key: audit.IotCardResourceKey(card), DisplayName: card.ICCID, Relation: constants.AuditResourceRelationAffected, Role: constants.AuditResourceRoleDeviceBindingTargetCard, IdentitySnapshot: audit.IotCardIdentitySnapshot(card), AfterData: map[string]any{"device_id": device.ID, "slot_position": binding.SlotPosition}, SubjectVisibility: constants.AuditSubjectResult, SubjectSummary: "设备导入并绑定 IoT 卡", SortOrder: index*2 + 1, }) } bindingID := strconv.FormatUint(uint64(binding.ID), 10) identity := map[string]any{ "id": binding.ID, "device_id": binding.DeviceID, "device_virtual_no": device.VirtualNo, "slot_position": binding.SlotPosition, "iot_card_id": binding.IotCardID, "is_current": binding.IsCurrent, } if card != nil { identity["iccid"] = card.ICCID identity["virtual_no"] = card.VirtualNo } resources = append(resources, audit.ResourceInput{ Type: constants.AuditResourceDeviceSIMBinding, ID: &bindingID, Key: bindingID, DisplayName: device.VirtualNo, Relation: constants.AuditResourceRelationAffected, Role: constants.AuditResourceRoleDeviceCreatedBinding, IdentitySnapshot: identity, AfterData: map[string]any{ "slot_position": binding.SlotPosition, "bind_status": constants.BindStatusBound, "is_current": binding.IsCurrent, }, SubjectVisibility: constants.AuditSubjectInternalOnly, SortOrder: index*2 + 2, }) } return h.auditWriter.Append(ctx, tx, audit.AppendInput{ EventID: audit.TaskEventID(constants.AuditResourceDeviceImportTask, device.ID, "item"), ActionCode: constants.AuditActionDeviceCreated, Summary: "导入创建设备", ScopeType: constants.AuditScopePlatform, Result: constants.AuditResultSuccess, Metadata: map[string]any{"import_task_id": task.ID, "import_task_no": task.TaskNo}, Resources: resources, }) } func (h *DeviceImportHandler) finishDeviceImportTask(ctx context.Context, task *model.DeviceImportTask, successCount, skipCount, failCount, status int, errorMessage string, items ...model.ImportResultItems) error { if h.auditWriter == nil { return pkgerrors.New(pkgerrors.CodeInvalidStatus, "设备导入任务统一审计接缝未配置") } return h.db.WithContext(ctx).Transaction(func(tx *gorm.DB) error { now := time.Now() updates := map[string]any{ "status": status, "total_count": task.TotalCount, "success_count": successCount, "skip_count": skipCount, "fail_count": failCount, "error_message": errorMessage, "completed_at": now, "updated_at": now, } if len(items) > 0 { updates["skipped_items"] = items[0] } if len(items) > 1 { updates["failed_items"] = items[1] } if err := tx.WithContext(ctx).Model(&model.DeviceImportTask{}).Where("id = ?", task.ID).Updates(updates).Error; err != nil { return err } rootID := audit.TaskEventID(constants.AuditResourceDeviceImportTask, task.ID, "completed") var childCount int64 if err := tx.WithContext(ctx).Model(&model.AuditEvent{}). Where("correlation_id = ? AND action_code = ? AND result = ?", task.TaskNo, deviceImportItemAction(task.OperationType), constants.AuditResultSuccess). Count(&childCount).Error; err != nil { return err } return h.auditWriter.WriteTask(ctx, tx, audit.TaskInput{ EventID: rootID, ActionCode: constants.AuditActionDeviceImportTaskCompleted, Summary: "完成设备导入任务", TaskID: task.ID, TaskNo: task.TaskNo, Result: batchAuditResult(int(childCount), failCount), CorrelationID: task.TaskNo, ParentEventID: audit.TaskEventID(constants.AuditResourceDeviceImportTask, task.ID, "created"), BatchTotal: task.TotalCount, SuccessCount: int(childCount), FailCount: failCount, IdentitySnapshot: map[string]any{ "id": task.ID, "task_no": task.TaskNo, "file_name": task.FileName, "operation_type": task.OperationType, "target_id": task.TargetID, "batch_no": task.BatchNo, "realname_policy": task.RealnamePolicy, }, BeforeData: map[string]any{"status": model.ImportTaskStatusProcessing}, AfterData: map[string]any{ "status": status, "total_count": task.TotalCount, "success_count": successCount, "skip_count": skipCount, "fail_count": failCount, }, Metadata: map[string]any{"skip_count": skipCount}, }) }) } func deviceImportItemAction(operationType string) string { switch operationType { case constants.DeviceImportOperationAssignShop: return constants.AuditActionDeviceAllocated case constants.DeviceImportOperationAssignSeries: return constants.AuditActionDeviceSeriesBound case constants.DeviceImportOperationRecall: return constants.AuditActionDeviceRecalled default: return constants.AuditActionDeviceCreated } } var ErrMissingDeviceNoColumn = stderrors.New("CSV 缺少 virtual_no 列")