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

feat: 资产标识符标准化、资产历史订单查询及导入虚拟号强制验证
All checks were successful
构建并部署到测试环境（无 SSH） / build-and-deploy (push) Successful in 7m21s
Details

Browse Source 
主要变更：
- 新增 AssetIdentifier 模型及 Store，统一管理资产标识符（ICCID/IMEI/SN 等）
- 新增迁移：asset_identifier 表、order 表新增 asset_identifier 字段、iot_card.virtual_no NOT NULL 约束
- 资产 Handler/Service/Route 全面重构，支持标识符路由查询与解析
- 新增资产历史订单查询接口，支持跨设备/卡/钱包维度的订单聚合
- 设备与物联卡导入任务强制校验虚拟号，缺失时直接拒绝
- Excel 工具函数优化，前端导入指引文档同步更新
- 归档三个 OpenSpec 提案：asset-identifier-standardization、asset-historical-orders、import-mandatory-virtual-no
- 更新 OpenAPI 文档及相关 DTO

Ultraworked with [Sisyphus](https://github.com/code-yeongyu/oh-my-openagent)

Co-authored-by: Sisyphus <clio-agent@sisyphuslabs.ai>
This commit is contained in:
huang
2026-04-07 17:39:36 +08:00
parent 7e489a19fb
commit 80c6f6c756
69 changed files with 3039 additions and 851 deletions
Download Patch File Download Diff File Expand all files Collapse all files

36
internal/task/device_import.go
View File

@@ -35,6 +35,7 @@ type DeviceImportHandler struct {
deviceSimBindingStore *postgres.DeviceSimBindingStore
iotCardStore *postgres.IotCardStore
assetWalletStore *postgres.AssetWalletStore
assetIdentifierStore *postgres.AssetIdentifierStore
storageService *storage.Service
logger *zap.Logger
}
@@ -47,6 +48,7 @@ func NewDeviceImportHandler(
deviceSimBindingStore *postgres.DeviceSimBindingStore,
iotCardStore *postgres.IotCardStore,
assetWalletStore *postgres.AssetWalletStore,
assetIdentifierStore *postgres.AssetIdentifierStore,
storageSvc *storage.Service,
logger *zap.Logger,
) *DeviceImportHandler {
@@ -58,6 +60,7 @@ func NewDeviceImportHandler(
deviceSimBindingStore: deviceSimBindingStore,
iotCardStore: iotCardStore,
assetWalletStore: assetWalletStore,
assetIdentifierStore: assetIdentifierStore,
storageService: storageSvc,
logger: logger,
}
@@ -98,7 +101,7 @@ func (h *DeviceImportHandler) HandleDeviceImport(ctx context.Context, task *asyn
zap.String("storage_key", importTask.StorageKey),
)
rows, totalCount, err := h.downloadAndParse(ctx, importTask)
parseResult, err := h.downloadAndParse(ctx, importTask)
if err != nil {
h.logger.Error("下载或解析 Excel 失败",
zap.Uint("task_id", importTask.ID),
@@ -108,9 +111,18 @@ func (h *DeviceImportHandler) HandleDeviceImport(ctx context.Context, task *asyn
return asynq.SkipRetry
}
result := h.processImport(ctx, importTask, rows, totalCount)
result := h.processImport(ctx, importTask, parseResult.Rows)
h.importTaskStore.UpdateResult(ctx, importTask.ID, totalCount, result.successCount, result.skipCount, result.failCount, 0, result.skippedItems, result.failedItems, nil)
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, "所有导入均失败")
@@ -128,24 +140,24 @@ func (h *DeviceImportHandler) HandleDeviceImport(ctx context.Context, task *asyn
return nil
}
func (h *DeviceImportHandler) downloadAndParse(ctx context.Context, task *model.DeviceImportTask) ([]utils.DeviceRow, int, error) {
func (h *DeviceImportHandler) downloadAndParse(ctx context.Context, task *model.DeviceImportTask) (*utils.DeviceParseResult, error) {
if h.storageService == nil {
return nil, 0, ErrStorageNotConfigured
return nil, ErrStorageNotConfigured
}
if task.StorageKey == "" {
return nil, 0, ErrStorageKeyEmpty
return nil, ErrStorageKeyEmpty
}
localPath, cleanup, err := h.storageService.DownloadToTemp(ctx, task.StorageKey)
if err != nil {
return nil, 0, err
return nil, err
}
defer cleanup()
if !strings.HasSuffix(strings.ToLower(task.FileName), ".xlsx") {
ext := filepath.Ext(task.FileName)
return nil, 0, fmt.Errorf("不支持的文件格式 %s,请上传Excel文件(.xlsx)", ext)
return nil, fmt.Errorf("不支持的文件格式 %s,请上传Excel文件(.xlsx)", ext)
}
return utils.ParseDeviceExcel(localPath)
@@ -159,7 +171,7 @@ type deviceImportResult struct {
failedItems model.ImportResultItems
}
func (h *DeviceImportHandler) processImport(ctx context.Context, task *model.DeviceImportTask, rows []utils.DeviceRow, totalCount int) *deviceImportResult {
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),
@@ -283,6 +295,11 @@ func (h *DeviceImportHandler) processBatch(ctx context.Context, task *model.Devi
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()
for i, cardID := range validCardIDs {
binding := &model.DeviceSimBinding{
@@ -297,7 +314,6 @@ func (h *DeviceImportHandler) processBatch(ctx context.Context, task *model.Devi
}
}
// 创建设备资产钱包（设备存在即应有钱包，避免购买时才发现缺失）
txWalletStore := postgres.NewAssetWalletStore(tx, nil)
deviceWallet := &model.AssetWallet{
ResourceType: constants.AssetWalletResourceTypeDevice,

162
internal/task/iot_card_import.go
View File

@@ -41,14 +41,15 @@ type PollingCallback interface {
}
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
db *gorm.DB
redis *redis.Client
importTaskStore *postgres.IotCardImportTaskStore
iotCardStore *postgres.IotCardStore
assetWalletStore *postgres.AssetWalletStore
assetIdentifierStore *postgres.AssetIdentifierStore
storageService *storage.Service
pollingCallback PollingCallback
logger *zap.Logger
}
func NewIotCardImportHandler(
@@ -57,19 +58,21 @@ func NewIotCardImportHandler(
importTaskStore *postgres.IotCardImportTaskStore,
iotCardStore *postgres.IotCardStore,
assetWalletStore *postgres.AssetWalletStore,
assetIdentifierStore *postgres.AssetIdentifierStore,
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,
db: db,
redis: redis,
importTaskStore: importTaskStore,
iotCardStore: iotCardStore,
assetWalletStore: assetWalletStore,
assetIdentifierStore: assetIdentifierStore,
storageService: storageSvc,
pollingCallback: pollingCallback,
logger: logger,
}
}
@@ -108,7 +111,7 @@ func (h *IotCardImportHandler) HandleIotCardImport(ctx context.Context, task *as
zap.String("storage_key", importTask.StorageKey),
)
cards, totalCount, err := h.downloadAndParse(ctx, importTask)
cards, totalCount, parseFailures, err := h.downloadAndParse(ctx, importTask)
if err != nil {
h.logger.Error("下载或解析 Excel 失败",
zap.Uint("task_id", importTask.ID),
@@ -124,6 +127,9 @@ func (h *IotCardImportHandler) HandleIotCardImport(ctx context.Context, task *as
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 {
@@ -142,29 +148,29 @@ func (h *IotCardImportHandler) HandleIotCardImport(ctx context.Context, task *as
return nil
}
func (h *IotCardImportHandler) downloadAndParse(ctx context.Context, task *model.IotCardImportTask) (model.CardListJSON, int, error) {
func (h *IotCardImportHandler) downloadAndParse(ctx context.Context, task *model.IotCardImportTask) (model.CardListJSON, int, model.ImportResultItems, error) {
if h.storageService == nil {
return nil, 0, ErrStorageNotConfigured
return nil, 0, nil, ErrStorageNotConfigured
}
if task.StorageKey == "" {
return nil, 0, ErrStorageKeyEmpty
return nil, 0, nil, ErrStorageKeyEmpty
}
localPath, cleanup, err := h.storageService.DownloadToTemp(ctx, task.StorageKey)
if err != nil {
return nil, 0, err
return nil, 0, nil, err
}
defer cleanup()
if !strings.HasSuffix(strings.ToLower(task.FileName), ".xlsx") {
ext := filepath.Ext(task.FileName)
return nil, 0, fmt.Errorf("不支持的文件格式 %s,请上传Excel文件(.xlsx)", ext)
return nil, 0, nil, fmt.Errorf("不支持的文件格式 %s,请上传Excel文件(.xlsx)", ext)
}
parseResult, err := utils.ParseCardExcel(localPath)
if err != nil {
return nil, 0, err
return nil, 0, nil, err
}
cards := make(model.CardListJSON, 0, len(parseResult.Cards))
@@ -176,7 +182,17 @@ func (h *IotCardImportHandler) downloadAndParse(ctx context.Context, task *model
})
}
return cards, parseResult.TotalCount, nil
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 {
@@ -287,42 +303,44 @@ func (h *IotCardImportHandler) processBatch(ctx context.Context, task *model.Iot
return
}
// 批量检查 virtual_no 唯一性
virtualNos := make([]string, 0)
virtualNos := make([]string, 0, len(newCards))
for _, card := range newCards {
if card.VirtualNo != "" {
virtualNos = append(virtualNos, card.VirtualNo)
}
virtualNos = append(virtualNos, card.VirtualNo)
}
existingVirtualNos := make(map[string]bool)
if len(virtualNos) > 0 {
existingVirtualNos, err = h.iotCardStore.ExistsByVirtualNoBatch(ctx, virtualNos)
if err != nil {
h.logger.Error("批量检查 virtual_no 是否存在失败",
zap.Error(err),
zap.Int("batch_size", len(virtualNos)),
)
}
existingVirtualNos, err = h.iotCardStore.ExistsByVirtualNoBatch(ctx, virtualNos)
if err != nil {
h.logger.Error("批量检查 virtual_no 是否存在失败",
zap.Error(err),
zap.Int("batch_size", len(virtualNos)),
)
}
// 批内去重：记录本批次已分配的 virtual_no
batchUsedVirtualNos := make(map[string]bool)
finalCards := make([]model.CardItem, 0, len(newCards))
for _, card := range newCards {
meta := cardMetaMap[card.ICCID]
if card.VirtualNo != "" {
if 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
if card.VirtualNo == "" {
result.failedItems = append(result.failedItems, model.ImportResultItem{
Line: meta.line,
ICCID: card.ICCID,
MSISDN: meta.msisdn,
Reason: "虚拟号(virtual_no)不能为空",
})
result.failCount++
continue
}
if 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)
}
@@ -330,9 +348,11 @@ func (h *IotCardImportHandler) processBatch(ctx context.Context, task *model.Iot
return
}
iotCards := make([]*model.IotCard, 0, len(finalCards))
now := time.Now()
createdCards := make([]*model.IotCard, 0, len(finalCards))
for _, card := range finalCards {
meta := cardMetaMap[card.ICCID]
iotCard := &model.IotCard{
ICCID: card.ICCID,
MSISDN: card.MSISDN,
@@ -349,34 +369,42 @@ func (h *IotCardImportHandler) processBatch(ctx context.Context, task *model.Iot
iotCard.BaseModel.Updater = task.Creator
iotCard.CreatedAt = now
iotCard.UpdatedAt = now
iotCards = append(iotCards, iotCard)
}
if err := h.iotCardStore.CreateBatch(ctx, iotCards); err != nil {
h.logger.Error("批量创建 IoT 卡失败",
zap.Error(err),
zap.Int("batch_size", len(iotCards)),
)
for _, card := range finalCards {
meta := cardMetaMap[card.ICCID]
txErr := h.db.Transaction(func(tx *gorm.DB) error {
if err := tx.Create(iotCard).Error; err != nil {
return err
}
txIdentifierStore := postgres.NewAssetIdentifierStore(tx)
if err := txIdentifierStore.Register(ctx, card.ICCID, model.AssetTypeIotCard, iotCard.ID); err != nil {
return fmt.Errorf("ICCID 已被占用: %s", card.ICCID)
}
if err := txIdentifierStore.Register(ctx, card.VirtualNo, model.AssetTypeIotCard, iotCard.ID); err != nil {
return fmt.Errorf("虚拟号已被占用: %s", card.VirtualNo)
}
return nil
})
if txErr != nil {
h.logger.Error("创建 IoT 卡失败",
zap.String("iccid", card.ICCID),
zap.Error(txErr),
)
result.failedItems = append(result.failedItems, model.ImportResultItem{
Line: meta.line,
ICCID: card.ICCID,
MSISDN: meta.msisdn,
Reason: "数据库写入失败",
Reason: txErr.Error(),
})
result.failCount++
continue
}
return
createdCards = append(createdCards, iotCard)
result.successCount++
}
result.successCount += len(finalCards)
// 为导入的卡批量创建资产钱包（钱包创建失败不影响卡导入结果，后续访问时 getOrCreateWallet 兜底）
h.batchCreateWallets(ctx, iotCards)
h.batchCreateWallets(ctx, createdCards)
if h.pollingCallback != nil {
h.pollingCallback.OnBatchCardsCreated(ctx, iotCards)
h.pollingCallback.OnBatchCardsCreated(ctx, createdCards)
}
}