break
8f738ffbe8
All checks were successful
构建并部署到测试环境（无 SSH） / build-and-deploy (push) Successful in 7m51s
402 lines
12 KiB
Go
402 lines
12 KiB
Go
package task
|
|
|
|
import (
|
|
"context"
|
|
stderrors "errors"
|
|
"fmt"
|
|
"path/filepath"
|
|
"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/model"
|
|
"github.com/break/junhong_cmp_fiber/internal/store/postgres"
|
|
"github.com/break/junhong_cmp_fiber/pkg/constants"
|
|
"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"`
|
|
}
|
|
|
|
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
|
|
logger *zap.Logger
|
|
}
|
|
|
|
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,
|
|
logger *zap.Logger,
|
|
) *DeviceImportHandler {
|
|
return &DeviceImportHandler{
|
|
db: db,
|
|
redis: redis,
|
|
importTaskStore: importTaskStore,
|
|
deviceStore: deviceStore,
|
|
deviceSimBindingStore: deviceSimBindingStore,
|
|
iotCardStore: iotCardStore,
|
|
assetWalletStore: assetWalletStore,
|
|
assetIdentifierStore: assetIdentifierStore,
|
|
storageService: storageSvc,
|
|
logger: logger,
|
|
}
|
|
}
|
|
|
|
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
|
|
}
|
|
|
|
if importTask.Status != model.ImportTaskStatusPending {
|
|
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),
|
|
)
|
|
|
|
parseResult, err := h.downloadAndParse(ctx, importTask)
|
|
if err != nil {
|
|
h.logger.Error("下载或解析 Excel 失败",
|
|
zap.Uint("task_id", importTask.ID),
|
|
zap.Error(err),
|
|
)
|
|
h.importTaskStore.UpdateStatus(ctx, importTask.ID, model.ImportTaskStatusFailed, err.Error())
|
|
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++
|
|
}
|
|
|
|
h.importTaskStore.UpdateResult(ctx, importTask.ID, parseResult.TotalCount, result.successCount, result.skipCount, result.failCount, 0, result.skippedItems, result.failedItems, nil)
|
|
|
|
if result.failCount > 0 && result.successCount == 0 {
|
|
h.importTaskStore.UpdateStatus(ctx, importTask.ID, model.ImportTaskStatusFailed, "所有导入均失败")
|
|
} else {
|
|
h.importTaskStore.UpdateStatus(ctx, importTask.ID, model.ImportTaskStatusCompleted, "")
|
|
}
|
|
|
|
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 nil
|
|
})
|
|
|
|
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++
|
|
}
|
|
}
|
|
|
|
var ErrMissingDeviceNoColumn = stderrors.New("CSV 缺少 virtual_no 列")
|