cgimport/importer/datainfo.go

package importer

import (
	"regexp"
	"strings"
	"sync"
	"time"

	"git.wecise.com/wecise/cgimport/schema"
	"git.wecise.com/wecise/util/cmap"
	"github.com/wecisecode/util/merrs"
)

type classdatainfo struct {
	*schema.ClassInfo
	insertcount   int64
	lastlogtime   time.Time
	lastlogicount int64
	mutex         sync.Mutex
}

var classdatainfos = cmap.NewSingle[string, *classdatainfo]()

// 根据数据修正类定义
func (odbci *ODBCImporter) ReviseClassStruct() (err error) {
	for _, classname := range schema.ClassNames {
		ci := schema.ClassInfos.GetIFPresent(classname)
		if ci == nil {
			return merrs.NewError("classinfo not found " + classname)
		}
		cdi, e := classdatainfos.GetWithNew(ci.Classaliasname, func() (cdi *classdatainfo, err error) {
			if odbci.client != nil {
				_, e := odbci.client.Query("select class,id from " + ci.Classfullname + " limit 1").Do()
				if e != nil {
					if !strings.Contains(e.Error(), "not find") {
						return nil, e
					}
					logger.Info("create class " + ci.Classfullname)
					_, e = odbci.client.Query(ci.Createmql).Do()
					if e != nil {
						return nil, e
					}
					// add graph tags
					_, e = odbci.client.Query(ci.Addtagmql, ci.Classaliasname, ci.Classaliasname, []string{ci.Classaliasname}).Do()
					if e != nil {
						return nil, e
					}
				}
			}
			cdi = &classdatainfo{ClassInfo: ci}
			return
		})
		if e != nil {
			return e
		}
		classdatainfos.Set(ci.Classfullname, cdi)
	}
	if odbci.client != nil {
		for _, createedgemql := range schema.CreateEdgeMqls {
			_, e := odbci.client.Query(createedgemql).Do()
			if e != nil && !strings.Contains(e.Error(), "already exist") {
				return e
			}
		}
	}
	return
}

func (odbci *ODBCImporter) reload() error {
	if odbci.client != nil {
		for i := len(schema.ClassNames) - 1; i >= 0; i-- {
			classname := schema.ClassNames[i]
			ci := schema.ClassInfos.GetIFPresent(classname)
			if ci == nil {
				continue
			}
			e := odbci.dropclass(ci.Classfullname)
			if e != nil {
				return e
			}
		}
		e := odbci.InitLocalDB(true)
		if e != nil {
			return e
		}
	}
	return nil
}

func (odbci *ODBCImporter) init() error {
	e := odbci.InitLocalDB(false)
	if e != nil {
		return e
	}
	return nil
}

func (odbci *ODBCImporter) dropclass(classnames ...string) error {
	for _, classname := range classnames {
		for retry := 2; retry >= 0; retry-- {
			_, e := odbci.client.Query(`delete from /matrix/tagdir where tags='` + classname + `'`).Do()
			_ = e
			_, e = odbci.client.Query(`delete from "` + classname + `" with version`).Do()
			_ = e
			_, e = odbci.client.Query(`drop class if exists "` + classname + `"`).Do()
			if e != nil {
				matchstr := regexp.MustCompile(`refer by ([^,]+)`).FindStringSubmatch(e.Error())
				if len(matchstr) >= 2 {
					e = odbci.dropclass(matchstr[1])
					if e != nil {
						return e
					}
				} else {
					matchstr := regexp.MustCompile(`has children \[([^\]]+)\]`).FindStringSubmatch(e.Error())
					if len(matchstr) >= 2 {
						e = odbci.dropclass(strings.Split(matchstr[1], ",")...)
						if e != nil {
							return e
						}
					}
				}
				if retry > 0 {
					continue
				}
				return e
			}
		}
		logger.Info("drop class " + classname)
	}
	return nil
}