Files
junhong_cmp_fiber/internal/task/phone_asset_unbind_import.go
break 5ed6b39deb
Some checks failed
构建并部署到测试环境(无 SSH) / build-and-deploy (push) Has been cancelled
feat(收口): 补齐 8 月迭代缺口并同步 Spec 与证据链
- 新增六对成对迁移 000232–000237:H5 弹窗类型、退款结算标识与申请人备注、优先轮询事实字段与两个新终态、通道阈值命中留痕、手机号最近解绑人、提现资格校验留痕
- 退款:原因必填与申请人备注、来源支付与渠道流水冻结、线下处理流水号补录审计、按订单查询可选退款方式、企微审批材料补齐且新增字段缺失映射即明确失败
- 优先轮询:人工关闭、有效期到期独立周期任务、失败与过期人工重触发、事实字段与异常重试查询、资产解析端点只读投影
- 通道阈值:命中事实同事务留痕与命中记录查询;员工账单:列表筛选与详情投影;商户池:列表投影与统计周期语义;H5:弹窗类型与类别排序
- 手机号:有效关联数量与最近解绑人、短信验证码失败次数限制;导出:佣金明细十五列与报表序号列
- 时间筛选:三处新增筛选纳入统一严格解析契约,员工账单产生时间参数改名
- 同步 12 份主 Spec 需求、两端点与异步任务证据链,门禁 context-health 与 OpenSpec 校验通过
2026-09-18 15:34:29 +08:00

437 lines
18 KiB
Go
Raw Blame History

