feat: 物料导入校验与事务入库编排

- 新增 MaterialImportService.Import:解析 → 逐行校验 → 事务入库 → 写导入记录
- 校验覆盖名称/规格长度、大类小类归属、单位名称或编码匹配、小类编码规则存在性
- 同一小类的规则只查一次,避免批量行数放大成逐行往返
- 校验失败整批不入库并返回逐行明细;入库失败整体回滚
- 导入记录成功失败都写,失败原因只存前 5 行行号摘要,明细不落库
This commit is contained in:
文睿 张 2026-10-10 15:33:13 +08:00
parent 0e3b1735a5
commit e853846070
2 changed files with 481 additions and 0 deletions

View File

@ -0,0 +1,284 @@
package service
import (
"fmt"
"io"
"strings"
"github.com/wmms/wmms/internal/model"
"github.com/wmms/wmms/internal/repository"
)
// MaterialImportService 负责一次物料批量导入:解析、逐行校验、事务入库、写记录。
// 首版只支持新增;校验阶段不生成编码,编码在入库事务里按小类规则分配。
type MaterialImportService struct {
materialRepo *repository.MaterialRepo
masterRepo *repository.MasterDataRepository
logRepo *repository.MaterialLogRepo
importRepo *repository.MaterialImportLogRepo
}
func NewMaterialImportService() *MaterialImportService {
return &MaterialImportService{
materialRepo: repository.NewMaterialRepo(),
masterRepo: repository.NewMasterDataRepository(),
logRepo: repository.NewMaterialLogRepo(),
importRepo: repository.NewMaterialImportLogRepo(),
}
}
// Import 处理一次上传。返回值约定:
// - 校验未通过:返回 result(含逐行错误)与 nil error,前端按明细展示;
// - 模板或文件不合法:返回 nil 与错误,前端提示原因;
// - 入库失败:返回 nil 与错误,事务已回滚,整批未落库。
//
// 三种情况都会写一条导入记录。
func (s *MaterialImportService) Import(organizationID int64, operator, fileName string, r io.Reader) (*model.MaterialImportResult, error) {
if organizationID <= 0 {
return nil, ErrInvalidOrganizationID
}
rows, err := ParseMaterialImportFile(r)
if err != nil {
s.recordImport(organizationID, operator, fileName, 0, 0, 0, model.MaterialImportStatusFailed, err.Error())
return nil, err
}
materials, importErrors, err := s.validateRows(organizationID, rows)
if err != nil {
s.recordImport(organizationID, operator, fileName, len(rows), 0, len(rows), model.MaterialImportStatusFailed, err.Error())
return nil, err
}
if len(importErrors) > 0 {
result := &model.MaterialImportResult{
TotalRows: len(rows),
SuccessRows: 0,
FailedRows: len(importErrors),
Errors: importErrors,
}
s.recordImport(organizationID, operator, fileName, result.TotalRows, 0, result.FailedRows,
model.MaterialImportStatusFailed, summarizeImportErrors(importErrors))
return result, nil
}
if err := s.persist(organizationID, operator, materials); err != nil {
s.recordImport(organizationID, operator, fileName, len(rows), 0, len(rows), model.MaterialImportStatusFailed, err.Error())
return nil, err
}
result := &model.MaterialImportResult{
TotalRows: len(rows),
SuccessRows: len(materials),
FailedRows: 0,
Errors: []model.MaterialImportError{},
}
s.recordImport(organizationID, operator, fileName, result.TotalRows, result.SuccessRows, 0, model.MaterialImportStatusSuccess, "")
return result, nil
}
// validateRows 逐行校验并组装待插入物料。返回的错误是查询主数据失败这类系统级
// 错误;逐行的问题以 importErrors 返回,两者互斥。
func (s *MaterialImportService) validateRows(organizationID int64, rows []model.MaterialImportRow) ([]*model.Material, []model.MaterialImportError, error) {
categories, err := s.masterRepo.ListCategories(organizationID)
if err != nil {
return nil, nil, err
}
units, err := s.masterRepo.ListUnits(organizationID)
if err != nil {
return nil, nil, err
}
majors := make(map[string]*model.MaterialCategory)
minors := make(map[string]*model.MaterialCategory) // key: 大类ID + "\x00" + 小类名
for i := range categories {
category := &categories[i]
if category.Status != "active" {
continue
}
if category.ParentID == nil {
majors[category.Name] = category
continue
}
minors[minorCategoryKey(*category.ParentID, category.Name)] = category
}
unitByName := make(map[string]*model.Unit)
unitByCode := make(map[string]*model.Unit)
for i := range units {
unit := &units[i]
if unit.Status != "active" {
continue
}
// 名称和编码各自只认第一条,避免同名单位覆盖成不确定的结果
if _, exists := unitByName[unit.Name]; !exists {
unitByName[unit.Name] = unit
}
if _, exists := unitByCode[unit.Code]; !exists {
unitByCode[unit.Code] = unit
}
}
materials := make([]*model.Material, 0, len(rows))
importErrors := make([]model.MaterialImportError, 0)
// 同一小类的规则只查一次:两万行通常只落在少数几个小类上,
// 逐行去查会把一次导入放大成两万次往返。
ruleAvailable := make(map[int64]bool)
for _, row := range rows {
reasons := make([]string, 0, 3)
name := row.Name
if name == "" {
reasons = append(reasons, "物料名称不能为空")
} else if len([]rune(name)) > 200 {
reasons = append(reasons, "物料名称超过 200 字")
}
if len([]rune(row.Spec)) > 200 {
reasons = append(reasons, "规格超过 200 字")
}
major := majors[row.MajorName]
if row.MajorName == "" {
reasons = append(reasons, "大类不能为空")
} else if major == nil {
reasons = append(reasons, fmt.Sprintf("大类「%s」不存在或已停用", row.MajorName))
}
var minor *model.MaterialCategory
if row.MinorName == "" {
reasons = append(reasons, "小类不能为空")
} else if major == nil {
// 大类已经报错,小类无法定位,不重复报
} else {
minor = minors[minorCategoryKey(major.ID, row.MinorName)]
if minor == nil {
reasons = append(reasons, fmt.Sprintf("小类「%s」在「%s」下不存在或已停用", row.MinorName, row.MajorName))
}
}
// 单位留空时按单条新增的口径默认「个」;填了则先按名称、再按编码匹配,
// 因为像「kg」和「千克」这种编码与名称不一致的单位两种写法都合理。
unitName := row.Unit
if unitName == "" {
unitName = "个"
}
unit := unitByName[unitName]
if unit == nil {
unit = unitByCode[unitName]
}
if unit == nil {
if row.Unit == "" {
reasons = append(reasons, "本单位缺少默认单位「个」,请在模板中填写单位")
} else {
reasons = append(reasons, fmt.Sprintf("单位「%s」不存在或已停用", row.Unit))
}
}
if minor != nil {
available, checked := ruleAvailable[minor.ID]
if !checked {
rule, err := s.masterRepo.FindActiveCodeRule(organizationID, minor.ID)
if err != nil {
return nil, nil, err
}
available = rule != nil
ruleAvailable[minor.ID] = available
}
if !available {
reasons = append(reasons, fmt.Sprintf("小类「%s」未配置编码规则", row.MinorName))
}
}
if len(reasons) > 0 {
importErrors = append(importErrors, model.MaterialImportError{
Row: row.Row,
Data: row.Data(),
Reason: strings.Join(reasons, ";"),
})
continue
}
var baseUnitID *int64
if unit != nil {
id := unit.ID
baseUnitID = &id
}
categoryID := minor.ID
materials = append(materials, &model.Material{
OrganizationID: organizationID,
Name: name,
Spec: row.Spec,
CategoryID: &categoryID,
BaseUnitID: baseUnitID,
Unit: unitName,
Status: "active",
Version: 1,
})
}
return materials, importErrors, nil
}
// persist 在单个事务里插入物料并写审计日志,任一步失败整批回滚。
func (s *MaterialImportService) persist(organizationID int64, operator string, materials []*model.Material) error {
tx, err := repository.BeginTx()
if err != nil {
return err
}
defer func() { _ = tx.Rollback() }()
if err := s.materialRepo.ImportTx(tx, organizationID, materials); err != nil {
return err
}
logs := make([]*model.MaterialLog, 0, len(materials))
for _, material := range materials {
logs = append(logs, &model.MaterialLog{
OperationType: "CREATE",
MaterialID: material.ID,
Operator: operator,
Description: "批量导入创建",
})
}
if err := s.logRepo.CreateBatchTx(tx, logs); err != nil {
return err
}
return tx.Commit()
}
// recordImport 写导入记录。记录失败只影响可追溯性,不应该让已经完成的导入
// 变成失败,所以这里吞掉错误。
func (s *MaterialImportService) recordImport(organizationID int64, operator, fileName string, total, success, failed int, status, reason string) {
if len([]rune(reason)) > 1000 {
reason = string([]rune(reason)[:1000])
}
_ = s.importRepo.Create(&model.MaterialImportLog{
OrganizationID: organizationID,
Operator: operator,
FileName: fileName,
TotalRows: total,
SuccessRows: success,
FailedRows: failed,
Status: status,
FailureReason: reason,
})
}
// summarizeImportErrors 生成给记录表用的失败原因摘要:前 5 行行号 + 总数。
func summarizeImportErrors(importErrors []model.MaterialImportError) string {
rows := make([]string, 0, 5)
for i, item := range importErrors {
if i == 5 {
break
}
rows = append(rows, fmt.Sprintf("%d", item.Row))
}
summary := fmt.Sprintf("共 %d 行校验未通过(行号:%s", len(importErrors), strings.Join(rows, "、"))
if len(importErrors) > 5 {
summary += " 等"
}
return summary + ")"
}
func minorCategoryKey(majorID int64, minorName string) string {
return fmt.Sprintf("%d\x00%s", majorID, minorName)
}

