eta_index_lib/models/base_from_kpler.go

package models

import (
	sql2 "database/sql"
	"eta/eta_index_lib/global"
	"eta/eta_index_lib/utils"
	"fmt"
	"strconv"
	"strings"
	"time"

	"gorm.io/gorm"
)

// BaseFromKpler Kpler数据
type BaseFromKpler struct{}

type BaseFromKplerData struct {
	BaseFromKplerDataId int `gorm:"column:base_from_kpler_data_id;primaryKey"`
	//BaseFromKplerDataId  int `orm:"column(base_from_kpler_data_id);pk"`
	BaseFromKplerIndexId int
	IndexCode            string
	DataTime             string
	Value                string
	CreateTime           time.Time
	ModifyTime           time.Time
	DataTimestamp        int64
}

func (e *BaseFromKplerData) AfterFind(db *gorm.DB) (err error) {
	e.DataTime = utils.GormDateStrToDateStr(e.DataTime)

	return
}

func GetBaseFromKplerDataByCondition(condition string, pars []interface{}) (list []*BaseFromKplerData, err error) {
	//o := orm.NewOrm()
	sql := `SELECT * FROM base_from_kpler_data WHERE 1=1 `
	if condition != "" {
		sql += condition
	}
	//_, err = o.Raw(sql, pars).QueryRows(&list)
	err = global.DEFAULT_DB.Raw(sql, pars...).Find(&list).Error
	return
}

// Add 添加
func (obj BaseFromKpler) Add(edbCode string) (err error) {
	//o := orm.NewOrm()

	var condition string
	var pars []interface{}
	if edbCode != "" {
		condition += " AND index_code=? "
		pars = append(pars, edbCode)
	}
	KplerBaseDataAll, err := GetBaseFromKplerDataByCondition(condition, pars)
	if err != nil && !utils.IsErrNoRow(err) {
		return
	}
	var isAdd bool
	addSql := ` INSERT INTO edb_data_kpler(edb_info_id,edb_code,data_time,value,create_time,modify_time,data_timestamp) values `
	existMap := make(map[string]string)
	for _, sv := range KplerBaseDataAll {
		eDate := sv.DataTime
		dataTime, err := time.Parse(utils.FormatDate, eDate)
		if err != nil {
			return err
		}
		timestamp := dataTime.UnixNano() / 1e6
		timeStr := fmt.Sprintf("%d", timestamp)
		if _, ok := existMap[eDate]; !ok {
			addSql += GetAddSql("0", edbCode, eDate, timeStr, sv.Value)
			isAdd = true
		}
		existMap[eDate] = sv.Value
	}
	if isAdd {
		addSql = strings.TrimRight(addSql, ",")
		utils.FileLog.Info("addSql:" + addSql)
		//_,err=o.Raw(addSql).Exec()
		err = global.DEFAULT_DB.Exec(addSql).Error
		if err != nil {
			return err
		}
	}
	return
}

