Files
iotlcm/server/internal/library/batchimport/runner.go
T

606 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 batchimport
import (
"context"
"fmt"
"io"
"os"
"strings"
"git.5iot.com/iotlcm-public/iotlcm-app/internal/consts"
"git.5iot.com/iotlcm-public/iotlcm-app/internal/dao"
"git.5iot.com/iotlcm-public/iotlcm-app/internal/library/batchmetadata"
"git.5iot.com/iotlcm-public/iotlcm-app/internal/model"
"git.5iot.com/iotlcm-public/iotlcm-app/internal/model/do"
"git.5iot.com/iotlcm-public/iotlcm-app/internal/model/entity"
"git.5iot.com/iotlcm-public/iotlcm-app/internal/service"
"git.5iot.com/iotlcm-public/iotlcm-app/utility/hash"
"git.5iot.com/iotlcm-public/iotlcm-app/utility/iot_excelize"
"github.com/gogf/gf/v2/crypto/gmd5"
"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/net/ghttp"
"github.com/gogf/gf/v2/os/gtime"
"github.com/xuri/excelize/v2"
)
func SubmitAndRun(ctx context.Context, req SubmitRequest, p Processor) (*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
result := &Result{
ImportStatus: consts.ImportStatusFailed,
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,
NotAddDate: 0,
})
if err != nil {
return nil, gerror.Wrap(err, "保存原始Excel失败")
}
result.ImportFileOssId = importOss.OssId
result.ImportFileUrl = importOss.Url
rows, err := p.ParseRows(ctx, importOss.Path)
if err != nil {
result.FailReason = userFacingFailReason(err)
if err = archiveFailedBatchAfterImport(ctx, req, p, result, nil, importOss); err != nil {
return nil, err
}
return result, nil
}
result.TotalRows = len(rows)
if err = p.ValidateRows(ctx, rows); err != nil {
for _, row := range rows {
row.AddErrorReason(userFacingFailReason(err))
}
}
fillResultCounts(result, rows)
startTime := gtime.Now()
endTime := gtime.Now()
if err = saveBatchAndExecute(ctx, &req, p, result, rows, importOss, startTime, endTime); err != nil {
markRowsFailed(result, rows, userFacingFailReason(err))
if err = saveBatchAndExecute(ctx, &req, p, result, rows, importOss, startTime, endTime); err != nil {
return nil, err
}
}
result.ExecutionResult = req.ExecutionResult
resultOss, downloadId, err := saveResultExcelAndDownload(ctx, req, p, result, rows, importOss, startTime, endTime)
if err != nil {
return nil, err
}
result.ResultDownloadId = downloadId
result.ResultFileOssId = resultOss.OssId
result.ResultFileUrl = resultOss.Url
if err = updateBatchResultFileInfo(ctx, req, p, result, importOss, resultOss, downloadId); err != nil {
return nil, err
}
return result, nil
}
func RunRows(ctx context.Context, req SubmitRequest, p Processor, rows []Row, sourceName string) (*Result, error) {
if err := batchmetadata.ValidateCreation(req.SourceChannel, req.ExecutionMode); err != nil {
return nil, err
}
cfg := p.Config()
batchNo, err := NextBatchNoContext(ctx, cfg.BatchNoPrefix)
if err != nil {
return nil, err
}
req.BatchNo = batchNo
result := &Result{
ImportStatus: consts.ImportStatusFailed,
BatchNo: batchNo,
TotalRows: len(rows),
}
if len(rows) == 0 {
result.FailReason = "导入数据为空"
} else if err = p.ValidateRows(ctx, rows); err != nil {
for _, row := range rows {
row.AddErrorReason(userFacingFailReason(err))
}
}
fillResultCounts(result, rows)
if result.FailReason != "" {
result.ImportStatus = consts.ImportStatusFailed
result.InsertedRows = 0
}
startTime := gtime.Now()
endTime := gtime.Now()
importOss := &model.SysOssSaveContentModel{SysOss: &entity.SysOss{}}
if sourceName != "" {
importOss.OriginalName = sourceName
}
if err = saveBatchAndExecute(ctx, &req, p, result, rows, importOss, startTime, endTime); err != nil {
markRowsFailed(result, rows, userFacingFailReason(err))
if err = saveBatchAndExecute(ctx, &req, p, result, rows, importOss, startTime, endTime); err != nil {
return nil, err
}
}
result.ExecutionResult = req.ExecutionResult
resultOss, downloadId, err := saveResultExcelAndDownload(ctx, req, p, result, rows, importOss, startTime, endTime)
if err != nil {
return nil, err
}
result.ResultDownloadId = downloadId
result.ResultFileOssId = resultOss.OssId
result.ResultFileUrl = resultOss.Url
if err = updateBatchResultFileInfo(ctx, req, p, result, importOss, resultOss, downloadId); err != nil {
return nil, err
}
return result, nil
}
func archiveFailedBatchAfterImport(ctx context.Context, req SubmitRequest, p Processor, result *Result, rows []Row, importOss *model.SysOssSaveContentModel) error {
result.ImportStatus = consts.ImportStatusFailed
result.InsertedRows = 0
if result.FailReason == "" {
result.FailReason = "导入失败"
}
startTime := gtime.Now()
endTime := gtime.Now()
if err := saveBatchAndExecute(ctx, &req, p, result, rows, importOss, startTime, endTime); err != nil {
return err
}
result.ExecutionResult = req.ExecutionResult
resultOss, downloadId, err := saveResultExcelAndDownload(ctx, req, p, result, rows, importOss, startTime, endTime)
if err != nil {
return err
}
result.ResultDownloadId = downloadId
result.ResultFileOssId = resultOss.OssId
result.ResultFileUrl = resultOss.Url
return updateBatchResultFileInfo(ctx, req, p, result, importOss, resultOss, downloadId)
}
func readUploadContent(file *ghttp.UploadFile) ([]byte, error) {
reader, err := file.Open()
if err != nil {
return nil, gerror.Wrap(err, "打开上传文件失败")
}
defer reader.Close()
content, err := io.ReadAll(reader)
if err != nil {
return nil, gerror.Wrap(err, "读取上传文件失败")
}
return content, nil
}
func fillResultCounts(result *Result, rows []Row) {
failCount := 0
for _, row := range rows {
if len(row.GetErrorReasons()) > 0 {
failCount++
}
}
result.ErrorRows = failCount
result.ValidRows = result.TotalRows - failCount
if result.ValidRows < 0 {
result.ValidRows = 0
}
if failCount == 0 {
result.ImportStatus = consts.ImportStatusSuccess
result.InsertedRows = len(rows)
return
}
result.ImportStatus = consts.ImportStatusFailed
result.InsertedRows = 0
}
func markRowsFailed(result *Result, rows []Row, reason string) {
if result == nil {
return
}
if reason == "" {
reason = "业务处理失败"
}
for _, row := range rows {
row.AddErrorReason(reason)
}
result.ImportStatus = consts.ImportStatusFailed
result.FailReason = reason
result.TotalRows = len(rows)
result.ValidRows = 0
result.ErrorRows = len(rows)
result.InsertedRows = 0
}
// saveBatchAndExecute writes stock data only when all rows pass validation.
// Failed validation keeps the batch and result file, but skips business inserts.
func saveBatchAndExecute(ctx context.Context, req *SubmitRequest, p Processor, result *Result, rows []Row, importOss *model.SysOssSaveContentModel, startTime, endTime *gtime.Time) error {
return g.DB().Transaction(ctx, func(ctx context.Context, tx gdb.TX) error {
now := gtime.Now()
failReason := result.FailReason
if failReason == "" && result.ErrorRows > 0 {
failReason = "存在校验失败数据,本次未写入"
}
batchType := batchTypeByImportStatus(req, p, result.ImportStatus)
result.BatchType = batchType
if result.ImportStatus == consts.ImportStatusSuccess && len(rows) > 0 && req.BeforeSave != nil {
if err := req.BeforeSave(ctx, tx, req, rows); err != nil {
return err
}
}
sourceBatchNo := req.SourceBatchNo
if sourceBatchNo == "" {
sourceBatchNo = result.BatchNo
}
if _, err := dao.BatchMain.Ctx(ctx).TX(tx).Data(do.BatchMain{
BatchNo: result.BatchNo,
OrderSn: "",
BatchType: batchType,
BatchStatus: batchStatusByImportStatus(result.ImportStatus),
BatchAsyncStatus: batchStatusByImportStatus(result.ImportStatus),
SourceType: req.SourceType,
SourceChannel: batchmetadata.Channel(req.SourceChannel),
ExecutionMode: batchmetadata.NormalizeMode(req.ExecutionMode),
SourceBatchNo: sourceBatchNo,
IsCrossDept: 0,
TargetDeptId: req.TargetDeptId,
DeptId: req.Operator.DeptId,
TotalCount: result.TotalRows,
ValidCount: result.ValidRows,
SuccessCount: result.InsertedRows,
FailCount: result.ErrorRows,
RunResult: runResultByImportStatus(result.ImportStatus),
FailReason: failReason,
BatchInfo: p.BuildBatchInfo(ctx, *req, result, importOss, nil, 0),
StartTime: startTime,
EndTime: endTime,
TenantId: req.Operator.TenantId,
CreatedDept: req.Operator.DeptId,
CreatedBy: req.Operator.UserId,
CreatedAt: now,
UpdatedBy: req.Operator.UserId,
UpdatedAt: now,
}).Insert(); err != nil {
return gerror.Wrap(err, "写入批次主表失败")
}
details := make([]do.BatchDetail, 0, len(rows))
for _, row := range rows {
details = append(details, do.BatchDetail{
BatchNo: result.BatchNo,
BatchType: batchType,
DetailStatus: detailStatus(row),
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,
CreatedAt: now,
UpdatedBy: req.Operator.UserId,
UpdatedAt: now,
})
}
if len(details) > 0 {
if _, err := dao.BatchDetail.Ctx(ctx).TX(tx).Data(details).Insert(); err != nil {
return gerror.Wrap(err, "写入批次明细失败")
}
}
if result.ImportStatus == consts.ImportStatusSuccess && len(rows) > 0 && !req.SkipExecute {
if err := p.Execute(ctx, tx, req, rows); err != nil {
return err
}
}
return nil
})
}
func saveResultExcelAndDownload(ctx context.Context, req SubmitRequest, p Processor, result *Result, rows []Row, importOss *model.SysOssSaveContentModel, startTime, endTime *gtime.Time) (*model.SysOssSaveContentModel, int64, error) {
cfg := p.Config()
resultRows := make([]iot_excelize.BatchResultRow, 0, len(rows))
for _, row := range rows {
resultRows = append(resultRows, iot_excelize.BatchResultRow{
RowNo: row.GetRowNo(),
ItemID: row.GetItemID(),
Status: displayRowStatus(row),
ErrorReason: strings.Join(row.GetErrorReasons(), ";"),
})
}
importName := ""
importOssId := ""
if importOss != nil {
importName = importOss.OriginalName
if importOss.OssId > 0 {
importOssId = fmt.Sprintf("%d", importOss.OssId)
}
}
infos := []iot_excelize.BatchInfoItem{
{Field: "批次号", Value: result.BatchNo},
{Field: "批次类型", Value: taskTypeName(p)},
{Field: "处理结果", Value: displayImportStatus(result.ImportStatus)},
{Field: "总行数", Value: fmt.Sprintf("%d", result.TotalRows)},
{Field: "校验通过行数", Value: fmt.Sprintf("%d", result.ValidRows)},
{Field: "失败行数", Value: fmt.Sprintf("%d", result.ErrorRows)},
{Field: "实际写入行数", Value: fmt.Sprintf("%d", result.InsertedRows)},
{Field: "原始文件名", Value: importName},
{Field: "原始文件OSS ID", Value: importOssId},
{Field: "结果文件OSS ID", Value: ""},
{Field: "操作人", Value: fmt.Sprintf("%d", req.Operator.UserId)},
{Field: "开始时间", Value: formatGTime(startTime)},
{Field: "结束时间", Value: formatGTime(endTime)},
{Field: "备注", Value: resultRemark(result)},
}
f, err := iot_excelize.BuildBatchResultExcel(resultRows, infos)
if err != nil {
return nil, 0, err
}
defer f.Close()
reader, err := iot_excelize.ToReadSeeker(f)
if err != nil {
return nil, 0, err
}
content, err := io.ReadAll(reader)
if err != nil {
return nil, 0, err
}
resultOss, err := service.SysOss().SaveContent(ctx, &model.SysOssSaveContentParam{
Content: content,
FileName: result.BatchNo + "-result.xlsx",
NewFileType: consts.SysUploadNewFileTypeUserFile,
SubDirName: cfg.ResultDir,
NotAddDate: 0,
})
if err != nil {
return nil, 0, gerror.Wrap(err, "保存结果Excel失败")
}
if err = fillResultOssId(ctx, req, resultOss); err != nil {
return nil, 0, err
}
now := gtime.Now()
downloadId, err := dao.TResourceBatchDownload.Ctx(ctx).Data(do.TResourceBatchDownload{
BatchNo: result.BatchNo,
TaskName: cfg.TaskName,
ResourceType: consts.ResourceTypeExcel,
ExportScene: cfg.ExportScene,
TotalCount: result.TotalRows,
SuccessCount: result.InsertedRows,
FailCount: result.ErrorRows,
DownloadStatus: consts.DownloadStatusGenerated,
OssId: resultOss.OssId,
FileName: resultOss.FileName,
FileExt: resultOss.FileSuffix,
FileUrl: resultOss.Url,
FilePath: resultOss.Path,
FileSize: resultOss.FileSize,
FileMd5: resultOss.Md5,
DownloadCount: 0,
MaxDownloadCount: 0,
DeptId: req.Operator.DeptId,
StartTime: startTime,
FinishTime: now,
TenantId: req.Operator.TenantId,
CreatedDept: req.Operator.DeptId,
CreatedBy: req.Operator.UserId,
CreatedAt: now,
UpdatedBy: req.Operator.UserId,
UpdatedAt: now,
}).InsertAndGetId()
if err != nil {
return nil, 0, gerror.Wrap(err, "写入结果下载任务失败")
}
return resultOss, downloadId, nil
}
func fillResultOssId(ctx context.Context, req SubmitRequest, resultOss *model.SysOssSaveContentModel) error {
f, err := excelize.OpenFile(resultOss.Path)
if err != nil {
return gerror.Wrap(err, "打开结果Excel失败")
}
defer f.Close()
const batchInfoSheet = "批次信息"
rows, err := f.GetRows(batchInfoSheet)
if err != nil {
return gerror.Wrap(err, "读取结果Excel批次信息失败")
}
found := false
for i, row := range rows {
if len(row) == 0 || strings.TrimSpace(row[0]) != "结果文件OSS ID" {
continue
}
cell, err := excelize.CoordinatesToCellName(2, i+1)
if err != nil {
return err
}
if err = f.SetCellStr(batchInfoSheet, cell, fmt.Sprintf("%d", resultOss.OssId)); err != nil {
return err
}
found = true
break
}
if !found {
return gerror.New("结果Excel缺少结果文件OSS ID字段")
}
if err = f.SaveAs(resultOss.Path); err != nil {
return gerror.Wrap(err, "回写结果Excel OSS ID失败")
}
fileInfo, err := os.Stat(resultOss.Path)
if err != nil {
return gerror.Wrap(err, "读取结果Excel文件信息失败")
}
md5, _ := gmd5.EncryptFile(resultOss.Path)
crc16 := hash.Crc16ChecksumFile(resultOss.Path)
sum16 := hash.Sum16ChecksumFile(resultOss.Path)
resultOss.FileSize = int(fileInfo.Size())
resultOss.Md5 = md5
resultOss.FileCrc16 = int(crc16)
resultOss.FileSum = int(sum16)
_, err = dao.SysOss.Ctx(ctx).
Where(dao.SysOss.Columns().OssId, resultOss.OssId).
Data(do.SysOss{
Md5: resultOss.Md5,
FileSize: resultOss.FileSize,
FileCrc16: resultOss.FileCrc16,
FileSum: resultOss.FileSum,
UpdatedAt: gtime.Now(),
UpdatedBy: req.Operator.UserId,
}).
Update()
if err != nil {
return gerror.Wrap(err, "更新结果Excel文件信息失败")
}
return nil
}
func updateBatchResultFileInfo(ctx context.Context, req SubmitRequest, p Processor, result *Result, importOss *model.SysOssSaveContentModel, resultOss *model.SysOssSaveContentModel, downloadId int64) error {
_, err := dao.BatchMain.Ctx(ctx).
Where(dao.BatchMain.Columns().BatchNo, result.BatchNo).
Data(do.BatchMain{
BatchInfo: p.BuildBatchInfo(ctx, req, result, importOss, resultOss, downloadId),
UpdatedBy: req.Operator.UserId,
UpdatedAt: gtime.Now(),
}).
Update()
if err != nil {
return gerror.Wrap(err, "更新批次结果文件信息失败")
}
return nil
}
func batchStatusByImportStatus(importStatus string) int {
if importStatus == consts.ImportStatusSuccess {
return consts.BatchAsyncStatusSuccess
}
return consts.BatchAsyncStatusFailed
}
func runResultByImportStatus(importStatus string) int {
if importStatus == consts.ImportStatusSuccess {
return consts.BatchRunResultSuccess
}
return consts.BatchRunResultFailed
}
func batchTypeByImportStatus(req *SubmitRequest, p Processor, importStatus string) int {
if req == nil {
return p.Config().BatchType
}
if importStatus != consts.ImportStatusSuccess && req.FailureBatchType > 0 {
return req.FailureBatchType
}
if req.BatchType > 0 {
return req.BatchType
}
return p.Config().BatchType
}
func detailStatus(row Row) int {
if len(row.GetErrorReasons()) > 0 {
return consts.BatchDetailStatusFailed
}
return consts.BatchDetailStatusSuccess
}
func displayRowStatus(row Row) string {
if len(row.GetErrorReasons()) > 0 {
return "失败"
}
return "成功"
}
func displayImportStatus(status string) string {
if status == consts.ImportStatusSuccess {
return "成功"
}
return "失败"
}
func resultRemark(result *Result) string {
if result != nil && result.ImportStatus == consts.ImportStatusSuccess {
return "导入成功"
}
if result != nil && result.FailReason != "" {
return result.FailReason
}
return "存在校验失败数据,本次未写入"
}
func userFacingFailReason(err error) string {
if err == nil {
return ""
}
reason := strings.TrimSpace(err.Error())
if reason == "" {
return "导入失败"
}
const maxLen = 500
if len([]rune(reason)) <= maxLen {
return reason
}
runes := []rune(reason)
return string(runes[:maxLen])
}
func taskTypeName(p Processor) string {
taskName := p.Config().TaskName
return strings.TrimSuffix(taskName, "结果")
}
func formatGTime(t *gtime.Time) string {
if t == nil {
return ""
}
return t.Format("Y-m-d H:i:s")
}