load continue

cgimport.conf

@@ -0,0 +1,8 @@
+
+[cgi]
+datapath=
+parallel=2
+
+[odbc]
+concurrent.limit=2
+

go.mod

@@ -7,7 +7,7 @@ require (
 	github.com/dgryski/go-farm v0.0.0-20200201041132-a6ae2369ad13
 	github.com/scylladb/go-set v1.0.3-0.20200225121959-cc7b2070d91e
 	github.com/spf13/cast v1.7.0
-	github.com/wecisecode/util v0.0.0-20250212093710-d44d3c1dbcac
+	github.com/wecisecode/util v0.0.1
 	modernc.org/sqlite v1.30.1
 )
 

go.sum

@@ -75,8 +75,8 @@ github.com/vmihailenco/msgpack/v5 v5.4.1 h1:cQriyiUvjTwOHg8QZaPihLWeRAAVoCpE00IU
 github.com/vmihailenco/msgpack/v5 v5.4.1/go.mod h1:GaZTsDaehaPpQVyxrf5mtQlH+pc21PIudVV/E3rRQok=
 github.com/vmihailenco/tagparser/v2 v2.0.0 h1:y09buUbR+b5aycVFQs/g70pqKVZNBmxwAhO7/IwNM9g=
 github.com/vmihailenco/tagparser/v2 v2.0.0/go.mod h1:Wri+At7QHww0WTrCBeu4J6bNtoV6mEfg5OIWRZA9qds=
-github.com/wecisecode/util v0.0.0-20250212093710-d44d3c1dbcac h1:4iVHQNUKZCiuYvXdP0xVq5T/KTKJd3StiDEhSxCundI=
-github.com/wecisecode/util v0.0.0-20250212093710-d44d3c1dbcac/go.mod h1:lZZE2bgkzjh4lH+fHuW+tV5dC97Ba6dDS57tsa+ZwYc=
+github.com/wecisecode/util v0.0.1 h1:iDaSQSy0jOCvcfgvM6jU2jRqkfdB21xlu3jBvx69gs0=
+github.com/wecisecode/util v0.0.1/go.mod h1:lZZE2bgkzjh4lH+fHuW+tV5dC97Ba6dDS57tsa+ZwYc=
 github.com/yuin/goldmark v1.1.27/go.mod h1:3hX8gzYuyVAZsxl0MRgGTJEmQBFcNTphYh9decYSb74=
 github.com/yuin/goldmark v1.2.1/go.mod h1:3hX8gzYuyVAZsxl0MRgGTJEmQBFcNTphYh9decYSb74=
 go.etcd.io/etcd/api/v3 v3.5.8 h1:Zf44zJszoU7zRV0X/nStPenegNXoFDWcB/MwrJbA+L4=

importer/cgistatus.go

@@ -12,6 +12,7 @@ import (
 )
 
 type ImportStatus struct {
+	LinesCount   int64
 	RecordsCount int64
 }
 