// Refresh 刷新Kpler指标数据
func (obj BaseFromKpler) Refresh(edbInfoId int, edbCode, startDate string) (err error) {
	source := obj.GetSource()
	//o := orm.NewOrm()
	if err != nil {
		return
	}
	edbInfoIdStr := strconv.Itoa(edbInfoId)
	//计算数据
	var condition string
	var pars []interface{}

	if edbCode != "" {
		condition += " AND index_code=? "
		pars = append(pars, edbCode)
	}

	if startDate != "" {
		condition += " AND data_time>=? "
		pars = append(pars, startDate)
	}

	KplerDataList, err := GetBaseFromKplerDataByCondition(condition, pars)
	if err != nil {
		return
	}

	// 真实数据的最大日期  ， 插入规则配置的日期
	var realDataMaxDate, edbDataInsertConfigDate time.Time
	var edbDataInsertConfig *EdbDataInsertConfig
	var isFindConfigDateRealData bool //是否找到配置日期的实际数据的值
	{
		edbDataInsertConfig, err = GetEdbDataInsertConfigByEdbId(edbInfoId)
		if err != nil && !utils.IsErrNoRow(err) {
			return
		}
		if edbDataInsertConfig != nil {
			edbDataInsertConfigDate = edbDataInsertConfig.Date
		}
	}

	var existCondition string
	var existPars []interface{}

	existCondition += " AND edb_info_id=? "
	existPars = append(existPars, edbInfoId)
	if startDate != "" {
		existCondition += " AND data_time>=? "
		existPars = append(existPars, startDate)
	}

	existList, err := GetEdbDataByCondition(source, 0, existCondition, existPars)
	if err != nil {
		return err
	}
	existMap := make(map[string]*EdbInfoSearchData)
	for _, v := range existList {
		existMap[v.DataTime] = v
	}
	addSql := ` INSERT INTO edb_data_kpler(edb_info_id,edb_code,data_time,value,create_time,modify_time,data_timestamp) values `
	var isAdd bool
	for _, v := range KplerDataList {
		item := v
		eDate := item.DataTime
		dataTime, err := time.ParseInLocation(utils.FormatDate, eDate, time.Local)
		if err != nil {
			return err
		}
		if findItem, ok := existMap[v.DataTime]; !ok {
			sValue := item.Value

			timestamp := dataTime.UnixNano() / 1e6
			timeStr := fmt.Sprintf("%d", timestamp)

			addSql += GetAddSql(edbInfoIdStr, edbCode, eDate, timeStr, sValue)
			isAdd = true
		} else {
			if findItem != nil && utils.SubFloatToString(findItem.Value, 30) != item.Value {
				err = ModifyEdbDataById(source, 0, findItem.EdbDataId, item.Value)
				if err != nil {
					return err
				}
			}
		}

		// 下面代码主要目的是处理掉手动插入的数据判断
		{
			if realDataMaxDate.IsZero() || dataTime.After(realDataMaxDate) {
				realDataMaxDate = dataTime
			}
			if edbDataInsertConfigDate.IsZero() || dataTime.Equal(edbDataInsertConfigDate) {
				isFindConfigDateRealData = true
			}
		}
	}

	// 处理手工数据补充的配置
	HandleConfigInsertEdbData(realDataMaxDate, edbDataInsertConfig, edbInfoId, source, 0, existMap, isFindConfigDateRealData)

	if isAdd {
		addSql = strings.TrimRight(addSql, ",")
		//_,err = o.Raw(assSql).Exec()
		err = global.DEFAULT_DB.Exec(addSql).Error
		if err != nil {
			fmt.Println("RefreshEdbDataFromKpler add Err", err.Error())
			return
		}
	}
	return
}

// GetSource 获取来源编码id
func (obj BaseFromKpler) GetSource() int {
	return utils.DATA_SOURCE_KPLER
}

// GetSourceName 获取来源名称
func (obj BaseFromKpler) GetSourceName() string {
	return utils.DATA_SOURCE_NAME_KPLER
}

type BaseFromKplerIndex struct {
	BaseFromKplerIndexId int `gorm:"column:base_from_kpler_index_id;primaryKey"`
	//BaseFromKplerIndexId int `orm:"column(base_from_kpler_index_id);pk"`
	ClassifyId    int
	IndexCode     string
	IndexName     string
	Frequency     string
	Unit          string
	Sort          int
	StartDate     string `description:"开始日期"`
	EndDate       string `description:"结束日期"`
	EndValue      float64
	CreateTime    time.Time
	ModifyTime    time.Time
	BaseFileName  string `description:"文件目录"`
	TerminalCode  string `description:"所属终端编码"`
	ApiQueryUrl   string `description:"API查询URL"`
	ExcelQueryUrl string `description:"Excel查询URL"`
	ProductNames  string `description:"产品名称"`
	FromZoneId    int    `description:"区域ID"`
	FromZoneName  string `description:"区域名称"`
	ToZoneId      int    `description:"区域ID"`
	ToZoneName    string `description:"区域名称"`
	FlowDirection string `description:"流向"`
	Granularity   string `description:"粒度"`
	Split         string `description:"拆分类型"`
}

