dirty and works

Author: medcl

csv2sql.yml

@@ -11,6 +11,13 @@ plugins:
 - name: csv2sql
   enabled: true
 
+- name: persist
+  enabled: true
+  driver: mysql
+  mysql:
+    connection: root:qeephp@tcp(localhost:3306)/ifish?charset=utf8&parseTime=true&loc=Local
+
+
 modules:
 - name: pipeline
   enabled: true
@@ -26,27 +33,33 @@ modules:
           joint: read_csv
           enabled: true
           parameters:
+            file_name: "test/import.xlsx"
             sheet_name: 基础数据导入表
-            data_start_from_index: 4
+            data_start_from_index: 3
             column_name:
-              - outer_code
-              - common_name
-              - scientific_name
-              - english_name
-              - chinese_name
-              - region_name
-              - category_name
-              - is_homemade
-              - aquatic_region_id
-              - inner_code
-              - produce_pattern
-              - feed_pattern
-              - produce_pattern
+            - id
+            - outer_code
+            - common_name
+            - scientific_name
+            - english_name
+            - chinese_name
+            - region_name
+            - aquatic_category_id
+            - category_name
+            - is_homemade
+            - aquatic_region_id
+            - inner_code
+            - produce_pattern
+            - feed_pattern
+            - produce_pattern
             row_format:
-              - INSERT INTO `aquatic_base_info` (`id`, `outer_code`, `common_name`, `scientific_name`, `english_name`, `chinese_name`, `region_name`, `aquatic_category_id`, `created_time`, `updated_time`)
-              - VALUES (<{base_info_id: }>, <{outer_code: }>, <{common_name: }>, <{scientific_name: }>, <{english_name: }>, <{chinese_name: }>, <{region_name: }>, <{aquatic_category_id: }>, <{created_time: }>, <{updated_time: }>);
+            - 'INSERT INTO `aquatic_base_info` (`id`, `outer_code`, `common_name`, `scientific_name`, `english_name`, `chinese_name`, `region_name`, `aquatic_category_id`)'
+            - 'VALUES (<{id: }>, <{outer_code: }>, <{common_name: }>, <{scientific_name: }>, <{english_name: }>, <{chinese_name: }>, <{region_name: }>, <{aquatic_category_id: }>);'
+
+        process:
+        - joint: import_sql
+          enabled: true
 
-        process: []
         error:
           joint: on_error
           enabled: true

executed_sql

