package batchimport import ( "context" "fmt" "io" "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/library/batchmetadata" "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/model/entity" "git.5iot.com/iotlcm-public/iotlcm-admin-svr/internal/service" "git.5iot.com/iotlcm-public/iotlcm-admin-svr/utility/hash" "git.5iot.com/iotlcm-public/iotlcm-admin-svr/utility/iot_excelize" "os" "strings" "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 = "存在校验失败数据,本次未写入" } if result.ImportStatus == consts.ImportStatusFailed && result.ErrorRows > 0 { markValidRowsNotExecuted(rows) } 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 := tx.Ctx(ctx).Model(dao.BatchMain.Table()).Data(do.BatchMain{ BatchNo: result.BatchNo, OrderSn: "", BatchType: batchType, BatchAsyncStatus: asyncStatusByImportStatus(result.ImportStatus), BatchStatus: consts.BatchStatusNormal, 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: 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 := tx.Ctx(ctx).Model(dao.BatchDetail.Table()).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{ SourceOssId: ossIDValue(sourceOssID(result, importOss)), ResultOssId: ossIDValue(resultOssID(result, resultOss)), 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 sourceOssID(result *Result, importOss *model.SysOssSaveContentModel) int64 { if result != nil && result.ImportFileOssId > 0 { return result.ImportFileOssId } if importOss != nil && importOss.OssId > 0 { return importOss.OssId } return 0 } func resultOssID(result *Result, resultOss *model.SysOssSaveContentModel) int64 { if result != nil && result.ResultFileOssId > 0 { return result.ResultFileOssId } if resultOss != nil && resultOss.OssId > 0 { return resultOss.OssId } return 0 } func ossIDValue(ossID int64) any { if ossID <= 0 { return nil } return ossID } func asyncStatusByImportStatus(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 } const NotExecutedReasonPrefix = "\u672c\u6279\u5b58\u5728\u6821\u9a8c\u5931\u8d25\uff0c\u6574\u6279\u672a\u6267\u884c\uff1a" func markValidRowsNotExecuted(rows []Row) { for _, row := range rows { if row != nil && len(row.GetErrorReasons()) == 0 { row.AddErrorReason(NotExecutedReasonPrefix + "\u672c\u5361\u6821\u9a8c\u901a\u8fc7") } } } func IsNotExecuted(row Row) bool { for _, reason := range row.GetErrorReasons() { if strings.HasPrefix(reason, NotExecutedReasonPrefix) { return true } } return false } func detailStatus(row Row) int { if IsNotExecuted(row) { return consts.BatchDetailStatusNotExecuted } if len(row.GetErrorReasons()) > 0 { return consts.BatchDetailStatusFailed } return consts.BatchDetailStatusSuccess } func displayRowStatus(row Row) string { if IsNotExecuted(row) { return "未执行" } 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") }