+ Importer Run() & Chunk

tribe-health · Jan 17, 2022 · 4983712 · 4983712
1 parent 23ffcf1
commit 4983712
Show file tree

Hide file tree

Showing 9 changed files with 153 additions and 36 deletions.
diff --git a/importer/from/source.go b/importer/from/source.go
@@ -19,8 +19,8 @@ const (
 type Source interface {
 	Data(row int, size int, cols []int) [][]interface{}
 	Columns() []Column
+	Chunk(size int, cols []int, cb func(line int, data [][]interface{}))
 	Inspect() Inspect
-	Bind()
 	Close() error
 }
 

diff --git a/importer/importer.go b/importer/importer.go
@@ -69,7 +69,7 @@ func (imp *Importer) AutoMapping(src from.Source) *Mapping {
 		if !ok {
 			continue
 		}
-		binding := Binding{Name: "", Axis: "", Rules: []string{}, Field: name, Label: imp.Columns[i].Label}
+		binding := Binding{Name: "", Axis: "", Col: -1, Rules: []string{}, Field: name, Label: imp.Columns[i].Label}
 		for _, suggest := range imp.Columns[i].Match {
 			if srcCol, has := sourceColumns[suggest]; has {
 				binding = Binding{
@@ -89,7 +89,7 @@ func (imp *Importer) AutoMapping(src from.Source) *Mapping {
 	return mapping
 }
 
-// DataGet 预览数据
+// DataGet 读取源数据记录
 func (imp *Importer) DataGet(src from.Source, page int, size int, mapping *Mapping) ([]string, [][]interface{}) {
 
 	row := (page-1)*size + mapping.RowStart
@@ -104,6 +104,15 @@ func (imp *Importer) DataGet(src from.Source, page int, size int, mapping *Mappi
 	return imp.DataClean(data, mapping.Columns)
 }
 
+// Chunk 遍历数据
+func (imp *Importer) Chunk(src from.Source, mapping *Mapping, cb func(line int, data [][]interface{})) {
+	cols := []int{}
+	for _, d := range mapping.Columns {
+		cols = append(cols, d.Col)
+	}
+	src.Chunk(imp.Option.ChunkSize, cols, cb)
+}
+
 // DataClean 清洗数据
 func (imp *Importer) DataClean(data [][]interface{}, bindings []*Binding) ([]string, [][]interface{}) {
 	columns := []string{}
@@ -189,7 +198,47 @@ func (imp *Importer) SaveAsTemplate(src from.Source) {
 }
 
 // Run 运行导入
-func (imp *Importer) Run() {}
+func (imp *Importer) Run(src from.Source, mapping *Mapping) map[string]int {
+	if mapping == nil {
+		mapping = imp.AutoMapping(src)
+	}
+
+	total := 0
+	failed := 0
+	imp.Chunk(src, mapping, func(line int, data [][]interface{}) {
+		length := len(data)
+		total = total + length
+		columns, data := imp.DataClean(data, mapping.Columns)
+		process, err := gou.ProcessOf(imp.Process, columns, data)
+		if err != nil {
+			failed = failed + length
+			xlog.Printf("导入失败 %d %s ", line, err.Error())
+			return
+		}
+
+		response, err := process.Exec()
+		if err != nil {
+			failed = failed + length
+			xlog.Printf("导入失败 %d %s ", line, err.Error())
+			return
+		}
+
+		if res, ok := response.(int); ok {
+			failed = failed + res
+			return
+		} else if res, ok := response.(int64); ok {
+			failed = failed + int(res)
+			return
+		}
+
+		xlog.Printf("导入处理器未返回失败结果 %#v %d %d", response, line, length)
+	})
+	return map[string]int{
+		"success": total - failed,
+		"failure": failed,
+		"total":   total,
+	}
+}
 
 // Start 运行导入(异步)
 func (imp *Importer) Start() {}

diff --git a/importer/importer_test.go b/importer/importer_test.go
@@ -40,7 +40,9 @@ func TestAutoMappingSimple(t *testing.T) {
 		assert.Equal(t, dst["name"], col.Field)
 		assert.Equal(t, dst["label"], col.Label)
 		assert.Equal(t, dst["rules"], col.Rules)
-		assert.NotEmpty(t, col.Axis)
+		if col.Field != "remark" {
+			assert.NotEmpty(t, col.Axis)
+		}
 	}
 }
 
@@ -59,8 +61,35 @@ func TestDataGetSimple(t *testing.T) {
 	}, columns)
 
 	assert.Equal(t, [][]interface{}{
-		{"SN202101120018", "张三", "男", "26", "13211000011", "彩绘湖北地图", "3", "65.5", "196.5", "", true},
-		{"", "李四", "男", "42", "13211000011", "景祐遁甲符应经", "1", "34.8", "34.8", "", false},
+		{"SN202101120018", "张三", "男", "26", "13211000011", "彩绘湖北地图", "3", "65.5", "196.5", "自动添加备注 @From 张三", true},
+		{"", "李四", "男", "42", "13211000011", "景祐遁甲符应经", "1", "34.8", "34.8", "自动添加备注 @From 李四", false},
 	}, data)
 
 }
+
+func TestDataChunkSimple(t *testing.T) {
+	simple := filepath.Join(config.Conf.Root, "imports", "assets", "simple.xlsx")
+	file := xlsx.Open(simple)
+	defer file.Close()
+
+	imp := Select("order")
+	mapping := imp.AutoMapping(file)
+	lines := []int{}
+	imp.Chunk(file, mapping, func(line int, data [][]interface{}) {
+		lines = append(lines, line)
+	})
+	assert.Equal(t, []int{3, 4}, lines)
+}
+
+func TestRunSimple(t *testing.T) {
+	simple := filepath.Join(config.Conf.Root, "imports", "assets", "simple.xlsx")
+	file := xlsx.Open(simple)
+	defer file.Close()
+
+	imp := Select("order")
+	mapping := imp.AutoMapping(file)
+	res := imp.Run(file, mapping)
+	assert.Equal(t, 2, res["failure"])
+	assert.Equal(t, 2, res["success"])
+	assert.Equal(t, 4, res["total"])
+}
diff --git a/importer/option.go b/importer/option.go
@@ -25,14 +25,14 @@ func (option *Option) UnmarshalJSON(source []byte) error {
 // OptionOf 解析配置
 func OptionOf(data map[string]interface{}) (*Option, error) {
 	option := &Option{
-		AutoMatching:   true,
+		UseTemplate:    true,
 		ChunkSize:      500,
 		MappingPreview: PreviewAuto,
 		DataPreview:    PreviewAuto,
 	}
 
-	if autoMatching, ok := data["autoMatching"].(bool); ok {
-		option.AutoMatching = autoMatching
+	if autoMatching, ok := data["useTemplate"].(bool); ok {
+		option.UseTemplate = autoMatching
 	}
 
 	chunkSize := any.Of(data["chunkSize"]).CInt()

diff --git a/importer/option_test.go b/importer/option_test.go
@@ -22,15 +22,15 @@ func TestOptionUnmarshalJSON(t *testing.T) {
 	var normal Option
 	err := jsoniter.Unmarshal(testDataOption["normal"], &normal)
 	assert.Nil(t, err)
-	assert.Equal(t, true, normal.AutoMatching)
+	assert.Equal(t, true, normal.UseTemplate)
 	assert.Equal(t, 200, normal.ChunkSize)
 	assert.Equal(t, "always", normal.MappingPreview)
 	assert.Equal(t, "never", normal.DataPreview)
 
 	var defaults Option
 	err = jsoniter.Unmarshal(testDataOption["defaults"], &defaults)
 	assert.Nil(t, err)
-	assert.Equal(t, true, defaults.AutoMatching)
+	assert.Equal(t, true, defaults.UseTemplate)
 	assert.Equal(t, 500, defaults.ChunkSize)
 	assert.Equal(t, "auto", defaults.MappingPreview)
 	assert.Equal(t, "auto", defaults.DataPreview)

diff --git a/importer/types.go b/importer/types.go
@@ -33,7 +33,7 @@ type Column struct {
 
 // Option 导入配置项定
 type Option struct {
-	AutoMatching   bool   `json:"autoMatching,omitempty"`   // 是否自动匹配已设定模板
+	UseTemplate    bool   `json:"useTemplate,omitempty"`    // 使用已匹配过的模板
 	ChunkSize      int    `json:"chunkSize,omitempty"`      // 每次处理记录数量
 	MappingPreview string `json:"mappingPreview,omitempty"` // 显示字段映射界面方式 auto 匹配模板失败显示, always 一直显示, never 不显示
 	DataPreview    string `json:"dataPreview,omitempty"`    // 数据预览界面方式 auto 有异常数据时显示, always 一直显示, never 不显示

diff --git a/importer/xlsx/xlsx.go b/importer/xlsx/xlsx.go
@@ -73,27 +73,65 @@ func (xlsx *Xlsx) Inspect() from.Inspect {
 // Data 读取数据
 func (xlsx *Xlsx) Data(row int, size int, cols []int) [][]interface{} {
 	data := [][]interface{}{}
-	for r := row; r < row+size; r++ {
-		row := []interface{}{}
-		end := true
-		for _, c := range cols {
-			axis := positionToAxis(r, c)
-			value, err := xlsx.File.GetCellValue(xlsx.SheetName, axis)
+	for line := row; line < row+size; line++ {
+		row, end := xlsx.readLine(line, cols)
+		if end {
+			break
+		}
+		data = append(data, row)
+	}
+	return data
+}
+
+// Chunk 遍历数据
+func (xlsx *Xlsx) Chunk(size int, cols []int, cb func(line int, data [][]interface{})) {
+	line := 0
+	data := [][]interface{}{}
+	for xlsx.Rows.Next() {
+		line++
+		if line < xlsx.RowStart {
+			continue
+		}
+		row, end := xlsx.readLine(line, cols)
+		if end {
+			cb(line, data)
+			break
+		}
+
+		data = append(data, row)
+		if line%size == 0 {
+			cb(line, data)
+			data = [][]interface{}{}
+		}
+	}
+
+	// 最后一批数据
+	if len(data) > 0 {
+		cb(line, data)
+	}
+}
+
+// readLine 读取给定行信息
+func (xlsx *Xlsx) readLine(line int, cols []int) ([]interface{}, bool) {
+	row := []interface{}{}
+	end := true
+	for _, c := range cols {
+		var value = ""
+		var err error
+		if c >= 0 {
+			axis := positionToAxis(line, c)
+			value, err = xlsx.File.GetCellValue(xlsx.SheetName, axis)
 			if err != nil {
 				xlog.Printf("读取数据出错 %s %s %s", xlsx.SheetName, axis, err.Error())
 				value = ""
 			}
-			row = append(row, value)
-			if value != "" {
-				end = false
-			}
 		}
-		if end {
-			break
+		row = append(row, value)
+		if value != "" {
+			end = false
 		}
-		data = append(data, row)
 	}
-	return data
+	return row, end
 }
 
 // Columns 读取列
@@ -142,10 +180,6 @@ func (xlsx *Xlsx) Columns() []from.Column {
 	return columns
 }
 
-// Bind 绑定映射表
-func (xlsx *Xlsx) Bind() {
-}
-
 func (xlsx *Xlsx) getMergeCells() {
 	cells, err := xlsx.File.GetMergeCells(xlsx.SheetName)
 	if err != nil {

diff --git a/tests/imports/order.imp.json b/tests/imports/order.imp.json
@@ -1,6 +1,6 @@
 {
   "title": "导入订单数据",
-  "process": "flows.order.Import",
+  "process": "scripts.rules.ImportData",
   "columns": [
     {
       "label": "订单号",
@@ -20,14 +20,14 @@
       "label": "性别",
       "name": "user.sex",
       "match": ["性别", "gender", "sex"],
-      "rules": ["scripts.rules.FmtUser"],
+      "rules": [],
       "nullable": true
     },
     {
       "label": "年龄",
       "name": "user.age",
       "match": ["年龄", "age"],
-      "rules": ["scripts.rules.FmtUser"],
+      "rules": [],
       "nullable": true
     },
     {
@@ -63,12 +63,12 @@
     {
       "label": "备注",
       "name": "remark",
-      "match": ["备注", "说明", "其他"]
+      "match": []
     }
   ],
   "option": {
     "autoMatching": true,
-    "chunkSize": 500,
+    "chunkSize": 3,
     "logging": "error",
     "mappingPreview": "auto",
     "dataPreview": "auto"

diff --git a/tests/libs/rules.js b/tests/libs/rules.js
@@ -9,6 +9,7 @@ function order_sn(value, row) {
 }
 
 function FmtUser(value, row) {
+  row[9] = "自动添加备注 @From " + value;
   return row;
 }
 
@@ -19,3 +20,7 @@ function FmtGoods(value, row) {
 function mobile(value, row) {
   return row;
 }
+
+function ImportData(columns, data) {
+  return 1;
+}