@@ -0,0 +1,9 @@
+INSERT INTO `aquatic_base_info` (`id`, `outer_code`, `common_name`, `scientific_name`, `english_name`, `chinese_name`, `region_name`, `aquatic_category_id`)VALUES ('100', '1700829701', '花鲈/鲈鱼/海鲈鱼', 'Lateolabrax japonicus', 'Japanese seabass', '中国花鲈', '', '');
+INSERT INTO `aquatic_base_info` (`id`, `outer_code`, `common_name`, `scientific_name`, `english_name`, `chinese_name`, `region_name`, `aquatic_category_id`)VALUES ('100', '1700829701', '花鲈/鲈鱼/海鲈鱼', 'Lateolabrax japonicus', 'Japanese seabass', '中国花鲈', '', '');
+INSERT INTO `aquatic_base_info` (`id`, `outer_code`, `common_name`, `scientific_name`, `english_name`, `chinese_name`, `region_name`, `aquatic_category_id`)VALUES ('100', '1700829701', '花鲈/鲈鱼/海鲈鱼', 'Lateolabrax japonicus', 'Japanese seabass', '中国花鲈', '', '');
+INSERT INTO `aquatic_base_info` (`id`, `outer_code`, `common_name`, `scientific_name`, `english_name`, `chinese_name`, `region_name`, `aquatic_category_id`)VALUES ('100', '1700829701', '花鲈/鲈鱼/海鲈鱼', 'Lateolabrax japonicus', 'Japanese seabass', '中国花鲈', '', '');
+INSERT INTO `aquatic_base_info` (`id`, `outer_code`, `common_name`, `scientific_name`, `english_name`, `chinese_name`, `region_name`, `aquatic_category_id`)VALUES (100, 1700829701, 花鲈/鲈鱼/海鲈鱼, Lateolabrax japonicus, Japanese seabass, 中国花鲈, , );
+INSERT INTO `aquatic_base_info` (`id`, `outer_code`, `common_name`, `scientific_name`, `english_name`, `chinese_name`, `region_name`, `aquatic_category_id`)VALUES ('100', '1700829701', '花鲈/鲈鱼/海鲈鱼', 'Lateolabrax japonicus', 'Japanese seabass', '中国花鲈', '', '');
+INSERT INTO `aquatic_base_info` (`id`, `outer_code`, `common_name`, `scientific_name`, `english_name`, `chinese_name`, `region_name`, `aquatic_category_id`)VALUES ('100', '1700829701', '花鲈/鲈鱼/海鲈鱼', 'Lateolabrax japonicus', 'Japanese seabass', '中国花鲈', '', '3');
+INSERT INTO `aquatic_base_info` (`id`, `outer_code`, `common_name`, `scientific_name`, `english_name`, `chinese_name`, `region_name`, `aquatic_category_id`)VALUES ('100', '1700829701', '花鲈/鲈鱼/海鲈鱼', 'Lateolabrax japonicus', 'Japanese seabass', '中国花鲈', '', '3');
+INSERT INTO `aquatic_base_info` (`id`, `outer_code`, `common_name`, `scientific_name`, `english_name`, `chinese_name`, `region_name`, `aquatic_category_id`)VALUES ('100', '1700829701', '花鲈/鲈鱼/海鲈鱼', 'Lateolabrax japonicus', 'Japanese seabass', '中国花鲈', '', '3');

main.go

@@ -21,6 +21,8 @@ import (
 	"github.com/infinitbyte/framework"
 	"github.com/infinitbyte/framework/core/module"
 	orm "github.com/infinitbyte/framework/core/persist"
+	q "github.com/infinitbyte/framework/core/queue"
+	pipe "github.com/infinitbyte/framework/core/pipeline"
 	"github.com/infinitbyte/framework/core/util"
 	"github.com/infinitbyte/framework/modules/api"
 	"github.com/infinitbyte/framework/modules/filter"
@@ -72,6 +74,10 @@ func main() {
 
 		orm.RegisterSchema(&model.Request{})
 
+		//manual trigger a pipeline event
+		context := pipe.Context{}
+		q.Push("primary", util.ToJSONBytes(context))
+
 	})
 
 }

pipelines/import_sql.go

@@ -0,0 +1,51 @@
+/*
+Copyright 2016 Medcl (m AT medcl.net)
+
+Licensed under the Apache License, Version 2.0 (the "License");
+you may not use this file except in compliance with the License.
+You may obtain a copy of the License at
+
+   http://www.apache.org/licenses/LICENSE-2.0
+
+Unless required by applicable law or agreed to in writing, software
+distributed under the License is distributed on an "AS IS" BASIS,
+WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+See the License for the specific language governing permissions and
+limitations under the License.
+*/
+
+package pipelines
+
+
+import (
+	"github.com/infinitbyte/framework/core/pipeline"
+	"github.com/infinitbyte/framework/core/persist"
+	log "github.com/cihub/seelog"
+	"github.com/infinitbyte/framework/core/util"
+)
+
+type ImportSQLJoint struct {
+	pipeline.Parameters
+}
+
+func (joint ImportSQLJoint) Name() string {
+	return "import_sql"
+}
+
+func (joint ImportSQLJoint) Process(c *pipeline.Context) error {
+
+	sql:=c.MustGetString(sqlKey)
+
+	log.Info("start execute:",sql)
+
+	util.FileAppendNewLine("executed_sql",sql)
+
+	err:=persist.RawQuery(sql)
+	if(err!=nil){
+		panic(err)
+	}
+
+	log.Info("sql execute success")
+
+	return nil
+}

pipelines/read_csv.go

@@ -22,14 +22,19 @@ import (
 	log "github.com/cihub/seelog"
 	"github.com/infinitbyte/framework/core/pipeline"
 	"strings"
+	"github.com/infinitbyte/framework/core/util"
 )
 
