break
178cc45bc2
All checks were successful
构建并部署到测试环境（无 SSH） / build-and-deploy (push) Successful in 8m1s
290 lines
10 KiB
Go
290 lines
10 KiB
Go
package device_import
|
||
|
||
import (
|
||
"context"
|
||
"path/filepath"
|
||
"strings"
|
||
"time"
|
||
|
||
"github.com/break/junhong_cmp_fiber/internal/model"
|
||
"github.com/break/junhong_cmp_fiber/internal/model/dto"
|
||
"github.com/break/junhong_cmp_fiber/internal/store"
|
||
"github.com/break/junhong_cmp_fiber/internal/store/postgres"
|
||
"github.com/break/junhong_cmp_fiber/pkg/constants"
|
||
"github.com/break/junhong_cmp_fiber/pkg/errors"
|
||
"github.com/break/junhong_cmp_fiber/pkg/middleware"
|
||
"github.com/break/junhong_cmp_fiber/pkg/queue"
|
||
"github.com/hibiken/asynq"
|
||
"gorm.io/gorm"
|
||
)
|
||
|
||
type Service struct {
|
||
db *gorm.DB
|
||
importTaskStore *postgres.DeviceImportTaskStore
|
||
queueClient *queue.Client
|
||
assetAudit AssetAuditService
|
||
}
|
||
|
||
type DeviceImportPayload struct {
|
||
TaskID uint `json:"task_id"`
|
||
}
|
||
|
||
func New(
|
||
db *gorm.DB,
|
||
importTaskStore *postgres.DeviceImportTaskStore,
|
||
queueClient *queue.Client,
|
||
assetAudit AssetAuditService,
|
||
) *Service {
|
||
return &Service{
|
||
db: db,
|
||
importTaskStore: importTaskStore,
|
||
queueClient: queueClient,
|
||
assetAudit: assetAudit,
|
||
}
|
||
}
|
||
|
||
func (s *Service) CreateImportTask(ctx context.Context, req *dto.ImportDeviceRequest) (*dto.ImportDeviceResponse, error) {
|
||
userID := middleware.GetUserIDFromContext(ctx)
|
||
if userID == 0 {
|
||
appErr := errors.New(errors.CodeUnauthorized, "未授权访问")
|
||
s.logDeviceImportAudit(ctx, newDeviceImportAuditParams(0, "", req, constants.AssetAuditResultDenied, appErr))
|
||
return nil, appErr
|
||
}
|
||
|
||
taskNo := s.importTaskStore.GenerateTaskNo(ctx)
|
||
fileName := filepath.Base(req.FileKey)
|
||
|
||
task := &model.DeviceImportTask{
|
||
TaskNo: taskNo,
|
||
OperationType: constants.DeviceImportOperationCreate,
|
||
Status: model.ImportTaskStatusPending,
|
||
BatchNo: req.BatchNo,
|
||
FileName: fileName,
|
||
StorageKey: req.FileKey,
|
||
RealnamePolicy: req.RealnamePolicy,
|
||
CreatorName: middleware.GetUsernameFromContext(ctx),
|
||
}
|
||
task.Creator = userID
|
||
task.Updater = userID
|
||
|
||
if err := s.importTaskStore.Create(ctx, task); err != nil {
|
||
appErr := errors.Wrap(errors.CodeInternalError, err, "创建导入任务失败")
|
||
s.logDeviceImportAudit(ctx, newDeviceImportAuditParams(0, taskNo, req, constants.AssetAuditResultFailed, appErr))
|
||
return nil, appErr
|
||
}
|
||
|
||
payload := DeviceImportPayload{TaskID: task.ID}
|
||
err := s.queueClient.EnqueueTask(
|
||
ctx,
|
||
constants.TaskTypeDeviceImport,
|
||
payload,
|
||
asynq.Queue(constants.QueueForTaskType(constants.TaskTypeDeviceImport)),
|
||
)
|
||
if err != nil {
|
||
s.importTaskStore.UpdateStatus(ctx, task.ID, model.ImportTaskStatusFailed, "任务入队失败: "+err.Error())
|
||
appErr := errors.Wrap(errors.CodeInternalError, err, "任务入队失败")
|
||
s.logDeviceImportAudit(ctx, newDeviceImportAuditParams(task.ID, taskNo, req, constants.AssetAuditResultFailed, appErr))
|
||
return nil, appErr
|
||
}
|
||
|
||
s.logDeviceImportAudit(ctx, newDeviceImportAuditParams(task.ID, taskNo, req, constants.AssetAuditResultSuccess, nil))
|
||
|
||
return &dto.ImportDeviceResponse{
|
||
TaskID: task.ID,
|
||
TaskNo: taskNo,
|
||
Message: "导入任务已创建，Worker 将异步处理文件",
|
||
}, nil
|
||
}
|
||
|
||
// CreateBatchAllocationTask 复用设备导入任务创建单列 CSV 批量操作任务。
|
||
func (s *Service) CreateBatchAllocationTask(ctx context.Context, req *dto.CreateDeviceBatchAllocationRequest) (*dto.CreateDeviceBatchAllocationResponse, error) {
|
||
userID := middleware.GetUserIDFromContext(ctx)
|
||
userType := middleware.GetUserTypeFromContext(ctx)
|
||
if userID == 0 || (userType != constants.UserTypeSuperAdmin && userType != constants.UserTypePlatform && userType != constants.UserTypeAgent) {
|
||
appErr := errors.New(errors.CodeForbidden, "仅平台和代理后台账号可创建设备CSV批量任务")
|
||
s.logDeviceImportAudit(ctx, newDeviceBatchAllocationAuditParams(0, "", req, constants.AssetAuditResultDenied, appErr))
|
||
return nil, appErr
|
||
}
|
||
if req == nil || !constants.IsDeviceImportOperation(req.OperationType) || req.OperationType == constants.DeviceImportOperationCreate {
|
||
return nil, errors.New(errors.CodeInvalidParam, "设备CSV批量任务参数不合法")
|
||
}
|
||
if (req.OperationType == constants.DeviceImportOperationRecall && req.TargetID != 0) ||
|
||
(req.OperationType != constants.DeviceImportOperationRecall && req.TargetID == 0) {
|
||
return nil, errors.New(errors.CodeInvalidParam, "设备CSV批量任务参数不合法")
|
||
}
|
||
if !strings.HasPrefix(req.FileKey, constants.DeviceBatchAllocationStoragePrefix+"/") || !strings.EqualFold(filepath.Ext(req.FileKey), ".csv") {
|
||
return nil, errors.New(errors.CodeInvalidParam, "设备CSV批量文件必须是指定目录下的CSV文件")
|
||
}
|
||
|
||
taskNo := s.importTaskStore.GenerateTaskNo(ctx)
|
||
var operatorShopID *uint
|
||
if userType == constants.UserTypeAgent {
|
||
shopID := middleware.GetShopIDFromContext(ctx)
|
||
if shopID == 0 {
|
||
return nil, errors.New(errors.CodeForbidden, "代理账号缺少店铺归属")
|
||
}
|
||
operatorShopID = &shopID
|
||
}
|
||
var targetID *uint
|
||
if req.OperationType != constants.DeviceImportOperationRecall {
|
||
targetID = &req.TargetID
|
||
}
|
||
task := &model.DeviceImportTask{
|
||
TaskNo: taskNo, OperationType: req.OperationType, TargetID: targetID,
|
||
OperatorType: userType, OperatorShopID: operatorShopID,
|
||
Status: model.ImportTaskStatusPending, StorageKey: req.FileKey, FileName: filepath.Base(req.FileKey),
|
||
CreatorName: middleware.GetUsernameFromContext(ctx),
|
||
}
|
||
task.Creator, task.Updater = userID, userID
|
||
if err := s.importTaskStore.Create(ctx, task); err != nil {
|
||
appErr := errors.Wrap(errors.CodeDatabaseError, err, "创建设备CSV批量任务失败")
|
||
s.logDeviceImportAudit(ctx, newDeviceBatchAllocationAuditParams(0, taskNo, req, constants.AssetAuditResultFailed, appErr))
|
||
return nil, appErr
|
||
}
|
||
if err := s.queueClient.EnqueueTask(ctx, constants.TaskTypeDeviceImport, DeviceImportPayload{TaskID: task.ID},
|
||
asynq.Queue(constants.QueueForTaskType(constants.TaskTypeDeviceImport)),
|
||
asynq.Timeout(constants.DeviceBatchAllocationTaskTimeout)); err != nil {
|
||
_ = s.importTaskStore.UpdateStatus(ctx, task.ID, model.ImportTaskStatusFailed, "任务入队失败")
|
||
appErr := errors.Wrap(errors.CodeInternalError, err, "设备CSV批量任务入队失败")
|
||
s.logDeviceImportAudit(ctx, newDeviceBatchAllocationAuditParams(task.ID, taskNo, req, constants.AssetAuditResultFailed, appErr))
|
||
return nil, appErr
|
||
}
|
||
s.logDeviceImportAudit(ctx, newDeviceBatchAllocationAuditParams(task.ID, taskNo, req, constants.AssetAuditResultSuccess, nil))
|
||
return &dto.CreateDeviceBatchAllocationResponse{
|
||
TaskID: task.ID, TaskNo: task.TaskNo, Message: "设备CSV批量任务已创建，Worker 将异步处理CSV文件",
|
||
}, nil
|
||
}
|
||
|
||
func (s *Service) List(ctx context.Context, req *dto.ListDeviceImportTaskRequest) (*dto.ListDeviceImportTaskResponse, error) {
|
||
page := req.Page
|
||
pageSize := req.PageSize
|
||
if page == 0 {
|
||
page = 1
|
||
}
|
||
if pageSize == 0 {
|
||
pageSize = constants.DefaultPageSize
|
||
}
|
||
|
||
opts := &store.QueryOptions{
|
||
Page: page,
|
||
PageSize: pageSize,
|
||
}
|
||
|
||
filters := make(map[string]interface{})
|
||
if req.Status != nil {
|
||
filters["status"] = *req.Status
|
||
}
|
||
if req.OperationType != "" {
|
||
filters["operation_type"] = req.OperationType
|
||
}
|
||
if req.BatchNo != "" {
|
||
filters["batch_no"] = req.BatchNo
|
||
}
|
||
if req.StartTime != nil {
|
||
filters["start_time"] = *req.StartTime
|
||
}
|
||
if req.EndTime != nil {
|
||
filters["end_time"] = *req.EndTime
|
||
}
|
||
|
||
tasks, total, err := s.importTaskStore.List(ctx, opts, filters)
|
||
if err != nil {
|
||
return nil, err
|
||
}
|
||
|
||
list := make([]*dto.DeviceImportTaskResponse, 0, len(tasks))
|
||
for _, task := range tasks {
|
||
list = append(list, s.toTaskResponse(task))
|
||
}
|
||
|
||
return &dto.ListDeviceImportTaskResponse{
|
||
List: list,
|
||
Total: total,
|
||
Page: page,
|
||
PageSize: pageSize,
|
||
}, nil
|
||
}
|
||
|
||
func (s *Service) GetByID(ctx context.Context, id uint) (*dto.DeviceImportTaskDetailResponse, error) {
|
||
task, err := s.importTaskStore.GetByID(ctx, id)
|
||
if err != nil {
|
||
return nil, errors.New(errors.CodeNotFound, "导入任务不存在")
|
||
}
|
||
|
||
resp := &dto.DeviceImportTaskDetailResponse{
|
||
DeviceImportTaskResponse: *s.toTaskResponse(task),
|
||
SkippedItems: make([]*dto.DeviceImportResultItemDTO, 0),
|
||
FailedItems: make([]*dto.DeviceImportResultItemDTO, 0),
|
||
WarningItems: make([]*dto.DeviceImportResultItemDTO, 0),
|
||
}
|
||
|
||
for _, item := range task.SkippedItems {
|
||
resp.SkippedItems = append(resp.SkippedItems, &dto.DeviceImportResultItemDTO{
|
||
Line: item.Line, VirtualNo: item.ICCID, DeviceIdentifier: item.ICCID, Reason: item.Reason,
|
||
})
|
||
}
|
||
|
||
for _, item := range task.FailedItems {
|
||
resp.FailedItems = append(resp.FailedItems, &dto.DeviceImportResultItemDTO{
|
||
Line: item.Line, VirtualNo: item.ICCID, DeviceIdentifier: item.ICCID, Reason: item.Reason,
|
||
})
|
||
}
|
||
|
||
for _, item := range task.WarningItems {
|
||
resp.WarningItems = append(resp.WarningItems, &dto.DeviceImportResultItemDTO{
|
||
Line: item.Line, VirtualNo: item.ICCID, DeviceIdentifier: item.ICCID, Reason: item.Reason,
|
||
})
|
||
}
|
||
|
||
return resp, nil
|
||
}
|
||
|
||
func (s *Service) toTaskResponse(task *model.DeviceImportTask) *dto.DeviceImportTaskResponse {
|
||
var startedAt, completedAt *time.Time
|
||
if task.StartedAt != nil {
|
||
startedAt = task.StartedAt
|
||
}
|
||
if task.CompletedAt != nil {
|
||
completedAt = task.CompletedAt
|
||
}
|
||
|
||
return &dto.DeviceImportTaskResponse{
|
||
ID: task.ID,
|
||
TaskNo: task.TaskNo,
|
||
OperationType: task.OperationType,
|
||
OperationName: constants.GetDeviceImportOperationName(task.OperationType),
|
||
TargetID: task.TargetID,
|
||
Status: task.Status,
|
||
StatusName: getStatusText(task.Status),
|
||
StatusText: getStatusText(task.Status),
|
||
BatchNo: task.BatchNo,
|
||
RealnamePolicy: task.RealnamePolicy,
|
||
FileName: task.FileName,
|
||
TotalCount: task.TotalCount,
|
||
SuccessCount: task.SuccessCount,
|
||
SkipCount: task.SkipCount,
|
||
FailCount: task.FailCount,
|
||
WarningCount: task.WarningCount,
|
||
StartedAt: startedAt,
|
||
CompletedAt: completedAt,
|
||
ErrorMessage: task.ErrorMessage,
|
||
CreatorName: task.CreatorName,
|
||
CreatedAt: task.CreatedAt,
|
||
}
|
||
}
|
||
|
||
func getStatusText(status int) string {
|
||
switch status {
|
||
case model.ImportTaskStatusPending:
|
||
return "待处理"
|
||
case model.ImportTaskStatusProcessing:
|
||
return "处理中"
|
||
case model.ImportTaskStatusCompleted:
|
||
return "已完成"
|
||
case model.ImportTaskStatusFailed:
|
||
return "失败"
|
||
default:
|
||
return "未知"
|
||
}
|
||
}
|