feat: 不轮询清单与批量开关轮询

- 新增不轮询清单 tb_card_polling_pause（迁移 000256 补写存量），所有改 enable_polling 的入口同事务写清单
- 卡、设备勾选批量开关轮询（单次 500，逐项结果，开启跳过风险停机与已销户）与 CSV 文件导入
- 订单套餐生效、改有效期后自动恢复轮询并按规则评估复机
- 不轮询清单查询与导出；卡、设备列表与导出增加是否轮询
- 失败告警停机被拒视图、导出与批量关闭轮询；开关轮询影响说明

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>

internal/task/polling_status_import.go

@@ -0,0 +1,195 @@
+package task
+
+import (
+	"bytes"
+	"context"
+	"encoding/csv"
+	"encoding/json"
+	stderrors "errors"
+	"io"
+	"os"
+	"strings"
+	"time"
+
+	"github.com/hibiken/asynq"
+	"go.uber.org/zap"
+	"gorm.io/gorm"
+
+	"github.com/break/junhong_cmp_fiber/internal/infrastructure/pollingpause"
+	"github.com/break/junhong_cmp_fiber/internal/model"
+	"github.com/break/junhong_cmp_fiber/internal/model/dto"
+	"github.com/break/junhong_cmp_fiber/pkg/constants"
+	"github.com/break/junhong_cmp_fiber/pkg/storage"
+	"github.com/break/junhong_cmp_fiber/pkg/utils"
+)
+
+// PollingToggler 改一个资产的轮询开关并返回逐项结果（由资产轮询服务实现）。
+type PollingToggler interface {
+	ToggleAsset(ctx context.Context, assetType string, assetID uint, enable, scoped bool) dto.PollingToggleItem
+}
+
+// pollingStatusImportPayload 与 API 侧 PollingStatusImportPayload 同一 JSON 契约。
+type pollingStatusImportPayload struct {
+	TaskID uint `json:"task_id"`
+}
+
+// PollingStatusImportHandler 轮询开关文件导入：逐行独立处理（规则同勾选批量），部分成功不回滚，逐行记录结果。
+type PollingStatusImportHandler struct {
+	db      *gorm.DB
+	storage *storage.Service
+	toggler PollingToggler
+	logger  *zap.Logger
+}
+
+// NewPollingStatusImportHandler 创建轮询开关导入任务处理器。
+func NewPollingStatusImportHandler(db *gorm.DB, storageService *storage.Service, toggler PollingToggler, logger *zap.Logger) *PollingStatusImportHandler {
+	if logger == nil {
+		logger = zap.NewNop()
+	}
+	return &PollingStatusImportHandler{db: db, storage: storageService, toggler: toggler, logger: logger}
+}
+
+// Handle 处理一个导入任务；任务只会从待处理被领取一次。
+func (h *PollingStatusImportHandler) Handle(ctx context.Context, t *asynq.Task) error {
+	var payload pollingStatusImportPayload
+	if err := json.Unmarshal(t.Payload(), &payload); err != nil || payload.TaskID == 0 {
+		h.logger.Warn("轮询开关导入任务载荷无效", zap.ByteString("payload", t.Payload()))
+		return nil
+	}
+	now := time.Now()
+	claimed := h.db.WithContext(ctx).Model(&model.PollingStatusImportTask{}).
+		Where("id = ? AND status = ?", payload.TaskID, model.ImportTaskStatusPending).
+		Updates(map[string]any{"status": model.ImportTaskStatusProcessing, "started_at": now, "updated_at": now})
+	if claimed.Error != nil {
+		return claimed.Error
+	}
+	if claimed.RowsAffected == 0 {
+		return nil
+	}
+	var task model.PollingStatusImportTask
+	if err := h.db.WithContext(ctx).First(&task, payload.TaskID).Error; err != nil {
+		return err
+	}
+	identifiers, err := h.downloadAndParse(ctx, task.StorageKey)
+	if err != nil {
+		h.fail(ctx, task.ID, err.Error())
+		return nil
+	}
+	rowCtx := pollingpause.WithOrigin(ctx, pollingpause.Origin{Source: constants.PollingToggleSourceImport, Reason: task.Reason,
+		ImportTaskID: task.ID, OperatorID: task.Creator})
+	results := make(model.PollingStatusImportResults, 0, len(identifiers))
+	success, skipped, failed := 0, 0, 0
+	for index, identifier := range identifiers {
+		item := h.processRow(rowCtx, task, index+1, identifier)
+		switch item.Status {
+		case constants.PollingToggleItemSuccess:
+			success++
+		case constants.PollingToggleItemSkipped:
+			skipped++
+		default:
+			failed++
+		}
+		results = append(results, item)
+	}
+	completed := time.Now()
+	if err := h.db.WithContext(context.WithoutCancel(ctx)).Model(&model.PollingStatusImportTask{}).Where("id = ?", task.ID).Updates(map[string]any{
+		"status": model.ImportTaskStatusCompleted, "total_count": len(identifiers), "success_count": success, "skip_count": skipped,
+		"fail_count": failed, "result_items": results, "completed_at": completed, "updated_at": completed,
+	}).Error; err != nil {
+		h.logger.Error("保存轮询开关导入结果失败", zap.Uint("task_id", task.ID), zap.Error(err))
+		return err
+	}
+	h.logger.Info("轮询开关导入完成", zap.Uint("task_id", task.ID), zap.Int("total", len(identifiers)), zap.Int("success", success),
+		zap.Int("skipped", skipped), zap.Int("failed", failed))
+	return nil
+}
+
+func (h *PollingStatusImportHandler) processRow(ctx context.Context, task model.PollingStatusImportTask, line int, identifier string) model.PollingStatusImportResultItem {
+	item := model.PollingStatusImportResultItem{Line: line, Identifier: identifier}
+	if identifier == "" {
+		item.Status, item.Reason = constants.PollingToggleItemFailed, "标识为空"
+		return item
+	}
+	assetID, err := h.resolve(ctx, task.AssetType, identifier)
+	if err != nil {
+		item.Status, item.Reason = constants.PollingToggleItemForbidden, "资产不存在"
+		if !stderrors.Is(err, gorm.ErrRecordNotFound) {
+			item.Status, item.Reason = constants.PollingToggleItemFailed, "查询资产失败"
+		}
+		return item
+	}
+	item.AssetID = assetID
+	result := h.toggler.ToggleAsset(ctx, task.AssetType, assetID, task.EnablePolling, false)
+	item.Status, item.Reason = result.Status, result.Reason
+	return item
+}
+
+// resolve 按 ICCID（卡）或 IMEI / 虚拟号（设备）查找资产。
+func (h *PollingStatusImportHandler) resolve(ctx context.Context, assetType, identifier string) (uint, error) {
+	if assetType == constants.AssetTypeDevice {
+		var device model.Device
+		err := h.db.WithContext(ctx).Select("id").Where("imei = ? OR virtual_no = ?", identifier, identifier).Order("id").First(&device).Error
+		return device.ID, err
+	}
+	var card model.IotCard
+	err := h.db.WithContext(ctx).Select("id").Where("iccid = ?", identifier).First(&card).Error
+	return card.ID, err
+}
+
+// downloadAndParse 下载 CSV 并取每行首列作为资产标识；首行为表头，空行忽略。
+func (h *PollingStatusImportHandler) downloadAndParse(ctx context.Context, key string) ([]string, error) {
+	if h.storage == nil {
+		return nil, stderrors.New("对象存储服务未配置")
+	}
+	localPath, cleanup, err := h.storage.DownloadToTemp(ctx, key)
+	if err != nil {
+		return nil, stderrors.New("下载导入文件失败")
+	}
+	defer cleanup()
+	data, err := os.ReadFile(localPath)
+	if err != nil {
+		return nil, stderrors.New("读取导入文件失败")
+	}
+	decoded, err := utils.DecodeTextToUTF8(data)
+	if err != nil {
+		return nil, stderrors.New("导入文件编码无法识别，请使用 UTF-8 或 GBK")
+	}
+	reader := csv.NewReader(bytes.NewReader(decoded))
+	reader.FieldsPerRecord = -1
+	reader.TrimLeadingSpace = true
+	identifiers := make([]string, 0)
+	header := true
+	for {
+		record, readErr := reader.Read()
+		if readErr == io.EOF {
+			break
+		}
+		if readErr != nil {
+			return nil, stderrors.New("导入文件不是有效的 CSV")
+		}
+		if header {
+			header = false
+			continue
+		}
+		if len(record) == 0 || strings.TrimSpace(strings.Join(record, "")) == "" {
+			continue
+		}
+		identifiers = append(identifiers, strings.TrimSpace(strings.TrimPrefix(record[0], "\uFEFF")))
+		if len(identifiers) > constants.PollingStatusImportMaxRows {
+			return nil, stderrors.New("导入文件超过 50000 行")
+		}
+	}
+	if len(identifiers) == 0 {
+		return nil, stderrors.New("导入文件没有数据行")
+	}
+	return identifiers, nil
+}
+
+func (h *PollingStatusImportHandler) fail(ctx context.Context, taskID uint, message string) {
+	now := time.Now()
+	if err := h.db.WithContext(context.WithoutCancel(ctx)).Model(&model.PollingStatusImportTask{}).Where("id = ?", taskID).Updates(map[string]any{
+		"status": model.ImportTaskStatusFailed, "error_message": message, "completed_at": now, "updated_at": now,
+	}).Error; err != nil {
+		h.logger.Error("标记轮询开关导入任务失败时出错", zap.Uint("task_id", taskID), zap.Error(err))
+	}
+}