+var fileName pipeline.ParaKey = "file_name"
 var rowFormat pipeline.ParaKey = "row_format"
 var sheetName pipeline.ParaKey = "sheet_name"
 var columnName pipeline.ParaKey = "column_name"
 var dataFromIndex pipeline.ParaKey = "data_start_from_index"
 
+var sqlKey pipeline.ParaKey = "sql"
+
 type ReadCsvJoint struct {
+	pipeline.Parameters
 }
 
 func (joint ReadCsvJoint) Name() string {
@@ -38,22 +43,25 @@ func (joint ReadCsvJoint) Name() string {
 
 func (joint ReadCsvJoint) Process(c *pipeline.Context) error {
 
+	log.Debug(joint.Data)
 	log.Debug(c.Data)
-	templates := c.MustGetStringArray(rowFormat)
+	templates := joint.MustGetStringArray(rowFormat)
 	log.Debug("row templates: ", templates)
 
-	xlsx, err := excelize.OpenFile("../test/Book1.xlsx")
+	xlsx, err := excelize.OpenFile(joint.MustGetString(fileName))
 	if err != nil {
 		log.Error(err)
 		return err
 	}
 	sheetMap := xlsx.GetSheetMap()
 	log.Debug("sheets: ", sheetMap)
 
-	colNames := c.MustGetStringArray(columnName)
-	dataOffset := c.MustGetInt(dataFromIndex)
+	colNames := joint.MustGetStringArray(columnName)
+	dataOffset := joint.MustGetInt(dataFromIndex)
+
+	rows := xlsx.GetRows(joint.MustGetString(sheetName))
 
-	rows := xlsx.GetRows(c.MustGetString(sheetName))
+	sql := ""
 	for offset, row := range rows {
 		if offset < dataOffset {
 			log.Debugf("%v < data offset: %v,　ignore", offset, dataOffset)
@@ -62,28 +70,45 @@ func (joint ReadCsvJoint) Process(c *pipeline.Context) error {
 
 		colMap := map[string]string{}
 
+		hit := false
 		for k, colCell := range row {
+			if colCell != "" {
+				hit = true
+			}
 			colName := colNames[k]
 			colMap[colName] = colCell
-			log.Debug("row:",offset, ": ", colName, "-", k, "-", colCell)
+			log.Trace("row:", offset, ": ", colName, "-", k, "-", colCell)
+		}
+
+		//ignore empty row
+		if !hit {
+			continue
 		}
 
 		for _, x := range templates {
-			line:=x
-			log.Debug("template:",line)
+			line := x
+			//line:=templates
+			log.Debug("template:", line)
 			for k, v := range colMap {
-				log.Debug(fmt.Sprintf("<{%v: }>",k),",", formatString(v))
+				log.Debug(fmt.Sprintf("<{%v: }>", k), ",", formatString(v))
 				line = strings.Replace(line, fmt.Sprintf("<{%v: }>", k), formatString(v), -1)
 			}
 			log.Debug(line)
+			sql = sql + line
+			log.Debug("sql:",sql)
 		}
 	}
 
+	c.Set(sqlKey, sql)
+
+	log.Info(sql)
+
 	return nil
 }
 
 func formatString(str string) string {
 	str = strings.Replace(str, "\"", "", -1)
-	str=fmt.Sprintf("'%s'",str)
+	str = strings.Replace(str, "'", "", -1)
+	str = fmt.Sprintf("'%s'", util.TrimSpaces(str))
 	return str
 }

plugin/plugin.go

@@ -35,6 +35,8 @@ func (module CSV2SQLPlugin) Start(cfg *Config) {
 
 	//register pipeline joints
 	pipeline.RegisterPipeJoint(pipelines.LoggingJoint{})
+	pipeline.RegisterPipeJoint(pipelines.ReadCsvJoint{})
+	pipeline.RegisterPipeJoint(pipelines.ImportSQLJoint{})
 }
 
 func (module CSV2SQLPlugin) Stop() error {