View File

@ -0,0 +1,197 @@
package service
import (
"bytes"
"regexp"
"strings"
"testing"
"github.com/DATA-DOG/go-sqlmock"
"github.com/wmms/wmms/pkg/database"
)
const importCategoryQuery = `SELECT id, organization_id, code, name, description, parent_id, status FROM material_categories WHERE organization_id = ? ORDER BY code`
const importUnitQuery = `SELECT id, organization_id, code, name, precision_scale, status FROM units WHERE organization_id = ? ORDER BY code`
const importRuleQuery = `SELECT id, organization_id, category_id, prefix, sequence_length, next_sequence, status FROM code_rules WHERE organization_id = ? AND category_id = ? AND status = 'active'`
func newImportServiceMock(t *testing.T) (sqlmock.Sqlmock, func()) {
t.Helper()
db, mock, err := sqlmock.New()
if err != nil {
t.Fatalf("open sql mock: %v", err)
}
previous := database.DB
database.DB = db
return mock, func() {
database.DB = previous
db.Close()
}
}
func expectImportMasterData(mock sqlmock.Sqlmock, withRule bool) {
mock.ExpectQuery(regexp.QuoteMeta(importCategoryQuery)).WithArgs(int64(1)).
WillReturnRows(sqlmock.NewRows([]string{"id", "organization_id", "code", "name", "description", "parent_id", "status"}).
AddRow(3, 1, "A", "机械结构类", "", nil, "active").
AddRow(12, 1, "A01", "标准件", "", 3, "active"))
mock.ExpectQuery(regexp.QuoteMeta(importUnitQuery)).WithArgs(int64(1)).
WillReturnRows(sqlmock.NewRows([]string{"id", "organization_id", "code", "name", "precision_scale", "status"}).
AddRow(1, 1, "个", "个", 0, "active"))
if withRule {
mock.ExpectQuery(regexp.QuoteMeta(importRuleQuery)).WithArgs(int64(1), int64(12)).
WillReturnRows(sqlmock.NewRows([]string{"id", "organization_id", "category_id", "prefix", "sequence_length", "next_sequence", "status"}).
AddRow(5, 1, 12, "A01", 4, 201, "active"))
} else {
mock.ExpectQuery(regexp.QuoteMeta(importRuleQuery)).WithArgs(int64(1), int64(12)).
WillReturnRows(sqlmock.NewRows([]string{"id", "organization_id", "category_id", "prefix", "sequence_length", "next_sequence"}))
}
}
// reason 用 interface{}:成功场景断言具体空串,失败场景传 sqlmock.AnyArg()
// 表示只校验"写了一条记录",不锁死文案。
func expectImportLog(mock sqlmock.Sqlmock, fileName, status string, total, success, failed int, reason interface{}) {
// 行数显式转 int64:database/sql 会把实参归一成 driver.Value(int→int64),
// sqlmock 按类型比对,传 Go int 匹配不上。
mock.ExpectExec(regexp.QuoteMeta(`INSERT INTO material_import_logs (organization_id, operator, file_name, total_rows, success_rows, failed_rows, status, failure_reason) VALUES (?, ?, ?, ?, ?, ?, ?, ?)`)).
WithArgs(int64(1), "tester", fileName, int64(total), int64(success), int64(failed), status, reason).
WillReturnResult(sqlmock.NewResult(9, 1))
}
func TestMaterialImportServiceReportsRowErrorsWithoutWritingMaterials(t *testing.T) {
mock, cleanup := newImportServiceMock(t)
defer cleanup()
expectImportMasterData(mock, true)
expectImportLog(mock, "物料.xlsx", "failed", 2, 0, 2, sqlmock.AnyArg())
data := buildTestWorkbook(t, [][]string{
{"", "", "", "机械结构类", "标准件"}, // 名称缺失
{"螺栓", "", "吨", "不存在的类", "标准件"}, // 大类不存在
})
result, err := NewMaterialImportService().Import(1, "tester", "物料.xlsx", bytes.NewReader(data))
if err != nil {
t.Fatalf("import: %v", err)
}
if result.SuccessRows != 0 || result.FailedRows != 2 || len(result.Errors) != 2 {
t.Fatalf("result = %+v", result)
}
if result.Errors[0].Row != 2 || !strings.Contains(result.Errors[0].Reason, "物料名称") {
t.Fatalf("first error = %+v", result.Errors[0])
}
if result.Errors[1].Row != 3 || !strings.Contains(result.Errors[1].Reason, "大类") {
t.Fatalf("second error = %+v", result.Errors[1])
}
if err := mock.ExpectationsWereMet(); err != nil {
t.Fatal(err)
}
}
func TestMaterialImportServiceReportsMissingCodeRule(t *testing.T) {
mock, cleanup := newImportServiceMock(t)
defer cleanup()
expectImportMasterData(mock, false)
expectImportLog(mock, "物料.xlsx", "failed", 1, 0, 1, sqlmock.AnyArg())
data := buildTestWorkbook(t, [][]string{
{"螺栓", "", "个", "机械结构类", "标准件"},
})
result, err := NewMaterialImportService().Import(1, "tester", "物料.xlsx", bytes.NewReader(data))
if err != nil {
t.Fatalf("import: %v", err)
}
if result.FailedRows != 1 || !strings.Contains(result.Errors[0].Reason, "编码规则") {
t.Fatalf("result = %+v", result)
}
if err := mock.ExpectationsWereMet(); err != nil {
t.Fatal(err)
}
}
func TestMaterialImportServiceRejectsInvalidTemplate(t *testing.T) {
mock, cleanup := newImportServiceMock(t)
defer cleanup()
expectImportLog(mock, "错的.xlsx", "failed", 0, 0, 0, sqlmock.AnyArg())
result, err := NewMaterialImportService().Import(1, "tester", "错的.xlsx", strings.NewReader("not an xlsx"))
if err == nil {
t.Fatal("期望返回模板错误")
}
if result != nil {
t.Fatalf("result = %+v, want nil", result)
}
if err := mock.ExpectationsWereMet(); err != nil {
t.Fatal(err)
}
}
func TestMaterialImportServiceImportsWhenAllRowsValid(t *testing.T) {
mock, cleanup := newImportServiceMock(t)
defer cleanup()
expectImportMasterData(mock, true)
// 事务内:锁规则 → 取已占用编码 → 推进序号 → 插入 → 回查 ID
// ImportTx 先把整批编码分配完并推进 code_rules,再分批插入。
mock.ExpectBegin()
mock.ExpectQuery(regexp.QuoteMeta(`SELECT id, organization_id, category_id, prefix, sequence_length, next_sequence FROM code_rules WHERE organization_id = ? AND category_id = ? AND status = 'active' FOR UPDATE`)).
WithArgs(int64(1), int64(12)).
WillReturnRows(sqlmock.NewRows([]string{"id", "organization_id", "category_id", "prefix", "sequence_length", "next_sequence"}).
AddRow(5, 1, 12, "A01", 4, 201))
mock.ExpectQuery(`SELECT material_code FROM materials WHERE organization_id = \? AND material_code LIKE \?`).
WithArgs(int64(1), "A01%").
WillReturnRows(sqlmock.NewRows([]string{"material_code"}))
mock.ExpectExec(regexp.QuoteMeta(`UPDATE code_rules SET next_sequence = ? WHERE id = ?`)).
WithArgs(int64(202), int64(5)).WillReturnResult(sqlmock.NewResult(0, 1))
mock.ExpectExec(`INSERT INTO materials`).WillReturnResult(sqlmock.NewResult(31, 1))
mock.ExpectQuery(`SELECT id, material_code FROM materials WHERE organization_id = \? AND material_code IN \(\?\)`).
WithArgs(int64(1), "A010201").
WillReturnRows(sqlmock.NewRows([]string{"id", "material_code"}).AddRow(31, "A010201"))
mock.ExpectExec(`INSERT INTO material_logs`).WillReturnResult(sqlmock.NewResult(1, 1))
mock.ExpectCommit()
expectImportLog(mock, "物料.xlsx", "success", 1, 1, 0, "")
data := buildTestWorkbook(t, [][]string{
{"内六角螺栓", "M8x30", "个", "机械结构类", "标准件"},
})
result, err := NewMaterialImportService().Import(1, "tester", "物料.xlsx", bytes.NewReader(data))
if err != nil {
t.Fatalf("import: %v", err)
}
if result.SuccessRows != 1 || result.FailedRows != 0 || len(result.Errors) != 0 {
t.Fatalf("result = %+v", result)
}
if err := mock.ExpectationsWereMet(); err != nil {
t.Fatal(err)
}
}
func TestMaterialImportServiceRollsBackWhenInsertFails(t *testing.T) {
mock, cleanup := newImportServiceMock(t)
defer cleanup()
expectImportMasterData(mock, true)
mock.ExpectBegin()
mock.ExpectQuery(`FOR UPDATE`).WillReturnRows(sqlmock.NewRows([]string{"id", "organization_id", "category_id", "prefix", "sequence_length", "next_sequence"}).
AddRow(5, 1, 12, "A01", 4, 201))
mock.ExpectQuery(`SELECT material_code FROM materials`).WillReturnRows(sqlmock.NewRows([]string{"material_code"}))
mock.ExpectExec(regexp.QuoteMeta(`UPDATE code_rules SET next_sequence = ? WHERE id = ?`)).WillReturnResult(sqlmock.NewResult(0, 1))
mock.ExpectExec(`INSERT INTO materials`).WillReturnResult(sqlmock.NewResult(0, 0))
// 插入后回查不到 ID(模拟插入未真正落库),persist 据此报错并回滚
mock.ExpectQuery(`SELECT id, material_code FROM materials`).WillReturnRows(sqlmock.NewRows([]string{"id", "material_code"}))
mock.ExpectRollback()
expectImportLog(mock, "物料.xlsx", "failed", 1, 0, 1, sqlmock.AnyArg())
data := buildTestWorkbook(t, [][]string{
{"内六角螺栓", "M8x30", "个", "机械结构类", "标准件"},
})
result, err := NewMaterialImportService().Import(1, "tester", "物料.xlsx", bytes.NewReader(data))
if err == nil {
t.Fatal("期望入库失败")
}
if result != nil {
t.Fatalf("result = %+v, want nil", result)
}
if err := mock.ExpectationsWereMet(); err != nil {
t.Fatal(err)
}
}