iotlcm/admin-svr/server/internal/logic/iot/simcardbatch/allocate/submitter.go

package allocate

import (
	"context"
	"path/filepath"
	"strings"

	"git.5iot.com/iotlcm-public/iotlcm-admin-svr/internal/consts"
	"git.5iot.com/iotlcm-public/iotlcm-admin-svr/internal/library/allocationunits"
	"git.5iot.com/iotlcm-public/iotlcm-admin-svr/internal/library/batchimport"
	"git.5iot.com/iotlcm-public/iotlcm-admin-svr/internal/library/batchmetadata"
	"git.5iot.com/iotlcm-public/iotlcm-admin-svr/internal/logic/iot/simcardbatch/common"
	"git.5iot.com/iotlcm-public/iotlcm-admin-svr/internal/model"
	"git.5iot.com/iotlcm-public/iotlcm-admin-svr/internal/service"

	"github.com/gogf/gf/v2/errors/gerror"
	"github.com/gogf/gf/v2/net/ghttp"
)

func CreatePendingExcel(ctx context.Context, file *ghttp.UploadFile, in *model.TSimCardAllocateExcelSubmitParam) (*PendingResult, error) {
	if in == nil {
		in = &model.TSimCardAllocateExcelSubmitParam{}
	}
	if file == nil {
		return nil, gerror.New("upload file is required")
	}
	if err := service.SysOss().CheckType(ctx, consts.SysUploadFileTypeFile, file); err != nil {
		return nil, err
	}
	if file.Size > allocateMaxFileSize(ctx) {
		return nil, gerror.Newf("uploaded file exceeds the limit of %d MB", allocateMaxFileSize(ctx)>>20)
	}

	options := SubmitOptions{
		TargetDeptId: in.TargetDeptId,
		Remark:       strings.TrimSpace(in.Remark),
	}
	processor := NewProcessor()
	parentBatchNo, err := batchimport.NextBatchNoContext(ctx, processor.Config().BatchNoPrefix)
	if err != nil {
		return nil, err
	}
	extension := strings.ToLower(filepath.Ext(file.Filename))
	if extension == "" {
		extension = ".xlsx"
	}
	importOss, err := batchimport.SaveUpload(ctx, file, parentBatchNo+"-import"+extension, processor.Config().ImportDir)
	if err != nil {
		return nil, gerror.Wrap(err, "save allocate import file failed")
	}

	req := batchimport.SubmitRequest{
		SourceChannel: batchmetadata.Console,
		ExecutionMode: batchmetadata.Bulk,
		BizParams:     &options,
		SourceType:    consts.BatchSourceTypeFile,
		TargetDeptId:  options.TargetDeptId,
		Operator:      common.CurrentOperator(ctx),
		BatchNo:       parentBatchNo,
	}
	stream := func(streamCtx context.Context, emit func([]batchimport.Row) error) (int, error) {
		chunkSize := allocateChunkSize(streamCtx)
		if extension == ".csv" {
			return streamAllocateCSV(streamCtx, importOss.Path, chunkSize, emit)
		}
		return streamAllocateExcel(streamCtx, importOss.Path, chunkSize, emit)
	}
	return createPendingSplit(ctx, req, processor, prepareOptions(options), parentBatchNo, importOss, stream)
}

func CreatePendingInput(ctx context.Context, in *model.TSimCardAllocatePrepareParam) (*PendingResult, error) {
	if in == nil {
		in = &model.TSimCardAllocatePrepareParam{}
	}
	if in.TargetDeptId <= 0 {
		return nil, gerror.New("target customer is required")
	}

	options := SubmitOptions{TargetDeptId: in.TargetDeptId}
	processor := NewProcessor()
	parentBatchNo, err := batchimport.NextBatchNoContext(ctx, processor.Config().BatchNoPrefix)
	if err != nil {
		return nil, err
	}
	req := batchimport.SubmitRequest{
		SourceChannel: batchmetadata.Console,
		ExecutionMode: batchmetadata.Bulk,
		BizParams:     &options,
		SourceType:    consts.BatchSourceTypeBatchInput,
		TargetDeptId:  options.TargetDeptId,
		Operator:      common.CurrentOperator(ctx),
		BatchNo:       parentBatchNo,
	}
	stream := func(streamCtx context.Context, emit func([]batchimport.Row) error) (int, error) {
		return streamAllocateInput(streamCtx, in.Input, options.TargetDeptId, allocateChunkSize(streamCtx), emit)
	}
	return createPendingSplit(ctx, req, processor, prepareOptions(options), parentBatchNo, nil, stream)
}

func prepareOptions(options SubmitOptions) batchimport.PendingOptions {
	return batchimport.PendingOptions{
		PrepareRows: func(ctx context.Context, rows []batchimport.Row) error {
			for _, item := range rows {
				row, ok := item.(*Row)
				if !ok || row == nil {
					return gerror.New("allocate row type is invalid")
				}
				if row.TargetDeptId <= 0 && options.TargetDeptId > 0 {
					row.TargetDeptId = options.TargetDeptId
				}
				if row.Remark == "" {
					row.Remark = options.Remark
				}
				if row.TargetDeptId <= 0 && row.TargetDeptKey == "" && row.TargetDeptName == "" {
					row.AddErrorReason("target customer is required")
				}
			}
			ids := []int64{}
			for _, item := range rows {
				if row, ok := item.(*Row); ok {
					ids = append(ids, row.TargetDeptId)
				}
			}
			units, err := allocationunits.Load(ctx, common.CurrentOperator(ctx).TenantId, ids)
			if err != nil {
				return err
			}
			for _, item := range rows {
				if row, ok := item.(*Row); ok {
					if unit, exists := units[row.TargetDeptId]; exists {
						row.TargetDeptKey = unit.Key
						row.TargetDeptName = unit.Name
					}
				}
			}
			return nil
		},
	}
}

func pendingResultFromBatch(result *batchimport.Result) *PendingResult {
	res := &PendingResult{
		BatchType:        consts.BatchTypeAllocate,
		BatchAsyncStatus: consts.BatchAsyncStatusPending,
	}
	if result == nil {
		return res
	}
	res.BatchNo = result.BatchNo
	res.BatchType = result.BatchType
	res.Total = result.TotalRows
	res.ValidCount = result.ValidRows
	res.FailCount = result.ErrorRows
	return res
}