This file contains ambiguous Unicode characters
This file contains Unicode characters that might be confused with other characters. If you think that this is intentional, you can safely ignore this warning. Use the Escape button to reveal them.
package task
import (
"bytes"
"context"
"encoding/csv"
stderrors "errors"
"fmt"
"io"
"os"
"strconv"
"strings"
"time"
"github.com/bytedance/sonic"
"github.com/hibiken/asynq"
"go.uber.org/zap"
"gorm.io/gorm"
"github.com/break/junhong_cmp_fiber/internal/infrastructure/audit"
"github.com/break/junhong_cmp_fiber/internal/model"
assetSvc "github.com/break/junhong_cmp_fiber/internal/service/asset"
"github.com/break/junhong_cmp_fiber/internal/store/postgres"
"github.com/break/junhong_cmp_fiber/pkg/auditcontext"
"github.com/break/junhong_cmp_fiber/pkg/constants"
pkgerrors "github.com/break/junhong_cmp_fiber/pkg/errors"
"github.com/break/junhong_cmp_fiber/pkg/sanitizer"
"github.com/break/junhong_cmp_fiber/pkg/storage"
"github.com/break/junhong_cmp_fiber/pkg/utils"
)
// PhoneAssetUnbindImportPayload 手机号—资产关联 CSV 解绑导入任务载荷。
type PhoneAssetUnbindImportPayload struct {
TaskID uint `json:"task_id"`
}
// PhoneAssetUnbindImportHandler 手机号—资产关联 CSV 解绑导入任务处理器。
// 逐行独立事务:成功行提交、失败行不解除任何关系;任务级失败与行级失败分开记录。
// 行内只填资产标识与可选备注,解绑原因取任务级必填字段。
type PhoneAssetUnbindImportHandler struct {
db *gorm.DB
taskStore *postgres.PhoneAssetUnbindImportTaskStore
associationStore *postgres.PhoneAssetAssociationStore
assetIdentifierStore *postgres.AssetIdentifierStore
iotCardStore *postgres.IotCardStore
deviceStore *postgres.DeviceStore
storageService *storage.Service
auditWriter *audit.Writer
logger *zap.Logger
}
// NewPhoneAssetUnbindImportHandler 创建解绑导入任务处理器。
func NewPhoneAssetUnbindImportHandler(
db *gorm.DB,
taskStore *postgres.PhoneAssetUnbindImportTaskStore,
associationStore *postgres.PhoneAssetAssociationStore,
assetIdentifierStore *postgres.AssetIdentifierStore,
iotCardStore *postgres.IotCardStore,
deviceStore *postgres.DeviceStore,
storageService *storage.Service,
logger *zap.Logger,
auditWriters ...*audit.Writer,
) *PhoneAssetUnbindImportHandler {
handler := &PhoneAssetUnbindImportHandler{
db: db, taskStore: taskStore, associationStore: associationStore,
assetIdentifierStore: assetIdentifierStore, iotCardStore: iotCardStore,
deviceStore: deviceStore, storageService: storageService, logger: logger,
}
if len(auditWriters) > 0 {
handler.auditWriter = auditWriters[0]
}
return handler
}
// Handle 处理手机号—资产关联 CSV 解绑导入任务。
func (h *PhoneAssetUnbindImportHandler) Handle(ctx context.Context, taskMessage *asynq.Task) error {
var payload PhoneAssetUnbindImportPayload
if err := sonic.Unmarshal(taskMessage.Payload(), &payload); err != nil {
h.logger.Error("解析手机号资产解绑导入任务载荷失败", zap.Error(err))
return asynq.SkipRetry
}
taskRecord, err := h.taskStore.GetByID(ctx, payload.TaskID)
if err != nil {
h.logger.Error("查询手机号资产解绑导入任务失败", zap.Uint("task_id", payload.TaskID), zap.Error(err))
return asynq.SkipRetry
}
if h.auditWriter == nil {
return pkgerrors.New(pkgerrors.CodeInvalidStatus, "手机号资产解绑导入统一审计接缝未配置")
}
ctx = auditcontext.With(ctx, auditcontext.Context{
ActorKind: constants.AuditActorSystemTask, ActorID: constants.TaskTypePhoneAssetUnbindImport,
ActorName: "手机号资产解绑导入任务", Source: constants.AuditSourceWorker,
CorrelationID: taskRecord.TaskNo,
ParentEventID: audit.TaskEventID(constants.AuditResourcePhoneAssetUnbindImportTask, taskRecord.ID, "completed"),
})
claimed, err := h.taskStore.ResetForProcessing(ctx, taskRecord.ID)
if err != nil {
return err
}
if !claimed {
h.logger.Info("手机号资产解绑导入任务已终结,跳过重复消费", zap.Uint("task_id", taskRecord.ID))
return nil
}
rows, err := h.downloadAndParse(ctx, taskRecord.StorageKey)
if err != nil {
h.logger.Warn("下载或解析手机号资产解绑导入CSV失败", zap.Uint("task_id", taskRecord.ID), zap.Error(err))
if finishErr := h.finishTask(ctx, taskRecord, nil, 0, 0, model.ImportTaskStatusFailed, err.Error()); finishErr != nil {
return finishErr
}
return asynq.SkipRetry
}
items, successCount, err := h.processRows(ctx, taskRecord, rows)
if err != nil {
message := "导入执行中断:" + err.Error()
h.logger.Error("手机号资产解绑导入行执行中断", zap.Uint("task_id", taskRecord.ID), zap.Error(err))
if finishErr := h.finishTask(ctx, taskRecord, nil, 0, 0, model.ImportTaskStatusFailed, message); finishErr != nil {
return finishErr
}
return asynq.SkipRetry
}
failCount := len(items) - successCount
if err := h.finishTask(ctx, taskRecord, items, successCount, failCount, model.ImportTaskStatusCompleted, ""); err != nil {
return err
}
h.logger.Info("手机号资产解绑导入任务完成",
zap.Uint("task_id", taskRecord.ID), zap.Int("success", successCount), zap.Int("fail", failCount))
return nil
}
// finishTask 在单事务内写任务终态、逐行明细与任务根审计事件。
// 任务级失败不产生行明细,与行级失败原因分开记录。
func (h *PhoneAssetUnbindImportHandler) finishTask(
ctx context.Context,
taskRecord *model.PhoneAssetUnbindImportTask,
items model.PhoneAssetUnbindImportResults,
successCount, failCount, status int,
errorMessage string,
) error {
return h.db.WithContext(ctx).Transaction(func(tx *gorm.DB) error {
store := h.taskStore.WithTx(tx)
if status == model.ImportTaskStatusFailed {
// 任务级失败:未命中非终态说明任务已被其他执行路径终结,按库内事实跳过重复收尾。
hit, err := store.MarkFailed(ctx, taskRecord.ID, errorMessage)
if err != nil {
return err
}
if !hit {
return nil
}
} else if err := store.Complete(ctx, taskRecord.ID, len(items), successCount, failCount, items); err != nil {
return err
}
result := batchAuditResult(successCount, failCount)
afterData := map[string]any{
"status": status, "total_count": len(items), "success_count": successCount, "fail_count": failCount,
}
if status == model.ImportTaskStatusFailed {
result = constants.AuditResultFailed
afterData["error_message"] = errorMessage
}
return h.auditWriter.WriteTask(ctx, tx, audit.TaskInput{
EventID: audit.TaskEventID(constants.AuditResourcePhoneAssetUnbindImportTask, taskRecord.ID, "completed"),
ActionCode: constants.AuditActionPhoneAssetUnbindImportTaskCompleted,
Summary: "完成手机号资产解绑导入任务", TaskID: taskRecord.ID, TaskNo: taskRecord.TaskNo,
Result: result, CorrelationID: taskRecord.TaskNo,
ParentEventID: audit.TaskEventID(constants.AuditResourcePhoneAssetUnbindImportTask, taskRecord.ID, "created"),
BatchTotal: len(items), SuccessCount: successCount, FailCount: failCount,
IdentitySnapshot: map[string]any{
"id": taskRecord.ID, "task_no": taskRecord.TaskNo, "file_name": taskRecord.FileName,
},
BeforeData: map[string]any{"status": model.ImportTaskStatusProcessing},
AfterData: afterData,
ErrorSummary: errorMessage,
})
})
}
// downloadAndParse 下载并解析导入 CSV编码、表头或格式问题一律按任务级失败返回。
func (h *PhoneAssetUnbindImportHandler) downloadAndParse(ctx context.Context, key string) ([]phoneAssetUnbindImportRow, error) {
if h.storageService == nil {
return nil, phoneAssetUnbindImportError("对象存储服务未配置")
}
if key == "" {
return nil, phoneAssetUnbindImportError("导入文件Key不能为空")
}
localPath, cleanup, err := h.storageService.DownloadToTemp(ctx, key)
if err != nil {
return nil, phoneAssetUnbindImportError("下载导入CSV失败")
}
defer cleanup()
// 不设行数与体积硬上限:体积沿用上传用途的既有校验,此处按文件实际大小读取。
data, err := os.ReadFile(localPath)
if err != nil {
return nil, phoneAssetUnbindImportError("读取导入CSV失败")
}
decoded, err := utils.DecodeTextToUTF8(data)
if err != nil {
return nil, phoneAssetUnbindImportError(constants.PhoneAssetUnbindImportErrorEncoding)
}
return parsePhoneAssetUnbindImportCSV(decoded)
}
// phoneAssetUnbindImportRow 是导入文件的单行业务事实;行号自数据首行起计,表头不计入。
type phoneAssetUnbindImportRow struct {
Line int
ColumnCountMatched bool
AssetIdentifier string
Remark string
}
// parsePhoneAssetUnbindImportCSV 解析固定列序的导入 CSV。
// 表头必须与固定列序完全一致,不一致即任务级失败且不进入逐行阶段;
// 数据行列数不符属行级「行格式错误」,因此必须关闭字段数一致性校验,
// 否则标准库在首条记录定型字段数后会让后续异常行直接返回 ErrFieldCount
// 把行级问题误升级为任务级失败且不产生行明细。
func parsePhoneAssetUnbindImportCSV(data []byte) ([]phoneAssetUnbindImportRow, error) {
reader := csv.NewReader(bytes.NewReader(data))
reader.TrimLeadingSpace = true
reader.FieldsPerRecord = -1
rows := make([]phoneAssetUnbindImportRow, 0)
line := 0
for {
record, err := reader.Read()
if err == io.EOF {
break
}
if err != nil {
// 关闭字段数校验后仍报错,说明是引号未闭合等真实 CSV 语法错误,属任务级失败。
return nil, phoneAssetUnbindImportError(constants.PhoneAssetUnbindImportErrorFileFormat)
}
if line == 0 {
if !matchPhoneAssetUnbindImportHeader(record) {
return nil, phoneAssetUnbindImportError(constants.PhoneAssetUnbindImportErrorFileFormat)
}
line++
continue
}
line++
row := phoneAssetUnbindImportRow{Line: line - 1}
if len(record) != len(constants.PhoneAssetUnbindImportColumns) {
// 列数不符的行不参与业务校验,直接以行格式错误记录并保留原状。
rows = append(rows, row)
continue
}
row.ColumnCountMatched = true
row.AssetIdentifier = strings.TrimSpace(record[0])
row.Remark = strings.TrimSpace(record[1])
rows = append(rows, row)
}
if len(rows) == 0 {
return nil, phoneAssetUnbindImportError(constants.PhoneAssetUnbindImportErrorNoDataRow)
}
return rows, nil
}
// matchPhoneAssetUnbindImportHeader 逐列比较表头与固定列序,仅容忍列内两侧空白差异。
func matchPhoneAssetUnbindImportHeader(record []string) bool {
columns := constants.PhoneAssetUnbindImportColumns
if len(record) != len(columns) {
return false
}
for index, column := range columns {
if strings.TrimSpace(record[index]) != column {
return false
}
}
return true
}
// processRows 逐行独立执行并按批更新进度计数;进度写失败不回滚已提交行。
// 返回错误表示行执行遇到基础设施故障,由调用方按任务级失败收尾。
func (h *PhoneAssetUnbindImportHandler) processRows(ctx context.Context, taskRecord *model.PhoneAssetUnbindImportTask, rows []phoneAssetUnbindImportRow) (model.PhoneAssetUnbindImportResults, int, error) {
items := make(model.PhoneAssetUnbindImportResults, 0, len(rows))
successCount, failCount := 0, 0
for index, row := range rows {
item, err := h.processRow(ctx, taskRecord, row)
if err != nil {
return nil, 0, err
}
items = append(items, item)
if item.Status == constants.PhoneAssetUnbindImportItemStatusSuccess {
successCount++
} else {
failCount++
}
if (index+1)%constants.PhoneAssetUnbindImportProgressBatchSize == 0 {
if err := h.taskStore.UpdateProgress(ctx, taskRecord.ID, len(rows), successCount, failCount); err != nil {
h.logger.Warn("更新手机号资产解绑导入进度失败", zap.Uint("task_id", taskRecord.ID), zap.Error(err))
}
}
}
return items, successCount, nil
}
// processRow 校验并执行单行;失败行只记录固定枚举原因,不解除任何关系。
func (h *PhoneAssetUnbindImportHandler) processRow(ctx context.Context, taskRecord *model.PhoneAssetUnbindImportTask, row phoneAssetUnbindImportRow) (model.PhoneAssetUnbindImportResultItem, error) {
item := model.PhoneAssetUnbindImportResultItem{Line: row.Line, AssetIdentifier: row.AssetIdentifier}
if !row.ColumnCountMatched {
return failedPhoneAssetUnbindImportItem(item, constants.PhoneAssetUnbindImportRowErrorFormat), nil
}
if row.AssetIdentifier == "" {
return failedPhoneAssetUnbindImportItem(item, constants.PhoneAssetUnbindImportRowErrorIdentifier), nil
}
if len(row.Remark) > constants.PhoneAssetUnbindImportRemarkMaxLength {
return failedPhoneAssetUnbindImportItem(item, constants.PhoneAssetUnbindImportRowErrorRemarkTooLong), nil
}
assetType, assetID, err := h.resolveAsset(ctx, row.AssetIdentifier)
if err != nil {
return item, err
}
if assetType == "" {
return failedPhoneAssetUnbindImportItem(item, constants.PhoneAssetUnbindImportRowErrorAssetMissing), nil
}
item.AssetType = assetType
item.AssetID = assetID
// 每行独立事务:成功行提交,失败行回滚并保留原状。
rowErr := h.db.WithContext(ctx).Transaction(func(tx *gorm.DB) error {
locked, err := h.associationStore.WithTx(tx).LockValidByAsset(ctx, tx, assetType, assetID)
if err != nil {
return pkgerrors.Wrap(pkgerrors.CodeDatabaseError, err, "锁定资产关联失败")
}
if len(locked) == 0 {
return pkgerrors.New(pkgerrors.CodeNotFound, constants.PhoneAssetUnbindImportRowErrorNoAssociation)
}
ids := make([]uint, 0, len(locked))
phones := make([]string, 0, len(locked))
for _, association := range locked {
ids = append(ids, association.ID)
phones = append(phones, association.Phone)
}
now := time.Now()
// CSV 导入由任务创建人发起Worker 侧没有请求上下文,名称快照取任务落库时的创建人名称。
affected, err := h.associationStore.WithTx(tx).InvalidateByIDs(ctx, tx, ids,
constants.PhoneAssetAssociationInvalidateMethodCSVImport, taskRecord.UnbindReason, taskRecord.Creator,
taskRecord.CreatorName, now)
if err != nil {
return pkgerrors.Wrap(pkgerrors.CodeDatabaseError, err, "解除手机号资产关联失败")
}
if int(affected) != len(locked) {
return pkgerrors.New(pkgerrors.CodeNotFound, constants.PhoneAssetUnbindImportRowErrorNoAssociation)
}
item.UnboundCount = len(locked)
item.AssociatedPhones = phones
return h.appendRowAudit(ctx, tx, taskRecord, row, locked, now)
})
if rowErr != nil {
var appErr *pkgerrors.AppError
if stderrors.As(rowErr, &appErr) && appErr.Code == pkgerrors.CodeNotFound {
return failedPhoneAssetUnbindImportItem(item, appErr.Message), nil
}
return item, rowErr
}
item.Status = constants.PhoneAssetUnbindImportItemStatusSuccess
return item, nil
}
// resolveAsset 按资产标识定位资产,未命中返回空类型;解析口径与后台关联列表一致。
func (h *PhoneAssetUnbindImportHandler) resolveAsset(ctx context.Context, identifier string) (string, uint, error) {
return assetSvc.ResolveIdentifier(ctx, h.assetIdentifierStore, h.iotCardStore, h.deviceStore, identifier)
}
// appendRowAudit 在行事务内写单行解除审计,手机号一律脱敏,完整快照只进任务明细。
func (h *PhoneAssetUnbindImportHandler) appendRowAudit(
ctx context.Context,
tx *gorm.DB,
taskRecord *model.PhoneAssetUnbindImportTask,
row phoneAssetUnbindImportRow,
locked []*model.PhoneAssetAssociation,
now time.Time,
) error {
if h.auditWriter == nil {
return pkgerrors.New(pkgerrors.CodeInvalidStatus, "手机号资产解绑导入统一审计接缝未配置")
}
metadata := map[string]any{
"import_task_id": taskRecord.ID, "import_task_no": taskRecord.TaskNo,
"line": row.Line, "unbound_count": len(locked),
}
if row.Remark != "" {
metadata["remark"] = row.Remark
}
resources := make([]audit.ResourceInput, 0, len(locked)+1)
for index, association := range locked {
relation := constants.AuditResourceRelationReference
if index == 0 {
relation = constants.AuditResourceRelationPrimary
}
associationID := strconv.FormatUint(uint64(association.ID), 10)
resources = append(resources, audit.ResourceInput{
Type: constants.AuditResourcePhoneAssetAssociation, ID: &associationID,
Key: associationID, DisplayName: sanitizer.MaskPhone(association.Phone),
Relation: relation, Role: constants.AuditResourceRolePhoneAssetAssociationTarget,
IdentitySnapshot: map[string]any{
"id": association.ID, "phone_masked": sanitizer.MaskPhone(association.Phone),
"asset_type": association.AssetType, "asset_id": association.AssetID,
"status": constants.PhoneAssetAssociationStatusInvalid, "source": association.Source,
"invalidated_at": now, "invalidation_method": constants.PhoneAssetAssociationInvalidateMethodCSVImport,
"invalidation_reason": taskRecord.UnbindReason,
},
BeforeData: map[string]any{"status": constants.PhoneAssetAssociationStatusValid},
AfterData: map[string]any{
"status": constants.PhoneAssetAssociationStatusInvalid, "invalidated_at": now,
"invalidation_method": constants.PhoneAssetAssociationInvalidateMethodCSVImport,
"invalidation_reason": taskRecord.UnbindReason,
},
SubjectVisibility: constants.AuditSubjectInternalOnly, SortOrder: index + 1,
})
}
assetID := strconv.FormatUint(uint64(locked[0].AssetID), 10)
resources = append(resources, audit.ResourceInput{
Type: locked[0].AssetType, ID: &assetID, Key: assetID, DisplayName: row.AssetIdentifier,
Relation: constants.AuditResourceRelationReference, Role: constants.AuditResourceRolePhoneAssetAssociationAsset,
IdentitySnapshot: map[string]any{"id": locked[0].AssetID},
SubjectVisibility: constants.AuditSubjectInternalOnly,
})
// 稳定事件 ID 由任务与行号决定,任务重复消费时同行为幂等重放。
return h.auditWriter.Append(ctx, tx, audit.AppendInput{
EventID: audit.TaskEventID(constants.AuditResourcePhoneAssetUnbindImportTask, taskRecord.ID, fmt.Sprintf("item:%d", row.Line)),
ActionCode: constants.AuditActionPhoneAssetAssociationImported,
Summary: "导入解除手机号资产关联",
Result: constants.AuditResultSuccess, Metadata: metadata,
ScopeType: constants.AuditScopePlatform, Resources: resources,
})
}
func failedPhoneAssetUnbindImportItem(item model.PhoneAssetUnbindImportResultItem, reason string) model.PhoneAssetUnbindImportResultItem {
item.Status, item.Reason = constants.PhoneAssetUnbindImportItemStatusFailed, reason
return item
}
// phoneAssetUnbindImportError 是任务级失败原因,与行级失败原因分开记录。
type phoneAssetUnbindImportError string
// Error 返回任务级失败原因原文。
func (e phoneAssetUnbindImportError) Error() string { return string(e) }