From 4983712c8a1b8aa50e6ba109f2dec20a927a480b Mon Sep 17 00:00:00 2001 From: Max Date: Mon, 17 Jan 2022 22:51:56 +0800 Subject: [PATCH] + Importer Run() & Chunk --- importer/from/source.go | 2 +- importer/importer.go | 55 +++++++++++++++++++++++++-- importer/importer_test.go | 35 ++++++++++++++++-- importer/option.go | 6 +-- importer/option_test.go | 4 +- importer/types.go | 2 +- importer/xlsx/xlsx.go | 72 ++++++++++++++++++++++++++---------- tests/imports/order.imp.json | 10 ++--- tests/libs/rules.js | 5 +++ 9 files changed, 154 insertions(+), 37 deletions(-) diff --git a/importer/from/source.go b/importer/from/source.go index f9f0bcee..b5a56e76 100644 --- 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 index 508a3d94..ad77f32a 100644 --- 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 index 9dff36a1..85381f02 100644 --- 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 index 8237e8aa..1b8d486f 100644 --- 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 index 595ab8eb..f170b9b5 100644 --- a/importer/option_test.go +++ b/importer/option_test.go @@ -22,7 +22,7 @@ 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) @@ -30,7 +30,7 @@ func TestOptionUnmarshalJSON(t *testing.T) { 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 index d9782556..efb9f4f8 100644 --- 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 index 8d35c6eb..dfe0db2c 100644 --- a/importer/xlsx/xlsx.go +++ b/importer/xlsx/xlsx.go @@ -73,21 +73,8 @@ 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) - if err != nil { - xlog.Printf("读取数据出错 %s %s %s", xlsx.SheetName, axis, err.Error()) - value = "" - } - row = append(row, value) - if value != "" { - end = false - } - } + for line := row; line < row+size; line++ { + row, end := xlsx.readLine(line, cols) if end { break } @@ -96,6 +83,57 @@ func (xlsx *Xlsx) Data(row int, size int, cols []int) [][]interface{} { 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 + } + } + return row, end +} + // Columns 读取列 func (xlsx *Xlsx) Columns() []from.Column { columns := []from.Column{} @@ -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 index 794fbf14..7c561c3d 100644 --- 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 index e78b1012..e247713f 100644 --- 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; +}