func (e *BaseFromKplerIndex) AfterFind(db *gorm.DB) (err error) {
	e.StartDate = utils.GormDateStrToDateStr(e.StartDate)
	e.EndDate = utils.GormDateStrToDateStr(e.EndDate)

	return
}

type BaseFromKplerIndexList struct {
	BaseFromKplerIndexId int `gorm:"column:base_from_kpler_index_id;primaryKey"`
	//BaseFromKplerIndexId int `orm:"column(base_from_kpler_index_id);pk"`
	IndexCode     string
	IndexName     string
	Frequency     string
	Unit          string
	Sort          int
	ClassifyId    int
	StartDate     string
	EndDate       string
	TerminalCode  string
	CreateTime    string
	ModifyTime    string
	ProductNames  string `description:"产品名称"`
	FromZoneId    int    `description:"区域ID"`
	FromZoneName  string `description:"区域名称"`
	ToZoneId      int    `description:"区域ID"`
	ToZoneName    string `description:"区域名称"`
	FlowDirection string `description:"流向"`
	Granularity   string `description:"粒度"`
	Split         string `description:"拆分类型"`
}

// AfterFind 在该模型上设置钩子函数，把日期转成正确的string，所以查询函数只能用Find函数，First或者Scan是不会触发该函数的来获取数据
func (m *BaseFromKplerIndexList) AfterFind(db *gorm.DB) (err error) {
	m.StartDate = utils.GormDateStrToDateStr(m.StartDate)
	m.EndDate = utils.GormDateStrToDateStr(m.EndDate)

	return
}

func (y *BaseFromKplerIndex) GetByIndexCodes(indexCodes []string) (list []*BaseFromKplerIndex, err error) {
	sql := ` SELECT * FROM base_from_kpler_index WHERE index_code in (?) `
	err = global.DEFAULT_DB.Raw(sql, indexCodes).Find(&list).Error
	return
}

func (y *BaseFromKplerData) GetByIndexCode(indexCode string) (list []*BaseFromKplerData, err error) {
	//o := orm.NewOrm()
	sql := ` SELECT * FROM base_from_kpler_data WHERE index_code=? `
	//_, err = o.Raw(sql, indexCode).QueryRows(&list)
	err = global.DEFAULT_DB.Raw(sql, indexCode).Find(&list).Error
	return
}

func (y *BaseFromKplerData) AddMulti(item []*BaseFromKplerData) (err error) {
	//o := orm.NewOrm()
	//_, err = o.InsertMulti(1, item)
	err = global.DEFAULT_DB.CreateInBatches(&item, 1).Error
	return
}

// Update 修改
func (y *BaseFromKplerData) Update(updateCols []string) (err error) {
	//o := orm.NewOrm()
	//_, err = o.Update(y, updateCols...)
	err = global.DEFAULT_DB.Model(&y).Select(updateCols).Updates(&y).Error
	return
}

// HandleKplerExcelData Kpler的excel数据
type HandleKplerExcelData struct {
	ClassifyName       string `description:"指标目录"`
	ParentClassifyName string `description:"父级指标目录"`
	ClassifySort       int    `description:"指标目录排序号"`
	IndexName          string `description:"指标名称"`
	IndexCode          string `description:"指标编码"`
	Unit               string `description:"单位"`
	Sort               int    `description:"排序号"`
	Frequency          string `description:"频度"`
	ProductNames       string `description:"产品名称"`
	FromZoneId         int    `description:"区域ID"`
	FromZoneName       string `description:"区域名称"`
	ToZoneId           int    `description:"区域ID"`
	ToZoneName         string `description:"区域名称"`
	FlowDirection      string `description:"流向"`
	Granularity        string `description:"粒度"`
	Split              string `description:"拆分类型"`
	SplitName          string `description:"拆分类型名称"`
	ExcelQueryUrl      string `description:"Excel查询URL"`
	ExcelDataMap       map[string]string
}

type HandleKplerExcelDataReq struct {
	List         []*HandleKplerExcelData
	TerminalCode string `description:"编码"`
}

