iotlcm/admin-svr/server/internal/library/batchimport/async.go

package batchimport

import (
	"context"
	"git.5iot.com/iotlcm-public/iotlcm-admin-svr/internal/library/batchmetadata"
	"strings"

	"git.5iot.com/iotlcm-public/iotlcm-admin-svr/internal/consts"
	"git.5iot.com/iotlcm-public/iotlcm-admin-svr/internal/dao"
	"git.5iot.com/iotlcm-public/iotlcm-admin-svr/internal/model"
	"git.5iot.com/iotlcm-public/iotlcm-admin-svr/internal/model/do"
	"git.5iot.com/iotlcm-public/iotlcm-admin-svr/internal/service"

	"github.com/gogf/gf/v2/database/gdb"
	"github.com/gogf/gf/v2/errors/gerror"
	"github.com/gogf/gf/v2/frame/g"
	"github.com/gogf/gf/v2/os/gtime"
)

type PendingOptions struct {
	PrepareRows func(ctx context.Context, rows []Row) error
}

const pendingDetailInsertBatchSize = 500

func CreatePendingUpload(ctx context.Context, req SubmitRequest, p PendingProcessor, options PendingOptions) (*Result, error) {
	if err := batchmetadata.ValidateCreation(req.SourceChannel, req.ExecutionMode); err != nil {
		return nil, err
	}
	cfg := p.Config()
	if req.File == nil {
		return nil, gerror.New("请选择上传文件")
	}
	if err := service.SysOss().CheckType(ctx, consts.SysUploadFileTypeFile, req.File); err != nil {
		return nil, err
	}
	if req.File.Size > cfg.MaxFileSize {
		return nil, gerror.Newf("上传文件大小超过限制，最大为%dM", cfg.MaxFileSize>>20)
	}

	batchNo, err := NextBatchNoContext(ctx, cfg.BatchNoPrefix)
	if err != nil {
		return nil, err
	}
	req.BatchNo = batchNo
	content, err := readUploadContent(req.File)
	if err != nil {
		return nil, err
	}
	importOss, err := service.SysOss().SaveContent(ctx, &model.SysOssSaveContentParam{
		Content:     content,
		FileName:    batchNo + "-import.xlsx",
		NewFileType: consts.SysUploadNewFileTypeUserFile,
		SubDirName:  cfg.ImportDir,
	})
	if err != nil {
		return nil, gerror.Wrap(err, "保存原始Excel失败")
	}

	rows, err := p.ParseRows(ctx, importOss.Path)
	if err != nil {
		return nil, err
	}
	return createPending(ctx, &req, p, rows, importOss, options)
}

func CreatePendingRows(ctx context.Context, req SubmitRequest, p PendingProcessor, rows []Row, options PendingOptions) (*Result, error) {
	if err := batchmetadata.ValidateCreation(req.SourceChannel, req.ExecutionMode); err != nil {
		return nil, err
	}
	return createPending(ctx, &req, p, rows, nil, options)
}

