eta_task/models/data_manage/edb_source.go

package data_manage

import (
	"eta_gn/eta_task/global"
	"eta_gn/eta_task/utils"
	"fmt"
	"strings"
)

var (
	EdbDataTableNameMap     map[int]string        // 指标来源对应数据表名
	EdbDataRefreshMethodMap map[int]string        // 指标来源对应的刷新指标方法
	EdbTableNameSourceMap   map[string]*EdbSource // 数据表名对应的指标来源
	EdbSourceIdMap          map[int]*EdbSource    // 指标来源
	EdbSourceExtendIdMap    map[string]int        // 指标来源字符串对应来源ID
	EdbNameSourceMap        map[string]*EdbSource // 数据来源名对应的指标来源
)

// EdbSource 指标来源表
type EdbSource struct {
	EdbSourceId      int    `gorm:"column:edb_source_id;primaryKey"` // `orm:"column(edb_source_id);pk"`
	SourceName       string `description:"指标来源名称"`
	TableName        string `description:"数据表名"`
	EdbAddMethod     string `description:"指标新增接口"`
	EdbRefreshMethod string `description:"指标刷新接口"`
	IsBase           int    `description:"是否为基础指标: 0-否; 1-是"`
	FromBridge       int    `description:"是否来源于桥接服务: 0-否; 1-是"`
	BridgeFlag       string `description:"桥接服务对象标识"`
	SourceExtend     string `description:"扩展字段做查询用"`
	EdbCodeRequired  int32  `gorm:"column:edb_code_required;type:tinyint(4);comment:指标编码是否必填: 0-否; 1-是;not null;default:0;"` // 指标编码是否必填: 0-否; 1-是
	IndexTableName   string `gorm:"column:index_table_name;type:varchar(128);comment:源指标表名;not null;"`                       // 源指标表名
	SourceNameEn     string `gorm:"column:source_name_en;type:varchar(128);comment:指标来源名称-英文;not null;"`                     // 指标来源名称-英文
}

// GetEdbSourceItemsByCondition 获取指标来源列表
func GetEdbSourceItemsByCondition(condition string, pars []interface{}, fieldArr []string, orderRule string) (items []*EdbSource, err error) {
	//o := orm.NewOrmUsingDB("data")
	fields := strings.Join(fieldArr, ",")
	if len(fieldArr) == 0 {
		fields = `*`
	}
	order := `ORDER BY edb_source_id ASC`
	if orderRule != "" {
		order = ` ORDER BY ` + orderRule
	}
	sql := fmt.Sprintf(`SELECT %s FROM edb_source WHERE 1=1 %s %s`, fields, condition, order)

	//_, err = o.Raw(sql, pars).QueryRows(&items)
	err = global.DmSQL["data"].Raw(sql, pars...).Find(&items).Error
	return
}

// GetEdbSourceItemByCondition 获取指标来源
func GetEdbSourceItemByCondition(condition string, pars []interface{}) (item *EdbSource, err error) {
	//o := orm.NewOrmUsingDB("data")
	sql := fmt.Sprintf(`SELECT * FROM edb_source WHERE 1=1 %s`, condition)
	//err = o.Raw(sql, pars).QueryRow(&item)

	err = global.DmSQL["data"].Raw(sql, pars...).First(&item).Error
	return
}

// InitEdbSourceVar 初始化时加载指标来源对应信息, 避免循环中查库, 注意edb_source表修改table_name的话需要重启服务
func InitEdbSourceVar() {
	EdbDataTableNameMap = make(map[int]string)
	EdbDataRefreshMethodMap = make(map[int]string)
	EdbTableNameSourceMap = make(map[string]*EdbSource)
	EdbNameSourceMap = make(map[string]*EdbSource)
	EdbSourceIdMap = make(map[int]*EdbSource)
	EdbSourceExtendIdMap = make(map[string]int)
	sources, e := GetEdbSourceItemsByCondition(``, make([]interface{}, 0), []string{}, "")
	if e != nil {
		utils.FileLog.Info("init source table err: %s", e.Error())
		return
	}
	for _, v := range sources {
		EdbDataTableNameMap[v.EdbSourceId] = v.TableName
		EdbDataRefreshMethodMap[v.EdbSourceId] = v.EdbRefreshMethod
		EdbTableNameSourceMap[v.TableName] = v
		EdbNameSourceMap[v.SourceName] = v

		EdbSourceIdMap[v.EdbSourceId] = v
		if v.SourceExtend != "" {
			arr := strings.Split(v.SourceExtend, ",")
			if len(arr) == 0 {
				continue
			}
			for _, s := range arr {
				EdbSourceExtendIdMap[s] = v.EdbSourceId
			}
		}
	}
}

// GetEdbSourceItemsSourceId
// @Description: 根据来源id获取指标来源
// @param sourceId
// @return item
// @return err
func GetEdbSourceItemsSourceId(sourceId int) (item *EdbSource, err error) {
	sql := `SELECT * FROM edb_source WHERE 1=1 AND edb_source_id = ? `
	err = global.DmSQL["data"].Raw(sql, sourceId).First(&item).Error

	return
}

