break
44fb21eb6a
All checks were successful
构建并部署到测试环境（无 SSH） / build-and-deploy (push) Successful in 7m53s
499 lines
14 KiB
Go
499 lines
14 KiB
Go
package task
|
||
|
||
import (
|
||
"context"
|
||
"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"
|
||
"github.com/break/junhong_cmp_fiber/pkg/validator"
|
||
)
|
||
|
||
var (
|
||
ErrStorageNotConfigured = errors.New("对象存储服务未配置")
|
||
ErrStorageKeyEmpty = errors.New("文件存储路径为空")
|
||
)
|
||
|
||
const batchSize = 1000
|
||
|
||
type IotCardImportPayload struct {
|
||
TaskID uint `json:"task_id"`
|
||
}
|
||
|
||
// PollingCallback 轮询回调接口
|
||
// 用于在卡创建/删除/状态变化时通知轮询系统
|
||
type PollingCallback interface {
|
||
// OnBatchCardsCreated 批量卡创建时的回调
|
||
OnBatchCardsCreated(ctx context.Context, cards []*model.IotCard)
|
||
}
|
||
|
||
type IotCardImportHandler struct {
|
||
db *gorm.DB
|
||
redis *redis.Client
|
||
importTaskStore *postgres.IotCardImportTaskStore
|
||
iotCardStore *postgres.IotCardStore
|
||
assetWalletStore *postgres.AssetWalletStore
|
||
storageService *storage.Service
|
||
pollingCallback PollingCallback
|
||
logger *zap.Logger
|
||
}
|
||
|
||
func NewIotCardImportHandler(
|
||
db *gorm.DB,
|
||
redis *redis.Client,
|
||
importTaskStore *postgres.IotCardImportTaskStore,
|
||
iotCardStore *postgres.IotCardStore,
|
||
assetWalletStore *postgres.AssetWalletStore,
|
||
storageSvc *storage.Service,
|
||
pollingCallback PollingCallback,
|
||
logger *zap.Logger,
|
||
) *IotCardImportHandler {
|
||
return &IotCardImportHandler{
|
||
db: db,
|
||
redis: redis,
|
||
importTaskStore: importTaskStore,
|
||
iotCardStore: iotCardStore,
|
||
assetWalletStore: assetWalletStore,
|
||
storageService: storageSvc,
|
||
pollingCallback: pollingCallback,
|
||
logger: logger,
|
||
}
|
||
}
|
||
|
||
func (h *IotCardImportHandler) HandleIotCardImport(ctx context.Context, task *asynq.Task) error {
|
||
var payload IotCardImportPayload
|
||
if err := sonic.Unmarshal(task.Payload(), &payload); err != nil {
|
||
h.logger.Error("解析 IoT 卡导入任务载荷失败",
|
||
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
|
||
}
|
||
|
||
switch importTask.Status {
|
||
case model.ImportTaskStatusPending:
|
||
// 正常首次处理
|
||
case model.ImportTaskStatusProcessing:
|
||
// 上次 worker 中途中断（重启/崩溃），重置计数重新处理
|
||
// 已入库的卡会被 ExistsByICCIDBatch 识别为 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("开始处理 IoT 卡导入任务",
|
||
zap.Uint("task_id", importTask.ID),
|
||
zap.String("task_no", importTask.TaskNo),
|
||
zap.String("storage_key", importTask.StorageKey),
|
||
)
|
||
|
||
cards, totalCount, parseFailures, 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
|
||
}
|
||
|
||
importTask.CardList = cards
|
||
importTask.TotalCount = totalCount
|
||
h.importTaskStore.UpdateCardList(ctx, importTask.ID, cards, totalCount)
|
||
|
||
result := h.processImport(ctx, importTask)
|
||
|
||
result.failedItems = append(parseFailures, result.failedItems...)
|
||
result.failCount += len(parseFailures)
|
||
|
||
h.importTaskStore.UpdateResult(ctx, importTask.ID, result.successCount, result.skipCount, result.failCount, result.skippedItems, result.failedItems)
|
||
|
||
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("IoT 卡导入任务完成",
|
||
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 *IotCardImportHandler) downloadAndParse(ctx context.Context, task *model.IotCardImportTask) (model.CardListJSON, int, model.ImportResultItems, error) {
|
||
if h.storageService == nil {
|
||
return nil, 0, nil, ErrStorageNotConfigured
|
||
}
|
||
|
||
if task.StorageKey == "" {
|
||
return nil, 0, nil, ErrStorageKeyEmpty
|
||
}
|
||
|
||
localPath, cleanup, err := h.storageService.DownloadToTemp(ctx, task.StorageKey)
|
||
if err != nil {
|
||
return nil, 0, nil, err
|
||
}
|
||
defer cleanup()
|
||
|
||
if !strings.HasSuffix(strings.ToLower(task.FileName), ".xlsx") {
|
||
ext := filepath.Ext(task.FileName)
|
||
return nil, 0, nil, fmt.Errorf("不支持的文件格式 %s,请上传Excel文件(.xlsx)", ext)
|
||
}
|
||
|
||
parseResult, err := utils.ParseCardExcel(localPath)
|
||
if err != nil {
|
||
return nil, 0, nil, err
|
||
}
|
||
|
||
cards := make(model.CardListJSON, 0, len(parseResult.Cards))
|
||
for _, card := range parseResult.Cards {
|
||
cards = append(cards, model.CardItem{
|
||
ICCID: card.ICCID,
|
||
MSISDN: card.MSISDN,
|
||
VirtualNo: card.VirtualNo,
|
||
})
|
||
}
|
||
|
||
parseFailures := make(model.ImportResultItems, 0, len(parseResult.ParseErrors))
|
||
for _, pe := range parseResult.ParseErrors {
|
||
parseFailures = append(parseFailures, model.ImportResultItem{
|
||
Line: pe.Line,
|
||
ICCID: pe.ICCID,
|
||
MSISDN: pe.MSISDN,
|
||
Reason: pe.Reason,
|
||
})
|
||
}
|
||
|
||
return cards, parseResult.TotalCount, parseFailures, nil
|
||
}
|
||
|
||
type importResult struct {
|
||
successCount int
|
||
skipCount int
|
||
failCount int
|
||
skippedItems model.ImportResultItems
|
||
failedItems model.ImportResultItems
|
||
}
|
||
|
||
func (h *IotCardImportHandler) processImport(ctx context.Context, task *model.IotCardImportTask) *importResult {
|
||
result := &importResult{
|
||
skippedItems: make(model.ImportResultItems, 0),
|
||
failedItems: make(model.ImportResultItems, 0),
|
||
}
|
||
|
||
cards := h.getCardsFromTask(task)
|
||
if len(cards) == 0 {
|
||
return result
|
||
}
|
||
|
||
for i := 0; i < len(cards); i += batchSize {
|
||
end := min(i+batchSize, len(cards))
|
||
batch := cards[i:end]
|
||
h.processBatch(ctx, task, batch, i+1, result)
|
||
// 每批完成后实时更新进度计数，让前端可以看到处理进度
|
||
h.importTaskStore.UpdateProgress(ctx, task.ID, result.successCount, result.skipCount, result.failCount)
|
||
}
|
||
|
||
return result
|
||
}
|
||
|
||
// getCardsFromTask 从任务中获取待导入的卡列表
|
||
func (h *IotCardImportHandler) getCardsFromTask(task *model.IotCardImportTask) []model.CardItem {
|
||
return []model.CardItem(task.CardList)
|
||
}
|
||
|
||
func (h *IotCardImportHandler) processBatch(ctx context.Context, task *model.IotCardImportTask, batch []model.CardItem, startLine int, result *importResult) {
|
||
type cardMeta struct {
|
||
line int
|
||
msisdn string
|
||
virtualNo string
|
||
}
|
||
validCards := make([]model.CardItem, 0)
|
||
cardMetaMap := make(map[string]cardMeta)
|
||
|
||
for i, card := range batch {
|
||
line := startLine + i
|
||
cardMetaMap[card.ICCID] = cardMeta{line: line, msisdn: card.MSISDN, virtualNo: card.VirtualNo}
|
||
|
||
validationResult := validator.ValidateICCID(card.ICCID, task.CarrierType)
|
||
if !validationResult.Valid {
|
||
result.failedItems = append(result.failedItems, model.ImportResultItem{
|
||
Line: line,
|
||
ICCID: card.ICCID,
|
||
MSISDN: card.MSISDN,
|
||
Reason: validationResult.Message,
|
||
})
|
||
result.failCount++
|
||
continue
|
||
}
|
||
validCards = append(validCards, card)
|
||
}
|
||
|
||
if len(validCards) == 0 {
|
||
return
|
||
}
|
||
|
||
validICCIDs := make([]string, len(validCards))
|
||
for i, card := range validCards {
|
||
validICCIDs[i] = card.ICCID
|
||
}
|
||
|
||
existingMap, err := h.iotCardStore.ExistsByICCIDBatch(ctx, validICCIDs)
|
||
if err != nil {
|
||
h.logger.Error("批量检查 ICCID 是否存在失败",
|
||
zap.Error(err),
|
||
zap.Int("batch_size", len(validICCIDs)),
|
||
)
|
||
for _, card := range validCards {
|
||
meta := cardMetaMap[card.ICCID]
|
||
result.failedItems = append(result.failedItems, model.ImportResultItem{
|
||
Line: meta.line,
|
||
ICCID: card.ICCID,
|
||
MSISDN: meta.msisdn,
|
||
Reason: "数据库查询失败",
|
||
})
|
||
result.failCount++
|
||
}
|
||
return
|
||
}
|
||
|
||
newCards := make([]model.CardItem, 0)
|
||
for _, card := range validCards {
|
||
meta := cardMetaMap[card.ICCID]
|
||
if existingMap[card.ICCID] {
|
||
result.skippedItems = append(result.skippedItems, model.ImportResultItem{
|
||
Line: meta.line,
|
||
ICCID: card.ICCID,
|
||
MSISDN: meta.msisdn,
|
||
Reason: "ICCID 已存在",
|
||
})
|
||
result.skipCount++
|
||
} else {
|
||
newCards = append(newCards, card)
|
||
}
|
||
}
|
||
|
||
if len(newCards) == 0 {
|
||
return
|
||
}
|
||
|
||
virtualNos := make([]string, 0, len(newCards))
|
||
for _, card := range newCards {
|
||
if card.VirtualNo != "" {
|
||
virtualNos = append(virtualNos, card.VirtualNo)
|
||
}
|
||
}
|
||
existingVirtualNos := make(map[string]bool)
|
||
existingVirtualNos, err = h.iotCardStore.ExistsByVirtualNoBatch(ctx, virtualNos)
|
||
if err != nil {
|
||
h.logger.Error("批量检查 virtual_no 是否存在失败",
|
||
zap.Error(err),
|
||
zap.Int("batch_size", len(virtualNos)),
|
||
)
|
||
}
|
||
batchUsedVirtualNos := make(map[string]bool)
|
||
|
||
finalCards := make([]model.CardItem, 0, len(newCards))
|
||
for _, card := range newCards {
|
||
meta := cardMetaMap[card.ICCID]
|
||
if card.VirtualNo != "" && (existingVirtualNos[card.VirtualNo] || batchUsedVirtualNos[card.VirtualNo]) {
|
||
result.failedItems = append(result.failedItems, model.ImportResultItem{
|
||
Line: meta.line,
|
||
ICCID: card.ICCID,
|
||
MSISDN: meta.msisdn,
|
||
Reason: "virtual_no 已被占用: " + card.VirtualNo,
|
||
})
|
||
result.failCount++
|
||
continue
|
||
}
|
||
batchUsedVirtualNos[card.VirtualNo] = true
|
||
finalCards = append(finalCards, card)
|
||
}
|
||
|
||
if len(finalCards) == 0 {
|
||
return
|
||
}
|
||
|
||
now := time.Now()
|
||
|
||
// 构建待批量写入的卡列表，同时记录原始信息用于失败回溯
|
||
type pendingEntry struct {
|
||
item model.CardItem
|
||
meta cardMeta
|
||
}
|
||
pending := make([]pendingEntry, 0, len(finalCards))
|
||
iotCards := make([]*model.IotCard, 0, len(finalCards))
|
||
|
||
for _, card := range finalCards {
|
||
meta := cardMetaMap[card.ICCID]
|
||
|
||
// 拆分 ICCID 为双列存储值（长度校验已在 ValidateICCID 通过，此处作防御性保留）
|
||
iccid19, iccid20 := utils.SplitICCID(card.ICCID)
|
||
if iccid19 == "" {
|
||
h.logger.Error("ICCID 长度异常，跳过写入",
|
||
zap.String("iccid", card.ICCID),
|
||
zap.Int("length", len(card.ICCID)),
|
||
)
|
||
result.failedItems = append(result.failedItems, model.ImportResultItem{
|
||
Line: meta.line,
|
||
ICCID: card.ICCID,
|
||
MSISDN: meta.msisdn,
|
||
Reason: "ICCID 长度异常，无法拆分",
|
||
})
|
||
result.failCount++
|
||
continue
|
||
}
|
||
|
||
iotCard := &model.IotCard{
|
||
ICCID: card.ICCID,
|
||
ICCID19: iccid19,
|
||
ICCID20: iccid20,
|
||
MSISDN: card.MSISDN,
|
||
VirtualNo: card.VirtualNo,
|
||
CarrierID: task.CarrierID,
|
||
CarrierType: task.CarrierType,
|
||
CarrierName: task.CarrierName,
|
||
BatchNo: task.BatchNo,
|
||
Status: constants.IotCardStatusInStock,
|
||
CardCategory: task.CardCategory,
|
||
ActivationStatus: constants.ActivationStatusInactive,
|
||
RealNameStatus: constants.RealNameStatusNotVerified,
|
||
NetworkStatus: constants.NetworkStatusOffline,
|
||
StopReason: constants.StopReasonNoPackage,
|
||
RealnamePolicy: task.RealnamePolicy,
|
||
}
|
||
iotCard.BaseModel.Creator = task.Creator
|
||
iotCard.BaseModel.Updater = task.Creator
|
||
iotCard.CreatedAt = now
|
||
iotCard.UpdatedAt = now
|
||
|
||
iotCards = append(iotCards, iotCard)
|
||
pending = append(pending, pendingEntry{item: card, meta: meta})
|
||
}
|
||
|
||
if len(iotCards) == 0 {
|
||
return
|
||
}
|
||
|
||
// 整批一个事务：批量写入 iot_card + asset_identifier
|
||
// 相比逐卡事务，将 N 个事务压缩为 1 个，大幅减少数据库往返
|
||
txErr := h.db.Transaction(func(tx *gorm.DB) error {
|
||
// 批量插入 iot_card，GORM 会通过 RETURNING 回填所有 ID
|
||
if err := tx.CreateInBatches(&iotCards, 500).Error; err != nil {
|
||
return err
|
||
}
|
||
// 用回填的 ID 构建 asset_identifier 列表
|
||
identifiers := make([]*model.AssetIdentifier, 0, len(iotCards)*2)
|
||
for _, c := range iotCards {
|
||
identifiers = append(identifiers, &model.AssetIdentifier{
|
||
Identifier: c.ICCID,
|
||
AssetType: model.AssetTypeIotCard,
|
||
AssetID: c.ID,
|
||
})
|
||
if c.VirtualNo != "" {
|
||
identifiers = append(identifiers, &model.AssetIdentifier{
|
||
Identifier: c.VirtualNo,
|
||
AssetType: model.AssetTypeIotCard,
|
||
AssetID: c.ID,
|
||
})
|
||
}
|
||
}
|
||
return tx.CreateInBatches(&identifiers, 500).Error
|
||
})
|
||
|
||
if txErr != nil {
|
||
h.logger.Error("批量创建 IoT 卡失败",
|
||
zap.Error(txErr),
|
||
zap.Int("batch_size", len(iotCards)),
|
||
)
|
||
for _, p := range pending {
|
||
result.failedItems = append(result.failedItems, model.ImportResultItem{
|
||
Line: p.meta.line,
|
||
ICCID: p.item.ICCID,
|
||
MSISDN: p.meta.msisdn,
|
||
Reason: "批量写入失败: " + txErr.Error(),
|
||
})
|
||
result.failCount++
|
||
}
|
||
return
|
||
}
|
||
|
||
result.successCount += len(iotCards)
|
||
|
||
h.batchCreateWallets(ctx, iotCards)
|
||
|
||
if h.pollingCallback != nil {
|
||
h.pollingCallback.OnBatchCardsCreated(ctx, iotCards)
|
||
}
|
||
}
|
||
|
||
// batchCreateWallets 批量为 IoT 卡创建资产钱包
|
||
func (h *IotCardImportHandler) batchCreateWallets(ctx context.Context, cards []*model.IotCard) {
|
||
if h.assetWalletStore == nil {
|
||
return
|
||
}
|
||
|
||
wallets := make([]*model.AssetWallet, 0, len(cards))
|
||
for _, card := range cards {
|
||
if card.ID == 0 {
|
||
continue
|
||
}
|
||
wallets = append(wallets, &model.AssetWallet{
|
||
ResourceType: constants.AssetWalletResourceTypeIotCard,
|
||
ResourceID: card.ID,
|
||
Balance: 0,
|
||
FrozenBalance: 0,
|
||
Currency: "CNY",
|
||
Status: constants.AssetWalletStatusNormal,
|
||
Version: 0,
|
||
})
|
||
}
|
||
|
||
if len(wallets) == 0 {
|
||
return
|
||
}
|
||
|
||
if err := h.db.WithContext(ctx).Create(&wallets).Error; err != nil {
|
||
h.logger.Warn("批量创建 IoT 卡资产钱包失败，后续访问时将自动创建",
|
||
zap.Int("count", len(wallets)),
|
||
zap.Error(err),
|
||
)
|
||
}
|
||
}
|