func (y *BaseFromKplerData) GetMaxAndMinDateByIndexCode(indexCode string) (item *EdbInfoMaxAndMinInfo, err error) {
	//o := orm.NewOrm()
	sql := ` SELECT MIN(data_time) AS min_date,MAX(data_time) AS max_date,MIN(value) AS min_value,MAX(value) AS max_value FROM base_from_kpler_data WHERE index_code=? `
	//err = o.Raw(sql, indexCode).QueryRow(&item)
	err = global.DEFAULT_DB.Raw(sql, indexCode).First(&item).Error
	sql = ` SELECT value AS latest_value FROM %s WHERE index_code=? ORDER BY data_time DESC LIMIT 1 `
	sql = fmt.Sprintf(sql, "base_from_kpler_data")
	//err = o.Raw(sql, indexCode).QueryRow(&latest_value)
	var latestValueNull sql2.NullFloat64
	err = global.DEFAULT_DB.Raw(sql, indexCode).Scan(&latestValueNull).Error
	if err == nil && latestValueNull.Valid {
		item.LatestValue = latestValueNull.Float64
	}

	return
}

func (y *BaseFromKplerIndex) ModifyIndexMaxAndMinDate(indexCode string, item *EdbInfoMaxAndMinInfo) (err error) {
	//o := orm.NewOrm()
	sql := ` UPDATE base_from_kpler_index SET start_date=?,end_date=?, end_value=?, modify_time=NOW() WHERE index_code=? `
	//_, err = o.Raw(sql, item.MinDate, item.MaxDate, item.LatestValue, indexCode).Exec()
	err = global.DEFAULT_DB.Exec(sql, item.MinDate, item.MaxDate, item.LatestValue, indexCode).Error
	return
}

func (y *BaseFromKplerIndex) GetByIndexCode(indexCode string) (item *BaseFromKplerIndex, err error) {
	//o := orm.NewOrm()
	sql := ` SELECT * FROM base_from_kpler_index WHERE index_code=? `
	//err = o.Raw(sql, indexCode).QueryRow(&item)
	err = global.DEFAULT_DB.Raw(sql, indexCode).First(&item).Error
	return
}

func (y *BaseFromKplerIndex) Add() (err error) {
	//o := orm.NewOrm()
	//lastId, err = o.Insert(y)
	err = global.DEFAULT_DB.Create(&y).Error
	if err != nil {
		return
	}
	return
}

// Update 修改
func (y *BaseFromKplerIndex) Update(updateCols []string) (err error) {
	//o := orm.NewOrm()
	//_, err = o.Update(y, updateCols...)
	err = global.DEFAULT_DB.Model(&y).Select(updateCols).Updates(&y).Error
	return
}

func (y *BaseFromKplerClassify) Add() (lastId int64, err error) {
	//o := orm.NewOrm()
	//lastId, err = o.Insert(y)
	err = global.DEFAULT_DB.Create(&y).Error
	if err != nil {
		return
	}
	lastId = int64(y.ClassifyId)
	return
}

// Update 修改
func (y *BaseFromKplerClassify) Update(updateCols []string) (err error) {
	//o := orm.NewOrm()
	//_, err = o.Update(y, updateCols...)
	err = global.DEFAULT_DB.Model(&y).Select(updateCols).Updates(&y).Error
	return
}

func (y *BaseFromKplerClassify) GetByClassifyName(classifyName string) (item *BaseFromKplerClassify, err error) {
	//o := orm.NewOrm()
	sql := ` SELECT * FROM base_from_kpler_classify WHERE classify_name=? `
	//err = o.Raw(sql, classifyName).QueryRow(&item)
	err = global.DEFAULT_DB.Raw(sql, classifyName).First(&item).Error
	return
}

func (y *BaseFromKplerClassify) GetParentClassify() (items []*BaseFromKplerClassify, err error) {
	//o := orm.NewOrm()
	sql := ` SELECT * FROM base_from_kpler_classify WHERE parent_id=0 `
	//_, err = o.Raw(sql).QueryRows(&items)
	err = global.DEFAULT_DB.Raw(sql).Find(&items).Error
	return
}