// GetEdbSourceBySourceId
// @Description: 根据来源id获取指标来源
// @param sourceId
// @return sourceItem
func GetEdbSourceBySourceId(sourceId int) (sourceItem *EdbSource) {
	sourceItem, ok := EdbSourceIdMap[sourceId]
	if !ok {
		item, err := GetEdbSourceItemsSourceId(sourceId)
		if err != nil {
			return
		}
		if item.EdbSourceId > 0 {
			sourceItem = item

			// 写入到内存中
			EdbSourceIdMap[sourceId] = sourceItem
			EdbDataTableNameMap[sourceId] = sourceItem.TableName
			EdbDataRefreshMethodMap[sourceItem.EdbSourceId] = sourceItem.EdbRefreshMethod
			EdbTableNameSourceMap[sourceItem.TableName] = sourceItem
			EdbNameSourceMap[sourceItem.SourceName] = sourceItem
		}
	}

	return
}

// GetEdbSourceItemsSourceName
// @Description: 根据来源名称获取指标来源
// @param sourceName
// @return item
// @return err
func GetEdbSourceItemsSourceName(sourceName string) (item *EdbSource, err error) {
	sql := `SELECT * FROM edb_source WHERE 1=1 AND source_name = ? `
	err = global.DmSQL["data"].Raw(sql, sourceName).First(&item).Error

	return
}

// GetEdbSourceBySourceName
// @Description: 根据来源名称获取指标来源
// @param sourceId
// @return sourceItem
func GetEdbSourceBySourceName(sourceName string) (sourceItem *EdbSource) {
	sourceItem, ok := EdbNameSourceMap[sourceName]
	if !ok {
		item, err := GetEdbSourceItemsSourceName(sourceName)
		if err != nil {
			return
		}
		if item.EdbSourceId > 0 {
			sourceItem = item

			// 写入到内存中
			EdbSourceIdMap[sourceItem.EdbSourceId] = sourceItem
			EdbDataTableNameMap[sourceItem.EdbSourceId] = sourceItem.TableName
			EdbDataRefreshMethodMap[sourceItem.EdbSourceId] = sourceItem.EdbRefreshMethod
			EdbTableNameSourceMap[sourceItem.TableName] = sourceItem
			EdbNameSourceMap[sourceItem.SourceName] = sourceItem
		}
	}

	return
}

// GetEdbSourceTableNameBySourceId
// @Description: 根据来源id获取指标来源的归属表
// @param sourceId
// @return sourceItem
func GetEdbSourceTableNameBySourceId(sourceId int) (tableName string) {
	sourceItem := GetEdbSourceBySourceId(sourceId)

	if sourceItem != nil {
		tableName = sourceItem.TableName
	}

	return
}

// GetEdbSourceRefreshMethodBySourceId
// @Description: 根据来源id获取指标来源的刷新路径
// @param sourceId
// @return sourceItem
func GetEdbSourceRefreshMethodBySourceId(sourceId int) (refreshMethod string) {
	sourceItem := GetEdbSourceBySourceId(sourceId)

	if sourceItem != nil {
		refreshMethod = sourceItem.EdbRefreshMethod
	}

	return
}

// AddEdbSource
// @Description: 添加一个新的数据源
// @param item
// @return err
func AddEdbSource(item *EdbSource, indexNamePrefix string) (err error) {
	o := global.DmSQL["data"].Begin()
	if err != nil {
		return
	}
	defer func() {
		if err != nil {
			_ = o.Rollback()
			return
		}
		_ = o.Commit()
	}()

	indexName1 := fmt.Sprintf(`INDEX_%s_EDB_CODE`, indexNamePrefix)
	indexName2 := fmt.Sprintf(`INDEX_%s_EDB_INFO_ID`, indexNamePrefix)
	sqlStatements := []string{
		fmt.Sprintf(`CREATE TABLE "%s"
(
"edb_data_id" INT IDENTITY(1, 1) NOT NULL,
"edb_info_id" INT,
"edb_code" VARCHAR(50),
"data_time" DATE,
"value" DOUBLE,
"create_time" TIMESTAMP(0),
"modify_time" TIMESTAMP(0),
"data_timestamp" BIGINT DEFAULT 0,
NOT CLUSTER PRIMARY KEY("edb_data_id"),
UNIQUE("edb_code", "data_time")) STORAGE(ON "MAIN", CLUSTERBTR) ;
`, item.TableName),
		fmt.Sprintf(`COMMENT ON COLUMN "%s"."create_time" IS '创建时间';`, item.TableName),
		fmt.Sprintf(`COMMENT ON COLUMN "%s"."data_time" IS '数据日期';`, item.TableName),
		fmt.Sprintf(`COMMENT ON COLUMN "%s"."data_timestamp" IS '数据日期时间戳';`, item.TableName),
		fmt.Sprintf(`COMMENT ON COLUMN "%s"."edb_code" IS '指标编码';`, item.TableName),
		fmt.Sprintf(`COMMENT ON COLUMN "%s"."edb_info_id" IS '指标id';`, item.TableName),
		fmt.Sprintf(`COMMENT ON COLUMN "%s"."modify_time" IS '修改时间';`, item.TableName),
		fmt.Sprintf(`COMMENT ON COLUMN "%s"."value" IS '数据值';`, item.TableName),

		fmt.Sprintf(`CREATE OR REPLACE  INDEX "%s" ON "%s"("edb_code" ASC) STORAGE(ON "MAIN", CLUSTERBTR) ;`, indexName1, item.TableName),
		fmt.Sprintf(`CREATE OR REPLACE  INDEX "%s" ON "%s"("edb_info_id" ASC) STORAGE(ON "MAIN", CLUSTERBTR) ;`, indexName2, item.TableName),
	}

	// 创建表和索引
	for _, sql := range sqlStatements {
		err = o.Exec(sql).Error
		if err != nil {
			return
		}
	}

	// 添加来源
	err = o.Create(item).Error
	if err != nil {
		return
	}

	return
}