func createPending(ctx context.Context, req *SubmitRequest, p PendingProcessor, rows []Row, importOss *model.SysOssSaveContentModel, options PendingOptions) (*Result, error) {
	if len(rows) == 0 {
		return nil, gerror.New("导入数据为空")
	}
	if err := validatePendingRowCount(len(rows), p.Config().MaxRows); err != nil {
		return nil, err
	}
	if err := p.ValidateRows(ctx, rows); err != nil {
		return nil, err
	}
	if options.PrepareRows != nil {
		if err := options.PrepareRows(ctx, rows); err != nil {
			return nil, err
		}
	}
	if strings.TrimSpace(req.BatchNo) == "" {
		batchNo, err := NextBatchNoContext(ctx, p.Config().BatchNoPrefix)
		if err != nil {
			return nil, err
		}
		req.BatchNo = batchNo
	}

	result := &Result{
		ImportStatus: consts.ImportStatusSuccess,
		BatchNo:      req.BatchNo,
		BatchType:    p.Config().BatchType,
		TotalRows:    len(rows),
	}
	if importOss != nil {
		result.ImportFileOssId = importOss.OssId
		result.ImportFileUrl = importOss.Url
	}

	if err := g.DB().Transaction(ctx, func(ctx context.Context, tx gdb.TX) error {
		sourceBatchNo := strings.TrimSpace(req.SourceBatchNo)
		if sourceBatchNo == "" {
			sourceBatchNo = req.BatchNo
		}
		if req.BeforeSave != nil {
			if err := req.BeforeSave(ctx, tx, req, rows); err != nil {
				return err
			}
		}
		if _, err := tx.Ctx(ctx).Model(dao.BatchMain.Table()).Data(do.BatchMain{
			BatchNo:          req.BatchNo,
			OrderSn:          "",
			BatchType:        p.Config().BatchType,
			BatchAsyncStatus: consts.BatchAsyncStatusPending,
			SourceType:       req.SourceType,
			SourceChannel:    batchmetadata.Channel(req.SourceChannel),
			ExecutionMode:    batchmetadata.NormalizeMode(req.ExecutionMode),
			SourceBatchNo:    sourceBatchNo,
			SourceOssId:      ossIDValue(sourceOssID(result, importOss)),
			IsCrossDept:      0,
			TargetDeptId:     req.TargetDeptId,
			DeptId:           req.Operator.DeptId,
			TotalCount:       len(rows),
			ValidCount:       0,
			SuccessCount:     0,
			FailCount:        0,
			RunResult:        consts.BatchRunResultPending,
			FailReason:       "",
			BatchInfo:        p.BuildBatchInfo(ctx, *req, result, importOss, nil, 0),
			TenantId:         req.Operator.TenantId,
			CreatedDept:      req.Operator.DeptId,
			CreatedBy:        req.Operator.UserId,
			UpdatedBy:        req.Operator.UserId,
		}).Insert(); err != nil {
			return gerror.Wrap(err, "写入待处理批次失败")
		}

		details := make([]do.BatchDetail, 0, len(rows))
		for _, row := range rows {
			details = append(details, do.BatchDetail{
				BatchNo:        req.BatchNo,
				BatchType:      p.Config().BatchType,
				DetailStatus:   consts.BatchDetailStatusPending,
				BlankOutStatus: consts.BlankOutStatusNo,
				RowNo:          row.GetRowNo(),
				ItemId:         row.GetItemID(),
				ErrorCode:      "",
				ErrorReason:    strings.Join(row.GetErrorReasons(), "；"),
				Snapshot:       row.Snapshot(),
				TenantId:       req.Operator.TenantId,
				CreatedDept:    req.Operator.DeptId,
				CreatedBy:      req.Operator.UserId,
				UpdatedBy:      req.Operator.UserId,
			})
		}
		for start := 0; start < len(details); start += pendingDetailInsertBatchSize {
			end := start + pendingDetailInsertBatchSize
			if end > len(details) {
				end = len(details)
			}
			if _, err := tx.Ctx(ctx).Model(dao.BatchDetail.Table()).Data(details[start:end]).Insert(); err != nil {
				return gerror.Wrapf(err, "写入待处理批次明细失败，范围:%d-%d", start+1, end)
			}
		}
		return nil
	}); err != nil {
		return nil, err
	}
	return result, nil
}

func validatePendingRowCount(rowCount, maxRows int) error {
	if maxRows > 0 && rowCount > maxRows {
		return gerror.Newf("单批最多支持%d张卡，当前共%d张，请拆分后重试", maxRows, rowCount)
	}
	return nil
}

func SaveAsyncResultFiles(
	ctx context.Context,
	req SubmitRequest,
	p Processor,
	result *Result,
	rows []Row,
	importOss *model.SysOssSaveContentModel,
	startTime, endTime *gtime.Time,
) (*model.SysOssSaveContentModel, int64, error) {
	resultOss, downloadId, err := saveResultExcelAndDownload(ctx, req, p, result, rows, importOss, startTime, endTime)
	if err != nil {
		return nil, 0, err
	}
	result.ResultDownloadId = downloadId
	result.ResultFileOssId = resultOss.OssId
	result.ResultFileUrl = resultOss.Url
	return resultOss, downloadId, nil
}