type KplerSearchEdbReq struct {
	ProductNames string `description:"产品名称"`
	FromZoneIds []int `description:"来源区域ID，对应Location" ` 
	FromZoneNames string `description:"来源区域名称"`
	ToZoneIds []int `description:"流向区域ID"`
	ToZoneNames string `description:"流向区域名称"`
	Split string `description:"拆分类型"`
	FlowDirection string `description:"流向，对应periodicity：export/import"`
	Granularity string `description:"粒度: daily/weekly/monthly/yearly"`
	Unit string `description:"单位"`
}

  // withIntraCountry := false
  // withIntraRegion := true
  // withForecast := true
  // withFreightView := false
  // withProductEstimation := false
type KplerFlowDataLibReq struct {
	Products string `description:"产品名称"`
	FromZones string `description:"来源区域ID，对应Location" ` 
	ToZones string `description:"流向区域ID"`
	Split string `description:"拆分类型"`
	FlowDirection string `description:"流向，对应periodicity：export/import"`
	Granularity string `description:"粒度: daily/weekly/monthly/yearly"`
	Unit string `description:"单位"`
	FromCountries string `description:"来源国家"`
	ToCountries string `description:"流向国家"`
	FromInstallations string `description:"来源安装"`
	ToInstallations string `description:"流向安装"`
	OnlyRealized string `description:"是否只查询已实现数据"`
	VesselTypes string `description:"船型"`
	VesselTypesAlt string `description:"船型"`
    WithIntraCountry string `description:"是否查询同国数据"`
    WithIntraRegion string `description:"是否查询同区域数据"`
    WithForecast string `description:"是否查询预测数据"`
    WithFreightView string `description:"是否查询运费数据"`
    WithProductEstimation string `description:"是否查询产品估算数据"`
    StartDate string `description:"开始日期"`
    EndDate string `description:"结束日期"`
}


type KplerFlowDataLibResp struct {
	Ret         int
	Msg         string
	ErrMsg      string
	ErrCode     string
	Data KplerFlowDataResp 
}

type KplerFlowDataResp struct {
	ApiQueryUrl string
	List []KplerFlowData 
}
type KplerFlowData struct {
	SplitItem string 
	IndexData []KplerIndexDataItem 
}

type KplerIndexItem struct {
	IndexCode string 
	IndexName string 
	Unit string 
	Frequency string 
	IndexData []KplerIndexDataItem 
}

type KplerIndexDataItem struct {
	DataTime string
	Value string 
}

// BaseFromKplerClassify Kpler原始数据分类表
type BaseFromKplerClassify struct {
	ClassifyId      int       `orm:"column(classify_id);pk" gorm:"primaryKey"`
	ClassifyName    string    `description:"分类名称"`
	ParentId        int       `description:"父级id"`
	SysUserId       int       `description:"创建人id"`
	SysUserRealName string    `description:"创建人姓名"`
	Level           int       `description:"层级"`
	Sort            int       `description:"排序字段，越小越靠前，默认值：10"`
	ModifyTime      time.Time `description:"修改时间"`
	CreateTime      time.Time `description:"创建时间"`
	ClassifyNameEn  string    `description:"英文分类名称"`
	ProductId       int       `description:"产品ID"`
	ProductName     string    `description:"产品名称"`
	AncestorFamilyId        int       `description:"父级产品ID"`
	AncestorFamilyName      string    `description:"父级产品名称"`
	AncestorGroupId         int       `description:"父级产品ID"`
	AncestorGroupName       string    `description:"父级产品名称"`
	AncestorProductId       int       `description:"父级产品ID"`
	AncestorProductName     string    `description:"父级产品名称"`
	AncestorGradeId         int       `description:"父级产品ID"`
	AncestorGradeName       string    `description:"父级产品名称"`
	ClassifyType    string    `description:"分类类型"`
}

