不允许购买第二个主套餐

internal/bootstrap/services.go

@@ -264,7 +264,7 @@ func initServices(s *stores, deps *Dependencies) *services {
 		Recharge:                   rechargeSvc.New(deps.DB, s.AssetWallet, s.AssetWalletTransaction, s.IotCard, s.Device, s.ShopSeriesAllocation, s.PackageSeries, s.CommissionRecord, wechatConfig, paymentLoader, deps.Logger),
 		PollingConfig:              pollingSvc.NewConfigService(s.PollingConfig, deps.Redis, deps.Logger),
 		PollingConcurrency:         pollingSvc.NewConcurrencyService(s.PollingConcurrencyConfig, deps.Redis),
-		PollingMonitoring:          pollingSvc.NewMonitoringService(deps.Redis),
+		PollingMonitoring:          pollingSvc.NewMonitoringServiceWithQueueMgr(deps.Redis, pollingQueueMgr, deps.Logger),
 		PollingAlert:               pollingSvc.NewAlertService(s.PollingAlertRule, s.PollingAlertHistory, deps.Redis, deps.Logger),
 		PollingCleanup:             pollingSvc.NewCleanupService(s.DataCleanupConfig, s.DataCleanupLog, deps.Logger),
 		PollingManualTrigger:       pollingSvc.NewManualTriggerService(s.PollingManualTriggerLog, s.IotCard, deps.Redis, deps.Logger),

internal/handler/app/client_asset.go

@@ -63,15 +63,16 @@ func NewClientAssetHandler(
 }
 
 type resolvedAssetContext struct {
-	CustomerID          uint
-	Identifier          string
-	Asset               *dto.AssetResolveResponse
-	Generation          int
-	WalletBalance       int64
-	SkipPermissionCtx   context.Context
-	IsAgentChannel      bool
-	SellerShopID        uint
-	HasAddonMainPackage bool
+	CustomerID           uint
+	Identifier           string
+	Asset                *dto.AssetResolveResponse
+	Generation           int
+	WalletBalance        int64
+	SkipPermissionCtx    context.Context
+	IsAgentChannel       bool
+	SellerShopID         uint
+	HasAddonMainPackage  bool
+	HasFormalMainPackage bool
 }
 
 // resolveAssetFromIdentifier 统一执行资产解析与归属校验
@@ -242,6 +243,16 @@ func (h *ClientAssetHandler) GetAvailablePackages(c *fiber.Ctx) error {
 	} else if err != gorm.ErrRecordNotFound {
 		return errors.Wrap(errors.CodeDatabaseError, err, "查询主套餐失败")
 	}
+	if hasBlockingMain, err := packagepkg.HasBlockingMainPackageForFormal(
+		h.db.WithContext(resolved.SkipPermissionCtx),
+		resolved.Asset.AssetType,
+		resolved.Asset.AssetID,
+		time.Now(),
+	); err != nil {
+		return errors.Wrap(errors.CodeDatabaseError, err, "查询主套餐失败")
+	} else {
+		resolved.HasFormalMainPackage = hasBlockingMain
+	}
 
 	listCtx := resolved.SkipPermissionCtx
 	if resolved.IsAgentChannel {
@@ -530,6 +541,9 @@ func buildClientPackageItem(
 	}
 
 	isAddon := pkg.PackageType == constants.PackageTypeAddon
+	if pkg.PackageType == constants.PackageTypeFormal && resolved.HasFormalMainPackage {
+		return dto.ClientPackageItem{}, false
+	}
 	if isAddon && !resolved.HasAddonMainPackage {
 		return dto.ClientPackageItem{}, false
 	}

internal/polling/queue_manager.go

@@ -186,3 +186,66 @@ func (m *PollingQueueManager) GetTotalQueueDepth(ctx context.Context, taskType s
 	}
 	return total, firstErr
 }
+
+// GetTotalDueCount 获取指定任务类型所有分片中已到期的队列数量。
+// 只统计 score <= now 的成员，用于后台监控展示真实待调度积压。
+func (m *PollingQueueManager) GetTotalDueCount(ctx context.Context, taskType string, now time.Time) (int64, error) {
+	var total int64
+	var firstErr error
+	for i := 0; i < m.shardCount; i++ {
+		key := constants.RedisPollingShardQueueKey(i, taskType)
+		count, err := m.redis.ZCount(ctx, key, "-inf", fmt.Sprintf("%d", now.Unix())).Result()
+		if err != nil {
+			m.logger.Warn("获取分片到期数量失败",
+				zap.Int("shard_id", i),
+				zap.String("task_type", taskType),
+				zap.Error(err))
+			if firstErr == nil {
+				firstErr = err
+			}
+			continue
+		}
+		total += count
+	}
+	return total, firstErr
+}
+
+// GetAverageWaitTime 获取指定任务类型所有分片最早若干任务的平均等待秒数。
+// 每个分片最多取 samplePerShard 条，避免监控接口扫描完整队列。
+func (m *PollingQueueManager) GetAverageWaitTime(ctx context.Context, taskType string, now time.Time, samplePerShard int64) (float64, error) {
+	if samplePerShard <= 0 {
+		samplePerShard = 10
+	}
+
+	var totalWait float64
+	var sampleCount int64
+	var firstErr error
+	nowUnix := now.Unix()
+
+	for i := 0; i < m.shardCount; i++ {
+		key := constants.RedisPollingShardQueueKey(i, taskType)
+		earliest, err := m.redis.ZRangeWithScores(ctx, key, 0, samplePerShard-1).Result()
+		if err != nil {
+			m.logger.Warn("获取分片最早任务失败",
+				zap.Int("shard_id", i),
+				zap.String("task_type", taskType),
+				zap.Error(err))
+			if firstErr == nil {
+				firstErr = err
+			}
+			continue
+		}
+		for _, z := range earliest {
+			waitTime := float64(nowUnix) - z.Score
+			if waitTime > 0 {
+				totalWait += waitTime
+			}
+			sampleCount++
+		}
+	}
+
+	if sampleCount == 0 {
+		return 0, firstErr
+	}
+	return totalWait / float64(sampleCount), firstErr
+}

internal/service/client_order/service.go

@@ -13,6 +13,7 @@ import (
 	"github.com/break/junhong_cmp_fiber/internal/model"
 	"github.com/break/junhong_cmp_fiber/internal/model/dto"
 	asset "github.com/break/junhong_cmp_fiber/internal/service/asset"
+	packagepkg "github.com/break/junhong_cmp_fiber/internal/service/package"
 	"github.com/break/junhong_cmp_fiber/internal/service/purchase_validation"
 	"github.com/break/junhong_cmp_fiber/internal/store/postgres"
 	"github.com/break/junhong_cmp_fiber/pkg/alipay"
@@ -1247,6 +1248,10 @@ func (s *Service) PayOrder(ctx context.Context, customerID uint, orderID uint, r
 		return &dto.ClientPayOrderResponse{PaymentMethod: model.PaymentMethodWallet}, nil
 
 	case model.PaymentMethodWechat:
+		if err := s.ensureOrderFormalPackagePayable(skipCtx, order); err != nil {
+			return nil, err
+		}
+
 		activeConfig, appID, err := s.resolveWechatConfig(skipCtx, req.AppType)
 		if err != nil {
 			return nil, err
@@ -1303,6 +1308,10 @@ func (s *Service) PayOrder(ctx context.Context, customerID uint, orderID uint, r
 		}, nil
 
 	case model.PaymentMethodAlipay:
+		if err := s.ensureOrderFormalPackagePayable(skipCtx, order); err != nil {
+			return nil, err
+		}
+
 		// 支付宝支付：不需要 app_type / OpenID
 		activeConfig, err := s.wechatConfigService.GetActiveConfig(skipCtx)
 		if err != nil {
@@ -1337,6 +1346,58 @@ func (s *Service) PayOrder(ctx context.Context, customerID uint, orderID uint, r
 	}
 }
 
+func (s *Service) ensureOrderFormalPackagePayable(ctx context.Context, order *model.Order) error {
+	if order == nil {
+		return errors.New(errors.CodeInvalidParam)
+	}
+
+	var items []*model.OrderItem
+	if err := s.db.WithContext(ctx).Where("order_id = ?", order.ID).Find(&items).Error; err != nil {
+		return errors.Wrap(errors.CodeDatabaseError, err, "查询订单明细失败")
+	}
+
+	hasFormal := false
+	for _, item := range items {
+		if item == nil || item.PackageType == nil {
+			continue
+		}
+		if *item.PackageType == constants.PackageTypeFormal {
+			hasFormal = true
+			break
+		}
+	}
+	if !hasFormal {
+		return nil
+	}
+
+	carrierType := ""
+	carrierID := uint(0)
+	if order.OrderType == model.OrderTypeSingleCard && order.IotCardID != nil {
+		carrierType = constants.AssetWalletResourceTypeIotCard
+		carrierID = *order.IotCardID
+	} else if order.OrderType == model.OrderTypeDevice && order.DeviceID != nil {
+		carrierType = constants.AssetWalletResourceTypeDevice
+		carrierID = *order.DeviceID
+	}
+	if carrierID == 0 {
+		return errors.New(errors.CodeInvalidParam, "订单缺少套餐载体")
+	}
+
+	hasBlockingMain, err := packagepkg.HasBlockingMainPackageForFormal(
+		s.db.WithContext(ctx),
+		carrierType,
+		carrierID,
+		time.Now(),
+	)
+	if err != nil {
+		return errors.Wrap(errors.CodeDatabaseError, err, "查询主套餐失败")
+	}
+	if hasBlockingMain {
+		return errors.New(errors.CodeMainPackageExists)
+	}
+	return nil
+}
+
 func (s *Service) getAssetGeneration(ctx context.Context, assetType string, assetID uint) (int, error) {
 	switch assetType {
 	case "card":

internal/service/order/service.go

@@ -25,6 +25,7 @@ import (
 	"github.com/redis/go-redis/v9"
 	"go.uber.org/zap"
 	"gorm.io/gorm"
+	"gorm.io/gorm/clause"
 )
 
 // WechatConfigServiceInterface 支付配置服务接口
@@ -602,7 +603,7 @@ func (s *Service) CreateH5Order(ctx context.Context, req *dto.CreateOrderRequest
 	purchaseRole := ""
 	var sellerShopID *uint = resourceShopID
 	var sellerCostPrice int64
-	var expiresAt *time.Time          // 待支付订单设置过期时间，立即支付的订单为 nil
+	var expiresAt *time.Time            // 待支付订单设置过期时间，立即支付的订单为 nil
 	var itemCostPriceMap map[uint]int64 // 各套餐成本单价映射，用于 OrderItem.CostPrice 字段
 
 	// 场景判断：offline（平台代购）、wallet（代理钱包支付）、其他（待支付）
@@ -2029,6 +2030,24 @@ func (s *Service) activatePackage(ctx context.Context, tx *gorm.DB, order *model
 	return nil
 }
 
+func (s *Service) lockPackageCarrier(ctx context.Context, tx *gorm.DB, carrierType string, carrierID uint) error {
+	switch carrierType {
+	case constants.AssetWalletResourceTypeIotCard, "card":
+		var card model.IotCard
+		if err := tx.WithContext(ctx).Clauses(clause.Locking{Strength: "UPDATE"}).First(&card, carrierID).Error; err != nil {
+			return errors.Wrap(errors.CodeDatabaseError, err, "锁定卡失败")
+		}
+	case constants.AssetWalletResourceTypeDevice:
+		var device model.Device
+		if err := tx.WithContext(ctx).Clauses(clause.Locking{Strength: "UPDATE"}).First(&device, carrierID).Error; err != nil {
+			return errors.Wrap(errors.CodeDatabaseError, err, "锁定设备失败")
+		}
+	default:
+		return errors.New(errors.CodeInvalidParam, "无效的套餐载体类型")
+	}
+	return nil
+}
+
 // validatePackageTypeMix 任务 8.1: 检查混买限制
 func (s *Service) validatePackageTypeMix(tx *gorm.DB, items []*model.OrderItem) error {
 	hasFormal := false
@@ -2204,15 +2223,17 @@ func (s *Service) markOrderCreated(ctx context.Context, idempotencyKey string, o
 
 // activateMainPackage 任务 8.2-8.4: 主套餐激活逻辑
 func (s *Service) activateMainPackage(ctx context.Context, tx *gorm.DB, order *model.Order, pkg *model.Package, carrierType string, carrierID uint, now time.Time) error {
-	// 检查是否有生效中主套餐
-	var activeMainPackage model.PackageUsage
-	err := tx.Where("status = ?", constants.PackageUsageStatusActive).
-		Where("master_usage_id IS NULL").
-		Where(carrierType+"_id = ?", carrierID).
-		Order("priority ASC").
-		First(&activeMainPackage).Error
+	if err := s.lockPackageCarrier(ctx, tx, carrierType, carrierID); err != nil {
+		return err
+	}
 
-	hasActiveMain := err == nil
+	hasBlockingMain, err := packagepkg.HasBlockingMainPackageForFormal(tx.WithContext(ctx), carrierType, carrierID, now)
+	if err != nil {
+		return errors.Wrap(errors.CodeDatabaseError, err, "查询主套餐失败")
+	}
+	if hasBlockingMain {
+		return errors.New(errors.CodeMainPackageExists)
+	}
 
 	var status int
 	var priority int
@@ -2221,80 +2242,63 @@ func (s *Service) activateMainPackage(ctx context.Context, tx *gorm.DB, order *m
 	var nextResetAt *time.Time
 	var pendingRealnameActivation bool
 
-	if hasActiveMain {
-		// 任务 8.3: 有生效中主套餐，新套餐排队
-		status = constants.PackageUsageStatusPending
-		// 查询当前最大优先级
-		var maxPriority int
-		tx.Model(&model.PackageUsage{}).
-			Where(carrierType+"_id = ?", carrierID).
-			Select("COALESCE(MAX(priority), 0)").
-			Scan(&maxPriority)
-		priority = maxPriority + 1
-		// 排队套餐暂不设置激活时间和过期时间（由激活任务处理）
-	} else {
-		// 任务 8.4: 无生效中主套餐，立即激活
-		status = constants.PackageUsageStatusActive
-		priority = 1
-		activatedAt = now
-		// 使用工具函数计算过期时间
-		expiresAt = packagepkg.CalculateExpiryTime(pkg.CalendarType, activatedAt, pkg.DurationMonths, pkg.DurationDays)
-		// 计算下次重置时间（基于套餐周期类型）
-		nextResetAt = packagepkg.CalculateNextResetTime(pkg.DataResetCycle, pkg.CalendarType, now, activatedAt)
-	}
+	status = constants.PackageUsageStatusActive
+	priority = 1
+	activatedAt = now
+	// 使用工具函数计算过期时间
+	expiresAt = packagepkg.CalculateExpiryTime(pkg.CalendarType, activatedAt, pkg.DurationMonths, pkg.DurationDays)
+	// 计算下次重置时间（基于套餐周期类型）
+	nextResetAt = packagepkg.CalculateNextResetTime(pkg.DataResetCycle, pkg.CalendarType, now, activatedAt)
 
 	// REALNAME-02: 后台囤货场景三维决策（卡类型 + 购买路径 + expiry_base + 实名状态）
-	// 注意：仅在 else 分支（立即激活）时才需要判断是否切换为等实名
-	if !hasActiveMain {
-		// 查询卡类型（行业卡永远直接激活，不等实名）
-		// 同时查询载体当前实名状态：已实名则跳过 pending，直接激活
-		var cardCategory string
-		var currentlyRealnamed bool
-		if carrierType == "iot_card" {
-			var card model.IotCard
-			if err := tx.Select("card_category", "real_name_status").First(&card, carrierID).Error; err == nil {
-				cardCategory = card.CardCategory
-				currentlyRealnamed = card.RealNameStatus == constants.RealNameStatusVerified
-			}
-			// err != nil 时 currentlyRealnamed 保守默认 false，不阻断购买流程
-		} else if carrierType == "device" {
-			// 子查询：统计该设备绑定的已实名卡数量（GORM 自动注入 deleted_at IS NULL）
-			var count int64
-			subQuery := tx.Model(&model.DeviceSimBinding{}).
-				Select("iot_card_id").
-				Where("device_id = ? AND bind_status = 1", carrierID)
-			if err := tx.Model(&model.IotCard{}).
-				Where("id IN (?) AND real_name_status = 1", subQuery).
-				Count(&count).Error; err == nil {
-				currentlyRealnamed = count > 0
-			}
-			// err != nil 时保守默认 false，不阻断购买流程
+	// 查询卡类型（行业卡永远直接激活，不等实名）
+	// 同时查询载体当前实名状态：已实名则跳过 pending，直接激活
+	var cardCategory string
+	var currentlyRealnamed bool
+	if carrierType == "iot_card" {
+		var card model.IotCard
+		if err := tx.Select("card_category", "real_name_status").First(&card, carrierID).Error; err == nil {
+			cardCategory = card.CardCategory
+			currentlyRealnamed = card.RealNameStatus == constants.RealNameStatusVerified
 		}
-
-		// 判断是否 C 端购买（C 端购买前已做实名前置检查，直接激活）
-		isCEndPurchase := order.BuyerType == model.BuyerTypePersonal
-
-		if cardCategory != "industry" && !isCEndPurchase {
-			// 后台囤货路径：按 expiry_base 决定是否等实名
-			if pkg.ExpiryBase != "from_purchase" {
-				if currentlyRealnamed {
-					s.logger.Info("购买时载体已实名，直接激活套餐",
-						zap.String("carrier_type", carrierType),
-						zap.Uint("carrier_id", carrierID))
-				} else {
-					// from_activation（默认）：未实名，等实名后激活
-					status = constants.PackageUsageStatusPending
-					pendingRealnameActivation = true
-					activatedAt = time.Time{}
-					expiresAt = time.Time{}
-					nextResetAt = nil
-				}
-			}
-			// from_purchase：立即激活，status 已为 Active，保持不变
+		// err != nil 时 currentlyRealnamed 保守默认 false，不阻断购买流程
+	} else if carrierType == "device" {
+		// 子查询：统计该设备绑定的已实名卡数量（GORM 自动注入 deleted_at IS NULL）
+		var count int64
+		subQuery := tx.Model(&model.DeviceSimBinding{}).
+			Select("iot_card_id").
+			Where("device_id = ? AND bind_status = 1", carrierID)
+		if err := tx.Model(&model.IotCard{}).
+			Where("id IN (?) AND real_name_status = 1", subQuery).
+			Count(&count).Error; err == nil {
+			currentlyRealnamed = count > 0
 		}
-		// 行业卡或 C 端购买：直接激活，status 已为 Active，保持不变
+		// err != nil 时保守默认 false，不阻断购买流程
 	}
 
+	// 判断是否 C 端购买（C 端购买前已做实名前置检查，直接激活）
+	isCEndPurchase := order.BuyerType == model.BuyerTypePersonal
+
+	if cardCategory != "industry" && !isCEndPurchase {
+		// 后台囤货路径：按 expiry_base 决定是否等实名
+		if pkg.ExpiryBase != "from_purchase" {
+			if currentlyRealnamed {
+				s.logger.Info("购买时载体已实名，直接激活套餐",
+					zap.String("carrier_type", carrierType),
+					zap.Uint("carrier_id", carrierID))
+			} else {
+				// from_activation（默认）：未实名，等实名后激活
+				status = constants.PackageUsageStatusPending
+				pendingRealnameActivation = true
+				activatedAt = time.Time{}
+				expiresAt = time.Time{}
+				nextResetAt = nil
+			}
+		}
+		// from_purchase：立即激活，status 已为 Active，保持不变
+	}
+	// 行业卡或 C 端购买：直接激活，status 已为 Active，保持不变
+
 	// 创建套餐使用记录
 	virtualTotalMBSnapshot, displayGainRatioSnapshot, enableVirtualDataSnapshot := model.BuildPackageUsageSnapshotValues(pkg)
 	retailAmount := order.TotalAmount

internal/service/package/addon_main_package.go

@@ -39,3 +39,47 @@ func FindAttachableMainPackageForAddon(tx *gorm.DB, carrierType string, carrierI
 	}
 	return &mainPackage, nil
 }
+
+// FindBlockingMainPackageForFormal 查询阻止购买新主套餐的现有主套餐。
+// 待生效、生效中、未过期的已用完主套餐都仍占用主套餐资格；只有过期或失效后才允许再次购买主套餐。
+func FindBlockingMainPackageForFormal(tx *gorm.DB, carrierType string, carrierID uint, now time.Time) (*model.PackageUsage, error) {
+	var mainPackage model.PackageUsage
+	query := tx.Model(&model.PackageUsage{}).
+		Where("master_usage_id IS NULL").
+		Where(
+			"(status IN ? OR (status = ? AND (expires_at IS NULL OR expires_at > ?)))",
+			[]int{
+				constants.PackageUsageStatusPending,
+				constants.PackageUsageStatusActive,
+			},
+			constants.PackageUsageStatusDepleted,
+			now,
+		).
+		Order("priority ASC, created_at ASC")
+
+	switch carrierType {
+	case constants.AssetWalletResourceTypeIotCard, "card":
+		query = query.Where("iot_card_id = ?", carrierID)
+	case constants.AssetWalletResourceTypeDevice:
+		query = query.Where("device_id = ?", carrierID)
+	default:
+		return nil, gorm.ErrRecordNotFound
+	}
+
+	if err := query.First(&mainPackage).Error; err != nil {
+		return nil, err
+	}
+	return &mainPackage, nil
+}
+
+// HasBlockingMainPackageForFormal 判断载体是否已有阻止新主套餐购买的主套餐。
+func HasBlockingMainPackageForFormal(tx *gorm.DB, carrierType string, carrierID uint, now time.Time) (bool, error) {
+	_, err := FindBlockingMainPackageForFormal(tx, carrierType, carrierID, now)
+	if err == nil {
+		return true, nil
+	}
+	if err == gorm.ErrRecordNotFound {
+		return false, nil
+	}
+	return false, err
+}

internal/service/polling/monitoring_service.go

@@ -114,14 +114,10 @@ func (s *MonitoringService) GetOverview(ctx context.Context) (*OverviewStats, er
 
 // GetQueueStatuses 获取所有队列状态
 func (s *MonitoringService) GetQueueStatuses(ctx context.Context) ([]*QueueStatus, error) {
-	taskTypes := []string{
-		constants.TaskTypePollingRealname,
-		constants.TaskTypePollingCarddata,
-		constants.TaskTypePollingPackage,
-	}
+	taskTypes := pollingMonitorTaskTypes()
 
 	result := make([]*QueueStatus, 0, len(taskTypes))
-	now := time.Now().Unix()
+	now := time.Now()
 
 	for _, taskType := range taskTypes {
 		status := &QueueStatus{
@@ -142,43 +138,43 @@ func (s *MonitoringService) GetQueueStatuses(ctx context.Context) ([]*QueueStatu
 
 		if s.queueMgr != nil {
 			status.QueueSize, _ = s.queueMgr.GetTotalQueueDepth(ctx, taskType)
+			status.DueCount, _ = s.queueMgr.GetTotalDueCount(ctx, taskType, now)
+			status.AvgWaitTime, _ = s.queueMgr.GetAverageWaitTime(ctx, taskType, now, 10)
 		} else {
 			status.QueueSize, _ = s.redis.ZCard(ctx, queueKey).Result()
+			status.DueCount, _ = s.redis.ZCount(ctx, queueKey, "-inf", formatInt64(now.Unix())).Result()
+			status.AvgWaitTime = s.getLegacyAverageWaitTime(ctx, queueKey, now.Unix())
 		}
 
-		// 获取到期数量（score <= now）
-		status.DueCount, _ = s.redis.ZCount(ctx, queueKey, "-inf", formatInt64(now)).Result()
-
 		// 获取手动触发队列待处理数
 		manualKey := constants.RedisPollingManualQueueKey(taskType)
 		status.ManualPending, _ = s.redis.LLen(ctx, manualKey).Result()
 
-		// 计算平均等待时间（取最早的10个任务的平均等待时间）
-		earliest, err := s.redis.ZRangeWithScores(ctx, queueKey, 0, 9).Result()
-		if err == nil && len(earliest) > 0 {
-			var totalWait float64
-			for _, z := range earliest {
-				waitTime := float64(now) - z.Score
-				if waitTime > 0 {
-					totalWait += waitTime
-				}
-			}
-			status.AvgWaitTime = totalWait / float64(len(earliest))
-		}
-
 		result = append(result, status)
 	}
 
 	return result, nil
 }
 
+// getLegacyAverageWaitTime 从旧版非分片队列计算平均等待时间。
+func (s *MonitoringService) getLegacyAverageWaitTime(ctx context.Context, queueKey string, now int64) float64 {
+	earliest, err := s.redis.ZRangeWithScores(ctx, queueKey, 0, 9).Result()
+	if err != nil || len(earliest) == 0 {
+		return 0
+	}
+	var totalWait float64
+	for _, z := range earliest {
+		waitTime := float64(now) - z.Score
+		if waitTime > 0 {
+			totalWait += waitTime
+		}
+	}
+	return totalWait / float64(len(earliest))
+}
+
 // GetTaskStatuses 获取所有任务统计
 func (s *MonitoringService) GetTaskStatuses(ctx context.Context) ([]*TaskStats, error) {
-	taskTypes := []string{
-		constants.TaskTypePollingRealname,
-		constants.TaskTypePollingCarddata,
-		constants.TaskTypePollingPackage,
-	}
+	taskTypes := pollingMonitorTaskTypes()
 
 	result := make([]*TaskStats, 0, len(taskTypes))
 
@@ -269,11 +265,26 @@ func (s *MonitoringService) getTaskTypeName(taskType string) string {
 		return "流量检查"
 	case constants.TaskTypePollingPackage:
 		return "套餐检查"
+	case constants.TaskTypePollingProtect:
+		return "保护期检查"
+	case constants.TaskTypePollingCardStatus:
+		return "卡状态检查"
 	default:
 		return taskType
 	}
 }
 
+// pollingMonitorTaskTypes 返回后台监控需要展示的全部轮询任务类型。
+func pollingMonitorTaskTypes() []string {
+	return []string{
+		constants.TaskTypePollingRealname,
+		constants.TaskTypePollingCarddata,
+		constants.TaskTypePollingPackage,
+		constants.TaskTypePollingProtect,
+		constants.TaskTypePollingCardStatus,
+	}
+}
+
 // parseInt64 解析字符串为 int64
 func parseInt64(s string) int64 {
 	var result int64

internal/service/purchase_validation/service.go

@@ -2,8 +2,10 @@ package purchase_validation
 
 import (
 	"context"
+	"time"
 
 	"github.com/break/junhong_cmp_fiber/internal/model"
+	packagepkg "github.com/break/junhong_cmp_fiber/internal/service/package"
 	"github.com/break/junhong_cmp_fiber/internal/service/packageprice"
 	"github.com/break/junhong_cmp_fiber/internal/store/postgres"
 	"github.com/break/junhong_cmp_fiber/pkg/constants"
@@ -82,6 +84,10 @@ func (s *Service) ValidateCardPurchase(ctx context.Context, cardID uint, package
 		return nil, err
 	}
 
+	if err := s.validatePackageUsageRules(ctx, constants.AssetWalletResourceTypeIotCard, card.ID, packages); err != nil {
+		return nil, err
+	}
+
 	return &PurchaseValidationResult{
 		Card:       card,
 		Packages:   packages,
@@ -113,6 +119,10 @@ func (s *Service) ValidateDevicePurchase(ctx context.Context, deviceID uint, pac
 		return nil, err
 	}
 
+	if err := s.validatePackageUsageRules(ctx, constants.AssetWalletResourceTypeDevice, device.ID, packages); err != nil {
+		return nil, err
+	}
+
 	return &PurchaseValidationResult{
 		Device:     device,
 		Packages:   packages,
@@ -179,6 +189,61 @@ func (s *Service) validatePackages(ctx context.Context, packageIDs []uint, serie
 	return packages, totalPrice, nil
 }
 
+// validatePackageUsageRules 校验主套餐和加油包的购买前置条件。
+func (s *Service) validatePackageUsageRules(ctx context.Context, carrierType string, carrierID uint, packages []*model.Package) error {
+	formalCount := 0
+	hasAddon := false
+	for _, pkg := range packages {
+		if pkg == nil {
+			continue
+		}
+		switch pkg.PackageType {
+		case constants.PackageTypeFormal:
+			formalCount++
+		case constants.PackageTypeAddon:
+			hasAddon = true
+		}
+		if formalCount > 0 && hasAddon {
+			return errors.New(errors.CodeMixedOrderForbidden)
+		}
+	}
+
+	if formalCount > 1 {
+		return errors.New(errors.CodeMainPackageExists, "一次只能购买一个主套餐")
+	}
+
+	if formalCount == 1 {
+		hasBlockingMain, err := packagepkg.HasBlockingMainPackageForFormal(
+			s.db.WithContext(ctx),
+			carrierType,
+			carrierID,
+			time.Now(),
+		)
+		if err != nil {
+			return errors.Wrap(errors.CodeDatabaseError, err, "查询主套餐失败")
+		}
+		if hasBlockingMain {
+			return errors.New(errors.CodeMainPackageExists)
+		}
+	}
+
+	if hasAddon {
+		if _, err := packagepkg.FindAttachableMainPackageForAddon(
+			s.db.WithContext(ctx),
+			carrierType,
+			carrierID,
+			time.Now(),
+		); err != nil {
+			if err == gorm.ErrRecordNotFound {
+				return errors.New(errors.CodeNoMainPackage)
+			}
+			return errors.Wrap(errors.CodeDatabaseError, err, "查询主套餐失败")
+		}
+	}
+
+	return nil
+}
+
 // validateAgentAllocation 校验卖家代理的分配记录上架状态，并返回分配记录
 func (s *Service) validateAgentAllocation(ctx context.Context, sellerShopID, packageID uint) (*model.ShopPackageAllocation, error) {
 	allocation, err := s.packageAllocationStore.GetByShopAndPackageForSystem(ctx, sellerShopID, packageID)
@@ -252,6 +317,10 @@ func (s *Service) ValidateAdminOfflineCardPurchase(ctx context.Context, cardID u
 		return nil, err
 	}
 
+	if err := s.validatePackageUsageRules(ctx, constants.AssetWalletResourceTypeIotCard, card.ID, packages); err != nil {
+		return nil, err
+	}
+
 	return &PurchaseValidationResult{
 		Card:       card,
 		Packages:   packages,
@@ -279,6 +348,10 @@ func (s *Service) ValidateAdminOfflineDevicePurchase(ctx context.Context, device
 		return nil, err
 	}
 
+	if err := s.validatePackageUsageRules(ctx, constants.AssetWalletResourceTypeDevice, device.ID, packages); err != nil {
+		return nil, err
+	}
+
 	return &PurchaseValidationResult{
 		Device:     device,
 		Packages:   packages,

internal/task/auto_purchase.go

@@ -10,6 +10,7 @@ import (
 	"github.com/redis/go-redis/v9"
 	"go.uber.org/zap"
 	"gorm.io/gorm"
+	"gorm.io/gorm/clause"
 
 	"github.com/break/junhong_cmp_fiber/internal/model"
 	packagepkg "github.com/break/junhong_cmp_fiber/internal/service/package"
@@ -496,14 +497,17 @@ func (h *AutoPurchaseHandler) activateMainPackage(
 	now time.Time,
 ) error {
 	_ = ctx
-	var activeMainPackage model.PackageUsage
-	err := tx.Where("status = ?", constants.PackageUsageStatusActive).
-		Where("master_usage_id IS NULL").
-		Where(carrierType+"_id = ?", carrierID).
-		Order("priority ASC").
-		First(&activeMainPackage).Error
+	if err := h.lockPackageCarrier(ctx, tx, carrierType, carrierID); err != nil {
+		return err
+	}
 
-	hasActiveMain := err == nil
+	hasBlockingMain, err := packagepkg.HasBlockingMainPackageForFormal(tx.WithContext(ctx), carrierType, carrierID, now)
+	if err != nil {
+		return err
+	}
+	if hasBlockingMain {
+		return errors.New("已有主套餐，不能重复购买主套餐")
+	}
 
 	var status int
 	var priority int
@@ -512,21 +516,11 @@ func (h *AutoPurchaseHandler) activateMainPackage(
 	var nextResetAt *time.Time
 	var pendingRealnameActivation bool
 
-	if hasActiveMain {
-		status = constants.PackageUsageStatusPending
-		var maxPriority int
-		tx.Model(&model.PackageUsage{}).
-			Where(carrierType+"_id = ?", carrierID).
-			Select("COALESCE(MAX(priority), 0)").
-			Scan(&maxPriority)
-		priority = maxPriority + 1
-	} else {
-		status = constants.PackageUsageStatusActive
-		priority = 1
-		activatedAt = now
-		expiresAt = packagepkg.CalculateExpiryTime(pkg.CalendarType, activatedAt, pkg.DurationMonths, pkg.DurationDays)
-		nextResetAt = packagepkg.CalculateNextResetTime(pkg.DataResetCycle, pkg.CalendarType, now, activatedAt)
-	}
+	status = constants.PackageUsageStatusActive
+	priority = 1
+	activatedAt = now
+	expiresAt = packagepkg.CalculateExpiryTime(pkg.CalendarType, activatedAt, pkg.DurationMonths, pkg.DurationDays)
+	nextResetAt = packagepkg.CalculateNextResetTime(pkg.DataResetCycle, pkg.CalendarType, now, activatedAt)
 
 	// REALNAME-02: 自动购包属于 C 端充值触发，不需要等实名（C 端已做前置实名检查）
 	// ExpiryBase 仅影响后台囤货路径，此处无需判断
@@ -580,6 +574,19 @@ func (h *AutoPurchaseHandler) activateMainPackage(
 	}).Error
 }
 
+func (h *AutoPurchaseHandler) lockPackageCarrier(ctx context.Context, tx *gorm.DB, carrierType string, carrierID uint) error {
+	switch carrierType {
+	case constants.AssetWalletResourceTypeIotCard, "card":
+		var card model.IotCard
+		return tx.WithContext(ctx).Clauses(clause.Locking{Strength: "UPDATE"}).First(&card, carrierID).Error
+	case constants.AssetWalletResourceTypeDevice:
+		var device model.Device
+		return tx.WithContext(ctx).Clauses(clause.Locking{Strength: "UPDATE"}).First(&device, carrierID).Error
+	default:
+		return errors.New("无效的套餐载体类型")
+	}
+}
+
 func (h *AutoPurchaseHandler) activateAddonPackage(
 	ctx context.Context,
 	tx *gorm.DB,

pkg/errors/codes.go

@@ -134,6 +134,7 @@ const (
 	CodeNoMainPackage             = 1162 // 必须有主套餐才能购买加油包
 	CodeRealnameRequired          = 1163 // 设备/卡必须先完成实名认证才能购买套餐
 	CodeMixedOrderForbidden       = 1164 // 同订单不能同时购买正式套餐和加油包
+	CodeMainPackageExists         = 1165 // 已有主套餐，不能重复购买主套餐
 
 	// 微信配置相关错误 (1170-1179)
 	CodeWechatConfigNotFound         = 1170 // 微信支付配置不存在
@@ -279,6 +280,7 @@ var allErrorCodes = []int{
 	CodeNoMainPackage,
 	CodeRealnameRequired,
 	CodeMixedOrderForbidden,
+	CodeMainPackageExists,
 	CodeWechatConfigNotFound,
 	CodeWechatConfigActive,
 	CodeWechatConfigHasPendingOrders,
@@ -413,6 +415,7 @@ var errorMessages = map[int]string{
 	CodeNoMainPackage:                     "必须有主套餐才能购买加油包",
 	CodeRealnameRequired:                  "设备/卡必须先完成实名认证才能购买套餐",
 	CodeMixedOrderForbidden:               "同订单不能同时购买正式套餐和加油包",
+	CodeMainPackageExists:                 "已有主套餐，不能重复购买主套餐",
 	CodeWechatConfigNotFound:              "微信支付配置不存在",
 	CodeWechatConfigActive:                "不能删除当前生效的支付配置，请先停用",
 	CodeWechatConfigHasPendingOrders:      "该配置存在未完成的支付订单，暂时无法删除",

scripts/migration/README.md

@@ -43,7 +43,7 @@ scripts/migration/
 把两个 CSV 放到 `resources/` 下(命名为 `cards.csv` / `devices.csv`):
 
 - **cards.csv**:`iccid_19`/`iccid_20` 至少填一个;仅填 `iccid_20` 时脚本自动派生前 19 位,但奇成主表只按 20 位精确查询;`is_industry` 可选
-- **devices.csv**:`imei` / `virtual_no` 至少填一个;`device_name` / `device_model` / `device_type` / `max_sim_slots` 可选;`sim_iccid_*` 引用 `cards.csv` 中的 ICCID(可填 19 或 20 位)
+- **devices.csv**:`imei` / `virtual_no` 至少填一个;`device_name` / `device_model` / `device_type` / `max_sim_slots` 可选;`sim_iccid_*` 引用 `cards.csv` 中的 ICCID(可填 19 或 20 位;若 19 位前缀对应多张 20 位卡,必须填完整 20 位)
 
 字段说明详见 [`resources/README.md`](resources/README.md)。
 
@@ -227,7 +227,7 @@ real_used_mb = total_bytes_cnt / (1 + flow_add_discount / 100)
 | `errors.csv` 出现 `carrier_not_in_mapping` | `mapping.yaml.carriers` 没覆盖到这个 account_id/account_name,重跑 `scan_legacy.py` 自动补 |
 | `errors.csv` 出现 `carrier_target_missing` | `mapping.yaml.carriers` 里有这一项但 `target_*` 还是 null,手动填 |
 | `errors.csv` 出现 `not_in_legacy` | 该 ICCID 在奇成 `tbl_card` 查不到,业务方确认 |
-| `errors.csv` 出现 `legacy_duplicate` | 同一 `iccid_19` 在奇成命中多条 `tbl_card`,业务方人工确认是哪张物理卡 |
+| `errors.csv` 出现 `legacy_duplicate` | 同一 ICCID 在奇成命中多条 `tbl_card`,业务方人工确认是哪张物理卡 |
 | `errors.csv` 出现 `carrier_length_conflict` | 业务方填的 `iccid_20` 与奇成查到的运营商不匹配(CTCC 不该有 20 位 / 非 CTCC 必须有 20 位) |
 | `errors.csv` 出现 `iccid_20_prefix_mismatch` | `iccid_20` 前 19 位 ≠ `iccid_19`,业务方校对 |
 | `warnings.csv` 出现 `agent_no_legacy` | `mapping.agents` 配了但奇成没有 tbl_agent 记录,业务方确认 |

scripts/migration/lib/csv_loader.py

@@ -54,7 +54,7 @@ class DeviceRow:
     line_no: int
     imei: str
     virtual_no: str                 # 留空时用 imei 兜底,见 _load_devices
-    sim_iccids: dict[int, str]      # slot → iccid(只保留非空)
+    sim_iccids: dict[int, str]      # slot → 完整 ICCID(只保留非空)
     device_name: str = ""
     device_model: str = ""
     device_type: str = ""
@@ -159,7 +159,7 @@ def _load_cards(path: Path, result: LoadResult) -> None:
         if "iccid_20" not in header:
             raise ValueError(f"{path} 必须包含 iccid_20 列(20 位运营商卡填写,CTCC 留空)")
 
-        seen_iccid_19: dict[str, int] = {}
+        seen_iccid_full: dict[str, int] = {}
         for idx, raw in enumerate(reader, start=2):
             iccid_19 = _normalize_cell(raw.get("iccid_19", ""))
             iccid_20 = _normalize_cell(raw.get("iccid_20", ""))
@@ -184,13 +184,14 @@ def _load_cards(path: Path, result: LoadResult) -> None:
                 result.errors.append(ErrorRow("cards.csv", idx, "iccid_20", iccid_20, err.code, err.message))
                 continue
 
-            if iccid_19 in seen_iccid_19:
+            iccid_full = iccid_20 or iccid_19
+            if iccid_full in seen_iccid_full:
                 result.errors.append(
-                    ErrorRow("cards.csv", idx, "iccid_19", iccid_19, "duplicate",
-                             f"与第 {seen_iccid_19[iccid_19]} 行 iccid_19 重复")
+                    ErrorRow("cards.csv", idx, "iccid", iccid_full, "duplicate",
+                             f"与第 {seen_iccid_full[iccid_full]} 行 ICCID 重复")
                 )
                 continue
-            seen_iccid_19[iccid_19] = idx
+            seen_iccid_full[iccid_full] = idx
 
             is_industry = _parse_bool(raw.get("is_industry", ""))
             if is_industry is None:
@@ -292,27 +293,59 @@ def _load_devices(path: Path, result: LoadResult) -> None:
 def _cross_check(result: LoadResult) -> None:
     """校验设备与卡之间的一致性,并把 bound_device_virtual_no 回填到 CardRow。
 
-    devices.csv 里 sim_iccid_X 业务方可能填 19 位也可能填 20 位,
-    统一截取前 19 位回查 cards.iccid_19。
+    devices.csv 里 sim_iccid_X 业务方可能填 19 位也可能填 20 位。
+    20 位按完整 ICCID 精确匹配;19 位只有在 cards.csv 中唯一对应一张卡时才允许。
     """
     from . import iccid_utils
 
-    cards_by_iccid_19: dict[str, CardRow] = {c.iccid_19: c for c in result.cards}
+    cards_by_full: dict[str, CardRow] = {c.iccid_full: c for c in result.cards}
+    cards_by_i19: dict[str, list[CardRow]] = {}
+    for card in result.cards:
+        cards_by_i19.setdefault(card.iccid_19, []).append(card)
     device_card_refs: dict[str, tuple[str, int]] = {}
 
     for device in result.devices:
         for slot, raw_iccid in device.sim_iccids.items():
-            key19 = iccid_utils.prefix_19(raw_iccid)
-            if key19 not in cards_by_iccid_19:
+            resolved_card: Optional[CardRow] = None
+            if len(raw_iccid) == 20:
+                resolved_card = cards_by_full.get(raw_iccid)
+                if resolved_card is None:
+                    prefix = iccid_utils.prefix_19(raw_iccid)
+                    candidates = cards_by_i19.get(prefix, [])
+                    if len(candidates) == 1 and not candidates[0].iccid_20:
+                        resolved_card = candidates[0]
+            elif len(raw_iccid) == 19:
+                candidates = cards_by_i19.get(raw_iccid, [])
+                if len(candidates) > 1:
+                    result.errors.append(
+                        ErrorRow(
+                            "devices.csv", device.line_no, f"sim_iccid_{slot}", raw_iccid,
+                            "ambiguous_iccid_19", "该 19 位前缀对应多张 20 位卡,请填写完整 20 位 ICCID",
+                        )
+                    )
+                    continue
+                if candidates:
+                    resolved_card = candidates[0]
+            else:
                 result.errors.append(
                     ErrorRow(
                         "devices.csv", device.line_no, f"sim_iccid_{slot}", raw_iccid,
-                        "card_not_found", "引用的 ICCID(前 19 位)在 cards.csv 中不存在",
+                        "sim_iccid_length", "设备槽位 ICCID 必须是 19 或 20 位",
                     )
                 )
                 continue
-            if key19 in device_card_refs:
-                prev_vno, prev_line = device_card_refs[key19]
+
+            if resolved_card is None:
+                result.errors.append(
+                    ErrorRow(
+                        "devices.csv", device.line_no, f"sim_iccid_{slot}", raw_iccid,
+                        "card_not_found", "引用的 ICCID 在 cards.csv 中不存在",
+                    )
+                )
+                continue
+            card_key = resolved_card.iccid_full
+            if card_key in device_card_refs:
+                prev_vno, prev_line = device_card_refs[card_key]
                 result.errors.append(
                     ErrorRow(
                         "devices.csv", device.line_no, f"sim_iccid_{slot}", raw_iccid,
@@ -320,7 +353,7 @@ def _cross_check(result: LoadResult) -> None:
                     )
                 )
                 continue
-            device_card_refs[key19] = (device.virtual_no, device.line_no)
-            cards_by_iccid_19[key19].bound_device_virtual_no = device.virtual_no
-            # 标准化:slot 里的值统一改写为 iccid_19,sql_builder 直接用它查 cards / card_metas
-            device.sim_iccids[slot] = key19
+            device_card_refs[card_key] = (device.virtual_no, device.line_no)
+            resolved_card.bound_device_virtual_no = device.virtual_no
+            # 标准化:slot 里的值统一改写为完整 ICCID,sql_builder 直接用它查 cards / card_metas
+            device.sim_iccids[slot] = card_key

scripts/migration/lib/legacy_query.py

@@ -11,7 +11,7 @@ from decimal import Decimal
 from typing import Any, Iterable, Iterator, Optional
 
 
-BATCH_SIZE = 500
+BATCH_SIZE = 200
 
 
 def _normalize_iccid_pairs(iccid_pairs: Iterable[tuple]) -> list[tuple[str, str, bool]]:
@@ -67,7 +67,7 @@ class LegacyCardUsage:
     奇成没有按套餐分段的用量明细,无更好做法。
     """
 
-    iccid: str                       # 业务方视角的 iccid_19
+    iccid: str                       # 业务方视角的完整 ICCID
     virtual_used_mb: Decimal         # 奇成 total_bytes_cnt 原始值(虚 MB)
     flow_add_discount_pct: Decimal   # 当前生效套餐虚比百分比(0~100)
     has_active_package: bool         # 是否在 tbl_card_life 中找到当前生效套餐
@@ -273,28 +273,32 @@ def fetch_card_usage(conn, iccid_pairs: Iterable[tuple]) -> dict[str, LegacyCard
         iccid_pairs: (iccid_19, iccid_20, allow_iccid_19_lookup) 列表。
 
     Returns:
-        {iccid_19: LegacyCardUsage}。
+        {完整 ICCID: LegacyCardUsage}。
     """
     pairs = _normalize_iccid_pairs(iccid_pairs)
     out: dict[str, LegacyCardUsage] = {}
     if not pairs:
         return out
 
-    full_to_19: dict[str, tuple[str, int]] = {}
-    related_keys_by_19: dict[str, set[str]] = {}
+    query_to_full: dict[str, tuple[str, int]] = {}
+    related_keys_by_full: dict[str, set[str]] = {}
     for i19, i20, allow_i19_lookup in pairs:
+        card_key = i20 or i19
         # tbl_card 主表同时查优先键和兜底键,结果选择时按 rank:
         # 0=业务方完整 ICCID 精确命中,1=20 位卡在奇成主表只存 19 位时兜底。
-        full_to_19[i20 or i19] = (i19, 0)
+        query_to_full[card_key] = (card_key, 0)
         if i20 and allow_i19_lookup:
-            full_to_19[i19] = (i19, 1)
-        related_keys_by_19.setdefault(i19, set()).add(i19)
+            query_to_full[i19] = (card_key, 1)
+        if not i20 or allow_i19_lookup:
+            related_keys_by_full.setdefault(card_key, set()).add(i19)
+        else:
+            related_keys_by_full.setdefault(card_key, set())
         if i20:
-            related_keys_by_19[i19].add(i20)
-    full_set = sorted(full_to_19.keys())
+            related_keys_by_full[card_key].add(i20)
+    full_set = sorted(query_to_full.keys())
 
-    # tbl_card 主表用 iccid_mark IN 全长精确匹配;tbl_card_life 从表用
-    # LEFT(iccid_mark, 19) 兜底,兼容奇成两张表 iccid 长度不一致的脏数据。
+    # tbl_card 主表用 iccid_mark IN 全长精确匹配;从表只在主表本身是 19 位时
+    # 按前缀兜底,避免同前缀 20 位卡互相串套餐。
     sql = """
         SELECT
             c.iccid_mark,
@@ -314,30 +318,33 @@ def fetch_card_usage(conn, iccid_pairs: Iterable[tuple]) -> dict[str, LegacyCard
               ON latest.iccid_mark = cl1.iccid_mark
              AND latest.max_expire = cl1.expire_date
             WHERE cl1.status = 1
-        ) cl ON LEFT(cl.iccid_mark, 19) = LEFT(c.iccid_mark, 19)
+        ) cl ON (
+            cl.iccid_mark = c.iccid_mark
+            OR (CHAR_LENGTH(c.iccid_mark) = 19 AND LEFT(cl.iccid_mark, 19) = c.iccid_mark)
+        )
         LEFT JOIN tbl_set_meal sm ON sm.id = cl.meal_id
         WHERE c.iccid_mark IN %s
     """
     hits: dict[str, list[tuple[int, dict]]] = {}
     with conn.cursor() as cur:
         for batch in _chunks(full_set):
-            batch_i19s = {full_to_19[x][0] for x in batch}
-            related_batch = sorted({x for i19 in batch_i19s for x in related_keys_by_19[i19]})
+            batch_keys = {query_to_full[x][0] for x in batch}
+            related_batch = sorted({x for key in batch_keys for x in related_keys_by_full[key]})
             cur.execute(sql, (tuple(related_batch), tuple(batch)))
             for row in cur.fetchall():
                 legacy_iccid = row["iccid_mark"]
                 if not legacy_iccid:
                     continue
-                matched = full_to_19.get(legacy_iccid)
+                matched = query_to_full.get(legacy_iccid)
                 if matched is None:
                     continue
-                i19, rank = matched
-                hits.setdefault(i19, []).append((rank, row))
-    for i19, rows in hits.items():
+                card_key, rank = matched
+                hits.setdefault(card_key, []).append((rank, row))
+    for card_key, rows in hits.items():
         best_rank = min(rank for rank, _row in rows)
         row = next(row for rank, row in rows if rank == best_rank)
-        out[i19] = LegacyCardUsage(
-            iccid=i19,
+        out[card_key] = LegacyCardUsage(
+            iccid=card_key,
             virtual_used_mb=_to_mb_decimal(row.get("total_bytes_cnt")),
             flow_add_discount_pct=_to_mb_decimal(row.get("flow_add_discount")),
             has_active_package=bool(row.get("has_active_package")),
@@ -388,31 +395,35 @@ def fetch_card_meta(
         - tbl_card 主表同时查优先键和显式允许的兜底键:有 iccid_20 时优先取
           20 位精确命中;只有业务方也显式填了 iccid_19 时,才允许退回 19 位。
           只填 iccid_20 的固定 20 位卡绝不拿 19 位前缀查主表。
-        - tbl_card_life / tbl_virtual_number 从表用 LEFT(iccid_mark, 19) 兜底,
-          应对奇成两张表 iccid 长度不一致的脏数据
+        - tbl_card_life / tbl_virtual_number 从表在主表本身是 19 位时才用
+          LEFT(iccid_mark, 19) 兜底,避免同前缀 20 位卡互相串数据
 
     Args:
         iccid_pairs: (iccid_19, iccid_20, allow_iccid_19_lookup) 列表。
 
     Returns:
-        (metas, duplicate_iccid_19):
-            - metas: {iccid_19: LegacyCardMeta} 单匹结果
-            - duplicate_iccid_19: 一份业务方 iccid_19 命中奇成 tbl_card 多条记录的 set,
+        (metas, duplicate_iccids):
+            - metas: {完整 ICCID: LegacyCardMeta} 单匹结果
+            - duplicate_iccids: 一份业务方完整 ICCID 命中奇成 tbl_card 多条记录的 set,
               这些卡不进 metas,sql_builder 会写 errors.csv 阻断
     """
     pairs = _normalize_iccid_pairs(iccid_pairs)
     if not pairs:
         return {}, set()
 
-    by_full: dict[str, tuple[str, int]] = {}     # tbl_card 查询键 → (iccid_19, rank)
-    related_keys_by_19: dict[str, set[str]] = {}
+    by_full: dict[str, tuple[str, int]] = {}     # tbl_card 查询键 → (完整 ICCID, rank)
+    related_keys_by_full: dict[str, set[str]] = {}
     for i19, i20, allow_i19_lookup in pairs:
-        by_full[i20 or i19] = (i19, 0)
+        card_key = i20 or i19
+        by_full[card_key] = (card_key, 0)
         if i20 and allow_i19_lookup:
-            by_full[i19] = (i19, 1)
-        related_keys_by_19.setdefault(i19, set()).add(i19)
+            by_full[i19] = (card_key, 1)
+        if not i20 or allow_i19_lookup:
+            related_keys_by_full.setdefault(card_key, set()).add(i19)
+        else:
+            related_keys_by_full.setdefault(card_key, set())
         if i20:
-            related_keys_by_19[i19].add(i20)
+            related_keys_by_full[card_key].add(i20)
     full_set = sorted(by_full.keys())
 
     sql = """
@@ -432,7 +443,10 @@ def fetch_card_meta(
             cl.expire_date             AS expire_date
         FROM tbl_card c
         LEFT JOIN tbl_vendor_category vc ON vc.category = c.category
-        LEFT JOIN tbl_virtual_number vn ON LEFT(vn.iccid_mark, 19) = LEFT(c.iccid_mark, 19)
+        LEFT JOIN tbl_virtual_number vn ON (
+            vn.iccid_mark = c.iccid_mark
+            OR (CHAR_LENGTH(c.iccid_mark) = 19 AND LEFT(vn.iccid_mark, 19) = c.iccid_mark)
+        )
         LEFT JOIN (
             SELECT cl1.iccid_mark, cl1.meal_id, cl1.meal_name, cl1.expire_date
             FROM tbl_card_life cl1
@@ -445,16 +459,19 @@ def fetch_card_meta(
               ON latest.iccid_mark = cl1.iccid_mark
              AND latest.max_expire = cl1.expire_date
             WHERE cl1.status = 1
-        ) cl ON LEFT(cl.iccid_mark, 19) = LEFT(c.iccid_mark, 19)
+        ) cl ON (
+            cl.iccid_mark = c.iccid_mark
+            OR (CHAR_LENGTH(c.iccid_mark) = 19 AND LEFT(cl.iccid_mark, 19) = c.iccid_mark)
+        )
         WHERE c.iccid_mark IN %s
     """
 
-    # 收集每个 iccid_19 命中的奇成行(可能多条 = 脏数据)
+    # 收集每个完整 ICCID 命中的奇成行(可能多条 = 脏数据)
     hits: dict[str, list[tuple[int, dict]]] = {}
     with conn.cursor() as cur:
         for batch in _chunks(full_set):
-            batch_i19s = {by_full[x][0] for x in batch}
-            related_batch = sorted({x for i19 in batch_i19s for x in related_keys_by_19[i19]})
+            batch_keys = {by_full[x][0] for x in batch}
+            related_batch = sorted({x for key in batch_keys for x in related_keys_by_full[key]})
             cur.execute(sql, (tuple(related_batch), tuple(batch)))
             for row in cur.fetchall():
                 legacy_iccid = row["iccid"]
@@ -463,12 +480,12 @@ def fetch_card_meta(
                 matched = by_full.get(legacy_iccid)
                 if matched is None:
                     continue
-                i19, rank = matched
-                hits.setdefault(i19, []).append((rank, row))
+                card_key, rank = matched
+                hits.setdefault(card_key, []).append((rank, row))
 
     metas: dict[str, LegacyCardMeta] = {}
     duplicates: set[str] = set()
-    for i19, rows in hits.items():
+    for card_key, rows in hits.items():
         best_rank = min(rank for rank, _row in rows)
         best_rows = [row for rank, row in rows if rank == best_rank]
         # 同一张 tbl_card 可能因 card_life / virtual_number 前缀兜底 JOIN 出多行,
@@ -477,12 +494,12 @@ def fetch_card_meta(
         for row in best_rows:
             rows_by_card_id[str(row.get("card_id") or row.get("iccid") or "")] = row
         if len(rows_by_card_id) > 1:
-            duplicates.add(i19)
+            duplicates.add(card_key)
             continue
         row = next(iter(rows_by_card_id.values()))
         expire = row.get("expire_date")
         expire_str = expire.strftime("%Y-%m-%d %H:%M:%S") if expire else None
-        metas[i19] = LegacyCardMeta(
+        metas[card_key] = LegacyCardMeta(
             legacy_raw_iccid=row["iccid"],
             account_id=row.get("account_id") or "",
             account_name=row.get("account_name") or "",

scripts/migration/lib/sql_builder.py

@@ -144,8 +144,8 @@ def _device_package_source_iccid(device: DeviceRow) -> str:
     )
 
 
-def _target_series_id_expr_for_iccid_19(
-    iccid_19: str,
+def _target_series_id_expr_for_iccid(
+    iccid: str,
     card_metas: dict[str, LegacyCardMeta],
     mapping: Mapping,
 ) -> str:
@@ -153,9 +153,9 @@ def _target_series_id_expr_for_iccid_19(
 
     系列归属以新系统目标套餐为准，避免奇成套餐系列与新库系列二次映射不一致。
     """
-    if not iccid_19:
+    if not iccid:
         return "NULL"
-    meta = card_metas.get(iccid_19)
+    meta = card_metas.get(iccid)
     if meta is None or not meta.current_meal_id:
         return "NULL"
     pkg_map = mapping.lookup_package(meta.current_meal_id)
@@ -200,7 +200,7 @@ def _runtime_asset_for_card(card: CardRow, devices_by_virtual_no: dict[str, Devi
     device = devices_by_virtual_no.get(card.bound_device_virtual_no)
     if device is None:
         return None
-    if card.iccid_19 != _device_package_source_iccid(device):
+    if card.iccid_full != _device_package_source_iccid(device):
         return None
     return "device", device.virtual_no
 
@@ -217,7 +217,7 @@ def write_step1(
     devices: list[DeviceRow],
     mapping: Mapping,
     card_metas: dict[str, LegacyCardMeta],
-    duplicate_iccid_19s: set[str],
+    duplicate_iccids: set[str],
     errors: list[ErrorRow],
 ) -> None:
     """生成 step1_* SQL + errors.csv + summary.txt。
@@ -225,21 +225,21 @@ def write_step1(
     会就地修改 errors 列表,把映射缺失等问题追加进去。
 
     Args:
-        card_metas: {iccid_19: LegacyCardMeta}
-        duplicate_iccid_19s: 同一 iccid_19 在奇成命中多条 tbl_card 记录的集合,直接报错阻断
+        card_metas: {完整 ICCID: LegacyCardMeta}
+        duplicate_iccids: 同一完整 ICCID 在奇成命中多条 tbl_card 记录的集合,直接报错阻断
     """
     user_id = mapping.migration_user_id
 
     # 预检 carrier 映射 + 运营商一致性 + 重复命中,把异常卡剔除
-    valid_cards = _filter_cards_by_carrier(cards, card_metas, duplicate_iccid_19s, mapping, errors)
+    valid_cards = _filter_cards_by_carrier(cards, card_metas, duplicate_iccids, mapping, errors)
 
-    cards_by_iccid_19 = {c.iccid_19: c for c in valid_cards}
+    cards_by_iccid = {c.iccid_full: c for c in valid_cards}
 
     _write_iot_cards(output_dir / "step1_01_iot_cards.sql", valid_cards, devices, card_metas, mapping, user_id)
     _write_devices(output_dir / "step1_02_devices.sql", devices, card_metas, mapping, user_id)
     _write_asset_identifiers(output_dir / "step1_03_asset_identifiers.sql", valid_cards, devices, card_metas)
     _write_asset_wallets(output_dir / "step1_04_asset_wallets.sql", valid_cards, devices)
-    _write_sim_bindings(output_dir / "step1_05_sim_bindings.sql", devices, cards_by_iccid_19)
+    _write_sim_bindings(output_dir / "step1_05_sim_bindings.sql", devices, cards_by_iccid)
     _write_shop_allocations(
         output_dir / "step1_06_shop_allocations.sql", valid_cards, devices, card_metas, mapping, user_id
     )
@@ -251,29 +251,29 @@ def write_step1(
 def _filter_cards_by_carrier(
     cards: list[CardRow],
     card_metas: dict[str, LegacyCardMeta],
-    duplicate_iccid_19s: set[str],
+    duplicate_iccids: set[str],
     mapping: Mapping,
     errors: list[ErrorRow],
 ) -> list[CardRow]:
     """筛掉:奇成查不到 / 命中多条 / carrier 映射缺失 / target_* 未填 / 运营商与 iccid 长度不一致 的卡。"""
     valid: list[CardRow] = []
     for c in cards:
-        if c.iccid_19 in duplicate_iccid_19s:
+        if c.iccid_full in duplicate_iccids:
             errors.append(ErrorRow(
-                "cards.csv", c.line_no, "iccid_19", c.iccid_19,
-                "legacy_duplicate", "奇成 tbl_card 中同一 iccid_19 命中多条记录,请业务方人工确认",
+                "cards.csv", c.line_no, "iccid", c.iccid_full,
+                "legacy_duplicate", "奇成 tbl_card 中同一 ICCID 命中多条记录,请业务方人工确认",
             ))
             continue
-        meta = card_metas.get(c.iccid_19)
+        meta = card_metas.get(c.iccid_full)
         if meta is None:
             errors.append(ErrorRow(
-                "cards.csv", c.line_no, "iccid_19", c.iccid_19,
+                "cards.csv", c.line_no, "iccid", c.iccid_full,
                 "not_in_legacy", "奇成 tbl_card 中找不到该 ICCID(iccid_19/iccid_20 都没命中)",
             ))
             continue
         if not meta.account_id:
             errors.append(ErrorRow(
-                "cards.csv", c.line_no, "iccid_19", c.iccid_19,
+                "cards.csv", c.line_no, "iccid", c.iccid_full,
                 "no_carrier_account", "奇成 tbl_card.account_id 为空,无法定位具体运营商账号映射",
             ))
             continue
@@ -320,7 +320,7 @@ def _write_iot_cards(
     with path.open("w", encoding="utf-8") as f:
         f.write(_file_header("step1_01 卡资产导入 (tb_iot_card)"))
         for c in cards:
-            meta = card_metas[c.iccid_19]
+            meta = card_metas[c.iccid_full]
             carrier = mapping.lookup_carrier(meta.account_id)
             assert carrier is not None  # 已被 _filter_cards_by_carrier 保证
 
@@ -332,7 +332,7 @@ def _write_iot_cards(
             if c.bound_device_virtual_no:
                 device = devices_by_virtual_no.get(c.bound_device_virtual_no)
                 shop_code = _resolve_device_shop_code(device, card_metas, mapping) if device else ""
-            series_id_expr = _target_series_id_expr_for_iccid_19(c.iccid_19, card_metas, mapping)
+            series_id_expr = _target_series_id_expr_for_iccid(c.iccid_full, card_metas, mapping)
             card_virtual_no = _card_virtual_no_for_import(meta, device_virtual_nos)
             is_standalone = "FALSE" if c.bound_device_virtual_no else "TRUE"
             device_virtual_no = c.bound_device_virtual_no or ""
@@ -408,7 +408,7 @@ def _write_devices(
             device_status = 2 if shop_code else 1
             shop_id_expr = _shop_id_sub(shop_code)
             source_iccid = _device_package_source_iccid(d)
-            series_id_expr = _target_series_id_expr_for_iccid_19(source_iccid, card_metas, mapping)
+            series_id_expr = _target_series_id_expr_for_iccid(source_iccid, card_metas, mapping)
             f.write(
                 "INSERT INTO tb_device (\n"
                 "    virtual_no, max_sim_slots, imei,\n"
@@ -480,7 +480,7 @@ def _write_asset_identifiers(
     with path.open("w", encoding="utf-8") as f:
         f.write(_file_header("step1_03 资产标识符注册 (tb_asset_identifier)"))
         for c in cards:
-            meta = card_metas[c.iccid_19]
+            meta = card_metas[c.iccid_full]
             card_virtual_no = _card_virtual_no_for_import(meta, device_virtual_nos)
             f.write(
                 "INSERT INTO tb_asset_identifier (identifier, asset_type, asset_id)\n"
@@ -563,13 +563,13 @@ def _write_asset_wallets(path: Path, cards: list[CardRow], devices: list[DeviceR
 def _write_sim_bindings(
     path: Path,
     devices: list[DeviceRow],
-    cards_by_iccid_19: dict[str, CardRow],
+    cards_by_iccid: dict[str, CardRow],
 ) -> None:
     with path.open("w", encoding="utf-8") as f:
         f.write(_file_header("step1_05 设备-卡绑定 (tb_device_sim_binding)"))
         for d in devices:
-            for slot, iccid_19 in sorted(d.sim_iccids.items()):
-                card = cards_by_iccid_19.get(iccid_19)
+            for slot, iccid in sorted(d.sim_iccids.items()):
+                card = cards_by_iccid.get(iccid)
                 if card is None:
                     continue  # 该卡可能被 carrier 校验剔除,跳过此 slot 绑定
                 is_current = "TRUE" if slot == d.current_slot else "FALSE"
@@ -632,7 +632,7 @@ def _write_shop_allocations(
         for c in cards:
             if c.bound_device_virtual_no:
                 continue
-            meta = card_metas[c.iccid_19]
+            meta = card_metas[c.iccid_full]
             shop_code = _resolve_card_shop_code(c, meta, mapping)
             if not shop_code:
                 continue
@@ -685,6 +685,11 @@ def _write_errors(path: Path, errors: list[ErrorRow]) -> None:
             writer.writerow(e.to_csv_row())
 
 
+def write_runtime_input_errors(output_dir: Path, errors: list[ErrorRow]) -> None:
+    """写运行态迁移的输入错误。"""
+    _write_errors(output_dir / "errors.csv", errors)
+
+
 def _write_summary_step1(path: Path, cards: list[CardRow], devices: list[DeviceRow], errors: list[ErrorRow]) -> None:
     bound = sum(1 for c in cards if c.bound_device_virtual_no)
     lines = [
@@ -723,32 +728,32 @@ def write_step2(
     warnings: list[tuple[str, str, str]] = []
     devices_by_virtual_no = _device_by_virtual_no(devices)
     # 卡 → package(从 card_meta.current_meal_id 经 mapping.packages 查 target)
-    resolved_packages: dict[str, tuple[int, str]] = {}  # iccid_19 → (target_package_id, expire_date)
+    resolved_packages: dict[str, tuple[int, str]] = {}  # 完整 ICCID → (target_package_id, expire_date)
     for c in cards:
-        meta = card_metas.get(c.iccid_19)
+        meta = card_metas.get(c.iccid_full)
         if meta is None or not meta.current_meal_id:
             continue
         pkg_map = mapping.lookup_package(meta.current_meal_id)
         if pkg_map is None:
-            warnings.append(("package_not_in_mapping", c.iccid_19,
+            warnings.append(("package_not_in_mapping", c.iccid_full,
                              f"奇成 meal_id={meta.current_meal_id} ({meta.current_meal_name}) 未在 mapping.yaml.packages"))
             continue
         try:
             pkg_map.require_target()
         except MissingTargetError as exc:
-            warnings.append(("package_target_missing", c.iccid_19, str(exc)))
+            warnings.append(("package_target_missing", c.iccid_full, str(exc)))
             continue
-        resolved_packages[c.iccid_19] = (pkg_map.target_package_id, meta.current_expire_date or "")
+        resolved_packages[c.iccid_full] = (pkg_map.target_package_id, meta.current_expire_date or "")
 
     # 用量预检:虚用量 > 0 但没找到当前生效套餐的卡,无法精确反算真用量
     # (按 flow_add_discount=0 处理,即真=虚,可能高估真用量;不阻断)
     for c in cards:
-        snap = usage_snapshots.get(c.iccid_19)
+        snap = usage_snapshots.get(c.iccid_full)
         if snap is None or snap.virtual_used_mb <= 0:
             continue
         if not snap.has_active_package:
             warnings.append((
-                "usage_without_active_package", c.iccid_19,
+                "usage_without_active_package", c.iccid_full,
                 f"奇成虚用量 {snap.virtual_used_mb}MB,但 tbl_card_life 中没有当前生效套餐"
                 "(status=1),无法精确反算真用量,按 flow_add_discount=0 处理(真=虚)",
             ))
@@ -805,7 +810,7 @@ def _write_migration_orders(
     with path.open("w", encoding="utf-8") as f:
         f.write(_file_header("step2_01 迁移伪订单 (tb_order)"))
         for c in cards:
-            if c.iccid_19 not in resolved_packages:
+            if c.iccid_full not in resolved_packages:
                 continue
             asset = _runtime_asset_for_card(c, devices_by_virtual_no)
             if asset is None:
@@ -873,16 +878,16 @@ def _write_package_usages(
     with path.open("w", encoding="utf-8") as f:
         f.write(_file_header("step2_02 套餐使用记录 (tb_package_usage)"))
         for c in cards:
-            if c.iccid_19 not in resolved_packages:
+            if c.iccid_full not in resolved_packages:
                 continue
             asset = _runtime_asset_for_card(c, devices_by_virtual_no)
             if asset is None:
                 continue
             asset_type, _asset_identifier = asset
-            target_pkg_id, expire = resolved_packages[c.iccid_19]
+            target_pkg_id, expire = resolved_packages[c.iccid_full]
             order_no = _make_card_order_no(c.iccid_full)
             expires_expr = f"{_sql_str(expire)}::timestamp" if expire else "NULL"
-            snap = usage_snapshots.get(c.iccid_19)
+            snap = usage_snapshots.get(c.iccid_full)
             used_mb = snap.real_used_mb_floor if snap is not None else 0
             effective_limit_expr = (
                 "CASE WHEN p.enable_virtual_data AND p.virtual_data_mb > 0 "
@@ -935,7 +940,7 @@ def _write_card_data_usage(path: Path, cards: list[CardRow], usage_snapshots: di
     with path.open("w", encoding="utf-8") as f:
         f.write(_file_header("step2_03 卡累计流量更新 (tb_iot_card 流量基线)"))
         for c in cards:
-            snap = usage_snapshots.get(c.iccid_19)
+            snap = usage_snapshots.get(c.iccid_full)
             if snap is None or snap.real_used_mb <= 0:
                 continue
             real_decimal = _sql_decimal(snap.real_used_mb)
@@ -967,9 +972,9 @@ def _write_asset_series_update(
     with path.open("w", encoding="utf-8") as f:
         f.write(_file_header("step2_06 资产套餐系列回填 (tb_iot_card / tb_device)"))
         for c in cards:
-            if c.iccid_19 not in resolved_packages:
+            if c.iccid_full not in resolved_packages:
                 continue
-            target_pkg_id, _expire = resolved_packages[c.iccid_19]
+            target_pkg_id, _expire = resolved_packages[c.iccid_full]
             f.write(
                 "UPDATE tb_iot_card c\n"
                 "SET series_id = p.series_id,\n"
@@ -1124,7 +1129,7 @@ def _write_summary_step2(path: Path, *, cards, resolved, devices_by_virtual_no,
     generated_count = sum(
         1
         for c in cards
-        if c.iccid_19 in resolved and _runtime_asset_for_card(c, devices_by_virtual_no) is not None
+        if c.iccid_full in resolved and _runtime_asset_for_card(c, devices_by_virtual_no) is not None
     )
     lines = [
         "奇成迁移 - 脚本2 关联数据导入摘要",

scripts/migration/migrate_assets.py

@@ -46,7 +46,7 @@ def main() -> int:
     iccid_pairs = [(c.iccid_19, c.iccid_20, c.allow_iccid_19_lookup) for c in cards]
     dsn = mapping_loader.load_legacy_dsn(config_dir)
     with legacy_query.connect_readonly(dsn) as conn:
-        card_metas, dup_iccid_19 = legacy_query.fetch_card_meta(conn, iccid_pairs)
+        card_metas, dup_iccids = legacy_query.fetch_card_meta(conn, iccid_pairs)
 
     sql_builder.write_step1(
         output_dir,
@@ -54,7 +54,7 @@ def main() -> int:
         devices=devices,
         mapping=mapping,
         card_metas=card_metas,
-        duplicate_iccid_19s=dup_iccid_19,
+        duplicate_iccids=dup_iccids,
         errors=errors,
     )
 

scripts/migration/migrate_runtime.py

@@ -37,14 +37,18 @@ def main() -> int:
     output_dir.mkdir(parents=True, exist_ok=True)
 
     mapping = mapping_loader.load_mapping(config_dir)
-    cards, devices, _errors = csv_loader.load_assets(resources_dir)
+    cards, devices, errors = csv_loader.load_assets(resources_dir)
+    if errors:
+        sql_builder.write_runtime_input_errors(output_dir, errors)
+        print(f"cards.csv/devices.csv 存在 {len(errors)} 个基础错误,已写入 {output_dir / 'errors.csv'}", file=sys.stderr)
+        return 1
 
     iccid_pairs = [(c.iccid_19, c.iccid_20, c.allow_iccid_19_lookup) for c in cards]
     agent_ids = [a.legacy_agent_id for a in mapping.agents.values() if (a.target_shop_code or "").strip()]
 
     dsn = mapping_loader.load_legacy_dsn(config_dir)
     with legacy_query.connect_readonly(dsn) as conn:
-        card_metas, _dup_iccid_19 = legacy_query.fetch_card_meta(conn, iccid_pairs)
+        card_metas, _dup_iccids = legacy_query.fetch_card_meta(conn, iccid_pairs)
         usage_snapshots = legacy_query.fetch_card_usage(conn, iccid_pairs)
         commission_snapshots = legacy_query.fetch_commission_accounts(conn, agent_ids) if agent_ids else {}
         agent_balances = legacy_query.fetch_agent_balances(conn, agent_ids) if agent_ids else {}

scripts/migration/resources/README.md

@@ -8,13 +8,13 @@
 ## cards.csv 卡资产清单
 
 每张要迁移的卡一行。`iccid_19` / `iccid_20` 至少填一个。只有 20 位完整 ICCID
-时可只填 `iccid_20`,脚本会自动派生前 19 位作为内部稳定 ID。
+时可只填 `iccid_20`,脚本会自动派生前 19 位写入 `iccid_19` 字段。
 只填 `iccid_20` 的卡,脚本查询奇成主表时只按 20 位精确匹配;只有同时显式填写
 `iccid_19` 和 `iccid_20` 时,才允许 20 位未命中后退回 19 位查询。
 
 | 列名 | 必填 | 说明 |
 |------|------|------|
-| `iccid_19` | 条件必填 | ICCID 前 19 位(内部稳定 ID;必须 19 位;允许奇成历史字母位)。`iccid_20` 已填时可留空 |
+| `iccid_19` | 条件必填 | ICCID 前 19 位(必须 19 位;允许奇成历史字母位)。`iccid_20` 已填时可留空 |
 | `iccid_20` | 视运营商 | 完整 20 位 ICCID;CMCC/CUCC/CBN(20 位运营商) **必填**,CTCC(19 位)**必须留空**;非空时前 19 位必须等于 `iccid_19` |
 | `is_industry` | 否 | 是否行业卡,留空 = `false`(普通卡) |
 
@@ -22,7 +22,7 @@
 `tbl_card.category` 只是奇成大类,仅辅助人工填写 `target_carrier_type`。填完映射后会做一致性校验:
 - CTCC 卡填了 `iccid_20` → `errors.csv: carrier_length_conflict` 阻断
 - 非 CTCC 卡 `iccid_20` 留空 → 同上阻断
-- `iccid_19` 在奇成 `tbl_card` 命中多条记录 → `errors.csv: legacy_duplicate` 阻断,请业务方人工确认
+- 同一完整 ICCID 在奇成 `tbl_card` 命中多条记录 → `errors.csv: legacy_duplicate` 阻断,请业务方人工确认
 
 **店铺归属不在 CSV 里填**,由 `mapping.yaml.agents[].target_shop_code` 通过奇成
 `tbl_card.agent_id` 自动推导。如果该卡在奇成没有 agent_id 或代理映射未填 shop_code,
@@ -40,7 +40,7 @@
 | `device_model` | 否 | 设备型号,写入 `tb_device.device_model` |
 | `device_type` | 否 | 设备类型,写入 `tb_device.device_type` |
 | `max_sim_slots` | 否 | 最大卡槽数,范围 1-4。不填时按非空 `sim_iccid_*` 的最大槽位推导 |
-| `sim_iccid_1` ~ `sim_iccid_4` | 否 | 各插槽 ICCID,引用 `cards.csv` 的卡。可填 19 或 20 位,内部统一按前 19 位回查 `cards.iccid_19`(至少要有 1 个) |
+| `sim_iccid_1` ~ `sim_iccid_4` | 否 | 各插槽 ICCID,引用 `cards.csv` 的卡。可填 19 或 20 位;20 位精确匹配完整 ICCID,19 位仅在该前缀唯一对应一张卡时允许(至少要有 1 个) |
 
 **店铺归属规则**:用 `sim_iccid_1`(主卡)的 agent_id 走 `mapping.agents`
 推导,与卡归属同口径。
@@ -52,5 +52,6 @@
 
 - `devices.csv` 任何 `sim_iccid_*` 引用的 ICCID **必须**在 `cards.csv` 中存在
 - 同一 ICCID 不允许在 `cards.csv` 中重复出现
+- 19 位前缀对应多张 20 位卡时,`devices.csv` 必须填写完整 20 位 ICCID
 - `imei` 和 `virtual_no` 至少要有一个(用作设备唯一标识)
 - 同一 ICCID 不能同时被多个设备/插槽引用

scripts/migration/scan_legacy.py

@@ -39,7 +39,7 @@ def _read_iccid_pairs(cards_csv: Path) -> list[tuple[str, str, bool]]:
     """
     if not cards_csv.exists():
         raise FileNotFoundError(f"找不到 {cards_csv},请按 resources/README.md 准备")
-    pairs: dict[str, tuple[str, bool]] = {}
+    pairs: dict[str, tuple[str, str, bool]] = {}
     first_seen_line: dict[str, int] = {}
     errors: list[str] = []
     with cards_csv.open("r", encoding="utf-8-sig", newline="") as f:
@@ -69,17 +69,19 @@ def _read_iccid_pairs(cards_csv: Path) -> list[tuple[str, str, bool]]:
             if i20 and i20[:19] != i19:
                 errors.append(f"第 {line_no} 行 iccid_20 前 19 位必须等于 iccid_19: {i20!r}")
                 continue
-            if i19 in pairs:
-                errors.append(f"第 {line_no} 行 iccid_19 与第 {first_seen_line[i19]} 行重复: {i19!r}")
+            # 有 iccid_20 的卡用 iccid_20 做唯一键，避免同批次卡派生出相同 iccid_19 误报重复
+            dedup_key = i20 if i20 else i19
+            if dedup_key in pairs:
+                errors.append(f"第 {line_no} 行 {'iccid_20' if i20 else 'iccid_19'} 与第 {first_seen_line[dedup_key]} 行重复: {dedup_key!r}")
                 continue
-            first_seen_line[i19] = line_no
-            pairs[i19] = (i20, allow_i19_lookup)
+            first_seen_line[dedup_key] = line_no
+            pairs[dedup_key] = (i19, i20, allow_i19_lookup)
     if errors:
         preview = "\n".join(f"  - {msg}" for msg in errors[:20])
         if len(errors) > 20:
             preview += f"\n  - 其余 {len(errors) - 20} 个错误已省略"
         raise ValueError(f"{cards_csv} 存在无效 ICCID,请先修正后再扫描:\n{preview}")
-    return [(i19, i20, allow_i19_lookup) for i19, (i20, allow_i19_lookup) in sorted(pairs.items())]
+    return [v for _, v in sorted(pairs.items())]
 
 
 def main() -> int:
@@ -104,13 +106,13 @@ def main() -> int:
 
     dsn = mapping_loader.load_legacy_dsn(config_dir)
     with legacy_query.connect_readonly(dsn) as conn:
-        card_metas, dup_iccid_19 = legacy_query.fetch_card_meta(conn, iccid_pairs)
+        card_metas, dup_iccids = legacy_query.fetch_card_meta(conn, iccid_pairs)
         meal_ids = {m.current_meal_id for m in card_metas.values() if m.current_meal_id}
         meal_metas = legacy_query.fetch_meal_meta(conn, meal_ids)
 
-    if dup_iccid_19:
-        print(f"警告:{len(dup_iccid_19)} 张卡的 iccid_19 在奇成 tbl_card 命中多条记录:")
-        for x in sorted(dup_iccid_19):
+    if dup_iccids:
+        print(f"警告:{len(dup_iccids)} 张卡的 ICCID 在奇成 tbl_card 命中多条记录:")
+        for x in sorted(dup_iccids):
             print(f"  - {x}")
         print("这些卡 migrate_assets 阶段会写 errors.csv 阻断,请业务方人工确认。")