break
5e552d99bc
Constraint: 在线热修前必须保存当前迭代分支全部有效代码进展 Confidence: medium Scope-risk: broad Directive: 后续修改需保持审计事件与业务事务边界一致 Tested: git diff --cached --check Not-tested: 未运行全量测试，提交用于切换分支前保存既有工作
576 lines
20 KiB
Go
576 lines
20 KiB
Go
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 列")
|