func GetAllBaseFromKplerClassify() (items []*BaseFromKplerClassify, err error) {
	sql := ` SELECT * FROM base_from_kpler_classify `
	err = global.DEFAULT_DB.Raw(sql).Find(&items).Error
	return
}

// 获取分类
func (y *BaseFromKplerClassify) GetByProductIds(productIds []int) (items []*BaseFromKplerClassify, err error) {
	sql := ` SELECT * FROM base_from_kpler_classify WHERE product_id in (?) `
	err = global.DEFAULT_DB.Raw(sql, productIds).Find(&items).Error
	return
}

// 获取分类
func (y *BaseFromKplerClassify) GetByProductName(productName string) (item *BaseFromKplerClassify, err error) {
	sql := ` SELECT * FROM base_from_kpler_classify WHERE product_name=? `
	err = global.DEFAULT_DB.Raw(sql, productName).First(&item).Error
	return
}

// 获取区域
// BaseFromKplerZone Kpler区域表
type BaseFromKplerZone struct {
	BaseFromKplerZoneId int       `orm:"column(base_from_kpler_zone_id);pk" gorm:"primaryKey"`
	ZoneName            string    `description:"区域名称"`
	ZoneType            string    `description:"区域类型"`
	ParentId            int       `description:"区域父ID"`
	AncestorId          int       `description:"区域祖先ID"`
	AncestorType        string    `description:"区域祖先类型"`
	AncestorName        string    `description:"区域祖先名称"`
	DescendantId        int       `description:"区域子ID"`
	DescendantName      string    `description:"区域子名称"`
	CreateTime          time.Time `description:"创建时间"`
	ModifyTime          time.Time `description:"修改时间"`
} 

// 获取区域
func (y *BaseFromKplerZone) GetByZoneIds(zoneIds []int) (items []*BaseFromKplerZone, err error) {
	sql := ` SELECT * FROM base_from_kpler_zone WHERE base_from_kpler_zone_id in (?) `
	err = global.DEFAULT_DB.Raw(sql, zoneIds).Find(&items).Error
	return
}

func (y *BaseFromKplerZone) Add() (lastId int64, err error) {
	err = global.DEFAULT_DB.Create(&y).Error
	if err != nil {
		return
	}
	lastId = int64(y.BaseFromKplerZoneId)
	return
}

type KplerProductLibResp struct {
	Ret         int
	Msg         string
	ErrMsg      string
	ErrCode     string
	Data []KplerProduct 
}

//bodystr := "Id (Product);Name;Type (Product);Family;Family Id;Group;Group Id;Product;Product Id;Grade;Grade Id;Density (Product);Density Unit;Energy Density;Energy Density Unit;Expansion Ratio
type KplerProduct struct {
	Id string
	Name string
	Type string
	Family string
	FamilyId string
	Group string
	GroupId string
	Product string
	ProductId string
	Grade string
	GradeId string
	Density string
	DensityUnit string
	EnergyDensity string
	EnergyDensityUnit string
	ExpansionRatio string
}

type KplerZoneLibResp struct {
	Ret         int
	Msg         string
	ErrMsg      string
	ErrCode     string
	Data []KplerZone 
}

// Ancestor Id;Ancestor Name;Ancestor Type;Descendant Id;Descendant Name;Descendant Type
type KplerZone struct {
	AncestorId string
	AncestorName string
	AncestorType string
	DescendantId string
	DescendantName string
}

type KplerZoneDataLibReq struct {
	AncestorName string `description:"祖先名称"`
	DescendantName string `description:"子名称"`
}

type KplerRefreshByApiReq struct {
	Frequency string `description:"频度"`
}

func GetApiKplerIndexesByFrequencyEndDateTime(frequency string, endDateTime string) (items []*BaseFromKplerIndex, err error) {
	sql := ` SELECT * FROM base_from_kpler_index WHERE frequency=? AND end_date < ? and api_query_url != '' order by base_from_kpler_index_id asc, api_query_url asc `
	err = global.DEFAULT_DB.Raw(sql, frequency, endDateTime).Find(&items).Error
	return
}