@@ -59,9 +60,10 @@ func (cgistatus *CGIStatus) WaitSaveDone() {
 func (cgistatus *CGIStatus) Save() (err error) {
 	cgistatus.rc.CallLast2Only(func() {
 		if !cgistatus.lastsavetime.Equal(time.Time{}) {
-			interval := time.Since(cgistatus.lastsavetime)
-			if interval < 1*time.Second {
-				t := time.NewTimer(1*time.Second - interval)
+			interval := 10 * time.Second
+			realinterval := time.Since(cgistatus.lastsavetime)
+			if realinterval < interval {
+				t := time.NewTimer(interval - realinterval)
 				select {
 				case <-t.C:
 				case v := <-cgistatus.waitdone:

importer/datainfo.go

@@ -58,7 +58,10 @@ func (odbci *ODBCImporter) ReviseClassStruct() (err error) {
 	if odbci.client != nil {
 		for _, createedgemql := range schema.CreateEdgeMqls {
 			_, e := odbci.client.Query(createedgemql).Do()
-			if e != nil && !strings.Contains(e.Error(), "already exist") {
+			if e != nil {
+				if strings.Contains(e.Error(), "already exist") {
+					continue
+				}
 				return e
 			}
 			logger.Info(createedgemql)
@@ -67,7 +70,7 @@ func (odbci *ODBCImporter) ReviseClassStruct() (err error) {
 	return
 }
 
-func (odbci *ODBCImporter) reload() error {
+func (odbci *ODBCImporter) rebuild() error {
 	if odbci.client != nil {
 		for i := len(schema.ClassNames) - 1; i >= 0; i-- {
 			classname := schema.ClassNames[i]

importer/importer.go

@@ -37,7 +37,7 @@ type Importer struct {
 	currentstarttime time.Time
 }
 
-func ImportDir(datapath string, parallel int, rebuild, reload bool) (totalfilescount, totalrecordscount int64, totalusetime time.Duration, filescount, recordscount int64, usetime time.Duration, err error) {
+func ImportDir(datapath string, parallel int, rebuild, reload bool) (totalfilescount, totallinecount, totalrecordscount int64, totalusetime time.Duration, filescount, linescount, recordscount int64, usetime time.Duration, err error) {
 	concurlimt := mcfg.GetInt("odbc.concurrent.limit", 100)
 	importer := &Importer{
 		datapath:     datapath,
@@ -52,8 +52,9 @@ func ImportDir(datapath string, parallel int, rebuild, reload bool) (totalfilesc
 	return importer.Import()
 }
 
-func (importer *Importer) Import() (totalfilescount, totalrecordscount int64, totalusetime time.Duration, filescount, recordscount int64, usetime time.Duration, err error) {
+func (importer *Importer) Import() (totalfilescount, totallinecount, totalrecordscount int64, totalusetime time.Duration, filescount, linescount, recordscount int64, usetime time.Duration, err error) {
 	if odbc.DevPhase&odbc.DP_PROCESSCONTINUE != 0 && !importer.reload {
+		// reload
 		err = importer.importstatus.Load()
 		if err != nil {
 			return
@@ -64,9 +65,9 @@ func (importer *Importer) Import() (totalfilescount, totalrecordscount int64, to
 		}
 	}
 	if importer.rebuild {
-		// reload
+		// rebuild
 		// 清除已有类
-		err = importer.odbcimporter.reload()
+		err = importer.odbcimporter.rebuild()
 		if err != nil {
 			return
 		}
@@ -76,36 +77,32 @@ func (importer *Importer) Import() (totalfilescount, totalrecordscount int64, to
 	if err != nil {
 		return
 	}
-	totalfilescount = int64(len(importer.importstatus.ImportStatus))
-	for _, v := range importer.importstatus.ImportStatus {
-		totalrecordscount += v.RecordsCount
-	}
 	totalusetime = importer.importstatus.TotalUseTime
 	importer.starttime = time.Now().Add(-totalusetime)
 	importer.currentstarttime = time.Now()
 
 	reedgefile := regexp.MustCompile("(?i).*edge.*.csv")
-	fc, rc, ut, e := importer.ImportEdgeFiles(reedgefile)
+	efc, elc, erc, ut, e := importer.ImportEdgeFiles(reedgefile, false)
 	if e != nil {
 		err = e
 		return
 	}
-	totalfilescount += fc
-	totalrecordscount += rc
-	filescount += fc
-	recordscount += rc
-	usetime += ut
-	totalusetime = importer.importstatus.TotalUseTime
-	fc, rc, ut, e = importer.ImportNonEdgeFiles(reedgefile)
+	afc, alc, arc, ut, e := importer.ImportNonEdgeFiles(reedgefile, true)
 	if e != nil {
 		err = e
 		return
 	}
-	totalfilescount += fc
-	totalrecordscount += rc
-	filescount += fc
-	recordscount += rc
-	usetime += ut
+	totalfilescount = int64(len(importer.importstatus.ImportStatus)) + efc
+	for _, v := range importer.importstatus.ImportStatus {
+		totallinecount += v.LinesCount
+		totalrecordscount += v.RecordsCount
+	}
+	totallinecount += elc
+	totalrecordscount += erc
+	filescount = afc + efc
+	linescount = alc + elc
+	recordscount = arc + erc
+	usetime = ut
 	totalusetime = importer.importstatus.TotalUseTime
 
 	importer.importstatus.WaitSaveDone()
@@ -113,24 +110,24 @@ func (importer *Importer) Import() (totalfilescount, totalrecordscount int64, to
 	return
 }
 
-func (importer *Importer) ImportEdgeFiles(reedgefile *regexp.Regexp) (filescount, recordscount int64, usetime time.Duration, err error) {
+func (importer *Importer) ImportEdgeFiles(reedgefile *regexp.Regexp, logstatus bool) (filescount, linecount, recordscount int64, usetime time.Duration, err error) {
 	return importer.ImportFiles(func(basedir string, fpath string) FWOP {
 		if !reedgefile.MatchString(filepath.Base(fpath)) {
 			// 忽略非EDGE文件
 			return FWOP_IGNORE
 		}
 		return FWOP_CONTINUE
-	})
+	}, logstatus)
 }
 
-func (importer *Importer) ImportNonEdgeFiles(reedgefile *regexp.Regexp) (filescount, recordscount int64, usetime time.Duration, err error) {
+func (importer *Importer) ImportNonEdgeFiles(reedgefile *regexp.Regexp, logstatus bool) (filescount, linecount, recordscount int64, usetime time.Duration, err error) {
 	return importer.ImportFiles(func(basedir string, fpath string) FWOP {
 		if reedgefile.MatchString(filepath.Base(fpath)) {
 			// 忽略EDGE文件
 			return FWOP_IGNORE
 		}
 		return FWOP_CONTINUE
-	})
+	}, logstatus)
 }
 
 type FWOP int
@@ -141,7 +138,7 @@ const (
 	FWOP_CONTINUE
 )
 
-func (importer *Importer) ImportFiles(fwop func(basedir string, fpath string) FWOP) (filescount, recordscount int64, usetime time.Duration, err error) {
+func (importer *Importer) ImportFiles(fwop func(basedir string, fpath string) FWOP, logstatus bool) (filescount, linescount, recordscount int64, usetime time.Duration, err error) {
 	// 遍历文件目录
 	var wg sync.WaitGroup
 	fw, e := filewalker.NewFileWalker([]string{importer.datapath}, ".*")
@@ -170,30 +167,38 @@ func (importer *Importer) ImportFiles(fwop func(basedir string, fpath string) FW
 		}
 		// 继续处理当前文件
 		filename := filepath.Join(basedir, fpath)
+		filescount++
 		wg.Add(1)
 		// 并发处理
 		importer.fileimportrc.ConcurCall(1,
 			func() {
 				defer wg.Done()
-				logger.Info("import", "file", filename)
 				importer.importstatus.mutex.RLock()
 				importstatus := importer.importstatus.ImportStatus[filename]
 				importer.importstatus.mutex.RUnlock()
-				importedrecordscount := int64(0)
+				linefrom, blockfrom := int64(0), int64(0)
 				if importstatus != nil {
-					importedrecordscount = importstatus.RecordsCount
-					return
+					linefrom, blockfrom = importstatus.LinesCount, importstatus.RecordsCount
 				}
-				records, e := importer.ImportFile(filename, importedrecordscount)
+				if linefrom == 0 {
+					logger.Info("import", "file", filename)
+				} else {
+					logger.Info("import", "file", filename, "from line", linefrom)
+				}
+				lines, records, e := importer.ImportFile(filename, linefrom, blockfrom, logstatus)
 				if e != nil {
 					err = e
 					return
 				}
-				atomic.AddInt64(&filescount, 1)
-				atomic.AddInt64(&recordscount, records)
+				atomic.AddInt64(&linescount, lines-linefrom)
+				atomic.AddInt64(&recordscount, records-blockfrom)
 				usetime = time.Since(importer.currentstarttime)
 				importer.importstatus.mutex.Lock()
-				importer.importstatus.ImportStatus[filename] = &ImportStatus{RecordsCount: importedrecordscount + records}
+				if logstatus {
+					importer.importstatus.ImportStatus[filename] = &ImportStatus{
+						LinesCount:   lines,
+						RecordsCount: records}
+				}
 				importer.importstatus.TotalUseTime = time.Since(importer.starttime)
 				importer.importstatus.mutex.Unlock()
 				importer.importstatus.Save()
@@ -214,16 +219,16 @@ func (importer *Importer) ImportFiles(fwop func(basedir string, fpath string) FW
 	return
 }
 
-func (importer *Importer) ImportFile(filepath string, skiprecordscount int64) (blockcount int64, err error) {
+func (importer *Importer) ImportFile(filepath string, linefrom, blockfrom int64, logstatus bool) (linecount, blockcount int64, err error) {
 	f, e := os.Open(filepath)
 	if e != nil {
-		return blockcount, merrs.NewError(e, merrs.SSMaps{{"filename": filepath}})
+		return linecount, blockcount, merrs.NewError(e, merrs.SSMaps{{"filename": filepath}})
 	}
 	defer f.Close()
-	return importer.importReader(filepath, f, skiprecordscount)
+	return importer.importReader(filepath, f, linefrom, blockfrom, logstatus)
 }
 
-func (importer *Importer) importReader(filename string, buf io.Reader, skiprecordscount int64) (blockcount int64, err error) {
+func (importer *Importer) importReader(filename string, buf io.Reader, linefrom, blockfrom int64, logstatus bool) (linecount, blockcount int64, err error) {
 	var filetype schema.FileType
 	switch {
 	case strings.Contains(filename, "_L1_"):
@@ -252,39 +257,67 @@ func (importer *Importer) importReader(filename string, buf io.Reader, skiprecor
 	}
 	br, e := reader.NewBlockReader(filename, filetype, buf)
 	if e != nil {
-		return blockcount, merrs.NewError(e, merrs.SSMaps{{"filename": filename}})
+		return linecount, blockcount, merrs.NewError(e, merrs.SSMaps{{"filename": filename}})
 	}
+	lastlogtime := time.Now()
+	skiplines := int(linefrom)
+	blockcount = blockfrom
+	doinglines := []int64{}
 	var wg sync.WaitGroup
 	defer importer.done()
 	defer wg.Wait()
-	n := int64(0)
 	for {
 		if err != nil {
 			break
 		}
-		block, line, linecount, e := br.ReadBlock()
+		block, line, linenumber, e := br.ReadBlock(skiplines)
+		linecount = int64(linenumber)
 		if e != nil {
-			return blockcount, merrs.NewError(e, merrs.SSMaps{{"filename": filename}, {"linecount": fmt.Sprint(linecount)}, {"line": line}})
+			return linecount, blockcount, merrs.NewError(e, merrs.SSMaps{{"filename": filename}, {"linecount": fmt.Sprint(linecount)}, {"line": line}})
 		}
 		if block == nil {
 			return
 		}
-		n++
-		if n <= skiprecordscount {
-			continue
-		}
+		blockcount++
 		wg.Add(1)
+		doingline := linecount
+		doingblock := blockcount
+		if logstatus {
+			doinglines = append(doinglines, doingline)
+		}
 		e = importer.odbcqueryrc.ConcurCall(1, func() {
 			defer wg.Done()
-			e = importer.importRecord(block, line, filename, filetype, linecount)
+			e = importer.importRecord(block, line, filename, filetype, int(doingline))
 			if e != nil {
-				err = merrs.NewError(e, merrs.SSMaps{{"filename": filename}, {"linecount": fmt.Sprint(linecount)}, {"line": line}})
+				err = merrs.NewError(e, merrs.SSMaps{{"filename": filename}, {"linecount": fmt.Sprint(doingline)}, {"line": line}})
 				return
 			}
-			atomic.AddInt64(&blockcount, 1)
+			if logstatus {
+				if doingline == doinglines[0] {
+					importer.importstatus.mutex.Lock()
+					importer.importstatus.ImportStatus[filename] = &ImportStatus{
+						LinesCount:   doingline,
+						RecordsCount: doingblock,
+					}
+					importer.importstatus.TotalUseTime = time.Since(importer.starttime)
+					importer.importstatus.Save()
+					doinglines = doinglines[1:]
+					if time.Since(lastlogtime) > 5*time.Second {
+						logger.Info("file", filename, "imported", doingblock, "records")
+						lastlogtime = time.Now()
+					}
+					importer.importstatus.mutex.Unlock()
+				} else {
+					for i, l := range doinglines {
+						if l == doingline {
+							doinglines = append(doinglines[:i], doinglines[i+1:]...)
+						}
+					}
+				}
+			}
 		})
 		if e != nil {
-			return blockcount, merrs.NewError(e, merrs.SSMaps{{"filename": filename}, {"linecount": fmt.Sprint(linecount)}, {"line": line}})
+			return linecount, blockcount, merrs.NewError(e, merrs.SSMaps{{"filename": filename}, {"linecount": fmt.Sprint(linecount)}, {"line": line}})
 		}
 	}
 	return

main.go

@@ -6,6 +6,7 @@ import (
 
 	"git.wecise.com/wecise/cgimport/importer"
 	"git.wecise.com/wecise/cgimport/odbc"
+	"github.com/wecisecode/util/mfmt"
 )
 
 // 获取配置信息
@@ -43,8 +44,9 @@ func main() {
 	logger.Info("datapath:   ", datapath)
 	logger.Info("parallel:   ", parallel)
 	logger.Info("reload:     ", reload)
+	logger.Info("rebuild:    ", rebuild)
 	// 导入
-	totalfilescount, totalrecordscount, totalusetime, filescount, recordscount, usetime, e := importer.ImportDir(datapath, parallel, rebuild, reload)
+	totalfilescount, totallinecount, totalrecordscount, totalusetime, filescount, linescount, recordscount, usetime, e := importer.ImportDir(datapath, parallel, rebuild, reload)
 	if e != nil {
 		logger.Error(e)
 		panic(e)
@@ -54,11 +56,13 @@ func main() {
 		return
 	}
 	// 输出统计信息
-	logger.Info("import", filescount, "files", recordscount, "records", "in", usetime)
-	logger.Info("total import", totalfilescount, "files", totalrecordscount, "records", "in", totalusetime)
+	logger.Info("import", filescount, "files", linescount, "lines", recordscount, "records", "in", mfmt.FormatDuration(usetime))
+	logger.Info("total import", totalfilescount, "files", totallinecount, "lines", totalrecordscount, "records", "in", mfmt.FormatDuration(totalusetime))
 	fmt.Println("access", odbc.LogFile, "for detail information")
 
 	// 验证
-	importer.Check()
+	if odbc.ODBCDebug || odbc.LogDebug {
+		importer.Check()
+	}
 	os.Exit(0)
 }

odbc/odbclient.go

@@ -15,7 +15,7 @@ var ODBError error
 
 var ODBServerPath string
 var Keyspace string
-var Debug bool
+var ODBCDebug bool
 
 var default_keyspace = `oktest`
 var default_odbpaths = `127.0.0.1:11001`
@@ -36,7 +36,6 @@ func Usage() string {
 命令行参数：` + CommandArgsInfo() + `
 odbpath=` + default_odbpaths + `    # 指定odbserver路径，默认通过环境变量ODBPATH或通过ETCD相关配置获取
 keyspace=` + default_keyspace + `    # 指定keyspace，默认通过环境变量KEYSPACE获取
-debug=true         # 开启调试模式，输出更多信息
 	
 环境变量需求：
 ODBPATH=` + default_odbpaths + `    # 指定odbserver路径，默认通过配置信息 odbc.odbpath 获取
@@ -63,7 +62,6 @@ keyspace=` + default_keyspace + `
 func LogConfigInfo() {
 	Logger.Info("odbpath:    ", ODBClient.Config().Hosts)
 	Logger.Info("keyspace:   ", ODBClient.Config().Keyspace)
-	Logger.Info("debug:      ", Debug)
 }
 
 func config_merge(a *odb.Config, b *odb.Config) *odb.Config {
@@ -91,14 +89,9 @@ func config_merge(a *odb.Config, b *odb.Config) *odb.Config {
 	return a
 }
 
-// 参数设置为程序默认配置
-// 可以通过命令行，环境变量 或 与应用同名的.conf配置文件 设置
-// 远程连接需要在白名单中增加本地IP
-func ODBC(odbcfgs ...*odb.Config) odb.Client {
-	odbcfg := config_merge(&odb.Config{}, default_config)
-	for _, c := range odbcfgs {
-		odbcfg = config_merge(odbcfg, c)
-	}
+var odbcfg = default_config
+
+func init() {
 	default_keyspace = odbcfg.Keyspace
 	default_odbpaths = strings.Join(odbcfg.Hosts, ",")
 	odbpaths := strset.New(strings.Split(
@@ -108,12 +101,21 @@ func ODBC(odbcfgs ...*odb.Config) odb.Client {
 	Keyspace = ucfg.CommandArgs.GetString("keyspace",
 		ucfg.Environs.GetString("KEYSPACE",
 			Config.GetString("odbc.keyspace", default_keyspace)))
-	Debug = ucfg.CommandArgs.GetBool("debug", false)
-	ODBClient, ODBError = odb.NewClient(config_merge(odbcfg, &odb.Config{
+	ODBCDebug = Config.GetBool("odbc.debug", false) || Config.GetString("odbc.debug") == "odbc.debug"
+	odbcfg = config_merge(odbcfg, &odb.Config{
 		Keyspace: Keyspace,
 		Hosts:    odbpaths,
-		Debug:    Debug,
-	}))
+		Debug:    ODBCDebug,
+	})
+	ODBServerPath = "[" + strings.Join(odbcfg.Hosts, ",") + "]"
+	Keyspace = odbcfg.Keyspace
+}
+
+// 可以通过命令行，环境变量 或 与应用同名的 .conf 配置文件 设置
+// 远程连接需要在白名单中增加本地IP
+// 此函数仅用于示范和简化 与 odbserver 建立连接的过程，建议应用根据实际使用场景调用 odb.NewClient
+func ODBC() odb.Client {
+	ODBClient, ODBError = odb.NewClient(odbcfg)
 	if ODBError != nil {
 		fmt.Print(Usage())
 		if strings.Contains(ODBError.Error(), "error: EOF") {
@@ -122,8 +124,6 @@ func ODBC(odbcfgs ...*odb.Config) odb.Client {
 		}
 		panic(ODBError)
 	}
-	ODBServerPath = "[" + strings.Join(ODBClient.Config().Hosts, ",") + "]"
-	Keyspace = ODBClient.Config().Keyspace
 	LogConfigInfo()
 	return ODBClient
 }

reader/blockreader.go

@@ -13,7 +13,7 @@ var mcfg = odbc.Config
 var logger = odbc.Logger
 
 type BlockReader interface {
-	ReadBlock() (block map[string]any, line string, linecount int, err error)
+	ReadBlock(skiplines int) (block map[string]any, line string, linecount int, err error)
 }
 
 func NewBlockReader(filename string, filetype schema.FileType, reader io.Reader) (BlockReader, error) {

reader/csvreader.go

@@ -24,7 +24,7 @@ func NewCSVBlockReader(filename string, filetype schema.FileType, reader io.Read
 	}
 }
 
-func (br *CSVBlockReader) ReadBlock() (block map[string]any, line string, linecount int, err error) {
+func (br *CSVBlockReader) ReadBlock(skiplines int) (block map[string]any, line string, linecount int, err error) {
 	classname := string(br.filetype)
 	ci := schema.ClassInfos.GetIFPresent(classname)
 	eof := false
@@ -46,6 +46,9 @@ func (br *CSVBlockReader) ReadBlock() (block map[string]any, line string, lineco
 			}
 			continue
 		}
+		if linecount <= skiplines {
+			continue
+		}
 		values := strings.Split(line, "^")
 		if len(values) != len(br.csvkeys) {
 			err = merrs.NewError(fmt.Sprint(br.filename, " format error, values count not match keys count, line ", br.linecount))

reader/txtreader.go

@@ -26,7 +26,7 @@ func NewTXTBlockReader(filename string, filetype schema.FileType, reader io.Read
 var regrecordstart = regexp.MustCompile(`^(?:[\.\/a-zA-Z0-9_]*:)?V:(\{.*)`)
 var regrecordend = regexp.MustCompile(`\}\s*$`)
 
-func (br *TXTBlockReader) ReadBlock() (block map[string]any, line string, linecount int, err error) {
+func (br *TXTBlockReader) ReadBlock(skiplines int) (block map[string]any, line string, linecount int, err error) {
 	eof := false
 	line = br.nextline
 	for {
@@ -41,6 +41,9 @@ func (br *TXTBlockReader) ReadBlock() (block map[string]any, line string, lineco
 				}
 				break
 			}
+			if linecount <= skiplines {
+				continue
+			}
 			if regrecordend.MatchString(line) {
 				break
 			}